diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs index 01143dd4d..c96b682a9 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -52,8 +52,20 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; /// ServerCluster.Enabled is not consulted — nothing else in the codebase consults it, /// so honouring it here would invent semantics. Disable the nodes, not the cluster row. /// +/// +/// Recovery load is asynchronous and retried (lmxopcua#868). The in-flight recovery read used to run +/// synchronously in PreStart. When the ConfigDb was unreachable as the singleton started, the throw +/// became an ActorInitializationException, the default strategy STOPPED the actor, and the +/// ClusterSingletonManager never re-creates a singleton that terminated on its own — so every later +/// deployment was told to a dead singleton and hung. The read now runs off the actor thread, a failure +/// retries with exponential backoff ( doubling to +/// , forever), and until it succeeds every message — deployments and acks +/// alike — is STASHED and replayed in order once the recovered state is in place, so a deploy arriving +/// during the outage is queued, never dropped. reports the +/// loading state (and the last error) to the node's health endpoint as Degraded. +/// /// -public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers +public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithUnboundedStash { public const string DeploymentsTopic = "deployments"; public const string DeploymentAcksTopic = "deployment-acks"; @@ -61,8 +73,18 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// apply-outcome timeout is derived from. public static readonly TimeSpan DefaultApplyDeadline = DeployTimings.ApplyDeadline; + /// First retry delay after a failed recovery load (lmxopcua#868). + public static readonly TimeSpan DefaultLoadRetryMin = TimeSpan.FromSeconds(1); + + /// Ceiling of the recovery-load retry backoff (lmxopcua#868). + public static readonly TimeSpan DefaultLoadRetryMax = TimeSpan.FromSeconds(30); + private readonly IDbContextFactory _dbFactory; private readonly TimeSpan _applyDeadline; + private readonly TimeSpan _loadRetryMin; + private readonly TimeSpan _loadRetryMax; + private readonly ConfigPublishCoordinatorReadiness? _readiness; + private int _failedLoads; /// /// Where central→node commands are handed off (per-cluster mesh Phase 2), or @@ -115,20 +137,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// Gets the timer scheduler for managing apply deadlines. public ITimerScheduler Timers { get; set; } = null!; + /// Holds every message that arrives before the recovery load completes (lmxopcua#868). + public IStash Stash { get; set; } = null!; + /// Creates actor props for the ConfigPublishCoordinator with the specified configuration. /// The database context factory for accessing configuration data. /// The timeout for waiting for apply acknowledgments from cluster nodes. - /// The Akka.NET used to spawn this actor. /// /// Central comm actor the dispatch is handed to, or to publish on the /// DistributedPubSub mediator directly (the original behaviour). /// + /// Where the recovery-load state is reported for health, or . + /// First recovery-load retry delay; = . + /// Recovery-load retry ceiling; = . + /// The Akka.NET used to spawn this actor. public static Props Props( IDbContextFactory dbFactory, TimeSpan? applyDeadline = null, - IActorRef? meshRouter = null) => - Akka.Actor.Props.Create(() => - new ConfigPublishCoordinator(dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter)); + IActorRef? meshRouter = null, + ConfigPublishCoordinatorReadiness? readiness = null, + TimeSpan? loadRetryMin = null, + TimeSpan? loadRetryMax = null) => + Akka.Actor.Props.Create(() => new ConfigPublishCoordinator( + dbFactory, applyDeadline ?? DefaultApplyDeadline, meshRouter, readiness, + loadRetryMin ?? DefaultLoadRetryMin, loadRetryMax ?? DefaultLoadRetryMax)); /// /// Initializes a new instance of the class. @@ -136,45 +168,167 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers /// The database context factory for persisting deployment state. /// The timeout for waiting for per-node apply acknowledgments. /// Central comm actor, or for the mediator. + /// Where the recovery-load state is reported, or . + /// First recovery-load retry delay. + /// Recovery-load retry ceiling. public ConfigPublishCoordinator( IDbContextFactory dbFactory, TimeSpan applyDeadline, - IActorRef? meshRouter = null) + IActorRef? meshRouter = null, + ConfigPublishCoordinatorReadiness? readiness = null, + TimeSpan? loadRetryMin = null, + TimeSpan? loadRetryMax = null) { _dbFactory = dbFactory; _applyDeadline = applyDeadline; _meshRouter = meshRouter; + _readiness = readiness; + _loadRetryMin = loadRetryMin ?? DefaultLoadRetryMin; + _loadRetryMax = loadRetryMax ?? DefaultLoadRetryMax; + Loading(); + } + + /// + /// The state before the recovery load has succeeded: only the load's own outcome and retry tick are + /// handled; everything else is stashed and replayed, in arrival order, once is in force. + /// + private void Loading() + { + Receive(HandleRecoveryLoaded); + Receive(HandleRecoveryLoadFailed); + Receive(_ => StartRecoveryLoad()); + Receive(_ => { /* DPS subscribe confirmation */ }); + Receive(msg => + { + _log.Info( + "DispatchDeployment {Id} queued: the coordinator has not yet loaded its recovery state from the " + + "ConfigDb ({Failed} failed attempt(s) so far); it is dispatched as soon as the load succeeds", + msg.DeploymentId, _failedLoads); + Stash.Stash(); + }); + ReceiveAny(_ => Stash.Stash()); + } + + /// The normal operating state, entered once the recovery load has been applied. + private void Ready() + { Receive(HandleDispatch); Receive(HandleAck); Receive(HandleDeadline); Receive(_ => { /* DPS subscribe confirmation */ }); + // A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed. + Receive(_ => { }); + Receive(_ => { }); + Receive(_ => { }); } /// protected override void PreStart() { - // On startup recover any deployment that was mid-flight when a prior singleton instance - // died. We re-derive _expectedAcks from NodeDeploymentState, replay the ACKs - // that already landed in the DB, and resume the deadline timer. // Subscribe to per-node ApplyAck broadcasts so DriverHostActors on remote members can // route their ACKs to whichever node currently hosts this singleton. DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self)); - using var db = _dbFactory.CreateDbContext(); + // lmxopcua#868: NO database access here. A throw from PreStart stops the actor for good (the + // singleton manager does not re-create it), so the recovery read runs asynchronously and retries. + _readiness?.MarkLoading(_failedLoads, lastError: null); + StartRecoveryLoad(); + } + + /// + protected override void PostStop() + { + // A restart re-enters PreStart (Loading again); a stop or hand-over leaves this node not hosting it. + _readiness?.MarkNotHosted(); + base.PostStop(); + } + + /// + /// Reads the deployment that was mid-flight when a prior singleton instance died, off the actor thread, + /// and pipes the outcome back. Touches no actor state. + /// + private void StartRecoveryLoad() + { + LoadRecoveryAsync(_dbFactory).PipeTo( + Self, + success: loaded => loaded, + failure: ex => new RecoveryLoadFailed(ex is AggregateException { InnerException: { } inner } ? inner : ex)); + } + + private static async Task LoadRecoveryAsync(IDbContextFactory dbFactory) + { + // Yield first so a factory or provider that throws synchronously still faults the TASK (piped back as + // RecoveryLoadFailed) rather than the caller on the actor thread. + await Task.Yield(); + await using var db = await dbFactory.CreateDbContextAsync().ConfigureAwait(false); // Projected: the row carries the whole ArtifactBlob (varbinary(max)), which this recovery never // reads — and this query runs on every singleton start and hand-over. - var inflight = db.Deployments + var inflight = await db.Deployments .Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks) .OrderByDescending(d => d.CreatedAtUtc) .Select(d => new { d.DeploymentId, d.RevisionHash, d.CreatedAtUtc }) - .FirstOrDefault(); + .FirstOrDefaultAsync() + .ConfigureAwait(false); + if (inflight is null) return new RecoveryLoaded(null); + + var nodeStates = await db.NodeDeploymentStates + .Where(x => x.DeploymentId == inflight.DeploymentId) + .AsNoTracking() + .Select(x => new RecoveredNodeState(x.NodeId, x.Status, x.FailureReason)) + .ToListAsync() + .ConfigureAwait(false); + return new RecoveryLoaded(new RecoveredDeployment( + inflight.DeploymentId, inflight.RevisionHash, inflight.CreatedAtUtc, nodeStates)); + } + + private void HandleRecoveryLoadFailed(RecoveryLoadFailed msg) + { + _failedLoads++; + // Exponential backoff, capped: min * 2^(n-1), with the shift bounded so it never overflows. + var factor = 1L << Math.Min(_failedLoads - 1, 20); + var delay = TimeSpan.FromTicks(Math.Min(_loadRetryMax.Ticks, _loadRetryMin.Ticks * factor)); + _readiness?.MarkLoading(_failedLoads, msg.Error.Message); + _log.Warning(msg.Error, + "ConfigPublishCoordinator could not load its recovery state from the ConfigDb (attempt {Attempt}); " + + "retrying in {Delay}. Deployments and acks are queued until it succeeds", + _failedLoads, delay); + Timers.StartSingleTimer(LoadRetryTimerKey, RetryRecoveryLoad.Instance, delay); + } + + /// + /// Applies the recovered state, switches to and replays every stashed message. The + /// recovery DECISIONS (re-dispatch, time out) run inline, before the replay, so a queued message for the + /// same deployment is judged against the recovered state rather than racing a self-sent one. + /// + private void HandleRecoveryLoaded(RecoveryLoaded msg) + { + Timers.Cancel(LoadRetryTimerKey); + if (_failedLoads > 0) + _log.Info("ConfigPublishCoordinator loaded its recovery state after {Failed} failed attempt(s)", _failedLoads); + _failedLoads = 0; + Become(Ready); + _readiness?.MarkReady(); + try + { + ApplyRecovery(msg.Inflight); + } + finally + { + Stash.UnstashAll(); + } + } + + /// + /// Recovers the deployment that was mid-flight when a prior singleton instance died: re-derives + /// _expectedAcks from NodeDeploymentState, replays the acks that already landed, and resumes + /// the deadline timer. + /// + private void ApplyRecovery(RecoveredDeployment? inflight) + { if (inflight is null) return; _current = new DeploymentId(inflight.DeploymentId); - var nodeStates = db.NodeDeploymentStates - .Where(x => x.DeploymentId == inflight.DeploymentId) - .AsNoTracking() - .ToList(); + var nodeStates = inflight.NodeStates; // Resume the deadline timer using the remaining time. The deadline runs from when the // deployment was first marked AwaitingApplyAcks (Deployment.CreatedAtUtc is a close enough @@ -201,12 +355,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers "before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " + "re-dispatching with {Remaining} of the apply deadline left", _current, remaining); - Self.Tell(new DispatchDeployment( + HandleDispatch(new DispatchDeployment( _current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId())); } else { - Self.Tell(new DeadlineElapsed(_current.Value)); + HandleDeadline(new DeadlineElapsed(_current.Value)); } return; } @@ -222,18 +376,25 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers if (remaining <= TimeSpan.Zero) { - Self.Tell(new DeadlineElapsed(_current.Value)); - } - else - { - Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining); + HandleDeadline(new DeadlineElapsed(_current.Value)); + return; } + + Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(_current.Value), remaining); _log.Info("Coordinator recovered in-flight deployment {Id} ({Acked}/{Total} acks landed)", _current, _acks.Count, _expectedAcks.Count); } private void HandleDispatch(DispatchDeployment msg) { + // A duplicate of the deployment already in flight with a seeded expected set (a dispatch queued while the + // recovery load ran, for the very row that load re-drove) must not reset the acks already gathered. + if (_current == msg.DeploymentId && _expectedAcks.Count > 0) + { + _log.Debug("Ignoring duplicate DispatchDeployment {Id}: already in flight", msg.DeploymentId); + return; + } + // Terminal-state guard (fix round 1): a late or re-delivered dispatch for a row that is already // Sealed / PartiallyFailed / TimedOut — e.g. one AdminOps closed as a lost dispatch while this message // was still queued — must never flip it back to AwaitingApplyAcks or seed an expected set for it. @@ -519,9 +680,27 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers } private const string DeadlineTimerKey = "apply-deadline"; + private const string LoadRetryTimerKey = "recovery-load-retry"; public sealed record DeadlineElapsed(DeploymentId DeploymentId); + /// The recovery read's result: the newest in-flight deployment, or null when none. + private sealed record RecoveryLoaded(RecoveredDeployment? Inflight); + + /// The recovery read failed (ConfigDb unreachable, ...); retried with backoff. + private sealed record RecoveryLoadFailed(Exception Error); + + /// The backoff timer's tick: run the recovery read again. + private sealed class RetryRecoveryLoad + { + public static readonly RetryRecoveryLoad Instance = new(); + } + + private sealed record RecoveredDeployment( + Guid DeploymentId, string RevisionHash, DateTime CreatedAtUtc, IReadOnlyList NodeStates); + + private sealed record RecoveredNodeState(string NodeId, NodeDeploymentStatus Status, string? FailureReason); + /// /// Stages a status change WITHOUT materialising the row: the row carries the whole ArtifactBlob /// (varbinary(max), multi-MB on a real fleet), which no status transition reads (fix round 1). A stub diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs new file mode 100644 index 000000000..0074a8eee --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs @@ -0,0 +1,50 @@ +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; + +/// Where the singleton stands on this node (lmxopcua#868). +public enum ConfigPublishCoordinatorState +{ + /// The singleton is not running on this node (the partner hosts it, or it has not started yet). + NotHosted, + + /// Running here, but its recovery state has not loaded from the ConfigDb yet; deployments are queued. + Loading, + + /// Running here with its recovery state loaded; deployments are dispatched. + Ready, +} + +/// One immutable observation of the coordinator's state. +/// The state. +/// Consecutive failed recovery loads (0 unless ). +/// The last load failure's message, or null. +/// When the state was entered. +public sealed record ConfigPublishCoordinatorSnapshot( + ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc); + +/// +/// Thread-safe state bridge from the actor to the node's health endpoint +/// (one DI singleton per admin node). The actor writes; the health check reads. +/// +public sealed class ConfigPublishCoordinatorReadiness +{ + private volatile ConfigPublishCoordinatorSnapshot _current = + new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow); + + /// The latest observation. + public ConfigPublishCoordinatorSnapshot Current => _current; + + /// The coordinator is loading (or retrying the load of) its recovery state. + /// Failed attempts so far. + /// The last failure's message, or null. + public void MarkLoading(int failedAttempts, string? lastError) + { + var since = _current.State == ConfigPublishCoordinatorState.Loading ? _current.SinceUtc : DateTime.UtcNow; + _current = new(ConfigPublishCoordinatorState.Loading, failedAttempts, lastError, since); + } + + /// The recovery state is loaded. + public void MarkReady() => _current = new(ConfigPublishCoordinatorState.Ready, 0, null, DateTime.UtcNow); + + /// The singleton stopped on this node. + public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow); +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index 9a9f07423..048613739 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -112,7 +112,11 @@ public static class ServiceCollectionExtensions // coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks // inside a props factory and races the spawn it is waiting for. var meshRouter = registry.Get(); - return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter); + // lmxopcua#868: optional — through IServiceProvider for GetService's null-for-absent contract + // (the Akka resolver makes no such promise), so harnesses that never register it still boot. + var readiness = resolver.GetService()?.GetService(); + return ConfigPublishCoordinator.Props( + dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter, readiness: readiness); }, singletonOptions); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs new file mode 100644 index 000000000..579eb660d --- /dev/null +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs @@ -0,0 +1,45 @@ +using Microsoft.Extensions.Diagnostics.HealthChecks; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; + +namespace ZB.MOM.WW.OtOpcUa.Host.Health; + +/// +/// The deploy coordinator's readiness (lmxopcua#868): Degraded while the +/// singleton on this node has not yet loaded its recovery state from the +/// ConfigDb (deployments are queued, not dispatched), Healthy otherwise — including on the admin node that does +/// not host the singleton. Never Unhealthy: the node still serves (the configdb probe owns "DB down"). +/// +/// The coordinator's state bridge; null on graphs that never register one. +public sealed class ConfigPublishCoordinatorHealthCheck(ConfigPublishCoordinatorReadiness? readiness) : IHealthCheck +{ + /// The check's registration name. + public const string Name = "config-publish"; + + /// + public Task CheckHealthAsync(HealthCheckContext context, CancellationToken cancellationToken = default) + { + if (readiness is null) + return Task.FromResult(HealthCheckResult.Healthy("Deploy coordinator readiness is not tracked on this node.")); + + var s = readiness.Current; + var data = new Dictionary(StringComparer.Ordinal) + { + ["state"] = s.State.ToString(), + ["sinceUtc"] = s.SinceUtc.ToString("O"), + ["failedLoadAttempts"] = s.FailedLoadAttempts, + }; + if (s.LastError is not null) data["lastError"] = s.LastError; + + return Task.FromResult(s.State switch + { + ConfigPublishCoordinatorState.Loading => HealthCheckResult.Degraded( + s.FailedLoadAttempts == 0 + ? "Deploy coordinator is loading its recovery state from the ConfigDb; deployments are queued until it completes." + : $"Deploy coordinator cannot load its recovery state from the ConfigDb ({s.FailedLoadAttempts} failed attempt(s), " + + $"last: {s.LastError}); retrying with backoff. Deployments are queued until it succeeds.", + data: data), + ConfigPublishCoordinatorState.Ready => HealthCheckResult.Healthy("Deploy coordinator ready.", data), + _ => HealthCheckResult.Healthy("Deploy coordinator singleton is not hosted on this node.", data), + }); + } +} diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs index e7fe87e91..24fec2e01 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/HealthEndpoints.cs @@ -87,6 +87,15 @@ public static class HealthEndpoints { ProbeQuery = static (db, ct) => db.Deployments.AsNoTracking().Take(1).ToListAsync(ct), }); + + // lmxopcua#868: Degraded (still 200) while the deploy coordinator singleton on this node is retrying + // its ConfigDb recovery load and queueing deployments; Healthy otherwise. Ready tier only. + checks.Add(new HealthCheckRegistration( + ConfigPublishCoordinatorHealthCheck.Name, + sp => new ConfigPublishCoordinatorHealthCheck( + sp.GetService()), + failureStatus: null, + tags: new[] { ZbHealthTags.Ready })); } checks diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs index 132ead6f6..0d5d88ab6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Program.cs @@ -176,7 +176,12 @@ else builder.Services.Configure(builder.Configuration.GetSection(AclOptions.SectionName)); if (hasAdmin) +{ builder.Services.AddOtOpcUaConfigDb(builder.Configuration); + // lmxopcua#868: the deploy coordinator singleton reports its ConfigDb recovery-load state here; the + // config-publish health check (AddOtOpcUaHealth) reads it — Degraded while deployments are queued. + builder.Services.AddSingleton(); +} else Log.Information("Driver-only node — no ConfigDb registered; configuration via ConfigSource:Mode=FetchAndCache"); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs new file mode 100644 index 000000000..0e702382a --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs @@ -0,0 +1,185 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests; + +/// +/// lmxopcua#868: the coordinator's recovery read used to run synchronously in PreStart. With the +/// ConfigDb unreachable when the singleton started, the actor died on initialization, was never re-created, +/// and every deployment hung. It now loads asynchronously with backoff, queues what arrives meanwhile and +/// reports until the load succeeds. +/// +public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTestBase +{ + private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('d', 64)); + private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5); + private static readonly TimeSpan RetryMin = TimeSpan.FromMilliseconds(50); + private static readonly TimeSpan RetryMax = TimeSpan.FromMilliseconds(200); + private const string Node = "site-a-1:4053"; + + [Fact] + public void Db_unreachable_at_start_does_not_kill_the_coordinator_and_a_deploy_sent_during_the_outage_completes_once_it_returns() + { + var dbName = Guid.NewGuid().ToString("N"); + var direct = NewInMemoryDbFactory(dbName); + var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName)) { Down = true }; + SeedNodes(direct, Node); + var deploymentId = SeedDeployment(direct, DeploymentStatus.Dispatching, DateTime.UtcNow); + var readiness = new ConfigPublishCoordinatorReadiness(); + + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + outage, NoTimeout, readiness: readiness, loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + Watch(coordinator); + + // Several failed loads: the actor keeps retrying (it is alive) and reports Loading with the error. + AwaitAssert(() => + { + outage.Attempts.ShouldBeGreaterThanOrEqualTo(3); + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Loading); + readiness.Current.FailedLoadAttempts.ShouldBeGreaterThanOrEqualTo(2); + readiness.Current.LastError.ShouldNotBeNull(); + readiness.Current.LastError.ShouldContain("unreachable"); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + // A deployment dispatched during the outage is queued, not processed and not dropped. + coordinator.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken); + using (var db = direct.CreateDbContext()) + { + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Dispatching); + db.NodeDeploymentStates.ShouldBeEmpty(); + } + + // The DB comes back: the load succeeds, the queued dispatch runs, and the node's ack seals it. + outage.Down = false; + AwaitAssert(() => + { + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready); + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks); + db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { Node }); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + AwaitAssert(() => + { + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + ExpectNoMsg(TimeSpan.FromMilliseconds(100), TestContext.Current.CancellationToken); // never Terminated + } + + [Fact] + public void In_flight_deployment_is_recovered_after_the_outage_and_an_ack_queued_during_it_is_not_lost() + { + var dbName = Guid.NewGuid().ToString("N"); + var direct = NewInMemoryDbFactory(dbName); + var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName)) { Down = true }; + SeedNodes(direct, Node, "site-a-2:4053"); + var deploymentId = SeedDeployment(direct, DeploymentStatus.AwaitingApplyAcks, DateTime.UtcNow); + using (var db = direct.CreateDbContext()) + { + db.NodeDeploymentStates.Add(new NodeDeploymentState + { NodeId = Node, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applying }); + db.NodeDeploymentStates.Add(new NodeDeploymentState + { NodeId = "site-a-2:4053", DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applied, AppliedAtUtc = DateTime.UtcNow }); + db.SaveChanges(); + } + + var readiness = new ConfigPublishCoordinatorReadiness(); + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + outage, NoTimeout, readiness: readiness, loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + AwaitAssert(() => outage.Attempts.ShouldBeGreaterThanOrEqualTo(2), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + // The last outstanding ack arrives while the coordinator is still loading: it is queued, then applied + // against the recovered expected set (site-a-2 already Applied in the DB) and seals the deployment. + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + outage.Down = false; + + AwaitAssert(() => + { + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready); + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + [Fact] + public void Readiness_is_NotHosted_after_the_coordinator_stops() + { + var readiness = new ConfigPublishCoordinatorReadiness(); + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(NewInMemoryDbFactory(), NoTimeout, readiness: readiness)); + AwaitAssert(() => readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + Watch(coordinator); + Sys.Stop(coordinator); + ExpectTerminated(coordinator, cancellationToken: TestContext.Current.CancellationToken); + readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.NotHosted); + } + + private static DeploymentId SeedDeployment( + IDbContextFactory dbFactory, DeploymentStatus status, DateTime createdAtUtc) + { + var id = DeploymentId.NewId(); + using var db = dbFactory.CreateDbContext(); + db.Deployments.Add(new Deployment + { + DeploymentId = id.Value, + RevisionHash = TestRevision.Value, + Status = status, + CreatedBy = "test", + CreatedAtUtc = createdAtUtc, + }); + db.SaveChanges(); + return id; + } + + private static void SeedNodes(IDbContextFactory dbFactory, params string[] nodeIds) + { + using var db = dbFactory.CreateDbContext(); + db.ServerClusters.Add(new ServerCluster + { + ClusterId = "SITE-A", Name = "Site A", Enterprise = "zb", Site = "site-a", + NodeCount = (byte)nodeIds.Length, RedundancyMode = RedundancyMode.None, CreatedBy = "test", + }); + foreach (var nodeId in nodeIds) + { + db.ClusterNodes.Add(new ClusterNode + { + NodeId = nodeId, ClusterId = "SITE-A", Host = nodeId.Split(':')[0], + ApplicationUri = $"urn:OtOpcUa:{nodeId}", Enabled = true, CreatedBy = "test", + }); + } + db.SaveChanges(); + } + + /// A factory that throws while — a ConfigDb that is unreachable. + private sealed class OutageDbFactory(IDbContextFactory inner) : IDbContextFactory + { + private volatile bool _down; + private int _attempts; + + public bool Down { get => _down; set => _down = value; } + + public int Attempts => Volatile.Read(ref _attempts); + + public OtOpcUaConfigDbContext CreateDbContext() + { + Interlocked.Increment(ref _attempts); + if (_down) throw new InvalidOperationException("ConfigDb unreachable (simulated outage)"); + return inner.CreateDbContext(); + } + } +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs index bd73313ad..91a947c8d 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs @@ -136,9 +136,9 @@ public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTes var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout)); var admin = Sys.ActorOf(AdminOperationsActor.Props(adminFactory, coordinator, Enumerable.Empty())); - // Arm the fault AFTER PreStart has run (it would otherwise fault the recovery read itself and - // the actor would be stopped, not restarted). Identify is processed after creation, so a - // resolved selection proves PreStart completed. + // Arm the fault AFTER PreStart has run. Since lmxopcua#868 the recovery read is asynchronous and + // retried, so a fault that lands on it instead of on HandleDispatch is also recovered (the dispatch is + // queued until the retried load succeeds) — either way the deployment must complete. await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget, TestContext.Current.CancellationToken); faulting.FailNext(); admin.Tell(new StartDeployment("joe", CorrelationId.NewId())); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs new file mode 100644 index 000000000..a9202432c --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs @@ -0,0 +1,65 @@ +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Diagnostics.HealthChecks; +using Microsoft.Extensions.Options; +using Shouldly; +using Xunit; +using ZB.MOM.WW.Health; +using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; +using ZB.MOM.WW.OtOpcUa.Host.Health; + +namespace ZB.MOM.WW.OtOpcUa.Host.Tests.Health; + +/// lmxopcua#868: the deploy coordinator reports Degraded (never Unhealthy) while it cannot load from the ConfigDb. +public sealed class ConfigPublishCoordinatorHealthCheckTests +{ + private static async Task Check(ConfigPublishCoordinatorReadiness? readiness) => + await new ConfigPublishCoordinatorHealthCheck(readiness).CheckHealthAsync(new HealthCheckContext(), CancellationToken.None); + + [Fact] + public async Task Loading_after_failed_attempts_is_degraded_and_names_the_error() + { + var r = new ConfigPublishCoordinatorReadiness(); + r.MarkLoading(3, "connection refused"); + var result = await Check(r); + result.Status.ShouldBe(HealthStatus.Degraded); + result.Description.ShouldNotBeNull(); + result.Description.ShouldContain("connection refused"); + result.Description.ShouldContain("queued"); + } + + [Fact] + public async Task Ready_is_healthy() + { + var r = new ConfigPublishCoordinatorReadiness(); + r.MarkLoading(2, "x"); + r.MarkReady(); + (await Check(r)).Status.ShouldBe(HealthStatus.Healthy); + } + + [Fact] + public async Task Not_hosted_and_untracked_are_healthy() + { + (await Check(new ConfigPublishCoordinatorReadiness())).Status.ShouldBe(HealthStatus.Healthy); + (await Check(null)).Status.ShouldBe(HealthStatus.Healthy); + } + + [Theory] + [InlineData(true, true)] + [InlineData(false, false)] + public void It_is_registered_on_the_ready_tier_only_on_admin_nodes(bool hasAdmin, bool registered) + { + var registrations = new ServiceCollection() + .AddOtOpcUaHealth(hasAdmin: hasAdmin, hasDriver: true) + .BuildServiceProvider() + .GetRequiredService>() + .Value.Registrations + .Where(x => x.Name == ConfigPublishCoordinatorHealthCheck.Name) + .ToList(); + if (!registered) + { + registrations.ShouldBeEmpty(); + return; + } + registrations.ShouldHaveSingleItem().Tags.ShouldBe([ZbHealthTags.Ready]); + } +}