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.
776 lines
40 KiB
C#
776 lines
40 KiB
C#
using System.Globalization;
|
||
using Microsoft.Data.Sqlite;
|
||
using Microsoft.Extensions.Hosting;
|
||
using Microsoft.Extensions.Logging;
|
||
using Microsoft.Extensions.Options;
|
||
using ZB.MOM.WW.ScadaBridge.Commons.Observability;
|
||
|
||
namespace ZB.MOM.WW.ScadaBridge.SiteEventLogging;
|
||
|
||
/// <summary>
|
||
/// Predicate the <see cref="EventLogPurgeService"/> consults at the top of every purge
|
||
/// tick. As of arch-review 2026-08-19 P4-4 it no longer GATES the purge — DataConnectionActor
|
||
/// writes <c>site_events</c> on both site nodes (register rows 89/90/97), so a standby that
|
||
/// skipped its own purge would grow its file unboundedly on an unreplicated site. Every node
|
||
/// now purges its own file every tick; this predicate is consulted only to annotate the tick's
|
||
/// log line with whether this node is the active one.
|
||
///
|
||
/// Registration is the Host's responsibility (it knows the cluster topology);
|
||
/// when no implementation is registered the purge service treats every tick as active for
|
||
/// logging purposes — backward compatible with hosts that never register this delegate.
|
||
/// </summary>
|
||
/// <returns><c>true</c> if this node is the active site member and should run the purge; <c>false</c> to skip.</returns>
|
||
public delegate bool SiteEventLogActiveNodeCheck();
|
||
|
||
/// <summary>
|
||
/// Predicate the <see cref="EventLogPurgeService"/> consults when a purge actually deleted
|
||
/// rows, to decide whether to add the "expect a transient replication backlog" operator note
|
||
/// to its purge log line.
|
||
///
|
||
/// <para>
|
||
/// <c>site_events</c> is one of LocalDb's replicated tables, and CDC captures a retention
|
||
/// DELETE exactly like any other row change — by design, since Phase 2 there is deliberately
|
||
/// no purge-exemption path. So on a replication-enabled node the purge briefly inflates the
|
||
/// oplog (health field <c>LocalDbOplogBacklog</c>, Prometheus <c>localdb_oplog_depth</c>),
|
||
/// which reads like a replication fault to an operator who does not know a purge just ran.
|
||
/// The log line is the breadcrumb that ties the two together; it is a message-wording choice
|
||
/// only and changes no purge behaviour.
|
||
/// </para>
|
||
///
|
||
/// <para>
|
||
/// Registration is the Host's responsibility — SiteEventLogging has no view of
|
||
/// <c>LocalDb:Replication</c> and must not grow one; the Host already owns that predicate
|
||
/// (<c>SiteLocalDbSetup.ReplicationIsConfigured</c>, the <c>PeerAddress</c>-OR-<c>ApiKey</c>
|
||
/// rule) and passes it in. When no implementation is registered the note is suppressed, which
|
||
/// matches the product default: replication is opt-in and off unless configured.
|
||
/// </para>
|
||
/// </summary>
|
||
/// <returns><c>true</c> if this node has LocalDb replication configured; <c>false</c> otherwise.</returns>
|
||
public delegate bool SiteEventLogReplicationCheck();
|
||
|
||
/// <summary>
|
||
/// Background service that periodically purges old events from the SQLite event log.
|
||
/// Enforces both time-based retention (default 30 days) and storage cap (default 1GB).
|
||
/// Runs on a background thread and does not block event recording.
|
||
/// </summary>
|
||
public class EventLogPurgeService : BackgroundService
|
||
{
|
||
/// <summary>Number of events deleted per cap-purge batch.</summary>
|
||
private const int CapPurgeBatchSize = 1000;
|
||
|
||
/// <summary>Number of events deleted per retention-purge batch — same size as <see cref="CapPurgeBatchSize"/>.</summary>
|
||
private const int RetentionPurgeBatchSize = 1000;
|
||
|
||
/// <summary>
|
||
/// Rows sampled from each end of <c>site_events</c> when sizing an average row. See
|
||
/// <see cref="EstimateSiteEventsSize"/>.
|
||
/// </summary>
|
||
private const int SizeSampleRows = 500;
|
||
|
||
/// <summary>
|
||
/// Per-row bookkeeping a table B-tree leaf spends on top of the row's own bytes: the
|
||
/// payload-length varint, the rowid varint and the 2-byte cell-pointer slot.
|
||
/// </summary>
|
||
private const int TableCellOverheadBytes = 16;
|
||
|
||
/// <summary>
|
||
/// Bytes at the end of a page SQLite will not fill with local payload — the
|
||
/// <c>pageSize - 35</c> maximum-local-payload rule for table leaves.
|
||
/// </summary>
|
||
private const int PageReservedBytes = 35;
|
||
|
||
/// <summary>
|
||
/// Average proportion of a table leaf page that actually holds rows. B-tree splits leave
|
||
/// pages part-empty; 0.78 is the measured mean across the workloads in
|
||
/// <c>docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md</c> §3.
|
||
/// </summary>
|
||
private const double TablePageFillFactor = 0.78;
|
||
|
||
/// <summary>Number of indexes on <c>site_events</c> (timestamp, event_type, instance_id, severity).</summary>
|
||
private const int IndexedColumnCount = 4;
|
||
|
||
/// <summary>Per-entry bookkeeping in an index B-tree leaf: key-length varint, rowid and cell pointer.</summary>
|
||
private const int IndexEntryOverheadBytes = 8;
|
||
|
||
/// <summary>Average fill of an index leaf page; measured alongside <see cref="TablePageFillFactor"/>.</summary>
|
||
private const double IndexPageFillFactor = 0.75;
|
||
|
||
/// <summary>Minimum spacing between ceiling-reached log lines; see <see cref="LogCeilingReached"/>.</summary>
|
||
private static readonly TimeSpan CeilingLogInterval = TimeSpan.FromMinutes(15);
|
||
|
||
/// <summary>
|
||
/// When the ceiling-reached line was last emitted, for the rate limit. Instance state and
|
||
/// only ever touched from the single purge loop, so no synchronisation is needed.
|
||
/// </summary>
|
||
private DateTimeOffset? _lastCeilingLogUtc;
|
||
|
||
private readonly SiteEventLogger _eventLogger;
|
||
private readonly SiteEventLogOptions _options;
|
||
private readonly ILogger<EventLogPurgeService> _logger;
|
||
private readonly SiteEventLogActiveNodeCheck _isActiveNode;
|
||
private readonly SiteEventLogReplicationCheck _isReplicationConfigured;
|
||
|
||
/// <summary>Initializes a new instance of <see cref="EventLogPurgeService"/>.</summary>
|
||
/// <param name="eventLogger">The concrete event logger providing lock-guarded database access.</param>
|
||
/// <param name="options">Site event log options (retention days, storage cap, purge interval).</param>
|
||
/// <param name="logger">Logger instance.</param>
|
||
/// <param name="isActiveNode">
|
||
/// Optional active-node check, consulted only to annotate the purge tick's log line
|
||
/// (arch-review 2026-08-19 P4-4 removed the early-exit this used to drive — every node
|
||
/// purges its own <c>site_events</c> file every tick, since DataConnectionActor writes
|
||
/// on both site nodes and a standby that skipped its own purge would grow unbounded on
|
||
/// an unreplicated site). When <c>null</c> every tick logs as active.
|
||
/// </param>
|
||
/// <param name="isReplicationConfigured">
|
||
/// Optional LocalDb-replication check. When <c>null</c> — non-clustered hosts,
|
||
/// unit tests — the purge log omits the replication-backlog operator note, matching
|
||
/// the product default that replication is opt-in and off unless configured.
|
||
/// See <see cref="SiteEventLogReplicationCheck"/>.
|
||
/// </param>
|
||
public EventLogPurgeService(
|
||
SiteEventLogger eventLogger,
|
||
IOptions<SiteEventLogOptions> options,
|
||
ILogger<EventLogPurgeService> logger,
|
||
SiteEventLogActiveNodeCheck? isActiveNode = null,
|
||
SiteEventLogReplicationCheck? isReplicationConfigured = null)
|
||
{
|
||
// Depend on the concrete recorder directly: purge must funnel database access
|
||
// through its lock-guarded WithConnection. Taking ISiteEventLogger and
|
||
// downcasting would throw InvalidCastException for any other implementation.
|
||
_eventLogger = eventLogger;
|
||
_options = options.Value;
|
||
_logger = logger;
|
||
_isActiveNode = isActiveNode ?? (static () => true);
|
||
_isReplicationConfigured = isReplicationConfigured ?? (static () => false);
|
||
}
|
||
|
||
/// <inheritdoc />
|
||
/// <remarks>
|
||
/// Deliberately a delay loop rather than a <see cref="PeriodicTimer"/>: the wait between
|
||
/// ticks is no longer a constant. A tick that stopped at
|
||
/// <see cref="SiteEventLogOptions.MaxPurgedRowsPerTick"/> still has expired rows to delete,
|
||
/// and waiting a full <see cref="SiteEventLogOptions.PurgeInterval"/> (a day) before
|
||
/// resuming would make a paced purge take longer to converge than the retention window it
|
||
/// enforces. Consequence worth knowing: the cadence is now anchored to the END of the
|
||
/// previous purge rather than to process start, so a long purge shifts the next tick by its
|
||
/// own duration. At the daily interval and a measured ~17 s for a full 250 000-row tick that
|
||
/// drift is immaterial.
|
||
/// </remarks>
|
||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||
{
|
||
_logger.LogInformation(
|
||
"Event log purge service started — retention: {Days} days, cap: {Cap} MB, interval: {Interval}, " +
|
||
"per-tick delete ceiling: {Ceiling}",
|
||
_options.RetentionDays, _options.MaxStorageMb, _options.PurgeInterval,
|
||
_options.MaxPurgedRowsPerTick > 0 ? _options.MaxPurgedRowsPerTick.ToString() : "unlimited");
|
||
|
||
// Run an initial purge on startup
|
||
var ceilingHit = RunPurge();
|
||
|
||
while (!stoppingToken.IsCancellationRequested)
|
||
{
|
||
var delay = ceilingHit ? _options.PurgeCatchUpInterval : _options.PurgeInterval;
|
||
try
|
||
{
|
||
await Task.Delay(delay, stoppingToken).ConfigureAwait(false);
|
||
}
|
||
catch (OperationCanceledException)
|
||
{
|
||
return;
|
||
}
|
||
|
||
ceilingHit = RunPurge();
|
||
}
|
||
}
|
||
|
||
/// <summary>Runs a single purge cycle: first by retention age, then by storage cap.</summary>
|
||
/// <returns>
|
||
/// <c>true</c> when the tick stopped at <see cref="SiteEventLogOptions.MaxPurgedRowsPerTick"/>
|
||
/// with rows still to purge, so the caller should schedule an accelerated catch-up tick.
|
||
/// </returns>
|
||
internal bool RunPurge()
|
||
{
|
||
try
|
||
{
|
||
// arch-review 2026-08-19 P4-4: purge no longer gates on the active-node check.
|
||
// DataConnectionActor writes site_events on BOTH site nodes (register rows
|
||
// 89/90/97), so the old "standby receives no writes" premise this gate was
|
||
// built on is stale — a standby that never purges its own file grows
|
||
// site_events forever (no retention, no cap) on an UNREPLICATED site (the rig's
|
||
// site-b/site-c posture). On a REPLICATED pair this was masked because the
|
||
// retention/cap DELETEs are CDC-captured and converge under LWW regardless of
|
||
// which node issues them (same reasoning as the existing purge-on-replicated
|
||
// design — see the replication-backlog note below and CLAUDE.md's LocalDb
|
||
// bullet); it is REQUIRED on an unreplicated pair because each node's file is
|
||
// independent. Both nodes now run retention + cap trim against their own file
|
||
// every tick. isActive is retained only to annotate the log line below (and
|
||
// defensively defaults to true if the check itself throws) — it no longer
|
||
// skips any work.
|
||
bool isActive;
|
||
try
|
||
{
|
||
isActive = _isActiveNode();
|
||
}
|
||
catch (Exception checkEx)
|
||
{
|
||
_logger.LogDebug(checkEx,
|
||
"Active-node check threw during purge tick; treating as active for logging purposes");
|
||
isActive = true;
|
||
}
|
||
|
||
_logger.LogDebug("Running event log purge tick (active={IsActive})", isActive);
|
||
|
||
var budget = new PurgeBudget(_options.MaxPurgedRowsPerTick);
|
||
PurgeByRetention(budget);
|
||
PurgeByStorageCap(budget);
|
||
|
||
// Register row 84: emitted ONCE per tick, from the tick's shared budget, rather
|
||
// than once per phase. A tick in which both the retention purge and the cap trim
|
||
// deleted rows used to log two identical Information notes; the number an operator
|
||
// wants next to the oplog spike is the tick total anyway, which is exactly what
|
||
// the shared budget already holds.
|
||
if (budget.Deleted > 0)
|
||
LogReplicationBacklogNote(budget.Deleted);
|
||
|
||
if (!budget.StoppedAtCeiling)
|
||
return false;
|
||
|
||
ScadaBridgeTelemetry.RecordSiteEventLogPurgeCeilingHit();
|
||
LogCeilingReached(budget.Deleted);
|
||
return true;
|
||
}
|
||
catch (Exception ex)
|
||
{
|
||
_logger.LogError(ex, "Error during event log purge");
|
||
// A failed tick must not claim a ceiling hit: the accelerated retry exists to finish
|
||
// KNOWN remaining work, and turning every fault into a 5-minute retry loop would
|
||
// hammer a broken database far harder than the daily cadence it replaced.
|
||
return false;
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Emits the ceiling-reached breadcrumb, rate-limited to one line per
|
||
/// <see cref="CeilingLogInterval"/>.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// Rate-limited because the whole point of the ceiling is that a large backlog produces a RUN
|
||
/// of ceiling-limited ticks, one every <see cref="SiteEventLogOptions.PurgeCatchUpInterval"/>
|
||
/// — an unthrottled line would turn a converging purge into log noise at exactly the moment an
|
||
/// operator is reading the log for something else. Warning rather than Information because the
|
||
/// state it describes is "retention is NOT yet satisfied on this node", which is worth
|
||
/// noticing; the counter
|
||
/// (<c>scadabridge.site.event_log.purge_ceiling_hits</c>) carries every hit for the series that
|
||
/// distinguishes converging from never-converging.
|
||
/// </remarks>
|
||
/// <param name="deletedThisTick">Rows the ceiling-limited tick deleted.</param>
|
||
private void LogCeilingReached(int deletedThisTick)
|
||
{
|
||
var now = DateTimeOffset.UtcNow;
|
||
if (_lastCeilingLogUtc is { } last && now - last < CeilingLogInterval)
|
||
return;
|
||
|
||
_lastCeilingLogUtc = now;
|
||
_logger.LogWarning(
|
||
"Event log purge stopped at its per-tick ceiling of {Ceiling} rows ({Deleted} deleted); " +
|
||
"expired events remain and the next purge runs in {CatchUp} instead of {Interval}. " +
|
||
"The ceiling exists so one tick's CDC oplog burst stays well inside LocalDb's " +
|
||
"MaxOplogRows budget on a replicated node; set ScadaBridge:SiteEventLog:MaxPurgedRowsPerTick " +
|
||
"to 0 to restore unbounded purges.",
|
||
_options.MaxPurgedRowsPerTick, deletedThisTick,
|
||
_options.PurgeCatchUpInterval, _options.PurgeInterval);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Deletes events older than the retention window, in bounded
|
||
/// <see cref="RetentionPurgeBatchSize"/>-row slices per DELETE statement rather than
|
||
/// one unbounded DELETE for the whole expired set — mirroring the batching already
|
||
/// used a few lines below in <see cref="PurgeByStorageCap"/> (<see cref="CapPurgeBatchSize"/>).
|
||
/// An expired backlog in the tens/hundreds of thousands of rows would otherwise hold a
|
||
/// single SQLite write transaction (and the shared <c>_writeLock</c>) open for the
|
||
/// whole delete, blocking every concurrent <see cref="SiteEventLogger.LogEventAsync"/>
|
||
/// writer-loop flush for that entire duration.
|
||
/// </summary>
|
||
/// <param name="budget">The tick's shared delete budget; slices are drawn from it.</param>
|
||
private void PurgeByRetention(PurgeBudget budget)
|
||
{
|
||
var cutoff = DateTimeOffset.UtcNow.AddDays(-_options.RetentionDays).ToString("o");
|
||
var totalDeleted = 0;
|
||
|
||
while (true)
|
||
{
|
||
// The slice is the smaller of the statement batch and what the tick has left, so the
|
||
// ceiling is enforced exactly rather than overshooting by up to one batch.
|
||
var slice = budget.NextSlice(RetentionPurgeBatchSize);
|
||
if (slice == 0)
|
||
{
|
||
// Only a ceiling hit if there is genuinely more to purge. An expired set that
|
||
// landed exactly on the budget is FINISHED, and claiming otherwise would arm a
|
||
// 5-minute catch-up tick that deletes nothing and logs that retention is behind.
|
||
budget.StoppedAtCeiling = ExpiredRowsRemain(cutoff);
|
||
break;
|
||
}
|
||
|
||
var deleted = _eventLogger.WithConnection(connection =>
|
||
{
|
||
using var cmd = connection.CreateCommand();
|
||
cmd.CommandText = $"""
|
||
DELETE FROM site_events WHERE id IN (
|
||
SELECT id FROM site_events
|
||
WHERE timestamp < $cutoff
|
||
LIMIT {slice}
|
||
)
|
||
""";
|
||
cmd.Parameters.AddWithValue("$cutoff", cutoff);
|
||
return cmd.ExecuteNonQuery();
|
||
});
|
||
|
||
totalDeleted += deleted;
|
||
budget.Record(deleted);
|
||
|
||
// Compared against the slice actually REQUESTED, not the constant: a short final
|
||
// slice that came back full still means "there may be more", and a full-size slice
|
||
// that came back short means the expired set is exhausted.
|
||
if (deleted < slice)
|
||
break;
|
||
}
|
||
|
||
if (totalDeleted > 0)
|
||
_logger.LogInformation("Purged {Count} events older than {Days} days", totalDeleted, _options.RetentionDays);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Whether at least one event older than <paramref name="cutoff"/> is still present.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// An existence probe rather than a <c>COUNT(*)</c>: the answer needed is boolean, and on a
|
||
/// node working through a multi-million-row backlog a count would walk the whole expired set
|
||
/// once per ceiling-limited tick to produce a number nothing reads. <c>idx_events_timestamp</c>
|
||
/// makes this one index seek.
|
||
/// </remarks>
|
||
/// <param name="cutoff">The retention cutoff, ISO-8601 round-trip formatted.</param>
|
||
/// <returns><c>true</c> if expired rows remain.</returns>
|
||
private bool ExpiredRowsRemain(string cutoff) =>
|
||
_eventLogger.WithConnection(connection =>
|
||
{
|
||
using var cmd = connection.CreateCommand();
|
||
cmd.CommandText = "SELECT EXISTS(SELECT 1 FROM site_events WHERE timestamp < $cutoff)";
|
||
cmd.Parameters.AddWithValue("$cutoff", cutoff);
|
||
return Convert.ToInt64(cmd.ExecuteScalar(), System.Globalization.CultureInfo.InvariantCulture) != 0;
|
||
});
|
||
|
||
/// <summary>
|
||
/// Emits the operator breadcrumb that explains the oplog backlog spike a purge produces on a
|
||
/// replication-enabled node — <b>once per purge tick</b>. No-op when this node has no
|
||
/// replication peer.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// Called from <see cref="RunPurge"/> rather than from each phase (register row 84). Both the
|
||
/// retention purge and the cap trim generate the same CDC cost, and a tick in which both
|
||
/// deleted rows used to emit two identical Information notes. One note carrying the tick's
|
||
/// combined total is what an operator correlating an oplog spike actually wants.
|
||
/// </para>
|
||
/// <para>
|
||
/// <c>site_events</c> is a replicated table, and a retention/cap DELETE is captured by CDC
|
||
/// exactly like an ordinary write — there is deliberately no purge-exemption path (LocalDb
|
||
/// Phase 2: CDC does all three jobs). The daily purge therefore pushes one oplog row per
|
||
/// deleted event, and the site health report's <c>LocalDbOplogBacklog</c> (Prometheus
|
||
/// <c>localdb_oplog_depth</c>) spikes until the peer acks them. That is expected and drains
|
||
/// on its own; without this line an operator correlating the spike has nothing in the log to
|
||
/// tie it to. Information level because it is only interesting next to the purge line it
|
||
/// follows — the spike itself is not a fault.
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <param name="deletedRows">
|
||
/// Rows deleted by the WHOLE tick — retention purge and cap trim combined, read off the
|
||
/// shared <see cref="PurgeBudget"/>. Always > 0 at the single call site.
|
||
/// </param>
|
||
private void LogReplicationBacklogNote(int deletedRows)
|
||
{
|
||
bool replicated;
|
||
try
|
||
{
|
||
replicated = _isReplicationConfigured();
|
||
}
|
||
catch (Exception checkEx)
|
||
{
|
||
// A log-wording predicate must never break the purge loop.
|
||
_logger.LogDebug(checkEx, "Replication check threw while composing the purge log note; note suppressed");
|
||
return;
|
||
}
|
||
|
||
if (!replicated)
|
||
return;
|
||
|
||
_logger.LogInformation(
|
||
"Purged {Count} site_events rows on a replication-enabled node — a transient LocalDb oplog backlog " +
|
||
"of up to that many rows is expected while the deletes replicate to the peer. It drains on its own; " +
|
||
"LocalDbReplicationConnected staying true with the backlog returning to ~0 is the healthy signature. " +
|
||
"A tick deletes at most {Ceiling} rows, so the spike is bounded by that number rather than by how " +
|
||
"much history had expired.",
|
||
deletedRows,
|
||
_options.MaxPurgedRowsPerTick > 0 ? _options.MaxPurgedRowsPerTick.ToString() : "unlimited");
|
||
}
|
||
|
||
/// <summary>
|
||
/// Trims the oldest events until <c>site_events</c>' own footprint is back under
|
||
/// <see cref="SiteEventLogOptions.MaxStorageMb"/>.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>The cap is table-scoped as of register row 81.</b> It used to compare
|
||
/// <see cref="SiteEventLogOptions.MaxStorageMb"/> against the whole consolidated LocalDb
|
||
/// file, which on a replication-enabled node made the trim unwinnable: every batch freed
|
||
/// event pages and added an equal number of oplog rows and row-version tombstones to the
|
||
/// same file, so the measured size did not fall and the progress guard stopped the trim
|
||
/// 2.5x over cap with 226 k rows still present. The cap now measures
|
||
/// <see cref="EstimateSiteEventsSize"/> — <c>site_events</c> and its four indexes only —
|
||
/// so replication bookkeeping can no longer consume, or block, the event log's budget.
|
||
/// </para>
|
||
/// <para>
|
||
/// Bytes-per-row is sampled ONCE per tick and held fixed for the whole loop, while the row
|
||
/// count is re-read after each batch. That split is deliberate: re-sampling row width every
|
||
/// batch would make the estimate jitter as the trim consumes the head of the table, and a
|
||
/// sample that happened to widen could stall the loop through the progress guard below —
|
||
/// the same class of self-defeating measurement row 81 was about.
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <param name="budget">The tick's shared delete budget, already drawn down by the retention purge.</param>
|
||
private void PurgeByStorageCap(PurgeBudget budget)
|
||
{
|
||
var capBytes = (long)_options.MaxStorageMb * 1024 * 1024;
|
||
var estimate = EstimateSiteEventsSize();
|
||
|
||
if (estimate.RowCount <= 0)
|
||
return;
|
||
|
||
// Wipe-prevention backstop, first of two — checked BEFORE the cap comparison, because a
|
||
// zero row width makes "site_events is over cap" unanswerable rather than false. An
|
||
// unusable width (empty sample, degenerate estimate) used to have no analogue: the old
|
||
// loop's answer to an unanswerable cap was to keep deleting. Refuse instead.
|
||
if (estimate.BytesPerRow <= 0)
|
||
{
|
||
_logger.LogWarning(
|
||
"Storage-cap trim skipped: site_events row width could not be sampled ({Rows} rows). " +
|
||
"No rows are deleted rather than trimming against an unusable size estimate.",
|
||
estimate.RowCount);
|
||
return;
|
||
}
|
||
|
||
if (estimate.EstimatedBytes <= capBytes)
|
||
return;
|
||
|
||
// Over cap but out of budget: flag the catch-up tick and stay silent rather than logging
|
||
// "purging oldest events" and then deleting nothing. Ordered AFTER the size probe so a
|
||
// budget spent by the retention purge on an under-cap database is not mistaken for
|
||
// outstanding cap work. The cap is re-evaluated from scratch on the catch-up tick.
|
||
if (budget.NextSlice(CapPurgeBatchSize) == 0)
|
||
{
|
||
budget.StoppedAtCeiling = true;
|
||
return;
|
||
}
|
||
|
||
_logger.LogWarning(
|
||
"site_events occupies about {Size:F1} MB across {Rows} rows — over the {Cap} MB cap. " +
|
||
"Trimming the oldest events. (The cap measures the event table and its indexes only, " +
|
||
"not the whole consolidated LocalDb file.)",
|
||
estimate.EstimatedBytes / (1024.0 * 1024.0), estimate.RowCount, _options.MaxStorageMb);
|
||
|
||
// Delete the oldest events in batches until the ESTIMATED TABLE size is under the cap.
|
||
// Wipe-prevention backstop, second of two: the loop also stops if the estimate fails to
|
||
// fall across an iteration. Replication bookkeeping can no longer trip that — the metric
|
||
// excludes it — so it now means what it says: the table is not shrinking, because a batch
|
||
// deleted nothing or because concurrent recorder writes are outrunning the trim.
|
||
var totalDeleted = 0;
|
||
var currentSizeBytes = estimate.EstimatedBytes;
|
||
|
||
while (currentSizeBytes > capBytes)
|
||
{
|
||
var slice = budget.NextSlice(CapPurgeBatchSize);
|
||
if (slice == 0)
|
||
{
|
||
budget.StoppedAtCeiling = true;
|
||
break;
|
||
}
|
||
|
||
var previousSizeBytes = currentSizeBytes;
|
||
|
||
var deleted = _eventLogger.WithConnection(connection =>
|
||
{
|
||
using var cmd = connection.CreateCommand();
|
||
// Order by timestamp, NOT by id. The id was a monotonic autoincrement
|
||
// when this was written, so "lowest id" meant "oldest"; since LocalDb
|
||
// Phase 1 it is a GUID and ordering by it would delete an ARBITRARY
|
||
// batch of events rather than the oldest ones. id remains in the ORDER
|
||
// BY only as a tie-break so the batch is deterministic.
|
||
cmd.CommandText = $"""
|
||
DELETE FROM site_events WHERE id IN (
|
||
SELECT id FROM site_events
|
||
ORDER BY timestamp ASC, id ASC
|
||
LIMIT {slice}
|
||
)
|
||
""";
|
||
var rows = cmd.ExecuteNonQuery();
|
||
|
||
// Reclaim free pages so page_count/freelist measurement reflects the
|
||
// delete. Effective because auto_vacuum = INCREMENTAL is set at schema
|
||
// creation; harmless otherwise.
|
||
using var vacuumCmd = connection.CreateCommand();
|
||
vacuumCmd.CommandText = "PRAGMA incremental_vacuum";
|
||
vacuumCmd.ExecuteNonQuery();
|
||
|
||
return rows;
|
||
});
|
||
|
||
if (deleted == 0)
|
||
break;
|
||
|
||
totalDeleted += deleted;
|
||
budget.Record(deleted);
|
||
|
||
// Row count re-read from the database, not decremented locally, so concurrent
|
||
// recorder writes are visible; row width stays fixed for the tick (see remarks).
|
||
currentSizeBytes = (long)(GetSiteEventsRowCount() * estimate.BytesPerRow);
|
||
|
||
if (currentSizeBytes >= previousSizeBytes)
|
||
{
|
||
_logger.LogWarning(
|
||
"Estimated site_events size did not decrease after a cap-purge batch " +
|
||
"({Size:F1} MB); stopping to avoid emptying the log. Deletes are landing, so " +
|
||
"this means new events are arriving at least as fast as the trim removes them.",
|
||
currentSizeBytes / (1024.0 * 1024.0));
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
/// <summary>
|
||
/// Returns the logical size of the WHOLE consolidated database file in bytes — only pages
|
||
/// that hold live data, excluding free pages on the freelist.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>This is no longer what the storage cap measures</b> (register row 81,
|
||
/// <c>docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md</c>). Since LocalDb
|
||
/// Phase 1/2 the file holds ten tables plus <c>__localdb_oplog</c>,
|
||
/// <c>__localdb_row_version</c> and <c>__localdb_snapshot_state</c>, so a whole-file
|
||
/// number is not <c>site_events</c>' size: on a replication-enabled node each cap-trim
|
||
/// batch freed ~390 KB of event pages while adding ~1000 oplog rows and ~1000 tombstones
|
||
/// to the same file, and the trim fought itself to a standstill 2.5x over cap.
|
||
/// </para>
|
||
/// <para>
|
||
/// It survives for two narrow jobs: it is a true UPPER BOUND on <c>site_events</c>' own
|
||
/// pages, which <see cref="EstimateSiteEventsSize"/> uses to clamp an over-eager model,
|
||
/// and it remains a useful diagnostic for the file as a whole.
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <returns>The logical size of the whole database file in bytes (used pages × page size).</returns>
|
||
internal long GetDatabaseSizeBytes()
|
||
{
|
||
return _eventLogger.WithConnection(connection => ReadLogicalFileBytes(connection));
|
||
}
|
||
|
||
/// <summary>Logical size of the whole file, on an already-held connection.</summary>
|
||
/// <param name="connection">An open connection to the consolidated database.</param>
|
||
/// <returns>Used pages × page size, in bytes.</returns>
|
||
private static long ReadLogicalFileBytes(SqliteConnection connection)
|
||
{
|
||
var pageCount = ReadPragmaLong(connection, "PRAGMA page_count");
|
||
var freeListCount = ReadPragmaLong(connection, "PRAGMA freelist_count");
|
||
var pageSize = ReadPragmaLong(connection, "PRAGMA page_size");
|
||
return Math.Max(0, pageCount - freeListCount) * pageSize;
|
||
}
|
||
|
||
/// <summary>Reads a single-value PRAGMA as a 64-bit integer.</summary>
|
||
/// <param name="connection">An open connection.</param>
|
||
/// <param name="pragma">The PRAGMA statement to run.</param>
|
||
/// <returns>The PRAGMA's value.</returns>
|
||
private static long ReadPragmaLong(SqliteConnection connection, string pragma)
|
||
{
|
||
using var cmd = connection.CreateCommand();
|
||
cmd.CommandText = pragma;
|
||
return Convert.ToInt64(cmd.ExecuteScalar(), CultureInfo.InvariantCulture);
|
||
}
|
||
|
||
/// <summary>
|
||
/// How many rows <c>site_events</c> currently holds, and how many bytes of the file one
|
||
/// average row occupies — the two numbers the storage cap is expressed in.
|
||
/// </summary>
|
||
/// <param name="RowCount">Live row count at the time of the estimate.</param>
|
||
/// <param name="BytesPerRow">
|
||
/// Modelled file bytes per row, table pages plus index pages. <c>0</c> when the table is
|
||
/// empty or the sample could not be sized, which the cap trim treats as "do not trim".
|
||
/// </param>
|
||
internal readonly record struct SiteEventsSizeEstimate(long RowCount, double BytesPerRow)
|
||
{
|
||
/// <summary>Modelled total bytes <c>site_events</c> occupies in the file.</summary>
|
||
public long EstimatedBytes => (long)(RowCount * BytesPerRow);
|
||
}
|
||
|
||
/// <summary>
|
||
/// Estimates how much of the consolidated file <c>site_events</c> — and only
|
||
/// <c>site_events</c> — occupies, by sampling average row width and modelling SQLite's
|
||
/// page layout.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// <para>
|
||
/// <b>Why a model and not a measurement.</b> SQLite can report per-table page usage
|
||
/// exactly through the <c>dbstat</c> virtual table, and that was the intended fix for
|
||
/// register row 81 — but <c>dbstat</c> is a compile-time option and the
|
||
/// <c>SQLitePCLRaw.lib.e_sqlite3</c> native build this solution ships does NOT enable it.
|
||
/// Probed directly (<c>SqliteCapabilityTests</c>): <c>PRAGMA compile_options</c> on
|
||
/// SQLite 3.53.3 lists no <c>ENABLE_DBSTAT_VTAB</c>, <c>pragma_module_list</c> returns ten
|
||
/// modules (fts3/4/5, rtree, geopoly) with neither <c>dbstat</c> nor <c>sqlite_dbpage</c>
|
||
/// among them, and selecting from <c>dbstat</c> fails with "no such table". So an exact
|
||
/// per-table page count is not obtainable from this provider at all, and the cap is
|
||
/// modelled instead. That trade is recorded in
|
||
/// <c>docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md</c>.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>The model.</b> Rows are sampled from both ends of the table (<see cref="SizeSampleRows"/>
|
||
/// from each — the head is what the oldest-first trim will actually delete, the tail catches a
|
||
/// workload whose event shape has drifted) to get an average payload width and an average
|
||
/// width across the four indexed columns. Table cost is then computed as whole pages —
|
||
/// <c>ceil(rows / rowsPerPage) × pageSize</c> — rather than as a flat bytes-per-row rate,
|
||
/// because that is the term that dominates once rows approach half a page: rows too wide to
|
||
/// pack two-per-page waste the remainder, which a linear model underestimates by ~40 %.
|
||
/// Index cost is added per row from the sampled indexed-column widths, so an event stream
|
||
/// that leaves <c>instance_id</c> null is charged less than one that fills it.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Index pages count against the cap</b> — deliberately. The four indexes are storage the
|
||
/// event log causes and owns (measured at ~34 % of the table's total footprint), and the
|
||
/// 1 GB default was sized against the 393 bytes/row figure that already included them.
|
||
/// Excluding them would silently raise real disk use by about a third at the same setting.
|
||
/// </para>
|
||
/// <para>
|
||
/// <b>Clamped by the file.</b> The whole-file logical size is a true upper bound on any one
|
||
/// table's pages, so the model is capped by it. That can only ever correct an
|
||
/// over-estimate downward; it cannot reintroduce the row-81 defect, because on a replicated
|
||
/// node the file is larger than <c>site_events</c>, not smaller, and so never binds.
|
||
/// </para>
|
||
/// <para>
|
||
/// Marked <c>virtual</c> so tests can substitute a degenerate estimate and exercise the
|
||
/// trim's backstops.
|
||
/// </para>
|
||
/// </remarks>
|
||
/// <returns>The current row count and modelled bytes per row.</returns>
|
||
internal virtual SiteEventsSizeEstimate EstimateSiteEventsSize()
|
||
{
|
||
return _eventLogger.WithConnection(connection =>
|
||
{
|
||
var rowCount = ReadPragmaLong(connection, "SELECT COUNT(*) FROM site_events");
|
||
if (rowCount <= 0)
|
||
return new SiteEventsSizeEstimate(0, 0);
|
||
|
||
double avgPayloadBytes;
|
||
double avgIndexedBytes;
|
||
using (var sampleCmd = connection.CreateCommand())
|
||
{
|
||
// octet_length (SQLite 3.43+) rather than length(), which counts CHARACTERS:
|
||
// a message with multi-byte UTF-8 would otherwise be under-sized. Pinned by
|
||
// SqliteCapabilityTests so a provider downgrade fails a test rather than
|
||
// silently shrinking every estimate.
|
||
sampleCmd.CommandText = $"""
|
||
SELECT AVG(payload_bytes), AVG(indexed_bytes) FROM (
|
||
SELECT payload_bytes, indexed_bytes FROM (
|
||
SELECT
|
||
octet_length(id) + octet_length(timestamp) + octet_length(event_type)
|
||
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0)
|
||
+ octet_length(source) + octet_length(message)
|
||
+ COALESCE(octet_length(details), 0) AS payload_bytes,
|
||
octet_length(timestamp) + octet_length(event_type)
|
||
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0)
|
||
AS indexed_bytes
|
||
FROM site_events LIMIT {SizeSampleRows}
|
||
)
|
||
UNION ALL
|
||
SELECT payload_bytes, indexed_bytes FROM (
|
||
SELECT
|
||
octet_length(id) + octet_length(timestamp) + octet_length(event_type)
|
||
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0)
|
||
+ octet_length(source) + octet_length(message)
|
||
+ COALESCE(octet_length(details), 0) AS payload_bytes,
|
||
octet_length(timestamp) + octet_length(event_type)
|
||
+ octet_length(severity) + COALESCE(octet_length(instance_id), 0)
|
||
AS indexed_bytes
|
||
FROM site_events ORDER BY rowid DESC LIMIT {SizeSampleRows}
|
||
)
|
||
)
|
||
""";
|
||
using var reader = sampleCmd.ExecuteReader();
|
||
if (!reader.Read() || reader.IsDBNull(0))
|
||
return new SiteEventsSizeEstimate(rowCount, 0);
|
||
|
||
avgPayloadBytes = reader.GetDouble(0);
|
||
avgIndexedBytes = reader.IsDBNull(1) ? 0 : reader.GetDouble(1);
|
||
}
|
||
|
||
var pageSize = ReadPragmaLong(connection, "PRAGMA page_size");
|
||
var usablePerPage = (pageSize - PageReservedBytes) * TablePageFillFactor;
|
||
var avgCellBytes = avgPayloadBytes + TableCellOverheadBytes;
|
||
|
||
// At least one row per page: a row wider than a page spills to overflow pages,
|
||
// which this floor charges as a whole page each — close enough, and never under.
|
||
var rowsPerPage = Math.Max(1, (long)(usablePerPage / Math.Max(1, avgCellBytes)));
|
||
var tableBytes = (long)Math.Ceiling(rowCount / (double)rowsPerPage) * pageSize;
|
||
|
||
var indexBytesPerRow =
|
||
(avgIndexedBytes + (IndexEntryOverheadBytes * IndexedColumnCount)) / IndexPageFillFactor;
|
||
|
||
var modelled = tableBytes + (long)(rowCount * indexBytesPerRow);
|
||
var clamped = Math.Min(modelled, ReadLogicalFileBytes(connection));
|
||
|
||
return new SiteEventsSizeEstimate(rowCount, clamped / (double)rowCount);
|
||
});
|
||
}
|
||
|
||
/// <summary>Live <c>site_events</c> row count.</summary>
|
||
/// <remarks>
|
||
/// Re-read after every cap-trim batch so the trim's progress check sees concurrent
|
||
/// recorder writes, not just its own deletes. <c>virtual</c> for the backstop tests.
|
||
/// </remarks>
|
||
/// <returns>Number of rows currently in <c>site_events</c>.</returns>
|
||
internal virtual long GetSiteEventsRowCount() =>
|
||
_eventLogger.WithConnection(connection => ReadPragmaLong(connection, "SELECT COUNT(*) FROM site_events"));
|
||
|
||
/// <summary>
|
||
/// One purge tick's shared delete allowance, drawn down by the retention purge and then the
|
||
/// storage-cap trim.
|
||
/// </summary>
|
||
/// <remarks>
|
||
/// Shared rather than per-phase on purpose: what the ceiling bounds is the CDC oplog burst a
|
||
/// TICK produces on a replicated node, and two independently-capped phases would let one tick
|
||
/// emit twice the intended maximum. The cap trim therefore runs on whatever the retention
|
||
/// purge left, and picks up on the accelerated catch-up tick if that was nothing.
|
||
/// </remarks>
|
||
/// <param name="ceiling">Rows this tick may delete; <c>0</c> or less means unlimited.</param>
|
||
private sealed class PurgeBudget(int ceiling)
|
||
{
|
||
private readonly bool _unlimited = ceiling <= 0;
|
||
|
||
/// <summary>Rows deleted so far this tick.</summary>
|
||
public int Deleted { get; private set; }
|
||
|
||
/// <summary>
|
||
/// Set when a phase ran out of budget with work still to do — the signal that the caller
|
||
/// should schedule an accelerated catch-up tick. Never set when the ceiling is unlimited.
|
||
/// </summary>
|
||
public bool StoppedAtCeiling { get; set; }
|
||
|
||
/// <summary>
|
||
/// The number of rows the next DELETE statement may take: <paramref name="batchSize"/>,
|
||
/// or what remains of the budget if that is smaller. <c>0</c> means the budget is spent.
|
||
/// </summary>
|
||
/// <param name="batchSize">The phase's own statement batch size.</param>
|
||
/// <returns>Rows the next statement may delete.</returns>
|
||
public int NextSlice(int batchSize) =>
|
||
_unlimited ? batchSize : Math.Max(0, Math.Min(batchSize, ceiling - Deleted));
|
||
|
||
/// <summary>Charges <paramref name="rows"/> against the budget.</summary>
|
||
/// <param name="rows">Rows the statement actually deleted.</param>
|
||
public void Record(int rows) => Deleted += rows;
|
||
}
|
||
}
|