From f7b9d342e471dba8cbb1187c32c87111ca82b80d Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 16:05:52 -0400 Subject: [PATCH] =?UTF-8?q?refactor(host):=20central=20singletons=20via=20?= =?UTF-8?q?CentralSingletonRegistrar=20=E2=80=94=20outbox/audit-ingest=20g?= =?UTF-8?q?ain=20drain=20tasks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Actors/AkkaHostedService.cs | 246 +++--------------- .../CoordinatedShutdownTests.cs | 37 +++ 2 files changed, 78 insertions(+), 205 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs index 9ee529c1..a51c8ef6 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs @@ -414,30 +414,29 @@ akka {{ var outboxAuditWriter = _serviceProvider .GetRequiredService(); - var outboxSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxActor( + // All central singletons below are created through CentralSingletonRegistrar.Start, + // which pins the actor to the active central node (ClusterSingletonManager), exposes a + // stable address (ClusterSingletonProxy), and — crucially — registers a PhaseClusterLeave + // GracefulStop drain task so an in-flight EF write drains before handover (review 01 + // [Medium]: notification-outbox and audit-log-ingest previously had NO drain task). Names + // are unchanged ({name}-singleton / {name}-proxy). The two SITE singletons stay + // hand-rolled below because they are role-scoped (.WithRole(siteRole)). + var outbox = CentralSingletonRegistrar.Start( + _actorSystem!, "notification-outbox", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxActor( _serviceProvider, outboxOptions, outboxAuditWriter, outboxLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("notification-outbox")); - _actorSystem!.ActorOf(outboxSingletonProps, "notification-outbox-singleton"); - - var outboxProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/notification-outbox-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("notification-outbox")); - var outboxProxy = _actorSystem.ActorOf(outboxProxyProps, "notification-outbox-proxy"); + _logger); // Hand the outbox proxy to the CentralCommunicationActor so forwarded // NotificationSubmit messages from sites are routed to the outbox singleton. - centralCommActor.Tell(new RegisterNotificationOutbox(outboxProxy)); + centralCommActor.Tell(new RegisterNotificationOutbox(outbox.Proxy)); // Hand the same proxy to the CommunicationService so the Central UI can // Ask the outbox actor directly (query, retry, discard, KPIs). - commService?.SetNotificationOutbox(outboxProxy); + commService?.SetNotificationOutbox(outbox.Proxy); _logger.LogInformation("NotificationOutbox singleton created and registered with CentralCommunicationActor"); // Audit Log — central singleton mirrors the Notification Outbox @@ -455,25 +454,17 @@ akka {{ var auditIngestLogger = _serviceProvider.GetRequiredService() .CreateLogger(); - var auditIngestSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor( + var auditIngest = CentralSingletonRegistrar.Start( + _actorSystem!, "audit-log-ingest", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor( _serviceProvider, auditIngestLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("audit-log-ingest")); - _actorSystem!.ActorOf(auditIngestSingletonProps, "audit-log-ingest-singleton"); - - var auditIngestProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/audit-log-ingest-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("audit-log-ingest")); - var auditIngestProxy = _actorSystem.ActorOf(auditIngestProxyProps, "audit-log-ingest-proxy"); + _logger); // Hand the audit-ingest proxy to the CentralCommunicationActor so audit // ingest commands forwarded by sites over ClusterClient are routed to the // singleton. Mirrors the RegisterNotificationOutbox wiring above. - centralCommActor.Tell(new RegisterAuditIngest(auditIngestProxy)); + centralCommActor.Tell(new RegisterAuditIngest(auditIngest.Proxy)); // Hand the proxy to the SiteStreamGrpcServer (if registered on this node) // so the IngestAuditEvents RPC routes incoming site batches to the singleton. @@ -481,7 +472,7 @@ akka {{ // node this resolves to null and the wiring is a no-op unless a future // central-hosted gRPC server (or a real site→central client) is added. var grpcServer = _serviceProvider.GetService(); - grpcServer?.SetAuditIngestActor(auditIngestProxy); + grpcServer?.SetAuditIngestActor(auditIngest.Proxy); _logger.LogInformation( "AuditLogIngestActor singleton created (gRPC server bound: {GrpcBound})", grpcServer is not null); @@ -528,61 +519,18 @@ akka {{ var siteCallAuditOptions = _serviceProvider .GetRequiredService>().Value; - var siteCallAuditSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor( + var siteCallAudit = CentralSingletonRegistrar.Start( + _actorSystem!, "site-call-audit", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor( _serviceProvider, siteCallAuditOptions, siteCallAuditLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("site-call-audit")); - var siteCallAuditSingletonManager = - _actorSystem!.ActorOf(siteCallAuditSingletonProps, "site-call-audit-singleton"); - - // Graceful-handover hook. The default singleton handover - // path waits for the actor's `ReceiveAsync` task to complete before - // signalling `HandOverDone` to the new oldest node — so an in-flight - // EF `UpsertAsync` IS waited for during a *clean* coordinated shutdown - // (the cluster-leave phase below fires before the singleton terminates). - // The risk here is the seam between in-flight async work - // and the cluster-leave + singleton-stop sequence: we bound it by - // issuing an explicit `GracefulStop` to the singleton manager early - // in `cluster-leave`, with a timeout that lets the running upsert + SQL - // round-trip drain before the handover-to-other-node race window - // opens. The timeout is bounded so a misbehaving upsert cannot stall - // coordinated shutdown indefinitely — exceeding it falls through to - // the existing PoisonPill termination path. Same pattern is suitable - // for the NotificationOutbox singleton; not added here to keep this - // change minimal. - var siteCallAuditShutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); - siteCallAuditShutdown.AddTask( - Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, - "drain-site-call-audit-singleton", - async () => - { - try - { - await siteCallAuditSingletonManager.GracefulStop(TimeSpan.FromSeconds(10)); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "SiteCallAudit singleton did not drain within the graceful-stop " - + "timeout; falling through to PoisonPill handover"); - } - return Akka.Done.Instance; - }); - - var siteCallAuditProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/site-call-audit-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("site-call-audit")); - var siteCallAuditProxy = _actorSystem.ActorOf(siteCallAuditProxyProps, "site-call-audit-proxy"); + _logger); // Hand the proxy to the CommunicationService so the Central UI can Ask // the Site Call Audit actor directly (query, KPIs, detail) — mirrors the // SetNotificationOutbox wiring above. - commService?.SetSiteCallAudit(siteCallAuditProxy); + commService?.SetSiteCallAudit(siteCallAudit.Proxy); // Hand the CentralCommunicationActor to the SiteCallAudit // actor so it can relay operator Retry/Discard on parked cached calls to @@ -590,7 +538,7 @@ akka {{ // Mirrors the RegisterAuditIngest / RegisterNotificationOutbox wiring; // the message is sent to the singleton proxy so it reaches whichever // central node currently hosts the singleton. - siteCallAuditProxy.Tell( + siteCallAudit.Proxy.Tell( new ZB.MOM.WW.ScadaBridge.SiteCallAudit.RegisterCentralCommunication(centralCommActor)); _logger.LogInformation( "SiteCallAuditActor singleton created and registered with CentralCommunicationActor"); @@ -616,42 +564,14 @@ akka {{ var auditLogOptions = _serviceProvider .GetRequiredService>(); - var auditPurgeSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeActor( + CentralSingletonRegistrar.Start( + _actorSystem!, "audit-log-purge", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeActor( _serviceProvider, auditPurgeOptions, auditLogOptions, auditPurgeLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("audit-log-purge")); - var auditPurgeSingletonManager = - _actorSystem!.ActorOf(auditPurgeSingletonProps, "audit-log-purge-singleton"); - - var auditPurgeShutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); - auditPurgeShutdown.AddTask( - Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, - "drain-audit-log-purge-singleton", - async () => - { - try - { - await auditPurgeSingletonManager.GracefulStop(TimeSpan.FromSeconds(10)); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "AuditLogPurge singleton did not drain within the graceful-stop " - + "timeout; falling through to PoisonPill handover"); - } - return Akka.Done.Instance; - }); - - var auditPurgeProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/audit-log-purge-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("audit-log-purge")); - _actorSystem.ActorOf(auditPurgeProxyProps, "audit-log-purge-proxy"); + _logger); _logger.LogInformation("AuditLogPurgeActor singleton created"); // SiteAuditReconciliationActor — self-healing fallback puller. Resolves @@ -667,43 +587,15 @@ akka {{ var auditReconClient = _serviceProvider .GetRequiredService(); - var auditReconSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationActor( + CentralSingletonRegistrar.Start( + _actorSystem!, "site-audit-reconciliation", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationActor( auditReconSites, auditReconClient, _serviceProvider, auditReconOptions, auditReconLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("site-audit-reconciliation")); - var auditReconSingletonManager = - _actorSystem!.ActorOf(auditReconSingletonProps, "site-audit-reconciliation-singleton"); - - var auditReconShutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); - auditReconShutdown.AddTask( - Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, - "drain-site-audit-reconciliation-singleton", - async () => - { - try - { - await auditReconSingletonManager.GracefulStop(TimeSpan.FromSeconds(10)); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "SiteAuditReconciliation singleton did not drain within the graceful-stop " - + "timeout; falling through to PoisonPill handover"); - } - return Akka.Done.Instance; - }); - - var auditReconProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/site-audit-reconciliation-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("site-audit-reconciliation")); - _actorSystem.ActorOf(auditReconProxyProps, "site-audit-reconciliation-proxy"); + _logger); _logger.LogInformation("SiteAuditReconciliationActor singleton created"); // KPI History — central singleton that periodically samples the @@ -726,41 +618,13 @@ akka {{ var kpiHistoryLogger = _serviceProvider.GetRequiredService() .CreateLogger(); - var kpiHistorySingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor( + CentralSingletonRegistrar.Start( + _actorSystem!, "kpi-history-recorder", + Props.Create(() => new ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor( _serviceProvider, kpiHistoryOptions, kpiHistoryLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("kpi-history-recorder")); - var kpiHistorySingletonManager = - _actorSystem!.ActorOf(kpiHistorySingletonProps, "kpi-history-recorder-singleton"); - - var kpiHistoryShutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); - kpiHistoryShutdown.AddTask( - Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, - "drain-kpi-history-recorder-singleton", - async () => - { - try - { - await kpiHistorySingletonManager.GracefulStop(TimeSpan.FromSeconds(10)); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "KpiHistoryRecorder singleton did not drain within the graceful-stop " - + "timeout; falling through to PoisonPill handover"); - } - return Akka.Done.Instance; - }); - - var kpiHistoryProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/kpi-history-recorder-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("kpi-history-recorder")); - _actorSystem.ActorOf(kpiHistoryProxyProps, "kpi-history-recorder-proxy"); + _logger); _logger.LogInformation("KpiHistoryRecorderActor singleton created (not readiness-gated)"); // Notify-and-fetch deploy transport — central singleton that periodically @@ -780,41 +644,13 @@ akka {{ var pendingPurgeCommunicationOptions = _serviceProvider.GetRequiredService>(); - var pendingPurgeSingletonProps = ClusterSingletonManager.Props( - singletonProps: Props.Create(() => new PendingDeploymentPurgeActor( + CentralSingletonRegistrar.Start( + _actorSystem!, "pending-deployment-purge", + Props.Create(() => new PendingDeploymentPurgeActor( _serviceProvider, pendingPurgeCommunicationOptions, pendingPurgeLogger)), - terminationMessage: PoisonPill.Instance, - settings: ClusterSingletonManagerSettings.Create(_actorSystem!) - .WithSingletonName("pending-deployment-purge")); - var pendingPurgeSingletonManager = - _actorSystem!.ActorOf(pendingPurgeSingletonProps, "pending-deployment-purge-singleton"); - - var pendingPurgeShutdown = Akka.Actor.CoordinatedShutdown.Get(_actorSystem); - pendingPurgeShutdown.AddTask( - Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, - "drain-pending-deployment-purge-singleton", - async () => - { - try - { - await pendingPurgeSingletonManager.GracefulStop(TimeSpan.FromSeconds(10)); - } - catch (Exception ex) - { - _logger.LogWarning(ex, - "PendingDeploymentPurge singleton did not drain within the graceful-stop " - + "timeout; falling through to PoisonPill handover"); - } - return Akka.Done.Instance; - }); - - var pendingPurgeProxyProps = ClusterSingletonProxy.Props( - singletonManagerPath: "/user/pending-deployment-purge-singleton", - settings: ClusterSingletonProxySettings.Create(_actorSystem) - .WithSingletonName("pending-deployment-purge")); - _actorSystem.ActorOf(pendingPurgeProxyProps, "pending-deployment-purge-proxy"); + _logger); _logger.LogInformation("PendingDeploymentPurgeActor singleton created (not readiness-gated)"); _logger.LogInformation("Central actors registered. CentralCommunicationActor created."); diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs index a6320bff..b5943cfe 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs @@ -42,6 +42,43 @@ public class CoordinatedShutdownTests Assert.Contains("run-coordinated-shutdown-when-down = on", content); } + [Fact] + public void AllCentralSingletons_RegisterThroughRegistrarWithDrain() + { + var hostProjectDir = FindHostProjectDirectory(); + Assert.NotNull(hostProjectDir); + var content = File.ReadAllText(Path.Combine(hostProjectDir!, "Actors", "AkkaHostedService.cs")); + + // Review 01 [Medium]: notification-outbox and audit-log-ingest were the + // only central singletons WITHOUT a cluster-leave drain task. All seven + // now go through CentralSingletonRegistrar.Start, which always adds the + // PhaseClusterLeave GracefulStop drain. + Assert.Contains("CentralSingletonRegistrar.Start(", content); + foreach (var name in new[] { "notification-outbox", "audit-log-ingest", "site-call-audit", + "audit-log-purge", "site-audit-reconciliation", + "kpi-history-recorder", "pending-deployment-purge" }) + { + Assert.Contains($"\"{name}\"", content); + } + + // No hand-rolled ClusterSingletonManager registrations remain in the + // CENTRAL branch. The only two left are the SITE singletons + // (deployment-manager, event-log-handler), which stay hand-rolled because + // they are role-scoped (.WithRole(siteRole)). + Assert.Equal(2, CountOccurrences(content, "ClusterSingletonManager.Props(")); + } + + private static int CountOccurrences(string haystack, string needle) + { + int count = 0, i = 0; + while ((i = haystack.IndexOf(needle, i, StringComparison.Ordinal)) >= 0) + { + count++; + i += needle.Length; + } + return count; + } + private static string? FindHostProjectDirectory() { var assemblyDir = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location)!;