Merge branch 'worktree-agent-ac58bf5116b74ff84' into arch-review-remediation

2026-08-14 20:14:13 -04:00
parent 628a14c299 2cfcd89052
commit 70cf718949
4 changed files with 358 additions and 41 deletions
@@ -30,6 +30,9 @@ public class EventLogPurgeService : BackgroundService
    /// <summary>Number of events deleted per cap-purge batch.</summary>
    private const int CapPurgeBatchSize = 1000;

+    /// <summary>Number of events deleted per retention-purge batch — same size as <see cref="CapPurgeBatchSize"/>.</summary>
+    private const int RetentionPurgeBatchSize = 1000;
+
    private readonly SiteEventLogger _eventLogger;
    private readonly SiteEventLogOptions _options;
    private readonly ILogger<EventLogPurgeService> _logger;
@@ -118,21 +121,46 @@ public class EventLogPurgeService : BackgroundService
        }
    }

+    /// <summary>
+    /// Deletes events older than the retention window, in bounded
+    /// <see cref="RetentionPurgeBatchSize"/>-row slices per DELETE statement rather than
+    /// one unbounded DELETE for the whole expired set — mirroring the batching already
+    /// used a few lines below in <see cref="PurgeByStorageCap"/> (<see cref="CapPurgeBatchSize"/>).
+    /// An expired backlog in the tens/hundreds of thousands of rows would otherwise hold a
+    /// single SQLite write transaction (and the shared <c>_writeLock</c>) open for the
+    /// whole delete, blocking every concurrent <see cref="SiteEventLogger.LogEventAsync"/>
+    /// writer-loop flush for that entire duration.
+    /// </summary>
    private void PurgeByRetention()
    {
        var cutoff = DateTimeOffset.UtcNow.AddDays(-_options.RetentionDays).ToString("o");
+        var totalDeleted = 0;

-        var deleted = _eventLogger.WithConnection(connection =>
+        while (true)
        {
-            using var cmd = connection.CreateCommand();
-            cmd.CommandText = "DELETE FROM site_events WHERE timestamp < $cutoff";
-            cmd.Parameters.AddWithValue("$cutoff", cutoff);
-            return cmd.ExecuteNonQuery();
-        });
+            var deleted = _eventLogger.WithConnection(connection =>
+            {
+                using var cmd = connection.CreateCommand();
+                cmd.CommandText = $"""
+                    DELETE FROM site_events WHERE id IN (
+                        SELECT id FROM site_events
+                        WHERE timestamp < $cutoff
+                        LIMIT {RetentionPurgeBatchSize}
+                    )
+                    """;
+                cmd.Parameters.AddWithValue("$cutoff", cutoff);
+                return cmd.ExecuteNonQuery();
+            });

-        if (deleted > 0)
+            totalDeleted += deleted;
+
+            if (deleted < RetentionPurgeBatchSize)
+                break;
+        }
+
+        if (totalDeleted > 0)
        {
-            _logger.LogInformation("Purged {Count} events older than {Days} days", deleted, _options.RetentionDays);
+            _logger.LogInformation("Purged {Count} events older than {Days} days", totalDeleted, _options.RetentionDays);
        }
    }

@@ -34,12 +34,23 @@ namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging;
 /// </remarks>
 public class SiteEventLogger : ISiteEventLogger, IDisposable
 {
+    /// <summary>
+    /// Maximum number of queued events committed together in a single transaction
+    /// per writer-loop drain. Mirrors <c>SqliteAuditWriter</c>'s batch size
+    /// (<c>ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs</c>,
+    /// <c>ProcessWriteQueueAsync</c>/<c>FlushBatch</c>) — see the comment on
+    /// <see cref="ProcessWriteQueueAsync"/> for why one transaction per drain
+    /// replaces the previous one-transaction-per-event shape.
+    /// </summary>
+    internal const int WriteBatchSize = 256;
+
    private readonly SqliteConnection _connection;
    private readonly ILogger<SiteEventLogger> _logger;
    private readonly object _writeLock = new();
    private readonly Channel<PendingEvent> _writeQueue;
    private readonly Task _writerLoop;
    private long _failedWriteCount;
+    private long _committedBatchCount;
    private bool _disposed;

    /// <summary>
@@ -102,6 +113,14 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
    /// <inheritdoc />
    public long FailedWriteCount => Interlocked.Read(ref _failedWriteCount);

+    /// <summary>
+    /// Number of write-batch transactions successfully committed since startup. Test-only
+    /// observability hook (not part of <see cref="ISiteEventLogger"/>) proving the
+    /// "drain up to <see cref="WriteBatchSize"/> queued events into one transaction" shape
+    /// described on <see cref="ProcessWriteQueueAsync"/> — production code does not read it.
+    /// </summary>
+    internal long CommittedBatchCount => Interlocked.Read(ref _committedBatchCount);
+
    /// <summary>
    /// Runs <paramref name="action"/> against the shared connection while holding the
    /// write lock, so purge / query / record callers on different threads never use
@@ -212,54 +231,132 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
        return pending.Completion.Task;
    }

+    /// <summary>
+    /// Drains up to <see cref="WriteBatchSize"/> queued events per iteration and commits
+    /// them in a single transaction, instead of one transaction per event. Mirrors the
+    /// reference shape in <c>SqliteAuditWriter.ProcessWriteQueueAsync</c>/<c>FlushBatch</c>
+    /// (<c>ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs</c>): pull the first
+    /// item with the blocking <c>ReadAllAsync</c> enumerator, then greedily <c>TryRead</c>
+    /// whatever else is already queued (non-blocking) up to the batch cap, so a burst of
+    /// concurrent <see cref="LogEventAsync"/> callers amortises SQLite's per-transaction
+    /// fsync/journal overhead across the whole burst instead of paying it per row.
+    /// </summary>
    private async Task ProcessWriteQueueAsync()
    {
-        await foreach (var pending in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
+        var batch = new List<PendingEvent>(WriteBatchSize);
+
+        await foreach (var first in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
        {
-            try
+            batch.Clear();
+            batch.Add(first);
+
+            while (batch.Count < WriteBatchSize && _writeQueue.Reader.TryRead(out var next))
            {
-                var written = WithConnection(connection =>
+                batch.Add(next);
+            }
+
+            FlushBatch(batch);
+        }
+    }
+
+    /// <summary>
+    /// Commits a drained batch of pending events in one transaction. All-or-nothing per
+    /// batch: <c>site_events</c> ids are freshly minted GUIDs per <see cref="LogEventAsync"/>
+    /// call (no cross-call replay/de-dup concern the way <c>SqliteAuditWriter</c> has), so
+    /// unlike that reference there is no per-row duplicate-key swallow here — any failure
+    /// rolls back the whole batch and faults every event in it. <c>site_events</c> is a
+    /// CDC-replicated table (LocalDb Phase 1); its capture triggers fire per row on INSERT
+    /// same as before, so batching the commit changes only how many rows land per
+    /// transaction, not per-row trigger/replication behavior.
+    /// </summary>
+    private void FlushBatch(IReadOnlyList<PendingEvent> batch)
+    {
+        bool written;
+        Exception? failure = null;
+
+        try
+        {
+            written = WithConnection(connection =>
+            {
+                using var transaction = connection.BeginTransaction();
+                try
                {
                    using var cmd = connection.CreateCommand();
+                    cmd.Transaction = transaction;
                    cmd.CommandText = """
                        INSERT INTO site_events (id, timestamp, event_type, severity, instance_id, source, message, details)
                        VALUES ($id, $timestamp, $event_type, $severity, $instance_id, $source, $message, $details)
                        """;
-                    cmd.Parameters.AddWithValue("$id", pending.Id);
-                    cmd.Parameters.AddWithValue("$timestamp", pending.Timestamp);
-                    cmd.Parameters.AddWithValue("$event_type", pending.EventType);
-                    cmd.Parameters.AddWithValue("$severity", pending.Severity);
-                    cmd.Parameters.AddWithValue("$instance_id", (object?)pending.InstanceId ?? DBNull.Value);
-                    cmd.Parameters.AddWithValue("$source", pending.Source);
-                    cmd.Parameters.AddWithValue("$message", pending.Message);
-                    cmd.Parameters.AddWithValue("$details", (object?)pending.Details ?? DBNull.Value);
-                    cmd.ExecuteNonQuery();
-                });
+                    var pId = cmd.Parameters.Add("$id", SqliteType.Text);
+                    var pTimestamp = cmd.Parameters.Add("$timestamp", SqliteType.Text);
+                    var pEventType = cmd.Parameters.Add("$event_type", SqliteType.Text);
+                    var pSeverity = cmd.Parameters.Add("$severity", SqliteType.Text);
+                    var pInstanceId = cmd.Parameters.Add("$instance_id", SqliteType.Text);
+                    var pSource = cmd.Parameters.Add("$source", SqliteType.Text);
+                    var pMessage = cmd.Parameters.Add("$message", SqliteType.Text);
+                    var pDetails = cmd.Parameters.Add("$details", SqliteType.Text);

-                if (written)
-                {
-                    pending.Completion.TrySetResult();
+                    foreach (var pending in batch)
+                    {
+                        pId.Value = pending.Id;
+                        pTimestamp.Value = pending.Timestamp;
+                        pEventType.Value = pending.EventType;
+                        pSeverity.Value = pending.Severity;
+                        pInstanceId.Value = (object?)pending.InstanceId ?? DBNull.Value;
+                        pSource.Value = pending.Source;
+                        pMessage.Value = pending.Message;
+                        pDetails.Value = (object?)pending.Details ?? DBNull.Value;
+                        cmd.ExecuteNonQuery();
+                    }
+
+                    transaction.Commit();
+                    Interlocked.Increment(ref _committedBatchCount);
                }
-                else
+                catch
                {
-                    // WithConnection returns false only when the logger has been
-                    // disposed mid-drain; the event was not persisted. Fault the
-                    // Task instead of reporting false
-                    // success for a dropped audit event.
-                    pending.Completion.TrySetException(
-                        new ObjectDisposedException(nameof(SiteEventLogger),
-                            "Event could not be recorded: the event logger was disposed before the write completed."));
+                    transaction.Rollback();
+                    throw;
                }
-            }
-            catch (Exception ex)
+            });
+        }
+        catch (Exception ex)
+        {
+            failure = ex;
+            written = false;
+        }
+
+        if (failure is not null)
+        {
+            // A write failure must be observable. Count every event in the failed
+            // batch (Health Monitoring reads FailedWriteCount) and fault each
+            // caller's Task instead of silently discarding the exception.
+            Interlocked.Add(ref _failedWriteCount, batch.Count);
+            _logger.LogError(failure, "Failed to record {Count} event(s) in batch (sqlite {SqliteError})",
+                batch.Count, DescribeSqliteError(failure));
+            foreach (var pending in batch)
            {
-                // A write failure must be observable. Count it
-                // (Health Monitoring reads FailedWriteCount) and fault the caller's
-                // Task instead of silently discarding the exception.
-                Interlocked.Increment(ref _failedWriteCount);
-                _logger.LogError(ex, "Failed to record event: {EventType} from {Source} (sqlite {SqliteError})",
-                    pending.EventType, pending.Source, DescribeSqliteError(ex));
-                pending.Completion.TrySetException(ex);
+                pending.Completion.TrySetException(failure);
+            }
+            return;
+        }
+
+        if (written)
+        {
+            foreach (var pending in batch)
+            {
+                pending.Completion.TrySetResult();
+            }
+        }
+        else
+        {
+            // WithConnection returns false only when the logger has been
+            // disposed mid-drain; none of the batch was persisted. Fault every
+            // Task instead of reporting false success for a dropped event.
+            var disposedEx = new ObjectDisposedException(nameof(SiteEventLogger),
+                "Event could not be recorded: the event logger was disposed before the write completed.");
+            foreach (var pending in batch)
+            {
+                pending.Completion.TrySetException(disposedEx);
            }
        }
    }