test: accept definitive session loss reasons during restart

docs/plans/2026-10-13-cache-concurrency-results.md

@@ -0,0 +1,9 @@
+# Gateway tag-presence concurrency coverage — 2026-10-13
+
+Approved tests-only fix for #701. Added 15 deterministic cases for overlapping positive/negative cold lookups, expiry refreshes, capacity crossing, successful-result isolation from faults/cancellation, and cancellation after asynchronous completion. Fake presence calls support an async hook and atomic counter. Production cache is unchanged; no single-flight promise.
+
+The synthetic 16-caller cold/expiry bursts produced 16 catalog RPCs. Assertions permit one through16; this records amplification without making it a requirement or a live-cost claim. Newclass15/15 and Gateway449/449 passed. Independent review found no actionable defects; diff check clean.
+
+First full solution run:14,614passed/609skipped/1failed across62suites. Existing restart/reconnect wire test failed, then isolated retry passed1/1. Full repeat with detailed console logging passed14,615/609skip/0fail across62suites, exit0, including OPC UA integration122/122 and host integration268passed/12skip. #702 tracks the unresolved intermittent failure; it is not closed merely because the repeat passed.
+
+Both full runs excluded configured live fixture variables from the environment; environment-gated tests skipped. Existing NU1510/xUnit duplicate-InlineData warnings remain pre-existing. No production behavior or deployment changed by this patch.

docs/plans/2026-10-13-restart-loss-reason-results.md

@@ -0,0 +1,9 @@
+# Restart loss reason test correction — #702
+
+The full integration reproduction failed at the restart wire test reason assertion: it required BadServerHalted but correctly received BadSessionIdInvalid. Production already supports that definitive loss reason.
+
+The test now accepts an explicit semantic set: four direct definitive session statuses and two sustained closed-channel statuses. All BadNoCommunication, retained-value, unhealthy-state, frozen-read-stamp, one-notification and explicit reinitialization/recovery assertions remain unchanged. Only test expectation/comments changed; production unchanged. Existing deterministic driver tests independently enforce direct-status ordering/count and the sustained-channel threshold.
+
+Verification: isolated restart1/1, deterministic status routes30/30, full OPC UA integration122/122. Full offline solution: 14,613 passed / 611 skipped / 0 failed, exit0, 61 reported suite summaries. Host integration266 passed /14 skipped; unavailable environment/fixture tests remain skips. Independent review found no important defects; git diff --check passed. Actual prior red run121pass/1fail, detailed console/TRX confirms the exact reason mismatch.
+
+Logs/TRX retained in the Codex thread workspace under work/issue702-*. No Gateway or deployment behavior changed by this test correction. Separate live gates remain open.

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs

@@ -15,10 +15,13 @@ public sealed class FakeHistorianGatewayClient : IHistorianGatewayClient
     public bool TagExistsResult = true;
     public int TagExistsCallCount;
     public Exception? TagExistsThrows;
+    /// <summary>Optional asynchronous catalog response for overlapping presence-cache tests.</summary>
+    public Func<string, CancellationToken, Task<bool>>? TagExistsSource;
     public Task<bool> TagExistsAsync(string tag, CancellationToken ct)
     {
         ct.ThrowIfCancellationRequested();
-        TagExistsCallCount++;
+        Interlocked.Increment(ref TagExistsCallCount);
+        if (TagExistsSource is { } source) return source(tag, ct);
         if (TagExistsThrows is { } error) throw error;
         return Task.FromResult(TagExistsResult);
     }

tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs

@@ -0,0 +1,209 @@
+using Grpc.Core;
+using Shouldly;
+using Xunit;
+
+namespace ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests;
+
+public sealed class GatewayTagPresenceCacheConcurrencyTests(ITestOutputHelper output)
+{
+    private static CancellationToken Ct => TestContext.Current.CancellationToken;
+    private const int BurstSize = 16;
+
+    private sealed class Clock : TimeProvider
+    {
+        public DateTimeOffset Current = new(2026, 10, 13, 8, 0, 0, TimeSpan.Zero);
+        public override DateTimeOffset GetUtcNow() => Current;
+    }
+
+    private static TaskCompletionSource<bool> Gate() => new(TaskCreationOptions.RunContinuationsAsynchronously);
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_cold_lookups_return_the_catalog_result_and_populate_the_cache(bool exists)
+    {
+        // Catches dropped/corrupted concurrent writes or a result cached with the wrong presence value.
+        var gate = Gate();
+        var fake = new FakeHistorianGatewayClient { TagExistsSource = (_, _) => gate.Task };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount;
+        // Measure amplification without requiring either coalescing or one RPC per caller.
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Cold {exists}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A warm lookup must not contact the catalog.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(exists);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task Overlapping_expired_lookups_refresh_presence_at_the_TTL_boundary(bool initial, int ttlSeconds)
+    {
+        // Catches an inclusive expiry comparison or failure to replace an expired positive/negative entry.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        var expires = clock.Current.AddSeconds(ttlSeconds);
+        clock.Current = expires.AddTicks(-1);
+        fake.TagExistsThrows = new InvalidOperationException("The entry must still be warm before expiry.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+
+        clock.Current = expires;
+        var gate = Gate();
+        fake.TagExistsThrows = null;
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = Enumerable.Range(0, BurstSize).Select(_ => cache.ExistsAsync("T", Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        var rpcCount = fake.TagExistsCallCount - 1;
+        rpcCount.ShouldBeInRange(1, BurstSize);
+        output.WriteLine($"Expired {initial}: {BurstSize} overlapping callers, {rpcCount} catalog RPCs.");
+        gate.SetResult(!initial);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == !initial);
+
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("A refreshed lookup must stay cached.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(!initial);
+    }
+
+    [Theory]
+    [InlineData(false)]
+    [InlineData(true)]
+    public async Task Overlapping_distinct_tag_completions_keep_the_cache_bounded(bool exists)
+    {
+        // Catches racing capacity checks that allow more than Capacity resident entries.
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = exists };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var tags = Enumerable.Range(0, GatewayTagPresenceCache.Capacity + 31).Select(i => $"T{i}").ToArray();
+        foreach (var tag in tags.Take(GatewayTagPresenceCache.Capacity - 1))
+            (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists);
+
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var reads = tags.Skip(GatewayTagPresenceCache.Capacity - 1).Select(tag => cache.ExistsAsync(tag, Ct)).ToArray();
+        reads.ShouldAllBe(read => !read.IsCompleted);
+        gate.SetResult(exists);
+        (await Task.WhenAll(reads).WaitAsync(Ct)).ShouldAllBe(value => value == exists);
+
+        // Failed probes cannot insert or evict entries, so count resident hits through the public behavior.
+        fake.TagExistsSource = null;
+        var uncached = new InvalidOperationException("Uncached catalog probe.");
+        fake.TagExistsThrows = uncached;
+        var hits = 0;
+        foreach (var tag in tags)
+        {
+            try { (await cache.ExistsAsync(tag, Ct)).ShouldBe(exists); hits++; }
+            catch (InvalidOperationException error) when (ReferenceEquals(error, uncached)) { }
+        }
+        hits.ShouldBe(GatewayTagPresenceCache.Capacity);
+    }
+
+    [Theory]
+    [InlineData(StatusCode.Unavailable)]
+    [InlineData(StatusCode.PermissionDenied)]
+    [InlineData(StatusCode.Internal)]
+    [InlineData(StatusCode.Cancelled)]
+    public async Task A_faulting_lookup_cannot_overwrite_an_overlapping_positive_result(StatusCode code)
+    {
+        // Catches converting metadata faults to false, including a fault finishing after a successful lookup.
+        var success = Gate();
+        var failure = Gate();
+        var calls = 0;
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, _) => Interlocked.Increment(ref calls) == 1 ? success.Task : failure.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var badRead = cache.ExistsAsync("T", Ct);
+        goodRead.IsCompleted.ShouldBeFalse();
+        badRead.IsCompleted.ShouldBeFalse();
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        failure.SetException(new RpcException(new Status(code, "Catalog fault.")));
+        (await Should.ThrowAsync<RpcException>(() => badRead.WaitAsync(Ct))).StatusCode.ShouldBe(code);
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive the failed lookup.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+    }
+
+    [Fact]
+    public async Task Cancellation_after_a_catalog_result_cannot_overwrite_an_overlapping_positive_result()
+    {
+        // Catches removal of the post-await token check when the client ignores cancellation and returns false.
+        var success = Gate();
+        var cancelled = Gate();
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var fake = new FakeHistorianGatewayClient
+        {
+            TagExistsSource = (_, token) => token == caller.Token ? cancelled.Task : success.Task,
+        };
+        var cache = new GatewayTagPresenceCache(fake, new Clock());
+        var goodRead = cache.ExistsAsync("T", Ct);
+        var cancelledRead = cache.ExistsAsync("T", caller.Token);
+        success.SetResult(true);
+        (await goodRead.WaitAsync(Ct)).ShouldBeTrue();
+        caller.Cancel();
+        cancelled.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => cancelledRead.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsThrows = new InvalidOperationException("The positive entry must survive cancellation.");
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        // Cached results still respect an already-cancelled caller.
+        await Should.ThrowAsync<OperationCanceledException>(() => cache.ExistsAsync("T", caller.Token));
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_faulting_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches faults renewing an expired negative or replacing an expired positive with a negative.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        var read = cache.ExistsAsync("T", Ct);
+        read.IsCompleted.ShouldBeFalse();
+        gate.SetException(new RpcException(new Status(StatusCode.Unavailable, "Catalog unavailable.")));
+        await Should.ThrowAsync<RpcException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+
+    [Theory]
+    [InlineData(false, 5)]
+    [InlineData(true, 60)]
+    public async Task An_expired_entry_is_rechecked_after_a_cancelled_refresh(bool initial, int ttlSeconds)
+    {
+        // Catches cancellation renewing expired entries when the catalog returns an absence result anyway.
+        var clock = new Clock();
+        var fake = new FakeHistorianGatewayClient { TagExistsResult = initial };
+        var cache = new GatewayTagPresenceCache(fake, clock);
+        (await cache.ExistsAsync("T", Ct)).ShouldBe(initial);
+        clock.Current = clock.Current.AddSeconds(ttlSeconds);
+        var gate = Gate();
+        fake.TagExistsSource = (_, _) => gate.Task;
+        using var caller = CancellationTokenSource.CreateLinkedTokenSource(Ct);
+        var read = cache.ExistsAsync("T", caller.Token);
+        read.IsCompleted.ShouldBeFalse();
+        caller.Cancel();
+        gate.SetResult(false);
+        await Should.ThrowAsync<OperationCanceledException>(() => read.WaitAsync(Ct));
+        fake.TagExistsSource = null;
+        fake.TagExistsResult = true;
+        (await cache.ExistsAsync("T", Ct)).ShouldBeTrue();
+        fake.TagExistsCallCount.ShouldBe(3);
+    }
+}

tests/Server/ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests/OpcUaClientDriverReconnectWireTests.cs

@@ -13,8 +13,8 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
 ///     comms-loss watchdog trips on). Recovery is the watchdog's sequence, replayed here as <c>DriverInstanceActor</c>
 ///     runs it: ShutdownAsync, InitializeAsync, re-subscribe — after which the value comes back Good. There is no
 ///     in-driver reconnect any more (#670: the SDK reconnect handler kept everything Good and Healthy forever).
-///     Since #676 a gracefully stopped server is a DEFINITIVE loss (<c>BadServerHalted</c>), so the driver also raises
-///     <c>ConnectionLost</c> and the host runs that sequence at once instead of after the watchdog's silence window.
+///     Since #676 definitive session loss also raises <c>ConnectionLost</c>, so the host runs that sequence at once
+///     instead of after the watchdog's silence window. Shutdown timing determines the SDK's loss status.
 /// </summary>
 [Trait("Category", "Integration")]
 public sealed class OpcUaClientDriverReconnectWireTests
@@ -72,12 +72,22 @@ public sealed class OpcUaClientDriverReconnectWireTests
             await Task.Delay(TimeSpan.FromSeconds(3), ct); // three keep-alive intervals
             drv.GetHealth().LastSuccessfulRead.ShouldBe(frozen);
 
-            // #676: a server that HALTED (graceful stop) is a definitive loss — measured here, the SDK reports
-            // BadServerHalted on every keep-alive and never re-opens the channel, so the session can never recover in
-            // place. The driver says so at once, so the host re-initialises it instead of waiting out the ~90 s watchdog.
+            // #676/#702: shutdown can report a halted server, an invalid/closed session, or a sustained closed
+            // channel. Each proves definitive loss; #702 reproduced BadSessionIdInvalid during this graceful stop.
+            // Deterministic ConnectionLossSignal tests cover the individual statuses and the sustained-channel gate.
             await WaitAsync(() => { lock (lost) return lost.Count > 0; },
                 "a halted upstream must raise ConnectionLost", TimeSpan.FromSeconds(10));
-            lock (lost) lost.Single().ShouldContain("BadServerHalted");
+            lock (lost)
+            {
+                var reason = lost.Single();
+                string[] definitiveStatuses =
+                [
+                    "BadSessionIdInvalid", "BadSessionClosed", "BadSessionNotActivated", "BadServerHalted",
+                    "BadSecureChannelClosed", "BadConnectionClosed",
+                ];
+                definitiveStatuses.ShouldContain(status => reason.EndsWith(status, StringComparison.Ordinal),
+                    $"the restart must report definitive loss, but reported: {reason}");
+            }
         });
 
         // The restarted server knows nothing of the old session, so the driver stays down on its own …