From a8d3790b090a8c963f83e996e4076974781935c5 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 11:07:59 -0500 Subject: [PATCH] =?UTF-8?q?fix(galaxy):=20#859=20final=20review=20?= =?UTF-8?q?=E2=80=94=20probe=20replay=20per=20platform=20session,=20teardo?= =?UTF-8?q?wn=20session=20bump?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit BLOCKER: the reconnect replay chose work from the registry entries' sessions, but ONE registry entry holds every probe handle and any probe publish on the new session (a seed in the reopen→replay gap, or the straddle retry) re-stamps it as current — the replay then skipped the probe watcher and the old platforms stayed bound to handle 0 (host status frozen); a publish after the snapshot made it re-subscribe everything (double advise, leaked handle). ReplayAsync now excludes the probe entry from its snapshot and calls the new PerPlatformProbeWatcher.ResubscribeStaleAsync on every reconnect, which re-subscribes exactly the platforms whose ProbeHandle.Session predates the current session. The replay-gap path keeps ResubscribeAsync (all platforms). Also: ClearRegistry (teardown) starts a new registry session, so a late PurgeParked from a teardown-orphaned call cannot reach the next runtime's parked values; the read path's unused pump local is gone (call kept). Tests go through the driver's reopen/replay: the probe old-session-number test is rewritten that way, plus the seed-in-the-gap scenario (only P and Q are re-subscribed, R keeps its new-session handle, all three resolve). Both checked red with the probe replay removed / re-subscribing all platforms. Refs #859, #861. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- .../GalaxyDriver.cs | 52 +++++++++--- .../Health/PerPlatformProbeWatcher.cs | 24 ++++++ .../Runtime/EarlyEventWindowTests.cs | 80 ++++++++++++++++--- 3 files changed, 136 insertions(+), 20 deletions(-) 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; } }