From ca5e79e922b480f6d8814c72da46fb276b874c5f Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 17:33:03 -0400 Subject: [PATCH] fix(communication): key gRPC client cache by (site, endpoint) so per-session failover cannot dispose shared channels --- .../Grpc/SiteStreamGrpcClientFactory.cs | 88 +++++++++---------- .../Grpc/SiteStreamGrpcClientFactoryTests.cs | 52 ++++++++--- 2 files changed, 83 insertions(+), 57 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs index 22b00ace..ec72182e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs @@ -5,13 +5,25 @@ using Microsoft.Extensions.Options; namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc; /// -/// Caches one per site identifier. +/// Caches one per (site, endpoint) pair. /// The DebugStreamBridgeActor uses this factory to obtain (or create) a -/// gRPC client for a given site before opening a streaming subscription. +/// gRPC client for a given site+node before opening a streaming subscription. +/// +/// Both of a site's node channels (NodeA and NodeB) coexist in the cache, so a +/// per-session NodeA↔NodeB failover flip never disposes a channel another debug +/// session is still using (arch review 02, High). no +/// longer disposes on endpoint mismatch; site *removal* () +/// is the only shared-disposal path — it disposes every endpoint for that site. +/// +/// +/// Trade-off: an edited gRPC address leaves the old endpoint's now-idle channel +/// cached until site removal or process shutdown — bounded at a handful of entries +/// per site (2 nodes + rare edits), each an idle HTTP/2 channel. +/// /// public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable { - private readonly ConcurrentDictionary _clients = new(); + private readonly ConcurrentDictionary<(string Site, string Endpoint), SiteStreamGrpcClient> _clients = new(); private readonly ILoggerFactory _loggerFactory; private readonly CommunicationOptions _options; @@ -38,44 +50,27 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable } /// - /// Returns the cached client for the site, or creates a new one. If a client is - /// already cached but bound to a *different* — the - /// NodeA→NodeB failover flip, or a site whose gRPC address was edited — the stale - /// client is disposed and replaced with one bound to the requested endpoint. - /// Keying purely by site identifier and ignoring the - /// endpoint on a cache hit defeated debug-stream node failover and meant a - /// corrected gRPC address never took effect without a central restart. + /// Returns the cached client for (site, endpoint), or creates one. Both node + /// channels for a site coexist under distinct keys, so requesting a different + /// endpoint for the same site adds a second entry rather than disposing the first — + /// per-session failover can never dispose a channel a concurrent session still holds. /// - /// Unique site identifier used as the cache key. - /// gRPC endpoint the returned client must be bound to. + /// Unique site identifier (first half of the cache key). + /// gRPC endpoint (second half of the cache key) the client is bound to. /// The cached or newly-created client bound to . - public virtual SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) - { - // Fast path: a client is cached and already bound to the requested endpoint. - if (_clients.TryGetValue(siteIdentifier, out var existing) && - string.Equals(existing.Endpoint, grpcEndpoint, StringComparison.Ordinal)) - { - return existing; - } + public virtual SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) => + _clients.GetOrAdd((siteIdentifier, grpcEndpoint), _ => CreateClient(grpcEndpoint)); - // Either no client is cached, or the cached one is bound to a different - // endpoint. AddOrUpdate atomically installs a client for the requested - // endpoint; the prior (stale) client, if any, is disposed afterwards. - SiteStreamGrpcClient? stale = null; - var client = _clients.AddOrUpdate( - siteIdentifier, - _ => CreateClient(grpcEndpoint), - (_, current) => - { - if (string.Equals(current.Endpoint, grpcEndpoint, StringComparison.Ordinal)) - return current; - stale = current; - return CreateClient(grpcEndpoint); - }); - - stale?.Dispose(); - return client; - } + /// + /// Returns the cached client for (site, endpoint), or null — never creates. + /// Teardown paths (DebugStreamBridgeActor.CleanupGrpc / failed-stream unsubscribe) + /// use this so cleanup can never open a fresh channel or touch another endpoint's client. + /// + /// Unique site identifier (first half of the cache key). + /// gRPC endpoint (second half of the cache key). + /// The cached client, or null when no client is cached for the pair. + public virtual SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) => + _clients.TryGetValue((siteIdentifier, grpcEndpoint), out var client) ? client : null; /// /// Creates a single . Overridable so tests @@ -91,19 +86,18 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable } /// - /// Removes and disposes the client for the given site. Site *address changes* are - /// now handled transparently by (it disposes and recreates - /// a client whose endpoint no longer matches). This method remains the disposal - /// path for full site *removal* — call it when a site record is deleted so its - /// cached gRPC client does not linger for the life of the process. + /// Removes and disposes every cached client for the given site — all of its node + /// endpoints. This is the only shared-disposal path: call it when a site record is + /// deleted so the site's gRPC channels do not linger for the life of the process. /// - /// Unique site identifier whose client should be removed. - /// A task that completes when the cached client has been removed and disposed. + /// Unique site identifier whose clients should be removed. + /// A task that completes when all of the site's cached clients have been disposed. public async Task RemoveSiteAsync(string siteIdentifier) { - if (_clients.TryRemove(siteIdentifier, out var client)) + foreach (var key in _clients.Keys.Where(k => k.Site == siteIdentifier).ToList()) { - await client.DisposeAsync(); + if (_clients.TryRemove(key, out var client)) + await client.DisposeAsync(); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs index 6ab28000..c84f2a96 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs @@ -81,19 +81,47 @@ public class SiteStreamGrpcClientFactoryTests } [Fact] - public void GetOrCreate_EndpointChanged_DisposesPriorClient() + public void GetOrCreate_DifferentEndpointSameSite_KeepsBothClientsAlive() { - // Communication-013 regression: a later edit to a site's gRPC address must - // invalidate (and dispose) the stale cached client, so the corrected - // endpoint takes effect without a central restart. + // Arch review 02 (High): the cache is keyed by (site, endpoint). Both node + // channels for a site coexist, so a per-session NodeA→NodeB failover flip + // must NOT dispose the other endpoint's channel — another debug session may + // be using it. (Supersedes Communication-013's dispose-on-endpoint-change.) using var factory = new TrackingEndpointFactory(); - var first = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5100"); - var second = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5200"); + var a = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5100"); + var b = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5200"); - Assert.NotSame(first, second); - Assert.True(first.Disposed, "stale client for the old endpoint should be disposed"); - Assert.False(second.Disposed, "fresh client for the new endpoint should still be live"); + Assert.NotSame(a, b); + Assert.False(a.Disposed, "the other endpoint's channel must stay alive"); + Assert.Same(a, factory.GetOrCreate("site-a", "http://localhost:5100")); // still cached + } + + [Fact] + public void TryGet_ReturnsCachedClientOrNull_WithoutCreating() + { + using var factory = new TrackingEndpointFactory(); + + Assert.Null(factory.TryGet("site-a", "http://localhost:5100")); + var a = factory.GetOrCreate("site-a", "http://localhost:5100"); + Assert.Same(a, factory.TryGet("site-a", "http://localhost:5100")); + Assert.Null(factory.TryGet("site-a", "http://localhost:5200")); // other endpoint not cached + Assert.Equal(1, factory.CreatedCount); // TryGet never creates + } + + [Fact] + public async Task RemoveSiteAsync_DisposesAllEndpointsForTheSite_OnlyThatSite() + { + using var factory = new TrackingEndpointFactory(); + var a = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5100"); + var b = (TrackingEndpointClient)factory.GetOrCreate("site-a", "http://localhost:5200"); + var other = (TrackingEndpointClient)factory.GetOrCreate("site-b", "http://localhost:5100"); + + await factory.RemoveSiteAsync("site-a"); + + Assert.True(a.Disposed); + Assert.True(b.Disposed); + Assert.False(other.Disposed); } [Fact] @@ -126,7 +154,11 @@ public class SiteStreamGrpcClientFactoryTests private sealed class TrackingEndpointFactory : SiteStreamGrpcClientFactory { public TrackingEndpointFactory() : base(NullLoggerFactory.Instance) { } + public int CreatedCount { get; private set; } protected override SiteStreamGrpcClient CreateClient(string grpcEndpoint) - => new TrackingEndpointClient(grpcEndpoint); + { + CreatedCount++; + return new TrackingEndpointClient(grpcEndpoint); + } } }