perf(sitelog): batched event-log commits and sliced retention purge
This commit is contained in:
@@ -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()
|
||||
{
|
||||
|
||||
@@ -0,0 +1,158 @@
|
||||
using Microsoft.Extensions.Logging.Abstractions;
|
||||
using Microsoft.Extensions.Options;
|
||||
|
||||
namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// WP1.7 (arch-review remediation, 2026-08-14): <see cref="SiteEventLogger"/>'s
|
||||
/// background writer loop must drain its bounded channel into batched transactions
|
||||
/// (up to <see cref="SiteEventLogger.WriteBatchSize"/> events per commit) instead of
|
||||
/// committing one transaction per event — mirroring the reference shape in
|
||||
/// <c>ZB.MOM.WW.ScadaBridge.AuditLog/Site/SqliteAuditWriter.cs</c>
|
||||
/// (<c>ProcessWriteQueueAsync</c>/<c>FlushBatch</c>).
|
||||
/// </summary>
|
||||
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<SiteEventLogger>.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<Task>(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<Task>(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<Exception>(() => 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}.");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user