dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(host): central singletons via CentralSingletonRegistrar — outbox/audit-ingest gain drain tasks

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-08 16:05:52 -04:00
parent c255ec31c9
commit f7b9d342e4
2 changed files with 78 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs
+41 -205
View File
@@ -414,30 +414,29 @@ akka {{
var outboxAuditWriter = _serviceProvider
.GetRequiredService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter>();
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<ILoggerFactory>()
.CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>();
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<ZB.MOM.WW.ScadaBridge.Communication.Grpc.SiteStreamGrpcServer>();
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<IOptions<ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditOptions>>().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<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Configuration.AuditLogOptions>>();
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<ZB.MOM.WW.ScadaBridge.AuditLog.Central.IPullAuditEventsClient>();
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<ILoggerFactory>()
.CreateLogger<ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor>();
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<IOptions<CommunicationOptions>>();
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.");
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs
+37
View File
@@ -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)!;