Servicebus receive_messages limited when called in loop

[kegs-equinor]

• Package Name: azure-servicebus
• Package Version: 7.14.1
• Operating System: All (SDK Issue)
• Python Version: 3.12.3

Describe the bug

When calling receiver.receive_messages(max_message_count=100) in a loop, the maximum received messages will never be higher than the amount received in the last call.

To Reproduce
Steps to reproduce the behavior:

receiver = client.get_queue_receiver()
with receiver:
  # The queue originally contains 100 messages before starting
  msgs = receiver.receive_messages(max_message_count=100)
  print(len(msgs))  # 100
  # Add a single message to the queue
  msgs = receiver.receive_messages(max_message_count=100)
  print(len(msgs))  # 1
  # Add 100 messages to the queue
  msgs = receiver.receive_messages(max_message_count=100)
  print(len(msgs)) #  1
  msgs = receiver.receive_messages(max_message_count=100)
  print(len(msgs)) #  1
  msgs = receiver.receive_messages(max_message_count=100)
  print(len(msgs)) #  1
  ...

Expected behavior
The amount of messages received should not depend on how many were received in the previous call. I know that the receive call may return early with less that all the messages, but it should not be capped at the result of the previous call.

Additional context
I am pretty sure that the azure.servicebus._pyamqp.link Link.flow is the problem:

    def flow(self, *, link_credit: Optional[int] = None, **kwargs: Any) -> None:
        # Given the desired link credit `link_credit`, the link credit sent via
        # FlowFrame is calculated as follows: The link credit to flow on the wire
        # `self.current_link_credit` is the desired link credit
        # `link_credit` minus the current link credit on the wire `self.total_link_credit`.
        self.current_link_credit = link_credit - self.total_link_credit if link_credit is not None else self.link_credit

        # If the link credit to flow is greater than 0 (i.e the desired link credit is greater than
        # the current link credit on the wire), then we will send a flow to issue more link credit.
        # Otherwise link credit on the wire is sufficient.
        if self.current_link_credit > 0:
            # Calculate the total link credit on the wire, by adding the credit we will flow to the total link credit.
            self.total_link_credit = (
                self.current_link_credit + self.total_link_credit if link_credit is not None else self.link_credit
            )
            self._outgoing_flow(**kwargs)

This function assumes that when setting the flow to the sender is cumulative. It is not

[github-actions]

Hello @kegs-equinor. I'm an AI assistant for the azure-sdk-for-python repository. I have some suggestions that you can try out while the team gets back to you.

• The behavior appears to be linked to how the underlying AMQP link credit is managed when a previous call returns fewer messages than the max requested. For subsequent calls, the available credit may be capped to the previous call’s result (if credit is not replenished).
• Since the SDK uses a cumulative flow mechanism (as shown in the cited code), it may not correctly account for replenishing credit when messages are depleted early. This is consistent with the implementation in azure.servicebus._pyamqp.link as documented in the code snippet.
• As a workaround, consider explicitly looping until the queue is empty, or try manually triggering a fresh receiver (which resets link credit) when expecting a larger number of messages.
• Please check if this issue persists in the latest release or file a detailed repro to help us evaluate the link flow behavior against AMQP specifications. For additional guidance and to report bugs, refer to our Issues page.

The team will get back to you shortly, hopefully this helps in the meantime.

[kegs-equinor]

I have tested my use case by redefining flow to only:

    def flow(self, *, link_credit: Optional[int] = None, **kwargs: Any) -> None:
        self.current_link_credit = link_credit if link_credit is not None else self.link_credit
        if self.current_link_credit > 0:
            self._outgoing_flow(**kwargs)

That seems to solve my issue.

[kashifkhan]

Thank you for the feedback @kegs-equinor,

we will investigate and get back to you asap.

[swathipil]

Hi @kegs-equinor - Would you be able to provide the following info:

• How large are the messages that you're receiving?
• Is the queue pre-populated with messages that you're receiving? Or is sending happening in parallel while receiving?
• Are you completing messages after receiving? Or are you receiving in ReceiveAndDelete mode?