feat(kpihistory): one-shot rollup backfill of existing samples on recorder start (plan #22 T6)
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
This commit is contained in:
@@ -28,7 +28,10 @@ namespace ZB.MOM.WW.ScadaBridge.KpiHistory;
|
|||||||
/// <c>KpiRollupHourly</c> table via
|
/// <c>KpiRollupHourly</c> table via
|
||||||
/// <see cref="IKpiHistoryRepository.FoldHourlyRollupsAsync"/> (idempotent upsert,
|
/// <see cref="IKpiHistoryRepository.FoldHourlyRollupsAsync"/> (idempotent upsert,
|
||||||
/// in-progress hour excluded), so long-range trend charts read a pre-thinned
|
/// in-progress hour excluded), so long-range trend charts read a pre-thinned
|
||||||
/// series.
|
/// series. A one-shot backfill timer additionally folds the whole raw-retention
|
||||||
|
/// window of already-recorded samples once shortly after start, so 30 d/90 d
|
||||||
|
/// charts aren't blank until enough wall-clock passes after deploy; it reuses the
|
||||||
|
/// same idempotent fold and runs at most once per actor lifetime.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
/// <remarks>
|
/// <remarks>
|
||||||
/// <para>
|
/// <para>
|
||||||
@@ -66,6 +69,18 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
private const string SampleTimerKey = "kpi-sample";
|
private const string SampleTimerKey = "kpi-sample";
|
||||||
private const string PurgeTimerKey = "kpi-purge";
|
private const string PurgeTimerKey = "kpi-purge";
|
||||||
private const string RollupTimerKey = "kpi-rollup";
|
private const string RollupTimerKey = "kpi-rollup";
|
||||||
|
private const string BackfillTimerKey = "kpi-rollup-backfill";
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Initial delay before the one-shot rollup backfill fires (see
|
||||||
|
/// <see cref="HandleBackfillTick"/>). Deliberately a touch longer than the periodic rollup
|
||||||
|
/// timer's own <c>10 s</c> initial delay so that, when both are armed at start, the periodic
|
||||||
|
/// tick claims the fold path first and the backfill simply defers (the two folds are
|
||||||
|
/// idempotent upserts, but they must not race on overlapping rows). Also comfortably beyond
|
||||||
|
/// the sub-second/second-scale <c>AwaitAssert</c> windows used by the unit tests, so the
|
||||||
|
/// one-shot never auto-fires mid-test — tests drive <see cref="BackfillTick"/> by hand.
|
||||||
|
/// </summary>
|
||||||
|
private static readonly TimeSpan BackfillInitialDelay = TimeSpan.FromSeconds(15);
|
||||||
|
|
||||||
private readonly IServiceProvider _serviceProvider;
|
private readonly IServiceProvider _serviceProvider;
|
||||||
private readonly KpiHistoryOptions _options;
|
private readonly KpiHistoryOptions _options;
|
||||||
@@ -102,6 +117,24 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private bool _rollupInFlight;
|
private bool _rollupInFlight;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// In-flight guard for the one-shot backfill fold. Set true when the backfill fold is
|
||||||
|
/// launched and cleared when its <see cref="BackfillComplete"/> arrives. While true, periodic
|
||||||
|
/// <see cref="RollupTick"/>s are skipped (like <see cref="_rollupInFlight"/>) so the (possibly
|
||||||
|
/// long) full-retention-window backfill never runs concurrently with a periodic lookback fold
|
||||||
|
/// on the same upsert rows.
|
||||||
|
/// </summary>
|
||||||
|
private bool _backfillInFlight;
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Once-only latch for the backfill. Set true when the backfill fold completes (success or
|
||||||
|
/// fault) so it runs at most once per actor lifetime — a fresh singleton on failover starts a
|
||||||
|
/// new instance and re-runs the (idempotent) backfill, which is the intended self-heal. Set on
|
||||||
|
/// completion rather than launch so that if the actor is torn down mid-backfill the next
|
||||||
|
/// lifetime still attempts it.
|
||||||
|
/// </summary>
|
||||||
|
private bool _backfillDone;
|
||||||
|
|
||||||
/// <summary>Akka timer scheduler, assigned by the actor system via <see cref="IWithTimers"/>.</summary>
|
/// <summary>Akka timer scheduler, assigned by the actor system via <see cref="IWithTimers"/>.</summary>
|
||||||
public ITimerScheduler Timers { get; set; } = null!;
|
public ITimerScheduler Timers { get; set; } = null!;
|
||||||
|
|
||||||
@@ -126,6 +159,12 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
Receive<PurgeComplete>(_ => { }); // best-effort: no actor state to reset on completion
|
Receive<PurgeComplete>(_ => { }); // best-effort: no actor state to reset on completion
|
||||||
Receive<RollupTick>(_ => HandleRollupTick());
|
Receive<RollupTick>(_ => HandleRollupTick());
|
||||||
Receive<RollupComplete>(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault)
|
Receive<RollupComplete>(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault)
|
||||||
|
Receive<BackfillTick>(_ => HandleBackfillTick());
|
||||||
|
Receive<BackfillComplete>(_ =>
|
||||||
|
{
|
||||||
|
_backfillInFlight = false; // lower the in-flight guard (success or fault)
|
||||||
|
_backfillDone = true; // latch: at most once per actor lifetime
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -161,6 +200,17 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
RollupTick.Instance,
|
RollupTick.Instance,
|
||||||
initialDelay: TimeSpan.FromSeconds(10),
|
initialDelay: TimeSpan.FromSeconds(10),
|
||||||
interval: _options.RollupInterval);
|
interval: _options.RollupInterval);
|
||||||
|
|
||||||
|
// One-shot backfill: fold the whole raw-retention window of existing samples into the
|
||||||
|
// hourly rollup table once shortly after start, so long-range (30 d/90 d) trend charts
|
||||||
|
// aren't blank until enough wall-clock passes after deploy. The periodic tick only folds
|
||||||
|
// the trailing RollupLookbackHours, so without this the deep history stays unrolled. The
|
||||||
|
// fold is an idempotent upsert, so re-running on a failover (a fresh singleton re-arms
|
||||||
|
// this timer) is safe.
|
||||||
|
Timers.StartSingleTimer(
|
||||||
|
BackfillTimerKey,
|
||||||
|
BackfillTick.Instance,
|
||||||
|
timeout: BackfillInitialDelay);
|
||||||
}
|
}
|
||||||
|
|
||||||
/// <inheritdoc />
|
/// <inheritdoc />
|
||||||
@@ -382,8 +432,11 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
/// </summary>
|
/// </summary>
|
||||||
private void HandleRollupTick()
|
private void HandleRollupTick()
|
||||||
{
|
{
|
||||||
if (_rollupInFlight)
|
if (_rollupInFlight || _backfillInFlight)
|
||||||
{
|
{
|
||||||
|
// Skip while either a prior periodic fold OR the one-shot full-window backfill is
|
||||||
|
// running, so two idempotent upserts never race on overlapping rows. A skipped
|
||||||
|
// periodic tick self-heals on the next fold (it re-processes the lookback window).
|
||||||
_logger.LogDebug("KPI rollup tick skipped — a rollup fold is already in flight.");
|
_logger.LogDebug("KPI rollup tick skipped — a rollup fold is already in flight.");
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -428,6 +481,82 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Handles the one-shot backfill tick. Runs at most once per actor lifetime: if the backfill
|
||||||
|
/// has already completed (<see cref="_backfillDone"/>) or is in flight
|
||||||
|
/// (<see cref="_backfillInFlight"/>) the tick is ignored. If a periodic fold currently holds
|
||||||
|
/// the fold path (<see cref="_rollupInFlight"/>) the backfill defers by re-arming its single
|
||||||
|
/// timer, so the full-window backfill and the periodic lookback fold never race on overlapping
|
||||||
|
/// upsert rows. Otherwise it raises the backfill guard and folds the whole raw-retention
|
||||||
|
/// window <c>[TruncateToHour(now) − RetentionDays, TruncateToHour(now))</c> off the actor
|
||||||
|
/// thread, piping a <see cref="BackfillComplete"/> back to lower the guard and set the latch.
|
||||||
|
/// </summary>
|
||||||
|
private void HandleBackfillTick()
|
||||||
|
{
|
||||||
|
if (_backfillDone || _backfillInFlight)
|
||||||
|
{
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
if (_rollupInFlight)
|
||||||
|
{
|
||||||
|
// A periodic fold is in flight; defer the one-shot backfill briefly so the two
|
||||||
|
// idempotent upserts don't contend on overlapping rows. Re-arm the single timer.
|
||||||
|
Timers.StartSingleTimer(BackfillTimerKey, BackfillTick.Instance, TimeSpan.FromSeconds(5));
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
_backfillInFlight = true;
|
||||||
|
var toHourUtc = TruncateToHour(DateTime.UtcNow);
|
||||||
|
var fromHourUtc = toHourUtc - TimeSpan.FromDays(_options.RetentionDays);
|
||||||
|
var cancellationToken = _shutdownCts?.Token ?? CancellationToken.None;
|
||||||
|
|
||||||
|
_logger.LogInformation(
|
||||||
|
"KPI rollup backfill starting — folding existing samples in [{FromHour:o}, {ToHour:o}) ({RetentionDays}-day window).",
|
||||||
|
fromHourUtc, toHourUtc, _options.RetentionDays);
|
||||||
|
|
||||||
|
RunBackfillPass(fromHourUtc, toHourUtc, cancellationToken).PipeTo(
|
||||||
|
Self,
|
||||||
|
success: () => BackfillComplete.Instance,
|
||||||
|
failure: ex =>
|
||||||
|
{
|
||||||
|
_logger.LogError(ex, "KPI rollup backfill faulted unexpectedly.");
|
||||||
|
return BackfillComplete.Instance;
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Runs the one-shot backfill fold: opens a DI scope, resolves the repository, and folds the
|
||||||
|
/// complete hours in <c>[<paramref name="fromHourUtc"/>, <paramref name="toHourUtc"/>)</c> into
|
||||||
|
/// the hourly rollup table via the same idempotent upsert the periodic fold uses (Task 3). The
|
||||||
|
/// whole body is wrapped so the returned task never faults — best-effort observability must
|
||||||
|
/// never disrupt startup. The fold method returns no count, so completion is logged with the
|
||||||
|
/// window and elapsed wall time only.
|
||||||
|
/// </summary>
|
||||||
|
private async Task RunBackfillPass(
|
||||||
|
DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken)
|
||||||
|
{
|
||||||
|
var startedAt = DateTime.UtcNow;
|
||||||
|
try
|
||||||
|
{
|
||||||
|
await using var scope = _serviceProvider.CreateAsyncScope();
|
||||||
|
var repository = scope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
|
||||||
|
await repository.FoldHourlyRollupsAsync(fromHourUtc, toHourUtc, cancellationToken);
|
||||||
|
|
||||||
|
_logger.LogInformation(
|
||||||
|
"KPI rollup backfill completed — folded [{FromHour:o}, {ToHour:o}) in {ElapsedMs:F0} ms.",
|
||||||
|
fromHourUtc, toHourUtc, (DateTime.UtcNow - startedAt).TotalMilliseconds);
|
||||||
|
}
|
||||||
|
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
|
||||||
|
{
|
||||||
|
// Shutdown interrupted the backfill; a failover re-runs it (idempotent). Not a failure.
|
||||||
|
}
|
||||||
|
catch (Exception ex)
|
||||||
|
{
|
||||||
|
_logger.LogError(ex, "KPI rollup backfill failed unexpectedly.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Truncates a UTC instant to the start of its hour (minutes/seconds/ticks dropped),
|
/// Truncates a UTC instant to the start of its hour (minutes/seconds/ticks dropped),
|
||||||
/// preserving <see cref="DateTimeKind.Utc"/>. Used to derive the fold window's whole-hour
|
/// preserving <see cref="DateTimeKind.Utc"/>. Used to derive the fold window's whole-hour
|
||||||
@@ -483,4 +612,26 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
|
|||||||
public static readonly RollupComplete Instance = new();
|
public static readonly RollupComplete Instance = new();
|
||||||
private RollupComplete() { }
|
private RollupComplete() { }
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Self-tick triggering the one-shot backfill fold over the full raw-retention window. Armed
|
||||||
|
/// once via a single timer in <see cref="PreStart"/> (and re-armed only to defer past an
|
||||||
|
/// in-flight periodic fold); runs at most once per actor lifetime.
|
||||||
|
/// </summary>
|
||||||
|
internal sealed class BackfillTick
|
||||||
|
{
|
||||||
|
public static readonly BackfillTick Instance = new();
|
||||||
|
private BackfillTick() { }
|
||||||
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Piped-back completion of the one-shot backfill fold; lets the fold run off the actor thread
|
||||||
|
/// and, on the actor thread, lowers the <c>_backfillInFlight</c> guard and latches
|
||||||
|
/// <c>_backfillDone</c> (fires on success and fault).
|
||||||
|
/// </summary>
|
||||||
|
internal sealed class BackfillComplete
|
||||||
|
{
|
||||||
|
public static readonly BackfillComplete Instance = new();
|
||||||
|
private BackfillComplete() { }
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -662,4 +662,106 @@ public class KpiHistoryRecorderActorTests : TestKit
|
|||||||
duration: TimeSpan.FromSeconds(3),
|
duration: TimeSpan.FromSeconds(3),
|
||||||
interval: TimeSpan.FromMilliseconds(50));
|
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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user