fix(kpi-history): fold raw samples one hour at a time; only a unique-key violation is the benign fold race (review 2026-10-21 06/PERF-6, 08/P4, 08/C3)
This commit is contained in:
Joseph Doherty
@@ -90,9 +90,9 @@ A timer fires every `SampleInterval` (default 60s; a first tick after five secon
|
||||
|
||||
Each source has a `SourceTimeout` (default 10 seconds) bounding the actual wait as well as requesting cancellation. A source that ignores cancellation keeps its task, cancellation source, and DI scope until it actually finishes; later ticks skip that registration instead of starting replacements. Healthy sources continue with separate scopes, so they do not share a busy `DbContext` with unfinished work. A root-DI singleton admits at most one unfinished collection per registration across recorder instances, including actor stop/recreation on the same provider. Admission is atomic; replacing an actor does not start another call against a still-running source. Different root providers have independent admission state.
|
||||
|
||||
**Hourly rollup tick.** A third timer (`kpi-rollup`) fires every `RollupInterval` (default 1h). On each tick the recorder opens a per-tick DI scope and calls `IKpiHistoryRepository.FoldHourlyRollupsAsync` over the window `[TruncateToHour(now) − RollupLookbackHours, TruncateToHour(now))` — an **exclusive** upper bound at the current hour start, so the in-progress hour is never folded. The fold groups raw `KpiSample` rows by `(series, hour)`, applies the per-metric aggregation intent (sum for rate metrics, last-value for gauges — see Query), and **idempotently upserts** each `(Source, Metric, Scope, ScopeKey, HourStartUtc)` row. Re-folding the trailing lookback window (default 3h) rather than only the last hour is the self-heal for a **singleton-failover-missed tick**: because the upsert overwrites the aggregate in place, re-running the same window produces identical values (no double-count), so any complete hour a missed tick skipped is refilled on the next fold. A `_rollupInFlight` guard (mirroring `_sampleInFlight`) skips a tick while a prior fold is still running so two idempotent upserts never race on overlapping rows. During a **failover overlap** — the old singleton incarnation's in-flight fold and the new node's first fold both `Add`-ing the same `(series, hour)` row — the loser of the unique `IX_KpiRollupHourly_Series` upsert race discards its **whole fold pass** (a single `SaveChanges`), logged at Information and repaired by the next idempotent fold; when reading fold logs after a failover, the failure grain is the **pass, not the row** (arch-review 04 round 2, R5).
|
||||
**Hourly rollup tick.** A third timer (`kpi-rollup`) fires every `RollupInterval` (default 1h). On each tick the recorder opens a per-tick DI scope and calls `IKpiHistoryRepository.FoldHourlyRollupsAsync` over the window `[TruncateToHour(now) − RollupLookbackHours, TruncateToHour(now))` — an **exclusive** upper bound at the current hour start, so the in-progress hour is never folded. The fold groups raw `KpiSample` rows by `(series, hour)`, applies the per-metric aggregation intent (sum for rate metrics, last-value for gauges — see Query), and **idempotently upserts** each `(Source, Metric, Scope, ScopeKey, HourStartUtc)` row. Re-folding the trailing lookback window (default 3h) rather than only the last hour is the self-heal for a **singleton-failover-missed tick**: because the upsert overwrites the aggregate in place, re-running the same window produces identical values (no double-count), so any complete hour a missed tick skipped is refilled on the next fold. A `_rollupInFlight` guard (mirroring `_sampleInFlight`) skips a tick while a prior fold is still running so two idempotent upserts never race on overlapping rows. During a **failover overlap** — the old singleton incarnation's in-flight fold and the new node's first fold both `Add`-ing the same `(series, hour)` row — the loser of the unique `IX_KpiRollupHourly_Series` upsert race discards its **whole fold pass** (a single `SaveChanges`), logged at Information and repaired by the next idempotent fold; when reading fold logs after a failover, the failure grain is the **pass, not the row** (arch-review 04 round 2, R5). Only a unique-key violation (SQL Server 2601/2627; the SQLite `UNIQUE constraint failed` fault on the test provider) is classified as that benign race — any other save failure propagates and the recorder logs it at Error, so a fold that faults for a real reason is no longer invisible.
|
||||
|
||||
**One-shot backfill.** So the 30 d / 90 d charts are not blank until enough wall-clock passes, the recorder runs a backfill on start (`kpi-rollup-backfill` timer) that folds the retention window, reusing the same `FoldHourlyRollupsAsync` path. The `FoldHourlyRollupsAsync` repository call materializes its whole window in memory, so the backfill folds the window in **≤24 h slices, oldest-first** — one bounded slice per fold — rather than handing the fold the full 90-day window at once (which at fleet volume is tens of millions of rows in one tracked pass). Between slices it lowers its in-flight guard so periodic folds **interleave** instead of stalling behind the historical pass; a periodic fold and a backfill slice strictly alternate through the single `_rollupInFlight`/`_backfillInFlight` gate pair, so the two idempotent upserts never race on overlapping rows. On a **failover restart** the plan first consults the `IKpiHistoryRepository.GetLatestRollupHourAsync` watermark: if the newest folded hour is already within `RollupLookbackHours` of now (the common case) the whole historical pass is **skipped** — the periodic fold's lookback covers the tail — otherwise the window is **shrunk to the un-rolled tail** `[watermark, TruncateToHour(now))` (re-folding the watermark hour itself as an idempotent safety margin) instead of the full `RetentionDays` floor. A `_backfillDone` latch makes the whole pass at-most-once per actor lifetime, and it defers (re-arms) while a periodic fold is in flight. Being an idempotent upsert, it is safe (idempotent) to re-run on a later failover restart; the watermark fast-path makes the failover re-run cheap as well.
|
||||
**One-shot backfill.** So the 30 d / 90 d charts are not blank until enough wall-clock passes, the recorder runs a backfill on start (`kpi-rollup-backfill` timer) that folds the retention window, reusing the same `FoldHourlyRollupsAsync` path. The `FoldHourlyRollupsAsync` repository call reads raw samples **one hour at a time** (an untracked projection of a single hour, grouped in memory by series), so resident memory is bounded by one hour of samples regardless of the window; the backfill still folds the window in **≤24 h slices, oldest-first** — one bounded slice per fold — rather than handing the fold the full 90-day window at once (which at fleet volume is tens of millions of rows in one tracked pass). Between slices it lowers its in-flight guard so periodic folds **interleave** instead of stalling behind the historical pass; a periodic fold and a backfill slice strictly alternate through the single `_rollupInFlight`/`_backfillInFlight` gate pair, so the two idempotent upserts never race on overlapping rows. On a **failover restart** the plan first consults the `IKpiHistoryRepository.GetLatestRollupHourAsync` watermark: if the newest folded hour is already within `RollupLookbackHours` of now (the common case) the whole historical pass is **skipped** — the periodic fold's lookback covers the tail — otherwise the window is **shrunk to the un-rolled tail** `[watermark, TruncateToHour(now))` (re-folding the watermark hour itself as an idempotent safety margin) instead of the full `RetentionDays` floor. A `_backfillDone` latch makes the whole pass at-most-once per actor lifetime, and it defers (re-arms) while a periodic fold is in flight. Being an idempotent upsert, it is safe (idempotent) to re-run on a later failover restart; the watermark fast-path makes the failover re-run cheap as well.
|
||||
|
||||
**Retention.** A daily purge timer (`PurgeInterval`, default 24h) now runs **both** purges: raw `KpiSample` rows older than `RetentionDays` (default 90) via `IKpiHistoryRepository.PurgeOlderThanAsync`, **and** hourly rollup rows older than `RollupRetentionDays` (default 365) via `IKpiHistoryRepository.PurgeRollupsOlderThanAsync` (a one-hour-sliced batched DELETE mirroring the raw purge). Rollups deliberately **outlive** raw samples so long-range trend charts have data after the raw rows are purged; the options validator enforces `RollupRetentionDays >= RetentionDays` so a window never falls into a hole where raw is purged but no rollup exists.
|
||||
|
||||
|
||||
+68
-38
@@ -107,27 +107,6 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
|
||||
return;
|
||||
}
|
||||
|
||||
// The caller passes a small trailing lookback (e.g. 3 h), so the window is bounded:
|
||||
// fetch it and group IN MEMORY. This deliberately avoids provider-specific
|
||||
// DATEADD/DATEPART hour-truncation translation (SQLite in tests, SQL Server in prod),
|
||||
// keeping the hour-bucket arithmetic identical across providers and trivially correct.
|
||||
// Projected, untracked fetch: the fold only reads these six fields and never
|
||||
// mutates a KpiSample. A tracked ToListAsync registered ~5k-90k read-only
|
||||
// entities in the change tracker per healthy 3h fold — all re-scanned by
|
||||
// DetectChanges on the final SaveChanges (arch-review 04 round 2, R2). A
|
||||
// projection is inherently untracked and materializes no entity at all.
|
||||
var samples = await _context.KpiSamples
|
||||
.Where(s => s.CapturedAtUtc >= from && s.CapturedAtUtc < to)
|
||||
.Select(s => new FoldSample(s.Source, s.Metric, s.Scope, s.ScopeKey, s.CapturedAtUtc, s.Value))
|
||||
.ToListAsync(cancellationToken);
|
||||
if (samples.Count == 0)
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var groups = samples.GroupBy(s => new SeriesHourKey(
|
||||
s.Source, s.Metric, s.Scope, s.ScopeKey, TruncateToHour(s.CapturedAtUtc)));
|
||||
|
||||
// Preload every rollup row already covering this window in ONE query and
|
||||
// index it by series+hour. The predecessor issued a
|
||||
// FirstOrDefaultAsync existence probe PER (series, hour) group — an N+1
|
||||
@@ -159,6 +138,74 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
|
||||
DateTime.SpecifyKind(row.HourStartUtc, DateTimeKind.Utc))] = row;
|
||||
}
|
||||
|
||||
// Read and fold the raw samples ONE HOUR AT A TIME (review 2026-10-21 06/PERF-6,
|
||||
// 08/P4). The periodic fold's 3 h lookback is small, but the startup backfill drains
|
||||
// 24 h slices: materializing a whole slice at once held ~0.3-0.7 M FoldSample records
|
||||
// resident on the active central node. Per-hour reads bound residency to one hour of
|
||||
// samples for the price of one cheap range seek per hour. Grouping stays IN MEMORY
|
||||
// within the hour: that avoids provider-specific DATEADD/DATEPART hour-truncation
|
||||
// translation (SQLite in tests, SQL Server in prod) and keeps the gauge last-value
|
||||
// pick trivially correct, and within one hour the series key alone is the group.
|
||||
//
|
||||
// Projected, untracked fetch: the fold only reads these six fields and never
|
||||
// mutates a KpiSample. A tracked ToListAsync registered ~5k-90k read-only
|
||||
// entities in the change tracker per healthy 3h fold — all re-scanned by
|
||||
// DetectChanges on the final SaveChanges (arch-review 04 round 2, R2). A
|
||||
// projection is inherently untracked and materializes no entity at all.
|
||||
for (var hourStart = from; hourStart < to; hourStart = hourStart.AddHours(1))
|
||||
{
|
||||
var hourEnd = hourStart.AddHours(1);
|
||||
var samples = await _context.KpiSamples
|
||||
.Where(s => s.CapturedAtUtc >= hourStart && s.CapturedAtUtc < hourEnd)
|
||||
.Select(s => new FoldSample(s.Source, s.Metric, s.Scope, s.ScopeKey, s.CapturedAtUtc, s.Value))
|
||||
.ToListAsync(cancellationToken);
|
||||
|
||||
var groups = samples.GroupBy(s => new SeriesHourKey(
|
||||
s.Source, s.Metric, s.Scope, s.ScopeKey, hourStart));
|
||||
FoldHour(groups, existingByKey);
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
catch (DbUpdateException ex) when (IsUniqueKeyViolation(ex))
|
||||
{
|
||||
// Failover-overlap race: the old singleton incarnation's in-flight fold and the
|
||||
// new node's first fold can both Add the same (series, hour) row; the unique
|
||||
// IX_KpiRollupHourly_Series then faults the loser's ENTIRE SaveChanges — the
|
||||
// failure grain is the PASS, not the row (arch-review 04 round 2, R5). A lost
|
||||
// upsert race is repaired identically by the next idempotent re-fold, so it is
|
||||
// classified at Information. ONLY a unique-key violation is: any other save
|
||||
// fault (truncation, constraint, connectivity) propagates to the recorder, which
|
||||
// logs it at Error — previously every DbUpdateException was swallowed here as a
|
||||
// benign race, so a backfill slice faulting for a real reason stayed un-rolled and
|
||||
// invisible until the next node start (review 2026-10-21 08/C3).
|
||||
_logger.LogInformation(ex,
|
||||
"KPI rollup fold lost a failover-overlap upsert race; pass discarded, next fold self-heals.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// True when <paramref name="ex"/> is a unique-key violation: SQL Server 2601/2627 in
|
||||
/// production, or the SQLite <c>UNIQUE constraint failed</c> fault on the test provider
|
||||
/// (matched by type name, as <c>ScadaBridgeDbContext</c> does, so this project takes no
|
||||
/// SQLite dependency).
|
||||
/// </summary>
|
||||
private static bool IsUniqueKeyViolation(DbUpdateException ex) =>
|
||||
ex.InnerException is Microsoft.Data.SqlClient.SqlException sql
|
||||
? sql.Number is 2601 or 2627
|
||||
: ex.InnerException?.GetType().Name == "SqliteException"
|
||||
&& ex.InnerException.Message.Contains("UNIQUE constraint failed", StringComparison.Ordinal);
|
||||
|
||||
/// <summary>
|
||||
/// Applies one hour's grouped samples to the rollup table: adds a row per new series-hour
|
||||
/// or overwrites the preloaded row in place.
|
||||
/// </summary>
|
||||
private void FoldHour(
|
||||
IEnumerable<IGrouping<SeriesHourKey, FoldSample>> groups,
|
||||
Dictionary<SeriesHourKey, KpiRollupHourly> existingByKey)
|
||||
{
|
||||
foreach (var group in groups)
|
||||
{
|
||||
var key = group.Key;
|
||||
@@ -216,23 +263,6 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
|
||||
existing.SampleCount = sampleCount;
|
||||
}
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await _context.SaveChangesAsync(cancellationToken);
|
||||
}
|
||||
catch (DbUpdateException ex)
|
||||
{
|
||||
// Failover-overlap race: the old singleton incarnation's in-flight fold and the
|
||||
// new node's first fold can both Add the same (series, hour) row; the unique
|
||||
// IX_KpiRollupHourly_Series then faults the loser's ENTIRE SaveChanges — the
|
||||
// failure grain is the PASS, not the row (arch-review 04 round 2, R5). The fold
|
||||
// only ever writes KpiRollupHourly rows, so any save fault here is a lost upsert
|
||||
// race or a transient the next idempotent re-fold repairs identically; classify
|
||||
// at Information instead of surfacing a scary error for a self-healing no-op.
|
||||
_logger.LogInformation(ex,
|
||||
"KPI rollup fold lost a failover-overlap upsert race; pass discarded, next fold self-heals.");
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
|
||||
+117
@@ -564,6 +564,123 @@ public class KpiHistoryRepositoryTests
|
||||
Assert.Equal(5, series[0].Value); // the fold's own value overwrote the side-inserted -1
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a multi-hour fold reads raw samples ONE HOUR AT A TIME, so a 24 h backfill
|
||||
/// slice never materializes the whole slice's samples at once (review 2026-10-21 06/PERF-6,
|
||||
/// 08/P4), and still folds every hour correctly.
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task FoldHourlyRollupsAsync_ReadsSamplesOneHourAtATime()
|
||||
{
|
||||
var recorder = new SampleReadWindowInterceptor();
|
||||
await using var ctx = SqliteTestHelper.CreateInMemoryContext(recorder);
|
||||
var repo = new KpiHistoryRepository(ctx);
|
||||
var samples = new List<KpiSample>();
|
||||
for (var hour = 0; hour < 5; hour++)
|
||||
{
|
||||
samples.Add(Sample("NotificationOutbox", "queueDepth", "Global", null, hour, Base.AddHours(hour).AddMinutes(5)));
|
||||
samples.Add(Sample("NotificationOutbox", "queueDepth", "Global", null, hour + 10, Base.AddHours(hour).AddMinutes(55)));
|
||||
}
|
||||
|
||||
await repo.RecordSamplesAsync(samples);
|
||||
|
||||
await repo.FoldHourlyRollupsAsync(Base, Base.AddHours(5));
|
||||
|
||||
Assert.NotEmpty(recorder.Windows);
|
||||
Assert.All(recorder.Windows, w => Assert.True(
|
||||
w <= TimeSpan.FromHours(1), $"a KpiSample read spanned {w}; expected at most one hour"));
|
||||
var series = await repo.GetHourlySeriesAsync(
|
||||
"NotificationOutbox", "queueDepth", "Global", null, Base, Base.AddHours(5));
|
||||
Assert.Equal(new[] { 10d, 11d, 12d, 13d, 14d }, series.Select(p => p.Value).ToArray());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Verifies that a save failure that is NOT a unique-key race propagates instead of being
|
||||
/// swallowed at Information as a benign failover race (review 2026-10-21 08/C3).
|
||||
/// </summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[Fact]
|
||||
public async Task Fold_NonRaceSaveFailure_Propagates()
|
||||
{
|
||||
await using var ctx = SqliteTestHelper.CreateInMemoryContext(new ThrowOnRollupSave());
|
||||
var repo = new KpiHistoryRepository(ctx);
|
||||
await repo.RecordSamplesAsync(new[]
|
||||
{
|
||||
Sample("NotificationOutbox", "queueDepth", "Global", null, 5, Base.AddMinutes(10)),
|
||||
});
|
||||
|
||||
await Assert.ThrowsAsync<DbUpdateException>(() => repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1)));
|
||||
}
|
||||
|
||||
/// <summary>Fails any save that writes a rollup row with a non-constraint fault.</summary>
|
||||
private sealed class ThrowOnRollupSave : Microsoft.EntityFrameworkCore.Diagnostics.SaveChangesInterceptor
|
||||
{
|
||||
/// <inheritdoc />
|
||||
public override ValueTask<InterceptionResult<int>> SavingChangesAsync(
|
||||
DbContextEventData eventData, InterceptionResult<int> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
if (eventData.Context is not null
|
||||
&& eventData.Context.ChangeTracker.Entries<KpiRollupHourly>().Any())
|
||||
{
|
||||
throw new DbUpdateException("simulated non-race failure (e.g. a truncation fault)");
|
||||
}
|
||||
|
||||
return base.SavingChangesAsync(eventData, result, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Records, for every reader command against <c>KpiSample</c>, the span between its
|
||||
/// smallest and largest DateTime parameter — i.e. the time window that read covers.
|
||||
/// </summary>
|
||||
private sealed class SampleReadWindowInterceptor : DbCommandInterceptor
|
||||
{
|
||||
/// <summary>The time window covered by each recorded KpiSample read.</summary>
|
||||
public List<TimeSpan> Windows { get; } = new();
|
||||
|
||||
private void Record(DbCommand command)
|
||||
{
|
||||
if (!command.CommandText.Contains("FROM \"KpiSample\"", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return;
|
||||
}
|
||||
|
||||
var instants = command.Parameters.Cast<DbParameter>()
|
||||
.Select(p => p.Value switch
|
||||
{
|
||||
DateTime d => (DateTime?)d,
|
||||
string text when DateTime.TryParse(text, out var parsed) => parsed,
|
||||
_ => null,
|
||||
})
|
||||
.Where(d => d is not null)
|
||||
.Select(d => d!.Value)
|
||||
.ToList();
|
||||
if (instants.Count >= 2)
|
||||
{
|
||||
Windows.Add(instants.Max() - instants.Min());
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override InterceptionResult<DbDataReader> ReaderExecuting(
|
||||
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result)
|
||||
{
|
||||
Record(command);
|
||||
return base.ReaderExecuting(command, eventData, result);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public override ValueTask<InterceptionResult<DbDataReader>> ReaderExecutingAsync(
|
||||
DbCommand command, CommandEventData eventData, InterceptionResult<DbDataReader> result,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
Record(command);
|
||||
return base.ReaderExecutingAsync(command, eventData, result, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
// Local mirror of the repo's private hour-truncation, for cutoff assertions.
|
||||
private static DateTime TruncateHour(DateTime utc) =>
|
||||
new(utc.Year, utc.Month, utc.Day, utc.Hour, 0, 0, DateTimeKind.Utc);
|
||||
|
||||
Reference in New Issue
Block a user