fix(management): purge terminal PendingSecuredWrites rows on the list sweep after a retention window (review 2026-10-21 08/G3)

2026-10-22 09:25:48 -04:00
parent baab6ac78e
commit d495494074
8 changed files with 237 additions and 1 deletions
@@ -172,6 +172,8 @@ The two-person authorization workflow for writes through the MxAccess Gateway. B

 **Server-side pending TTL (arch-review S2).** A `Pending` secured write is not valid indefinitely. Both `ApproveSecuredWriteCommand` and `RejectSecuredWriteCommand` call a shared `TryExpireIfStaleAsync` helper immediately after the `Pending`-status check: if the row's `SubmittedAtUtc` is older than the configured `SecuredWritePendingTtl` (`ManagementServiceOptions`, default **24 h**), the helper atomically transitions it `Pending → Expired` via `ISecuredWriteRepository.TryMarkExpiredAsync` (compare-and-swap — no verifier recorded, `VerifierUser` stays `NULL`), emits a best-effort `SecuredWriteExpire` audit row (system actor, `Discarded` status — the closest terminal `AuditStatus`), and the command fails with an "expired" error. A stale setpoint therefore can **never** be approved or executed days later, and no `WriteTagRequest` is relayed. A non-positive TTL disables expiry. `ListSecuredWritesCommand` additionally runs an **opportunistic sweep**: on every read it walks the returned page and calls the same helper for each `Pending` row, so overdue writes surface as `Expired` the next time anyone views the queue — no dedicated expiry-timer singleton is needed because the ManagementActor already runs on every central node and the CAS makes concurrent multi-node sweeps idempotent.

+**Terminal-row purge (review 2026-10-21 08/G3).** The same list sweep also purges decided/executed rows — `Approved`, `Rejected`, `Expired`, `Executed`, `Failed` — whose last lifecycle instant (`ExecutedAtUtc`, else `DecidedAtUtc`, else `SubmittedAtUtc`) is older than `SecuredWriteTerminalRetention` (default **90 days**), via `ISecuredWriteRepository.PurgeTerminalAsync` (bounded set-based delete, 1 000 rows per pass, so a backlog converges over successive passes). `Pending` rows are never purged — they expire first. The pass is throttled per node to once per `SecuredWriteTerminalPurgeInterval` (default 1 h), is idempotent across both central nodes, and is best-effort (a failure is logged and never fails the list). `PendingSecuredWrites` is operational state; the append-only `AuditLog` `SecuredWrite*` rows remain the lifecycle history after a purge. Previously no component deleted rows, so the table and the list's offset-paged scan grew forever.
+
 ### External Systems

 - **ListExternalSystems** / **GetExternalSystem**: Query external system definitions.
@@ -302,6 +304,8 @@ The ManagementActor receives the following services and repositories via DI (inj
 | `ScadaBridge:ManagementService` | `ManagementServiceOptions` | `CommandTimeout` (`TimeSpan`, default 30 s) — Ask timeout the HTTP endpoint applies when forwarding to the `ManagementActor`. A non-positive configured value falls back to the 30 s default. |
 | | | `LongRunningCommandTimeout` (`TimeSpan`, default 5 min) — Ask timeout applied to long-running commands (`ImportBundle`, `PreviewBundle`, `ExportBundle`, `MgmtDeployArtifacts`, `MgmtDeployInstance`, `MgmtDeploySite`); all other commands use `CommandTimeout`. A non-positive configured value falls back to the 5 min default. |
 | | | `SecuredWritePendingTtl` (`TimeSpan`, default 24 h) — age after which a `Pending` secured write is transitioned to `Expired` and can no longer be approved/executed; enforced at approve/reject and swept opportunistically on list. A non-positive value disables expiry (arch-review S2). |
+| | | `SecuredWriteTerminalRetention` (`TimeSpan`, default 90 days) — age (from the row's last lifecycle instant) after which an `Approved`/`Rejected`/`Expired`/`Executed`/`Failed` secured-write row is purged on the list sweep. A non-positive value disables the purge (review 2026-10-21 08/G3). |
+| | | `SecuredWriteTerminalPurgeInterval` (`TimeSpan`, default 1 h, must be ≥ 0) — minimum interval between purge passes on one node; zero purges on every list. |

 ## Dependencies

@@ -153,4 +153,18 @@ public interface ISecuredWriteRepository
        string? siteId,
        IReadOnlyCollection<string>? permittedSiteIds = null,
        CancellationToken ct = default);
+
+    /// <summary>
+    /// Deletes up to <paramref name="batchSize"/> rows in a post-Pending status
+    /// (<c>Approved</c>, <c>Rejected</c>, <c>Expired</c>, <c>Executed</c>, <c>Failed</c>) whose
+    /// last lifecycle instant — <c>ExecutedAtUtc</c>, else <c>DecidedAtUtc</c>, else
+    /// <c>SubmittedAtUtc</c> — is older than <paramref name="olderThanUtc"/>. <c>Pending</c>
+    /// rows are never purged (they are expired first). The table is operational state; the
+    /// append-only <c>AuditLog</c> keeps the lifecycle history (review 2026-10-21 08/G3).
+    /// </summary>
+    /// <param name="olderThanUtc">Rows whose last lifecycle instant is before this are purged.</param>
+    /// <param name="batchSize">Maximum rows deleted in this call.</param>
+    /// <param name="ct">Cancellation token.</param>
+    /// <returns>A task that resolves to the number of rows deleted.</returns>
+    Task<int> PurgeTerminalAsync(DateTime olderThanUtc, int batchSize, CancellationToken ct = default);
 }
@@ -188,4 +188,25 @@ WHERE Id = {id}

        return await query.CountAsync(ct);
    }
+
+    /// <inheritdoc />
+    public async Task<int> PurgeTerminalAsync(DateTime olderThanUtc, int batchSize, CancellationToken ct = default)
+    {
+        if (batchSize <= 0)
+        {
+            return 0;
+        }
+
+        // Bounded set-based delete (DELETE ... WHERE Id IN (SELECT TOP(n) ...)): never a
+        // whole-table delete in one statement, so a first purge of a long-lived table cannot
+        // hold a large lock. Pending is excluded — it is expired first, then purged later.
+        return await _context.Set<PendingSecuredWrite>()
+            .Where(p => TerminalStatuses.Contains(p.Status)
+                && (p.ExecutedAtUtc ?? p.DecidedAtUtc ?? p.SubmittedAtUtc) < olderThanUtc)
+            .OrderBy(p => p.Id)
+            .Take(batchSize)
+            .ExecuteDeleteAsync(ct);
+    }
+
+    private static readonly string[] TerminalStatuses = ["Approved", "Rejected", "Expired", "Executed", "Failed"];
 }
@@ -1860,9 +1860,56 @@ public class ManagementActor : ReceiveActor
                await TryExpireIfStaleAsync(sp, repo, row);
        }

+        await TryPurgeTerminalSecuredWritesAsync(sp, repo);
+
        return new SecuredWriteListResult(rows.Select(ToSecuredWriteDto).ToList(), totalCount);
    }

+    // Last terminal-purge pass on this node (UTC ticks); throttles the purge that rides the
+    // list sweep to once per SecuredWriteTerminalPurgeInterval.
+    private static long _lastSecuredWritePurgeTicks;
+
+    private const int SecuredWritePurgeBatchSize = 1000;
+
+    /// <summary>
+    /// Purges decided/executed secured-write rows older than
+    /// <see cref="ManagementServiceOptions.SecuredWriteTerminalRetention"/> (review 2026-10-21
+    /// 08/G3 — the table used to grow forever and the list's offset paging scanned it all).
+    /// Rides the same opportunistic list sweep as Pending expiry rather than a dedicated
+    /// timer, throttled per node; one bounded batch per pass, so a backlog converges over
+    /// successive passes. Both central nodes may purge concurrently — the set-based delete is
+    /// idempotent. Best-effort: a failure is logged and never fails the list.
+    /// </summary>
+    private static async Task TryPurgeTerminalSecuredWritesAsync(IServiceProvider sp, ISecuredWriteRepository repo)
+    {
+        var options = sp.GetService<IOptions<ManagementServiceOptions>>()?.Value ?? new ManagementServiceOptions();
+        var retention = options.SecuredWriteTerminalRetention;
+        if (retention <= TimeSpan.Zero)
+            return;
+
+        var now = DateTime.UtcNow;
+        var last = Interlocked.Read(ref _lastSecuredWritePurgeTicks);
+        if (options.SecuredWriteTerminalPurgeInterval > TimeSpan.Zero
+            && now.Ticks - last < options.SecuredWriteTerminalPurgeInterval.Ticks)
+            return;
+        // One winner per interval on this node; a concurrent list skips the pass.
+        if (Interlocked.CompareExchange(ref _lastSecuredWritePurgeTicks, now.Ticks, last) != last)
+            return;
+
+        try
+        {
+            var deleted = await repo.PurgeTerminalAsync(now - retention, SecuredWritePurgeBatchSize);
+            if (deleted > 0)
+                sp.GetService<ILogger<ManagementActor>>()?.LogInformation(
+                    "Purged {Count} terminal secured-write row(s) older than {Retention}.", deleted, retention);
+        }
+        catch (Exception ex)
+        {
+            sp.GetService<ILogger<ManagementActor>>()?.LogWarning(
+                ex, "Best-effort terminal secured-write purge failed; the list is unaffected.");
+        }
+    }
+
    // ========================================================================
    // Site handlers
    // ========================================================================
@@ -10,4 +10,10 @@ public class ManagementServiceOptions

    /// <summary>Age after which a Pending secured write can no longer be approved; it is transitioned to Expired at the next decision/list touch. Default 24 hours. A non-positive value disables server-side expiry.</summary>
    public TimeSpan SecuredWritePendingTtl { get; set; } = TimeSpan.FromHours(24);
+
+    /// <summary>Age after which a decided/executed secured-write row (Approved/Rejected/Expired/Executed/Failed, measured from its last lifecycle instant) is purged from <c>PendingSecuredWrites</c>; the AuditLog keeps the history. Default 90 days. A non-positive value disables the purge.</summary>
+    public TimeSpan SecuredWriteTerminalRetention { get; set; } = TimeSpan.FromDays(90);
+
+    /// <summary>Minimum interval between terminal secured-write purge passes on one node (the purge rides the Secured Writes list sweep). Default 1 hour; zero runs it on every list.</summary>
+    public TimeSpan SecuredWriteTerminalPurgeInterval { get; set; } = TimeSpan.FromHours(1);
 }
@@ -11,7 +11,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService;
 /// section fails fast at boot with a clear, key-naming message.
 /// <para>
 /// <see cref="ManagementServiceOptions.SecuredWritePendingTtl"/> is intentionally NOT required positive:
-/// a non-positive value legitimately disables server-side secured-write expiry.
+/// a non-positive value legitimately disables server-side secured-write expiry. The same holds for
+/// <see cref="ManagementServiceOptions.SecuredWriteTerminalRetention"/> (non-positive disables the terminal-row purge).
 /// </para>
 /// </summary>
 public sealed class ManagementServiceOptionsValidator : OptionsValidatorBase<ManagementServiceOptions>
@@ -26,5 +27,9 @@ public sealed class ManagementServiceOptionsValidator : OptionsValidatorBase<Man
        builder.RequireThat(options.LongRunningCommandTimeout > FlowDebug.FlowDebugRunner.ResponseGrace,
            $"ScadaBridge:ManagementService:LongRunningCommandTimeout must exceed the 5-second response grace " +
            $"(was {options.LongRunningCommandTimeout}); it is the Ask timeout for transport/deploy commands.");
+
+        builder.RequireThat(options.SecuredWriteTerminalPurgeInterval >= TimeSpan.Zero,
+            $"ScadaBridge:ManagementService:SecuredWriteTerminalPurgeInterval must not be negative " +
+            $"(was {options.SecuredWriteTerminalPurgeInterval}); zero purges on every Secured Writes list.");
    }
 }
@@ -358,6 +358,53 @@ public class SecuredWriteRepositoryTests : IClassFixture<MsSqlMigrationFixture>
        Assert.Equal(4, rows.Count);
    }

+    /// <summary>
+    /// Review 2026-10-21 08/G3: the terminal purge deletes decided/executed rows whose last
+    /// lifecycle instant is before the cutoff, and never touches Pending or recent rows.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task PurgeTerminalAsync_DeletesOnlyOldTerminalRows()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+
+        var siteId = NewSiteId();
+        var old = new DateTime(2001, 1, 1, 0, 0, 0, DateTimeKind.Utc);
+        long oldExecuted, oldRejected, oldPending, recentlyExecuted;
+        await using (var context = CreateContext())
+        {
+            var repo = new SecuredWriteRepository(context);
+            oldExecuted = await repo.AddAsync(With(NewRow(siteId, "Executed"), old, executedAt: old.AddMinutes(1)));
+            oldRejected = await repo.AddAsync(With(NewRow(siteId, "Rejected"), old, decidedAt: old.AddMinutes(1)));
+            oldPending = await repo.AddAsync(With(NewRow(siteId, "Pending"), old));
+            // Submitted long ago but executed recently: its LAST lifecycle instant is recent.
+            recentlyExecuted = await repo.AddAsync(With(NewRow(siteId, "Executed"), old, executedAt: DateTime.UtcNow));
+        }
+
+        await using (var purgeContext = CreateContext())
+        {
+            var deleted = await new SecuredWriteRepository(purgeContext)
+                .PurgeTerminalAsync(new DateTime(2002, 1, 1, 0, 0, 0, DateTimeKind.Utc), batchSize: 1000);
+            Assert.True(deleted >= 2);
+        }
+
+        await using var verify = CreateContext();
+        var verifyRepo = new SecuredWriteRepository(verify);
+        Assert.Null(await verifyRepo.GetAsync(oldExecuted));
+        Assert.Null(await verifyRepo.GetAsync(oldRejected));
+        Assert.NotNull(await verifyRepo.GetAsync(oldPending));
+        Assert.NotNull(await verifyRepo.GetAsync(recentlyExecuted));
+    }
+
+    private static PendingSecuredWrite With(
+        PendingSecuredWrite row, DateTime submittedAt, DateTime? decidedAt = null, DateTime? executedAt = null)
+    {
+        row.SubmittedAtUtc = submittedAt;
+        row.DecidedAtUtc = decidedAt;
+        row.ExecutedAtUtc = executedAt;
+        return row;
+    }
+
    // --- helpers ------------------------------------------------------------

    private ScadaBridgeDbContext CreateContext()
@@ -0,0 +1,92 @@
+using Akka.Actor;
+using Akka.TestKit.Xunit2;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+using NSubstitute;
+using NSubstitute.ExceptionExtensions;
+using ZB.MOM.WW.ScadaBridge.Commons.Entities.SecuredWrites;
+using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
+using ZB.MOM.WW.ScadaBridge.Commons.Messages.Management;
+
+namespace ZB.MOM.WW.ScadaBridge.ManagementService.Tests;
+
+/// <summary>
+/// Review 2026-10-21 08/G3: terminal <c>PendingSecuredWrites</c> rows were never purged.
+/// The Secured Writes list sweep (which already expires overdue Pending rows) now also
+/// purges decided/executed rows older than <c>SecuredWriteTerminalRetention</c>.
+/// </summary>
+public class SecuredWriteTerminalPurgeTests : TestKit
+{
+    private readonly ISecuredWriteRepository _repo = Substitute.For<ISecuredWriteRepository>();
+    private readonly ServiceCollection _services = new ServiceCollection().AddTestConfigurationTransaction();
+
+    /// <summary>Initializes the harness with an empty secured-write table.</summary>
+    public SecuredWriteTerminalPurgeTests()
+    {
+        _services.AddScoped(_ => _repo);
+        _repo.QueryAsync(Arg.Any<string?>(), Arg.Any<string?>(), Arg.Any<int>(), Arg.Any<int>(),
+                Arg.Any<IReadOnlyCollection<string>?>(), Arg.Any<CancellationToken>())
+            .Returns(Array.Empty<PendingSecuredWrite>());
+    }
+
+    private void UseOptions(ManagementServiceOptions options) =>
+        _services.AddSingleton<IOptions<ManagementServiceOptions>>(Options.Create(options));
+
+    private IActorRef CreateActor() =>
+        Sys.ActorOf(Props.Create(() => new ManagementActor(
+            _services.BuildServiceProvider(), NullLogger<ManagementActor>.Instance)));
+
+    private static ManagementEnvelope List() =>
+        new(new AuthenticatedUser("admin", "Admin", new[] { "Administrator" }, Array.Empty<string>()),
+            new ListSecuredWritesCommand(null, null), Guid.NewGuid().ToString("N"));
+
+    /// <summary>The list sweep purges terminal rows older than the retention window.</summary>
+    [Fact]
+    public void List_PurgesTerminalRowsOlderThanRetention()
+    {
+        UseOptions(new ManagementServiceOptions
+        {
+            SecuredWriteTerminalRetention = TimeSpan.FromDays(30),
+            SecuredWriteTerminalPurgeInterval = TimeSpan.Zero,
+        });
+        DateTime? cutoff = null;
+        _repo.PurgeTerminalAsync(Arg.Do<DateTime>(c => cutoff = c), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(3);
+
+        CreateActor().Tell(List());
+
+        ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
+        Assert.NotNull(cutoff);
+        Assert.InRange(cutoff!.Value, DateTime.UtcNow.AddDays(-30).AddMinutes(-1), DateTime.UtcNow.AddDays(-30).AddMinutes(1));
+    }
+
+    /// <summary>A non-positive retention disables the purge.</summary>
+    [Fact]
+    public void List_WithRetentionDisabled_DoesNotPurge()
+    {
+        UseOptions(new ManagementServiceOptions
+        {
+            SecuredWriteTerminalRetention = TimeSpan.Zero,
+            SecuredWriteTerminalPurgeInterval = TimeSpan.Zero,
+        });
+
+        CreateActor().Tell(List());
+
+        ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
+        _repo.DidNotReceive().PurgeTerminalAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>());
+    }
+
+    /// <summary>A purge failure is best-effort — the list still answers.</summary>
+    [Fact]
+    public void List_PurgeFailure_DoesNotFailTheList()
+    {
+        UseOptions(new ManagementServiceOptions { SecuredWriteTerminalPurgeInterval = TimeSpan.Zero });
+        _repo.PurgeTerminalAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .ThrowsAsync(new InvalidOperationException("db down"));
+
+        CreateActor().Tell(List());
+
+        ExpectMsg<ManagementSuccess>(TimeSpan.FromSeconds(5));
+    }
+}