Merge row 34: site_events purge burst measured (2.73M-row worst case = 2.7x MaxOplogRows) and paced — shared 250k/tick budget, 5-min catch-up, ceiling-hit counter

src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs

@@ -111,6 +111,21 @@ public static class ScadaBridgeTelemetry
         Meter.CreateCounter<long>("scadabridge.central.transport.import_sessions.rejected", unit: "1",
             description: "Bundle-import sessions rejected because the concurrent-import-session cap was exhausted.");
 
+    /// <summary>
+    /// Register row 34: incremented once per site-event-log purge tick that stopped at
+    /// <c>SiteEventLogOptions.MaxPurgedRowsPerTick</c> with expired rows still to delete.
+    ///
+    /// A single hit is routine (a backlog is being worked through on accelerated catch-up
+    /// ticks) — what the series is for is the SHAPE. A short burst of hits that stops is a
+    /// backlog converging; hits that never stop mean the event log is being written faster
+    /// than one paced tick can purge, i.e. the ceiling or the catch-up interval is now
+    /// mis-sized for this site's event rate. The paired log line is rate-limited and so
+    /// cannot answer that question on its own.
+    /// </summary>
+    private static readonly Counter<long> _siteEventLogPurgeCeilingHits =
+        Meter.CreateCounter<long>("scadabridge.site.event_log.purge_ceiling_hits", unit: "1",
+            description: "Site event log purge ticks that stopped at the per-tick delete ceiling with work remaining.");
+
     // ---------------- Histograms ----------------
 
     /// <summary>
@@ -272,6 +287,13 @@ public static class ScadaBridgeTelemetry
     /// </summary>
     public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1);
 
+    /// <summary>
+    /// Records one site-event-log purge tick that stopped at its per-tick delete ceiling
+    /// (register row 34). Call ONLY when rows genuinely remain to be purged — a tick that
+    /// finished the work and happened to land exactly on the ceiling has nothing to report.
+    /// </summary>
+    public static void RecordSiteEventLogPurgeCeilingHit() => _siteEventLogPurgeCeilingHits.Add(1);
+
     /// <summary>
     /// Records that a central→site gRPC subscriber stream opened (increments
     /// <c>scadabridge.site.stream.subscribers.active</c>). Renamed from

src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs

@@ -1,6 +1,7 @@
 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;
 
@@ -59,6 +60,15 @@ public class EventLogPurgeService : BackgroundService
     /// <summary>Number of events deleted per retention-purge batch — same size as <see cref="CapPurgeBatchSize"/>.</summary>
     private const int RetentionPurgeBatchSize = 1000;
 
+    /// <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;
@@ -100,25 +110,50 @@ public class EventLogPurgeService : BackgroundService
     }
 
     /// <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}",
-            _options.RetentionDays, _options.MaxStorageMb, _options.PurgeInterval);
-
-        using var timer = new PeriodicTimer(_options.PurgeInterval);
+            "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
-        RunPurge();
+        var ceilingHit = RunPurge();
 
-        while (await timer.WaitForNextTickAsync(stoppingToken).ConfigureAwait(false))
+        while (!stoppingToken.IsCancellationRequested)
         {
-            RunPurge();
+            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>
-    internal void RunPurge()
+    /// <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
         {
@@ -144,18 +179,62 @@ public class EventLogPurgeService : BackgroundService
             if (!isActive)
             {
                 _logger.LogDebug("Skipping event log purge tick — this node is not the active site member");
-                return;
+                return false;
             }
 
-            PurgeByRetention();
-            PurgeByStorageCap();
+            var budget = new PurgeBudget(_options.MaxPurgedRowsPerTick);
+            PurgeByRetention(budget);
+            PurgeByStorageCap(budget);
+
+            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
@@ -166,13 +245,26 @@ public class EventLogPurgeService : BackgroundService
     /// whole delete, blocking every concurrent <see cref="SiteEventLogger.LogEventAsync"/>
     /// writer-loop flush for that entire duration.
     /// </summary>
-    private void PurgeByRetention()
+    /// <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();
@@ -180,7 +272,7 @@ public class EventLogPurgeService : BackgroundService
                     DELETE FROM site_events WHERE id IN (
                         SELECT id FROM site_events
                         WHERE timestamp < $cutoff
-                        LIMIT {RetentionPurgeBatchSize}
+                        LIMIT {slice}
                     )
                     """;
                 cmd.Parameters.AddWithValue("$cutoff", cutoff);
@@ -188,8 +280,12 @@ public class EventLogPurgeService : BackgroundService
             });
 
             totalDeleted += deleted;
+            budget.Record(deleted);
 
-            if (deleted < RetentionPurgeBatchSize)
+            // 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;
         }
 
@@ -200,6 +296,26 @@ public class EventLogPurgeService : BackgroundService
         }
     }
 
+    /// <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. No-op when this node has no replication peer.
@@ -234,12 +350,16 @@ public class EventLogPurgeService : BackgroundService
 
         _logger.LogInformation(
             "Purged {Count} site_events rows on a replication-enabled node — a transient LocalDb oplog backlog " +
-            "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.",
-            deletedRows);
+            "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");
     }
 
-    private void PurgeByStorageCap()
+    /// <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 currentSizeBytes = GetDatabaseSizeBytes();
@@ -247,6 +367,16 @@ public class EventLogPurgeService : BackgroundService
         if (currentSizeBytes <= 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(
             "Event log size {Size:F1} MB exceeds cap {Cap} MB — purging oldest events",
             currentSizeBytes / (1024.0 * 1024.0), _options.MaxStorageMb);
@@ -259,6 +389,13 @@ public class EventLogPurgeService : BackgroundService
 
         while (currentSizeBytes > capBytes)
         {
+            var slice = budget.NextSlice(CapPurgeBatchSize);
+            if (slice == 0)
+            {
+                budget.StoppedAtCeiling = true;
+                break;
+            }
+
             var previousSizeBytes = currentSizeBytes;
 
             var deleted = _eventLogger.WithConnection(connection =>
@@ -273,7 +410,7 @@ public class EventLogPurgeService : BackgroundService
                     DELETE FROM site_events WHERE id IN (
                         SELECT id FROM site_events
                         ORDER BY timestamp ASC, id ASC
-                        LIMIT {CapPurgeBatchSize}
+                        LIMIT {slice}
                     )
                     """;
                 var rows = cmd.ExecuteNonQuery();
@@ -292,6 +429,7 @@ public class EventLogPurgeService : BackgroundService
                 break;
 
             totalDeleted += deleted;
+            budget.Record(deleted);
             currentSizeBytes = GetDatabaseSizeBytes();
 
             if (currentSizeBytes >= previousSizeBytes)
@@ -342,4 +480,42 @@ public class EventLogPurgeService : BackgroundService
             return usedPages * pageSize;
         });
     }
+
+    /// <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;
+    }
 }

src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptions.cs

@@ -28,6 +28,43 @@ public class SiteEventLogOptions
     /// <summary>Interval between purge runs; default 24 hours.</summary>
     public TimeSpan PurgeInterval { get; set; } = TimeSpan.FromHours(24);
     /// <summary>
+    /// Ceiling on how many <c>site_events</c> rows ONE purge tick may delete, across the
+    /// retention purge and the storage-cap trim combined. Default 250 000; <c>0</c> means
+    /// unlimited (the pre-pacing behaviour, kept as an escape hatch).
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why a ceiling exists at all.</b> <c>site_events</c> is CDC-captured on a
+    /// replication-enabled node and there is deliberately no purge-exemption path, so a purge
+    /// queues one oplog row per deleted event. Measured worst case (register row 34,
+    /// <c>docs/plans/2026-08-15-site-events-purge-burst-measurement.md</c>): a 1 GB event log
+    /// holds ~2.73 million rows at 393 bytes/row, and expiring all of them in one tick queued
+    /// 2 731 947 oplog rows — <b>2.7x over LocalDb's <c>MaxOplogRows</c> default of
+    /// 1 000 000</b>. Crossing that ceiling makes LocalDb's maintenance tick prune the oplog
+    /// and set <c>needs_snapshot</c>, which costs a FULL snapshot resync of all ten replicated
+    /// tables. The ceiling keeps a single tick's burst safely inside that budget (250 000 is
+    /// a 4x margin) while <see cref="PurgeCatchUpInterval"/> keeps retention converging.
+    /// </para>
+    /// <para>
+    /// The remainder is not dropped: a tick that hits the ceiling schedules the next tick after
+    /// <see cref="PurgeCatchUpInterval"/> instead of <see cref="PurgeInterval"/>, so a backlog
+    /// drains over minutes rather than days while every individual burst stays bounded.
+    /// </para>
+    /// </remarks>
+    public int MaxPurgedRowsPerTick { get; set; } = 250_000;
+    /// <summary>
+    /// Delay before the next purge tick when the previous tick stopped at
+    /// <see cref="MaxPurgedRowsPerTick"/>; default 5 minutes. Ignored when the ceiling was not
+    /// reached — an ordinary tick waits <see cref="PurgeInterval"/>.
+    /// </summary>
+    /// <remarks>
+    /// Sized so the peer drains one tick's burst before the next one lands: 250 000 tombstones
+    /// measured at ~10 800 rows/s over a loopback pair (≈23 s), leaving an order of magnitude of
+    /// headroom for a real inter-node link. A worst-case 2.73 M-row backlog therefore converges
+    /// in ~11 ticks ≈ 55 minutes instead of 11 days at the daily interval.
+    /// </remarks>
+    public TimeSpan PurgeCatchUpInterval { get; set; } = TimeSpan.FromMinutes(5);
+    /// <summary>
     /// Bound on the background write queue. Default 10 000 events.
     /// Overflow uses <c>BoundedChannelFullMode.DropOldest</c> — callers never block; the
     /// dropped event's <c>Task</c> is faulted and <c>FailedWriteCount</c> is incremented

src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptionsValidator.cs

@@ -41,6 +41,29 @@ public sealed class SiteEventLogOptionsValidator : OptionsValidatorBase<SiteEven
             $"ScadaBridge:SiteEventLog:PurgeInterval must be a positive duration " +
             $"(was {options.PurgeInterval}); it is the purge PeriodicTimer period.");
 
+        // A NEGATIVE ceiling is a config error, not a synonym for "unlimited": 0 already
+        // carries that meaning explicitly, and silently treating -1 the same way would hide
+        // a typo that removes the only bound on a purge burst's oplog cost.
+        builder.RequireThat(options.MaxPurgedRowsPerTick >= 0,
+            $"ScadaBridge:SiteEventLog:MaxPurgedRowsPerTick must be 0 (unlimited) or greater " +
+            $"(was {options.MaxPurgedRowsPerTick}); it bounds how many rows one purge tick deletes.");
+
+        builder.RequireThat(options.PurgeCatchUpInterval > TimeSpan.Zero,
+            $"ScadaBridge:SiteEventLog:PurgeCatchUpInterval must be a positive duration " +
+            $"(was {options.PurgeCatchUpInterval}); it is the delay before the next tick when a " +
+            "purge stopped at MaxPurgedRowsPerTick.");
+
+        // A catch-up longer than the ordinary interval would make hitting the ceiling SLOW the
+        // purge down, which inverts the whole point of the accelerated retry. Only compared when
+        // both are positive (each owned above), so a bad value fires exactly one message.
+        builder.RequireThat(
+            !(options.PurgeInterval > TimeSpan.Zero
+              && options.PurgeCatchUpInterval > TimeSpan.Zero
+              && options.PurgeCatchUpInterval > options.PurgeInterval),
+            $"ScadaBridge:SiteEventLog:PurgeCatchUpInterval ({options.PurgeCatchUpInterval}) must be " +
+            $"less than or equal to PurgeInterval ({options.PurgeInterval}): it is an ACCELERATED " +
+            "retry after a ceiling-limited tick, so a longer value would delay convergence instead.");
+
         // The hard page-size ceiling clamps a caller-supplied PageSize; a ceiling
         // below the default QueryPageSize would silently clamp the default below
         // itself. Only meaningful when both are positive (owned above), so stay