diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index c7674793e..cc6dce90b 100644 --- a/docs/drivers/Galaxy.md +++ b/docs/drivers/Galaxy.md @@ -172,7 +172,7 @@ Every (re)open of `StreamAlarms` replays the gateway's active-alarm snapshot. `G - **Replay gaps** (`galaxy.events.replay_gaps`): the gateway signals that events were evicted from its replay ring before we could read them. The driver responds by re-issuing `SubscribeBulk` for every tracked subscription — MXAccess delivers an initial value per item on subscribe, so re-subscribing *is* the re-snapshot the contract asks for. It does **not** restart the event pump (the stream carrying the sentinel is healthy) and it honours `Reconnect:ReplayOnSessionLost=false` by logging and standing down, on the grounds that an operator who delegated recovery to the gateway did not ask for a re-subscribe on a different trigger. ⚠️ This cannot currently fire: the sentinel is only emitted on a stream opened with `after_worker_sequence > 0`, and the driver opens with `0`, which the gateway treats as "fresh stream, not a resume". Enabling resume is a bigger change than it looks — the sequence is only meaningful **within one gateway session**, and the reconnect path recreates the session, so a remembered sequence replayed against a fresh session would set the server's live watermark *above* every event it is about to send, silencing the stream rather than recovering it. - **Undecodable values** (`galaxy.events.undecodable`): a value kind this driver has no CLR representation for is published as **`BadTypeMismatch` with a null value**, and the first occurrence of each distinct kind is logged. It is *not* published as a null carrying the event's own quality: an MXAccess event can be Good-quality while the payload is one we cannot map, and a good null is indistinguishable from a genuine null reading — a client would consume a value it never received as a real one, with the failure visible only in a log. Non-zero here means the gateway is sending a wire type `MxValueDecoder` has no mapping for and those readings are being lost; the fix is a decoder mapping, not a quality change. -- **Initial values can overtake the subscribe reply (#859)** (`galaxy.events.early_parked` / `early_replayed` / `early_overflow`): mxaccessgw gives no reply-before-event ordering — the worker frames the SubscribeBulk reply on a thread-pool continuation while the STA is already pumping the OnDataChange that Advise queued, and the reply (unary Invoke) and the event (StreamEvents) are separate HTTP/2 streams anyway. An event for an unregistered handle is therefore only "stale after unsubscribe" when no subscribe is in flight. Every SubscribeBulk call site (`SubscribeAsync`, the reconnect/replay-gap `ResubscribeAllAsync`, `PerPlatformProbeWatcher`) opens `SubscriptionRegistry.BeginPendingSubscribe()` before the RPC and closes it after registering; inside the window unknown-handle events are parked (bounded at 100 000), promoted by the `Register`/`Rebind` that binds the handle, and replayed on the pump's dispatch thread before any later event for that handle. When the last window closes, unclaimed parked events are discarded — the stale drop is unchanged outside it. +- **Initial values can overtake the subscribe reply (#859)** (`galaxy.events.early_parked` / `early_replayed` / `early_overflow`): mxaccessgw gives no reply-before-event ordering — the worker frames the SubscribeBulk reply on a thread-pool continuation while the STA is already pumping the OnDataChange that Advise queued, and the reply (unary Invoke) and the event (StreamEvents) are separate HTTP/2 streams anyway. An event for an unregistered handle is therefore only "stale after unsubscribe" when no subscribe is in flight. Every SubscribeBulk call site (`SubscribeAsync`, the subscribe-once `ReadAsync` — which waits only for the initial value and would otherwise read `BadTimeout` — the reconnect/replay-gap `ResubscribeAllAsync` (one window per subscription, not across the whole replay), `PerPlatformProbeWatcher`) opens `SubscriptionRegistry.BeginPendingSubscribe()` before the RPC and closes it right after registering, never later; inside the window unknown-handle events are parked, promoted by the `Register`/`Rebind` that binds the handle — map writes and promotion are one critical section, so no later event can resolve in between — and replayed on the pump's dispatch thread before any later event for that handle. Held events (parked + promoted-but-unreplayed) are bounded at 100 000 together. When the last window closes, unclaimed parked events are discarded — the stale drop is unchanged outside it. Item handles are per gateway session and reused by the next one, so held events carry the pump's **epoch**: each new `EventPump` (one per session) discards everything held and fences the old pump, so an old-session value can never land on the tag that now owns a reused handle number. - **Duplicate `attributeRef` (#605 / 06-S-13)**: two authored RawPaths dialling the same Galaxy attribute are both dialled, but the reverse map is first-wins, so value fan-out and alarm `ConditionId` surface under the first RawPath only. The loser is logged at Warning and recorded in the skipped-tag ledger with the winner path in the reason. - **Subscribe-once reads are bounded (#605 / 06-S-3)**: `ReadAsync` on the production path (SubscribeBulk, first event, Unsubscribe) carries an internal 30 s deadline in addition to the caller token; slots that never publish come back `BadTimeout`, so `CancellationToken.None` can no longer hang on a silent tag. - **Supervisory advise is retried after a thrown failure (#605 / 06-S-12)**: the writer marks an item handle advised before the call; a thrown advise (not just a non-OK status) now unmarks it, fails the write closed and re-advises on the next write. diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs index 7263d5c45..05f7f9c12 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -996,10 +996,6 @@ public sealed class GalaxyDriver CancellationToken cancellationToken) { var tagCount = 0; - - // #859: the re-issued handles' initial values (the whole point of a re-snapshot) can - // overtake each SubscribeBulk reply; park them until the matching Rebind promotes them. - using var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); foreach (var (subscriptionId, oldBindings) in entries) { if (Volatile.Read(ref _runtimeGeneration) != generation) @@ -1023,6 +1019,12 @@ public sealed class GalaxyDriver .ToArray(); if (refs.Length == 0) continue; + // #859: the re-issued handles' initial values (the whole point of a re-snapshot) can + // overtake this SubscribeBulk reply; park them until the Rebind below promotes them. + // Scoped to this one iteration (the using is disposed at the end of the loop body): a + // window held across all N round-trips would park every unrelated unknown-handle event + // for the whole replay and could exhaust the bound before the genuine early values. + using var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); var results = await _subscriber! .SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -1457,75 +1459,89 @@ public sealed class GalaxyDriver var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs; IReadOnlyList results; - // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow - // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the - // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call. - var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken); + List bindings; + + // #859: a read waits ONLY for the initial value, and that value can overtake the SubscribeBulk + // reply — dropped as stale, a slow-changing attribute would read BadTimeout. Open the early-event + // window before the RPC and close it right after registration (the finally below), not after the + // value wait: a window held for the whole wait would park every unrelated unknown-handle event. + var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); try { - results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false); - } - catch (OperationCanceledException) when (!callerToken.IsCancellationRequested) - { - // Internal deadline fired while the subscribe RPC was still in flight. - pump.OnDataChange -= handler; - var subscriber = readSubscriber; - var logger = _logger; - _ = subscribeTask.ContinueWith(t => + // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow + // gateway can never surface an OperationCanceledException to a caller that did not cancel, and the + // handles it eventually creates are known (and unsubscribed below) rather than lost to a cancelled call. + var subscribeTask = readSubscriber.SubscribeBulkAsync(fullReferences, bufferedIntervalMs, callerToken); + try { - if (!t.IsCompletedSuccessfully) return Task.CompletedTask; - var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray(); - return late.Length == 0 - ? Task.CompletedTask - : subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith( - u => logger.LogWarning(u.Exception, - "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length), - TaskContinuationOptions.OnlyOnFaulted); - }, TaskScheduler.Default).Unwrap(); - var now = DateTime.UtcNow; - var timedOut = new DataValueSnapshot[fullReferences.Count]; - for (var i = 0; i < timedOut.Length; i++) - timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now); - return timedOut; - } - catch - { - pump.OnDataChange -= handler; - throw; - } - - // Register bindings so the pump knows to dispatch events for these handles. - // Index the SubscribeBulk results once and correlate to references in O(1) - // instead of FirstOrDefault per element (O(n²) over the batch). On the - // 50k-tag soak path this turns a 2.5G-comparison loop into a single - // Dictionary build + linear scan. - var resultIndex = BuildResultIndex(results); - var bindings = new List(fullReferences.Count); - for (var i = 0; i < fullReferences.Count; i++) - { - var fullRef = fullReferences[i]; - var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true } - ? match.ItemHandle - : 0; - bindings.Add(new TagBinding(fullRef, itemHandle)); - - // Tags the gw rejected up front — complete with Bad status now so the - // wait below doesn't time out on them. - if (itemHandle <= 0 - && pendingByRef.TryGetValue(fullRef, out var rejectedTcs)) + results = await subscribeTask.WaitAsync(cancellationToken).ConfigureAwait(false); + } + catch (OperationCanceledException) when (!callerToken.IsCancellationRequested) { - rejectedTcs.TrySetResult(new DataValueSnapshot( - Value: null, - StatusCode: StatusCodeMap.Bad, - SourceTimestampUtc: null, - ServerTimestampUtc: DateTime.UtcNow)); + // Internal deadline fired while the subscribe RPC was still in flight. + pump.OnDataChange -= handler; + var subscriber = readSubscriber; + var logger = _logger; + _ = subscribeTask.ContinueWith(t => + { + if (!t.IsCompletedSuccessfully) return Task.CompletedTask; + var late = t.Result.Where(r => r.WasSuccessful && r.ItemHandle > 0).Select(r => r.ItemHandle).ToArray(); + return late.Length == 0 + ? Task.CompletedTask + : subscriber.UnsubscribeBulkAsync(late, CancellationToken.None).ContinueWith( + u => logger.LogWarning(u.Exception, + "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length), + TaskContinuationOptions.OnlyOnFaulted); + }, TaskScheduler.Default).Unwrap(); + var now = DateTime.UtcNow; + var timedOut = new DataValueSnapshot[fullReferences.Count]; + for (var i = 0; i < timedOut.Length; i++) + timedOut[i] = new DataValueSnapshot(null, StatusCodeMap.BadTimeout, null, now); + return timedOut; + } + catch + { + pump.OnDataChange -= handler; + throw; + } + + // Register bindings so the pump knows to dispatch events for these handles. + // Index the SubscribeBulk results once and correlate to references in O(1) + // instead of FirstOrDefault per element (O(n²) over the batch). On the + // 50k-tag soak path this turns a 2.5G-comparison loop into a single + // Dictionary build + linear scan. + var resultIndex = BuildResultIndex(results); + bindings = new List(fullReferences.Count); + for (var i = 0; i < fullReferences.Count; i++) + { + var fullRef = fullReferences[i]; + var itemHandle = resultIndex.TryGetValue(fullRef, out var match) && match is { WasSuccessful: true } + ? match.ItemHandle + : 0; + bindings.Add(new TagBinding(fullRef, itemHandle)); + + // Tags the gw rejected up front — complete with Bad status now so the + // wait below doesn't time out on them. + if (itemHandle <= 0 + && pendingByRef.TryGetValue(fullRef, out var rejectedTcs)) + { + rejectedTcs.TrySetResult(new DataValueSnapshot( + Value: null, + StatusCode: StatusCodeMap.Bad, + SourceTimestampUtc: null, + ServerTimestampUtc: DateTime.UtcNow)); + } + } + if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) + { + pump.OnDataChange -= handler; + await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false); + throw TornDownDuringSubscribe(); } } - if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) + finally { - pump.OnDataChange -= handler; - await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false); - throw TornDownDuringSubscribe(); + earlyEventWindow.Dispose(); } try diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs index 701ab73a5..a7d5388b8 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/EventPump.cs @@ -138,9 +138,16 @@ internal sealed class EventPump : IAsyncDisposable SingleWriter = false, }); _clientTag = new KeyValuePair("galaxy.client", clientName ?? ""); + + // One pump per gateway session: a new epoch discards whatever an earlier session left held + // (#859 review — item handles are reused across sessions), and this pump never parks or + // replays under any other epoch. + _epoch = _registry.StartNewEpoch(); _registry.ParkedEventsReady += WakeForReadyEvents; } + private readonly int _epoch; + /// /// Marker written into the channel to wake the dispatch loop after a registration promoted /// parked early events (#859). Never dispatched itself; compared by reference. @@ -149,9 +156,12 @@ internal sealed class EventPump : IAsyncDisposable /// /// Wakes the dispatch loop so promoted early events replay even if no further event arrives. + /// Runs on the registering thread, possibly under the driver's _registryGate or the probe + /// watcher's _syncLock — safe because the channel keeps the default + /// AllowSynchronousContinuations = false, so the write never runs the dispatch loop inline. /// A failed write is harmless: the channel is either full (the loop is busy and replays the - /// ready queue before its next data-change dispatch) or completed (the pump is dead, and the - /// next pump on this registry replays them). + /// ready queue before its next dispatch) or completed (the pump is dead; its held events + /// belong to a session the next pump's new epoch discards). /// private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel); @@ -353,11 +363,13 @@ internal sealed class EventPump : IAsyncDisposable private void DispatchDataChange(MxEvent ev) { - var outcome = _registry.ResolveOrPark(ev, out var subscribers); + var outcome = _registry.ResolveOrPark(ev, _epoch, out var subscribers); - // Replay promoted early events (#859) AFTER the resolve and BEFORE fanning this event out: - // any registration that let this event resolve has already moved its parked (older) events - // to the ready queue, so they reach the handler first and per-handle order is preserved. + // Replay promoted early events (#859) AFTER the resolve and BEFORE fanning this event out. + // A registration writes the reverse map and promotes its parked events in ONE critical + // section under the registry's park gate, and ResolveOrPark resolves under that gate + // whenever anything is held. So if this event resolved, every older parked event for its + // handle is already in the ready queue, and replaying it first preserves per-handle order. DispatchReadyEvents(); switch (outcome) @@ -386,7 +398,7 @@ internal sealed class EventPump : IAsyncDisposable /// private void DispatchReadyEvents() { - var ready = _registry.TakeReadyEvents(); + var ready = _registry.TakeReadyEvents(_epoch); if (ready.Count == 0) return; foreach (var early in ready) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs index 1bb5da9c4..ad65f06d3 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Runtime/SubscriptionRegistry.cs @@ -70,17 +70,27 @@ internal sealed class SubscriptionRegistry /// The tag bindings for the subscription. public void Register(long subscriptionId, IReadOnlyList bindings) { - var entry = new SubscriptionEntry(subscriptionId, bindings); - _bySubscriptionId[subscriptionId] = entry; - foreach (var binding in bindings) + bool promoted; + // The reverse-map writes and the early-event promotion are ONE critical section under the + // park gate (#859 review): a dispatch that resolves a handle through the map must never + // run between the write and the promotion, or a later live event would overtake the + // parked initial value. ResolveOrPark takes the same gate whenever anything is held. + lock (_parkGate) { - if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected - _subscribersByItemHandle.AddOrUpdate( - binding.ItemHandle, - _ => [subscriptionId], - (_, set) => set.Add(subscriptionId)); + var entry = new SubscriptionEntry(subscriptionId, bindings); + _bySubscriptionId[subscriptionId] = entry; + foreach (var binding in bindings) + { + if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected + _subscribersByItemHandle.AddOrUpdate( + binding.ItemHandle, + _ => [subscriptionId], + (_, set) => set.Add(subscriptionId)); + } + AfterMapWriteForTests?.Invoke(); + promoted = PromoteParkedEventsLocked(bindings); } - PromoteParkedEvents(bindings); + if (promoted) ParkedEventsReady?.Invoke(); } /// @@ -143,8 +153,8 @@ internal sealed class SubscriptionRegistry lock (_parkGate) { _parkedByItemHandle.Clear(); - _parkedCount = 0; _ready.Clear(); + Volatile.Write(ref _heldCount, 0); } } @@ -170,60 +180,110 @@ internal sealed class SubscriptionRegistry // the handle — so per-handle value order is preserved. When the last in-flight subscribe ends, // whatever is still parked belonged to no subscription and is discarded, so the genuine // stale-after-unsubscribe drop is unchanged outside the window. + // + // Item handles are only meaningful within one gateway session, and a new session reuses small + // integers. Every held event therefore belongs to an EPOCH: each EventPump (one per session — + // reconnect replay recreates it) starts a new epoch, which discards everything held, and an + // event parked or replayed by a pump of an older epoch is dropped. Otherwise an old-session + // value could be promoted onto whichever tag now owns the reused handle number. /// - /// Upper bound on events parked across all in-flight subscribes. Sized to hold the initial - /// values of a large bulk subscribe (MXAccess sends one per advised item); overflow is - /// dropped and counted by the pump, i.e. it degrades to the pre-#859 behaviour. + /// Upper bound on events held (parked + promoted-but-not-yet-replayed) across all in-flight + /// subscribes. Sized to hold the initial values of a large bulk subscribe (MXAccess sends + /// one per advised item); overflow is dropped and counted by the pump, i.e. it degrades to + /// the pre-#859 behaviour. /// public const int MaxParkedEvents = 100_000; private readonly Lock _parkGate = new(); private readonly Dictionary> _parkedByItemHandle = new(); private readonly Queue _ready = new(); - private int _parkedCount; + + // Parked + ready. Written only under _parkGate; read lock-free (Volatile) by the dispatch fast path. + private int _heldCount; + + // Open windows. Written only under _parkGate (so the 0→N / N→0 transitions and the discard of + // the previous window's leftovers are atomic); read lock-free by the dispatch fast path. private int _pendingSubscribes; + private int _epoch; /// - /// Raised (outside any lock) when a registration promoted parked events to the ready queue, - /// so the can wake its dispatch loop even if no further event arrives. + /// Test seam (#859 review): invoked inside the registration critical section, between the + /// reverse-map writes and the promotion of parked events. + /// + internal Action? AfterMapWriteForTests { get; set; } + + /// + /// Raised when a registration promoted parked events to the ready queue, so the + /// can wake its dispatch loop even if no further event arrives. + /// Raised after the park gate is released but still on the registering thread, which may hold + /// the driver's _registryGate or the probe watcher's _syncLock: a handler must + /// only signal (the pump's handler is a channel TryWrite, safe there because the + /// channel keeps the default AllowSynchronousContinuations = false, so the dispatch + /// loop never runs inline under those locks). /// public event Action? ParkedEventsReady; /// Gets the number of subscribes currently inside their reply window. public int PendingSubscribeCount => Volatile.Read(ref _pendingSubscribes); - /// Gets the number of events currently parked for not-yet-registered item handles. - public int ParkedEventCount + /// Gets the number of events currently held (parked or awaiting replay). + public int ParkedEventCount => Volatile.Read(ref _heldCount); + + /// + /// Starts a new session epoch: discards every held event and makes events from any earlier + /// epoch unparkable and unreplayable. Called by each new . + /// + /// The new epoch. + public int StartNewEpoch() { - get { lock (_parkGate) return _parkedCount; } + lock (_parkGate) + { + _parkedByItemHandle.Clear(); + _ready.Clear(); + Volatile.Write(ref _heldCount, 0); + return ++_epoch; + } } /// /// Opens the early-event window for one subscribe. Call BEFORE issuing SubscribeBulk - /// and dispose AFTER the returned handles are registered (or the attempt was abandoned): - /// while any window is open, events for unknown handles are parked rather than dropped. + /// and dispose AFTER the returned handles are registered (or the attempt was abandoned) — + /// not later: while any window is open, events for unknown handles are parked, not dropped. /// /// A token that closes the window when disposed (idempotent). public IDisposable BeginPendingSubscribe() { - Interlocked.Increment(ref _pendingSubscribes); + lock (_parkGate) + { + // 0→1: anything still parked belongs to an earlier, closed window. EndPendingSubscribe + // already discards on N→0 under this same gate, so this is belt-and-braces. + if (_pendingSubscribes == 0) DiscardParkedLocked(); + Volatile.Write(ref _pendingSubscribes, _pendingSubscribes + 1); + } return new PendingSubscribeScope(this); } private void EndPendingSubscribe() { - if (Interlocked.Decrement(ref _pendingSubscribes) != 0) return; lock (_parkGate) { - // Re-checked under the gate: a subscribe that began after the decrement may already - // have parked its own early events, which must survive. - if (Volatile.Read(ref _pendingSubscribes) != 0) return; - _parkedByItemHandle.Clear(); - _parkedCount = 0; + Volatile.Write(ref _pendingSubscribes, _pendingSubscribes - 1); + // N→0: what is still parked was claimed by no registration. Promoted (ready) events are + // kept — they belong to a registered handle and are awaiting replay. + if (_pendingSubscribes == 0) DiscardParkedLocked(); } } + private void DiscardParkedLocked() + { + if (_parkedByItemHandle.Count == 0) return; + var parked = 0; + foreach (var list in _parkedByItemHandle.Values) parked += list.Count; + _parkedByItemHandle.Clear(); + Volatile.Write(ref _heldCount, _heldCount - parked); + } + /// /// Resolve the subscribers for 's item handle, or — when the handle is /// unknown while a subscribe is in flight — park the event for replay. Atomic with the @@ -231,22 +291,49 @@ internal sealed class SubscriptionRegistry /// promotion already ran (it would otherwise be stranded until the window closed). /// /// The data-change event. + /// The calling pump's epoch (). /// The subscribers when the outcome is . /// What happened to the event. public EarlyEventOutcome ResolveOrPark( - MxEvent ev, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers) + MxEvent ev, int epoch, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers) { - subscribers = ResolveSubscribers(ev.ItemHandle); - if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; - if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; - - lock (_parkGate) + // Lock-free fast path ONLY while nothing is held: then no older event for this handle can be + // waiting, so resolving through the map cannot overtake one. While anything is held, the + // resolve happens under the gate, i.e. never inside a registration's map-write→promotion + // critical section. + if (Volatile.Read(ref _heldCount) == 0) { - // A registration may have bound the handle between the lookup above and the gate. subscribers = ResolveSubscribers(ev.ItemHandle); if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; if (Volatile.Read(ref _pendingSubscribes) == 0) return EarlyEventOutcome.Stale; - if (_parkedCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow; + } + + lock (_parkGate) + { + if (epoch != _epoch) + { + subscribers = []; + return EarlyEventOutcome.Stale; // a dying pump of an earlier session + } + + subscribers = ResolveSubscribers(ev.ItemHandle); + if (subscribers.Count > 0) + { + // Defensive: registration promotes atomically with its map writes, so a resolved + // handle should hold no parked list — but if one ever does, queue behind it rather + // than overtake it (promotion moves the whole list in order). Not bounded: dropping + // this newer value would leave the older parked one as the node's last word. + if (_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var pending)) + { + pending.Add(ev); + Volatile.Write(ref _heldCount, _heldCount + 1); + subscribers = []; + return EarlyEventOutcome.Parked; + } + return EarlyEventOutcome.Resolved; + } + if (_pendingSubscribes == 0) return EarlyEventOutcome.Stale; + if (_heldCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow; if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var list)) { @@ -254,43 +341,42 @@ internal sealed class SubscriptionRegistry _parkedByItemHandle[ev.ItemHandle] = list; } list.Add(ev); - _parkedCount++; + Volatile.Write(ref _heldCount, _heldCount + 1); return EarlyEventOutcome.Parked; } } /// /// Dequeue every event a registration promoted, in arrival order. Called only from the - /// EventPump's single dispatch thread. + /// EventPump's single dispatch thread. A pump of an earlier epoch gets nothing. /// + /// The calling pump's epoch. /// The promoted events; empty when none are waiting. - public IReadOnlyList TakeReadyEvents() + public IReadOnlyList TakeReadyEvents(int epoch) { + if (Volatile.Read(ref _heldCount) == 0) return []; lock (_parkGate) { - if (_ready.Count == 0) return []; + if (epoch != _epoch || _ready.Count == 0) return []; var taken = _ready.ToArray(); _ready.Clear(); + Volatile.Write(ref _heldCount, _heldCount - taken.Length); return taken; } } - private void PromoteParkedEvents(IReadOnlyList bindings) + private bool PromoteParkedEventsLocked(IReadOnlyList bindings) { + if (_parkedByItemHandle.Count == 0) return false; var promoted = false; - lock (_parkGate) + foreach (var binding in bindings) { - if (_parkedCount == 0) return; - foreach (var binding in bindings) - { - if (binding.ItemHandle <= 0) continue; - if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var list)) continue; - foreach (var ev in list) _ready.Enqueue(ev); - _parkedCount -= list.Count; - promoted = true; - } + if (binding.ItemHandle <= 0) continue; + if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var list)) continue; + foreach (var ev in list) _ready.Enqueue(ev); // stays counted in _heldCount until replayed + promoted = true; } - if (promoted) ParkedEventsReady?.Invoke(); + return promoted; } private sealed class PendingSubscribeScope(SubscriptionRegistry owner) : IDisposable @@ -328,32 +414,38 @@ internal sealed class SubscriptionRegistry /// The new tag bindings after reconnection. public void Rebind(long subscriptionId, IReadOnlyList newBindings) { - if (!_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) return; - - // Drop this subscription from every reverse-map set it currently appears in. The - // pre-reconnect item handles are stale once the gw re-issues fresh ones. - foreach (var binding in oldEntry.Bindings) + bool promoted; + lock (_parkGate) // map writes + promotion atomic — see Register { - if (binding.ItemHandle <= 0) continue; - if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue; - var remaining = set.Remove(subscriptionId); - if (remaining.IsEmpty) + if (!_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) return; + + // Drop this subscription from every reverse-map set it currently appears in. The + // pre-reconnect item handles are stale once the gw re-issues fresh ones. + foreach (var binding in oldEntry.Bindings) { - _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _); + if (binding.ItemHandle <= 0) continue; + if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue; + var remaining = set.Remove(subscriptionId); + if (remaining.IsEmpty) + { + _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _); + } + else _subscribersByItemHandle[binding.ItemHandle] = remaining; } - else _subscribersByItemHandle[binding.ItemHandle] = remaining; - } - _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, newBindings); - foreach (var binding in newBindings) - { - if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected - _subscribersByItemHandle.AddOrUpdate( - binding.ItemHandle, - _ => [subscriptionId], - (_, set) => set.Add(subscriptionId)); + _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, newBindings); + foreach (var binding in newBindings) + { + if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected + _subscribersByItemHandle.AddOrUpdate( + binding.ItemHandle, + _ => [subscriptionId], + (_, set) => set.Add(subscriptionId)); + } + AfterMapWriteForTests?.Invoke(); + promoted = PromoteParkedEventsLocked(newBindings); } - PromoteParkedEvents(newBindings); + if (promoted) ParkedEventsReady?.Invoke(); } /// diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs index f835d7491..e970b6661 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests/Runtime/EarlyEventWindowTests.cs @@ -7,6 +7,7 @@ using Xunit; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.OtOpcUa.Core.Abstractions; using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Config; +using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Health; using ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Runtime; namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; @@ -22,8 +23,6 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.Galaxy.Tests.Runtime; /// public sealed class EarlyEventWindowTests { - private const int WaitMs = 10_000; - private static MxEvent DataChange(int handle, double value) => new() { Family = MxEventFamily.OnDataChange, @@ -33,15 +32,11 @@ public sealed class EarlyEventWindowTests SourceTimestamp = Timestamp.FromDateTime(DateTime.UtcNow), }; - private static async Task WaitUntilAsync(Func predicate, string because) - { - var deadline = DateTime.UtcNow.AddMilliseconds(WaitMs); - while (!predicate()) - { - if (DateTime.UtcNow > deadline) throw new TimeoutException(because); - await Task.Delay(5); - } - } + private static GalaxyDriverOptions Opts() => new( + new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), + new GalaxyMxAccessOptions("OtOpcUa-A"), + new GalaxyRepositoryOptions(), + new GalaxyReconnectOptions()); // ===== Registry ===== @@ -50,8 +45,9 @@ public sealed class EarlyEventWindowTests public void Unknown_handle_without_a_pending_subscribe_is_stale() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); - registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Stale); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Stale); registry.ParkedEventCount.ShouldBe(0); } @@ -60,22 +56,23 @@ public sealed class EarlyEventWindowTests public void Parked_events_are_promoted_in_order_on_register() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); var woken = 0; registry.ParkedEventsReady += () => woken++; using (registry.BeginPendingSubscribe()) { - registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Parked); - registry.ResolveOrPark(DataChange(7, 2), out _).ShouldBe(EarlyEventOutcome.Parked); - registry.ResolveOrPark(DataChange(8, 9), out _).ShouldBe(EarlyEventOutcome.Parked); - registry.TakeReadyEvents().ShouldBeEmpty(); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(7, 2), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(8, 9), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.TakeReadyEvents(epoch).ShouldBeEmpty(); registry.Register(1, [new TagBinding("Tank.Level", 7)]); } woken.ShouldBe(1); - registry.TakeReadyEvents().Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]); - registry.TakeReadyEvents().ShouldBeEmpty(); + registry.TakeReadyEvents(epoch).Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]); + registry.TakeReadyEvents(epoch).ShouldBeEmpty(); // Handle 8 was never registered: closing the last window discarded it. registry.ParkedEventCount.ShouldBe(0); } @@ -85,15 +82,16 @@ public sealed class EarlyEventWindowTests public void Rebind_promotes_parked_events_for_the_new_handles() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); registry.Register(1, [new TagBinding("Tank.Level", 7)]); using (registry.BeginPendingSubscribe()) { - registry.ResolveOrPark(DataChange(70, 5), out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(70, 5), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); registry.Rebind(1, [new TagBinding("Tank.Level", 70)]); } - registry.TakeReadyEvents().Single().ItemHandle.ShouldBe(70); + registry.TakeReadyEvents(epoch).Single().ItemHandle.ShouldBe(70); } /// Closing one of two overlapping windows keeps the other subscribe's parked events. @@ -101,10 +99,11 @@ public sealed class EarlyEventWindowTests public void Closing_one_of_two_windows_keeps_parked_events() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); var first = registry.BeginPendingSubscribe(); var second = registry.BeginPendingSubscribe(); - registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); first.Dispose(); first.Dispose(); // idempotent — must not close the second window registry.PendingSubscribeCount.ShouldBe(1); @@ -112,7 +111,26 @@ public sealed class EarlyEventWindowTests second.Dispose(); registry.ParkedEventCount.ShouldBe(0); - registry.ResolveOrPark(DataChange(7, 1), out _).ShouldBe(EarlyEventOutcome.Stale); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Stale); + } + + /// + /// Review #6: a window opening right after the previous one closed never inherits its + /// leftovers (the N→0 discard and the 0→1 open are serialised under the park gate). + /// + [Fact] + public void A_new_window_never_inherits_the_previous_windows_leftovers() + { + var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); + var first = registry.BeginPendingSubscribe(); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + first.Dispose(); + + using var second = registry.BeginPendingSubscribe(); + registry.ParkedEventCount.ShouldBe(0); + registry.Register(1, [new TagBinding("Reused", 7)]); + registry.TakeReadyEvents(epoch).ShouldBeEmpty("the closed window's event must not reach the new owner of handle 7"); } /// Parking is bounded; past the bound the event is dropped as overflow. @@ -120,29 +138,77 @@ public sealed class EarlyEventWindowTests public void Parking_is_bounded() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); using var window = registry.BeginPendingSubscribe(); var ev = DataChange(7, 1); for (var i = 0; i < SubscriptionRegistry.MaxParkedEvents; i++) { - registry.ResolveOrPark(ev, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(ev, epoch, out _).ShouldBe(EarlyEventOutcome.Parked); } - registry.ResolveOrPark(ev, out _).ShouldBe(EarlyEventOutcome.Overflow); + registry.ResolveOrPark(ev, epoch, out _).ShouldBe(EarlyEventOutcome.Overflow); registry.ParkedEventCount.ShouldBe(SubscriptionRegistry.MaxParkedEvents); } + /// Review #5: promoted-but-not-yet-replayed events still count toward the bound. + [Fact] + public void Ready_events_count_toward_the_bound() + { + var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); + using var window = registry.BeginPendingSubscribe(); + var ev = DataChange(7, 1); + for (var i = 0; i < SubscriptionRegistry.MaxParkedEvents; i++) registry.ResolveOrPark(ev, epoch, out _); + registry.Register(1, [new TagBinding("Tank.Level", 7)]); // all promoted to ready + + registry.ResolveOrPark(DataChange(8, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Overflow); + + registry.TakeReadyEvents(epoch).Count.ShouldBe(SubscriptionRegistry.MaxParkedEvents); + registry.ParkedEventCount.ShouldBe(0); + registry.ResolveOrPark(DataChange(8, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked, + "positive control: once replayed, the room is back"); + } + /// A registered handle resolves normally even while a window is open. [Fact] public void Registered_handle_resolves_while_a_window_is_open() { var registry = new SubscriptionRegistry(); + var epoch = registry.StartNewEpoch(); registry.Register(1, [new TagBinding("Tank.Level", 7)]); using var window = registry.BeginPendingSubscribe(); - registry.ResolveOrPark(DataChange(7, 1), out var subs).ShouldBe(EarlyEventOutcome.Resolved); + registry.ResolveOrPark(DataChange(7, 1), epoch, out var subs).ShouldBe(EarlyEventOutcome.Resolved); subs.Single().FullReference.ShouldBe("Tank.Level"); } + /// + /// Review #3: held events belong to a session epoch. A new epoch (a new pump after a + /// reconnect) discards parked AND promoted-but-unreplayed events, and an older pump can + /// neither park nor replay — so a reused handle number never receives an old-session value. + /// + [Fact] + public void A_new_epoch_discards_held_events_and_fences_the_old_epoch() + { + var registry = new SubscriptionRegistry(); + var oldEpoch = registry.StartNewEpoch(); + using var window = registry.BeginPendingSubscribe(); + registry.ResolveOrPark(DataChange(7, 111), oldEpoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(8, 112), oldEpoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.Register(1, [new TagBinding("Old.A", 8)]); // 8 promoted to ready, never replayed + + var newEpoch = registry.StartNewEpoch(); + registry.ParkedEventCount.ShouldBe(0); + registry.TakeReadyEvents(newEpoch).ShouldBeEmpty(); + registry.Register(2, [new TagBinding("New.B", 7)]); // new session reuses handle 7 + registry.TakeReadyEvents(newEpoch).ShouldBeEmpty("the old session's value must not reach New.B"); + + registry.ResolveOrPark(DataChange(9, 1), oldEpoch, out _).ShouldBe(EarlyEventOutcome.Stale); + registry.TakeReadyEvents(oldEpoch).ShouldBeEmpty(); + registry.ResolveOrPark(DataChange(9, 1), newEpoch, out _).ShouldBe(EarlyEventOutcome.Parked, + "positive control: the current epoch still parks"); + } + // ===== EventPump ===== private sealed class ChannelSubscriber : IGalaxySubscriber @@ -163,7 +229,7 @@ public sealed class EarlyEventWindowTests } } - public ValueTask EmitAsync(MxEvent ev) => _events.Writer.WriteAsync(ev); + public void Emit(MxEvent ev) => _events.Writer.TryWrite(ev).ShouldBeTrue(); } /// @@ -182,14 +248,14 @@ public sealed class EarlyEventWindowTests using (registry.BeginPendingSubscribe()) { - await subscriber.EmitAsync(DataChange(7, 42)); - await WaitUntilAsync(() => registry.ParkedEventCount == 1, "the early event must be parked, not dropped"); + subscriber.Emit(DataChange(7, 42)); + await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the early event must be parked, not dropped"); delivered.ShouldBeEmpty(); registry.Register(1, [new TagBinding("Tank.Level", 7)]); } - await WaitUntilAsync(() => !delivered.IsEmpty, "the parked initial value must be replayed"); + await TestWait.UntilAsync(() => !delivered.IsEmpty, "the parked initial value must be replayed"); var args = delivered.Single(); args.FullReference.ShouldBe("Tank.Level"); ((double)args.Snapshot.Value!).ShouldBe(42.0); @@ -208,17 +274,56 @@ public sealed class EarlyEventWindowTests using (registry.BeginPendingSubscribe()) { - await subscriber.EmitAsync(DataChange(7, 1)); - await subscriber.EmitAsync(DataChange(7, 2)); - await WaitUntilAsync(() => registry.ParkedEventCount == 2, "both early events must be parked"); + subscriber.Emit(DataChange(7, 1)); + subscriber.Emit(DataChange(7, 2)); + await TestWait.UntilAsync(() => registry.ParkedEventCount == 2, "both early events must be parked"); registry.Register(1, [new TagBinding("Tank.Level", 7)]); - await subscriber.EmitAsync(DataChange(7, 3)); + subscriber.Emit(DataChange(7, 3)); } - await WaitUntilAsync(() => delivered.Count >= 3, "all three values must be delivered"); + await TestWait.UntilAsync(() => delivered.Count >= 3, "all three values must be delivered", + () => string.Join(",", delivered)); delivered.ShouldBe([1.0, 2.0, 3.0]); } + /// + /// Review #2: a live event arriving while a registration has written the reverse map but not + /// yet promoted (a 50k-binding Register is milliseconds of work) must NOT overtake the parked + /// initial value. The hook runs inside that window, emits the newer value and gives the pump + /// ample time to (wrongly) dispatch it; the pump must instead wait for the promotion. + /// + [Fact] + public async Task Pump_does_not_let_a_live_event_overtake_a_registration_in_progress() + { + var subscriber = new ChannelSubscriber(); + var registry = new SubscriptionRegistry(); + var delivered = new ConcurrentQueue(); + await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Early"); + pump.OnDataChange += (_, e) => delivered.Enqueue((double)e.Snapshot.Value!); + pump.Start(); + + var deliveredDuringRegistration = -1; + using (registry.BeginPendingSubscribe()) + { + subscriber.Emit(DataChange(7, 1)); + await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the initial value must be parked"); + + registry.AfterMapWriteForTests = () => + { + subscriber.Emit(DataChange(7, 2)); + Thread.Sleep(300); + deliveredDuringRegistration = delivered.Count; + }; + registry.Register(1, [new TagBinding("Tank.Level", 7)]); + registry.AfterMapWriteForTests = null; + } + + await TestWait.UntilAsync(() => delivered.Count >= 2, "both values must be delivered", + () => string.Join(",", delivered)); + deliveredDuringRegistration.ShouldBe(0, "nothing may dispatch for the handle mid-registration"); + delivered.ShouldBe([1.0, 2.0]); + } + /// /// A genuinely stale event (no subscribe in flight) is still dropped; the positive control /// that follows it proves the pump processed it rather than not having got to it yet. @@ -234,35 +339,72 @@ public sealed class EarlyEventWindowTests pump.OnDataChange += (_, e) => delivered.Enqueue(e); pump.Start(); - await subscriber.EmitAsync(DataChange(99, 5)); // unsubscribed handle - await subscriber.EmitAsync(DataChange(7, 6)); // positive control, strictly after + subscriber.Emit(DataChange(99, 5)); // unsubscribed handle + subscriber.Emit(DataChange(7, 6)); // positive control, strictly after - await WaitUntilAsync(() => !delivered.IsEmpty, "the positive control must arrive"); + await TestWait.UntilAsync(() => !delivered.IsEmpty, "the positive control must arrive"); delivered.Single().FullReference.ShouldBe("Tank.Level"); registry.ParkedEventCount.ShouldBe(0); // Registering 99 now must not resurrect the stale event. registry.Register(2, [new TagBinding("Other", 99)]); - registry.TakeReadyEvents().ShouldBeEmpty(); + registry.ParkedEventCount.ShouldBe(0); } - // ===== GalaxyDriver end-to-end ===== + /// + /// Review #3, pump level: an event the old session's pump parked is never replayed by the + /// next session's pump onto the tag that now owns the reused handle number. + /// + [Fact] + public async Task Pump_after_reconnect_never_replays_an_old_session_event_onto_a_reused_handle() + { + var registry = new SubscriptionRegistry(); + var delivered = new ConcurrentQueue(); + using var window = registry.BeginPendingSubscribe(); // e.g. a subscribe hanging across the reconnect + + var oldStream = new ChannelSubscriber(); + var oldPump = new EventPump(oldStream, registry, channelCapacity: 16, clientName: "Old"); + oldPump.Start(); + oldStream.Emit(DataChange(7, 111)); + await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the old session's event must be parked"); + await oldPump.DisposeAsync(); + + var newStream = new ChannelSubscriber(); + await using var newPump = new EventPump(newStream, registry, channelCapacity: 16, clientName: "New"); + newPump.OnDataChange += (_, e) => delivered.Enqueue(e); + newPump.Start(); + + registry.Register(2, [new TagBinding("New.Tag", 7)]); // new session reuses handle 7 + newStream.Emit(DataChange(7, 222)); // positive control on the new session + + await TestWait.UntilAsync(() => !delivered.IsEmpty, "the new session's value must arrive"); + await TestWait.UntilAsync(() => registry.ParkedEventCount == 0, "nothing may stay held"); + delivered.Select(e => (double)e.Snapshot.Value!).ShouldBe([222.0]); + } + + // ===== GalaxyDriver / probe watcher end-to-end ===== /// /// A gateway fake that writes each new handle's initial value to the event stream and only - /// returns the SubscribeBulk reply once the driver's pump has consumed that event — the + /// returns the SubscribeBulk reply once the consumer's pump has parked that event — the /// event deterministically precedes the reply. /// - private sealed class EventBeforeReplySubscriber(Func parkedCount) : IGalaxySubscriber + private sealed class EventBeforeReplySubscriber : IGalaxySubscriber { private readonly Channel _events = Channel.CreateUnbounded(); private int _nextHandle = 100; + /// Gets or sets the consumer's held-event count, polled before replying. + public Func ParkedCount { get; set; } = () => 0; + + /// Gets the handles unsubscribed so far. + public ConcurrentQueue Unsubscribed { get; } = new(); + public async Task> SubscribeBulkAsync( IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) { var results = new List(); - var before = parkedCount(); + var before = ParkedCount(); foreach (var fullRef in fullReferences) { var handle = Interlocked.Increment(ref _nextHandle); @@ -270,12 +412,17 @@ public sealed class EarlyEventWindowTests results.Add(new SubscribeResult { TagAddress = fullRef, ItemHandle = handle, WasSuccessful = true }); } - await WaitUntilAsync(() => parkedCount() >= before + fullReferences.Count, - "the initial values must reach the pump (and be parked) before the reply is returned"); + await TestWait.UntilAsync(() => ParkedCount() >= before + fullReferences.Count, + "the initial values must reach the pump (and be parked) before the reply is returned", + () => $"held={ParkedCount()} before={before}"); return results; } - public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) => Task.CompletedTask; + public Task UnsubscribeBulkAsync(IReadOnlyList itemHandles, CancellationToken cancellationToken) + { + foreach (var h in itemHandles) Unsubscribed.Enqueue(h); + return Task.CompletedTask; + } public async IAsyncEnumerable StreamEventsAsync([EnumeratorCancellation] CancellationToken cancellationToken) { @@ -290,26 +437,61 @@ public sealed class EarlyEventWindowTests [Fact] public async Task Driver_publishes_an_initial_value_that_arrived_before_the_subscribe_reply() { - GalaxyDriver? driverRef = null; - var subscriber = new EventBeforeReplySubscriber(() => driverRef!.ParkedEarlyEventCountForTests); + var subscriber = new EventBeforeReplySubscriber(); using var driver = new GalaxyDriver( - "g", - new GalaxyDriverOptions( - new GalaxyGatewayOptions("https://mxgw.test:5001", "key"), - new GalaxyMxAccessOptions("OtOpcUa-A"), - new GalaxyRepositoryOptions(), - new GalaxyReconnectOptions()), - hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); - driverRef = driver; + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + subscriber.ParkedCount = () => driver.ParkedEarlyEventCountForTests; var delivered = new ConcurrentQueue(); driver.OnDataChange += (_, e) => delivered.Enqueue(e); var handle = await driver.SubscribeAsync( ["Tank.Level", "Tank.Temp"], TimeSpan.FromSeconds(1), TestContext.Current.CancellationToken); - await WaitUntilAsync(() => delivered.Count >= 2, "both initial values must be published"); + await TestWait.UntilAsync(() => delivered.Count >= 2, "both initial values must be published"); delivered.Select(e => e.FullReference).OrderBy(r => r).ShouldBe(["Tank.Level", "Tank.Temp"]); delivered.ShouldAllBe(e => e.SubscriptionHandle.Equals(handle)); driver.ParkedEarlyEventCountForTests.ShouldBe(0); } + + /// + /// Review #1: the subscribe-once read waits ONLY for the initial value, so an initial value + /// that overtakes the reply must still complete the read Good — not BadTimeout. + /// + [Fact] + public async Task Driver_read_returns_an_initial_value_that_arrived_before_the_subscribe_reply() + { + var subscriber = new EventBeforeReplySubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + subscriber.ParkedCount = () => driver.ParkedEarlyEventCountForTests; + + var snapshots = await TestWait.ForAsync( + driver.ReadAsync(["Tank.Level"], TestContext.Current.CancellationToken), + "the read must complete"); + + var snapshot = snapshots.ShouldHaveSingleItem(); + snapshot.StatusCode.ShouldBe(StatusCodeMap.Good); + ((double)snapshot.Value!).ShouldBe(1010.0); + driver.ParkedEarlyEventCountForTests.ShouldBe(0); + } + + /// The probe watcher's initial ScanState overtaking its reply still reaches the fan-out. + [Fact] + public async Task Probe_watcher_initial_value_that_arrived_before_the_reply_is_dispatched() + { + var registry = new SubscriptionRegistry(); + var subscriber = new EventBeforeReplySubscriber { ParkedCount = () => registry.ParkedEventCount }; + var delivered = new ConcurrentQueue(); + await using var pump = new EventPump(subscriber, registry, channelCapacity: 16, clientName: "Probe"); + pump.OnDataChange += (_, e) => delivered.Enqueue(e); + pump.Start(); + using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry); + + await watcher.SyncPlatformsAsync(["PlatformA"], TestContext.Current.CancellationToken); + + await TestWait.UntilAsync(() => !delivered.IsEmpty, "the initial ScanState must be dispatched"); + var args = delivered.Single(); + args.FullReference.ShouldBe("PlatformA.ScanState"); + ((GalaxySubscriptionHandle)args.SubscriptionHandle).SubscriptionId.ShouldBe(SubscriptionRegistry.ProbeSubscriptionId); + } }