Files
ScadaBridge/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs
T
Joseph Doherty d9a35846d9 perf(siteeventlog): pace the site_events purge burst — per-tick delete ceiling + accelerated catch-up (row 34)
Measured the unbounded purge burst and found it can force a full ten-table
snapshot resync, then bounded it. No purge-exemption path was added — the
deletes are still CDC-captured like any other write, which stays ruled out
by design; what changed is how many a single tick performs.

MEASURED (harness: real EventLogPurgeService over a real LocalDb file with
site_events genuinely RegisterReplicated, plus a two-node loopback gRPC
replication pair; docker rig untouched):

  393 bytes/row incl. indexes => a 1 GB event log holds ~2,731,947 rows.
  Expiring all of them in ONE tick:  285 s, 2,731,947 oplog rows,
  284 MB oplog payload, 2.73 M row-version tombstones (7-day retention),
  i.e. 2.7x LocalDb's MaxOplogRows default of 1,000,000. Crossing that makes
  the 60 s maintenance tick prune the oplog and set needs_snapshot — a FULL
  snapshot of all ten replicated tables, bought by a routine retention purge.

  The resident backlog is produced-minus-drained, so a healthy peer blunts it:
  on loopback the same burst peaked at 391,424 and cleared in 6.0 min. That is
  the problem, not the reassurance — the old safety margin was a function of
  peer throughput and vanished with the peer down. A bounded tick is safe at
  any peer speed.

  Drain is not a bottleneck: 7.6k-19.8k tombstones/s (20k / 500k / 2.73M runs).

SHIPPED:
  - SiteEventLogOptions.MaxPurgedRowsPerTick (default 250,000; 0 = unlimited
    escape hatch, negatives rejected). One SHARED budget across the retention
    purge and the cap trim — two independently-capped phases would let a tick
    emit twice the intended burst. 4x margin under MaxOplogRows, which is also
    shared with every other replicated table's ordinary traffic.
  - SiteEventLogOptions.PurgeCatchUpInterval (default 5 min, validated > 0 and
    <= PurgeInterval). Without it pacing would make retention converge in 11
    DAYS at the daily cadence; with it the worst case converges in ~11 ticks
    (~55 min), every burst bounded. 5 min rather than 1 so ticks cannot outrun
    the peer's drain and re-accumulate the backlog.
  - A ceiling is claimed only when work genuinely remains (bounded EXISTS probe),
    so an expired set landing exactly on the budget does not arm a no-op retry.
  - Visibility: rate-limited (15 min) Warning naming the ceiling and the
    accelerated interval, plus an unthrottled counter
    scadabridge.site.event_log.purge_ceiling_hits — its SHAPE distinguishes a
    backlog converging from a site whose event rate outruns one paced tick.
  - The purge loop is a delay loop rather than a PeriodicTimer, so its cadence
    now anchors to the END of the previous purge. Documented.

ROW-33 INTERPLAY VERIFIED: pacing lowers each spike and lengthens total purge
duration, but a full 250k tick drains in ~23 s (~4 min even at 10x a real
link's penalty) against the 15-minute sustained dwell and a 5-minute gap to
the next tick. The UNPACED 2.73M burst had only 2.5x of margin (6.0 min), so
pacing makes the badge assumption robust rather than merely preserving it.

Docs travel together: Component-SiteEventLogging.md, topology-guide.md
("Reading the replication backlog" + the row-33 badge note), CLAUDE.md's
purge-burst sentence, and a dated measurement/decision note under docs/plans/.

Fail-before: the new behavioural tests do not compile against the pre-fix
service (RunPurge was void); with PurgeBudget neutralised to always-unlimited,
6 of the new tests fail and all 19 pre-existing ones pass.

Tests: SiteEventLogging 93/93, Commons 715/715, HealthMonitoring 115/115,
Host 505/505. Solution build clean.
2026-08-16 04:05:21 -04:00

805 lines
32 KiB
C#

using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests;
public class EventLogPurgeServiceTests : IDisposable
{
private readonly SiteEventLogger _eventLogger;
private readonly TestLocalDb _localDb;
private readonly string _dbPath;
private readonly SiteEventLogOptions _options;
/// <summary>
/// SiteEventLogging-023: stop flag for the concurrent stress test. Declared as
/// a <c>volatile</c> field so every writer thread observes the main thread's
/// `_stop = true` write without depending on JIT/runtime quirks. A plain
/// <c>bool</c> local would be legal-cached in a register inside the tight
/// <c>while (!_stop)</c> loop under release-mode optimisation.
/// </summary>
private volatile bool _stop;
public EventLogPurgeServiceTests()
{
_dbPath = Path.Combine(Path.GetTempPath(), $"test_purge_{Guid.NewGuid()}.db");
_options = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = 1024
};
_localDb = TestLocalDb.Create(_dbPath);
_eventLogger = new SiteEventLogger(
Options.Create(_options),
NullLogger<SiteEventLogger>.Instance,
_localDb.Db);
}
public void Dispose()
{
_eventLogger.Dispose();
_localDb.Dispose();
TestLocalDb.DeleteFiles(_dbPath);
GC.SuppressFinalize(this);
}
private EventLogPurgeService CreatePurgeService(
SiteEventLogOptions? optionsOverride = null,
SiteEventLogActiveNodeCheck? isActiveNode = null,
SiteEventLogReplicationCheck? isReplicationConfigured = null,
ILogger<EventLogPurgeService>? logger = null)
{
var opts = optionsOverride ?? _options;
return new EventLogPurgeService(
_eventLogger,
Options.Create(opts),
logger ?? NullLogger<EventLogPurgeService>.Instance,
isActiveNode,
isReplicationConfigured);
}
private void InsertEventWithTimestamp(DateTimeOffset timestamp)
{
_eventLogger.WithConnection(connection =>
{
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", timestamp.ToString("o"));
cmd.ExecuteNonQuery();
});
}
private long GetEventCount()
{
return _eventLogger.WithConnection(connection =>
{
using var cmd = connection.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM site_events";
return (long)cmd.ExecuteScalar()!;
});
}
[Fact]
public void PurgeByRetention_DeletesOldEvents()
{
// Insert an old event (31 days ago) and a recent one
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService();
purge.RunPurge();
Assert.Equal(1, GetEventCount());
}
[Fact]
public void PurgeByRetention_KeepsRecentEvents()
{
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-29));
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-1));
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService();
purge.RunPurge();
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()
{
// Insert enough events to have some data
for (int i = 0; i < 100; i++)
{
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
}
// Set an artificially small cap to trigger purge
var smallCapOptions = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = 0 // 0 MB cap forces purge
};
var purge = CreatePurgeService(smallCapOptions);
purge.RunPurge();
// All events should be purged since cap is 0
Assert.Equal(0, GetEventCount());
}
[Fact]
public void GetDatabaseSizeBytes_ReturnsPositiveValue()
{
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService();
var size = purge.GetDatabaseSizeBytes();
Assert.True(size > 0);
}
/// <summary>
/// Base instant for bulk-seeded events: one day ago, evaluated once so a run is
/// self-consistent.
/// <para>
/// Must stay INSIDE the retention window (<c>RetentionDays</c>, default 30).
/// <see cref="EventLogPurgeService.RunPurge"/> applies the retention purge before
/// the storage-cap purge, so a fixed calendar date in the past would be deleted
/// wholesale as stale and the storage-cap assertions would silently test an empty
/// table rather than cap-trimming behaviour.
/// </para>
/// </summary>
private static readonly DateTimeOffset BulkSeedStart =
DateTimeOffset.UtcNow.AddDays(-1);
/// <summary>
/// Timestamp of the i-th bulk-seeded event. Each row is one second newer than the
/// last, so "oldest" is unambiguous — the storage-cap purge orders by timestamp,
/// and previously every bulk row shared the same <c>UtcNow</c>, which left the
/// oldest-first guarantee untestable.
/// </summary>
private static DateTimeOffset BulkSeedTimestamp(int index) =>
BulkSeedStart.AddSeconds(index);
private void InsertBulkEvents(int count)
{
// Each event carries a sizeable details payload so the database grows
// measurably and the storage cap can be exercised against a realistic file.
var details = new string('x', 2000);
_eventLogger.WithConnection(connection =>
{
for (int i = 0; i < count; i++)
{
using var cmd = connection.CreateCommand();
cmd.CommandText = """
INSERT INTO site_events (id, timestamp, event_type, severity, source, message, details)
VALUES ($id, $ts, 'script', 'Info', 'Test', 'Bulk event', $details)
""";
cmd.Parameters.AddWithValue("$id", Guid.NewGuid().ToString("N"));
cmd.Parameters.AddWithValue("$ts", BulkSeedTimestamp(i).ToString("o"));
cmd.Parameters.AddWithValue("$details", details);
cmd.ExecuteNonQuery();
}
});
}
/// <summary>
/// Oldest surviving event timestamp. Replaces the former <c>MIN(id)</c> probe:
/// ids are GUIDs since LocalDb Phase 1 and carry no ordering, so age is measured
/// on the timestamp column the purge itself orders by.
/// </summary>
private string MinEventTimestamp()
{
return _eventLogger.WithConnection(connection =>
{
using var cmd = connection.CreateCommand();
cmd.CommandText = "SELECT MIN(timestamp) FROM site_events";
var result = cmd.ExecuteScalar();
return result as string ?? "";
});
}
[Fact]
public void PurgeByStorageCap_StopsWhenUnderCap_DoesNotEmptyTable()
{
// Regression test for SiteEventLogging-001 / -002:
// a realistic non-zero cap must trim the oldest events to the budget,
// not delete the entire table.
InsertBulkEvents(3000);
var purge = CreatePurgeService();
var totalSize = purge.GetDatabaseSizeBytes();
// Cap at roughly half the current database size — purge must keep some rows.
var capBytes = totalSize / 2;
var capOptions = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = (int)Math.Max(1, capBytes / (1024 * 1024))
};
var cappedPurge = CreatePurgeService(capOptions);
cappedPurge.RunPurge();
var remaining = GetEventCount();
Assert.True(remaining > 0, "Storage-cap purge must not delete the entire table.");
Assert.True(remaining < 3000, "Storage-cap purge must remove some events when over cap.");
// The database must actually be back under the cap after purge.
var finalSize = cappedPurge.GetDatabaseSizeBytes();
var finalCapBytes = (long)capOptions.MaxStorageMb * 1024 * 1024;
Assert.True(finalSize <= finalCapBytes,
$"Database size {finalSize} must be at or below cap {finalCapBytes} after purge.");
}
[Fact]
public void PurgeByStorageCap_RemovesOldestEventsFirst()
{
// Regression test for SiteEventLogging-002: only the oldest events should be
// removed when trimming to the cap.
//
// "Oldest" used to mean "lowest autoincrement id". Since LocalDb Phase 1 the id
// is a GUID with no ordering, so both the purge and this test key on timestamp.
// Ordering by id here would delete an arbitrary batch, which is precisely the
// regression this test now guards.
const int eventCount = 3000;
InsertBulkEvents(eventCount);
var purge = CreatePurgeService();
var totalSize = purge.GetDatabaseSizeBytes();
var capOptions = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = (int)Math.Max(1, (totalSize / 2) / (1024 * 1024))
};
var oldestBefore = MinEventTimestamp();
var cappedPurge = CreatePurgeService(capOptions);
cappedPurge.RunPurge();
var oldestAfter = MinEventTimestamp();
// The surviving rows must be the newest ones — the oldest timestamp advanced.
Assert.True(
string.CompareOrdinal(oldestAfter, oldestBefore) > 0,
$"Oldest events must be purged first; oldest went from '{oldestBefore}' to '{oldestAfter}'.");
// The newest event must still be present.
var newestTimestamp = BulkSeedTimestamp(eventCount - 1).ToString("o");
var newestPresent = _eventLogger.WithConnection(connection =>
{
using var cmd = connection.CreateCommand();
cmd.CommandText = "SELECT COUNT(*) FROM site_events WHERE timestamp = $ts";
cmd.Parameters.AddWithValue("$ts", newestTimestamp);
return (long)cmd.ExecuteScalar()!;
});
Assert.Equal(1L, newestPresent);
}
[Fact]
public async Task StartAsync_DoesNotBlock_OnTheInitialPurge()
{
// SiteEventLogging-014 (re-triaged): on .NET 8+ BackgroundService runs
// ExecuteAsync on a thread-pool thread — the synchronous prelude (the
// initial RunPurge()) does NOT execute on the host startup thread, so
// StartAsync returns promptly and host startup / the /health/ready gate is
// not blocked even by a large initial purge. This test pins that behaviour:
// StartAsync returns fast, and the initial purge still happens shortly
// afterwards on the background scheduler.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
Assert.Equal(1, GetEventCount());
var purge = CreatePurgeService();
using var cts = new CancellationTokenSource();
var sw = System.Diagnostics.Stopwatch.StartNew();
await purge.StartAsync(cts.Token);
sw.Stop();
Assert.True(sw.ElapsedMilliseconds < 1000,
$"StartAsync blocked for {sw.ElapsedMilliseconds} ms — the initial purge " +
"must not run on the host startup thread.");
// The initial purge still runs on the background scheduler.
var deadline = DateTime.UtcNow.AddSeconds(5);
while (GetEventCount() != 0 && DateTime.UtcNow < deadline)
{
await Task.Delay(25);
}
Assert.Equal(0, GetEventCount());
await cts.CancelAsync();
await purge.StopAsync(CancellationToken.None);
}
[Fact]
public async Task PurgeByStorageCap_ConcurrentWritesDoNotCorruptConnection()
{
// Regression test for SiteEventLogging-003: purge running on a background
// thread while events are recorded on other threads must not throw
// "DataReader already open" / "connection busy" from a shared connection.
InsertBulkEvents(2000);
var purge = CreatePurgeService();
var totalSize = purge.GetDatabaseSizeBytes();
var capOptions = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = (int)Math.Max(1, (totalSize / 2) / (1024 * 1024))
};
var exceptions = new System.Collections.Concurrent.ConcurrentBag<Exception>();
// SiteEventLogging-023: must be volatile so the writer threads observe the
// main thread's `stop = true` flip after the purge task completes. Without
// it, a release-mode JIT is permitted to cache the `stop = false` read in
// a register inside the tight `while (!stop)` loop and never see the flip,
// causing the writer tasks to hang past xUnit's per-test timeout instead
// of asserting `Empty(exceptions)`.
_stop = false;
var purgeTask = Task.Run(() =>
{
try
{
var p = CreatePurgeService(capOptions);
for (int i = 0; i < 20; i++) p.RunPurge();
}
catch (Exception ex) { exceptions.Add(ex); }
});
var writeTasks = Enumerable.Range(0, 4).Select(_ => Task.Run(async () =>
{
try
{
while (!_stop)
{
await _eventLogger.LogEventAsync("script", "Info", null, "Concurrent", "Concurrent write");
}
}
catch (Exception ex) { exceptions.Add(ex); }
})).ToArray();
await purgeTask;
_stop = true;
await Task.WhenAll(writeTasks);
Assert.Empty(exceptions);
}
// ── SiteEventLogging-019: purge runs only on the active node ──
[Fact]
public void RunPurge_OnStandbyNode_SkipsAllWork()
{
// SiteEventLogging-019: per design, the daily purge runs on the active
// node only. The standby's local SQLite receives no writes, so purging
// there is unnecessary; we gate the purge tick on the injected
// active-node check and early-exit when it returns false. The row
// inserted here is well past retention, so a real purge would delete
// it — the standby gate must leave it intact.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
Assert.Equal(1, GetEventCount());
var purge = CreatePurgeService(isActiveNode: () => false);
purge.RunPurge();
Assert.Equal(1, GetEventCount());
}
[Fact]
public void RunPurge_OnActiveNode_RunsTheRetentionPurge()
{
// SiteEventLogging-019: when the active-node check returns true the
// service runs the purge as before. Pinned alongside the standby case
// so a future regression that inverts the gate is caught.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService(isActiveNode: () => true);
purge.RunPurge();
Assert.Equal(1, GetEventCount());
}
[Fact]
public void RunPurge_WithNullCheck_FallsBackToRunning()
{
// SiteEventLogging-019: when no active-node check is supplied (the
// default for non-clustered hosts and pre-existing tests), the service
// preserves the pre-fix "run on every tick" behaviour rather than
// silently skipping every tick. Backward compatibility guard.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
var purge = CreatePurgeService(isActiveNode: null);
purge.RunPurge();
Assert.Equal(0, GetEventCount());
}
// ── R7: replication-backlog operator note on the purge log line ──
/// <summary>
/// Fragment unique to the R7 note, matched against the rendered log message.
/// Deliberately not the whole sentence — the wording is operator prose and may be
/// reworded; what must not regress is that the note fires (or does not) per the
/// replication predicate.
/// </summary>
private const string BacklogNoteFragment = "transient LocalDb oplog backlog";
[Fact]
public void PurgeByRetention_OnReplicatedNode_LogsTheOplogBacklogNote()
{
// R7: site_events is a replicated table and the retention DELETE is CDC-captured
// like any other write — no purge-exemption path by design — so the purge inflates
// LocalDbOplogBacklog / localdb_oplog_depth until the peer acks. The Information
// line is the breadcrumb that lets an operator tie the spike to the purge instead
// of reading it as a replication fault.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
var logger = new CapturingLogger();
var purge = CreatePurgeService(isReplicationConfigured: () => true, logger: logger);
purge.RunPurge();
Assert.Equal(0, GetEventCount());
var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
Assert.Equal(LogLevel.Information, note.Level);
}
[Fact]
public void PurgeByRetention_OnUnreplicatedNode_DoesNotLogTheNote()
{
// The other half of the gate: site-b/site-c on the rig run with no replication
// configured, have no CDC triggers at all, and therefore no backlog to explain.
// Emitting the note there would be noise pointing at a metric that reads null.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
var logger = new CapturingLogger();
var purge = CreatePurgeService(isReplicationConfigured: () => false, logger: logger);
purge.RunPurge();
Assert.Equal(0, GetEventCount());
Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
}
[Fact]
public void PurgeByRetention_WithNoRowsDeleted_DoesNotLogTheNote()
{
// The note is a companion to a purge line, not a tick heartbeat: a daily tick that
// deleted nothing produces no oplog rows and so must stay silent, otherwise the
// breadcrumb loses all correlating value.
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var logger = new CapturingLogger();
var purge = CreatePurgeService(isReplicationConfigured: () => true, logger: logger);
purge.RunPurge();
Assert.Equal(1, GetEventCount());
Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
}
[Fact]
public void PurgeByStorageCap_OnReplicatedNode_LogsTheOplogBacklogNote()
{
// A cap-driven trim deletes rows through the same replicated table and is usually
// the larger of the two bursts, so it carries the same note.
for (int i = 0; i < 100; i++)
{
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
}
var capOptions = new SiteEventLogOptions
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = 0 // 0 MB cap forces the cap purge; retention deletes nothing here
};
var logger = new CapturingLogger();
var purge = CreatePurgeService(capOptions, isReplicationConfigured: () => true, logger: logger);
purge.RunPurge();
Assert.Equal(0, GetEventCount());
var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
Assert.Equal(LogLevel.Information, note.Level);
}
[Fact]
public void RunPurge_WhenReplicationCheckThrows_StillPurgesAndSwallows()
{
// Defensive: the replication predicate only selects log wording. A throw from it
// must never escape the purge — the rows are already deleted by the time it is
// consulted, and an exception here would surface as "Error during event log purge"
// for a purge that in fact succeeded.
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
var logger = new CapturingLogger();
var purge = CreatePurgeService(
isReplicationConfigured: () => throw new InvalidOperationException("boom"),
logger: logger);
purge.RunPurge();
Assert.Equal(0, GetEventCount());
Assert.DoesNotContain(logger.Entries, e => e.Level == LogLevel.Error);
Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
}
// ── Register row 34: per-tick delete ceiling ──
/// <summary>
/// Fragment unique to the ceiling-reached line. As with the backlog note, the operator prose
/// may be reworded; what must not regress is that the line fires when a tick is cut short.
/// </summary>
private const string CeilingLineFragment = "per-tick ceiling";
/// <summary>The counter incremented once per ceiling-limited tick.</summary>
private const string CeilingHitCounter = "scadabridge.site.event_log.purge_ceiling_hits";
/// <summary>Seeds <paramref name="count"/> events one second apart, all past retention.</summary>
private void InsertExpiredEvents(int count)
{
var expiredBase = DateTimeOffset.UtcNow.AddDays(-31);
_eventLogger.WithConnection(connection =>
{
using var tx = connection.BeginTransaction();
using var cmd = connection.CreateCommand();
cmd.Transaction = tx;
cmd.CommandText = """
INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
""";
var id = cmd.Parameters.Add("$id", Microsoft.Data.Sqlite.SqliteType.Text);
var ts = cmd.Parameters.Add("$ts", Microsoft.Data.Sqlite.SqliteType.Text);
for (var i = 0; i < count; i++)
{
id.Value = Guid.NewGuid().ToString("N");
ts.Value = expiredBase.AddSeconds(i).ToString("o");
cmd.ExecuteNonQuery();
}
tx.Commit();
});
}
/// <summary>Options with the per-tick ceiling set and a short catch-up interval.</summary>
private SiteEventLogOptions CeilingOptions(int ceiling, int maxStorageMb = 1024) => new()
{
DatabasePath = _dbPath,
RetentionDays = 30,
MaxStorageMb = maxStorageMb,
MaxPurgedRowsPerTick = ceiling,
PurgeCatchUpInterval = TimeSpan.FromSeconds(1),
};
[Fact]
public void PurgeByRetention_StopsAtThePerTickCeiling()
{
// Row 34 core behaviour. site_events is CDC-captured on a replicated node with no
// purge-exemption path, so an unbounded tick queues one oplog row per deleted event —
// measured at 2,731,947 rows for a full 1 GB log, 2.7x over LocalDb's MaxOplogRows
// default, which costs a full ten-table snapshot resync. The ceiling bounds the burst.
InsertExpiredEvents(2500);
var purge = CreatePurgeService(CeilingOptions(1000));
var ceilingHit = purge.RunPurge();
Assert.True(ceilingHit, "A tick cut short by the ceiling must ask for an accelerated catch-up tick.");
Assert.Equal(1500, GetEventCount());
}
[Fact]
public void PurgeByRetention_WithCeiling_ConvergesAcrossSuccessiveTicks()
{
// The remainder must be picked up, or pacing would trade a burst for a retention policy
// that never completes. 2500 expired rows at 1000/tick = three ticks, the third of which
// finishes the work and reports no ceiling hit.
InsertExpiredEvents(2500);
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService(CeilingOptions(1000));
Assert.True(purge.RunPurge());
Assert.Equal(1501, GetEventCount());
Assert.True(purge.RunPurge());
Assert.Equal(501, GetEventCount());
Assert.False(purge.RunPurge(), "The final tick clears the backlog and must not ask for another.");
Assert.Equal(1, GetEventCount());
}
[Fact]
public void PurgeByRetention_ExpiredSetLandingExactlyOnTheCeiling_IsNotACeilingHit()
{
// Exhausting the budget is not the same as having work left. A tick that deleted the last
// expired row and happened to land on the ceiling must not arm a catch-up tick that will
// delete nothing and log that retention is behind.
InsertExpiredEvents(1000);
var purge = CreatePurgeService(CeilingOptions(1000));
Assert.False(purge.RunPurge());
Assert.Equal(0, GetEventCount());
}
[Fact]
public void MaxPurgedRowsPerTick_Zero_RestoresUnboundedPurges()
{
// The documented escape hatch: 0 means unlimited, i.e. byte-for-byte the pre-row-34
// behaviour, for an operator who would rather take one long burst than a paced one.
InsertExpiredEvents(2500);
var purge = CreatePurgeService(CeilingOptions(0));
Assert.False(purge.RunPurge());
Assert.Equal(0, GetEventCount());
}
[Fact]
public void RunPurge_HittingTheCeiling_LogsTheCeilingLineAndCountsIt()
{
// Visibility requirement: a ceiling-limited tick is silent progress otherwise — the log
// line names the ceiling and the accelerated retry, and the counter carries every hit so
// "converging" can be told from "never converging".
InsertExpiredEvents(2500);
using var probe = new CounterProbe(CeilingHitCounter);
var logger = new CapturingLogger();
var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
purge.RunPurge();
var line = Assert.Single(
logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
Assert.Equal(LogLevel.Warning, line.Level);
Assert.Equal(1, probe.Total);
}
[Fact]
public void RunPurge_RepeatedCeilingHits_RateLimitTheLogLineButNotTheCounter()
{
// A large backlog produces a RUN of ceiling-limited ticks, one every catch-up interval.
// Logging each would bury the rest of the log; the counter is the unthrottled signal.
InsertExpiredEvents(3000);
using var probe = new CounterProbe(CeilingHitCounter);
var logger = new CapturingLogger();
var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
purge.RunPurge();
purge.RunPurge();
Assert.Single(logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
Assert.Equal(2, probe.Total);
}
[Fact]
public void PurgeByStorageCap_SharesTheTickBudgetWithRetention()
{
// The ceiling bounds the TICK, not each phase: two independently-capped phases would let
// one tick emit twice the intended oplog burst. Here the retention purge consumes the
// whole budget, so the cap trim does not run at all — and must not log that it is
// purging oldest events while deleting nothing.
InsertExpiredEvents(1200);
for (var i = 0; i < 200; i++)
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var logger = new CapturingLogger();
// A 0 MB cap means the cap trim would empty whatever the retention purge left, if it ran.
var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0), logger: logger);
Assert.True(purge.RunPurge());
Assert.Equal(400, GetEventCount());
Assert.DoesNotContain(logger.Entries, e => e.Message.Contains("exceeds cap", StringComparison.Ordinal));
}
[Fact]
public void PurgeByStorageCap_WithCeiling_StopsAtTheBudgetAndAsksForCatchUp()
{
// The cap path pays the same CDC cost and is usually the larger burst of the two, so it
// draws from the same budget and reports the same ceiling hit when it runs out.
for (var i = 0; i < 2500; i++)
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0));
Assert.True(purge.RunPurge());
Assert.Equal(1500, GetEventCount());
}
/// <summary>
/// Accumulates measurements on one counter instrument for as long as it is alive.
/// </summary>
/// <remarks>
/// A <c>Counter&lt;long&gt;</c> is not observable, so a listener started after the fact sees
/// nothing — the probe must be constructed BEFORE the code under test runs. Tests in this
/// class run sequentially (xUnit parallelises across classes, not within one) and no other
/// class in the assembly touches this instrument, so the process-global meter is safe to
/// read absolutely here.
/// </remarks>
private sealed class CounterProbe : IDisposable
{
private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
private long _total;
public CounterProbe(string instrumentName)
{
_listener.InstrumentPublished = (instrument, listener) =>
{
if (instrument.Name == instrumentName)
listener.EnableMeasurementEvents(instrument);
};
_listener.SetMeasurementEventCallback<long>(
(_, measurement, _, _) => Interlocked.Add(ref _total, measurement));
_listener.Start();
}
/// <summary>Measurements recorded since construction.</summary>
public long Total => Interlocked.Read(ref _total);
public void Dispose() => _listener.Dispose();
}
/// <summary>
/// Minimal <see cref="ILogger{TCategoryName}"/> that records level + rendered message.
/// The purge service has no other observable output for a log-only change, and the
/// suite has no shared capturing logger to reuse.
/// </summary>
private sealed class CapturingLogger : ILogger<EventLogPurgeService>
{
public List<(LogLevel Level, string Message)> Entries { get; } = [];
public IDisposable? BeginScope<TState>(TState state) where TState : notnull => null;
public bool IsEnabled(LogLevel logLevel) => true;
public void Log<TState>(
LogLevel logLevel,
EventId eventId,
TState state,
Exception? exception,
Func<TState, Exception?, string> formatter)
{
Entries.Add((logLevel, formatter(state, exception)));
}
}
}