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 7399b7f92..e5887e7cb 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Galaxy/GalaxyDriver.cs @@ -690,7 +690,13 @@ public sealed class GalaxyDriver private void ClearRegistry() { - lock (_registryGate) _subscriptions.Clear(); + lock (_registryGate) + { + _subscriptions.Clear(); + // A new session too: a late PurgeParked from a call the teardown orphaned (captured the old + // session) must never reach the next runtime's parked values. + _subscriptions.StartNewSession(); + } } /// @@ -1036,27 +1042,49 @@ public sealed class GalaxyDriver // 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; + // The probe entry is excluded: ONE registry entry holds every probe handle and any probe publish + // on the new session re-stamps it, so its session says nothing about which platforms are stale. + // The probe watcher tracks the session per handle and is driven separately below. + var session = _subscriptions.CurrentSession; + var entries = _subscriptions.SnapshotEntriesNotIn(session) + .Where(e => e.SubscriptionId != SubscriptionRegistry.ProbeSubscriptionId) + .ToArray(); + var probeWatcher = _probeWatcher; + if (entries.Length == 0 && probeWatcher is null) 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); + if (entries.Length > 0) + { + _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.Length); + } return; } - var tagCount = await ResubscribeAllAsync(entries, Volatile.Read(ref _runtimeGeneration), cancellationToken) - .ConfigureAwait(false); + var generation = Volatile.Read(ref _runtimeGeneration); + var tagCount = entries.Length == 0 + ? 0 + : await ResubscribeAllAsync(entries, generation, cancellationToken).ConfigureAwait(false); + + // Probes on EVERY reconnect, whatever the registry entry's session: re-subscribes exactly the + // platforms whose handles predate the current session (a platform a seed already advised on the + // new session is left alone — no double advise), and republishes the fenced bindings. + if (probeWatcher is not null + && Volatile.Read(ref _runtimeGeneration) == generation + && _subscriptions.CurrentSession == session) + { + tagCount += await probeWatcher.ResubscribeStaleAsync(cancellationToken).ConfigureAwait(false); + } _logger.LogInformation( "GalaxyDriver {InstanceId} replay completed — {SubCount} subscriptions, {TagCount} tags re-subscribed", - _driverInstanceId, entries.Count, tagCount); + _driverInstanceId, entries.Length, tagCount); } /// @@ -1555,7 +1583,9 @@ public sealed class GalaxyDriver var generation = Volatile.Read(ref _runtimeGeneration); // see SubscribeAsync var readSubscriber = _subscriber!; - var pump = EnsureEventPumpStarted(); + // A live pump must exist (values arrive through it); the read listens on ReadDataChange, not + // on this instance, so it survives a reconnect's pump restart. + EnsureEventPumpStarted(); var subscriptionId = _subscriptions.NextSubscriptionId(); // Pre-allocate one TaskCompletionSource per full-reference so the 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 19448404f..fe887b084 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 @@ -150,6 +150,30 @@ public sealed class PerPlatformProbeWatcher : IDisposable return platforms.Length; } + /// + /// The reconnect replay: re-issue SubscribeBulk only for platforms whose handle predates the + /// registry's current session, then republish. A platform a seed already advised on the new + /// session is left alone (re-subscribing it would double-advise and leak a handle), and every + /// reconnect calls this regardless of the shared registry entry's session — that entry is + /// re-stamped by any probe publish, so it cannot say which platforms are stale (#859). + /// + /// A cancellation token to cancel the operation. + /// The number of probe references re-subscribed. + public async Task ResubscribeStaleAsync(CancellationToken cancellationToken) + { + ObjectDisposedException.ThrowIf(_disposed, this); + string[] stale; + lock (_syncLock) + { + var session = CurrentSession; + stale = [.. _itemHandlesByPlatform.Where(kvp => kvp.Value.Session != session).Select(kvp => kvp.Key)]; + } + if (stale.Length == 0) return 0; + + await SubscribeAndRecordAsync(stale, cancellationToken).ConfigureAwait(false); + return stale.Length; + } + private async Task SubscribeAndRecordAsync(IReadOnlyList platforms, CancellationToken cancellationToken) { // #861 / #859 I-A: a reply that straddled a gateway reconnect carries dead handles; re-issue. 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 69989c2ad..1d0ade328 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 @@ -949,24 +949,86 @@ public sealed class EarlyEventWindowTests [Fact] public async Task Probe_watcher_never_publishes_or_unsubscribes_old_session_numbers() { - var registry = new SubscriptionRegistry(); var subscriber = new ScriptedSubscriber(); + using var driver = await InitializedDriverAsync(subscriber); + var watcher = driver.ProbeWatcherForTests.ShouldNotBeNull(); + var registry = driver.SubscriptionsForTests; 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(); + await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken); // gateway session swapped // The new session reuses number 1 for a user tag. - registry.Register(10, [new TagBinding("User.Tag", 1)]); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 1)])); + await driver.SubscribeAsync(["User.Tag"], TimeSpan.Zero, TestContext.Current.CancellationToken); - await watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken); // drops Q + await watcher.SyncPlatformsAsync(["P"], TestContext.Current.CancellationToken); // a seed in the gap 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"); + registry.ResolveSubscribers(1).Single().FullReference.ShouldBe("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"); + var replayed = new ConcurrentQueue(); + subscriber.Script.Enqueue((refs, _) => + { + foreach (var r in refs) replayed.Enqueue(r); + return Task.FromResult>([Ok(refs[0], 7)]); + }); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); // the driver's replay + + replayed.ToArray().ShouldBe(["P.ScanState"]); + watcher.CurrentBindings.Single().ItemHandle.ShouldBe(7); + registry.ResolveSubscribers(7).Single().FullReference.ShouldBe("P.ScanState"); + } + + /// + /// Final review blocker: a probe seed in the gap between the session swap and the replay + /// re-stamps the shared probe registry entry as current. The replay must still re-subscribe the + /// platforms whose handles predate the swap (they would otherwise freeze at handle 0) — and only + /// those: the platform the seed just advised on the new session is not advised twice. + /// + [Fact] + public async Task Replay_after_a_seed_in_the_gap_resubscribes_exactly_the_stale_platforms() + { + var subscriber = new ScriptedSubscriber(); + using var driver = await InitializedDriverAsync(subscriber); + var watcher = driver.ProbeWatcherForTests.ShouldNotBeNull(); + var registry = driver.SubscriptionsForTests; + subscriber.Script.Enqueue((refs, _) => Task.FromResult>( + [.. refs.Select((r, i) => Ok(r, i + 1))])); // P → 1, Q → 2 + await watcher.SyncPlatformsAsync(["P", "Q"], TestContext.Current.CancellationToken); + + await driver.InvokeReopenForTestAsync(TestContext.Current.CancellationToken); + subscriber.Script.Enqueue((refs, _) => Task.FromResult>([Ok(refs[0], 20)])); + await watcher.SyncPlatformsAsync(["P", "Q", "R"], TestContext.Current.CancellationToken); // seed adds R on the new session + registry.ResolveSubscribers(20).Single().FullReference.ShouldBe("R.ScanState"); + + var replayed = new ConcurrentQueue(); + subscriber.Script.Enqueue((refs, _) => + { + foreach (var r in refs) replayed.Enqueue(r); + return Task.FromResult>([.. refs.Select((r, i) => Ok(r, 30 + i))]); + }); + await driver.InvokeReplayForTestAsync(TestContext.Current.CancellationToken); + + replayed.OrderBy(r => r).ToArray().ShouldBe(["P.ScanState", "Q.ScanState"], "R was advised on the new session already"); + var bindings = watcher.CurrentBindings.ToDictionary(b => b.FullReference, b => b.ItemHandle); + bindings["R.ScanState"].ShouldBe(20); + bindings["P.ScanState"].ShouldBeGreaterThanOrEqualTo(30); + bindings["Q.ScanState"].ShouldBeGreaterThanOrEqualTo(30); + foreach (var (reference, handle) in bindings) + { + registry.ResolveSubscribers(handle).Single().FullReference.ShouldBe(reference); + } + } + + private static async Task InitializedDriverAsync(ScriptedSubscriber subscriber) + { + var driver = new GalaxyDriver( + "g", Opts(), hierarchySource: null, dataReader: null, dataWriter: null, subscriber: subscriber) + { + RecreateOverrideForTests = _ => Task.CompletedTask, + }; + await driver.InitializeAsync("{}", TestContext.Current.CancellationToken); + return driver; } }