From 1da5d0bac43229b756a2e470356e06ad2caba278 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 06:23:35 -0400 Subject: [PATCH] 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) --- docs/requirements/Component-AuditLog.md | 24 ++- .../Site/SqliteAuditWriter.cs | 157 ++++++++++++------ .../Site/SqliteAuditWriterOptions.cs | 24 ++- .../Site/SqliteAuditWriterOptionsValidator.cs | 8 + .../SqliteAuditWriterSaturatedException.cs | 24 +++ .../Site/SqliteAuditWriterRetentionTests.cs | 80 ++++++++- 6 files changed, 261 insertions(+), 56 deletions(-) create mode 100644 src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterSaturatedException.cs diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index 79eeeb90..e12e0d89 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -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 diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs index 36b87431..e1ad577f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs @@ -134,10 +134,11 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable _writeQueue = Channel.CreateBounded( 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 /// /// 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 + /// immediately. /// /// The audit event to persist. /// Cancellation token. @@ -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.")); - } + /// + /// Number of calls refused because the write queue was full + /// since construction. Observability seam; the health-report count is the + /// SiteAuditWriteFailures bump makes per refusal. + /// + 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. /// internal int FlushCountForTests => _flushCountForTests; + + /// + /// 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). + /// + internal object WriteLockForTests => _writeLock; private int _flushCountForTests; private void FlushBatch(IReadOnlyList batch) @@ -1181,7 +1194,52 @@ public class SqliteAuditWriter : IAuditWriter, ISiteAuditQueue, IAsyncDisposable } /// - public Task PurgeExpiredAsync(DateTime olderThanUtc, CancellationToken ct = default) + /// + /// Paged (review 2026-10-21 06/MEM-1). The purge deletes at most + /// rows per transaction and runs a + /// bounded PRAGMA incremental_vacuum(N) after each page, taking _writeLock + /// 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. + /// + public async Task 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(); + } + } + + /// + /// Number of purge-page transactions committed since construction. Test-only + /// observability seam for asserting the purge is paged; production code never reads it. + /// + internal int PurgePageCountForTests => Volatile.Read(ref _purgePageCountForTests); + private int _purgePageCountForTests; + + /// + /// Deletes ONE page (at most rows) of purge-eligible rows in a + /// single transaction under _writeLock, then reclaims a bounded number of freed + /// pages. Returns the number of canonical rows deleted. + /// + 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 diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptions.cs index ab8c1f8f..a205cf7a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptions.cs @@ -29,8 +29,11 @@ public sealed class SqliteAuditWriterOptions /// /// Capacity of the bounded write queue. Set high enough that ordinary - /// script bursts never fill it; - /// applies when the writer falls behind. + /// script bursts never fill it. When the writer falls behind and the queue is + /// full, WriteAsync fails fast with + /// instead of parking an unbounded number of waiters (review 2026-10-21 06/MEM-1); + /// turns that into its documented best-effort + /// drop-oldest ring + SiteAuditWriteFailures counter bump. /// public int ChannelCapacity { get; set; } = 4096; @@ -40,6 +43,23 @@ public sealed class SqliteAuditWriterOptions /// Soft flush interval the writer enforces when fewer than BatchSize events are queued. public int FlushIntervalMs { get; set; } = 50; + /// + /// 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 + /// site_events). + /// + public int PurgeBatchSize { get; set; } = 5000; + + /// + /// Free pages reclaimed by the bounded PRAGMA incremental_vacuum(N) 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. + /// + public int PurgeVacuumPagesPerBatch { get; set; } = 1000; + /// /// Poll cadence, in seconds, for the SiteAuditBacklogReporter hosted service /// that probes the SQLite audit backlog and pushes the snapshot onto the site health diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptionsValidator.cs index 0f798e64..04950f1b 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterOptionsValidator.cs @@ -37,6 +37,14 @@ public sealed class SqliteAuditWriterOptionsValidator : OptionsValidatorBase 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). diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterSaturatedException.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterSaturatedException.cs new file mode 100644 index 00000000..85239709 --- /dev/null +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriterSaturatedException.cs @@ -0,0 +1,24 @@ +namespace ZB.MOM.WW.ScadaBridge.AuditLog.Site; + +/// +/// Thrown (as a faulted task) by 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 AuditEvent) bounded; +/// catches it like any other primary failure, stashes the event in its drop-oldest ring and +/// bumps the SiteAuditWriteFailures health counter (review 2026-10-21 06/MEM-1). +/// Audit writes are best-effort: this never reaches the user-facing action. +/// +public sealed class SqliteAuditWriterSaturatedException : InvalidOperationException +{ + /// Initializes a new instance with the queue capacity that was exhausted. + /// The configured . + public SqliteAuditWriterSaturatedException(int capacity) + : base($"Audit write queue is full ({capacity} pending events); the background writer has fallen behind.") + { + Capacity = capacity; + } + + /// The write-queue capacity that was exhausted. + public int Capacity { get; } +} diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs index 320dc603..60767f2b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/SqliteAuditWriterRetentionTests.cs @@ -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.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))); } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task PurgeExpired_IsPaged_AndStillRemovesEveryEligibleRow() + { + await using var writer = CreateWriter(purgeBatchSize: 10); + var ids = new List(); + 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); + } + + /// + /// Review 2026-10-21 06/MEM-1: with the background writer parked (as behind a long purge + /// page) and the bounded queue full, WriteAsync must fail fast with + /// instead of parking a waiter that + /// holds its event for as long as the writer stays stuck. + /// + /// A task that represents the asynchronous operation. + [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(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); }