feat(comm): ack site health reports end-to-end (SiteHealthReportAck, additive contract)

src/ZB.MOM.WW.ScadaBridge.Commons/Messages/Health/SiteHealthReportAck.cs

@@ -0,0 +1,10 @@
+namespace ZB.MOM.WW.ScadaBridge.Commons.Messages.Health;
+
+/// <summary>
+/// Acknowledgement for a <see cref="SiteHealthReport"/> forwarded site→central.
+/// Review 01 [Medium]: the transport was fire-and-forget, so the sender's
+/// interval-counter restore logic could never observe a loss. The ack makes
+/// delivery observable end-to-end (central processed the report).
+/// </summary>
+public sealed record SiteHealthReportAck(
+    string SiteId, long SequenceNumber, bool Accepted, string? Error = null);

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs

@@ -394,6 +394,15 @@ public class CentralCommunicationActor : ReceiveActor
         {
             // No-op in non-clustered hosts (TestKit).
         }
+
+        // Ack the site so its AkkaHealthReportTransport Ask completes and the
+        // report-loss counter-restore path can observe delivery (review 01
+        // [Medium]). Guarded so the peer-replica path (SiteHealthReportReplica,
+        // which arrives without an Ask sender) never dead-letters an ack.
+        if (!Sender.IsNobody())
+        {
+            Sender.Tell(new SiteHealthReportAck(report.SiteId, report.SequenceNumber, Accepted: true));
+        }
     }
 
     /// <summary>

src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteCommunicationActor.cs

@@ -390,11 +390,28 @@ public class SiteCommunicationActor : ReceiveActor, IWithTimers
         // Internal: send heartbeat tick
         Receive<SendHeartbeat>(_ => SendHeartbeatToCentral());
 
-        // Internal: forward health report to central
+        // Internal: forward health report to central. The original Sender (the
+        // AkkaHealthReportTransport's Ask) is forwarded as the ClusterClient.Send
+        // sender so the central SiteHealthReportAck routes straight back to the
+        // waiting Ask — making report delivery observable end-to-end (review 01
+        // [Medium]). Mirrors the NotificationSubmit ack pattern above.
         Receive<SiteHealthReport>(msg =>
         {
-            _centralClient?.Tell(
+            if (_centralClient == null)
-                new ClusterClient.Send("/user/central-communication", msg), Self);
+            {
+                // No ClusterClient registered yet. A non-accepted ack makes the
+                // sender's counter-restore path treat this tick as a loss.
+                _log.Warning(
+                    "Cannot forward SiteHealthReport #{0} — no central ClusterClient registered",
+                    msg.SequenceNumber);
+                Sender.Tell(new SiteHealthReportAck(
+                    msg.SiteId, msg.SequenceNumber, Accepted: false,
+                    Error: "Central ClusterClient not registered"));
+                return;
+            }
+
+            _centralClient.Tell(
+                new ClusterClient.Send("/user/central-communication", msg), Sender);
         });
 
     }

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/HealthReportAckTests.cs

@@ -0,0 +1,66 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+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.Actors;
+using ZB.MOM.WW.ScadaBridge.HealthMonitoring;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
+
+/// <summary>
+/// Review 01 [Medium]: site→central health reports were fire-and-forget, so a lost
+/// report was invisible to the sender. These tests pin the new end-to-end ack:
+/// the site actor replies not-accepted when it has no central ClusterClient, and the
+/// central actor processes + acks a report it receives.
+/// </summary>
+public class HealthReportAckTests : TestKit
+{
+    private readonly CommunicationOptions _options = new();
+
+    private static SiteHealthReport SampleReport(long seq, string siteId = "site-a") =>
+        new(siteId, seq, DateTimeOffset.UtcNow,
+            new Dictionary<string, ConnectionHealth>(),
+            new Dictionary<string, TagResolutionStatus>(),
+            0, 0, new Dictionary<string, int>(), 0, 0, 0, 0);
+
+    [Fact]
+    public void SiteCommunicationActor_NoCentralClient_RepliesNotAccepted()
+    {
+        var dmProbe = CreateTestProbe();
+        var siteComm = Sys.ActorOf(Props.Create(() =>
+            new SiteCommunicationActor("site-a", _options, dmProbe.Ref)));
+        // No RegisterCentralClient sent => _centralClient is null.
+        siteComm.Tell(SampleReport(seq: 7));
+        var ack = ExpectMsg<SiteHealthReportAck>();
+        Assert.False(ack.Accepted);
+        Assert.Equal(7, ack.SequenceNumber);
+        Assert.Equal("site-a", ack.SiteId);
+    }
+
+    [Fact]
+    public void CentralCommunicationActor_OnReport_ProcessesAndAcks()
+    {
+        var mockRepo = Substitute.For<ISiteRepository>();
+        mockRepo.GetAllSitesAsync(Arg.Any<CancellationToken>()).Returns(new List<Site>());
+        var aggregator = Substitute.For<ICentralHealthAggregator>();
+
+        var services = new ServiceCollection();
+        services.AddScoped(_ => mockRepo);
+        services.AddSingleton(aggregator);
+        var sp = services.BuildServiceProvider();
+
+        var siteClientFactory = Substitute.For<ISiteClientFactory>();
+        var actor = Sys.ActorOf(Props.Create(() => new CentralCommunicationActor(sp, siteClientFactory)));
+
+        actor.Tell(SampleReport(seq: 3));
+        var ack = ExpectMsg<SiteHealthReportAck>();
+        Assert.True(ack.Accepted);
+        Assert.Equal(3, ack.SequenceNumber);
+        AwaitAssert(() => aggregator.Received().ProcessReport(Arg.Is<SiteHealthReport>(r => r.SequenceNumber == 3)));
+    }
+}