fix(communication): evict idle SiteStreamGrpcClientFactory (site, endpoint) clients (ScadaBridge#67)

Each cached client records when it was last handed out; a periodic sweep (half of the new
CommunicationOptions.GrpcClientIdleTtl, default 30 min, 0 disables) disposes and removes entries
unused for the TTL that have no live subscription (SiteStreamGrpcClient.ActiveSubscriptionCount).
Lookup and the sweep's check-and-remove share one lock so a handed-out client is never disposed
under its caller; an evicted client is re-created on next GetOrCreate.
This commit is contained in:
Joseph Doherty
2026-11-09 12:07:28 -05:00
parent e1734675e9
commit e2b15bd520
6 changed files with 300 additions and 14 deletions
+1 -1
View File
@@ -85,7 +85,7 @@ Delivered 2026-07-10 (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.m
- **Alarms-only seed (WP2.3)**: the seed/reconcile fan-out sets `DebugSnapshotRequest.AlarmsOnly` (wire: `DebugSnapshotRequestDto.alarms_only`, field 3, additive), so the site builds and ships only the alarm half of the snapshot. The fan-out discarded every attribute row anyway, and an instance's attribute surface dwarfs its alarm set. A pre-WP2.3 site ignores the flag and returns the full snapshot, which reads identically. - **Alarms-only seed (WP2.3)**: the seed/reconcile fan-out sets `DebugSnapshotRequest.AlarmsOnly` (wire: `DebugSnapshotRequestDto.alarms_only`, field 3, additive), so the site builds and ships only the alarm half of the snapshot. The fan-out discarded every attribute row anyway, and an instance's attribute surface dwarfs its alarm set. A pre-WP2.3 site ignores the flag and returns the full snapshot, which reads identically.
- **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it) or, for a peer that defers its headers, when the FIRST EVENT arrives — whichever comes first, and at most once. A header **timeout** is deliberately not a connect signal: an unreachable or wedged site produces exactly that shape, and reporting it as connected cleared `_streamDown`, consumed the pending re-seed and fanned a whole-site snapshot out at a site that never answered. Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a CONNECT- OR FAILOVER-DRIVEN seed already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A **tick's own fan-out never arms that skip** — when it did, steady state alternated fan-out/skip and the effective reconcile rate was one per TWO intervals (120 s by default), halving both the not-reporting refresh and the alarm reconcile backstop for no benefit: with no reconnect in play there is nothing duplicated to suppress. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch. - **Failover & drift**: a re-seed runs **once per successful (re)connect**, not once per reconnect ATTEMPT — the connect signal is `SiteStreamGrpcClient.SubscribeSiteAsync`'s `onConnected` callback, raised when the site's response headers arrive (the site flushes them as soon as its relay is attached, so no event can be missed after it) or, for a peer that defers its headers, when the FIRST EVENT arrives — whichever comes first, and at most once. A header **timeout** is deliberately not a connect signal: an unreachable or wedged site produces exactly that shape, and reporting it as connected cleared `_streamDown`, consumed the pending re-seed and fanned a whole-site snapshot out at a site that never answered. Fanning a whole-site snapshot out per retry meant N snapshots against a site that was, by definition of the retry, unreachable. A periodic **reconcile snapshot** (default 60s, **jittered** by `LiveAlarmCacheReconcileJitterFraction` so aggregators started together do not stampede one boundary) remains the drift backstop, but it is **skipped when a CONNECT- OR FAILOVER-DRIVEN seed already ran in that window** and **publishes only when the snapshot actually changed the cache** (a diff, not an unconditional viewer fan-out). Staleness stays bounded at two intervals: the skip consumes its flag, so the next tick always fans out. A **tick's own fan-out never arms that skip** — when it did, steady state alternated fan-out/skip and the effective reconcile rate was one per TWO intervals (120 s by default), halving both the not-reporting refresh and the alarm reconcile backstop for no benefit: with no reconnect in play there is nothing duplicated to suppress. A fan-out that fails as a whole now retries on its own **backoff** timer (`reconnectDelay` doubling, capped at 8× the reconcile interval) instead of waiting a full interval. `[PERM]` (`docs/plans/2026-05-29-native-alarms-design.md`): the cache is **purely in-memory** — no EF entity/table/migration, no persisted central alarm store — so a new active node simply re-seeds from scratch.
- **Stream terminations are a triad, and every ending hits exactly one leg.** `SiteStreamGrpcClient.ConsumeStreamAsync` classifies a stream's end as a fault (`onError`), a graceful server-side end (`onCompleted`), or our own teardown (neither). Only a cancellation **we asked for** takes the silent leg — the `RpcException(Cancelled)` filter is guarded by `cts.IsCancellationRequested`. A **foreign** `Cancelled` (the peer cancelled, or the channel was disposed underneath us) falls through to `onError`; unguarded, it fired none of the three, so the aggregator kept `_streamDown = false` — `IsLive` stuck true and the reconcile tick's reopen, which only runs on a stream known to be down, never fired. - **Stream terminations are a triad, and every ending hits exactly one leg.** `SiteStreamGrpcClient.ConsumeStreamAsync` classifies a stream's end as a fault (`onError`), a graceful server-side end (`onCompleted`), or our own teardown (neither). Only a cancellation **we asked for** takes the silent leg — the `RpcException(Cancelled)` filter is guarded by `cts.IsCancellationRequested`. A **foreign** `Cancelled` (the peer cancelled, or the channel was disposed underneath us) falls through to `onError`; unguarded, it fired none of the three, so the aggregator kept `_streamDown = false` — `IsLive` stuck true and the reconcile tick's reopen, which only runs on a stream known to be down, never fired.
- **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200), `LiveAlarmCachePublishCoalesce` (default 250ms; `0` = publish per delta — legacy — batches an alarm storm into one snapshot copy + one viewer fan-out per window; arch review 02 round 2, N6), `LiveAlarmCacheReconcileJitterFraction` (default 0.2 = up to +20% per tick; `0` disables). Stream channel sizing lives alongside the other gRPC limits: `GrpcInstanceStreamChannelCapacity` (default 1000) and `GrpcSiteAlarmStreamChannelCapacity` (default 20 000). - **Options** (`Communication` section, `CommunicationOptions`; eagerly validated by `CommunicationOptionsValidator` / `ValidateOnStart`): `LiveAlarmCacheLinger` (default 30s), `LiveAlarmCacheReconcileInterval` (default 60s), `LiveAlarmCacheSeedConcurrency` (default 8), `LiveAlarmCacheMaxSubscribersPerSite` (default 200), `LiveAlarmCachePublishCoalesce` (default 250ms; `0` = publish per delta — legacy — batches an alarm storm into one snapshot copy + one viewer fan-out per window; arch review 02 round 2, N6), `LiveAlarmCacheReconcileJitterFraction` (default 0.2 = up to +20% per tick; `0` disables). `GrpcClientIdleTtl` (default 30 min; `0` disables) bounds the central `SiteStreamGrpcClientFactory` cache: a `(site, endpoint)` client unused for the TTL with no live subscription is disposed with its channel by a periodic sweep and re-created on demand (ScadaBridge#67). Stream channel sizing lives alongside the other gRPC limits: `GrpcInstanceStreamChannelCapacity` (default 1000) and `GrpcSiteAlarmStreamChannelCapacity` (default 20 000).
- **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link), and counter `scadabridge.site.alarm_cache.buffer_dropped` (deltas evicted from the aggregator's **bounded** 20 000-entry pre-seed buffer, drop-oldest — non-zero means a fan-out ran long enough for the delta storm behind it to exceed the cap; the fan-out's snapshot is authoritative for the evicted rows). - **Telemetry** (`ScadaBridgeTelemetry` meter): observable gauge `scadabridge.site.alarm_cache.aggregators.active` (running per-site aggregators) and counter `scadabridge.site.alarm_cache.reconnects` (site-wide stream reconnects — a NodeA↔NodeB flip or reconcile-driven reopen; a sustained climb signals a flapping site link), and counter `scadabridge.site.alarm_cache.buffer_dropped` (deltas evicted from the aggregator's **bounded** 20 000-entry pre-seed buffer, drop-oldest — non-zero means a fan-out ran long enough for the delta storm behind it to exceed the cap; the fan-out's snapshot is authoritative for the evicted rows).
- **Accepted limitations (arch review 02 round 2, N8):** - **Accepted limitations (arch review 02 round 2, N8):**
- **Standby-node aggregators**: the live cache is per-node and `SetActorSystem` is wired on every central node, so browsing the standby node directly (diagnostic ports, e.g. 9002) starts a second, fully-functional read-only aggregator + `SubscribeSite` stream there. Accepted — not gated: the aggregator is read-only, bounded (one stream/site, viewer-capped), and torn down by the viewer linger; gating `SetActorSystem` behind the active check would break the diagnostic-browse path and buy nothing. The `[PERM]` "lives only on the active central node" claim is corrected to "per-node; in routine operation only the active node hosts viewers". - **Standby-node aggregators**: the live cache is per-node and `SetActorSystem` is wired on every central node, so browsing the standby node directly (diagnostic ports, e.g. 9002) starts a second, fully-functional read-only aggregator + `SubscribeSite` stream there. Accepted — not gated: the aggregator is read-only, bounded (one stream/site, viewer-capped), and torn down by the viewer linger; gating `SetActorSystem` behind the active check would break the diagnostic-browse path and buy nothing. The `[PERM]` "lives only on the active central node" claim is corrected to "per-node; in routine operation only the active node hosts viewers".
@@ -223,6 +223,16 @@ public class CommunicationOptions
/// </summary> /// </summary>
public TimeSpan LiveAlarmCacheLinger { get; set; } = TimeSpan.FromSeconds(30); public TimeSpan LiveAlarmCacheLinger { get; set; } = TimeSpan.FromSeconds(30);
/// <summary>
/// How long a cached central-to-site stream client (one per <c>(site, gRPC endpoint)</c>) may sit
/// unused - no <c>GetOrCreate</c>/<c>TryGet</c> and no live subscription - before
/// <see cref="Grpc.SiteStreamGrpcClientFactory"/> disposes it and its HTTP/2 channel. Reclaims the
/// channel of an endpoint that was renamed or removed from a site's address pair; a live client
/// is simply re-created on next use (ScadaBridge#67). <see cref="TimeSpan.Zero"/> disables
/// idle eviction (clients then live until site removal or shutdown). Default 30 minutes.
/// </summary>
public TimeSpan GrpcClientIdleTtl { get; set; } = TimeSpan.FromMinutes(30);
/// <summary> /// <summary>
/// Cadence of the per-site aggregator's periodic reconcile snapshot fan-out. The /// Cadence of the per-site aggregator's periodic reconcile snapshot fan-out. The
/// backstop that corrects instance-set drift (enable/disable/deploy/delete) and any /// backstop that corrects instance-set drift (enable/disable/deploy/delete) and any
@@ -113,6 +113,10 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase<Communi
+ "entries (a site node must list at least one central gRPC endpoint; central nodes " + "entries (a site node must list at least one central gRPC endpoint; central nodes "
+ "leave it empty)."); + "leave it empty).");
// Idle TTL of the cached site-stream gRPC clients; zero disables eviction, negative is invalid.
builder.RequireThat(options.GrpcClientIdleTtl >= TimeSpan.Zero,
$"ScadaBridge:Communication:GrpcClientIdleTtl must be zero (disabled) or a positive duration (was {options.GrpcClientIdleTtl}).");
// ── Aggregated live alarm cache ────────────────────────────────────────── // ── Aggregated live alarm cache ──────────────────────────────────────────
// Linger drives a Timer dueTime; TimeSpan.Zero is valid (stop the aggregator // Linger drives a Timer dueTime; TimeSpan.Zero is valid (stop the aggregator
// immediately when the last viewer leaves), only a negative value is invalid. // immediately when the last viewer leaves), only a negative value is invalid.
@@ -131,6 +131,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable
ConnectedHeaderTimeout = options.ConnectedHeaderTimeout; ConnectedHeaderTimeout = options.ConnectedHeaderTimeout;
} }
/// <summary>
/// Number of live subscriptions currently registered on this client. The owning
/// <see cref="SiteStreamGrpcClientFactory"/> never evicts a client while this is non-zero.
/// </summary>
public virtual int ActiveSubscriptionCount => _subscriptions.Count;
/// <summary> /// <summary>
/// Registers a CancellationTokenSource for a correlation ID. Test-only. /// Registers a CancellationTokenSource for a correlation ID. Test-only.
/// </summary> /// </summary>
@@ -16,14 +16,29 @@ namespace ZB.MOM.WW.ScadaBridge.Communication.Grpc;
/// is the only shared-disposal path — it disposes every endpoint for that site. /// is the only shared-disposal path — it disposes every endpoint for that site.
/// </para> /// </para>
/// <para> /// <para>
/// Trade-off: an edited gRPC address leaves the old endpoint's now-idle channel /// Idle eviction (ScadaBridge#67): an edited or removed gRPC address would otherwise leave the
/// cached until site removal or process shutdown — bounded at a handful of entries /// old endpoint's channel resident until site removal or shutdown. Each entry records when it
/// per site (2 nodes + rare edits), each an idle HTTP/2 channel. /// was last handed out; a periodic sweep (<see cref="CommunicationOptions.GrpcClientIdleTtl"/>)
/// disposes entries that have been unused for the TTL AND have no live subscription. Lookups
/// and the sweep's check-and-remove share one lock, so a client can never be disposed after
/// being handed out; callers re-request via <see cref="GetOrCreate"/> on every (re)connect, and
/// an evicted client is simply re-created.
/// </para> /// </para>
/// </summary> /// </summary>
public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
{ {
private readonly ConcurrentDictionary<(string Site, string Endpoint), SiteStreamGrpcClient> _clients = new(); private sealed class Entry(SiteStreamGrpcClient client, long lastUsedTimestamp)
{
public SiteStreamGrpcClient Client { get; } = client;
public long LastUsedTimestamp { get; set; } = lastUsedTimestamp;
}
private readonly ConcurrentDictionary<(string Site, string Endpoint), Entry> _clients = new();
// Serialises hand-out (lookup + touch) against the idle sweep's check-and-remove.
private readonly object _gate = new();
private readonly TimeProvider _timeProvider;
private readonly TimeSpan _idleTtl;
private readonly ITimer? _sweepTimer;
private readonly ILoggerFactory _loggerFactory; private readonly ILoggerFactory _loggerFactory;
private readonly CommunicationOptions _options; private readonly CommunicationOptions _options;
private readonly ISitePskProvider? _pskProvider; private readonly ISitePskProvider? _pskProvider;
@@ -64,11 +79,20 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
public SiteStreamGrpcClientFactory( public SiteStreamGrpcClientFactory(
ILoggerFactory loggerFactory, ILoggerFactory loggerFactory,
IOptions<CommunicationOptions> options, IOptions<CommunicationOptions> options,
ISitePskProvider? pskProvider) ISitePskProvider? pskProvider,
TimeProvider? timeProvider = null)
{ {
_loggerFactory = loggerFactory; _loggerFactory = loggerFactory;
_options = options.Value; _options = options.Value;
_pskProvider = pskProvider; _pskProvider = pskProvider;
_timeProvider = timeProvider ?? TimeProvider.System;
_idleTtl = _options.GrpcClientIdleTtl;
if (_idleTtl > TimeSpan.Zero)
{
// Sweep at half the TTL, so an entry is reclaimed within 1.5 x TTL of its last use.
var period = TimeSpan.FromTicks(Math.Max(_idleTtl.Ticks / 2, TimeSpan.FromSeconds(1).Ticks));
_sweepTimer = _timeProvider.CreateTimer(_ => EvictIdle(), null, period, period);
}
} }
/// <summary> /// <summary>
@@ -80,8 +104,23 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
/// <param name="siteIdentifier">Unique site identifier (first half of the cache key).</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> /// <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> /// <returns>The cached or newly-created client bound to <paramref name="grpcEndpoint"/>.</returns>
public virtual SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint) => public virtual SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
_clients.GetOrAdd((siteIdentifier, grpcEndpoint), key => CreateClient(key.Site, key.Endpoint)); {
lock (_gate)
{
var key = (siteIdentifier, grpcEndpoint);
if (!_clients.TryGetValue(key, out var entry))
{
entry = new Entry(CreateClient(siteIdentifier, grpcEndpoint), _timeProvider.GetTimestamp());
_clients[key] = entry;
}
else
{
entry.LastUsedTimestamp = _timeProvider.GetTimestamp();
}
return entry.Client;
}
}
/// <summary> /// <summary>
/// Returns the cached client for <c>(site, endpoint)</c>, or <c>null</c> — never creates. /// Returns the cached client for <c>(site, endpoint)</c>, or <c>null</c> — never creates.
@@ -91,8 +130,66 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
/// <param name="siteIdentifier">Unique site identifier (first half of the cache key).</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).</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> /// <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) => public virtual SiteStreamGrpcClient? TryGet(string siteIdentifier, string grpcEndpoint)
_clients.TryGetValue((siteIdentifier, grpcEndpoint), out var client) ? client : null; {
lock (_gate)
{
if (!_clients.TryGetValue((siteIdentifier, grpcEndpoint), out var entry))
return null;
entry.LastUsedTimestamp = _timeProvider.GetTimestamp();
return entry.Client;
}
}
/// <summary>
/// Disposes and removes every cached client that has been unused for at least
/// <see cref="CommunicationOptions.GrpcClientIdleTtl"/> and has no live subscription.
/// Run periodically by the factory's sweep timer; internal so tests can drive it directly.
/// Thread-safe: the check-and-remove runs under the same lock as <see cref="GetOrCreate"/>,
/// and disposal happens outside it.
/// </summary>
/// <returns>The number of clients evicted.</returns>
internal int EvictIdle()
{
if (_idleTtl <= TimeSpan.Zero)
return 0;
List<((string Site, string Endpoint) Key, SiteStreamGrpcClient Client)>? evicted = null;
lock (_gate)
{
var now = _timeProvider.GetTimestamp();
foreach (var kv in _clients)
{
if (_timeProvider.GetElapsedTime(kv.Value.LastUsedTimestamp, now) < _idleTtl)
continue;
if (kv.Value.Client.ActiveSubscriptionCount > 0)
continue; // a live stream is using the channel - it is not idle
if (_clients.TryRemove(kv.Key, out var removed))
(evicted ??= []).Add((kv.Key, removed.Client));
}
}
if (evicted is null)
return 0;
foreach (var (key, client) in evicted)
{
_loggerFactory.CreateLogger<SiteStreamGrpcClientFactory>().LogInformation(
"Evicting idle site stream gRPC client for site {SiteId} endpoint {Endpoint}",
key.Site, key.Endpoint);
// Synchronous dispose: the client releases everything synchronously, and the
// timer callback must not block on or leak an async path. Fault-isolated.
try { client.Dispose(); }
catch (Exception ex)
{
_loggerFactory.CreateLogger<SiteStreamGrpcClientFactory>().LogWarning(
ex, "Disposing the idle site stream gRPC client for site {SiteId} endpoint {Endpoint} failed",
key.Site, key.Endpoint);
}
}
return evicted.Count;
}
/// <summary> /// <summary>
/// Creates a single <see cref="SiteStreamGrpcClient"/>. Overridable so tests /// Creates a single <see cref="SiteStreamGrpcClient"/>. Overridable so tests
@@ -119,8 +216,13 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
{ {
foreach (var key in _clients.Keys.Where(k => k.Site == siteIdentifier).ToList()) foreach (var key in _clients.Keys.Where(k => k.Site == siteIdentifier).ToList())
{ {
if (_clients.TryRemove(key, out var client)) Entry? entry;
await DisposeClientAsync(client, key); lock (_gate)
{
_clients.TryRemove(key, out entry);
}
if (entry is not null)
await DisposeClientAsync(entry.Client, key);
} }
// Drop the cached preshared key too, so a site removed and re-added under the same // Drop the cached preshared key too, so a site removed and re-added under the same
@@ -136,9 +238,10 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
/// <returns>A value task that completes when all clients have been disposed.</returns> /// <returns>A value task that completes when all clients have been disposed.</returns>
public async ValueTask DisposeAsync() public async ValueTask DisposeAsync()
{ {
_sweepTimer?.Dispose();
foreach (var entry in _clients) foreach (var entry in _clients)
{ {
await DisposeClientAsync(entry.Value, entry.Key); await DisposeClientAsync(entry.Value.Client, entry.Key);
} }
_clients.Clear(); _clients.Clear();
} }
@@ -174,11 +277,12 @@ public class SiteStreamGrpcClientFactory : IAsyncDisposable, IDisposable
/// </summary> /// </summary>
public void Dispose() public void Dispose()
{ {
_sweepTimer?.Dispose();
foreach (var entry in _clients) foreach (var entry in _clients)
{ {
try try
{ {
entry.Value.Dispose(); entry.Value.Client.Dispose();
} }
catch (Exception ex) catch (Exception ex)
{ {
@@ -0,0 +1,162 @@
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc;
/// <summary>
/// ScadaBridge#67: <see cref="SiteStreamGrpcClientFactory"/> evicts (and disposes) clients whose
/// <c>(site, endpoint)</c> has gone unused for <see cref="CommunicationOptions.GrpcClientIdleTtl"/>,
/// so a renamed or removed address does not leave its channel resident for the life of the process.
/// </summary>
public class SiteStreamGrpcClientFactoryIdleEvictionTests
{
private static readonly TimeSpan Ttl = TimeSpan.FromMinutes(10);
/// <summary>Test clock: only advances when told to.</summary>
private sealed class ManualTimeProvider : TimeProvider
{
private long _ticks = TimeSpan.FromHours(1).Ticks;
public override long TimestampFrequency => TimeSpan.TicksPerSecond;
public override long GetTimestamp() => Interlocked.Read(ref _ticks);
public void Advance(TimeSpan by) => Interlocked.Add(ref _ticks, by.Ticks);
}
private sealed class TrackingClient(string endpoint) : SiteStreamGrpcClient(endpoint)
{
public int DisposeCount;
public override void Dispose() { Interlocked.Increment(ref DisposeCount); base.Dispose(); }
public override ValueTask DisposeAsync() { Interlocked.Increment(ref DisposeCount); return base.DisposeAsync(); }
}
private sealed class Factory(ManualTimeProvider clock, TimeSpan ttl) : SiteStreamGrpcClientFactory(
NullLoggerFactory.Instance,
Options.Create(new CommunicationOptions { GrpcClientIdleTtl = ttl }),
pskProvider: null,
clock)
{
public int Created;
protected override SiteStreamGrpcClient CreateClient(string siteIdentifier, string grpcEndpoint)
{
Interlocked.Increment(ref Created);
return new TrackingClient(grpcEndpoint);
}
}
/// <summary>An unused client is disposed and removed once the TTL elapses; a fresh one is created on next use.</summary>
[Fact]
public void IdleClient_IsEvictedAndDisposed_ThenRecreatedOnDemand()
{
var clock = new ManualTimeProvider();
using var factory = new Factory(clock, Ttl);
var old = (TrackingClient)factory.GetOrCreate("site-a", "http://old:5100");
clock.Advance(Ttl + TimeSpan.FromSeconds(1));
Assert.Equal(1, factory.EvictIdle());
Assert.Equal(1, old.DisposeCount);
Assert.Null(factory.TryGet("site-a", "http://old:5100"));
var fresh = factory.GetOrCreate("site-a", "http://old:5100");
Assert.NotSame(old, fresh);
}
/// <summary>A client used within the TTL survives the sweep - and using it resets the idle clock.</summary>
[Fact]
public void RecentlyUsedClient_IsNotEvicted()
{
var clock = new ManualTimeProvider();
using var factory = new Factory(clock, Ttl);
var kept = (TrackingClient)factory.GetOrCreate("site-a", "http://a:5100");
var idle = (TrackingClient)factory.GetOrCreate("site-a", "http://b:5100");
clock.Advance(TimeSpan.FromMinutes(8));
Assert.Same(kept, factory.GetOrCreate("site-a", "http://a:5100")); // touch
clock.Advance(TimeSpan.FromMinutes(8));
Assert.Equal(1, factory.EvictIdle());
Assert.Equal(0, kept.DisposeCount);
Assert.Equal(1, idle.DisposeCount);
Assert.Same(kept, factory.TryGet("site-a", "http://a:5100"));
}
/// <summary>A client with a live subscription is never evicted, however long since it was handed out.</summary>
[Fact]
public void ClientWithActiveSubscription_IsNotEvicted()
{
var clock = new ManualTimeProvider();
using var factory = new Factory(clock, Ttl);
var busy = (TrackingClient)factory.GetOrCreate("site-a", "http://a:5100");
using var cts = new CancellationTokenSource();
busy.RegisterSubscription("corr-1", cts);
clock.Advance(Ttl * 3);
Assert.Equal(0, factory.EvictIdle());
Assert.Equal(0, busy.DisposeCount);
// Once the subscription is gone it becomes evictable again.
busy.Unsubscribe("corr-1");
Assert.Equal(1, factory.EvictIdle());
Assert.Equal(1, busy.DisposeCount);
}
/// <summary>A zero TTL disables idle eviction entirely.</summary>
[Fact]
public void ZeroTtl_DisablesEviction()
{
var clock = new ManualTimeProvider();
using var factory = new Factory(clock, TimeSpan.Zero);
var client = (TrackingClient)factory.GetOrCreate("site-a", "http://a:5100");
clock.Advance(TimeSpan.FromDays(30));
Assert.Equal(0, factory.EvictIdle());
Assert.Equal(0, client.DisposeCount);
}
/// <summary>Hand-out racing the sweep never throws and never returns null; evicted entries are re-created.</summary>
[Fact]
public async Task ConcurrentGetOrCreateAndEviction_IsThreadSafe()
{
var clock = new ManualTimeProvider();
using var factory = new Factory(clock, TimeSpan.FromSeconds(1));
var stop = false;
var evictor = Task.Run(() =>
{
while (!Volatile.Read(ref stop))
{
clock.Advance(TimeSpan.FromSeconds(2));
factory.EvictIdle();
}
});
var getters = Enumerable.Range(0, 4).Select(_ => Task.Run(() =>
{
for (var i = 0; i < 20_000; i++)
{
Assert.NotNull(factory.GetOrCreate("site-a", "http://a:5100"));
factory.TryGet("site-a", "http://a:5100");
}
})).ToArray();
await Task.WhenAll(getters);
Volatile.Write(ref stop, true);
await evictor;
Assert.True(factory.Created >= 1);
}
/// <summary>The validator rejects a negative idle TTL and accepts zero.</summary>
[Fact]
public void Validator_RejectsNegativeIdleTtl()
{
var validator = new CommunicationOptionsValidator();
var bad = validator.Validate(null, new CommunicationOptions { GrpcClientIdleTtl = TimeSpan.FromSeconds(-1) });
var zero = validator.Validate(null, new CommunicationOptions { GrpcClientIdleTtl = TimeSpan.Zero });
Assert.True(bad.Failed);
Assert.Contains("GrpcClientIdleTtl", string.Join(" ", bad.Failures!));
Assert.DoesNotContain("GrpcClientIdleTtl", string.Join(" ", zero.Failures ?? []));
}
}