feat(kpihistory): hourly rollup recorder tick + rollup purge + options (plan #22 T4)

Add a third Akka periodic timer (kpi-rollup) to KpiHistoryRecorderActor that folds the trailing RollupLookbackHours of raw KpiSample rows into the hourly KpiRollupHourly table via the idempotent FoldHourlyRollupsAsync upsert, with the in-progress hour excluded (toHourUtc = current hour start, exclusive). A _rollupInFlight guard coalesces overlapping ticks (mirrors _sampleInFlight), the fold is best-effort (no exception escapes a tick), and a missed/failover tick self-heals via the lookback re-fold. The daily purge now runs BOTH the raw PurgeOlderThanAsync (RetentionDays) and PurgeRollupsOlderThanAsync (RollupRetentionDays), isolated so one failure never skips the other. New KpiHistoryOptions: RollupInterval (1h), RollupLookbackHours (3), RollupRetentionDays (365), RollupThresholdHours (168, consumed by Task 5). Validator adds bounds incl. the coherence rule RollupRetentionDays >= RetentionDays so long-range trends have no data hole. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
2026-07-10 11:58:50 -04:00
parent 09f67d1c65
commit 4f145fd62a
5 changed files with 710 additions and 20 deletions
@@ -107,6 +107,10 @@ public class KpiHistoryRecorderActorTests : TestKit
        private readonly object _gate = new();
        private readonly List<KpiSample> _recorded = new();
        private DateTime? _purgeCutoff;
+        private DateTime? _rollupPurgeCutoff;
+        private DateTime? _foldFromHourUtc;
+        private DateTime? _foldToHourUtc;
+        private int _foldCount;

        public IReadOnlyList<KpiSample> Recorded
        {
@@ -120,6 +124,30 @@ public class KpiHistoryRecorderActorTests : TestKit
            get { lock (_gate) { return _purgeCutoff; } }
        }

+        /// <summary>Cut-off handed to <see cref="PurgeRollupsOlderThanAsync"/> (rollup retention).</summary>
+        public DateTime? RollupPurgeCutoff
+        {
+            get { lock (_gate) { return _rollupPurgeCutoff; } }
+        }
+
+        /// <summary>Inclusive lower bound handed to <see cref="FoldHourlyRollupsAsync"/>.</summary>
+        public DateTime? FoldFromHourUtc
+        {
+            get { lock (_gate) { return _foldFromHourUtc; } }
+        }
+
+        /// <summary>Exclusive upper bound handed to <see cref="FoldHourlyRollupsAsync"/>.</summary>
+        public DateTime? FoldToHourUtc
+        {
+            get { lock (_gate) { return _foldToHourUtc; } }
+        }
+
+        /// <summary>Number of times <see cref="FoldHourlyRollupsAsync"/> was entered.</summary>
+        public int FoldCount
+        {
+            get { lock (_gate) { return _foldCount; } }
+        }
+
        public Task RecordSamplesAsync(
            IReadOnlyCollection<KpiSample> samples, CancellationToken cancellationToken = default)
        {
@@ -140,6 +168,29 @@ public class KpiHistoryRecorderActorTests : TestKit
            lock (_gate) { _purgeCutoff = before; }
            return Task.FromResult(0);
        }
+
+        public Task FoldHourlyRollupsAsync(
+            DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default)
+        {
+            lock (_gate)
+            {
+                _foldFromHourUtc = fromHourUtc;
+                _foldToHourUtc = toHourUtc;
+                _foldCount++;
+            }
+            return Task.CompletedTask;
+        }
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetHourlySeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default)
+        {
+            lock (_gate) { _rollupPurgeCutoff = before; }
+            return Task.CompletedTask;
+        }
    }

    /// <summary>
@@ -174,6 +225,121 @@ public class KpiHistoryRecorderActorTests : TestKit

        public Task<int> PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
            Task.FromResult(0);
+
+        public Task FoldHourlyRollupsAsync(
+            DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetHourlySeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
+    }
+
+    /// <summary>
+    /// Repository fake whose <see cref="FoldHourlyRollupsAsync"/> blocks on a manual-reset gate
+    /// until the test releases it, holding a rollup fold "in flight" so an overlapping-tick
+    /// scenario can be driven deterministically. Counts how many times the fold was entered.
+    /// </summary>
+    private sealed class GatedFoldRepository : IKpiHistoryRepository
+    {
+        private readonly ManualResetEventSlim _release = new(initialState: false);
+        private int _foldCount;
+
+        /// <summary>Number of times <see cref="FoldHourlyRollupsAsync"/> has been entered.</summary>
+        public int FoldCount => Volatile.Read(ref _foldCount);
+
+        /// <summary>Releases all blocked folds so the gated passes can complete.</summary>
+        public void Release() => _release.Set();
+
+        public Task RecordSamplesAsync(
+            IReadOnlyCollection<KpiSample> samples, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetRawSeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task<int> PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
+            Task.FromResult(0);
+
+        public Task FoldHourlyRollupsAsync(
+            DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default)
+        {
+            Interlocked.Increment(ref _foldCount);
+            // Block on a threadpool thread (PipeTo runs the pass off the actor thread), holding
+            // the fold in flight until the test opens the gate.
+            return Task.Run(() => _release.Wait(cancellationToken), cancellationToken);
+        }
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetHourlySeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
+    }
+
+    /// <summary>
+    /// Repository fake whose <see cref="FoldHourlyRollupsAsync"/> throws on the first call and
+    /// succeeds (recording the window) on every subsequent call — drives a faulted first fold
+    /// followed by a healthy second fold on the same actor instance.
+    /// </summary>
+    private sealed class ThrowOnceFoldRepository : IKpiHistoryRepository
+    {
+        private readonly object _gate = new();
+        private int _foldCount;
+        private DateTime? _lastFoldToHourUtc;
+
+        /// <summary>Number of times <see cref="FoldHourlyRollupsAsync"/> has been entered (incl. the throw).</summary>
+        public int FoldCount
+        {
+            get { lock (_gate) { return _foldCount; } }
+        }
+
+        /// <summary>Exclusive upper bound of the most recent successful fold.</summary>
+        public DateTime? LastFoldToHourUtc
+        {
+            get { lock (_gate) { return _lastFoldToHourUtc; } }
+        }
+
+        public Task RecordSamplesAsync(
+            IReadOnlyCollection<KpiSample> samples, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetRawSeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task<int> PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
+            Task.FromResult(0);
+
+        public Task FoldHourlyRollupsAsync(
+            DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default)
+        {
+            lock (_gate)
+            {
+                _foldCount++;
+                if (_foldCount == 1)
+                    throw new InvalidOperationException("simulated first-fold failure");
+                _lastFoldToHourUtc = toHourUtc;
+            }
+            return Task.CompletedTask;
+        }
+
+        public Task<IReadOnlyList<KpiSeriesPoint>> GetHourlySeriesAsync(
+            string source, string metric, string scope, string? scopeKey,
+            DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) =>
+            Task.FromResult<IReadOnlyList<KpiSeriesPoint>>(Array.Empty<KpiSeriesPoint>());
+
+        public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) =>
+            Task.CompletedTask;
    }

    private IServiceProvider BuildServiceProvider(
@@ -337,4 +503,163 @@ public class KpiHistoryRecorderActorTests : TestKit
            duration: TimeSpan.FromSeconds(3),
            interval: TimeSpan.FromMilliseconds(50));
    }
+
+    // =====================================================================
+    // Hourly rollup fold (Task 4)
+    // =====================================================================
+
+    [Fact]
+    public void RollupTick_FoldsTrailingLookback_WithInProgressHourExcluded()
+    {
+        // The fold's upper bound must be the CURRENT hour start (exclusive) so the in-progress
+        // hour is never folded, and its lower bound must be RollupLookbackHours earlier so a
+        // missed/failover tick self-heals via the idempotent upsert.
+        var repository = new RecordingRepository();
+        var sp = BuildServiceProvider(repository, new HealthySource());
+        const int lookbackHours = 3;
+        var actor = CreateActor(sp, new KpiHistoryOptions
+        {
+            SampleInterval = TimeSpan.FromHours(1),
+            PurgeInterval = TimeSpan.FromHours(1),
+            RollupInterval = TimeSpan.FromHours(1),
+            RollupLookbackHours = lookbackHours,
+        });
+
+        actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+
+        AwaitAssert(
+            () =>
+            {
+                Assert.NotNull(repository.FoldToHourUtc);
+                Assert.NotNull(repository.FoldFromHourUtc);
+
+                var toHour = repository.FoldToHourUtc!.Value;
+                var fromHour = repository.FoldFromHourUtc!.Value;
+
+                // toHourUtc is the truncated current hour (in-progress hour excluded).
+                Assert.Equal(DateTimeKind.Utc, toHour.Kind);
+                Assert.Equal(0, toHour.Minute);
+                Assert.Equal(0, toHour.Second);
+                Assert.Equal(0, toHour.Millisecond);
+
+                var expectedToHour = new DateTime(
+                    DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day,
+                    DateTime.UtcNow.Hour, 0, 0, DateTimeKind.Utc);
+                // Within one hour of "now truncated" (guards a rare hour-boundary crossing mid-test).
+                Assert.True(
+                    Math.Abs((toHour - expectedToHour).TotalHours) <= 1.0,
+                    $"fold toHour {toHour:o} should be the current hour start {expectedToHour:o}");
+
+                // fromHour is exactly lookbackHours before toHour.
+                Assert.Equal(toHour - TimeSpan.FromHours(lookbackHours), fromHour);
+            },
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+    }
+
+    [Fact]
+    public void PurgeTick_PurgesBothRawSamplesAndRollups_WithTheirOwnCutoffs()
+    {
+        // The daily purge must drop raw samples at RetentionDays AND rollups at the longer
+        // RollupRetentionDays — each with its own cut-off, both best-effort.
+        var repository = new RecordingRepository();
+        var sp = BuildServiceProvider(repository, new HealthySource());
+        const int retentionDays = 90;
+        const int rollupRetentionDays = 365;
+        var actor = CreateActor(sp, new KpiHistoryOptions
+        {
+            SampleInterval = TimeSpan.FromHours(1),
+            PurgeInterval = TimeSpan.FromHours(1),
+            RetentionDays = retentionDays,
+            RollupRetentionDays = rollupRetentionDays,
+        });
+
+        actor.Tell(KpiHistoryRecorderActor.PurgeTick.Instance);
+
+        AwaitAssert(
+            () =>
+            {
+                Assert.NotNull(repository.PurgeCutoff);
+                Assert.NotNull(repository.RollupPurgeCutoff);
+
+                var expectedRaw = DateTime.UtcNow - TimeSpan.FromDays(retentionDays);
+                var expectedRollup = DateTime.UtcNow - TimeSpan.FromDays(rollupRetentionDays);
+
+                Assert.True(
+                    Math.Abs((repository.PurgeCutoff!.Value - expectedRaw).TotalMinutes) < 1.0,
+                    $"raw purge cutoff {repository.PurgeCutoff:o} should be within 1 minute of {expectedRaw:o}");
+                Assert.True(
+                    Math.Abs((repository.RollupPurgeCutoff!.Value - expectedRollup).TotalMinutes) < 1.0,
+                    $"rollup purge cutoff {repository.RollupPurgeCutoff:o} should be within 1 minute of {expectedRollup:o}");
+            },
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+    }
+
+    [Fact]
+    public void OverlappingRollupTick_WhileFirstFoldInFlight_DoesNotStartSecondFold()
+    {
+        // The _rollupInFlight guard must coalesce a tick that arrives while a prior fold is
+        // still running. With a gated fold held open, a second RollupTick must NOT spawn a
+        // second fold — so FoldHourlyRollupsAsync is entered exactly once until the gate opens.
+        var repository = new GatedFoldRepository();
+        var sp = BuildServiceProvider(repository, new HealthySource());
+        var actor = CreateActor(sp);
+
+        // First tick: raises the guard, enters the (gated, blocking) fold — held in flight.
+        actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+        AwaitAssert(
+            () => Assert.Equal(1, repository.FoldCount),
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+
+        // Second tick while the first fold is still in flight: must be skipped by the guard.
+        actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+
+        Assert.Equal(1, repository.FoldCount);
+        Thread.Sleep(300);
+        Assert.Equal(1, repository.FoldCount);
+
+        // Release the gate so the first fold completes and the guard is lowered; a fresh tick
+        // must now run a new fold (guard correctly reset, not stuck). Re-send the tick on each
+        // poll so we don't race the asynchronous RollupComplete that lowers the guard.
+        repository.Release();
+        AwaitAssert(
+            () =>
+            {
+                actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+                Assert.Equal(2, repository.FoldCount);
+            },
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+    }
+
+    [Fact]
+    public void FaultedRollupFold_DoesNotCrashActor_AndSubsequentTickStillFolds()
+    {
+        // A fold that throws must be caught (best-effort) — it must neither crash the singleton
+        // nor stop future rollup ticks. ThrowOnceFoldRepository throws on the first fold and
+        // succeeds afterward, so a healthy second fold on the SAME actor proves the loop lives.
+        var repository = new ThrowOnceFoldRepository();
+        var sp = BuildServiceProvider(repository, new HealthySource());
+        var actor = CreateActor(sp);
+
+        // First tick: fold throws — caught, actor lives, no successful fold recorded.
+        actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+        AwaitAssert(
+            () => Assert.Null(repository.LastFoldToHourUtc),
+            duration: TimeSpan.FromSeconds(2),
+            interval: TimeSpan.FromMilliseconds(50));
+
+        // Second tick to the SAME actor: fold now succeeds. Re-send on each poll so we don't
+        // race the asynchronous RollupComplete that lowers the guard after the faulted fold.
+        AwaitAssert(
+            () =>
+            {
+                actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
+                Assert.NotNull(repository.LastFoldToHourUtc);
+            },
+            duration: TimeSpan.FromSeconds(3),
+            interval: TimeSpan.FromMilliseconds(50));
+    }
 }