feat(audit-log): permanently-abandoned reconciliation rows leave a durable ReconciliationAbandoned audit record
When a pulled AuditEvent fails to insert on every retry up to the permanent- abandon threshold, the reconciliation actor now writes ONE synthetic ReconciliationAbandoned row (new AuditKind) alongside the Critical log line — fresh EventId, Status=Failed, channel preserved from the lost row, Extra carrying the abandoned EventId + source site + final error — via the same ScadaBridgeAuditEventFactory the ingest path uses. Best-effort in its own try/catch so it never blocks the cursor twice. The permanent loss is now queryable in the Audit Log, not only in a rotating log file.
This commit is contained in:
@@ -138,6 +138,7 @@ row per lifecycle event across all channels.
|
|||||||
| `SecuredWriteApprove` | A verifier wins the approval CAS for a pending secured write. |
|
| `SecuredWriteApprove` | A verifier wins the approval CAS for a pending secured write. |
|
||||||
| `SecuredWriteReject` | A verifier rejects a pending secured write (`Status=Discarded`). |
|
| `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). |
|
| `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
|
Inbound API is intentionally collapsed to a single `InboundRequest` (or
|
||||||
`InboundAuthFailure` for auth rejections) row per request rather than a
|
`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
|
- **Reconciliation as fallback.** If two consecutive reconciliation cycles
|
||||||
report a non-draining backlog, the supervisor restarts the telemetry actor
|
report a non-draining backlog, the supervisor restarts the telemetry actor
|
||||||
and a `SiteAuditTelemetryStalled` event fires.
|
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
|
- **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"
|
exists. A retry that arrives after the original row is purged inserts a "new"
|
||||||
row — vanishingly rare and harmless.
|
row — vanishingly rare and harmless.
|
||||||
|
|||||||
@@ -1,9 +1,11 @@
|
|||||||
|
using System.Text.Json;
|
||||||
using Akka.Actor;
|
using Akka.Actor;
|
||||||
using Microsoft.Extensions.DependencyInjection;
|
using Microsoft.Extensions.DependencyInjection;
|
||||||
using Microsoft.Extensions.Logging;
|
using Microsoft.Extensions.Logging;
|
||||||
using Microsoft.Extensions.Options;
|
using Microsoft.Extensions.Options;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
using ZB.MOM.WW.ScadaBridge.Commons.Interfaces.Repositories;
|
||||||
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Audit;
|
||||||
|
using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
namespace ZB.MOM.WW.ScadaBridge.AuditLog.Central;
|
||||||
|
|
||||||
@@ -290,6 +292,13 @@ public class SiteAuditReconciliationActor : ReceiveActor
|
|||||||
attempts);
|
attempts);
|
||||||
_failedInsertAttempts.Remove(evt.EventId);
|
_failedInsertAttempts.Remove(evt.EventId);
|
||||||
advanceForThisRow = true;
|
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
|
else
|
||||||
{
|
{
|
||||||
@@ -324,6 +333,68 @@ public class SiteAuditReconciliationActor : ReceiveActor
|
|||||||
UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream);
|
UpdateStalledState(site.SiteId, draining: !nonDraining, eventStream);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Writes ONE synthetic <see cref="AuditKind.ReconciliationAbandoned"/> 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
|
||||||
|
/// <see cref="ScadaBridgeAuditEventFactory"/> the ingest path uses (fresh
|
||||||
|
/// <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>
|
||||||
|
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<AuditChannel>(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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>
|
/// <summary>
|
||||||
/// Flips the per-site stalled flag based on whether this tick drained the
|
/// 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
|
/// queue. A "draining" cycle is one where the server reported no more rows
|
||||||
|
|||||||
@@ -22,5 +22,15 @@ public enum AuditKind
|
|||||||
SecuredWriteSubmit,
|
SecuredWriteSubmit,
|
||||||
SecuredWriteApprove,
|
SecuredWriteApprove,
|
||||||
SecuredWriteReject,
|
SecuredWriteReject,
|
||||||
SecuredWriteExecute
|
SecuredWriteExecute,
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// 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
|
||||||
|
/// <c>EventId</c>, 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)
|
||||||
|
/// </summary>
|
||||||
|
ReconciliationAbandoned
|
||||||
}
|
}
|
||||||
|
|||||||
+115
@@ -113,6 +113,69 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
|
|||||||
Task.FromResult<IReadOnlyList<string>>(Array.Empty<string>());
|
Task.FromResult<IReadOnlyList<string>>(Array.Empty<string>());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// Recording stub whose <see cref="InsertIfNotExistsAsync"/> throws for ONE
|
||||||
|
/// poison <see cref="AuditEvent.EventId"/> (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).
|
||||||
|
/// </summary>
|
||||||
|
private sealed class PoisonInsertRepo : IAuditLogRepository
|
||||||
|
{
|
||||||
|
private readonly Guid _poisonId;
|
||||||
|
private readonly Exception _fault;
|
||||||
|
public List<AuditEvent> Inserted { get; } = new();
|
||||||
|
private readonly HashSet<Guid> _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<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>
|
/// <summary>
|
||||||
/// In-memory enumerator returning a static list of sites.
|
/// In-memory enumerator returning a static list of sites.
|
||||||
/// </summary>
|
/// </summary>
|
||||||
@@ -457,4 +520,56 @@ public class SiteAuditReconciliationActorTests : TestKit, IClassFixture<MsSqlMig
|
|||||||
Assert.False(second.Stalled);
|
Assert.False(second.Stalled);
|
||||||
Assert.Equal("siteA", second.SiteId);
|
Assert.Equal("siteA", second.SiteId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// ---------------------------------------------------------------------
|
||||||
|
// 9. Task 17: a permanently-abandoned row leaves a durable
|
||||||
|
// ReconciliationAbandoned audit record (not just a Critical log line),
|
||||||
|
// and the cursor advances past it.
|
||||||
|
// ---------------------------------------------------------------------
|
||||||
|
|
||||||
|
[Fact]
|
||||||
|
public void PermanentAbandonment_EmitsReconciliationAbandonedRow_AndAdvancesCursor()
|
||||||
|
{
|
||||||
|
var siteId = "siteAbandon";
|
||||||
|
var poisonTime = new DateTime(2026, 5, 20, 10, 5, 0, DateTimeKind.Utc);
|
||||||
|
var poison = NewEvent(siteId, poisonTime);
|
||||||
|
|
||||||
|
var sites = new StaticEnumerator(new SiteEntry(siteId, "http://x:8083"));
|
||||||
|
// Five pulls each returning the same poison row so its per-EventId attempt
|
||||||
|
// counter reaches MaxPermanentInsertAttempts (5); afterwards the scripted
|
||||||
|
// client returns empty and the actor settles.
|
||||||
|
var responses = Enumerable.Range(0, 5)
|
||||||
|
.Select(_ => 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));
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user