From 74aa872c071ce647971404427dd9b180cea9a9b2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 06:22:19 -0400 Subject: [PATCH] feat(audit-log): daily site SQLite retention purge job (closes unbounded site DB growth) SiteAuditRetentionService (site-only IHostedService) ticks PurgeExpiredAsync on InitialDelay(5m)/PurgeInterval(24h, clamped >=1m), cutoff = UtcNow - RetentionDays. Registered site-only in AddAuditLogHealthMetricsBridge; options bound from AuditLog:SiteRetention. Per-tick failures swallowed. Added a test-only PurgeIntervalOverride (mirrors SiteCallAuditOptions) so the ms-cadence tick test can bypass the 1-min production clamp. Doc updated. (PLAN-04 Task 3, S1/U2) --- docs/requirements/Component-AuditLog.md | 16 ++- .../ServiceCollectionExtensions.cs | 11 ++ .../Site/SiteAuditRetentionOptions.cs | 25 +++- .../Site/SiteAuditRetentionService.cs | 133 ++++++++++++++++++ .../Site/SiteAuditRetentionServiceTests.cs | 114 +++++++++++++++ 5 files changed, 292 insertions(+), 7 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionService.cs create mode 100644 tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 650b24ea..859cbb55 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -409,9 +409,19 @@ MS SQL for direct-write events). Unredacted secrets never persist. partition switch-out loop. Values are validated to be in `[30, RetentionDays]`; keys that are not a recognized `AuditChannel` enum name are rejected at startup. -- **Sites:** daily site job; default 7-day retention (configurable, min 1, - max 90). Respects the hard `ForwardState` invariant — `Pending` rows are - never purged on age alone. +- **Sites:** `SiteAuditRetentionService` (site-only `IHostedService`) runs the + site SQLite retention purge on a timer — first tick after + `AuditLog:SiteRetention:InitialDelay` (default 5 min, short so a daily-recycled + node still purges soon after start), then every + `AuditLog:SiteRetention:PurgeInterval` (default 24 h, clamped to ≥ 1 min). + Each tick calls `ISiteAuditQueue.PurgeExpiredAsync(UtcNow − RetentionDays)`, + which deletes eligible rows in one transaction (sidecar first, then the + canonical row) and then runs `PRAGMA incremental_vacuum` to return the freed + pages to the OS. `AuditLog:SiteRetention:RetentionDays` defaults to 7 and is + clamped to `[1, 90]`. Respects the hard `ForwardState` invariant — a `Pending` + row is never purged on age alone; only `Forwarded`/`Reconciled` rows older than + the cutoff are removed. Per-tick failures are swallowed and logged so a transient + SQLite fault never tears the service down. ## Security & Tamper-Evidence diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs index 10534063..6a72a30f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/ServiceCollectionExtensions.cs @@ -101,6 +101,11 @@ public static class ServiceCollectionExtensions .Bind(config.GetSection(SiteWriterSectionName)); services.AddOptions() .Bind(config.GetSection(SiteTelemetrySectionName)); + // Retention purge options (the SiteAuditRetentionService hosted job is + // registered site-only in AddAuditLogHealthMetricsBridge). Binding here is + // inert on central — no hosted service consumes it there. + services.AddOptions() + .Bind(config.GetSection(SiteAuditRetentionOptions.SectionName)); // SqliteAuditWriter is a singleton with a single owned SqliteConnection // and a background writer Task; multiple instances would race on the @@ -316,6 +321,12 @@ public static class ServiceCollectionExtensions // free of hosted-service registrations that would resolve a missing // ISiteHealthCollector on central. services.AddHostedService(); + // Site-only retention purge: runs ISiteAuditQueue.PurgeExpiredAsync on a + // daily timer so the ephemeral site SQLite store does not grow unbounded. + // Registered here (not in AddAuditLog, which also runs on central) so the + // hosted service only starts on site composition roots. Guarded by the same + // SiteAuditBacklogReporter sentinel above against double-registration. + services.AddHostedService(); return services; } diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionOptions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionOptions.cs index 3aee06a9..aec5dfbc 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionOptions.cs @@ -10,8 +10,12 @@ namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; /// public class SiteAuditRetentionOptions { - /// Configuration section path bound in the site host. - public const string SectionName = "ScadaBridge:AuditLog:SiteRetention"; + /// + /// Configuration section path bound in the site host. Follows the component's + /// existing AuditLog:* convention (sibling of AuditLog:SiteWriter, + /// AuditLog:Purge, …), not an application-root-absolute path. + /// + public const string SectionName = "AuditLog:SiteRetention"; /// /// Retention window in days. On each purge tick, Forwarded/Reconciled rows whose @@ -42,14 +46,27 @@ public class SiteAuditRetentionOptions /// public TimeSpan PurgeInterval { get; set; } = TimeSpan.FromHours(24); + /// + /// Test-only override for the purge cadence — bypasses the + /// clamp so unit tests can drop the period to + /// milliseconds. Production config never sets this; leave null. Mirrors + /// SiteCallAuditOptions.PurgeIntervalOverride. + /// + public TimeSpan? PurgeIntervalOverride { get; set; } + private static readonly TimeSpan MinPurgeInterval = TimeSpan.FromMinutes(1); /// - /// Effective purge period: clamped to at least + /// Effective purge period: the test override when set (bypassing the clamp), + /// otherwise clamped to at least /// (the Timer/Akka zero-interval spin footgun). /// public TimeSpan ResolvedPurgeInterval => - PurgeInterval < MinPurgeInterval ? MinPurgeInterval : PurgeInterval; + PurgeIntervalOverride is { } o + ? o + : PurgeInterval < MinPurgeInterval + ? MinPurgeInterval + : PurgeInterval; /// /// Delay before the first purge tick. Default 5 minutes — short so a node that diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionService.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionService.cs new file mode 100644 index 00000000..0b76827f --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SiteAuditRetentionService.cs @@ -0,0 +1,133 @@ +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; + +/// +/// Site-side hosted service that runs the SQLite retention purge on a timer +/// (PLAN-04 Task 3, S1/U2). Each tick calls +/// with a cutoff of +/// UtcNow - , +/// deleting Forwarded/Reconciled rows past the retention window and reclaiming the +/// freed pages. Without it the ephemeral site store grows unbounded until the next +/// deployment recreates it. +/// +/// +/// +/// Cadence. First tick after +/// (default 5 min — short so a node that recycles daily still purges shortly after +/// start), then every +/// (default 24 h). Mirrors the loop shape of . +/// +/// +/// Failure containment. Each purge is wrapped in a try/catch so a transient +/// SQLite error never tears down the hosted service — the next tick retries. +/// +/// +public sealed class SiteAuditRetentionService : IHostedService, IDisposable +{ + private readonly ISiteAuditQueue _queue; + private readonly SiteAuditRetentionOptions _options; + private readonly ILogger _logger; + private CancellationTokenSource? _cts; + private Task? _loop; + + /// Initializes a new instance of . + /// The site audit queue whose retention purge is driven. + /// Retention cadence + window options. + /// Logger instance. + public SiteAuditRetentionService( + ISiteAuditQueue queue, + IOptions options, + ILogger logger) + { + _queue = queue ?? throw new ArgumentNullException(nameof(queue)); + _options = (options ?? throw new ArgumentNullException(nameof(options))).Value; + _logger = logger ?? throw new ArgumentNullException(nameof(logger)); + } + + /// Starts the background purge loop. + /// Cancellation token signalling host shutdown. + /// A task that represents the asynchronous operation. + public Task StartAsync(CancellationToken ct) + { + // Linked CTS so both StopAsync and the host shutdown token abort the loop. + _cts = CancellationTokenSource.CreateLinkedTokenSource(ct); + _loop = Task.Run(() => RunLoopAsync(_cts.Token)); + return Task.CompletedTask; + } + + private async Task RunLoopAsync(CancellationToken ct) + { + // InitialDelay before the first purge — short so a daily-recycled node + // still purges soon after start rather than waiting a full interval it may + // never reach. + try + { + await Task.Delay(_options.InitialDelay, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return; + } + + await SafePurgeAsync(ct).ConfigureAwait(false); + + while (!ct.IsCancellationRequested) + { + try + { + await Task.Delay(_options.ResolvedPurgeInterval, ct).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + break; + } + + await SafePurgeAsync(ct).ConfigureAwait(false); + } + } + + private async Task SafePurgeAsync(CancellationToken ct) + { + try + { + var cutoff = DateTime.UtcNow.AddDays(-_options.ResolvedRetentionDays); + var purged = await _queue.PurgeExpiredAsync(cutoff, ct).ConfigureAwait(false); + if (purged > 0) + { + _logger.LogInformation( + "Site audit retention purge removed {Count} row(s) older than {Cutoff:o}.", + purged, cutoff); + } + } + catch (OperationCanceledException) + { + // Shutdown — let the outer loop exit cleanly. + throw; + } + catch (Exception ex) + { + // Catch-all is deliberate: a transient SQLite fault must not tear down + // the hosted service; the next tick retries. + _logger.LogWarning(ex, "SiteAuditRetentionService purge failed; next tick will retry."); + } + } + + /// Signals the purge loop to stop and waits for it to complete. + /// Cancellation token (not used; the internal CTS governs shutdown). + /// A task that represents the asynchronous operation. + public Task StopAsync(CancellationToken ct) + { + _cts?.Cancel(); + return _loop ?? Task.CompletedTask; + } + + /// Releases the internal . + public void Dispose() + { + _cts?.Dispose(); + } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs new file mode 100644 index 00000000..77652d78 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SiteAuditRetentionServiceTests.cs @@ -0,0 +1,114 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.AuditLog.Site; +using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Services; +using ZB.MOM.WW.ScadaBridge.Commons.Types; +using AuditEvent = ZB.MOM.WW.Audit.AuditEvent; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site; + +/// +/// Tests for (PLAN-04 Task 3). The hosted +/// service runs the site SQLite retention purge on a timer: first tick after +/// , then every +/// , passing a cutoff of +/// UtcNow - ResolvedRetentionDays. Per-tick failures are swallowed so a +/// transient SQLite error never tears the service down. +/// +public class SiteAuditRetentionServiceTests +{ + [Fact] + public async Task Tick_Purges_With_RetentionCutoff() + { + var queue = new RecordingSiteAuditQueue(); + var options = Options.Create(new SiteAuditRetentionOptions + { + RetentionDays = 7, + // Override bypasses the 1-min production clamp so the loop ticks in ms. + PurgeIntervalOverride = TimeSpan.FromMilliseconds(50), + InitialDelay = TimeSpan.Zero, + }); + using var svc = new SiteAuditRetentionService(queue, options, NullLogger.Instance); + + await svc.StartAsync(CancellationToken.None); + await WaitUntilAsync(() => queue.PurgeCalls.Count >= 1, TimeSpan.FromSeconds(5)); + await svc.StopAsync(CancellationToken.None); + + var cutoff = queue.PurgeCalls.First(); + Assert.InRange(cutoff, + DateTime.UtcNow.AddDays(-7).AddMinutes(-1), + DateTime.UtcNow.AddDays(-7).AddMinutes(1)); + } + + [Fact] + public async Task Tick_SwallowsExceptions_AndKeepsTicking() + { + var queue = new RecordingSiteAuditQueue { ThrowOnFirstCall = true }; + var options = Options.Create(new SiteAuditRetentionOptions + { + RetentionDays = 7, + // Override bypasses the 1-min production clamp so the loop ticks in ms. + PurgeIntervalOverride = TimeSpan.FromMilliseconds(50), + InitialDelay = TimeSpan.Zero, + }); + using var svc = new SiteAuditRetentionService(queue, options, NullLogger.Instance); + + await svc.StartAsync(CancellationToken.None); + // The first tick throws; a second tick must still arrive (service survives). + await WaitUntilAsync(() => queue.PurgeCalls.Count >= 2, TimeSpan.FromSeconds(5)); + await svc.StopAsync(CancellationToken.None); + + Assert.True(queue.PurgeCalls.Count >= 2); + } + + private static async Task WaitUntilAsync(Func condition, TimeSpan timeout) + { + var sw = Stopwatch.StartNew(); + while (!condition()) + { + if (sw.Elapsed > timeout) + { + throw new TimeoutException("Condition was not met within the timeout."); + } + + await Task.Delay(10); + } + } + + /// Stub queue that records purge cutoffs and can throw on the first tick. + private sealed class RecordingSiteAuditQueue : ISiteAuditQueue + { + private readonly ConcurrentQueue _purgeCalls = new(); + private int _calls; + + public IReadOnlyCollection PurgeCalls => _purgeCalls; + public bool ThrowOnFirstCall { get; init; } + + public Task PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default) + { + var n = Interlocked.Increment(ref _calls); + _purgeCalls.Enqueue(olderThanUtc); + if (ThrowOnFirstCall && n == 1) + { + throw new InvalidOperationException("Simulated transient purge failure."); + } + + return Task.FromResult(0); + } + + public Task> ReadPendingAsync(int limit, CancellationToken ct = default) + => throw new NotSupportedException(); + public Task> ReadPendingCachedTelemetryAsync(int limit, CancellationToken ct = default) + => throw new NotSupportedException(); + public Task MarkForwardedAsync(IReadOnlyList eventIds, CancellationToken ct = default) + => throw new NotSupportedException(); + public Task> ReadPendingSinceAsync(DateTime sinceUtc, int batchSize, CancellationToken ct = default) + => throw new NotSupportedException(); + public Task MarkReconciledAsync(IReadOnlyList eventIds, CancellationToken ct = default) + => throw new NotSupportedException(); + public Task GetBacklogStatsAsync(CancellationToken ct = default) + => throw new NotSupportedException(); + } +}