perf(siteeventlog): pace the site_events purge burst — per-tick delete ceiling + accelerated catch-up (row 34)

Measured the unbounded purge burst and found it can force a full ten-table
snapshot resync, then bounded it. No purge-exemption path was added — the
deletes are still CDC-captured like any other write, which stays ruled out
by design; what changed is how many a single tick performs.

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

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

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

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

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

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

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

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

Tests: SiteEventLogging 93/93, Commons 715/715, HealthMonitoring 115/115,
Host 505/505. Solution build clean.

tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs

@@ -567,6 +567,217 @@ public class EventLogPurgeServiceTests : IDisposable
         Assert.DoesNotContain(logger.Entries, e => e.Message.Contains(BacklogNoteFragment, StringComparison.Ordinal));
     }
 
+    // ── Register row 34: per-tick delete ceiling ──
+
+    /// <summary>
+    /// Fragment unique to the ceiling-reached line. As with the backlog note, the operator prose
+    /// may be reworded; what must not regress is that the line fires when a tick is cut short.
+    /// </summary>
+    private const string CeilingLineFragment = "per-tick ceiling";
+
+    /// <summary>The counter incremented once per ceiling-limited tick.</summary>
+    private const string CeilingHitCounter = "scadabridge.site.event_log.purge_ceiling_hits";
+
+    /// <summary>Seeds <paramref name="count"/> events one second apart, all past retention.</summary>
+    private void InsertExpiredEvents(int count)
+    {
+        var expiredBase = DateTimeOffset.UtcNow.AddDays(-31);
+        _eventLogger.WithConnection(connection =>
+        {
+            using var tx = connection.BeginTransaction();
+            using var cmd = connection.CreateCommand();
+            cmd.Transaction = tx;
+            cmd.CommandText = """
+                INSERT INTO site_events (id, timestamp, event_type, severity, source, message)
+                VALUES ($id, $ts, 'script', 'Info', 'Test', 'Test message')
+                """;
+            var id = cmd.Parameters.Add("$id", Microsoft.Data.Sqlite.SqliteType.Text);
+            var ts = cmd.Parameters.Add("$ts", Microsoft.Data.Sqlite.SqliteType.Text);
+            for (var i = 0; i < count; i++)
+            {
+                id.Value = Guid.NewGuid().ToString("N");
+                ts.Value = expiredBase.AddSeconds(i).ToString("o");
+                cmd.ExecuteNonQuery();
+            }
+            tx.Commit();
+        });
+    }
+
+    /// <summary>Options with the per-tick ceiling set and a short catch-up interval.</summary>
+    private SiteEventLogOptions CeilingOptions(int ceiling, int maxStorageMb = 1024) => new()
+    {
+        DatabasePath = _dbPath,
+        RetentionDays = 30,
+        MaxStorageMb = maxStorageMb,
+        MaxPurgedRowsPerTick = ceiling,
+        PurgeCatchUpInterval = TimeSpan.FromSeconds(1),
+    };
+
+    [Fact]
+    public void PurgeByRetention_StopsAtThePerTickCeiling()
+    {
+        // Row 34 core behaviour. site_events is CDC-captured on a replicated node with no
+        // purge-exemption path, so an unbounded tick queues one oplog row per deleted event —
+        // measured at 2,731,947 rows for a full 1 GB log, 2.7x over LocalDb's MaxOplogRows
+        // default, which costs a full ten-table snapshot resync. The ceiling bounds the burst.
+        InsertExpiredEvents(2500);
+
+        var purge = CreatePurgeService(CeilingOptions(1000));
+        var ceilingHit = purge.RunPurge();
+
+        Assert.True(ceilingHit, "A tick cut short by the ceiling must ask for an accelerated catch-up tick.");
+        Assert.Equal(1500, GetEventCount());
+    }
+
+    [Fact]
+    public void PurgeByRetention_WithCeiling_ConvergesAcrossSuccessiveTicks()
+    {
+        // The remainder must be picked up, or pacing would trade a burst for a retention policy
+        // that never completes. 2500 expired rows at 1000/tick = three ticks, the third of which
+        // finishes the work and reports no ceiling hit.
+        InsertExpiredEvents(2500);
+        InsertEventWithTimestamp(DateTimeOffset.UtcNow);
+
+        var purge = CreatePurgeService(CeilingOptions(1000));
+
+        Assert.True(purge.RunPurge());
+        Assert.Equal(1501, GetEventCount());
+        Assert.True(purge.RunPurge());
+        Assert.Equal(501, GetEventCount());
+        Assert.False(purge.RunPurge(), "The final tick clears the backlog and must not ask for another.");
+        Assert.Equal(1, GetEventCount());
+    }
+
+    [Fact]
+    public void PurgeByRetention_ExpiredSetLandingExactlyOnTheCeiling_IsNotACeilingHit()
+    {
+        // Exhausting the budget is not the same as having work left. A tick that deleted the last
+        // expired row and happened to land on the ceiling must not arm a catch-up tick that will
+        // delete nothing and log that retention is behind.
+        InsertExpiredEvents(1000);
+
+        var purge = CreatePurgeService(CeilingOptions(1000));
+
+        Assert.False(purge.RunPurge());
+        Assert.Equal(0, GetEventCount());
+    }
+
+    [Fact]
+    public void MaxPurgedRowsPerTick_Zero_RestoresUnboundedPurges()
+    {
+        // The documented escape hatch: 0 means unlimited, i.e. byte-for-byte the pre-row-34
+        // behaviour, for an operator who would rather take one long burst than a paced one.
+        InsertExpiredEvents(2500);
+
+        var purge = CreatePurgeService(CeilingOptions(0));
+
+        Assert.False(purge.RunPurge());
+        Assert.Equal(0, GetEventCount());
+    }
+
+    [Fact]
+    public void RunPurge_HittingTheCeiling_LogsTheCeilingLineAndCountsIt()
+    {
+        // Visibility requirement: a ceiling-limited tick is silent progress otherwise — the log
+        // line names the ceiling and the accelerated retry, and the counter carries every hit so
+        // "converging" can be told from "never converging".
+        InsertExpiredEvents(2500);
+
+        using var probe = new CounterProbe(CeilingHitCounter);
+        var logger = new CapturingLogger();
+        var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
+        purge.RunPurge();
+
+        var line = Assert.Single(
+            logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
+        Assert.Equal(LogLevel.Warning, line.Level);
+        Assert.Equal(1, probe.Total);
+    }
+
+    [Fact]
+    public void RunPurge_RepeatedCeilingHits_RateLimitTheLogLineButNotTheCounter()
+    {
+        // A large backlog produces a RUN of ceiling-limited ticks, one every catch-up interval.
+        // Logging each would bury the rest of the log; the counter is the unthrottled signal.
+        InsertExpiredEvents(3000);
+
+        using var probe = new CounterProbe(CeilingHitCounter);
+        var logger = new CapturingLogger();
+        var purge = CreatePurgeService(CeilingOptions(1000), logger: logger);
+        purge.RunPurge();
+        purge.RunPurge();
+
+        Assert.Single(logger.Entries, e => e.Message.Contains(CeilingLineFragment, StringComparison.Ordinal));
+        Assert.Equal(2, probe.Total);
+    }
+
+    [Fact]
+    public void PurgeByStorageCap_SharesTheTickBudgetWithRetention()
+    {
+        // The ceiling bounds the TICK, not each phase: two independently-capped phases would let
+        // one tick emit twice the intended oplog burst. Here the retention purge consumes the
+        // whole budget, so the cap trim does not run at all — and must not log that it is
+        // purging oldest events while deleting nothing.
+        InsertExpiredEvents(1200);
+        for (var i = 0; i < 200; i++)
+            InsertEventWithTimestamp(DateTimeOffset.UtcNow);
+
+        var logger = new CapturingLogger();
+        // A 0 MB cap means the cap trim would empty whatever the retention purge left, if it ran.
+        var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0), logger: logger);
+
+        Assert.True(purge.RunPurge());
+        Assert.Equal(400, GetEventCount());
+        Assert.DoesNotContain(logger.Entries, e => e.Message.Contains("exceeds cap", StringComparison.Ordinal));
+    }
+
+    [Fact]
+    public void PurgeByStorageCap_WithCeiling_StopsAtTheBudgetAndAsksForCatchUp()
+    {
+        // The cap path pays the same CDC cost and is usually the larger burst of the two, so it
+        // draws from the same budget and reports the same ceiling hit when it runs out.
+        for (var i = 0; i < 2500; i++)
+            InsertEventWithTimestamp(DateTimeOffset.UtcNow);
+
+        var purge = CreatePurgeService(CeilingOptions(1000, maxStorageMb: 0));
+
+        Assert.True(purge.RunPurge());
+        Assert.Equal(1500, GetEventCount());
+    }
+
+    /// <summary>
+    /// Accumulates measurements on one counter instrument for as long as it is alive.
+    /// </summary>
+    /// <remarks>
+    /// A <c>Counter&lt;long&gt;</c> is not observable, so a listener started after the fact sees
+    /// nothing — the probe must be constructed BEFORE the code under test runs. Tests in this
+    /// class run sequentially (xUnit parallelises across classes, not within one) and no other
+    /// class in the assembly touches this instrument, so the process-global meter is safe to
+    /// read absolutely here.
+    /// </remarks>
+    private sealed class CounterProbe : IDisposable
+    {
+        private readonly System.Diagnostics.Metrics.MeterListener _listener = new();
+        private long _total;
+
+        public CounterProbe(string instrumentName)
+        {
+            _listener.InstrumentPublished = (instrument, listener) =>
+            {
+                if (instrument.Name == instrumentName)
+                    listener.EnableMeasurementEvents(instrument);
+            };
+            _listener.SetMeasurementEventCallback<long>(
+                (_, measurement, _, _) => Interlocked.Add(ref _total, measurement));
+            _listener.Start();
+        }
+
+        /// <summary>Measurements recorded since construction.</summary>
+        public long Total => Interlocked.Read(ref _total);
+
+        public void Dispose() => _listener.Dispose();
+    }
+
     /// <summary>
     /// Minimal <see cref="ILogger{TCategoryName}"/> that records level + rendered message.
     /// The purge service has no other observable output for a log-only change, and the

tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogOptionsValidatorTests.cs

@@ -47,4 +47,51 @@ public class SiteEventLogOptionsValidatorTests
         Assert.True(result.Failed);
         Assert.Contains("MaxQueryPageSize", result.FailureMessage);
     }
+
+    // ── Register row 34: the per-tick purge ceiling ──
+
+    [Fact]
+    public void NegativeMaxPurgedRowsPerTick_IsRejected()
+    {
+        // 0 already means "unlimited" explicitly, so a negative value is a typo, not a synonym.
+        // Accepting it would silently remove the only bound on a purge tick's CDC oplog burst.
+        var result = Validate(new SiteEventLogOptions { MaxPurgedRowsPerTick = -1 });
+
+        Assert.True(result.Failed);
+        Assert.Contains("MaxPurgedRowsPerTick", result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroMaxPurgedRowsPerTick_IsAccepted()
+    {
+        // The documented escape hatch back to unbounded purges.
+        var result = Validate(new SiteEventLogOptions { MaxPurgedRowsPerTick = 0 });
+
+        Assert.True(result.Succeeded, result.FailureMessage);
+    }
+
+    [Fact]
+    public void ZeroPurgeCatchUpInterval_IsRejected()
+    {
+        // It is a Task.Delay period on the purge loop; zero would spin a ceiling-limited node.
+        var result = Validate(new SiteEventLogOptions { PurgeCatchUpInterval = TimeSpan.Zero });
+
+        Assert.True(result.Failed);
+        Assert.Contains("PurgeCatchUpInterval", result.FailureMessage);
+    }
+
+    [Fact]
+    public void PurgeCatchUpInterval_LongerThanPurgeInterval_IsRejected()
+    {
+        // The catch-up is an ACCELERATED retry; a longer value would make hitting the ceiling
+        // delay convergence instead of hastening it — the exact inverse of its purpose.
+        var result = Validate(new SiteEventLogOptions
+        {
+            PurgeInterval = TimeSpan.FromMinutes(10),
+            PurgeCatchUpInterval = TimeSpan.FromHours(1),
+        });
+
+        Assert.True(result.Failed);
+        Assert.Contains("PurgeCatchUpInterval", result.FailureMessage);
+    }
 }