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

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<Guid>(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<Guid>();
+            // 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
+                    // it there is nothing to look up, ever, so this row is
-                    // bad row doesn't block the rest of the batch. The audit
+                    // abandoned immediately rather than after the grace period.
-                    // row stays Pending (still not in emittedEventIds) and
+                    abandonedEventIds.Add(auditRow.EventId);
-                    // central reconciliation will pick it up.
-                    _logger.LogWarning(
-                        "Cached-telemetry drain: audit row {EventId} ({Action}) has no CorrelationId; skipping.",
-                        auditRow.EventId, auditRow.Action);
                     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
+                    // batch — the audit half is best-effort. A throw is a
-                    // this row; it stays Pending for the next drain.
+                    // STORE fault (locked, corrupt) rather than a verdict about
-                    _logger.LogWarning(ex,
+                    // this row, so the row is never abandoned on this path: it
-                        "Cached-telemetry drain: tracking lookup threw for {EventId} (TrackedOperationId {Tid}, sqlite {SqliteError}); skipping.",
+                    // stays Pending and retries once the store recovers.
-                        auditRow.EventId, auditRow.CorrelationId, SqliteErrorCodes.Describe(ex));
+                    lookupFailures++;
+                    firstLookupFailure ??= ex;
                     continue;
                 }
 
                 if (snapshot is null)
                 {
-                    // No tracking row — possible if the audit row is older
+                    // No tracking row. Within the grace window this is the
-                    // than the tracking retention window, or the tracking
+                    // ordinary write race (the audit row landed first), so
-                    // store was reset. The audit half remains valid and will
+                    // retry. Past it the snapshot is gone for good — tracking
-                    // be picked up by central reconciliation; skip the
+                    // retention elapsed, or the two stores were reset
-                    // combined push for this row.
+                    // independently — and retrying forever would wedge the
-                    _logger.LogWarning(
+                    // queue behind these rows.
-                        "Cached-telemetry drain: no tracking snapshot for {EventId} (TrackedOperationId {Tid}); skipping.",
+                    if (auditRow.OccurredAtUtc <= abandonBefore)
-                        auditRow.EventId, auditRow.CorrelationId);
+                    {
+                        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
+                // Nothing resolvable in this read. Any permanently-unresolvable
-                // tracking snapshot). Leave them Pending and try again next
+                // rows have just been marked Forwarded above, so the next drain
-                // drain — the underlying race normally resolves on its own.
+                // sees past them rather than re-reading the same head of queue.
                 return;
             }
 
@@ -358,6 +396,59 @@ public class SiteAuditTelemetryActor : ReceiveActor
         }
     }
 
+    /// <summary>
+    /// Marks cached rows whose operational half can never be built as
+    /// Forwarded, so they leave the cached-drain queue.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// <b>Why they must leave.</b> Leaving an unresolvable row Pending — the
+    /// previous behaviour — is not a harmless skip. The queue is read
+    /// oldest-first with a fixed <c>BatchSize</c>, 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).
+    /// </para>
+    /// <para>
+    /// <b>Why Forwarded is the honest state.</b> 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:
+    /// <c>ReadPendingSinceAsync</c> 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
+    /// (<c>SiteCalls</c>) half — and that is unrecoverable regardless, because
+    /// the tracking row it would have been built from no longer exists.
+    /// </para>
+    /// <para>
+    /// 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.
+    /// </para>
+    /// </remarks>
+    private async Task AbandonUnresolvableAsync(
+        IReadOnlyList<Guid> 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<AuditEvent> events)
     {
         var batch = new AuditEventBatch();

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.
     /// </summary>
     public int IdleIntervalSeconds { get; set; } = 30;
+
+    /// <summary>
+    /// How long a cached-telemetry audit row may go without a resolvable
+    /// <c>OperationTracking</c> snapshot before the drain ABANDONS its
+    /// operational half and marks the row Forwarded. Default: 300 s.
+    /// </summary>
+    /// <remarks>
+    /// <para>
+    /// 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
+    /// <see cref="BatchSize"/>, so a batch's worth of permanently-unresolvable
+    /// rows at the head STARVES every newer cached row behind them.
+    /// </para>
+    /// <para>
+    /// Abandoning costs only the operational (<c>SiteCalls</c>) 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.
+    /// </para>
+    /// <para>
+    /// Set to 0 to abandon on the first failed lookup. Negative values are
+    /// treated as 0.
+    /// </para>
+    /// </remarks>
+    public int CachedTrackingGraceSeconds { get; set; } = 300;
 }

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
+        // Assert — exactly one push containing ONLY the valid row: the orphan
-        // is skipped and stays Pending (not in MarkForwardedAsync's id list).
+        // 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<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == valid.EventId),
             Arg.Any<CancellationToken>());
+
+        // ...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<IReadOnlyList<Guid>>(g => g.Count == 1 && g[0] == orphan.EventId),
+            Arg.Any<CancellationToken>());
+    }
+
+    [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<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { fresh }));
+        _trackingStore.GetStatusAsync(
+                new TrackedOperationId(fresh.CorrelationId!.Value), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<TrackingStatusSnapshot?>(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<int>(), Arg.Any<CancellationToken>());
+        }, TimeSpan.FromSeconds(6));
+
+        await _queue.DidNotReceive().MarkForwardedAsync(
+            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
+    }
+
+    [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<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(stuck));
+        foreach (var row in stuck)
+        {
+            _trackingStore.GetStatusAsync(
+                    new TrackedOperationId(row.CorrelationId!.Value), Arg.Any<CancellationToken>())
+                .Returns(Task.FromResult<TrackingStatusSnapshot?>(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<IReadOnlyList<Guid>>(g => g.Count == stuck.Count),
+                Arg.Any<CancellationToken>());
+        }, 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<CachedTelemetryBatch>(), Arg.Any<CancellationToken>());
+    }
+
+    [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<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(Array.Empty<AuditEvent>()));
+        _queue.ReadPendingCachedTelemetryAsync(Arg.Any<int>(), Arg.Any<CancellationToken>())
+            .Returns(Task.FromResult<IReadOnlyList<AuditEvent>>(new[] { row }));
+        _trackingStore.GetStatusAsync(
+                new TrackedOperationId(row.CorrelationId!.Value), Arg.Any<CancellationToken>())
+            .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<int>(), Arg.Any<CancellationToken>());
+        }, TimeSpan.FromSeconds(6));
+
+        await _queue.DidNotReceive().MarkForwardedAsync(
+            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
     }
 
     [Fact]