From 7b95ef375205d7a3c99b634f1d1416c2191db589 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Mon, 13 Jul 2026 10:03:33 -0400 Subject: [PATCH] fix(kpi): backfill skips or shrinks to the rollup watermark on failover restarts (plan R2-04 T4) --- docs/requirements/Component-KpiHistory.md | 2 +- .../KpiHistoryRecorderActor.cs | 118 +++++++++++++++--- .../KpiHistoryRecorderActorTests.cs | 106 +++++++++++++++- 3 files changed, 204 insertions(+), 22 deletions(-) diff --git a/docs/requirements/Component-KpiHistory.md b/docs/requirements/Component-KpiHistory.md index 86099993..d553856a 100644 --- a/docs/requirements/Component-KpiHistory.md +++ b/docs/requirements/Component-KpiHistory.md @@ -90,7 +90,7 @@ A timer fires every `SampleInterval` (default 60s; an immediate first tick prime **Hourly rollup tick.** A third timer (`kpi-rollup`) fires every `RollupInterval` (default 1h). On each tick the recorder opens a per-tick DI scope and calls `IKpiHistoryRepository.FoldHourlyRollupsAsync` over the window `[TruncateToHour(now) − RollupLookbackHours, TruncateToHour(now))` — an **exclusive** upper bound at the current hour start, so the in-progress hour is never folded. The fold groups raw `KpiSample` rows by `(series, hour)`, applies the per-metric aggregation intent (sum for rate metrics, last-value for gauges — see Query), and **idempotently upserts** each `(Source, Metric, Scope, ScopeKey, HourStartUtc)` row. Re-folding the trailing lookback window (default 3h) rather than only the last hour is the self-heal for a **singleton-failover-missed tick**: because the upsert overwrites the aggregate in place, re-running the same window produces identical values (no double-count), so any complete hour a missed tick skipped is refilled on the next fold. A `_rollupInFlight` guard (mirroring `_sampleInFlight`) skips a tick while a prior fold is still running so two idempotent upserts never race on overlapping rows. -**One-shot backfill.** So the 30 d / 90 d charts are not blank until enough wall-clock passes, the recorder runs a **single** backfill on start (`kpi-rollup-backfill` timer) that folds the full retention window once, `[TruncateToHour(now) − RetentionDays, TruncateToHour(now))`, reusing the same `FoldHourlyRollupsAsync` path. A `_backfillDone` latch makes it at-most-once per actor lifetime, and it defers (re-arms) while a periodic fold is in flight; being an idempotent upsert, it is cheap and safe to re-run on a later failover restart. +**One-shot backfill.** So the 30 d / 90 d charts are not blank until enough wall-clock passes, the recorder runs a backfill on start (`kpi-rollup-backfill` timer) that folds the retention window, reusing the same `FoldHourlyRollupsAsync` path. The `FoldHourlyRollupsAsync` repository call materializes its whole window in memory, so the backfill folds the window in **≤24 h slices, oldest-first** — one bounded slice per fold — rather than handing the fold the full 90-day window at once (which at fleet volume is tens of millions of rows in one tracked pass). Between slices it lowers its in-flight guard so periodic folds **interleave** instead of stalling behind the historical pass; a periodic fold and a backfill slice strictly alternate through the single `_rollupInFlight`/`_backfillInFlight` gate pair, so the two idempotent upserts never race on overlapping rows. On a **failover restart** the plan first consults the `IKpiHistoryRepository.GetLatestRollupHourAsync` watermark: if the newest folded hour is already within `RollupLookbackHours` of now (the common case) the whole historical pass is **skipped** — the periodic fold's lookback covers the tail — otherwise the window is **shrunk to the un-rolled tail** `[watermark, TruncateToHour(now))` (re-folding the watermark hour itself as an idempotent safety margin) instead of the full `RetentionDays` floor. A `_backfillDone` latch makes the whole pass at-most-once per actor lifetime, and it defers (re-arms) while a periodic fold is in flight. Being an idempotent upsert, it is safe (idempotent) to re-run on a later failover restart; the watermark fast-path makes the failover re-run cheap as well. **Retention.** A daily purge timer (`PurgeInterval`, default 24h) now runs **both** purges: raw `KpiSample` rows older than `RetentionDays` (default 90) via `IKpiHistoryRepository.PurgeOlderThanAsync`, **and** hourly rollup rows older than `RollupRetentionDays` (default 365) via `IKpiHistoryRepository.PurgeRollupsOlderThanAsync` (a one-hour-sliced batched DELETE mirroring the raw purge). Rollups deliberately **outlive** raw samples so long-range trend charts have data after the raw rows are purged; the options validator enforces `RollupRetentionDays >= RetentionDays` so a window never falls into a hole where raw is purged but no rollup exists. diff --git a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs index 1dca2278..1624b55a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs @@ -186,6 +186,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers Receive(_ => HandleRollupTick()); Receive(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault) Receive(_ => HandleBackfillTick()); + Receive(HandleBackfillPlan); Receive(_ => { _backfillInFlight = false; // release the fold path BETWEEN slices (R1) @@ -546,26 +547,22 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers if (_backfillSlices.Count == 0) { - // Build the slice plan once: the full raw-retention window in bounded 24 h folds, - // oldest-first. Never hand the repository fold the whole window at once (R1). - var toHourUtc = TruncateToHour(DateTime.UtcNow); - var fromHourUtc = toHourUtc - TimeSpan.FromDays(_options.RetentionDays); - EnqueueBackfillSlices(fromHourUtc, toHourUtc); + // No slice plan yet: resolve the rollup watermark off the actor thread, then plan on + // the actor thread (Receive). Raise the guard so periodic ticks defer + // while planning is in flight; the plan handler lowers it. On a failover restart the + // watermark lets the plan SKIP or SHRINK to the un-rolled tail instead of re-folding + // the whole raw-retention window (arch-review 04 round 2, R1). + _backfillInFlight = true; + var planCancellationToken = _shutdownCts?.Token ?? CancellationToken.None; - if (_backfillSlices.Count == 0) - { - // Degenerate window (no whole hours to fold) — nothing to backfill. - _backfillDone = true; - return; - } - - _logger.LogInformation( - "KPI rollup backfill starting — folding existing samples in [{FromHour:o}, {ToHour:o}) " - + "({RetentionDays}-day window) across {SliceCount} slice(s) of up to {SliceHours} h.", - fromHourUtc, toHourUtc, _options.RetentionDays, _backfillSlices.Count, BackfillSliceWidth.TotalHours); - - // Start draining via the timer (uniform path with the between-slice re-arm). - Timers.StartSingleTimer(BackfillTimerKey, BackfillTick.Instance, TimeSpan.Zero); + RunBackfillPlanPass(planCancellationToken).PipeTo( + Self, + success: watermark => new BackfillPlan(watermark), + failure: ex => + { + _logger.LogError(ex, "KPI rollup backfill plan pass faulted unexpectedly."); + return new BackfillPlan(null); // null watermark ⇒ plan the full retention window + }); return; } @@ -605,6 +602,82 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers } } + /// + /// Plans the backfill from the resolved rollup watermark (actor thread). The window floor is + /// the newer of the watermark (re-folding the watermark hour itself is a cheap idempotent + /// safety margin) and the raw-retention floor; a null watermark (fresh install) keeps the full + /// retention window. If the floor is already within + /// of now the periodic fold's lookback + /// covers everything from there forward, so the whole historical pass is skipped (the failover + /// fast-path); otherwise the tail [floor, TruncateToHour(now)) is enumerated into + /// bounded slices and draining is self-armed. Either way the plan-pass guard is lowered here + /// (arch-review 04 round 2, R1). + /// + private void HandleBackfillPlan(BackfillPlan plan) + { + var toHourUtc = TruncateToHour(DateTime.UtcNow); + var retentionFloor = toHourUtc - TimeSpan.FromDays(_options.RetentionDays); + // Shrink to the un-rolled tail; re-fold the watermark hour itself (idempotent) as a safety + // margin. A null watermark (fresh install) keeps the full retention window. + var floor = plan.Watermark is { } w && w > retentionFloor ? w : retentionFloor; + + if (floor >= toHourUtc - TimeSpan.FromHours(_options.RollupLookbackHours)) + { + // Failover fast-path: rollups are already current — the periodic fold's lookback + // window covers everything from the watermark forward, so the full historical pass is + // skipped entirely (arch-review 04 round 2, R1). + _backfillInFlight = false; + _backfillDone = true; + _logger.LogInformation( + "KPI rollup backfill skipped — rollups current through {Watermark:o}.", plan.Watermark); + return; + } + + EnqueueBackfillSlices(floor, toHourUtc); + _backfillInFlight = false; + + if (_backfillSlices.Count == 0) + { + // Degenerate window (no whole hours to fold) — nothing to backfill. + _backfillDone = true; + return; + } + + _logger.LogInformation( + "KPI rollup backfill starting — folding [{FromHour:o}, {ToHour:o}) across {SliceCount} " + + "slice(s) of up to {SliceHours} h (watermark {Watermark:o}).", + floor, toHourUtc, _backfillSlices.Count, BackfillSliceWidth.TotalHours, plan.Watermark); + + // Start draining via the timer (uniform path with the between-slice re-arm). + Timers.StartSingleTimer(BackfillTimerKey, BackfillTick.Instance, TimeSpan.Zero); + } + + /// + /// Resolves the newest folded rollup hour (the backfill watermark) off the actor thread in a + /// fresh DI scope. Never faults — on cancellation or error it returns null, which the + /// plan treats as "no rollups", keeping the full retention window (best-effort, mirrors the + /// other passes). + /// + private async Task RunBackfillPlanPass(CancellationToken cancellationToken) + { + try + { + await using var scope = _serviceProvider.CreateAsyncScope(); + var repository = scope.ServiceProvider.GetRequiredService(); + return await repository.GetLatestRollupHourAsync(cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown interrupted the lookup; a fresh active node re-plans. Not a failure. + return null; + } + catch (Exception ex) + { + _logger.LogError(ex, "KPI rollup backfill watermark lookup failed; planning the full retention window."); + return null; + } + } + /// /// Runs one backfill fold slice: opens a DI scope, resolves the repository, and folds the /// complete hours in [, ) into @@ -706,6 +779,13 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers private BackfillTick() { } } + /// + /// Piped-back result of the backfill plan pass: the newest folded rollup hour (the watermark), + /// or null when no rollups exist. Consumed on the actor thread to skip or shrink the + /// backfill to the un-rolled tail (arch-review 04 round 2, R1). + /// + internal sealed record BackfillPlan(DateTime? Watermark); + /// /// Piped-back completion of ONE backfill fold slice; lets the slice run off the actor thread /// and, on the actor thread, lowers the _backfillInFlight guard BETWEEN slices — then diff --git a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs index 392cdccc..a9f31382 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs @@ -112,6 +112,8 @@ public class KpiHistoryRecorderActorTests : TestKit private DateTime? _foldToHourUtc; private int _foldCount; private readonly List<(DateTime FromHourUtc, DateTime ToHourUtc)> _foldWindows = new(); + private DateTime? _latestRollupHour; + private int _latestRollupHourQueryCount; public IReadOnlyList Recorded { @@ -200,8 +202,27 @@ public class KpiHistoryRecorderActorTests : TestKit return Task.CompletedTask; } - public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) => - Task.FromResult(null); + /// Watermark the fake returns from (settable). + public DateTime? LatestRollupHour + { + get { lock (_gate) { return _latestRollupHour; } } + set { lock (_gate) { _latestRollupHour = value; } } + } + + /// Number of times the backfill plan pass queried the rollup watermark. + public int LatestRollupHourQueryCount + { + get { lock (_gate) { return _latestRollupHourQueryCount; } } + } + + public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) + { + lock (_gate) + { + _latestRollupHourQueryCount++; + return Task.FromResult(_latestRollupHour); + } + } } /// @@ -985,4 +1006,85 @@ public class KpiHistoryRecorderActorTests : TestKit duration: TimeSpan.FromSeconds(3), interval: TimeSpan.FromMilliseconds(50)); } + + [Fact] + public void Backfill_Skips_WhenRollupsAlreadyCurrent() + { + // Watermark within RollupLookbackHours of now (the common failover case): the backfill + // must be skipped entirely — ZERO folds, the done-latch set — because the periodic fold's + // lookback window already covers everything from the watermark forward (arch-review 04 + // round 2, R1). + var repository = new RecordingRepository(); + var sp = BuildServiceProvider(repository, new HealthySource()); + var actor = CreateActor(sp, new KpiHistoryOptions + { + SampleInterval = TimeSpan.FromHours(1), + PurgeInterval = TimeSpan.FromHours(1), + RollupInterval = TimeSpan.FromHours(1), + RetentionDays = 90, + RollupLookbackHours = 3, + }); + + // Rollups current through one hour ago — well inside the 3 h lookback tail. + var nowHour = new DateTime( + DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day, + DateTime.UtcNow.Hour, 0, 0, DateTimeKind.Utc); + repository.LatestRollupHour = nowHour - TimeSpan.FromHours(1); + + actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance); + + // The plan pass must consult the watermark… + AwaitAssert( + () => Assert.True(repository.LatestRollupHourQueryCount >= 1), + duration: TimeSpan.FromSeconds(3), + interval: TimeSpan.FromMilliseconds(50)); + + // …and then skip: no slices are ever folded. + Thread.Sleep(300); + Assert.Equal(0, repository.FoldCount); + + // A second BackfillTick is a no-op (the done-latch is set). + actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance); + Thread.Sleep(300); + Assert.Equal(0, repository.FoldCount); + } + + [Fact] + public void Backfill_ShrinksToWatermark_InsteadOfRetentionFloor() + { + // RetentionDays = 90 but the watermark is only 2 days old: the first slice must start at + // the WATERMARK hour (inclusive — re-folding the newest already-folded hour is a cheap + // idempotent safety margin), NOT 90 days ago, and there must be <= 3 slices total + // (arch-review 04 round 2, R1). + var repository = new RecordingRepository(); + var sp = BuildServiceProvider(repository, new HealthySource()); + var actor = CreateActor(sp, new KpiHistoryOptions + { + SampleInterval = TimeSpan.FromHours(1), + PurgeInterval = TimeSpan.FromHours(1), + RollupInterval = TimeSpan.FromHours(1), + RetentionDays = 90, + RollupLookbackHours = 3, + }); + + var nowHour = new DateTime( + DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day, + DateTime.UtcNow.Hour, 0, 0, DateTimeKind.Utc); + var watermark = nowHour - TimeSpan.FromDays(2); + repository.LatestRollupHour = watermark; + + actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance); + + AwaitAssert( + () => + { + var windows = repository.FoldWindows; + Assert.NotEmpty(windows); + // Shrunk to the un-rolled tail: first slice starts AT the watermark, not 90 d ago. + Assert.Equal(watermark, windows[0].FromHourUtc); + Assert.True(windows.Count <= 3, $"expected <= 3 slices from the 2-day tail, got {windows.Count}"); + }, + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(50)); + } }