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

fix(health): acked SendAsync transport — report-loss counter restore is live, not dead code

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-08 16:12:48 -04:00
parent 17af376a8e
commit 6300d6d399
5 changed files with 142 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs
+8 -5
View File
@@ -140,10 +140,13 @@ public class HealthReportSender : BackgroundService
var seq = Interlocked.Increment(ref _sequenceNumber);
// CollectReport atomically read-and-resets
// the per-interval error counters via Interlocked.Exchange. If
// the Send below throws, those counts are otherwise lost
// forever — neither in the un-sent report nor in the now-zeroed
// collector. Snapshot the freshly-collected report so that on a
// the per-interval error counters via Interlocked.Exchange. The
// transport is now an ACKED round-trip (SendAsync throws on
// timeout/nack — review 01 [Medium]), so this restore path is live
// rather than dead code. If the SendAsync below throws, those
// counts are otherwise lost forever — neither in the un-sent
// report nor in the now-zeroed collector. Snapshot the
// freshly-collected report so that on a
// transport failure we can atomically restore the counts back
// into the collector via Interlocked.Add, so the next
// successful report includes them. Concurrent increments
@@ -157,7 +160,7 @@ public class HealthReportSender : BackgroundService
try
{
_transport.Send(reportWithSeq);
await _transport.SendAsync(reportWithSeq, stoppingToken).ConfigureAwait(false);
}
catch
{
src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/IHealthReportTransport.cs
+10 -3
View File
@@ -4,13 +4,20 @@ namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring;
/// <summary>
/// Abstraction for sending health reports to central.
/// In production, implemented via Akka remoting (Tell, fire-and-forget).
/// In production, implemented over Akka ClusterClient with an acked round-trip
/// (review 01 [Medium]): the transport Asks and awaits a
/// <see cref="SiteHealthReportAck"/>, so a lost/rejected report surfaces as a
/// fault the sender can observe (and its per-interval counter-restore fires).
/// </summary>
public interface IHealthReportTransport
{
/// <summary>
/// Sends a health report to central (fire-and-forget).
/// Sends a health report to central and completes only once central has
/// acknowledged processing. MUST THROW on non-delivery (Ask timeout, actor
/// system not started, or a not-accepted ack) — the sender's counter-restore
/// path in <c>HealthReportSender</c> depends on that exception.
/// </summary>
/// <param name="report">The site health report to send.</param>
void Send(SiteHealthReport report);
/// <param name="cancellationToken">Cancels the in-flight send.</param>
Task SendAsync(SiteHealthReport report, CancellationToken cancellationToken);
}
src/ZB.MOM.WW.ScadaBridge.Host/AkkaHealthReportTransport.cs
+13 -6
View File
@@ -6,13 +6,17 @@ using ZB.MOM.WW.ScadaBridge.Host.Actors;
namespace ZB.MOM.WW.ScadaBridge.Host;
/// <summary>
/// Sends SiteHealthReport to the local SiteCommunicationActor via Akka ActorSelection.
/// The SiteCommunicationActor forwards it to central.
/// Sends SiteHealthReport to the local SiteCommunicationActor via Akka ActorSelection,
/// which forwards it to central. The send is an acked round-trip (review 01 [Medium]):
/// it Asks and awaits the central <see cref="SiteHealthReportAck"/>, so a lost or
/// rejected report faults the sender's task (driving its counter-restore path).
/// </summary>
public class AkkaHealthReportTransport : IHealthReportTransport
{
private readonly AkkaHostedService _akkaService;
private static readonly TimeSpan AckTimeout = TimeSpan.FromSeconds(10);
/// <summary>
/// Initializes a new <see cref="AkkaHealthReportTransport"/> backed by the given Akka hosted service.
/// </summary>
@@ -23,12 +27,15 @@ public class AkkaHealthReportTransport : IHealthReportTransport
}
/// <inheritdoc />
public void Send(SiteHealthReport report)
public async Task SendAsync(SiteHealthReport report, CancellationToken cancellationToken)
{
var actorSystem = _akkaService.ActorSystem;
if (actorSystem == null) return;
var actorSystem = _akkaService.ActorSystem
?? throw new InvalidOperationException("Actor system not started — health report not sent");
var siteComm = actorSystem.ActorSelection("/user/site-communication");
siteComm.Tell(report, ActorRefs.NoSender);
var ack = await siteComm.Ask<SiteHealthReportAck>(report, AckTimeout, cancellationToken)
.ConfigureAwait(false);
if (!ack.Accepted)
throw new InvalidOperationException($"Health report #{report.SequenceNumber} rejected: {ack.Error}");
}
}
tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthReportSenderTests.cs
+7 -2
View File
@@ -13,7 +13,11 @@ public class HealthReportSenderTests
private class FakeTransport : IHealthReportTransport
{
public List<SiteHealthReport> SentReports { get; } = [];
public void Send(SiteHealthReport report) => SentReports.Add(report);
public Task SendAsync(SiteHealthReport report, CancellationToken cancellationToken)
{
SentReports.Add(report);
return Task.CompletedTask;
}
}
private class FakeSiteIdentityProvider : ISiteIdentityProvider
@@ -387,7 +391,7 @@ public class HealthReportSenderTests
private int _callCount;
public List<SiteHealthReport> SentReports { get; } = [];
public void Send(SiteHealthReport report)
public Task SendAsync(SiteHealthReport report, CancellationToken cancellationToken)
{
var n = Interlocked.Increment(ref _callCount);
if (n == 1)
@@ -395,6 +399,7 @@ public class HealthReportSenderTests
throw new InvalidOperationException("transport temporarily unavailable");
}
SentReports.Add(report);
return Task.CompletedTask;
}
}
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/AkkaHealthReportTransportTests.cs
+104
View File
@@ -0,0 +1,104 @@
using Akka.Actor;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.ClusterInfrastructure;
using ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
using ZB.MOM.WW.ScadaBridge.Commons.Types;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
using ZB.MOM.WW.ScadaBridge.Communication;
using ZB.MOM.WW.ScadaBridge.Host.Actors;
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
/// <summary>
/// Review 01 [Medium]: the health-report transport is now an ACKED round-trip.
/// SendAsync completes when central acks Accepted:true, throws when the ack is
/// not-accepted, and throws when no actor answers (Ask timeout / system not
/// started) — the fault the sender's counter-restore path depends on.
/// </summary>
public sealed class AkkaHealthReportTransportTests
{
private sealed class AckActor : ReceiveActor
{
public AckActor(bool accepted) => Receive<SiteHealthReport>(r =>
Sender.Tell(new SiteHealthReportAck(
r.SiteId, r.SequenceNumber, accepted, accepted ? null : "rejected by test")));
}
private static SiteHealthReport SampleReport(long seq = 1) =>
new("site-a", seq, DateTimeOffset.UtcNow,
new Dictionary<string, ConnectionHealth>(),
new Dictionary<string, TagResolutionStatus>(),
0, 0, new Dictionary<string, int>(), 0, 0, 0, 0);
private static AkkaHostedService BuildService(out ActorSystem system)
{
var port = FreePort();
var self = $"akka.tcp://scadabridge@127.0.0.1:{port}";
var svc = new AkkaHostedService(
new ServiceCollection().BuildServiceProvider(),
Options.Create(new NodeOptions { Role = "Site", NodeHostname = "127.0.0.1", RemotingPort = port }),
Options.Create(new ClusterOptions
{
SeedNodes = new List<string> { self },
SplitBrainResolverStrategy = "keep-oldest",
StableAfter = TimeSpan.FromSeconds(3),
HeartbeatInterval = TimeSpan.FromMilliseconds(500),
FailureDetectionThreshold = TimeSpan.FromSeconds(2),
MinNrOfMembers = 1,
DownIfAlone = true,
}),
Options.Create(new CommunicationOptions()),
NullLogger<AkkaHostedService>.Instance);
system = svc.GetOrCreateActorSystem();
return svc;
}
[Fact]
public async Task SendAsync_AckedAccepted_Completes()
{
var svc = BuildService(out var system);
try
{
system.ActorOf(Props.Create(() => new AckActor(true)), "site-communication");
var transport = new AkkaHealthReportTransport(svc);
await transport.SendAsync(SampleReport(), CancellationToken.None);
}
finally { await system.Terminate(); }
}
[Fact]
public async Task SendAsync_AckedNotAccepted_Throws()
{
var svc = BuildService(out var system);
try
{
system.ActorOf(Props.Create(() => new AckActor(false)), "site-communication");
var transport = new AkkaHealthReportTransport(svc);
await Assert.ThrowsAsync<InvalidOperationException>(
() => transport.SendAsync(SampleReport(), CancellationToken.None));
}
finally { await system.Terminate(); }
}
[Fact]
public async Task SendAsync_NoResponder_Throws()
{
var svc = BuildService(out var system);
try
{
// No actor at /user/site-communication => the Ask times out and faults.
var transport = new AkkaHealthReportTransport(svc);
await Assert.ThrowsAnyAsync<Exception>(
() => transport.SendAsync(SampleReport(), CancellationToken.None));
}
finally { await system.Terminate(); }
}
private static int FreePort()
{
var l = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0);
l.Start(); var p = ((System.Net.IPEndPoint)l.LocalEndpoint).Port; l.Stop(); return p;
}
}