diff --git a/docs/drivers/Galaxy.md b/docs/drivers/Galaxy.md index 0022abed1..303f7cb2a 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. 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. +- **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 `ReopenAsync` calls `SubscriptionRegistry.StartNewSession()` both before and after the gateway session swap (`ReplayAsync` does it only when no reopen ran), and the replay re-subscribes only subscriptions still holding the previous session's handles (one registered after the swap is already live). The pump is always restarted, even with nothing to replay. `StartNewSession()` 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` (a read rebound by the replay releases the handles the registry holds, not its original ones); an old session's pump can no longer resolve through the map. Reads listen on the driver, not on a pump instance, so a read survives the reconnect's pump restart. `PerPlatformProbeWatcher` records the session with each probe handle: only current-session handles are published (fenced `TryUpsert`), unsubscribed or purged, and a seed reply that straddled a reconnect is re-issued. Purges are fenced by the session captured before the unsubscribe RPC. **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 7e77a70ce..7399b7f92 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -250,6 +250,9 @@ public sealed class GalaxyDriver /// Test seam: true while an is running. internal bool HasEventPumpForTests { get { lock (_pumpLock) return _eventPump is not null; } } + /// Test seam — the current pump instance (identity only), to observe a restart. + internal object? EventPumpInstanceForTests { get { lock (_pumpLock) return _eventPump; } } + /// Test seam: the number of driver subscriptions the registry still tracks. internal int TrackedSubscriptionCountForTests => _subscriptions.TrackedSubscriptionCount; @@ -671,13 +674,17 @@ public sealed class GalaxyDriver private async Task UnsubscribeAndPurgeAsync( IGalaxySubscriber subscriber, IReadOnlyList itemHandles, CancellationToken cancellationToken) { + // Captured BEFORE the RPC (#859 I-B): a slow or failed unsubscribe can finish after a reconnect, + // when these numbers may be the NEW session's — the purge is then a no-op instead of deleting + // their parked initial values. + var session = _subscriptions.CurrentSession; try { await subscriber.UnsubscribeBulkAsync(itemHandles, cancellationToken).ConfigureAwait(false); } finally { - _subscriptions.PurgeParked(itemHandles); + _subscriptions.PurgeParked(itemHandles, session); } } @@ -948,14 +955,49 @@ public sealed class GalaxyDriver /// private async Task ReopenAsync(CancellationToken cancellationToken) { - if (_ownedMxSession is null) return; - var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false); - await _ownedMxSession.RecreateAsync(clientOptions, cancellationToken).ConfigureAwait(false); + if (_ownedMxSession is null && RecreateOverrideForTests is null) return; + + // #859 I-C: fence the registry around the swap itself, not later in ReplayAsync — a + // SubscribeBulk issued between RecreateAsync and the replay goes to the NEW gateway session. + // BEFORE: from here on nothing registered so far counts as live, so an UnsubscribeAsync during + // reopen/backoff never sends an old number (which the new session may reuse). + _subscriptions.StartNewSession(); + if (RecreateOverrideForTests is { } recreate) + { + await recreate(cancellationToken).ConfigureAwait(false); + } + else + { + var clientOptions = await BuildClientOptionsAsync(_options.Gateway, cancellationToken).ConfigureAwait(false); + await _ownedMxSession!.RecreateAsync(clientOptions, cancellationToken).ConfigureAwait(false); + } + // AFTER: a subscribe that captured the session between the two bumps may have dialled the old + // (dead) session, so it is refused and re-issued; one that captures it from here on dials the + // recreated session (GatewayGalaxySubscriber reads GalaxyMxSession.Session at call time), so + // it registers normally and the replay leaves it alone instead of re-subscribing it. + Volatile.Write(ref _sessionOpenedByReopen, _subscriptions.StartNewSession()); + // The recreated session invalidates every prior gw item handle; drop the writer's handle/advise // caches so the next write re-AddItems + re-AdviseSupervisory against the fresh session. _dataWriter?.InvalidateHandleCaches(); } + /// + /// The registry session opened for the recreated gateway session, or -1. + /// starts a new session itself only when no reopen did (the injected- + /// subscriber path, and tests); a second bump after a reopen would fence subscribes that already + /// registered against the recreated session. + /// + private int _sessionOpenedByReopen = -1; + + /// Test seam — replaces the gateway session swap inside . + internal Func? RecreateOverrideForTests { get; set; } + + /// Test-visible entry into . + /// Cancellation token. + /// A task that represents the asynchronous operation. + internal Task InvokeReopenForTestAsync(CancellationToken cancellationToken) => ReopenAsync(cancellationToken); + /// /// Replay callback. Walks every active subscription, re-issues SubscribeBulk for /// its tag list, and rebinds the @@ -977,32 +1019,38 @@ public sealed class GalaxyDriver // 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; - - // Honor ReplayOnSessionLost. When operators opt out (false) we skip the - // per-tag SubscribeBulk fan-out — they're delegating to the gateway's - // session-level ReplaySubscriptions or accept post-reconnect tag loss. We - // still restart the EventPump so a future Subscribe call lands on a live - // consumer. - if (!_options.Reconnect.ReplayOnSessionLost) + // held events, and fence any subscribe whose reply straddled the reconnect. ReopenAsync + // already did this around the session swap; only when no reopen ran (injected subscriber, + // tests) is it done here — FIRST, before the pump restart and even with nothing to replay. + if (Interlocked.Exchange(ref _sessionOpenedByReopen, -1) != _subscriptions.CurrentSession) { - RestartEventPumpForReplay(); - _logger.LogInformation( - "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false ({SubCount} subscriptions tracked)", - _driverInstanceId, entries.Count); - return; + _subscriptions.StartNewSession(); } // The stream-fault that triggered this recovery left the old pump's RunAsync loop // exited and its channel completed; EventPump.Start() is a no-op on a non-null but - // completed loop. Recreate the pump so the replayed subscriptions have a consumer. + // completed loop. Recreate the pump BEFORE anything can return early (#859 P-1): with no + // subscriptions to replay the dead pump used to stay in _eventPump, so every later + // Subscribe/Read reused it and was silently deaf while health stayed Healthy. RestartEventPumpForReplay(); + // Only subscriptions still holding the previous session's handles: one registered after the + // reopen is already live on the recreated session (re-subscribing it would double-advise). + var entries = _subscriptions.SnapshotEntriesNotIn(_subscriptions.CurrentSession); + if (entries.Count == 0) return; + + // Honor ReplayOnSessionLost. When operators opt out (false) the per-tag SubscribeBulk fan-out + // is skipped, and the recreated gateway session carries NONE of the old advises — those + // subscriptions stay deaf (their handles are dead and no longer resolve) until re-subscribed. + if (!_options.Reconnect.ReplayOnSessionLost) + { + _logger.LogWarning( + "GalaxyDriver {InstanceId} reconnect replay skipped — ReplayOnSessionLost=false; {SubCount} subscription(s) " + + "receive no further data until they are re-subscribed (the recreated gateway session has none of their advises).", + _driverInstanceId, entries.Count); + return; + } + var tagCount = await ResubscribeAllAsync(entries, Volatile.Read(ref _runtimeGeneration), cancellationToken) .ConfigureAwait(false); @@ -1032,6 +1080,9 @@ public sealed class GalaxyDriver CancellationToken cancellationToken) { var tagCount = 0; + // M-b: the session this pass belongs to, read ONCE. Any later change fences the rest of the + // pass (the new session's own replay takes over). + var session = _subscriptions.CurrentSession; foreach (var (subscriptionId, oldBindings) in entries) { if (Volatile.Read(ref _runtimeGeneration) != generation) @@ -1039,6 +1090,7 @@ public sealed class GalaxyDriver _logger.LogDebug("GalaxyDriver {InstanceId} re-subscribe abandoned — runtime was torn down.", _driverInstanceId); return tagCount; } + if (_subscriptions.CurrentSession != session) return tagCount; // a newer session's replay owns the rest if (subscriptionId == SubscriptionRegistry.ProbeSubscriptionId && _probeWatcher is { } probeWatcher) { @@ -1061,7 +1113,6 @@ 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); @@ -1079,18 +1130,41 @@ public sealed class GalaxyDriver // Rebind so the EventPump fan-out reverse map points at the post-reconnect // handles; otherwise events on the new handles miss every subscription. Not after a // teardown: the fresh handles belong to a subscriber that is already being released. + SubscriptionRegistry.BindOutcome outcome; lock (_registryGate) { if (Volatile.Read(ref _runtimeGeneration) != generation) return tagCount; - // #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)) + outcome = _subscriptions.TryRebind(subscriptionId, newBindings, session); + } + + if (outcome == SubscriptionRegistry.BindOutcome.SessionChanged) + { + // #861: another reconnect during the pass makes these handles dead too; the newer + // session's own replay re-subscribes everything, so stop here. + _logger.LogDebug( + "GalaxyDriver {InstanceId} re-subscribe abandoned — the gateway session changed again.", + _driverInstanceId); + return tagCount; + } + if (outcome == SubscriptionRegistry.BindOutcome.UnknownId) + { + // M-a: the subscription (or a subscribe-once read) was removed while its SubscribeBulk + // was in flight. Nothing will ever unsubscribe what was just advised — release it now. + var orphaned = newBindings.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); + if (orphaned.Length > 0) { - _logger.LogDebug( - "GalaxyDriver {InstanceId} re-subscribe abandoned — the gateway session changed again.", - _driverInstanceId); - return tagCount; + try + { + await UnsubscribeAndPurgeAsync(_subscriber!, orphaned, CancellationToken.None).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "GalaxyDriver {InstanceId} could not release {Count} handle(s) re-subscribed for a subscription removed during the pass.", + _driverInstanceId, orphaned.Length); + } } + continue; } tagCount += refs.Length; } @@ -1446,6 +1520,13 @@ public sealed class GalaxyDriver return ReadViaSubscribeOnceAsync(dialledRefs, cancellationToken); } + /// + /// Raw (attributeRef-domain) data changes for in-flight subscribe-once reads. Raised by + /// for whichever pump is current, so a read survives a reconnect's + /// pump restart. + /// + private event EventHandler? ReadDataChange; + /// Upper bound on one subscribe-once read; internal so tests can shrink it. internal TimeSpan ReadViaSubscribeDeadline { get; set; } = TimeSpan.FromSeconds(30); @@ -1500,7 +1581,7 @@ public sealed class GalaxyDriver tcs.TrySetResult(args.Snapshot); } }; - pump.OnDataChange += handler; + ReadDataChange += handler; var bufferedIntervalMs = _options.MxAccess.PublishingIntervalMs; IReadOnlyList results; @@ -1525,7 +1606,7 @@ public sealed class GalaxyDriver catch (OperationCanceledException) when (!callerToken.IsCancellationRequested) { // Internal deadline fired while the subscribe RPC was still in flight. - pump.OnDataChange -= handler; + ReadDataChange -= handler; var subscriber = readSubscriber; var logger = _logger; _ = subscribeTask.ContinueWith(t => @@ -1550,7 +1631,7 @@ public sealed class GalaxyDriver } catch { - pump.OnDataChange -= handler; + ReadDataChange -= handler; throw; } @@ -1584,7 +1665,7 @@ public sealed class GalaxyDriver var registered = TryRegisterForGeneration(subscriptionId, bindings, generation, session); if (registered != RegisterOutcome.Registered) { - pump.OnDataChange -= handler; + ReadDataChange -= handler; earlyEventWindow.Dispose(); // before any further round-trip if (registered == RegisterOutcome.TornDown) { @@ -1637,16 +1718,20 @@ public sealed class GalaxyDriver } finally { - pump.OnDataChange -= handler; + ReadDataChange -= handler; // 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. 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 && !staleSession) + // Unsubscribe what the registry held at removal, NOT the local `bindings` (#859 B-1): a + // reconnect replay during the wait may have rebound this read to new-session handles, and + // the original numbers may now be other tags' live items. If the entry still holds dead + // old-session handles (not yet rebound), release nothing (#861). Only when the entry was + // already gone (teardown cleared the registry) fall back to the original bindings. + IReadOnlyList source = removed ?? bindings; + if (removed is not null && !liveInCurrentSession) source = []; + var liveHandles = source.Where(b => b.ItemHandle > 0).Select(b => b.ItemHandle).ToArray(); + if (liveHandles.Length > 0) { try { @@ -2277,6 +2362,14 @@ public sealed class GalaxyDriver // supervisor Healthy transition, so a 90 s Degraded tripped a re-init under flowing data). StampLiveness(); + // Subscribe-once reads listen here (attributeRef domain), not on a pump instance: a reconnect + // replaces the pump, and a read attached to the dead one never saw the replayed value (#859 B-1). + try { ReadDataChange?.Invoke(this, args); } + catch (Exception ex) + { + _logger.LogWarning(ex, "GalaxyDriver {InstanceId} read listener threw — continuing.", _driverInstanceId); + } + // Probe traffic rides the reserved registry entry and stays in the attributeRef domain — // ScanState attributes are discovered platform attributes, never authored RawPaths. It is // the watcher's alone: the host never subscribed it and must not see it on OnDataChange. 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 d98bae7e0..19448404f 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 @@ -46,7 +46,10 @@ public sealed class PerPlatformProbeWatcher : IDisposable // Tracked platform → gw item handle. Item handle 0 means the gw rejected the subscribe; // we keep the entry so SyncPlatformsAsync doesn't try to subscribe it again on every call. - private readonly ConcurrentDictionary _itemHandlesByPlatform = + // Each handle carries the registry session that issued it (#859 I-A): item handles are per gateway + // session and the next session reuses the numbers, so an old-session handle must never be published, + // unsubscribed or purged against the current session. + private readonly ConcurrentDictionary _itemHandlesByPlatform = new(StringComparer.OrdinalIgnoreCase); private readonly Lock _syncLock = new(); // Written under _syncLock by StopAsync; read under it wherever a write-back must not outlive a stop. @@ -99,7 +102,14 @@ public sealed class PerPlatformProbeWatcher : IDisposable internal IReadOnlyList CurrentBindings => BuildBindings(); private IReadOnlyList BuildBindings() => - [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value))]; + [.. _itemHandlesByPlatform.Select(kvp => new TagBinding(kvp.Key + ProbeSuffix, kvp.Value.ItemHandle))]; + + /// The bindings to publish for : another session's handles bind as 0 (no events). + private IReadOnlyList BuildBindingsFor(int session) => + [.. _itemHandlesByPlatform.Select(kvp => new TagBinding( + kvp.Key + ProbeSuffix, kvp.Value.Session == session ? kvp.Value.ItemHandle : 0))]; + + private int CurrentSession => _registry?.CurrentSession ?? 0; /// /// Mirror the tracked handle set into the driver registry (no-op without one, or once stopped — @@ -107,15 +117,20 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// private void PublishBindings() { - lock (_syncLock) PublishBindingsLocked(); + lock (_syncLock) PublishBindingsLocked(CurrentSession); } - private void PublishBindingsLocked() + /// False when the registry's session is no longer (nothing published). + private bool PublishBindingsLocked(int session) { - if (_registry is null || _disposed) return; - var bindings = BuildBindings(); - if (bindings.Count == 0) _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId); - else _registry.Upsert(SubscriptionRegistry.ProbeSubscriptionId, bindings); + if (_registry is null || _disposed) return true; + if (_itemHandlesByPlatform.IsEmpty) + { + _registry.Remove(SubscriptionRegistry.ProbeSubscriptionId); + return true; + } + // Fenced (#859 I-A): only handles issued by `session` are bound, and only while it is current. + return _registry.TryUpsert(SubscriptionRegistry.ProbeSubscriptionId, BuildBindingsFor(session), session); } /// @@ -136,12 +151,31 @@ public sealed class PerPlatformProbeWatcher : IDisposable } private async Task SubscribeAndRecordAsync(IReadOnlyList platforms, CancellationToken cancellationToken) + { + // #861 / #859 I-A: a reply that straddled a gateway reconnect carries dead handles; re-issue. + for (var attempt = 1; ; attempt++) + { + if (await TrySubscribeAndRecordAsync(platforms, cancellationToken).ConfigureAwait(false)) return; + if (attempt >= MaxSessionAttempts) + { + _logger.LogWarning( + "PerPlatformProbeWatcher probe subscribe straddled {Attempts} gateway reconnects — giving up until the next replay/seed", attempt); + return; + } + } + } + + private const int MaxSessionAttempts = 3; + + /// False when the reply straddled a session change (nothing recorded; retry). + private async Task TrySubscribeAndRecordAsync(IReadOnlyList platforms, CancellationToken cancellationToken) { var probeAddresses = platforms.Select(p => p + ProbeSuffix).ToArray(); // #859: the initial ScanState value can overtake the SubscribeBulk reply; park it until // PublishBindingsLocked registers the handles below (window closes at method exit). using var earlyEventWindow = _registry?.BeginPendingSubscribe(); + var session = CurrentSession; // the session this SubscribeBulk is issued on // PR 6.3 — use the configured bufferedUpdateIntervalMs (defaults to 0 = gw cadence // when the driver hasn't overridden MxAccess.PublishingIntervalMs). Probe ScanState // changes are rare so a coarser interval is usually fine; deployments that need @@ -167,8 +201,15 @@ public sealed class PerPlatformProbeWatcher : IDisposable stopped = _disposed; if (!stopped) { - foreach (var (platform, itemHandle, _) in acquired) _itemHandlesByPlatform[platform] = itemHandle; - PublishBindingsLocked(); + if (_registry is not null && _registry.CurrentSession != session) + { + // Straddled a reconnect: these handles died with the old session and their numbers may + // be reused — neither record, publish nor unsubscribe them. + return false; + } + foreach (var (platform, itemHandle, _) in acquired) + _itemHandlesByPlatform[platform] = new ProbeHandle(itemHandle, session); + if (!PublishBindingsLocked(session)) return false; } } @@ -184,7 +225,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable "PerPlatformProbeWatcher could not release {Count} probe handle(s) acquired after stop", orphans.Length); } } - return; + return true; } foreach (var (platform, itemHandle, error) in acquired) @@ -194,6 +235,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable _logger.LogWarning("PerPlatformProbeWatcher subscribe failed for {Platform}: {Error}", platform, error); } } + return true; } /// @@ -228,12 +270,29 @@ public sealed class PerPlatformProbeWatcher : IDisposable // race on the membership view. List toAdd; List<(string Platform, int ItemHandle)> toRemove; + int removeSession; + bool droppedWithoutRelease; lock (_syncLock) { toAdd = [.. desired.Where(p => !_itemHandlesByPlatform.ContainsKey(p))]; + // Captured before the RPC (#859 I-B); only handles issued by the CURRENT session are released — + // an older session's numbers may now belong to unrelated live items (#859 I-A). + removeSession = CurrentSession; toRemove = [.. _itemHandlesByPlatform - .Where(kvp => !desired.Contains(kvp.Key) && kvp.Value > 0) - .Select(kvp => (kvp.Key, kvp.Value))]; + .Where(kvp => !desired.Contains(kvp.Key) + && kvp.Value.ItemHandle > 0 && kvp.Value.Session == removeSession) + .Select(kvp => (kvp.Key, kvp.Value.ItemHandle))]; + var deadRemovals = _itemHandlesByPlatform + .Where(kvp => !desired.Contains(kvp.Key) && !(kvp.Value.ItemHandle > 0 && kvp.Value.Session == removeSession)) + .Select(kvp => kvp.Key).ToArray(); + // Dropped platforms with no releasable handle (rejected, or an older session's dead number): + // forget them without any unsubscribe. + foreach (var platform in deadRemovals) + { + _itemHandlesByPlatform.TryRemove(platform, out _); + _aggregator.Remove(platform); + } + droppedWithoutRelease = deadRemovals.Length > 0; // Drop removed entries from the membership map up-front so a concurrent // OnProbeValueChanged for them is silently ignored. The unsubscribe RPC @@ -245,6 +304,8 @@ public sealed class PerPlatformProbeWatcher : IDisposable } } + if (droppedWithoutRelease && toRemove.Count == 0) PublishBindings(); // keep the fan-out map in step + if (toRemove.Count > 0) { // Drop the dead handles from the fan-out map before the (slow, fallible) unsubscribe. @@ -264,7 +325,7 @@ public sealed class PerPlatformProbeWatcher : IDisposable { // #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)]); + _registry?.PurgeParked([.. toRemove.Select(t => t.ItemHandle)], removeSession); } } @@ -334,7 +395,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable { if (_disposed) return; _disposed = true; - liveHandles = [.. _itemHandlesByPlatform.Values.Where(h => h > 0)]; + var session = CurrentSession; // never release an older session's (reusable) numbers + liveHandles = [.. _itemHandlesByPlatform.Values + .Where(h => h.ItemHandle > 0 && h.Session == session).Select(h => h.ItemHandle)]; _itemHandlesByPlatform.Clear(); _registry?.Remove(SubscriptionRegistry.ProbeSubscriptionId); } @@ -347,4 +410,9 @@ public sealed class PerPlatformProbeWatcher : IDisposable /// Disposes the probe watcher and unsubscribes all tracked platforms (blocking on the round-trip). public void Dispose() => StopAsync(CancellationToken.None).GetAwaiter().GetResult(); + + /// A probe item handle and the registry session that issued it. + /// The gateway item handle (0 = rejected). + /// The registry session the handle belongs to. + private readonly record struct ProbeHandle(int ItemHandle, int Session); } 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 303dde0c5..d7a76d37c 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 @@ -83,7 +83,7 @@ internal sealed class SubscriptionRegistry /// 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; + Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindOutcome.Bound; /// /// Replace an existing subscription's bindings with the item handles a post-reconnect @@ -100,13 +100,36 @@ internal sealed class SubscriptionRegistry /// 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; + /// + /// ; when the subscription was + /// removed meanwhile (the caller must release the handles it just acquired — nothing will + /// ever unsubscribe them, M-a); when the session moved on. + /// + public BindOutcome TryRebind(long subscriptionId, IReadOnlyList newBindings, int expectedSession) => + Bind(subscriptionId, newBindings, requireExisting: true, expectedSession); - private enum BindResult { Bound, UnknownId, SessionChanged } + /// + /// fenced by (the probe watcher, #859 I-A). + /// + /// The subscription identifier. + /// The tag bindings, all issued by . + /// The session the bindings' handles belong to. + /// False when the session changed (nothing published). + public bool TryUpsert(long subscriptionId, IReadOnlyList bindings, int expectedSession) => + Bind(subscriptionId, bindings, requireExisting: false, expectedSession) == BindOutcome.Bound; - private BindResult Bind( + /// Outcome of a fenced bind. + public enum BindOutcome + { + /// Bound. + Bound, + /// The subscription id is no longer tracked (rebind only). + UnknownId, + /// The gateway session changed since the handles were issued. + SessionChanged, + } + + private BindOutcome Bind( long subscriptionId, IReadOnlyList bindings, bool requireExisting, int? expectedSession) { bool promoted; @@ -117,7 +140,7 @@ internal sealed class SubscriptionRegistry // session check is in here too, so it is atomic with StartNewSession's invalidation. lock (_parkGate) { - if (expectedSession is { } session && session != _session) return BindResult.SessionChanged; + if (expectedSession is { } session && session != _session) return BindOutcome.SessionChanged; if (_bySubscriptionId.TryGetValue(subscriptionId, out var oldEntry)) { @@ -128,7 +151,7 @@ internal sealed class SubscriptionRegistry } else if (requireExisting) { - return BindResult.UnknownId; + return BindOutcome.UnknownId; } _bySubscriptionId[subscriptionId] = new SubscriptionEntry(subscriptionId, bindings, _session); @@ -144,7 +167,7 @@ internal sealed class SubscriptionRegistry promoted = PromoteParkedEventsLocked(bindings); } if (promoted) ParkedEventsReady?.Invoke(); - return BindResult.Bound; + return BindOutcome.Bound; } private void RemoveFromReverseMap(SubscriptionEntry entry) @@ -397,7 +420,7 @@ internal sealed class SubscriptionRegistry // head of its handle's queue. var queue = _parkedByItemHandle[head.ItemHandle]; queue.Dequeue(); - head.Gone = true; + head.Release(); if (queue.Count == 0) _parkedByItemHandle.Remove(head.ItemHandle); Volatile.Write(ref _heldCount, _heldCount - 1); } @@ -405,7 +428,7 @@ internal sealed class SubscriptionRegistry private void DiscardAllHeldLocked() { - foreach (var parked in _parkOrder) parked.Gone = true; + foreach (var parked in _parkOrder) parked.Release(); _parkOrder.Clear(); _parkedByItemHandle.Clear(); _ready.Clear(); @@ -423,15 +446,21 @@ internal sealed class SubscriptionRegistry /// worker sequence on the SubscribeBulk reply (mxaccessgw, cross-repo). /// /// The handles whose unsubscribe completed. - public void PurgeParked(IReadOnlyList itemHandles) + /// + /// The session the handles (and the unsubscribe) belong to, captured BEFORE the RPC. A slow or + /// failed unsubscribe can complete after a reconnect; purging then would delete the NEW + /// session's parked initial values for reused numbers, so it is a no-op (#859 I-B). + /// + public void PurgeParked(IReadOnlyList itemHandles, int session) { if (Volatile.Read(ref _heldCount) == 0 || itemHandles.Count == 0) return; lock (_parkGate) { + if (session != _session) return; foreach (var handle in itemHandles) { if (!_parkedByItemHandle.Remove(handle, out var queue)) continue; - foreach (var parked in queue) parked.Gone = true; + foreach (var parked in queue) parked.Release(); Volatile.Write(ref _heldCount, _heldCount - queue.Count); } } @@ -489,8 +518,7 @@ internal sealed class SubscriptionRegistry { foreach (var parked in stranded) { - parked.Gone = true; - _ready.Enqueue(parked.Event); + _ready.Enqueue(parked.Release()); } } replayFirst = TakeReadyLocked(); @@ -548,8 +576,7 @@ internal sealed class SubscriptionRegistry 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); + _ready.Enqueue(parked.Release()); // stays counted in _heldCount until replayed } promoted = true; } @@ -562,10 +589,19 @@ internal sealed class SubscriptionRegistry 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; } + /// Gets the event; null once it has left the parked set (M-c — not pinned by the trim index). + public MxEvent? Event { get; private set; } = ev; + /// Gets whether the event has left the parked set (promoted, purged or trimmed). + public bool Gone => Event is null; + + /// Takes the event out of this entry, so the trim index no longer pins it. + /// The event. + public MxEvent Release() + { + var ev = Event!; + Event = null; + return ev; + } } private sealed class PendingSubscribeScope(SubscriptionRegistry owner, long sequence) : IDisposable @@ -593,6 +629,16 @@ internal sealed class SubscriptionRegistry public IReadOnlyList<(long SubscriptionId, IReadOnlyList Bindings)> SnapshotEntries() => [.. _bySubscriptionId.Values.Select(entry => (entry.SubscriptionId, entry.Bindings))]; + /// + /// limited to subscriptions whose handles were NOT issued by + /// — the reconnect replay's work list. A subscription registered after + /// the reconnect already holds live handles; re-subscribing it would double-advise (#859 I-C). + /// + /// The current session. + /// The stale subscriptions with their bindings. + public IReadOnlyList<(long SubscriptionId, IReadOnlyList Bindings)> SnapshotEntriesNotIn(int session) => + [.. _bySubscriptionId.Values.Where(e => e.Session != session).Select(e => (e.SubscriptionId, e.Bindings))]; + /// /// Per-subscription bookkeeping. is an index /// over keyed by item handle so ResolveSubscribers 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 b68651989..69989c2ad 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 @@ -210,7 +210,7 @@ public sealed class EarlyEventWindowTests 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.PurgeParked([7], session); registry.ParkedEventCount.ShouldBe(1); registry.Register(1, [new TagBinding("Reused", 7), new TagBinding("Other", 8)]); @@ -297,7 +297,7 @@ public sealed class EarlyEventWindowTests 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.TryRebind(2, [new TagBinding("Old.B", 7)], session).ShouldBe(SubscriptionRegistry.BindOutcome.Bound); registry.TakeReadyEvents(session).ShouldHaveSingleItem().Value.DoubleValue.ShouldBe(5.0); registry.ResolveSubscribers(7).ShouldBe([(2L, "Old.B")]); @@ -314,7 +314,7 @@ public sealed class EarlyEventWindowTests registry.TryRegister(2, [new TagBinding("Late", 9)], before).ShouldBeFalse(); registry.Contains(2).ShouldBeFalse(); - registry.TryRebind(1, [new TagBinding("Tank.Level", 9)], before).ShouldBeFalse(); + registry.TryRebind(1, [new TagBinding("Tank.Level", 9)], before).ShouldBe(SubscriptionRegistry.BindOutcome.SessionChanged); registry.ResolveSubscribers(9).ShouldBeEmpty(); registry.TryRegister(3, [new TagBinding("Fresh", 9)], registry.CurrentSession).ShouldBeTrue("positive control"); @@ -683,9 +683,10 @@ public sealed class EarlyEventWindowTests subscriber.Script.Enqueue(async (refs, _) => { var tag = refs[0]; - var before = registry.ParkedEventCount; + // The previous iteration's promoted value stays counted until the pump replays it. + await TestWait.UntilAsync(() => registry.ParkedEventCount == 0, "earlier promoted values must be replayed first"); subscriber.Emit(DataChange(newHandle[tag], newValue[tag])); - await TestWait.UntilAsync(() => registry.ParkedEventCount > before, + await TestWait.UntilAsync(() => registry.ParkedEventCount > 0, $"{tag}'s new-session initial value must PARK, not resolve to the old owner of handle {newHandle[tag]}"); return [Ok(tag, newHandle[tag])]; }); @@ -778,4 +779,194 @@ public sealed class EarlyEventWindowTests driver.SubscriptionsForTests.PendingSubscribeCount.ShouldBe(0); never.SetResult([]); // let the abandoned RPC finish cleanly } + + // ===== Third review (B-1, I-A, I-B, I-C, P-1, M-a) ===== + + /// + /// B-1: a read rebound by a reconnect replay while it waited unsubscribes its NEW handle, not the + /// original number (which may by now be another tag's live item). + /// + [Fact] + public async Task Read_rebound_by_a_replay_unsubscribes_the_new_handle_not_the_original() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 1)])); + subscriber.Script.Enqueue((refs, _) => + { + subscriber.Emit(DataChange(9, 99.0)); // the replayed initial value (may overtake the reply) + return Task.FromResult>([Ok(refs[0], 9)]); + }); + + var read = driver.ReadAsync(["Tank.R"], TestContext.Current.CancellationToken); + await TestWait.UntilAsync(() => driver.TrackedSubscriptionCountForTests == 1, "the read must be registered and waiting"); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + var snapshot = (await TestWait.ForAsync(read, "the replayed value must complete the read")).ShouldHaveSingleItem(); + ((double)snapshot.Value!).ShouldBe(99.0); + subscriber.Unsubscribed.ToArray().ShouldBe([9], "the read must release its NEW handle, never the original number 1"); + } + + /// P-1: a reconnect with nothing to replay still replaces the dead pump; a later subscribe gets data. + [Fact] + public async Task Reconnect_with_no_subscriptions_restarts_the_pump() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + var delivered = new ConcurrentQueue(); + driver.OnDataChange += (_, e) => delivered.Enqueue(e); + var first = await driver.SubscribeAsync(["Tank.Old"], TimeSpan.Zero, TestContext.Current.CancellationToken); + await driver.UnsubscribeAsync(first, TestContext.Current.CancellationToken); + var deadPump = driver.EventPumpInstanceForTests.ShouldNotBeNull(); + + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + driver.EventPumpInstanceForTests.ShouldNotBeSameAs(deadPump, "the faulted pump must be replaced even with no entries"); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 42)])); + await driver.SubscribeAsync(["Tank.New"], TimeSpan.Zero, TestContext.Current.CancellationToken); + subscriber.Emit(DataChange(42, 4.2)); + await TestWait.UntilAsync(() => !delivered.IsEmpty, "a subscribe after the reconnect must receive data"); + delivered.Single().FullReference.ShouldBe("Tank.New"); + } + + /// + /// I-C: the reopen fences the registry around the session swap. An unsubscribe during the + /// reopen/backoff never sends an old number; a subscribe issued after the swap registers normally + /// and the replay does not re-subscribe it (no double advise). + /// + [Fact] + public async Task Subscribe_after_the_session_swap_but_before_replay_is_live_and_not_replayed() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) + { + RecreateOverrideForTests = _ => Task.CompletedTask, + }; + 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], 3)])); + await driver.SubscribeAsync(["Tank.A"], TimeSpan.Zero, TestContext.Current.CancellationToken); + var c = await driver.SubscribeAsync(["Tank.C"], TimeSpan.Zero, TestContext.Current.CancellationToken); + + await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken); // gateway session swapped + + await driver.UnsubscribeAsync(c, TestContext.Current.CancellationToken); + subscriber.Unsubscribed.ShouldNotContain(3, "an old-session number must never reach the new session"); + + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 50)])); + await driver.SubscribeAsync(["Tank.B"], TimeSpan.Zero, TestContext.Current.CancellationToken); + var callsBeforeReplay = Volatile.Read(ref subscriber.SubscribeCalls); + + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 51)])); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + (Volatile.Read(ref subscriber.SubscribeCalls) - callsBeforeReplay).ShouldBe(1, "only Tank.A (old session) is re-subscribed"); + var registry = driver.SubscriptionsForTests; + registry.ResolveSubscribers(50).Single().FullReference.ShouldBe("Tank.B"); + registry.ResolveSubscribers(51).Single().FullReference.ShouldBe("Tank.A"); + subscriber.Emit(DataChange(50, 5.0)); + await TestWait.UntilAsync(() => !delivered.IsEmpty, "the post-swap subscription must stay live across the replay"); + delivered.Single().ShouldBe(("Tank.B", 5.0)); + } + + /// + /// M-a: a subscription removed while the replay's SubscribeBulk for it was in flight gets the + /// freshly advised handles released (nothing else would ever unsubscribe them). + /// + [Fact] + public async Task Replay_releases_handles_for_a_subscription_removed_mid_replay() + { + var subscriber = new ScriptedSubscriber(); + using var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 1)])); + var handle = await driver.SubscribeAsync(["Tank.Gone"], TimeSpan.Zero, TestContext.Current.CancellationToken); + + subscriber.Script.Enqueue(async (refs, _) => + { + await driver.UnsubscribeAsync(handle, CancellationToken.None); // removed mid-replay + return [Ok(refs[0], 77)]; + }); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + subscriber.Unsubscribed.ToArray().ShouldBe([77], "the re-advised handle is released; the dead number 1 is not sent"); + driver.SubscriptionsForTests.ResolveSubscribers(77).ShouldBeEmpty(); + } + + /// I-B: a purge that completes after a reconnect does not delete the new session's parked values. + [Fact] + public void A_purge_for_an_older_session_is_a_no_op() + { + var registry = new SubscriptionRegistry(); + var oldSession = registry.CurrentSession; + var newSession = registry.StartNewSession(); + using var window = registry.BeginPendingSubscribe(); + registry.ResolveOrPark(DataChange(7, 1), newSession, out _, out _).ShouldBe(EarlyEventOutcome.Parked); + + registry.PurgeParked([7], oldSession); // a slow old-session unsubscribe finishing late + registry.ParkedEventCount.ShouldBe(1, "the new session's parked initial value must survive"); + + registry.PurgeParked([7], newSession); // positive control + registry.ParkedEventCount.ShouldBe(0); + } + + /// I-A: a probe seed whose reply straddles a reconnect records nothing dead and is re-issued. + [Fact] + public async Task Probe_seed_straddling_a_reconnect_is_reissued_and_never_publishes_the_dead_handle() + { + var registry = new SubscriptionRegistry(); + var subscriber = new ScriptedSubscriber(); + var gate = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + subscriber.Script.Enqueue(async (refs, _) => + { + await gate.Task; + return [Ok(refs[0], 5)]; // old-session handle + }); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 6)])); + using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry); + + var sync = watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken); + await TestWait.UntilAsync(() => Volatile.Read(ref subscriber.SubscribeCalls) == 1, "the seed's SubscribeBulk must be in flight"); + registry.StartNewSession(); + gate.SetResult(); + await TestWait.ForAsync(sync, "the seed must complete"); + + Volatile.Read(ref subscriber.SubscribeCalls).ShouldBe(2); + watcher.CurrentBindings.Single().ItemHandle.ShouldBe(6); + registry.ResolveSubscribers(5).ShouldBeEmpty(); + registry.ResolveSubscribers(6).Single().FullReference.ShouldBe("P.ScanState"); + subscriber.Unsubscribed.ShouldBeEmpty(); + } + + /// + /// I-A: after a reconnect (before the replay rebinds probes), a seed neither re-publishes the old + /// session's probe numbers nor unsubscribes a removed platform's old number. + /// + [Fact] + public async Task Probe_watcher_never_publishes_or_unsubscribes_old_session_numbers() + { + var registry = new SubscriptionRegistry(); + var subscriber = new ScriptedSubscriber(); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>( + [.. refs.Select((r, i) => Ok(r, i + 1))])); // P → 1, Q → 2 + using var watcher = new PerPlatformProbeWatcher(subscriber, new HostStatusAggregator(), registry); + await watcher.SyncPlatformsAsync(["P", "Q"], TestContext.Current.CancellationToken); + + registry.StartNewSession(); + // The new session reuses number 1 for a user tag. + registry.Register(10, [new TagBinding("User.Tag", 1)]); + + await watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken); // drops Q + + subscriber.Unsubscribed.ShouldBeEmpty("Q's handle 2 is an old-session number"); + registry.ResolveSubscribers(1).ShouldBe([(10L, "User.Tag")], "P's old-session number must not be re-published"); + + (await watcher.ResubscribeAsync(TestContext.Current.CancellationToken)).ShouldBe(1); // the replay + var fresh = watcher.CurrentBindings.Single().ItemHandle; + registry.ResolveSubscribers(fresh).Single().FullReference.ShouldBe("P.ScanState"); + } }