Joseph Doherty
6c11362d9f
P4-2: SqliteAuditWriter.FlushBatch completed each pending row's Task (TrySetResult) inside the per-row insert loop, before transaction.Commit() ran. A later failure in the same batch (another row's insert throwing, or the commit itself failing) rolled the whole transaction back, but the earlier rows' Tasks had already completed successfully -- the caller and FallbackAuditWriter (whose ring-buffer fallback exists precisely for this outage) both saw false success while the rows were actually gone from the database; the outer catch's TrySetException became a silent no-op against an already-completed TaskCompletionSource. Every row's Completion is now deferred and only resolved (TrySetResult for the whole batch, or TrySetException for the whole batch) after transaction.Commit() succeeds or the transaction rolls back, mirroring SiteEventLogger.FlushBatch. P4-4: EventLogPurgeService gated the daily site_events purge on the active-node check, early-exiting on the standby, on the stale premise that "the standby receives no writes". DataConnectionActor runs on both site nodes and writes site_events regardless of active/standby status (register rows 89/90/97), so an unreplicated standby (the rig's site-b/site-c posture) grew its site_events file forever with no retention and no cap. The gate is removed; every node now purges its own file every tick. Safe on a replicated pair because the retention/cap deletes are CDC-captured and converge under LWW regardless of which node issues them; required on an unreplicated pair because each node's file is independent. The active-node check is retained only to annotate the purge tick's log line. Tests: new SqliteAuditWriterCommitFailureTests forces a genuine non-swallowed mid-batch INSERT failure via a real SQLite trigger + custom scalar function on the writer's own connection (reflection seam), proving (a) a row whose own insert succeeded still faults when a later row in the same batch fails, and (b) FallbackAuditWriter rings every faulted row and increments the failure counter for each -- both assertions fail against the pre-fix code (proof-of-failure verified by temporarily reverting the fix). EventLogPurgeServiceTests' standby case is inverted to assert the purge now runs (not skips) on a standby node.
845 lines
34 KiB
C#
845 lines
34 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);
|
|
}
|
|
|
|
// ── arch-review 2026-08-19 P4-4: purge runs on every node, not just the active one ──
|
|
// (supersedes the old SiteEventLogging-019 "active-node-only gate" — DataConnectionActor
|
|
// writes site_events on both site nodes, so a standby that skipped its own purge would
|
|
// grow its file unbounded on an unreplicated site.)
|
|
|
|
[Fact]
|
|
public void RunPurge_OnStandbyNode_StillRunsThePurge()
|
|
{
|
|
// P4-4 regression: before the fix, a standby (isActiveNode returning false) early-exited
|
|
// the purge tick entirely, leaving this well-past-retention row in place forever on an
|
|
// unreplicated standby. The gate is gone — the standby purges its own file exactly like
|
|
// the active node.
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
|
|
Assert.Equal(1, GetEventCount());
|
|
|
|
var purge = CreatePurgeService(isActiveNode: () => false);
|
|
purge.RunPurge();
|
|
|
|
Assert.Equal(0, GetEventCount());
|
|
}
|
|
|
|
[Fact]
|
|
public void RunPurge_OnActiveNode_RunsTheRetentionPurge()
|
|
{
|
|
// Pinned alongside the standby case above so a future regression that reintroduces
|
|
// asymmetric active/standby purge behaviour is caught from either side.
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
|
|
|
|
var purge = CreatePurgeService(isActiveNode: () => true);
|
|
purge.RunPurge();
|
|
|
|
Assert.Equal(1, GetEventCount());
|
|
}
|
|
|
|
[Fact]
|
|
public void RunPurge_WithNullCheck_StillRuns()
|
|
{
|
|
// When no active-node check is supplied (the default for non-clustered hosts and
|
|
// pre-existing tests), the service runs the purge — unaffected by P4-4 since the
|
|
// check was never gating anything for this caller either.
|
|
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_WhenBothPhasesDeleteRows_LogsTheBacklogNoteOnceForTheWholeTick()
|
|
{
|
|
// Register row 84. The note used to be emitted by each phase, so a tick in which the
|
|
// retention purge AND the cap trim both deleted rows produced two identical Information
|
|
// lines. It is now emitted once, from the tick's shared delete budget — which also makes
|
|
// the number it reports the figure an operator actually wants next to an oplog spike:
|
|
// the whole tick's contribution, not one phase's.
|
|
const int expiredRows = 50;
|
|
const int recentRows = 50;
|
|
for (int i = 0; i < expiredRows; i++)
|
|
{
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
|
|
}
|
|
for (int i = 0; i < recentRows; i++)
|
|
{
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow);
|
|
}
|
|
|
|
var capOptions = new SiteEventLogOptions
|
|
{
|
|
DatabasePath = _dbPath,
|
|
RetentionDays = 30,
|
|
MaxStorageMb = 0 // retention takes the expired rows; the cap trim takes the rest
|
|
};
|
|
|
|
var logger = new CapturingLogger();
|
|
var purge = CreatePurgeService(capOptions, isReplicationConfigured: () => true, logger: logger);
|
|
purge.RunPurge();
|
|
|
|
// Both phases really did delete — otherwise this would pass for the wrong reason.
|
|
Assert.Contains(logger.Entries, e =>
|
|
e.Message.Contains($"Purged {expiredRows} events older than", StringComparison.Ordinal));
|
|
Assert.Contains(logger.Entries, e => e.Message.Contains("over the 0 MB cap", StringComparison.Ordinal));
|
|
Assert.Equal(0, GetEventCount());
|
|
|
|
var note = Assert.Single(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
|
|
Assert.Equal(LogLevel.Information, note.Level);
|
|
Assert.Contains($"Purged {expiredRows + recentRows} site_events rows", note.Message, StringComparison.Ordinal);
|
|
}
|
|
|
|
[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<long></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)));
|
|
}
|
|
}
|
|
}
|