Joseph Doherty
bbc1b55cc4
The T4 test asserted 'reflection Invoke returned AND the row is still present', but the yielded continuation runs RunPurge on a pool thread and can delete the row before the test thread reaches its assert — lost 2/2 full-suite sweeps on the integration branch, green in isolation (row-40 class, introduced by this wave). The purge now blocks inside the isActiveNode check (consulted before any delete) until the test releases it: fixed code returns from Invoke while the gate is held; unfixed code blocks on the gate itself and the bounded wait fails. Proof of failure re-run: with Task.Yield() removed the test fails with the intended message after the 10 s bound; 3/3 green with it restored.
949 lines
40 KiB
C#
949 lines
40 KiB
C#
using System.Reflection;
|
|
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 ExecuteAsync_YieldsBeforeRunningTheInitialPurgeSynchronously()
|
|
{
|
|
// arch-review 2026-08-19 P4-5 — timing-independent companion to
|
|
// StartAsync_DoesNotBlock_OnTheInitialPurge above.
|
|
//
|
|
// That test times BackgroundService.StartAsync as a whole and passes even on the
|
|
// UNFIXED code, because .NET 8+ already invokes ExecuteAsync via Task.Run — the
|
|
// synchronous prelude runs on a pool thread, not the caller's, so StartAsync itself
|
|
// never blocks regardless of whether ExecuteAsync yields first. Verified empirically
|
|
// against this repo's actual net10.0 target (a BackgroundService whose ExecuteAsync
|
|
// does Thread.Sleep(1500) before its first await still returns from StartAsync in
|
|
// ~2 ms). So P4-5's literal "StartAsync blocks" symptom does not reproduce here, and
|
|
// a wall-clock test cannot prove the fix.
|
|
//
|
|
// What CAN be proven without any timing is the ExecuteAsync-level contract itself:
|
|
// invoking the protected ExecuteAsync directly via reflection — bypassing
|
|
// BackgroundService's Task.Run wrapper entirely — still runs it synchronously, on
|
|
// the calling thread, up through its first genuinely-suspending await (that is what
|
|
// "async Task method" means; MethodInfo.Invoke does not change it). On the unfixed
|
|
// code ExecuteAsync's first statement is RunPurge() with no await before it, so
|
|
// Invoke would not return until the initial purge had fully run — deleting the
|
|
// expired row before this method's Assert ever executes. With Task.Yield() as the
|
|
// very first statement, Invoke returns as soon as MoveNext() hits the (always
|
|
// IsCompleted == false) yield point, before RunPurge has been called at all.
|
|
//
|
|
// Made deterministic at integration (2026-08-20): "Invoke returned, row still present"
|
|
// on its own is a RACE — the yielded continuation runs RunPurge on a pool thread and
|
|
// can delete the row before this thread reaches its assert (lost 2/2 full-suite
|
|
// sweeps, green in isolation). So the purge is parked on a gate instead: RunPurge
|
|
// consults the isActiveNode check BEFORE any delete, and the check blocks until the
|
|
// test releases it. Fixed code: Invoke returns while the gate is held (the continuation
|
|
// is the one waiting). Unfixed code: Invoke itself blocks on the gate, so the
|
|
// bounded wait below fails — no timing assumption in either direction.
|
|
InsertEventWithTimestamp(DateTimeOffset.UtcNow.AddDays(-31));
|
|
Assert.Equal(1, GetEventCount());
|
|
|
|
using var gate = new ManualResetEventSlim(initialState: false);
|
|
var purge = CreatePurgeService(isActiveNode: () => { gate.Wait(); return true; });
|
|
var executeAsync = typeof(EventLogPurgeService).GetMethod(
|
|
"ExecuteAsync", BindingFlags.NonPublic | BindingFlags.Instance);
|
|
Assert.NotNull(executeAsync);
|
|
|
|
using var cts = new CancellationTokenSource();
|
|
Task? executeTask = null;
|
|
var invoke = Task.Factory.StartNew(
|
|
() => executeTask = (Task)executeAsync!.Invoke(purge, [cts.Token])!,
|
|
TaskCreationOptions.LongRunning);
|
|
|
|
// With the gate held, Invoke can only return if ExecuteAsync yielded before RunPurge.
|
|
var returned = await Task.WhenAny(invoke, Task.Delay(TimeSpan.FromSeconds(10))) == invoke;
|
|
Assert.True(returned,
|
|
"ExecuteAsync did not return control before running the initial purge — " +
|
|
"it is running RunPurge synchronously ahead of its first await.");
|
|
Assert.Equal(1, GetEventCount());
|
|
|
|
// Release the purge; it still runs, just after control was yielded back.
|
|
gate.Set();
|
|
var deadline = DateTime.UtcNow.AddSeconds(5);
|
|
while (GetEventCount() != 0 && DateTime.UtcNow < deadline)
|
|
{
|
|
await Task.Delay(25);
|
|
}
|
|
Assert.Equal(0, GetEventCount());
|
|
|
|
await cts.CancelAsync();
|
|
try
|
|
{
|
|
await executeTask!;
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
// Expected: the loop's Task.Delay observes the cancellation.
|
|
}
|
|
}
|
|
|
|
[Fact]
|
|
public void RunPurge_ActiveNodeCheckThrows_TickLogsAsInactive_NotActive()
|
|
{
|
|
// arch-review 2026-08-19 P4-5: a throwing isActive predicate must fail CLOSED
|
|
// (logged as inactive) rather than silently claiming "active". The predicate no
|
|
// longer gates any purge work (P4-4 — every node purges its own file every tick
|
|
// regardless), so a wrong value here cannot skip work; but an operator correlating
|
|
// this line against the real active-node state must not be misled into thinking the
|
|
// check actually succeeded and reported active.
|
|
var logger = new CapturingLogger();
|
|
var purge = CreatePurgeService(
|
|
isActiveNode: () => throw new InvalidOperationException("boom"),
|
|
logger: logger);
|
|
|
|
purge.RunPurge();
|
|
|
|
var warnLine = Assert.Single(
|
|
logger.Entries,
|
|
e => e.Message.Contains("Active-node check threw", StringComparison.Ordinal));
|
|
Assert.Equal(LogLevel.Warning, warnLine.Level);
|
|
|
|
var tickLine = Assert.Single(
|
|
logger.Entries,
|
|
e => e.Message.Contains("Running event log purge tick", StringComparison.Ordinal));
|
|
Assert.Contains("active=False", tickLine.Message, StringComparison.Ordinal);
|
|
}
|
|
|
|
[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)));
|
|
}
|
|
}
|
|
}
|