fix(auditlog): page the site audit purge, bound incremental_vacuum per page, fail fast instead of parking writers on a full queue (review 2026-10-21 06/MEM-1)

2026-10-22 06:23:35 -04:00
parent 268b55e350
commit 1da5d0bac4
6 changed files with 261 additions and 56 deletions
@@ -574,6 +574,17 @@ MS SQL for direct-write events). Unredacted secrets never persist.
  (disk full, schema lock, transient IO), events buffer in a small in-memory
  ring (default 1024 rows); oldest are discarded with a Site Event Log warning
  per drop.
+- **A saturated writer fails fast; it never parks callers.** The site writer's
+  bounded queue (`AuditLog:SiteWriter:ChannelCapacity`, default 4096) used to
+  make `WriteAsync` *await* room once full — and because every hot-path caller is
+  fire-and-continue, a writer stuck behind a long purge accumulated an unbounded
+  number of parked waiters, each holding its event, while reporting healthy. A
+  full queue now faults the write immediately with
+  `SqliteAuditWriterSaturatedException`; `FallbackAuditWriter` treats it like
+  any other primary failure — the event goes to the drop-oldest ring above and
+  `SiteAuditWriteFailures` is bumped — so the degradation is the documented
+  best-effort drop with a counter, never unbounded memory (review 2026-10-21
+  06/MEM-1). Audit failure still never aborts the action.
 - **Reconciliation as fallback.** If two consecutive reconciliation cycles
  report a non-draining backlog, the supervisor restarts the telemetry actor
  and a `SiteAuditTelemetryStalled` event fires.
@@ -655,9 +666,16 @@ MS SQL for direct-write events). Unredacted secrets never persist.
  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
+  which deletes eligible rows **in pages** of at most
+  `AuditLog:SiteWriter:PurgeBatchSize` rows (default 5000) — one transaction per
+  page, sidecar first, then the canonical row — and after each page runs a
+  **bounded** `PRAGMA incremental_vacuum(N)` (`AuditLog:SiteWriter:PurgeVacuumPagesPerBatch`,
+  default 1000 pages) to return freed pages to the OS. The write lock is taken per
+  page and released between pages, so the batched hot-path writer interleaves with a
+  multi-million-row purge instead of stalling behind one giant DELETE plus an
+  unbounded vacuum (review 2026-10-21 06/MEM-1 — the same hazard register row 34
+  closed for `site_events`). One call still removes every eligible row; only the
+  lock grain changed. `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
@@ -134,10 +134,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
        _writeQueue = Channel.CreateBounded<PendingAuditEvent>(
            new BoundedChannelOptions(_options.ChannelCapacity)
            {
-                // The hot-path enqueue must back-pressure if the background
-                // writer falls behind; a higher-level fallback
-                // handles truly catastrophic primary failure with a drop-oldest
-                // ring buffer.
+                // Wait mode is only what the (never-awaited) async write path would
+                // do; WriteAsync uses TryWrite exclusively and fails fast with
+                // SqliteAuditWriterSaturatedException when the queue is full, so no
+                // caller ever parks on room (review 2026-10-21 06/MEM-1). The
+                // FallbackAuditWriter's drop-oldest ring absorbs the overflow.
                FullMode = BoundedChannelFullMode.Wait,
                SingleReader = true,
                SingleWriter = false,
@@ -286,7 +287,8 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable

    /// <summary>
    /// Enqueues an audit event for asynchronous batched persistence to SQLite.
-    /// Back-pressure is applied when the write channel is full.
+    /// Never parks the caller on queue room: a full queue faults the returned task with
+    /// <see cref="SqliteAuditWriterSaturatedException"/> immediately.
    /// </summary>
    /// <param name="evt">The audit event to persist.</param>
    /// <param name="ct">Cancellation token.</param>
@@ -300,35 +302,40 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
        // alongside the canonical row in the same transaction.
        var pending = new PendingAuditEvent(evt, AuditForwardState.Pending);

-        // CreateBounded(FullMode=Wait) means WriteAsync will await room rather
-        // than throw when full — exactly the hot-path back-pressure semantics
-        // we want.
-        if (!_writeQueue.Writer.TryWrite(pending))
+        if (_writeQueue.Writer.TryWrite(pending))
        {
-            // The writer is either completed (logger disposed) or the channel
-            // is at capacity. Fall back to the async path which honours the
-            // FullMode=Wait policy.
-            return WriteSlowPathAsync(pending, ct);
+            return pending.Completion.Task;
        }

-        return pending.Completion.Task;
+        // TryWrite fails for exactly two reasons: the writer is completed (disposed)
+        // or the queue is at capacity. The second used to fall into an awaited
+        // WriteAsync on the channel — one parked waiter, holding its AuditEvent, per
+        // fire-and-continue script-thread caller for as long as the background writer
+        // stayed stuck (e.g. behind a multi-minute retention purge). That count was
+        // unbounded and invisible. Fail fast instead: FallbackAuditWriter routes the
+        // event to its bounded drop-oldest ring and bumps SiteAuditWriteFailures, which
+        // is the documented best-effort degradation (review 2026-10-21 06/MEM-1).
+        if (Volatile.Read(ref _writerCompleted))
+        {
+            return Task.FromException(new ObjectDisposedException(nameof(SqliteAuditWriter),
+                "Event could not be recorded: the audit writer has been disposed."));
+        }
+
+        Interlocked.Increment(ref _saturatedRejections);
+        return Task.FromException(new SqliteAuditWriterSaturatedException(_options.ChannelCapacity));
    }

-    private async Task WriteSlowPathAsync(PendingAuditEvent pending, CancellationToken ct)
-    {
-        try
-        {
-            await _writeQueue.Writer.WriteAsync(pending, ct).ConfigureAwait(false);
-        }
-        catch (ChannelClosedException)
-        {
-            pending.Completion.TrySetException(
-                new ObjectDisposedException(nameof(SqliteAuditWriter),
-                    "Event could not be recorded: the audit writer has been disposed."));
-        }
+    /// <summary>
+    /// Number of <see cref="WriteAsync"/> calls refused because the write queue was full
+    /// since construction. Observability seam; the health-report count is the
+    /// <c>SiteAuditWriteFailures</c> bump <see cref="FallbackAuditWriter"/> makes per refusal.
+    /// </summary>
+    public long SaturatedRejectionCount => Interlocked.Read(ref _saturatedRejections);
+    private long _saturatedRejections;

-        await pending.Completion.Task.ConfigureAwait(false);
-    }
+    // Set (before the channel is completed) once DisposeAsync starts, so a refused
+    // TryWrite can tell "disposed" from "full" without racing the channel's state.
+    private bool _writerCompleted;

    private async Task ProcessWriteQueueAsync()
    {
@@ -397,6 +404,12 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
    /// commit per event; production code never reads this.
    /// </summary>
    internal int FlushCountForTests => _flushCountForTests;
+
+    /// <summary>
+    /// Test-only: the write lock FlushBatch / the flips / the purge serialise on, so a test can
+    /// park the background writer deterministically (e.g. to fill the write queue).
+    /// </summary>
+    internal object WriteLockForTests => _writeLock;
    private int _flushCountForTests;

    private void FlushBatch(IReadOnlyList<PendingAuditEvent> batch)
@@ -1181,7 +1194,52 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
    }

    /// <inheritdoc />
-    public Task<int> PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default)
+    /// <remarks>
+    /// <b>Paged (review 2026-10-21 06/MEM-1).</b> The purge deletes at most
+    /// <see cref="SqliteAuditWriterOptions.PurgeBatchSize"/> rows per transaction and runs a
+    /// bounded <c>PRAGMA incremental_vacuum(N)</c> after each page, taking <c>_writeLock</c>
+    /// per page and releasing it in between, so the batched hot-path writer interleaves with
+    /// a multi-million-row purge rather than stalling behind one giant DELETE + an unbounded
+    /// vacuum. Every eligible row is still removed by one call; only the lock grain changed.
+    /// </remarks>
+    public async Task<int> PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default)
+    {
+        // 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);
+        var pageSize = Math.Max(1, _options.PurgeBatchSize);
+
+        var total = 0;
+        while (true)
+        {
+            ct.ThrowIfCancellationRequested();
+            var purged = PurgePage(cutoff, pageSize);
+            total += purged;
+            if (purged < pageSize)
+            {
+                return total;
+            }
+
+            // Lock already released by PurgePage; yield so the background writer (and any
+            // flip/read waiting on _writeLock) gets the lock before the next page.
+            await Task.Yield();
+        }
+    }
+
+    /// <summary>
+    /// Number of purge-page transactions committed since construction. Test-only
+    /// observability seam for asserting the purge is paged; production code never reads it.
+    /// </summary>
+    internal int PurgePageCountForTests => Volatile.Read(ref _purgePageCountForTests);
+    private int _purgePageCountForTests;
+
+    /// <summary>
+    /// Deletes ONE page (at most <paramref name="pageSize"/> rows) of purge-eligible rows in a
+    /// single transaction under <c>_writeLock</c>, then reclaims a bounded number of freed
+    /// pages. Returns the number of canonical rows deleted.
+    /// </summary>
+    private int PurgePage(string cutoff, int pageSize)
    {
        // Runs on the WRITE connection under _writeLock (a DELETE mutates the file),
        // mirroring MarkForwardedAsync / MarkReconciledAsync. Serialising with the
@@ -1190,24 +1248,18 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
        {
            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.
+                    // Capture this page's 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 the page off
+                    // the SQLite 999-parameter limit. The hard invariant is encoded here:
+                    // only Forwarded/Reconciled rows qualify — Pending (and the pull-owed
+                    // PushAbandoned/Rejected states) are never purged on age alone.
                    using (var createCmd = _connection.CreateCommand())
                    {
                        createCmd.Transaction = transaction;
@@ -1216,11 +1268,13 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                            CREATE TEMP TABLE purge_ids AS
                                SELECT EventId FROM audit_forward_state
                                WHERE ForwardState IN ($forwarded, $reconciled)
-                                  AND OccurredAtUtc < $cutoff;
+                                  AND OccurredAtUtc < $cutoff
+                                LIMIT $limit;
                            """;
                        createCmd.Parameters.AddWithValue("$forwarded", AuditForwardState.Forwarded.ToString());
                        createCmd.Parameters.AddWithValue("$reconciled", AuditForwardState.Reconciled.ToString());
                        createCmd.Parameters.AddWithValue("$cutoff", cutoff);
+                        createCmd.Parameters.AddWithValue("$limit", pageSize);
                        createCmd.ExecuteNonQuery();
                    }

@@ -1276,6 +1330,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                    }

                    transaction.Commit();
+                    Interlocked.Increment(ref _purgePageCountForTests);
                }
                catch
                {
@@ -1284,16 +1339,19 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                }
            }

-            // 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.
+            // Reclaim a BOUNDED number of freed pages back to the OS. A bare
+            // `PRAGMA incremental_vacuum;` empties the whole freelist in one step, which
+            // after a large purge held _writeLock for as long as the delete itself did.
+            // auto_vacuum=INCREMENTAL (set in InitializeSchema) makes this a cheap
+            // freelist trim. 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.CommandText = string.Create(System.Globalization.CultureInfo.InvariantCulture,
+                        $"PRAGMA incremental_vacuum({Math.Max(1, _options.PurgeVacuumPagesPerBatch)});");
                    vacuumCmd.ExecuteNonQuery();
                }
                catch (Exception ex)
@@ -1303,7 +1361,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
                }
            }

-            return Task.FromResult(purged);
+            return purged;
        }
    }

@@ -1387,6 +1445,7 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable
        lock (_writeLock)
        {
            if (_disposed) return;
+            Volatile.Write(ref _writerCompleted, true);
            // Stop accepting new events. Completing the channel writer is the
            // shutdown signal: WriteAsync calls observe the completion and
            // fault, and the writer loop drains any already-buffered items
@@ -29,8 +29,11 @@ public sealed class SqliteAuditWriterOptions

    /// <summary>
    /// Capacity of the bounded write queue. Set high enough that ordinary
-    /// script bursts never fill it; <see cref="System.Threading.Channels.BoundedChannelFullMode.Wait"/>
-    /// applies when the writer falls behind.
+    /// script bursts never fill it. When the writer falls behind and the queue is
+    /// full, <c>WriteAsync</c> fails fast with <see cref="SqliteAuditWriterSaturatedException"/>
+    /// instead of parking an unbounded number of waiters (review 2026-10-21 06/MEM-1);
+    /// <see cref="FallbackAuditWriter"/> turns that into its documented best-effort
+    /// drop-oldest ring + <c>SiteAuditWriteFailures</c> counter bump.
    /// </summary>
    public int ChannelCapacity { get; set; } = 4096;

@@ -40,6 +43,23 @@ public sealed class SqliteAuditWriterOptions
    /// <summary>Soft flush interval the writer enforces when fewer than BatchSize events are queued.</summary>
    public int FlushIntervalMs { get; set; } = 50;

+    /// <summary>
+    /// Maximum rows the retention purge deletes in ONE transaction. The purge loops page by
+    /// page and releases the write lock between pages, so the batched hot-path writer
+    /// interleaves with a large purge instead of stalling behind one multi-minute DELETE
+    /// (review 2026-10-21 06/MEM-1 — the same hazard class register row 34 fixed for
+    /// <c>site_events</c>).
+    /// </summary>
+    public int PurgeBatchSize { get; set; } = 5000;
+
+    /// <summary>
+    /// Free pages reclaimed by the bounded <c>PRAGMA incremental_vacuum(N)</c> run after each
+    /// purge page. Bounds the time the vacuum holds the write lock; pages not reclaimed in
+    /// one step stay on the freelist and are reused by later inserts or reclaimed by the
+    /// next purge.
+    /// </summary>
+    public int PurgeVacuumPagesPerBatch { get; set; } = 1000;
+
    /// <summary>
    /// Poll cadence, in seconds, for the <c>SiteAuditBacklogReporter</c> hosted service
    /// that probes the SQLite audit backlog and pushes the snapshot onto the site health
@@ -37,6 +37,14 @@ public sealed class SqliteAuditWriterOptionsValidator : OptionsValidatorBase<Sql
            $"AuditLog:SiteWriter:{nameof(SqliteAuditWriterOptions.FlushIntervalMs)} " +
            $"({options.FlushIntervalMs}) must be > 0.");

+        builder.RequireThat(options.PurgeBatchSize > 0,
+            $"AuditLog:SiteWriter:{nameof(SqliteAuditWriterOptions.PurgeBatchSize)} " +
+            $"({options.PurgeBatchSize}) must be > 0.");
+
+        builder.RequireThat(options.PurgeVacuumPagesPerBatch > 0,
+            $"AuditLog:SiteWriter:{nameof(SqliteAuditWriterOptions.PurgeVacuumPagesPerBatch)} " +
+            $"({options.PurgeVacuumPagesPerBatch}) must be > 0.");
+
        // BacklogPollIntervalSeconds is deliberately unchecked: a non-positive
        // value falls back to the reporter's 30s default (see the option's
        // remarks + register row 21 resolution).
@@ -0,0 +1,24 @@
+namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site;
+
+/// <summary>
+/// Thrown (as a faulted task) by <see cref="SqliteAuditWriter.WriteAsync"/> when the bounded
+/// write queue is full because the background writer has fallen behind — e.g. it is parked
+/// behind a long retention-purge page. Failing fast instead of awaiting room keeps the number
+/// of parked callers (each holding its <c>AuditEvent</c>) bounded; <see cref="FallbackAuditWriter"/>
+/// catches it like any other primary failure, stashes the event in its drop-oldest ring and
+/// bumps the <c>SiteAuditWriteFailures</c> health counter (review 2026-10-21 06/MEM-1).
+/// Audit writes are best-effort: this never reaches the user-facing action.
+/// </summary>
+public sealed class SqliteAuditWriterSaturatedException : InvalidOperationException
+{
+    /// <summary>Initializes a new instance with the queue capacity that was exhausted.</summary>
+    /// <param name="capacity">The configured <see cref="SqliteAuditWriterOptions.ChannelCapacity"/>.</param>
+    public SqliteAuditWriterSaturatedException(int capacity)
+        : base($"Audit write queue is full ({capacity} pending events); the background writer has fallen behind.")
+    {
+        Capacity = capacity;
+    }
+
+    /// <summary>The write-queue capacity that was exhausted.</summary>
+    public int Capacity { get; }
+}
@@ -38,9 +38,15 @@ public class SqliteAuditWriterRetentionTests : IDisposable
        }
    }

-    private SqliteAuditWriter CreateWriter()
+    private SqliteAuditWriter CreateWriter(int purgeBatchSize = 5000, int channelCapacity = 4096, int batchSize = 256)
    {
-        var options = new SqliteAuditWriterOptions { DatabasePath = _dbPath };
+        var options = new SqliteAuditWriterOptions
+        {
+            DatabasePath = _dbPath,
+            PurgeBatchSize = purgeBatchSize,
+            ChannelCapacity = channelCapacity,
+            BatchSize = batchSize,
+        };
        return new SqliteAuditWriter(
            Options.Create(options),
            NullLogger<SqliteAuditWriter>.Instance,
@@ -95,4 +101,74 @@ public class SqliteAuditWriterRetentionTests : IDisposable
        Assert.Equal(1, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7)));
        Assert.Equal(0, await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7)));
    }
+
+    /// <summary>
+    /// Review 2026-10-21 06/MEM-1: the purge deletes in bounded pages (one transaction
+    /// each, write lock released between them) and still removes every eligible row.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task PurgeExpired_IsPaged_AndStillRemovesEveryEligibleRow()
+    {
+        await using var writer = CreateWriter(purgeBatchSize: 10);
+        var ids = new List<Guid>();
+        for (var i = 0; i < 35; i++)
+        {
+            ids.Add(await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10)));
+        }
+        var pendingId = await WriteEventAsync(writer, DateTime.UtcNow.AddDays(-10));
+        await writer.MarkForwardedAsync(ids);
+
+        var purged = await writer.PurgeExpiredAsync(DateTime.UtcNow.AddDays(-7));
+
+        Assert.Equal(35, purged);
+        Assert.Equal(4, writer.PurgePageCountForTests); // 10 + 10 + 10 + 5
+        var pending = await writer.ReadPendingAsync(10);
+        Assert.Equal(pendingId, Assert.Single(pending).EventId);
+    }
+
+    /// <summary>
+    /// Review 2026-10-21 06/MEM-1: with the background writer parked (as behind a long purge
+    /// page) and the bounded queue full, <c>WriteAsync</c> must fail fast with
+    /// <see cref="SqliteAuditWriterSaturatedException"/> instead of parking a waiter that
+    /// holds its event for as long as the writer stays stuck.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task WriteAsync_QueueFull_FailsFast_InsteadOfParking()
+    {
+        await using var writer = CreateWriter(channelCapacity: 1, batchSize: 1);
+        Task first, third;
+        System.Threading.Monitor.Enter(writer.WriteLockForTests);
+        try
+        {
+            first = writer.WriteAsync(NewEvent());
+            // Wait until the writer loop has dequeued `first` and is blocked on the lock.
+            Assert.True(SpinWait.SpinUntil(() =>
+            {
+                var probe = writer.WriteAsync(NewEvent());
+                if (probe.IsFaulted) { probe.Exception!.Handle(_ => true); return false; }
+                return true;
+            }, TimeSpan.FromSeconds(5)));
+            // The queue (capacity 1) now holds the probe; the next write cannot fit.
+            third = writer.WriteAsync(NewEvent());
+            Assert.True(third.IsFaulted, "a full queue must fault immediately, not park");
+            Assert.IsType<SqliteAuditWriterSaturatedException>(third.Exception!.InnerException);
+            Assert.True(writer.SaturatedRejectionCount >= 1);
+        }
+        finally
+        {
+            System.Threading.Monitor.Exit(writer.WriteLockForTests);
+        }
+
+        // Lock released: the parked writer resumes and the accepted event lands.
+        await first.WaitAsync(TimeSpan.FromSeconds(5));
+    }
+
+    private static AuditEvent NewEvent() => ScadaBridgeAuditEventFactory.Create(
+        eventId: Guid.NewGuid(),
+        occurredAtUtc: DateTime.UtcNow,
+        channel: AuditChannel.ApiOutbound,
+        kind: AuditKind.ApiCall,
+        status: AuditStatus.Delivered);
 }