From 6300d6d399d2fd10bf11eb911add0427558f575b Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 16:12:48 -0400 Subject: [PATCH] =?UTF-8?q?fix(health):=20acked=20SendAsync=20transport=20?= =?UTF-8?q?=E2=80=94=20report-loss=20counter=20restore=20is=20live,=20not?= =?UTF-8?q?=20dead=20code?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../HealthReportSender.cs | 13 ++- .../IHealthReportTransport.cs | 13 ++- .../AkkaHealthReportTransport.cs | 19 +++- .../HealthReportSenderTests.cs | 9 +- .../AkkaHealthReportTransportTests.cs | 104 ++++++++++++++++++ 5 files changed, 142 insertions(+), 16 deletions(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Host.Tests/AkkaHealthReportTransportTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs index 671ad86f..d5335f08 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/HealthReportSender.cs @@ -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 { diff --git a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/IHealthReportTransport.cs b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/IHealthReportTransport.cs index b76ee761..cf4ae906 100644 --- a/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/IHealthReportTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.HealthMonitoring/IHealthReportTransport.cs @@ -4,13 +4,20 @@ namespace ZB.MOM.WW.ScadaBridge.HealthMonitoring; /// /// 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 +/// , so a lost/rejected report surfaces as a +/// fault the sender can observe (and its per-interval counter-restore fires). /// public interface IHealthReportTransport { /// - /// 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 HealthReportSender depends on that exception. /// /// The site health report to send. - void Send(SiteHealthReport report); + /// Cancels the in-flight send. + Task SendAsync(SiteHealthReport report, CancellationToken cancellationToken); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/AkkaHealthReportTransport.cs b/src/ZB.MOM.WW.ScadaBridge.Host/AkkaHealthReportTransport.cs index 40b59c7d..80ec12b0 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/AkkaHealthReportTransport.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/AkkaHealthReportTransport.cs @@ -6,13 +6,17 @@ using ZB.MOM.WW.ScadaBridge.Host.Actors; namespace ZB.MOM.WW.ScadaBridge.Host; /// -/// 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 , so a lost or +/// rejected report faults the sender's task (driving its counter-restore path). /// public class AkkaHealthReportTransport : IHealthReportTransport { private readonly AkkaHostedService _akkaService; + private static readonly TimeSpan AckTimeout = TimeSpan.FromSeconds(10); + /// /// Initializes a new backed by the given Akka hosted service. /// @@ -23,12 +27,15 @@ public class AkkaHealthReportTransport : IHealthReportTransport } /// - 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(report, AckTimeout, cancellationToken) + .ConfigureAwait(false); + if (!ack.Accepted) + throw new InvalidOperationException($"Health report #{report.SequenceNumber} rejected: {ack.Error}"); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthReportSenderTests.cs b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthReportSenderTests.cs index fc8c26c0..5ea6a969 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthReportSenderTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.HealthMonitoring.Tests/HealthReportSenderTests.cs @@ -13,7 +13,11 @@ public class HealthReportSenderTests private class FakeTransport : IHealthReportTransport { public List 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 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; } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/AkkaHealthReportTransportTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/AkkaHealthReportTransportTests.cs new file mode 100644 index 00000000..9f7ef0d2 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/AkkaHealthReportTransportTests.cs @@ -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; + +/// +/// 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. +/// +public sealed class AkkaHealthReportTransportTests +{ + private sealed class AckActor : ReceiveActor + { + public AckActor(bool accepted) => Receive(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(), + new Dictionary(), + 0, 0, new Dictionary(), 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 { 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.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( + () => 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( + () => 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; + } +}