feat(pending-updates): short-circuit an unchanged fleet on a DB rowversion watermark (#62)

Owner decision (b), 2026-11-09. PendingUpdatesEvaluator re-flattened every deployed instance on each cache-TTL expiry even when nothing had changed. - Migration AddConfigGraphRowVersions: a non-concurrency rowversion column + index on the 18 configuration tables the evaluation reads (Templates + 5 member tables, Instances + 4 override/binding tables, DataConnections, DeployedConfigSnapshots, Sites, SharedScripts, ExternalSystemDefinitions, ExternalSystemMethods, DatabaseConnectionDefinitions). SQL Server only in the model; SQLite test contexts are untouched. Script: docs/plans/sql/AddConfigGraphRowVersions.sql. Joins the coordinated central-pair migration window (#107 item 4). - IConfigGraphWatermarkReader / ConfigGraphWatermarkReader: ONE query of MAX(RowVersion) + COUNT_BIG(*) per table (count catches deletes, which raise no rowversion). rowversion is engine-bumped for any writer, so a peer central node's edit moves it — not the per-process watermark MG-3 removed. - The secret store (outside the DB) gets its own stamp: SHA-256 over ListAsync(includeDeleted: true) metadata. - Evaluator re-serves the last CLEAN snapshot while watermark + secret stamp are unchanged; every PendingUpdatesFullEvaluationEvery'th (default 10, validated >= 1, 1 = off) is full regardless — the safety net for a same-KEK re-wrap or an input table missing from the set. Always full when the token is unavailable/unreadable, after an evaluation that could not compute a site's artifact hash, on Re-check, and after a deploy completion. It only re-serves a real earlier answer, never manufactures one. Tests: SQL Server (real migrated DB) — token stable without writes, moves for a peer-connection INSERT/UPDATE and a non-max DELETE, moves for a local EF write, a detached Update still saves (not a concurrency token), and a guard that every table the evaluator's repository reads touch is covered. Evaluator — reuse without flatten (re-served answer still shows the stale instance), watermark move, 4 secret changes, every-Nth, 5 no-token modes, unclean evaluation not reused, cache Re-check and deploy completion force full. Mutation-proven: token compare, clean check, cache force flag, row count, and an uncovered read each turn tests red. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 22:38:55 -05:00
parent 63ec45e2e3
commit a4806576af
20 changed files with 3963 additions and 9 deletions
@@ -0,0 +1,198 @@
+using System.Text.RegularExpressions;
+using Microsoft.Data.SqlClient;
+using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
+using ZB.MOM.WW.ScadaBridge.Commons.Types.Templates;
+using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
+using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests.Migrations;
+using Xunit;
+
+namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
+
+/// <summary>
+/// Gitea #62: the database-side config watermark behind the pending-updates short-circuit,
+/// against a real migrated SQL Server database (the <c>AddConfigGraphRowVersions</c>
+/// migration included). The "peer node" in these tests is a separate raw
+/// <see cref="SqlConnection"/> that never touches this process's DbContext — the exact shape
+/// of a write made by the other central node, which a per-process watermark never sees
+/// (arch-review 2026-09-09 MG-3).
+/// </summary>
+public class ConfigGraphWatermarkReaderTests : IClassFixture<MsSqlMigrationFixture>
+{
+    private readonly MsSqlMigrationFixture _fixture;
+
+    /// <summary>Initializes the suite with the shared MS SQL migration fixture.</summary>
+    /// <param name="fixture">The shared MS SQL migration fixture.</param>
+    public ConfigGraphWatermarkReaderTests(MsSqlMigrationFixture fixture) => _fixture = fixture;
+
+    private ScadaBridgeDbContext CreateContext(SqlReaderCommandRecorder? recorder = null)
+    {
+        var builder = new DbContextOptionsBuilder<ScadaBridgeDbContext>().UseSqlServer(_fixture.ConnectionString);
+        if (recorder is not null)
+            builder.AddInterceptors(recorder);
+        return new ScadaBridgeDbContext(builder.Options);
+    }
+
+    private async Task<string> ReadAsync()
+    {
+        await using var context = CreateContext();
+        var token = await new ConfigGraphWatermarkReader(context).ReadAsync();
+        Assert.NotNull(token);
+        return token!;
+    }
+
+    private void PeerExecute(string sql, params SqlParameter[] parameters)
+    {
+        using var connection = _fixture.OpenConnection();
+        using var command = connection.CreateCommand();
+        command.CommandText = sql;
+        command.Parameters.AddRange(parameters);
+        command.ExecuteNonQuery();
+    }
+
+    /// <summary>
+    /// Verifies that the token is stable with no writes, and moves for a peer-node insert, update
+    /// and delete made on a connection this process's context never saw.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task Token_IsStableWithoutWrites_AndMovesForPeerInsertUpdateAndDelete()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+        var ident = "wm-" + Guid.NewGuid().ToString("N")[..8];
+
+        var t0 = await ReadAsync();
+        Assert.Equal(t0, await ReadAsync()); // no write → identical token (the short-circuit case)
+
+        // Peer INSERT.
+        PeerExecute("INSERT INTO [Sites] ([Name], [SiteIdentifier]) VALUES (@n, @i);",
+            new SqlParameter("@n", ident), new SqlParameter("@i", ident));
+        var t1 = await ReadAsync();
+        Assert.NotEqual(t0, t1);
+        Assert.Equal(t1, await ReadAsync());
+
+        // Peer UPDATE of a column no application write path ever "bumps" anything for.
+        PeerExecute("UPDATE [Sites] SET [Description] = N'edited by the peer' WHERE [SiteIdentifier] = @i;",
+            new SqlParameter("@i", ident));
+        var t2 = await ReadAsync();
+        Assert.NotEqual(t1, t2);
+
+        // Peer DELETE of a row that is NOT the table's newest — so MAX(RowVersion) stays put and
+        // only the row count can catch it. A second, newer row is inserted first to guarantee that.
+        var newer = ident + "-b";
+        PeerExecute("INSERT INTO [Sites] ([Name], [SiteIdentifier]) VALUES (@n, @i);",
+            new SqlParameter("@n", newer), new SqlParameter("@i", newer));
+        var t3 = await ReadAsync();
+        PeerExecute("DELETE FROM [Sites] WHERE [SiteIdentifier] = @i;", new SqlParameter("@i", ident));
+        var t4 = await ReadAsync();
+        Assert.NotEqual(t3, t4);
+    }
+
+    /// <summary>Verifies that an EF write through this process's context also moves the token.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task Token_MovesForALocalEfWrite_AndTheRowVersionIsNotAConcurrencyToken()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+        var ident = "wm-" + Guid.NewGuid().ToString("N")[..8];
+        var before = await ReadAsync();
+
+        int id;
+        await using (var context = CreateContext())
+        {
+            var site = new Site(ident, ident);
+            context.Sites.Add(site);
+            await context.SaveChangesAsync();
+            id = site.Id;
+        }
+
+        var afterInsert = await ReadAsync();
+        Assert.NotEqual(before, afterInsert);
+
+        // A detached Update of a stale copy must still save: the column is change detection only,
+        // never an optimistic-concurrency check that would fail a save the old schema accepted.
+        await using (var context = CreateContext())
+        {
+            context.Sites.Update(new Site(ident, ident) { Id = id, Description = "detached update" });
+            await context.SaveChangesAsync();
+        }
+
+        Assert.NotEqual(afterInsert, await ReadAsync());
+    }
+
+    /// <summary>
+    /// Guard for the covered set: every table the pending-updates evaluator's repository reads
+    /// touch must carry the watermark column, so a new input table cannot silently escape it.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task EveryTableThePendingUpdatesReadsTouch_IsCoveredByTheWatermark()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+        // Seed one site → template → instance so the split-query child reads actually execute
+        // (EF skips a split query's child statements when the root row is absent).
+        var ident = "wm-" + Guid.NewGuid().ToString("N")[..8];
+        int siteId, templateId, instanceId;
+        await using (var seed = CreateContext())
+        {
+            var site = new Site(ident, ident);
+            var template = new ZB.MOM.WW.ScadaBridge.Commons.Entities.Templates.Template(ident);
+            seed.Sites.Add(site);
+            seed.Templates.Add(template);
+            await seed.SaveChangesAsync();
+            var instance = new ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances.Instance(ident)
+            {
+                SiteId = site.Id,
+                TemplateId = template.Id,
+            };
+            seed.Instances.Add(instance);
+            await seed.SaveChangesAsync();
+            (siteId, templateId, instanceId) = (site.Id, template.Id, instance.Id);
+        }
+
+        var recorder = new SqlReaderCommandRecorder();
+        await using var context = CreateContext(recorder);
+        var watermark = new TemplateGraphWatermark();
+        var sites = new SiteRepository(context, watermark);
+        var templates = new TemplateEngineRepository(context, watermark);
+        var deployments = new DeploymentManagerRepository(context);
+        var external = new ExternalSystemRepository(context);
+
+        // PendingUpdatesEvaluator → ArtifactDeploymentService.ComputeCurrentArtifactHashesAsync
+        await sites.GetAllSitesAsync();
+        await templates.GetAllSharedScriptsAsync();
+        await external.GetAllExternalSystemsAsync();
+        await external.GetAllDatabaseConnectionsAsync();
+        await deployments.GetAllExternalSystemMethodsAsync();
+        await sites.GetDataConnectionsBySiteIdAsync(siteId);
+        // → DeploymentService.GetInstanceStalenessAsync / FlatteningPipeline.FlattenCoreAsync
+        await deployments.GetDeployedInstanceSummariesAsync();
+        await deployments.GetDeployedRevisionHashesAsync();
+        await deployments.GetDeployedSnapshotsByInstanceIdsAsync([instanceId]);
+        await templates.GetInstanceByIdAsync(instanceId);
+        await templates.GetTemplateWithChildrenAsync(templateId);
+
+        var covered = ConfigGraphRowVersions.EntityTypes
+            .Select(t => context.Model.FindEntityType(t)!.GetTableName()!)
+            .ToHashSet(StringComparer.OrdinalIgnoreCase);
+        var read = recorder.Commands
+            .SelectMany(c => Regex.Matches(c, @"(?:FROM|JOIN)\s+(?:\[\w+\]\.)?\[(\w+)\]").Select(m => m.Groups[1].Value))
+            .ToHashSet(StringComparer.OrdinalIgnoreCase);
+
+        Assert.NotEmpty(read);
+        Assert.Empty(read.Except(covered));
+        // Positive control: the regex really does see the tables (a vacuous empty set would pass above).
+        Assert.Contains("Templates", read);
+        Assert.Contains("InstanceAttributeOverrides", read);
+        Assert.Contains("DeployedConfigSnapshots", read);
+    }
+
+    /// <summary>Verifies that a non-SQL-Server provider yields no token, so callers never short-circuit on it.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task NonSqlServerProvider_ReturnsNull()
+    {
+        await using var context = SqliteTestHelper.CreateInMemoryContext();
+        Assert.Null(await new ConfigGraphWatermarkReader(context).ReadAsync());
+    }
+}