using Akka.Actor;
using Akka.TestKit.Xunit2;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Kpi;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
namespace ZB.MOM.WW.ScadaBridge.KpiHistory.Tests;
///
/// K4 tests for . The actor's internal
/// SampleTick/PurgeTick messages are exposed to this assembly via
/// InternalsVisibleTo so a tick can be driven deterministically without
/// racing the periodic timer. Hand-rolled fakes (no mocking lib in this test
/// project) record what the recorder hands to the repository.
///
public class KpiHistoryRecorderActorTests : TestKit
{
///
/// A healthy sample source that returns a fixed set of samples, stamping each with the
/// capturedAtUtc the recorder supplies.
///
private sealed class HealthySource : IKpiSampleSource
{
public string Source => KpiSources.NotificationOutbox;
public Task> CollectAsync(
DateTime capturedAtUtc, CancellationToken cancellationToken = default)
{
IReadOnlyList samples = new List
{
new()
{
Source = Source,
Metric = "QueueDepth",
Scope = KpiScopes.Global,
ScopeKey = null,
Value = 7,
CapturedAtUtc = capturedAtUtc,
},
new()
{
Source = Source,
Metric = "ParkedCount",
Scope = KpiScopes.Global,
ScopeKey = null,
Value = 2,
CapturedAtUtc = capturedAtUtc,
},
};
return Task.FromResult(samples);
}
}
/// A source whose always throws — must not abort the pass.
private sealed class ThrowingSource : IKpiSampleSource
{
public string Source => KpiSources.SiteCallAudit;
public Task> CollectAsync(
DateTime capturedAtUtc, CancellationToken cancellationToken = default) =>
throw new InvalidOperationException("simulated source failure");
}
///
/// A source that throws on the first call and returns a
/// healthy sample on every subsequent call. Used to drive a faulted first tick followed
/// by a healthy second tick on the same actor instance.
///
private sealed class ThrowOnceSource : IKpiSampleSource
{
private int _callCount;
public string Source => KpiSources.SiteCallAudit;
public Task> CollectAsync(
DateTime capturedAtUtc, CancellationToken cancellationToken = default)
{
if (Interlocked.Increment(ref _callCount) == 1)
throw new InvalidOperationException("simulated first-call source failure");
IReadOnlyList samples = new[]
{
new KpiSample
{
Source = Source,
Metric = "RecoveredSample",
Scope = KpiScopes.Global,
ScopeKey = null,
Value = 1,
CapturedAtUtc = capturedAtUtc,
},
};
return Task.FromResult(samples);
}
}
///
/// Recording repository fake. Captures the samples handed to
/// and the cut-off handed to
/// .
///
private sealed class RecordingRepository : IKpiHistoryRepository
{
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;
private readonly List<(DateTime FromHourUtc, DateTime ToHourUtc)> _foldWindows = new();
public IReadOnlyList Recorded
{
get { lock (_gate) { return _recorded.ToArray(); } }
}
// PurgeOlderThanAsync runs on a threadpool thread; guard the field with
// the same _gate lock used by _recorded so test-thread reads are race-free.
public DateTime? PurgeCutoff
{
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; } }
}
/// Every fold window handed to , in call order.
public IReadOnlyList<(DateTime FromHourUtc, DateTime ToHourUtc)> FoldWindows
{
get { lock (_gate) { return _foldWindows.ToArray(); } }
}
public Task RecordSamplesAsync(
IReadOnlyCollection samples, CancellationToken cancellationToken = default)
{
lock (_gate)
{
_recorded.AddRange(samples);
}
return 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)
{
lock (_gate) { _purgeCutoff = before; }
return Task.FromResult(0);
}
public Task FoldHourlyRollupsAsync(
DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken = default)
{
lock (_gate)
{
_foldFromHourUtc = fromHourUtc;
_foldToHourUtc = toHourUtc;
_foldWindows.Add((fromHourUtc, 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;
}
public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(null);
}
///
/// Repository fake whose blocks on a manual-reset gate
/// until the test releases it, holding a sample pass "in flight" so an overlapping-tick
/// scenario can be driven deterministically. Counts how many times the write was entered.
///
private sealed class GatedRepository : IKpiHistoryRepository
{
private readonly ManualResetEventSlim _release = new(initialState: false);
private int _writeCount;
/// Number of times has been entered.
public int WriteCount => Volatile.Read(ref _writeCount);
/// Releases all blocked writes so the gated passes can complete.
public void Release() => _release.Set();
public Task RecordSamplesAsync(
IReadOnlyCollection samples, CancellationToken cancellationToken = default)
{
Interlocked.Increment(ref _writeCount);
// Block on a threadpool thread (PipeTo runs the pass off the actor thread), holding
// the pass in flight until the test opens the gate.
return Task.Run(() => _release.Wait(cancellationToken), cancellationToken);
}
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) =>
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;
public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(null);
}
///
/// 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;
public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(null);
}
///
/// 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;
public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(null);
}
///
/// Repository fake for the backfill-slice interleaving test. Classifies each fold as a backfill
/// slice (24 h window) or a periodic fold (the distinct RollupLookbackHours-wide window)
/// by window width. Backfill slices complete immediately (counted); the FIRST periodic fold
/// blocks in flight until released, so the test can observe that a periodic fold claimed the
/// fold path between backfill slices while backfill slices remained.
///
private sealed class InterleaveFoldRepository : IKpiHistoryRepository
{
private readonly TimeSpan _periodicWindowWidth;
private readonly object _gate = new();
private readonly SemaphoreSlim _firstPeriodicRelease = new(initialCount: 0);
private int _backfillFoldCount;
private int _periodicFoldCount;
private int _backfillCountAtFirstPeriodic = -1;
public InterleaveFoldRepository(TimeSpan periodicWindowWidth) =>
_periodicWindowWidth = periodicWindowWidth;
/// Number of backfill-slice folds (24 h windows) entered so far.
public int BackfillFoldCount { get { lock (_gate) { return _backfillFoldCount; } } }
/// Number of periodic folds (the lookback-width window) entered so far.
public int PeriodicFoldCount { get { lock (_gate) { return _periodicFoldCount; } } }
/// Backfill-slice count observed at the instant the first periodic fold entered.
public int BackfillCountAtFirstPeriodic { get { lock (_gate) { return _backfillCountAtFirstPeriodic; } } }
/// Releases the first (held-in-flight) periodic fold so the deferred backfill resumes.
public void ReleaseFirstPeriodic() => _firstPeriodicRelease.Release();
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)
{
var isPeriodic = (toHourUtc - fromHourUtc) == _periodicWindowWidth;
if (isPeriodic)
{
bool blockThis;
lock (_gate)
{
_periodicFoldCount++;
blockThis = _periodicFoldCount == 1;
if (blockThis)
{
_backfillCountAtFirstPeriodic = _backfillFoldCount;
}
}
// Hold the first periodic fold in flight (on a threadpool thread — PipeTo runs the
// pass off the actor thread) so the between-slice interleave state is observable.
return blockThis
? Task.Run(() => _firstPeriodicRelease.Wait(cancellationToken), cancellationToken)
: Task.CompletedTask;
}
lock (_gate) { _backfillFoldCount++; }
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;
public Task GetLatestRollupHourAsync(CancellationToken cancellationToken = default) =>
Task.FromResult(null);
}
private IServiceProvider BuildServiceProvider(
IKpiHistoryRepository repository, params IKpiSampleSource[] sources)
{
var services = new ServiceCollection();
// Mirror production: sources + repository are scoped, so the recorder opens a fresh
// scope per tick and resolves there.
foreach (var source in sources)
{
var captured = source;
services.AddScoped(_ => captured);
}
services.AddScoped(_ => repository);
return services.BuildServiceProvider();
}
///
/// Creates the recorder with both timers set to a long interval so neither periodic timer
/// fires during a test — ticks are sent manually instead.
///
private IActorRef CreateActor(IServiceProvider serviceProvider, KpiHistoryOptions? options = null)
{
return Sys.ActorOf(Props.Create(() => new KpiHistoryRecorderActor(
serviceProvider,
options ?? new KpiHistoryOptions
{
SampleInterval = TimeSpan.FromHours(1),
PurgeInterval = TimeSpan.FromHours(1),
},
NullLogger.Instance)));
}
[Fact]
public void SampleTick_WritesHealthySourceSamples_AndThrowingSourceDoesNotAbortTick()
{
var repository = new RecordingRepository();
// Order the throwing source FIRST so the test also proves a throw early in the
// enumeration does not suppress a later healthy source.
var sp = BuildServiceProvider(repository, new ThrowingSource(), new HealthySource());
var actor = CreateActor(sp);
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
AwaitAssert(
() =>
{
// The healthy source's two samples were written despite the throwing source.
Assert.Equal(2, repository.Recorded.Count);
Assert.Contains(repository.Recorded, s => s.Metric == "QueueDepth" && s.Value == 7);
Assert.Contains(repository.Recorded, s => s.Metric == "ParkedCount" && s.Value == 2);
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
}
[Fact]
public void PurgeTick_CallsPurgeWithCutoff_AtUtcNowMinusRetentionDays()
{
var repository = new RecordingRepository();
var sp = BuildServiceProvider(repository, new HealthySource());
const int retentionDays = 30;
var actor = CreateActor(sp, new KpiHistoryOptions
{
SampleInterval = TimeSpan.FromHours(1),
PurgeInterval = TimeSpan.FromHours(1),
RetentionDays = retentionDays,
});
actor.Tell(KpiHistoryRecorderActor.PurgeTick.Instance);
AwaitAssert(
() =>
{
Assert.NotNull(repository.PurgeCutoff);
var expected = DateTime.UtcNow - TimeSpan.FromDays(retentionDays);
Assert.True(
Math.Abs((repository.PurgeCutoff!.Value - expected).TotalMinutes) < 1.0,
$"purge cutoff {repository.PurgeCutoff:o} should be within 1 minute of {expected:o}");
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
}
[Fact]
public void FaultedTick_DoesNotCrashActor_AndSubsequentTickStillRuns()
{
// ThrowOnceSource throws on the first CollectAsync call and returns a healthy
// sample on every subsequent call. This lets us send two ticks to the SAME
// actor instance and verify that:
// • The first tick (faulted source) records nothing but does not crash the actor.
// • The second tick reaches the same actor and records the recovered sample,
// proving the singleton's message loop is still alive after a faulted pass.
var repository = new RecordingRepository();
var sp = BuildServiceProvider(repository, new ThrowOnceSource());
var actor = CreateActor(sp);
// First tick: source throws on first call — caught per-source, nothing written, actor lives.
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
AwaitAssert(
() => Assert.Empty(repository.Recorded),
duration: TimeSpan.FromSeconds(2),
interval: TimeSpan.FromMilliseconds(50));
// Second tick to the SAME actor: source now returns a healthy sample. Re-send the tick
// on each poll so we don't race the (asynchronous) SampleComplete that lowers the
// in-flight guard after the faulted first pass — a tick that lands before the guard
// clears is harmlessly skipped, and the next poll's tick runs. The recovered sample
// being recorded proves the actor's message loop is still alive after a faulted pass.
// (>= 1 rather than exactly-1: a later re-sent tick could run an extra recovering pass
// once the guard clears, which is not what this test pins — KH-001's one-pass-per-tick
// guard is pinned by OverlappingTick_WhileFirstPassInFlight_DoesNotStartSecondPass.)
AwaitAssert(
() =>
{
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
Assert.NotEmpty(repository.Recorded);
},
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
}
[Fact]
public void OverlappingTick_WhileFirstPassInFlight_DoesNotStartSecondPass()
{
// KpiHistory-001 regression: the in-flight guard must coalesce a tick that arrives
// while a prior sample pass is still awaiting its DB write. With a gated repository
// holding the first write open, a second SampleTick must NOT spawn a second pass —
// so RecordSamplesAsync is entered exactly once until the gate is released.
var repository = new GatedRepository();
var sp = BuildServiceProvider(repository, new HealthySource());
var actor = CreateActor(sp);
// First tick: raises the guard, enters the (gated, blocking) write — held in flight.
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
AwaitAssert(
() => Assert.Equal(1, repository.WriteCount),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
// Second tick while the first pass is still in flight: must be skipped by the guard.
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
// Give the second tick ample time to (wrongly) start a pass; the write count must
// stay at 1, proving no second concurrent pass was launched.
Assert.Equal(1, repository.WriteCount);
Thread.Sleep(300);
Assert.Equal(1, repository.WriteCount);
// Release the gate so the first pass completes and the guard is lowered; a fresh
// tick must now run a new pass (guard correctly reset, not stuck). Re-send the tick on
// each poll so we don't race the (asynchronous) SampleComplete that lowers the guard —
// a tick that lands before the guard clears is harmlessly skipped, the next one runs.
repository.Release();
AwaitAssert(
() =>
{
actor.Tell(KpiHistoryRecorderActor.SampleTick.Instance);
Assert.Equal(2, repository.WriteCount);
},
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));
}
// =====================================================================
// One-shot rollup backfill (Task 6)
// =====================================================================
[Fact]
public void BackfillTick_FoldsFullRetentionWindow_Once()
{
// The one-shot backfill must fold the WHOLE raw-retention window once, so long-range
// (30 d/90 d) charts aren't blank until enough wall-clock passes after deploy. It now
// does so in bounded <=24 h slices (R1) rather than a single unbounded fold: the union
// of the slice windows must cover exactly [truncate(now) − RetentionDays, truncate(now)),
// each slice must be <=24 h, and the slices must be contiguous and oldest-first. A second
// BackfillTick to the same actor must be a no-op (the _backfillDone latch makes the whole
// pass run at most once per actor lifetime).
var repository = new RecordingRepository();
var sp = BuildServiceProvider(repository, new HealthySource());
const int retentionDays = 5;
var actor = CreateActor(sp, new KpiHistoryOptions
{
SampleInterval = TimeSpan.FromHours(1),
PurgeInterval = TimeSpan.FromHours(1),
RollupInterval = TimeSpan.FromHours(1),
RetentionDays = retentionDays,
});
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
AwaitAssert(
() =>
{
var windows = repository.FoldWindows;
// RetentionDays = 5 → exactly 5 contiguous 24 h slices.
Assert.Equal(retentionDays, windows.Count);
var expectedToHour = new DateTime(
DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day,
DateTime.UtcNow.Hour, 0, 0, DateTimeKind.Utc);
var expectedFromHour = expectedToHour - TimeSpan.FromDays(retentionDays);
// Oldest-first, contiguous, each <=24 h, union == full retention window.
Assert.Equal(expectedFromHour, windows[0].FromHourUtc);
Assert.Equal(expectedToHour, windows[^1].ToHourUtc);
for (var i = 0; i < windows.Count; i++)
{
Assert.True(windows[i].ToHourUtc - windows[i].FromHourUtc <= TimeSpan.FromHours(24),
$"slice {i} width {windows[i].ToHourUtc - windows[i].FromHourUtc} exceeds 24 h");
if (i > 0)
{
Assert.Equal(windows[i - 1].ToHourUtc, windows[i].FromHourUtc); // contiguous
}
}
},
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
// Second BackfillTick to the SAME actor: the once-only latch must suppress the whole pass,
// so no further slices are folded. (No periodic RollupTick is sent, and the periodic/backfill
// auto-timers don't fire within this window, so the fold windows are driven solely by the
// backfill under test.)
var foldedCount = repository.FoldWindows.Count;
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
Thread.Sleep(300);
Assert.Equal(foldedCount, repository.FoldWindows.Count);
}
[Fact]
public void Backfill_SlicesRetentionWindow_IntoBoundedDayFolds_OldestFirst()
{
// RetentionDays = 3 → 3 fold calls, each window <=24 h, contiguous, oldest-first, whose
// union is exactly [TruncateToHour(now) − 3 d, TruncateToHour(now)). Pre-R1 a single fold
// call spanned the whole 3-day window (arch-review 04 round 2, R1).
var repository = new RecordingRepository();
var sp = BuildServiceProvider(repository, new HealthySource());
const int retentionDays = 3;
var actor = CreateActor(sp, new KpiHistoryOptions
{
SampleInterval = TimeSpan.FromHours(1),
PurgeInterval = TimeSpan.FromHours(1),
RollupInterval = TimeSpan.FromHours(1),
RetentionDays = retentionDays,
});
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
AwaitAssert(
() =>
{
var windows = repository.FoldWindows;
Assert.Equal(retentionDays, windows.Count); // 3 × 24 h slices
var expectedToHour = new DateTime(
DateTime.UtcNow.Year, DateTime.UtcNow.Month, DateTime.UtcNow.Day,
DateTime.UtcNow.Hour, 0, 0, DateTimeKind.Utc);
var expectedFromHour = expectedToHour - TimeSpan.FromDays(retentionDays);
Assert.Equal(expectedFromHour, windows[0].FromHourUtc); // oldest first
Assert.Equal(expectedToHour, windows[^1].ToHourUtc); // newest last
for (var i = 0; i < windows.Count; i++)
{
Assert.True(windows[i].ToHourUtc - windows[i].FromHourUtc <= TimeSpan.FromHours(24),
$"slice {i} exceeds 24 h");
if (i > 0)
{
Assert.Equal(windows[i - 1].ToHourUtc, windows[i].FromHourUtc); // contiguous
}
}
},
duration: TimeSpan.FromSeconds(5),
interval: TimeSpan.FromMilliseconds(50));
}
[Fact]
public async Task PeriodicRollupTick_Interleaves_BetweenBackfillSlices()
{
// A periodic fold must be able to claim the fold path BETWEEN backfill slices: while the
// historical backfill runs slice-by-slice, a RollupTick that lands in a between-slice gap
// runs its (recent-window) fold and the remaining backfill slices defer behind it, then
// ALL backfill slices still complete afterwards. Pre-R1 _backfillInFlight blocked every
// periodic fold for the whole (single-fold) pass, so no interleaving was possible
// (arch-review 04 round 2, R1).
const int retentionDays = 8; // 8 backfill slices of 24 h
const int lookbackHours = 2; // periodic fold window width — distinct from the 24 h slices
var repository = new InterleaveFoldRepository(TimeSpan.FromHours(lookbackHours));
var sp = BuildServiceProvider(repository, new HealthySource());
var actor = CreateActor(sp, new KpiHistoryOptions
{
SampleInterval = TimeSpan.FromHours(1),
PurgeInterval = TimeSpan.FromHours(1),
RollupInterval = TimeSpan.FromHours(24), // periodic auto-timer won't fire during the test
RetentionDays = retentionDays,
RollupLookbackHours = lookbackHours,
});
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
// Continuously offer periodic ticks; one will win a between-slice gap and interleave.
using var stopFlood = new CancellationTokenSource();
var flood = Task.Run(() =>
{
while (!stopFlood.IsCancellationRequested)
{
actor.Tell(KpiHistoryRecorderActor.RollupTick.Instance);
Thread.Sleep(1);
}
});
try
{
// A periodic fold interleaved BEFORE the backfill finished all its slices.
AwaitAssert(
() =>
{
Assert.True(repository.PeriodicFoldCount >= 1, "no periodic fold interleaved");
Assert.True(
repository.BackfillCountAtFirstPeriodic < retentionDays,
"the periodic fold did not interleave before the backfill completed");
},
duration: TimeSpan.FromSeconds(10),
interval: TimeSpan.FromMilliseconds(20));
}
finally
{
stopFlood.Cancel();
await flood;
}
// The first periodic fold is held in flight, deferring the remaining backfill slices;
// release it so the deferred backfill resumes and ALL slices still complete.
repository.ReleaseFirstPeriodic();
AwaitAssert(
() => Assert.Equal(retentionDays, repository.BackfillFoldCount),
duration: TimeSpan.FromSeconds(10),
interval: TimeSpan.FromMilliseconds(50));
}
[Fact]
public void BackfillInFlight_SkipsPeriodicRollupTick()
{
// While a backfill slice fold is in flight, a periodic RollupTick must be skipped so the
// two idempotent upserts never race on overlapping rows (unchanged semantics — a RollupTick
// arriving while a *slice* is in flight is still skipped). With a gated fold holding the
// single backfill slice open, a RollupTick must NOT enter a second fold — the fold count
// stays at 1 until the gate is released. RetentionDays = 1 → exactly one 24 h backfill slice.
var repository = new GatedFoldRepository();
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 = 1,
});
// Backfill tick: raises the backfill guard, enters the (gated, blocking) fold — held in flight.
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
AwaitAssert(
() => Assert.Equal(1, repository.FoldCount),
duration: TimeSpan.FromSeconds(3),
interval: TimeSpan.FromMilliseconds(50));
// Periodic rollup tick while the backfill is 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 backfill completes and its guard is lowered; a fresh periodic
// tick must now run a new fold (guards correctly reset, not stuck). Re-send on each poll so
// we don't race the asynchronous BackfillSliceComplete 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));
}
}