dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

fix(auditlog)+fix(configdb): budget the ingest fallback loop; index + varchar-type the SiteCalls list query (arch-review 2026-08-19 P5-1 + P5-2)

Browse Source 
P5-1: AuditLogIngestActor.OnCachedTelemetryAsync's per-entry fallback loop
(used when the cached-telemetry batch fast path fails) ran with no
budget/cancellation, unlike OnIngestAsync's fallback. A slow/wedged DB could
freeze the central ingest singleton's mailbox for minutes. The loop now runs
under its own fresh IngestFallbackBudget CancellationTokenSource (mirroring
OnIngestAsync), passes the token into BeginTransactionAsync/
InsertIfNotExistsAsync/UpsertAsync, and stops early on budget exhaustion,
logging how many entries were left for the periodic reconciliation pull to
pick up.

P5-2: SiteCallAuditRepository.QueryAsync ordered by
(CreatedAtUtc DESC, TrackedOperationId DESC) with no index leading
unconditionally on CreatedAtUtc, forcing a full scan+sort on an unfiltered
page. Added IX_SiteCalls_Created_TrackedOperationId (migration
AddSiteCallsListSortIndex + matching hand-written production-parity script
under docs/plans/sql/). Also switched the query from FromSqlInterpolated
(which binds string filters as nvarchar(4000)) to FromSqlRaw with explicitly
SqlDbType.VarChar-typed parameters, matching the SiteCalls columns'
IsUnicode(false) declaration and the write path's existing Param(...)
pattern, so seeks — including the keyset predicate — are no longer blocked
by an implicit conversion.

Tests: new AuditLogIngestActorCombinedTelemetryTests case proves the
fallback budget bounds the wait (hanging repo stub, ~5s not indefinite),
that remaining entries are skipped, and that a non-default/cancellable
token reaches the repository call; verified fail-unfixed by reverting the
actor change (test times out at its 20s bound). Existing
SiteCallAuditRepositoryTests QueryAsync/keyset/stuck-cutoff cases (real
MSSQL) confirm the raw-SQL rewrite still returns correct pages.
This commit is contained in:
Joseph Doherty
2026-08-19 14:44:44 -04:00
parent 6c11362d9f
commit eca0ab768d
8 changed files with 2488 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditLogIngestActorCombinedTelemetryTests.cs
+167 -3
View File
@@ -89,15 +89,24 @@ public class AuditLogIngestActorCombinedTelemetryTests : TestKit, IClassFixture<
/// DataProtection wiring (we never write secret columns in these tests).
/// </summary>
private IServiceProvider BuildServiceProvider(
Func<ScadaBridgeDbContext, ISiteCallAuditRepository>? siteCallRepoFactory = null)
Func<ScadaBridgeDbContext, ISiteCallAuditRepository>? siteCallRepoFactory = null,
Func<ScadaBridgeDbContext, IAuditLogRepository>? auditRepoFactory = null)
{
var services = new ServiceCollection();
services.AddDbContext<ScadaBridgeDbContext>(opts =>
opts.UseSqlServer(_fixture.ConnectionString)
.ConfigureWarnings(w => w.Ignore(
Microsoft.EntityFrameworkCore.Diagnostics.RelationalEventId.PendingModelChangesWarning)));
services.AddScoped<IAuditLogRepository>(sp =>
new AuditLogRepository(sp.GetRequiredService<ScadaBridgeDbContext>()));
if (auditRepoFactory is null)
{
services.AddScoped<IAuditLogRepository>(sp =>
new AuditLogRepository(sp.GetRequiredService<ScadaBridgeDbContext>()));
}
else
{
services.AddScoped(sp =>
auditRepoFactory(sp.GetRequiredService<ScadaBridgeDbContext>()));
}
if (siteCallRepoFactory is null)
{
services.AddScoped<ISiteCallAuditRepository>(sp =>
@@ -338,6 +347,161 @@ public class AuditLogIngestActorCombinedTelemetryTests : TestKit, IClassFixture<
Assert.DoesNotContain(siteCallRows, r => r.TrackedOperationId == poisonTrackedId);
}
/// <summary>
/// arch-review 2026-08-19 P5-1: the per-entry fallback loop (used when the
/// batch fast path fails) must run under its own short
/// <see cref="AuditLogIngestActor.IngestFallbackBudget"/>, not an unbounded
/// wait — otherwise a single slow/wedged downstream call can freeze this
/// ReceiveAsync handler, and with it the central ingest singleton's mailbox,
/// for as long as the SQL retry ladder allows (minutes). Forces the batch
/// fast path to fail immediately (via <see cref="BatchFailingAuditRepo"/>) so
/// the per-entry fallback runs, then hangs the first entry's SiteCalls
/// upsert (via <see cref="HangingSiteCallRepo"/>) to prove the fallback's own
/// token — not <see cref="CancellationToken.None"/> — reaches the repository
/// call and bounds the wait.
/// </summary>
/// <remarks>
/// Pre-fix, the per-entry loop passed no token to
/// <c>BeginTransactionAsync</c>/<c>InsertIfNotExistsAsync</c>/<c>UpsertAsync</c>,
/// so <see cref="HangingSiteCallRepo"/>'s <c>Task.Delay(Timeout.InfiniteTimeSpan, ct)</c>
/// would never observe a cancellation and hang forever — this test's bounded
/// <c>ExpectMsg</c> below would time out and fail, which is the fail-unfixed
/// proof for this fix.
/// </remarks>
[SkippableFact]
public async Task Receive_CachedTelemetry_PerEntryFallbackHangs_BudgetBoundsWait_RemainderSkipped()
{
Skip.IfNot(_fixture.Available, _fixture.SkipReason);
var siteId = NewSiteId();
var entries = Enumerable.Range(0, 3).Select(_ =>
{
var (audit, siteCall) = NewEntry(siteId);
return new CachedTelemetryEntry(audit, siteCall);
}).ToList();
var hangingRepo = new HangingSiteCallRepo();
var sp = BuildServiceProvider(
siteCallRepoFactory: _ => hangingRepo,
auditRepoFactory: ctx => new BatchFailingAuditRepo(new AuditLogRepository(ctx)));
var actor = CreateActor(sp);
var stopwatch = System.Diagnostics.Stopwatch.StartNew();
actor.Tell(new IngestCachedTelemetryCommand(entries), TestActor);
// Budget is IngestFallbackBudget (5s). Bound the wait comfortably above
// that but far below "hangs indefinitely" — on unfixed code (no token
// passed to the fallback loop) this ExpectMsg times out and the test
// fails, which is the point: it proves the mailbox would otherwise be
// frozen for the life of the hang, not just for one short budget.
var reply = ExpectMsg<IngestCachedTelemetryReply>(TimeSpan.FromSeconds(20));
stopwatch.Stop();
Assert.True(
stopwatch.Elapsed < TimeSpan.FromSeconds(15),
$"handler took {stopwatch.Elapsed} — the per-entry fallback budget should bound the wait to ~{AuditLogIngestActor.IngestFallbackBudget.TotalSeconds}s, not the full hang.");
// Only the first entry pays the hang; once the budget is exhausted the
// remaining two are skipped without even being attempted. None of the
// three entries commit (the first was cancelled mid-transaction; the
// rest were never started).
Assert.Empty(reply.AcceptedEventIds);
Assert.Equal(1, hangingRepo.CallCount);
// The token reaching the repository call is the fallback's OWN fresh
// token — non-default and cancellable — never CancellationToken.None.
Assert.NotNull(hangingRepo.ReceivedToken);
Assert.NotEqual(CancellationToken.None, hangingRepo.ReceivedToken!.Value);
Assert.True(hangingRepo.ReceivedToken.Value.CanBeCanceled);
await using var read = CreateReadContext();
var siteCallCount = await read.Set<SiteCall>().CountAsync(s => s.SourceSite == siteId);
Assert.Equal(0, siteCallCount);
}
/// <summary>
/// Test double — forces <see cref="AuditLogIngestActor"/>'s cached-telemetry
/// batch fast path (<c>TryIngestCachedBatchAsync</c>) to fail immediately by
/// throwing from <see cref="InsertManyIfNotExistsAsync"/>, so tests can reach
/// the per-entry fallback loop without waiting out the real 20s
/// <see cref="AuditLogIngestActor.IngestBudget"/>. Every other member
/// delegates to a real <see cref="AuditLogRepository"/> so the fallback path
/// itself still exercises genuine SQL.
/// </summary>
private sealed class BatchFailingAuditRepo : IAuditLogRepository
{
private readonly IAuditLogRepository _inner;
public BatchFailingAuditRepo(IAuditLogRepository inner) { _inner = inner; }
public Task<int> InsertManyIfNotExistsAsync(
IReadOnlyList<AuditEvent> events, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
throw new InvalidOperationException("simulated batch ingest failure — forces the per-entry fallback");
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) =>
_inner.InsertIfNotExistsAsync(evt, ct);
public Task<IReadOnlyList<AuditEvent>> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
_inner.QueryAsync(filter, paging, ct);
public Task<long> SwitchOutPartitionAsync(
DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
_inner.SwitchOutPartitionAsync(monthBoundary, commandTimeout, ct);
public Task<long> PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
_inner.PurgeChannelOlderThanAsync(channel, threshold, batchSize, commandTimeout, ct);
public Task<IReadOnlyList<DateTime>> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default) =>
_inner.GetPartitionBoundariesOlderThanAsync(threshold, ct);
public Task<ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot> GetKpiSnapshotAsync(
TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) =>
_inner.GetKpiSnapshotAsync(window, nowUtc, ct);
public Task<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default) =>
_inner.GetExecutionTreeAsync(executionId, ct);
public Task<IReadOnlyList<string>> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
_inner.GetDistinctSourceNodesAsync(ct);
public Task<long> BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
_inner.BackfillSourceNodeAsync(sentinel, before, batchSize, ct);
}
/// <summary>
/// Test double — <see cref="UpsertAsync"/> hangs on
/// <c>Task.Delay(Timeout.InfiniteTimeSpan, ct)</c> until the caller's token
/// cancels, recording the call count and the token it was handed. Used to
/// prove the per-entry fallback loop's own budget bounds the wait (arch-review
/// 2026-08-19 P5-1) rather than blocking forever on a wedged downstream call.
/// </summary>
private sealed class HangingSiteCallRepo : ISiteCallAuditRepository
{
private int _callCount;
public int CallCount => _callCount;
public CancellationToken? ReceivedToken { get; private set; }
public async Task UpsertAsync(SiteCall siteCall, CancellationToken ct = default)
{
Interlocked.Increment(ref _callCount);
ReceivedToken = ct;
await Task.Delay(Timeout.InfiniteTimeSpan, ct).ConfigureAwait(false);
}
public Task<SiteCall?> GetAsync(TrackedOperationId id, CancellationToken ct = default) =>
throw new NotSupportedException();
public Task<IReadOnlyList<SiteCall>> QueryAsync(
SiteCallQueryFilter filter, SiteCallPaging paging, CancellationToken ct = default) =>
throw new NotSupportedException();
public Task<int> PurgeTerminalAsync(DateTime olderThanUtc, CancellationToken ct = default) =>
throw new NotSupportedException();
public Task<SiteCallKpiSnapshot> ComputeKpisAsync(
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
throw new NotSupportedException();
public Task<IReadOnlyList<SiteCallSiteKpiSnapshot>> ComputePerSiteKpisAsync(
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
throw new NotSupportedException();
public Task<IReadOnlyList<SiteCallNodeKpiSnapshot>> ComputePerNodeKpisAsync(
DateTime stuckCutoff, DateTime intervalSince, CancellationToken ct = default) =>
throw new NotSupportedException();
}
/// <summary>
/// Test double — throws unconditionally from <see cref="UpsertAsync"/> so
/// the dual-write transaction is forced to roll back. Lets the AuditLog