diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index 659b51fa..ffad9ae9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -141,9 +141,9 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable private void InitializeSchema() { // auto_vacuum must be set before any table is created for it to take - // effect on a fresh database. INCREMENTAL lets a future - // `PRAGMA incremental_vacuum` shrink the file after the 7-day retention - // purge — see alog.md §10. + // effect on a fresh database. INCREMENTAL lets the `PRAGMA + // incremental_vacuum` run at the end of PurgeExpiredAsync shrink the file + // after the ~7-day retention purge — see alog.md §10. using (var pragmaCmd = _connection.CreateCommand()) { pragmaCmd.CommandText = "PRAGMA auto_vacuum = INCREMENTAL"; @@ -821,6 +821,117 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable OnDiskBytes: onDiskBytes)); } + /// + public Task PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default) + { + // Runs on the WRITE connection under _writeLock (a DELETE mutates the file), + // mirroring MarkForwardedAsync / MarkReconciledAsync. Serialising with the + // batched writer means the purge never races an in-flight FlushBatch. + lock (_writeLock) + { + ObjectDisposedException.ThrowIf(_disposed, this); + + // Cutoff uses the same round-trip "o" encoding the writer stores + // OccurredAtUtc in, so string comparison stays monotonic (see FlushBatch). + var cutoff = EnsureUtc(olderThanUtc).ToString( + "o", System.Globalization.CultureInfo.InvariantCulture); + + int purged; + using (var transaction = _connection.BeginTransaction()) + { + try + { + // Capture the eligible EventIds from the sidecar FIRST: the + // Forwarded/Reconciled + age criteria live on + // audit_forward_state, and it is the FK child of audit_event, so + // it must be deleted before the parent. A temp table (rather than + // an IN (...) list) keeps an arbitrarily large purge set off the + // SQLite 999-parameter limit. The hard invariant is encoded here: + // only Forwarded/Reconciled rows qualify — Pending is never purged + // on age alone. + using (var createCmd = _connection.CreateCommand()) + { + createCmd.Transaction = transaction; + createCmd.CommandText = """ + DROP TABLE IF EXISTS purge_ids; + CREATE TEMP TABLE purge_ids AS + SELECT EventId FROM audit_forward_state + WHERE ForwardState IN ($forwarded, $reconciled) + AND OccurredAtUtc < $cutoff; + """; + createCmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString()); + createCmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString()); + createCmd.Parameters.AddWithValue("$cutoff", cutoff); + createCmd.ExecuteNonQuery(); + } + + // Delete the sidecar (FK child) first, then the canonical row. + using (var delSidecar = _connection.CreateCommand()) + { + delSidecar.Transaction = transaction; + delSidecar.CommandText = + "DELETE FROM audit_forward_state WHERE EventId IN (SELECT EventId FROM purge_ids);"; + delSidecar.ExecuteNonQuery(); + } + + using (var delEvent = _connection.CreateCommand()) + { + delEvent.Transaction = transaction; + delEvent.CommandText = + "DELETE FROM audit_event WHERE EventId IN (SELECT EventId FROM purge_ids);"; + delEvent.ExecuteNonQuery(); + } + + // changes() reports rows affected by the most recent DELETE on + // this connection (the audit_event delete above); read it before + // the DROP so no later statement can perturb the count. + using (var changesCmd = _connection.CreateCommand()) + { + changesCmd.Transaction = transaction; + changesCmd.CommandText = "SELECT changes();"; + purged = Convert.ToInt32(changesCmd.ExecuteScalar(), + System.Globalization.CultureInfo.InvariantCulture); + } + + using (var dropCmd = _connection.CreateCommand()) + { + dropCmd.Transaction = transaction; + dropCmd.CommandText = "DROP TABLE IF EXISTS purge_ids;"; + dropCmd.ExecuteNonQuery(); + } + + transaction.Commit(); + } + catch + { + transaction.Rollback(); + throw; + } + } + + // Reclaim the freed pages back to the OS. auto_vacuum=INCREMENTAL (set in + // InitializeSchema) makes this a bounded incremental step rather than a + // full VACUUM rewrite. Runs outside the transaction; a vacuum failure must + // not fault the purge itself, so it is best-effort. + if (purged > 0) + { + try + { + using var vacuumCmd = _connection.CreateCommand(); + vacuumCmd.CommandText = "PRAGMA incremental_vacuum;"; + vacuumCmd.ExecuteNonQuery(); + } + catch (Exception ex) + { + _logger.LogDebug(ex, + "SqliteAuditWriter incremental_vacuum after retention purge failed (non-fatal)."); + } + } + + return Task.FromResult(purged); + } + } + private static DateTime EnsureUtc(DateTime value) => value.Kind == DateTimeKind.Utc ? value diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs index aa5baa61..17058c91 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Interfaces/Services/ISiteAuditQueue.cs @@ -142,4 +142,25 @@ public interface ISiteAuditQueue /// Cancellation token. /// A task that resolves to a point-in-time snapshot of the site audit queue's pending count, oldest timestamp, and on-disk file size. Task GetBacklogStatsAsync(CancellationToken ct = default); + + /// + /// Retention purge: permanently deletes rows that have already left the site + /// ( or + /// ) + /// whose is strictly older than + /// , and reclaims the freed pages. + /// + /// + /// Hard invariant: + /// a row still in + /// is NEVER purged on age alone — it has not yet reached central, so dropping it + /// would silently lose audit data. Only Forwarded/Reconciled rows are eligible. + /// Idempotent: a second call with the same cutoff removes nothing further and + /// returns 0. The site store is ephemeral (~7-day retention), so this bounds its + /// unbounded growth (arch-review 04, S1/U2). + /// + /// Exclusive upper bound on ; rows at or after this instant are retained. + /// Cancellation token. + /// A task that resolves to the number of canonical rows deleted. + Task PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default); } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs new file mode 100644 index 00000000..9df9f090 --- /dev/null +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs @@ -0,0 +1,92 @@ +using Microsoft.Extensions.Logging.Abstractions; +using Microsoft.Extensions.Options; +using ZB.MOM.WW.ScadaBridge.AuditLog.Site; +using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; + +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Tests.Site; + +/// +/// Tests for (arch-review 04, S1/U2). +/// The site SQLite store is ephemeral (~7-day retention); the retention purge deletes +/// rows that have already left the site (Forwarded/Reconciled) and are older than the +/// cutoff, while honoring the hard invariant: a +/// row is NEVER purged on age alone — dropping +/// an un-forwarded row would silently lose audit data that never reached central. +/// Fixture mirrors (file-backed so the +/// post-purge incremental_vacuum has a real file to shrink). +/// +public class SqliteAuditWriterRetentionTests : IDisposable +{ + private readonly string _dbPath; + + public SqliteAuditWriterRetentionTests() + { + _dbPath = Path.Combine(Path.GetTempPath(), + $"audit-retention-{Guid.NewGuid():N}.db"); + } + + public void Dispose() + { + if (File.Exists(_dbPath)) + { + try { File.Delete(_dbPath); } catch { /* test cleanup best-effort */ } + } + } + + private SqliteAuditWriter CreateWriter() + { + var options = new SqliteAuditWriterOptions { DatabasePath = _dbPath }; + return new SqliteAuditWriter( + Options.Create(options), + NullLogger.Instance, + new FakeNodeIdentityProvider()); + } + + private static async Task WriteEventAsync(SqliteAuditWriter writer, DateTime occurredAtUtc) + { + var id = Guid.NewGuid(); + var evt = ScadaBridgeAuditEventFactory.Create( + eventId: id, + occurredAtUtc: occurredAtUtc, + channel: AuditChannel.ApiOutbound, + kind: AuditKind.ApiCall, + status: AuditStatus.Delivered); + await writer.WriteAsync(evt); + return id; + } + + [Fact] + public async Task PurgeExpired_DeletesForwardedAndReconciled_OlderThanCutoff() + { + await using var writer = CreateWriter(); + var oldForwarded = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)); + var oldReconciled = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)); + var oldPending = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)); + var fresh = await WriteEventAsync(writer, DateTime.UtcNow); + await writer.MarkForwardedAsync(new[] { oldForwarded, oldReconciled, fresh }); + await writer.MarkReconciledAsync(new[] { oldReconciled }); + + var purged = await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7)); + + Assert.Equal(2, purged); // oldForwarded + oldReconciled + // Hard ForwardState invariant: the old *Pending* row survives on age alone. + var pending = await writer.ReadPendingAsync(10); + Assert.Contains(pending, e => e.EventId == oldPending); + // Fresh forwarded row inside the window survives; only oldPending remains Pending. + var stats = await writer.GetBacklogStatsAsync(); + Assert.Equal(1, stats.PendingCount); + } + + [Fact] + public async Task PurgeExpired_IsIdempotent() + { + await using var writer = CreateWriter(); + var id = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)); + await writer.MarkForwardedAsync(new[] { id }); + Assert.Equal(1, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7))); + Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7))); + } +}