diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs index f2d4b207..bcdc2a37 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/CentralCommunicationActor.cs @@ -197,6 +197,7 @@ public class CentralCommunicationActor : ReceiveActor Receive(HandleHeartbeat); Receive(HandleSiteHealthReport); Receive(r => ProcessLocally(r.Report)); + Receive(r => MarkHeartbeatLocally(r.Heartbeat)); Receive(_ => { /* 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). + } + } + + /// + /// Marks a site heartbeat on the local aggregator without re-broadcasting. + /// Used for both site-originated heartbeats and peer-replicated ones. + /// + private void MarkHeartbeatLocally(HeartbeatMessage heartbeat) { var aggregator = _serviceProvider.GetService(); aggregator?.MarkHeartbeat(heartbeat.SiteId, heartbeat.Timestamp); @@ -665,6 +691,15 @@ internal record SiteAddressCacheLoaded( IReadOnlyDictionary> SiteContacts, IReadOnlyCollection KnownSiteIds); +/// +/// Peer-replication envelope for a site heartbeat, fanned out over the same +/// DistributedPubSub topic as 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. +/// +public sealed record SiteHeartbeatReplica(HeartbeatMessage Heartbeat); + /// /// Notification sent to debug view subscribers when the stream is terminated /// due to site disconnection. diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs index 40aeb4ef..14987b82 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/CentralCommunicationActorTests.cs +++ b/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(); + mockRepo.GetAllSitesAsync(Arg.Any()) + .Returns(new List()); + + var aggregator = Substitute.For(); + + var services = new ServiceCollection(); + services.AddScoped(_ => mockRepo); + services.AddSingleton(aggregator); + var sp = services.BuildServiceProvider(); + + var siteClientFactory = Substitute.For(); + 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() {