initial_credits does not bound the number of unsettled deliveries

Client 2.0.0a3 on the v2_main branch.

Describe the bug

initial_credits does not bound how many deliveries a consumer holds unsettled. The count grows by initial_credits - 1 on every settlement, until the queue runs dry. initial_credits = 1 is the only value that holds its bound.

With the default DEFAULT_INITIAL_CREDITS = 100, a consumer ends up holding almost the whole queue. Raising MESSAGES to 1000 in the script below gives a peak of 989. The peak tracks the queue length, not the credit.

This contradicts the contract stated in consumer.py:

Credit is tied to settlement rather than to delivery, which is what bounds the number of unsettled deliveries in flight to initial_credits

Mechanism. Consumer._replenish_credit() grants flow(self._initial_credits) on every settlement, and ReceiverLink.flow() sends that absolute link-credit together with the receiver's current delivery-count. The sender's available credit is delivery_count_rcv + link_credit - delivery_count_snd, so re-granting the full window against a count that has already advanced re-opens the whole window instead of adding one: one delivery is settled and initial_credits more become allowed.

The docstring of _replenish_credit assumes the count "has advanced by one delivery" since the last flow. With initial_credits > 1 the initial grant alone makes that untrue, because the broker already has that many deliveries in flight.

unpause() grants flow(self._initial_credits) the same way.

Reproduction steps

1. Start RabbitMQ 4.x. Tested on 4.3.5.
2. pip install rabbitmq-amqp-python-client==2.0.0a3
3. Run the script below.

"""initial_credits does not bound the number of unsettled deliveries.

    in flight = published - settled - message_count(queue)

QueueInfo.message_count is what is still ready in the queue, so the remainder is
what the broker has delivered and this client has not settled.

Run against any RabbitMQ 4.x:
    pip install rabbitmq-amqp-python-client==2.0.0a3
    python repro.py [host] [user] [password] [vhost]
"""

import sys
import threading
import time
import uuid

from rabbitmq_amqp_python_client import Connection, ConnectionParameters, Context, Message

MESSAGES = 400
DELAY = 0.02
HOST, USER, PASSWORD, VHOST = (sys.argv[1:] + ["localhost", "guest", "guest", "/"][len(sys.argv) - 1 :])
PARAMETERS = ConnectionParameters(
    host=HOST, port=5672, user=USER, password=PASSWORD, virtual_host=VHOST
)


class Fixture:
    """A fresh queue holding MESSAGES ready messages, and a consumer over it."""

    def __init__(self) -> None:
        self.queue = f"repro-{uuid.uuid4()}"
        self.publishing = Connection(PARAMETERS)
        self.management = self.publishing.management()
        self.management.queue(self.queue).declare()
        publisher = self.publishing.publisher_builder().queue(self.queue).build()
        for index in range(MESSAGES):
            publisher.publish(Message(body=str(index).encode()))
        publisher.close()
        self.consuming: Connection | None = None
        self.consumer = None

    def consume(self, handler, credits: int) -> None:
        self.consuming = Connection(PARAMETERS)
        self.consumer = (
            self.consuming.consumer_builder()
            .queue(self.queue)
            .message_handler(handler)
            .initial_credits(credits)
            .build()
        )

    def in_flight(self, settled_now) -> int:
        # Read the queue before the settled count. The other order overestimates,
        # because settlements land between the two reads.
        ready = self.management.queue_info(self.queue).message_count
        return MESSAGES - settled_now() - ready

    def close(self) -> None:
        if self.consumer is not None:
            self.consumer.close()
        if self.consuming is not None:
            self.consuming.close()
        self.management.queue(self.queue).delete()
        self.publishing.close()


def in_flight_without_settling(credits: int) -> int:
    """Nothing settles, so nothing replenishes credit: in flight must equal credits."""
    fixture = Fixture()
    blocked = threading.Event()
    fixture.consume(lambda context, message: blocked.wait(), credits)
    time.sleep(4)
    observed = fixture.in_flight(lambda: 0)
    blocked.set()
    time.sleep(0.5)
    fixture.close()
    return observed


def peak_in_flight(credits: int) -> int:
    """Highest number of delivered-but-unsettled messages seen while draining."""
    fixture = Fixture()
    settled = 0
    lock = threading.Lock()

    def handler(context: Context, _message: Message) -> None:
        nonlocal settled
        time.sleep(DELAY)
        context.accept()
        with lock:
            settled += 1

    fixture.consume(handler, credits)
    def settled_now() -> int:
        with lock:
            return settled

    peak = 0
    deadline = time.monotonic() + 120
    while time.monotonic() < deadline:
        peak = max(peak, fixture.in_flight(settled_now))
        if settled_now() >= MESSAGES:
            break
        time.sleep(0.05)
    fixture.close()
    return peak


print(f"{MESSAGES} messages, handler delay {DELAY * 1000:.0f} ms\n")
print("control, handler never settles, initial_credits=3 -> in flight", in_flight_without_settling(3))
print()
print(f"{'initial_credits':>15} {'peak in flight':>15} {'expected':>9}")
for credits in (1, 2, 3, 5, 10, 100):
    print(f"{credits:>15} {peak_in_flight(credits):>15} {credits:>9}")

Output:

400 messages, handler delay 20 ms

control, handler never settles, initial_credits=3 -> in flight 3

initial_credits  peak in flight  expected
              1               1         1
              2             199         2
              3             266         3
              5             321         5
             10             360        10
            100             396       100

Expected behavior

At most initial_credits deliveries unsettled per consumer, as the module docstring states.

AmqpConsumer in rabbitmq-amqp-java-client holds that invariant explicitly: it keeps credit + unsettled at or below initialCredits.

Additional context

• consumer.py and link.py are byte-identical on v2_main HEAD, so this is the current code.
• Reproduces on classic and quorum queues. With a quorum queue and initial_credits=2 the peak is 201, and with 3 it is 267.
• The peak matches messages * (initial_credits - 1) / initial_credits + 1 for every value tested, which is what "one settled, initial_credits more allowed" predicts.
• Measured peaks vary by about 2 messages between runs. The sampling reads the queue and the settled count at slightly different moments, so the derived figure carries that much skew. Both controls below are stable across runs.
• Two controls: initial_credits = 1 holds its bound, and a handler that never settles holds at exactly initial_credits. The second shows the broker honours the grant, so the overshoot enters through replenishment and not through the initial grant.
• QueueInfo.message_count was cross-checked against the HTTP management API, which reports ready and unacked separately. The derived figure matches the broker's own unacked count.
• Not caused by concurrent handlers. The handler above settles inline on the delivery loop thread.

Start with _replenish_credit and unpause in consumer.py, then trace ReceiverLink.flow in link.py and the delivery-count/link-credit interaction described in the report. Use the supplied reproduction against RabbitMQ 4.x and add or run coverage for settled and never-settled consumers. Done means unsettled deliveries never exceed initial_credits for every tested credit value.