diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 953f8380..c6bb605d 100644 --- a/docs/requirements/Component-Communication.md +++ b/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. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index 7e120287..9d32bac4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/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 /// private readonly CancellationTokenSource _lifecycleCts = new(); + /// + /// Process-wide monotonic stamp for (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. + /// + private static long s_loadSequence; + + /// Sequence of the DB load currently in flight; 0 when none (the overlap guard). + private long _inFlightLoadSequence; + + /// A refresh requested while a load was in flight; re-run once the load completes. + private bool _refreshPending; + + /// Sequence of the last result handed to the transport; older results are dropped. + private long _lastAppliedLoadSequence; + /// /// Proxy for the central NotificationOutboxActor cluster singleton. /// Set via — the Host creates the singleton proxy @@ -103,10 +120,11 @@ public class CentralCommunicationActor : ReceiveActor // Periodic refresh trigger Receive(_ => 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(HandleSiteAddressLoadFailed); Receive(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>()?.Value.QueryTimeout + ?? TimeSpan.FromSeconds(30); + Task.Run(async () => { - using var scope = _serviceProvider.CreateScope(); - var repo = scope.ServiceProvider.GetRequiredService(); - 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(); - 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(); + 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(); + 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; /// /// Every configured site id, address-bearing or not, for aggregator pruning. /// gRPC endpoint pairs per site (from GrpcNodeA/GrpcNodeBAddress). +/// +/// 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. +/// public sealed record SiteAddressCacheLoaded( IReadOnlyCollection KnownSiteIds, - IReadOnlyDictionary GrpcContacts); + IReadOnlyDictionary GrpcContacts, + long Sequence); + +/// A failed (or deadline-cancelled) site-address DB load, piped back to release the overlap guard. +/// The failed load's sequence stamp. +/// The failure. +/// The per-load deadline when the load was cancelled by it; otherwise null. +internal sealed record SiteAddressLoadFailed(long Sequence, Exception Cause, TimeSpan? TimedOutAfter); /// /// A site's gRPC node-pair endpoints, as loaded from Site.GrpcNodeAAddress/ diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs new file mode 100644 index 00000000..d2842868 --- /dev/null +++ b/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; + +/// +/// 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. +/// +public class CentralCommunicationActorAddressRefreshTests : TestKit +{ + /// Initializes the test kit with a quieter Akka log level. + 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(); + var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, transport))); + return (actor, transport); + } + + /// Verifies that refresh ticks arriving while a load is in flight do not start more loads, and coalesce into one re-run. + [Fact] + public void RefreshWhileLoadInFlight_DoesNotStackLoads_AndCoalescesIntoOneRerun() + { + var gate = new TaskCompletionSource>(TaskCreationOptions.RunContinuationsAsynchronously); + var calls = 0; + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(_ => + Interlocked.Increment(ref calls) == 1 + ? gate.Task + : Task.FromResult>(new List { 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 { GrpcSite("site-a") }); + + // The wedged load completes, then exactly ONE coalesced re-run picks up the newer state. + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(c => c.GrpcContacts.ContainsKey("site-b"))), + TimeSpan.FromSeconds(3)); + Thread.Sleep(300); + Assert.Equal(2, Volatile.Read(ref calls)); + } + + /// Verifies that a load which never completes is cancelled at its deadline so the next refresh can run. + [Fact] + public void HungLoad_IsCancelledAtItsDeadline_AndTheNextRefreshRuns() + { + var calls = 0; + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(async ci => + { + if (Interlocked.Increment(ref calls) == 1) + { + await Task.Delay(Timeout.Infinite, ci.Arg()); + } + return (IReadOnlyList)new List { 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(c => c.GrpcContacts.ContainsKey("site-a"))), + TimeSpan.FromSeconds(3)); + } + + /// Verifies that a result older than the last applied one is dropped rather than reconciled. + [Fact] + public void OlderResult_ArrivingAfterANewerOne_IsDropped() + { + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(new List()); + var (actor, transport) = CreateActor(repo); + + static SiteAddressCacheLoaded Loaded(string site, long seq) => new( + new[] { site }, + new Dictionary { [site] = new("http://" + site + ":8083", null) }, + seq); + + actor.Tell(Loaded("site-new", long.MaxValue - 1)); + AwaitAssert(() => transport.Received().ReconcileSites( + Arg.Is(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(c => c.GrpcContacts.ContainsKey("site-old"))); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs index 04207de3..1f2203aa 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SitePairChannelProviderBadAddressTests.cs +++ b/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"); }