fix(controlplane,runtime): #868 fourth review — SQL poison allowlist, memory-counted acks survive restart, adopted revisions unconfirmed

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
2026-11-11 13:31:49 -05:00
parent ca6d131078
commit ec3b835e4a
7 changed files with 367 additions and 30 deletions
@@ -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"));
@@ -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."));
+}
@@ -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);
+        }
+    }
+}