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;
+ }
+}