From c255ec31c9eec1a22a645e1af9f830d66b360551 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 8 Jul 2026 15:59:01 -0400 Subject: [PATCH] =?UTF-8?q?feat(host):=20CentralSingletonRegistrar=20?= =?UTF-8?q?=E2=80=94=20canonical=20singleton=20manager+proxy+drain=20helpe?= =?UTF-8?q?r?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../Actors/CentralSingletonRegistrar.cs | 61 +++++++++++++++++++ .../CentralSingletonRegistrarTests.cs | 60 ++++++++++++++++++ 2 files changed, 121 insertions(+) create mode 100644 src/ZB.MOM.WW.ScadaBridge.Host/Actors/CentralSingletonRegistrar.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralSingletonRegistrarTests.cs diff --git a/src/ZB.MOM.WW.ScadaBridge.Host/Actors/CentralSingletonRegistrar.cs b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/CentralSingletonRegistrar.cs new file mode 100644 index 00000000..8c24dce1 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.Host/Actors/CentralSingletonRegistrar.cs @@ -0,0 +1,61 @@ +using Akka.Actor; +using Akka.Cluster.Tools.Singleton; +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.ScadaBridge.Host.Actors; + +/// +/// Registers a central cluster singleton with the canonical naming scheme +/// ({name}-singleton / {name}-proxy), 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]). +/// +internal static class CentralSingletonRegistrar +{ + internal sealed record Handle(IActorRef Manager, IActorRef Proxy); + + internal static Handle Start( + ActorSystem system, + string name, + Props singletonProps, + ILogger logger, + TimeSpan? drainTimeout = null) + { + var manager = system.ActorOf( + ClusterSingletonManager.Props( + singletonProps, + PoisonPill.Instance, + ClusterSingletonManagerSettings.Create(system).WithSingletonName(name)), + $"{name}-singleton"); + + var timeout = drainTimeout ?? TimeSpan.FromSeconds(10); + Akka.Actor.CoordinatedShutdown.Get(system).AddTask( + Akka.Actor.CoordinatedShutdown.PhaseClusterLeave, + $"drain-{name}-singleton", + async () => + { + try + { + await manager.GracefulStop(timeout); + } + catch (Exception ex) + { + logger.LogWarning(ex, + "{Singleton} singleton did not drain within the graceful-stop timeout; " + + "falling through to PoisonPill handover", name); + } + return Akka.Done.Instance; + }); + + var proxy = system.ActorOf( + ClusterSingletonProxy.Props( + $"/user/{name}-singleton", + ClusterSingletonProxySettings.Create(system).WithSingletonName(name)), + $"{name}-proxy"); + + return new Handle(manager, proxy); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralSingletonRegistrarTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralSingletonRegistrarTests.cs new file mode 100644 index 00000000..21ceb8be --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.Host.Tests/CentralSingletonRegistrarTests.cs @@ -0,0 +1,60 @@ +using Akka.Actor; +using Akka.Configuration; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.ScadaBridge.Host.Actors; + +namespace ZB.MOM.WW.ScadaBridge.Host.Tests; + +public class CentralSingletonRegistrarTests +{ + private sealed class EchoActor : ReceiveActor + { + public EchoActor() => ReceiveAny(m => Sender.Tell(m)); + } + + [Fact] + public async Task Start_CreatesManagerAndProxy_WithCanonicalNames_AndDrainTask() + { + using var system = CreateSingleNodeClusterSystem(); + var handle = CentralSingletonRegistrar.Start( + system, "test-widget", Props.Create(() => new EchoActor()), + NullLogger.Instance, drainTimeout: TimeSpan.FromSeconds(2)); + + // Canonical naming preserved: -singleton / -proxy. + Assert.EndsWith("/user/test-widget-singleton", handle.Manager.Path.ToString()); + Assert.EndsWith("/user/test-widget-proxy", handle.Proxy.Path.ToString()); + + // Singleton answers through the proxy once the member is Up. + var echo = await handle.Proxy.Ask("hi", TimeSpan.FromSeconds(20)); + Assert.Equal("hi", echo); + + // The drain task registered on PhaseClusterLeave must stop the manager + // during coordinated shutdown. + await Akka.Actor.CoordinatedShutdown.Get(system).Run( + Akka.Actor.CoordinatedShutdown.ClrExitReason.Instance); + // Manager terminated => drain ran (GracefulStop completed or PoisonPill fallback). + Assert.True(handle.Manager.IsNobody() || system.WhenTerminated.IsCompleted); + } + + private static ActorSystem CreateSingleNodeClusterSystem() + { + var port = FreePort(); + var config = ConfigurationFactory.ParseString($@" +akka {{ + actor.provider = cluster + remote.dot-netty.tcp {{ hostname = ""127.0.0.1"", port = {port} }} + cluster {{ + seed-nodes = [""akka.tcp://registrar-test@127.0.0.1:{port}""] + roles = [""Central""] + min-nr-of-members = 1 + }} +}}"); + return ActorSystem.Create("registrar-test", config); + } + + private static int FreePort() + { + var l = new System.Net.Sockets.TcpListener(System.Net.IPAddress.Loopback, 0); + l.Start(); var p = ((System.Net.IPEndPoint)l.LocalEndpoint).Port; l.Stop(); return p; + } +}