diff --git a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptions.cs b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptions.cs index 15ff01f2..67d55056 100644 --- a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptions.cs @@ -35,4 +35,40 @@ public sealed class KpiHistoryOptions /// [2, 5000]). At least two points are required to draw a line. /// public int DefaultMaxSeriesPoints { get; set; } = 200; + + /// + /// How often the recorder folds the trailing raw KpiSample rows into the + /// hourly KpiRollupHourly table (default 1 hour). Must be strictly + /// positive. Each tick re-folds the trailing + /// hours via an idempotent upsert, so a tick + /// missed during a singleton-failover handover self-heals on the next fold. + /// + public TimeSpan RollupInterval { get; set; } = TimeSpan.FromHours(1); + + /// + /// How many trailing whole hours each rollup fold re-processes (default 3, + /// range [1, 168]). Folding a lookback window rather than just the last + /// hour lets an idempotent re-fold recover the one or more complete hours a + /// missed/failover tick skipped. Never includes the in-progress hour (the fold's + /// upper bound is the current hour start, exclusive). + /// + public int RollupLookbackHours { get; set; } = 3; + + /// + /// Central retention window in days for the hourly rollup rows (default 365, + /// range [1, 3650]). Rollups outlive raw samples so long-range trend + /// charts have data after raw KpiSample rows are purged; the validator + /// therefore requires RollupRetentionDays >= + /// so a window never falls into a hole where raw is purged but no rollup was + /// written. Dropped by the rollup purge sweep. + /// + public int RollupRetentionDays { get; set; } = 365; + + /// + /// Query-routing boundary in hours (default 168 = 7 days, minimum 24). Trend + /// queries whose window is <= this width read raw KpiSample rows + /// (preserving intra-minute detail); wider windows read the pre-aggregated + /// hourly rollups. Consumed by the Central UI query service's range routing. + /// + public int RollupThresholdHours { get; set; } = 168; } diff --git a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptionsValidator.cs b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptionsValidator.cs index f8e48214..7d78758c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptionsValidator.cs +++ b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryOptionsValidator.cs @@ -28,6 +28,21 @@ public sealed class KpiHistoryOptionsValidator : OptionsValidatorBaseInclusive upper bound for . public const int MaxDefaultMaxSeriesPoints = 5000; + /// Inclusive lower bound for . + public const int MinRollupLookbackHours = 1; + + /// Inclusive upper bound for (a week). + public const int MaxRollupLookbackHours = 168; + + /// Inclusive lower bound for . + public const int MinRollupRetentionDays = 1; + + /// Inclusive upper bound for . + public const int MaxRollupRetentionDays = 3650; + + /// Inclusive lower bound for (one day). + public const int MinRollupThresholdHours = 24; + /// protected override void Validate(ValidationBuilder builder, KpiHistoryOptions options) { @@ -54,5 +69,36 @@ public sealed class KpiHistoryOptionsValidator : OptionsValidatorBase MaxDefaultMaxSeriesPoints), $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.DefaultMaxSeriesPoints)} ({options.DefaultMaxSeriesPoints}) " + $"must be in [{MinDefaultMaxSeriesPoints}, {MaxDefaultMaxSeriesPoints}]."); + + builder.RequireThat(options.RollupInterval > TimeSpan.Zero, + $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.RollupInterval)} ({options.RollupInterval}) " + + "must be > 0; it is the hourly rollup fold cadence."); + + // Valid when RollupLookbackHours is within [Min, Max] inclusive. + builder.RequireThat( + !(options.RollupLookbackHours < MinRollupLookbackHours + || options.RollupLookbackHours > MaxRollupLookbackHours), + $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.RollupLookbackHours)} ({options.RollupLookbackHours}) " + + $"must be in [{MinRollupLookbackHours}, {MaxRollupLookbackHours}] hours."); + + // Valid when RollupRetentionDays is within [Min, Max] inclusive. + builder.RequireThat( + !(options.RollupRetentionDays < MinRollupRetentionDays + || options.RollupRetentionDays > MaxRollupRetentionDays), + $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.RollupRetentionDays)} ({options.RollupRetentionDays}) " + + $"must be in [{MinRollupRetentionDays}, {MaxRollupRetentionDays}] days."); + + // Coherence: rollups must be retained AT LEAST as long as raw samples, else a + // long-range chart hits a hole where raw is purged but the rollup was already + // dropped (or never written). Enforced independently of the range check above. + builder.RequireThat( + options.RollupRetentionDays >= options.RetentionDays, + $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.RollupRetentionDays)} ({options.RollupRetentionDays}) " + + $"must be >= {nameof(KpiHistoryOptions.RetentionDays)} ({options.RetentionDays}) so rollups outlive " + + "raw samples and long-range trends have no data hole."); + + builder.RequireThat(options.RollupThresholdHours >= MinRollupThresholdHours, + $"ScadaBridge:KpiHistory:{nameof(KpiHistoryOptions.RollupThresholdHours)} ({options.RollupThresholdHours}) " + + $"must be >= {MinRollupThresholdHours} hours; it is the raw-vs-rollup query-routing boundary."); } } diff --git a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs index 18a887e9..7a2cd7bd 100644 --- a/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.KpiHistory/KpiHistoryRecorderActor.cs @@ -18,9 +18,17 @@ namespace ZB.MOM.WW.ScadaBridge.KpiHistory; /// Bulk-writes the combined batch through /// . /// -/// A separate daily timer (default 1 d) runs the retention purge, dropping rows -/// older than via -/// . +/// A separate daily timer (default 1 d) runs the retention purge, dropping raw +/// sample rows older than via +/// and hourly rollup rows +/// older than the longer via +/// . A third hourly +/// timer (default 1 h) folds the trailing +/// hours of raw samples into the +/// KpiRollupHourly table via +/// (idempotent upsert, +/// in-progress hour excluded), so long-range trend charts read a pre-thinned +/// series. /// /// /// @@ -57,6 +65,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers { private const string SampleTimerKey = "kpi-sample"; private const string PurgeTimerKey = "kpi-purge"; + private const string RollupTimerKey = "kpi-rollup"; private readonly IServiceProvider _serviceProvider; private readonly KpiHistoryOptions _options; @@ -82,6 +91,17 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers /// private bool _sampleInFlight; + /// + /// In-flight guard for the hourly rollup fold. Set true at the start of a fold pass and + /// cleared when the pass's arrives. While true, further + /// s are skipped so folds never overlap — the fold is a + /// group-by-and-upsert over a lookback window that can outlast the (default 1 h) + /// on a slow/recovering store; without this + /// guard a subsequent tick would launch a second concurrent fold contending for the same + /// upsert key. Mirrors . + /// + private bool _rollupInFlight; + /// Akka timer scheduler, assigned by the actor system via . public ITimerScheduler Timers { get; set; } = null!; @@ -104,6 +124,8 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers Receive(_ => _sampleInFlight = false); // lower the in-flight guard (success or fault) Receive(_ => HandlePurgeTick()); Receive(_ => { }); // best-effort: no actor state to reset on completion + Receive(_ => HandleRollupTick()); + Receive(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault) } /// @@ -129,6 +151,16 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers PurgeTick.Instance, initialDelay: PurgeTimerSchedule.InitialDelay(_options.PurgeInterval), interval: _options.PurgeInterval); + + // Fold the trailing hours into the hourly rollup table on a short initial delay + // (so a fresh active node catches up promptly after a failover), then settle into + // the periodic cadence. The fold re-processes a lookback window via an idempotent + // upsert, so a missed tick self-heals on the next fold. + Timers.StartPeriodicTimer( + RollupTimerKey, + RollupTick.Instance, + initialDelay: TimeSpan.FromSeconds(10), + interval: _options.RollupInterval); } /// @@ -249,16 +281,20 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers } /// - /// Handles a purge tick: computes the retention cut-off on the actor thread, then runs - /// the bulk delete off-thread and pipes a completion back to . Purges - /// are daily and idempotent, so no in-flight guard is needed. + /// Handles a purge tick: computes both retention cut-offs on the actor thread (raw samples + /// at , hourly rollups at the longer + /// ), then runs both bulk deletes + /// off-thread and pipes a completion back to . Purges are daily and + /// idempotent, so no in-flight guard is needed. /// private void HandlePurgeTick() { - var before = DateTime.UtcNow - TimeSpan.FromDays(_options.RetentionDays); + var now = DateTime.UtcNow; + var sampleBefore = now - TimeSpan.FromDays(_options.RetentionDays); + var rollupBefore = now - TimeSpan.FromDays(_options.RollupRetentionDays); var cancellationToken = _shutdownCts?.Token ?? CancellationToken.None; - RunPurgePass(before, cancellationToken).PipeTo( + RunPurgePass(sampleBefore, rollupBefore, cancellationToken).PipeTo( Self, success: deleted => { @@ -266,7 +302,7 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers { _logger.LogInformation( "KPI history purge removed {DeletedCount} sample(s) older than {Cutoff:o}.", - deleted, before); + deleted, sampleBefore); } return PurgeComplete.Instance; @@ -280,22 +316,53 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers /// /// Runs a single purge sweep: opens a DI scope, resolves the repository, and bulk-deletes - /// rows captured before , returning the deleted count. The whole - /// body is wrapped so the returned task never faults — on failure the exception is logged - /// and 0 is returned, mirroring 's best-effort contract. + /// expired raw samples (before ) and expired hourly rollups + /// (before ), returning the deleted raw-sample count. The two + /// deletes are isolated so a failure of one never skips the other, and the whole body is + /// wrapped so the returned task never faults — on failure the exception is logged and 0 is + /// returned, mirroring 's best-effort contract. /// - private async Task RunPurgePass(DateTime before, CancellationToken cancellationToken) + private async Task RunPurgePass( + DateTime sampleBefore, DateTime rollupBefore, CancellationToken cancellationToken) { try { await using var scope = _serviceProvider.CreateAsyncScope(); var repository = scope.ServiceProvider.GetRequiredService(); - return await repository.PurgeOlderThanAsync(before, cancellationToken); - } - catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) - { - // Shutdown interrupted the purge; the next active sweep retries. Not a failure. - return 0; + + var deleted = 0; + + // Raw-sample purge — isolated so a fault here does not skip the rollup purge below. + try + { + deleted = await repository.PurgeOlderThanAsync(sampleBefore, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown interrupted the purge; the next active sweep retries. Not a failure. + return deleted; + } + catch (Exception ex) + { + _logger.LogError(ex, "KPI history raw-sample purge failed unexpectedly."); + } + + // Rollup purge — retained longer than raw; isolated so its failure does not affect + // the raw-purge outcome already reported. + try + { + await repository.PurgeRollupsOlderThanAsync(rollupBefore, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown interrupted the rollup purge; the next active sweep retries. Not a failure. + } + catch (Exception ex) + { + _logger.LogError(ex, "KPI history rollup purge failed unexpectedly."); + } + + return deleted; } catch (Exception ex) { @@ -304,6 +371,71 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers } } + /// + /// Handles a rollup tick. If a fold is already in flight the tick is skipped (logged at + /// debug) so folds never overlap; otherwise the in-flight guard is raised and the fold is + /// launched off-thread with a piped back to + /// to lower the guard on the actor thread. The fold window's upper bound is the current + /// hour start (exclusive) so the in-progress hour is never folded, and its lower bound is + /// earlier so a missed/failover tick + /// self-heals via the idempotent upsert. + /// + private void HandleRollupTick() + { + if (_rollupInFlight) + { + _logger.LogDebug("KPI rollup tick skipped — a rollup fold is already in flight."); + return; + } + + _rollupInFlight = true; + var toHourUtc = TruncateToHour(DateTime.UtcNow); + var fromHourUtc = toHourUtc - TimeSpan.FromHours(_options.RollupLookbackHours); + var cancellationToken = _shutdownCts?.Token ?? CancellationToken.None; + + RunRollupPass(fromHourUtc, toHourUtc, cancellationToken).PipeTo( + Self, + success: () => RollupComplete.Instance, + failure: ex => + { + _logger.LogError(ex, "KPI rollup fold faulted unexpectedly."); + return RollupComplete.Instance; + }); + } + + /// + /// Runs a single rollup fold: opens a DI scope, resolves the repository, and folds the + /// complete hours in [, ) + /// into the hourly rollup table via the idempotent upsert. The whole body is wrapped so the + /// returned task never faults — best-effort observability must never disrupt anything. + /// + private async Task RunRollupPass( + DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken) + { + try + { + await using var scope = _serviceProvider.CreateAsyncScope(); + var repository = scope.ServiceProvider.GetRequiredService(); + await repository.FoldHourlyRollupsAsync(fromHourUtc, toHourUtc, cancellationToken); + } + catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested) + { + // Shutdown interrupted the fold; the next active node re-folds the lookback window. Not a failure. + } + catch (Exception ex) + { + _logger.LogError(ex, "KPI rollup fold failed unexpectedly."); + } + } + + /// + /// Truncates a UTC instant to the start of its hour (minutes/seconds/ticks dropped), + /// preserving . Used to derive the fold window's whole-hour + /// boundaries so the in-progress hour is excluded. + /// + private static DateTime TruncateToHour(DateTime utc) => + new(utc.Year, utc.Month, utc.Day, utc.Hour, 0, 0, DateTimeKind.Utc); + /// Self-tick triggering a sampling pass across all registered sources. internal sealed class SampleTick { @@ -334,4 +466,21 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers public static readonly PurgeComplete Instance = new(); private PurgeComplete() { } } + + /// Self-tick triggering an hourly rollup fold over the trailing lookback window. + internal sealed class RollupTick + { + public static readonly RollupTick Instance = new(); + private RollupTick() { } + } + + /// + /// Piped-back completion of a rollup fold; lets the fold run off the actor thread and + /// lowers the _rollupInFlight guard on the actor thread (fires on success and fault). + /// + internal sealed class RollupComplete + { + public static readonly RollupComplete Instance = new(); + private RollupComplete() { } + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryOptionsValidatorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryOptionsValidatorTests.cs index e3969029..b1585220 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryOptionsValidatorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryOptionsValidatorTests.cs @@ -56,7 +56,10 @@ public class KpiHistoryOptionsValidatorTests [InlineData(3650)] // max public void Validate_RetentionDays_InRange_Passes(int value) { - var opts = new KpiHistoryOptions { RetentionDays = value }; + // Keep RollupRetentionDays coherent (>= RetentionDays) so this test isolates the + // RetentionDays range rule and doesn't trip the rollup-retention coherence rule when + // RetentionDays is set above the default RollupRetentionDays (365). + var opts = new KpiHistoryOptions { RetentionDays = value, RollupRetentionDays = value }; Assert.True(NewValidator().Validate(null, opts).Succeeded); } @@ -132,6 +135,137 @@ public class KpiHistoryOptionsValidatorTests f => f.Contains(nameof(KpiHistoryOptions.DefaultMaxSeriesPoints), StringComparison.Ordinal)); } + // --------------------------------------------------------------------- + // RollupInterval > TimeSpan.Zero + // --------------------------------------------------------------------- + + [Theory] + [InlineData(0)] + [InlineData(-1)] + public void Validate_RollupInterval_NotPositive_Fails(int hours) + { + var opts = new KpiHistoryOptions { RollupInterval = TimeSpan.FromHours(hours) }; + var result = NewValidator().Validate(null, opts); + + Assert.False(result.Succeeded); + Assert.Contains( + result.Failures!, + f => f.Contains(nameof(KpiHistoryOptions.RollupInterval), StringComparison.Ordinal)); + } + + // --------------------------------------------------------------------- + // RollupLookbackHours in [1, 168] + // --------------------------------------------------------------------- + + [Theory] + [InlineData(1)] // min + [InlineData(3)] // default + [InlineData(168)] // max + public void Validate_RollupLookbackHours_InRange_Passes(int value) + { + var opts = new KpiHistoryOptions { RollupLookbackHours = value }; + Assert.True(NewValidator().Validate(null, opts).Succeeded); + } + + [Theory] + [InlineData(0)] + [InlineData(169)] + [InlineData(-1)] + public void Validate_RollupLookbackHours_OutOfRange_Fails(int value) + { + var opts = new KpiHistoryOptions { RollupLookbackHours = value }; + var result = NewValidator().Validate(null, opts); + + Assert.False(result.Succeeded); + Assert.Contains( + result.Failures!, + f => f.Contains(nameof(KpiHistoryOptions.RollupLookbackHours), StringComparison.Ordinal)); + } + + // --------------------------------------------------------------------- + // RollupRetentionDays in [1, 3650] AND >= RetentionDays + // --------------------------------------------------------------------- + + [Theory] + [InlineData(90)] // == RetentionDays default + [InlineData(365)] // default + [InlineData(3650)] // max + public void Validate_RollupRetentionDays_InRange_Passes(int value) + { + var opts = new KpiHistoryOptions { RollupRetentionDays = value }; + Assert.True(NewValidator().Validate(null, opts).Succeeded); + } + + [Theory] + [InlineData(0)] + [InlineData(3651)] + [InlineData(-1)] + public void Validate_RollupRetentionDays_OutOfRange_Fails(int value) + { + var opts = new KpiHistoryOptions { RollupRetentionDays = value }; + var result = NewValidator().Validate(null, opts); + + Assert.False(result.Succeeded); + Assert.Contains( + result.Failures!, + f => f.Contains(nameof(KpiHistoryOptions.RollupRetentionDays), StringComparison.Ordinal)); + } + + [Fact] + public void Validate_RollupRetentionDays_LessThanRetentionDays_Fails() + { + // Coherence rule: rollups must outlive raw samples, else long-range trends hit a hole. + var opts = new KpiHistoryOptions + { + RetentionDays = 90, + RollupRetentionDays = 30, // in [1,3650] but < RetentionDays + }; + var result = NewValidator().Validate(null, opts); + + Assert.False(result.Succeeded); + Assert.Contains( + result.Failures!, + f => f.Contains(nameof(KpiHistoryOptions.RollupRetentionDays), StringComparison.Ordinal) + && f.Contains(nameof(KpiHistoryOptions.RetentionDays), StringComparison.Ordinal)); + } + + [Fact] + public void Validate_RollupRetentionDays_EqualToRetentionDays_Passes() + { + // Boundary of the coherence rule (>=): equal retention is allowed. + var opts = new KpiHistoryOptions { RetentionDays = 120, RollupRetentionDays = 120 }; + Assert.True(NewValidator().Validate(null, opts).Succeeded); + } + + // --------------------------------------------------------------------- + // RollupThresholdHours >= 24 + // --------------------------------------------------------------------- + + [Theory] + [InlineData(24)] // min + [InlineData(168)] // default + [InlineData(2160)] // 90 d + public void Validate_RollupThresholdHours_AtOrAboveMin_Passes(int value) + { + var opts = new KpiHistoryOptions { RollupThresholdHours = value }; + Assert.True(NewValidator().Validate(null, opts).Succeeded); + } + + [Theory] + [InlineData(23)] + [InlineData(0)] + [InlineData(-1)] + public void Validate_RollupThresholdHours_BelowMin_Fails(int value) + { + var opts = new KpiHistoryOptions { RollupThresholdHours = value }; + var result = NewValidator().Validate(null, opts); + + Assert.False(result.Succeeded); + Assert.Contains( + result.Failures!, + f => f.Contains(nameof(KpiHistoryOptions.RollupThresholdHours), StringComparison.Ordinal)); + } + [Fact] public void Validate_AllRulesViolated_AccumulatesEveryFailure() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs index 27f4bf60..f17e2125 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.KpiHistory.Tests/KpiHistoryRecorderActorTests.cs @@ -107,6 +107,10 @@ public class KpiHistoryRecorderActorTests : TestKit private readonly object _gate = new(); private readonly List _recorded = new(); private DateTime? _purgeCutoff; + private DateTime? _rollupPurgeCutoff; + private DateTime? _foldFromHourUtc; + private DateTime? _foldToHourUtc; + private int _foldCount; public IReadOnlyList Recorded { @@ -120,6 +124,30 @@ public class KpiHistoryRecorderActorTests : TestKit get { lock (_gate) { return _purgeCutoff; } } } + /// Cut-off handed to (rollup retention). + public DateTime? RollupPurgeCutoff + { + get { lock (_gate) { return _rollupPurgeCutoff; } } + } + + /// Inclusive lower bound handed to . + public DateTime? FoldFromHourUtc + { + get { lock (_gate) { return _foldFromHourUtc; } } + } + + /// Exclusive upper bound handed to . + public DateTime? FoldToHourUtc + { + get { lock (_gate) { return _foldToHourUtc; } } + } + + /// Number of times was entered. + public int FoldCount + { + get { lock (_gate) { return _foldCount; } } + } + public Task RecordSamplesAsync( IReadOnlyCollection 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> GetHourlySeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) + { + lock (_gate) { _rollupPurgeCutoff = before; } + return Task.CompletedTask; + } } /// @@ -174,6 +225,121 @@ public class KpiHistoryRecorderActorTests : TestKit public Task PurgeOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) => Task.FromResult(0); + + public Task FoldHourlyRollupsAsync( + DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default) => + Task.CompletedTask; + + public Task> GetHourlySeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) => + Task.CompletedTask; + } + + /// + /// Repository fake whose 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. + /// + private sealed class GatedFoldRepository : IKpiHistoryRepository + { + private readonly ManualResetEventSlim _release = new(initialState: false); + private int _foldCount; + + /// Number of times has been entered. + public int FoldCount => Volatile.Read(ref _foldCount); + + /// Releases all blocked folds so the gated passes can complete. + public void Release() => _release.Set(); + + public Task RecordSamplesAsync( + IReadOnlyCollection samples, CancellationToken cancellationToken = default) => + Task.CompletedTask; + + public Task> GetRawSeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + public Task 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> GetHourlySeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + public Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default) => + Task.CompletedTask; + } + + /// + /// Repository fake whose 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. + /// + private sealed class ThrowOnceFoldRepository : IKpiHistoryRepository + { + private readonly object _gate = new(); + private int _foldCount; + private DateTime? _lastFoldToHourUtc; + + /// Number of times has been entered (incl. the throw). + public int FoldCount + { + get { lock (_gate) { return _foldCount; } } + } + + /// Exclusive upper bound of the most recent successful fold. + public DateTime? LastFoldToHourUtc + { + get { lock (_gate) { return _lastFoldToHourUtc; } } + } + + public Task RecordSamplesAsync( + IReadOnlyCollection samples, CancellationToken cancellationToken = default) => + Task.CompletedTask; + + public Task> GetRawSeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + public Task 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> GetHourlySeriesAsync( + string source, string metric, string scope, string? scopeKey, + DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default) => + Task.FromResult>(Array.Empty()); + + 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)); + } }