Fix disconnect watcher rescheduling a finished asyncio.Timeout (#59)

[aersam]

Fixes #59

Reproduced (deterministically)

The trigger is BaseHTTPMiddleware in front together with a server whose send() takes a loop iteration or two (Granian; a real socket write). It is not specific to failing queries:

• Under Granian with tests/_granian_app.py's @app.middleware("http"), ~97 % of all responses (both /ok and a query failing before the first byte) logged Task exception was never retrieved: RuntimeError('Cannot change state of finished Timeout') from watch_for_disconnect (598 out of 600 requests). Without the middleware: 0.
• In-process with plain asyncio: BaseHTTPMiddleware(PostgresJsonResponse(...)) and a FakeClient whose send() takes 2 loop iterations fails on every run (1 iteration does not).

Mechanism: BaseHTTPMiddleware's receive_or_disconnect() awaits the real receive() inside an anyio task group and cancels that group once response_sent is set. That cancellation and our watcher.cancel() from the finally hit the watcher together; asyncio delivers them as one CancelledError (the second Task.cancel() only sets _must_cancel on a task whose waiter is already done), anyio recognises it as its own and swallows it, and receive_or_disconnect() returns a synthetic http.disconnect. Traced: the watcher comes back from receive() with cancelling() == 1 and <Timeout [finished]>, so reschedule() raises (it only accepts state ENTERED; EXPIRING/EXPIRED/EXITED raise Cannot change state of … Timeout, CREATED raises has not been entered). So Task.cancel() alone is not enough to keep the watcher from running on: whatever receive() is wrapped in may absorb it.

Other interleavings checked:

• (a) disconnect delivered in the same iteration in which the body ends, with a plain receive(): Task.cancel() on a task whose awaited future is done sets _must_cancel, the next step throws CancelledError at the await, reschedule() is not reached. Safe; only a receive() that swallows the cancellation breaks this (above).
• (b) EXPIRING: only the watcher itself reschedules, once, so the scope can't be expiring when it does. Not reachable.
• (c) outer cancellation and (d) exceptions in the stream: both go through the same finally, i.e. reduce to (a).

Fix

PostgresJsonResponse.__call__: a streamed flag, set in the finally right before watcher.cancel() (still inside the scope). The watcher only keeps calling receive(), and only calls reschedule(), while it is unset, so a watcher whose cancellation was swallowed returns quietly instead of rescheduling an exited scope (or receiving on after the response).

CancelOnDisconnectRoute (defensive): its watcher waits on our own asyncio.Event, which can't swallow the cancellation, so the race doesn't happen there (also checked under Granian + BaseHTTPMiddleware: 0 errors on main for /cancellable-ok). It gets the same handled guard anyway.

Version 0.16.1 (uv.lock follows).

Tests

• test_pg_json_cancel.py::test_a_receive_that_swallows_the_watchers_cancellation_does_not_break_it[failing query / ok query]: BaseHTTPMiddleware + FakeClient(send_iterations=2), collects unhandled loop errors (new tests._asgi.unhandled_loop_errors(), with gc.collect()). Fails on main, passes with the fix.
• test_pg_json_granian.py::test_finished_responses_leave_no_failed_watcher_behind[/ok, /fails-at-once]: 20 requests each against real Granian with the middleware, asserts no never retrieved in the server log. Fails on main, passes with the fix.
• test_cancel_on_disconnect.py::test_a_watcher_whose_cancellation_is_swallowed_leaves_the_finished_scope_alone: forces the swallow with an Event whose wait() turns CancelledError into a return (the path can't be hit otherwise). Fails without the guard, passes with it.
• tests/test_pg_json_*.py + tests/test_cancel_on_disconnect.py: 75 passed locally; ty check pgdevkit and ruff check on the changed files are clean.

🤖 Generated with Claude Code