diff --git a/docs/Redundancy.md b/docs/Redundancy.md index 71f229959..1349ee036 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -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. 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 c2b6770e1..3e0cb2c17 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -98,9 +98,17 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU private readonly ConfigPublishCoordinatorReadiness? _readiness; private int _failedLoads; - /// Deployments whose dispatch was queued while Loading (only to word the recovery log correctly). + /// Deployments whose dispatch was queued while Loading (never sent: recovery re-dispatches them). private readonly HashSet _queuedDispatches = new(); + /// + /// 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 (). + /// + private readonly Dictionary _unpersisted = new(NodeIdComparer); + private string? _lastPersistError; + /// /// Where central→node commands are handed off (per-cluster mesh Phase 2), or /// to publish on the mediator directly. @@ -230,7 +238,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU { Receive(HandleDispatch); Receive(HandleAck); - Receive(m => HandleAck(m.Ack)); + Receive(HandleRetryAckPersist); Receive(HandleDeadline); Receive(_ => { /* 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(); } + /// + 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); + } + /// 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); + + /// The dispatch. + /// The apply deadline to arm; null = the full . + 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(); + } + + /// + /// 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 and retried on a backoff. + /// + /// when the ack is persisted and counted. + 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(); + /// A retry tick: only the node's CURRENT unpersisted ack is retried; a superseded one is dropped (L3). + 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); + + /// min * 2^(attempt-1), capped at the max — shared by the load and ack-persist retries. + 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)); } /// Seals (all Applied) or marks partially failed — called once every expected ack is in. @@ -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 /// An ack whose persist failed, re-delivered by its retry timer. private sealed record RetryAckPersist(ApplyAck Ack); + /// A received ack not yet persisted, and how many persists of it have failed. + private sealed record PendingPersist(ApplyAck Ack, int Attempts); + + /// Why recovery cannot interpret a row (review L5), or null when it can. + 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); /// The recovery read's result: the newest in-flight deployment, or null when none. @@ -784,9 +900,10 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU /// stub stays tracked so can add to the same UPDATE. /// 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); reads the current one. /// private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry? TrackStub( - OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null) + OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null, bool onlyIfInFlight = false) { var tracked = db.ChangeTracker.Entries().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 diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs index 0074a8eee..f96bab29a 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinatorReadiness.cs @@ -18,8 +18,11 @@ public enum ConfigPublishCoordinatorState /// Consecutive failed recovery loads (0 unless ). /// The last load failure's message, or null. /// When the state was entered. +/// Received acks the coordinator could not persist yet and is retrying. +/// The last ack-persist failure's message, or null. 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); /// /// Thread-safe state bridge from the 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, + }; } + /// The ack-persist backlog changed (review L6): acks received but not yet written, and why. + /// Acks awaiting a persist retry. + /// The last persist failure's message, or null. + public void MarkAckPersistBacklog(int pending, string? lastError) => + _current = _current with { PendingAckPersists = pending, LastPersistError = pending > 0 ? lastError : null }; + /// The recovery state is loaded. - 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, + }; /// The singleton stopped on this node. public void MarkNotHosted() => _current = new(ConfigPublishCoordinatorState.NotHosted, 0, null, DateTime.UtcNow); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs index 579eb660d..94ff59990 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Host/Health/ConfigPublishCoordinatorHealthCheck.cs @@ -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." diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs index 254a7ee21..56ea2ad3b 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -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("outcome", "degraded")); _log.Warning( "DriverHost {Node}: deployment {Id} (rev {Rev}) applied DEGRADED — {Reason}. Acking Applied+degraded; re-dispatching the same revision re-applies it", diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs index ae10d1966..bb3045a67 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs @@ -116,8 +116,9 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes } /// - /// 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 DriverHostActor 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. /// @@ -214,6 +215,214 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); } + /// + /// 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. + /// + [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); + } + + /// + /// 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. + /// + [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); + } + + /// + /// 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). + /// + [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); + } + + /// + /// 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". + /// + [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); + } + + /// + /// 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. + /// + [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); + } + + /// Second review L6: an ack-persist backlog is visible on readiness while Ready, and clears. + [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); + } + + /// A single-node deployment dispatched and AwaitingApplyAcks through an outage-capable factory. + private (IDbContextFactory 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); + /// The next context creations throw, then the DB is reachable again. + 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(); } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs index a9202432c..560028524 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Host.Tests/Health/ConfigPublishCoordinatorHealthCheckTests.cs @@ -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() { diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs index 7af4f16e1..99f261ee1 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorDegradedAckTests.cs @@ -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 { "driver" }, + opcUaPublishActor: publish.Ref)); + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + var rebuild = publish.FishForMessage(_ => 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(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();