diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index ad0858c2..18d6aef8 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -392,6 +392,15 @@ reports `MoreAvailable` and the cursor advances, mirroring `SiteCallAuditActor`. **Mixed versions:** a pre-PERF-1 site still serves `Forwarded` rows; central tolerates that unchanged (idempotent on `EventId`, merely more pages). +**Per-site state follows the site list (review 2026-10-21 06/MEM-2).** Each tick drops +the cursor, non-draining counter and stalled latch of any site no longer returned by +`ISiteEnumerator` (a latched site publishes `Stalled=false` as it goes), and the +`PullAuditEvents` / `PullSiteCalls` / `PullTraceEvents` invokers evict and dispose any +cached `(site, endpoint)` gRPC channel that is not a live site's primary or fallback +address on every enumeration. The in-memory abandoned-`EventId` short-circuit set is +capped at 10 000 entries (cleared when full — it is an optimisation only; the +deterministic synthetic id is what keeps re-abandonment idempotent). + **`PullTraceEvents` is a point read, not a reconciliation path (2026-09-05).** The Flow Debugger's value-change run cannot get its steps inline (the device echo triggers the run after the write reply has returned), so central polls the site diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcChannelCachePruning.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcChannelCachePruning.cs new file mode 100644 index 00000000..3ab919be --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcChannelCachePruning.cs @@ -0,0 +1,66 @@ +using System.Collections.Concurrent; +using Grpc.Net.Client; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; + +/// +/// Optional capability of a pull invoker that caches one per +/// (site, endpoint): 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). +/// +public interface IGrpcChannelCachePruner +{ + /// Retains only channels whose (site, endpoint) appears in . + /// The current site enumeration. + void RetainOnly(IReadOnlyList liveSites); +} + +/// Shared implementation of . +internal static class GrpcChannelCachePruning +{ + /// + /// Removes and disposes every channel in whose key is neither a + /// live site's primary nor its fallback endpoint. + /// + /// The invoker's channel cache. + /// The current site enumeration. + /// The number of channels pruned. + public static int RetainOnly( + ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel> channels, + IReadOnlyList 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; + } +} diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs index 4f553a54..a563c524 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullAuditEventsClient.cs @@ -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 — it creates a fresh channel for the -/// new address. Unlike SiteStreamGrpcClientFactory (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 -/// . 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 , 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. /// 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)); + /// + public void RetainOnly(IReadOnlyList liveSites) => + GrpcChannelCachePruning.RetainOnly(_channels, liveSites); + /// Disposes all cached channels. public void Dispose() { diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs index 4d5179ad..faae334e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullSiteCallsClient.cs @@ -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 PullSiteCallsAsync 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 . The channel for a previous address lingers idle -/// until (idle channels hold no streams — a minor cache -/// footprint cost, not a correctness or liveness gap). Sibling of +/// to . The channel for a previous address (or a deleted +/// site) is evicted and disposed by , which the client calls +/// with every site enumeration (review 2026-10-21 06/MEM-2). Sibling of /// . /// 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)); + /// + public void RetainOnly(IReadOnlyList liveSites) => + GrpcChannelCachePruning.RetainOnly(_channels, liveSites); + /// Disposes all cached channels. public void Dispose() { diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs index 6ca51d4f..f9e0d7ac 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/GrpcPullTraceEventsClient.cs @@ -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 /// per (site, endpoint), keepalive from , /// per-site PSK call credentials — the same shape as . /// -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)); + /// + public void RetainOnly(IReadOnlyList liveSites) => + GrpcChannelCachePruning.RetainOnly(_channels, liveSites); + /// Disposes every cached gRPC channel. public void Dispose() { diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index 55065999..ef60235d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -181,7 +181,51 @@ public class SiteAuditReconciliationActor : ReceiveActor private ICancelable? _timer; - private void RememberAbandoned(Guid eventId) => _abandonedEventIds.Add(eventId); + /// + /// Upper bound on . The set is only a short-circuit (see its + /// doc comment — determinism of 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). + /// + internal const int MaxRememberedAbandonedEventIds = 10_000; + + internal void RememberAbandoned(Guid eventId) + { + if (_abandonedEventIds.Count >= MaxRememberedAbandonedEventIds) + { + _abandonedEventIds.Clear(); + } + _abandonedEventIds.Add(eventId); + } + + /// Test seam: current size of the abandoned-id short-circuit set. + internal int AbandonedEventIdCountForTests => _abandonedEventIds.Count; + + /// + /// 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 CentralHealthAggregator.PruneUnknownSites). + /// A pruned site that was latched stalled publishes Stalled=false so the health + /// surface does not keep a badge for a site that no longer exists. + /// + private void PruneUnknownSites(IReadOnlyList sites, Akka.Event.EventStream eventStream) + { + var live = new HashSet(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)); + } + } + } + + /// Test seam: number of sites with a held reconciliation cursor. + internal int TrackedSiteCountForTests => _cursors.Count; /// /// 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; diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs new file mode 100644 index 00000000..b57f93ad --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/ReconciliationStatePruningTests.cs @@ -0,0 +1,149 @@ +using System.Collections.Concurrent; +using Akka.Actor; +using Akka.TestKit.Xunit2; +using Grpc.Net.Client; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using NSubstitute; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.AuditLog.Central; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; +using ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; +using ProtoPullRequest = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsRequest; +using ProtoPullResponse = ZB.MOM.WW.ScadaBridge.Communication.Grpc.PullAuditEventsResponse; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Central; + +/// +/// Review 2026-10-21 06/MEM-2: per-site reconciliation state and the per-(site, endpoint) gRPC +/// pull-channel caches are pruned for sites absent from the enumeration, and the +/// abandoned-EventId short-circuit set is bounded. +/// +public class ReconciliationStatePruningTests : TestKit +{ + private sealed class MutableEnumerator : ISiteEnumerator + { + public volatile IReadOnlyList Sites = Array.Empty(); + public Task> EnumerateAsync(CancellationToken ct = default) => Task.FromResult(Sites); + } + + private sealed class PruneRecordingInvoker + : GrpcPullAuditEventsClient.IPullAuditEventsInvoker, IGrpcChannelCachePruner + { + public List> Retained { get; } = new(); + public Task InvokeAsync( + string siteId, string endpoint, ProtoPullRequest request, CancellationToken ct) => + Task.FromResult(new ProtoPullResponse()); + public void RetainOnly(IReadOnlyList liveSites) => Retained.Add(liveSites); + } + + private sealed class AlwaysMoreClient : IPullAuditEventsClient + { + public Task PullAsync( + string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) => + Task.FromResult(new PullAuditEventsResponse( + new[] + { + ScadaBridgeAuditEventFactory.Create( + eventId: Guid.NewGuid(), + occurredAtUtc: DateTime.UtcNow, + channel: AuditChannel.ApiOutbound, + kind: AuditKind.ApiCall, + status: AuditStatus.Delivered, + sourceSiteId: siteId), + }, + MoreAvailable: true)); + } + + /// The shared pruning helper disposes channels whose (site, endpoint) is no longer enumerated. + [Fact] + public void RetainOnly_DropsChannelsOfDeletedSitesAndStaleAddresses() + { + var channels = new ConcurrentDictionary<(string Site, string Endpoint), GrpcChannel>(); + channels[("a", "http://a1:8083")] = GrpcChannel.ForAddress("http://a1:8083"); + channels[("a", "http://a-old:8083")] = GrpcChannel.ForAddress("http://a-old:8083"); + channels[("a", "http://a2:8083")] = GrpcChannel.ForAddress("http://a2:8083"); + channels[("gone", "http://g:8083")] = GrpcChannel.ForAddress("http://g:8083"); + + var pruned = GrpcChannelCachePruning.RetainOnly(channels, + new[] { new SiteEntry("a", "http://a1:8083", "http://a2:8083") }); + + Assert.Equal(2, pruned); + Assert.Equal( + new[] { ("a", "http://a1:8083"), ("a", "http://a2:8083") }.OrderBy(k => k.Item2), + channels.Keys.Select(k => (k.Site, k.Endpoint)).OrderBy(k => k.Item2)); + foreach (var c in channels.Values) c.Dispose(); + } + + /// The pull client hands every enumeration it makes to a pruning-capable invoker. + /// A task that represents the asynchronous operation. + [Fact] + public async Task PullClient_PrunesTheInvokerCache_WithEachEnumeration() + { + var sites = new MutableEnumerator { Sites = new[] { new SiteEntry("a", "http://a:8083") } }; + var invoker = new PruneRecordingInvoker(); + var sut = new GrpcPullAuditEventsClient(sites, invoker, NullLogger.Instance); + + await sut.PullAsync("a", DateTime.MinValue, null, 10, CancellationToken.None); + + Assert.Same(sites.Sites, Assert.Single(invoker.Retained)); + } + + /// A site removed from the enumeration loses its latched stalled state (published false). + [Fact] + public void RemovedSite_IsPruned_AndItsStalledLatchCleared() + { + var sites = new MutableEnumerator { Sites = new[] { new SiteEntry("doomed", "http://d:8083") } }; + var services = new ServiceCollection(); + services.AddScoped(_ => Substitute.For()); + var probe = CreateTestProbe(); + Sys.EventStream.Subscribe(probe.Ref, typeof(SiteAuditTelemetryStalledChanged)); + + var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor( + sites, + new AlwaysMoreClient(), + services.BuildServiceProvider(), + Options.Create(new SiteAuditReconciliationOptions + { + ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(100), + MaxPagesPerTick = 1, + StalledAfterNonDrainingCycles = 1, + }), + NullLogger.Instance))); + + Assert.True(probe.ExpectMsg(TimeSpan.FromSeconds(5)).Stalled); + + sites.Sites = Array.Empty(); + + var cleared = probe.ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.Equal("doomed", cleared.SiteId); + Assert.False(cleared.Stalled); + + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + } + + /// The abandoned-EventId short-circuit set never grows past its bound. + [Fact] + public void AbandonedEventIdSet_IsBounded() + { + var actorRef = ActorOfAsTestActorRef(Props.Create(() => new SiteAuditReconciliationActor( + new MutableEnumerator(), + new AlwaysMoreClient(), + new ServiceCollection().BuildServiceProvider(), + Options.Create(new SiteAuditReconciliationOptions { ReconciliationIntervalSeconds = 3600 }), + NullLogger.Instance))); + var actor = actorRef.UnderlyingActor; + + for (var i = 0; i < SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds + 5; i++) + { + actor.RememberAbandoned(Guid.NewGuid()); + } + + Assert.True(actor.AbandonedEventIdCountForTests <= SiteAuditReconciliationActor.MaxRememberedAbandonedEventIds); + } +}