read_next(): task cancellation can be lost in the fetch/wait path — reader keeps polling forever, wait_for() never returns

[majkelx]

Symptom (production incident, 2026-07-21, halina @ OCA)

A caller wraps read_next() in a 2 s timeout (wait_for_psce(reader.read_next(), 2); serverish 2.0.5, nats-py 2.15.0, Python 3.10). Out of 18 readers started in the same second, 14 timed out normally ("stream is empty" path) and 4 never returned: the outer wait_for hung forever, while the inner read_next() task kept running its non-nowait wait loop.

Wire evidence captured 2 days into the hang — the reader was still polling every 2 s:

PUB $JS.API.CONSUMER.MSG.NEXT.tic_journal.SUZtJ4Cn...   ← fetch(1, timeout=2.0), msg_reader.py:326
HMSG ... (408 status)
PUB $JS.API.CONSUMER.INFO.tic_journal.SUZtJ4Cn...       ← consumer_info() after TimeoutError, :333
MSG {"num_pending":0,"consumer_seq":0,...}
(repeat every ~2 s, for 2 days)

Consumer created at 2026-07-21T16:05:00.906Z (the moment the run started), deliver_policy: by_start_time, num_pending: 0 — an "empty" subject, so the reader sat in the read_batch wait loop (msg_reader.py:316-343).

Analysis

asyncio.wait_for(task, 2) on timeout cancels the inner task and awaits it before raising TimeoutError. If the task swallows/loses the CancelledError, wait_for never returns — which is exactly what the caller observed. Since the same cancellation path works hundreds of times a day (every "stream is empty" break), this is a race, most likely around pull_subscription.fetch(1, timeout=...) (nats-py's fetch does its own timeout/cleanup handling internally) or consumer_info() — a CancelledError delivered inside those awaits appears to be consumed in some interleaving, after which read_next()'s while True continues as if nothing happened.

The except Exception clauses in read_batch / @async_shield do not catch CancelledError (BaseException in 3.10), so the swallow is either inside nats-py or in an interleaving where the cancel lands between fetch cycles and gets converted (e.g. into the caught asyncio.TimeoutError at msg_reader.py:330? — worth checking: if cancellation is requested while fetch is waiting, nats-py may surface it as its own TimeoutError, which serverish treats as "timeout is normal", losing the cancellation).

Suggested hardening

1. In the wait loop, after every except asyncio.TimeoutError from fetch, check asyncio.current_task().cancelling() (py3.11+) or track an explicit cancellation flag — and re-raise CancelledError if the task was cancelled. Cheap and closes the "cancel converted to TimeoutError" hole regardless of where it happens.
2. Consider making read_next cancellation-safe by design: run the fetch in a subtask and await asyncio.wait({subtask, cancel_event}) style, so external cancel always wins.
3. Repro starting point: the stall-proxy tooling from the 2026-05-03 ocam investigation (oca_monitor/scratch/nats_stall_proxy.py) — inject delayed 408s around the cancel moment.

Related: #17 (reader behaviour on connection issues), #19 (journal hang).