From b437c34f924a689177b3ab8b867f78ed6b5edb95 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:22:20 -0400 Subject: [PATCH] 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 --- .../Coordinators/ConfigPublishCoordinator.cs | 99 ++++++-- .../ConfigPublishCoordinatorRecoveryTests.cs | 234 ++++++++++++++++++ 2 files changed, 316 insertions(+), 17 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs 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 b621cc64..3c99e998 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -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, diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs new file mode 100644 index 00000000..86f6efa7 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs @@ -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; + +/// +/// Review 03 §2.2: 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 — used to be recovered with an EMPTY expected +/// set, at which point the deadline hit the 0 == 0 "race" branch and returned without touching +/// the row. Every later StartDeployment was then refused AnotherDeploymentInFlight forever. +/// +public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTestBase +{ + private static readonly RevisionHash TestRevision = RevisionHash.Parse(new string('c', 64)); + private static readonly TimeSpan NoTimeout = TimeSpan.FromMinutes(5); + + /// + /// 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. + /// + [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); + } + + /// + /// A lost dispatch that is already past the apply deadline is closed as TimedOut with a reason + /// naming the condition — never left in Dispatching — and the next deployment proceeds + /// (positive control: it is dispatched, acked and sealed through the same coordinator). + /// + [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); + } + + /// + /// An AwaitingApplyAcks 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. + /// + [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); + } + + /// + /// The brief's scenario end to end: HandleDispatch throws once (a transient fault seeding the + /// expected set). The coordinator restarts, recovers the Dispatching row, re-dispatches it and + /// the deployment completes; a second StartDeployment is then Accepted (positive control: a + /// normal deploy seals). + /// + [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())); + + // 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(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(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 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(); + } + + /// Throws from exactly one CreateDbContext call after . + private sealed class OneShotFaultingDbFactory(IDbContextFactory inner) + : IDbContextFactory + { + 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(); + } + } +}