fix(auditlog): reconciliation abandons only row-intrinsic failures and only when a sibling row inserted (review 2026-10-21 follow-up M1)

This commit is contained in:
Joseph Doherty
2026-10-22 09:31:26 -04:00
parent 6b8a91cc83
commit 1f87942ce3
4 changed files with 225 additions and 73 deletions
+10 -4
View File
@@ -329,9 +329,14 @@ batch forever, failing each set-based insert at central into the per-row fallbac
transient refusal still lands, and a genuinely poison row is recorded centrally as transient refusal still lands, and a genuinely poison row is recorded centrally as
a `ReconciliationAbandoned` row before its cursor retires it. A transport fault (no a `ReconciliationAbandoned` row before its cursor retires it. A transport fault (no
answer) is never counted as a refusal. Symmetrically, the reconciliation puller counts answer) is never counted as a refusal. Symmetrically, the reconciliation puller counts
only non-transient failures toward its permanent-abandon threshold: a timeout, only **row-intrinsic** failures (the same narrow set the push path reports as rejected:
connection loss, deadlock or throttling error holds the cursor and retries without truncation, NULL into NOT NULL, conversion/overflow, constraint) toward its
ever abandoning the row. A cached-lifecycle row whose tracking permanent-abandon threshold: a timeout, connection loss, deadlock, throttling error, or
schema drift after a missed central migration (207/208) holds the cursor and retries
without ever abandoning the row. And even a row-intrinsic failure abandons a row only on a
tick where at least one *sibling* row of the same page inserted — a page where every row
fails identically is a database problem, and abandoning it would let the site mark every
row `Reconciled` and purge it (review 2026-10-21 follow-up M1). A cached-lifecycle row whose tracking
snapshot is gone past `CachedTrackingGraceSeconds` leaves the cached drain as snapshot is gone past `CachedTrackingGraceSeconds` leaves the cached drain as
`PushAbandoned` (it used to be marked `Forwarded`, which only worked because the `PushAbandoned` (it used to be marked `Forwarded`, which only worked because the
pull re-served every `Forwarded` row). Cadence is short (default 5 s) when pull re-served every `Forwarded` row). Cadence is short (default 5 s) when
@@ -665,7 +670,8 @@ MS SQL for direct-write events). Unredacted secrets never persist.
still report a non-draining backlog after paging up to `MaxPagesPerTick`, a still report a non-draining backlog after paging up to `MaxPagesPerTick`, a
`SiteAuditTelemetryStalled` event fires. `SiteAuditTelemetryStalled` event fires.
- **Permanent-abandonment record.** A reconciliation pull row that fails to - **Permanent-abandonment record.** A reconciliation pull row that fails to
insert on every retry up to the permanent-abandon threshold is dropped and the insert with a row-intrinsic error on every retry up to the permanent-abandon
threshold, while a sibling row of its page inserts, is dropped and the
cursor advances past it (so one broken row can't block the site forever). The cursor advances past it (so one broken row can't block the site forever). The
loss is not silent: alongside the Critical log line the actor writes ONE loss is not silent: alongside the Critical log line the actor writes ONE
synthetic `ReconciliationAbandoned` audit row (fresh `EventId`, `Status=Failed`, synthetic `ReconciliationAbandoned` audit row (fresh `EventId`, `Status=Failed`,
@@ -391,7 +391,12 @@ public class SiteAuditReconciliationActor : ReceiveActor
/// when the batch throws does the page fall back to the per-row path, which keeps the /// when the batch throws does the page fall back to the per-row path, which keeps the
/// documented poison-row isolation: the failing row is retried up to /// documented poison-row isolation: the failing row is retried up to
/// <see cref="MaxPermanentInsertAttempts"/> times with the cursor held back, then /// <see cref="MaxPermanentInsertAttempts"/> times with the cursor held back, then
/// permanently abandoned. Re-running rows the failed batch may already have committed is /// permanently abandoned. Only a row-intrinsic failure
/// (<see cref="AuditIngestErrorClassifier.IsPermanentRowRejection"/>) counts toward that
/// threshold, and a row is abandoned only on a tick where at least one sibling row of the same
/// page inserted — a page where every row fails is a database problem (schema drift after a
/// missed migration, a narrowed column), and abandoning it would let the site mark every row
/// <c>Reconciled</c> and purge it (review 2026-10-21 follow-up M1). Re-running rows the failed batch may already have committed is
/// safe — every insert is idempotent on <see cref="AuditEvent.EventId"/>, which is also /// safe — every insert is idempotent on <see cref="AuditEvent.EventId"/>, which is also
/// what lets a pre-PERF-1 site that still re-serves <c>Forwarded</c> rows be pulled /// what lets a pre-PERF-1 site that still re-serves <c>Forwarded</c> rows be pulled
/// unchanged. /// unchanged.
@@ -449,6 +454,39 @@ public class SiteAuditReconciliationActor : ReceiveActor
} }
} }
// Per-row fallback (only when the set-based insert threw). Every row is attempted FIRST and
// its outcome recorded, so the abandonment decision below can see whether ANY sibling in
// this page inserted (review 2026-10-21 follow-up M1).
var rowFailures = new Dictionary<Guid, Exception>();
var anyRowInserted = false;
if (!batchCommitted)
{
foreach (var evt in response.Events)
{
if (_abandonedEventIds.Contains(evt.EventId))
{
continue;
}
try
{
// Idempotent repository write: duplicate EventIds (from a
// concurrent push, or a retry of this very pull) collapse to
// a no-op on InsertIfNotExistsAsync.
// IngestedAtUtc is a DetailsJson field on the canonical record —
// stamp it via the projection helper.
var ingested = AuditRowProjection.WithIngestedAtUtc(evt, nowUtc);
await repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false);
_failedInsertAttempts.Remove(evt.EventId);
anyRowInserted = true;
}
catch (Exception ex)
{
rowFailures[evt.EventId] = ex;
}
}
}
foreach (var evt in response.Events) foreach (var evt in response.Events)
{ {
var advanceForThisRow = false; var advanceForThisRow = false;
@@ -464,78 +502,72 @@ public class SiteAuditReconciliationActor : ReceiveActor
// row collapse into the existing one anyway. // row collapse into the existing one anyway.
advanceForThisRow = true; advanceForThisRow = true;
} }
else if (batchCommitted) else if (batchCommitted || !rowFailures.TryGetValue(evt.EventId, out var ex))
{ {
advanceForThisRow = true; advanceForThisRow = true;
} }
else else
{ {
try // Per-row verdict so one bad event does not abandon the rest of the batch.
// Only a ROW-INTRINSIC failure (AuditIngestErrorClassifier.IsPermanentRowRejection
// — the same narrow set the push path reports back as rejected) counts toward
// MaxPermanentInsertAttempts; below the threshold the cursor is HELD BACK so the
// next tick re-pulls and retries; at the threshold the row is permanently
// abandoned (LogCritical + cursor advances past) to keep a truly broken row from
// blocking all subsequent progress for the site.
if (!AuditIngestErrorClassifier.IsPermanentRowRejection(ex))
{ {
// Idempotent repository write: duplicate EventIds (from a // A database problem (timeout, connection loss, deadlock, throttling, a
// concurrent push, or a retry of this very pull) collapse to // maintenance window) or anything not provably about THIS row — e.g. schema
// a no-op on InsertIfNotExistsAsync. // drift after a missed central migration (207 / 208 invalid column/object) —
// IngestedAtUtc is a DetailsJson field on the canonical record — // is not a verdict about the row: hold the cursor and retry, but never count
// stamp it via the projection helper. // it toward permanent abandonment. Abandoning such rows flips the site's
var ingested = AuditRowProjection.WithIngestedAtUtc(evt, nowUtc); // cursor past them, the site marks them Reconciled, and its retention purge
await repository.InsertIfNotExistsAsync(ingested).ConfigureAwait(false); // deletes them — a silent loss of every row of the outage (review 2026-10-21
_failedInsertAttempts.Remove(evt.EventId); // 06/PERF-1 adversarial follow-up; follow-up M1).
advanceForThisRow = true; _logger.LogWarning(
ex,
"Reconciliation ingest of AuditEvent {EventId} from site {SiteId} failed with a non-row-specific error; cursor held back, not counted toward abandonment.",
evt.EventId,
site.SiteId);
hasUnresolvedFailure = true;
} }
catch (Exception ex) else
{ {
// Per-row catch so one bad event does not abandon var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1;
// the rest of the batch. Track the failure count per EventId — _failedInsertAttempts[evt.EventId] = attempts;
// below MaxPermanentInsertAttempts the cursor is HELD BACK so
// the next tick re-pulls and retries; at the threshold the row if (attempts >= MaxPermanentInsertAttempts && anyRowInserted)
// is permanently abandoned (LogCritical + cursor advances past)
// to keep a truly broken row from blocking all subsequent
// progress for the site.
if (AuditIngestErrorClassifier.IsTransient(ex))
{ {
// A database problem (timeout, connection loss, deadlock, throttling, a _logger.LogCritical(
// maintenance window) is not a verdict about this row: hold the cursor
// and retry, but never count it toward permanent abandonment — a long
// outage must not turn healthy rows into ReconciliationAbandoned losses
// (review 2026-10-21 06/PERF-1 adversarial follow-up).
_logger.LogWarning(
ex, ex,
"Reconciliation ingest of AuditEvent {EventId} from site {SiteId} hit a transient database fault; cursor held back, not counted toward abandonment.", "Permanently abandoning AuditEvent {EventId} from site {SiteId} after {Attempts} consecutive insert failures; cursor will advance past it.",
evt.EventId, evt.EventId,
site.SiteId); site.SiteId,
hasUnresolvedFailure = true; attempts);
_failedInsertAttempts.Remove(evt.EventId);
RememberAbandoned(evt.EventId);
advanceForThisRow = true;
await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc)
.ConfigureAwait(false);
} }
else else
{ {
var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; // At the threshold but no sibling in this page inserted: every row failing
_failedInsertAttempts[evt.EventId] = attempts; // identically is a DATABASE problem (e.g. a column narrowed by a migration
// that 245/8114s every row), not one poison row — hold, never abandon. The
if (attempts >= MaxPermanentInsertAttempts) // row is abandoned on a later tick whose page also carries a row that
{ // inserts, which proves the database itself accepts writes.
_logger.LogCritical( _logger.LogError(
ex, ex,
"Permanently abandoning AuditEvent {EventId} from site {SiteId} after {Attempts} consecutive insert failures; cursor will advance past it.", "Reconciliation ingest failed for AuditEvent {EventId} from site {SiteId} (attempt {Attempts}/{Max}{NoSibling}); cursor held back for retry.",
evt.EventId, evt.EventId,
site.SiteId, site.SiteId,
attempts); attempts,
_failedInsertAttempts.Remove(evt.EventId); MaxPermanentInsertAttempts,
RememberAbandoned(evt.EventId); anyRowInserted ? string.Empty : ", no sibling row inserted this tick");
advanceForThisRow = true; hasUnresolvedFailure = true;
await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc)
.ConfigureAwait(false);
}
else
{
_logger.LogError(
ex,
"Reconciliation ingest failed for AuditEvent {EventId} from site {SiteId} (attempt {Attempts}/{Max}); cursor held back for retry.",
evt.EventId,
site.SiteId,
attempts,
MaxPermanentInsertAttempts);
hasUnresolvedFailure = true;
}
} }
} }
} }
@@ -78,15 +78,15 @@ public class AuditIngestRejectionTests : TestKit
private sealed class OneRowClient : IPullAuditEventsClient private sealed class OneRowClient : IPullAuditEventsClient
{ {
private readonly AuditEvent _row; private readonly AuditEvent[] _rows;
public OneRowClient(AuditEvent row) => _row = row; public OneRowClient(params AuditEvent[] rows) => _rows = rows;
public System.Collections.Concurrent.ConcurrentQueue<DateTime> Since { get; } = new(); public System.Collections.Concurrent.ConcurrentQueue<DateTime> Since { get; } = new();
public Task<ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse> PullAsync( public Task<ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse> PullAsync(
string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct)
{ {
Since.Enqueue(sinceUtc); Since.Enqueue(sinceUtc);
return Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse( return Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Messages.Integration.PullAuditEventsResponse(
new[] { _row }, MoreAvailable: false)); _rows, MoreAvailable: false));
} }
} }
@@ -129,4 +129,95 @@ public class AuditIngestRejectionTests : TestKit
Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
Arg.Any<CancellationToken>()); Arg.Any<CancellationToken>());
} }
private SiteAuditReconciliationOptions FastOptions() => new()
{
ReconciliationIntervalOverride = TimeSpan.FromMilliseconds(30),
};
private void RunTicks(IAuditLogRepository repo, OneRowClient client, int ticks)
{
var actor = Sys.ActorOf(Props.Create(() => new SiteAuditReconciliationActor(
new OneSite(), client, StubAuditRepositoryProvider.Build(repo),
Microsoft.Extensions.Options.Options.Create(FastOptions()),
NullLogger<SiteAuditReconciliationActor>.Instance)));
AwaitCondition(() => client.Since.Count >= ticks, TimeSpan.FromSeconds(10));
Watch(actor);
Sys.Stop(actor);
ExpectTerminated(actor);
}
/// <summary>
/// Follow-up M1: schema drift after a missed central migration (207 invalid column name on
/// every insert) is not a ROW verdict — it never counts toward abandonment, so the site's
/// rows are never flipped Reconciled and purged.
/// </summary>
[Fact]
public void Reconciliation_SchemaDriftOnEveryRow_NeverAbandons()
{
var a = NewEvent();
var b = NewEvent();
var repo = Substitute.For<IAuditLogRepository>();
repo.InsertManyIfNotExistsAsync(Arg.Any<IReadOnlyList<AuditEvent>>(), Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(207));
repo.InsertIfNotExistsAsync(Arg.Any<AuditEvent>(), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(207));
var client = new OneRowClient(a, b);
RunTicks(repo, client, 12);
Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s));
repo.DidNotReceive().InsertIfNotExistsAsync(
Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
Arg.Any<CancellationToken>());
}
/// <summary>
/// Follow-up M1: a ROW-classified error (8114 conversion) that hits EVERY row of the page —
/// e.g. a column whose type a migration changed — is a database problem, not one poison row.
/// With no sibling inserting, nothing is abandoned however many ticks pass.
/// </summary>
[Fact]
public void Reconciliation_PermanentErrorOnEveryRow_NoSiblingSucceeds_NeverAbandons()
{
var a = NewEvent();
var b = NewEvent();
var repo = Substitute.For<IAuditLogRepository>();
repo.InsertManyIfNotExistsAsync(Arg.Any<IReadOnlyList<AuditEvent>>(), Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(8114));
repo.InsertIfNotExistsAsync(Arg.Is<AuditEvent>(e => e.EventId == a.EventId || e.EventId == b.EventId), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(8114));
var client = new OneRowClient(a, b);
RunTicks(repo, client, 12);
Assert.All(client.Since, s => Assert.Equal(DateTime.MinValue, s));
repo.DidNotReceive().InsertIfNotExistsAsync(
Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
Arg.Any<CancellationToken>());
}
/// <summary>
/// Follow-up M1, positive half: the same row-classified error on ONE row while a sibling in the
/// page inserts IS a poison row — it is abandoned (synthetic row written) after the threshold.
/// </summary>
[Fact]
public void Reconciliation_PermanentErrorOnOneRow_WithSiblingSuccess_Abandons()
{
var ok = NewEvent();
var poison = NewEvent();
var repo = Substitute.For<IAuditLogRepository>();
repo.InsertManyIfNotExistsAsync(Arg.Any<IReadOnlyList<AuditEvent>>(), Arg.Any<TimeSpan?>(), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(8114));
repo.InsertIfNotExistsAsync(Arg.Is<AuditEvent>(e => e.EventId == poison.EventId), Arg.Any<CancellationToken>())
.ThrowsAsync(SqlExceptionFactory.Create(8114));
var client = new OneRowClient(ok, poison);
RunTicks(repo, client, 8);
repo.Received().InsertIfNotExistsAsync(
Arg.Is<AuditEvent>(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)),
Arg.Any<CancellationToken>());
}
} }
@@ -5,6 +5,7 @@ using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using ZB.MOM.WW.ScadaBridge.AuditLog.Central; using ZB.MOM.WW.ScadaBridge.AuditLog.Central;
using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport;
using ZB.MOM.WW.Audit; using ZB.MOM.WW.Audit;
using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Entities; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Entities;
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
@@ -254,13 +255,29 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
/// <summary>Gets the audit events recorded by this stub.</summary> /// <summary>Gets the audit events recorded by this stub.</summary>
public List<AuditEvent> Inserted { get; } = new(); public List<AuditEvent> Inserted { get; } = new();
private readonly HashSet<Guid> _seen = new(); private readonly HashSet<Guid> _seen = new();
private readonly Guid _poisonId;
private readonly Guid _healthyId;
/// <summary>Initializes the stub.</summary>
/// <param name="poisonId">The row that fails with a row-intrinsic (permanent) SQL error.</param>
/// <param name="healthyId">The one non-synthetic row that inserts; every other row fails transiently.</param>
public AlwaysFailUnlessSyntheticRepo(Guid poisonId, Guid healthyId)
{
_poisonId = poisonId;
_healthyId = healthyId;
}
/// <inheritdoc /> /// <inheritdoc />
public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default)
{ {
if (!evt.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)) if (evt.EventId == _poisonId)
{ {
throw new InvalidOperationException("simulated permanent insert failure"); throw SqlExceptionFactory.Create(8152);
}
if (evt.EventId != _healthyId
&& !evt.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal))
{
throw new TimeoutException("simulated transient insert failure");
} }
if (_seen.Add(evt.EventId)) if (_seen.Add(evt.EventId))
{ {
@@ -900,12 +917,15 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
// Five pulls each returning the same poison row so its per-EventId attempt // Five pulls each returning the same poison row so its per-EventId attempt
// counter reaches MaxPermanentInsertAttempts (5); afterwards the scripted // counter reaches MaxPermanentInsertAttempts (5); afterwards the scripted
// client returns empty and the actor settles. // client returns empty and the actor settles.
// A healthy sibling (earlier timestamp) rides every page: it proves the database
// accepts writes, which is what lets the poison row be abandoned at all (follow-up M1).
// The poison fault is a ROW-intrinsic SQL error (8152 string truncation).
var healthy = NewEvent(siteId, poisonTime.AddSeconds(-1));
var responses = Enumerable.Range(0, 5) var responses = Enumerable.Range(0, 5)
.Select(_ => new PullAuditEventsResponse(new[] { poison }, MoreAvailable: false)) .Select(_ => new PullAuditEventsResponse(new[] { healthy, poison }, MoreAvailable: false))
.ToArray(); .ToArray();
var client = new ScriptedPullClient().Script(siteId, responses); var client = new ScriptedPullClient().Script(siteId, responses);
var repo = new PoisonInsertRepo( var repo = new PoisonInsertRepo(poison.EventId, SqlExceptionFactory.Create(8152));
poison.EventId, new InvalidOperationException("poison row will never insert"));
var actor = CreateActor(sites, client, repo, FastTickOptions()); var actor = CreateActor(sites, client, repo, FastTickOptions());
@@ -965,6 +985,9 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
var siteId = "siteReabandon"; var siteId = "siteReabandon";
var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc); var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc);
var poison = NewEvent(siteId, poisonTime); var poison = NewEvent(siteId, poisonTime);
// A row that always inserts rides every page so the poison row is abandonable
// (an all-rows-failing page is a database problem and never abandons — follow-up M1).
var healthy = NewEvent(siteId, poisonTime.AddSeconds(-1));
// 10 scripted responses: enough for exactly two 5-attempt abandon // 10 scripted responses: enough for exactly two 5-attempt abandon
// cycles of the poison row (the cursor never advances between them, // cycles of the poison row (the cursor never advances between them,
@@ -974,12 +997,12 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
// row count is bounded and deterministic. // row count is bounded and deterministic.
var responses = Enumerable.Range(0, 10) var responses = Enumerable.Range(0, 10)
.Select(i => new PullAuditEventsResponse( .Select(i => new PullAuditEventsResponse(
new[] { poison, NewEvent(siteId, poisonTime.AddSeconds(i + 1)) }, new[] { healthy, poison, NewEvent(siteId, poisonTime.AddSeconds(i + 1)) },
MoreAvailable: false)) MoreAvailable: false))
.ToArray(); .ToArray();
var client = new ScriptedPullClient().Script(siteId, responses); var client = new ScriptedPullClient().Script(siteId, responses);
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083")); var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083"));
var repo = new AlwaysFailUnlessSyntheticRepo(); var repo = new AlwaysFailUnlessSyntheticRepo(poison.EventId, healthy.EventId);
var actor = CreateActor(sites, client, repo, FastTickOptions()); var actor = CreateActor(sites, client, repo, FastTickOptions());