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

CLAUDE.md

@@ -134,13 +134,8 @@ Other peers in the `scadaproj` family (see `scadaproj/CLAUDE.md` for details): `
     - **Operational constraints (read before upgrading a site pair):** stop and start both nodes TOGETHER — rolling one at a time is no longer supported, since the legacy `SfBufferSnapshot` compatibility handler went with the replicator. And a node offline longer than `LocalDb:Replication:TombstoneRetention` (default 7 days) can resurrect deleted rows on rejoin. See `docs/deployment/topology-guide.md`.
     - `LocalDb:Replication:MaxBatchSize` batches by ROW COUNT, not bytes, against a 4 MB gRPC cap — the rig pins it to **16** (~70 KB worst-case `config_json` x 16 ~= 1.1 MB). The 500 default would allow ~35 MB.
 - All timestamps are UTC throughout the system.
-- Inter-cluster communication uses **three** transports, not two: **ClusterClient** for command/control (deployments, lifecycle, subscribe/unsubscribe handshake, snapshots); **gRPC** server-streaming for real-time data (attribute values, alarm states); and **plain token-gated HTTP** for the deployment config itself — notify-and-fetch, the site pulls the config from `DeploymentConfigEndpoints` (`ManagementService/DeploymentConfigEndpoints.cs`) with an `X-Deployment-Token` header, `AllowAnonymous` with the per-deployment token as the entire security boundary. Both CentralCommunicationActor and SiteCommunicationActor registered with receptionist (**per node, not as a singleton** — contact rotation reaches whichever node answers). Central creates one ClusterClient per site using NodeA/NodeB as contact points. Sites configure multiple central contact points for failover. Addresses cached in CentralCommunicationActor, refreshed periodically (60s) and on admin changes. Heartbeats serve health monitoring only. **Discovery is asymmetric by design:** central discovers sites from the *database* (`Site.NodeAAddress`/`NodeBAddress`, refreshable at runtime), sites discover central from *appsettings* (`ScadaBridge:Communication:CentralContactPoints`, static — restart required). **Central never buffers for an unreachable site** — the send is dropped with a warning and the caller's Ask times out; a `ConnectionStateChanged` mechanism built for this was deleted as dead code.
-- **All clusters share ONE ActorSystem name**, `"scadabridge"` — hardcoded at `AkkaHostedService.cs:191`. Central and each site are separate clusters *only* by seed-node partitioning. This is required, not incidental: Akka.Remote address matching means a ClusterClient could not reach a differently-named system.
-- **`ActiveNodeEvaluator.SelfIsOldestUp` is THE single definition of "active node"** (`Communication/ClusterState/ActiveNodeEvaluator.cs`) — the **oldest Up member** in a role scope, and explicitly **never `cluster.State.Leader`**: leadership (lowest address) is an Akka-internal concept that diverges from singleton placement permanently once the original first node restarts and rejoins, and both sides claim it during a partition. The equivalence *oldest-Up == where `ClusterSingletonManager` places singletons* **is** the design. `ClusterActivityEvaluator.SelfIsOldest`, the S&F delivery gate, `/health/active` and the heartbeat `IsActive` stamp all delegate here.
-- Site nodes carry **two Akka roles**: the base `Site` plus a site-specific `site-{SiteId}` (`BuildRoles`, `AkkaHostedService.cs:386`). Singletons scope to the **site-specific** role.
-- **Neither inter-cluster transport is authenticated or encrypted.** Akka remoting sets no `enable-ssl`, no secure cookie, no `trusted-selection-paths`; the gRPC listener is **h2c** and `LocalDbSyncAuthInterceptor` gates *only* `/localdb_sync.v1.LocalDbSync/`, so the whole `SiteStreamService` surface — including the `PullAuditEvents`/`PullSiteCalls` RPCs that return audit rows — is callable by anyone who can reach :8083. The boundary assumes a trusted network. (The HTTP fetch token and the LocalDb sync interceptor are the only authenticated pieces.)
-- **Akka frame size is the default 128 KB with `log-frame-size-exceeding` off**, and no custom serializer is configured — so payload carried over ClusterClient is JSON-escaped a second time by the default Newtonsoft serializer, roughly doubling it. Over the limit the transport drops **that one message** without tearing down the association (heartbeats keep flowing, the site still reports healthy) and the central Ask simply times out. See `docs/known-issues/2026-06-26-deploy-config-exceeds-akka-frame-size.md`; `DeployArtifactsCommand` still carries payload and remains exposed.
-- gRPC streaming channel — **note the direction is inverted from the data flow**: data moves site→central, but each **site node hosts the gRPC server** (`SiteStreamGrpcServer`, Kestrel h2c, port 8083, mapped **only in the Site branch** of `Program.cs`) and **central is the client**, dialling in. There is **no gRPC server on central at all**, which is why the two `Ingest*` unary RPCs — documented as a "central-side ingest surface" — are dead in practice (acknowledged in `AkkaHostedService.cs:510-519`); sites reach central over ClusterClient instead. Central creates per-site `SiteStreamGrpcClient` via `SiteStreamGrpcClientFactory`, keyed **`(siteId, endpoint)`** — the key was widened from site-only to fix an arch-review High where one session's NodeA→NodeB flip disposed a channel another session was still using. Site entity has GrpcNodeAAddress/GrpcNodeBAddress fields. Proto: `sitestream.proto`, **6 RPCs** (2 server-streaming: `SubscribeInstance`, `SubscribeSite` (site-wide, alarm-only); 4 unary: `IngestAuditEvents`, `IngestCachedTelemetry`, `PullAuditEvents`, `PullSiteCalls`), `SiteStreamEvent` (oneof: AttributeValueUpdate, AlarmStateUpdate). Field numbers are never reused; evolution is additive only (`AlarmStateUpdate` grew 7→23 fields for the native-alarm mirror). Generated C# is **vendored** under `Communication/SiteStreamGrpc/` with the `<Protobuf>` include commented out — regeneration is a manual toggle-build-copy-untoggle. DebugStreamEvent message removed (no longer flows through ClusterClient).
+- Inter-cluster communication uses two transports: ClusterClient for command/control (deployments, lifecycle, subscribe/unsubscribe handshake, snapshots) and gRPC server-streaming for real-time data (attribute values, alarm states). Both CentralCommunicationActor and SiteCommunicationActor registered with receptionist. Central creates one ClusterClient per site using NodeA/NodeB as contact points. Sites configure multiple central contact points for failover. Addresses cached in CentralCommunicationActor, refreshed periodically (60s) and on admin changes. Heartbeats serve health monitoring only.
+- gRPC streaming channel: SiteStreamGrpcServer on each site node (Kestrel HTTP/2, port 8083); central creates per-site SiteStreamGrpcClient via SiteStreamGrpcClientFactory. Site entity has GrpcNodeAAddress/GrpcNodeBAddress fields. Proto: sitestream.proto with SiteStreamService, SiteStreamEvent (oneof: AttributeValueUpdate, AlarmStateUpdate). DebugStreamEvent message removed (no longer flows through ClusterClient).
 - Native alarms: a read-only mirror of native alarms from OPC UA Alarms & Conditions servers and the MxAccess Gateway, unified onto an A&C-style condition model (`AlarmConditionState`: orthogonal Active/Acked/Confirmed/Shelved/Suppressed + 0–1000 severity) plus an `AlarmKind` discriminator (Computed/NativeOpcUa/NativeMxAccess). New DCL capability seam `IAlarmSubscribableConnection` (implemented by the OPC UA and MxGateway adapters); the `DataConnectionActor` opens ONE alarm feed per connection and routes transitions to instances by source-object reference. A `NativeAlarmActor` (peer to the computed `AlarmActor` under `InstanceActor`) mirrors one source binding: snapshot atomic-swap on (re)subscribe, retention (drops once inactive+acked), per-source cap, and site SQLite persistence (`native_alarm_state`, survives failover, cleared on redeploy/undeploy — mirrors static overrides). State streams to central over the additively-enriched gRPC `AlarmStateUpdate` (the existing computed `AlarmStateChanged` was enriched additively) and seeds via the DebugView snapshot. Authoring: `TemplateNativeAlarmSource` / `InstanceNativeAlarmSourceOverride` entities flatten to `ResolvedNativeAlarmSource` (inherit/compose/override); management commands + ManagementActor handlers + CLI (`template/instance native-alarm-source`) + Central UI (template editor tab + instance override panel) + enriched DebugView alarm table. Read-only — no ack-back; no central tables.
 - OPC UA / MxGateway UX (M7): operator **Alarm Summary** page (`/monitoring/alarms`, RequireDeployment, read-only) fans out the existing per-instance `DebugViewSnapshot` Ask (SemaphoreSlim-capped, partial-results tolerant) and aggregates client-side — no central alarm store; shared `AlarmStateBadges` component. **Aggregated live stream shipped 2026-07-10** (`docs/plans/2026-07-10-aggregated-live-alarm-stream-plan.md`): a transient in-memory per-site central live cache (`ISiteAlarmLiveCache`) fed by a site-wide, alarm-only `SubscribeSite` gRPC stream (seed-then-stream), pushing near-real-time deltas to the page over the Blazor circuit with the 15s poll kept as fallback + NotReporting authority — still no persisted central alarm store. OPC UA node browser gains `BrowseNext` continuation paging ("Load more"), a bounded recursive address-space **search** (`IAddressSpaceSearchable` seam; depth + result caps; substring on DisplayName/path), and **type-info** (DataType/ValueRank/Writable on `BrowseNode` for Variables). Attribute-override **CSV bulk import** (`OverrideCsvParser`, all-or-nothing) via InstanceConfigure `InputFile` + CLI `instance import-overrides --file` (native-alarm-source-override CSV deferred). **Verify-endpoint** probe (temporary `RealOpcUaClient`, short timeout, captures an untrusted server cert but NEVER trusts it) + **site-local cert trust**: per-node `CertStoreActor` (runs on every site node, not a singleton) writing the `.der` into the node's OPC UA trusted-peer PKI store; DeploymentManager broadcasts `TrustServerCertCommand`/`RemoveServerCertCommand` to BOTH site nodes so PKI stores stay consistent across failover; Admin-gated cert-management UI (`/design/connections/{id}/certificates`). No central persistence of cert trust (follow-up).
 
@@ -215,11 +210,9 @@ Other peers in the `scadaproj` family (see `scadaproj/CLAUDE.md` for details): `
 ### Cluster & Failover
 - Keep-oldest split-brain resolver with `down-if-alone = on`, 15s stable-after.
 - Both nodes are seed nodes. `min-nr-of-members = 1`.
-- Failure detection: 2s heartbeat, 10s threshold. Total failover ~25s (drill-measured 27s, 0 routing blips — `docker/failover-drill.sh`).
+- Failure detection: 2s heartbeat, 10s threshold. Total failover ~25s.
 - CoordinatedShutdown for graceful singleton handover.
 - Automatic dual-node recovery from persistent storage.
-- **Active/standby is decided by `ActiveNodeEvaluator.SelfIsOldestUp`, never by cluster leadership** — see the Architecture note above. `/health/active` is **central-only** (site nodes map no `/health/*` at all) and backs both Traefik's active-node routing and `IActiveNodeGate`, so the proxy and the Inbound API always agree on which node is active. Central never needs to know which *site* node is active: ClusterClient contact rotation reaches either receptionist and the site-internal `ClusterSingletonProxy` lands the work on the active node for free. The **exception is gRPC**, which picks `GrpcNodeAAddress`/`GrpcNodeBAddress` explicitly and flips on error.
-- **KNOWN GAP — two-node keep-oldest has a total-outage hole.** Only the *first* seed listed in `Cluster:SeedNodes` may self-join to form a new cluster, so a lone restarted non-first-seed node (first seed still down) loops on `InitJoin` forever — never `Up`, never routable. After the oldest/active node crashes, the younger survivor self-downs and **cannot re-bootstrap alone**; recovery is operator-driven. `failover-drill.sh` has `DRILL_MODE=active` specifically *"to make the registered gap observable — not to pretend it is covered."* Closing it is the registered deferred keep-oldest topology/strategy decision. See `docs/requirements/Component-ClusterInfrastructure.md:113`.
 
 ### UI & Monitoring
 - Central UI: Blazor Server (ASP.NET Core + SignalR) with Bootstrap CSS. No third-party component frameworks (no Blazorise, MudBlazor, Radzen, etc.). Build custom Blazor components for tables, grids, forms, etc.

Directory.Packages.props

@@ -108,9 +108,9 @@
     <PackageVersion Include="ZB.MOM.WW.Secrets.Abstractions" Version="0.2.3" />
     <PackageVersion Include="ZB.MOM.WW.Secrets.Ui" Version="0.2.3" />
     <PackageVersion Include="ZB.MOM.WW.Secrets.Replicator.SqlServer" Version="0.2.3" />
-    <PackageVersion Include="ZB.MOM.WW.LocalDb" Version="0.1.3" />
-    <PackageVersion Include="ZB.MOM.WW.LocalDb.Replication" Version="0.1.3" />
-    <PackageVersion Include="ZB.MOM.WW.LocalDb.Contracts" Version="0.1.3" />
+    <PackageVersion Include="ZB.MOM.WW.LocalDb" Version="0.1.1" />
+    <PackageVersion Include="ZB.MOM.WW.LocalDb.Replication" Version="0.1.1" />
+    <PackageVersion Include="ZB.MOM.WW.LocalDb.Contracts" Version="0.1.1" />
   </ItemGroup>
 
   <!--

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]