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

fix(auditlog): P5-6 — deterministic synthetic ReconciliationAbandoned EventId

Browse Source 
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).
This commit is contained in:
Joseph Doherty
2026-08-19 16:20:05 -04:00
parent abbbcd1a8d
commit 88fbc9e351
2 changed files with 331 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs
+149 -41
View File
@@ -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
/// </summary>
private readonly Dictionary<Guid, int> _failedInsertAttempts = new();
/// <summary>
/// EventIds already permanently abandoned earlier in this process's
/// lifetime — a cheap short-circuit ONLY, not a correctness mechanism.
/// </summary>
/// <remarks>
/// <para>
/// The cursor can be held at <c>since</c> indefinitely by an unrelated
/// co-batch row that keeps failing (see <c>hasUnresolvedFailure</c> in
/// <see cref="PullSiteAsync"/>), 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
/// <see cref="MaxPermanentInsertAttempts"/>-attempt retry cycle (since
/// <see cref="_failedInsertAttempts"/> 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.
/// </para>
/// <para>
/// This is deliberately NOT what makes re-abandonment safe —
/// <see cref="DeriveAbandonmentEventId"/>'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.
/// </para>
/// </remarks>
private readonly HashSet<Guid> _abandonedEventIds = new();
/// <summary>
/// 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
/// </summary>
private const int MaxPermanentInsertAttempts = 5;
/// <summary>
/// Fixed namespace GUID mixed into <see cref="DeriveAbandonmentEventId"/>'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.
/// </summary>
private static readonly Guid AbandonmentEventIdNamespace =
new("f2b6a6d1-6e2b-4b8b-9b3e-9a6b6a2c1d4e");
private ICancelable? _timer;
/// <summary>
@@ -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
/// <see cref="ScadaBridgeAuditEventFactory"/> the ingest path uses (fresh
/// <c>EventId</c>, <c>Actor="system"</c>, channel preserved from the lost row's
/// <see cref="ScadaBridgeAuditEventFactory"/> the ingest path uses (a
/// <see cref="DeriveAbandonmentEventId"/>-derived <c>EventId</c>,
/// <c>Actor="system"</c>, channel preserved from the lost row's
/// <c>Category</c> where parseable) and carries the abandoned <c>EventId</c>,
/// the source site, and the final error in <c>Extra</c>. 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.
/// </summary>
/// <remarks>
/// The synthetic <c>EventId</c> is deterministic (derived from
/// <c>lost.EventId</c>), not fresh, specifically so this call is idempotent:
/// a co-batch row that never resolves can hold the reconciliation cursor at
/// <c>since</c> indefinitely, causing the SAME lost row to be re-pulled and
/// re-abandoned on a later tick (its <see cref="_failedInsertAttempts"/> 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
/// <see cref="IAuditLogRepository.InsertIfNotExistsAsync"/> collapses every
/// repeat into the first insert instead of minting an unbounded number of
/// duplicate rows for one lost event.
/// </remarks>
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
}
}
/// <summary>
/// Derives a deterministic synthetic <see cref="ZB.MOM.WW.Audit.AuditEvent.EventId"/>
/// for the <see cref="AuditKind.ReconciliationAbandoned"/> row minted when
/// <paramref name="lostEventId"/> is permanently abandoned.
/// </summary>
/// <remarks>
/// <para>
/// Pure function of <paramref name="lostEventId"/> 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 <see cref="IAuditLogRepository.InsertIfNotExistsAsync"/>
/// even after a singleton failover wipes every in-memory tracking
/// dictionary this actor keeps (<see cref="_failedInsertAttempts"/>,
/// <see cref="_abandonedEventIds"/>, the cursors themselves).
/// </para>
/// <para>
/// Built as a name-based (RFC 4122 "version 5"-shaped) GUID: SHA-256 over
/// a fixed namespace GUID concatenated with <paramref name="lostEventId"/>'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.
/// </para>
/// </remarks>
internal static Guid DeriveAbandonmentEventId(Guid lostEventId)
{
Span<byte> input = stackalloc byte[32];
AbandonmentEventIdNamespace.TryWriteBytes(input[..16]);
lostEventId.TryWriteBytes(input[16..]);
Span<byte> 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);
}
/// <summary>
/// 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
tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs
+182
View File
@@ -176,6 +176,69 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
Task.FromResult<IReadOnlyList<string>>(Array.Empty<string>());
}
/// <summary>
/// Recording stub whose <see cref="InsertIfNotExistsAsync"/> throws for EVERY
/// row except a synthetic <see cref="AuditKind.ReconciliationAbandoned"/> row
/// (identified by its <c>Action</c> suffix, mirroring <see cref="PoisonInsertRepo"/>'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 <c>MaxPermanentInsertAttempts</c>, so it keeps
/// <c>hasUnresolvedFailure</c> true and holds the cursor at <c>since</c> forever
/// (P5-6's defect trigger). Dedupes inserted rows on <see cref="AuditEvent.EventId"/>
/// exactly like the real <c>InsertIfNotExistsAsync</c>, so a test asserting on
/// <see cref="Inserted"/>'s count is exercising the same idempotency contract
/// production relies on.
/// </summary>
private sealed class AlwaysFailUnlessSyntheticRepo : IAuditLogRepository
{
public List<AuditEvent> Inserted { get; } = new();
private readonly HashSet<Guid> _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<IReadOnlyList<AuditEvent>> QueryAsync(
AuditLogQueryFilter filter, AuditLogPaging paging, CancellationToken ct = default) =>
Task.FromResult<IReadOnlyList<AuditEvent>>(Inserted);
public Task<long> SwitchOutPartitionAsync(DateTime monthBoundary, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
Task.FromResult(0L);
public Task<long> PurgeChannelOlderThanAsync(
string channel, DateTime threshold, int batchSize, TimeSpan? commandTimeout = null, CancellationToken ct = default) =>
Task.FromResult(0L);
public Task<long> BackfillSourceNodeAsync(
string sentinel, DateTime before, int batchSize, CancellationToken ct = default) =>
Task.FromResult(0L);
public Task<IReadOnlyList<DateTime>> GetPartitionBoundariesOlderThanAsync(
DateTime threshold, CancellationToken ct = default) =>
Task.FromResult<IReadOnlyList<DateTime>>(Array.Empty<DateTime>());
public Task<ZB.MOM.WW.ScadaBridge.Commons.Types.AuditLogKpiSnapshot> 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<IReadOnlyList<ExecutionTreeNode>> GetExecutionTreeAsync(
Guid executionId, CancellationToken ct = default) =>
Task.FromResult<IReadOnlyList<ExecutionTreeNode>>(Array.Empty<ExecutionTreeNode>());
public Task<IReadOnlyList<string>> GetDistinctSourceNodesAsync(CancellationToken ct = default) =>
Task.FromResult<IReadOnlyList<string>>(Array.Empty<string>());
}
/// <summary>
/// In-memory enumerator returning a static list of sites.
/// </summary>
@@ -727,4 +790,123 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
StopReconciliationActor(actor);
}
// ---------------------------------------------------------------------
// 10. P5-6: deterministic synthetic ReconciliationAbandoned EventId
// ---------------------------------------------------------------------
/// <summary>
/// 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 <see cref="AuditEvent.EventId"/> each time (modelling a
/// site that keeps producing new events with transient insert failures).
/// Per <c>PullSiteAsync</c>'s <c>hasUnresolvedFailure</c> rule the rotating
/// row never individually reaches <c>MaxPermanentInsertAttempts</c>, so it
/// holds the reconciliation cursor at <c>since</c> forever — the poison row
/// keeps being re-pulled with a reset attempt counter (its
/// <c>_failedInsertAttempts</c> entry was cleared by its first abandonment)
/// and is permanently abandoned again roughly every
/// <c>MaxPermanentInsertAttempts</c> ticks. Ten ticks reproduces two full
/// abandon cycles for the SAME poisoned EventId. Pre-fix, each cycle mints a
/// fresh random synthetic <c>EventId</c>, so two DISTINCT rows land in the
/// repository. Post-fix, <c>DeriveAbandonmentEventId</c> derives the same id
/// both times and <c>InsertIfNotExistsAsync</c>'s idempotency collapses the
/// second insert into a no-op, leaving exactly one row.
/// </summary>
[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);
}
/// <summary>
/// Pure-function coverage of <see cref="SiteAuditReconciliationActor.DeriveAbandonmentEventId"/>
/// directly: distinct lost rows must derive distinct synthetic ids (so two
/// different permanently-abandoned rows don't collide into one central row).
/// </summary>
[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);
}
/// <summary>
/// Pure-function coverage proving the derivation depends ONLY on the lost
/// row's <see cref="AuditEvent.EventId"/> VALUE — not on object identity, not
/// on any clock/random/process state. Two independently-parsed <see cref="Guid"/>
/// 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 <paramref name="lostEventId"/>, 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.
/// </summary>
[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);
}
}