Deregister pull consumer heartbeat callbacks on stop

[caspervonb]

Pull message streams and batches register disconnect/reconnect callbacks on the client for the heartbeat timer (ADR-37) but never deregister them. Every created-then-stopped consumer therefore leaks two callbacks plus the stream object they bind, growing unbounded for any pattern that re-creates the message stream over one connection.

Removes them from the cleanup paths that already tear down the subscription, using the remove_*_callback methods added in #957 (this PR is stacked on that branch).

Closes #962.

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 2m 28s —— View job

---

PR Review: Deregister pull consumer heartbeat callbacks on stop

• Gather context and read changed files
• Analyze implementation correctness
• Check test coverage
• Post review findings

---

The fix is correct and minimal. The _deregister_callbacks() helper is properly idempotent (nulling self._client prevents double-removal), it's wired into all existing teardown paths (_cleanup for PullMessageStream, the except clause and pending-exhaustion guard for PullMessageBatch), and self._client = None is initialized unconditionally in both constructors so the attribute is always present.

One issue worth addressing: The except (StopAsyncIteration, asyncio.TimeoutError) block in PullMessageBatch.__anext__ doesn't catch asyncio.CancelledError, which is a BaseException since Python 3.8. A cancelled task will propagate the error without calling _deregister_callbacks() or unsubscribing — the same leak this PR fixes for normal termination. The same gap exists in PullMessageStream._cleanup too (but the _cleanup path is not async-cancelled itself since it's called explicitly). Left an inline comment with a suggestion. This is pre-existing for the subscription, but since _deregister_callbacks is new here it's a gap in the otherwise complete fix.

Tests cover the two main paths well: explicit stop() on PullMessageStream and iterator exhaustion on PullMessageBatch. The double-stop idempotency check is a nice touch.

[synadia-claude-reviewer]

The except (StopAsyncIteration, asyncio.TimeoutError) clause doesn't catch asyncio.CancelledError (a BaseException since Python 3.8). If the task iterating the batch is cancelled mid-flight, _deregister_callbacks() is skipped and the heartbeat callbacks leak — the same bug this PR fixes for the normal termination path.

This is technically a pre-existing gap (the subscription also isn't unsubscribed on cancellation), but it means the fix is incomplete for cancellation. Consider a try/finally wrapping both the unsubscribe and deregister calls.

[caspervonb]

Review comments addressed, PTAL @claude

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 1m 43s —— View job

---

Re-reviewing PR #963

• Gather context from previous review
• Read updated changed files
• Verify CancelledError fix in PullMessageBatch.__anext__
• Check other teardown paths
• Post updated review findings

---

The previous review comment has been fully addressed. Here's what changed and how it holds up:

PullMessageBatch.__anext__ — cancellation gap fixed

The previous structure used except (StopAsyncIteration, asyncio.TimeoutError) which didn't catch asyncio.CancelledError (a BaseException). The refactor in pull.py:126–231 correctly addresses this with a try/finally approach:

delivering = False
try:
    while True:
        ...
        delivering = True
        return js_msg
except (StopAsyncIteration, asyncio.TimeoutError):
    raise StopAsyncIteration
finally:
    if not delivering and not self._terminated:
        self._terminated = True
        self._deregister_callbacks()
        await self._subscription.unsubscribe()

The delivering flag is the key invariant: it's only True on the happy path (returning a message). For every other exit — exhaustion, timeout, cancellation, unexpected exception — the finally block runs cleanup. This is correct and covers the previously reported gap.

One subtle detail worth noting: _deregister_callbacks() is intentionally called before await self._subscription.unsubscribe() (and the comment explains why: the await could itself be interrupted by a second cancellation). This means even in a double-cancellation scenario, the heartbeat callbacks are removed while only the subscription cleanup might be lost — a sensible priority ordering.

Test coverage — new cancellation test

test_fetch_deregisters_heartbeat_callbacks_on_cancellation (test_consumer.py:432–458) directly validates the fixed path: a task iterating a batch with no messages is cancelled mid-flight, and the test confirms CancelledError propagates correctly and that the disconnect/reconnect callbacks are gone afterward. This is exactly the right regression test.

The two other new tests (test_messages_deregisters_heartbeat_callbacks_on_stop, test_fetch_deregisters_heartbeat_callbacks_on_exhaustion) cover the normal teardown paths, including the idempotent double-stop case.

Outstanding gap (pre-existing, out of scope)

PullMessageStream.__anext__ still catches only Exception at pull.py:351:

try:
    raw_msg = await self._subscription.next()
except Exception:
    await self._cleanup()
    raise StopAsyncIteration

asyncio.CancelledError bypasses this, leaving _request_task, _heartbeat_task, and the heartbeat callbacks alive on cancellation. This was flagged as pre-existing in the earlier review and remains unaddressed — but since the inline comment focused on PullMessageBatch, it's reasonable to track separately.

Summary

The specific issue raised in review is fully resolved. The fix is minimal, correct, and well-tested. The PullMessageStream cancellation gap can be a follow-up in a future PR.