Fix PullSubscription.fetch hang due to orphan lingering request

[phemmer]

This fixes an issue in the nats-py code where a race condition occurs that leaves an orphaned lingering request on the server, causing fetch to hang until the timeout expires.

fixes #933

[synadia-claude-reviewer]

Claude finished @phemmer's task in 4m 20s —— View job

---

PR Review: Fix PullSubscription.fetch hang due to orphan lingering request

• Gather context
• Read PR diff and changed files
• Analyze correctness and asyncio patterns
• Post review feedback

---

Review

The fix is correct and the root cause analysis is accurate. The lingering pull request was being published with the original full expires (computed at the top of fetch()) even when most of the client-side timeout had already elapsed — specifically when the server's 408 for the no-wait probe arrives in the narrow window just before the Python asyncio timer fires. Using the remaining deadline for the lingering request's expires closes that window.

client.py changes

Deadline computation (lines 1279–1291): The two early-exit guards (deadline <= 0 and remaining_expires <= 0) are correct. The 100µs offset (- 100_000) is preserved, maintaining the invariant that the server sends its 408 slightly before the client's asyncio timer fires. The elif expires: fallback is only reachable when timeout is None (since _time_until returns None iff timeout is None), and in that case expires is also None, so no expiry is set — matching original behaviour.

_fetch_one is not fixed (batch=1): See the inline comment. The fix only applies to _fetch_n. _fetch_one still publishes with the full original expires. The probability is lower (no no-wait probe to consume time), but the same class of race exists.

Early break for NO_MESSAGES/REQUEST_TIMEOUT in the second loop (lines 1358–1364): This is a correct improvement. Previously, the second for _ in range(needed) loop had no explicit handling for these status codes — it would fall through to _is_processable_msg, which returns False for temporary errors, so the loop body would simply do nothing and continue to the next iteration, burning through remaining timeout unnecessarily. The new break correctly stops consuming messages once the server signals the pull request is exhausted.

Pre-existing issue at line 1338 (not introduced by this PR): elif status == api.StatusCode.NO_MESSAGES or status: — the or status is always truthy for any non-None status string, which makes the elif len(msgs) == 0: raise ... branch unreachable dead code. Worth a separate fix, but out of scope here.

test_js.py changes

The test faithfully reproduces the race: short first fetch on an empty stream forces the timeout path, then a subsequent fetch verifies the orphan doesn't capture the newly published message. The elapsed < 1.0 assertion with a 3 s fetch timeout gives plenty of headroom without being flaky.

[synadia-claude-reviewer]

The elif expires: fallback is only reachable when timeout is None (since deadline is None iff timeout is None). When timeout is None, expires is also None, so neither branch sets next_req["expires"] — which is correct.

However, _fetch_one (batch=1) has the same class of orphan bug and is not fixed here. In _fetch_one, the single lingering request is published with the original expires (computed from the full timeout at the call site), and start_time is measured only after that publish. If the Python asyncio timer fires before the server's 408 arrives, the lingering request is abandoned on the server with up to timeout_ns - 100_000 ns remaining. The probability is lower than in _fetch_n (no no-wait probe to consume time first), but the race is real.

[phemmer]

Added another fix. Completely separate issue, but related, so throwing it in this same PR. ab04fa6
Basically if there are messages in _pending_queue at the start of fetch(), it would call the server to ask for more, and then hang for the full timeout.

Also one thing that might be worth discussing: I preserved the original behavior in that if messages are sitting in _pending_queue when fetch() is called, it still calls the server to request more messages. I don't know that this is the ideal behavior. I would think that if messages are available in the local buffer, that we would want to return them immediately without having to perform a network call.

[phemmer]

This might also be worth discussing. This and line 1296 below raise asyncio.TimeoutError. This is consistent with line 1234 above. However I question whether this is the right behavior. There's also FetchTimeoutError. So the current code is consistent, but is it consistently wrong? Should it be FetchTimeoutError? And fix line 1234 as well?

[superlevure]

FYI we just upgraded to 2.15.0 in production and we're hitting many asyncio.TimeoutError that are uncaught because we expect nats.errors.TimeoutError instead. We have to revert to 2.14.0 because of this since we can't update all our applications to catch this new Exception.

[superlevure]

Tentative fix: #1008

[caspervonb]

Merged, looks good. Thank you very much!