feat(audit-log): add site SQLite PurgeExpiredAsync honoring the ForwardState invariant
Retention purge deletes Forwarded/Reconciled rows older than the cutoff and reclaims pages via incremental_vacuum; Pending rows are never purged on age alone (would lose audit data that never reached central). Temp-table purge set avoids the SQLite 999-param limit; runs under _writeLock. (PLAN-04 Task 1, S1/U2)
This commit is contained in:
Joseph Doherty
@@ -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));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<int> 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
|
||||
|
||||
@@ -142,4 +142,25 @@ public interface ISiteAuditQueue
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>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.</returns>
|
||||
Task<SiteAuditBacklogSnapshot> GetBacklogStatsAsync(CancellationToken ct = default);
|
||||
|
||||
/// <summary>
|
||||
/// Retention purge: permanently deletes rows that have already left the site
|
||||
/// (<see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState.Forwarded"/> or
|
||||
/// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState.Reconciled"/>)
|
||||
/// whose <see cref="AuditEvent.OccurredAtUtc"/> is strictly older than
|
||||
/// <paramref name="olderThanUtc"/>, and reclaims the freed pages.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <b>Hard <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState"/> invariant:</b>
|
||||
/// a row still in <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Enums.AuditForwardState.Pending"/>
|
||||
/// 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).
|
||||
/// </remarks>
|
||||
/// <param name="olderThanUtc">Exclusive upper bound on <see cref="AuditEvent.OccurredAtUtc"/>; rows at or after this instant are retained.</param>
|
||||
/// <param name="ct">Cancellation token.</param>
|
||||
/// <returns>A task that resolves to the number of canonical rows deleted.</returns>
|
||||
Task<int> PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default);
|
||||
}
|
||||
|
||||
@@ -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;
|
||||
|
||||
/// <summary>
|
||||
/// Tests for <see cref="SqliteAuditWriter.PurgeExpiredAsync"/> (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 <see cref="AuditForwardState"/> invariant: a
|
||||
/// <see cref="AuditForwardState.Pending"/> row is NEVER purged on age alone — dropping
|
||||
/// an un-forwarded row would silently lose audit data that never reached central.
|
||||
/// Fixture mirrors <see cref="SqliteAuditWriterBacklogStatsTests"/> (file-backed so the
|
||||
/// post-purge incremental_vacuum has a real file to shrink).
|
||||
/// </summary>
|
||||
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<SqliteAuditWriter>.Instance,
|
||||
new FakeNodeIdentityProvider());
|
||||
}
|
||||
|
||||
private static async Task<Guid> 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)));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user