diff --git a/docs/DesignDecisions.md b/docs/DesignDecisions.md index 2d1e479..2068527 100644 --- a/docs/DesignDecisions.md +++ b/docs/DesignDecisions.md @@ -244,14 +244,39 @@ authenticated call reads. The schema bootstrap now runs once, from the drain's ` When the channel is full the newest event is dropped and counted rather than blocking the producer: a stalled audit database must cost audit completeness, not gateway availability. Drops are logged once and reported in aggregate on each sweep. Shutdown drains what is buffered under a -2-second cap. Where no hosted service runs — the `apikey` admin CLI — the writer falls back to the -synchronous path, so audit is never buffered into a channel nobody drains. +2-second cap. + +Every other failure mode degrades to synchronous writes rather than to silent loss. The writer +falls back to the direct path whenever nothing is draining: before the drain attaches, after it +detaches, where no hosted service runs at all (the `apikey` admin CLI), and when the channel has +been completed — so no attach/detach sequence can leave producers filling a buffer with no reader. +If the drain loop itself dies it detaches the writer on the way out, which reverts every producer +to the direct path. A batch that will not commit is retried one event at a time, so an unwritable +row costs only itself instead of the up-to-63 good events sharing its transaction. + +**All** audit is channelled, including admin and CRUD records — dashboard key create/revoke/rotate, +session Close/Kill, and the library-forwarded API-key lifecycle entries. The alternative considered +was keeping those on the synchronous writer and channelling only high-volume denial audit. It was +rejected because a single dashboard key-create emits two records through two different seams (the +library's `create-key` via `IApiKeyAuditStore`, and the enriching `dashboard-create-key` via +`IAuditWriter`); splitting them across two durability regimes gives an auditor a per-producer +matrix to reason about instead of one rule. The residual exposure is explicit: **if the gateway +process dies between the enqueue and the batch commit, buffered audit events are lost.** The window +is bounded by drain latency — the drain wakes on every write and commits immediately, so it is +sub-millisecond under normal load — and it does not apply to the `apikey` CLI, which writes +synchronously. Audit is a best-effort record of what the gateway did, not a write-ahead log of what +it is about to do; a deployment that needs crash-durable admin audit should ship the events off-box +rather than rely on this table. `MxGateway:Security:AuditRetentionDays` (default 90, minimum 1) bounds the table: the drain sweeps at startup and hourly, deleting older rows. Retention cannot be configured off. The sweep compares -through SQLite's `datetime()` rather than on the stored ISO-8601 text, because timestamps written -from a non-UTC offset do not sort lexicographically against a UTC cutoff; a row whose timestamp -cannot be parsed yields NULL and is kept rather than deleted. +through SQLite's `datetime()` rather than on the stored ISO-8601 text. Text comparison is correct +only while every row is UTC-normalized — which the canonical model guarantees for rows written +through the store, but not for rows that entered the table any other way — and on a mixed-format +column it silently deletes live audit, because `2026-05-17T09:00:00-05:00` is two hours after a +`2026-05-17T12:00:00+00:00` cutoff yet sorts before it. Comparing instants is correct however the +text got there, and a timestamp `datetime()` cannot parse yields NULL, so undateable audit is kept +rather than swept. ## Authorization diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/AuditDrainService.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/AuditDrainService.cs new file mode 100644 index 0000000..d4519fb --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/AuditDrainService.cs @@ -0,0 +1,274 @@ +using ZB.MOM.WW.Audit; +using ZB.MOM.WW.MxGateway.Server.Configuration; + +namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; + +/// +/// Drains onto the durable , +/// owns the one-time schema bootstrap, and sweeps audit rows past their retention window. +/// +/// +/// Batching is the point: up to buffered events are committed in a +/// single transaction, so a burst of constraint denials costs a handful of commits instead of +/// one per denied tag. The bootstrap runs here — before the writer starts enqueueing — so no +/// audit write ever pays a CREATE TABLE IF NOT EXISTS round-trip. +/// +/// Every failure mode ends in synchronous audit rather than silent loss: a batch that will not +/// commit is retried one event at a time so only the offending row is dropped, and a drain loop +/// that dies detaches the writer, which reverts every producer to the direct write path. +/// +/// +/// The channel writer whose buffered events are drained. +/// The durable sink events are committed to. +/// Security options carrying the audit retention window. +/// Clock used for the retention cutoff and sweep interval. +/// Logger for bootstrap, drain and sweep diagnostics. +public sealed class AuditDrainService( + ChannelAuditWriter writer, + IAuditEventSink sink, + SecurityOptions security, + TimeProvider timeProvider, + ILogger logger) : BackgroundService +{ + /// Maximum number of audit events committed in one transaction per drain pass. + public const int MaxBatchSize = 64; + + /// How often the retention sweep runs while the gateway is up. + public static readonly TimeSpan RetentionSweepInterval = TimeSpan.FromHours(1); + + /// Upper bound on how long shutdown waits for the remaining buffered events. + private static readonly TimeSpan ShutdownDrainCap = TimeSpan.FromSeconds(2); + + /// + /// Bootstraps the audit table, runs one retention sweep, then attaches the drain so the + /// writer switches from synchronous write-through to enqueueing. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public override async Task StartAsync(CancellationToken cancellationToken) + { + try + { + await sink.EnsureInitializedAsync(cancellationToken).ConfigureAwait(false); + await SweepRetentionAsync(cancellationToken).ConfigureAwait(false); + } + catch (Exception exception) + { + // Audit is best-effort: a bootstrap failure must not take the gateway down. The + // sink's own latch will retry the schema check on the first write. + logger.LogWarning(exception, "Audit store bootstrap failed; audit writes will retry the schema check."); + } + + writer.AttachDrain(); + + await base.StartAsync(cancellationToken).ConfigureAwait(false); + } + + /// + /// Detaches the drain (so late writes go straight to the sink) and gives the buffered + /// events a bounded window to reach the store. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public override async Task StopAsync(CancellationToken cancellationToken) + { + writer.DetachDrain(); + writer.CompleteWriting(); + + await base.StopAsync(cancellationToken).ConfigureAwait(false); + + using CancellationTokenSource drainCap = new(ShutdownDrainCap); + try + { + await DrainPendingAsync(drainCap.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + logger.LogWarning( + "Shutdown drain exceeded {CapSeconds}s; remaining buffered audit events were not persisted.", + ShutdownDrainCap.TotalSeconds); + } + } + + /// + /// Commits every event currently buffered, in batches of at most . + /// + /// Token to observe for cancellation. + /// The number of events persisted. + /// + /// The drain was cancelled — at shutdown this is the 2-second cap expiring, which the caller + /// reports as unpersisted audit rather than as a store fault. + /// + public async Task DrainPendingAsync(CancellationToken cancellationToken) + { + int persisted = 0; + List batch = new(MaxBatchSize); + + while (!cancellationToken.IsCancellationRequested) + { + batch.Clear(); + while (batch.Count < MaxBatchSize && writer.Reader.TryRead(out AuditEvent? auditEvent)) + { + batch.Add(auditEvent); + } + + if (batch.Count == 0) + { + break; + } + + try + { + await sink.InsertBatchAsync(batch, cancellationToken).ConfigureAwait(false); + persisted += batch.Count; + } + catch (OperationCanceledException) + { + // Cancellation is the shutdown cap, not a store fault: surface it so StopAsync + // reports unpersisted audit instead of misreporting it as a failed write. + throw; + } + catch (Exception exception) + { + logger.LogWarning( + exception, + "Failed to commit a batch of {Count} audit events; retrying them individually.", + batch.Count); + + persisted += await InsertIndividuallyAsync(batch, cancellationToken).ConfigureAwait(false); + } + } + + return persisted; + } + + /// + /// Deletes audit rows older than MxGateway:Security:AuditRetentionDays, and reports the + /// running total of audit events dropped by channel pressure since startup. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + public async Task SweepRetentionAsync(CancellationToken cancellationToken) + { + DateTimeOffset cutoff = timeProvider.GetUtcNow() - TimeSpan.FromDays(security.AuditRetentionDays); + + try + { + int deleted = await sink.DeleteOlderThanAsync(cutoff, cancellationToken).ConfigureAwait(false); + if (deleted > 0) + { + logger.LogInformation( + "Audit retention sweep removed {Deleted} events older than {Cutoff:o} ({RetentionDays} days).", + deleted, + cutoff, + security.AuditRetentionDays); + } + } + catch (Exception exception) + { + logger.LogWarning(exception, "Audit retention sweep failed; it will be retried on the next interval."); + } + + long dropped = writer.DroppedCount; + if (dropped > 0) + { + logger.LogWarning( + "{Dropped} audit events have been dropped since startup because the audit channel was full.", + dropped); + } + } + + /// + protected override async Task ExecuteAsync(CancellationToken stoppingToken) + { + await Task.WhenAll( + DrainLoopAsync(stoppingToken), + RetentionLoopAsync(stoppingToken)).ConfigureAwait(false); + } + + // Re-inserts a failed batch one event at a time so a single unwritable row costs only itself + // rather than the up-to-MaxBatchSize good events that happened to share its transaction. + private async Task InsertIndividuallyAsync( + IReadOnlyList batch, + CancellationToken cancellationToken) + { + int persisted = 0; + + foreach (AuditEvent auditEvent in batch) + { + try + { + await sink.InsertAsync(auditEvent, cancellationToken).ConfigureAwait(false); + persisted++; + } + catch (OperationCanceledException) + { + throw; + } + catch (Exception exception) + { + logger.LogWarning( + exception, + "Dropped audit event {EventId} (action {Action}); it could not be persisted individually.", + auditEvent.EventId, + auditEvent.Action); + } + } + + logger.LogWarning( + "Recovered {Persisted} of {Count} audit events from a failed batch.", + persisted, + batch.Count); + + return persisted; + } + + private async Task DrainLoopAsync(CancellationToken stoppingToken) + { + try + { + while (await writer.Reader.WaitToReadAsync(stoppingToken).ConfigureAwait(false)) + { + await DrainPendingAsync(stoppingToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown; StopAsync performs the final bounded drain. + } + catch (Exception exception) + { + // A dead drain loop would silently discard every later audit write, because producers + // keep enqueueing into a channel nobody reads. Detaching (below) reverts them to the + // synchronous path, so audit degrades in latency rather than disappearing. + logger.LogError(exception, "Audit drain loop failed; reverting to synchronous audit writes."); + } + finally + { + writer.DetachDrain(); + } + } + + private async Task RetentionLoopAsync(CancellationToken stoppingToken) + { + try + { + using PeriodicTimer timer = new(RetentionSweepInterval, timeProvider); + + while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) + { + await SweepRetentionAsync(stoppingToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Shutdown. + } + catch (Exception exception) + { + // Retention is unbounded growth if it stops: say so loudly rather than letting the + // audit table grow forever behind a silently dead timer loop. + logger.LogError(exception, "Audit retention loop failed; expired audit rows will no longer be swept."); + } + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs index 3b204cf..d85c5a2 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/ChannelAuditWriter.cs @@ -1,6 +1,5 @@ using System.Threading.Channels; using ZB.MOM.WW.Audit; -using ZB.MOM.WW.MxGateway.Server.Configuration; namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; @@ -27,6 +26,8 @@ namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; /// straight through to . Enqueueing into a channel nobody will /// ever read would silently discard audit in the processes that have no hosted services — the /// apikey admin CLI and the DI-only tests — so those keep the original synchronous path. +/// The same fallback covers a completed channel, so no combination of attach/detach can leave +/// producers writing into a buffer that will never be read. /// /// public sealed class ChannelAuditWriter : IAuditWriter @@ -79,8 +80,9 @@ public sealed class ChannelAuditWriter : IAuditWriter public void AttachDrain() => Volatile.Write(ref _drainAttached, 1); /// - /// Marks the drain as no longer running, so writes revert to the synchronous path. Called - /// at shutdown so late audit is still persisted rather than buffered into a dead channel. + /// Marks the drain as no longer running, so writes revert to the synchronous path. Called at + /// shutdown, and whenever the drain loop dies, so late audit is still persisted rather than + /// buffered into a channel with no reader. /// public void DetachDrain() => Volatile.Write(ref _drainAttached, 0); @@ -100,11 +102,13 @@ public sealed class ChannelAuditWriter : IAuditWriter return _directWriter.WriteAsync(auditEvent, cancellationToken); } - // Under DropWrite a full channel still reports success (the discard surfaces through the - // itemDropped callback); a false here means the channel is completed, which is also a drop. + // Under DropWrite a full channel still reports SUCCESS — the discard surfaces through the + // itemDropped callback. So a false here does not mean "full", it means the channel has + // been completed and no drain will ever read it again (shutdown, or a re-attach onto a + // dead channel). Writing through is the only outcome that keeps the event. if (!_channel.Writer.TryWrite(auditEvent)) { - RecordDrop(auditEvent); + return _directWriter.WriteAsync(auditEvent, cancellationToken); } return Task.CompletedTask; @@ -130,204 +134,3 @@ public sealed class ChannelAuditWriter : IAuditWriter } } } - -/// -/// Drains onto the durable , -/// owns the one-time schema bootstrap, and sweeps audit rows past their retention window. -/// -/// -/// Batching is the point: up to buffered events are committed in a -/// single transaction, so a burst of constraint denials costs a handful of commits instead of -/// one per denied tag. The bootstrap runs here — before the writer starts enqueueing — so no -/// audit write ever pays a CREATE TABLE IF NOT EXISTS round-trip. -/// -/// The channel writer whose buffered events are drained. -/// The durable sink events are committed to. -/// Security options carrying the audit retention window. -/// Clock used for the retention cutoff and sweep interval. -/// Logger for bootstrap, drain and sweep diagnostics. -public sealed class AuditDrainService( - ChannelAuditWriter writer, - IAuditEventSink sink, - SecurityOptions security, - TimeProvider timeProvider, - ILogger logger) : BackgroundService -{ - /// Maximum number of audit events committed in one transaction per drain pass. - public const int MaxBatchSize = 64; - - /// How often the retention sweep runs while the gateway is up. - public static readonly TimeSpan RetentionSweepInterval = TimeSpan.FromHours(1); - - /// Upper bound on how long shutdown waits for the remaining buffered events. - private static readonly TimeSpan ShutdownDrainCap = TimeSpan.FromSeconds(2); - - /// - /// Bootstraps the audit table, runs one retention sweep, then attaches the drain so the - /// writer switches from synchronous write-through to enqueueing. - /// - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - public override async Task StartAsync(CancellationToken cancellationToken) - { - try - { - await sink.EnsureInitializedAsync(cancellationToken).ConfigureAwait(false); - await SweepRetentionAsync(cancellationToken).ConfigureAwait(false); - } - catch (Exception exception) - { - // Audit is best-effort: a bootstrap failure must not take the gateway down. The - // sink's own latch will retry the schema check on the first write. - logger.LogWarning(exception, "Audit store bootstrap failed; audit writes will retry the schema check."); - } - - writer.AttachDrain(); - - await base.StartAsync(cancellationToken).ConfigureAwait(false); - } - - /// - /// Detaches the drain (so late writes go straight to the sink) and gives the buffered - /// events a bounded window to reach the store. - /// - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - public override async Task StopAsync(CancellationToken cancellationToken) - { - writer.DetachDrain(); - writer.CompleteWriting(); - - await base.StopAsync(cancellationToken).ConfigureAwait(false); - - using CancellationTokenSource drainCap = new(ShutdownDrainCap); - try - { - await DrainPendingAsync(drainCap.Token).ConfigureAwait(false); - } - catch (OperationCanceledException) - { - logger.LogWarning( - "Shutdown drain exceeded {CapSeconds}s; remaining buffered audit events were not persisted.", - ShutdownDrainCap.TotalSeconds); - } - } - - /// - /// Commits every event currently buffered, in batches of at most . - /// - /// Token to observe for cancellation. - /// The number of events persisted. - public async Task DrainPendingAsync(CancellationToken cancellationToken) - { - int persisted = 0; - List batch = new(MaxBatchSize); - - while (!cancellationToken.IsCancellationRequested) - { - batch.Clear(); - while (batch.Count < MaxBatchSize && writer.Reader.TryRead(out AuditEvent? auditEvent)) - { - batch.Add(auditEvent); - } - - if (batch.Count == 0) - { - break; - } - - try - { - await sink.InsertBatchAsync(batch, cancellationToken).ConfigureAwait(false); - persisted += batch.Count; - } - catch (Exception exception) - { - // Best-effort: the batch is lost, but the drain keeps going so one poisoned - // batch cannot stall every later audit event behind it. - logger.LogWarning( - exception, - "Failed to persist a batch of {Count} audit events; audit writes are best-effort and were suppressed.", - batch.Count); - } - } - - return persisted; - } - - /// - /// Deletes audit rows older than MxGateway:Security:AuditRetentionDays, and reports - /// any audit events dropped by channel pressure since the last sweep. - /// - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - public async Task SweepRetentionAsync(CancellationToken cancellationToken) - { - DateTimeOffset cutoff = timeProvider.GetUtcNow() - TimeSpan.FromDays(security.AuditRetentionDays); - - try - { - int deleted = await sink.DeleteOlderThanAsync(cutoff, cancellationToken).ConfigureAwait(false); - if (deleted > 0) - { - logger.LogInformation( - "Audit retention sweep removed {Deleted} events older than {Cutoff:o} ({RetentionDays} days).", - deleted, - cutoff, - security.AuditRetentionDays); - } - } - catch (Exception exception) - { - logger.LogWarning(exception, "Audit retention sweep failed; it will be retried on the next interval."); - } - - long dropped = writer.DroppedCount; - if (dropped > 0) - { - logger.LogWarning( - "{Dropped} audit events have been dropped since startup because the audit channel was full.", - dropped); - } - } - - /// - protected override async Task ExecuteAsync(CancellationToken stoppingToken) - { - await Task.WhenAll( - DrainLoopAsync(stoppingToken), - RetentionLoopAsync(stoppingToken)).ConfigureAwait(false); - } - - private async Task DrainLoopAsync(CancellationToken stoppingToken) - { - try - { - while (await writer.Reader.WaitToReadAsync(stoppingToken).ConfigureAwait(false)) - { - await DrainPendingAsync(stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) - { - // Shutdown; StopAsync performs the final bounded drain. - } - } - - private async Task RetentionLoopAsync(CancellationToken stoppingToken) - { - using PeriodicTimer timer = new(RetentionSweepInterval, timeProvider); - - try - { - while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false)) - { - await SweepRetentionAsync(stoppingToken).ConfigureAwait(false); - } - } - catch (OperationCanceledException) - { - // Shutdown. - } - } -} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/IAuditEventSink.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/IAuditEventSink.cs new file mode 100644 index 0000000..7f12fb5 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/IAuditEventSink.cs @@ -0,0 +1,38 @@ +using ZB.MOM.WW.Audit; + +namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; + +/// +/// Durable sink the audit pipeline persists canonical s through. +/// It exists so the write path () and the batching drain +/// () depend on the storage contract rather than on the +/// concrete . +/// +public interface IAuditEventSink +{ + /// + /// Bootstraps the backing storage. Called once at startup so no write path pays a schema + /// round-trip; implementations must be idempotent and safe to call concurrently. + /// + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task EnsureInitializedAsync(CancellationToken cancellationToken); + + /// Persists a single canonical audit event. + /// The canonical event to persist. + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken); + + /// Persists a batch of canonical audit events as one unit of work. + /// The canonical events to persist. + /// Token to observe for cancellation. + /// A task that represents the asynchronous operation. + Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken); + + /// Deletes every audit row that occurred strictly before . + /// The retention cutoff; rows older than this are removed. + /// Token to observe for cancellation. + /// The number of rows deleted. + Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken); +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs index db83c62..3ad7f91 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Security/Audit/SqliteCanonicalAuditStore.cs @@ -5,41 +5,6 @@ using ZB.MOM.WW.Auth.ApiKeys.Sqlite; namespace ZB.MOM.WW.MxGateway.Server.Security.Audit; -/// -/// Durable sink the audit pipeline persists canonical s through. -/// It exists so the write path () and the batching drain -/// () depend on the storage contract rather than on the -/// concrete . -/// -public interface IAuditEventSink -{ - /// - /// Bootstraps the backing storage. Called once at startup so no write path pays a schema - /// round-trip; implementations must be idempotent and safe to call concurrently. - /// - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - Task EnsureInitializedAsync(CancellationToken cancellationToken); - - /// Persists a single canonical audit event. - /// The canonical event to persist. - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - Task InsertAsync(AuditEvent auditEvent, CancellationToken cancellationToken); - - /// Persists a batch of canonical audit events as one unit of work. - /// The canonical events to persist. - /// Token to observe for cancellation. - /// A task that represents the asynchronous operation. - Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken); - - /// Deletes every audit row that occurred strictly before . - /// The retention cutoff; rows older than this are removed. - /// Token to observe for cancellation. - /// The number of rows deleted. - Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken); -} - /// /// MxGateway-owned, append-only SQLite store for canonical /// s. It writes to a NEW audit_event table in the @@ -178,10 +143,14 @@ public sealed class SqliteCanonicalAuditStore(AuthSqliteConnectionFactory connec /// /// /// The comparison goes through SQLite's datetime() rather than comparing the stored - /// ISO-8601 text directly: rows written from a non-UTC (the - /// library's CreatedUtc is caller-supplied) do not sort lexicographically against a - /// UTC cutoff. A row whose timestamp datetime() cannot parse yields NULL and is - /// therefore never deleted — audit that cannot be dated is kept, not swept. + /// ISO-8601 text directly. Text comparison is only correct while every row is UTC-normalized + /// ISO-8601 — which guarantees for rows written through + /// this store, but not for rows that entered the table any other way (a repair script, an + /// older schema, a future producer). On a mixed-format column a text comparison silently + /// deletes live audit: 2026-05-17T09:00:00-05:00 is two hours AFTER a + /// 2026-05-17T12:00:00+00:00 cutoff yet sorts before it. Comparing instants is correct + /// regardless of how the text got there, and anything datetime() cannot parse yields + /// NULL and is therefore never deleted — audit that cannot be dated is kept, not swept. /// public async Task DeleteOlderThanAsync(DateTimeOffset cutoffUtc, CancellationToken cancellationToken) { diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs index c96dce5..3b5a273 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Security/Audit/ChannelAuditWriterTests.cs @@ -1,3 +1,4 @@ +using Microsoft.Data.Sqlite; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Time.Testing; using ZB.MOM.WW.Audit; @@ -10,10 +11,11 @@ namespace ZB.MOM.WW.MxGateway.Tests.Security.Audit; /// /// Tests the bounded, asynchronous audit path: (enqueue-only -/// once a drain is attached, drops rather than blocks when the channel is full) and -/// (batched drain, one-time table bootstrap, retention sweep). -/// The channel makes the already-documented best-effort audit contract explicit and bounded: -/// a partially denied bulk RPC no longer pays a SQLite round-trip per denied tag. +/// once a drain is attached, drops rather than blocks when the channel is full, writes through +/// whenever nothing is draining) and (batched drain, poison-batch +/// isolation, one-time table bootstrap, retention sweep). The channel makes the already-documented +/// best-effort audit contract explicit and bounded: a partially denied bulk RPC no longer pays a +/// SQLite round-trip per denied tag. /// public sealed class ChannelAuditWriterTests : IDisposable { @@ -71,6 +73,29 @@ public sealed class ChannelAuditWriterTests : IDisposable Assert.All(sink.BatchSizes, size => Assert.True(size <= AuditDrainService.MaxBatchSize)); } + /// + /// A batch that will not commit is retried one event at a time, so a single unwritable row + /// costs only itself instead of taking the up-to-63 good events sharing its transaction. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task DrainPendingAsync_WhenBatchFails_RetriesIndividuallyAndKeepsSurvivors() + { + CountingAuditSink sink = new() { PoisonAction = "poison" }; + (ChannelAuditWriter writer, AuditDrainService drain) = CreateWriterAndDrain(sink); + writer.AttachDrain(); + + await writer.WriteAsync(MakeEvent("good-1"), CancellationToken.None); + await writer.WriteAsync(MakeEvent("poison"), CancellationToken.None); + await writer.WriteAsync(MakeEvent("good-2"), CancellationToken.None); + + int persisted = await drain.DrainPendingAsync(CancellationToken.None); + + // The batch commit fails on the poison event; the per-event retry still lands both others. + Assert.Equal(2, persisted); + Assert.Equal(["good-1", "good-2"], sink.Events.Select(auditEvent => auditEvent.Action)); + } + /// /// When the bounded channel is full the write is dropped rather than blocking the caller, /// and the drop is counted. A stalled or slow audit database must never stall an RPC. @@ -89,13 +114,12 @@ public sealed class ChannelAuditWriterTests : IDisposable // Every call must complete synchronously: the channel never blocks a producer. Task write = writer.WriteAsync(MakeEvent($"denied-{index}"), CancellationToken.None); Assert.True(write.IsCompletedSuccessfully); + await write; } Assert.Equal(overflow, writer.DroppedCount); // The channel still holds exactly its capacity; nothing reached the sink (no drain ran). Assert.Equal(0, sink.InsertBatchCalls); - - await Task.CompletedTask; } /// @@ -114,6 +138,26 @@ public sealed class ChannelAuditWriterTests : IDisposable Assert.Equal("dashboard-create-key", Assert.Single(sink.Events).Action); } + /// + /// A completed channel has no future reader, so an attached writer must write through rather + /// than discard. This is the re-attach-after-shutdown footgun: enqueueing would lose 100% of + /// audit silently. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_AfterChannelCompleted_WritesThroughInsteadOfDiscarding() + { + CountingAuditSink sink = new(); + (ChannelAuditWriter writer, _) = CreateWriterAndDrain(sink); + writer.AttachDrain(); + writer.CompleteWriting(); + + await writer.WriteAsync(MakeEvent("constraint-denied"), CancellationToken.None); + + Assert.Equal("constraint-denied", Assert.Single(sink.Events).Action); + Assert.Equal(0, writer.DroppedCount); + } + /// /// The retention sweep deletes rows older than AuditRetentionDays, measured from the /// injected clock, and runs once at startup after the one-time table bootstrap. @@ -137,29 +181,42 @@ public sealed class ChannelAuditWriterTests : IDisposable } /// - /// The real SQLite sink deletes only rows older than the cutoff, and its bootstrap latch - /// means the CREATE TABLE round-trip is not repeated on every subsequent write. + /// The sweep compares instants, not stored text, and never deletes a row it cannot date. /// /// A task that represents the asynchronous operation. [Fact] - public async Task SqliteStore_DeleteOlderThan_RemovesOnlyRowsBeyondTheRetentionWindow() + public async Task SqliteStore_DeleteOlderThan_ComparesInstantsAndKeepsUnparseableRows() { - SqliteCanonicalAuditStore store = CreateStore(); - DateTimeOffset now = new(2026, 8, 15, 12, 0, 0, TimeSpan.Zero); + (SqliteCanonicalAuditStore store, AuthSqliteConnectionFactory factory) = CreateStore(); + + DateTimeOffset cutoff = new(2026, 5, 17, 12, 0, 0, TimeSpan.Zero); await store.InsertBatchAsync( [ - MakeEvent("stale", now.AddDays(-120)), - // A non-UTC offset must still compare correctly against the cutoff. - MakeEvent("stale-offset", now.AddDays(-100).ToOffset(TimeSpan.FromHours(-5))), - MakeEvent("fresh", now.AddDays(-1)), + MakeEvent("swept", new DateTimeOffset(2026, 5, 17, 11, 0, 0, TimeSpan.Zero)), + MakeEvent("kept-fresh", new DateTimeOffset(2026, 5, 18, 0, 0, 0, TimeSpan.Zero)), ], CancellationToken.None); - int deleted = await store.DeleteOlderThanAsync(now.AddDays(-90), CancellationToken.None); + // Both discriminating rows are written as raw text, because AuditEvent.OccurredAtUtc + // normalizes to UTC and so cannot express them. They stand for audit that reached the + // table any other way — a repair script, an older schema, a future producer. + // + // 09:00 at -05:00 is 14:00 UTC, two hours AFTER the cutoff, yet its text sorts BEFORE + // "2026-05-17T12:00:00.0000000+00:00". A lexicographic sweep deletes it; comparing + // instants keeps it. + await InsertRawRowAsync(factory, "kept-offset", "2026-05-17T09:00:00.1234567-05:00"); - Assert.Equal(2, deleted); - Assert.Equal("fresh", Assert.Single(await store.ListRecentAsync(10, CancellationToken.None)).Action); + // Undateable audit is kept, never guessed at. This text sorts BELOW the cutoff, so a + // lexicographic sweep deletes it, while datetime() yields NULL and leaves it alone. + await InsertRawRowAsync(factory, "kept-unparseable", "0000-not-a-timestamp"); + + int deleted = await store.DeleteOlderThanAsync(cutoff, CancellationToken.None); + + Assert.Equal(1, deleted); + Assert.Equal( + ["kept-fresh", "kept-offset", "kept-unparseable"], + (await ListActionsAsync(factory)).OrderBy(action => action, StringComparer.Ordinal)); } private static (ChannelAuditWriter Writer, AuditDrainService Drain) CreateWriterAndDrain( @@ -179,11 +236,51 @@ public sealed class ChannelAuditWriterTests : IDisposable return (writer, drain); } - private SqliteCanonicalAuditStore CreateStore() + private (SqliteCanonicalAuditStore Store, AuthSqliteConnectionFactory Factory) CreateStore() { TempDatabaseDirectory directory = TempDatabaseDirectory.Create("mxgateway-channel-audit"); _tempDirectories.Add(directory); - return new SqliteCanonicalAuditStore(new AuthSqliteConnectionFactory(directory.DatabasePath())); + AuthSqliteConnectionFactory factory = new(directory.DatabasePath()); + return (new SqliteCanonicalAuditStore(factory), factory); + } + + // Writes an audit row whose occurred_at_utc bypasses the store's DateTimeOffset formatting, + // so the sweep can be shown to leave undateable audit alone. + private static async Task InsertRawRowAsync( + AuthSqliteConnectionFactory factory, + string action, + string occurredAtUtc) + { + await using SqliteConnection connection = await factory.OpenConnectionAsync(CancellationToken.None); + await using SqliteCommand command = connection.CreateCommand(); + command.CommandText = + """ + INSERT INTO audit_event (event_id, occurred_at_utc, actor, action, outcome) + VALUES ($event_id, $occurred_at_utc, 'operator01', $action, 'Denied'); + """; + command.Parameters.AddWithValue("$event_id", Guid.NewGuid().ToString()); + command.Parameters.AddWithValue("$occurred_at_utc", occurredAtUtc); + command.Parameters.AddWithValue("$action", action); + + await command.ExecuteNonQueryAsync(CancellationToken.None); + } + + // Reads actions straight from SQL: ListRecentAsync would throw on the unparseable timestamp. + private static async Task> ListActionsAsync(AuthSqliteConnectionFactory factory) + { + await using SqliteConnection connection = await factory.OpenConnectionAsync(CancellationToken.None); + await using SqliteCommand command = connection.CreateCommand(); + command.CommandText = "SELECT action FROM audit_event;"; + + List actions = []; + + await using SqliteDataReader reader = await command.ExecuteReaderAsync(CancellationToken.None); + while (await reader.ReadAsync(CancellationToken.None)) + { + actions.Add(reader.GetString(0)); + } + + return actions; } private static AuditEvent MakeEvent(string action, DateTimeOffset? occurredAtUtc = null) => new() @@ -209,22 +306,27 @@ public sealed class ChannelAuditWriterTests : IDisposable /// /// In-memory that records every call, so a test can prove the - /// write path did NOT touch the sink and that the drain batched what it did write. + /// write path did NOT touch the sink and that the drain batched what it did write. Setting + /// makes any write containing that action fail, modelling a row + /// the store refuses. /// private sealed class CountingAuditSink : IAuditEventSink { private readonly Lock _gate = new(); + /// Gets the action whose presence makes a write fail, or null to accept everything. + public string? PoisonAction { get; init; } + /// Gets the events handed to the sink, in the order they were written. public List Events { get; } = []; - /// Gets the size of each batch the sink was asked to insert. + /// Gets the size of each batch the sink accepted. public List BatchSizes { get; } = []; /// Gets the cutoffs the sink was asked to delete below. public List DeleteCutoffs { get; } = []; - /// Gets the number of calls. + /// Gets the number of accepted calls. public int InsertBatchCalls => BatchSizes.Count; /// Gets the number of calls. @@ -248,6 +350,13 @@ public sealed class ChannelAuditWriterTests : IDisposable /// public Task InsertBatchAsync(IReadOnlyList auditEvents, CancellationToken cancellationToken) { + if (PoisonAction is not null + && auditEvents.Any(auditEvent => auditEvent.Action == PoisonAction)) + { + return Task.FromException( + new InvalidOperationException($"Refused a write of {auditEvents.Count} audit events.")); + } + lock (_gate) { BatchSizes.Add(auditEvents.Count);