From ec3b835e4a1a93ffe9db7e7c745c413af3f175c1 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 11 Nov 2026 13:31:49 -0500 Subject: [PATCH] =?UTF-8?q?fix(controlplane,runtime):=20#868=20fourth=20re?= =?UTF-8?q?view=20=E2=80=94=20SQL=20poison=20allowlist,=20memory-counted?= =?UTF-8?q?=20acks=20survive=20restart,=20adopted=20revisions=20unconfirme?= =?UTF-8?q?d?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit M1 IsPoison classified every SQL Server transient as poison: Microsoft.Data.SqlClient never sets DbException.IsTransient, so a deadlock (1205), timeout (-2 / Win32 258), dropped connection or AG failover was counted from memory and the node's row left Applying. Poison is now an ALLOWLIST: a DbUpdateException (not concurrency) carrying a SqlException whose number is 8152, 2628, 547, 515, 2627 or 2601. Everything else retries, bounded by the deadline deferral cap. IsPoison is public for tests. Acks counted only from memory (poison branch and the deferral cap) are kept in _memoryOnlyAcks and re-sent by PreRestart alongside _unpersisted, so a restart before the terminal write does not lose them. M2 Revisions adopted without an outcome-reporting apply are unconfirmed, so a same-revision dispatch re-applies instead of short-circuiting a clean Applied over an empty address space: TryRecoverFromStale (restores only the revision marker) and RestoreApplied when the artifact read fails (null blob, LoadArtifactBlob swallows the error) or the restore throws. L1 The re-drive comment documents the CreatedAtUtc recompute after a second restart / failover. L2 _currentRevisionDegraded renamed _currentRevisionUnconfirmed; the stale TryBootFromCache remark corrected. L3 docs/Redundancy.md: a Direct node running from cache that still cannot reach the ConfigDb acks a same-revision re-dispatch Failed (PartiallyFailed, not Sealed); Stale recovery and failed restores are unconfirmed too. Test harness: SqlExceptionFactory builds real SqlExceptions (reflection on SqlError / SqlErrorCollection / SqlException.CreateException); ConstraintEnforcingDbFactory now throws DbUpdateException over SQL 2628 (over-length), 547 (poison) and 1205 (transient), plus FailNextContexts. Tests (all fail on the previous code): poison allowlist theory (1205, 40613, 10054, 4060 not poison) + timeout/concurrency/providerless not poison; a deadlocked ack write is retried and recorded; a memory-counted ack survives a coordinator restart; Stale-recovered and failed-restore revisions re-apply on re-dispatch. Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk --- docs/Redundancy.md | 8 +- .../Coordinators/ConfigPublishCoordinator.cs | 51 ++++++-- .../Drivers/DriverHostActor.cs | 37 ++++-- .../ConfigPublishCoordinatorDbOutageTests.cs | 91 ++++++++++++++ .../Harness/ConstraintEnforcingDbFactory.cs | 30 +++-- .../Harness/SqlExceptionFactory.cs | 63 ++++++++++ ...DriverHostActorUnconfirmedRevisionTests.cs | 117 ++++++++++++++++++ 7 files changed, 367 insertions(+), 30 deletions(-) create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/SqlExceptionFactory.cs create mode 100644 tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnconfirmedRevisionTests.cs diff --git a/docs/Redundancy.md b/docs/Redundancy.md index 150c15cb5..db8b6f525 100644 --- a/docs/Redundancy.md +++ b/docs/Redundancy.md @@ -687,7 +687,13 @@ deployment green fleet-wide. Now: whose ConfigDb is unreachable) restores the cached revision — the cache is written before the outcome — and sends **no ack**, so central times that deployment out even though the node runs it. The restored revision is marked *unconfirmed*, so **re-deploying it re-applies** and acks the real outcome rather than - short-circuiting a clean `Applied`. + short-circuiting a clean `Applied`. ⚠️ A `Direct` node running from cache that **still cannot reach the + ConfigDb** cannot read the artifact to re-apply, so it acks that same-revision re-dispatch **`Failed`** and + the deployment reads **`PartiallyFailed`**, not `Sealed` — the honest answer; re-deploy once the node reaches + the ConfigDb again. + - The same *unconfirmed* marking applies to a revision adopted by **Stale recovery** (which restores only the + revision marker, not drivers or the address space) and to one whose **bootstrap restore failed**: a dispatch + of it re-applies instead of acking clean over an empty address space. - **Upgrade note (lmxopcua#868): do not deploy while the central pair is mid rolling-upgrade across this change.** An old-build fused central node still writes its own row `Applied` before the outcome, which a new-build coordinator's recovery would count as a clean ack. Finish upgrading both central nodes, then deploy. 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 898c6e282..b508feb84 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -109,6 +109,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU private readonly Dictionary _unpersisted = new(NodeIdComparer); private string? _lastPersistError; + /// + /// Acks counted without a persisted row (a rejected write, or the deferral cap) — re-sent across a restart, since + /// no row records them (fourth review M1). Dropped once a later persist of that node succeeds. + /// + private readonly Dictionary _memoryOnlyAcks = new(NodeIdComparer); + /// When the current deadline was first deferred on an unpersisted ack (bounds the deferral). private DateTime? _deadlineDeferredSinceUtc; @@ -272,6 +278,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU // 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); + // Fourth review M1: so are acks counted only FROM MEMORY (rejected writes, the deferral cap) — the outcome they + // fed has not been written yet, and without this the next incarnation would wait on their nodes forever. + foreach (var ack in _memoryOnlyAcks.Values) Self.Tell(ack); base.PreRestart(reason, message); } @@ -430,8 +439,11 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU // A FULL deadline (third review #3, superseding the second review's "remaining time"): a node needs its // whole outcome timeout (DeployTimings.ApplyDeadline - ApplyOutcomeSafetyMargin) plus that margin to // report — i.e. the full ApplyDeadline — so ANY remaining time shorter than that let the nodes apply - // and the deployment time out moments later. Bounded anyway: there is no dispatched-at column, so a - // SECOND restart recomputes from CreatedAtUtc and, past the deadline by then, closes the row instead. + // and the deployment time out moments later. Bounded anyway: there is no dispatched-at column, so the + // fresh deadline lives only in this incarnation's timer. After a SECOND restart or singleton failover, + // recovery recomputes "remaining" from Deployment.CreatedAtUtc again — by then usually past the + // deadline — so the row is no longer re-driven: if its expected set was seeded it is timed out from the + // seeded rows (acks that landed are kept), otherwise it is closed TimedOut as a lost dispatch. HandleDispatch(new DispatchDeployment( _current.Value, RevisionHash.Parse(inflight.RevisionHash), CorrelationId.NewId())); } @@ -637,6 +649,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU "counting it from memory. Its NodeDeploymentState row stays unrecorded", msg.NodeId, msg.DeploymentId, msg.Outcome); DropUnpersisted(msg.NodeId); + _memoryOnlyAcks[msg.NodeId] = msg; CountAck(msg); return true; } @@ -659,6 +672,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU } DropUnpersisted(msg.NodeId); + _memoryOnlyAcks.Remove(msg.NodeId); CountAck(msg); return true; } @@ -680,21 +694,34 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU } /// - /// A non-transient rejection of the write itself: an EF (not a concurrency - /// conflict) with no transient database error, timeout or socket failure underneath. Connectivity failures - /// surface as other exception types (or as transient s) and stay - /// retryable. + /// SQL Server errors that reject the WRITE ITSELF and fail identically on every retry (fourth review M1): string + /// or binary data truncated (8152, 2628), FK/CHECK conflict (547), NULL into a NOT NULL column (515), unique + /// key/index violation (2627, 2601). /// - internal static bool IsPoison(Exception ex) + private static readonly int[] PoisonSqlNumbers = [8152, 2628, 547, 515, 2627, 2601]; + + /// + /// Whether an ack write failure is POISON — never retried — by an ALLOWLIST: an EF + /// (not a concurrency conflict) carrying a whose error + /// numbers include one of . Everything else — deadlocks (1205), timeouts (-2, + /// Win32 258), dropped connections, failovers, unknown providers — RETRIES, bounded by the deadline deferral + /// cap. (Microsoft.Data.SqlClient does not set DbException.IsTransient, so a deny-list on it classified + /// every SQL Server transient as poison.) + /// + /// The failure. + /// only for a known, deterministic rejection. + public static bool IsPoison(Exception ex) { if (ex is not DbUpdateException || ex is DbUpdateConcurrencyException) return false; for (var e = ex.InnerException; e is not null; e = e.InnerException) { - if (e is System.Data.Common.DbException { IsTransient: true } - or TimeoutException or System.Net.Sockets.SocketException or System.IO.IOException) - return false; + if (e is not Microsoft.Data.SqlClient.SqlException sql) continue; + if (Array.IndexOf(PoisonSqlNumbers, sql.Number) >= 0) return true; + foreach (Microsoft.Data.SqlClient.SqlError error in sql.Errors) + if (Array.IndexOf(PoisonSqlNumbers, error.Number) >= 0) return true; + return false; } - return true; + return false; } /// A retry tick: only the node's CURRENT unpersisted ack is retried; a superseded one is dropped (L3). @@ -818,6 +845,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU foreach (var pending in _unpersisted.Values.ToList()) { DropUnpersisted(pending.Ack.NodeId); + _memoryOnlyAcks[pending.Ack.NodeId] = pending.Ack; CountAck(pending.Ack); } } @@ -907,6 +935,7 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU _lastPersistError = null; ReportPersistBacklog(); } + _memoryOnlyAcks.Clear(); _current = null; _expectedAcks.Clear(); _acks.Clear(); 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 b16f68634..039128cf6 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Runtime/Drivers/DriverHostActor.cs @@ -290,15 +290,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS private RevisionHash? _currentRevision; /// - /// The last address-space apply of reported DEGRADED (or no - /// outcome in time). A re-dispatch of that same revision then RE-APPLIES instead of taking the + /// is not known to be cleanly served: its last address-space apply reported + /// DEGRADED (or no outcome in time), or it was ADOPTED without an outcome-reporting apply (cache boot, Stale + /// recovery, a failed bootstrap restore — lmxopcua#868). A re-dispatch of that same revision then RE-APPLIES instead of taking the /// idempotent "already at this revision" short-circuit: the short-circuit would ack a clean Applied /// for an address space known to be partial, and the re-apply is what repairs it (a degraded /// non-rebuild apply leaves the publish actor's composition un-advanced, so it re-diffs the missing /// nodes). Reset by every clean outcome and by the ConfigDb boot path (an Applied row is a final outcome); /// SET by the cache boot paths, whose cached revision may never have reported an outcome (lmxopcua#868). /// - private bool _currentRevisionDegraded; + private bool _currentRevisionUnconfirmed; /// Applied acks held until the address-space apply reports its outcome, keyed by /// the apply's correlation id (echoed back in ). @@ -1438,7 +1439,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Unconfirmed (lmxopcua#868 third review #2): the cache is written BEFORE the address-space outcome, so a // restart in that window boots a revision central never got an ack for. Marking it unconfirmed makes a // re-dispatch of it RE-APPLY (and ack the real outcome) instead of short-circuiting a clean Applied. - _currentRevisionDegraded = true; + _currentRevisionUnconfirmed = true; BecomeSteady(); _log.Info( @@ -1475,9 +1476,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS /// /// /// This does not make the node current. _currentRevision is set from the - /// cached artifact, so a subsequent dispatch of that same revision correctly no-ops, - /// while any NEW revision still requires central — the cache holds the past, not the - /// future. Dispatch handling is unchanged. + /// cached artifact but marked UNCONFIRMED (, lmxopcua#868): the + /// cache is written before the address-space outcome, so central may never have had an ack for it. A + /// re-dispatch of that same revision therefore RE-APPLIES — which, while the ConfigDb is still unreachable, + /// acks Failed (the deployment reads PartiallyFailed, not Sealed) — and any NEW revision still requires + /// central: the cache holds the past, not the future. /// /// /// Never throws: a fault here must degrade to Stale, which is exactly where the node @@ -1520,7 +1523,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // matching how a normally-booted node behaves. _currentRevision = revision; // Unconfirmed, as in BootstrapFromCache (third review #2): a re-dispatch of it re-applies. - _currentRevisionDegraded = true; + _currentRevisionUnconfirmed = true; BecomeSteady(); _isRunningFromCache = true; @@ -2817,7 +2820,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // The short-circuit claims a CLEAN Applied, so it is only honest while the current revision's // address space is known-good — not after a degraded apply, and not while its outcome is still // pending. Either way the dispatch re-applies instead, which is also what repairs a degraded apply. - var currentRevisionUnconfirmed = _currentRevisionDegraded + var currentRevisionUnconfirmed = _currentRevisionUnconfirmed || _pendingAcks.Values.Any(p => p.Revision == msg.RevisionHash); if (_currentRevision is { } cur && cur == msg.RevisionHash && !currentRevisionUnconfirmed) { @@ -3081,7 +3084,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS { if (_opcUaPublishActor is null) { - _currentRevisionDegraded = false; + _currentRevisionUnconfirmed = false; UpsertNodeDeploymentState(deploymentId, NodeDeploymentStatus.Applied, failureReason: null); SendAck(deploymentId, ApplyAckOutcome.Applied, failureReason: null, correlation); return; @@ -3138,7 +3141,7 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS // Only the CURRENT revision's flag is ours to set — an older apply's late outcome must not mark a // newer revision degraded (or clean). if (_currentRevision is { } cur && cur == pending.Revision) - _currentRevisionDegraded = degraded; + _currentRevisionUnconfirmed = 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 @@ -3593,6 +3596,10 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS try { var appliedBlob = ReconcileDrivers(deploymentId); + // Null = the artifact could not be read (LoadArtifactBlob logs and swallows the failure), so the drivers + // were NOT restored: the adopted revision is unconfirmed (fourth review M2), and a re-dispatch of it + // re-applies instead of short-circuiting a clean Applied. + if (appliedBlob is null) _currentRevisionUnconfirmed = true; _opcUaPublishActor?.Tell(new ZB.MOM.WW.OtOpcUa.Runtime.OpcUa.OpcUaPublishActor.RebuildAddressSpace(correlation, deploymentId, appliedBlob)); PushDesiredSubscriptions(deploymentId); // Populate the node-local cache from the artifact this restored node is now serving. The @@ -3609,6 +3616,9 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS } catch (Exception ex) { + // The revision was adopted from the row but its served state was NOT restored (fourth review M2): mark it + // unconfirmed so a re-dispatch of it re-applies instead of short-circuiting a clean Applied. + _currentRevisionUnconfirmed = true; _log.Warning(ex, "DriverHost {Node}: failed to restore served state for {Id} on bootstrap", _localNode, deploymentId); } } @@ -4330,6 +4340,11 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS if (latestSealed is not null) { _currentRevision = RevisionHash.Parse(latestSealed.RevisionHash); + // Adopted, NOT applied (fourth review M2): Stale recovery restores only the revision marker — drivers + // and the address space stay empty until a dispatch (see PushDesiredSubscriptions' note). Unconfirmed, + // so a dispatch of this same revision re-applies rather than short-circuiting a clean Applied over an + // empty address space. + _currentRevisionUnconfirmed = true; UpsertNodeDeploymentState(new DeploymentId(latestSealed.DeploymentId), NodeDeploymentStatus.Applied, failureReason: null); } 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 202969bc7..ae2aa5bb4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorDbOutageTests.cs @@ -495,6 +495,97 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); } + /// + /// Fourth review M1: poison is an ALLOWLIST of SQL Server rejections. Microsoft.Data.SqlClient never sets + /// DbException.IsTransient, so the old deny-list called every SQL Server transient poison. + /// + [Theory] + [InlineData(2628, true)] + [InlineData(8152, true)] + [InlineData(547, true)] + [InlineData(515, true)] + [InlineData(2627, true)] + [InlineData(2601, true)] + [InlineData(1205, false)] // deadlock victim + [InlineData(40613, false)] // database unavailable (failover) + [InlineData(10054, false)] // connection forcibly closed + [InlineData(4060, false)] // cannot open database + public void Only_known_SQL_rejections_are_poison(int number, bool poison) + { + var ex = new DbUpdateException("An error occurred while saving the entity changes.", + SqlExceptionFactory.Create(number, $"error {number}")); + ConfigPublishCoordinator.IsPoison(ex).ShouldBe(poison); + } + + [Fact] + public void A_timeout_a_concurrency_conflict_and_a_providerless_update_error_are_not_poison() + { + ConfigPublishCoordinator.IsPoison(new DbUpdateException("save failed", SqlExceptionFactory.Timeout())).ShouldBeFalse(); + ConfigPublishCoordinator.IsPoison(new DbUpdateConcurrencyException("conflict")).ShouldBeFalse(); + ConfigPublishCoordinator.IsPoison(new DbUpdateException("no provider detail")).ShouldBeFalse(); + ConfigPublishCoordinator.IsPoison(SqlExceptionFactory.Create(2628, "not wrapped by EF")).ShouldBeFalse(); + } + + /// + /// Fourth review M1: a deadlock on the ack write is retried and the row is RECORDED once it clears — it used to + /// be classed poison, counted from memory, and the node's row left Applying for good. + /// + [Fact] + public void A_deadlocked_ack_write_is_retried_and_recorded_not_counted_from_memory() + { + var (db, deploymentId, coordinator) = DispatchedOnConstraintDb(NoTimeout); + db.FailNodeStateUpdatesTransiently = true; + + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + Thread.Sleep(TimeSpan.FromMilliseconds(400)); + db.FailNodeStateUpdatesTransiently = false; + + AwaitAssert(() => + { + using var ctx = db.CreateDbContext(); + ctx.Deployments.Single().Status.ShouldBe(DeploymentStatus.Sealed); + ctx.NodeDeploymentStates.Single().Status.ShouldBe(NodeDeploymentStatus.Applied); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + /// + /// Fourth review M1: an ack counted only from memory (its write was rejected) must survive a coordinator restart + /// before the outcome is written — no row records it, and its node will not resend. + /// + [Fact] + public void A_memory_counted_ack_survives_a_coordinator_restart() + { + var db = new ConstraintEnforcingDbFactory(Guid.NewGuid().ToString("N")); + SeedNodes(db, Node, "site-a-2:4053"); + var deploymentId = SeedDeployment(db, DeploymentStatus.Dispatching, DateTime.UtcNow); + var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props( + db, NoTimeout, loadRetryMin: RetryMin, loadRetryMax: RetryMax)); + coordinator.Tell(new DispatchDeployment(deploymentId, TestRevision, CorrelationId.NewId())); + AwaitAssert(() => + { + using var ctx = db.CreateDbContext(); + ctx.Deployments.Single().Status.ShouldBe(DeploymentStatus.AwaitingApplyAcks); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + + // Node 1's write is rejected: counted from memory only. + db.PoisonNodeStateUpdates = true; + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse(Node), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + Thread.Sleep(TimeSpan.FromMilliseconds(300)); + db.PoisonNodeStateUpdates = false; + + // Restart the coordinator: a dispatch whose guard read cannot open the DB throws in the handler. + db.FailNextContexts(1); + coordinator.Tell(new DispatchDeployment(DeploymentId.NewId(), TestRevision, CorrelationId.NewId())); + Thread.Sleep(TimeSpan.FromMilliseconds(500)); + + coordinator.Tell(new ApplyAck(deploymentId, NodeId.Parse("site-a-2:4053"), ApplyAckOutcome.Applied, null, CorrelationId.NewId())); + AwaitAssert(() => + { + using var ctx = db.CreateDbContext(); + ctx.Deployments.Single(d => d.DeploymentId == deploymentId.Value).Status.ShouldBe(DeploymentStatus.Sealed); + }, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + private (ConstraintEnforcingDbFactory Db, DeploymentId Id, IActorRef Coordinator) DispatchedOnConstraintDb(TimeSpan applyDeadline) { var db = new ConstraintEnforcingDbFactory(Guid.NewGuid().ToString("N")); diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/ConstraintEnforcingDbFactory.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/ConstraintEnforcingDbFactory.cs index a3d29603b..eabed3ad4 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/ConstraintEnforcingDbFactory.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/ConstraintEnforcingDbFactory.cs @@ -8,9 +8,10 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; /// /// An in-memory ConfigDb that behaves like SQL Server where the InMemory provider does not (lmxopcua#868 third /// review): a FailureReason longer than the column refuses the whole SaveChanges with a -/// . It can also reject (, a non-transient -/// ) or fail transiently () every -/// update of an existing NodeDeploymentState row. +/// over SQL error 2628. It can also reject (, +/// error 547) or fail transiently (, deadlock 1205) every update of +/// an existing NodeDeploymentState row — each a real SqlException inside EF's +/// , the shape SQL Server produces (fourth review M1). /// public sealed class ConstraintEnforcingDbFactory(string dbName) : IDbContextFactory { @@ -22,8 +23,18 @@ public sealed class ConstraintEnforcingDbFactory(string dbName) : IDbContextFact /// Every update of an existing node row fails as if the connection dropped. public bool FailNodeStateUpdatesTransiently { get => _interceptor.Transient; set => _interceptor.Transient = value; } + private int _failNextContexts; + + /// The next context creations throw (an unreachable server). + public void FailNextContexts(int count) => Volatile.Write(ref _failNextContexts, count); + /// - public OtOpcUaConfigDbContext CreateDbContext() => new(new DbContextOptionsBuilder() + public OtOpcUaConfigDbContext CreateDbContext() => + Volatile.Read(ref _failNextContexts) > 0 && Interlocked.Decrement(ref _failNextContexts) >= 0 + ? throw new InvalidOperationException("ConfigDb unreachable (simulated)") + : Create(); + + private OtOpcUaConfigDbContext Create() => new(new DbContextOptionsBuilder() .UseInMemoryDatabase(dbName) .AddInterceptors(_interceptor) .Options); @@ -57,11 +68,16 @@ public sealed class ConstraintEnforcingDbFactory(string dbName) : IDbContextFact _ => null, }; if (reason is { Length: > FailureReasons.MaxLength }) - throw new DbUpdateException("String or binary data would be truncated in column 'FailureReason'."); + throw new DbUpdateException("An error occurred while saving the entity changes.", + SqlExceptionFactory.Create(2628, "String or binary data would be truncated in column 'FailureReason'.")); if (e.Entity is NodeDeploymentState && e.State == EntityState.Modified) { - if (Poison) throw new DbUpdateException("The UPDATE statement conflicted with a CHECK constraint."); - if (Transient) throw new InvalidOperationException("A network-related error occurred (simulated)."); + if (Poison) + throw new DbUpdateException("An error occurred while saving the entity changes.", + SqlExceptionFactory.Create(547, "The UPDATE statement conflicted with a CHECK constraint.")); + if (Transient) + throw new DbUpdateException("An error occurred while saving the entity changes.", + SqlExceptionFactory.Create(1205, "Transaction was deadlocked on lock resources and has been chosen as the deadlock victim.")); } } } diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/SqlExceptionFactory.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/SqlExceptionFactory.cs new file mode 100644 index 000000000..479ac7689 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/Harness/SqlExceptionFactory.cs @@ -0,0 +1,63 @@ +using System.ComponentModel; +using System.Reflection; +using Microsoft.Data.SqlClient; + +namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness; + +/// +/// Builds real s for tests. The type has no public constructor, so this goes through its +/// internal factory by reflection — the same objects SQL Server failures produce, carrying a real error number. +/// +public static class SqlExceptionFactory +{ + private const BindingFlags NonPublic = BindingFlags.NonPublic | BindingFlags.Instance; + + /// A with one error of the given number. + /// The SQL Server error number (e.g. 2628, 1205, -2). + /// The error text. + /// An inner exception (e.g. a for a timeout). + /// The exception. + public static SqlException Create(int number, string message, Exception? inner = null) + { + var errorCtor = typeof(SqlError).GetConstructors(NonPublic) + .OrderByDescending(c => c.GetParameters().Length) + .First(); + var args = errorCtor.GetParameters().Select(p => p.Name switch + { + "infoNumber" => (object?)number, + "errorMessage" => message, + "server" => "test-sql", + "procedure" => string.Empty, + "exception" => inner, + "errorClass" => (byte)16, + _ => p.ParameterType == typeof(string) ? string.Empty + : p.ParameterType.IsValueType ? Activator.CreateInstance(p.ParameterType) : null, + }).ToArray(); + var error = (SqlError)errorCtor.Invoke(args); + + var collection = (SqlErrorCollection)Activator.CreateInstance(typeof(SqlErrorCollection), nonPublic: true)!; + typeof(SqlErrorCollection).GetMethod("Add", NonPublic)!.Invoke(collection, [error]); + + var create = typeof(SqlException) + .GetMethods(BindingFlags.NonPublic | BindingFlags.Static) + .Where(m => m.Name == "CreateException") + .Select(m => (Method: m, Params: m.GetParameters())) + .Where(x => x.Params.Length >= 2 && x.Params[0].ParameterType == typeof(SqlErrorCollection) + && x.Params[1].ParameterType == typeof(string)) + .OrderBy(x => x.Params.Length) + .First(); + var createArgs = create.Params.Select((p, i) => i switch + { + 0 => collection, + 1 => (object?)"16.0.0", + _ => p.ParameterType == typeof(Exception) ? inner + : p.ParameterType.IsValueType ? Activator.CreateInstance(p.ParameterType) : null, + }).ToArray(); + return (SqlException)create.Method.Invoke(null, createArgs)!; + } + + /// A command timeout as SQL Server reports it: number -2 over a Win32 258 (WAIT_TIMEOUT). + /// The exception. + public static SqlException Timeout() => + Create(-2, "Execution Timeout Expired.", new Win32Exception(258, "The wait operation timed out.")); +} diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnconfirmedRevisionTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnconfirmedRevisionTests.cs new file mode 100644 index 000000000..df40125f8 --- /dev/null +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.Runtime.Tests/Drivers/DriverHostActorUnconfirmedRevisionTests.cs @@ -0,0 +1,117 @@ +using Akka.Actor; +using Microsoft.EntityFrameworkCore; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy; +using ZB.MOM.WW.OtOpcUa.Commons.Types; +using ZB.MOM.WW.OtOpcUa.Configuration; +using ZB.MOM.WW.OtOpcUa.Configuration.Entities; +using ZB.MOM.WW.OtOpcUa.Configuration.Enums; +using ZB.MOM.WW.OtOpcUa.Runtime.Drivers; +using ZB.MOM.WW.OtOpcUa.Runtime.OpcUa; +using ZB.MOM.WW.OtOpcUa.Runtime.Tests.Harness; + +namespace ZB.MOM.WW.OtOpcUa.Runtime.Tests.Drivers; + +/// +/// lmxopcua#868 fourth review M2: a revision a node ADOPTS without an outcome-reporting apply — Stale recovery, a +/// failed bootstrap restore — is not known to be served. A dispatch of that same revision must RE-APPLY (the +/// publish actor gets a rebuild) instead of short-circuiting a clean Applied over an empty address space. +/// +public sealed class DriverHostActorUnconfirmedRevisionTests : RuntimeActorTestBase +{ + private static readonly NodeId TestNode = NodeId.Parse("driver-test"); + private static readonly RevisionHash RevA = RevisionHash.Parse(new string('a', 64)); + + [Fact] + public void A_revision_adopted_by_Stale_recovery_is_reapplied_on_redispatch() + { + var db = new SwitchableDbFactory(Guid.NewGuid().ToString("N")); + var deploymentId = Seed(db); + db.Down = true; + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, CreateTestProbe().Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + + // Bootstrap fails (DB down) ⇒ Stale. Bring the DB back and drive the recovery tick by hand. + // Barrier: Stale answers RouteNodeWrite with a "stale" refusal, so a reply proves the host is in Stale. + var asker = CreateTestProbe(); + actor.Tell(new DriverHostActor.RouteNodeWrite("eq-x/speed", 0.0, ZB.MOM.WW.OtOpcUa.Commons.OpcUa.AddressSpaceRealm.Uns), asker.Ref); + asker.ExpectMsg(PresenceBudget, cancellationToken: TestContext.Current.CancellationToken) + .Reason!.ShouldContain("stale"); + db.Down = false; + actor.Tell(DriverHostActor.RetryConfigDbConnection.Instance); + AwaitAssert(() => + { + using var ctx = db.CreateDbContext(); + ctx.NodeDeploymentStates.Single(r => r.NodeId == TestNode.Value).Status.ShouldBe(NodeDeploymentStatus.Applied); + }, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + publish.ExpectNoMsg(TimeSpan.FromMilliseconds(200), TestContext.Current.CancellationToken); // recovery restores nothing + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + publish.FishForMessage( + m => m.DeploymentId == deploymentId && m.ReportOutcome, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + [Fact] + public void A_revision_whose_bootstrap_restore_failed_is_reapplied_on_redispatch() + { + var db = new SwitchableDbFactory(Guid.NewGuid().ToString("N")); + var deploymentId = Seed(db); + using (var ctx = db.CreateDbContext()) + { + ctx.NodeDeploymentStates.Add(new NodeDeploymentState + { + NodeId = TestNode.Value, DeploymentId = deploymentId.Value, Status = NodeDeploymentStatus.Applied, + StartedAtUtc = DateTime.UtcNow, AppliedAtUtc = DateTime.UtcNow, + }); + ctx.SaveChanges(); + } + db.FailRestore = true; // the restore's artifact read throws; the bootstrap's own row read does not + var publish = CreateTestProbe(); + var actor = Sys.ActorOf(DriverHostActor.Props( + db, TestNode, CreateTestProbe().Ref, localRoles: new HashSet { "driver" }, + opcUaPublishActor: publish.Ref)); + AwaitCondition(() => db.RestoreFailures > 0, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + db.FailRestore = false; + + actor.Tell(new DispatchDeployment(deploymentId, RevA, CorrelationId.NewId())); + + publish.FishForMessage( + m => m.DeploymentId == deploymentId && m.ReportOutcome, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken); + } + + private static DeploymentId Seed(IDbContextFactory db) => + V3ArtifactSeeder.SeedV3Deployment(db, RevA, + rawTags: new[] { (Driver: "drv-1", DriverName: "Modbus", Device: "dev1", Tag: "speed", DataType: "Double", Writable: false) }, + unsRefs: new[] { (Area: "filling", Line: "line1", Equip: "station1", BackingTag: "speed", Effective: (string?)null) }); + + /// In-memory ConfigDb that can be down, or fail only the bootstrap restore's artifact read. + private sealed class SwitchableDbFactory(string dbName) : IDbContextFactory + { + private volatile bool _down; + private volatile bool _failRestore; + private int _attempts; + private int _restoreFailures; + + public bool Down { get => _down; set => _down = value; } + public bool FailRestore { get => _failRestore; set => _failRestore = value; } + public int Attempts => Volatile.Read(ref _attempts); + public int RestoreFailures => Volatile.Read(ref _restoreFailures); + + public OtOpcUaConfigDbContext CreateDbContext() + { + Interlocked.Increment(ref _attempts); + if (_down) throw new InvalidOperationException("ConfigDb unreachable (simulated)"); + if (_failRestore && new System.Diagnostics.StackTrace().ToString().Contains("RestoreApplied")) + { + Interlocked.Increment(ref _restoreFailures); + throw new InvalidOperationException("ConfigDb read failed during restore (simulated)"); + } + return new OtOpcUaConfigDbContext(new DbContextOptionsBuilder() + .UseInMemoryDatabase(dbName).Options); + } + } +}