diff --git a/CLAUDE.md b/CLAUDE.md index fd96eff5..d421103a 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -83,7 +83,7 @@ spec for each is `docs/requirements/Component-.md`, and `README.md` carrie - `site_events.id` changed from autoincrement INTEGER to an application-minted **GUID**. Last-writer-wins keys on the primary key, so two nodes independently minting `id=1,2,3…` would destroy each other's events rather than merge them. The event-log read path uses a composite `(timestamp, id)` keyset cursor with an **opaque string** continuation token; `EventLogEntry.Id` and both `ContinuationToken`s are `string`/`string?` on the site↔central Akka DTOs. - `ScadaBridge:OperationTracking:ConnectionString`, `ScadaBridge:SiteEventLog:DatabasePath` and — as of Phase 2 — `ScadaBridge:StoreAndForward:SqliteDbPath` + `ScadaBridge:Database:SiteDbPath` are all **migration-only** — nothing reads them but `SiteLocalDbLegacyMigrator`, which copies a pre-Phase-1 file in once (deterministic `mig-{NodeName}-{legacyId}` event ids, `INSERT OR IGNORE`, runs AFTER `RegisterReplicated` so migrated rows replicate) and renames it `.migrated`. Delete the keys once a node has migrated. - This incidentally fixes a data-loss bug: both legacy databases defaulted to CWD-relative paths **outside** the mounted volume and were discarded on every container recreate. - - **Replication is default-OFF and opt-in** via `LocalDb:Replication:PeerAddress` + a matching `ApiKey` on both nodes. `LocalDbSyncAuthInterceptor` is **fail-closed**: no configured key means no sync stream is accepted at all, so a key typo does not degrade to unauthenticated replication — the pair simply stops converging. The sync endpoint shares the existing site gRPC h2c listener (8083); no new port. Rig posture: **site-a replicated, site-b/site-c deliberately not**, so both states are proven side-by-side. Status surfaces on the site health report as `LocalDbReplicationConnected` / `LocalDbOplogBacklog` (both nullable — null means "no data", NOT "disconnected with an empty backlog") and as `localdb_*` Prometheus series. **A transient backlog spike is expected daily and is not a fault:** the `site_events` retention purge (and the storage-cap trim) is CDC-captured like any other write — there is deliberately no purge-exemption path — so the backlog jumps by the deleted batch size and drains as the peer acks it; the purge logs an Information note saying so on a replication-enabled node (`SiteEventLogReplicationCheck`, supplied by the Host from `SiteLocalDbSetup.ReplicationIsConfigured` so SiteEventLogging never reads LocalDb config itself). Healthy = `LocalDbReplicationConnected` stays true and the backlog returns to ~0; see `docs/deployment/topology-guide.md` → *Reading the replication backlog*. Note `ZbTelemetryOptions.Meters` is an **allowlist** (`SiteServiceRegistration.ObservedMeters`); an unlisted meter exports nothing, silently. + - **Replication is default-OFF and opt-in** via `LocalDb:Replication:PeerAddress` + a matching `ApiKey` on both nodes. `LocalDbSyncAuthInterceptor` is **fail-closed**: no configured key means no sync stream is accepted at all, so a key typo does not degrade to unauthenticated replication — the pair simply stops converging. The sync endpoint shares the existing site gRPC h2c listener (8083); no new port. Rig posture: **site-a replicated, site-b/site-c deliberately not**, so both states are proven side-by-side. Status surfaces on the site health report as `LocalDbReplicationConnected` / `LocalDbOplogBacklog` (both nullable — null means "no data", NOT "disconnected with an empty backlog") and as `localdb_*` Prometheus series. **A transient backlog spike is expected daily and is not a fault:** the `site_events` retention purge (and the storage-cap trim) is CDC-captured like any other write — there is deliberately no purge-exemption path — so the backlog jumps by the deleted batch size and drains as the peer acks it; the purge logs an Information note saying so on a replication-enabled node (`SiteEventLogReplicationCheck`, supplied by the Host from `SiteLocalDbSetup.ReplicationIsConfigured` so SiteEventLogging never reads LocalDb config itself). Healthy = `LocalDbReplicationConnected` stays true and the backlog returns to ~0; see `docs/deployment/topology-guide.md` → *Reading the replication backlog*. **That spike is PACED as of row 34 (2026-08-15)** — `ScadaBridge:SiteEventLog:MaxPurgedRowsPerTick` (default **250 000**, `0` = unlimited escape hatch) is a per-tick delete ceiling **shared** by the retention purge and the cap trim, and a ceiling-limited tick reschedules after `PurgeCatchUpInterval` (default **5 min**) instead of the daily `PurgeInterval` so retention still converges (~11 ticks ≈ 55 min for the worst case, vs 11 days without the acceleration). Measured worst case, unpaced: a 1 GB event log holds **~2.73 M rows at 393 B/row**, and expiring all of them in one tick took 285 s and queued **2 731 947 oplog rows — 2.7x LocalDb's `MaxOplogRows` default of 1 000 000**, i.e. enough to make the maintenance tick prune the oplog and flag a **full ten-table snapshot resync**. **The subtlety that makes pacing the right fix rather than a bigger `MaxOplogRows`:** the *resident* backlog is `produced − drained`, so a healthy peer blunts it — on a loopback pair the same 2.73 M burst peaked at only **391 k** and drained in 6.0 min. The overrun is therefore **conditional on peer throughput** (breached outright with the peer down; breached at ~1.5 M if the peer runs at half loopback speed), and a 250 k ceiling makes the outcome independent of the peer instead. Drain speed itself is not a bottleneck (7.6–19.8 k tombstones/s measured), so a full 250 k tick clears far inside row 33's 15-minute sustained dwell — verified at 2x a tick (500 k drained in 46 s). The purge loop is now a delay loop rather than a `PeriodicTimer`, so its cadence anchors to the END of the previous purge. Ceiling hits log a rate-limited (15 min) Warning and always increment `scadabridge.site.event_log.purge_ceiling_hits` — read the counter's shape: a run that stops is a backlog converging, hits that never stop mean the ceiling is mis-sized for that site's event rate. Measurement + decision record: `docs/plans/2026-08-15-site-events-purge-burst-measurement.md`. Note `ZbTelemetryOptions.Meters` is an **allowlist** (`SiteServiceRegistration.ObservedMeters`); an unlisted meter exports nothing, silently. - Design: **scadaproj** `docs/plans/2026-07-19-scadabridge-localdb-design.md` (that doc lives in the umbrella repo, not here); Phase 1 plan + Phase 2 gate are here under `docs/plans/`. Phase 2 **deleted** `SiteReplicationActor`, its `ReplicationMessages`, StoreAndForward's `ReplicationService`, and `StoreAndForwardStorage.ReplaceAllAsync` — do not reintroduce them: - CDC replication does all three jobs now: config deploys reach the standby as ordinary row changes — **the standby makes no fetch at all** during a deploy (`SiteReconciliationActor`'s node-STARTUP fetch when central reports gaps is a different, surviving path) — and buffer mutations replicate via triggers on `sf_messages`. `ReplaceAllAsync` was a destructive delete-all-then-insert-all resync and is **unsafe to reintroduce**: a mass DELETE on a replicated table would be captured and shipped to the peer. LocalDb's snapshot resync merges per row under LWW and never deletes, which is also why the old N1 directional-authority guard is gone — there is no wipe left to gate. diff --git a/docs/deployment/topology-guide.md b/docs/deployment/topology-guide.md index 8836cd8c..b2e69bff 100644 --- a/docs/deployment/topology-guide.md +++ b/docs/deployment/topology-guide.md @@ -223,19 +223,35 @@ one of the ten replicated tables, and its retention DELETE is captured by CDC ex ordinary write — by design, since LocalDb Phase 2 there is deliberately no purge-exemption path (the same property that makes a mass DELETE dangerous, which is why `ReplaceAllAsync` was deleted rather than reinstated). One oplog row is therefore queued per deleted event, and the backlog jumps -by the size of the day's expired batch. +by the size of the batch the tick deleted. + +**The spike is bounded per tick** (register row 34, measured: +`docs/plans/2026-08-15-site-events-purge-burst-measurement.md`). One purge tick deletes at most +`ScadaBridge:SiteEventLog:MaxPurgedRowsPerTick` rows — default **250 000**, across the retention +purge and the cap trim combined — so the height of the spike is a property of the *configuration*, +not of how much history had expired. That matters because the burst used to be unbounded: a full +1 GB event log holds ~2.73 million rows, and expiring all of them at once queued 2 731 947 oplog +rows, **2.7x LocalDb's `MaxOplogRows` (default 1 000 000)** — enough for the maintenance tick to +prune the oplog and flag a *full ten-table snapshot resync*. A healthy peer blunts that (the oplog +drains while the purge runs — measured peak was 391 000, not 2.73 M), which is precisely the problem: +the old safety margin was a function of how fast the peer happened to be, and vanished when the peer +was down. Set `MaxPurgedRowsPerTick` to `0` to restore the old unbounded behaviour. - **When.** Every `ScadaBridge:SiteEventLog:PurgeInterval` (default **24 h**), plus once at - startup. The timer is anchored to the **active node's process start**, not to a wall-clock hour, - so the burst lands at a different time of day after each failover or restart — do not expect it - at a fixed hour. The storage-cap trim (default 1 GB) can produce the same shape off-schedule, and - is usually the larger of the two. + startup — **except** after a tick that hit the ceiling, which schedules the next tick after + `ScadaBridge:SiteEventLog:PurgeCatchUpInterval` (default **5 minutes**) instead. So a large + backlog appears as a *run* of bounded spikes minutes apart rather than one huge one; a + worst-case 2.73 M-row backlog converges in ~11 ticks ≈ 55 minutes. The cadence is anchored to the + **end of the previous purge**, not to a wall-clock hour, so the burst lands at a different time of + day after each failover or restart — do not expect it at a fixed hour. The storage-cap trim + (default 1 GB) can produce the same shape off-schedule, and draws from the same per-tick budget. - **Where it shows.** Only on a node with replication configured — the rig's site-a. site-b/site-c have no capture triggers at all and report no backlog for a purge. - **What healthy looks like.** `LocalDbReplicationConnected` stays **true** across the spike, and - the backlog drains back to ~0 as the peer acks the batch — within seconds to a couple of minutes - depending on batch size (delta messages are bounded by `LocalDb:Replication:MaxBatchBytes`, - default 2 MB, and secondarily by `MaxBatchSize`). No dead letters, no schema-mismatch errors. + the backlog drains back to ~0 as the peer acks the batch. Measured on a loopback pair at + ~10 800 tombstones/second, so a full 250 000-row tick drains in well under a minute (delta + messages are bounded by `LocalDb:Replication:MaxBatchBytes`, default 2 MB, and secondarily by + `MaxBatchSize`). No dead letters, no schema-mismatch errors. - **What is actually wrong.** `LocalDbReplicationConnected` **false** while the backlog climbs, a backlog that keeps rising across successive readings rather than draining, or a backlog that never returns near zero between bursts. Those point at the sync stream — an ApiKey mismatch @@ -248,13 +264,30 @@ logs an Information line next to the purge count: ``` Purged 41230 events older than 30 days Purged 41230 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; +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 250000 rows, so the spike is bounded by that number rather than by how much +history had expired. ``` An unreplicated node logs only the first line. If a backlog spike has no such line near it in the active node's log, the purge is *not* the explanation and the spike is worth investigating. +**A tick that hit the ceiling adds a Warning**, rate-limited to one line per 15 minutes so a +converging backlog does not flood the log: + +``` +Event log purge stopped at its per-tick ceiling of 250000 rows (250000 deleted); expired events +remain and the next purge runs in 00:05:00 instead of 1.00:00:00. +``` + +Every hit — throttled line or not — increments the counter +`scadabridge_site_event_log_purge_ceiling_hits`. **Read the counter's shape, not a single hit:** a +short run of hits that stops is a backlog converging (normal, e.g. the first purge after enabling +replication on a long-running site); hits that never stop mean the site writes events faster than +one paced tick can purge, i.e. `MaxPurgedRowsPerTick` or `PurgeCatchUpInterval` is mis-sized for +that site's event rate. + **Two dashboard surfaces make this reviewable rather than a log-archaeology exercise** (register row 33): @@ -263,7 +296,13 @@ row 33): above `ScadaBridge:HealthMonitoring:LocalDbBacklogThreshold` (default **1000**) continuously for `ScadaBridge:HealthMonitoring:LocalDbBacklogSustainedFor` (default **15 minutes**). The daily purge burst described above drains in seconds to a couple of minutes, so it never reaches the - dwell and never badges the site — which is the entire reason the dwell exists. A bare threshold + dwell and never badges the site — which is the entire reason the dwell exists. **The per-tick + ceiling reinforces that rather than threatening it:** pacing lowers the spike (250 000 rows + maximum instead of an unbounded 2.7 M) at the cost of more ticks, and a 250 000-row tick was + measured draining in well under a minute — even allowing 10x for a real inter-node link that is + ~4 minutes, against a 15-minute dwell and a 5-minute gap to the next tick. (The *unpaced* 2.73 M + burst drained in 6.0 minutes on loopback: inside the dwell, but with only 2.5x of margin, so a + materially slower link would have badged.) A bare threshold would fire once a day on every replicated site and be ignored within a week. The dwell is validated to be at least one `ReportInterval`, since a shorter window could be satisfied by a single sample and would collapse back into a bare threshold. diff --git a/docs/plans/2026-08-15-site-events-purge-burst-measurement.md b/docs/plans/2026-08-15-site-events-purge-burst-measurement.md new file mode 100644 index 00000000..d8d72360 --- /dev/null +++ b/docs/plans/2026-08-15-site-events-purge-burst-measurement.md @@ -0,0 +1,214 @@ +# `site_events` purge burst — measurement + pacing decision (register row 34) + +**Date:** 2026-08-15 +**Row:** deferred-work register #34 — *"`site_events` purge burst is unbounded in the oplog and unmeasured at scale."* +**Outcome:** measured; **pacing implemented** (a per-tick delete ceiling with an accelerated +catch-up tick). One separate defect found on the storage-cap path — see *New issue* at the end. + +--- + +## 1. What the row asked + +`site_events` lives in the consolidated LocalDb file and is CDC-replicated on a replication-enabled +site. The daily retention purge (30 days) and the storage-cap trim (1 GB) slice their deletes at +1000 rows per DELETE statement, but the **whole** expired set went in one tick, so the oplog gained +one row per deleted event with no pacing. The wire path is bounded (`MaxBatchBytes` 2 MB; the rig +also pins `MaxBatchSize: 64` to bound the DB read page) but the oplog **write** was not, and nothing +had been measured. + +The design commitment that had to survive, and did: **there is no purge-exemption path.** Purge +deletes are captured like any other write — the same property that made `ReplaceAllAsync` unsafe to +reintroduce. Nothing below adds an exemption or an oplog bypass; the fix bounds how much one tick +deletes, not what gets captured. + +## 2. How it was measured + +A throwaway console harness (`dotnet` console app, scratchpad — not committed) running the **real** +`EventLogPurgeService` over a **real** `ZB.MOM.WW.LocalDb` file with `site_events` genuinely +`RegisterReplicated`, plus a two-node loopback replication pair for the drain leg: + +| Phase | What it does | +| --- | --- | +| `calibrate` | Seeds realistically-shaped rows, measures `page_count`/`freelist_count`/`page_size` delta ⇒ bytes per row ⇒ rows at the 1 GB cap. | +| `burst` | Seeds N expired rows, registers CDC, parks the oplog at "everything acked", runs one real `RunPurge()`. Records purge wall time, oplog rows, oplog payload bytes, row-version tombstones, file size before/after. | +| `cap` | Same but with rows *inside* retention and a small `MaxStorageMb`, so only the storage-cap trim runs. Run twice — replicated and not — to isolate the CDC contribution. | +| `drain` | Two full replication stacks (initiator + passive `MapZbLocalDbSync`) over Kestrel h2c on loopback, both nodes registered for `site_events`. Seeds through the normal write path so the pair converges naturally, then purges on node A and polls the O(1) backlog expression until it returns to 0 and node B's table is empty. | + +Row content mirrors the post-WP3.2 event stream: mostly short Info rows, 1-in-10 carrying a details +payload, instance ids on 2-in-3. `docker/` was not touched. + +Machine: macOS (darwin 25.5.0), .NET 10, Release build, server GC, local SSD. Loopback drain numbers +are therefore an **upper bound on throughput** — a real inter-node link is slower, and the margins +below are stated with that in mind. + +## 3. Numbers + +### 3.1 How big the worst case actually is + +``` +calibrate: 200,000 rows + bytes per row (incl. the four indexes) 393 + rows at the 1 GB storage cap 2,731,947 +``` + +The 1 GB cap is the real bound on a single retention burst: the worst realistic case is a log at the +cap in which everything has expired (a site restarted after a long outage, a clock correction, or +simply the first purge after `site_events` retention was ever satisfied). + +### 3.2 The unpaced retention burst + +| Seeded expired rows | Purge wall time | Oplog rows queued | Oplog payload | Row-version tombstones | File before → after | +| ---: | ---: | ---: | ---: | ---: | --- | +| 500,000 | 36.2 s | 500,000 | 52.0 MB | 500,000 | 254 MB → 129 MB | +| **2,731,947** (1 GB cap) | **285.1 s** | **2,731,947** | **284.1 MB** | **2,731,947** | 1.39 GB → 707 MB | + +**The headline finding:** the worst-case burst *queues* **2.7x LocalDb's `MaxOplogRows` default of +1,000,000**. Crossing that ceiling is not cosmetic — `OplogStore.EnforceCapsAsync`, which the +replication maintenance service runs every 60 s, prunes the oldest oplog rows **and sets +`needs_snapshot = 1`**, the database-wide flag. The next handshake therefore serves a **full +snapshot of all ten registered tables**. A routine retention purge could, on its own, force a full +resync of the site's entire replicated state. + +**How much of that 2.73 M is actually resident depends on the peer** — the oplog is drained +concurrently, so the *peak backlog* is the burst minus what the peer acks while the purge runs. §3.3 +measures it: on a loopback pair the peak was 391 k, not 2.73 M, because the peer kept up. That does +not make the risk theoretical, it makes it **conditional on peer speed**, which is exactly the +property a design should not depend on: + +- **peer down / disconnected** — nothing drains, the full 2.73 M is resident. Over the cap outright. +- **peer slower than ~63 % of loopback** — the production deficit outruns the drain and the peak + crosses 1 M. Arithmetic from §3.3: the purge produced 8,548 rows/s and the loopback peer consumed + 7,585 rows/s; halve the peer's rate and the peak lands ≈1.5 M. +- **peer at loopback speed** — peak 391 k, under the cap, but still 391x the row-33 badge threshold. + +Pacing removes the dependency entirely: a 250 k tick cannot exceed the cap at *any* peer speed, +including a peer that is down. + +Secondary costs at worst case: 285 s of purge (the write lock is taken and released 2,732 times, so +recorder flushes interleave — but they contend for nearly five minutes), 284 MB of oplog payload, +and 2.73 M row-version tombstones retained for `TombstoneRetention` (7 days). The file was still +707 MB *after* deleting a 1 GB event log, because the replication bookkeeping the delete generated +is itself in that file. + +### 3.3 Drain + +Two-node loopback pair, `MaxBatchSize: 64` (the rig's pin), `MaxBatchBytes` at its 2 MB default. +Both nodes registered for `site_events`; rows are seeded through the normal write path so the pair +converges naturally before the purge runs. + +| Rows | Insert leg (seed + replicate) | Purge | Backlog → 0 | **Peak backlog** | Tombstones/s | Inside the 15-min dwell? | +| ---: | ---: | ---: | ---: | ---: | ---: | --- | +| 20,000 | 1.3 s (15.0 k/s) | 0.5 s | 1.0 s | 17,424 | 19,807 | yes | +| 500,000 | 65.7 s (7.6 k/s) | 34.4 s | 46.3 s | 150,912 | 10,795 | yes | +| **2,731,947** (1 GB cap) | 996.9 s (2.7 k/s) | 319.6 s | **360.2 s** | **391,424** | 7,585 | yes (6.0 min) | + +**Drain is not the bottleneck, and a healthy peer blunts the burst.** At every size the pump +consumed tombstones almost as fast as the purge produced them, so the resident backlog stayed far +below the burst size — 391 k peak on a 2.73 M burst. The tombstone leg is roughly 4x cheaper per row +than the insert leg, as expected: a delete oplog entry carries a null `row_json`. + +**That is exactly why the unpaced burst was still unsafe**, though. The peak is +`produced − drained`, so it is a function of the *peer's* throughput, and the margin evaporates as +soon as the peer is slower than loopback or absent (see §3.2). A design whose oplog-cap safety +depends on the peer keeping up is one bad link away from a full ten-table resync. + +**Row-33 interplay (the sustained badge).** The badge needs the backlog above 1000 continuously for +15 minutes. + +- **Paced (shipped default).** A full 250 000-row tick drains in ~23 s at the measured rate; even + assuming a real link is **10x slower** than loopback it clears in ~4 minutes — comfortably inside + the dwell, with the next tick 5 minutes away. Pacing lowers each spike and lengthens *total* + purge duration, but every individual spike still drains between ticks, so the badge assumption + holds with a large margin. **Verified, not assumed:** the 500 k measurement is 2x a paced tick and + drained in 46 s. +- **Unpaced (previous behaviour).** 6.0 minutes on loopback — inside the dwell, but only 2.5x of + margin, at a peak of 391 k (391x the badge threshold). At a 3x-slower link the same burst badges. + So pacing does not merely preserve the row-33 assumption; it is what makes it robust. + +### 3.4 The storage-cap trim + +`MaxStorageMb = 64`, 400,000 rows seeded inside the retention window (so only the cap path runs): + +| Replicated? | Events deleted | Remaining | Measured size before → after | Reached the cap? | +| --- | ---: | ---: | --- | --- | +| no | 248,000 | 152,000 | 158.0 MB → 66.98 MB | **yes** (cap 67.11 MB) | +| yes | 174,000 | 226,000 | 203.6 MB → 167.7 MB | **no — stopped 2.5x over cap** | + +See *New issue* below. + +## 4. Decision — pace it + +Pacing is warranted, on the strength of §3.2: an unbounded tick can exceed `MaxOplogRows` and buy a +full ten-table snapshot resync. Drain *speed* (§3.3) was not the reason — that is healthy — but the +fact that the safety margin **depends on** drain speed is: with the peer down, the whole 2.73 M +burst is resident and the cap is breached outright. Bounding the tick makes the outcome independent +of the peer, which is the property the previous design lacked. + +**Shipped:** + +- `SiteEventLogOptions.MaxPurgedRowsPerTick`, default **250 000**, `0` = unlimited. + - Ceiling on rows deleted by **one tick**, **shared** between the retention purge and the cap + trim. Two independently-capped phases would let a tick emit twice the intended burst. + - `0` (not a negative, which the validator rejects) is the documented escape hatch back to the + pre-row-34 behaviour, byte-for-byte. + - **Why 250 000:** a 4x margin under `MaxOplogRows` = 1 000 000, and that margin is shared with + every *other* replicated table's ordinary traffic, which also occupies the same oplog. At the + measured rates one such tick costs ~17 s of purge and ~23 s of drain. +- `SiteEventLogOptions.PurgeCatchUpInterval`, default **5 minutes**. + - A tick that stopped at the ceiling schedules the next tick after this instead of + `PurgeInterval`. Without it, pacing would make retention converge in 11 **days** at the daily + cadence — a policy that no longer enforces its own window. With it, the worst case converges in + ~11 ticks ≈ **55 minutes**, every burst bounded. + - **Why 5 minutes rather than 1:** the interval must exceed the peer's drain time for one tick, or + ticks outrun the drain and the backlog accumulates across them — reintroducing the very problem + being fixed. 250 k drains in ~23 s on loopback; 5 minutes leaves ~13x headroom for a slower + real link. + - Validated `> 0` and `<= PurgeInterval` (a longer catch-up would *delay* convergence, inverting + its purpose). +- **The ceiling is not claimed when the work is finished.** An expired set that lands exactly on the + budget is done; the retention phase runs a bounded `EXISTS` probe before flagging a catch-up tick, + so a converged node never arms a 5-minute retry that deletes nothing. +- **Visibility.** A ceiling-limited tick logs a Warning naming the ceiling, the rows deleted and the + accelerated interval — **rate-limited to one line per 15 minutes**, because a large backlog + produces a run of such ticks. Every hit increments + `scadabridge.site.event_log.purge_ceiling_hits` (`ScadaBridgeTelemetry`, whose meter is already in + the site `ObservedMeters` allowlist). The counter is the unthrottled signal, and its **shape** is + the diagnosis: a run that stops = a backlog converging; hits that never stop = the site writes + events faster than one paced tick purges, so the ceiling or the catch-up interval is mis-sized. +- **Log wording updated.** The existing replication-backlog Information note now says the spike is + bounded by the ceiling rather than by how much history expired. + +**Consequence accepted:** the purge loop is now a `Task.Delay` loop rather than a `PeriodicTimer`, +because the wait is no longer a constant. The cadence is therefore anchored to the **end** of the +previous purge rather than to process start. At a daily interval and a measured worst-case paced +tick of ~17 s the drift is immaterial; the topology guide and component doc say so. + +**Explicitly not done:** no purge-exemption path, no oplog bypass, no "skip capture for deletes" +flag. Those are ruled out by design and the measurement gives no reason to revisit that. + +## 5. New issue found — the storage cap cannot be met on a replicated node with an undrained oplog + +Not row 34, and not fixed here. `EventLogPurgeService.GetDatabaseSizeBytes()` measures +`PRAGMA page_count - freelist_count` — i.e. the **whole consolidated LocalDb file**, which since +Phase 1/2 holds ten tables *plus* `__localdb_oplog` and `__localdb_row_version`. On a replicated +node the cap trim therefore fights itself: each 1000-row delete frees ~390 KB of `site_events` pages +and adds ~1000 oplog rows plus ~1000 row-version tombstones to the same file. Measured (§3.4) the +trim stopped **2.5x over a 64 MB cap** with 226 k rows still present, tripped by the existing +"size did not decrease across an iteration" guard — which is doing its job (it prevents the table +being wiped) but leaves the cap unenforced. + +Severity is moderate, not acute: with a live peer the oplog drains on ack and successive ticks +recover ground, and tombstones expire after 7 days. But two real consequences follow — a node whose +peer is down cannot enforce its event-log storage cap at all, and the "1 GB event log" budget is in +practice shared with replication bookkeeping that the event log does not own. A correct fix would +measure `site_events`' own pages (e.g. `dbstat`) rather than the file, which is a change to the +cap's semantics and deserves its own decision. Reported for the register; no row added here. + +## 6. Verification + +- Fail-before: the new behavioural tests do not compile against pre-fix `EventLogPurgeService` + (`RunPurge` was `void`); with the fix restored but `PurgeBudget` neutralised to always-unlimited, + **6 of the new tests fail** and the 19 pre-existing ones pass — so the assertions are carried by + the ceiling, not incidentally. +- `dotnet build ZB.MOM.WW.ScadaBridge.slnx` clean. +- `ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests`: 93 passed / 0 failed. diff --git a/docs/requirements/Component-SiteEventLogging.md b/docs/requirements/Component-SiteEventLogging.md index 7cacf025..7a40c737 100644 --- a/docs/requirements/Component-SiteEventLogging.md +++ b/docs/requirements/Component-SiteEventLogging.md @@ -94,12 +94,35 @@ Each event entry contains: documented trade of not configuring a peer, unchanged by this policy. - **Retention**: 30 days. A **daily background job** runs on the active node and deletes all events older than 30 days. Hard delete — no archival. Deletes are sliced into bounded 1000-row batches per DELETE statement rather than one unbounded statement, so a large expired backlog does not hold the write lock (and every concurrent recorder flush) for the duration of the purge. - **Storage cap**: A configurable maximum database size (default: 1 GB) is enforced. If the storage cap is reached before the 30-day retention window, the oldest events are purged first. This prevents disk exhaustion from alarm storms, script failure loops, or connection flapping. +- **Per-tick delete ceiling** (`MaxPurgedRowsPerTick`, default **250 000**; `0` = unlimited): one + purge tick deletes at most this many rows, **shared** by the retention purge and the storage-cap + trim — two independently-capped phases would let a tick emit twice the intended burst. A tick that + stops at the ceiling schedules the next tick after `PurgeCatchUpInterval` (default **5 minutes**) + instead of `PurgeInterval`, so retention still converges: a worst-case backlog drains over + ~11 accelerated ticks (≈55 minutes) rather than 11 days at the daily cadence, with every + individual burst bounded. The ceiling exists because the purge is CDC-captured (next bullet) and + an unbounded tick could exceed LocalDb's oplog budget; measured evidence and the chosen defaults + are in `docs/plans/2026-08-15-site-events-purge-burst-measurement.md`. A ceiling-limited tick logs + a Warning (rate-limited to one line per 15 minutes) and increments + `scadabridge.site.event_log.purge_ceiling_hits` — the counter is the unthrottled series that + distinguishes a backlog converging from a site whose event rate outruns one paced tick. + Consequence of the accelerated retry: the purge loop is a delay loop, not a `PeriodicTimer`, so + the cadence is anchored to the **end** of the previous purge rather than to process start. - **Purge deletes are CDC-captured on a replicated site — the resulting backlog spike is expected.** Retention and cap deletes are ordinary row changes on a replicated table; there is deliberately no purge-exemption path (LocalDb Phase 2 — CDC does all three jobs, and a table-wide exemption is the same mechanism that made `ReplaceAllAsync` unsafe). One oplog row is queued per deleted event, so the site health report's `LocalDbOplogBacklog` (Prometheus `localdb_oplog_depth`) jumps by the size - of the batch at purge time and drains as the peer acks it. **Healthy signature:** + of the batch at purge time and drains as the peer acks it. **The height of that jump is bounded by + `MaxPurgedRowsPerTick`, not by how much history expired** — measured (register row 34): a full 1 GB + event log holds ~2.73 million rows at 393 bytes/row, and purging it unbounded queued 2 731 947 + oplog rows, **2.7x LocalDb's `MaxOplogRows` default of 1 000 000**, which would make the + maintenance tick prune the oplog and flag a full ten-table snapshot resync. How much of that is + *resident* depends on the peer — the oplog drains concurrently, and a loopback peer held the same + burst to a 391 000 peak — so the overrun is conditional on peer throughput and unconditional when + the peer is down. Bounding the tick is what makes the outcome independent of the peer. Pacing the + tick, not exempting the purge from capture, is the fix — the exemption remains ruled out. + **Healthy signature:** `LocalDbReplicationConnected` stays true across the spike and the backlog returns to ~0; a backlog that keeps climbing, or climbs while `LocalDbReplicationConnected` is false, is a genuine replication fault and not the purge. The purge logs an Information line naming the row count and diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index cce04d8a..12d808e3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -111,6 +111,21 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("scadabridge.central.transport.import_sessions.rejected", unit: "1", description: "Bundle-import sessions rejected because the concurrent-import-session cap was exhausted."); + /// + /// Register row 34: incremented once per site-event-log purge tick that stopped at + /// SiteEventLogOptions.MaxPurgedRowsPerTick 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. + /// + private static readonly Counter _siteEventLogPurgeCeilingHits = + Meter.CreateCounter("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 ---------------- /// @@ -272,6 +287,13 @@ public static class ScadaBridgeTelemetry /// public static void RecordImportSessionRejected() => _transportImportSessionsRejected.Add(1); + /// + /// 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. + /// + public static void RecordSiteEventLogPurgeCeilingHit() => _siteEventLogPurgeCeilingHits.Add(1); + /// /// Records that a central→site gRPC subscriber stream opened (increments /// scadabridge.site.stream.subscribers.active). Renamed from diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs index 7519c77b..fad27c36 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs +++ b/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 /// Number of events deleted per retention-purge batch — same size as . private const int RetentionPurgeBatchSize = 1000; + /// Minimum spacing between ceiling-reached log lines; see . + private static readonly TimeSpan CeilingLogInterval = TimeSpan.FromMinutes(15); + + /// + /// 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. + /// + private DateTimeOffset? _lastCeilingLogUtc; + private readonly SiteEventLogger _eventLogger; private readonly SiteEventLogOptions _options; private readonly ILogger _logger; @@ -100,25 +110,50 @@ public class EventLogPurgeService : BackgroundService } /// + /// + /// Deliberately a delay loop rather than a : the wait between + /// ticks is no longer a constant. A tick that stopped at + /// still has expired rows to delete, + /// and waiting a full (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. + /// 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(); } } /// Runs a single purge cycle: first by retention age, then by storage cap. - internal void RunPurge() + /// + /// true when the tick stopped at + /// with rows still to purge, so the caller should schedule an accelerated catch-up tick. + /// + 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; } } + /// + /// Emits the ceiling-reached breadcrumb, rate-limited to one line per + /// . + /// + /// + /// Rate-limited because the whole point of the ceiling is that a large backlog produces a RUN + /// of ceiling-limited ticks, one every + /// — 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 + /// (scadabridge.site.event_log.purge_ceiling_hits) carries every hit for the series that + /// distinguishes converging from never-converging. + /// + /// Rows the ceiling-limited tick deleted. + 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); + } + /// /// Deletes events older than the retention window, in bounded /// -row slices per DELETE statement rather than @@ -166,13 +245,26 @@ public class EventLogPurgeService : BackgroundService /// whole delete, blocking every concurrent /// writer-loop flush for that entire duration. /// - private void PurgeByRetention() + /// The tick's shared delete budget; slices are drawn from it. + 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 } } + /// + /// Whether at least one event older than is still present. + /// + /// + /// An existence probe rather than a COUNT(*): 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. idx_events_timestamp + /// makes this one index seek. + /// + /// The retention cutoff, ISO-8601 round-trip formatted. + /// true if expired rows remain. + 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; + }); + /// /// 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() + /// The tick's shared delete budget, already drawn down by the retention purge. + 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; }); } + + /// + /// One purge tick's shared delete allowance, drawn down by the retention purge and then the + /// storage-cap trim. + /// + /// + /// 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. + /// + /// Rows this tick may delete; 0 or less means unlimited. + private sealed class PurgeBudget(int ceiling) + { + private readonly bool _unlimited = ceiling <= 0; + + /// Rows deleted so far this tick. + public int Deleted { get; private set; } + + /// + /// 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. + /// + public bool StoppedAtCeiling { get; set; } + + /// + /// The number of rows the next DELETE statement may take: , + /// or what remains of the budget if that is smaller. 0 means the budget is spent. + /// + /// The phase's own statement batch size. + /// Rows the next statement may delete. + public int NextSlice(int batchSize) => + _unlimited ? batchSize : Math.Max(0, Math.Min(batchSize, ceiling - Deleted)); + + /// Charges against the budget. + /// Rows the statement actually deleted. + public void Record(int rows) => Deleted += rows; + } } diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptions.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptions.cs index 483ac83e..00c19ce9 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptions.cs @@ -28,6 +28,43 @@ public class SiteEventLogOptions /// Interval between purge runs; default 24 hours. public TimeSpan PurgeInterval { get; set; } = TimeSpan.FromHours(24); /// + /// Ceiling on how many site_events rows ONE purge tick may delete, across the + /// retention purge and the storage-cap trim combined. Default 250 000; 0 means + /// unlimited (the pre-pacing behaviour, kept as an escape hatch). + /// + /// + /// + /// Why a ceiling exists at all. site_events 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, + /// docs/plans/2026-08-15-site-events-purge-burst-measurement.md): 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 — 2.7x over LocalDb's MaxOplogRows default of + /// 1 000 000. Crossing that ceiling makes LocalDb's maintenance tick prune the oplog + /// and set needs_snapshot, 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 keeps retention converging. + /// + /// + /// The remainder is not dropped: a tick that hits the ceiling schedules the next tick after + /// instead of , so a backlog + /// drains over minutes rather than days while every individual burst stays bounded. + /// + /// + public int MaxPurgedRowsPerTick { get; set; } = 250_000; + /// + /// Delay before the next purge tick when the previous tick stopped at + /// ; default 5 minutes. Ignored when the ceiling was not + /// reached — an ordinary tick waits . + /// + /// + /// 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. + /// + public TimeSpan PurgeCatchUpInterval { get; set; } = TimeSpan.FromMinutes(5); + /// /// Bound on the background write queue. Default 10 000 events. /// Overflow uses BoundedChannelFullMode.DropOldest — callers never block; the /// dropped event's Task is faulted and FailedWriteCount is incremented diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptionsValidator.cs index 702ecd59..aa852b1c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogOptionsValidator.cs @@ -41,6 +41,29 @@ public sealed class SiteEventLogOptionsValidator : OptionsValidatorBase= 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 diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs index 77a47f13..a26f7cdd 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/EventLogPurgeServiceTests.cs +++ b/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 ── + + /// + /// 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. + /// + private const string CeilingLineFragment = "per-tick ceiling"; + + /// The counter incremented once per ceiling-limited tick. + private const string CeilingHitCounter = "scadabridge.site.event_log.purge_ceiling_hits"; + + /// Seeds events one second apart, all past retention. + 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(); + }); + } + + /// Options with the per-tick ceiling set and a short catch-up interval. + 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()); + } + + /// + /// Accumulates measurements on one counter instrument for as long as it is alive. + /// + /// + /// A Counter<long> 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. + /// + 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( + (_, measurement, _, _) => Interlocked.Add(ref _total, measurement)); + _listener.Start(); + } + + /// Measurements recorded since construction. + public long Total => Interlocked.Read(ref _total); + + public void Dispose() => _listener.Dispose(); + } + /// /// Minimal that records level + rendered message. /// The purge service has no other observable output for a log-only change, and the diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogOptionsValidatorTests.cs index 865eca8d..7cd7db96 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogOptionsValidatorTests.cs +++ b/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); + } }