fix(controlplane): condition the dispatch flip on the guard's RowVersion; one deadline source; align IdentityModel pins

Re-review 1: HandleDispatch's terminal guard and its AwaitingApplyAcks flip read the row separately, so a lost-dispatch closure committed between them was silently reopened. The flip now carries the RowVersion the guard read; a change in between is a concurrency conflict (nothing seeded, nothing published, coordinator reset). Pinned on real SQL Server with a closure injected between the two reads (fails without the fix). AdminOps' apply deadline and the coordinator's now both read DeployTimings.ApplyDeadline, passed explicitly to both singletons. Microsoft.IdentityModel.Protocols(.OpenIdConnect) pinned to 8.11.0 and referenced directly by the Security project, so the whole IdentityModel family resolves 8.11.0 (fresh restore clean of NU190x/NU1605). Refs #733 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 14:08:42 -04:00
parent dc3323fcca
commit 5097444d56
6 changed files with 138 additions and 24 deletions
@@ -36,8 +36,10 @@
    <PackageVersion Include="libplctag" Version="1.5.2" />
    <PackageVersion Include="MessagePack" Version="2.5.301" />
    <!-- First-party bearer validation for /auth/token's JWTs (R03:2.4, Lane C fix round 1). Same 10.0.x servicing
-         line as the other Microsoft.AspNetCore.* pins; its Microsoft.IdentityModel.* dependencies resolve UP to the
-         8.11.0 pins below (no downgrade, no NU1605). -->
+         line as the other Microsoft.AspNetCore.* pins. It depends on Microsoft.IdentityModel.Protocols(.OpenIdConnect)
+         8.0.1; those two are pinned to 8.11.0 below (and referenced directly by the Security project, the JwtBearer
+         entry point, so the pin applies without transitive pinning) to keep the whole IdentityModel family on one
+         version with Microsoft.IdentityModel.Tokens / System.IdentityModel.Tokens.Jwt. -->
    <PackageVersion Include="Microsoft.AspNetCore.Authentication.JwtBearer" Version="10.0.7" />
    <PackageVersion Include="Microsoft.AspNetCore.Authorization" Version="10.0.7" />
    <PackageVersion Include="Microsoft.AspNetCore.DataProtection" Version="10.0.7" />
@@ -86,6 +88,8 @@
    <PackageVersion Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.7" />
    <PackageVersion Include="Microsoft.Extensions.Options" Version="10.0.7" />
    <PackageVersion Include="Microsoft.Extensions.Options.ConfigurationExtensions" Version="10.0.7" />
+    <PackageVersion Include="Microsoft.IdentityModel.Protocols" Version="8.11.0" />
+    <PackageVersion Include="Microsoft.IdentityModel.Protocols.OpenIdConnect" Version="8.11.0" />
    <PackageVersion Include="Microsoft.IdentityModel.Tokens" Version="8.11.0" />
    <PackageVersion Include="Microsoft.NET.Test.Sdk" Version="17.12.0" />
    <PackageVersion Include="Microsoft.Playwright" Version="1.51.0" />
@@ -11,7 +11,6 @@ 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.Configuration.Validation;
-using ZB.MOM.WW.OtOpcUa.ControlPlane.Coordinators;
 using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
 using ZB.MOM.WW.OtOpcUa.Core.Scripting;

@@ -55,7 +54,7 @@ public sealed class AdminOperationsActor : ReceiveActor
    ///     Central comm actor commands are handed to, or <see langword="null"/> to publish on the
    ///     DistributedPubSub mediator directly (the behaviour before the per-cluster mesh).
    /// </param>
-    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="ConfigPublishCoordinator.DefaultApplyDeadline"/>.</param>
+    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</param>
    /// <returns>Props configured to create an AdminOperationsActor.</returns>
    public static Props Props(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
@@ -73,7 +72,7 @@ public sealed class AdminOperationsActor : ReceiveActor
    /// <param name="probes">Driver probes registered in DI; keyed by DriverType (case-insensitive).</param>
    /// <param name="tagConfigValidationMode">Deploy-gate TagConfig strictness mode (Warn default | Error opt-in).</param>
    /// <param name="meshRouter">Central comm actor, or <see langword="null"/> for the mediator.</param>
-    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="ConfigPublishCoordinator.DefaultApplyDeadline"/>.</param>
+    /// <param name="applyDeadline">The coordinator's apply deadline; defaults to <see cref="DeployTimings.ApplyDeadline"/> (the coordinator's default source).</param>
    public AdminOperationsActor(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory,
        IActorRef coordinator,
@@ -82,7 +81,9 @@ public sealed class AdminOperationsActor : ReceiveActor
        IActorRef? meshRouter = null,
        TimeSpan? applyDeadline = null)
    {
-        _applyDeadline = applyDeadline ?? ConfigPublishCoordinator.DefaultApplyDeadline;
+        // The same source the coordinator's default reads (ConfigPublishCoordinator.DefaultApplyDeadline IS
+        // DeployTimings.ApplyDeadline), so the lost-dispatch closure and the coordinator's own deadline agree.
+        _applyDeadline = applyDeadline ?? DeployTimings.ApplyDeadline;
        _dbFactory = dbFactory;
        _coordinator = coordinator;
        _probesByType = probes.ToDictionary(p => p.DriverType, StringComparer.OrdinalIgnoreCase);
@@ -237,21 +237,28 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
        // Terminal-state guard (fix round 1): a late or re-delivered dispatch for a row that is already
        // Sealed / PartiallyFailed / TimedOut — e.g. one AdminOps closed as a lost dispatch while this message
        // was still queued — must never flip it back to AwaitingApplyAcks or seed an expected set for it.
+        //
+        // The RowVersion read HERE is the one the AwaitingApplyAcks flip below is conditioned on (re-review 1):
+        // reading a fresh one at flip time left a window in which AdminOps' lost-dispatch closure could commit
+        // TimedOut between the guard and the flip, and the flip would silently reopen the row. With the guard's
+        // RowVersion, any change since the guard read makes the flip a concurrency conflict instead.
+        byte[]? guardRowVersion;
        using (var check = _dbFactory.CreateDbContext())
        {
-            var status = check.Deployments
+            var guard = check.Deployments
                .AsNoTracking()
                .Where(d => d.DeploymentId == msg.DeploymentId.Value)
-                .Select(d => (DeploymentStatus?)d.Status)
+                .Select(d => new { d.Status, d.RowVersion })
                .FirstOrDefault();
-            if (status is DeploymentStatus.Sealed or DeploymentStatus.PartiallyFailed or DeploymentStatus.TimedOut)
+            if (guard?.Status is DeploymentStatus.Sealed or DeploymentStatus.PartiallyFailed or DeploymentStatus.TimedOut)
            {
                _log.Warning(
                    "Ignoring DispatchDeployment {Id}: the deployment is already {Status} (a late or duplicated dispatch)",
-                    msg.DeploymentId, status);
+                    msg.DeploymentId, guard.Status);
                if (_current == msg.DeploymentId) ResetForNext();
                return;
            }
+            guardRowVersion = guard?.RowVersion;
        }

        _current = msg.DeploymentId;
@@ -290,8 +297,22 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
                    });
                }
            }
-            UpdateDeploymentStatus(db, msg.DeploymentId, DeploymentStatus.AwaitingApplyAcks);
-            db.SaveChanges();
+            UpdateDeploymentStatus(db, msg.DeploymentId, DeploymentStatus.AwaitingApplyAcks, rowVersion: guardRowVersion);
+            try
+            {
+                db.SaveChanges();
+            }
+            catch (DbUpdateConcurrencyException)
+            {
+                // The row moved after the guard read (on SQL Server: AdminOps closed it as a lost dispatch). The
+                // expected-set rows and the flip were one SaveChanges, so nothing was written; nothing is
+                // published and the coordinator is free for the next deployment.
+                _log.Warning(
+                    "DispatchDeployment {Id} abandoned: the deployment changed state while it was being dispatched " +
+                    "(closed as a lost dispatch?) — not reopening it", msg.DeploymentId);
+                ResetForNext();
+                return;
+            }
        }

        // Per-cluster mesh Phase 2: hand off to the transport router rather than publishing
@@ -483,9 +504,9 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
    ///     stub stays tracked so <see cref="SetFailureReason"/> can add to the same UPDATE.
    /// </summary>
    private static void UpdateDeploymentStatus(
-        OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false)
+        OtOpcUaConfigDbContext db, DeploymentId id, DeploymentStatus status, bool sealNow = false, byte[]? rowVersion = null)
    {
-        var entry = TrackStub(db, id);
+        var entry = TrackStub(db, id, rowVersion);
        if (entry is null) return;
        entry.Entity.Status = status;
        entry.Property(x => x.Status).IsModified = true;
@@ -507,20 +528,30 @@ public sealed class ConfigPublishCoordinator : ReceiveActor, IWithTimers
    }

    /// <summary>The tracked blob-free stub for a deployment, attaching one on first use; null when the row is gone.</summary>
+    /// <param name="db">The context.</param>
+    /// <param name="id">The deployment.</param>
+    /// <param name="rowVersion">
+    ///     The RowVersion the update must be conditioned on, when the caller already read one (the dispatch
+    ///     guard); <see langword="null"/> reads the current one.
+    /// </param>
    private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry<Deployment>? TrackStub(
-        OtOpcUaConfigDbContext db, DeploymentId id)
+        OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null)
    {
        var tracked = db.ChangeTracker.Entries<Deployment>().FirstOrDefault(e => e.Entity.DeploymentId == id.Value);
        if (tracked is not null) return tracked;
-        var stamp = db.Deployments
-            .AsNoTracking()
-            .Where(x => x.DeploymentId == id.Value)
-            .Select(x => new { x.RowVersion })
-            .FirstOrDefault();
-        if (stamp is null) return null;
+        if (rowVersion is null)
+        {
+            var stamp = db.Deployments
+                .AsNoTracking()
+                .Where(x => x.DeploymentId == id.Value)
+                .Select(x => new { x.RowVersion })
+                .FirstOrDefault();
+            if (stamp is null) return null;
+            rowVersion = stamp.RowVersion;
+        }
        return db.Deployments.Attach(new Deployment
        {
-            DeploymentId = id.Value, RevisionHash = string.Empty, CreatedBy = string.Empty, RowVersion = stamp.RowVersion,
+            DeploymentId = id.Value, RevisionHash = string.Empty, CreatedBy = string.Empty, RowVersion = rowVersion,
        });
    }

@@ -7,6 +7,7 @@ using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Options;
 using ZB.MOM.WW.OtOpcUa.Configuration;
 using ZB.MOM.WW.OtOpcUa.Cluster;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Deploy;
 using ZB.MOM.WW.OtOpcUa.Commons.Messages.Mesh;
 using ZB.MOM.WW.OtOpcUa.ControlPlane.AdminOperations;
 using ZB.MOM.WW.OtOpcUa.ControlPlane.Audit;
@@ -111,7 +112,7 @@ public static class ServiceCollectionExtensions
                // coordinator. Deliberately NOT an ActorSelection.ResolveOne here: that blocks
                // inside a props factory and races the spawn it is waiting for.
                var meshRouter = registry.Get<CentralCommunicationActorKey>();
-                return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: null, meshRouter: meshRouter);
+                return ConfigPublishCoordinator.Props(dbFactory, applyDeadline: DeployTimings.ApplyDeadline, meshRouter: meshRouter);
            },
            singletonOptions);

@@ -128,7 +129,10 @@ public static class ServiceCollectionExtensions
                    resolver.GetService<IConfiguration>()?["Deployment:TagConfigValidationMode"],
                    ignoreCase: true, out var m) ? m : TagConfigValidationMode.Warn;
                var meshRouter = registry.Get<CentralCommunicationActorKey>();
-                return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter);
+                // The SAME deadline source as the coordinator above: AdminOps closes a lost dispatch only once it
+                // is past the coordinator's deadline, so the two must never drift.
+                return AdminOperationsActor.Props(dbFactory, coordinator, probes, mode, meshRouter,
+                    applyDeadline: DeployTimings.ApplyDeadline);
            },
            singletonOptions);

@@ -11,6 +11,9 @@

    <ItemGroup>
        <PackageReference Include="Microsoft.AspNetCore.Authentication.JwtBearer"/>
+        <!-- Direct, to promote JwtBearer's 8.0.1 IdentityModel.Protocols dependencies to the central 8.11.0 pin. -->
+        <PackageReference Include="Microsoft.IdentityModel.Protocols"/>
+        <PackageReference Include="Microsoft.IdentityModel.Protocols.OpenIdConnect"/>
        <PackageReference Include="Microsoft.IdentityModel.Tokens"/>
        <PackageReference Include="System.IdentityModel.Tokens.Jwt"/>
        <PackageReference Include="ZB.MOM.WW.Auth.Abstractions"/>
@@ -311,6 +311,77 @@ public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTes
        }, duration: PresenceBudget);
    }

+    /// <summary>
+    ///     Re-review 1 — the reopen race. AdminOps' lost-dispatch closure commits TimedOut AFTER the coordinator's
+    ///     terminal guard read the row as Dispatching but BEFORE the AwaitingApplyAcks flip. The flip is
+    ///     conditioned on the guard's RowVersion, so on SQL Server it becomes a concurrency conflict and the row
+    ///     stays TimedOut with no expected set. Positive control: the coordinator is not wedged — the next live
+    ///     dispatch is seeded and flipped.
+    /// </summary>
+    [Fact]
+    [Trait("Category", "SchemaCompliance")]
+    public async Task On_sql_server_a_closure_between_guard_and_flip_is_not_reopened()
+    {
+        using var sql = new SqlServerTestDatabase();
+        var inner = sql.Factory();
+        SeedNodes(inner, "site-a-1:4053");
+        var interposer = new InterposingDbFactory(inner);
+        var coordinator = Sys.ActorOf(ConfigPublishCoordinator.Props(interposer, NoTimeout));
+        await Sys.ActorSelection(coordinator.Path).ResolveOne(PresenceBudget);   // PreStart ran over an empty table
+
+        // The race proper: a new Dispatching row whose dispatch is interleaved with an AdminOps-style closure.
+        var victim = SeedDeployment(inner, DeploymentStatus.Dispatching, DateTime.UtcNow);
+        interposer.OnSecondCreate(() =>
+        {
+            using var db = sql.NewContext();
+            db.Database.ExecuteSqlRaw(
+                "UPDATE Deployment SET Status = {0}, FailureReason = N'closed as lost' WHERE DeploymentId = {1}",
+                (int)DeploymentStatus.TimedOut, victim.Value);
+        });
+        coordinator.Tell(new DispatchDeployment(victim, TestRevision, CorrelationId.NewId()));
+        AwaitAssert(() => interposer.Fired.ShouldBeTrue(), duration: PresenceBudget);
+
+        // Positive control + ordering: a live row dispatched after it IS flipped, so the victim's dispatch was processed.
+        var live = SeedDeployment(inner, DeploymentStatus.Dispatching, DateTime.UtcNow);
+        coordinator.Tell(new DispatchDeployment(live, TestRevision, CorrelationId.NewId()));
+        AwaitAssert(() =>
+        {
+            using var db = sql.NewContext();
+            db.Deployments.AsNoTracking().Single(d => d.DeploymentId == live.Value).Status
+                .ShouldBe(DeploymentStatus.AwaitingApplyAcks);
+        }, duration: PresenceBudget);
+
+        using var verify = sql.NewContext();
+        verify.Deployments.AsNoTracking().Single(d => d.DeploymentId == victim.Value).Status
+            .ShouldBe(DeploymentStatus.TimedOut, "the closure must not be reopened");
+        verify.NodeDeploymentStates.Count(s => s.DeploymentId == victim.Value).ShouldBe(0);
+    }
+
+    /// <summary>Runs an action just before handing out the SECOND context after arming (HandleDispatch: guard, then flip).</summary>
+    private sealed class InterposingDbFactory(IDbContextFactory<OtOpcUaConfigDbContext> inner)
+        : IDbContextFactory<OtOpcUaConfigDbContext>
+    {
+        private Action? _action;
+        private int _countdown;
+        public volatile bool Fired;
+
+        public void OnSecondCreate(Action action)
+        {
+            _action = action;
+            Volatile.Write(ref _countdown, 2);
+        }
+
+        public OtOpcUaConfigDbContext CreateDbContext()
+        {
+            if (Volatile.Read(ref _countdown) > 0 && Interlocked.Decrement(ref _countdown) == 0)
+            {
+                _action?.Invoke();
+                Fired = true;
+            }
+            return inner.CreateDbContext();
+        }
+    }
+
    private static DeploymentId SeedDeployment(
        IDbContextFactory<OtOpcUaConfigDbContext> dbFactory, DeploymentStatus status, DateTime createdAtUtc)
    {