diff --git a/Directory.Packages.props b/Directory.Packages.props index ed2dca67..d63e819e 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -36,8 +36,10 @@ + 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. --> @@ -86,6 +88,8 @@ + + diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs index 81f09dae..69e878db 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/AdminOperations/AdminOperationsActor.cs @@ -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 to publish on the /// DistributedPubSub mediator directly (the behaviour before the per-cluster mesh). /// - /// The coordinator's apply deadline; defaults to . + /// The coordinator's apply deadline; defaults to (the coordinator's default source). /// Props configured to create an AdminOperationsActor. public static Props Props( IDbContextFactory dbFactory, @@ -73,7 +72,7 @@ public sealed class AdminOperationsActor : ReceiveActor /// Driver probes registered in DI; keyed by DriverType (case-insensitive). /// Deploy-gate TagConfig strictness mode (Warn default | Error opt-in). /// Central comm actor, or for the mediator. - /// The coordinator's apply deadline; defaults to . + /// The coordinator's apply deadline; defaults to (the coordinator's default source). public AdminOperationsActor( IDbContextFactory 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); 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 49b54df4..d408343c 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/Coordinators/ConfigPublishCoordinator.cs @@ -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 can add to the same UPDATE. /// 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 } /// The tracked blob-free stub for a deployment, attaching one on first use; null when the row is gone. + /// The context. + /// The deployment. + /// + /// The RowVersion the update must be conditioned on, when the caller already read one (the dispatch + /// guard); reads the current one. + /// private static Microsoft.EntityFrameworkCore.ChangeTracking.EntityEntry? TrackStub( - OtOpcUaConfigDbContext db, DeploymentId id) + OtOpcUaConfigDbContext db, DeploymentId id, byte[]? rowVersion = null) { var tracked = db.ChangeTracker.Entries().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, }); } diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs index f838a004..9a9f0742 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs +++ b/src/Server/ZB.MOM.WW.OtOpcUa.ControlPlane/ServiceCollectionExtensions.cs @@ -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(); - 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()?["Deployment:TagConfigValidationMode"], ignoreCase: true, out var m) ? m : TagConfigValidationMode.Warn; var meshRouter = registry.Get(); - 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); diff --git a/src/Server/ZB.MOM.WW.OtOpcUa.Security/ZB.MOM.WW.OtOpcUa.Security.csproj b/src/Server/ZB.MOM.WW.OtOpcUa.Security/ZB.MOM.WW.OtOpcUa.Security.csproj index 507aca6c..8ee03814 100644 --- a/src/Server/ZB.MOM.WW.OtOpcUa.Security/ZB.MOM.WW.OtOpcUa.Security.csproj +++ b/src/Server/ZB.MOM.WW.OtOpcUa.Security/ZB.MOM.WW.OtOpcUa.Security.csproj @@ -11,6 +11,9 @@ + + + diff --git a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs index 0added3d..8c93fb1c 100644 --- a/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs +++ b/tests/Server/ZB.MOM.WW.OtOpcUa.ControlPlane.Tests/ConfigPublishCoordinatorRecoveryTests.cs @@ -311,6 +311,77 @@ public sealed class ConfigPublishCoordinatorRecoveryTests : ControlPlaneActorTes }, duration: PresenceBudget); } + /// + /// 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. + /// + [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); + } + + /// Runs an action just before handing out the SECOND context after arming (HandleDispatch: guard, then flip). + private sealed class InterposingDbFactory(IDbContextFactory inner) + : IDbContextFactory + { + 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 dbFactory, DeploymentStatus status, DateTime createdAtUtc) {