Merge #868 fourth review fixes: poison only by SQL error allowlist, memory-counted acks survive restart, Stale-recovered revisions unconfirmed
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
+7
-1
@@ -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.
|
||||
|
||||
+40
-11
@@ -109,6 +109,12 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers, IWithU
|
||||
private readonly Dictionary<NodeId, PendingPersist> _unpersisted = new(NodeIdComparer);
|
||||
private string? _lastPersistError;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
private readonly Dictionary<NodeId, ApplyAck> _memoryOnlyAcks = new(NodeIdComparer);
|
||||
|
||||
/// <summary>When the current deadline was first deferred on an unpersisted ack (bounds the deferral).</summary>
|
||||
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
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// A non-transient rejection of the write itself: an EF <see cref="DbUpdateException"/> (not a concurrency
|
||||
/// conflict) with no transient database error, timeout or socket failure underneath. Connectivity failures
|
||||
/// surface as other exception types (or as transient <see cref="System.Data.Common.DbException"/>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).
|
||||
/// </summary>
|
||||
internal static bool IsPoison(Exception ex)
|
||||
private static readonly int[] PoisonSqlNumbers = [8152, 2628, 547, 515, 2627, 2601];
|
||||
|
||||
/// <summary>
|
||||
/// Whether an ack write failure is POISON — never retried — by an ALLOWLIST: an EF <see cref="DbUpdateException"/>
|
||||
/// (not a concurrency conflict) carrying a <see cref="Microsoft.Data.SqlClient.SqlException"/> whose error
|
||||
/// numbers include one of <see cref="PoisonSqlNumbers"/>. 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 <c>DbException.IsTransient</c>, so a deny-list on it classified
|
||||
/// every SQL Server transient as poison.)
|
||||
/// </summary>
|
||||
/// <param name="ex">The failure.</param>
|
||||
/// <returns><see langword="true"/> only for a known, deterministic rejection.</returns>
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>A retry tick: only the node's CURRENT unpersisted ack is retried; a superseded one is dropped (L3).</summary>
|
||||
@@ -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();
|
||||
|
||||
@@ -290,15 +290,16 @@ public sealed class DriverHostActor : ReceiveActor, IWithTimers, IWithUnboundedS
|
||||
private RevisionHash? _currentRevision;
|
||||
|
||||
/// <summary>
|
||||
/// The last address-space apply of <see cref="_currentRevision"/> reported DEGRADED (or no
|
||||
/// outcome in time). A re-dispatch of that same revision then RE-APPLIES instead of taking the
|
||||
/// <see cref="_currentRevision"/> 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).
|
||||
/// </summary>
|
||||
private bool _currentRevisionDegraded;
|
||||
private bool _currentRevisionUnconfirmed;
|
||||
|
||||
/// <summary>Applied acks held until the address-space apply reports its outcome, keyed by
|
||||
/// the apply's correlation id (echoed back in <see cref="OpcUa.OpcUaPublishActor.RebuildCompleted"/>).</summary>
|
||||
@@ -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
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>This does not make the node current.</b> <c>_currentRevision</c> 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 (<see cref="_currentRevisionUnconfirmed"/>, 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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;
|
||||
@@ -2821,7 +2824,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)
|
||||
{
|
||||
@@ -3085,7 +3088,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;
|
||||
@@ -3142,7 +3145,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
|
||||
@@ -3597,6 +3600,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
|
||||
@@ -3613,6 +3620,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);
|
||||
}
|
||||
}
|
||||
@@ -4334,6 +4344,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);
|
||||
}
|
||||
|
||||
+91
@@ -495,6 +495,97 @@ public sealed class ConfigPublishCoordinatorDbOutageTests : ControlPlaneActorTes
|
||||
}, duration: PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Fourth review M1: poison is an ALLOWLIST of SQL Server rejections. Microsoft.Data.SqlClient never sets
|
||||
/// <c>DbException.IsTransient</c>, so the old deny-list called every SQL Server transient poison.
|
||||
/// </summary>
|
||||
[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();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
[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"));
|
||||
|
||||
+23
-7
@@ -8,9 +8,10 @@ namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
|
||||
/// <summary>
|
||||
/// An in-memory ConfigDb that behaves like SQL Server where the InMemory provider does not (lmxopcua#868 third
|
||||
/// review): a <c>FailureReason</c> longer than the column refuses the whole <c>SaveChanges</c> with a
|
||||
/// <see cref="DbUpdateException"/>. It can also reject (<see cref="PoisonNodeStateUpdates"/>, a non-transient
|
||||
/// <see cref="DbUpdateException"/>) or fail transiently (<see cref="FailNodeStateUpdatesTransiently"/>) every
|
||||
/// update of an existing <c>NodeDeploymentState</c> row.
|
||||
/// <see cref="DbUpdateException"/> over SQL error 2628. It can also reject (<see cref="PoisonNodeStateUpdates"/>,
|
||||
/// error 547) or fail transiently (<see cref="FailNodeStateUpdatesTransiently"/>, deadlock 1205) every update of
|
||||
/// an existing <c>NodeDeploymentState</c> row — each a real <c>SqlException</c> inside EF's
|
||||
/// <see cref="DbUpdateException"/>, the shape SQL Server produces (fourth review M1).
|
||||
/// </summary>
|
||||
public sealed class ConstraintEnforcingDbFactory(string dbName) : IDbContextFactory<OtOpcUaConfigDbContext>
|
||||
{
|
||||
@@ -22,8 +23,18 @@ public sealed class ConstraintEnforcingDbFactory(string dbName) : IDbContextFact
|
||||
/// <summary>Every update of an existing node row fails as if the connection dropped.</summary>
|
||||
public bool FailNodeStateUpdatesTransiently { get => _interceptor.Transient; set => _interceptor.Transient = value; }
|
||||
|
||||
private int _failNextContexts;
|
||||
|
||||
/// <summary>The next <paramref name="count"/> context creations throw (an unreachable server).</summary>
|
||||
public void FailNextContexts(int count) => Volatile.Write(ref _failNextContexts, count);
|
||||
|
||||
/// <inheritdoc />
|
||||
public OtOpcUaConfigDbContext CreateDbContext() => new(new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
|
||||
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<OtOpcUaConfigDbContext>()
|
||||
.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."));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,63 @@
|
||||
using System.ComponentModel;
|
||||
using System.Reflection;
|
||||
using Microsoft.Data.SqlClient;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
|
||||
|
||||
/// <summary>
|
||||
/// Builds real <see cref="SqlException"/>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.
|
||||
/// </summary>
|
||||
public static class SqlExceptionFactory
|
||||
{
|
||||
private const BindingFlags NonPublic = BindingFlags.NonPublic | BindingFlags.Instance;
|
||||
|
||||
/// <summary>A <see cref="SqlException"/> with one error of the given number.</summary>
|
||||
/// <param name="number">The SQL Server error number (e.g. 2628, 1205, -2).</param>
|
||||
/// <param name="message">The error text.</param>
|
||||
/// <param name="inner">An inner exception (e.g. a <see cref="Win32Exception"/> for a timeout).</param>
|
||||
/// <returns>The exception.</returns>
|
||||
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)!;
|
||||
}
|
||||
|
||||
/// <summary>A command timeout as SQL Server reports it: number -2 over a Win32 258 (WAIT_TIMEOUT).</summary>
|
||||
/// <returns>The exception.</returns>
|
||||
public static SqlException Timeout() =>
|
||||
Create(-2, "Execution Timeout Expired.", new Win32Exception(258, "The wait operation timed out."));
|
||||
}
|
||||
+117
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// 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.
|
||||
/// </summary>
|
||||
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<string> { "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<DriverHostActor.NodeWriteResult>(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<OpcUaPublishActor.RebuildAddressSpace>(
|
||||
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<string> { "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<OpcUaPublishActor.RebuildAddressSpace>(
|
||||
m => m.DeploymentId == deploymentId && m.ReportOutcome, PresenceBudget, cancellationToken: TestContext.Current.CancellationToken);
|
||||
}
|
||||
|
||||
private static DeploymentId Seed(IDbContextFactory<OtOpcUaConfigDbContext> 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) });
|
||||
|
||||
/// <summary>In-memory ConfigDb that can be down, or fail only the bootstrap restore's artifact read.</summary>
|
||||
private sealed class SwitchableDbFactory(string dbName) : IDbContextFactory<OtOpcUaConfigDbContext>
|
||||
{
|
||||
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<OtOpcUaConfigDbContext>()
|
||||
.UseInMemoryDatabase(dbName).Options);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user