diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index c08b9494..19ff1e8b 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -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 a `ReconciliationAbandoned` row before its cursor retires it. A transport fault (no answer) is never counted as a refusal. Symmetrically, the reconciliation puller counts -only non-transient failures toward its permanent-abandon threshold: a timeout, -connection loss, deadlock or throttling error holds the cursor and retries without -ever abandoning the row. A cached-lifecycle row whose tracking +only **row-intrinsic** failures (the same narrow set the push path reports as rejected: +truncation, NULL into NOT NULL, conversion/overflow, constraint) toward its +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 `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 @@ -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 `SiteAuditTelemetryStalled` event fires. - **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 loss is not silent: alongside the Critical log line the actor writes ONE synthetic `ReconciliationAbandoned` audit row (fresh `EventId`, `Status=Failed`, diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index 4dd3e3e2..0fb43806 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -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 /// documented poison-row isolation: the failing row is retried up to /// 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 + /// () 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 + /// Reconciled 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 , which is also /// what lets a pre-PERF-1 site that still re-serves Forwarded rows be pulled /// 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(); + 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) { var advanceForThisRow = false; @@ -464,78 +502,72 @@ public class SiteAuditReconciliationActor : ReceiveActor // row collapse into the existing one anyway. advanceForThisRow = true; } - else if (batchCommitted) + else if (batchCommitted || !rowFailures.TryGetValue(evt.EventId, out var ex)) { advanceForThisRow = true; } 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 - // 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); - advanceForThisRow = true; + // A database problem (timeout, connection loss, deadlock, throttling, a + // maintenance window) or anything not provably about THIS row — e.g. schema + // drift after a missed central migration (207 / 208 invalid column/object) — + // is not a verdict about the row: hold the cursor and retry, but never count + // it toward permanent abandonment. Abandoning such rows flips the site's + // cursor past them, the site marks them Reconciled, and its retention purge + // deletes them — a silent loss of every row of the outage (review 2026-10-21 + // 06/PERF-1 adversarial follow-up; follow-up M1). + _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 - // the rest of the batch. Track the failure count per EventId — - // below MaxPermanentInsertAttempts 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.IsTransient(ex)) + var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; + _failedInsertAttempts[evt.EventId] = attempts; + + if (attempts >= MaxPermanentInsertAttempts && anyRowInserted) { - // A database problem (timeout, connection loss, deadlock, throttling, a - // 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( + _logger.LogCritical( 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, - site.SiteId); - hasUnresolvedFailure = true; + site.SiteId, + attempts); + _failedInsertAttempts.Remove(evt.EventId); + RememberAbandoned(evt.EventId); + advanceForThisRow = true; + + await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) + .ConfigureAwait(false); } else { - var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; - _failedInsertAttempts[evt.EventId] = attempts; - - if (attempts >= MaxPermanentInsertAttempts) - { - _logger.LogCritical( - ex, - "Permanently abandoning AuditEvent {EventId} from site {SiteId} after {Attempts} consecutive insert failures; cursor will advance past it.", - evt.EventId, - site.SiteId, - attempts); - _failedInsertAttempts.Remove(evt.EventId); - RememberAbandoned(evt.EventId); - advanceForThisRow = 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; - } + // At the threshold but no sibling in this page inserted: every row failing + // 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 + // row is abandoned on a later tick whose page also carries a row that + // inserts, which proves the database itself accepts writes. + _logger.LogError( + ex, + "Reconciliation ingest failed for AuditEvent {EventId} from site {SiteId} (attempt {Attempts}/{Max}{NoSibling}); cursor held back for retry.", + evt.EventId, + site.SiteId, + attempts, + MaxPermanentInsertAttempts, + anyRowInserted ? string.Empty : ", no sibling row inserted this tick"); + hasUnresolvedFailure = true; } } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs index 1b33c01d..6a94c5a1 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/AuditIngestRejectionTests.cs @@ -78,15 +78,15 @@ public class AuditIngestRejectionTests : TestKit private sealed class OneRowClient : IPullAuditEventsClient { - private readonly AuditEvent _row; - public OneRowClient(AuditEvent row) => _row = row; + private readonly AuditEvent[] _rows; + public OneRowClient(params AuditEvent[] rows) => _rows = rows; public System.Collections.Concurrent.ConcurrentQueue Since { get; } = new(); public Task PullAsync( string siteId, DateTime sinceUtc, string? afterId, int batchSize, CancellationToken ct) { Since.Enqueue(sinceUtc); 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(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), Arg.Any()); } + + 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.Instance))); + + AwaitCondition(() => client.Since.Count >= ticks, TimeSpan.FromSeconds(10)); + Watch(actor); + Sys.Stop(actor); + ExpectTerminated(actor); + } + + /// + /// 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. + /// + [Fact] + public void Reconciliation_SchemaDriftOnEveryRow_NeverAbandons() + { + var a = NewEvent(); + var b = NewEvent(); + var repo = Substitute.For(); + repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(207)); + repo.InsertIfNotExistsAsync(Arg.Any(), Arg.Any()) + .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(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), + Arg.Any()); + } + + /// + /// 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. + /// + [Fact] + public void Reconciliation_PermanentErrorOnEveryRow_NoSiblingSucceeds_NeverAbandons() + { + var a = NewEvent(); + var b = NewEvent(); + var repo = Substitute.For(); + repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(8114)); + repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == a.EventId || e.EventId == b.EventId), Arg.Any()) + .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(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), + Arg.Any()); + } + + /// + /// 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. + /// + [Fact] + public void Reconciliation_PermanentErrorOnOneRow_WithSiblingSuccess_Abandons() + { + var ok = NewEvent(); + var poison = NewEvent(); + var repo = Substitute.For(); + repo.InsertManyIfNotExistsAsync(Arg.Any>(), Arg.Any(), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(8114)); + repo.InsertIfNotExistsAsync(Arg.Is(e => e.EventId == poison.EventId), Arg.Any()) + .ThrowsAsync(SqlExceptionFactory.Create(8114)); + var client = new OneRowClient(ok, poison); + + RunTicks(repo, client, 8); + + repo.Received().InsertIfNotExistsAsync( + Arg.Is(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), + Arg.Any()); + } } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs index 5e673b91..d1c19e4e 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs @@ -5,6 +5,7 @@ using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.AuditLog.Central; +using ZB.MOM.WW.ScadaBridge.AuditLog.Tests.TestSupport; using ZB.MOM.WW.Audit; using ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Entities; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; @@ -254,13 +255,29 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixtureGets the audit events recorded by this stub. public List Inserted { get; } = new(); private readonly HashSet _seen = new(); + private readonly Guid _poisonId; + private readonly Guid _healthyId; + + /// Initializes the stub. + /// The row that fails with a row-intrinsic (permanent) SQL error. + /// The one non-synthetic row that inserts; every other row fails transiently. + public AlwaysFailUnlessSyntheticRepo(Guid poisonId, Guid healthyId) + { + _poisonId = poisonId; + _healthyId = healthyId; + } /// 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)) { @@ -900,12 +917,15 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture new PullAuditEventsResponse(new[] { poison }, MoreAvailable: false)) + .Select(_ => new PullAuditEventsResponse(new[] { healthy, poison }, MoreAvailable: false)) .ToArray(); var client = new ScriptedPullClient().Script(siteId, responses); - var repo = new PoisonInsertRepo( - poison.EventId, new InvalidOperationException("poison row will never insert")); + var repo = new PoisonInsertRepo(poison.EventId, SqlExceptionFactory.Create(8152)); var actor = CreateActor(sites, client, repo, FastTickOptions()); @@ -965,6 +985,9 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture new PullAuditEventsResponse( - new[] { poison, NewEvent(siteId, poisonTime.AddSeconds(i + 1)) }, + new[] { healthy, poison, NewEvent(siteId, poisonTime.AddSeconds(i + 1)) }, MoreAvailable: false)) .ToArray(); var client = new ScriptedPullClient().Script(siteId, responses); 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());