Merge pull request 'Cover concurrent gateway tag cache lookups' (#703) from test/historian-cache-concurrency into master

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.

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);
+    }
+}