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 c96b682a9..c2b6770e1 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -64,6 +64,18 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators; /// during the outage is queued, never dropped. reports the /// loading state (and the last error) to the node's health endpoint as Degraded. /// +/// +/// The stash is node-local. A singleton hand-over while the coordinator is still Loading discards what it +/// queued (the old incarnation stops with its mailbox). That loses nothing durable: a queued dispatch's row is +/// Dispatching with no expected-ack rows, which the new incarnation's recovery (or AdminOps' lost-dispatch +/// closure) re-drives or closes; a queued ack is the one exception — its node does not resend, so that node is +/// named in the deadline's timeout reason. +/// +/// +/// An ack is held in memory only once it is persisted. NodeDeploymentState is the system of record +/// (driver-only nodes never write their own row), so a failed PersistNodeAck re-queues the ack on a timer +/// instead of throwing — a throw would restart the actor and drop it. +/// /// public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithUnboundedStash { @@ -86,6 +98,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU private readonly ConfigPublishCoordinatorReadiness? _readiness; private int _failedLoads; + /// Deployments whose dispatch was queued while Loading (only to word the recovery log correctly). + private readonly HashSet _queuedDispatches = new(); + /// /// Where central→node commands are handed off (per-cluster mesh Phase 2), or /// to publish on the mediator directly. @@ -204,6 +219,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU "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); + _queuedDispatches.Add(msg.DeploymentId); Stash.Stash(); }); ReceiveAny(_ => Stash.Stash()); @@ -214,6 +230,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU { Receive(HandleDispatch); Receive(HandleAck); + Receive(m => HandleAck(m.Ack)); Receive(HandleDeadline); Receive(_ => { /* DPS subscribe confirmation */ }); // A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed. @@ -296,26 +313,38 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU } /// - /// 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. + /// Applies the recovered state, then switches to and replays every stashed message. A + /// decision that must see the queued messages first (the deadline of a recovered deployment that is already + /// past it) is SELF-SENT, so it lands behind the replay; one that must precede them (re-driving a lost + /// dispatch, so a queued duplicate of it is recognised) runs inline. A throw while applying is treated as a + /// failed load — the actor stays Loading and retries on the backoff, never a tight restart loop. /// 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 + catch (Exception ex) { - Stash.UnstashAll(); + _log.Error(ex, "ConfigPublishCoordinator failed to apply its recovered state; retrying the load"); + Timers.Cancel(DeadlineTimerKey); + _current = null; + _expectedAcks.Clear(); + _acks.Clear(); + _degraded.Clear(); + HandleRecoveryLoadFailed(new RecoveryLoadFailed(ex)); + return; } + + if (_failedLoads > 0) + _log.Info("ConfigPublishCoordinator loaded its recovery state after {Failed} failed attempt(s)", _failedLoads); + _failedLoads = 0; + _queuedDispatches.Clear(); + Become(Ready); + _readiness?.MarkReady(); + Stash.UnstashAll(); } /// @@ -350,16 +379,25 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU // closed as TimedOut with a reason that names the condition. if (remaining > TimeSpan.Zero) { - _log.Warning( - "Coordinator recovered deployment {Id} with NO expected-ack rows — the dispatch was lost " + - "before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " + - "re-dispatching with {Remaining} of the apply deadline left", - _current, remaining); + if (_queuedDispatches.Contains(_current.Value)) + _log.Info( + "Coordinator dispatching deployment {Id}, queued while its recovery state was loading", + _current); + else + _log.Warning( + "Coordinator recovered deployment {Id} with NO expected-ack rows — the dispatch was lost " + + "before any node was expected (coordinator restart or singleton hand-over mid-dispatch); " + + "re-dispatching with {Remaining} of the apply deadline left", + _current, remaining); HandleDispatch(new DispatchDeployment( _current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId())); } else { + // Deliberately INLINE, unlike the populated-set deadline below: with no expected set no queued ack + // can count, so there is nothing to let the replay apply first — and a queued dispatch of this very + // row must meet the TimedOut row (terminal-state guard ignores it), not publish to every node and + // be timed out the moment a self-sent deadline caught up with it. HandleDeadline(new DeadlineElapsed(_current.Value)); } return; @@ -374,9 +412,20 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId)); } + // Every expected ack is already in the DB (a seal whose SaveChanges failed, or a fused node's own + // DriverHostActor row): finish it now. Waiting for the deadline would leave it in flight — the deadline + // can only time out an INCOMPLETE set. + if (_acks.Count == _expectedAcks.Count) + { + _log.Info("Coordinator recovered deployment {Id} with every ack already landed; completing it", _current); + CompleteFromAcks(); + return; + } + if (remaining <= TimeSpan.Zero) { - HandleDeadline(new DeadlineElapsed(_current.Value)); + // Self-sent, so acks queued during the load (which arrived in time) are applied first. + Self.Tell(new DeadlineElapsed(_current.Value)); return; } @@ -516,20 +565,41 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU return; } + // Persist FIRST: the row is the system of record (driver-only nodes never write their own, and nodes do not + // resend). A throw here used to restart the actor and drop the ack; it is re-queued instead. + try + { + PersistNodeAck(msg); + } + catch (Exception ex) + { + _log.Warning(ex, + "Could not persist the ApplyAck from {Node} for {Id}; retrying in {Delay}", + msg.NodeId, msg.DeploymentId, _loadRetryMin); + Timers.StartSingleTimer(AckRetryTimerKey(msg), new RetryAckPersist(msg), _loadRetryMin); + return; + } + _acks[msg.NodeId] = msg.Outcome; // An old node never sets Degraded, so its ack takes the clean branch exactly as before. if (IsDegradedAck(msg)) _degraded.Add(msg.NodeId); else _degraded.Remove(msg.NodeId); - PersistNodeAck(msg); if (_acks.Count < _expectedAcks.Count) return; + CompleteFromAcks(); + } + /// Seals (all Applied) or marks partially failed — called once every expected ack is in. + private void CompleteFromAcks() + { if (_acks.Values.All(o => o == ApplyAckOutcome.Applied)) SealDeployment(); else MarkPartiallyFailed(); } + private static string AckRetryTimerKey(ApplyAck msg) => $"ack-retry:{msg.DeploymentId.Value:N}:{msg.NodeId.Value}"; + private void PersistNodeAck(ApplyAck msg) { using var db = _dbFactory.CreateDbContext(); @@ -612,7 +682,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU // everyone-acked race, it is a lost dispatch (see PreStart) that must be closed, not skipped. if (_expectedAcks.Count > 0 && _acks.Count == _expectedAcks.Count) { - // Race: every node acked just as the deadline fired. Already sealed/failed elsewhere. + // Every ack is in but nothing completed it (a completion normally resets _current, so the stale-id + // check above catches the ack-vs-deadline race). Complete it rather than leave it in flight forever. + CompleteFromAcks(); return; } @@ -682,6 +754,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU private const string DeadlineTimerKey = "apply-deadline"; private const string LoadRetryTimerKey = "recovery-load-retry"; + /// An ack whose persist failed, re-delivered by its retry timer. + private sealed record RetryAckPersist(ApplyAck Ack); + public sealed record DeadlineElapsed(DeploymentId DeploymentId); /// The recovery read's result: the newest in-flight deployment, or null when none. diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs index 0e702382a..ae10d1966 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs @@ -115,6 +115,105 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); } + /// + /// Review finding 1: a recovered deployment whose every expected ack is already in the DB (a seal whose + /// SaveChanges failed, or a fused node's own row) must be completed — inside the deadline it used to wait for + /// acks that had already landed, past it the deadline's "race" guard returned without sealing; either way it + /// sat in AwaitingApplyAcks forever and AdminOps refused every later deploy. + /// + [Theory] + [InlineData(false, false, DeploymentStatus.Sealed)] + [InlineData(true, false, DeploymentStatus.Sealed)] + [InlineData(false, true, DeploymentStatus.PartiallyFailed)] + [InlineData(true, true, DeploymentStatus.PartiallyFailed)] + public void Recovered_deployment_with_every_ack_already_persisted_is_completed(bool pastDeadline, bool oneFailed, DeploymentStatus expected) + { + var dbFactory = NewInMemoryDbFactory(); + SeedNodes(dbFactory, Node, "site-a-2:4053"); + var deadline = TimeSpan.FromMinutes(1); + var deploymentId = SeedDeployment(dbFactory, DeploymentStatus.AwaitingApplyAcks, + pastDeadline ? DateTime.UtcNow.AddMinutes(-10) : DateTime.UtcNow); + SeedNodeState(dbFactory, deploymentId, Node, NodeDeploymentStatus.Applied); + SeedNodeState(dbFactory, deploymentId, "site-a-2:4053", oneFailed ? NodeDeploymentStatus.Failed : NodeDeploymentStatus.Applied); + + Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, deadline)); + + AwaitAssert(() => + { + using var db = dbFactory.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(expected); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + /// + /// Review finding 2: a recovered deployment that is already past its deadline must first apply the acks that + /// were queued while the load was failing (they arrived in time); the deadline used to run inline before the + /// replay, so they were discarded as stale and the deployment was timed out. + /// + [Fact] + public void Ack_queued_during_the_outage_is_applied_before_a_past_deadline_recovery_times_out() + { + 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.AwaitingApplyAcks, DateTime.UtcNow.AddMinutes(-10)); + SeedNodeState(direct, deploymentId, Node, NodeDeploymentStatus.Applying); + + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + outage, TimeSpan.FromMinutes(1), loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + AwaitAssert(() => outage.Attempts.ShouldBeGreaterThanOrEqualTo(2), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + outage.Down = false; + + AwaitAssert(() => + { + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + db.NodeDeploymentStates.Single().Status.ShouldBe(NodeDeploymentStatus.Applied); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + /// + /// Review finding 3: an ack whose persist fails (ConfigDb blip while Ready) is retried, not dropped. It used to + /// throw, restart the actor and lose the ack — the node never resends and never writes its own row. + /// + [Fact] + public void Ack_whose_persist_fails_is_retried_not_dropped() + { + var dbName = Guid.NewGuid().ToString("N"); + var direct = NewInMemoryDbFactory(dbName); + var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName)); + 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)); + coordinator.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitAssert(() => + { + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + outage.Down = true; + var before = outage.Attempts; + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + AwaitAssert(() => outage.Attempts.ShouldBeGreaterThanOrEqualTo(before + 2), + duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + outage.Down = false; + + AwaitAssert(() => + { + using var db = direct.CreateDbContext(); + db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + db.NodeDeploymentStates.Single().Status.ShouldBe(NodeDeploymentStatus.Applied); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + [Fact] public void Readiness_is_NotHosted_after_the_coordinator_stops() { @@ -146,6 +245,18 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes return id; } + private static void SeedNodeState( + IDbContextFactory dbFactory, DeploymentId deploymentId, string nodeId, NodeDeploymentStatus status) + { + using var db = dbFactory.CreateDbContext(); + db.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = nodeId, DeploymentId = deploymentId.Value, Status = status, + AppliedAtUtc = status == NodeDeploymentStatus.Applying ? null : DateTime.UtcNow, + }); + db.SaveChanges(); + } + private static void SeedNodes(IDbContextFactory dbFactory, params string[] nodeIds) { using var db = dbFactory.CreateDbContext();