Batching pipelines: back-pressure sees their depth, faulted receivers rebuild, OOM can't fault the block (CritterWatch#942)

[jeremydmiller]

Context

CritterWatch RC.7 moved telemetry ingest to BatchMessagesOf and a production fleet (10.8k agents, 117 msg/s steady inflow) went from 2.7 GiB to its 6 GiB GC hard limit in two minutes, then split non-deterministically between a hard restart and a zombie: the SQS receive loop polling forever against a terminally faulted receive block, deleting nothing, 934 identical This Block<Envelope> has faulted errors while status reported green. Full mechanical analysis in JasperFx/CritterWatch#942.

Three composable defects, all fixed here. Verified first that neither 6.24.8 nor current JasperFx (#506 closed via #509 — loud-only) addresses any of them.

1. BatchMessagesOf severs the back-pressure chain

The transport receive block (bounded, the only stage BackPressureAgent watches) drains instantly into the batching channel and onward to the batch handler's local queue — deliberately unbounded (GH-3287) and excluded from back-pressure. QueueCount reads ~0 while the real backlog (every member pinned via Envelope.Batch with payload + deserialized graph, ~20–30× wire size each) grows without bound.

Fix — accounting, not bounding: BatchingPendingCounts tracks members in flight per originating listener address; ListeningAgent.QueueCount adds it to the receive block's depth, so the existing latch oscillates exactly as it did under per-message handling. Increment in BatchingProcessor.HandleAsync; settle exactly once per grouped envelope at both receivers' CompleteAsync batch terminals (Envelope.BatchPendingSettled guard). Nothing blocks, and local publishers (null Listener) are never counted — the GH-3287 no-deadlock rule for self-cascading handlers is preserved.

2. Retention amplification

SqsListener now releases Message.Body (UTF-16 base64, ~2.7× wire size, previously pinned for the envelope's whole life) once mapping into the envelope succeeds; the mapping-failure DLQ path keeps it. Deliberately not done: nulling Envelope.Data post-deserialization — dead-letter persistence and durable retry re-upserts store Data and would be corrupted. (The complementary consumer-side fix — CritterWatch's brotli serializer no longer overwriting Data with the decompressed payload — ships in CritterWatch.)

3. An OOM in a recovery path faults the block terminally, forever

Every catch rung in BufferedReceiver.executeAsync, both onBlockErrors, and DurableReceiver's requeue-on-failure allocate (logging), so the same memory pressure that fails a message can throw again inside recovery, hit the block's error rung, and fault it — no un-fault exists, QueueCount freezes (a latched listener never resumes), every post throws (an Accepting listener receives and drops forever), and StartAsync's _receiver ??= re-attached each new transport listener to the same corpse.

Fix: all recovery rungs are exception-safe; receivers record the #506 terminal fault (null-item OnError) as IFaultTrackingReceiver.HasFaulted; IListeningAgent.ReceiverHasFaulted exposes it for health checks (default-interface false, non-breaking); StartAsync disposes and rebuilds a faulted receiver; and BackPressureAgent's 2s check force-restarts on fault — covering both the frozen-latched and the Accepting-but-dropping zombie.

Known gaps (on purpose)

• Durable-mode endpoints with back-pressure disabled have no automatic fault watchdog (BackPressureAgent hosts the timer); RestartAsync(force: true) remains the manual remediation.
• Listener health: detect dead/hung receive-loop Tasks + extract a shared polling-loop abstraction #3236 receive-loop health reporting doesn't yet read ReceiverHasFaulted.
• No JasperFx changes required; optional follow-ups there would be lifting Failure onto IBlock and an un-fault primitive.

Tests

CoreTests 2292 passed / 0 failed (net9.0), including 6 new BatchingPendingCountsTests and 3 new BackPressureAgent watchdog tests (forces_a_full_rebuild_when_the_receiver_has_terminally_faulted, faulted_receiver_recovery_fires_even_while_accepting, no_rebuild_when_the_receiver_is_healthy). Wolverine.AmazonSqs builds; its LocalStack integration suite was not run.

🤖 Generated with Claude Code

https://claude.ai/code/session_016v2Aijyo8MX2AdPUZL5VtG