diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index 9d32bac4..ee0136b1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -125,10 +125,19 @@ public class CentralCommunicationActor : ReceiveActor // operators could not distinguish "no sites configured" from "database is down". Log at // Warning and release the overlap guard. Receive(HandleSiteAddressLoadFailed); + // The load task converts its own faults to SiteAddressLoadFailed, so a Status.Failure + // here is unexpected — but it must still release the overlap guard, or every later + // refresh would be coalesced behind a load that never reports back. Receive(failure => + { _log.Warning(failure.Cause, "Failed to load site addresses from the database; the per-site gRPC channel " - + "cache was not refreshed and may be stale or empty")); + + "cache was not refreshed and may be stale or empty"); + if (_inFlightLoadSequence != 0) + { + CompleteLoad(_inFlightLoadSequence); + } + }); // Health monitoring: heartbeats and health reports from sites Receive(HandleHeartbeat); @@ -401,10 +410,13 @@ public class CentralCommunicationActor : ReceiveActor Task.Run(async () => { - using var loadCts = CancellationTokenSource.CreateLinkedTokenSource(lifecycle); - loadCts.CancelAfter(deadline); + CancellationTokenSource? loadCts = null; try { + // Inside the try: a throw here (e.g. a lifecycle CTS disposed by a racing + // PostStop) must still report back, or the overlap guard would never release. + loadCts = CancellationTokenSource.CreateLinkedTokenSource(lifecycle); + loadCts.CancelAfter(deadline); using var scope = _serviceProvider.CreateScope(); var repo = scope.ServiceProvider.GetRequiredService(); var sites = await repo.GetAllSitesAsync(loadCts.Token).ConfigureAwait(false); @@ -428,8 +440,13 @@ public class CentralCommunicationActor : ReceiveActor } catch (Exception ex) { - var timedOut = ex is OperationCanceledException && !lifecycle.IsCancellationRequested; - return new SiteAddressLoadFailed(sequence, ex, timedOut ? deadline : null); + var stopping = lifecycle.IsCancellationRequested; + var timedOut = ex is OperationCanceledException && !stopping; + return new SiteAddressLoadFailed(sequence, ex, timedOut ? deadline : null, ActorStopping: stopping); + } + finally + { + loadCts?.Dispose(); } }).PipeTo(self); } @@ -449,7 +466,13 @@ public class CentralCommunicationActor : ReceiveActor private void HandleSiteAddressLoadFailed(SiteAddressLoadFailed failed) { - if (failed.TimedOutAfter is { } after) + if (failed.ActorStopping) + { + // Cancelled because the (previous incarnation of the) actor stopped or restarted — + // expected, not a database problem. + _log.Debug("Site-address load {0} cancelled by actor shutdown/restart", failed.Sequence); + } + else if (failed.TimedOutAfter is { } after) { _log.Warning( "Loading site addresses from the database did not complete within {0}; the per-site " @@ -590,7 +613,8 @@ public sealed record SiteAddressCacheLoaded( /// 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); +/// True when the actor's lifecycle token was cancelled (stop/restart) — logged at Debug. +internal sealed record SiteAddressLoadFailed(long Sequence, Exception Cause, TimeSpan? TimedOutAfter, bool ActorStopping = false); /// /// 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 index d2842868..fda310a1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs @@ -116,4 +116,42 @@ public class CentralCommunicationActorAddressRefreshTests : TestKit transport.DidNotReceive().ReconcileSites( Arg.Is(c => c.GrpcContacts.ContainsKey("site-old"))); } + + /// Verifies that an unexpected Status.Failure releases the overlap guard so later refreshes still load. + [Fact] + public void UnexpectedStatusFailure_ReleasesTheOverlapGuard() + { + var calls = 0; + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(_ => + { + Interlocked.Increment(ref calls); + return new TaskCompletionSource>().Task; // never completes + }); + + var (actor, _) = CreateActor(repo); + AwaitCondition(() => Volatile.Read(ref calls) == 1, TimeSpan.FromSeconds(3)); + + actor.Tell(new Status.Failure(new InvalidOperationException("unexpected pipe fault"))); + actor.Tell(new RefreshSiteAddresses()); + + AwaitCondition(() => Volatile.Read(ref calls) == 2, TimeSpan.FromSeconds(3)); + } + + /// Verifies that a load cancelled by the actor lifecycle (stop/restart) is not logged as a Warning. + [Fact] + public void LoadCancelledByActorLifecycle_IsNotAWarning() + { + var repo = Substitute.For(); + repo.GetAllSitesAsync(Arg.Any()).Returns(new List()); + var (actor, _) = CreateActor(repo); + + // A previous incarnation's load, cancelled by its PostStop, lands on the restarted actor. + EventFilter.Warning(contains: "site addresses").Expect(0, TimeSpan.FromSeconds(1), () => + actor.Tell(new SiteAddressLoadFailed(1, new OperationCanceledException(), null, ActorStopping: true))); + + // A genuine DB failure still warns. + EventFilter.Warning(contains: "site addresses").ExpectOne(TimeSpan.FromSeconds(3), () => + actor.Tell(new SiteAddressLoadFailed(2, new InvalidOperationException("db down"), null))); + } }