fix(kpi): backfill skips or shrinks to the rollup watermark on failover restarts (plan R2-04 T4)

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.
 

src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs

@@ -186,6 +186,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
         Receive<RollupTick>(_ => HandleRollupTick());
         Receive<RollupComplete>(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault)
         Receive<BackfillTick>(_ => HandleBackfillTick());
+        Receive<BackfillPlan>(HandleBackfillPlan);
         Receive<BackfillSliceComplete>(_ =>
         {
             _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<BackfillPlan>). 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)
+            RunBackfillPlanPass(planCancellationToken).PipeTo(
+                Self,
+                success: watermark => new BackfillPlan(watermark),
+                failure: ex =>
                 {
-                // 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);
+                    _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
         }
     }
 
+    /// <summary>
+    /// 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
+    /// <see cref="KpiHistoryOptions.RollupLookbackHours"/> 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 <c>[floor, TruncateToHour(now))</c> 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).
+    /// </summary>
+    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);
+    }
+
+    /// <summary>
+    /// 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 <c>null</c>, which the
+    /// plan treats as "no rollups", keeping the full retention window (best-effort, mirrors the
+    /// other passes).
+    /// </summary>
+    private async Task<DateTime?> RunBackfillPlanPass(CancellationToken cancellationToken)
+    {
+        try
+        {
+            await using var scope = _serviceProvider.CreateAsyncScope();
+            var repository = scope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
+            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;
+        }
+    }
+
     /// <summary>
     /// Runs one backfill fold slice: opens a DI scope, resolves the repository, and folds the
     /// complete hours in <c>[<paramref name="fromHourUtc"/>, <paramref name="toHourUtc"/>)</c> into
@@ -706,6 +779,13 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
         private BackfillTick() { }
     }
 
+    /// <summary>
+    /// Piped-back result of the backfill plan pass: the newest folded rollup hour (the watermark),
+    /// or <c>null</c> 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).
+    /// </summary>
+    internal sealed record BackfillPlan(DateTime? Watermark);
+
     /// <summary>
     /// Piped-back completion of ONE backfill fold slice; lets the slice run off the actor thread
     /// and, on the actor thread, lowers the <c>_backfillInFlight</c> guard BETWEEN slices — then

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<KpiSample> Recorded
         {
@@ -200,8 +202,27 @@ public class KpiHistoryRecorderActorTests : TestKit
             return Task.CompletedTask;
         }
 
-        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
-            Task.FromResult<DateTime?>(null);
+        /// <summary>Watermark the fake returns from <see cref="GetLatestRollupHourAsync"/> (settable).</summary>
+        public DateTime? LatestRollupHour
+        {
+            get { lock (_gate) { return _latestRollupHour; } }
+            set { lock (_gate) { _latestRollupHour = value; } }
+        }
+
+        /// <summary>Number of times the backfill plan pass queried the rollup watermark.</summary>
+        public int LatestRollupHourQueryCount
+        {
+            get { lock (_gate) { return _latestRollupHourQueryCount; } }
+        }
+
+        public Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default)
+        {
+            lock (_gate)
+            {
+                _latestRollupHourQueryCount++;
+                return Task.FromResult(_latestRollupHour);
+            }
+        }
     }
 
     /// <summary>
@@ -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));
+    }
 }