diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index cc6dce90b..0022abed1 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 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. +- **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. Unclaimed parked events are trimmed by **window sequence**, not only when every window has closed (concurrent reads keep some window open almost permanently): each window gets a monotonically increasing sequence when it opens, each parked event is stamped with the newest sequence, and once every window that opened at or before that stamp has closed the event is discarded — outside any window the stale drop is unchanged. **Sessions (#861):** item handles are per gateway session and the next session reuses the numbers. On reconnect `ReplayAsync` first calls `SubscriptionRegistry.StartNewSession()`, which discards everything held, **invalidates the reverse map** (subscriptions are kept; new-session events park until the replay's per-subscription `Rebind` promotes them, instead of resolving to the tag that owned the number before), and bumps a session number. Registration and rebind are refused for a stale session, so a `SubscribeAsync` whose reply straddled the reconnect is **re-issued on the new session** (up to 3 attempts), a subscribe-once read returns `BadCommunicationError`, and dead old-session handle numbers are never sent to `UnsubscribeBulk`; an old session's pump can no longer resolve through the map. **Interim (I-3):** a completed `UnsubscribeBulk` purges anything parked for its handles, so a trailing event of a removed item is not promoted onto a reused handle number within one session. It cannot catch an event still in flight after the unsubscribe, and it also drops a genuine early value for a number reused that fast (a timeout rather than another tag's value). The complete fix needs a worker sequence on the SubscribeBulk reply so the driver can tell pre-subscribe events from post-subscribe ones — an optional cross-repo change in mxaccessgw. - **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 05f7f9c12..7e77a70ce 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -256,6 +256,9 @@ public sealed class GalaxyDriver /// Test seam — the registry's parked early-event count (#859). internal int ParkedEarlyEventCountForTests => _subscriptions.ParkedEventCount; + /// Test seam — the driver's subscription registry (#859/#861 window + session tests). + internal SubscriptionRegistry SubscriptionsForTests => _subscriptions; + /// /// Server-pushed data-change notification. Fires from the /// 's background loop; handlers should be cheap (or queue @@ -644,13 +647,37 @@ public sealed class GalaxyDriver /// Register only while is still the /// live runtime. Atomic with the teardown's registry Clear (both under ). /// - private bool TryRegisterForGeneration(long subscriptionId, IReadOnlyList bindings, int generation) + private RegisterOutcome TryRegisterForGeneration( + long subscriptionId, IReadOnlyList bindings, int generation, int session) { lock (_registryGate) { - if (Volatile.Read(ref _runtimeGeneration) != generation) return false; - _subscriptions.Register(subscriptionId, bindings); - return true; + if (Volatile.Read(ref _runtimeGeneration) != generation) return RegisterOutcome.TornDown; + // #861: the reply may have straddled a reconnect — its handles then belong to the dead + // session and collide with the new session's numbers. Checked atomically with + // StartNewSession inside the registry. + return _subscriptions.TryRegister(subscriptionId, bindings, session) + ? RegisterOutcome.Registered + : RegisterOutcome.SessionChanged; + } + } + + private enum RegisterOutcome { Registered, TornDown, SessionChanged } + + /// + /// UnsubscribeBulk for , then purge any event parked for them + /// (#859 I-3 interim — a trailing event must not be promoted onto a reused handle number). + /// + private async Task UnsubscribeAndPurgeAsync( + IGalaxySubscriber subscriber, IReadOnlyList itemHandles, CancellationToken cancellationToken) + { + try + { + await subscriber.UnsubscribeBulkAsync(itemHandles, cancellationToken).ConfigureAwait(false); + } + finally + { + _subscriptions.PurgeParked(itemHandles); } } @@ -667,7 +694,7 @@ public sealed class GalaxyDriver { var live = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); if (live.Length == 0) return; - try { await subscriber.UnsubscribeBulkAsync(live, CancellationToken.None).ConfigureAwait(false); } + try { await UnsubscribeAndPurgeAsync(subscriber, live, CancellationToken.None).ConfigureAwait(false); } catch (Exception ex) { _logger.LogWarning(ex, @@ -945,6 +972,15 @@ public sealed class GalaxyDriver // anyway would start a fresh pump underneath the shutdown. cancellationToken.ThrowIfCancellationRequested(); if (_subscriber is null) return; + + // The session was recreated: every item handle the registry holds is dead and its number + // may be reused by the new session (#859 I-1 / #861). Invalidate the reverse map (the + // subscriptions stay; the per-iteration Rebind below re-binds each one, and new-session + // events park meanwhile instead of resolving to the old owner of a reused number), discard + // held events, and fence any subscribe whose reply straddled the reconnect. Done FIRST and + // unconditionally — before the pump restart and even with nothing to replay. + _subscriptions.StartNewSession(); + var entries = _subscriptions.SnapshotEntries(); if (entries.Count == 0) return; @@ -1025,6 +1061,7 @@ public sealed class GalaxyDriver // 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 session = _subscriptions.CurrentSession; var results = await _subscriber! .SubscribeBulkAsync(refs, _options.MxAccess.PublishingIntervalMs, cancellationToken) .ConfigureAwait(false); @@ -1045,7 +1082,15 @@ public sealed class GalaxyDriver lock (_registryGate) { if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount; - _subscriptions.Rebind(subscriptionId, newBindings); + // #861: another reconnect during this round-trip makes these handles dead too; the + // newer session's own replay re-subscribes everything, so stop here. + if (!_subscriptions.TryRebind(subscriptionId, newBindings, session)) + { + _logger.LogDebug( + "GalaxyDriver {InstanceId} re-subscribe abandoned — the gateway session changed again.", + _driverInstanceId); + return tagCount; + } } tagCount += refs.Length; } @@ -1466,6 +1511,7 @@ public sealed class GalaxyDriver // 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(); + var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on try { // The RPC gets only the CALLER's token: the internal deadline bounds our wait, not the RPC, so a slow @@ -1485,10 +1531,13 @@ public sealed class GalaxyDriver _ = subscribeTask.ContinueWith(t => { if (!t.IsCompletedSuccessfully) return Task.CompletedTask; + // A late reply from a session that has since been replaced carries dead handle + // numbers the new session may reuse — never unsubscribe those (#861). + if (_subscriptions.CurrentSession != session) 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( + : UnsubscribeAndPurgeAsync(subscriber, late, CancellationToken.None).ContinueWith( u => logger.LogWarning(u.Exception, "GalaxyDriver.ReadViaSubscribeOnceAsync late UnsubscribeBulk failed for {Count} handle(s).", late.Length), TaskContinuationOptions.OnlyOnFaulted); @@ -1532,11 +1581,28 @@ public sealed class GalaxyDriver ServerTimestampUtc: DateTime.UtcNow)); } } - if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) + var registered = TryRegisterForGeneration(subscriptionId, bindings, generation, session); + if (registered != RegisterOutcome.Registered) { pump.OnDataChange -= handler; - await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false); - throw TornDownDuringSubscribe(); + earlyEventWindow.Dispose(); // before any further round-trip + if (registered == RegisterOutcome.TornDown) + { + await ReleaseOrphanedHandlesAsync(readSubscriber, bindings).ConfigureAwait(false); + throw TornDownDuringSubscribe(); + } + + // #861: the reply straddled a reconnect. Its handles died with the old session and + // their numbers may already belong to new-session items, so they are neither + // registered nor unsubscribed. A read is one-shot: report the lost session per tag. + _logger.LogInformation( + "GalaxyDriver {InstanceId} subscribe-once read of {Count} tag(s) straddled a gateway reconnect — returning BadCommunicationError.", + _driverInstanceId, fullReferences.Count); + var lostAt = DateTime.UtcNow; + var lost = new DataValueSnapshot[fullReferences.Count]; + for (var i = 0; i < lost.Length; i++) + lost[i] = new DataValueSnapshot(null, StatusCodeMap.BadCommunicationError, null, lostAt); + return lost; } } finally @@ -1575,13 +1641,16 @@ public sealed class GalaxyDriver // Drop the bindings + unsubscribe the live handles. UnsubscribeBulkAsync's // failure isn't fatal — the registry is already cleared, so any straggling // event from the gw would be a no-op fan-out. - _subscriptions.Remove(subscriptionId); + var removed = _subscriptions.Remove(subscriptionId, out var liveInCurrentSession); + // A reconnect during the wait left these handles in the dead session; their numbers may now + // be new-session items, so they must not be unsubscribed (#861). + var staleSession = removed is not null && !liveInCurrentSession; var liveHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); - if (liveHandles.Length > 0) + if (liveHandles.Length > 0 && !staleSession) { try { - await readSubscriber.UnsubscribeBulkAsync(liveHandles, CancellationToken.None) + await UnsubscribeAndPurgeAsync(readSubscriber, liveHandles, CancellationToken.None) .ConfigureAwait(false); } catch (Exception ex) @@ -1653,7 +1722,14 @@ public sealed class GalaxyDriver { // Empty subscriptions register but never bind anything — keeps Unsubscribe // symmetric for callers that conditionally add tags later. - if (!TryRegisterForGeneration(subscriptionId, [], generation)) throw TornDownDuringSubscribe(); + RegisterOutcome outcome; + do + { + // No handles, so a session change only means "read the session again". + outcome = TryRegisterForGeneration(subscriptionId, [], generation, _subscriptions.CurrentSession); + } + while (outcome == RegisterOutcome.SessionChanged); + if (outcome == RegisterOutcome.TornDown) throw TornDownDuringSubscribe(); return new GalaxySubscriptionHandle(subscriptionId); } @@ -1670,16 +1746,63 @@ public sealed class GalaxyDriver var requested = (int)Math.Max(0, publishingInterval.TotalMilliseconds); var bufferedIntervalMs = requested > 0 ? requested : _options.MxAccess.PublishingIntervalMs; - // #859: MXAccess's initial value for a new handle can reach the event pump before this - // reply is registered (the gateway gives no reply-before-event ordering, and the reply and - // the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events - // are parked rather than dropped as stale, and close it only after TryRegisterForGeneration - // (method scope), which promotes them for in-order replay. - using var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); - var results = await subscriber - .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) - .ConfigureAwait(false); + for (var attempt = 1; ; attempt++) + { + // #859: MXAccess's initial value for a new handle can reach the event pump before this + // reply is registered (the gateway gives no reply-before-event ordering, and the reply and + // the event ride separate RPCs). Open the early-event window BEFORE the RPC so such events + // are parked rather than dropped as stale; it closes once the registration (which promotes + // them for in-order replay) is done. + var earlyEventWindow = _subscriptions.BeginPendingSubscribe(); + try + { + var session = _subscriptions.CurrentSession; // #861 — the session this SubscribeBulk is issued on + var results = await subscriber + .SubscribeBulkAsync(dialledRefs, bufferedIntervalMs, cancellationToken) + .ConfigureAwait(false); + var bindings = BuildSubscribeBindings(dialledRefs, results); + switch (TryRegisterForGeneration(subscriptionId, bindings, generation, session)) + { + case RegisterOutcome.Registered: + _ = pump; // keep the pump alive for the subscription's lifetime + return new GalaxySubscriptionHandle(subscriptionId); + + case RegisterOutcome.TornDown: + // The teardown cleared the registry while SubscribeBulk was in flight; registering + // now would carry dead handles into the next runtime. Release what we got and fail. + earlyEventWindow.Dispose(); // M-3: not held across the release round-trip + await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false); + throw TornDownDuringSubscribe(); + + default: + // #861: the reply straddled a gateway reconnect, so its handles died with the old + // session (and are NOT unsubscribed — the new session may reuse the numbers). The + // reconnect replay only re-subscribes what was registered when it snapshotted, so + // this subscription would be lost: issue it again on the new session. + if (attempt >= MaxSubscribeSessionAttempts) + { + throw new InvalidOperationException( + $"Galaxy subscribe straddled {attempt} consecutive gateway reconnects — giving up; retry later."); + } + _logger.LogInformation( + "GalaxyDriver {InstanceId} subscribe of {Count} tag(s) straddled a gateway reconnect — re-issuing on the new session (attempt {Attempt}).", + _driverInstanceId, dialledRefs.Length, attempt + 1); + continue; + } + } + finally + { + earlyEventWindow.Dispose(); + } + } + } + + /// Bound on re-issuing a subscribe whose reply straddled a gateway reconnect (#861). + private const int MaxSubscribeSessionAttempts = 3; + + private List BuildSubscribeBindings(string[] dialledRefs, IReadOnlyList results) + { // Build the binding list in input order. Failed entries (gw rejected the tag) are // recorded with a non-positive ItemHandle so the caller can detect partial failure // by inspecting the returned handle's diagnostic context — full per-tag error @@ -1703,16 +1826,7 @@ public sealed class GalaxyDriver fullRef, match?.ErrorMessage ?? ""); } } - - if (!TryRegisterForGeneration(subscriptionId, bindings, generation)) - { - // The teardown cleared the registry while SubscribeBulk was in flight; registering now - // would carry dead handles into the next runtime. Release what we got and fail the call. - await ReleaseOrphanedHandlesAsync(subscriber, bindings).ConfigureAwait(false); - throw TornDownDuringSubscribe(); - } - _ = pump; // keep the pump alive for the subscription's lifetime - return new GalaxySubscriptionHandle(subscriptionId); + return bindings; } /// @@ -1727,15 +1841,19 @@ public sealed class GalaxyDriver nameof(handle)); } - var bindings = _subscriptions.Remove(gsh.SubscriptionId); + var bindings = _subscriptions.Remove(gsh.SubscriptionId, out var liveInCurrentSession); if (bindings is null) return; // already removed or never registered + // Not yet rebound after a reconnect: its handles died with the old session and the new one may + // have reused the numbers for other items — unsubscribing them would unadvise those (#861). + if (!liveInCurrentSession) return; + var liveItemHandles = bindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); if (liveItemHandles.Length == 0 || _subscriber is null) return; try { - await _subscriber.UnsubscribeBulkAsync(liveItemHandles, cancellationToken).ConfigureAwait(false); + await UnsubscribeAndPurgeAsync(_subscriber, liveItemHandles, cancellationToken).ConfigureAwait(false); } catch (Exception ex) { diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs index 2173a5a00..d98bae7e0 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/Health/PerPlatformProbeWatcher.cs @@ -260,6 +260,12 @@ public sealed class PerPlatformProbeWatcher : IDisposable "PerPlatformProbeWatcher unsubscribe failed for {Count} probe(s); aggregator entries already cleared.", toRemove.Count); } + finally + { + // #859 I-3 interim: no trailing event for a removed probe handle may be promoted onto + // a reused handle number. + _registry?.PurgeParked([.. toRemove.Select(t => t.ItemHandle)]); + } } if (toAdd.Count == 0) return; 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 a7d5388b8..e2be7600e 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 @@ -139,14 +139,14 @@ internal sealed class EventPump : IAsyncDisposable }); _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(); + // One pump per gateway session. The owner calls SubscriptionRegistry.StartNewSession() BEFORE + // creating the pump for a new session (reconnect); this pump binds to that session and never + // parks or replays under any other (#859 review — item handles are reused across sessions). + _session = _registry.CurrentSession; _registry.ParkedEventsReady += WakeForReadyEvents; } - private readonly int _epoch; + private readonly int _session; /// /// Marker written into the channel to wake the dispatch loop after a registration promoted @@ -161,7 +161,7 @@ internal sealed class EventPump : IAsyncDisposable /// 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 dispatch) or completed (the pump is dead; its held events - /// belong to a session the next pump's new epoch discards). + /// belong to a session the next StartNewSession discards). /// private void WakeForReadyEvents() => _channel.Writer.TryWrite(WakeSentinel); @@ -363,14 +363,14 @@ internal sealed class EventPump : IAsyncDisposable private void DispatchDataChange(MxEvent ev) { - var outcome = _registry.ResolveOrPark(ev, _epoch, out var subscribers); + var outcome = _registry.ResolveOrPark(ev, _session, out var subscribers, out var replayFirst); - // 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(); + // Replay promoted early events (#859) 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 — and takes the ready queue — under that + // same gate whenever anything is held. So if this event resolved, every older parked event + // for its handle is in replayFirst, and replaying it first preserves per-handle order. + ReplayEarly(replayFirst); switch (outcome) { @@ -396,9 +396,10 @@ internal sealed class EventPump : IAsyncDisposable /// thread, so it is serialised with live dispatch. A promoted event whose subscription was /// removed again before the replay resolves to nobody and is dropped like any stale event. /// - private void DispatchReadyEvents() + private void DispatchReadyEvents() => ReplayEarly(_registry.TakeReadyEvents(_session)); + + private void ReplayEarly(IReadOnlyList ready) { - 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 ad65f06d3..303dde0c5 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 @@ -68,17 +68,70 @@ internal sealed class SubscriptionRegistry /// /// The subscription identifier. /// The tag bindings for the subscription. - public void Register(long subscriptionId, IReadOnlyList bindings) + public void Register(long subscriptionId, IReadOnlyList bindings) => + Bind(subscriptionId, bindings, requireExisting: false, expectedSession: null); + + /// + /// , but only if the gateway session is still + /// — the session the SubscribeBulk that produced + /// was issued on (#861). Item handles are per session and the next + /// session reuses the numbers, so registering a reply that straddled a reconnect would bind + /// dead handles that collide with live ones. + /// + /// The subscription identifier. + /// The tag bindings for the subscription. + /// The captured before the SubscribeBulk. + /// False (nothing registered) when the session changed in the meantime. + public bool TryRegister(long subscriptionId, IReadOnlyList bindings, int expectedSession) => + Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindResult.Bound; + + /// + /// Replace an existing subscription's bindings with the item handles a post-reconnect + /// SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new + /// handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the + /// subscription id is unknown (it was unsubscribed during the reconnect window). + /// + /// The subscription identifier. + /// The new tag bindings after reconnection. + public void Rebind(long subscriptionId, IReadOnlyList newBindings) => + Bind(subscriptionId, newBindings, requireExisting: true, expectedSession: null); + + /// , but only while the session is still (#861). + /// The subscription identifier. + /// The new tag bindings. + /// The session the SubscribeBulk was issued on. + /// False when the session changed (nothing rebound); true otherwise, including the unknown-id no-op. + public bool TryRebind(long subscriptionId, IReadOnlyList newBindings, int expectedSession) => + Bind(subscriptionId, newBindings, requireExisting: true, expectedSession) != BindResult.SessionChanged; + + private enum BindResult { Bound, UnknownId, SessionChanged } + + private BindResult Bind( + long subscriptionId, IReadOnlyList bindings, bool requireExisting, int? expectedSession) { 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. + // parked initial value. ResolveOrPark takes the same gate whenever anything is held. The + // session check is in here too, so it is atomic with StartNewSession's invalidation. lock (_parkGate) { - var entry = new SubscriptionEntry(subscriptionId, bindings); - _bySubscriptionId[subscriptionId] = entry; + if (expectedSession is { } session && session != _session) return BindResult.SessionChanged; + + if (_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) + { + // Drop this subscription from every reverse-map set it currently appears in. Only + // this id is removed, so another subscription that now owns the same handle number + // (a reused handle in a new session) keeps its entry. + RemoveFromReverseMap(oldEntry); + } + else if (requireExisting) + { + return BindResult.UnknownId; + } + + _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, bindings, _session); foreach (var binding in bindings) { if (binding.ItemHandle <= 0) continue; // failed gw subscribe — no events expected @@ -91,6 +144,22 @@ internal sealed class SubscriptionRegistry promoted = PromoteParkedEventsLocked(bindings); } if (promoted) ParkedEventsReady?.Invoke(); + return BindResult.Bound; + } + + private void RemoveFromReverseMap(SubscriptionEntry entry) + { + foreach (var binding in entry.Bindings) + { + if (binding.ItemHandle <= 0) continue; + if (!_subscribersByItemHandle.TryGetValue(binding.ItemHandle, out var set)) continue; + var remaining = set.Remove(entry.SubscriptionId); + if (remaining.IsEmpty) + { + _subscribersByItemHandle.TryRemove(binding.ItemHandle, out _); + } + else _subscribersByItemHandle[binding.ItemHandle] = remaining; + } } /// @@ -99,23 +168,27 @@ internal sealed class SubscriptionRegistry /// /// The subscription identifier. /// The bindings for the subscription, or null if not found. - public IReadOnlyList? Remove(long subscriptionId) + public IReadOnlyList? Remove(long subscriptionId) => Remove(subscriptionId, out _); + + /// + /// Remove a subscription, also reporting whether its handles belong to the CURRENT gateway + /// session. A subscription not yet rebound after a reconnect still carries the previous + /// session's handles, whose numbers the new session reuses — the caller must not send those + /// to UnsubscribeBulk, or it would unadvise some other live item. + /// + /// The subscription identifier. + /// True when the bindings' handles were issued by the current session. + /// The bindings for the subscription, or null if not found. + public IReadOnlyList? Remove(long subscriptionId, out bool liveInCurrentSession) { - if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null; - - foreach (var binding in entry.Bindings) + lock (_parkGate) { - 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; + liveInCurrentSession = false; + if (!_bySubscriptionId.TryRemove(subscriptionId, out var entry)) return null; + RemoveFromReverseMap(entry); + liveInCurrentSession = entry.Session == _session; + return entry.Bindings; } - - return entry.Bindings; } /// @@ -140,7 +213,6 @@ internal sealed class SubscriptionRegistry return result; } - /// /// Drop every subscription and the whole reverse map. Used by the driver's runtime teardown: /// once the gateway session is closed every item handle in here is dead, and the host @@ -148,13 +220,11 @@ internal sealed class SubscriptionRegistry /// public void Clear() { - _bySubscriptionId.Clear(); - _subscribersByItemHandle.Clear(); lock (_parkGate) { - _parkedByItemHandle.Clear(); - _ready.Clear(); - Volatile.Write(ref _heldCount, 0); + _bySubscriptionId.Clear(); + _subscribersByItemHandle.Clear(); + DiscardAllHeldLocked(); } } @@ -177,15 +247,21 @@ internal sealed class SubscriptionRegistry // While at least one subscribe is in flight such events are parked here (bounded) instead of // dropped; a registration that binds a parked handle promotes its events to a ready queue that // the EventPump's dispatch loop replays — on that same single thread, BEFORE any later event for - // 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. + // the handle — so per-handle value order is preserved. // - // 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. + // Unclaimed parked events are trimmed by WINDOW SEQUENCE, not only when every window has + // closed (#859 second review — concurrent reads keep some window open almost permanently): + // each window gets a monotonically increasing sequence when it opens, and each parked event is + // stamped with the newest sequence at park time. An event belonging to window W is generated + // after W opened, so its stamp is >= W's sequence; once every window with a sequence <= the + // stamp has closed, nothing can claim it and it is discarded. + // + // Item handles are only meaningful within one gateway SESSION, and a new session reuses small + // integers. StartNewSession (called on reconnect, before the replay re-subscribes) discards + // every held event, invalidates the reverse map (so a new-session event can never resolve to + // the tag that owned that number in the old session — it parks until the replay's Rebind), + // and fences: an event parked or replayed by an older session's pump is dropped, and a + // subscribe whose reply straddled the reconnect cannot register its dead handles (#861). /// /// Upper bound on events held (parked + promoted-but-not-yet-replayed) across all in-flight @@ -196,16 +272,18 @@ internal sealed class SubscriptionRegistry public const int MaxParkedEvents = 100_000; private readonly Lock _parkGate = new(); - private readonly Dictionary> _parkedByItemHandle = new(); + private readonly Dictionary> _parkedByItemHandle = new(); + private readonly Queue _parkOrder = new(); // every parked event, in park order (trim index) private readonly Queue _ready = new(); + private readonly SortedSet _openWindows = new(); + private long _windowSequence; // 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. + // Open windows. Written only under _parkGate; read lock-free by the dispatch fast path. private int _pendingSubscribes; - private int _epoch; + private int _session; /// /// Test seam (#859 review): invoked inside the registration critical section, between the @@ -213,6 +291,27 @@ internal sealed class SubscriptionRegistry /// internal Action? AfterMapWriteForTests { get; set; } + /// + /// Test seam (M-1): parks unconditionally, producing the parked list for a + /// resolved handle that the atomic registration otherwise makes impossible. + /// + /// The event to park. + internal void ForceParkForTests(MxEvent ev) + { + lock (_parkGate) + { + var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev); + if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue)) + { + queue = new Queue(); + _parkedByItemHandle[ev.ItemHandle] = queue; + } + queue.Enqueue(entry); + _parkOrder.Enqueue(entry); + Volatile.Write(ref _heldCount, _heldCount + 1); + } + } + /// /// Raised when a registration promoted parked events to the ready queue, so the /// can wake its dispatch loop even if no further event arrives. @@ -230,78 +329,141 @@ internal sealed class SubscriptionRegistry /// Gets the number of events currently held (parked or awaiting replay). public int ParkedEventCount => Volatile.Read(ref _heldCount); + /// Gets the current gateway-session number (bumped by ). + public int CurrentSession => Volatile.Read(ref _session); + /// - /// Starts a new session epoch: discards every held event and makes events from any earlier - /// epoch unparkable and unreplayable. Called by each new . + /// A new gateway session replaced the old one (reconnect). Discards every held event, + /// invalidates the reverse map — subscriptions are KEPT, but their old-session handles no + /// longer resolve, so new-session events park until the replay rebinds each subscription — + /// and bumps the session number that fences old pumps and straddling subscribes. NOT for the + /// same-session replay-gap re-snapshot, whose handles stay valid until rebound. /// - /// The new epoch. - public int StartNewEpoch() + /// The new session number. + public int StartNewSession() { lock (_parkGate) { - _parkedByItemHandle.Clear(); - _ready.Clear(); - Volatile.Write(ref _heldCount, 0); - return ++_epoch; + _subscribersByItemHandle.Clear(); + DiscardAllHeldLocked(); + // Volatile: the dispatch fast path reads it lock-free so a dying pump of the old session + // can never resolve an old-session event through the map once the new session exists. + Volatile.Write(ref _session, _session + 1); + return _session; } } /// /// 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) — - /// not later: while any window is open, events for unknown handles are parked, not dropped. + /// not later: while a window is open, events for unknown handles are parked, not dropped. /// /// A token that closes the window when disposed (idempotent). public IDisposable BeginPendingSubscribe() { + long sequence; 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); + sequence = ++_windowSequence; + _openWindows.Add(sequence); + Volatile.Write(ref _pendingSubscribes, _openWindows.Count); } - return new PendingSubscribeScope(this); + return new PendingSubscribeScope(this, sequence); } - private void EndPendingSubscribe() + private void EndPendingSubscribe(long sequence) { lock (_parkGate) { - 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(); + _openWindows.Remove(sequence); + Volatile.Write(ref _pendingSubscribes, _openWindows.Count); + TrimUnclaimableLocked(); } } - private void DiscardParkedLocked() + /// + /// Discard parked events no open window can claim: those stamped before the oldest open + /// window's sequence (all of them once no window is open). Promoted (ready) events are kept — + /// they belong to a registered handle and await replay. Amortised O(1) per parked event. + /// + private void TrimUnclaimableLocked() { - if (_parkedByItemHandle.Count == 0) return; - var parked = 0; - foreach (var list in _parkedByItemHandle.Values) parked += list.Count; + var oldestOpen = _openWindows.Count == 0 ? long.MaxValue : _openWindows.Min; + while (_parkOrder.TryPeek(out var head) && head.Stamp < oldestOpen) + { + _parkOrder.Dequeue(); + if (head.Gone) continue; // already promoted or purged + // Per-handle queues are in park order too, so a live head of the global order is the + // head of its handle's queue. + var queue = _parkedByItemHandle[head.ItemHandle]; + queue.Dequeue(); + head.Gone = true; + if (queue.Count == 0) _parkedByItemHandle.Remove(head.ItemHandle); + Volatile.Write(ref _heldCount, _heldCount - 1); + } + } + + private void DiscardAllHeldLocked() + { + foreach (var parked in _parkOrder) parked.Gone = true; + _parkOrder.Clear(); _parkedByItemHandle.Clear(); - Volatile.Write(ref _heldCount, _heldCount - parked); + _ready.Clear(); + Volatile.Write(ref _heldCount, 0); + } + + /// + /// Drop parked events for handles that were just unsubscribed (#859 second review, I-3 + /// interim). A trailing event for a removed handle can be parked during another subscribe's + /// window; if MXAccess reuses that handle number for the other subscribe's item, promotion + /// would hand the old item's value to the new tag. Call when UnsubscribeBulk for + /// completes. It cannot catch an event still in flight + /// after that, nor tell a genuine early value for a reused number from a trailing one (it + /// drops both — a timeout is safer than another tag's value); a complete fix needs a + /// worker sequence on the SubscribeBulk reply (mxaccessgw, cross-repo). + /// + /// The handles whose unsubscribe completed. + public void PurgeParked(IReadOnlyList itemHandles) + { + if (Volatile.Read(ref _heldCount) == 0 || itemHandles.Count == 0) return; + lock (_parkGate) + { + foreach (var handle in itemHandles) + { + if (!_parkedByItemHandle.Remove(handle, out var queue)) continue; + foreach (var parked in queue) parked.Gone = true; + Volatile.Write(ref _heldCount, _heldCount - queue.Count); + } + } } /// /// 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 - /// promotion a registration performs, so an event is never parked after its handle's - /// promotion already ran (it would otherwise be stranded until the window closed). + /// unknown while a subscribe is in flight — park the event for replay. Also hands back the + /// promoted events that must be dispatched BEFORE this one (one gate acquisition per event + /// while anything is held). /// + /// + /// ⚠️ The lock-free fast path relies on a single-dispatcher invariant: only the ONE live + /// 's dispatch thread ever parks (increments _heldCount), and it + /// is the thread calling this. So when it reads _heldCount == 0, no older event for + /// this handle can be waiting anywhere — it would have to have been parked by this same + /// thread earlier — and resolving through the map cannot overtake one. Registrations only + /// move events from parked to ready (count unchanged) and the dispatch thread only removes + /// them. A second concurrently-dispatching pump on one registry would break this. + /// /// The data-change event. - /// The calling pump's epoch (). + /// The calling pump's session ( at its creation). /// The subscribers when the outcome is . + /// Promoted early events to dispatch before this one, in order. /// What happened to the event. public EarlyEventOutcome ResolveOrPark( - MxEvent ev, int epoch, out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers) + MxEvent ev, int session, + out IReadOnlyList<(long SubscriptionId, string FullReference)> subscribers, + out IReadOnlyList replayFirst) { - // 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) + replayFirst = []; + if (Volatile.Read(ref _heldCount) == 0 && Volatile.Read(ref _session) == session) { subscribers = ResolveSubscribers(ev.ItemHandle); if (subscribers.Count > 0) return EarlyEventOutcome.Resolved; @@ -310,7 +472,7 @@ internal sealed class SubscriptionRegistry lock (_parkGate) { - if (epoch != _epoch) + if (session != _session) { subscribers = []; return EarlyEventOutcome.Stale; // a dying pump of an earlier session @@ -319,28 +481,34 @@ internal sealed class SubscriptionRegistry 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)) + // Defensive (M-1): registration promotes atomically with its map writes, so a + // resolved handle should hold no parked list. If one ever does, promote it now + // (fail safe) — parking this LIVE event behind it would lose it to a later trim if + // no registration ever claimed the list. + if (_parkedByItemHandle.Remove(ev.ItemHandle, out var stranded)) { - pending.Add(ev); - Volatile.Write(ref _heldCount, _heldCount + 1); - subscribers = []; - return EarlyEventOutcome.Parked; + foreach (var parked in stranded) + { + parked.Gone = true; + _ready.Enqueue(parked.Event); + } } + replayFirst = TakeReadyLocked(); return EarlyEventOutcome.Resolved; } + + replayFirst = TakeReadyLocked(); if (_pendingSubscribes == 0) return EarlyEventOutcome.Stale; if (_heldCount >= MaxParkedEvents) return EarlyEventOutcome.Overflow; - if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var list)) + var entry = new ParkedEvent(_windowSequence, ev.ItemHandle, ev); + if (!_parkedByItemHandle.TryGetValue(ev.ItemHandle, out var queue)) { - list = []; - _parkedByItemHandle[ev.ItemHandle] = list; + queue = new Queue(); + _parkedByItemHandle[ev.ItemHandle] = queue; } - list.Add(ev); + queue.Enqueue(entry); + _parkOrder.Enqueue(entry); Volatile.Write(ref _heldCount, _heldCount + 1); return EarlyEventOutcome.Parked; } @@ -348,23 +516,28 @@ internal sealed class SubscriptionRegistry /// /// Dequeue every event a registration promoted, in arrival order. Called only from the - /// EventPump's single dispatch thread. A pump of an earlier epoch gets nothing. + /// EventPump's single dispatch thread (the wake path). A pump of an earlier session gets nothing. /// - /// The calling pump's epoch. + /// The calling pump's session. /// The promoted events; empty when none are waiting. - public IReadOnlyList TakeReadyEvents(int epoch) + public IReadOnlyList TakeReadyEvents(int session) { if (Volatile.Read(ref _heldCount) == 0) return []; lock (_parkGate) { - if (epoch != _epoch || _ready.Count == 0) return []; - var taken = _ready.ToArray(); - _ready.Clear(); - Volatile.Write(ref _heldCount, _heldCount - taken.Length); - return taken; + return session != _session ? [] : TakeReadyLocked(); } } + private IReadOnlyList TakeReadyLocked() + { + if (_ready.Count == 0) return []; + var taken = _ready.ToArray(); + _ready.Clear(); + Volatile.Write(ref _heldCount, _heldCount - taken.Length); + return taken; + } + private bool PromoteParkedEventsLocked(IReadOnlyList bindings) { if (_parkedByItemHandle.Count == 0) return false; @@ -372,21 +545,37 @@ internal sealed class SubscriptionRegistry 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); // stays counted in _heldCount until replayed + if (!_parkedByItemHandle.Remove(binding.ItemHandle, out var queue)) continue; + foreach (var parked in queue) + { + parked.Gone = true; // stays counted in _heldCount until replayed + _ready.Enqueue(parked.Event); + } promoted = true; } return promoted; } - private sealed class PendingSubscribeScope(SubscriptionRegistry owner) : IDisposable + private sealed class ParkedEvent(long stamp, int itemHandle, MxEvent ev) + { + /// Gets the newest window sequence when the event was parked. + public long Stamp { get; } = stamp; + /// Gets the event's item handle. + public int ItemHandle { get; } = itemHandle; + /// Gets the event. + public MxEvent Event { get; } = ev; + /// Gets or sets whether the event has left the parked set (promoted, purged or trimmed). + public bool Gone { get; set; } + } + + private sealed class PendingSubscribeScope(SubscriptionRegistry owner, long sequence) : IDisposable { private int _disposed; /// Closes this subscribe's early-event window (once). public void Dispose() { - if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe(); + if (Interlocked.Exchange(ref _disposed, 1) == 0) owner.EndPendingSubscribe(sequence); } } @@ -404,50 +593,6 @@ internal sealed class SubscriptionRegistry public IReadOnlyList<(long SubscriptionId, IReadOnlyList Bindings)> SnapshotEntries() => [.. _bySubscriptionId.Values.Select(entry => (entry.SubscriptionId, entry.Bindings))]; - /// - /// Replace an existing subscription's bindings with the item handles a post-reconnect - /// SubscribeBulk returned, rebuilding the reverse fan-out map so events on the new - /// handles dispatch and the now-dead pre-reconnect handles are dropped. No-op when the - /// subscription id is unknown (it was unsubscribed during the reconnect window). - /// - /// The subscription identifier. - /// The new tag bindings after reconnection. - public void Rebind(long subscriptionId, IReadOnlyList newBindings) - { - bool promoted; - lock (_parkGate) // map writes + promotion atomic — see Register - { - 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) - { - 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; - } - - _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); - } - if (promoted) ParkedEventsReady?.Invoke(); - } - /// /// Per-subscription bookkeeping. is an index /// over keyed by item handle so ResolveSubscribers @@ -463,14 +608,18 @@ internal sealed class SubscriptionRegistry public IReadOnlyList Bindings { get; } /// Gets the index of full references by item handle. public IReadOnlyDictionary FullRefByItemHandle { get; } + /// Gets the gateway session whose handles carries. + public int Session { get; } /// Initializes a new subscription entry. /// The subscription identifier. /// The tag bindings for the subscription. - public SubscriptionEntry(long subscriptionId, IReadOnlyList bindings) + /// The session the bindings' handles were issued by. + public SubscriptionEntry(long subscriptionId, IReadOnlyList bindings, int session) { SubscriptionId = subscriptionId; Bindings = bindings; + Session = session; var index = new Dictionary(bindings.Count); foreach (var binding in bindings) { 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 e970b6661..b68651989 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 @@ -45,9 +45,9 @@ public sealed class EarlyEventWindowTests public void Unknown_handle_without_a_pending_subscribe_is_stale() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; - registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Stale); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Stale); registry.ParkedEventCount.ShouldBe(0); } @@ -56,15 +56,15 @@ public sealed class EarlyEventWindowTests public void Parked_events_are_promoted_in_order_on_register() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; var woken = 0; registry.ParkedEventsReady += () => woken++; using (registry.BeginPendingSubscribe()) { - 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.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(7, 2), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(8, 9), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); registry.TakeReadyEvents(epoch).ShouldBeEmpty(); registry.Register(1, [new TagBinding("Tank.Level", 7)]); @@ -82,12 +82,12 @@ public sealed class EarlyEventWindowTests public void Rebind_promotes_parked_events_for_the_new_handles() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; registry.Register(1, [new TagBinding("Tank.Level", 7)]); using (registry.BeginPendingSubscribe()) { - registry.ResolveOrPark(DataChange(70, 5), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(70, 5), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); registry.Rebind(1, [new TagBinding("Tank.Level", 70)]); } @@ -99,11 +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 epoch = registry.CurrentSession; var first = registry.BeginPendingSubscribe(); var second = registry.BeginPendingSubscribe(); - registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); first.Dispose(); first.Dispose(); // idempotent — must not close the second window registry.PendingSubscribeCount.ShouldBe(1); @@ -111,7 +111,7 @@ public sealed class EarlyEventWindowTests second.Dispose(); registry.ParkedEventCount.ShouldBe(0); - registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Stale); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Stale); } /// @@ -122,9 +122,9 @@ public sealed class EarlyEventWindowTests public void A_new_window_never_inherits_the_previous_windows_leftovers() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; var first = registry.BeginPendingSubscribe(); - registry.ResolveOrPark(DataChange(7, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(7, 1), epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); first.Dispose(); using var second = registry.BeginPendingSubscribe(); @@ -138,35 +138,104 @@ public sealed class EarlyEventWindowTests public void Parking_is_bounded() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; using var window = registry.BeginPendingSubscribe(); var ev = DataChange(7, 1); for (var i = 0; i < SubscriptionRegistry.MaxParkedEvents; i++) { - registry.ResolveOrPark(ev, epoch, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(ev, epoch, out _, out _).ShouldBe(EarlyEventOutcome.Parked); } - registry.ResolveOrPark(ev, epoch, out _).ShouldBe(EarlyEventOutcome.Overflow); + registry.ResolveOrPark(ev, epoch, out _, out _).ShouldBe(EarlyEventOutcome.Overflow); registry.ParkedEventCount.ShouldBe(SubscriptionRegistry.MaxParkedEvents); } - /// Review #5: promoted-but-not-yet-replayed events still count toward the bound. + /// + /// Review #5 + M-2: promoted-but-not-yet-replayed events stay counted until the dispatch thread + /// takes them, and the next ResolveOrPark hands them back (in order) before anything else. + /// [Fact] - public void Ready_events_count_toward_the_bound() + public void Ready_events_stay_counted_until_handed_back_by_the_next_resolve() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var session = registry.CurrentSession; 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(7, 1), session, out _, out _); + registry.ResolveOrPark(DataChange(7, 2), session, out _, out _); + registry.Register(1, [new TagBinding("Tank.Level", 7)]); // promoted to ready + registry.ParkedEventCount.ShouldBe(2); - registry.ResolveOrPark(DataChange(8, 1), epoch, out _).ShouldBe(EarlyEventOutcome.Overflow); + registry.ResolveOrPark(DataChange(8, 3), session, out _, out var replayFirst).ShouldBe(EarlyEventOutcome.Parked); + replayFirst.Select(e => e.Value.DoubleValue).ShouldBe([1.0, 2.0]); + registry.ParkedEventCount.ShouldBe(1, "only the newly parked event for handle 8 remains held"); + } - registry.TakeReadyEvents(epoch).Count.ShouldBe(SubscriptionRegistry.MaxParkedEvents); + /// + /// I-2: windows that overlap so the open count never reaches 0 (concurrent reads) must not + /// accumulate unclaimed events — each is trimmed once every window that could claim it closed. + /// + [Fact] + public void Overlapping_windows_that_never_all_close_do_not_accumulate_unclaimed_events() + { + var registry = new SubscriptionRegistry(); + var session = registry.CurrentSession; + var open = registry.BeginPendingSubscribe(); + for (var i = 0; i < 1_000; i++) + { + // An unknown-handle event nobody will ever claim (e.g. a trailing event after a read's Remove). + registry.ResolveOrPark(DataChange(10_000 + i, i), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + var next = registry.BeginPendingSubscribe(); // overlap: the count never drops to 0 + open.Dispose(); + open = next; + registry.ParkedEventCount.ShouldBe(0, $"iteration {i}: the event's only possible owners have all closed"); + } + + // Positive control: an event parked inside the still-open window survives and is claimable. + registry.ResolveOrPark(DataChange(7, 42), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + var later = registry.BeginPendingSubscribe(); + later.Dispose(); // a window opened AFTER the park closing must not trim it + registry.ParkedEventCount.ShouldBe(1); + registry.Register(1, [new TagBinding("Tank.Level", 7)]); + registry.TakeReadyEvents(session).Single().Value.DoubleValue.ShouldBe(42.0); + open.Dispose(); + } + + /// I-3 interim: an unsubscribe purges events parked for its handles. + [Fact] + public void Purge_drops_events_parked_for_unsubscribed_handles() + { + var registry = new SubscriptionRegistry(); + var session = registry.CurrentSession; + using var window = registry.BeginPendingSubscribe(); + registry.ResolveOrPark(DataChange(7, 1), session, out _, out _); // trailing event of a removed item + registry.ResolveOrPark(DataChange(8, 2), session, out _, out _); // positive control + + registry.PurgeParked([7]); + + registry.ParkedEventCount.ShouldBe(1); + registry.Register(1, [new TagBinding("Reused", 7), new TagBinding("Other", 8)]); + registry.TakeReadyEvents(session).Select(e => e.ItemHandle).ShouldBe([8]); + } + + /// + /// M-1: a live event for a RESOLVED handle that (defensively) still has a parked list is never + /// parked behind it — the list is promoted and the event resolves, so no trim can lose it. + /// + [Fact] + public void A_resolved_handle_with_a_stranded_parked_list_promotes_it_and_resolves() + { + var registry = new SubscriptionRegistry(); + var session = registry.CurrentSession; + var window = registry.BeginPendingSubscribe(); + registry.Register(1, [new TagBinding("Tank.Level", 7)]); + registry.ForceParkForTests(DataChange(7, 1)); // the "should be impossible" stranded list + + registry.ResolveOrPark(DataChange(7, 2), session, out var subs, out var replayFirst) + .ShouldBe(EarlyEventOutcome.Resolved); + subs.Single().FullReference.ShouldBe("Tank.Level"); + replayFirst.Select(e => e.Value.DoubleValue).ShouldBe([1.0]); + window.Dispose(); 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. @@ -174,11 +243,11 @@ public sealed class EarlyEventWindowTests public void Registered_handle_resolves_while_a_window_is_open() { var registry = new SubscriptionRegistry(); - var epoch = registry.StartNewEpoch(); + var epoch = registry.CurrentSession; registry.Register(1, [new TagBinding("Tank.Level", 7)]); using var window = registry.BeginPendingSubscribe(); - registry.ResolveOrPark(DataChange(7, 1), epoch, out var subs).ShouldBe(EarlyEventOutcome.Resolved); + registry.ResolveOrPark(DataChange(7, 1), epoch, out var subs, out _).ShouldBe(EarlyEventOutcome.Resolved); subs.Single().FullReference.ShouldBe("Tank.Level"); } @@ -188,25 +257,71 @@ public sealed class EarlyEventWindowTests /// 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() + public void A_new_session_discards_held_events_and_fences_the_old_session() { var registry = new SubscriptionRegistry(); - var oldEpoch = registry.StartNewEpoch(); + var oldSession = registry.CurrentSession; 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.ResolveOrPark(DataChange(7, 111), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.ResolveOrPark(DataChange(8, 112), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked); registry.Register(1, [new TagBinding("Old.A", 8)]); // 8 promoted to ready, never replayed - var newEpoch = registry.StartNewEpoch(); + var newSession = registry.StartNewSession(); registry.ParkedEventCount.ShouldBe(0); - registry.TakeReadyEvents(newEpoch).ShouldBeEmpty(); + registry.TakeReadyEvents(newSession).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.TakeReadyEvents(newSession).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"); + registry.ResolveOrPark(DataChange(7, 1), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Stale, + "a dying old-session pump must not resolve through the map, even for a registered number"); + registry.ResolveOrPark(DataChange(9, 1), oldSession, out _, out _).ShouldBe(EarlyEventOutcome.Stale); + registry.TakeReadyEvents(oldSession).ShouldBeEmpty(); + registry.ResolveOrPark(DataChange(9, 1), newSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked, + "positive control: the current session still parks"); + } + + /// + /// I-1: a new session invalidates the reverse map but keeps the subscriptions, so a new-session + /// event on a number an old binding still holds PARKS (instead of resolving to the old tag) and + /// is promoted by the replay's Rebind to the right tag. + /// + [Fact] + public void A_new_session_invalidates_old_handles_so_colliding_numbers_park_until_rebind() + { + var registry = new SubscriptionRegistry(); + registry.Register(1, [new TagBinding("Old.A", 7)]); + registry.Register(2, [new TagBinding("Old.B", 8)]); + + var session = registry.StartNewSession(); + registry.TrackedSubscriptionCount.ShouldBe(2, "subscriptions survive for the replay to rebind"); + using var window = registry.BeginPendingSubscribe(); + // New session: Old.B is re-issued as handle 7 — the number Old.A still holds until its own Rebind. + registry.ResolveOrPark(DataChange(7, 5), session, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + registry.TryRebind(2, [new TagBinding("Old.B", 7)], session).ShouldBeTrue(); + + registry.TakeReadyEvents(session).ShouldHaveSingleItem().Value.DoubleValue.ShouldBe(5.0); + registry.ResolveSubscribers(7).ShouldBe([(2L, "Old.B")]); + } + + /// #861: a registration or rebind carrying a stale session is refused. + [Fact] + public void Registration_with_a_stale_session_is_refused() + { + var registry = new SubscriptionRegistry(); + var before = registry.CurrentSession; + registry.Register(1, [new TagBinding("Tank.Level", 7)]); + registry.StartNewSession(); + + registry.TryRegister(2, [new TagBinding("Late", 9)], before).ShouldBeFalse(); + registry.Contains(2).ShouldBeFalse(); + registry.TryRebind(1, [new TagBinding("Tank.Level", 9)], before).ShouldBeFalse(); + registry.ResolveSubscribers(9).ShouldBeEmpty(); + registry.TryRegister(3, [new TagBinding("Fresh", 9)], registry.CurrentSession).ShouldBeTrue("positive control"); + + registry.Remove(1, out var live).ShouldNotBeNull(); + live.ShouldBeFalse("never rebound in the new session — its handle 7 must not be unsubscribed"); + registry.Remove(3, out live).ShouldNotBeNull(); + live.ShouldBeTrue(); } // ===== EventPump ===== @@ -368,6 +483,7 @@ public sealed class EarlyEventWindowTests oldStream.Emit(DataChange(7, 111)); await TestWait.UntilAsync(() => registry.ParkedEventCount == 1, "the old session's event must be parked"); await oldPump.DisposeAsync(); + registry.StartNewSession(); // what GalaxyDriver.ReplayAsync does before recreating the pump var newStream = new ChannelSubscriber(); await using var newPump = new EventPump(newStream, registry, channelCapacity: 16, clientName: "New"); @@ -494,4 +610,172 @@ public sealed class EarlyEventWindowTests args.FullReference.ShouldBe("PlatformA.ScanState"); ((GalaxySubscriptionHandle)args.SubscriptionHandle).SubscriptionId.ShouldBe(SubscriptionRegistry.ProbeSubscriptionId); } + + // ===== Reconnect / session (#859 I-1, #861) ===== + + /// + /// A gateway fake whose SubscribeBulk calls follow a per-call script (default: fresh handles). + /// Events go on one stream that survives pump restarts. + /// + private sealed class ScriptedSubscriber : IGalaxySubscriber + { + private readonly Channel _events = Channel.CreateUnbounded(); + private int _nextHandle = 1_000; + + public ConcurrentQueue, CancellationToken, Task>>> Script { get; } = new(); + public ConcurrentQueue Unsubscribed { get; } = new(); + public int SubscribeCalls; + + public Task> SubscribeBulkAsync( + IReadOnlyList fullReferences, int bufferedUpdateIntervalMs, CancellationToken cancellationToken) + { + Interlocked.Increment(ref SubscribeCalls); + if (Script.TryDequeue(out var step)) return step(fullReferences, cancellationToken); + return Task.FromResult>( + [.. fullReferences.Select(r => Ok(r, Interlocked.Increment(ref _nextHandle)))]); + } + + 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) + { + await foreach (var ev in _events.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + { + yield return ev; + } + } + + public void Emit(MxEvent ev) => _events.Writer.TryWrite(ev).ShouldBeTrue(); + } + + private static SubscribeResult Ok(string tag, int handle) => + new() { TagAddress = tag, ItemHandle = handle, WasSuccessful = true }; + + /// + /// I-1 end to end: after a reconnect the new session hands Tank.A the number Tank.B held in the + /// old session (and vice versa), and each initial value arrives BEFORE its SubscribeBulk reply. + /// Every value must reach its own tag — not the tag that owned the number before the reconnect. + /// + [Fact] + public async Task Replay_with_colliding_handles_and_event_before_reply_delivers_each_value_to_its_own_tag() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + var registry = driver.SubscriptionsForTests; + var delivered = new ConcurrentQueue<(string Ref, double Value)>(); + driver.OnDataChange += (_, e) => delivered.Enqueue((e.FullReference, (double)e.Snapshot.Value!)); + + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 1)])); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 2)])); + await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken); + await driver.SubscribeAsync(["Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken); + + // New session: swapped numbers; each initial value is on the stream (and parked) before the reply. + var newHandle = new Dictionary { ["Tank.A"] = 2, ["Tank.B"] = 1 }; + var newValue = new Dictionary { ["Tank.A"] = 10.0, ["Tank.B"] = 20.0 }; + for (var i = 0; i < 2; i++) + { + subscriber.Script.Enqueue(async (refs, _) => + { + var tag = refs[0]; + var before = registry.ParkedEventCount; + subscriber.Emit(DataChange(newHandle[tag], newValue[tag])); + await TestWait.UntilAsync(() => registry.ParkedEventCount > before, + $"{tag}'s new-session initial value must PARK, not resolve to the old owner of handle {newHandle[tag]}"); + return [Ok(tag, newHandle[tag])]; + }); + } + + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + await TestWait.UntilAsync(() => delivered.Count >= 2, "both replayed initial values must be published", + () => string.Join(", ", delivered)); + delivered.OrderBy(d => d.Ref).ShouldBe([("Tank.A", 10.0), ("Tank.B", 20.0)]); + registry.ResolveSubscribers(1).Single().FullReference.ShouldBe("Tank.B"); + registry.ResolveSubscribers(2).Single().FullReference.ShouldBe("Tank.A"); + } + + /// + /// #861: a SubscribeAsync whose reply returns after a reconnect does not register the dead + /// old-session handles (nor unsubscribe their numbers) — it re-issues on the new session. + /// + [Fact] + public async Task Subscribe_whose_reply_straddles_a_reconnect_is_reissued_on_the_new_session() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + var registry = driver.SubscriptionsForTests; + var replyGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscriber.Script.Enqueue(async (refs, _) => + { + await replyGate.Task; + return [Ok(refs[0], 500)]; // an old-session handle + }); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 600)])); + + var subscribe = driver.SubscribeAsync(["Tank.X"], TimeSpan.Zero, TestContext.Current.CancellationToken); + await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the first SubscribeBulk must be in flight"); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); // the reconnect + replyGate.SetResult(); + + var handle = await TestWait.ForAsync(subscribe, "the subscribe must complete on the new session"); + Volatile.Read(ref subscriber.SubscribeCalls).ShouldBe(2); + registry.ResolveSubscribers(500).ShouldBeEmpty("the dead old-session handle must not be registered"); + registry.ResolveSubscribers(600).ShouldBe([(((GalaxySubscriptionHandle)handle).SubscriptionId, "Tank.X")]); + subscriber.Unsubscribed.ShouldNotContain(500, "its number may belong to a new-session item"); + } + + /// #861: a subscribe-once read whose reply straddles a reconnect reports the lost session. + [Fact] + public async Task Read_whose_reply_straddles_a_reconnect_returns_bad_communication_and_leaves_old_handles_alone() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + var registry = driver.SubscriptionsForTests; + var replyGate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscriber.Script.Enqueue(async (refs, _) => + { + await replyGate.Task; + return [Ok(refs[0], 700)]; + }); + + var read = driver.ReadAsync(["Tank.Y"], TestContext.Current.CancellationToken); + await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the read's SubscribeBulk must be in flight"); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + replyGate.SetResult(); + + var snapshot = (await TestWait.ForAsync(read, "the read must complete")).ShouldHaveSingleItem(); + snapshot.StatusCode.ShouldBe(StatusCodeMap.BadCommunicationError); + registry.ResolveSubscribers(700).ShouldBeEmpty(); + subscriber.Unsubscribed.ShouldBeEmpty(); + registry.PendingSubscribeCount.ShouldBe(0, "the read's early-event window must be closed"); + } + + /// The read's internal-deadline path closes its early-event window. + [Fact] + public async Task Read_deadline_path_closes_its_window() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) + { + ReadViaSubscribeDeadline = TimeSpan.FromMilliseconds(100), + }; + var never = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + subscriber.Script.Enqueue((_, _) => never.Task); + + var snapshot = (await TestWait.ForAsync( + driver.ReadAsync(["Tank.Z"], TestContext.Current.CancellationToken), "the read must time out")).ShouldHaveSingleItem(); + + snapshot.StatusCode.ShouldBe(StatusCodeMap.BadTimeout); + driver.SubscriptionsForTests.PendingSubscribeCount.ShouldBe(0); + never.SetResult([]); // let the abandoned RPC finish cleanly + } }