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

refactor(host): CentralSingletonRegistrar -> SingletonRegistrar with optional role scope — unblocks site-singleton drains (plan R2-01 T10)

Browse Source
This commit is contained in:
Joseph Doherty
2026-07-13 10:49:49 -04:00
parent c2707fdcf8
commit 80e5b36852
4 changed files with 55 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.Host/Actors/AkkaHostedService.cs
+10 -10
View File
@@ -456,14 +456,14 @@ akka {{
var outboxAuditWriter = _serviceProvider
.GetRequiredService<ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services.ICentralAuditWriter>();
// All central singletons below are created through CentralSingletonRegistrar.Start,
// All central singletons below are created through SingletonRegistrar.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(
// are unchanged ({name}-singleton / {name}-proxy). Site singletons use the same
// registrar with `role: siteRole` (Task 11 / round-2 N5).
var outbox = SingletonRegistrar.Start(
_actorSystem!, "notification-outbox",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.NotificationOutbox.NotificationOutboxActor(
_serviceProvider,
@@ -496,7 +496,7 @@ akka {{
var auditIngestLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
.CreateLogger<ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor>();
var auditIngest = CentralSingletonRegistrar.Start(
var auditIngest = SingletonRegistrar.Start(
_actorSystem!, "audit-log-ingest",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogIngestActor(
_serviceProvider,
@@ -561,7 +561,7 @@ akka {{
var siteCallAuditOptions = _serviceProvider
.GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditOptions>>().Value;
var siteCallAudit = CentralSingletonRegistrar.Start(
var siteCallAudit = SingletonRegistrar.Start(
_actorSystem!, "site-call-audit",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.SiteCallAudit.SiteCallAuditActor(
_serviceProvider,
@@ -606,7 +606,7 @@ akka {{
var auditLogOptions = _serviceProvider
.GetRequiredService<IOptions<ZB.MOM.WW.ScadaBridge.AuditLog.Configuration.AuditLogOptions>>();
CentralSingletonRegistrar.Start(
SingletonRegistrar.Start(
_actorSystem!, "audit-log-purge",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.AuditLogPurgeActor(
_serviceProvider,
@@ -629,7 +629,7 @@ akka {{
var auditReconClient = _serviceProvider
.GetRequiredService<ZB.MOM.WW.ScadaBridge.AuditLog.Central.IPullAuditEventsClient>();
CentralSingletonRegistrar.Start(
SingletonRegistrar.Start(
_actorSystem!, "site-audit-reconciliation",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.AuditLog.Central.SiteAuditReconciliationActor(
auditReconSites,
@@ -660,7 +660,7 @@ akka {{
var kpiHistoryLogger = _serviceProvider.GetRequiredService<ILoggerFactory>()
.CreateLogger<ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor>();
CentralSingletonRegistrar.Start(
SingletonRegistrar.Start(
_actorSystem!, "kpi-history-recorder",
Props.Create(() => new ZB.MOM.WW.ScadaBridge.KpiHistory.KpiHistoryRecorderActor(
_serviceProvider,
@@ -686,7 +686,7 @@ akka {{
var pendingPurgeCommunicationOptions =
_serviceProvider.GetRequiredService<IOptions<CommunicationOptions>>();
CentralSingletonRegistrar.Start(
SingletonRegistrar.Start(
_actorSystem!, "pending-deployment-purge",
Props.Create(() => new PendingDeploymentPurgeActor(
_serviceProvider,
src/ZB.MOM.WW.ScadaBridge.Host/Actors/CentralSingletonRegistrar.cs → src/ZB.MOM.WW.ScadaBridge.Host/Actors/SingletonRegistrar.cs
+22 -11
View File
@@ -5,21 +5,24 @@ using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.ScadaBridge.Host.Actors;
/// <summary>
/// Registers a central cluster singleton with the canonical naming scheme
/// Registers a cluster singleton with the canonical naming scheme
/// (<c>{name}-singleton</c> / <c>{name}-proxy</c>), a PoisonPill termination
/// message, and a PhaseClusterLeave drain task that GracefulStops the manager
/// so in-flight EF work completes before handover. Extracted from five
/// copy-pasted ~60-line blocks (review 01 [Low]) whose drift left the two
/// busiest singletons (notification-outbox, audit-log-ingest) without drain
/// tasks (review 01 [Medium]).
/// so in-flight EF (central) or SQLite (site) work completes before handover.
/// Extracted from copy-pasted ~60-line blocks (review 01 [Low]) whose drift
/// left the two busiest singletons (notification-outbox, audit-log-ingest)
/// without drain tasks (review 01 [Medium]). The optional <c>role</c> maps to
/// <see cref="ClusterSingletonManagerSettings"/>/<see cref="ClusterSingletonProxySettings"/>
/// <c>.WithRole</c>, so role-scoped site singletons get the same drain
/// (review 01 round-2 N5).
/// </summary>
internal static class CentralSingletonRegistrar
internal static class SingletonRegistrar
{
internal sealed record Handle(IActorRef Manager, IActorRef Proxy);
/// <summary>
/// Creates the <see cref="ClusterSingletonManager"/> and proxy for a central
/// cluster singleton under the canonical naming scheme, and registers a
/// Creates the <see cref="ClusterSingletonManager"/> and proxy for a cluster
/// singleton under the canonical naming scheme, and registers a
/// PhaseClusterLeave drain task that GracefulStops the manager on shutdown.
/// </summary>
/// <param name="system">The actor system to register the singleton on.</param>
@@ -27,19 +30,24 @@ internal static class CentralSingletonRegistrar
/// <param name="singletonProps">The <see cref="Props"/> used to create the singleton's managed actor.</param>
/// <param name="logger">Logger used to report drain progress/failures.</param>
/// <param name="drainTimeout">Maximum time to wait for the drain task to complete; defaults to 10 seconds.</param>
/// <param name="role">Optional cluster role to scope the singleton to (applied to both the manager and proxy settings).</param>
/// <returns>A <see cref="Handle"/> carrying the singleton manager and proxy actor refs.</returns>
internal static Handle Start(
ActorSystem system,
string name,
Props singletonProps,
ILogger logger,
TimeSpan? drainTimeout = null)
TimeSpan? drainTimeout = null,
string? role = null)
{
var managerSettings = ClusterSingletonManagerSettings.Create(system).WithSingletonName(name);
if (role is not null) managerSettings = managerSettings.WithRole(role);
var manager = system.ActorOf(
ClusterSingletonManager.Props(
singletonProps,
PoisonPill.Instance,
ClusterSingletonManagerSettings.Create(system).WithSingletonName(name)),
managerSettings),
$"{name}-singleton");
var timeout = drainTimeout ?? TimeSpan.FromSeconds(10);
@@ -61,10 +69,13 @@ internal static class CentralSingletonRegistrar
return Akka.Done.Instance;
});
var proxySettings = ClusterSingletonProxySettings.Create(system).WithSingletonName(name);
if (role is not null) proxySettings = proxySettings.WithRole(role);
var proxy = system.ActorOf(
ClusterSingletonProxy.Props(
$"/user/{name}-singleton",
ClusterSingletonProxySettings.Create(system).WithSingletonName(name)),
proxySettings),
$"{name}-proxy");
return new Handle(manager, proxy);
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CoordinatedShutdownTests.cs
+2 -2
View File
@@ -51,9 +51,9 @@ public class CoordinatedShutdownTests
// 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
// now go through SingletonRegistrar.Start, which always adds the
// PhaseClusterLeave GracefulStop drain.
Assert.Contains("CentralSingletonRegistrar.Start(", content);
Assert.Contains("SingletonRegistrar.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" })
tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralSingletonRegistrarTests.cs → tests/ZB.MOM.WW.ScadaBridge.Host.Tests/SingletonRegistrarTests.cs
+21 -2
View File
@@ -5,7 +5,7 @@ using ZB.MOM.WW.ScadaBridge.Host.Actors;
namespace ZB.MOM.WW.ScadaBridge.Host.Tests;
public class CentralSingletonRegistrarTests
public class SingletonRegistrarTests
{
private sealed class EchoActor : ReceiveActor
{
@@ -16,7 +16,7 @@ public class CentralSingletonRegistrarTests
public async Task Start_CreatesManagerAndProxy_WithCanonicalNames_AndDrainTask()
{
using var system = CreateSingleNodeClusterSystem();
var handle = CentralSingletonRegistrar.Start(
var handle = SingletonRegistrar.Start(
system, "test-widget", Props.Create(() => new EchoActor()),
NullLogger.Instance, drainTimeout: TimeSpan.FromSeconds(2));
@@ -36,6 +36,25 @@ public class CentralSingletonRegistrarTests
Assert.True(handle.Manager.IsNobody() || system.WhenTerminated.IsCompleted);
}
[Fact]
public async Task Start_WithRoleScope_CreatesRoleScopedSingleton_ThatAnswers()
{
// Round-2 N5: role scoping is what kept the two SITE singletons
// (deployment-manager, event-log-handler) hand-rolled and drain-less.
using var system = CreateSingleNodeClusterSystem(); // roles = ["Central"]
var handle = SingletonRegistrar.Start(
system, "role-widget", Props.Create(() => new EchoActor()),
NullLogger.Instance, drainTimeout: TimeSpan.FromSeconds(2), role: "Central");
Assert.EndsWith("/user/role-widget-singleton", handle.Manager.Path.ToString());
Assert.EndsWith("/user/role-widget-proxy", handle.Proxy.Path.ToString());
// The node holds the role => manager instantiates the singleton and
// the role-scoped proxy resolves it.
var echo = await handle.Proxy.Ask<string>("hi", TimeSpan.FromSeconds(20));
Assert.Equal("hi", echo);
}
private static ActorSystem CreateSingleNodeClusterSystem()
{
var port = FreePort();