fix(communication): site-address refresh gets an overlap guard, per-load deadline and monotonic sequence (review 2026-10-21 03/M-1)

docs/requirements/Component-Communication.md

@@ -359,7 +359,7 @@ Central discovers site addresses through the **configuration database**, not run
 
 - Each site record in the Sites table includes optional **GrpcNodeAAddress** and **GrpcNodeBAddress** fields containing the site nodes' gRPC endpoints (e.g., `http://host:8083`), used for **both** central→site command/control (`SiteCommandService`) and site→central streaming (`SiteStreamService`). The legacy **NodeAAddress** / **NodeBAddress** Akka fields are retained but no longer drive cross-cluster dialling — Akka remoting is now intra-cluster only.
 - The **CentralCommunicationActor** loads all site addresses from the database at startup and builds a per-site gRPC-endpoint cache. `GrpcSiteTransport` uses it (via a `SitePairChannelProvider`) to dial each site's `SiteCommandService` as a **per-site NodeA→NodeB failover channel pair**; the **SiteStreamGrpcClientFactory** uses the same `GrpcNodeAAddress` / `GrpcNodeBAddress` for per-site streaming channels.
-- The address cache is **refreshed every 60 seconds** and **on-demand** when site records are added, edited, or deleted via the Central UI or CLI. Channels are rebuilt when a site's gRPC endpoints change.
+- The address cache is **refreshed every 60 seconds** and **on-demand** when site records are added, edited, or deleted via the Central UI or CLI. Command-plane channels are rebuilt when a site's gRPC endpoints change. At most **one** DB load is in flight: a tick or on-demand refresh that arrives mid-load is coalesced into a single re-run after it, each load is bounded by a deadline (`QueryTimeout`, default 30 s) so a wedged read cannot pin the refresh, and every result carries a monotonic sequence so an older result arriving after a newer one is dropped rather than re-applying stale addresses.
 - A **malformed** `GrpcNodeAAddress`/`GrpcNodeBAddress` (not an absolute `http://`/`https://` URL with a host, or one the gRPC channel cannot be built from) is **rejected per node at refresh time**: `SitePairChannelProvider.UpdateSite` builds every replacement channel *before* parking the old one, so a rejected side keeps its previous channel and endpoint (the other side's valid change still applies), and the rejection is logged at **Error** naming the site and column (once per distinct bad value). `GrpcSiteTransport.ReconcileSites` reconciles each site in its own try/catch and `CentralCommunicationActor` guards the reconcile call, so one bad row can neither stop the other sites in the tick from being refreshed nor restart the actor. Write-path validation (Central UI/CLI/bundle import) is the first line of defence; this is the runtime backstop.
 - When routing a command to a site, central calls `SiteCommandService` on the site's active gRPC endpoint; on error it flips to the other node endpoint. The failover is an explicit NodeA→NodeB channel-pair flip, not internal transport routing.
 - **Heartbeats** from sites serve **health monitoring only** — they do not serve as a registration or address discovery mechanism.

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs

@@ -3,6 +3,7 @@ using Akka.Cluster.Tools.PublishSubscribe;
 using Akka.Event;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
 using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
 using ZB.MOM.WW.ScadaBridge.Commons.Messages.Audit;
 using ZB.MOM.WW.ScadaBridge.Commons.Messages.Communication;
@@ -58,6 +59,22 @@ public class CentralCommunicationActor : ReceiveActor
     /// </summary>
     private readonly CancellationTokenSource _lifecycleCts = new();
 
+    /// <summary>
+    /// Process-wide monotonic stamp for <see cref="SiteAddressCacheLoaded"/> (review 2026-10-21
+    /// 03/M-1). Static so it stays monotonic across actor restarts: a late result piped from a
+    /// previous incarnation's load can never outrank a newer one.
+    /// </summary>
+    private static long s_loadSequence;
+
+    /// <summary>Sequence of the DB load currently in flight; 0 when none (the overlap guard).</summary>
+    private long _inFlightLoadSequence;
+
+    /// <summary>A refresh requested while a load was in flight; re-run once the load completes.</summary>
+    private bool _refreshPending;
+
+    /// <summary>Sequence of the last result handed to the transport; older results are dropped.</summary>
+    private long _lastAppliedLoadSequence;
+
     /// <summary>
     /// Proxy <see cref="IActorRef"/> for the central NotificationOutboxActor cluster singleton.
     /// Set via <see cref="RegisterNotificationOutbox"/> — the Host creates the singleton proxy
@@ -103,10 +120,11 @@ public class CentralCommunicationActor : ReceiveActor
         // Periodic refresh trigger
         Receive<RefreshSiteAddresses>(_ => LoadSiteAddressesFromDb());
 
-        // A faulted LoadSiteAddressesFromDb task is piped here as a
-        // Status.Failure. Without this handler the failure was an unhandled message
-        // (debug-level only) and the refresh failed silently — operators could not
-        // distinguish "no sites configured" from "database is down". Log at Warning.
+        // A faulted (or deadline-cancelled) LoadSiteAddressesFromDb task. Without this the
+        // failure was an unhandled message (debug-level only) and the refresh failed silently —
+        // operators could not distinguish "no sites configured" from "database is down". Log at
+        // Warning and release the overlap guard.
+        Receive<SiteAddressLoadFailed>(HandleSiteAddressLoadFailed);
         Receive<Status.Failure>(failure =>
             _log.Warning(failure.Cause,
                 "Failed to load site addresses from the database; the per-site gRPC channel "
@@ -348,49 +366,118 @@ public class CentralCommunicationActor : ReceiveActor
 
     private void LoadSiteAddressesFromDb()
     {
+        // Overlap guard (review 2026-10-21 03/M-1): a wedged DB used to stack one Task.Run + DI
+        // scope + DbContext per 60 s tick, and their late completions could land out of order.
+        // A tick (or an admin-change refresh) that arrives mid-load is coalesced into ONE re-run
+        // once the current load finishes, so an on-demand refresh is never lost.
+        if (_inFlightLoadSequence != 0)
+        {
+            _refreshPending = true;
+            return;
+        }
+
         var self = Self;
         // Pass the actor's lifecycle CT into the repository
         // call so a hung database query is cancelled when the actor stops
         // rather than leaving the piped task to accumulate. Captured locally
         // because the lifecycle CTS may have been disposed by PostStop on a
         // racing late tick; treat that as "actor gone, give up".
-        CancellationToken ct;
+        CancellationToken lifecycle;
         try
         {
-            ct = _lifecycleCts.Token;
+            lifecycle = _lifecycleCts.Token;
         }
         catch (ObjectDisposedException)
         {
             return;
         }
 
+        var sequence = Interlocked.Increment(ref s_loadSequence);
+        _inFlightLoadSequence = sequence;
+        _refreshPending = false;
+        // Per-load deadline: a hung read is abandoned instead of pinning the overlap guard.
+        var deadline = _serviceProvider.GetService<IOptions<CommunicationOptions>>()?.Value.QueryTimeout
+            ?? TimeSpan.FromSeconds(30);
+
         Task.Run(async () =>
         {
-            using var scope = _serviceProvider.CreateScope();
-            var repo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
-            var sites = await repo.GetAllSitesAsync(ct).ConfigureAwait(false);
-
-            // gRPC-endpoint cache (the GrpcNodeA/BAddress columns, NOT the legacy Akka
-            // NodeA/BAddress ones, which no longer drive cross-cluster dialling).
-            var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
-            foreach (var site in sites)
+            using var loadCts = CancellationTokenSource.CreateLinkedTokenSource(lifecycle);
+            loadCts.CancelAfter(deadline);
+            try
             {
-                var grpcA = string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) ? null : site.GrpcNodeAAddress;
-                var grpcB = string.IsNullOrWhiteSpace(site.GrpcNodeBAddress) ? null : site.GrpcNodeBAddress;
-                if (grpcA is not null || grpcB is not null)
-                    grpcContacts[site.SiteIdentifier] = new SiteGrpcEndpoints(grpcA, grpcB);
-            }
+                using var scope = _serviceProvider.CreateScope();
+                var repo = scope.ServiceProvider.GetRequiredService<ISiteRepository>();
+                var sites = await repo.GetAllSitesAsync(loadCts.Token).ConfigureAwait(false);
 
-            // Carry the full configured-site identifier set (not just the
-            // gRPC-address-bearing subset) so the aggregator prunes only
-            // genuinely-deleted sites and never an addressless-but-configured one.
-            var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
-            return new SiteAddressCacheLoaded(knownSiteIds, grpcContacts);
+                // gRPC-endpoint cache (the GrpcNodeA/BAddress columns, NOT the legacy Akka
+                // NodeA/BAddress ones, which no longer drive cross-cluster dialling).
+                var grpcContacts = new Dictionary<string, SiteGrpcEndpoints>();
+                foreach (var site in sites)
+                {
+                    var grpcA = string.IsNullOrWhiteSpace(site.GrpcNodeAAddress) ? null : site.GrpcNodeAAddress;
+                    var grpcB = string.IsNullOrWhiteSpace(site.GrpcNodeBAddress) ? null : site.GrpcNodeBAddress;
+                    if (grpcA is not null || grpcB is not null)
+                        grpcContacts[site.SiteIdentifier] = new SiteGrpcEndpoints(grpcA, grpcB);
+                }
+
+                // Carry the full configured-site identifier set (not just the
+                // gRPC-address-bearing subset) so the aggregator prunes only
+                // genuinely-deleted sites and never an addressless-but-configured one.
+                var knownSiteIds = sites.Select(s => s.SiteIdentifier).ToList();
+                return (object)new SiteAddressCacheLoaded(knownSiteIds, grpcContacts, sequence);
+            }
+            catch (Exception ex)
+            {
+                var timedOut = ex is OperationCanceledException && !lifecycle.IsCancellationRequested;
+                return new SiteAddressLoadFailed(sequence, ex, timedOut ? deadline : null);
+            }
         }).PipeTo(self);
     }
 
+    private void CompleteLoad(long sequence)
+    {
+        if (sequence != _inFlightLoadSequence)
+        {
+            return; // a previous incarnation's / superseded load — the guard belongs to another one
+        }
+        _inFlightLoadSequence = 0;
+        if (_refreshPending)
+        {
+            LoadSiteAddressesFromDb();
+        }
+    }
+
+    private void HandleSiteAddressLoadFailed(SiteAddressLoadFailed failed)
+    {
+        if (failed.TimedOutAfter is { } after)
+        {
+            _log.Warning(
+                "Loading site addresses from the database did not complete within {0}; the per-site "
+                + "gRPC channel cache was not refreshed and may be stale or empty", after);
+        }
+        else
+        {
+            _log.Warning(failed.Cause,
+                "Failed to load site addresses from the database; the per-site gRPC channel "
+                + "cache was not refreshed and may be stale or empty");
+        }
+        CompleteLoad(failed.Sequence);
+    }
+
     private void HandleSiteAddressCacheLoaded(SiteAddressCacheLoaded msg)
     {
+        CompleteLoad(msg.Sequence);
+
+        // Never let an older read overwrite a newer one (review 2026-10-21 03/M-1): applying
+        // stale addresses would park freshly built channels and rebuild from the old endpoints.
+        if (msg.Sequence <= _lastAppliedLoadSequence)
+        {
+            _log.Debug("Dropping stale site-address load {0} (last applied {1})",
+                msg.Sequence, _lastAppliedLoadSequence);
+            return;
+        }
+        _lastAppliedLoadSequence = msg.Sequence;
+
         // Per-site transport resource reconciliation (build/drop gRPC channel
         // pairs). Runs on the actor thread each refresh tick. Guarded (review 2026-10-21 03/R-1):
         // a throw here would restart this actor, whose PreStart re-reads the Sites table at
@@ -490,9 +577,20 @@ public record RefreshSiteAddresses;
 /// </summary>
 /// <param name="KnownSiteIds">Every configured site id, address-bearing or not, for aggregator pruning.</param>
 /// <param name="GrpcContacts">gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress).</param>
+/// <param name="Sequence">
+/// Monotonic load stamp; the actor drops a result whose sequence is not newer than the last one it
+/// applied, so an out-of-order completion never re-applies stale addresses.
+/// </param>
 public sealed record SiteAddressCacheLoaded(
     IReadOnlyCollection<string> KnownSiteIds,
-    IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts);
+    IReadOnlyDictionary<string, SiteGrpcEndpoints> GrpcContacts,
+    long Sequence);
+
+/// <summary>A failed (or deadline-cancelled) site-address DB load, piped back to release the overlap guard.</summary>
+/// <param name="Sequence">The failed load's sequence stamp.</param>
+/// <param name="Cause">The failure.</param>
+/// <param name="TimedOutAfter">The per-load deadline when the load was cancelled by it; otherwise null.</param>
+internal sealed record SiteAddressLoadFailed(long Sequence, Exception Cause, TimeSpan? TimedOutAfter);
 
 /// <summary>
 /// A site's gRPC node-pair endpoints, as loaded from <c>Site.GrpcNodeAAddress</c>/

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs

@@ -0,0 +1,119 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Options;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Communication.Actors;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
+
+/// <summary>
+/// The 60 s site-address refresh (review 2026-10-21 03/M-1): at most one DB load in flight, a
+/// per-load deadline so a wedged read cannot pin the refresh forever, and a monotonic sequence so a
+/// late (older) result never overwrites a newer one.
+/// </summary>
+public class CentralCommunicationActorAddressRefreshTests : TestKit
+{
+    /// <summary>Initializes the test kit with a quieter Akka log level.</summary>
+    public CentralCommunicationActorAddressRefreshTests() : base(@"akka.loglevel = WARNING") { }
+
+    private static Site GrpcSite(string id) =>
+        new("Test " + id, id) { GrpcNodeAAddress = "http://a:8083" };
+
+    private (IActorRef Actor, ISiteCommandTransport Transport) CreateActor(
+        ISiteRepository repo, TimeSpan? loadTimeout = null)
+    {
+        var services = new ServiceCollection();
+        services.AddScoped(_ => repo);
+        if (loadTimeout is { } t)
+        {
+            services.AddSingleton(Options.Create(new CommunicationOptions { QueryTimeout = t }));
+        }
+        var sp = services.BuildServiceProvider();
+        var transport = Substitute.For<ISiteCommandTransport>();
+        var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport)));
+        return (actor, transport);
+    }
+
+    /// <summary>Verifies that refresh ticks arriving while a load is in flight do not start more loads, and coalesce into one re-run.</summary>
+    [Fact]
+    public void RefreshWhileLoadInFlight_DoesNotStackLoads_AndCoalescesIntoOneRerun()
+    {
+        var gate = new TaskCompletionSource<IReadOnlyList<Site>>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var calls = 0;
+        var repo = Substitute.For<ISiteRepository>();
+        repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(_ =>
+            Interlocked.Increment(ref calls) == 1
+                ? gate.Task
+                : Task.FromResult<IReadOnlyList<Site>>(new List<Site> { GrpcSite("site-b") }));
+
+        var (actor, transport) = CreateActor(repo);
+        AwaitCondition(() => Volatile.Read(ref calls) == 1, TimeSpan.FromSeconds(3));
+
+        for (var i = 0; i < 5; i++) actor.Tell(new RefreshSiteAddresses());
+        Thread.Sleep(300);
+        Assert.Equal(1, Volatile.Read(ref calls));
+
+        gate.SetResult(new List<Site> { GrpcSite("site-a") });
+
+        // The wedged load completes, then exactly ONE coalesced re-run picks up the newer state.
+        AwaitAssert(() => transport.Received().ReconcileSites(
+            Arg.Is<SiteAddressCacheLoaded>(c => c.GrpcContacts.ContainsKey("site-b"))),
+            TimeSpan.FromSeconds(3));
+        Thread.Sleep(300);
+        Assert.Equal(2, Volatile.Read(ref calls));
+    }
+
+    /// <summary>Verifies that a load which never completes is cancelled at its deadline so the next refresh can run.</summary>
+    [Fact]
+    public void HungLoad_IsCancelledAtItsDeadline_AndTheNextRefreshRuns()
+    {
+        var calls = 0;
+        var repo = Substitute.For<ISiteRepository>();
+        repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(async ci =>
+        {
+            if (Interlocked.Increment(ref calls) == 1)
+            {
+                await Task.Delay(Timeout.Infinite, ci.Arg<CancellationToken>());
+            }
+            return (IReadOnlyList<Site>)new List<Site> { GrpcSite("site-a") };
+        });
+
+        var (actor, transport) = CreateActor(repo, loadTimeout: TimeSpan.FromMilliseconds(300));
+        AwaitCondition(() => Volatile.Read(ref calls) == 1, TimeSpan.FromSeconds(3));
+
+        // Let the deadline expire, then a fresh tick must be able to load.
+        Thread.Sleep(600);
+        actor.Tell(new RefreshSiteAddresses());
+
+        AwaitAssert(() => transport.Received().ReconcileSites(
+            Arg.Is<SiteAddressCacheLoaded>(c => c.GrpcContacts.ContainsKey("site-a"))),
+            TimeSpan.FromSeconds(3));
+    }
+
+    /// <summary>Verifies that a result older than the last applied one is dropped rather than reconciled.</summary>
+    [Fact]
+    public void OlderResult_ArrivingAfterANewerOne_IsDropped()
+    {
+        var repo = Substitute.For<ISiteRepository>();
+        repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>());
+        var (actor, transport) = CreateActor(repo);
+
+        static SiteAddressCacheLoaded Loaded(string site, long seq) => new(
+            new[] { site },
+            new Dictionary<string, SiteGrpcEndpoints> { [site] = new("http://" + site + ":8083", null) },
+            seq);
+
+        actor.Tell(Loaded("site-new", long.MaxValue - 1));
+        AwaitAssert(() => transport.Received().ReconcileSites(
+            Arg.Is<SiteAddressCacheLoaded>(c => c.GrpcContacts.ContainsKey("site-new"))),
+            TimeSpan.FromSeconds(3));
+
+        actor.Tell(Loaded("site-old", long.MaxValue - 2));
+        Thread.Sleep(300);
+        transport.DidNotReceive().ReconcileSites(
+            Arg.Is<SiteAddressCacheLoaded>(c => c.GrpcContacts.ContainsKey("site-old")));
+    }
+}

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs

@@ -101,7 +101,7 @@ public class SitePairChannelProviderBadAddressTests : TestKit
             ["site-good"] = new("http://localhost:15222", null),
         };
 
-        transport.ReconcileSites(new SiteAddressCacheLoaded(new[] { " ", "site-good" }, contacts));
+        transport.ReconcileSites(new SiteAddressCacheLoaded(new[] { " ", "site-good" }, contacts, 1));
 
         _ = await CurrentChannel(provider, "site-good");
     }