fix(communication): replicate heartbeat marks to the peer central node (closes post-failover offline blind window)

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

@@ -197,6 +197,7 @@ public class CentralCommunicationActor : ReceiveActor
         Receive<HeartbeatMessage>(HandleHeartbeat);
         Receive<SiteHealthReport>(HandleSiteHealthReport);
         Receive<SiteHealthReportReplica>(r => ProcessLocally(r.Report));
+        Receive<SiteHeartbeatReplica>(r => MarkHeartbeatLocally(r.Heartbeat));
         Receive<SubscribeAck>(_ => { /* DistributedPubSub subscribe confirmation */ });
 
         // Route enveloped messages to sites
@@ -371,6 +372,31 @@ public class CentralCommunicationActor : ReceiveActor
     }
 
     private void HandleHeartbeat(HeartbeatMessage heartbeat)
+    {
+        MarkHeartbeatLocally(heartbeat);
+
+        // Fan the heartbeat out to the peer central node so BOTH aggregators mark
+        // it, regardless of which central node the site's ClusterClient delivered
+        // to. Without this, a heartbeat that only ever reaches one node leaves the
+        // other node's aggregator blind to that site's liveness after a failover
+        // (arch review 02, Low). MarkHeartbeat is idempotent (timestamp overwrite),
+        // so self-delivery via pub-sub is harmless.
+        try
+        {
+            DistributedPubSub.Get(Context.System).Mediator.Tell(
+                new Publish(HealthReportTopic, new SiteHeartbeatReplica(heartbeat)));
+        }
+        catch
+        {
+            // No-op in non-clustered hosts (TestKit).
+        }
+    }
+
+    /// <summary>
+    /// Marks a site heartbeat on the local aggregator without re-broadcasting.
+    /// Used for both site-originated heartbeats and peer-replicated ones.
+    /// </summary>
+    private void MarkHeartbeatLocally(HeartbeatMessage heartbeat)
     {
         var aggregator = _serviceProvider.GetService<ICentralHealthAggregator>();
         aggregator?.MarkHeartbeat(heartbeat.SiteId, heartbeat.Timestamp);
@@ -665,6 +691,15 @@ internal record SiteAddressCacheLoaded(
     IReadOnlyDictionary<string, IReadOnlyList<string>> SiteContacts,
     IReadOnlyCollection<string> KnownSiteIds);
 
+/// <summary>
+/// Peer-replication envelope for a site heartbeat, fanned out over the same
+/// DistributedPubSub topic as <see cref="SiteHealthReportReplica"/> so both
+/// central aggregators mark heartbeats regardless of which node the site's
+/// ClusterClient delivered to. Only ever travels central↔central (same assembly
+/// version, rolled together), so the default reflective serializer is safe.
+/// </summary>
+public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat);
+
 /// <summary>
 /// Notification sent to debug view subscribers when the stream is terminated
 /// due to site disconnection.

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

@@ -148,6 +148,33 @@ public class CentralCommunicationActorTests : TestKit
         AwaitAssert(() => aggregator.Received(1).MarkHeartbeat("site1", timestamp));
     }
 
+    [Fact]
+    public void HeartbeatReplica_MarksLocalAggregator_WithoutRebroadcast()
+    {
+        // Task 16 (arch review 02): a heartbeat replicated from the peer central
+        // node marks the local aggregator (so post-failover both nodes know the
+        // site's liveness) but is NOT re-published (it arrives via pub-sub already).
+        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 centralActor = Sys.ActorOf(
+            Props.Create(() => new CentralCommunicationActor(sp, siteClientFactory)));
+
+        var ts = DateTimeOffset.UtcNow;
+        centralActor.Tell(new SiteHeartbeatReplica(new HeartbeatMessage("site-1", "host-a", true, ts)));
+
+        AwaitAssert(() => aggregator.Received(1).MarkHeartbeat("site-1", ts));
+    }
+
     [Fact]
     public void RefreshSiteAddresses_UpdatesCache()
     {