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

Merge #868 second review fixes: node row stays Applying until outcome, pending ack persists survive deadline/restart, write-once terminal states

Browse Source 
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
2026-11-11 12:54:47 -05:00
parent ff85b8511a 494dcb5b87
commit 74496b4fec
8 changed files with 475 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/Redundancy.md
+5
View File
@@ -676,6 +676,11 @@ deployment green fleet-wide. Now:
no second rebuild), reporting clean only if that pass is clean.
- **A node with a ConfigDb also writes the reason on its own row** (`ApplyAck.DegradedReasonPrefix`), because
central only records it when it *accepts* the ack — a timed-out deployment's late ack is discarded.
**That row stays `Applying` until the outcome is known** (lmxopcua#868): it used to be written `Applied`/clean
the moment the revision advanced, and `ConfigPublishCoordinator`'s recovery counts any non-`Applying` row as
an ack, so a coordinator restart inside the ~105 s outcome window sealed the deployment clean while the node
was about to report degraded. `/fleet` therefore shows the node *Applying* until its outcome lands; a restart
of the node in that window replays the orphan `Applying` row (re-applies) instead of restoring it.
- **Mixed-version fleets.** An old node never sets the flag, so its acks are treated exactly as before
(clean). A new node acking an old central: the old coordinator ignores the unknown field and seals as it
always did — no worse than today.
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs
+160 -39
View File
@@ -98,9 +98,17 @@ 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>
/// <summary>Deployments whose dispatch was queued while Loading (never sent: recovery re-dispatches them).</summary>
private readonly HashSet<DeploymentId> _queuedDispatches = new();
/// <summary>
/// Acks of the current deployment that arrived but could not be persisted yet (review M2), one per node — the
/// newest wins. They are retried on a backoff, block the deadline from deciding without them, and are
/// re-sent to Self across a restart (<see cref="PreRestart"/>).
/// </summary>
private readonly Dictionary<NodeId, PendingPersist> _unpersisted = new(NodeIdComparer);
private string? _lastPersistError;
/// <summary>
/// Where central→node commands are handed off (per-cluster mesh Phase 2), or
/// <see langword="null"/> to publish on the mediator directly.
@@ -230,7 +238,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
{
Receive<DispatchDeployment>(HandleDispatch);
Receive<ApplyAck>(HandleAck);
Receive<RetryAckPersist>(m => HandleAck(m.Ack));
Receive<RetryAckPersist>(HandleRetryAckPersist);
Receive<DeadlineElapsed>(HandleDeadline);
Receive<SubscribeAck>(_ => { /* DPS subscribe confirmation */ });
// A late load outcome / retry tick cannot normally arrive once Ready; ignored, never stashed.
@@ -252,6 +260,15 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
StartRecoveryLoad();
}
/// <inheritdoc />
protected override void PreRestart(Exception reason, object message)
{
// Timers die with this incarnation; the mailbox survives. Re-send every ack that is still unpersisted so the
// next incarnation (which stashes while Loading) persists and counts it — its node will not resend (M2).
foreach (var pending in _unpersisted.Values) Self.Tell(pending.Ack);
base.PreRestart(reason, message);
}
/// <inheritdoc />
protected override void PostStop()
{
@@ -302,8 +319,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
{
_failedLoads++;
// Exponential backoff, capped: min * 2^(n-1), with the shift bounded so it never overflows.
var factor = 1L << Math.Min(_failedLoads - 1, 20);
var delay = TimeSpan.FromTicks(Math.Min(_loadRetryMax.Ticks, _loadRetryMin.Ticks * factor));
var delay = Backoff(_failedLoads);
_readiness?.MarkLoading(_failedLoads, msg.Error.Message);
_log.Warning(msg.Error,
"ConfigPublishCoordinator could not load its recovery state from the ConfigDb (attempt {Attempt}); " +
@@ -329,11 +345,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
catch (Exception ex)
{
_log.Error(ex, "ConfigPublishCoordinator failed to apply its recovered state; retrying the load");
Timers.Cancel(DeadlineTimerKey);
_current = null;
_expectedAcks.Clear();
_acks.Clear();
_degraded.Clear();
ResetForNext();
HandleRecoveryLoadFailed(new RecoveryLoadFailed(ex));
return;
}
@@ -359,6 +371,21 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
_current = new DeploymentId(inflight.DeploymentId);
var nodeStates = inflight.NodeStates;
// Review L5: a row recovery cannot interpret (blank revision on a row it must re-dispatch, blank node id) would
// throw on every load and keep the coordinator Loading forever. Close it with a reason and carry on.
var malformed = DescribeMalformed(inflight);
if (malformed is not null)
{
_log.Error("Coordinator cannot recover deployment {Id}: {Problem}. Closing it as PartiallyFailed", _current, malformed);
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed, onlyIfInFlight: true);
SetFailureReason(db, Truncate(
$"The deploy coordinator could not recover this deployment: {malformed}. Redeploy."));
db.SaveChanges();
ResetForNext();
return;
}
// 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).
@@ -377,21 +404,27 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
// 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)
if (_queuedDispatches.Contains(_current.Value))
{
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);
// Review L4: its dispatch was QUEUED here while the load was failing — it was never sent, so however
// long the outage lasted it is not a lost dispatch. Send it now with a full, fresh deadline.
_log.Info(
"Coordinator dispatching deployment {Id}, queued while its recovery state was loading",
_current);
HandleDispatch(new DispatchDeployment(
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()));
}
else 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);
// The REMAINING time (review L4): CreatedAtUtc never moves, so a re-drive must not restart the clock.
HandleDispatch(new DispatchDeployment(
_current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId()), remaining);
}
else
{
// Deliberately INLINE, unlike the populated-set deadline below: with no expected set no queued ack
@@ -413,7 +446,8 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
}
// 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
// DriverHostActor row — which that actor writes Applied only once the address-space outcome is known, so a
// non-Applying row is always a FINAL outcome; review M1): 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)
{
@@ -434,7 +468,11 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
_current, _acks.Count, _expectedAcks.Count);
}
private void HandleDispatch(DispatchDeployment msg)
private void HandleDispatch(DispatchDeployment msg) => HandleDispatch(msg, deadline: null);
/// <param name="msg">The dispatch.</param>
/// <param name="deadline">The apply deadline to arm; null = the full <see cref="_applyDeadline"/>.</param>
private void HandleDispatch(DispatchDeployment msg, TimeSpan? deadline)
{
// A duplicate of the deployment already in flight with a seeded expected set (a dispatch queued while the
// recovery load ran, for the very row that load re-drove) must not reset the acks already gathered.
@@ -529,7 +567,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
// directly, so this actor no longer knows which transport is in force.
if (_meshRouter is not null) _meshRouter.Tell(new MeshCommand(DeploymentsTopic, msg));
else DistributedPubSub.Get(Context.System).Mediator.Tell(new Publish(DeploymentsTopic, msg));
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(msg.DeploymentId), _applyDeadline);
Timers.StartSingleTimer(DeadlineTimerKey, new DeadlineElapsed(msg.DeploymentId), deadline ?? _applyDeadline);
if (_expectedAcks.Count == 0)
{
@@ -567,26 +605,68 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
// 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.
if (!TryPersistAck(msg)) return;
if (_acks.Count < _expectedAcks.Count) return;
CompleteFromAcks();
}
/// <summary>
/// Persists an ack and, only on success, counts it (and drops any older unpersisted ack of that node, with its
/// retry timer — review L3). On failure the ack is held in <see cref="_unpersisted"/> and retried on a backoff.
/// </summary>
/// <returns><see langword="true"/> when the ack is persisted and counted.</returns>
private bool TryPersistAck(ApplyAck msg)
{
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;
var attempts = _unpersisted.TryGetValue(msg.NodeId, out var prior) ? prior.Attempts + 1 : 1;
_unpersisted[msg.NodeId] = new PendingPersist(msg, attempts);
_lastPersistError = ex.Message;
var delay = Backoff(attempts);
// Review L7: the stack trace once, then one line per retry.
if (attempts == 1)
_log.Warning(ex, "Could not persist the ApplyAck from {Node} for {Id}; retrying in {Delay}",
msg.NodeId, msg.DeploymentId, delay);
else
_log.Warning("Still cannot persist the ApplyAck from {Node} for {Id} (attempt {Attempt}: {Error}); retrying in {Delay}",
msg.NodeId, msg.DeploymentId, attempts, ex.Message, delay);
Timers.StartSingleTimer(AckRetryTimerKey(msg.NodeId), new RetryAckPersist(msg), delay);
ReportPersistBacklog();
return false;
}
if (_unpersisted.Remove(msg.NodeId))
{
Timers.Cancel(AckRetryTimerKey(msg.NodeId));
if (_unpersisted.Count == 0) _lastPersistError = null;
ReportPersistBacklog();
}
_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);
return true;
}
if (_acks.Count < _expectedAcks.Count) return;
CompleteFromAcks();
/// <summary>A retry tick: only the node's CURRENT unpersisted ack is retried; a superseded one is dropped (L3).</summary>
private void HandleRetryAckPersist(RetryAckPersist msg)
{
if (!_unpersisted.TryGetValue(msg.Ack.NodeId, out var pending) || !ReferenceEquals(pending.Ack, msg.Ack)) return;
HandleAck(msg.Ack);
}
private void ReportPersistBacklog() => _readiness?.MarkAckPersistBacklog(_unpersisted.Count, _lastPersistError);
/// <summary>min * 2^(attempt-1), capped at the max — shared by the load and ack-persist retries.</summary>
private TimeSpan Backoff(int attempt)
{
var factor = 1L << Math.Min(Math.Max(attempt, 1) - 1, 20);
return TimeSpan.FromTicks(Math.Min(_loadRetryMax.Ticks, _loadRetryMin.Ticks * factor));
}
/// <summary>Seals (all Applied) or marks partially failed — called once every expected ack is in.</summary>
@@ -598,7 +678,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
MarkPartiallyFailed();
}
private static string AckRetryTimerKey(ApplyAck msg) => $"ack-retry:{msg.DeploymentId.Value:N}:{msg.NodeId.Value}";
private static string AckRetryTimerKey(NodeId node) => $"ack-retry:{node.Value.ToUpperInvariant()}";
private void PersistNodeAck(ApplyAck msg)
{
@@ -634,7 +714,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
{
if (_current is null) return;
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.Sealed, sealNow: true, onlyIfInFlight: true);
if (_degraded.Count > 0)
{
// Still Sealed: every node DID apply this revision and is running it, and "latest sealed" is
@@ -663,7 +743,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
{
if (_current is null) return;
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.PartiallyFailed, onlyIfInFlight: true);
db.SaveChanges();
_log.Warning("Deployment {Id} partially failed; acks={Acks}", _current.Value,
string.Join(",", _acks.Select(kv => $"{kv.Key.Value}={kv.Value}")));
@@ -678,6 +758,22 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
msg.DeploymentId, _current);
return;
}
// Review M2: an ack that ARRIVED but is not yet persisted must never be timed out as "no ack from" its node.
// Persist the backlog now; if the DB still refuses, defer the decision (the terminal write needs it too).
if (_unpersisted.Count > 0)
{
foreach (var pending in _unpersisted.Values.ToList()) TryPersistAck(pending.Ack);
if (_unpersisted.Count > 0)
{
var delay = Backoff(_unpersisted.Values.Min(p => p.Attempts));
_log.Warning(
"Deployment {Id} reached its apply deadline with {Count} received ack(s) not yet persisted; deferring " +
"the outcome {Delay} until they are", _current.Value, _unpersisted.Count, delay);
Timers.StartSingleTimer(DeadlineTimerKey, msg, delay);
return;
}
}
// 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)
@@ -689,7 +785,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
}
using var db = _dbFactory.CreateDbContext();
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut);
UpdateDeploymentStatus(db, _current.Value, DeploymentStatus.TimedOut, onlyIfInFlight: true);
if (_expectedAcks.Count == 0)
{
SetFailureReason(db, Truncate(
@@ -745,6 +841,13 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
private void ResetForNext()
{
Timers.Cancel(DeadlineTimerKey);
foreach (var node in _unpersisted.Keys) Timers.Cancel(AckRetryTimerKey(node));
if (_unpersisted.Count > 0)
{
_unpersisted.Clear();
_lastPersistError = null;
ReportPersistBacklog();
}
_current = null;
_expectedAcks.Clear();
_acks.Clear();
@@ -757,6 +860,19 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
/// <summary>An ack whose persist failed, re-delivered by its retry timer.</summary>
private sealed record RetryAckPersist(ApplyAck Ack);
/// <summary>A received ack not yet persisted, and how many persists of it have failed.</summary>
private sealed record PendingPersist(ApplyAck Ack, int Attempts);
/// <summary>Why recovery cannot interpret a row (review L5), or null when it can.</summary>
private static string? DescribeMalformed(RecoveredDeployment d)
{
if (d.NodeStates.Count == 0 && string.IsNullOrWhiteSpace(d.RevisionHash))
return "its RevisionHash is blank, so it cannot be re-dispatched";
if (d.NodeStates.Any(n => string.IsNullOrWhiteSpace(n.NodeId)))
return "one of its NodeDeploymentState rows has a blank NodeId";
return null;
}
public sealed record DeadlineElapsed(DeploymentId DeploymentId);
/// <summary>The recovery read's result: the newest in-flight deployment, or null when none.</summary>
@@ -784,9 +900,10 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
/// stub stays tracked so <see cref="SetFailureReason"/> can add to the same UPDATE.
/// </summary>
private static void UpdateDeploymentStatus(
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null)
OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null,
bool onlyIfInFlight = false)
{
var entry = TrackStub(db, id, rowVersion);
var entry = TrackStub(db, id, rowVersion, onlyIfInFlight);
if (entry is null) return;
entry.Entity.Status = status;
entry.Property(x => x.Status).IsModified = true;
@@ -801,7 +918,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
private void SetFailureReason(OtOpcUaConfigDbContext db, string reason)
{
if (_current is null) return;
var entry = TrackStub(db, _current.Value);
var entry = TrackStub(db, _current.Value, onlyIfInFlight: true);
if (entry is null) return;
entry.Entity.FailureReason = reason;
entry.Property(x => x.FailureReason).IsModified = true;
@@ -815,7 +932,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
/// guard); <see langword="null"/> reads the current one.
/// </param>
private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry<Deployment>? TrackStub(
OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null)
OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null, bool onlyIfInFlight = false)
{
var tracked = db.ChangeTracker.Entries<Deployment>().FirstOrDefault(e => e.Entity.DeploymentId == id.Value);
if (tracked is not null) return tracked;
@@ -824,9 +941,13 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
var stamp = db.Deployments
.AsNoTracking()
.Where(x => x.DeploymentId == id.Value)
.Select(x => new { x.RowVersion })
.Select(x => new { x.RowVersion, x.Status })
.FirstOrDefault();
if (stamp is null) return null;
// Review L8: a terminal outcome is written once. A row something else already closed (AdminOps' lost-
// dispatch closure, a previous incarnation) is left as it is; the RowVersion check covers the race.
if (onlyIfInFlight && stamp.Status is not (DeploymentStatus.Dispatching or DeploymentStatus.AwaitingApplyAcks))
return null;
rowVersion = stamp.RowVersion;
}
return db.Deployments.Attach(new Deployment
src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs
+18 -3
View File
@@ -18,8 +18,11 @@ public enum ConfigPublishCoordinatorState
/// <param name="FailedLoadAttempts">Consecutive failed recovery loads (0 unless <see cref="ConfigPublishCoordinatorState.Loading"/>).</param>
/// <param name="LastError">The last load failure's message, or null.</param>
/// <param name="SinceUtc">When the state was entered.</param>
/// <param name="PendingAckPersists">Received acks the coordinator could not persist yet and is retrying.</param>
/// <param name="LastPersistError">The last ack-persist failure's message, or null.</param>
public sealed record ConfigPublishCoordinatorSnapshot(
ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc);
ConfigPublishCoordinatorState State, int FailedLoadAttempts, string? LastError, DateTime SinceUtc,
int PendingAckPersists = 0, string? LastPersistError = null);
/// <summary>
/// Thread-safe state bridge from the <see cref="ConfigPublishCoordinator"/> actor to the node's health endpoint
@@ -39,11 +42,23 @@ public sealed class ConfigPublishCoordinatorReadiness
public void MarkLoading(int failedAttempts, string? lastError)
{
var since = _current.State == ConfigPublishCoordinatorState.Loading ? _current.SinceUtc : DateTime.UtcNow;
_current = new(ConfigPublishCoordinatorState.Loading, failedAttempts, lastError, since);
_current = _current with
{
State = ConfigPublishCoordinatorState.Loading, FailedLoadAttempts = failedAttempts, LastError = lastError, SinceUtc = since,
};
}
/// <summary>The ack-persist backlog changed (review L6): acks received but not yet written, and why.</summary>
/// <param name="pending">Acks awaiting a persist retry.</param>
/// <param name="lastError">The last persist failure's message, or null.</param>
public void MarkAckPersistBacklog(int pending, string? lastError) =>
_current = _current with { PendingAckPersists = pending, LastPersistError = pending > 0 ? lastError : null };
/// <summary>The recovery state is loaded.</summary>
public void MarkReady() => _current = new(ConfigPublishCoordinatorState.Ready, 0, null, DateTime.UtcNow);
public void MarkReady() => _current = _current with
{
State = ConfigPublishCoordinatorState.Ready, FailedLoadAttempts = 0, LastError = null, SinceUtc = DateTime.UtcNow,
};
/// <summary>The singleton stopped on this node.</summary>
public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow);
src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs
+6
View File
@@ -29,9 +29,15 @@ public sealed class ConfigPublishCoordinatorHealthCheck(ConfigPublishCoordinator
["failedLoadAttempts"] = s.FailedLoadAttempts,
};
if (s.LastError is not null) data["lastError"] = s.LastError;
data["pendingAckPersists"] = s.PendingAckPersists;
if (s.LastPersistError is not null) data["lastPersistError"] = s.LastPersistError;
return Task.FromResult(s.State switch
{
ConfigPublishCoordinatorState.Ready when s.PendingAckPersists > 0 => HealthCheckResult.Degraded(
$"Deploy coordinator cannot persist {s.PendingAckPersists} received deploy ack(s) to the ConfigDb " +
$"(last: {s.LastPersistError}); retrying with backoff. The deployment's outcome waits for them.",
data: data),
ConfigPublishCoordinatorState.Loading => HealthCheckResult.Degraded(
s.FailedLoadAttempts == 0
? "Deploy coordinator is loading its recovery state from the ConfigDb; deployments are queued until it completes."
src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs
+14 -6
View File
@@ -2984,7 +2984,13 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
}
_currentRevision = msg.Revision;
UpsertNodeDeploymentState(msg.DeploymentId, NodeDeploymentStatus.Applied, failureReason: null);
// lmxopcua#868 review M1: the own row stays APPLYING (written by BeginDirectApply) until the
// address-space outcome is known — AwaitAddressSpaceOutcomeThenAck / ReleasePendingAck write the
// final Applied (clean, or with the degraded reason). Writing Applied/null here, up to ~105 s before
// that outcome, made the row read as a CLEAN ack: a coordinator restart in that window recovered it
// as complete and sealed the deployment clean while this node was about to report degraded (#601).
// A restart of THIS node in the window now replays the orphan Applying row (re-applies), which is
// the honest answer to "outcome unknown".
// The Applied ack is NOT sent here any more. It used to go out right now, before the
// address-space apply below had even been dequeued, so a degraded apply (swallowed sink faults,
// _lastApplied not advanced) still sealed the deployment green fleet-wide. The ack is now held
@@ -3069,6 +3075,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (_opcUaPublishActor is null)
{
_currentRevisionDegraded = false;
UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applied, failureReason: null);
SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation);
return;
}
@@ -3126,13 +3133,14 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
if (_currentRevision is { } cur && cur == pending.Revision)
_currentRevisionDegraded = degraded;
// The own row has been Applying since the apply began (lmxopcua#868 review M1); the FINAL outcome is
// written here, clean or degraded. Central adds the reason from the ack only when it ACCEPTS the ack (a
// timed-out or superseded deployment's ack is discarded), so a node with a ConfigDb records it itself; a
// driver-only node no-ops here.
UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied,
degraded ? ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply degraded") : null);
if (degraded)
{
// The apply wrote this node's own row Applied/null before the outcome was known. Central
// adds the reason from the ack only when it ACCEPTS the ack (a timed-out or superseded deployment's ack
// is discarded), so a node with a ConfigDb records it itself; a driver-only node no-ops here.
UpsertNodeDeploymentState(pending.DeploymentId, NodeDeploymentStatus.Applied,
ApplyAck.DegradedReasonPrefix + (reason ?? "address-space apply degraded"));
OtOpcUaTelemetry.DeploymentApplied.Add(1, new KeyValuePair<string, object?>("outcome", "degraded"));
_log.Warning(
"DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it",
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();