diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs index b467efce..2b10cc51 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs @@ -108,10 +108,22 @@ public class DebugStreamService : IDisposable instanceUniqueName = instance.UniqueName; siteIdentifier = site.SiteIdentifier; - grpcNodeAAddress = site.GrpcNodeAAddress - ?? throw new InvalidOperationException($"Site {siteIdentifier} has no GrpcNodeAAddress configured."); - grpcNodeBAddress = site.GrpcNodeBAddress - ?? throw new InvalidOperationException($"Site {siteIdentifier} has no GrpcNodeBAddress configured."); + + // Single-endpoint site (arch-review 2026-08-19 P3-3, applying the same N9 + // degeneration SiteAlarmLiveCacheService.ResolveSiteAsync already uses): reuse the + // sole configured endpoint for both slots, so the bridge actor's NodeA↔NodeB + // failover flip degenerates to a reconnect against the same node. Requiring BOTH + // addresses made Debug View the only surface that refused a one-node site. + var nodeA = site.GrpcNodeAAddress; + var nodeB = site.GrpcNodeBAddress; + if (string.IsNullOrWhiteSpace(nodeA) && string.IsNullOrWhiteSpace(nodeB)) + { + throw new InvalidOperationException( + $"Site {siteIdentifier} has no gRPC node addresses configured."); + } + + grpcNodeAAddress = !string.IsNullOrWhiteSpace(nodeA) ? nodeA! : nodeB!; + grpcNodeBAddress = !string.IsNullOrWhiteSpace(nodeB) ? nodeB! : nodeA!; } var sessionId = Guid.NewGuid().ToString("N"); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs index b8128f9f..f9973899 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SitePairChannelProvider.cs @@ -53,6 +53,14 @@ public sealed class SitePairChannelProvider : IDisposable private static readonly TimeSpan MaxBackoff = TimeSpan.FromSeconds(60); private static readonly TimeSpan FailbackSweepInterval = TimeSpan.FromSeconds(1); + /// + /// How long a replaced/removed channel is PARKED before it is disposed (arch-review + /// 2026-08-19 P3-2). Sized to exceed the longest command deadline + /// (CommunicationOptions.DeploymentTimeout and friends, minutes at most), so no call + /// that was already in flight when the channel was replaced can still be using it. + /// + internal static readonly TimeSpan ChannelDisposalGrace = TimeSpan.FromMinutes(5); + private readonly ISitePskProvider _pskProvider; private readonly CommunicationOptions _options; private readonly ILogger _logger; @@ -60,6 +68,14 @@ public sealed class SitePairChannelProvider : IDisposable private readonly Func> _reachabilityProbe; private readonly ConcurrentDictionary _sites = new(StringComparer.Ordinal); + + /// + /// Channels replaced by or dropped by , + /// each with the instant it becomes safe to dispose. FIFO order is disposal order because + /// the grace window is constant. + /// + private readonly ConcurrentQueue<(GrpcChannel Channel, DateTime DisposeAfterUtc)> _graveyard = new(); + private readonly CancellationTokenSource _shutdown = new(); private Task? _failbackLoop; private readonly object _loopGate = new(); @@ -98,8 +114,9 @@ public sealed class SitePairChannelProvider : IDisposable /// /// (Re)builds a site's channel pair from its gRPC node endpoints. A no-op when neither endpoint - /// changed. Disposes and rebuilds a channel whose endpoint changed, and resets stickiness to the - /// preferred node when the pair is (re)created. Called on the DB refresh tick. + /// changed. Parks (see ) and rebuilds a channel whose endpoint + /// changed, and resets stickiness to the preferred node when the pair is (re)created. Called on + /// the DB refresh tick. /// /// Site identifier. /// NodeA gRPC base address, or null when unconfigured. @@ -107,6 +124,7 @@ public sealed class SitePairChannelProvider : IDisposable public void UpdateSite(string siteId, string? nodeAEndpoint, string? nodeBEndpoint) { ArgumentException.ThrowIfNullOrWhiteSpace(siteId); + SweepGraveyard(); var pair = _sites.GetOrAdd(siteId, id => new SitePair(id)); lock (pair.Gate) @@ -120,13 +138,13 @@ public sealed class SitePairChannelProvider : IDisposable if (changedA) { - pair.ChannelA?.Dispose(); + Park(pair.ChannelA); pair.ChannelA = string.IsNullOrWhiteSpace(nodeAEndpoint) ? null : BuildChannel(nodeAEndpoint, siteId); pair.NodeAEndpoint = nodeAEndpoint; } if (changedB) { - pair.ChannelB?.Dispose(); + Park(pair.ChannelB); pair.ChannelB = string.IsNullOrWhiteSpace(nodeBEndpoint) ? null : BuildChannel(nodeBEndpoint, siteId); pair.NodeBEndpoint = nodeBEndpoint; } @@ -140,7 +158,10 @@ public sealed class SitePairChannelProvider : IDisposable } } - /// Disposes a removed site's channels and invalidates its cached preshared key. + /// + /// Parks a removed site's channels for deferred disposal (see ) + /// and invalidates its cached preshared key. + /// /// Site identifier to remove. public void RemoveSite(string siteId) { @@ -149,12 +170,14 @@ public sealed class SitePairChannelProvider : IDisposable return; } + SweepGraveyard(); + if (_sites.TryRemove(siteId, out var pair)) { lock (pair.Gate) { - pair.ChannelA?.Dispose(); - pair.ChannelB?.Dispose(); + Park(pair.ChannelA); + Park(pair.ChannelB); pair.ChannelA = null; pair.ChannelB = null; } @@ -284,6 +307,7 @@ public sealed class SitePairChannelProvider : IDisposable { while (await timer.WaitForNextTickAsync(ct).ConfigureAwait(false)) { + SweepGraveyard(); var now = DateTime.UtcNow; foreach (var kvp in _sites) { @@ -317,6 +341,67 @@ public sealed class SitePairChannelProvider : IDisposable } } + /// + /// Parks a channel that is no longer reachable through instead of + /// disposing it inline (arch-review 2026-08-19 P3-2). + /// + /// Why deferred. reads the channel under + /// pair.Gate but invokes the RPC OUTSIDE it, so an inline + /// Dispose from the 60 s DB address refresh (or an admin edit) aborted whatever was + /// in flight — a multi-minute DeployArtifactsCommand included — with + /// or . Neither is + /// , so the cross-node retry deliberately never engaged + /// and the command simply failed. + /// + /// + /// Why a timed park rather than reference counting. A refcount would have to be taken + /// and released around every ExecuteAsync call — bookkeeping on the hot path, plus a + /// quiescence handshake — to buy nothing a fixed window does not already give: the window + /// only has to outlast the longest command deadline. It also matches the precedent set by + /// , which likewise keeps a superseded endpoint's + /// channel alive rather than disposing it under a concurrent session. Cost is a handful of + /// idle HTTP/2 channels for at most . + /// + /// + /// The channel to park; null is ignored. + private void Park(GrpcChannel? channel) + { + if (channel is not null) + { + _graveyard.Enqueue((channel, DateTime.UtcNow + ChannelDisposalGrace)); + } + } + + /// + /// Disposes every parked channel whose grace window has elapsed. Called from the DB refresh + /// path (/) and from the failback loop, so a + /// park always drains even on a provider that never runs a call again. + /// + /// True to dispose every parked channel regardless of its window (shutdown, and tests). + internal void SweepGraveyard(bool force = false) + { + var now = DateTime.UtcNow; + while (_graveyard.TryPeek(out var parked) && (force || now >= parked.DisposeAfterUtc)) + { + if (!_graveyard.TryDequeue(out parked)) + { + break; + } + + try + { + parked.Channel.Dispose(); + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Disposing a parked gRPC channel faulted"); + } + } + } + + /// Test/diagnostic accessor: how many replaced channels are awaiting deferred disposal. + internal int ParkedChannelCount => _graveyard.Count; + private void FlipTo(SitePair pair, bool toIsA) { lock (pair.Gate) @@ -396,6 +481,10 @@ public sealed class SitePairChannelProvider : IDisposable } } _sites.Clear(); + + // Process shutdown — nothing can still be in flight, so drain the graveyard regardless + // of the remaining grace on each entry. + SweepGraveyard(force: true); } /// Per-site mutable channel state, guarded by . diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs index 9cd78de9..42c5539e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs @@ -138,7 +138,18 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable /// Registers a subscription's CancellationTokenSource for a correlation ID. /// If an entry already exists for that correlation ID (a reconnect race where two /// calls briefly share an ID), the prior CTS is - /// cancelled and disposed so it cannot leak. Internal for testability. + /// cancelled only — never disposed from here. Internal for testability. + /// + /// Cancel-only is the contract (arch-review 2026-08-19 P3-1, the client half of the + /// row-47 class): the prior handler's is still running and + /// still reads cts.Token (opening the call, ReadAllAsync, the + /// IsCancellationRequested checks), so disposing it from the REPLACING path throws + /// inside the prior stream — and, worse, out of the + /// prior.Cancel() call itself into the NEW SubscribeAsync, where it reads as a + /// fault of the current generation and burns a reconnect. Disposal belongs to the OWNING + /// path: / dispose the CTS they + /// created once their stream loop has fully unwound. + /// /// /// Unique identifier for the subscription. /// CancellationTokenSource for managing the subscription lifecycle. @@ -146,12 +157,30 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable { if (_subscriptions.TryGetValue(correlationId, out var prior) && !ReferenceEquals(prior, cts)) { - prior.Cancel(); - prior.Dispose(); + CancelQuietly(prior); } _subscriptions[correlationId] = cts; } + /// + /// Cancels a subscription token source, swallowing . + /// A CTS whose owning stream loop has already unwound and disposed it is simply a + /// subscription that no longer needs cancelling; the caller must never see that race as a + /// failure (arch-review 2026-08-19 P3-1). + /// + /// The token source to cancel. + private static void CancelQuietly(CancellationTokenSource cts) + { + try + { + cts.Cancel(); + } + catch (ObjectDisposedException) + { + // Already disposed by its owning stream loop — nothing left to cancel. + } + } + /// /// Removes the subscription entry for a correlation ID only if the stored CTS is /// exactly the one supplied. A racing replacement stream may already own the slot, @@ -213,13 +242,23 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable onEvent(domainEvent); } - await ConsumeStreamAsync( - correlationId, - cts, - () => _client.SubscribeInstance(request, cancellationToken: cts.Token), - frame => ForEachEvent(frame, Deliver), - onError, - onCompleted); + // This call OWNS `cts` — it created it, so it disposes it once the stream loop has + // fully unwound (P3-1). No other path may dispose it while this one is still reading + // cts.Token; a replacing subscription cancels only. + try + { + await ConsumeStreamAsync( + correlationId, + cts, + () => _client.SubscribeInstance(request, cancellationToken: cts.Token), + frame => ForEachEvent(frame, Deliver), + onError, + onCompleted); + } + finally + { + cts.Dispose(); + } } /// @@ -288,14 +327,22 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable onAlarmEvent(alarm); } - await ConsumeStreamAsync( - correlationId, - cts, - () => _client.SubscribeSite(request, cancellationToken: cts.Token), - frame => ForEachEvent(frame, Deliver), - onError, - onCompleted, - onConnected); + // Owns `cts` — see SubscribeAsync. + try + { + await ConsumeStreamAsync( + correlationId, + cts, + () => _client.SubscribeSite(request, cancellationToken: cts.Token), + frame => ForEachEvent(frame, Deliver), + onError, + onCompleted, + onConnected); + } + finally + { + cts.Dispose(); + } } /// @@ -449,15 +496,17 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable } /// - /// Cancels an active subscription by correlation ID. + /// Cancels an active subscription by correlation ID. Cancel-only: the subscription's own + /// / call disposes the token + /// source once its stream loop unwinds (P3-1). Cancelling an entry whose owner already + /// disposed it — or unsubscribing after the client was disposed — is a silent no-op. /// /// Unique identifier of the subscription to cancel. public virtual void Unsubscribe(string correlationId) { if (_subscriptions.TryRemove(correlationId, out var cts)) { - cts.Cancel(); - cts.Dispose(); + CancelQuietly(cts); } } @@ -601,17 +650,23 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable }; /// - /// Releases all subscription CancellationTokenSources and the underlying - /// gRPC channel. All teardown here is synchronous (CTS disposal and - /// ), so a synchronous - /// can release everything without sync-over-async blocking. + /// Cancels every live subscription and releases the underlying gRPC channel. All teardown + /// here is synchronous (CTS cancellation and ), so a + /// synchronous can release everything without sync-over-async blocking. + /// + /// Cancel-only, per-entry guarded (P3-1). The token sources are owned by their + /// / calls, which dispose them + /// as their loops unwind in response to this cancellation; disposing them here raced those + /// loops, and one aborted the whole foreach — which in + /// turn aborted 's client loop and left + /// every later client (and its channel) undisposed. + /// /// private void ReleaseResources() { foreach (var cts in _subscriptions.Values) { - cts.Cancel(); - cts.Dispose(); + CancelQuietly(cts); } _subscriptions.Clear(); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs index 2401b470..712cf457 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs @@ -120,7 +120,7 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable foreach (var key in _clients.Keys.Where(k => k.Site == siteIdentifier).ToList()) { if (_clients.TryRemove(key, out var client)) - await client.DisposeAsync(); + await DisposeClientAsync(client, key); } // Drop the cached preshared key too, so a site removed and re-added under the same @@ -129,16 +129,39 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable } /// - /// Asynchronously disposes all cached clients and clears the cache. + /// Asynchronously disposes all cached clients and clears the cache. Each client's disposal + /// is fault-isolated: one throwing client must not abort the loop and strand every later + /// client's channel undisposed (arch-review 2026-08-19 P3-1). /// /// A value task that completes when all clients have been disposed. public async ValueTask DisposeAsync() { - foreach (var client in _clients.Values) + foreach (var entry in _clients) + { + await DisposeClientAsync(entry.Value, entry.Key); + } + _clients.Clear(); + } + + /// + /// Disposes one client, logging and swallowing any failure so the caller's disposal loop + /// continues to the remaining clients. + /// + /// The client to dispose. + /// Cache key of the client, for diagnostics. + /// A task that completes when the client has been disposed (or its failure logged). + private async Task DisposeClientAsync(SiteStreamGrpcClient client, (string Site, string Endpoint) key) + { + try { await client.DisposeAsync(); } - _clients.Clear(); + catch (Exception ex) + { + _loggerFactory.CreateLogger().LogWarning( + ex, "Disposing the site stream gRPC client for site {SiteId} endpoint {Endpoint} failed", + key.Site, key.Endpoint); + } } /// @@ -147,12 +170,22 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable /// a stall/deadlock risk during host shutdown). Each /// releases all of its resources /// synchronously, so we dispose them directly with no async path. + /// Fault-isolated per client, exactly as is. /// public void Dispose() { - foreach (var client in _clients.Values) + foreach (var entry in _clients) { - client.Dispose(); + try + { + entry.Value.Dispose(); + } + catch (Exception ex) + { + _loggerFactory.CreateLogger().LogWarning( + ex, "Disposing the site stream gRPC client for site {SiteId} endpoint {Endpoint} failed", + entry.Key.Site, entry.Key.Endpoint); + } } _clients.Clear(); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs index 5a69a6af..f3ca559c 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs @@ -164,4 +164,93 @@ public class DebugStreamServiceTests : TestKit DebugStreamBridgeActor.ConsumerLivenessCheckInterval = previousCheck; } } + + // --- arch-review 2026-08-19 P3-3: single-endpoint sites --- + + [Theory] + [InlineData("http://localhost:5100", null)] + [InlineData(null, "http://localhost:5200")] + public async Task StartStreamAsync_SingleEndpointSite_UsesTheSoleAddressForBothNodeSlots( + string? nodeA, string? nodeB) + { + // Debug View used to hard-require BOTH gRPC node addresses and threw before ever + // creating the bridge actor, though single-endpoint sites are supported everywhere else + // (SiteAlarmLiveCacheService's N9 degeneration, SitePairChannelProvider's null-tolerant + // pair). The sole configured address now fills both slots, so the bridge actor's + // NodeA↔NodeB flip degenerates to a reconnect against the same node. + var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 }; + var site = new Site("Site One", "site-1") + { + Id = 3, + GrpcNodeAAddress = nodeA, + GrpcNodeBAddress = nodeB + }; + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstanceByIdAsync(7, Arg.Any()).Returns(instance); + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(3, Arg.Any()).Returns(site); + + var services = new ServiceCollection(); + services.AddScoped(_ => instanceRepo); + services.AddScoped(_ => siteRepo); + using var provider = services.BuildServiceProvider(); + + var commProbe = CreateTestProbe(); + var commService = new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + commService.SetCommunicationActor(commProbe.Ref); + + using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory( + new Grpc.MockSiteStreamGrpcClient()); + using var service = new DebugStreamService( + commService, provider, grpcFactory, NullLogger.Instance); + service.SetActorSystem(Sys); + + // Act — resolution must get past the address check and start a bridge actor. + var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { }); + + commProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + var bridgeActor = commProbe.LastSender; + + // The stream it opened dials the ONE configured address. + Assert.Equal(new[] { nodeA ?? nodeB }, grpcFactory.RequestedEndpoints); + + // Unwind the session. + bridgeActor.Tell(new DebugStreamTerminated("site-1", "corr")); + await Assert.ThrowsAsync(() => startTask); + } + + [Fact] + public async Task StartStreamAsync_SiteWithNoGrpcAddresses_StillThrows() + { + // Degenerating a single endpoint must not degenerate into accepting NONE. + var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 }; + var site = new Site("Site One", "site-1") { Id = 3 }; + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstanceByIdAsync(7, Arg.Any()).Returns(instance); + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(3, Arg.Any()).Returns(site); + + var services = new ServiceCollection(); + services.AddScoped(_ => instanceRepo); + services.AddScoped(_ => siteRepo); + using var provider = services.BuildServiceProvider(); + + var commService = new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + commService.SetCommunicationActor(CreateTestProbe().Ref); + + using var grpcFactory = new SiteStreamGrpcClientFactory(NullLoggerFactory.Instance); + using var service = new DebugStreamService( + commService, provider, grpcFactory, NullLogger.Instance); + service.SetActorSystem(Sys); + + var ex = await Assert.ThrowsAsync(() => + service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { })); + Assert.Contains("no gRPC node addresses", ex.Message); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderDisposalTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderDisposalTests.cs new file mode 100644 index 00000000..ff0e459e --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderDisposalTests.cs @@ -0,0 +1,105 @@ +using Grpc.Net.Client; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Communication.Grpc; + +namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; + +/// +/// Arch-review 2026-08-19 P3-2 — a channel replaced by the 60 s DB address refresh (or an admin +/// edit) must not be disposed out from under an in-flight +/// call. ExecuteAsync reads the channel under the pair gate but runs the RPC outside it, so an +/// inline dispose aborted whatever was running — a multi-minute DeployArtifactsCommand +/// included — with a status the cross-node retry deliberately never engages on. +/// +public class SitePairChannelProviderDisposalTests +{ + private static SitePairChannelProvider Build() => new( + new NoKeyProvider(), + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + + [Fact] + public async Task UpdateSite_WhileCallInFlight_ParksTheOldChannelAndLetsTheCallComplete() + { + using var provider = Build(); + provider.UpdateSite("site-1", "http://localhost:15101", nodeBEndpoint: null); + + var entered = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + GrpcChannel? inFlightChannel = null; + + // A long-running command (deploy) against the site's current channel. + var call = provider.ExecuteAsync("site-1", async (channel, ct) => + { + inFlightChannel = channel; + entered.SetResult(); + await release.Task; + // A disposed GrpcChannel throws here — this is the assertion that the address + // refresh did not pull the channel out from under the call. + _ = channel.CreateCallInvoker(); + return 42; + }, CancellationToken.None); + + await entered.Task.WaitAsync(TimeSpan.FromSeconds(5)); + + // The DB refresh tick re-points the site at a new address, mid-call. + provider.UpdateSite("site-1", "http://localhost:15102", nodeBEndpoint: null); + + // The superseded channel is PARKED, not disposed synchronously. + Assert.Equal(1, provider.ParkedChannelCount); + + release.SetResult(); + Assert.Equal(42, await call.WaitAsync(TimeSpan.FromSeconds(5))); + + // Once the grace window elapses the park drains and the channel really is disposed — + // deferred disposal, not a leak. + provider.SweepGraveyard(force: true); + Assert.Equal(0, provider.ParkedChannelCount); + Assert.Throws(() => inFlightChannel!.CreateCallInvoker()); + } + + [Fact] + public void RemoveSite_ParksBothChannelsRatherThanDisposingThemInline() + { + using var provider = Build(); + provider.UpdateSite("site-1", "http://localhost:15111", "http://localhost:15112"); + + provider.RemoveSite("site-1"); + + // Same hazard as UpdateSite: an admin deleting a site while a command is in flight. + Assert.Equal(2, provider.ParkedChannelCount); + + provider.SweepGraveyard(force: true); + Assert.Equal(0, provider.ParkedChannelCount); + } + + [Fact] + public void UpdateSite_UnchangedEndpoints_ParksNothing() + { + using var provider = Build(); + provider.UpdateSite("site-1", "http://localhost:15121", "http://localhost:15122"); + provider.UpdateSite("site-1", "http://localhost:15121", "http://localhost:15122"); + + Assert.Equal(0, provider.ParkedChannelCount); + } + + [Fact] + public void Dispose_DrainsParkedChannelsRegardlessOfRemainingGrace() + { + var provider = Build(); + provider.UpdateSite("site-1", "http://localhost:15131", nodeBEndpoint: null); + provider.UpdateSite("site-1", "http://localhost:15132", nodeBEndpoint: null); + Assert.Equal(1, provider.ParkedChannelCount); + + provider.Dispose(); + + Assert.Equal(0, provider.ParkedChannelCount); + } + + private sealed class NoKeyProvider : ISitePskProvider + { + public ValueTask GetAsync(string siteId, CancellationToken ct) => new("k"); + public void Invalidate(string siteId) { } + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryDisposeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryDisposeTests.cs index fecf5741..210a4f23 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryDisposeTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryDisposeTests.cs @@ -66,6 +66,74 @@ public class SiteStreamGrpcClientFactoryDisposeTests }); } + /// + /// Test client whose disposal always throws — stands in for a client whose + /// ReleaseResources hit an on a subscription + /// token source (arch-review 2026-08-19 P3-1). + /// + private sealed class ThrowingClient : SiteStreamGrpcClient + { + public override void Dispose() => throw new ObjectDisposedException(nameof(CancellationTokenSource)); + + public override ValueTask DisposeAsync() => + throw new ObjectDisposedException(nameof(CancellationTokenSource)); + } + + /// Hands out one throwing client followed by tracking clients. + private sealed class OneThrowingClientFactory : SiteStreamGrpcClientFactory + { + private readonly ConcurrentBag _created = new(); + private int _next; + + public OneThrowingClientFactory() : base(NullLoggerFactory.Instance) { } + + public IReadOnlyCollection Created => _created.ToList(); + + protected override SiteStreamGrpcClient CreateClient(string siteIdentifier, string grpcEndpoint) + { + if (Interlocked.Increment(ref _next) == 1) + { + return new ThrowingClient(); + } + + var client = new TrackingClient(); + _created.Add(client); + return client; + } + } + + [Fact] + public async Task DisposeAsync_OneClientThrowing_StillDisposesEveryOtherClient() + { + // The disposal loop used to abort on the first throwing client, leaving every LATER + // client — and its gRPC channel — undisposed for the life of the process. + var factory = new OneThrowingClientFactory(); + factory.GetOrCreate("site-a", "http://localhost:5100"); // throwing + factory.GetOrCreate("site-b", "http://localhost:5200"); + factory.GetOrCreate("site-c", "http://localhost:5300"); + + await factory.DisposeAsync(); + + Assert.Equal(2, factory.Created.Count); + Assert.All(factory.Created, c => + Assert.True(c.AsyncDisposeCalled, "a throwing client must not strand its peers undisposed")); + } + + [Fact] + public void Dispose_OneClientThrowing_StillDisposesEveryOtherClient() + { + var factory = new OneThrowingClientFactory(); + factory.GetOrCreate("site-a", "http://localhost:5100"); // throwing + factory.GetOrCreate("site-b", "http://localhost:5200"); + factory.GetOrCreate("site-c", "http://localhost:5300"); + + factory.Dispose(); + + Assert.Equal(2, factory.Created.Count); + Assert.All(factory.Created, c => + Assert.True(c.SyncDisposeCalled, "a throwing client must not strand its peers undisposed")); + } + [Fact] public void Dispose_DoesNotDeadlock_UnderSingleThreadedSynchronizationContext() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs index 84978f08..38010d95 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs @@ -677,26 +677,91 @@ public class SiteStreamGrpcClientTests // --- Communication-003 regression tests --- [Fact] - public void RegisterSubscription_ReusedCorrelationId_CancelsAndDisposesPriorCts() + public void RegisterSubscription_ReusedCorrelationId_CancelsPriorCts_WithoutDisposingIt() { // Two SubscribeAsync calls briefly sharing a correlation ID (reconnect race). - // Inserting the second must cancel + dispose the first so it does not leak. + // Inserting the second must CANCEL the first — and must NOT dispose it (arch-review + // 2026-08-19 P3-1, re-pinning the earlier cancel-AND-dispose contract): the prior + // handler's ConsumeStreamAsync is still running and still reads prior.Token, so + // disposing from the replacing path throws ObjectDisposedException inside that loop. + // The owning SubscribeAsync/SubscribeSiteAsync disposes it as its loop unwinds. var client = SiteStreamGrpcClient.CreateForTesting(); - var first = new CancellationTokenSource(); - var second = new CancellationTokenSource(); + using var first = new CancellationTokenSource(); + using var second = new CancellationTokenSource(); client.RegisterSubscription("corr-shared", first); client.RegisterSubscription("corr-shared", second); Assert.True(first.IsCancellationRequested); - // Disposed CTS throws ObjectDisposedException when its token is touched. - Assert.Throws(() => _ = first.Token); + // Still usable by the prior handler that owns it — no ObjectDisposedException. + Assert.True(first.Token.IsCancellationRequested); // The second (live) CTS must remain intact. Assert.False(second.IsCancellationRequested); } + [Fact] + public void RegisterSubscription_ReplacingAnAlreadyDisposedPriorCts_DoesNotThrowIntoTheNewSubscription() + { + // The prior handler's loop already unwound and disposed its own CTS while a replacement + // subscription is registering. The unguarded prior.Cancel() threw ObjectDisposedException + // straight out of the NEW SubscribeAsync, which reads it as a fault of the CURRENT + // generation and burns a reconnect on a phantom failure. + var client = SiteStreamGrpcClient.CreateForTesting(); + + var prior = new CancellationTokenSource(); + client.RegisterSubscription("corr-shared", prior); + prior.Dispose(); + + using var replacement = new CancellationTokenSource(); + client.RegisterSubscription("corr-shared", replacement); + + // The replacement owns the slot and is untouched. + Assert.False(replacement.IsCancellationRequested); + client.Unsubscribe("corr-shared"); + Assert.True(replacement.IsCancellationRequested); + } + + [Fact] + public void Unsubscribe_AfterClientDisposal_IsANoOp() + { + // Disposal cancels every live subscription and clears the map; a late Unsubscribe from + // the consuming actor's teardown must not throw (an ObjectDisposedException here used to + // escape into the caller's cleanup path). + var client = SiteStreamGrpcClient.CreateForTesting(); + using var cts = new CancellationTokenSource(); + client.AddSubscriptionForTesting("corr-1", cts); + + client.Dispose(); + Assert.True(cts.IsCancellationRequested); + + client.Unsubscribe("corr-1"); + client.Unsubscribe("never-registered"); + } + + [Fact] + public void Dispose_WithAnAlreadyDisposedSubscriptionCts_StillCancelsTheRest() + { + // ReleaseResources iterated the live CTSes unguarded, so one already-disposed entry + // aborted the whole loop — leaving later subscriptions uncancelled AND (via the + // exception escaping into SiteStreamGrpcClientFactory's disposal loop) later clients' + // channels undisposed. + var client = SiteStreamGrpcClient.CreateForTesting(); + + var disposedEarly = new CancellationTokenSource(); + using var live = new CancellationTokenSource(); + client.AddSubscriptionForTesting("corr-disposed", disposedEarly); + client.AddSubscriptionForTesting("corr-live", live); + disposedEarly.Dispose(); + + // Deterministic regardless of dictionary iteration order: the disposed entry is always + // visited, so the unguarded loop always threw out of Dispose(). + Assert.Null(Record.Exception(client.Dispose)); + + Assert.True(live.IsCancellationRequested); + } + [Fact] public void RemoveSubscription_OnlyRemovesOwnCts_NotAReplacement() {