diff --git a/docs/known-issues/2026-07-20-cached-telemetry-drain-hot-loop.md b/docs/known-issues/2026-07-20-cached-telemetry-drain-hot-loop.md index d44037a5..fcadfde8 100644 --- a/docs/known-issues/2026-07-20-cached-telemetry-drain-hot-loop.md +++ b/docs/known-issues/2026-07-20-cached-telemetry-drain-hot-loop.md @@ -1,8 +1,55 @@ # Cached-telemetry drain hot-loops forever on a row whose tracking snapshot is gone -**Date:** 2026-07-20 · **Status:** OPEN · **Severity:** Medium (log flood + wasted I/O; no data loss) +**Date:** 2026-07-20 · **Status:** RESOLVED (2026-07-20) · **Severity:** High — revised up from Medium on investigation (permanent stall of the cached-telemetry path, not just a log flood) · **Area:** AuditLog / Site Telemetry +## Resolution (2026-07-20) + +Fixed by letting an unresolvable row LEAVE the queue. +`SiteAuditTelemetryActor.OnCachedDrainAsync` now abandons the operational half of a cached row +whose tracking snapshot is still unresolvable after a grace period +(`SiteAuditTelemetryOptions.CachedTrackingGraceSeconds`, default **300 s**) and marks it +`Forwarded`. A row with no `CorrelationId` at all is abandoned immediately — it can never resolve. + +**Marking `Forwarded` does not drop the audit data.** That state's role in this machine is "no +longer owed by the drain, still eligible for reconciliation", which is exactly the situation: +`ISiteAuditQueue.ReadPendingSinceAsync` covers `Forwarded` rows as well as `Pending` ones and +central dedups on `EventId`, so the reconciliation pull still delivers them. What is genuinely lost +is the operational (`SiteCalls`) half — unrecoverable regardless, since the tracking row it would +have been built from no longer exists. + +Three further behaviours, each deliberate: + +- **Inside the grace window the row is still retried.** A missing snapshot is normally a brief + write race (the audit row lands microseconds before the tracking row); abandoning on the first + failed lookup would discard the operational half of every cached call that lost that race. +- **A tracking-store THROW never abandons.** A throw is a store fault (locked, corrupt, + mid-restore), not a verdict about the row. Those rows stay `Pending` however old they are. +- **Logging is per drain pass, not per row.** One summary line each for abandoned, lookup-failed + (with the first exception attached) and deferred rows. The deferred case dropped to Debug — being + inside the grace window is normal operation, not a warning. + +### Severity was revised UP during the fix + +The original write-up called this a log flood with "no data loss", which understated it. The queue +is read **oldest-first with a fixed `BatchSize` (default 256)**, so once a batch's worth of +permanently-unresolvable rows collects at the head, every drain re-reads exactly those rows, fails +identically, and **never reaches the newer rows behind them**. The cached-telemetry path stalls +permanently. The audit half still reached central by reconciliation, so "no data loss" held — but +"Medium" did not. + +**Fixed in:** `SiteAuditTelemetryActor.cs` (`AbandonUnresolvableAsync` + the rewritten skip +branches), `SiteAuditTelemetryOptions.cs` (`CachedTrackingGraceSeconds`). +**Tests:** `SiteAuditTelemetryActorTests` — `CachedDrain_OrphanRow_PastGrace_IsAbandoned_...`, +`..._InsideGrace_IsRetried_NotAbandoned`, `..._FullBatchOfUnresolvableRows_DoesNotStarveTheQueue`, +`..._TrackingStoreThrow_DoesNotAbandonTheRow`. The pre-existing +`CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash` was **retargeted, not deleted** — +it had pinned the defect ("skipped and stays Pending"), so its orphan assertion is inverted while +the half that still holds is kept verbatim. Verified non-vacuous: with abandonment disabled the two +abandonment tests go red; the two must-NOT-abandon guards stay green in both directions. + +The diagnosis below is retained as the record of how it was found. + ## Summary `SiteAuditTelemetryActor`'s cached-telemetry drain reads Pending audit rows, looks up each row's diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs index dbc9e463..5d6d2bf3 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryActor.cs @@ -261,6 +261,18 @@ public class SiteAuditTelemetryActor : ReceiveActor var batch = new CachedTelemetryBatch(); var emittedEventIds = new List(pending.Count); + // Rows whose operational half can never be built. They are marked + // Forwarded so they LEAVE this queue — see AbandonUnresolvableAsync + // for why that is safe and why leaving them Pending is not. + var abandonedEventIds = new List(); + // Counted, not logged per row: a whole batch can fail identically + // (tracking store down, tracking retention elapsed), and one line + // per row per drain is what turned this into a log flood. + var deferredNoSnapshot = 0; + var lookupFailures = 0; + Exception? firstLookupFailure = null; + var graceSeconds = Math.Max(0, _options.CachedTrackingGraceSeconds); + var abandonBefore = DateTime.UtcNow - TimeSpan.FromSeconds(graceSeconds); foreach (var auditRow in pending) { @@ -268,13 +280,9 @@ public class SiteAuditTelemetryActor : ReceiveActor { // CorrelationId carries the TrackedOperationId for cached // rows — see CachedCallLifecycleBridge.BuildPacket. Without - // it we can't look up the tracking row; log + skip so the - // bad row doesn't block the rest of the batch. The audit - // row stays Pending (still not in emittedEventIds) and - // central reconciliation will pick it up. - _logger.LogWarning( - "Cached-telemetry drain: audit row {EventId} ({Action}) has no CorrelationId; skipping.", - auditRow.EventId, auditRow.Action); + // it there is nothing to look up, ever, so this row is + // abandoned immediately rather than after the grace period. + abandonedEventIds.Add(auditRow.EventId); continue; } @@ -288,24 +296,32 @@ public class SiteAuditTelemetryActor : ReceiveActor catch (Exception ex) { // A tracking-store throw must NOT abort the rest of the - // batch — the audit half is best-effort. Log and skip - // this row; it stays Pending for the next drain. - _logger.LogWarning(ex, - "Cached-telemetry drain: tracking lookup threw for {EventId} (TrackedOperationId {Tid}, sqlite {SqliteError}); skipping.", - auditRow.EventId, auditRow.CorrelationId, SqliteErrorCodes.Describe(ex)); + // batch — the audit half is best-effort. A throw is a + // STORE fault (locked, corrupt) rather than a verdict about + // this row, so the row is never abandoned on this path: it + // stays Pending and retries once the store recovers. + lookupFailures++; + firstLookupFailure ??= ex; continue; } if (snapshot is null) { - // No tracking row — possible if the audit row is older - // than the tracking retention window, or the tracking - // store was reset. The audit half remains valid and will - // be picked up by central reconciliation; skip the - // combined push for this row. - _logger.LogWarning( - "Cached-telemetry drain: no tracking snapshot for {EventId} (TrackedOperationId {Tid}); skipping.", - auditRow.EventId, auditRow.CorrelationId); + // No tracking row. Within the grace window this is the + // ordinary write race (the audit row landed first), so + // retry. Past it the snapshot is gone for good — tracking + // retention elapsed, or the two stores were reset + // independently — and retrying forever would wedge the + // queue behind these rows. + if (auditRow.OccurredAtUtc <= abandonBefore) + { + abandonedEventIds.Add(auditRow.EventId); + } + else + { + deferredNoSnapshot++; + } + continue; } @@ -314,11 +330,33 @@ public class SiteAuditTelemetryActor : ReceiveActor emittedEventIds.Add(auditRow.EventId); } + if (lookupFailures > 0) + { + _logger.LogWarning(firstLookupFailure, + "Cached-telemetry drain: tracking lookup failed for {Count} of {Total} rows " + + "(sqlite {SqliteError}); they stay Pending and retry next drain. First failure attached.", + lookupFailures, pending.Count, SqliteErrorCodes.Describe(firstLookupFailure!)); + } + + if (deferredNoSnapshot > 0) + { + _logger.LogDebug( + "Cached-telemetry drain: {Count} row(s) have no tracking snapshot yet and are inside the " + + "{Grace}s grace window; retrying next drain.", + deferredNoSnapshot, graceSeconds); + } + + if (abandonedEventIds.Count > 0) + { + await AbandonUnresolvableAsync(abandonedEventIds, graceSeconds, ct) + .ConfigureAwait(false); + } + if (batch.Packets.Count == 0) { - // Every row in this read was skipped (no CorrelationId / no - // tracking snapshot). Leave them Pending and try again next - // drain — the underlying race normally resolves on its own. + // Nothing resolvable in this read. Any permanently-unresolvable + // rows have just been marked Forwarded above, so the next drain + // sees past them rather than re-reading the same head of queue. return; } @@ -358,6 +396,59 @@ public class SiteAuditTelemetryActor : ReceiveActor } } + /// + /// Marks cached rows whose operational half can never be built as + /// Forwarded, so they leave the cached-drain queue. + /// + /// + /// + /// Why they must leave. Leaving an unresolvable row Pending — the + /// previous behaviour — is not a harmless skip. The queue is read + /// oldest-first with a fixed BatchSize, so once a batch's worth of + /// unresolvable rows collects at the head, every drain re-reads exactly + /// those rows, fails identically, and never sees the newer rows behind + /// them. The cached-telemetry path stalls permanently, and each pass logs + /// once per row (measured at ~2 800 warnings/minute on a rig). + /// + /// + /// Why Forwarded is the honest state. Its role in this state machine + /// is "no longer owed by the drain, still eligible for reconciliation" — + /// which is exactly the situation here. The audit half is NOT dropped: + /// ReadPendingSinceAsync covers Forwarded rows as well as Pending + /// ones and central dedups on EventId, so the reconciliation pull still + /// delivers them. What is genuinely lost is the operational + /// (SiteCalls) half — and that is unrecoverable regardless, because + /// the tracking row it would have been built from no longer exists. + /// + /// + /// A failure to mark is swallowed: the rows simply stay Pending and the + /// next drain retries. Escalating here would take down the audit drain over + /// a best-effort cleanup. + /// + /// + private async Task AbandonUnresolvableAsync( + IReadOnlyList eventIds, int graceSeconds, CancellationToken ct) + { + try + { + await _queue.MarkForwardedAsync(eventIds, ct).ConfigureAwait(false); + _logger.LogWarning( + "Cached-telemetry drain: abandoned the operational half of {Count} row(s) with no " + + "resolvable tracking snapshot after {Grace}s; marked Forwarded so they no longer block " + + "the queue. The audit half still reaches central via the reconciliation pull. This is " + + "expected after tracking retention elapses or the tracking store is reset independently " + + "of the audit store.", + eventIds.Count, graceSeconds); + } + catch (Exception ex) + { + _logger.LogWarning(ex, + "Cached-telemetry drain: could not mark {Count} unresolvable row(s) Forwarded " + + "(sqlite {SqliteError}); they stay Pending and will be retried.", + eventIds.Count, SqliteErrorCodes.Describe(ex)); + } + } + private static AuditEventBatch BuildBatch(IReadOnlyList events) { var batch = new AuditEventBatch(); diff --git a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs index e159e4c4..72318e2d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs +++ b/src/ZB.MOM.WW.ScadaBridge.AuditLog/Site/Telemetry/SiteAuditTelemetryOptions.cs @@ -25,4 +25,33 @@ public sealed class SiteAuditTelemetryOptions /// Longer interval avoids hammering an idle SQLite + gRPC channel. /// public int IdleIntervalSeconds { get; set; } = 30; + + /// + /// How long a cached-telemetry audit row may go without a resolvable + /// OperationTracking snapshot before the drain ABANDONS its + /// operational half and marks the row Forwarded. Default: 300 s. + /// + /// + /// + /// A missing tracking snapshot is normally a brief write race — the audit + /// row lands microseconds before the tracking row — so the drain retries + /// within this grace period. Past it the snapshot is almost certainly gone + /// for good (tracking retention elapsed while central was unreachable, or + /// the two stores were reset independently), and retrying forever is + /// actively harmful: the queue is read oldest-first with a fixed + /// , so a batch's worth of permanently-unresolvable + /// rows at the head STARVES every newer cached row behind them. + /// + /// + /// Abandoning costs only the operational (SiteCalls) half, which is + /// unrecoverable anyway once the tracking row is gone. The audit half is + /// NOT lost: the reconciliation pull covers Forwarded rows as well as + /// Pending ones, and central dedups on EventId. + /// + /// + /// Set to 0 to abandon on the first failed lookup. Negative values are + /// treated as 0. + /// + /// + public int CachedTrackingGraceSeconds { get; set; } = 300; } diff --git a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs index c19ed203..38416762 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.AuditLog.Tests/Site/Telemetry/SiteAuditTelemetryActorTests.cs @@ -362,11 +362,22 @@ public class SiteAuditTelemetryActorTests : TestKit } [Fact] - public async Task CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash() + public async Task CachedDrain_OrphanRow_PastGrace_IsAbandoned_AndTheValidRowStillFlows() { + // RETARGETED, not rewritten from scratch: this test previously asserted + // the orphan "is skipped and stays Pending", which is precisely the + // defect fixed here (known-issues/2026-07-20-cached-telemetry-drain- + // hot-loop.md) — a Pending row that can never resolve is re-read every + // drain forever and starves the queue behind it. The half of the test + // that still holds (the valid row flows, the drain does not crash) is + // kept verbatim; only the orphan's fate is inverted. + // + // NewCachedEvent stamps occurredAtUtc in the past, well beyond the + // default 300 s grace, so this orphan is past the abandon threshold. + // // Arrange — two cached audit rows: one with a tracking snapshot, one // orphaned (the tracking store returns null). The orphaned row must be - // skipped without aborting the batch — the valid row still flows. + // abandoned without aborting the batch — the valid row still flows. var orphan = NewCachedEvent(AuditKind.CachedSubmit); var valid = NewCachedEvent(AuditKind.CachedResolve); @@ -403,8 +414,8 @@ public class SiteAuditTelemetryActorTests : TestKit // Act CreateActorWithCachedDrain(); - // Assert — exactly one push containing ONLY the valid row; the orphan - // is skipped and stays Pending (not in MarkForwardedAsync's id list). + // Assert — exactly one push containing ONLY the valid row: the orphan + // has no operational half to send. await AwaitAssertAsync(async () => { await _client.Received(1).IngestCachedTelemetryAsync( @@ -415,9 +426,131 @@ public class SiteAuditTelemetryActorTests : TestKit Assert.Single(capturedBatch!.Packets); Assert.Equal(valid.EventId.ToString(), capturedBatch.Packets[0].AuditEvent.EventId); + // The valid row is marked Forwarded because central ack'd it... await _queue.Received(1).MarkForwardedAsync( Arg.Is>(g => g.Count == 1 && g[0] == valid.EventId), Arg.Any()); + + // ...and the orphan is marked Forwarded too, in its own call, so it + // LEAVES the cached queue. Its audit half is still delivered by the + // reconciliation pull, which covers Forwarded rows as well as Pending. + await _queue.Received(1).MarkForwardedAsync( + Arg.Is>(g => g.Count == 1 && g[0] == orphan.EventId), + Arg.Any()); + } + + [Fact] + public async Task CachedDrain_OrphanRow_InsideGrace_IsRetried_NotAbandoned() + { + // The other side of the grace window, and the reason abandonment is not + // immediate: a missing snapshot is normally a brief write race (the + // audit row lands microseconds before the tracking row). Abandoning on + // the first failed lookup would throw away the operational half of + // every cached call that happened to lose that race. + var fresh = ScadaBridgeAuditEventFactory.Create( + channel: AuditChannel.ApiOutbound, + kind: AuditKind.CachedSubmit, + status: AuditStatus.Submitted, + eventId: Guid.NewGuid(), + occurredAtUtc: DateTime.UtcNow, // inside the grace window + target: "ERP.GetOrder", + sourceSiteId: "site-1", + correlationId: Guid.NewGuid()); + + _queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + _queue.ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(new[] { fresh })); + _trackingStore.GetStatusAsync( + new TrackedOperationId(fresh.CorrelationId!.Value), Arg.Any()) + .Returns(Task.FromResult(null)); + + CreateActorWithCachedDrain(); + + // Give the drain several ticks to prove it keeps retrying rather than + // abandoning. A positive control guards against vacuity: the row must + // actually have been READ more than once, otherwise "never marked + // Forwarded" would also be true of a drain that never ran at all. + await AwaitAssertAsync(async () => + { + await _queue.Received(Quantity.Within(2, int.MaxValue)) + .ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()); + }, TimeSpan.FromSeconds(6)); + + await _queue.DidNotReceive().MarkForwardedAsync( + Arg.Any>(), Arg.Any()); + } + + [Fact] + public async Task CachedDrain_FullBatchOfUnresolvableRows_DoesNotStarveTheQueue() + { + // The actual regression. The queue is read oldest-first with a fixed + // BatchSize, so before the fix a batch's worth of permanently + // unresolvable rows at the head meant every subsequent drain re-read + // exactly those rows, failed identically, and never reached the newer + // rows behind them — a permanent stall of the cached-telemetry path, + // plus one log line per row per pass (~2,800/min measured on a rig). + var stuck = Enumerable.Range(0, 4).Select(_ => NewCachedEvent()).ToList(); + + _queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + _queue.ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(stuck)); + foreach (var row in stuck) + { + _trackingStore.GetStatusAsync( + new TrackedOperationId(row.CorrelationId!.Value), Arg.Any()) + .Returns(Task.FromResult(null)); + } + + CreateActorWithCachedDrain(); + + // Every row is abandoned in ONE call, so the head of the queue clears + // and the next drain can see past it. + await AwaitAssertAsync(async () => + { + await _queue.Received(1).MarkForwardedAsync( + Arg.Is>(g => g.Count == stuck.Count), + Arg.Any()); + }, TimeSpan.FromSeconds(5)); + + // Nothing was pushed — there was no operational half to build for any + // of them — but the drain still made progress rather than spinning. + await _client.DidNotReceive().IngestCachedTelemetryAsync( + Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task CachedDrain_TrackingStoreThrow_DoesNotAbandonTheRow() + { + // A throw is a STORE fault (locked, corrupt, mid-restore), not a + // verdict about this row: the snapshot may well exist and be readable + // a second later. Abandoning here would discard the operational half + // of every in-flight cached call during a transient SQLite lock — so + // the throw path deliberately never abandons, however old the row is. + var row = NewCachedEvent(); // deliberately older than the grace window + + _queue.ReadPendingAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(Array.Empty())); + _queue.ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()) + .Returns(Task.FromResult>(new[] { row })); + _trackingStore.GetStatusAsync( + new TrackedOperationId(row.CorrelationId!.Value), Arg.Any()) + .ThrowsAsync(new InvalidOperationException("tracking store unavailable")); + + CreateActorWithCachedDrain(); + + // Positive control again: prove the drain actually ran repeatedly, so + // "never abandoned" is a real observation rather than an artefact of a + // drain that never executed. + await AwaitAssertAsync(async () => + { + await _queue.Received(Quantity.Within(2, int.MaxValue)) + .ReadPendingCachedTelemetryAsync(Arg.Any(), Arg.Any()); + }, TimeSpan.FromSeconds(6)); + + await _queue.DidNotReceive().MarkForwardedAsync( + Arg.Any>(), Arg.Any()); } [Fact]