fix(audit): let an unresolvable cached row leave the drain queue

A cached-telemetry audit row whose OperationTracking snapshot could not be
resolved was skipped and left Pending. Nothing ever removed it, so the next
drain re-read it, failed identically, and logged again — forever.

The severity was worse than the original write-up said. The queue is read
oldest-first with a fixed BatchSize (default 256), so once a batch's worth of
permanently-unresolvable rows collected at the head, every drain re-read
exactly those rows and NEVER REACHED the newer rows behind them. That is a
permanent stall of the cached-telemetry path, not the "log flood, no data loss"
the issue was first filed as. Measured on the rig at ~2,800 warnings/minute,
surviving both a process restart and a container restart.

Fix: after a grace period (CachedTrackingGraceSeconds, default 300 s) the
operational half is abandoned and the row marked Forwarded. A row with no
CorrelationId is abandoned immediately — it can never resolve.

Marking Forwarded does not drop audit data. That state means "no longer owed by
the drain, still eligible for reconciliation", which is exactly this situation:
ReadPendingSinceAsync covers Forwarded as well as Pending and central dedups on
EventId, so the reconciliation pull still delivers the audit half. What is lost
is the operational (SiteCalls) half, which is unrecoverable anyway once the
tracking row is gone.

Three deliberate boundaries:
- Inside the grace window the row is still retried — a missing snapshot is
  normally a brief write race, and 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, however old the row: a throw is a
  store fault (locked, corrupt, mid-restore), not a verdict about the row.
- Logging is per drain pass, not per row. Deferred rows dropped to Debug —
  being inside the grace window is normal operation, not a warning.

CachedDrain_OrphanRow_NoTrackingSnapshot_IsSkipped_DoesNotCrash was RETARGETED
rather than deleted: it pinned the defect ("skipped and stays Pending"), so its
orphan assertion is inverted and the half that still holds is kept verbatim.
Three tests added, including the starvation regression. Both "must NOT abandon"
tests carry a positive control on the read count, so they cannot pass by virtue
of a drain that never ran.

Verified non-vacuous: with abandonment disabled the two abandonment tests go
red and the two guards stay green. Build 0 warnings; AuditLog 358, Host 330,
SiteRuntime 512, StoreAndForward 130, Communication 312, Commons 684,
Integration 94 — all pass.

Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts

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
-                    // 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
         }
     }
 
+    /// <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
-        // 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<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]