dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix(communication): key gRPC client cache by (site, endpoint) so per-session failover cannot dispose shared channels

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-08 17:33:03 -04:00
parent 76c10c5de6
commit ca5e79e922
2 changed files with 83 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClientFactory.cs
+41 -47
View File
@@ -5,13 +5,25 @@ using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
/// <summary>
/// Caches one <see cref="SiteStreamGrpcClient"/> per site identifier.
/// Caches one <see cref="SiteStreamGrpcClient"/> per <c>(site, endpoint)</c> 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.
/// <para>
/// 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). <see cref="GetOrCreate"/> no
/// longer disposes on endpoint mismatch; site *removal* (<see cref="RemoveSiteAsync"/>)
/// is the only shared-disposal path — it disposes every endpoint for that site.
/// </para>
/// <para>
/// 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.
/// </para>
/// </summary>
public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
{
private readonly ConcurrentDictionary<string, SiteStreamGrpcClient> _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
}
/// <summary>
/// Returns the cached client for the site, or creates a new one. If a client is
/// already cached but bound to a *different* <paramref name="grpcEndpoint"/> — 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 <c>(site, endpoint)</c>, 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.
/// </summary>
/// <param name="siteIdentifier">Unique site identifier used as the cache key.</param>
/// <param name="grpcEndpoint">gRPC endpoint the returned client must be bound to.</param>
/// <param name="siteIdentifier">Unique site identifier (first half of the cache key).</param>
/// <param name="grpcEndpoint">gRPC endpoint (second half of the cache key) the client is bound to.</param>
/// <returns>The cached or newly-created client bound to <paramref name="grpcEndpoint"/>.</returns>
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;
}
/// <summary>
/// Returns the cached client for <c>(site, endpoint)</c>, or <c>null</c> — 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.
/// </summary>
/// <param name="siteIdentifier">Unique site identifier (first half of the cache key).</param>
/// <param name="grpcEndpoint">gRPC endpoint (second half of the cache key).</param>
/// <returns>The cached client, or <c>null</c> when no client is cached for the pair.</returns>
public virtual SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint) =>
_clients.TryGetValue((siteIdentifier, grpcEndpoint), out var client) ? client : null;
/// <summary>
/// Creates a single <see cref="SiteStreamGrpcClient"/>. Overridable so tests
@@ -91,19 +86,18 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
}
/// <summary>
/// Removes and disposes the client for the given site. Site *address changes* are
/// now handled transparently by <see cref="GetOrCreate"/> (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.
/// </summary>
/// <param name="siteIdentifier">Unique site identifier whose client should be removed.</param>
/// <returns>A task that completes when the cached client has been removed and disposed.</returns>
/// <param name="siteIdentifier">Unique site identifier whose clients should be removed.</param>
/// <returns>A task that completes when all of the site's cached clients have been disposed.</returns>
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();
}
}
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientFactoryTests.cs
+42 -10
View File
@@ -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);
}
}
}