fix(controlplane): recover a lost dispatch instead of wedging the deploy pipeline

A Deployment row left in Dispatching with no NodeDeploymentState rows (HandleDispatch threw before its SaveChanges, or the process died between the AdminOps row write and the coordinator's Tell) was recovered with an empty expected set; the deadline then hit the 0 == 0 "race" branch and returned without touching the row, so every later StartDeployment was refused AnotherDeploymentInFlight until someone hand-edited SQL. - PreStart: a recovered row with no expected-ack rows is a lost dispatch. Inside the apply deadline it is re-driven through HandleDispatch (bounded: CreatedAtUtc never moves); past it the row is closed TimedOut with a reason naming the condition. - HandleDispatch is idempotent on a re-dispatch (rows already seeded ARE the expected set). - HandleDeadline takes the race branch only when _expectedAcks.Count > 0. - The PreStart recovery read projects ArtifactBlob away (R03:3.5). Refs #733 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:22:20 -04:00
parent ad2b5c4044
commit b437c34f92
2 changed files with 316 additions and 17 deletions
@@ -161,9 +161,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
        DistributedPubSub.Get(Context.System).Mediator.Tell(new Subscribe(DeploymentAcksTopic, Self));

        using var db = _dbFactory.CreateDbContext();
+        // 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
            .Where(d => d.Status == DeploymentStatus.Dispatching || d.Status == DeploymentStatus.AwaitingApplyAcks)
            .OrderByDescending(d => d.CreatedAtUtc)
+            .Select(d => new { d.DeploymentId, d.RevisionHash, d.CreatedAtUtc })
            .FirstOrDefault();
        if (inflight is null) return;

@@ -173,6 +176,41 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
            .AsNoTracking()
            .ToList();

+        // 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
+        // proxy — we don't track a separate "dispatched at" column).
+        var elapsed = DateTime.UtcNow - inflight.CreatedAtUtc;
+        var remaining = _applyDeadline - elapsed;
+
+        if (nodeStates.Count == 0)
+        {
+            // The expected set and the AwaitingApplyAcks flip are written in ONE SaveChanges by
+            // HandleDispatch, so a row with no NodeDeploymentState rows is a dispatch that never completed:
+            // HandleDispatch threw before its SaveChanges (this actor restarted and the message was
+            // dropped), the process died between AdminOps' row write and the coordinator's Tell, or the
+            // Tell was lost across a singleton hand-over. Recovering it with an EMPTY expected set used to
+            // let the deadline take the "everyone acked" race branch (0 == 0) and leave the row in flight
+            // forever, refusing every later StartDeployment. Inside the deadline the dispatch is simply
+            // re-driven through the normal handler (bounded: CreatedAtUtc never moves, so a dispatch that
+            // keeps throwing ends up on the other branch once the deadline passes); past it the row is
+            // 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);
+                Self.Tell(new DispatchDeployment(
+                    _current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
+            }
+            else
+            {
+                Self.Tell(new DeadlineElapsed(_current.Value));
+            }
+            return;
+        }
+
        _expectedAcks = nodeStates.Select(s => NodeId.Parse(s.NodeId)).ToHashSet(NodeIdComparer);
        foreach (var s in nodeStates.Where(s => s.Status != NodeDeploymentStatus.Applying))
        {
@@ -182,11 +220,6 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
            if (IsDegraded(s.Status, s.FailureReason)) _degraded.Add(NodeId.Parse(s.NodeId));
        }

-        // 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
-        // proxy — we don't track a separate "dispatched at" column).
-        var elapsed = DateTime.UtcNow - inflight.CreatedAtUtc;
-        var remaining = _applyDeadline - elapsed;
        if (remaining <= TimeSpan.Zero)
        {
            Self.Tell(new DeadlineElapsed(_current.Value));
@@ -208,19 +241,34 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
        // Seed NodeDeploymentState rows so a failover knows which nodes were expected to ack.
        using (var db = _dbFactory.CreateDbContext())
        {
-            // Discovery reuses this context rather than opening a second one — it reads the same
-            // ClusterNode table the NodeDeploymentState rows below are FK-bound to, so a single
-            // context keeps the expected set and the seeded rows consistent by construction.
-            _expectedAcks = DiscoverDriverNodes(db);
-
-            foreach (var node in _expectedAcks)
+            // Idempotent on a re-dispatch (PreStart re-drives a lost dispatch through this handler): rows
+            // already seeded for this deployment ARE the expected set — re-discovering would both duplicate
+            // the primary key and let a node toggled since the first attempt change who is expected.
+            var seeded = db.NodeDeploymentStates
+                .AsNoTracking()
+                .Where(x => x.DeploymentId == msg.DeploymentId.Value)
+                .Select(x => x.NodeId)
+                .ToList();
+            if (seeded.Count > 0)
            {
-                db.NodeDeploymentStates.Add(new NodeDeploymentState
+                _expectedAcks = seeded.Select(NodeId.Parse).ToHashSet(NodeIdComparer);
+            }
+            else
+            {
+                // Discovery reuses this context rather than opening a second one — it reads the same
+                // ClusterNode table the NodeDeploymentState rows below are FK-bound to, so a single
+                // context keeps the expected set and the seeded rows consistent by construction.
+                _expectedAcks = DiscoverDriverNodes(db);
+
+                foreach (var node in _expectedAcks)
                {
-                    NodeId = node.Value,
-                    DeploymentId = msg.DeploymentId.Value,
-                    Status = NodeDeploymentStatus.Applying,
-                });
+                    db.NodeDeploymentStates.Add(new NodeDeploymentState
+                    {
+                        NodeId = node.Value,
+                        DeploymentId = msg.DeploymentId.Value,
+                        Status = NodeDeploymentStatus.Applying,
+                    });
+                }
            }
            UpdateDeploymentStatus(db, msg.DeploymentId, DeploymentStatus.AwaitingApplyAcks);
            db.SaveChanges();
@@ -360,7 +408,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
                msg.DeploymentId, _current);
            return;
        }
-        if (_acks.Count == _expectedAcks.Count)
+        // Only a NON-EMPTY expected set can be "fully acked": with nothing expected, 0 == 0 is not the
+        // 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.
            return;
@@ -368,6 +418,21 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers

        using var db = _dbFactory.CreateDbContext();
        UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut);
+        if (_expectedAcks.Count == 0)
+        {
+            var d = db.Deployments.FirstOrDefault(x => x.DeploymentId == _current.Value.Value);
+            if (d is not null)
+                d.FailureReason = Truncate(
+                    "Dispatch was lost before any node was expected to ack (coordinator restart or singleton " +
+                    "hand-over mid-dispatch) and the apply deadline has passed. Redeploy.");
+            db.SaveChanges();
+            _log.Warning(
+                "Deployment {Id} timed out with NO expected-ack rows: the dispatch was lost before any node was " +
+                "expected and the {Deadline} apply deadline has passed. Closed as TimedOut — redeploy",
+                _current.Value, _applyDeadline);
+            ResetForNext();
+            return;
+        }
        db.SaveChanges();
        // Name the silent nodes, not just the counts. Since the expected set became DB-derived this
        // is the failure an operator meets after deploying while a configured node is switched off,
@@ -0,0 +1,234 @@
+using Akka.Actor;
+using Microsoft.EntityFrameworkCore;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
+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.AdminOperations;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
+
+/// <summary>
+///     Review 03 §2.2: a <c>Deployment</c> row left in <c>Dispatching</c> with NO <c>NodeDeploymentState</c>
+///     rows — <c>HandleDispatch</c> threw before its <c>SaveChanges</c>, or the process died between the
+///     AdminOps row write and the coordinator's <c>Tell</c> — used to be recovered with an EMPTY expected
+///     set, at which point the deadline hit the <c>0 == 0</c> "race" branch and returned without touching
+///     the row. Every later <c>StartDeployment</c> was then refused <c>AnotherDeploymentInFlight</c> forever.
+/// </summary>
+public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTestBase
+{
+    private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('c', 64));
+    private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5);
+
+    /// <summary>
+    ///     A lost dispatch that is still inside the apply deadline is re-driven: PreStart re-dispatches it,
+    ///     the expected set is seeded from the enabled rows, and the node's ack seals it.
+    /// </summary>
+    [Fact]
+    public void Recovered_row_with_no_expected_set_is_redispatched_and_then_seals()
+    {
+        var dbFactory = NewInMemoryDbFactory();
+        var deploymentId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow);
+        SeedNodes(dbFactory, "site-a-1:4053");
+
+        // No DispatchDeployment is ever sent — PreStart alone has to recover it.
+        var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
+
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            db.NodeDeploymentStates.Select(s => s.NodeId).ShouldBe(new[] { "site-a-1:4053" });
+            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
+        }, duration: PresenceBudget);
+
+        actor.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-1:4053"),
+            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
+
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
+        }, duration: PresenceBudget);
+    }
+
+    /// <summary>
+    ///     A lost dispatch that is already past the apply deadline is closed as <c>TimedOut</c> with a reason
+    ///     naming the condition — never left in <c>Dispatching</c> — and the next deployment proceeds
+    ///     (positive control: it is dispatched, acked and sealed through the same coordinator).
+    /// </summary>
+    [Fact]
+    public void Recovered_row_past_deadline_with_no_expected_set_is_timed_out_not_wedged()
+    {
+        var dbFactory = NewInMemoryDbFactory();
+        var staleId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow.AddMinutes(-10));
+        SeedNodes(dbFactory, "site-a-1:4053");
+
+        var actor = Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
+
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            var row = db.Deployments.Single(d => d.DeploymentId == staleId.Value);
+            row.Status.ShouldBe(DeploymentStatus.TimedOut);
+            row.FailureReason.ShouldNotBeNull();
+            row.FailureReason.ShouldContain("lost");
+        }, duration: PresenceBudget);
+
+        // Positive control: the coordinator is not wedged — a fresh deployment runs end to end.
+        var nextId = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow);
+        actor.Tell(new DispatchDeployment(nextId, TestRevision, CorrelationId.NewId()));
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            db.Deployments.Single(d => d.DeploymentId == nextId.Value).Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
+        }, duration: PresenceBudget);
+
+        actor.Tell(new ApplyAck(nextId, NodeId.Parse("site-a-1:4053"),
+            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            db.Deployments.Single(d => d.DeploymentId == nextId.Value).Status.ShouldBe(DeploymentStatus.Sealed);
+        }, duration: PresenceBudget);
+    }
+
+    /// <summary>
+    ///     An <c>AwaitingApplyAcks</c> row with no rows (the process died between the status flip and the
+    ///     empty-fleet seal) is re-driven too, and with no enabled nodes it seals empty exactly as a fresh
+    ///     dispatch would — the recovery keys on the missing expected set, not on the status value.
+    /// </summary>
+    [Fact]
+    public void Recovered_AwaitingApplyAcks_row_with_no_rows_and_no_fleet_seals_empty()
+    {
+        var dbFactory = NewInMemoryDbFactory();
+        SeedDeployment(dbFactory, DeploymentStatus.AwaitingApplyAcks, DateTime.UtcNow);
+        Sys.ActorOf(ConfigPublishCoordinator.Props(dbFactory, NoTimeout));
+
+        AwaitAssert(() =>
+        {
+            using var db = dbFactory.CreateDbContext();
+            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
+            db.NodeDeploymentStates.ShouldBeEmpty();
+        }, duration: PresenceBudget);
+    }
+
+    /// <summary>
+    ///     The brief's scenario end to end: <c>HandleDispatch</c> throws once (a transient fault seeding the
+    ///     expected set). The coordinator restarts, recovers the <c>Dispatching</c> row, re-dispatches it and
+    ///     the deployment completes; a second <c>StartDeployment</c> is then Accepted (positive control: a
+    ///     normal deploy seals).
+    /// </summary>
+    [Fact]
+    public async Task HandleDispatch_throw_is_recovered_and_the_next_StartDeployment_succeeds()
+    {
+        var dbName = Guid.NewGuid().ToString("N");
+        var adminFactory = NewInMemoryDbFactory(dbName);
+        var faulting = new OneShotFaultingDbFactory(NewInMemoryDbFactory(dbName));
+        SeedNodes(adminFactory, "site-a-1:4053");
+
+        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(faulting, NoTimeout));
+        var admin = Sys.ActorOf(AdminOperationsActor.Props(adminFactory, coordinator, Enumerable.Empty<IDriverProbe>()));
+
+        // 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.
+        await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget);
+        faulting.FailNext();
+        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
+        var first = ExpectMsg<StartDeploymentResult>(PresenceBudget);
+        first.Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
+        first.DeploymentId.ShouldNotBeNull();
+
+        // The fault fired, the actor restarted, PreStart re-dispatched: the expected set lands.
+        AwaitAssert(() =>
+        {
+            faulting.Faulted.ShouldBeTrue();
+            using var db = adminFactory.CreateDbContext();
+            db.NodeDeploymentStates.Count(s => s.DeploymentId == first.DeploymentId.Value.Value).ShouldBe(1);
+            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks);
+        }, duration: PresenceBudget);
+
+        coordinator.Tell(new ApplyAck(first.DeploymentId.Value, NodeId.Parse("site-a-1:4053"),
+            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
+        AwaitAssert(() =>
+        {
+            using var db = adminFactory.CreateDbContext();
+            db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
+        }, duration: PresenceBudget);
+
+        // Positive control: nothing is wedged — the next deploy is Accepted and seals too.
+        admin.Tell(new StartDeployment("joe", CorrelationId.NewId()));
+        var second = ExpectMsg<StartDeploymentResult>(PresenceBudget);
+        second.Outcome.ShouldBe(StartDeploymentOutcome.Accepted);
+        coordinator.Tell(new ApplyAck(second.DeploymentId!.Value, NodeId.Parse("site-a-1:4053"),
+            ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
+        AwaitAssert(() =>
+        {
+            using var db = adminFactory.CreateDbContext();
+            db.Deployments.Count(d => d.Status == DeploymentStatus.Sealed).ShouldBe(2);
+        }, duration: PresenceBudget);
+    }
+
+    private static DeploymentId SeedDeployment(
+        IDbContextFactory<OtOpcUaConfigDbContext> 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<OtOpcUaConfigDbContext> 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();
+    }
+
+    /// <summary>Throws from exactly one <c>CreateDbContext</c> call after <see cref="FailNext"/>.</summary>
+    private sealed class OneShotFaultingDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
+        : IDbContextFactory<OtOpcUaConfigDbContext>
+    {
+        private int _armed;
+        public bool Faulted { get; private set; }
+
+        public void FailNext() => Interlocked.Exchange(ref _armed, 1);
+
+        public OtOpcUaConfigDbContext CreateDbContext()
+        {
+            if (Interlocked.Exchange(ref _armed, 0) == 1)
+            {
+                Faulted = true;
+                throw new InvalidOperationException("injected transient fault seeding the expected set");
+            }
+            return inner.CreateDbContext();
+        }
+    }
+}