From 88fbc9e35153ffb7ff6635d34d58139381d225d2 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 19 Aug 2026 16:20:05 -0400 Subject: [PATCH] =?UTF-8?q?fix(auditlog):=20P5-6=20=E2=80=94=20determinist?= =?UTF-8?q?ic=20synthetic=20ReconciliationAbandoned=20EventId?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SiteAuditReconciliationActor minted a fresh random EventId for the synthetic ReconciliationAbandoned row every time a permanently-abandoned row was re-abandoned. Because the reconciliation cursor is held at `since` by ANY co-batch unresolved row (see PullSiteAsync's hasUnresolvedFailure gate), a poisoned row's per-EventId attempt counter restarts every time it reappears in a stuck batch, so it gets re-abandoned roughly every MaxPermanentInsertAttempts ticks — minting an unbounded number of duplicate central audit rows for the SAME lost event. Fix: DeriveAbandonmentEventId(Guid lostEventId) derives the synthetic EventId deterministically (SHA-256 over a fixed namespace GUID + the lost row's bytes, RFC4122-v5-shaped). Re-abandonment of the same lost row now always produces the same synthetic EventId, so IAuditLogRepository.InsertIfNotExistsAsync's existing idempotency collapses every repeat into the first insert. Also adds an in-memory _abandonedEventIds set as a CHEAP SHORT-CIRCUIT ONLY (skips the wasted ~25 min retry-then-abandon cycle for a row already known abandoned this process's lifetime) — it is not the correctness mechanism; it resets on singleton restart/failover, at which point the deterministic id alone still keeps re-abandonment idempotent. Tests (SiteAuditReconciliationActorTests): - ReAbandonment_SameLostEventId_YieldsExactlyOneCentralRow: end-to-end repro of the defect trigger (poison row + a rotating-EventId co-batch row that never itself reaches the threshold, holding the cursor forever) across two full abandon cycles. Proof of failure: run against the unfixed source (git stash of the source file only, pure-function tests temporarily #if false'd to keep the file compiling), the test failed with exactly the predicted defect shape — 2 distinct synthetic rows for one lost EventId. Passes post-fix (1 row). - DeriveAbandonmentEventId_DifferentLostRows_YieldDistinctSyntheticIds and DeriveAbandonmentEventId_IsPureAndStableAcrossIndependentInvocations: pure-function coverage of the derivation directly, including a pinned golden-value regression guard proving it is a function of the input GUID's value alone (no clock/randomness/process state). --- .../Central/SiteAuditReconciliationActor.cs | 190 ++++++++++++++---- .../SiteAuditReconciliationActorTests.cs | 182 +++++++++++++++++ 2 files changed, 331 insertions(+), 41 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index a68f7ee5..fd83b09f 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -1,3 +1,4 @@ +using System.Security.Cryptography; using System.Text.Json; using Akka.Actor; using Microsoft.Extensions.DependencyInjection; @@ -125,6 +126,34 @@ public class SiteAuditReconciliationActor : ReceiveActor /// private readonly Dictionary _failedInsertAttempts = new(); + /// + /// EventIds already permanently abandoned earlier in this process's + /// lifetime — a cheap short-circuit ONLY, not a correctness mechanism. + /// + /// + /// + /// The cursor can be held at since indefinitely by an unrelated + /// co-batch row that keeps failing (see hasUnresolvedFailure in + /// ), which means an already-abandoned row can + /// keep reappearing in the pulled batch tick after tick. Without this + /// set, each reappearance would re-run the full + /// -attempt retry cycle (since + /// was cleared when the row was first + /// abandoned) purely to re-derive the same conclusion. Checking this set + /// first skips straight to "already resolved" instead of burning another + /// ~25 minutes per reappearance. + /// + /// + /// This is deliberately NOT what makes re-abandonment safe — + /// 's determinism is. This set is + /// in-memory only and is empty again after a singleton restart/failover, + /// at which point the full retry-then-abandon cycle simply runs again; + /// the deterministic EventId still makes that re-run's synthetic insert + /// collapse into the SAME row instead of minting a duplicate. + /// + /// + private readonly HashSet _abandonedEventIds = new(); + /// /// Number of consecutive central-insert failures before a row /// is permanently abandoned with a Critical log entry and the cursor is @@ -133,6 +162,15 @@ public class SiteAuditReconciliationActor : ReceiveActor /// private const int MaxPermanentInsertAttempts = 5; + /// + /// Fixed namespace GUID mixed into 's + /// hash input. Any fixed value works — it exists only to separate this + /// derivation's ID space from other GUID-from-GUID derivations that might + /// ever reuse the same technique elsewhere in the codebase. + /// + private static readonly Guid AbandonmentEventIdNamespace = + new("f2b6a6d1-6e2b-4b8b-9b3e-9a6b6a2c1d4e"); + private ICancelable? _timer; /// @@ -285,54 +323,70 @@ public class SiteAuditReconciliationActor : ReceiveActor foreach (var evt in response.Events) { var advanceForThisRow = false; - try + + if (_abandonedEventIds.Contains(evt.EventId)) { - // 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); + // Cheap short-circuit ONLY (see the field's doc comment): this row was + // already permanently abandoned earlier in this process's lifetime and + // keeps reappearing because an unrelated co-batch row is holding the + // cursor at `since`. Skip straight to "resolved" instead of re-running + // the whole MaxPermanentInsertAttempts retry cycle — the insert is known + // to fail, and DeriveAbandonmentEventId would make a re-run's synthetic + // row collapse into the existing one anyway. advanceForThisRow = 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. - var attempts = _failedInsertAttempts.GetValueOrDefault(evt.EventId) + 1; - _failedInsertAttempts[evt.EventId] = attempts; - - if (attempts >= MaxPermanentInsertAttempts) + try { - _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); + // 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; - - await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) - .ConfigureAwait(false); } - else + catch (Exception ex) { - _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; + // 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. + 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); + _abandonedEventIds.Add(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; + } } } @@ -372,14 +426,27 @@ public class SiteAuditReconciliationActor : ReceiveActor /// row when a pulled event is permanently abandoned — so the loss is /// queryable in the Audit Log rather than living only in the Critical log line. /// The synthetic row is built through the same - /// the ingest path uses (fresh - /// EventId, Actor="system", channel preserved from the lost row's + /// the ingest path uses (a + /// -derived EventId, + /// Actor="system", channel preserved from the lost row's /// Category where parseable) and carries the abandoned EventId, /// the source site, and the final error in Extra. It runs in its own /// try/catch: the synthetic row is small and well-formed, so the row-specific /// fault that killed the original will not recur; if it somehow does, log and /// continue — the abandonment cursor must never be blocked a second time. /// + /// + /// The synthetic EventId is deterministic (derived from + /// lost.EventId), not fresh, specifically so this call is idempotent: + /// a co-batch row that never resolves can hold the reconciliation cursor at + /// since indefinitely, causing the SAME lost row to be re-pulled and + /// re-abandoned on a later tick (its entry + /// was cleared on the first abandonment, so it restarts). Every re-abandonment + /// of the same lost row now derives the same synthetic id, so + /// collapses every + /// repeat into the first insert instead of minting an unbounded number of + /// duplicate rows for one lost event. + /// private async Task EmitAbandonmentRecordAsync( IAuditLogRepository repository, ZB.MOM.WW.Audit.AuditEvent lost, @@ -403,6 +470,7 @@ public class SiteAuditReconciliationActor : ReceiveActor }); var synthetic = ScadaBridgeAuditEventFactory.Create( + eventId: DeriveAbandonmentEventId(lost.EventId), channel: channel, kind: AuditKind.ReconciliationAbandoned, status: AuditStatus.Failed, @@ -428,6 +496,46 @@ public class SiteAuditReconciliationActor : ReceiveActor } } + /// + /// Derives a deterministic synthetic + /// for the row minted when + /// is permanently abandoned. + /// + /// + /// + /// Pure function of alone — no clock, no + /// randomness, no actor state. Same input always produces the same output, + /// including across a process restart, which is what makes re-abandonment + /// idempotent under + /// even after a singleton failover wipes every in-memory tracking + /// dictionary this actor keeps (, + /// , the cursors themselves). + /// + /// + /// Built as a name-based (RFC 4122 "version 5"-shaped) GUID: SHA-256 over + /// a fixed namespace GUID concatenated with 's + /// bytes, truncated to 16 bytes with the version/variant bits set so the + /// result is a well-formed GUID. This is not a cryptographic use of + /// SHA-256 — collision resistance against an adversarial input is not the + /// goal, only a stable, well-distributed GUID-to-GUID mapping so distinct + /// lost rows get distinct synthetic ids. + /// + /// + internal static Guid DeriveAbandonmentEventId(Guid lostEventId) + { + Span input = stackalloc byte[32]; + AbandonmentEventIdNamespace.TryWriteBytes(input[..16]); + lostEventId.TryWriteBytes(input[16..]); + + Span hash = stackalloc byte[32]; + SHA256.HashData(input, hash); + + var guidBytes = hash[..16].ToArray(); + guidBytes[6] = (byte)((guidBytes[6] & 0x0F) | 0x50); // version 5 + guidBytes[8] = (byte)((guidBytes[8] & 0x3F) | 0x80); // RFC 4122 variant + return new Guid(guidBytes); + } + /// /// Flips the per-site stalled flag based on whether this tick drained the /// queue. A "draining" cycle is one where the server reported no more rows 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 3cd50a0c..54634777 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs @@ -176,6 +176,69 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture>(Array.Empty()); } + /// + /// Recording stub whose throws for EVERY + /// row except a synthetic row + /// (identified by its Action suffix, mirroring 's + /// assertion pattern). Used to model a site whose backlog contains one row that + /// NEVER inserts (the eventual permanent-abandon target) alongside a co-batch + /// row that also never inserts but changes EventId every pull — i.e. it never + /// individually reaches MaxPermanentInsertAttempts, so it keeps + /// hasUnresolvedFailure true and holds the cursor at since forever + /// (P5-6's defect trigger). Dedupes inserted rows on + /// exactly like the real InsertIfNotExistsAsync, so a test asserting on + /// 's count is exercising the same idempotency contract + /// production relies on. + /// + private sealed class AlwaysFailUnlessSyntheticRepo : IAuditLogRepository + { + public List Inserted { get; } = new(); + private readonly HashSet _seen = new(); + + public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) + { + if (!evt.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)) + { + throw new InvalidOperationException("simulated permanent insert failure"); + } + if (_seen.Add(evt.EventId)) + { + Inserted.Add(evt); + } + return Task.CompletedTask; + } + + public Task> QueryAsync( + AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) => + Task.FromResult>(Inserted); + + public Task SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) => + Task.FromResult(0L); + + public Task PurgeChannelOlderThanAsync( + string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) => + Task.FromResult(0L); + + public Task BackfillSourceNodeAsync( + string sentinel, DateTime before, int batchSize, CancellationToken ct = default) => + Task.FromResult(0L); + + public Task> GetPartitionBoundariesOlderThanAsync( + DateTime threshold, CancellationToken ct = default) => + Task.FromResult>(Array.Empty()); + + public Task GetKpiSnapshotAsync( + TimeSpan window, DateTime? nowUtc = null, CancellationToken ct = default) => + Task.FromResult(new ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot(0L, 0L, 0L, nowUtc ?? DateTime.UtcNow)); + + public Task> GetExecutionTreeAsync( + Guid executionId, CancellationToken ct = default) => + Task.FromResult>(Array.Empty()); + + public Task> GetDistinctSourceNodesAsync(CancellationToken ct = default) => + Task.FromResult>(Array.Empty()); + } + /// /// In-memory enumerator returning a static list of sites. /// @@ -727,4 +790,123 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture + /// Reproduces the P5-6 defect end to end: a permanently-poisoned row sits + /// in a batch alongside a co-batch row that ALSO fails on every pull but + /// carries a fresh each time (modelling a + /// site that keeps producing new events with transient insert failures). + /// Per PullSiteAsync's hasUnresolvedFailure rule the rotating + /// row never individually reaches MaxPermanentInsertAttempts, so it + /// holds the reconciliation cursor at since forever — the poison row + /// keeps being re-pulled with a reset attempt counter (its + /// _failedInsertAttempts entry was cleared by its first abandonment) + /// and is permanently abandoned again roughly every + /// MaxPermanentInsertAttempts ticks. Ten ticks reproduces two full + /// abandon cycles for the SAME poisoned EventId. Pre-fix, each cycle mints a + /// fresh random synthetic EventId, so two DISTINCT rows land in the + /// repository. Post-fix, DeriveAbandonmentEventId derives the same id + /// both times and InsertIfNotExistsAsync's idempotency collapses the + /// second insert into a no-op, leaving exactly one row. + /// + [Fact] + public void ReAbandonment_SameLostEventId_YieldsExactlyOneCentralRow() + { + var siteId = "siteReabandon"; + var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc); + var poison = NewEvent(siteId, poisonTime); + + // 10 scripted responses: enough for exactly two 5-attempt abandon + // cycles of the poison row (the cursor never advances between them, + // so the SAME poison row + a freshly-guid'd rotating sibling are + // re-served every tick). Once exhausted, ScriptedPullClient returns + // empty responses, so no further cycles occur — the test's eventual + // row count is bounded and deterministic. + var responses = Enumerable.Range(0, 10) + .Select(i => new PullAuditEventsResponse( + new[] { 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 actor = CreateActor(sites, client, repo, FastTickOptions()); + + // Wait for all 10 scripted pulls to have been consumed, i.e. both + // abandon cycles have run to completion. + AwaitAssert( + () => Assert.True(client.Calls.Count >= 10, $"expected >= 10 pull calls, got {client.Calls.Count}"), + duration: TimeSpan.FromSeconds(10), + interval: TimeSpan.FromMilliseconds(100)); + + // Give the actor's mailbox a chance to fully drain the 10th tick's + // processing (the call count above only proves PullAsync was invoked, + // not that the resulting insert(s) have landed yet) before reading + // the repository state. + AwaitAssert( + () => Assert.True( + repo.Inserted.Any(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)), + "expected at least one synthetic ReconciliationAbandoned row"), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(100)); + + StopReconciliationActor(actor); + + var synthetic = repo.Inserted + .Where(e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)) + .ToList(); + + // The defect: pre-fix this is 2 (a fresh random EventId per abandon + // cycle defeats InsertIfNotExistsAsync's dedup). Post-fix it is 1. + Assert.Single(synthetic); + Assert.Contains(poison.EventId.ToString(), synthetic[0].DetailsJson); + } + + /// + /// Pure-function coverage of + /// directly: distinct lost rows must derive distinct synthetic ids (so two + /// different permanently-abandoned rows don't collide into one central row). + /// + [Fact] + public void DeriveAbandonmentEventId_DifferentLostRows_YieldDistinctSyntheticIds() + { + var lostA = Guid.NewGuid(); + var lostB = Guid.NewGuid(); + + var derivedA = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostA); + var derivedB = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostB); + + Assert.NotEqual(derivedA, derivedB); + Assert.NotEqual(lostA, derivedA); + Assert.NotEqual(lostB, derivedB); + } + + /// + /// Pure-function coverage proving the derivation depends ONLY on the lost + /// row's VALUE — not on object identity, not + /// on any clock/random/process state. Two independently-parsed + /// values that are equal by value must derive the identical synthetic id, and + /// the exact value is pinned as a regression guard: since the function has no + /// external input besides , this literal must + /// reproduce identically in any process — this run, CI, or a reconciliation + /// singleton after a failover restart — which is the "stable across process + /// restarts" contract the fix depends on for idempotent re-abandonment. + /// + [Fact] + public void DeriveAbandonmentEventId_IsPureAndStableAcrossIndependentInvocations() + { + var lostEventId = Guid.Parse("6b2f1e3a-8c9d-4a2b-9e3f-1a2b3c4d5e6f"); + var roundTripped = Guid.Parse(lostEventId.ToString("D")); + + var first = SiteAuditReconciliationActor.DeriveAbandonmentEventId(lostEventId); + var second = SiteAuditReconciliationActor.DeriveAbandonmentEventId(roundTripped); + + Assert.Equal(first, second); + Assert.Equal(Guid.Parse("36baac53-f4c6-835e-9a77-d0b350a4e0d9"), first); + } }