feat(deploy): wire periodic PendingDeployment purge + SQL Server same-id re-stage test

Notify-and-fetch follow-ups: - PendingDeploymentPurgeActor: a central cluster singleton (not readiness-gated, best-effort) that sweeps expired PendingDeployment staging rows on CommunicationOptions.PendingDeploymentPurgeInterval (default 1h). Modeled on the kpi-history-recorder pattern: self-scheduling timer, per-tick DI scope -> IDeploymentManagerRepository, continue-on-error. Wired in AkkaHostedService.RegisterCentralActors (manager + proxy + drain); resolves the deferred TODO in DeploymentService. Correctness never depends on it (supersession bounds rows to <=1/instance; the fetch endpoint enforces the TTL), so it is deliberately absent from RequiredSingletonsHealthCheck. - SQL Server integration test for StagePendingIfAbsentAsync re-staging an instance's OWN DeploymentId over an expired row against the real UNIQUE index on DeploymentId — confirms EF orders DELETE before INSERT in one SaveChanges (SQLite's constraint timing differs from SQL Server's). Plus a same-instance supersession variant on real SQL Server. Tests: 2 TestKit actor tests + 2 SQL Server integration tests (both ran green against the infra MSSQL container); 235 Communication + 15 PendingDeployment tests pass; Host builds 0 warnings.
2026-06-26 23:19:29 -04:00
parent d9f5fbb664
commit 06f2df4f89
6 changed files with 431 additions and 4 deletions
@@ -0,0 +1,98 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NSubstitute;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Communication.Actors;
+
+namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;
+
+/// <summary>
+/// Tests for <see cref="PendingDeploymentPurgeActor"/> — the notify-and-fetch central
+/// singleton that sweeps expired PendingDeployment staging rows. The actor is thin glue
+/// over <c>IDeploymentManagerRepository.PurgeExpiredPendingDeploymentsAsync</c> (whose
+/// row-level semantics are covered by the repository tests); these tests pin the actor's
+/// own policy: it opens a per-tick DI scope, calls the purge with a now-ish threshold,
+/// and — being best-effort hygiene — survives a throwing purge and keeps ticking.
+/// </summary>
+/// <remarks>
+/// The substitute's return is driven by a closure with an <see cref="System.Threading.Interlocked"/>
+/// counter rather than NSubstitute's <c>ReceivedCalls()</c>, because the repeating
+/// scheduler timer records calls on a background thread concurrently with the test
+/// thread's assertions — querying NSubstitute's call list under that concurrency can
+/// throw "collection was modified". A fenced counter is safe to read while the timer runs.
+/// </remarks>
+public class PendingDeploymentPurgeActorTests : TestKit
+{
+    private IActorRef CreateActor(IDeploymentManagerRepository repo, TimeSpan interval)
+    {
+        var services = new ServiceCollection();
+        // Mirror AddConfigurationDatabase: IDeploymentManagerRepository is scoped, so the
+        // actor opens a fresh scope per tick and resolves the repository there.
+        services.AddScoped(_ => repo);
+        var sp = services.BuildServiceProvider();
+
+        var options = Options.Create(new CommunicationOptions { PendingDeploymentPurgeInterval = interval });
+        return Sys.ActorOf(Props.Create(() => new PendingDeploymentPurgeActor(
+            sp, options, NullLogger<PendingDeploymentPurgeActor>.Instance)));
+    }
+
+    [Fact]
+    public void Tick_PurgesExpiredRows_WithNowThreshold()
+    {
+        var calls = 0;
+        DateTimeOffset captured = default;
+        var repo = Substitute.For<IDeploymentManagerRepository>();
+        repo.PurgeExpiredPendingDeploymentsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
+            .Returns(ci =>
+            {
+                captured = ci.Arg<DateTimeOffset>();
+                Interlocked.Increment(ref calls);
+                return 3;
+            });
+
+        // Large interval so the PreStart timer never fires during the test — only the
+        // single manual tick below drives a purge, making the count deterministic.
+        var actor = CreateActor(repo, TimeSpan.FromHours(1));
+
+        var before = DateTimeOffset.UtcNow;
+        actor.Tell(PendingDeploymentPurgeActor.PurgeTick.Instance);
+
+        AwaitAssert(
+            () => Assert.Equal(1, Volatile.Read(ref calls)),
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+
+        // The purge threshold is "now" — bounded by the window around the tick. Reading
+        // `captured` after the fenced counter read is safe (Interlocked establishes the
+        // happens-before with the write that precedes the increment).
+        var after = DateTimeOffset.UtcNow;
+        Assert.InRange(captured, before, after);
+    }
+
+    [Fact]
+    public void Tick_PurgeThrows_ActorSurvives_AndRetriesOnNextInterval()
+    {
+        var calls = 0;
+        var repo = Substitute.For<IDeploymentManagerRepository>();
+        repo.PurgeExpiredPendingDeploymentsAsync(Arg.Any<DateTimeOffset>(), Arg.Any<CancellationToken>())
+            .Returns<int>(_ =>
+            {
+                Interlocked.Increment(ref calls);
+                throw new InvalidOperationException("simulated purge failure");
+            });
+
+        // Fast interval so the self-scheduling timer fires repeatedly; the actor must
+        // swallow each per-tick exception and keep ticking (best-effort hygiene — a
+        // crash here would drop the singleton until failover).
+        CreateActor(repo, TimeSpan.FromMilliseconds(100));
+
+        AwaitAssert(
+            () => Assert.True(Volatile.Read(ref calls) >= 2,
+                $"expected >= 2 ticks despite failures, got {Volatile.Read(ref calls)}"),
+            duration: TimeSpan.FromSeconds(5),
+            interval: TimeSpan.FromMilliseconds(50));
+    }
+}
@@ -0,0 +1,141 @@
+using Microsoft.EntityFrameworkCore;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Deployment;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Instances;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.Sites;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.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.Repositories;
+
+/// <summary>
+/// SQL Server integration coverage for the notify-and-fetch <see cref="PendingDeployment"/>
+/// staging store's <b>same-DeploymentId re-stage</b> path. The SQLite in-memory fixture
+/// (<c>PendingDeploymentRepositoryTests</c>) covers the logic, but the delete-before-insert
+/// ordering that lets <see cref="DeploymentManagerRepository.StagePendingIfAbsentAsync"/>
+/// re-stage an instance's OWN <c>DeploymentId</c> over an expired row depends on EF emitting
+/// the DELETE before the INSERT within a single <c>SaveChanges</c> — against the production
+/// UNIQUE index on <c>DeploymentId</c>. SQLite's constraint timing differs from SQL Server's
+/// (SQLite defers/relaxes within a transaction where SQL Server enforces per-statement), so
+/// this class asserts the behaviour against the real migrated schema via
+/// <see cref="MsSqlMigrationFixture"/> rather than the SQLite provider.
+/// </summary>
+public class PendingDeploymentRepositoryIntegrationTests : IClassFixture<MsSqlMigrationFixture>
+{
+    private readonly MsSqlMigrationFixture _fixture;
+
+    public PendingDeploymentRepositoryIntegrationTests(MsSqlMigrationFixture fixture)
+    {
+        _fixture = fixture;
+    }
+
+    [SkippableFact]
+    public async Task StagePendingIfAbsent_ExpiredRowSameDeploymentId_ReStages_NoUniqueViolation()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+
+        var instanceId = await SeedInstanceAsync("ReStageInst");
+        var now = DateTimeOffset.UtcNow;
+
+        // Seed an EXPIRED pending row carrying DeploymentId "D1". A startup reconcile re-stages
+        // the DeployedConfigSnapshot's OWN DeploymentId, so the fresh row reuses "D1" — the exact
+        // collision the unique index would reject if the DELETE didn't precede the INSERT.
+        await using (var seedContext = CreateContext())
+        {
+            seedContext.Set<PendingDeployment>().Add(new PendingDeployment(
+                "D1", instanceId, "rev-old", "{\"old\":true}", "tok-old",
+                now.AddMinutes(-20), now.AddMinutes(-10)));
+            await seedContext.SaveChangesAsync();
+        }
+
+        // Re-stage "D1" over the expired row. Against the real UNIQUE index on DeploymentId this
+        // MUST succeed (EF orders delete-before-insert in the single SaveChanges) rather than
+        // throwing SqlException 2627/2601.
+        await using (var context = CreateContext())
+        {
+            var repo = new DeploymentManagerRepository(context);
+            var staged = await repo.StagePendingIfAbsentAsync(
+                instanceId, "D1", "rev-new", "{\"new\":true}", "tok-new",
+                now, now.AddMinutes(10));
+            Assert.True(staged);
+        }
+
+        // Exactly one row for "D1" survives — the fresh one.
+        await using (var readContext = CreateContext())
+        {
+            var rows = await readContext.Set<PendingDeployment>()
+                .Where(p => p.DeploymentId == "D1")
+                .ToListAsync();
+            Assert.Single(rows);
+            Assert.Equal("tok-new", rows[0].Token);
+            Assert.Equal("{\"new\":true}", rows[0].ConfigurationJson);
+            Assert.True(rows[0].ExpiresAtUtc > now);
+        }
+    }
+
+    [SkippableFact]
+    public async Task AddPendingDeployment_NewDeployDifferentId_SupersedesPriorRow_OnRealSqlServer()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+
+        var instanceId = await SeedInstanceAsync("SupersedeInst");
+        var now = DateTimeOffset.UtcNow;
+
+        await using (var context = CreateContext())
+        {
+            var repo = new DeploymentManagerRepository(context);
+            await repo.AddPendingDeploymentAsync(new PendingDeployment(
+                "dep-A", instanceId, "rev-a", "{\"v\":1}", "tok-a", now, now.AddMinutes(5)));
+            await repo.SaveChangesAsync();
+        }
+
+        // A newer deploy for the SAME instance with a DIFFERENT DeploymentId must replace the
+        // prior row (delete-then-insert under the per-instance operation lock) with no FK or
+        // unique conflict on real SQL Server.
+        await using (var context = CreateContext())
+        {
+            var repo = new DeploymentManagerRepository(context);
+            await repo.AddPendingDeploymentAsync(new PendingDeployment(
+                "dep-B", instanceId, "rev-b", "{\"v\":2}", "tok-b", now, now.AddMinutes(5)));
+            await repo.SaveChangesAsync();
+        }
+
+        await using (var readContext = CreateContext())
+        {
+            var rows = await readContext.Set<PendingDeployment>()
+                .Where(p => p.InstanceId == instanceId)
+                .ToListAsync();
+            Assert.Single(rows);
+            Assert.Equal("dep-B", rows[0].DeploymentId);
+            Assert.Equal("{\"v\":2}", rows[0].ConfigurationJson);
+        }
+    }
+
+    /// <summary>
+    /// Seeds an Instance (with its required Site + Template) against the fixture database and
+    /// returns its generated id. The PendingDeployment → Instance FK requires a real parent row.
+    /// </summary>
+    private async Task<int> SeedInstanceAsync(string uniqueName)
+    {
+        await using var context = CreateContext();
+        var site = new Site($"Site-{uniqueName}", $"S-{uniqueName}");
+        var template = new Template($"T-{uniqueName}");
+        context.Sites.Add(site);
+        context.Templates.Add(template);
+        await context.SaveChangesAsync();
+
+        var instance = new Instance(uniqueName) { SiteId = site.Id, TemplateId = template.Id };
+        context.Instances.Add(instance);
+        await context.SaveChangesAsync();
+        return instance.Id;
+    }
+
+    private ScadaBridgeDbContext CreateContext()
+    {
+        var options = new DbContextOptionsBuilder<ScadaBridgeDbContext>()
+            .UseSqlServer(_fixture.ConnectionString)
+            .Options;
+        return new ScadaBridgeDbContext(options);
+    }
+}