diff --git a/docs/requirements/Component-AuditLog.md b/docs/requirements/Component-AuditLog.md index c1575305..b2f14675 100644 --- a/docs/requirements/Component-AuditLog.md +++ b/docs/requirements/Component-AuditLog.md @@ -138,6 +138,7 @@ row per lifecycle event across all channels. | `SecuredWriteApprove` | A verifier wins the approval CAS for a pending secured write. | | `SecuredWriteReject` | A verifier rejects a pending secured write (`Status=Discarded`). | | `SecuredWriteExecute` | The approved write was relayed to the site MxGateway — terminal outcome (`Delivered`-equivalent on success, `Failed` on error). | +| `ReconciliationAbandoned` | Central-direct synthetic row (`Status=Failed`) written when a reconciliation pull row failed to insert on every retry up to the permanent-abandon threshold and central advanced its cursor past it. `Extra` carries the abandoned `EventId`, the source site, and the final error — so the permanent loss is queryable, not only in the Critical log line. | Inbound API is intentionally collapsed to a single `InboundRequest` (or `InboundAuthFailure` for auth rejections) row per request rather than a @@ -394,6 +395,15 @@ MS SQL for direct-write events). Unredacted secrets never persist. - **Reconciliation as fallback.** If two consecutive reconciliation cycles report a non-draining backlog, the supervisor restarts the telemetry actor and 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 + 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`, + channel preserved from the lost row where parseable) whose `Extra` carries the + abandoned `EventId`, the source site, and the final error — so the permanent + loss is queryable in the Audit Log itself. That write is best-effort in its own + try/catch and never blocks the cursor a second time. - **No dedup horizon.** `EventId` PK enforces uniqueness only while a row exists. A retry that arrives after the original row is purged inserts a "new" row — vanishingly rare and harmless. diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs index 217aba01..870b812a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Central/SiteAuditReconciliationActor.cs @@ -1,9 +1,11 @@ +using System.Text.Json; using Akka.Actor; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories; using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit; +using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central; @@ -290,6 +292,13 @@ public class SiteAuditReconciliationActor : ReceiveActor attempts); _failedInsertAttempts.Remove(evt.EventId); advanceForThisRow = true; + + // Task 17: the permanent loss must be queryable in the Audit Log + // itself, not only in a rotating log file. Leave a durable + // synthetic ReconciliationAbandoned row recording the lost id, + // the site, and the final error. + await EmitAbandonmentRecordAsync(repository, evt, site.SiteId, ex, nowUtc) + .ConfigureAwait(false); } else { @@ -324,6 +333,68 @@ public class SiteAuditReconciliationActor : ReceiveActor UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream); } + /// + /// Writes ONE synthetic audit + /// row when a pulled event is permanently abandoned (Task 17) — 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 + /// 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. + /// + private async Task EmitAbandonmentRecordAsync( + IAuditLogRepository repository, + ZB.MOM.WW.Audit.AuditEvent lost, + string siteId, + Exception error, + DateTime nowUtc) + { + try + { + // Category is the channel name (BuildCategory); preserve it where it + // parses, else ApiOutbound so the row still lands (mirrors the Task-14 + // fallback convention). + var channel = Enum.TryParse(lost.Category, out var parsed) + ? parsed + : AuditChannel.ApiOutbound; + + var extra = JsonSerializer.Serialize(new + { + abandonedEventId = lost.EventId, + sourceSiteId = siteId, + error = error.Message, + }); + + var synthetic = ScadaBridgeAuditEventFactory.Create( + channel: channel, + kind: AuditKind.ReconciliationAbandoned, + status: AuditStatus.Failed, + occurredAtUtc: nowUtc, + actor: "system", + sourceNode: lost.SourceNode, + sourceSiteId: siteId, + errorMessage: error.Message, + extra: extra, + ingestedAtUtc: nowUtc); + + await repository.InsertIfNotExistsAsync(synthetic).ConfigureAwait(false); + } + catch (Exception ex) + { + _logger.LogError( + ex, + "Failed to write the synthetic ReconciliationAbandoned audit row for the " + + "permanently-abandoned AuditEvent {EventId} (site {SiteId}); the loss is recorded " + + "only in the Critical log line.", + lost.EventId, + siteId); + } + } + /// /// 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/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Enums/AuditKind.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Enums/AuditKind.cs index f6cf2790..30005e2a 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Enums/AuditKind.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Types/Enums/AuditKind.cs @@ -22,5 +22,15 @@ public enum AuditKind SecuredWriteSubmit, SecuredWriteApprove, SecuredWriteReject, - SecuredWriteExecute + SecuredWriteExecute, + + /// + /// A reconciliation pull row that failed to insert on every retry up to the + /// permanent-abandon threshold, so central advanced its cursor past it and + /// the row is permanently lost from the mirror. The reconciliation actor + /// emits ONE synthetic audit row of this kind (carrying the abandoned + /// EventId, source site, and final error) so the loss is queryable in + /// the Audit Log itself, not only in a rotating log file. (arch-review 04, Task 17) + /// + ReconciliationAbandoned } 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 341d24e5..bb4b9eef 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Central/SiteAuditReconciliationActorTests.cs @@ -113,6 +113,69 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture>(Array.Empty()); } + /// + /// Recording stub whose throws for ONE + /// poison (every attempt) but records every + /// other row — including the synthetic ReconciliationAbandoned row the actor + /// emits once it gives up on the poison row (Task 17). + /// + private sealed class PoisonInsertRepo : IAuditLogRepository + { + private readonly Guid _poisonId; + private readonly Exception _fault; + public List Inserted { get; } = new(); + private readonly HashSet _seen = new(); + + public PoisonInsertRepo(Guid poisonId, Exception fault) + { + _poisonId = poisonId; + _fault = fault; + } + + public Task InsertIfNotExistsAsync(AuditEvent evt, CancellationToken ct = default) + { + if (evt.EventId == _poisonId) + { + throw _fault; + } + 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. /// @@ -457,4 +520,56 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture new PullAuditEventsResponse(new[] { poison }, MoreAvailable: false)) + .ToArray(); + var client = new ScriptedPullClient().Script(siteId, responses); + var repo = new PoisonInsertRepo( + poison.EventId, new InvalidOperationException("poison row will never insert")); + + CreateActor(sites, client, repo, FastTickOptions()); + + // After the 5th failed attempt the actor abandons the row and writes ONE + // synthetic ReconciliationAbandoned row (fresh EventId) whose details carry + // the abandoned id + the site. + AwaitAssert( + () => + { + var synthetic = repo.Inserted.SingleOrDefault( + e => e.Action.EndsWith(".ReconciliationAbandoned", StringComparison.Ordinal)); + Assert.NotNull(synthetic); + Assert.NotEqual(poison.EventId, synthetic!.EventId); + Assert.Contains(poison.EventId.ToString(), synthetic.DetailsJson); + Assert.Contains(siteId, synthetic.DetailsJson); + }, + duration: TimeSpan.FromSeconds(8), + interval: TimeSpan.FromMilliseconds(100)); + + // The cursor advanced past MinValue to the abandoned row's timestamp — the + // poison row no longer pins the site's reconciliation forever. A later tick + // must pull with since = poisonTime (poll for it; the abandonment tick and + // the first advanced-cursor pull are separate ticks). + AwaitAssert( + () => Assert.Contains(client.Calls, c => c.SinceUtc == poisonTime), + duration: TimeSpan.FromSeconds(5), + interval: TimeSpan.FromMilliseconds(100)); + } }