dohertj2/lmxopcua
1
0
Fork 0
Code Issues 5 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(controlplane,runtime): #868 second review — final-outcome node rows, unpersisted acks survive deadline and restart

Browse Source 
M1 DriverHostActor wrote its own NodeDeploymentState row Applied/null as soon
   as the revision advanced, up to ~105 s before the address-space outcome.
   The coordinator's recovery counts any non-Applying row as an ack, so a
   coordinator restart in that window sealed CLEAN while the node was about to
   report degraded (#601). The row now stays Applying until ReleasePendingAck
   (or the no-publish-actor path) writes the final Applied, clean or with the
   degraded reason. A node restart in the window replays the orphan Applying
   row; /fleet shows Applying until the outcome (docs/Redundancy.md).
M2 Acks whose persist failed are tracked in _unpersisted (newest per node).
   HandleDeadline persists that backlog first and defers its decision while
   any remain, so a node that acked is never timed out as silent; PreRestart
   re-sends them to Self so a restart (timers lost) does not drop them.
L3 A successful persist cancels the node's retry timer and drops its older
   pending ack; a retry only fires for the node's current pending ack.
L4 A dispatch queued during Loading is sent with a fresh deadline however long
   the outage lasted (never closed "Dispatch was lost"); the inline re-drive of
   a genuinely lost dispatch arms the REMAINING deadline.
L5 A recovered row with a blank RevisionHash (re-dispatch) or blank NodeId is
   closed PartiallyFailed with a reason instead of failing every load.
L6 Readiness carries PendingAckPersists + LastPersistError; the config-publish
   health check is Degraded while Ready with a backlog.
L7 Ack-persist retries back off (shared Backoff); stack trace on the first
   failure only.
L8 Terminal writes (Sealed / PartiallyFailed / TimedOut / failure reason) only
   apply to a row still Dispatching/AwaitingApplyAcks.

Tests (all fail on the previous code): ControlPlane DbOutage — deadline does
not time out an unpersisted ack; unpersisted ack survives a restart; stale
retry does not overwrite a newer ack; queued dispatch through an outage longer
than the deadline is sent; malformed row closed and coordinator carries on;
readiness reports the persist backlog. Runtime DegradedAck — own row stays
Applying until the outcome. Host — Ready + backlog is Degraded.

Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 12:54:30 -05:00
parent ff85b8511a
commit 494dcb5b87
8 changed files with 475 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs
+217 -2
View File
@@ -116,8 +116,9 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes
}
/// <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
/// Review finding 1: a recovered deployment whose every expected ack is already in the DB as a FINAL outcome
/// (a seal whose SaveChanges failed, or a fused node's own row — which <c>DriverHostActor</c> writes Applied only
/// once the address-space outcome is known; a pre-outcome row stays Applying, second review M1) 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>
@@ -214,6 +215,214 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
/// <summary>
/// Second review M2: an ack that arrived but whose persist is still failing at the deadline must not be timed
/// out as "no ack from" its node. The deadline is deferred until the backlog persists, then the deployment seals.
/// </summary>
[Fact]
public void Deadline_does_not_time_out_a_received_but_unpersisted_ack()
{
var (direct, outage, deploymentId, coordinator, _) = DispatchedSingleNode(TimeSpan.FromSeconds(2));
outage.Down = true;
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
// Hold the DB down past the 2 s apply deadline.
Thread.Sleep(TimeSpan.FromSeconds(3.5));
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>
/// Second review M2: an unpersisted ack must survive an actor restart (its retry timer does not). Here a second
/// dispatch's guard read throws while the DB is down, restarting the coordinator.
/// </summary>
[Fact]
public void Unpersisted_ack_survives_a_coordinator_restart()
{
var (direct, outage, deploymentId, coordinator, _) = DispatchedSingleNode(NoTimeout);
outage.Down = true;
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
var before = outage.Attempts;
AwaitAssert(() => outage.Attempts.ShouldBeGreaterThan(before),
duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
// Its guard read throws (DB down) => the coordinator restarts.
coordinator.Tell(new DispatchDeployment(DeploymentId.NewId(), TestRevision, CorrelationId.NewId()));
Thread.Sleep(TimeSpan.FromMilliseconds(500));
outage.Down = false;
AwaitAssert(() =>
{
using var db = direct.CreateDbContext();
db.Deployments.Single(d => d.DeploymentId == deploymentId.Value).Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
/// <summary>
/// Second review L3: a retry of a node's OLDER unpersisted ack must not overwrite the newer ack that persisted
/// after it (it used to flip the node to Failed and turn a clean deploy PartiallyFailed).
/// </summary>
[Fact]
public void A_stale_ack_retry_does_not_overwrite_a_newer_persisted_ack()
{
var dbName = Guid.NewGuid().ToString("N");
var direct = NewInMemoryDbFactory(dbName);
var outage = new OutageDbFactory(NewInMemoryDbFactory(dbName));
SeedNodes(direct, Node, "site-a-2:4053");
var deploymentId = SeedDeployment(direct, DeploymentStatus.Dispatching, DateTime.UtcNow);
var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(
outage, NoTimeout, 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);
// Node 1's first answer (Failed) cannot be persisted; its corrected answer (Applied) can.
outage.FailNext(1);
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Failed, "first try", CorrelationId.NewId()));
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
Thread.Sleep(TimeSpan.FromMilliseconds(600)); // several retry periods
using (var db = direct.CreateDbContext())
db.NodeDeploymentStates.Single(s => s.NodeId == Node).Status.ShouldBe(NodeDeploymentStatus.Applied);
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-2:4053"), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = direct.CreateDbContext();
db.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
/// <summary>
/// Second review L4: a dispatch QUEUED while the load failed for longer than the apply deadline was never sent;
/// it must be dispatched (fresh deadline) once the load succeeds, not closed "Dispatch was lost".
/// </summary>
[Fact]
public void Dispatch_queued_through_an_outage_longer_than_the_deadline_is_sent_not_closed_as_lost()
{
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 deadline = TimeSpan.FromSeconds(3);
var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(
outage, deadline, loadRetryMin: RetryMin, loadRetryMax: RetryMax));
coordinator.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId()));
Thread.Sleep(deadline + TimeSpan.FromSeconds(1));
outage.Down = false;
AwaitAssert(() =>
{
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);
}
/// <summary>
/// Second review L5: an in-flight row recovery cannot interpret (blank RevisionHash on a row it must re-dispatch)
/// used to throw on every load and keep the coordinator Loading forever. It is closed with a reason, and the
/// coordinator serves the next deployment.
/// </summary>
[Fact]
public void Malformed_in_flight_row_is_closed_with_a_reason_and_the_coordinator_carries_on()
{
var dbFactory = NewInMemoryDbFactory();
SeedNodes(dbFactory, Node);
var bad = DeploymentId.NewId();
using (var db = dbFactory.CreateDbContext())
{
db.Deployments.Add(new Deployment
{
DeploymentId = bad.Value, RevisionHash = " ", Status = DeploymentStatus.Dispatching,
CreatedBy = "test", CreatedAtUtc = DateTime.UtcNow,
});
db.SaveChanges();
}
var readiness = new ConfigPublishCoordinatorReadiness();
var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(
dbFactory, NoTimeout, readiness: readiness, loadRetryMin: RetryMin, loadRetryMax: RetryMax));
AwaitAssert(() =>
{
readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready);
using var db = dbFactory.CreateDbContext();
var row = db.Deployments.Single(d => d.DeploymentId == bad.Value);
row.Status.ShouldBe(DeploymentStatus.PartiallyFailed);
row.FailureReason.ShouldNotBeNull();
row.FailureReason.ShouldContain("RevisionHash");
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
var next = SeedDeployment(dbFactory, DeploymentStatus.Dispatching, DateTime.UtcNow.AddSeconds(1));
coordinator.Tell(new DispatchDeployment(next, TestRevision, CorrelationId.NewId()));
coordinator.Tell(new ApplyAck(next, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
using var db = dbFactory.CreateDbContext();
db.Deployments.Single(d => d.DeploymentId == next.Value).Status.ShouldBe(DeploymentStatus.Sealed);
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
/// <summary>Second review L6: an ack-persist backlog is visible on readiness while Ready, and clears.</summary>
[Fact]
public void Readiness_reports_the_ack_persist_backlog()
{
var (_, outage, deploymentId, coordinator, readiness) = DispatchedSingleNode(NoTimeout);
outage.Down = true;
coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId()));
AwaitAssert(() =>
{
readiness.Current.State.ShouldBe(ConfigPublishCoordinatorState.Ready);
readiness.Current.PendingAckPersists.ShouldBe(1);
readiness.Current.LastPersistError.ShouldNotBeNull();
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
outage.Down = false;
AwaitAssert(() =>
{
readiness.Current.PendingAckPersists.ShouldBe(0);
readiness.Current.LastPersistError.ShouldBeNull();
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
/// <summary>A single-node deployment dispatched and AwaitingApplyAcks through an outage-capable factory.</summary>
private (IDbContextFactory<OtOpcUaConfigDbContext> Direct, OutageDbFactory Outage, DeploymentId Id, IActorRef Coordinator,
ConfigPublishCoordinatorReadiness Readiness) DispatchedSingleNode(TimeSpan applyDeadline)
{
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, applyDeadline, 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);
return (direct, outage, deploymentId, coordinator, readiness);
}
[Fact]
public void Readiness_is_NotHosted_after_the_coordinator_stops()
{
@@ -281,15 +490,21 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes
{
private volatile bool _down;
private int _attempts;
private int _failNext;
public bool Down { get => _down; set => _down = value; }
public int Attempts => Volatile.Read(ref _attempts);
/// <summary>The next <paramref name="count"/> context creations throw, then the DB is reachable again.</summary>
public void FailNext(int count) => Volatile.Write(ref _failNext, count);
public OtOpcUaConfigDbContext CreateDbContext()
{
Interlocked.Increment(ref _attempts);
if (_down) throw new InvalidOperationException("ConfigDb unreachable (simulated outage)");
if (Volatile.Read(ref _failNext) > 0 && Interlocked.Decrement(ref _failNext) >= 0)
throw new InvalidOperationException("ConfigDb unreachable (simulated blip)");
return inner.CreateDbContext();
}
}
tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs
+15
View File
@@ -36,6 +36,21 @@ public sealed class ConfigPublishCoordinatorHealthCheckTests
(await Check(r)).Status.ShouldBe(HealthStatus.Healthy);
}
[Fact]
public async Task Ready_with_an_ack_persist_backlog_is_degraded_and_clears()
{
var r = new ConfigPublishCoordinatorReadiness();
r.MarkReady();
r.MarkAckPersistBacklog(2, "timeout");
var result = await Check(r);
result.Status.ShouldBe(HealthStatus.Degraded);
result.Description.ShouldNotBeNull();
result.Description.ShouldContain("timeout");
r.MarkAckPersistBacklog(0, null);
(await Check(r)).Status.ShouldBe(HealthStatus.Healthy);
}
[Fact]
public async Task Not_hosted_and_untracked_are_healthy()
{
tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs
+40 -3
View File
@@ -174,12 +174,49 @@ public sealed class DriverHostActorDegradedAckTests : RuntimeActorTestBase
publish.ExpectNoMsg(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken);
}
[Fact]
public void The_local_row_stays_Applying_until_the_address_space_outcome_is_known()
{
// lmxopcua#868 second review M1: the node used to write its own row Applied/null the moment the revision
// advanced, up to ~105 s before the outcome. ConfigPublishCoordinator's recovery counts a non-Applying row
// as an ack, so a coordinator restart in that window sealed the deployment CLEAN while the node was about
// to report degraded. The row must carry only the final outcome.
var (db, deploymentId) = SeedOneTagDeployment();
var coordinator = CreateTestProbe();
var publish = CreateTestProbe();
var actor = Sys.ActorOf(DriverHostActor.Props(
db, TestNode, coordinator.Ref, localRoles: new HashSet<string> { "driver" },
opcUaPublishActor: publish.Ref));
actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId()));
var rebuild = publish.FishForMessage<OpcUaPublishActor.RebuildAddressSpace>(_ => true, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
coordinator.ExpectNoMsg(TimeSpan.FromMilliseconds(300), TestContext.Current.CancellationToken);
// The ack is held: the row must still read Applying (the apply's synchronous writes are done by now — the
// rebuild Tell above is issued after them on the actor thread).
using (var ctx = db.CreateDbContext())
ctx.NodeDeploymentStates.Single(r => r.NodeId == TestNode.Value && r.DeploymentId == deploymentId.Value)
.Status.ShouldBe(ZB.MOM.WW.OtOpcUa.Configuration.Enums.NodeDeploymentStatus.Applying);
publish.Reply(new OpcUaPublishActor.RebuildCompleted(rebuild.Correlation, rebuild.DeploymentId, false, null));
coordinator.ExpectMsg<ApplyAck>(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken).Degraded.ShouldBeFalse();
AwaitAssert(() =>
{
using var ctx = db.CreateDbContext();
var row = ctx.NodeDeploymentStates.Single(r => r.NodeId == TestNode.Value && r.DeploymentId == deploymentId.Value);
row.Status.ShouldBe(ZB.MOM.WW.OtOpcUa.Configuration.Enums.NodeDeploymentStatus.Applied);
row.FailureReason.ShouldBeNull();
row.AppliedAtUtc.ShouldNotBeNull();
}, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
}
[Fact]
public void A_degraded_apply_records_its_reason_on_the_local_NodeDeploymentState_row()
{
// #601 review — a node WITH a ConfigDb writes its own row Applied/null before the outcome is known.
// Central later adds the reason from the ack, but only when it accepts that ack (a stale/timed-out
// deployment's ack is discarded), so the node must not leave its own row reading clean.
// #601 review — a node WITH a ConfigDb records its own row. Central later adds the reason from the ack,
// but only when it accepts that ack (a stale/timed-out deployment's ack is discarded), so the node must
// not leave its own row reading clean.
var (db, deploymentId) = SeedOneTagDeployment();
var coordinator = CreateTestProbe();
var publish = CreateTestProbe();