feat(host): CentralSingletonRegistrar — canonical singleton manager+proxy+drain helper
This commit is contained in:
@@ -0,0 +1,61 @@
|
|||||||
|
using Akka.Actor;
|
||||||
|
using Akka.Cluster.Tools.Singleton;
|
||||||
|
using Microsoft.Extensions.Logging;
|
||||||
|
|
||||||
|
namespace ZB.MOM.WW.ScadaBridge.Host.Actors;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Registers a central 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]).
|
||||||
|
/// </summary>
|
||||||
|
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);
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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: <name>-singleton / <name>-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<string>("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;
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user