diff --git a/docs/requirements/Component-KpiHistory.md b/docs/requirements/Component-KpiHistory.md index 96c5eea2..3f6251e8 100644 --- a/docs/requirements/Component-KpiHistory.md +++ b/docs/requirements/Component-KpiHistory.md @@ -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. diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs index 7fb4703f..a0692df4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs +++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/KpiHistoryRepository.cs @@ -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."); + } + } + + /// + /// True when is a unique-key violation: SQL Server 2601/2627 in + /// production, or the SQLite UNIQUE constraint failed fault on the test provider + /// (matched by type name, as ScadaBridgeDbContext does, so this project takes no + /// SQLite dependency). + /// + 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); + + /// + /// Applies one hour's grouped samples to the rollup table: adds a row per new series-hour + /// or overwrites the preloaded row in place. + /// + private void FoldHour( + IEnumerable> groups, + Dictionary 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."); - } } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs index 95248e40..39c1aa97 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/KpiHistoryRepositoryTests.cs @@ -564,6 +564,123 @@ public class KpiHistoryRepositoryTests Assert.Equal(5, series[0].Value); // the fold's own value overwrote the side-inserted -1 } + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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(); + 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()); + } + + /// + /// 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). + /// + /// A task that represents the asynchronous operation. + [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(() => repo.FoldHourlyRollupsAsync(Base, Base.AddHours(1))); + } + + /// Fails any save that writes a rollup row with a non-constraint fault. + private sealed class ThrowOnRollupSave : Microsoft.EntityFrameworkCore.Diagnostics.SaveChangesInterceptor + { + /// + public override ValueTask> SavingChangesAsync( + DbContextEventData eventData, InterceptionResult result, + CancellationToken cancellationToken = default) + { + if (eventData.Context is not null + && eventData.Context.ChangeTracker.Entries().Any()) + { + throw new DbUpdateException("simulated non-race failure (e.g. a truncation fault)"); + } + + return base.SavingChangesAsync(eventData, result, cancellationToken); + } + } + + /// + /// Records, for every reader command against KpiSample, the span between its + /// smallest and largest DateTime parameter — i.e. the time window that read covers. + /// + private sealed class SampleReadWindowInterceptor : DbCommandInterceptor + { + /// The time window covered by each recorded KpiSample read. + public List Windows { get; } = new(); + + private void Record(DbCommand command) + { + if (!command.CommandText.Contains("FROM \"KpiSample\"", StringComparison.OrdinalIgnoreCase)) + { + return; + } + + var instants = command.Parameters.Cast() + .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()); + } + } + + /// + public override InterceptionResult ReaderExecuting( + DbCommand command, CommandEventData eventData, InterceptionResult result) + { + Record(command); + return base.ReaderExecuting(command, eventData, result); + } + + /// + public override ValueTask> ReaderExecutingAsync( + DbCommand command, CommandEventData eventData, InterceptionResult 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);