merge(r2): r2-plan04

This commit is contained in:
Joseph Doherty
2026-07-13 11:09:36 -04:00
26 changed files with 3427 additions and 142 deletions
@@ -61,32 +61,42 @@ public sealed class SiteAuditRetentionService : IHostedService, IDisposable
private async Task RunLoopAsync(CancellationToken ct)
{
// InitialDelay before the first purge — short so a daily-recycled node
// still purges soon after start rather than waiting a full interval it may
// never reach.
try
{
await Task.Delay(_options.InitialDelay, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
return;
}
await SafePurgeAsync(ct).ConfigureAwait(false);
while (!ct.IsCancellationRequested)
{
// InitialDelay before the first purge — short so a daily-recycled node
// still purges soon after start rather than waiting a full interval it may
// never reach.
try
{
await Task.Delay(_options.ResolvedPurgeInterval, ct).ConfigureAwait(false);
await Task.Delay(_options.InitialDelay, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break;
return;
}
await SafePurgeAsync(ct).ConfigureAwait(false);
while (!ct.IsCancellationRequested)
{
try
{
await Task.Delay(_options.ResolvedPurgeInterval, ct).ConfigureAwait(false);
}
catch (OperationCanceledException)
{
break;
}
await SafePurgeAsync(ct).ConfigureAwait(false);
}
}
catch (OperationCanceledException)
{
// Shutdown landed mid-purge: SafePurgeAsync rethrows OCE by design so the purge
// aborts promptly, but the loop task must complete CLEANLY — StopAsync hands
// _loop straight to the host, and a canceled task there is shutdown-log noise
// (arch-review 04 round 2, R7). Cancellation here IS the clean exit.
}
}
@@ -51,6 +51,14 @@ public partial class KpiTrendChart
/// The series to plot, assumed ordered ascending by
/// <see cref="KpiSeriesPoint.BucketStartUtc"/>. <c>null</c> or empty renders
/// the unavailable placeholder rather than an empty chart.
/// <para>
/// The per-point value's meaning depends on the metric's
/// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi.KpiMetricAggregationCatalog"/>
/// classification: a Rate-classified series carries <b>per-bucket totals</b> (the
/// bucketer sums each bucket's deltas), while a Gauge series carries
/// last-value-per-bucket readings. The chart is agnostic to which — it plots the
/// supplied values verbatim (arch-review 04 round 2, R3).
/// </para>
/// </summary>
[Parameter] public IReadOnlyList<KpiSeriesPoint>? Points { get; set; }
@@ -58,7 +66,10 @@ public partial class KpiTrendChart
/// <c>data-test</c> slug. Required-ish; defaults to "Trend" if blank.</summary>
[Parameter] public string Title { get; set; } = "Trend";
/// <summary>Optional unit suffix appended to value labels (e.g. "s").</summary>
/// <summary>Optional unit suffix appended to value labels (e.g. "s"). For a
/// Rate-classified metric the supplied text must state the per-bucket-total
/// semantics (e.g. "delivered/bucket"), not a per-minute/per-interval reading,
/// so the label matches the summed values in <see cref="Points"/> (R3).</summary>
[Parameter] public string? Unit { get; set; }
/// <summary>
@@ -76,12 +76,18 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
{
var effectiveMax = maxPoints ?? _options.DefaultMaxSeriesPoints;
// Per-metric reduction intent: the same catalog that drives the hourly fold drives
// the chart-time bucket reduction, so a Rate series is summed per bucket on BOTH the
// raw-minute and hourly-rollup routes — one unit ("events per chart bucket") on both
// sides of the RollupThresholdHours routing boundary (arch-review 04 round 2, R3).
var aggregation = KpiMetricAggregationCatalog.Resolve(source, metric);
// Test-seam ctor: use the injected repository directly.
if (_injectedRepository is not null)
{
var injectedSeries = await FetchSeriesAsync(
_injectedRepository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax);
return KpiSeriesBucketer.Bucket(injectedSeries, fromUtc, toUtc, effectiveMax, aggregation);
}
// Production: a fresh scope (and thus a fresh DbContext) per query so a
@@ -90,7 +96,7 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
var repository = serviceScope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
var series = await FetchSeriesAsync(
repository, source, metric, scope, scopeKey, fromUtc, toUtc, cancellationToken);
return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax);
return KpiSeriesBucketer.Bucket(series, fromUtc, toUtc, effectiveMax, aggregation);
}
/// <summary>
@@ -103,6 +109,13 @@ public sealed class KpiHistoryQueryService : IKpiHistoryQueryService
/// the caller's <see cref="KpiSeriesBucketer.Bucket"/> step is agnostic to the source
/// (a 90 d rollup can still exceed the point ceiling, so bucketing still applies).
/// The boundary is strict: exactly <c>RollupThresholdHours</c> stays on the raw path.
/// <para>
/// The routing boundary no longer changes a Rate chart's magnitude: both paths reduce to
/// per-bucket sums via the catalog-driven aggregation applied in
/// <see cref="GetSeriesAsync"/>. A residual ≤~1.2× native-resolution difference exists only
/// for an unbucketed just-over-threshold rollup series (plotted at native per-hour sums) vs.
/// an at-threshold bucketed raw series — documented, not hidden (arch-review 04 round 2, R3).
/// </para>
/// </summary>
private Task<IReadOnlyList<KpiSeriesPoint>> FetchSeriesAsync(
IKpiHistoryRepository repository,
@@ -109,6 +109,16 @@ public interface IKpiHistoryRepository
string source, string metric, string scope, string? scopeKey,
DateTime fromUtc, DateTime toUtc, CancellationToken cancellationToken = default);
/// <summary>
/// Returns the newest <c>KpiRollupHourly.HourStartUtc</c> across all series, or
/// <c>null</c> when no rollups exist. The recorder's one-shot backfill consults
/// this watermark to skip (or shrink to) the un-rolled tail instead of re-folding
/// the whole raw-retention window on every failover (arch-review 04 round 2, R1).
/// </summary>
/// <param name="cancellationToken">Cancellation token.</param>
/// <returns>A task resolving to the newest folded hour start, or <c>null</c> when empty.</returns>
Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default);
/// <summary>
/// Bulk-deletes hourly rollup rows whose
/// <see cref="ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi.KpiRollupHourly.HourStartUtc"/> is
@@ -42,10 +42,10 @@ public enum KpiRollupAggregation
/// Keying by <c>(source, metric)</c> — not metric alone — is deliberate:
/// <c>deliveredLastInterval</c> is emitted by both
/// <see cref="KpiSources.NotificationOutbox"/> and <see cref="KpiSources.SiteCallAudit"/>,
/// so a metric name is not globally unique. Every metric string is taken from the
/// emitting source verbatim (shared <see cref="KpiMetrics"/> constants where they exist,
/// otherwise the source's own private literal), so this catalog stays a faithful mirror
/// of what actually lands in <c>KpiSample.Metric</c>.
/// so a metric name is not globally unique. Every catalogued metric string is now a shared
/// <see cref="KpiMetrics"/> constant — the emitting source consumes the same constant — so an
/// emitter rename is a <strong>compile error</strong>, not a silent Gauge downgrade (R4), and
/// this catalog stays a faithful mirror of what actually lands in <c>KpiSample.Metric</c>.
/// </para>
/// <para>
/// The catalog stores only the <see cref="KpiRollupAggregation.Rate"/> pairs; anything
@@ -57,12 +57,6 @@ public enum KpiRollupAggregation
/// </remarks>
public static class KpiMetricAggregationCatalog
{
// Metric literals for the sources that keep their names private (no shared
// KpiMetrics constant exists for these). Charted metrics use the KpiMetrics catalog.
private const string SiteCallAuditDeliveredLastInterval = "deliveredLastInterval";
private const string SiteHealthAlarmEvalErrors = "alarmEvalErrors";
private const string SiteHealthEventLogWriteFailures = "eventLogWriteFailures";
/// <summary>
/// The <c>(source, metric)</c> pairs classified as <see cref="KpiRollupAggregation.Rate"/>
/// (sum-per-hour). Every other emitted metric is a <see cref="KpiRollupAggregation.Gauge"/>.
@@ -99,11 +93,11 @@ public static class KpiMetricAggregationCatalog
{
(KpiSources.NotificationOutbox, KpiMetrics.NotificationOutbox.DeliveredLastInterval),
(KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.FailedLastInterval),
(KpiSources.SiteCallAudit, SiteCallAuditDeliveredLastInterval),
(KpiSources.SiteCallAudit, KpiMetrics.SiteCallAudit.DeliveredLastInterval),
(KpiSources.SiteHealth, KpiMetrics.SiteHealth.ScriptErrors),
(KpiSources.SiteHealth, SiteHealthAlarmEvalErrors),
(KpiSources.SiteHealth, KpiMetrics.SiteHealth.AlarmEvalErrors),
(KpiSources.SiteHealth, KpiMetrics.SiteHealth.DeadLetters),
(KpiSources.SiteHealth, SiteHealthEventLogWriteFailures),
(KpiSources.SiteHealth, KpiMetrics.SiteHealth.EventLogWriteFailures),
};
/// <summary>
@@ -18,11 +18,16 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
/// <strong>not</strong> a rename — changing any value would orphan historical samples.
/// </para>
/// <para>
/// Only the metrics a trend page actually charts are catalogued here. Sources may emit
/// additional internal metrics (e.g. the Notification Outbox <c>stuckCount</c> /
/// <c>oldestPendingAgeSeconds</c>, the Site Call Audit <c>deliveredLastInterval</c> /
/// <c>stuck</c> / <c>oldestPendingAgeSeconds</c>, and the bulk of the Site Health catalog)
/// that no page references; those stay private to their source until a page charts them.
/// This catalog now carries two kinds of metric: those a trend page charts, <em>and</em>
/// those the <see cref="KpiMetricAggregationCatalog"/> classifies for the hourly rollup fold
/// (e.g. <c>SiteCallAudit/deliveredLastInterval</c>, <c>SiteHealth/alarmEvalErrors</c>,
/// <c>SiteHealth/eventLogWriteFailures</c>) even if no page charts them yet. The C4/R4 lesson
/// is that <strong>any metric named in two places needs one shared symbol</strong>: a private
/// literal on the emitter plus a matching private literal in the catalog drifts silently on a
/// rename. Sources may still keep genuinely single-use internal metrics (e.g. the Notification
/// Outbox <c>stuckCount</c> / <c>oldestPendingAgeSeconds</c>, the Site Call Audit <c>stuck</c> /
/// <c>oldestPendingAgeSeconds</c>, and the bulk of the Site Health catalog) private until a page
/// or the catalog references them.
/// Constants are grouped by source via nested static classes mirroring
/// <see cref="KpiSources"/>.
/// </para>
@@ -65,6 +70,11 @@ public static class KpiMetrics
/// <summary>Cached calls that failed permanently in the last KPI interval.</summary>
public const string FailedLastInterval = "failedLastInterval";
/// <summary>Cached calls delivered in the last KPI interval. Not charted by a trend
/// page today, but named in the rollup aggregation catalog as a Rate metric — so it
/// is a shared symbol to keep the emitter and catalog in lock-step (R4).</summary>
public const string DeliveredLastInterval = "deliveredLastInterval";
}
/// <summary>
@@ -98,6 +108,16 @@ public static class KpiMetrics
/// <summary>Script-execution error count reported by the site.</summary>
public const string ScriptErrors = "scriptErrors";
/// <summary>Alarm-evaluation error count reported by the site. Named in the rollup
/// aggregation catalog as a Rate metric — shared symbol so an emitter rename is a
/// compile error, not a silent Gauge downgrade (R4).</summary>
public const string AlarmEvalErrors = "alarmEvalErrors";
/// <summary>Event-log write-failure count reported by the site. Named in the rollup
/// aggregation catalog as a Rate metric — shared symbol so an emitter rename is a
/// compile error, not a silent Gauge downgrade (R4).</summary>
public const string EventLogWriteFailures = "eventLogWriteFailures";
/// <summary>Summed store-and-forward buffer depth across all buffers.</summary>
public const string SfBufferDepth = "sfBufferDepth";
}
@@ -3,8 +3,14 @@ namespace ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
/// <summary>
/// Pure, deterministic downsampling helper for KPI series charting ("KPI History &amp; Trends").
/// Reduces a raw <see cref="KpiSeriesPoint"/> series to at most <c>maxPoints</c> points using
/// last-value-per-bucket / gauge semantics — suitable for step/area charts where the most
/// recent value in a window best represents that window.
/// <em>per-metric</em> bucket reduction driven by <see cref="KpiRollupAggregation"/>:
/// <see cref="KpiRollupAggregation.Gauge"/> (the default) keeps the <strong>last value</strong>
/// in each bucket — suitable for step/area charts where the most recent reading best represents
/// the window; <see cref="KpiRollupAggregation.Rate"/> <strong>sums per bucket</strong> — each
/// raw point of a Rate series is a per-interval delta, so the bucket's true total is the sum of
/// its deltas. Summing on both the raw-minute and hourly-rollup paths charts the same unit
/// ("events per chart bucket") on both sides of the raw/rollup routing boundary, eliminating the
/// ~60× magnitude jump at that boundary (arch-review 04 round 2, R3).
/// </summary>
public static class KpiSeriesBucketer
{
@@ -27,12 +33,20 @@ public static class KpiSeriesBucketer
/// <param name="fromUtc">UTC start of the query window (inclusive).</param>
/// <param name="toUtc">UTC end of the query window (inclusive on the right edge).</param>
/// <param name="maxPoints">Maximum number of output points. Must be ≥ 2.</param>
/// <param name="aggregation">
/// How each bucket is reduced. <see cref="KpiRollupAggregation.Gauge"/> (default) keeps the
/// last value in the bucket — the historical behavior. <see cref="KpiRollupAggregation.Rate"/>
/// sums the values in the bucket — each raw point of a Rate series is a per-interval delta, so
/// the bucket total is the sum of its deltas (arch-review 04 round 2, R3).
/// </param>
/// <returns>
/// An <see cref="IReadOnlyList{T}"/> of at most <paramref name="maxPoints"/> bucketed points,
/// ordered by <see cref="KpiSeriesPoint.BucketStartUtc"/> ascending.
/// Returns <paramref name="raw"/> unchanged (same reference) when
/// <c>raw.Count &lt;= maxPoints</c>; callers must not mutate the underlying
/// collection in that case, as it is the same object passed in.
/// For <see cref="KpiRollupAggregation.Gauge"/> returns <paramref name="raw"/> unchanged
/// (same reference) when <c>raw.Count &lt;= maxPoints</c>; callers must not mutate the
/// underlying collection in that case. <see cref="KpiRollupAggregation.Rate"/> always runs
/// the bucketing pass — two deltas landing in one bucket must sum even in a short series —
/// so it never returns the input reference.
/// </returns>
/// <exception cref="ArgumentOutOfRangeException">
/// Thrown when <paramref name="maxPoints"/> &lt; 2 or
@@ -44,7 +58,8 @@ public static class KpiSeriesBucketer
IReadOnlyList<KpiSeriesPoint> raw,
DateTime fromUtc,
DateTime toUtc,
int maxPoints)
int maxPoints,
KpiRollupAggregation aggregation = KpiRollupAggregation.Gauge)
{
if (maxPoints < 2)
throw new ArgumentOutOfRangeException(nameof(maxPoints),
@@ -54,11 +69,15 @@ public static class KpiSeriesBucketer
throw new ArgumentOutOfRangeException(nameof(toUtc),
toUtc, "toUtc must be strictly greater than fromUtc.");
// Normal runtime case — empty or short series: return as-is.
// Normal runtime case — empty series: return as-is.
if (raw is null || raw.Count == 0)
return Array.Empty<KpiSeriesPoint>();
if (raw.Count <= maxPoints)
// Short-series early return is Gauge-only: for Rate, two deltas landing in one bucket
// must still sum, so a short Rate series cannot be returned verbatim (arch-review 04
// round 2, R3). For well-spaced short Rate series the per-point buckets make the sum an
// identity, so the output still equals the input anyway.
if (aggregation == KpiRollupAggregation.Gauge && raw.Count <= maxPoints)
return raw;
// Divide the window into maxPoints equal-width buckets.
@@ -67,12 +86,11 @@ public static class KpiSeriesBucketer
double windowTicks = (double)(toUtc.Ticks - fromUtc.Ticks);
double bucketWidthTicks = windowTicks / maxPoints;
// For each bucket, track the candidate point: the one with the
// maximum BucketStartUtc (last value within the bucket).
// We use a fixed-size array indexed by bucket number.
// Nullable KpiSeriesPoint[] with 'hasValue' flags is fine since the
// struct is small.
// For each bucket, track the candidate point: for Gauge the one with the maximum
// BucketStartUtc (last value within the bucket); for Rate a running sum of the bucket's
// deltas. Fixed-size arrays indexed by bucket number.
var best = new KpiSeriesPoint[maxPoints];
var sums = new double[maxPoints];
var occupied = new bool[maxPoints];
foreach (var point in raw)
@@ -89,13 +107,19 @@ public static class KpiSeriesBucketer
if (bucketIndex >= maxPoints)
bucketIndex = maxPoints - 1;
if (aggregation == KpiRollupAggregation.Rate)
{
// Rate: accumulate the bucket's per-interval deltas.
sums[bucketIndex] += point.Value;
occupied[bucketIndex] = true;
}
// Keep the last point in iteration order within this bucket. Because the stored
// candidate's BucketStartUtc is the bucket-START timestamp (not the raw point's
// capture time), the comparison below is true for essentially any in-bucket point,
// so each later-in-iteration point overwrites the previous one. For the ascending-
// sorted input this method requires, last-in-iteration IS the largest-timestamp
// point — i.e. last-value-per-bucket semantics.
if (!occupied[bucketIndex] ||
else if (!occupied[bucketIndex] ||
point.BucketStartUtc > best[bucketIndex].BucketStartUtc)
{
best[bucketIndex] = new KpiSeriesPoint(
@@ -109,8 +133,13 @@ public static class KpiSeriesBucketer
var result = new List<KpiSeriesPoint>(maxPoints);
for (int i = 0; i < maxPoints; i++)
{
if (occupied[i])
result.Add(best[i]);
if (!occupied[i])
continue;
result.Add(aggregation == KpiRollupAggregation.Rate
? new KpiSeriesPoint(
fromUtc + TimeSpan.FromTicks((long)(i * bucketWidthTicks)), sums[i])
: best[i]);
}
return result;
@@ -97,5 +97,14 @@ public class SiteCallEntityTypeConfiguration : IEntityTypeConfiguration<SiteCall
.HasDatabaseName("IX_SiteCalls_NonTerminal")
.HasFilter("[TerminalAtUtc] IS NULL")
.IncludeProperties(nameof(SiteCall.SourceSite), nameof(SiteCall.SourceNode), nameof(SiteCall.Status));
// Terminal backs the daily retention purge's predicate (TerminalAtUtc IS NOT NULL
// AND TerminalAtUtc < cutoff) and Task 11's MIN() slice anchor: filtered to the
// terminal population so the purge seeks the expired tail instead of full-scanning
// the 365-day table (arch-review 04 round 2, R6). Complements IX_SiteCalls_NonTerminal,
// which is filtered to IS NULL and unusable for this predicate.
builder.HasIndex(s => s.TerminalAtUtc)
.HasDatabaseName("IX_SiteCalls_Terminal")
.HasFilter("[TerminalAtUtc] IS NOT NULL");
}
}
@@ -0,0 +1,28 @@
using Microsoft.EntityFrameworkCore.Migrations;
#nullable disable
namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
{
/// <inheritdoc />
public partial class AddSiteCallsTerminalIndex : Migration
{
/// <inheritdoc />
protected override void Up(MigrationBuilder migrationBuilder)
{
migrationBuilder.CreateIndex(
name: "IX_SiteCalls_Terminal",
table: "SiteCalls",
column: "TerminalAtUtc",
filter: "[TerminalAtUtc] IS NOT NULL");
}
/// <inheritdoc />
protected override void Down(MigrationBuilder migrationBuilder)
{
migrationBuilder.DropIndex(
name: "IX_SiteCalls_Terminal",
table: "SiteCalls");
}
}
}
@@ -167,6 +167,10 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Migrations
SqlServerIndexBuilderExtensions.IncludeProperties(b.HasIndex("CreatedAtUtc"), new[] { "SourceSite", "SourceNode", "Status" });
b.HasIndex("TerminalAtUtc")
.HasDatabaseName("IX_SiteCalls_Terminal")
.HasFilter("[TerminalAtUtc] IS NOT NULL");
b.HasIndex("SourceSite", "CreatedAtUtc")
.IsDescending(false, true)
.HasDatabaseName("IX_SiteCalls_Source_Created");
@@ -1,4 +1,6 @@
using Microsoft.EntityFrameworkCore;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using ZB.MOM.WW.ScadaBridge.Commons.Entities.Kpi;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
using ZB.MOM.WW.ScadaBridge.Commons.Types.Kpi;
@@ -13,14 +15,22 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Repositories;
public sealed class KpiHistoryRepository : IKpiHistoryRepository
{
private readonly ScadaBridgeDbContext _context;
private readonly ILogger<KpiHistoryRepository> _logger;
/// <summary>
/// Initializes a new instance of the <see cref="KpiHistoryRepository"/> class.
/// </summary>
/// <param name="context">The EF Core database context.</param>
public KpiHistoryRepository(ScadaBridgeDbContext context)
/// <param name="logger">
/// Optional logger; defaults to <see cref="NullLogger{T}"/> (mirrors
/// <c>SiteCallAuditRepository</c> — MS.DI resolves <see cref="ILogger{T}"/> automatically,
/// so no registration churn). Used to classify the self-healing failover fold-race.
/// </param>
public KpiHistoryRepository(
ScadaBridgeDbContext context, ILogger<KpiHistoryRepository>? logger = null)
{
_context = context ?? throw new ArgumentNullException(nameof(context));
_logger = logger ?? NullLogger<KpiHistoryRepository>.Instance;
}
/// <inheritdoc />
@@ -101,8 +111,14 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
// 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)
{
@@ -164,7 +180,22 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
}
}
await _context.SaveChangesAsync(cancellationToken);
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 />
@@ -186,6 +217,15 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
.ToListAsync(cancellationToken);
}
/// <inheritdoc />
public async Task<DateTime?> GetLatestRollupHourAsync(CancellationToken cancellationToken = default)
{
// Single MAX over the unique IX_KpiRollupHourly_Series population (the rollup
// table is small — one row per series-hour). Null when no rollups exist.
return await _context.KpiRollupHourly
.MaxAsync(r => (DateTime?)r.HourStartUtc, cancellationToken);
}
/// <inheritdoc />
public async Task PurgeRollupsOlderThanAsync(DateTime before, CancellationToken cancellationToken = default)
{
@@ -212,4 +252,8 @@ public sealed class KpiHistoryRepository : IKpiHistoryRepository
/// <summary>In-memory grouping key: one KPI series (four-tuple) within one UTC hour.</summary>
private readonly record struct SeriesHourKey(
string Source, string Metric, string Scope, string? ScopeKey, DateTime HourStartUtc);
/// <summary>Narrow, untracked projection of one KpiSample row for the fold (R2).</summary>
private readonly record struct FoldSample(
string Source, string Metric, string Scope, string? ScopeKey, DateTime CapturedAtUtc, double Value);
}
@@ -246,9 +246,29 @@ OPTION (RECOMPILE);";
/// <inheritdoc />
public async Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default)
{
return await _context.Database.ExecuteSqlInterpolatedAsync(
$"DELETE FROM dbo.SiteCalls WHERE TerminalAtUtc IS NOT NULL AND TerminalAtUtc < {olderThanUtc};",
ct);
// Time-sliced batches (arch-review 04 round 2, R6 — the one maintenance DELETE
// that missed round 1's batching pass): each DELETE covers at most one DAY of
// terminal rows, capping the lock/log footprint per statement. Steady state
// (daily purge, 365-day retention) is a single slice; only catch-up after an
// outage runs several. One-day (not one-hour) slices are proportionate to
// SiteCalls volume, which is far below KpiSample's. The MIN() anchor and the
// DELETE predicate both seek IX_SiteCalls_Terminal (filtered IS NOT NULL).
var total = 0;
var floor = await _context.SiteCalls
.Where(s => s.TerminalAtUtc != null && s.TerminalAtUtc < olderThanUtc)
.MinAsync(s => s.TerminalAtUtc, ct);
while (floor is not null && floor < olderThanUtc)
{
var ceiling = floor.Value.AddDays(1) < olderThanUtc ? floor.Value.AddDays(1) : olderThanUtc;
total += await _context.Database.ExecuteSqlInterpolatedAsync(
$"DELETE FROM dbo.SiteCalls WHERE TerminalAtUtc IS NOT NULL AND TerminalAtUtc < {ceiling};",
ct);
floor = await _context.SiteCalls
.Where(s => s.TerminalAtUtc != null && s.TerminalAtUtc < olderThanUtc)
.MinAsync(s => s.TerminalAtUtc, ct);
}
return total;
}
// Terminal status string literals for the interval-throughput KPIs. The
@@ -40,7 +40,7 @@ public sealed class SiteHealthKpiSampleSource : IKpiSampleSource
private const string MetricConnectionsUp = "connectionsUp";
private const string MetricConnectionsDown = KpiMetrics.SiteHealth.ConnectionsDown;
private const string MetricScriptErrors = KpiMetrics.SiteHealth.ScriptErrors;
private const string MetricAlarmEvalErrors = "alarmEvalErrors";
private const string MetricAlarmEvalErrors = KpiMetrics.SiteHealth.AlarmEvalErrors;
private const string MetricSfBufferDepth = KpiMetrics.SiteHealth.SfBufferDepth;
private const string MetricDeadLetters = KpiMetrics.SiteHealth.DeadLetters;
private const string MetricParkedMessages = "parkedMessages";
@@ -48,7 +48,7 @@ public sealed class SiteHealthKpiSampleSource : IKpiSampleSource
private const string MetricEnabledInstances = "enabledInstances";
private const string MetricDisabledInstances = "disabledInstances";
private const string MetricAuditBacklogPending = "auditBacklogPending";
private const string MetricEventLogWriteFailures = "eventLogWriteFailures";
private const string MetricEventLogWriteFailures = KpiMetrics.SiteHealth.EventLogWriteFailures;
private readonly ICentralHealthAggregator _aggregator;
@@ -31,7 +31,12 @@ namespace ZB.MOM.WW.ScadaBridge.KpiHistory;
/// 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.
/// same idempotent fold and runs at most once per actor lifetime. The backfill is
/// sliced into bounded 24 h fold windows (oldest-first) rather than one unbounded
/// pass, and lowers its in-flight guard between slices so periodic folds interleave
/// — periodic folds and backfill slices strictly alternate through the single
/// <c>_rollupInFlight</c>/<c>_backfillInFlight</c> gate pair, so the two idempotent
/// upserts never race on overlapping rows (arch-review 04 round 2, R1).
/// </summary>
/// <remarks>
/// <para>
@@ -118,11 +123,32 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
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.
/// Width of one backfill fold slice (24 h). The repository fold materializes its whole
/// window in memory, so the one-shot backfill must never hand it the full raw-retention
/// window (90 d ≈ tens of millions of rows at fleet volume — arch-review 04 round 2, R1).
/// One day per fold bounds each pass to roughly what a steady-state day writes, and the
/// guard is lowered between slices so periodic folds interleave instead of stalling
/// behind the historical pass.
/// </summary>
private static readonly TimeSpan BackfillSliceWidth = TimeSpan.FromHours(24);
/// <summary>
/// Queue of remaining backfill fold slices (each a bounded <see cref="BackfillSliceWidth"/>
/// window), oldest-first. Built once when the backfill plan is enumerated and drained one
/// slice per fold; a periodic <see cref="RollupTick"/> may claim the fold path between slices
/// (the two idempotent upserts strictly alternate through the
/// <see cref="_rollupInFlight"/>/<see cref="_backfillInFlight"/> gate pair, never racing on
/// overlapping rows — arch-review 04 round 2, R1).
/// </summary>
private readonly Queue<(DateTime FromHourUtc, DateTime ToHourUtc)> _backfillSlices = new();
/// <summary>
/// In-flight guard for a single backfill fold slice. Set true when a slice fold is launched
/// and cleared when its <see cref="BackfillSliceComplete"/> arrives. While true, periodic
/// <see cref="RollupTick"/>s are skipped (like <see cref="_rollupInFlight"/>) so a backfill
/// slice never runs concurrently with a periodic lookback fold on the same upsert rows; it is
/// lowered BETWEEN slices so periodic folds interleave instead of stalling behind the whole
/// historical pass.
/// </summary>
private bool _backfillInFlight;
@@ -160,10 +186,21 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
Receive<RollupTick>(_ => HandleRollupTick());
Receive<RollupComplete>(_ => _rollupInFlight = false); // lower the in-flight guard (success or fault)
Receive<BackfillTick>(_ => HandleBackfillTick());
Receive<BackfillComplete>(_ =>
Receive<BackfillPlan>(HandleBackfillPlan);
Receive<BackfillSliceComplete>(_ =>
{
_backfillInFlight = false; // lower the in-flight guard (success or fault)
_backfillDone = true; // latch: at most once per actor lifetime
_backfillInFlight = false; // release the fold path BETWEEN slices (R1)
if (_backfillSlices.Count == 0)
{
_backfillDone = true; // latch: at most once per actor lifetime (unchanged semantics)
_logger.LogInformation("KPI rollup backfill completed — all slices folded.");
return;
}
// Next slice via the timer, not inline: a RollupTick already queued in the mailbox
// is processed first, so periodic folds interleave; the existing _rollupInFlight
// deferral branch in HandleBackfillTick then re-arms the backfill behind it.
Timers.StartSingleTimer(BackfillTimerKey, BackfillTick.Instance, TimeSpan.Zero);
});
}
@@ -483,13 +520,15 @@ 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
/// has already completed (<see cref="_backfillDone"/>) or a slice 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.
/// timer, so a backfill slice and the periodic lookback fold never race on overlapping upsert
/// rows. When no slice plan exists yet it enumerates the raw-retention window
/// <c>[TruncateToHour(now) RetentionDays, TruncateToHour(now))</c> into bounded
/// <see cref="BackfillSliceWidth"/> windows (oldest-first) and self-arms to start draining;
/// otherwise it dequeues ONE slice, raises the guard, and folds it off the actor thread,
/// piping a <see cref="BackfillSliceComplete"/> back to release the guard between slices.
/// </summary>
private void HandleBackfillTick()
{
@@ -500,38 +539,152 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
if (_rollupInFlight)
{
// A periodic fold is in flight; defer the one-shot backfill briefly so the two
// A periodic fold is in flight; defer the next backfill slice 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;
}
if (_backfillSlices.Count == 0)
{
// No slice plan yet: resolve the rollup watermark off the actor thread, then plan on
// the actor thread (Receive<BackfillPlan>). Raise the guard so periodic ticks defer
// while planning is in flight; the plan handler lowers it. On a failover restart the
// watermark lets the plan SKIP or SHRINK to the un-rolled tail instead of re-folding
// the whole raw-retention window (arch-review 04 round 2, R1).
_backfillInFlight = true;
var planCancellationToken = _shutdownCts?.Token ?? CancellationToken.None;
RunBackfillPlanPass(planCancellationToken).PipeTo(
Self,
success: watermark => new BackfillPlan(watermark),
failure: ex =>
{
_logger.LogError(ex, "KPI rollup backfill plan pass faulted unexpectedly.");
return new BackfillPlan(null); // null watermark ⇒ plan the full retention window
});
return;
}
// Dequeue and fold exactly ONE bounded slice.
var slice = _backfillSlices.Dequeue();
_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(
RunBackfillPass(slice.FromHourUtc, slice.ToHourUtc, cancellationToken).PipeTo(
Self,
success: () => BackfillComplete.Instance,
success: () => BackfillSliceComplete.Instance,
failure: ex =>
{
_logger.LogError(ex, "KPI rollup backfill faulted unexpectedly.");
return BackfillComplete.Instance;
_logger.LogError(ex, "KPI rollup backfill slice faulted unexpectedly.");
return BackfillSliceComplete.Instance;
});
}
/// <summary>
/// Runs the one-shot backfill fold: opens a DI scope, resolves the repository, and folds the
/// Enumerates <c>[<paramref name="fromHourUtc"/>, <paramref name="toHourUtc"/>)</c> into
/// contiguous <see cref="BackfillSliceWidth"/> windows (the last possibly shorter),
/// oldest-first, and enqueues them onto <see cref="_backfillSlices"/>.
/// </summary>
private void EnqueueBackfillSlices(DateTime fromHourUtc, DateTime toHourUtc)
{
var sliceStart = fromHourUtc;
while (sliceStart < toHourUtc)
{
var sliceEnd = sliceStart + BackfillSliceWidth;
if (sliceEnd > toHourUtc)
{
sliceEnd = toHourUtc;
}
_backfillSlices.Enqueue((sliceStart, sliceEnd));
sliceStart = sliceEnd;
}
}
/// <summary>
/// Plans the backfill from the resolved rollup watermark (actor thread). The window floor is
/// the newer of the watermark (re-folding the watermark hour itself is a cheap idempotent
/// safety margin) and the raw-retention floor; a null watermark (fresh install) keeps the full
/// retention window. If the floor is already within
/// <see cref="KpiHistoryOptions.RollupLookbackHours"/> of now the periodic fold's lookback
/// covers everything from there forward, so the whole historical pass is skipped (the failover
/// fast-path); otherwise the tail <c>[floor, TruncateToHour(now))</c> is enumerated into
/// bounded slices and draining is self-armed. Either way the plan-pass guard is lowered here
/// (arch-review 04 round 2, R1).
/// </summary>
private void HandleBackfillPlan(BackfillPlan plan)
{
var toHourUtc = TruncateToHour(DateTime.UtcNow);
var retentionFloor = toHourUtc - TimeSpan.FromDays(_options.RetentionDays);
// Shrink to the un-rolled tail; re-fold the watermark hour itself (idempotent) as a safety
// margin. A null watermark (fresh install) keeps the full retention window.
var floor = plan.Watermark is { } w && w > retentionFloor ? w : retentionFloor;
if (floor >= toHourUtc - TimeSpan.FromHours(_options.RollupLookbackHours))
{
// Failover fast-path: rollups are already current — the periodic fold's lookback
// window covers everything from the watermark forward, so the full historical pass is
// skipped entirely (arch-review 04 round 2, R1).
_backfillInFlight = false;
_backfillDone = true;
_logger.LogInformation(
"KPI rollup backfill skipped — rollups current through {Watermark:o}.", plan.Watermark);
return;
}
EnqueueBackfillSlices(floor, toHourUtc);
_backfillInFlight = false;
if (_backfillSlices.Count == 0)
{
// Degenerate window (no whole hours to fold) — nothing to backfill.
_backfillDone = true;
return;
}
_logger.LogInformation(
"KPI rollup backfill starting — folding [{FromHour:o}, {ToHour:o}) across {SliceCount} "
+ "slice(s) of up to {SliceHours} h (watermark {Watermark:o}).",
floor, toHourUtc, _backfillSlices.Count, BackfillSliceWidth.TotalHours, plan.Watermark);
// Start draining via the timer (uniform path with the between-slice re-arm).
Timers.StartSingleTimer(BackfillTimerKey, BackfillTick.Instance, TimeSpan.Zero);
}
/// <summary>
/// Resolves the newest folded rollup hour (the backfill watermark) off the actor thread in a
/// fresh DI scope. Never faults — on cancellation or error it returns <c>null</c>, which the
/// plan treats as "no rollups", keeping the full retention window (best-effort, mirrors the
/// other passes).
/// </summary>
private async Task<DateTime?> RunBackfillPlanPass(CancellationToken cancellationToken)
{
try
{
await using var scope = _serviceProvider.CreateAsyncScope();
var repository = scope.ServiceProvider.GetRequiredService<IKpiHistoryRepository>();
return await repository.GetLatestRollupHourAsync(cancellationToken);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
// Shutdown interrupted the lookup; a fresh active node re-plans. Not a failure.
return null;
}
catch (Exception ex)
{
_logger.LogError(ex, "KPI rollup backfill watermark lookup failed; planning the full retention window.");
return null;
}
}
/// <summary>
/// Runs one backfill fold slice: 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.
/// the hourly rollup table via the same idempotent upsert the periodic fold uses. 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 each slice is logged at Debug with its
/// window and elapsed wall time (the queue-drained completion is logged at Information).
/// </summary>
private async Task RunBackfillPass(
DateTime fromHourUtc, DateTime toHourUtc, CancellationToken cancellationToken)
@@ -543,8 +696,8 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
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.",
_logger.LogDebug(
"KPI rollup backfill slice folded [{FromHour:o}, {ToHour:o}) in {ElapsedMs:F0} ms.",
fromHourUtc, toHourUtc, (DateTime.UtcNow - startedAt).TotalMilliseconds);
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
@@ -614,9 +767,11 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
}
/// <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.
/// Self-tick triggering the one-shot backfill: on the first tick it enumerates the full
/// raw-retention window into bounded slices, and each subsequent tick folds one slice. Armed
/// once via a single timer in <see cref="PreStart"/> (and re-armed to drain the next slice or
/// to defer past an in-flight periodic fold); the whole pass runs at most once per actor
/// lifetime.
/// </summary>
internal sealed class BackfillTick
{
@@ -625,13 +780,21 @@ public class KpiHistoryRecorderActor : ReceiveActor, IWithTimers
}
/// <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
/// Piped-back result of the backfill plan pass: the newest folded rollup hour (the watermark),
/// or <c>null</c> when no rollups exist. Consumed on the actor thread to skip or shrink the
/// backfill to the un-rolled tail (arch-review 04 round 2, R1).
/// </summary>
internal sealed record BackfillPlan(DateTime? Watermark);
/// <summary>
/// Piped-back completion of ONE backfill fold slice; lets the slice run off the actor thread
/// and, on the actor thread, lowers the <c>_backfillInFlight</c> guard BETWEEN slices — then
/// either drains the next slice (self-armed timer) or, when the queue is empty, latches
/// <c>_backfillDone</c> (fires on success and fault).
/// </summary>
internal sealed class BackfillComplete
internal sealed class BackfillSliceComplete
{
public static readonly BackfillComplete Instance = new();
private BackfillComplete() { }
public static readonly BackfillSliceComplete Instance = new();
private BackfillSliceComplete() { }
}
}
@@ -42,7 +42,7 @@ public sealed class SiteCallAuditKpiSampleSource : IKpiSampleSource
private const string MetricBuffered = KpiMetrics.SiteCallAudit.Buffered;
private const string MetricParked = KpiMetrics.SiteCallAudit.Parked;
private const string MetricFailedLastInterval = KpiMetrics.SiteCallAudit.FailedLastInterval;
private const string MetricDeliveredLastInterval = "deliveredLastInterval";
private const string MetricDeliveredLastInterval = KpiMetrics.SiteCallAudit.DeliveredLastInterval;
private const string MetricStuck = "stuck";
private const string MetricOldestPendingAgeSeconds = "oldestPendingAgeSeconds";