diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
index 5668d75f..13bf874d 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
@@ -30,6 +30,9 @@ public class EventLogPurgeService : BackgroundService
/// Number of events deleted per cap-purge batch.
private const int CapPurgeBatchSize = 1000;
+ /// Number of events deleted per retention-purge batch — same size as .
+ private const int RetentionPurgeBatchSize = 1000;
+
private readonly SiteEventLogger _eventLogger;
private readonly SiteEventLogOptions _options;
private readonly ILogger _logger;
@@ -118,21 +121,46 @@ public class EventLogPurgeService : BackgroundService
}
}
+ ///
+ /// Deletes events older than the retention window, in bounded
+ /// -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 ().
+ /// An expired backlog in the tens/hundreds of thousands of rows would otherwise hold a
+ /// single SQLite write transaction (and the shared _writeLock) open for the
+ /// whole delete, blocking every concurrent
+ /// writer-loop flush for that entire duration.
+ ///
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);
}
}
diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs
index 95f2361c..169477c0 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogger.cs
@@ -34,12 +34,23 @@ namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging;
///
public class SiteEventLogger : ISiteEventLogger, IDisposable
{
+ ///
+ /// Maximum number of queued events committed together in a single transaction
+ /// per writer-loop drain. Mirrors SqliteAuditWriter's batch size
+ /// (ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs,
+ /// ProcessWriteQueueAsync/FlushBatch) — see the comment on
+ /// for why one transaction per drain
+ /// replaces the previous one-transaction-per-event shape.
+ ///
+ internal const int WriteBatchSize = 256;
+
private readonly SqliteConnection _connection;
private readonly ILogger _logger;
private readonly object _writeLock = new();
private readonly Channel _writeQueue;
private readonly Task _writerLoop;
private long _failedWriteCount;
+ private long _committedBatchCount;
private bool _disposed;
///
@@ -102,6 +113,14 @@ public class SiteEventLogger : ISiteEventLogger, IDisposable
///
public long FailedWriteCount => Interlocked.Read(ref _failedWriteCount);
+ ///
+ /// Number of write-batch transactions successfully committed since startup. Test-only
+ /// observability hook (not part of ) proving the
+ /// "drain up to queued events into one transaction" shape
+ /// described on — production code does not read it.
+ ///
+ internal long CommittedBatchCount => Interlocked.Read(ref _committedBatchCount);
+
///
/// Runs 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;
}
+ ///
+ /// Drains up to queued events per iteration and commits
+ /// them in a single transaction, instead of one transaction per event. Mirrors the
+ /// reference shape in SqliteAuditWriter.ProcessWriteQueueAsync/FlushBatch
+ /// (ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs): pull the first
+ /// item with the blocking ReadAllAsync enumerator, then greedily TryRead
+ /// whatever else is already queued (non-blocking) up to the batch cap, so a burst of
+ /// concurrent callers amortises SQLite's per-transaction
+ /// fsync/journal overhead across the whole burst instead of paying it per row.
+ ///
private async Task ProcessWriteQueueAsync()
{
- await foreach (var pending in _writeQueue.Reader.ReadAllAsync().ConfigureAwait(false))
+ var batch = new List(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);
+ }
+ }
+
+ ///
+ /// Commits a drained batch of pending events in one transaction. All-or-nothing per
+ /// batch: site_events ids are freshly minted GUIDs per
+ /// call (no cross-call replay/de-dup concern the way SqliteAuditWriter 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. site_events 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.
+ ///
+ private void FlushBatch(IReadOnlyList 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);
}
}
}
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs
index 22240faa..494385e6 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs
@@ -106,6 +106,40 @@ public class EventLogPurgeServiceTests : IDisposable
Assert.Equal(3, GetEventCount());
}
+ [Fact]
+ public void PurgeByRetention_OverBatchSize_DeletesAllExpiredRowsAcrossSlices()
+ {
+ // WP1.7: the retention purge must slice a large expired backlog into
+ // bounded DELETE batches (1000 rows/iteration, mirroring the storage-cap
+ // purge a few lines below in the same file) rather than one unbounded
+ // DELETE. Seed more than one slice's worth of expired rows plus a
+ // handful of recent rows, and assert the whole expired set is gone —
+ // not just the first 1000 rows — while the recent rows survive.
+ const int expiredCount = 2500; // > one 1000-row slice, not an exact multiple
+ var expiredBase = DateTimeOffset.UtcNow.AddDays(-31);
+ _eventLogger.WithConnection(connection =>
+ {
+ for (var i = 0; i < expiredCount; i++)
+ {
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = """
+ INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
+ VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
+ """;
+ cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
+ cmd.Parameters.AddWithValue("$ts", expiredBase.AddSeconds(i).ToString("o"));
+ cmd.ExecuteNonQuery();
+ }
+ });
+ InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-1));
+ InsertEventWithTimestamp(DateTimeOffset.UtcNow);
+
+ var purge = CreatePurgeService();
+ purge.RunPurge();
+
+ Assert.Equal(2, GetEventCount());
+ }
+
[Fact]
public void PurgeByStorageCap_DeletesOldestWhenOverCap()
{
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs
new file mode 100644
index 00000000..c274f871
--- /dev/null
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLoggerBatchingTests.cs
@@ -0,0 +1,158 @@
+using Microsoft.Extensions.Logging.Abstractions;
+using Microsoft.Extensions.Options;
+
+namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests;
+
+///
+/// WP1.7 (arch-review remediation, 2026-08-14): 's
+/// background writer loop must drain its bounded channel into batched transactions
+/// (up to events per commit) instead of
+/// committing one transaction per event — mirroring the reference shape in
+/// ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs
+/// (ProcessWriteQueueAsync/FlushBatch).
+///
+public class SiteEventLoggerBatchingTests : IDisposable
+{
+ private readonly SiteEventLogger _logger;
+ private readonly string _dbPath;
+ private readonly TestLocalDb _localDb;
+
+ public SiteEventLoggerBatchingTests()
+ {
+ _dbPath = Path.Combine(Path.GetTempPath(), $"test_batch_{Guid.NewGuid()}.db");
+ var options = Options.Create(new SiteEventLogOptions { DatabasePath = _dbPath });
+ _localDb = TestLocalDb.Create(_dbPath);
+ _logger = new SiteEventLogger(options, NullLogger.Instance, _localDb.Db);
+ }
+
+ public void Dispose()
+ {
+ _logger.Dispose();
+ _localDb.Dispose();
+ TestLocalDb.DeleteFiles(_dbPath);
+ }
+
+ [Fact]
+ public async Task LogEventAsync_SingleEvent_CommitsOneBatch()
+ {
+ await _logger.LogEventAsync("script", "Info", null, "Source", "one event");
+
+ Assert.Equal(1, _logger.CommittedBatchCount);
+ }
+
+ [Fact]
+ public async Task ProcessWriteQueue_BurstOfQueuedEvents_CommitsFarFewerBatchesThanEvents()
+ {
+ // Deterministically force a large burst of events to be sitting in the
+ // channel BEFORE the writer loop is free to drain them, so the drain is
+ // not at the mercy of scheduler timing (the same "hold the connection
+ // busy" technique SiteEventLoggerAsyncTests uses to prove the caller
+ // never blocks). While the write lock is held, the writer loop's first
+ // FlushBatch call blocks trying to acquire it — every event enqueued
+ // during that window queues up untouched. Once released, the writer
+ // loop's next iterations greedily TryRead everything already queued
+ // into batches capped at WriteBatchSize (256), so a burst well above
+ // that cap must still land in only a handful of transactions.
+ var busyStarted = new ManualResetEventSlim(false);
+ var releaseBusy = new ManualResetEventSlim(false);
+
+ var busyThread = new Thread(() =>
+ {
+ _logger.WithConnection(_ =>
+ {
+ busyStarted.Set();
+ releaseBusy.Wait(TimeSpan.FromSeconds(10));
+ });
+ });
+ busyThread.Start();
+ Assert.True(busyStarted.Wait(TimeSpan.FromSeconds(5)), "Busy thread did not start.");
+
+ // > WriteBatchSize (256) so a correctly batching writer must still take
+ // at least ceil(burstCount / WriteBatchSize) = 2 post-release batches,
+ // plus at most one partial batch already blocked on the lock — nowhere
+ // near burstCount (300) transactions.
+ const int burstCount = 300;
+ var tasks = new List(burstCount);
+ for (var i = 0; i < burstCount; i++)
+ {
+ tasks.Add(_logger.LogEventAsync("script", "Info", null, "Burst", $"event {i}"));
+ }
+
+ releaseBusy.Set();
+ busyThread.Join(TimeSpan.FromSeconds(10));
+
+ await Task.WhenAll(tasks).WaitAsync(TimeSpan.FromSeconds(30));
+
+ var batchesUsed = _logger.CommittedBatchCount;
+ var expectedMax = (long)Math.Ceiling(burstCount / (double)SiteEventLogger.WriteBatchSize) + 1;
+
+ Assert.True(batchesUsed <= expectedMax,
+ $"Expected at most {expectedMax} batch transactions for {burstCount} queued events " +
+ $"(WriteBatchSize={SiteEventLogger.WriteBatchSize}), but {batchesUsed} were committed — " +
+ "the writer loop is not batching the drain into few transactions.");
+ Assert.True(batchesUsed < burstCount,
+ $"{batchesUsed} batches were committed for {burstCount} events — that is one-transaction-per-event, not batched.");
+
+ var count = _logger.WithConnection(connection =>
+ {
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = "SELECT COUNT(*) FROM site_events";
+ return (long)cmd.ExecuteScalar()!;
+ });
+ Assert.Equal(burstCount, count);
+ }
+
+ [Fact]
+ public async Task ProcessWriteQueue_BatchFailure_FaultsEveryQueuedEvent_NoneSilentlyDropped()
+ {
+ // Batching introduces a risk absent from the old one-transaction-per-event
+ // shape: a failure could silently drop or under-report events that were
+ // swept into the same failed batch as the one that actually errored,
+ // instead of faulting every affected Task and counting every affected
+ // event in FailedWriteCount. Force a burst to queue up while the writer
+ // is blocked on the shared lock, then make every subsequent write fail
+ // (drop the table) before releasing — regardless of how the burst splits
+ // across batch(es) once released, every queued event's Task must still
+ // fault and FailedWriteCount must still account for all of them.
+ var busyStarted = new ManualResetEventSlim(false);
+ var releaseBusy = new ManualResetEventSlim(false);
+
+ var busyThread = new Thread(() =>
+ {
+ _logger.WithConnection(connection =>
+ {
+ busyStarted.Set();
+ releaseBusy.Wait(TimeSpan.FromSeconds(10));
+
+ // Drop the table while still holding the lock, immediately before
+ // releasing it, so the burst below (already fully queued during
+ // the busy window) hits a missing table as ONE batch.
+ using var cmd = connection.CreateCommand();
+ cmd.CommandText = "DROP TABLE site_events";
+ cmd.ExecuteNonQuery();
+ });
+ });
+ busyThread.Start();
+ Assert.True(busyStarted.Wait(TimeSpan.FromSeconds(5)), "Busy thread did not start.");
+
+ const int burstCount = 10;
+ var tasks = new List(burstCount);
+ for (var i = 0; i < burstCount; i++)
+ {
+ tasks.Add(_logger.LogEventAsync("script", "Info", null, "Burst", $"event {i}"));
+ }
+
+ releaseBusy.Set();
+ busyThread.Join(TimeSpan.FromSeconds(10));
+
+ // Every event in the batch must fault — none silently "succeeds" while
+ // its siblings fail, and none is silently dropped.
+ foreach (var task in tasks)
+ {
+ await Assert.ThrowsAnyAsync(() => task).WaitAsync(TimeSpan.FromSeconds(10));
+ }
+
+ Assert.True(_logger.FailedWriteCount >= burstCount,
+ $"Expected FailedWriteCount to account for all {burstCount} events in the rolled-back batch, got {_logger.FailedWriteCount}.");
+ }
+}