Merge #868 review fixes: recovered fully-acked deployments seal, stashed acks beat a past deadline, failed ack writes retry

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
2026-11-11 12:38:57 -05:00
parent 2bdff975fa 19c7b5d7e4
commit ff85b8511a
2 changed files with 204 additions and 18 deletions
@@ -64,6 +64,18 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
 /// during the outage is queued, never dropped. <see cref="ConfigPublishCoordinatorReadiness"/> reports the
 /// loading state (and the last error) to the node's health endpoint as Degraded.
 /// </para>
+/// <para>
+/// <b>The stash is node-local.</b> 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
+/// <c>Dispatching</c> 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.
+/// </para>
+/// <para>
+/// <b>An ack is held in memory only once it is persisted.</b> <c>NodeDeploymentState</c> is the system of record
+/// (driver-only nodes never write their own row), so a failed <c>PersistNodeAck</c> re-queues the ack on a timer
+/// instead of throwing — a throw would restart the actor and drop it.
+/// </para>
 /// </summary>
 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;

+    /// <summary>Deployments whose dispatch was queued while Loading (only to word the recovery log correctly).</summary>
+    private readonly HashSet<DeploymentId> _queuedDispatches = new();
+
    /// <summary>
    ///     Where central→node commands are handed off (per-cluster mesh Phase 2), or
    ///     <see langword="null"/> 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<DispatchDeployment>(HandleDispatch);
        Receive<ApplyAck>(HandleAck);
+        Receive<RetryAckPersist>(m => HandleAck(m.Ack));
        Receive<DeadlineElapsed>(HandleDeadline);
        Receive<SubscribeAck>(_ => { /* 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
    }

    /// <summary>
-    ///     Applies the recovered state, switches to <see cref="Ready"/> 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 <see cref="Ready"/> 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.
    /// </summary>
    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();
    }

    /// <summary>
@@ -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();
+    }

+    /// <summary>Seals (all Applied) or marks partially failed — called once every expected ack is in.</summary>
+    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";

+    /// <summary>An ack whose persist failed, re-delivered by its retry timer.</summary>
+    private sealed record RetryAckPersist(ApplyAck Ack);
+
    public sealed record DeadlineElapsed(DeploymentId DeploymentId);

    /// <summary>The recovery read's result: the newest in-flight deployment, or null when none.</summary>
@@ -115,6 +115,105 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes
        }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
    }

+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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);
+    }
+
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> dbFactory, params string[] nodeIds)
    {
        using var db = dbFactory.CreateDbContext();