[Service Bus] Drain receive link and release messages on receiver close (#42917)#47999
Open
pranz1996 wants to merge 5 commits into
Open
[Service Bus] Drain receive link and release messages on receiver close (#42917)#47999pranz1996 wants to merge 5 commits into
pranz1996 wants to merge 5 commits into
Conversation
On close, a PEEK_LOCK receiver dropped messages that had been prefetched into the client buffer but not yet returned to the application, leaving them locked at the broker until lock expiry (delaying redelivery and inflating delivery_count). Both the sync (ReceiveClient.close) and async (ReceiveClientAsync.close_async) pyamqp paths now release buffered-but- unconsumed messages with the `released` disposition before clearing the buffer. Gated on PEEK_LOCK (ReceiverSettleMode.Second) and guarded so a faulted link cannot block close. Adds deterministic regression tests. Fixes Azure#42917 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: bef8092b-2a9f-4862-aef1-680eeb2e97f0
Contributor
There was a problem hiding this comment.
Pull request overview
Fixes delayed redelivery of buffered PEEK_LOCK messages when Service Bus receivers close.
Changes:
- Releases buffered messages during synchronous and asynchronous close.
- Adds seven regression tests.
- Documents the fix in the changelog.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
CHANGELOG.md |
Documents the receiver-close fix. |
_pyamqp/client.py |
Drains buffered messages during synchronous close. |
_pyamqp/aio/_client_async.py |
Implements asynchronous buffer draining. |
test_receiver_drain_on_close.py |
Tests settlement modes, empty buffers, and failures. |
Reworks the Azure#42917 fix to keep the vendored _pyamqp package untouched (it is shared with azure-eventhub and maintained in lockstep). The earlier edit to _pyamqp/client.py and _client_async.py is reverted; drain-on-close now lives in the Service Bus transport layer: - PyamqpTransport.drain_receive_link_and_release_messages (+ async twin) sends a drain flow (flow link_credit=0, drain=True), pumps the connection until quiescent (bounded by RECEIVE_LINK_DRAIN_TIMEOUT), then releases buffered/in-flight messages with the `released` disposition. No-op for the deprecated uamqp transport. - ServiceBusReceiver._close_handler calls it, gated to non-session PEEK_LOCK receivers (parity with the .NET SDK). Draining in-flight (not just already-buffered) messages fully eliminates the delayed redelivery and inflated delivery count in the reporter's scenario. Fixes Azure#42917 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: bef8092b-2a9f-4862-aef1-680eeb2e97f0
pranz1996
changed the title
- Always release buffered deliveries when the link is open. The previous early-return on zero link credit also skipped releasing already-buffered messages, re-introducing the bug for prefetch (prefetch_count > 0) receivers whose credit is exhausted while _received_messages still has deliveries. Only the drain flow is now gated on outstanding credit. - Harden drain quiescence: pump with batch=outstanding so a multi-frame (fragmented) transfer is assembled within a listen() cycle, and require two consecutive idle cycles before concluding (still deadline-bounded). pyamqp does not surface the broker drain response, so queue-growth remains the signal. - Tests: add no-credit-still-releases (sync/async) and empty-buffer no-op; assert closed link settles nothing. Fixes Azure#42917 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: bef8092b-2a9f-4862-aef1-680eeb2e97f0
) The drain deadline was only checked between listen() calls, so a large socket_timeout (or a continuously-available batch) could let a single blocking read overrun RECEIVE_LINK_DRAIN_TIMEOUT. Each listen() now waits at most the remaining drain budget (wait = min(socket_timeout, remaining)), so close() is bounded by the cap regardless of socket_timeout. The bounded tests now use a listen mock that actually blocks for its wait argument with a large socket_timeout and assert close() returns within the cap. Fixes Azure#42917 Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: bef8092b-2a9f-4862-aef1-680eeb2e97f0
pranz1996
requested review from
EldertGrootenboer,
SwayGom,
j7nw4r,
sagar0207 and
skarri-microsoft
as code owners
Member
Author
|
@microsoft-github-policy-service agree company="Microsoft" |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Fixes #42917. A
PEEK_LOCKService Bus receiver, on close, dropped messagesthat had been prefetched into the client buffer or were still in flight on
the wire (leftover link credit), leaving them locked at the broker until lock
expiry — delaying redelivery and inflating
delivery_count(risking prematuredead-lettering under repeated receiver churn).
Repro
The reporter's scenario, reproduced against a live namespace:
prefetch_count=0, receiver recreated each loop;receive_messagesreturns partialbatches, leaving outstanding link credit;
Bug metric: messages redelivered with
delivery_count > 0— i.e. leftlocked at close and redelivered only after the lock expired.
Why the test queue uses
LockDuration = 10s: this is a test-environmentchoice, not part of the fix. The default lock duration is up to 5 minutes, so a
lock-expiry redelivery would take minutes to observe. A short 10s lock makes the
bug surface within seconds. The fix itself uses no
sleepand nolock-duration-dependent timing — the reporter's
asyncio.sleep(10)workaroundis removed entirely.
Fix (transport layer — vendored
_pyamqpuntouched)The
_pyamqppackage is vendored and shared withazure-eventhub, so the fixlives in the Service Bus transport layer instead of the vendored client:
PyamqpTransport.drain_receive_link_and_release_messages(and the asynctwin) sends a drain flow (
flow(link_credit=0, drain=True)), pumps theconnection until quiescent (bounded by
RECEIVE_LINK_DRAIN_TIMEOUT = 5s, whichexits early on quiescence — not a sleep), then releases every buffered/in-flight
message with the
releaseddisposition. No-op for the deprecateduamqptransport.
ServiceBusReceiver._close_handlercalls it, gated to non-sessionPEEK_LOCK receivers (parity with the .NET SDK; Event Hubs /
RECEIVE_AND_DELETE use
Firstmode where the drain is a no-op).released(notmodified/abandon) means the broker redelivers immediatelywithout incrementing
delivery_count.Tested after the fix
Unit (
tests/unittests/test_receiver_drain_on_close.py, 18 tests):transport drain/release mechanics (drain flow sent, in-flight pulled in,
released; skipped on no-credit / closed / null link; failure-guarded; timeout-
bounded) + receiver gating (PEEK_LOCK-non-session drains; RECEIVE_AND_DELETE and
session skip) — sync and async. Verified all behavior tests fail when the fix
is disabled. Full offline unit-test suite passes; pylint/mypy clean on changed
files.
Live A/B on a real namespace (queue LockDuration 10s), reporter's scenario,
no
sleepworkaround, ~60–90s per run — lock-expiry redeliveries(
delivery_count > 0) per run:POST_DELAYamplifier)Skipped paths: RECEIVE_AND_DELETE and session receivers receive all messages
and close cleanly (drain correctly skipped).
Are there any breaking changes?
No. Behavior only improves (locks released sooner); no public API change.