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); + } + } +}