Joseph Doherty
bedf3c55f0
Fold the whole raw-retention window of already-recorded KpiSample rows into KpiRollupHourly once shortly after the recorder singleton starts, so long-range (30 d/90 d) trend charts aren't blank until enough wall-clock passes after deploy. Reuses Task 3's idempotent FoldHourlyRollupsAsync over [TruncateToHour(now) - RetentionDays, TruncateToHour(now)); runs at most once per actor lifetime (_backfillDone latch) and defers past / blocks the periodic lookback fold via a dedicated _backfillInFlight guard so the two idempotent upserts never race on overlapping rows. Best-effort: no exception escapes; logs start, window, and elapsed on completion. No Host or options change. Claude-Session: https://claude.ai/code/session_01MtdgwpEeCUn6cUA5f1LMPj
768 lines
34 KiB
C#
768 lines
34 KiB
C#
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;
|
|
|
|
/// <summary>
|
|
/// K4 tests for <see cref="KpiHistoryRecorderActor"/>. The actor's internal
|
|
/// <c>SampleTick</c>/<c>PurgeTick</c> messages are exposed to this assembly via
|
|
/// <c>InternalsVisibleTo</c> 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.
|
|
/// </summary>
|
|
public class KpiHistoryRecorderActorTests : TestKit
|
|
{
|
|
/// <summary>
|
|
/// A healthy sample source that returns a fixed set of samples, stamping each with the
|
|
/// <c>capturedAtUtc</c> the recorder supplies.
|
|
/// </summary>
|
|
private sealed class HealthySource : IKpiSampleSource
|
|
{
|
|
public string Source => KpiSources.NotificationOutbox;
|
|
|
|
public Task<IReadOnlyList<KpiSample>> CollectAsync(
|
|
DateTime capturedAtUtc, CancellationToken cancellationToken = default)
|
|
{
|
|
IReadOnlyList<KpiSample> samples = new List<KpiSample>
|
|
{
|
|
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);
|
|
}
|
|
}
|
|
|
|
/// <summary>A source whose <see cref="CollectAsync"/> always throws — must not abort the pass.</summary>
|
|
private sealed class ThrowingSource : IKpiSampleSource
|
|
{
|
|
public string Source => KpiSources.SiteCallAudit;
|
|
|
|
public Task<IReadOnlyList<KpiSample>> CollectAsync(
|
|
DateTime capturedAtUtc, CancellationToken cancellationToken = default) =>
|
|
throw new InvalidOperationException("simulated source failure");
|
|
}
|
|
|
|
/// <summary>
|
|
/// A source that throws on the first <see cref="CollectAsync"/> 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 <em>same</em> actor instance.
|
|
/// </summary>
|
|
private sealed class ThrowOnceSource : IKpiSampleSource
|
|
{
|
|
private int _callCount;
|
|
|
|
public string Source => KpiSources.SiteCallAudit;
|
|
|
|
public Task<IReadOnlyList<KpiSample>> CollectAsync(
|
|
DateTime capturedAtUtc, CancellationToken cancellationToken = default)
|
|
{
|
|
if (Interlocked.Increment(ref _callCount) == 1)
|
|
throw new InvalidOperationException("simulated first-call source failure");
|
|
|
|
IReadOnlyList<KpiSample> samples = new[]
|
|
{
|
|
new KpiSample
|
|
{
|
|
Source = Source,
|
|
Metric = "RecoveredSample",
|
|
Scope = KpiScopes.Global,
|
|
ScopeKey = null,
|
|
Value = 1,
|
|
CapturedAtUtc = capturedAtUtc,
|
|
},
|
|
};
|
|
return Task.FromResult(samples);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Recording repository fake. Captures the samples handed to
|
|
/// <see cref="RecordSamplesAsync"/> and the cut-off handed to
|
|
/// <see cref="PurgeOlderThanAsync"/>.
|
|
/// </summary>
|
|
private sealed class RecordingRepository : IKpiHistoryRepository
|
|
{
|
|
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
|
|
{
|
|
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; } }
|
|
}
|
|
|
|
/// <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)
|
|
{
|
|
lock (_gate)
|
|
{
|
|
_recorded.AddRange(samples);
|
|
}
|
|
return 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)
|
|
{
|
|
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>
|
|
/// Repository fake whose <see cref="RecordSamplesAsync"/> 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.
|
|
/// </summary>
|
|
private sealed class GatedRepository : IKpiHistoryRepository
|
|
{
|
|
private readonly ManualResetEventSlim _release = new(initialState: false);
|
|
private int _writeCount;
|
|
|
|
/// <summary>Number of times <see cref="RecordSamplesAsync"/> has been entered.</summary>
|
|
public int WriteCount => Volatile.Read(ref _writeCount);
|
|
|
|
/// <summary>Releases all blocked writes so the gated passes can complete.</summary>
|
|
public void Release() => _release.Set();
|
|
|
|
public Task RecordSamplesAsync(
|
|
IReadOnlyCollection<KpiSample> 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<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) =>
|
|
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(
|
|
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();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Creates the recorder with both timers set to a long interval so neither periodic timer
|
|
/// fires during a test — ticks are sent manually instead.
|
|
/// </summary>
|
|
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<KpiHistoryRecorderActor>.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: the fold's
|
|
// upper bound is the current hour start (in-progress hour excluded) and its lower bound is
|
|
// RetentionDays earlier. A second BackfillTick to the same actor must be a no-op (the
|
|
// _backfillDone latch makes it run at most once per actor lifetime).
|
|
var repository = new RecordingRepository();
|
|
var sp = BuildServiceProvider(repository, new HealthySource());
|
|
const int retentionDays = 90;
|
|
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(
|
|
() =>
|
|
{
|
|
Assert.Equal(1, repository.FoldCount);
|
|
Assert.NotNull(repository.FoldToHourUtc);
|
|
Assert.NotNull(repository.FoldFromHourUtc);
|
|
|
|
var toHour = repository.FoldToHourUtc!.Value;
|
|
var fromHour = repository.FoldFromHourUtc!.Value;
|
|
|
|
// toHour 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,
|
|
$"backfill toHour {toHour:o} should be the current hour start {expectedToHour:o}");
|
|
|
|
// fromHour is exactly RetentionDays (the full raw-retention window) before toHour.
|
|
Assert.Equal(toHour - TimeSpan.FromDays(retentionDays), fromHour);
|
|
},
|
|
duration: TimeSpan.FromSeconds(3),
|
|
interval: TimeSpan.FromMilliseconds(50));
|
|
|
|
// Second BackfillTick to the SAME actor: the once-only latch must suppress it, so the
|
|
// fold count stays at 1. (No periodic RollupTick is sent, and the periodic/backfill
|
|
// auto-timers don't fire within this sub-10 s window, so FoldCount is driven solely by
|
|
// the two backfill ticks under test.)
|
|
actor.Tell(KpiHistoryRecorderActor.BackfillTick.Instance);
|
|
Thread.Sleep(300);
|
|
Assert.Equal(1, repository.FoldCount);
|
|
}
|
|
|
|
[Fact]
|
|
public void BackfillInFlight_SkipsPeriodicRollupTick()
|
|
{
|
|
// While the (possibly long) full-window backfill fold is in flight, a periodic RollupTick
|
|
// must be skipped so the two idempotent upserts never race on overlapping rows. With a
|
|
// gated fold holding the backfill open, a RollupTick must NOT enter a second fold — the
|
|
// fold count stays at 1 until the gate is released.
|
|
var repository = new GatedFoldRepository();
|
|
var sp = BuildServiceProvider(repository, new HealthySource());
|
|
var actor = CreateActor(sp);
|
|
|
|
// 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 BackfillComplete 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));
|
|
}
|
|
}
|