fix(auditlog): prune per-site reconciliation state and pull-channel caches for sites absent from the enumeration; bound the abandoned-id set (review 2026-10-21 06/MEM-2)

2026-10-22 06:40:54 -04:00
parent 9e7e19c3c1
commit 38167bca88
7 changed files with 299 additions and 12 deletions
@@ -0,0 +1,66 @@
+using System.Collections.Concurrent;
+using Grpc.Net.Client;
+
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
+
+/// <summary>
+/// Optional capability of a pull invoker that caches one <see cref="GrpcChannel"/> per
+/// <c>(site, endpoint)</c>: drop (and dispose) every cached channel whose key is not in the
+/// current site enumeration. The pull clients call it with each enumeration they already make
+/// to resolve an endpoint, so a deleted site or an edited address no longer leaves a dead
+/// channel (and its HTTP/2 connection pool) alive until process exit (review 2026-10-21 06/MEM-2).
+/// </summary>
+public interface IGrpcChannelCachePruner
+{
+    /// <summary>Retains only channels whose <c>(site, endpoint)</c> appears in <paramref name="liveSites"/>.</summary>
+    /// <param name="liveSites">The current site enumeration.</param>
+    void RetainOnly(IReadOnlyList<SiteEntry> liveSites);
+}
+
+/// <summary>Shared implementation of <see cref="IGrpcChannelCachePruner.RetainOnly"/>.</summary>
+internal static class GrpcChannelCachePruning
+{
+    /// <summary>
+    /// Removes and disposes every channel in <paramref name="channels"/> whose key is neither a
+    /// live site's primary nor its fallback endpoint.
+    /// </summary>
+    /// <param name="channels">The invoker's channel cache.</param>
+    /// <param name="liveSites">The current site enumeration.</param>
+    /// <returns>The number of channels pruned.</returns>
+    public static int RetainOnly(
+        ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> channels,
+        IReadOnlyList<SiteEntry> liveSites)
+    {
+        if (channels.IsEmpty)
+        {
+            return 0;
+        }
+
+        var live = new HashSet<(string, string)>();
+        foreach (var site in liveSites)
+        {
+            if (!string.IsNullOrWhiteSpace(site.GrpcEndpoint))
+            {
+                live.Add((site.SiteId, site.GrpcEndpoint));
+            }
+            if (!string.IsNullOrWhiteSpace(site.FallbackGrpcEndpoint))
+            {
+                live.Add((site.SiteId, site.FallbackGrpcEndpoint!));
+            }
+        }
+
+        var pruned = 0;
+        foreach (var key in channels.Keys)
+        {
+            if (!live.Contains(key) && channels.TryRemove(key, out var stale))
+            {
+                // A call racing this on the stale channel fails as a tolerable transport
+                // fault and the next tick re-resolves — acceptable for an address that is
+                // no longer configured.
+                stale.Dispose();
+                pruned++;
+            }
+        }
+        return pruned;
+    }
+}
@@ -194,6 +194,8 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient
        string siteId, CancellationToken ct)
    {
        var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false);
+        // Review 2026-10-21 06/MEM-2: drop channels for sites/addresses no longer enumerated.
+        (_invoker as IGrpcChannelCachePruner)?.RetainOnly(sites);
        foreach (var site in sites)
        {
            if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) &&
@@ -256,14 +258,13 @@ public sealed class GrpcPullAuditEventsClient : IPullAuditEventsClient
 /// endpoint string, so a changed site address (NodeA→NodeB failover flip / an
 /// edited gRPC address) is reached as soon as the resolver hands the new
 /// endpoint to <see cref="InvokeAsync"/> — it creates a fresh channel for the
-/// new address. Unlike <c>SiteStreamGrpcClientFactory</c> (keyed by siteId,
-/// which actively evicts a re-keyed client), the channel for the previous
-/// address is NOT actively evicted here; it lingers idle until
-/// <see cref="Dispose"/>. Idle channels hold no streams, so this is a minor
-/// cache footprint cost, not a correctness or liveness gap.
+/// new address. The channel for a previous address (or a deleted site) is
+/// evicted and disposed by <see cref="RetainOnly"/>, which the client calls with
+/// every site enumeration it already makes (review 2026-10-21 06/MEM-2) — before
+/// that it lingered, with its HTTP/2 connection pool, until process exit.
 /// </summary>
 public sealed class GrpcPullAuditEventsInvoker
-    : GrpcPullAuditEventsClient.IPullAuditEventsInvoker, IDisposable
+    : GrpcPullAuditEventsClient.IPullAuditEventsInvoker, IDisposable, IGrpcChannelCachePruner
 {
    private readonly ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> _channels = new();
    private readonly CommunicationOptions _options;
@@ -344,6 +345,10 @@ public sealed class GrpcPullAuditEventsInvoker
            },
        }.WithMessageLimits().WithSiteCredentials(_pskProvider, siteId));

+    /// <inheritdoc />
+    public void RetainOnly(IReadOnlyList<SiteEntry> liveSites) =>
+        GrpcChannelCachePruning.RetainOnly(_channels, liveSites);
+
    /// <summary>Disposes all cached channels.</summary>
    public void Dispose()
    {
@@ -214,6 +214,8 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient
        string siteId, CancellationToken ct)
    {
        var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false);
+        // Review 2026-10-21 06/MEM-2: drop channels for sites/addresses no longer enumerated.
+        (_invoker as IGrpcChannelCachePruner)?.RetainOnly(sites);
        foreach (var site in sites)
        {
            if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) &&
@@ -275,13 +277,13 @@ public sealed class GrpcPullSiteCallsClient : IPullSiteCallsClient
 /// issues the unary <c>PullSiteCallsAsync</c> call. The cache is keyed by
 /// endpoint string, so a changed site address (NodeA→NodeB failover flip / an
 /// edited gRPC address) is reached as soon as the resolver hands the new endpoint
-/// to <see cref="InvokeAsync"/>. The channel for a previous address lingers idle
-/// until <see cref="Dispose"/> (idle channels hold no streams — a minor cache
-/// footprint cost, not a correctness or liveness gap). Sibling of
+/// to <see cref="InvokeAsync"/>. The channel for a previous address (or a deleted
+/// site) is evicted and disposed by <see cref="RetainOnly"/>, which the client calls
+/// with every site enumeration (review 2026-10-21 06/MEM-2). Sibling of
 /// <see cref="GrpcPullAuditEventsInvoker"/>.
 /// </summary>
 public sealed class GrpcPullSiteCallsInvoker
-    : GrpcPullSiteCallsClient.IPullSiteCallsInvoker, IDisposable
+    : GrpcPullSiteCallsClient.IPullSiteCallsInvoker, IDisposable, IGrpcChannelCachePruner
 {
    private readonly ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> _channels = new();
    private readonly CommunicationOptions _options;
@@ -358,6 +360,10 @@ public sealed class GrpcPullSiteCallsInvoker
            },
        }.WithMessageLimits().WithSiteCredentials(_pskProvider, siteId));

+    /// <inheritdoc />
+    public void RetainOnly(IReadOnlyList<SiteEntry> liveSites) =>
+        GrpcChannelCachePruning.RetainOnly(_channels, liveSites);
+
    /// <summary>Disposes all cached channels.</summary>
    public void Dispose()
    {
@@ -135,6 +135,8 @@ public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient
    private async Task<(string? Primary, string? Fallback)> ResolveEndpointsAsync(string siteId, CancellationToken ct)
    {
        var sites = await _sites.EnumerateAsync(ct).ConfigureAwait(false);
+        // Review 2026-10-21 06/MEM-2: drop channels for sites/addresses no longer enumerated.
+        (_invoker as IGrpcChannelCachePruner)?.RetainOnly(sites);
        foreach (var site in sites)
        {
            if (string.Equals(site.SiteId, siteId, StringComparison.Ordinal) && !string.IsNullOrWhiteSpace(site.GrpcEndpoint))
@@ -170,7 +172,7 @@ public sealed class GrpcPullTraceEventsClient : IPullTraceEventsClient
 /// <see cref="GrpcChannel"/> per (site, endpoint), keepalive from <see cref="CommunicationOptions"/>,
 /// per-site PSK call credentials — the same shape as <see cref="GrpcPullAuditEventsInvoker"/>.
 /// </summary>
-public sealed class GrpcPullTraceEventsInvoker : GrpcPullTraceEventsClient.IPullTraceEventsInvoker, IDisposable
+public sealed class GrpcPullTraceEventsInvoker : GrpcPullTraceEventsClient.IPullTraceEventsInvoker, IDisposable, IGrpcChannelCachePruner
 {
    private readonly ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> _channels = new();
    private readonly CommunicationOptions _options;
@@ -222,6 +224,10 @@ public sealed class GrpcPullTraceEventsInvoker : GrpcPullTraceEventsClient.IPull
            },
        }.WithMessageLimits().WithSiteCredentials(_pskProvider, siteId));

+    /// <inheritdoc />
+    public void RetainOnly(IReadOnlyList<SiteEntry> liveSites) =>
+        GrpcChannelCachePruning.RetainOnly(_channels, liveSites);
+
    /// <summary>Disposes every cached gRPC channel.</summary>
    public void Dispose()
    {
@@ -181,7 +181,51 @@ public class SiteAuditReconciliationActor : ReceiveActor

    private ICancelable? _timer;

-    private void RememberAbandoned(Guid eventId) => _abandonedEventIds.Add(eventId);
+    /// <summary>
+    /// Upper bound on <see cref="_abandonedEventIds"/>. The set is only a short-circuit (see its
+    /// doc comment — determinism of <see cref="DeriveAbandonmentEventId"/> is what makes
+    /// re-abandonment safe), so clearing it when it grows past this costs at most one re-run of
+    /// the retry cycle per forgotten row, never a duplicate or a lost row (review 2026-10-21
+    /// 06/MEM-2: it used to grow one entry per abandoned row for the singleton's lifetime).
+    /// </summary>
+    internal const int MaxRememberedAbandonedEventIds = 10_000;
+
+    internal void RememberAbandoned(Guid eventId)
+    {
+        if (_abandonedEventIds.Count >= MaxRememberedAbandonedEventIds)
+        {
+            _abandonedEventIds.Clear();
+        }
+        _abandonedEventIds.Add(eventId);
+    }
+
+    /// <summary>Test seam: current size of the abandoned-id short-circuit set.</summary>
+    internal int AbandonedEventIdCountForTests => _abandonedEventIds.Count;
+
+    /// <summary>
+    /// Drops per-site state (cursor, non-draining counter, stalled latch) for sites no longer in
+    /// the enumeration — a deleted or renamed site otherwise kept its entries for the singleton's
+    /// lifetime (review 2026-10-21 06/MEM-2, mirroring <c>CentralHealthAggregator.PruneUnknownSites</c>).
+    /// A pruned site that was latched stalled publishes <c>Stalled=false</c> so the health
+    /// surface does not keep a badge for a site that no longer exists.
+    /// </summary>
+    private void PruneUnknownSites(IReadOnlyList<SiteEntry> sites, Akka.Event.EventStream eventStream)
+    {
+        var live = new HashSet<string>(sites.Select(s => s.SiteId), StringComparer.Ordinal);
+        foreach (var siteId in _cursors.Keys.Concat(_nonDrainingCycles.Keys).Concat(_stalled.Keys)
+                     .Where(id => !live.Contains(id)).Distinct().ToList())
+        {
+            _cursors.Remove(siteId);
+            _nonDrainingCycles.Remove(siteId);
+            if (_stalled.Remove(siteId, out var wasStalled) && wasStalled)
+            {
+                eventStream.Publish(new SiteAuditTelemetryStalledChanged(siteId, Stalled: false));
+            }
+        }
+    }
+
+    /// <summary>Test seam: number of sites with a held reconciliation cursor.</summary>
+    internal int TrackedSiteCountForTests => _cursors.Count;

    /// <summary>
    /// Initializes the reconciliation actor with its dependencies and registers the tick handler.
@@ -258,6 +302,8 @@ public class SiteAuditReconciliationActor : ReceiveActor
            return;
        }

+        PruneUnknownSites(sites, eventStream);
+
        if (sites.Count == 0)
        {
            return;