fix(controlplane): validate and compose a deployment from ONE snapshot read

StartDeployment read the whole config DB twice with no transaction — DraftSnapshotFactory for the validator, then ConfigComposer for the artifact, plus clusters/nodes and Script.SourceCode again in between. An AdminUI save landing between the two shipped an artifact the DraftValidator never saw. - DraftSnapshot carries ServerClusters, ClusterNodes and ApiKeys; DraftSnapshotFactory reads every table inside one SNAPSHOT transaction on a relational provider (no-op on InMemory or under an ambient transaction). - ConfigComposer.Compose(DraftSnapshot) is pure over the snapshot (ordinal-sorted keys); SnapshotAndFlattenAsync(db) delegates to it, byte-identical. - AdminOperationsActor validates topology, estimates compile cost and composes the artifact from the one snapshot it holds. - Migration EnableSnapshotIsolation: ALTER DATABASE CURRENT SET ALLOW_SNAPSHOT_ISOLATION ON (suppressTransaction; no model change). Tests: the SQL Server snapshot read ignores a row committed mid-read (positive control: a later read sees it); an edit injected between the Equipment and Tags reads is absent from the shipped artifact and rejected by the next deploy; Compose(snapshot) == SnapshotAndFlattenAsync(db). Refs #733 Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
2026-10-22 06:36:12 -04:00
parent b437c34f92
commit 73e74e270d
10 changed files with 2077 additions and 36 deletions
@@ -0,0 +1,117 @@
+using Microsoft.Data.SqlClient;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Diagnostics;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Configuration.Entities;
+using ZB.MOM.WW.OtOpcUa.Configuration.Enums;
+using ZB.MOM.WW.OtOpcUa.Configuration.Validation;
+
+namespace ZB.MOM.WW.OtOpcUa.Configuration.Tests;
+
+/// <summary>
+///     R03:3.2 on the real provider: the migrated config database allows SNAPSHOT isolation (migration
+///     <c>EnableSnapshotIsolation</c>) and <see cref="DraftSnapshotFactory.FromConfigDbAsync"/> reads under
+///     it. The InMemory tests cannot see either — the provider has no transactions — so this is the one
+///     place the deploy path's "one consistent read" is proven against SQL Server.
+/// </summary>
+[Trait("Category", "SchemaCompliance")]
+[Collection(nameof(SchemaComplianceCollection))]
+public sealed class DraftSnapshotFactorySqlServerTests
+{
+    private readonly SchemaComplianceFixture _fixture;
+
+    /// <summary>Initializes a new instance of the <see cref="DraftSnapshotFactorySqlServerTests"/> class.</summary>
+    /// <param name="fixture">The migrated test database.</param>
+    public DraftSnapshotFactorySqlServerTests(SchemaComplianceFixture fixture) => _fixture = fixture;
+
+    /// <summary>The migration leaves the database with snapshot isolation allowed.</summary>
+    [Fact]
+    public void Migrated_database_allows_snapshot_isolation()
+    {
+        using var conn = _fixture.OpenConnection();
+        using var cmd = new SqlCommand(
+            "SELECT snapshot_isolation_state_desc FROM sys.databases WHERE name = DB_NAME();", conn);
+        cmd.ExecuteScalar().ShouldBe("ON");
+    }
+
+    /// <summary>
+    ///     A row committed by another connection while the snapshot read is in progress is NOT observed by
+    ///     the tables read after it — every table reflects the instant the transaction started. Positive
+    ///     control: a read started after the commit does see the row.
+    /// </summary>
+    [Fact]
+    public async Task FromConfigDb_reads_every_table_as_of_one_instant()
+    {
+        var clusterId = $"SNAP-{Guid.NewGuid():N}"[..24];
+        var scriptId = $"late-{Guid.NewGuid():N}";
+        await using (var seed = NewContext())
+        {
+            seed.ServerClusters.Add(new ServerCluster
+            {
+                ClusterId = clusterId, Name = "snapshot-read", Enterprise = "zb", Site = "test",
+                NodeCount = 1, RedundancyMode = RedundancyMode.None, CreatedBy = "test",
+            });
+            await seed.SaveChangesAsync();
+        }
+
+        try
+        {
+            var options = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>()
+                .UseSqlServer(_fixture.ConnectionString)
+                .AddInterceptors(new LateEditInterceptor(_fixture.ConnectionString, clusterId, scriptId))
+                .Options;
+
+            await using (var db = new OtOpcUaConfigDbContext(options))
+            {
+                var snapshot = await DraftSnapshotFactory.FromConfigDbAsync(db);
+                // The interceptor committed a Script while ServerClusters (the FIRST table read) was being
+                // materialised. Scripts is read later in the same transaction and must not see it.
+                snapshot.ServerClusters.ShouldContain(c => c.ClusterId == clusterId);
+                snapshot.Scripts.ShouldNotContain(s => s.ScriptId == scriptId);
+            }
+
+            await using (var verify = NewContext())
+            {
+                // Positive control: the edit really landed, and a read started afterwards sees it.
+                (await verify.Scripts.AnyAsync(s => s.ScriptId == scriptId)).ShouldBeTrue();
+                var later = await DraftSnapshotFactory.FromConfigDbAsync(verify);
+                later.Scripts.ShouldContain(s => s.ScriptId == scriptId);
+            }
+        }
+        finally
+        {
+            await using var cleanup = NewContext();
+            await cleanup.Scripts.Where(s => s.ScriptId == scriptId).ExecuteDeleteAsync();
+            await cleanup.ServerClusters.Where(c => c.ClusterId == clusterId).ExecuteDeleteAsync();
+        }
+    }
+
+    private OtOpcUaConfigDbContext NewContext() => new(
+        new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseSqlServer(_fixture.ConnectionString).Options);
+
+    /// <summary>
+    ///     Commits a Script on a SEPARATE connection the first time the snapshot read materialises the
+    ///     seeded ServerCluster row — i.e. mid-read, after the first table and before Scripts.
+    /// </summary>
+    private sealed class LateEditInterceptor(string connectionString, string clusterId, string scriptId)
+        : IMaterializationInterceptor
+    {
+        private int _fired;
+
+        public object InitializedInstance(MaterializationInterceptionData materializationData, object entity)
+        {
+            if (entity is ServerCluster c && c.ClusterId == clusterId && Interlocked.Exchange(ref _fired, 1) == 0)
+            {
+                using var other = new OtOpcUaConfigDbContext(
+                    new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseSqlServer(connectionString).Options);
+                other.Scripts.Add(new Script
+                {
+                    ScriptId = scriptId, Name = "late edit", SourceCode = "return 1;", SourceHash = "late-hash",
+                });
+                other.SaveChanges();
+            }
+            return entity;
+        }
+    }
+}
@@ -103,6 +103,36 @@ public sealed class DraftSnapshotFactoryTests : IDisposable
        snapshot.ActiveReservations[0].ReleasedAt.ShouldBeNull();
    }

+    /// <summary>
+    ///     R03:3.2 — the deploy path validates AND composes from ONE snapshot, so the tables the composer
+    ///     needs beyond the validator's set (clusters, nodes, API keys) must ride the same read.
+    /// </summary>
+    [Fact]
+    public async Task FromConfigDb_carries_clusters_nodes_and_api_keys_for_the_composer()
+    {
+        _db.ServerClusters.Add(new ServerCluster
+        {
+            ClusterId = "SITE-A", Name = "Site A", Enterprise = "zb", Site = "site-a",
+            NodeCount = 1, RedundancyMode = RedundancyMode.None, CreatedBy = "test",
+        });
+        _db.ClusterNodes.Add(new ClusterNode
+        {
+            NodeId = "site-a-1:4053", ClusterId = "SITE-A", Host = "site-a-1",
+            ApplicationUri = "urn:OtOpcUa:site-a-1", Enabled = true, CreatedBy = "test",
+        });
+        _db.ApiKeys.Add(new ApiKey
+        {
+            ApiKeyId = "k1", Name = "scada", SecretHash = "hash", Principal = "scadabridge", Enabled = true,
+        });
+        await _db.SaveChangesAsync();
+
+        var snapshot = await DraftSnapshotFactory.FromConfigDbAsync(_db);
+
+        snapshot.ServerClusters.ShouldHaveSingleItem().ClusterId.ShouldBe("SITE-A");
+        snapshot.ClusterNodes.ShouldHaveSingleItem().NodeId.ShouldBe("site-a-1:4053");
+        snapshot.ApiKeys.ShouldHaveSingleItem().ApiKeyId.ShouldBe("k1");
+    }
+
    private void SeedEquipment(string equipmentId)
    {
        var uuid = Guid.NewGuid();
@@ -0,0 +1,133 @@
+using System.IO.Compression;
+using System.Text;
+using Akka.Actor;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.EntityFrameworkCore.Diagnostics;
+using Microsoft.EntityFrameworkCore.Storage;
+using Shouldly;
+using Xunit;
+using ZB.MOM.WW.OtOpcUa.Commons.Messages.Admin;
+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.ControlPlane.AdminOperations;
+using ZB.MOM.WW.OtOpcUa.ControlPlane.Tests.Harness;
+using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
+
+namespace ZB.MOM.WW.OtOpcUa.ControlPlane.Tests;
+
+/// <summary>
+///     R03:3.2 — <c>StartDeployment</c> used to read the whole config DB twice: once for the validator,
+///     once for the composer. An edit landing between the two shipped an artifact the validator never saw.
+///     The deploy now validates and composes ONE snapshot, so an edit that lands mid-deploy is either
+///     validated (it was in the snapshot) or absent from the artifact (it was not) — never shipped unseen.
+/// </summary>
+public sealed class AdminOperationsActorSingleSnapshotTests : ControlPlaneActorTestBase
+{
+    private const string LateEquipmentId = "EQ-late-and-NOT-canonical";
+
+    /// <summary>
+    ///     An Equipment with a non-canonical id (the validator rejects those with <c>EquipmentIdNotDerived</c>)
+    ///     is committed by another context while the snapshot read is in progress, AFTER the Equipment table
+    ///     has been read. The validator cannot see it, so the deploy is Accepted — and the artifact must not
+    ///     carry it either. Positive control: the row really landed in the DB, and a deploy started after it
+    ///     is Rejected on exactly that rule.
+    /// </summary>
+    [Fact]
+    public void Edit_landing_between_validate_and_compose_cannot_ship_unvalidated()
+    {
+        // One shared root: an InMemory store is scoped to the internal service provider, and a context
+        // carrying an interceptor builds its own provider — without the shared root the intercepting
+        // context would read an EMPTY database and the whole test would be vacuous.
+        var dbName = Guid.NewGuid().ToString("N");
+        var root = new InMemoryDatabaseRoot();
+        var plain = new RootedDbFactory(dbName, root, interceptor: null);
+        using (var db = plain.CreateDbContext())
+        {
+            // A Tag to materialise — Tags is read AFTER Equipment by the snapshot factory, so an insert
+            // fired on the first Tag lands "between" the two tables of one read.
+            db.DriverInstances.Add(new DriverInstance
+            {
+                DriverInstanceId = "d", ClusterId = "", Name = "drv", DriverType = "Modbus", DriverConfig = "{}",
+            });
+            db.Devices.Add(new Device { DeviceId = "dev", DriverInstanceId = "d", Name = "dev", DeviceConfig = "{}" });
+            db.Tags.Add(new Tag
+            {
+                TagId = "tag-speed", DeviceId = "dev", Name = "speed", DataType = "Float",
+                AccessLevel = TagAccessLevel.Read, TagConfig = "{}",
+            });
+            db.SaveChanges();
+        }
+
+        var injecting = new RootedDbFactory(dbName, root, new LateEquipmentInterceptor(plain));
+        var coordinator = CreateTestProbe("coord");
+        var actor = Sys.ActorOf(AdminOperationsActor.Props(injecting, coordinator.Ref, Enumerable.Empty<IDriverProbe>()));
+
+        actor.Tell(new StartDeployment("joe", CorrelationId.NewId()));
+
+        var reply = ExpectMsg<StartDeploymentResult>(PresenceBudget);
+        reply.Outcome.ShouldBe(StartDeploymentOutcome.Accepted, reply.Message);
+        coordinator.ExpectMsg<DispatchDeployment>(PresenceBudget);
+
+        using (var verify = plain.CreateDbContext())
+        {
+            // The injection happened...
+            verify.Equipment.Any(e => e.EquipmentId == LateEquipmentId).ShouldBeTrue();
+            // ...and the shipped artifact does not contain what the validator never saw.
+            var row = verify.Deployments.Single();
+            Inflate(row.ArtifactBlob).ShouldNotContain(LateEquipmentId);
+            // The coordinator is a probe here, so seal the row by hand to let the next deploy through.
+            row.Status = DeploymentStatus.Sealed;
+            verify.SaveChanges();
+        }
+
+        // Positive control: now that the row is visible to a fresh read, the gate rejects it.
+        actor.Tell(new StartDeployment("joe", CorrelationId.NewId()));
+        var second = ExpectMsg<StartDeploymentResult>(PresenceBudget);
+        second.Outcome.ShouldBe(StartDeploymentOutcome.Rejected);
+        second.Message.ShouldNotBeNull();
+        second.Message.ShouldContain("EquipmentIdNotDerived");
+    }
+
+    private static string Inflate(byte[] blob)
+    {
+        using var gz = new GZipStream(new MemoryStream(blob), CompressionMode.Decompress);
+        using var reader = new StreamReader(gz, Encoding.UTF8);
+        return reader.ReadToEnd();
+    }
+
+    /// <summary>On the first Tag materialised, commits a non-canonical Equipment through ANOTHER context.</summary>
+    private sealed class LateEquipmentInterceptor(IDbContextFactory<OtOpcUaConfigDbContext> other)
+        : IMaterializationInterceptor
+    {
+        private int _fired;
+
+        public object InitializedInstance(MaterializationInterceptionData materializationData, object entity)
+        {
+            if (entity is Tag && Interlocked.Exchange(ref _fired, 1) == 0)
+            {
+                using var db = other.CreateDbContext();
+                db.Equipment.Add(new Equipment
+                {
+                    EquipmentUuid = Guid.NewGuid(), EquipmentId = LateEquipmentId, Name = "late",
+                    UnsLineId = "line-a", MachineCode = "m",
+                });
+                db.SaveChanges();
+            }
+            return entity;
+        }
+    }
+
+    private sealed class RootedDbFactory(string dbName, InMemoryDatabaseRoot root, IInterceptor? interceptor)
+        : IDbContextFactory<OtOpcUaConfigDbContext>
+    {
+        public OtOpcUaConfigDbContext CreateDbContext()
+        {
+            var builder = new DbContextOptionsBuilder<OtOpcUaConfigDbContext>().UseInMemoryDatabase(dbName, root);
+            if (interceptor is not null) builder.AddInterceptors(interceptor);
+            return new OtOpcUaConfigDbContext(builder.Options);
+        }
+    }
+}
@@ -222,6 +222,47 @@ public sealed class ConfigComposerTests : ControlPlaneActorTestBase
        withoutOverride.ResilienceConfig.ShouldBeNull("an unset override stays null ⇒ tier defaults at spawn");
    }

+    /// <summary>
+    ///     R03:3.2 — the deploy path composes from the <see cref="Configuration.Validation.DraftSnapshot"/> it
+    ///     validated rather than re-reading the DB. <see cref="ConfigComposer.Compose"/> over the factory's
+    ///     snapshot must therefore be byte-identical to the DB overload, or the drift probe (which still
+    ///     uses the DB overload) would report every deployment as drifted.
+    /// </summary>
+    [Fact]
+    public async Task Compose_over_the_factory_snapshot_is_byte_identical_to_the_db_overload()
+    {
+        var f = NewInMemoryDbFactory();
+        await using (var db = f.CreateDbContext())
+        {
+            db.ServerClusters.Add(NewCluster("cluster-b"));
+            db.ServerClusters.Add(NewCluster("cluster-a"));
+            db.ApiKeys.Add(new ApiKey
+            {
+                ApiKeyId = "k1", Name = "scada", SecretHash = "h", Principal = "scadabridge", CreatedUtc = FixedTimestamp,
+            });
+            db.Scripts.Add(new Script { ScriptId = "s-1", Name = "s", SourceCode = "return 1;", SourceHash = "x" });
+            await db.SaveChangesAsync();
+        }
+
+        var viaDb = await ConfigComposer.SnapshotAndFlattenAsync(f.CreateDbContext());
+        var snapshot = await Configuration.Validation.DraftSnapshotFactory.FromConfigDbAsync(f.CreateDbContext());
+        var viaSnapshot = ConfigComposer.Compose(snapshot);
+
+        viaSnapshot.RevisionHash.ShouldBe(viaDb.RevisionHash);
+        viaSnapshot.Blob.ShouldBe(viaDb.Blob);
+        // The composed content is the snapshot's, not something re-read: the keys and clusters are in it.
+        var json = Inflate(viaSnapshot.Blob);
+        json.ShouldContain("\"ApiKeyId\":\"k1\"");
+        json.IndexOf("cluster-a", StringComparison.Ordinal).ShouldBeLessThan(json.IndexOf("cluster-b", StringComparison.Ordinal));
+    }
+
+    private static string Inflate(byte[] blob)
+    {
+        using var gz = new GZipStream(new MemoryStream(blob), CompressionMode.Decompress);
+        using var reader = new StreamReader(gz, Encoding.UTF8);
+        return reader.ReadToEnd();
+    }
+
    private static readonly DateTime FixedTimestamp = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc);

    private static ServerCluster NewCluster(string id) => new()