Request idle heartbeats for ordered consumers

[caspervonb]

An ordered consumer recovers by resetting when its inner message stream ends (OrderedMessageStream.__anext__). Nothing ended that stream when the consumer stopped answering: messages() left the heartbeat unset, so no monitor ran and no idle_heartbeat was requested from the server, and the iterator waited forever.

The monitor could not have ended it anyway — a missed window only reset the pending counts and asked again, which loops indefinitely against a consumer that no longer exists. That retry is kept for the first miss, where a batch may simply have been lost; a second consecutive miss is now terminal.

This is what has been stalling the nats-jetstream Windows job for hours since the job was added in #998. test_ordered_consumer_messages_server_restart restarts the server, after which the consumer (memory storage) is gone; its pull requests go unanswered and the iterator never returns. That test now bounds its wait too, so a regression fails instead of hanging the run.

Not reproducible on Linux or macOS, where the reconnect happens to produce a reply that ends the inner stream. Verified by checking the new tests fail against main.

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 4m 10s —— View job

---

PR Review

• Gather context and read PR diff
• Review ordered.py changes
• Review pull.py changes
• Review test changes
• Post inline comments for specific issues
• Post overall summary

---

The fix is correct and well-motivated. The root cause analysis is accurate: messages() left heartbeats unset on the inner PullMessageStream, so no monitor ran and a gone consumer left the iterator waiting forever. Making heartbeats mandatory for ordered consumers and adding a two-strike termination policy in the monitor are the right primitives.

Logic / correctness

The two-strike approach (_MAX_MISSED_HEARTBEATS = 2) is sound. After the first miss, _missed_heartbeats stays at 1 until a real message or heartbeat resets it to 0 in __anext__. If the server never responds again, the second miss fires and terminates. The _heartbeat_deadline reset after the first miss is correct (placed after _send_request(), so the clock starts when the retry actually went out).

The min(_DEFAULT_HEARTBEAT, max_wait / 2) formula for the default heartbeat ensures idle_heartbeat < expires (the server rejects the other way around) across all three parametrized test cases.

The contextlib.suppress(Exception) around unsubscribe() in the terminal path is correct: this task must not call _cleanup() itself because _cleanup() would cancel and then await the very task it is running on.

One medium issue: the disconnect/reconnect callbacks registered in __init__ are never removed in _cleanup() (see inline comment). This was a pre-existing bug, but the PR makes heartbeats universal for ordered consumers, so the callback leak now occurs on every inner-stream reset rather than only when the caller opted into heartbeats.

One minor issue: the terminal path calls unsubscribe() to wake __anext__, then __anext__'s exception handler calls _cleanup() which calls unsubscribe() again (see inline comment). Probably fine if unsubscribe() is idempotent, but worth confirming.

Tests cover the key cases well: auto-heartbeat computation across the (None, 10.0, 4.0) max-wait variants, explicit heartbeat passthrough, the two-miss termination path (via a mocked-silent request loop), and the server-restart recovery with a 60s timeout guard. The 60s bound on the server-restart test is the direct fix for the Windows CI hang — a good pragmatic choice.

[synadia-claude-reviewer]

The disconnect/reconnect callbacks registered just below this (in the same __init__ block) are never removed in _cleanup(). For ordered consumers, each inner stream reset creates a fresh PullMessageStream and registers new callbacks, while the old (now-dead) ones accumulate on the client. Since this PR makes heartbeats mandatory for ordered consumers, the leak now occurs on every reset rather than only when the caller explicitly sets heartbeat.

Consider deregistering in _cleanup():

client.remove_disconnected_callback(self._pause_heartbeat_timer)
client.remove_reconnected_callback(self._resume_heartbeat_timer)

(Worth tracking as a follow-up if nats-core doesn't currently expose remove variants.)

[caspervonb]

Confirmed, and the amplification you describe is right — this PR turns it from opt-in to every inner-stream reset.

Not fixing it here though: #963 ("Deregister pull consumer heartbeat callbacks on stop") already adds _deregister_callbacks() to both PullMessageBatch and PullMessageStream and calls it from _cleanup(). Two PRs editing the same _cleanup() would only conflict.

remove_disconnected_callback / remove_reconnected_callback do exist in nats-core (client/__init__.py:1553 and :1571), so that fix is sound as written. The terminal path added here still reaches _cleanup() via __anext__, so it will be covered once #963 lands.

The practical upshot is that #963 matters more after this PR than before it.

[synadia-claude-reviewer]

When the terminal path fires here, unsubscribe() is called to wake the pending __anext__. That __anext__'s exception handler then calls _cleanup(), which calls unsubscribe() a second time (line 523). This is likely benign if unsubscribe() is idempotent, but any error from the second call in _cleanup() is not suppressed — worth confirming.

[caspervonb]

Confirmed idempotent, so no change needed. Subscription.unsubscribe() is guarded (nats-core/src/nats/client/subscription.py:294):

if not self._closed:
    await self._client._unsubscribe(self._sid)
    self._pending_queue.shutdown(immediate=True)
    ...
    self._closed = True

The second call from _cleanup() is a no-op that cannot raise, so there is nothing for the unsuppressed path to surface.

[synadia-claude-reviewer]

@pytest.mark.asyncio is redundant — nats-jetstream uses asyncio_mode = "auto". Other tests in this file don't carry it. Nit only.

[caspervonb]

The premise is off — 33 of the 34 async tests in this file carry @pytest.mark.asyncio, so it is the convention here rather than the exception.

There was a real bug underneath it though. Inserting this test above test_consumer_reset_to_seq took over that test's decorator, leaving it without one and pulling an unrelated test into the diff. Harmless under asyncio_mode = "auto", but wrong.

Fixed in 4403aeb: the marker is back on test_consumer_reset_to_seq, the count is 34/34, and the diff to this file is now purely additive.

[caspervonb]

PTAL @claude

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 3m 22s —— View job

---

PR Review

• Read changed files and git diff
• Review ordered.py changes
• Review pull.py changes
• Review test changes
• Assess previous reviewer's comments and author responses
• Post final review

---

The fix is correct and ready to merge. Root cause, implementation, and test coverage are all sound. Addressing each aspect in turn:

ordered.py

The _create_inner_stream change is the core fix. Two things are correct:

1. Always set heartbeat. The formula min(_DEFAULT_HEARTBEAT, max_wait / 2) guarantees idle_heartbeat < expires for all parametrized cases — the server enforces this. The test covers all three (None, 10.0, 4.0) cases and asserts inner._heartbeat < inner._expires explicitly. The edge case max_wait = 0 is not a concern; the server already rejects expires = 0.

2. Explicit heartbeat pass-through. self._heartbeat if self._heartbeat is not None else ... correctly preserves a caller-supplied value. The dedicated test confirms this is untouched.

pull.py

The two-strike mechanism (_MAX_MISSED_HEARTBEATS = 2) is correct:

• _missed_heartbeats increments in the monitor, resets to 0 in __anext__ on any message (including status 100/heartbeats, which hit the reset code before the case "100": continue). This means a single late heartbeat fully resets the counter, which is the right semantics.
• After the first miss, _heartbeat_deadline is reset after _send_request() (line 493), so the clock restarts cleanly when the retry actually goes out.
• The terminal path sets _terminated = True first, suppresses any exception from unsubscribe(), and returns without calling _cleanup(). The comment explains why: _cleanup() would cancel and then await the task it is currently running on.

One pre-existing behaviour worth noting (not introduced here): _heartbeat_monitor doesn't check _heartbeat_paused before comparing against _heartbeat_deadline. If a disconnect happens after the deadline has already passed, the monitor can fire a miss during the outage even though silence is expected. The pause saves _heartbeat_remaining for resume but doesn't prevent the check from firing. For ordered consumers this is benign — a spurious miss triggers a reset, which is the intended recovery path. For plain pull consumers it incorrectly consumes a strike. This predates the PR and is out of scope, but worth tracking alongside #963.

Tests

test_messages_end_after_repeated_missed_heartbeats: Effective white-box test. One practical note: the test uses heartbeat=0.2, max_wait=5.0. The initial pull request has expires=5s and the server heartbeats it every 0.2s. Each heartbeat resets _missed_heartbeats = 0, so the first miss cannot fire until after the pull expires (~5s), then two 0.4s windows pass — total runtime ~6s. The 10s wait_for covers it, but a smaller max_wait (e.g., 0.5s) would make the test much faster if speed becomes a concern later.

test_ordered_consumer_messages_request_heartbeats: Correctly calls the private _create_inner_stream() to inspect the heartbeat value before any request goes out. The finally block stops both the inner and outer streams cleanly.

test_ordered_consumer_messages_server_restart: The 60s wait_for is the direct fix for the Windows CI hang. 60s is generous but appropriate — recovery involves reconnect, consumer recreation, and delivery; trimming it risks flakiness on slow CI.

Previous Review Comments

All resolved satisfactorily:

• Callback leak (Initialize signature_cb when only user_jwt defined #284): Confirmed pre-existing, deferred to Deregister pull consumer heartbeat callbacks on stop #963 which already has _deregister_callbacks() in both PullMessageBatch and PullMessageStream. Splitting the _cleanup() edits across two PRs would only conflict.
• Double unsubscribe() (Fix connect() #484): Confirmed idempotent via if not self._closed: guard in Subscription.unsubscribe(). Second call is a no-op that cannot raise.
• @pytest.mark.asyncio nit (#1087): The marker follows file convention (33/34 tests carry it). The underlying bug — marker borrowed from the next test — was fixed in 4403aeb, making the diff purely additive.

Verdict: LGTM. The fix is minimal, well-reasoned, and correctly addresses the root cause. No blocking issues.

[scottf]

LGTM