diff --git a/docs/plans/2026-10-13-cache-concurrency-results.md b/docs/plans/2026-10-13-cache-concurrency-results.md new file mode 100644 index 00000000..b3ba5cf9 --- /dev/null +++ b/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. diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs index 80decb3f..52c526ef 100644 --- a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/FakeHistorianGatewayClient.cs +++ b/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; + /// Optional asynchronous catalog response for overlapping presence-cache tests. + public Func>? TagExistsSource; public Task 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); } diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Historian.Gateway.Tests/GatewayTagPresenceCacheConcurrencyTests.cs new file mode 100644 index 00000000..90feb181 --- /dev/null +++ b/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 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(() => 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(() => 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(() => 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(() => 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(() => read.WaitAsync(Ct)); + fake.TagExistsSource = null; + fake.TagExistsResult = true; + (await cache.ExistsAsync("T", Ct)).ShouldBeTrue(); + fake.TagExistsCallCount.ShouldBe(3); + } +}