dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(communication): site-address load guard cannot wedge on an unexpected fault; lifecycle-cancelled loads log at Debug (review 2026-10-21 03/M-1 follow-up)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-22 08:29:02 -04:00
parent 4456a04e81
commit e0ebebac11
2 changed files with 69 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs
+31 -7
View File
@@ -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<SiteAddressLoadFailed>(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<Status.Failure>(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<HeartbeatMessage>(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<ISiteRepository>();
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(
/// <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);
/// <param name="ActorStopping">True when the actor's lifecycle token was cancelled (stop/restart) — logged at Debug.</param>
internal sealed record SiteAddressLoadFailed(long Sequence, Exception Cause, TimeSpan? TimedOutAfter, bool ActorStopping = false);
/// <summary>
/// A site's gRPC node-pair endpoints, as loaded from <c>Site.GrpcNodeAAddress</c>/
tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorAddressRefreshTests.cs
+38
View File
@@ -116,4 +116,42 @@ public class CentralCommunicationActorAddressRefreshTests : TestKit
transport.DidNotReceive().ReconcileSites(
Arg.Is<SiteAddressCacheLoaded>(c => c.GrpcContacts.ContainsKey("site-old")));
}
/// <summary>Verifies that an unexpected Status.Failure releases the overlap guard so later refreshes still load.</summary>
[Fact]
public void UnexpectedStatusFailure_ReleasesTheOverlapGuard()
{
var calls = 0;
var repo = Substitute.For<ISiteRepository>();
repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(_ =>
{
Interlocked.Increment(ref calls);
return new TaskCompletionSource<IReadOnlyList<Site>>().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));
}
/// <summary>Verifies that a load cancelled by the actor lifecycle (stop/restart) is not logged as a Warning.</summary>
[Fact]
public void LoadCancelledByActorLifecycle_IsNotAWarning()
{
var repo = Substitute.For<ISiteRepository>();
repo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>());
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)));
}
}