From d495494074e84b4a9772c589bee43b21f2b73798 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 09:25:48 -0400 Subject: [PATCH] fix(management): purge terminal PendingSecuredWrites rows on the list sweep after a retention window (review 2026-10-21 08/G3) --- .../Component-ManagementService.md | 4 + .../Repositories/ISecuredWriteRepository.cs | 14 +++ .../Repositories/SecuredWriteRepository.cs | 21 +++++ .../ManagementActor.cs | 47 ++++++++++ .../ManagementServiceOptions.cs | 6 ++ .../ManagementServiceOptionsValidator.cs | 7 +- .../SecuredWriteRepositoryTests.cs | 47 ++++++++++ .../SecuredWriteTerminalPurgeTests.cs | 92 +++++++++++++++++++ 8 files changed, 237 insertions(+), 1 deletion(-) create mode 100644 tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/SecuredWriteTerminalPurgeTests.cs diff --git a/docs/requirements/Component-ManagementService.md b/docs/requirements/Component-ManagementService.md index 758c93a3..83ded24c 100644 --- a/docs/requirements/Component-ManagementService.md +++ b/docs/requirements/Component-ManagementService.md @@ -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 diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISecuredWriteRepository.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISecuredWriteRepository.cs index dbc51ff7..7d611ad1 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISecuredWriteRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Repositories/ISecuredWriteRepository.cs @@ -153,4 +153,18 @@ public interface ISecuredWriteRepository string? siteId, IReadOnlyCollection? permittedSiteIds = null, CancellationToken ct = default); + + /// + /// Deletes up to rows in a post-Pending status + /// (Approved, Rejected, Expired, Executed, Failed) whose + /// last lifecycle instant — ExecutedAtUtc, else DecidedAtUtc, else + /// SubmittedAtUtc — is older than . Pending + /// rows are never purged (they are expired first). The table is operational state; the + /// append-only AuditLog keeps the lifecycle history (review 2026-10-21 08/G3). + /// + /// Rows whose last lifecycle instant is before this are purged. + /// Maximum rows deleted in this call. + /// Cancellation token. + /// A task that resolves to the number of rows deleted. + Task PurgeTerminalAsync(DateTime olderThanUtc, int batchSize, CancellationToken ct = default); } diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SecuredWriteRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SecuredWriteRepository.cs index 7897a2f1..d258745c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SecuredWriteRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/SecuredWriteRepository.cs @@ -188,4 +188,25 @@ WHERE Id = {id} return await query.CountAsync(ct); } + + /// + public async Task 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() + .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"]; } diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs index afe1fb59..57cc52f7 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementActor.cs @@ -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; + + /// + /// Purges decided/executed secured-write rows older than + /// (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. + /// + private static async Task TryPurgeTerminalSecuredWritesAsync(IServiceProvider sp, ISecuredWriteRepository repo) + { + var options = sp.GetService>()?.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>()?.LogInformation( + "Purged {Count} terminal secured-write row(s) older than {Retention}.", deleted, retention); + } + catch (Exception ex) + { + sp.GetService>()?.LogWarning( + ex, "Best-effort terminal secured-write purge failed; the list is unaffected."); + } + } + // ======================================================================== // Site handlers // ======================================================================== diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptions.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptions.cs index 2e3bc44a..4c329de4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptions.cs @@ -10,4 +10,10 @@ public class ManagementServiceOptions /// 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. public TimeSpan SecuredWritePendingTtl { get; set; } = TimeSpan.FromHours(24); + + /// Age after which a decided/executed secured-write row (Approved/Rejected/Expired/Executed/Failed, measured from its last lifecycle instant) is purged from PendingSecuredWrites; the AuditLog keeps the history. Default 90 days. A non-positive value disables the purge. + public TimeSpan SecuredWriteTerminalRetention { get; set; } = TimeSpan.FromDays(90); + + /// 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. + public TimeSpan SecuredWriteTerminalPurgeInterval { get; set; } = TimeSpan.FromHours(1); } diff --git a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptionsValidator.cs index 7aefbf13..fdf11ac5 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ManagementService/ManagementServiceOptionsValidator.cs @@ -11,7 +11,8 @@ namespace ZB.MOM.WW.ScadaBridge.ManagementService; /// section fails fast at boot with a clear, key-naming message. /// /// 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 +/// (non-positive disables the terminal-row purge). /// /// public sealed class ManagementServiceOptionsValidator : OptionsValidatorBase @@ -26,5 +27,9 @@ public sealed class ManagementServiceOptionsValidator : OptionsValidatorBase 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."); } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SecuredWriteRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SecuredWriteRepositoryTests.cs index 9ab62fd7..9c504eac 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SecuredWriteRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/SecuredWriteRepositoryTests.cs @@ -358,6 +358,53 @@ public class SecuredWriteRepositoryTests : IClassFixture Assert.Equal(4, rows.Count); } + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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() diff --git a/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/SecuredWriteTerminalPurgeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/SecuredWriteTerminalPurgeTests.cs new file mode 100644 index 00000000..93c82038 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.ManagementService.Tests/SecuredWriteTerminalPurgeTests.cs @@ -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; + +/// +/// Review 2026-10-21 08/G3: terminal PendingSecuredWrites rows were never purged. +/// The Secured Writes list sweep (which already expires overdue Pending rows) now also +/// purges decided/executed rows older than SecuredWriteTerminalRetention. +/// +public class SecuredWriteTerminalPurgeTests : TestKit +{ + private readonly ISecuredWriteRepository _repo = Substitute.For(); + private readonly ServiceCollection _services = new ServiceCollection().AddTestConfigurationTransaction(); + + /// Initializes the harness with an empty secured-write table. + public SecuredWriteTerminalPurgeTests() + { + _services.AddScoped(_ => _repo); + _repo.QueryAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any(), + Arg.Any?>(), Arg.Any()) + .Returns(Array.Empty()); + } + + private void UseOptions(ManagementServiceOptions options) => + _services.AddSingleton>(Options.Create(options)); + + private IActorRef CreateActor() => + Sys.ActorOf(Props.Create(() => new ManagementActor( + _services.BuildServiceProvider(), NullLogger.Instance))); + + private static ManagementEnvelope List() => + new(new AuthenticatedUser("admin", "Admin", new[] { "Administrator" }, Array.Empty()), + new ListSecuredWritesCommand(null, null), Guid.NewGuid().ToString("N")); + + /// The list sweep purges terminal rows older than the retention window. + [Fact] + public void List_PurgesTerminalRowsOlderThanRetention() + { + UseOptions(new ManagementServiceOptions + { + SecuredWriteTerminalRetention = TimeSpan.FromDays(30), + SecuredWriteTerminalPurgeInterval = TimeSpan.Zero, + }); + DateTime? cutoff = null; + _repo.PurgeTerminalAsync(Arg.Do(c => cutoff = c), Arg.Any(), Arg.Any()) + .Returns(3); + + CreateActor().Tell(List()); + + ExpectMsg(TimeSpan.FromSeconds(5)); + Assert.NotNull(cutoff); + Assert.InRange(cutoff!.Value, DateTime.UtcNow.AddDays(-30).AddMinutes(-1), DateTime.UtcNow.AddDays(-30).AddMinutes(1)); + } + + /// A non-positive retention disables the purge. + [Fact] + public void List_WithRetentionDisabled_DoesNotPurge() + { + UseOptions(new ManagementServiceOptions + { + SecuredWriteTerminalRetention = TimeSpan.Zero, + SecuredWriteTerminalPurgeInterval = TimeSpan.Zero, + }); + + CreateActor().Tell(List()); + + ExpectMsg(TimeSpan.FromSeconds(5)); + _repo.DidNotReceive().PurgeTerminalAsync(Arg.Any(), Arg.Any(), Arg.Any()); + } + + /// A purge failure is best-effort — the list still answers. + [Fact] + public void List_PurgeFailure_DoesNotFailTheList() + { + UseOptions(new ManagementServiceOptions { SecuredWriteTerminalPurgeInterval = TimeSpan.Zero }); + _repo.PurgeTerminalAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("db down")); + + CreateActor().Tell(List()); + + ExpectMsg(TimeSpan.FromSeconds(5)); + } +}