High-Load Infrastructure

FlowController (backpressure)

FlowController

Publish-side flow control.

Usage::

fc = FlowController(BackpressureConfig(max_in_flight=500, rate_limit=1000))

# Before publish:
if fc.acquire(timeout=5.0):
    transport.publish(envelope)
    fc.release()
else:
    # message dropped or error raised (depends on on_blocked)
    ...

# Register with transport callbacks:
transport.on_blocked(fc.on_blocked)
transport.on_unblocked(fc.on_unblocked)

Source code in src/rabbitkit/highload/backpressure.py

class FlowController:
    """Publish-side flow control.

    Usage::

        fc = FlowController(BackpressureConfig(max_in_flight=500, rate_limit=1000))

        # Before publish:
        if fc.acquire(timeout=5.0):
            transport.publish(envelope)
            fc.release()
        else:
            # message dropped or error raised (depends on on_blocked)
            ...

        # Register with transport callbacks:
        transport.on_blocked(fc.on_blocked)
        transport.on_unblocked(fc.on_unblocked)
    """

    def __init__(self, config: BackpressureConfig | None = None) -> None:
        self._config = config or BackpressureConfig()
        self._blocked = False
        self._in_flight = 0
        self._lock = threading.Lock()
        self._unblock_event = threading.Event()
        self._unblock_event.set()  # start unblocked
        self._slot_event = threading.Event()
        self._slot_event.set()  # start with slots available

        # Async equivalents
        self._async_lock: asyncio.Lock | None = None  # lazily created
        self._async_unblock_event: asyncio.Event | None = None  # lazily created
        self._async_slot_event: asyncio.Event | None = None  # lazily created

        # Rate limiter (optional)
        # Sync path uses threading.Lock-based bucket; async path uses asyncio.Lock
        # so acquiring tokens never blocks the event loop.
        self._rate_limiter: _TokenBucket | None = None
        self._async_rate_limiter: _AsyncTokenBucket | None = None
        if self._config.rate_limit is not None:
            poll = self._config.poll_interval_ms / 1000.0
            self._rate_limiter = _TokenBucket(self._config.rate_limit, poll_interval=poll)
            self._async_rate_limiter = _AsyncTokenBucket(self._config.rate_limit)

        # Select the on_blocked strategy ONCE (public API stays a string for
        # backward compat — the Strategy selection happens here).
        _strategies: dict[str, type[_BlockedPolicy]] = {
            "wait": _WaitPolicy,
            "raise": _RaisePolicy,
            "drop": _DropPolicy,
        }
        self._policy = _strategies[self._config.on_blocked]()

    def _ensure_async_primitives(self) -> None:
        """Lazily create asyncio primitives (must be called in event loop)."""
        if self._async_lock is None:
            self._async_lock = asyncio.Lock()
        if self._async_unblock_event is None:
            self._async_unblock_event = asyncio.Event()
            if not self._blocked:
                self._async_unblock_event.set()
        if self._async_slot_event is None:
            self._async_slot_event = asyncio.Event()
            self._async_slot_event.set()

    # ── Connection blocked/unblocked callbacks ───────────────────────────

    def on_blocked(self) -> None:
        """Called when RabbitMQ signals connection.blocked."""
        self._blocked = True
        self._unblock_event.clear()
        if self._async_unblock_event is not None:
            self._async_unblock_event.clear()
        logger.warning("Connection blocked — backpressure active")

    def on_unblocked(self) -> None:
        """Called when RabbitMQ signals connection.unblocked."""
        self._blocked = False
        self._unblock_event.set()
        if self._async_unblock_event is not None:
            self._async_unblock_event.set()
        logger.info("Connection unblocked — backpressure released")

    # ── Properties ───────────────────────────────────────────────────────

    @property
    def is_blocked(self) -> bool:
        """True if connection is currently blocked by RabbitMQ."""
        return self._blocked

    @property
    def in_flight(self) -> int:
        """Current number of in-flight (unconfirmed) publishes."""
        return self._in_flight

    # ── Sync acquire / release ───────────────────────────────────────────

    def acquire(self, timeout: float | None = None) -> bool:
        """Acquire a publish slot.

        Checks (in order): not blocked, in-flight < max, rate OK.

        Returns True if slot acquired, False if dropped.
        Raises BackpressureError if ``on_blocked == "raise"`` and blocked.

        With ``on_blocked="wait"`` a race loss (the slot we waited for was
        taken by another contender) re-loops instead of silently dropping —
        mirroring the async path (I-9). The loop is bounded by the deadline
        derived from *timeout* / ``blocked_timeout``.
        """
        effective_timeout = timeout if timeout is not None else self._config.blocked_timeout

        # 1. Check blocked state
        if self._blocked:
            if not self._policy.handle(self, _REASON_BLOCKED, effective_timeout):
                return False

        # 2. Check in-flight limit + rate limiter. The rate-limiter wait (which
        # sleeps) must happen OUTSIDE self._lock (C-5): under the lock we only do
        # non-blocking checks, then release the lock before any blocking wait.
        # On a race loss (the slot/token we waited for was taken by another
        # contender) we re-loop instead of dropping (I-9), bounded by deadline.
        deadline = None if effective_timeout is None else time.monotonic() + effective_timeout

        def _remaining() -> float | None:
            return None if deadline is None else max(0.0, deadline - time.monotonic())

        while True:
            rate_needed = False
            at_limit = False
            with self._lock:
                if self._in_flight < self._config.max_in_flight:
                    if self._rate_limiter is not None and not self._rate_limiter.acquire():
                        # No token right now; the (possibly blocking) wait happens
                        # outside the lock below. Fall through to the rate policy.
                        rate_needed = True
                    else:
                        self._in_flight += 1
                        if self._in_flight >= self._config.max_in_flight:
                            self._slot_event.clear()
                        return True
                else:
                    at_limit = True
                    self._slot_event.clear()

            if rate_needed:
                # Policy dispatch outside the lock (C-5): wait/raise/drop.
                if not self._policy.handle(self, _REASON_RATE, _remaining()):
                    return False
                # wait() consumed a token atomically; now claim the in-flight slot.
                with self._lock:
                    if self._in_flight < self._config.max_in_flight:
                        self._in_flight += 1
                        if self._in_flight >= self._config.max_in_flight:
                            self._slot_event.clear()
                        return True
                # The slot was taken while we waited for a token; wait for a slot
                # then re-loop and re-claim under the lock (I-9: was a silent drop).
                if not self._slot_event.wait(timeout=_remaining()):
                    return False
                continue

            if at_limit:
                # Policy dispatch outside the lock (C-5): wait/raise/drop.
                if not self._policy.handle(self, _REASON_SLOT, _remaining()):
                    return False
                # Re-loop and re-claim under the lock. If we lose the race again
                # (still at limit), the loop clears the event and re-waits — this
                # is the I-9 fix (previously we returned False on a single loss).
                continue

    def release(self) -> None:
        """Release a publish slot after confirm/nack/timeout."""
        with self._lock:
            if self._in_flight > 0:
                self._in_flight -= 1
            self._slot_event.set()

    # ── Async acquire / release ──────────────────────────────────────────

    async def acquire_async(self, timeout: float | None = None) -> bool:
        """Async variant of ``acquire``."""
        self._ensure_async_primitives()
        assert self._async_lock is not None
        assert self._async_unblock_event is not None
        effective_timeout = timeout if timeout is not None else self._config.blocked_timeout

        # 1. Check blocked state
        if self._blocked:
            if not await self._policy.handle_async(self, _REASON_BLOCKED, effective_timeout):
                return False

        # 2. Check in-flight limit + rate. Rate-token waits and in-flight slot
        # waits happen OUTSIDE the lock; on a race loss we re-wait (loop) instead
        # of dropping (H-P4). With on_blocked="wait" a missing rate token now
        # actually waits, mirroring the sync semantics.
        deadline = None if effective_timeout is None else time.monotonic() + effective_timeout

        def _remaining() -> float | None:
            return None if deadline is None else max(0.0, deadline - time.monotonic())

        while True:
            rate_needed = False
            at_limit = False
            async with self._async_lock:
                if self._in_flight < self._config.max_in_flight:
                    if self._async_rate_limiter is not None and not await self._async_rate_limiter.acquire():
                        # No token right now; the (possibly blocking) wait happens
                        # outside the lock below. Fall through to the rate policy.
                        rate_needed = True
                    else:
                        self._in_flight += 1
                        if self._in_flight >= self._config.max_in_flight:
                            assert self._async_slot_event is not None
                            self._async_slot_event.clear()
                        return True
                else:
                    at_limit = True
                    assert self._async_slot_event is not None
                    self._async_slot_event.clear()

            if rate_needed:
                # Policy dispatch outside the lock (C-5): wait/raise/drop.
                if not await self._policy.handle_async(self, _REASON_RATE, _remaining()):
                    return False
                # Token consumed by wait(); claim the slot without another acquire.
                async with self._async_lock:
                    if self._in_flight < self._config.max_in_flight:
                        self._in_flight += 1
                        if self._in_flight >= self._config.max_in_flight:
                            assert self._async_slot_event is not None
                            self._async_slot_event.clear()
                        return True
                # The slot was taken while we waited for a token; wait for a slot
                # then RE-LOOP and re-claim under the lock, exactly like the sync
                # path (I-9/perf-M-1). The re-loop re-checks the rate bucket
                # non-blockingly and, if it's empty again (another contender took
                # the refill), re-enters the bounded _REASON_RATE wait — it must
                # NOT drop. A previous version demanded a second token HERE, under
                # the lock, and dropped (_REASON_RATE_RETRY) when the bucket was
                # momentarily empty — silently failing a waiter with almost its
                # whole deadline remaining, the exact single-race-loss drop this
                # loop exists to prevent (caught by the contender stress test
                # failing ~1% of runs under CPU load, in ~70ms of a 10s budget).
                assert self._async_slot_event is not None
                _rem = _remaining()
                if _rem is not None and _rem <= 0:
                    return False
                try:
                    async with asyncio.timeout(_rem):
                        await self._async_slot_event.wait()
                except TimeoutError:
                    return False
                continue

            if at_limit:
                # Policy dispatch outside the lock (C-5): wait/raise/drop.
                assert self._async_slot_event is not None
                if not await self._policy.handle_async(self, _REASON_SLOT, _remaining()):
                    return False
                # Re-loop and re-claim under the lock. If we lose the race again
                # (still at limit), the loop clears the event and re-waits - this
                # is the H-P4 fix (previously we returned False on a single loss).
                continue

    async def release_async(self) -> None:
        """Async variant of ``release``."""
        self._ensure_async_primitives()
        assert self._async_lock is not None
        assert self._async_slot_event is not None
        async with self._async_lock:
            if self._in_flight > 0:
                self._in_flight -= 1
            self._async_slot_event.set()

Attributes

is_blocked: bool property

True if connection is currently blocked by RabbitMQ.

in_flight: int property

Current number of in-flight (unconfirmed) publishes.

Methods:

on_blocked() -> None

Called when RabbitMQ signals connection.blocked.

Source code in src/rabbitkit/highload/backpressure.py

def on_blocked(self) -> None:
    """Called when RabbitMQ signals connection.blocked."""
    self._blocked = True
    self._unblock_event.clear()
    if self._async_unblock_event is not None:
        self._async_unblock_event.clear()
    logger.warning("Connection blocked — backpressure active")

on_unblocked() -> None

Called when RabbitMQ signals connection.unblocked.

Source code in src/rabbitkit/highload/backpressure.py

def on_unblocked(self) -> None:
    """Called when RabbitMQ signals connection.unblocked."""
    self._blocked = False
    self._unblock_event.set()
    if self._async_unblock_event is not None:
        self._async_unblock_event.set()
    logger.info("Connection unblocked — backpressure released")

acquire(timeout: float | None = None) -> bool

Acquire a publish slot.

Checks (in order): not blocked, in-flight < max, rate OK.

Returns True if slot acquired, False if dropped. Raises BackpressureError if on_blocked == "raise" and blocked.

With on_blocked="wait" a race loss (the slot we waited for was taken by another contender) re-loops instead of silently dropping — mirroring the async path (I-9). The loop is bounded by the deadline derived from timeout / blocked_timeout.

Source code in src/rabbitkit/highload/backpressure.py

def acquire(self, timeout: float | None = None) -> bool:
    """Acquire a publish slot.

    Checks (in order): not blocked, in-flight < max, rate OK.

    Returns True if slot acquired, False if dropped.
    Raises BackpressureError if ``on_blocked == "raise"`` and blocked.

    With ``on_blocked="wait"`` a race loss (the slot we waited for was
    taken by another contender) re-loops instead of silently dropping —
    mirroring the async path (I-9). The loop is bounded by the deadline
    derived from *timeout* / ``blocked_timeout``.
    """
    effective_timeout = timeout if timeout is not None else self._config.blocked_timeout

    # 1. Check blocked state
    if self._blocked:
        if not self._policy.handle(self, _REASON_BLOCKED, effective_timeout):
            return False

    # 2. Check in-flight limit + rate limiter. The rate-limiter wait (which
    # sleeps) must happen OUTSIDE self._lock (C-5): under the lock we only do
    # non-blocking checks, then release the lock before any blocking wait.
    # On a race loss (the slot/token we waited for was taken by another
    # contender) we re-loop instead of dropping (I-9), bounded by deadline.
    deadline = None if effective_timeout is None else time.monotonic() + effective_timeout

    def _remaining() -> float | None:
        return None if deadline is None else max(0.0, deadline - time.monotonic())

    while True:
        rate_needed = False
        at_limit = False
        with self._lock:
            if self._in_flight < self._config.max_in_flight:
                if self._rate_limiter is not None and not self._rate_limiter.acquire():
                    # No token right now; the (possibly blocking) wait happens
                    # outside the lock below. Fall through to the rate policy.
                    rate_needed = True
                else:
                    self._in_flight += 1
                    if self._in_flight >= self._config.max_in_flight:
                        self._slot_event.clear()
                    return True
            else:
                at_limit = True
                self._slot_event.clear()

        if rate_needed:
            # Policy dispatch outside the lock (C-5): wait/raise/drop.
            if not self._policy.handle(self, _REASON_RATE, _remaining()):
                return False
            # wait() consumed a token atomically; now claim the in-flight slot.
            with self._lock:
                if self._in_flight < self._config.max_in_flight:
                    self._in_flight += 1
                    if self._in_flight >= self._config.max_in_flight:
                        self._slot_event.clear()
                    return True
            # The slot was taken while we waited for a token; wait for a slot
            # then re-loop and re-claim under the lock (I-9: was a silent drop).
            if not self._slot_event.wait(timeout=_remaining()):
                return False
            continue

        if at_limit:
            # Policy dispatch outside the lock (C-5): wait/raise/drop.
            if not self._policy.handle(self, _REASON_SLOT, _remaining()):
                return False
            # Re-loop and re-claim under the lock. If we lose the race again
            # (still at limit), the loop clears the event and re-waits — this
            # is the I-9 fix (previously we returned False on a single loss).
            continue

release() -> None

Release a publish slot after confirm/nack/timeout.

Source code in src/rabbitkit/highload/backpressure.py

def release(self) -> None:
    """Release a publish slot after confirm/nack/timeout."""
    with self._lock:
        if self._in_flight > 0:
            self._in_flight -= 1
        self._slot_event.set()

acquire_async(timeout: float | None = None) -> bool async

Async variant of acquire.

Source code in src/rabbitkit/highload/backpressure.py

async def acquire_async(self, timeout: float | None = None) -> bool:
    """Async variant of ``acquire``."""
    self._ensure_async_primitives()
    assert self._async_lock is not None
    assert self._async_unblock_event is not None
    effective_timeout = timeout if timeout is not None else self._config.blocked_timeout

    # 1. Check blocked state
    if self._blocked:
        if not await self._policy.handle_async(self, _REASON_BLOCKED, effective_timeout):
            return False

    # 2. Check in-flight limit + rate. Rate-token waits and in-flight slot
    # waits happen OUTSIDE the lock; on a race loss we re-wait (loop) instead
    # of dropping (H-P4). With on_blocked="wait" a missing rate token now
    # actually waits, mirroring the sync semantics.
    deadline = None if effective_timeout is None else time.monotonic() + effective_timeout

    def _remaining() -> float | None:
        return None if deadline is None else max(0.0, deadline - time.monotonic())

    while True:
        rate_needed = False
        at_limit = False
        async with self._async_lock:
            if self._in_flight < self._config.max_in_flight:
                if self._async_rate_limiter is not None and not await self._async_rate_limiter.acquire():
                    # No token right now; the (possibly blocking) wait happens
                    # outside the lock below. Fall through to the rate policy.
                    rate_needed = True
                else:
                    self._in_flight += 1
                    if self._in_flight >= self._config.max_in_flight:
                        assert self._async_slot_event is not None
                        self._async_slot_event.clear()
                    return True
            else:
                at_limit = True
                assert self._async_slot_event is not None
                self._async_slot_event.clear()

        if rate_needed:
            # Policy dispatch outside the lock (C-5): wait/raise/drop.
            if not await self._policy.handle_async(self, _REASON_RATE, _remaining()):
                return False
            # Token consumed by wait(); claim the slot without another acquire.
            async with self._async_lock:
                if self._in_flight < self._config.max_in_flight:
                    self._in_flight += 1
                    if self._in_flight >= self._config.max_in_flight:
                        assert self._async_slot_event is not None
                        self._async_slot_event.clear()
                    return True
            # The slot was taken while we waited for a token; wait for a slot
            # then RE-LOOP and re-claim under the lock, exactly like the sync
            # path (I-9/perf-M-1). The re-loop re-checks the rate bucket
            # non-blockingly and, if it's empty again (another contender took
            # the refill), re-enters the bounded _REASON_RATE wait — it must
            # NOT drop. A previous version demanded a second token HERE, under
            # the lock, and dropped (_REASON_RATE_RETRY) when the bucket was
            # momentarily empty — silently failing a waiter with almost its
            # whole deadline remaining, the exact single-race-loss drop this
            # loop exists to prevent (caught by the contender stress test
            # failing ~1% of runs under CPU load, in ~70ms of a 10s budget).
            assert self._async_slot_event is not None
            _rem = _remaining()
            if _rem is not None and _rem <= 0:
                return False
            try:
                async with asyncio.timeout(_rem):
                    await self._async_slot_event.wait()
            except TimeoutError:
                return False
            continue

        if at_limit:
            # Policy dispatch outside the lock (C-5): wait/raise/drop.
            assert self._async_slot_event is not None
            if not await self._policy.handle_async(self, _REASON_SLOT, _remaining()):
                return False
            # Re-loop and re-claim under the lock. If we lose the race again
            # (still at limit), the loop clears the event and re-waits - this
            # is the H-P4 fix (previously we returned False on a single loss).
            continue

release_async() -> None async

Async variant of release.

Source code in src/rabbitkit/highload/backpressure.py

async def release_async(self) -> None:
    """Async variant of ``release``."""
    self._ensure_async_primitives()
    assert self._async_lock is not None
    assert self._async_slot_event is not None
    async with self._async_lock:
        if self._in_flight > 0:
            self._in_flight -= 1
        self._async_slot_event.set()

BatchPublisher

BatchPublisher

Buffer outgoing envelopes and flush as a batch.

When flush_interval_ms > 0 (default 50 ms), a background timer fires periodically to flush any buffered envelopes even if batch_size has not been reached. The timer starts lazily on the first call to add() and is cancelled by close() / close_async().

max_in_flight is reserved for future async-confirm support and has no runtime effect in the current synchronous-confirm model.

NOTE (throughput): this is a buffering/timing helper, not wire-level batching. flush publishes each buffered envelope via publish_fn, so if publish_fn awaits a confirm per message the confirms do not pipeline — you get ergonomics, not extra throughput. For high-volume confirmed publishing, pipeline confirms yourself (publish many, then await) or use the transactional outbox; for safety-critical events, always use the outbox.

Usage::

bp = BatchPublisher(
    config=BatchPublishConfig(batch_size=50, flush_interval_ms=100),
    publish_fn=transport.publish,
    confirm_fn=transport.wait_for_confirms,  # optional
)
bp.add(envelope1)
bp.add(envelope2)
...
bp.flush()   # publishes all buffered
bp.close()   # flush remaining + cancel timer

Source code in src/rabbitkit/highload/batch.py

class BatchPublisher:
    """Buffer outgoing envelopes and flush as a batch.

    When ``flush_interval_ms > 0`` (default 50 ms), a background timer
    fires periodically to flush any buffered envelopes even if
    ``batch_size`` has not been reached.  The timer starts lazily on the
    first call to ``add()`` and is cancelled by ``close()`` / ``close_async()``.

    ``max_in_flight`` is reserved for future async-confirm support and has
    no runtime effect in the current synchronous-confirm model.

    NOTE (throughput): this is a *buffering/timing* helper, not wire-level
    batching. ``flush`` publishes each buffered envelope via ``publish_fn``, so
    if ``publish_fn`` awaits a confirm per message the confirms do not pipeline —
    you get ergonomics, not extra throughput. For high-volume confirmed
    publishing, pipeline confirms yourself (publish many, then await) or use the
    transactional outbox; for safety-critical events, always use the outbox.

    Usage::

        bp = BatchPublisher(
            config=BatchPublishConfig(batch_size=50, flush_interval_ms=100),
            publish_fn=transport.publish,
            confirm_fn=transport.wait_for_confirms,  # optional
        )
        bp.add(envelope1)
        bp.add(envelope2)
        ...
        bp.flush()   # publishes all buffered
        bp.close()   # flush remaining + cancel timer
    """

    def __init__(
        self,
        publish_fn: Callable[[MessageEnvelope], Any],
        config: BatchPublishConfig | None = None,
        confirm_fn: Callable[[], Any] | None = None,
    ) -> None:
        self._config = config or BatchPublishConfig()
        self._publish_fn = publish_fn
        self._confirm_fn = confirm_fn
        self._buffer: list[MessageEnvelope] = []
        self._lock = threading.Lock()
        self._timer: threading.Timer | None = None
        self._closed = False
        self._flush_task: asyncio.Task[None] | None = None
        # I-7: an asyncio.Lock guards _buffer/_flush_task mutations in the async
        # path so the background interval loop and add_async/flush_async/close_async
        # cannot race and lose messages. Lazily created inside the event loop.
        self._async_lock: asyncio.Lock | None = None

    @property
    def pending(self) -> int:
        """Number of envelopes buffered but not yet flushed."""
        return len(self._buffer)

    def _ensure_async_lock(self) -> asyncio.Lock:
        if self._async_lock is None:
            self._async_lock = asyncio.Lock()
        return self._async_lock

    # ── Timer helpers ────────────────────────────────────────────────────

    def _schedule_timer(self) -> None:
        # MH-1: defensive — only arm a new timer when none is already running
        # so a concurrent add()/timer-callback cannot double-schedule an orphan
        # timer that fires forever (leaking daemon threads).
        if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
            interval = self._config.flush_interval_ms / 1000.0
            self._timer = threading.Timer(interval, self._timer_callback)
            self._timer.daemon = True
            self._timer.start()

    def _timer_callback(self) -> None:
        # I-7: clear the timer slot under the lock so concurrent flush()/add()
        # see a consistent None and cannot double-schedule.
        with self._lock:
            self._timer = None
        self.flush()
        # MH-1: reschedule UNDER the lock with a None-guard. A concurrent add()
        # that armed a timer during flush() (outside the lock) leaves
        # self._timer set, so we skip rescheduling here instead of starting an
        # orphan timer that fires forever.
        with self._lock:
            if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
                self._schedule_timer()

    # ── Sync API ─────────────────────────────────────────────────────────

    def add(self, envelope: MessageEnvelope) -> None:
        """Add an envelope to the batch buffer.

        Auto-flushes when ``batch_size`` is reached.  Starts the interval
        timer on the first call when ``flush_interval_ms > 0``.
        """
        with self._lock:
            self._buffer.append(envelope)
            should_flush = len(self._buffer) >= self._config.batch_size
            if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
                self._schedule_timer()
        if should_flush:
            self.flush()

    def flush(self) -> int:
        """Publish all buffered envelopes.

        Returns the number of envelopes published.
        """
        with self._lock:
            if not self._buffer:
                return 0
            count = len(self._buffer)
            batch = list(self._buffer)
            self._buffer.clear()

        for envelope in batch:
            self._publish_fn(envelope)

        if self._confirm_fn is not None:
            self._confirm_fn()

        logger.debug("Batch-published %d envelopes", count)
        return count

    def close(self) -> int:
        """Flush remaining envelopes, cancel the interval timer, and clean up.

        Returns the number of envelopes flushed.
        """
        self._closed = True
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None
        # I-7: also cancel a stray async interval task so closing via the sync
        # API does not leave the async loop running.
        if self._flush_task is not None:
            self._flush_task.cancel()
            self._flush_task = None
        return self.flush()

    # ── Async API ────────────────────────────────────────────────────────

    async def _interval_loop_async(self) -> None:
        interval = self._config.flush_interval_ms / 1000.0
        try:
            while True:
                await asyncio.sleep(interval)
                await self.flush_async()
        except asyncio.CancelledError:
            pass

    async def add_async(self, envelope: MessageEnvelope) -> None:
        """Async: add envelope; auto-flush at batch_size.

        Starts the async interval loop on the first call when
        ``flush_interval_ms > 0``.
        """
        # I-7: guard _buffer / _flush_task mutations with the async lock so a
        # concurrent interval-loop flush_async cannot interleave with this add
        # and lose/duplicate buffered envelopes.
        should_flush = False
        async with self._ensure_async_lock():
            self._buffer.append(envelope)
            if len(self._buffer) >= self._config.batch_size:
                should_flush = True
            elif self._config.flush_interval_ms > 0 and self._flush_task is None:
                self._flush_task = asyncio.create_task(self._interval_loop_async())
        if should_flush:
            await self.flush_async()

    async def flush_async(self) -> int:
        """Async: publish all buffered envelopes."""
        # I-7: take the batch snapshot under the async lock so concurrent
        # flush_async / add_async calls cannot both grab the same buffer.
        async with self._ensure_async_lock():
            if not self._buffer:
                return 0
            count = len(self._buffer)
            batch = list(self._buffer)
            self._buffer.clear()

        for envelope in batch:
            result = self._publish_fn(envelope)
            if hasattr(result, "__await__"):
                await result

        if self._confirm_fn is not None:
            result = self._confirm_fn()
            if hasattr(result, "__await__"):
                await result

        logger.debug("Async batch-published %d envelopes", count)
        return count

    async def close_async(self) -> int:
        """Async: cancel the interval loop, flush remaining, and clean up."""
        self._closed = True
        async with self._ensure_async_lock():
            if self._flush_task is not None:
                self._flush_task.cancel()
                self._flush_task = None
        # I-7: also cancel a stray sync timer so closing via the async API does
        # not leave the sync timer firing after shutdown.
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None
        return await self.flush_async()

Attributes

pending: int property

Number of envelopes buffered but not yet flushed.

Methods:

add(envelope: MessageEnvelope) -> None

Add an envelope to the batch buffer.

Auto-flushes when batch_size is reached. Starts the interval timer on the first call when flush_interval_ms > 0.

Source code in src/rabbitkit/highload/batch.py

def add(self, envelope: MessageEnvelope) -> None:
    """Add an envelope to the batch buffer.

    Auto-flushes when ``batch_size`` is reached.  Starts the interval
    timer on the first call when ``flush_interval_ms > 0``.
    """
    with self._lock:
        self._buffer.append(envelope)
        should_flush = len(self._buffer) >= self._config.batch_size
        if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
            self._schedule_timer()
    if should_flush:
        self.flush()

flush() -> int

Publish all buffered envelopes.

Returns the number of envelopes published.

Source code in src/rabbitkit/highload/batch.py

def flush(self) -> int:
    """Publish all buffered envelopes.

    Returns the number of envelopes published.
    """
    with self._lock:
        if not self._buffer:
            return 0
        count = len(self._buffer)
        batch = list(self._buffer)
        self._buffer.clear()

    for envelope in batch:
        self._publish_fn(envelope)

    if self._confirm_fn is not None:
        self._confirm_fn()

    logger.debug("Batch-published %d envelopes", count)
    return count

close() -> int

Flush remaining envelopes, cancel the interval timer, and clean up.

Returns the number of envelopes flushed.

Source code in src/rabbitkit/highload/batch.py

def close(self) -> int:
    """Flush remaining envelopes, cancel the interval timer, and clean up.

    Returns the number of envelopes flushed.
    """
    self._closed = True
    if self._timer is not None:
        self._timer.cancel()
        self._timer = None
    # I-7: also cancel a stray async interval task so closing via the sync
    # API does not leave the async loop running.
    if self._flush_task is not None:
        self._flush_task.cancel()
        self._flush_task = None
    return self.flush()

add_async(envelope: MessageEnvelope) -> None async

Async: add envelope; auto-flush at batch_size.

Starts the async interval loop on the first call when flush_interval_ms > 0.

Source code in src/rabbitkit/highload/batch.py

async def add_async(self, envelope: MessageEnvelope) -> None:
    """Async: add envelope; auto-flush at batch_size.

    Starts the async interval loop on the first call when
    ``flush_interval_ms > 0``.
    """
    # I-7: guard _buffer / _flush_task mutations with the async lock so a
    # concurrent interval-loop flush_async cannot interleave with this add
    # and lose/duplicate buffered envelopes.
    should_flush = False
    async with self._ensure_async_lock():
        self._buffer.append(envelope)
        if len(self._buffer) >= self._config.batch_size:
            should_flush = True
        elif self._config.flush_interval_ms > 0 and self._flush_task is None:
            self._flush_task = asyncio.create_task(self._interval_loop_async())
    if should_flush:
        await self.flush_async()

flush_async() -> int async

Async: publish all buffered envelopes.

Source code in src/rabbitkit/highload/batch.py

async def flush_async(self) -> int:
    """Async: publish all buffered envelopes."""
    # I-7: take the batch snapshot under the async lock so concurrent
    # flush_async / add_async calls cannot both grab the same buffer.
    async with self._ensure_async_lock():
        if not self._buffer:
            return 0
        count = len(self._buffer)
        batch = list(self._buffer)
        self._buffer.clear()

    for envelope in batch:
        result = self._publish_fn(envelope)
        if hasattr(result, "__await__"):
            await result

    if self._confirm_fn is not None:
        result = self._confirm_fn()
        if hasattr(result, "__await__"):
            await result

    logger.debug("Async batch-published %d envelopes", count)
    return count

close_async() -> int async

Async: cancel the interval loop, flush remaining, and clean up.

Source code in src/rabbitkit/highload/batch.py

async def close_async(self) -> int:
    """Async: cancel the interval loop, flush remaining, and clean up."""
    self._closed = True
    async with self._ensure_async_lock():
        if self._flush_task is not None:
            self._flush_task.cancel()
            self._flush_task = None
    # I-7: also cancel a stray sync timer so closing via the async API does
    # not leave the sync timer firing after shutdown.
    if self._timer is not None:
        self._timer.cancel()
        self._timer = None
    return await self.flush_async()

BatchAcker

BatchAcker

Accumulate delivery tags and ack in batches.

Uses multiple=True on the maximum delivery tag in the batch.

When flush_interval_ms > 0 (default 200 ms), a background timer fires periodically to ack any buffered tags even if batch_size has not been reached. The timer starts lazily on the first call to add() and is cancelled by close() / close_async().

Ownership rules: - Channel-scoped — NEVER cross channels - Handlers MUST NOT call msg.ack() when BatchAcker is active - Compatible with AUTO and NACK_ON_ERROR policies only

Usage (sync / pika) — ack_fn MUST NOT be a raw channel.basic_ack. The interval timer fires flush() from a background threading.Timer thread, not pika's connection I/O thread; pika channel methods are not thread-safe, so a direct channel.basic_ack reference here is a real cross-thread violation waiting to happen the first time the timer (rather than add()) triggers the flush. Marshal onto the I/O thread instead, e.g. via connection.add_callback_threadsafe::

def safe_ack(delivery_tag: int, multiple: bool = False) -> None:
    connection.add_callback_threadsafe(
        lambda: channel.basic_ack(delivery_tag=delivery_tag, multiple=multiple)
    )

ba = BatchAcker(
    config=BatchAckConfig(batch_size=50, flush_interval_ms=200),
    ack_fn=safe_ack,
)
ba.add(delivery_tag=1)
ba.add(delivery_tag=2)
...
ba.flush()  # ack(max_tag, multiple=True)
ba.close()  # flush remaining + cancel timer

The async path (add_async/flush_async) has no such hazard — _interval_loop_async schedules via asyncio.create_task on the same event loop the aio-pika channel already runs on, so an aio-pika channel.basic_ack (a coroutine function) can be passed directly as ack_fn.

Source code in src/rabbitkit/highload/batch.py

class BatchAcker:
    """Accumulate delivery tags and ack in batches.

    Uses ``multiple=True`` on the maximum delivery tag in the batch.

    When ``flush_interval_ms > 0`` (default 200 ms), a background timer
    fires periodically to ack any buffered tags even if ``batch_size`` has
    not been reached.  The timer starts lazily on the first call to
    ``add()`` and is cancelled by ``close()`` / ``close_async()``.

    **Ownership rules:**
    - Channel-scoped — NEVER cross channels
    - Handlers MUST NOT call ``msg.ack()`` when BatchAcker is active
    - Compatible with AUTO and NACK_ON_ERROR policies only

    Usage (sync / pika) — ``ack_fn`` MUST NOT be a raw ``channel.basic_ack``.
    The interval timer fires ``flush()`` from a background
    ``threading.Timer`` thread, not pika's connection I/O thread; pika
    channel methods are not thread-safe, so a direct ``channel.basic_ack``
    reference here is a real cross-thread violation waiting to happen the
    first time the timer (rather than ``add()``) triggers the flush.
    Marshal onto the I/O thread instead, e.g. via
    ``connection.add_callback_threadsafe``::

        def safe_ack(delivery_tag: int, multiple: bool = False) -> None:
            connection.add_callback_threadsafe(
                lambda: channel.basic_ack(delivery_tag=delivery_tag, multiple=multiple)
            )

        ba = BatchAcker(
            config=BatchAckConfig(batch_size=50, flush_interval_ms=200),
            ack_fn=safe_ack,
        )
        ba.add(delivery_tag=1)
        ba.add(delivery_tag=2)
        ...
        ba.flush()  # ack(max_tag, multiple=True)
        ba.close()  # flush remaining + cancel timer

    The async path (``add_async``/``flush_async``) has no such hazard —
    ``_interval_loop_async`` schedules via ``asyncio.create_task`` on the
    same event loop the aio-pika channel already runs on, so an aio-pika
    ``channel.basic_ack`` (a coroutine function) can be passed directly as
    ``ack_fn``.
    """

    def __init__(
        self,
        ack_fn: Callable[..., Any],
        config: BatchAckConfig | None = None,
    ) -> None:
        self._config = config or BatchAckConfig()
        self._ack_fn = ack_fn
        self._tags: list[int] = []
        self._max_tag: int = 0  # O(1) max tracking (perf)
        self._lock = threading.Lock()
        self._timer: threading.Timer | None = None
        self._closed = False
        self._flush_task: asyncio.Task[None] | None = None
        # I-7: asyncio.Lock guarding _tags/_flush_task in the async path.
        # Lazily created inside the event loop.
        self._async_lock: asyncio.Lock | None = None

    @property
    def pending(self) -> int:
        """Number of delivery tags buffered."""
        return len(self._tags)

    def _ensure_async_lock(self) -> asyncio.Lock:
        if self._async_lock is None:
            self._async_lock = asyncio.Lock()
        return self._async_lock

    # ── Timer helpers ────────────────────────────────────────────────────

    def _schedule_timer(self) -> None:
        # MH-1: defensive — only arm a new timer when none is already running
        # so a concurrent add()/timer-callback cannot double-schedule an orphan
        # timer that fires forever (leaking daemon threads).
        if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
            interval = self._config.flush_interval_ms / 1000.0
            self._timer = threading.Timer(interval, self._timer_callback)
            self._timer.daemon = True
            self._timer.start()

    def _timer_callback(self) -> None:
        # I-7: clear the timer slot under the lock for consistency.
        with self._lock:
            self._timer = None
        self.flush()
        # MH-1: reschedule UNDER the lock with a None-guard so a concurrent
        # add() that armed a timer during flush() doesn't leave us starting an
        # orphan timer that fires forever.
        with self._lock:
            if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
                self._schedule_timer()

    # ── Sync API ─────────────────────────────────────────────────────────

    def add(self, delivery_tag: int) -> None:
        """Add a delivery tag to the batch.

        Auto-flushes when ``batch_size`` is reached.  Starts the interval
        timer on the first call when ``flush_interval_ms > 0``.
        """
        with self._lock:
            self._tags.append(delivery_tag)
            if delivery_tag > self._max_tag:
                self._max_tag = delivery_tag
            should_flush = len(self._tags) >= self._config.batch_size
            if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
                self._schedule_timer()
        if should_flush:
            self.flush()

    def flush(self) -> int:
        """Ack all buffered tags using the max tag with multiple=True.

        Returns the number of tags acked.
        """
        with self._lock:
            if not self._tags:
                return 0
            count = len(self._tags)
            max_tag = max(self._tags)
            self._tags.clear()

        self._ack_fn(max_tag, multiple=True)
        logger.debug("Batch-acked %d messages (max_tag=%d)", count, max_tag)
        return count

    def close(self) -> int:
        """Flush remaining tags, cancel the interval timer, and clean up."""
        self._closed = True
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None
        # I-7: also cancel a stray async interval task.
        if self._flush_task is not None:
            self._flush_task.cancel()
            self._flush_task = None
        return self.flush()

    # ── Async API ────────────────────────────────────────────────────────

    async def _interval_loop_async(self) -> None:
        interval = self._config.flush_interval_ms / 1000.0
        try:
            while True:
                await asyncio.sleep(interval)
                await self.flush_async()
        except asyncio.CancelledError:
            pass

    async def add_async(self, delivery_tag: int) -> None:
        """Async: add tag; auto-flush at batch_size.

        Starts the async interval loop on the first call when
        ``flush_interval_ms > 0``.
        """
        # I-7: guard _tags / _flush_task mutations with the async lock.
        should_flush = False
        async with self._ensure_async_lock():
            self._tags.append(delivery_tag)
            if len(self._tags) >= self._config.batch_size:
                should_flush = True
            elif self._config.flush_interval_ms > 0 and self._flush_task is None:
                self._flush_task = asyncio.create_task(self._interval_loop_async())
        if should_flush:
            await self.flush_async()

    async def flush_async(self) -> int:
        """Async: ack all buffered tags."""
        # I-7: take the batch snapshot under the async lock.
        async with self._ensure_async_lock():
            if not self._tags:
                return 0
            count = len(self._tags)
            max_tag = max(self._tags)
            self._tags.clear()

        result = self._ack_fn(max_tag, multiple=True)
        if hasattr(result, "__await__"):
            await result

        logger.debug("Async batch-acked %d messages (max_tag=%d)", count, max_tag)
        return count

    async def close_async(self) -> int:
        """Async: cancel the interval loop, flush remaining, and clean up."""
        self._closed = True
        async with self._ensure_async_lock():
            if self._flush_task is not None:
                self._flush_task.cancel()
                self._flush_task = None
        # I-7: also cancel a stray sync timer.
        if self._timer is not None:
            self._timer.cancel()
            self._timer = None
        return await self.flush_async()

Attributes

pending: int property

Number of delivery tags buffered.

Methods:

add(delivery_tag: int) -> None

Add a delivery tag to the batch.

Auto-flushes when batch_size is reached. Starts the interval timer on the first call when flush_interval_ms > 0.

Source code in src/rabbitkit/highload/batch.py

def add(self, delivery_tag: int) -> None:
    """Add a delivery tag to the batch.

    Auto-flushes when ``batch_size`` is reached.  Starts the interval
    timer on the first call when ``flush_interval_ms > 0``.
    """
    with self._lock:
        self._tags.append(delivery_tag)
        if delivery_tag > self._max_tag:
            self._max_tag = delivery_tag
        should_flush = len(self._tags) >= self._config.batch_size
        if self._timer is None and self._config.flush_interval_ms > 0 and not self._closed:
            self._schedule_timer()
    if should_flush:
        self.flush()

flush() -> int

Ack all buffered tags using the max tag with multiple=True.

Returns the number of tags acked.

Source code in src/rabbitkit/highload/batch.py

def flush(self) -> int:
    """Ack all buffered tags using the max tag with multiple=True.

    Returns the number of tags acked.
    """
    with self._lock:
        if not self._tags:
            return 0
        count = len(self._tags)
        max_tag = max(self._tags)
        self._tags.clear()

    self._ack_fn(max_tag, multiple=True)
    logger.debug("Batch-acked %d messages (max_tag=%d)", count, max_tag)
    return count

close() -> int

Flush remaining tags, cancel the interval timer, and clean up.

Source code in src/rabbitkit/highload/batch.py

def close(self) -> int:
    """Flush remaining tags, cancel the interval timer, and clean up."""
    self._closed = True
    if self._timer is not None:
        self._timer.cancel()
        self._timer = None
    # I-7: also cancel a stray async interval task.
    if self._flush_task is not None:
        self._flush_task.cancel()
        self._flush_task = None
    return self.flush()

add_async(delivery_tag: int) -> None async

Async: add tag; auto-flush at batch_size.

Starts the async interval loop on the first call when flush_interval_ms > 0.

Source code in src/rabbitkit/highload/batch.py

async def add_async(self, delivery_tag: int) -> None:
    """Async: add tag; auto-flush at batch_size.

    Starts the async interval loop on the first call when
    ``flush_interval_ms > 0``.
    """
    # I-7: guard _tags / _flush_task mutations with the async lock.
    should_flush = False
    async with self._ensure_async_lock():
        self._tags.append(delivery_tag)
        if len(self._tags) >= self._config.batch_size:
            should_flush = True
        elif self._config.flush_interval_ms > 0 and self._flush_task is None:
            self._flush_task = asyncio.create_task(self._interval_loop_async())
    if should_flush:
        await self.flush_async()

flush_async() -> int async

Async: ack all buffered tags.

Source code in src/rabbitkit/highload/batch.py

async def flush_async(self) -> int:
    """Async: ack all buffered tags."""
    # I-7: take the batch snapshot under the async lock.
    async with self._ensure_async_lock():
        if not self._tags:
            return 0
        count = len(self._tags)
        max_tag = max(self._tags)
        self._tags.clear()

    result = self._ack_fn(max_tag, multiple=True)
    if hasattr(result, "__await__"):
        await result

    logger.debug("Async batch-acked %d messages (max_tag=%d)", count, max_tag)
    return count

close_async() -> int async

Async: cancel the interval loop, flush remaining, and clean up.

Source code in src/rabbitkit/highload/batch.py

async def close_async(self) -> int:
    """Async: cancel the interval loop, flush remaining, and clean up."""
    self._closed = True
    async with self._ensure_async_lock():
        if self._flush_task is not None:
            self._flush_task.cancel()
            self._flush_task = None
    # I-7: also cancel a stray sync timer.
    if self._timer is not None:
        self._timer.cancel()
        self._timer = None
    return await self.flush_async()

Worker Pools

SyncWorkerPool

Thread pool for concurrent sync message processing.

Wraps a handler callback to execute it in a thread pool with limited concurrency.

Usage::

pool = SyncWorkerPool(config=WorkerConfig(worker_count=4))
pool.start()
# Use pool.submit(callback, message) instead of callback(message)
pool.stop()

Source code in src/rabbitkit/concurrency.py

class SyncWorkerPool:
    """Thread pool for concurrent sync message processing.

    Wraps a handler callback to execute it in a thread pool with
    limited concurrency.

    Usage::

        pool = SyncWorkerPool(config=WorkerConfig(worker_count=4))
        pool.start()
        # Use pool.submit(callback, message) instead of callback(message)
        pool.stop()
    """

    def __init__(self, config: WorkerConfig | None = None) -> None:
        self._config = config or WorkerConfig()
        self._executor: _DaemonWorkerPool | None = None
        self._futures: set[concurrent.futures.Future[Any]] = set()
        # H12: message associated with each in-flight future, so a future
        # abandoned at the stop_timeout deadline can be logged by delivery
        # tag/message id instead of just a bare count.
        self._future_messages: dict[concurrent.futures.Future[Any], RabbitMessage] = {}
        self._futures_lock = threading.Lock()

    @property
    def worker_count(self) -> int:
        """Return the configured worker count."""
        return self._config.worker_count

    def start(self) -> None:
        """Start the worker pool.

        Uses daemon worker threads (see :class:`_DaemonWorkerPool`) so a stuck
        handler cannot keep the process alive past ``stop()`` — important for
        k8s graceful shutdown where ``terminationGracePeriodSeconds`` must be
        honored without relying on SIGKILL.
        """
        if self._config.worker_count <= 1:
            return  # No pool needed for single worker
        self._executor = _DaemonWorkerPool(
            max_workers=self._config.worker_count,
            thread_name_prefix="rabbitkit-worker",
            max_queue_size=self._config.max_queue_size,
        )
        logger.info("SyncWorkerPool started with %d workers", self._config.worker_count)

    def stop(self, timeout: float | None = None, pump: Callable[[], None] | None = None) -> None:
        """Stop the worker pool, waiting for in-flight tasks.

        Cancels pending (not-yet-started) futures and bounds the wait for
        running ones by ``timeout`` (default ``WorkerConfig.stop_timeout``).
        Because workers are daemon threads, any task that does not finish in
        time is abandoned and the process can still exit cleanly — SIGKILL is
        no longer required as a backstop.

        H2: when *pump* is given (``SyncBroker.stop()`` passes
        ``SyncTransport.pump``), the wait is polled in short slices with
        *pump* called between them, instead of one blocking
        ``concurrent.futures.wait(timeout=effective)``. A worker thread
        finishing its handler acks/nacks by marshaling onto the transport's
        owner thread via ``_run_on_io_thread`` — without something draining
        the connection's I/O loop while this method blocks, those marshaled
        callbacks would never run (they'd eventually time out on the worker
        side, or — before this fix — the transport fell back to an unsafe
        inline cross-thread call). *pump* must be safe to call from whichever
        thread calls ``stop()`` (i.e. ``stop()`` must run on the transport's
        owner thread when *pump* is given). Without *pump*, behavior is
        unchanged: a single blocking wait for the full *effective* timeout.
        """
        if self._executor is None:
            return
        effective = timeout if timeout is not None else self._config.stop_timeout
        with self._futures_lock:
            futures_snapshot = list(self._futures)
        if pump is None or not futures_snapshot:
            _done, not_done = concurrent.futures.wait(futures_snapshot, timeout=effective)
        else:
            poll = 0.05
            deadline = time.monotonic() + effective
            not_done = set(futures_snapshot)
            while not_done:
                pump()
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    break
                _done, not_done = concurrent.futures.wait(not_done, timeout=min(poll, remaining))
            pump()  # final drain in case the last ack was just marshaled
        if not_done:
            # H12: identify WHICH deliveries were abandoned (delivery_tag /
            # message_id), not just a count — the handler thread keeps
            # running (daemon, not killed) and may still ack/nack/produce
            # side effects later, so operators need to be able to correlate
            # an abandoned delivery with what shows up (or doesn't) downstream.
            with self._futures_lock:
                abandoned = [self._future_messages.get(f) for f in not_done]
            for message in abandoned:
                if message is None:
                    # Not tracked — e.g. a future submitted directly via the
                    # underlying executor rather than through submit().
                    continue
                logger.warning(
                    "SyncWorkerPool: handler for delivery_tag=%s message_id=%s did not "
                    "complete within stop_timeout; abandoning (its daemon thread keeps "
                    "running — ensure handlers are idempotent under at-least-once delivery)",
                    message.delivery_tag,
                    message.message_id,
                )
            logger.warning(
                "SyncWorkerPool: %d tasks did not complete within timeout; "
                "abandoning (daemon threads will not block process exit)",
                len(not_done),
            )
        # cancel_futures=True abandons queued-but-unstarted work; daemon threads
        # ensure the process exits even if a running handler is wedged.
        self._executor.shutdown(wait=False, cancel_futures=True)
        self._executor = None
        with self._futures_lock:
            self._futures.clear()
            self._future_messages.clear()
        logger.info("SyncWorkerPool stopped")

    def submit(
        self,
        callback: Callable[[RabbitMessage], None],
        message: RabbitMessage,
    ) -> None:
        """Submit a message for processing.

        If worker_count=1 (default), runs synchronously in the current thread.
        Otherwise, submits to the thread pool.
        """
        if self._executor is None:
            # Single worker mode -- run directly
            callback(message)
            return

        future = self._executor.submit(callback, message)
        with self._futures_lock:
            self._futures.add(future)
            self._future_messages[future] = message
        # Self-cleaning: O(1) add + discard instead of rebuilding the list
        # on every message. Callback fires on the worker thread (or inline if
        # already done) and removes the future under the lock.
        future.add_done_callback(self._discard_future)

    def _discard_future(self, future: concurrent.futures.Future[Any]) -> None:
        with self._futures_lock:
            self._futures.discard(future)
            self._future_messages.pop(future, None)

    @property
    def pending_count(self) -> int:
        """Number of tasks currently pending/running."""
        if self._executor is None:
            return 0
        with self._futures_lock:
            return len(self._futures)

Attributes

worker_count: int property

Return the configured worker count.

pending_count: int property

Number of tasks currently pending/running.

Methods:

start() -> None

Start the worker pool.

Uses daemon worker threads (see :class:_DaemonWorkerPool) so a stuck handler cannot keep the process alive past stop() — important for k8s graceful shutdown where terminationGracePeriodSeconds must be honored without relying on SIGKILL.

Source code in src/rabbitkit/concurrency.py

def start(self) -> None:
    """Start the worker pool.

    Uses daemon worker threads (see :class:`_DaemonWorkerPool`) so a stuck
    handler cannot keep the process alive past ``stop()`` — important for
    k8s graceful shutdown where ``terminationGracePeriodSeconds`` must be
    honored without relying on SIGKILL.
    """
    if self._config.worker_count <= 1:
        return  # No pool needed for single worker
    self._executor = _DaemonWorkerPool(
        max_workers=self._config.worker_count,
        thread_name_prefix="rabbitkit-worker",
        max_queue_size=self._config.max_queue_size,
    )
    logger.info("SyncWorkerPool started with %d workers", self._config.worker_count)

stop(timeout: float | None = None, pump: Callable[[], None] | None = None) -> None

Stop the worker pool, waiting for in-flight tasks.

Cancels pending (not-yet-started) futures and bounds the wait for running ones by timeout (default WorkerConfig.stop_timeout). Because workers are daemon threads, any task that does not finish in time is abandoned and the process can still exit cleanly — SIGKILL is no longer required as a backstop.

H2: when pump is given (SyncBroker.stop() passes SyncTransport.pump), the wait is polled in short slices with pump called between them, instead of one blocking concurrent.futures.wait(timeout=effective). A worker thread finishing its handler acks/nacks by marshaling onto the transport's owner thread via _run_on_io_thread — without something draining the connection's I/O loop while this method blocks, those marshaled callbacks would never run (they'd eventually time out on the worker side, or — before this fix — the transport fell back to an unsafe inline cross-thread call). pump must be safe to call from whichever thread calls stop() (i.e. stop() must run on the transport's owner thread when pump is given). Without pump, behavior is unchanged: a single blocking wait for the full effective timeout.

Source code in src/rabbitkit/concurrency.py

def stop(self, timeout: float | None = None, pump: Callable[[], None] | None = None) -> None:
    """Stop the worker pool, waiting for in-flight tasks.

    Cancels pending (not-yet-started) futures and bounds the wait for
    running ones by ``timeout`` (default ``WorkerConfig.stop_timeout``).
    Because workers are daemon threads, any task that does not finish in
    time is abandoned and the process can still exit cleanly — SIGKILL is
    no longer required as a backstop.

    H2: when *pump* is given (``SyncBroker.stop()`` passes
    ``SyncTransport.pump``), the wait is polled in short slices with
    *pump* called between them, instead of one blocking
    ``concurrent.futures.wait(timeout=effective)``. A worker thread
    finishing its handler acks/nacks by marshaling onto the transport's
    owner thread via ``_run_on_io_thread`` — without something draining
    the connection's I/O loop while this method blocks, those marshaled
    callbacks would never run (they'd eventually time out on the worker
    side, or — before this fix — the transport fell back to an unsafe
    inline cross-thread call). *pump* must be safe to call from whichever
    thread calls ``stop()`` (i.e. ``stop()`` must run on the transport's
    owner thread when *pump* is given). Without *pump*, behavior is
    unchanged: a single blocking wait for the full *effective* timeout.
    """
    if self._executor is None:
        return
    effective = timeout if timeout is not None else self._config.stop_timeout
    with self._futures_lock:
        futures_snapshot = list(self._futures)
    if pump is None or not futures_snapshot:
        _done, not_done = concurrent.futures.wait(futures_snapshot, timeout=effective)
    else:
        poll = 0.05
        deadline = time.monotonic() + effective
        not_done = set(futures_snapshot)
        while not_done:
            pump()
            remaining = deadline - time.monotonic()
            if remaining <= 0:
                break
            _done, not_done = concurrent.futures.wait(not_done, timeout=min(poll, remaining))
        pump()  # final drain in case the last ack was just marshaled
    if not_done:
        # H12: identify WHICH deliveries were abandoned (delivery_tag /
        # message_id), not just a count — the handler thread keeps
        # running (daemon, not killed) and may still ack/nack/produce
        # side effects later, so operators need to be able to correlate
        # an abandoned delivery with what shows up (or doesn't) downstream.
        with self._futures_lock:
            abandoned = [self._future_messages.get(f) for f in not_done]
        for message in abandoned:
            if message is None:
                # Not tracked — e.g. a future submitted directly via the
                # underlying executor rather than through submit().
                continue
            logger.warning(
                "SyncWorkerPool: handler for delivery_tag=%s message_id=%s did not "
                "complete within stop_timeout; abandoning (its daemon thread keeps "
                "running — ensure handlers are idempotent under at-least-once delivery)",
                message.delivery_tag,
                message.message_id,
            )
        logger.warning(
            "SyncWorkerPool: %d tasks did not complete within timeout; "
            "abandoning (daemon threads will not block process exit)",
            len(not_done),
        )
    # cancel_futures=True abandons queued-but-unstarted work; daemon threads
    # ensure the process exits even if a running handler is wedged.
    self._executor.shutdown(wait=False, cancel_futures=True)
    self._executor = None
    with self._futures_lock:
        self._futures.clear()
        self._future_messages.clear()
    logger.info("SyncWorkerPool stopped")

submit(callback: Callable[[RabbitMessage], None], message: RabbitMessage) -> None

Submit a message for processing.

If worker_count=1 (default), runs synchronously in the current thread. Otherwise, submits to the thread pool.

Source code in src/rabbitkit/concurrency.py

def submit(
    self,
    callback: Callable[[RabbitMessage], None],
    message: RabbitMessage,
) -> None:
    """Submit a message for processing.

    If worker_count=1 (default), runs synchronously in the current thread.
    Otherwise, submits to the thread pool.
    """
    if self._executor is None:
        # Single worker mode -- run directly
        callback(message)
        return

    future = self._executor.submit(callback, message)
    with self._futures_lock:
        self._futures.add(future)
        self._future_messages[future] = message
    # Self-cleaning: O(1) add + discard instead of rebuilding the list
    # on every message. Callback fires on the worker thread (or inline if
    # already done) and removes the future under the lock.
    future.add_done_callback(self._discard_future)

AsyncWorkerPool

Semaphore-based concurrent async message processing.

Limits the number of concurrently processing async handlers.

Usage::

pool = AsyncWorkerPool(config=WorkerConfig(worker_count=4))
pool.start()
await pool.submit(callback, message)
await pool.stop()

Source code in src/rabbitkit/concurrency.py

class AsyncWorkerPool:
    """Semaphore-based concurrent async message processing.

    Limits the number of concurrently processing async handlers.

    Usage::

        pool = AsyncWorkerPool(config=WorkerConfig(worker_count=4))
        pool.start()
        await pool.submit(callback, message)
        await pool.stop()
    """

    def __init__(self, config: WorkerConfig | None = None) -> None:
        self._config = config or WorkerConfig()
        self._semaphore: asyncio.Semaphore | None = None
        self._tasks: set[asyncio.Task[None]] = set()
        # H12: message behind each task, so a task abandoned at the
        # stop_timeout deadline can be nacked (redelivery) and logged by
        # delivery tag/message id instead of just cancelled and forgotten.
        self._task_messages: dict[asyncio.Task[None], RabbitMessage] = {}
        self._running = False

    @property
    def worker_count(self) -> int:
        """Return the configured worker count."""
        return self._config.worker_count

    def start(self) -> None:
        """Start the worker pool."""
        if self._config.worker_count <= 1:
            return
        self._semaphore = asyncio.Semaphore(self._config.worker_count)
        self._running = True
        logger.info("AsyncWorkerPool started with %d workers", self._config.worker_count)

    async def stop(self, timeout: float | None = None) -> None:
        """Stop the worker pool, waiting for in-flight tasks.

        R-TaskGroup: rather than ``asyncio.wait`` + a manual ``task.cancel()``
        loop, we ``gather(*tasks, return_exceptions=True)`` bounded by
        ``asyncio.timeout`` (the 3.11+ idiom). Tasks that don't finish before
        the deadline are cancelled and awaited once more so their
        ``CancelledError`` is consumed rather than leaking as "Task was
        destroyed but it is pending" warnings.

        H12: a task cancelled by the deadline is *not* guaranteed to have
        reached its own ``ack``/``nack`` — ``CancelledError`` is a
        ``BaseException``, so it skips right past the pipeline's
        ``except Exception`` handling and the message is left unsettled.
        Rather than leave that to the implicit requeue-on-unacked-channel-close
        behavior when the transport eventually disconnects, any message still
        unsettled after a task is abandoned is nacked (requeue) here and
        logged by delivery tag — an explicit, immediate, observable
        redelivery instead of an implicit one.
        """
        self._running = False
        if not self._tasks:
            return
        effective = timeout if timeout is not None else self._config.stop_timeout
        tasks = list(self._tasks)
        task_messages = dict(self._task_messages)
        try:
            async with asyncio.timeout(effective):
                await asyncio.gather(*tasks, return_exceptions=True)
        except TimeoutError:
            # Deadline elapsed: cancel any still-running tasks and drain them
            # so cancellation propagates cleanly (no pending-task warnings).
            # In practice `asyncio.timeout` firing cancels this method's own
            # await of gather(), and gather's cancellation handling already
            # cancels + drains every task before re-raising -- so `not_done`
            # is normally empty by the time we get here; this loop is a
            # defensive backstop, not the primary abandonment path.
            not_done = [t for t in tasks if not t.done()]
            for task in not_done:  # pragma: no cover — gather already drains before raising
                task.cancel()  # pragma: no cover
            if not_done:  # pragma: no cover
                await asyncio.gather(*not_done, return_exceptions=True)  # pragma: no cover
            for message in task_messages.values():
                if message.is_settled:
                    continue  # handler reached its own ack/nack before being cut off
                logger.warning(
                    "AsyncWorkerPool: handler for delivery_tag=%s message_id=%s did not "
                    "complete within stop_timeout; abandoning (task cancelled) and nacking "
                    "for redelivery — ensure handlers are idempotent under at-least-once delivery",
                    message.delivery_tag,
                    message.message_id,
                )
                try:
                    await message.nack_async(requeue=True)
                except Exception:
                    logger.warning(
                        "nack on abandoned handler's message raised", exc_info=True
                    )
        finally:
            self._tasks.clear()
            self._task_messages.clear()
        logger.info("AsyncWorkerPool stopped")

    async def submit(
        self,
        callback: Callable[[RabbitMessage], Awaitable[None]],
        message: RabbitMessage,
    ) -> None:
        """Submit a message for concurrent processing.

        If worker_count=1, runs directly (no semaphore).
        Otherwise, acquires semaphore slot before executing.

        H12: refuses (nacks for redelivery) instead of scheduling an unawaited
        task when the pool isn't running — e.g. a delivery callback firing
        after ``stop()`` already cleared ``_tasks``. Nothing would ever await
        that task, so it would silently race the event loop's own shutdown
        instead of the message being cleanly settled.
        """
        if self._semaphore is None:
            await callback(message)
            return

        if not self._running:
            logger.warning(
                "AsyncWorkerPool.submit() called while stopped (delivery_tag=%s, "
                "message_id=%s); nacking for redelivery instead of orphaning an "
                "unawaited task",
                message.delivery_tag,
                message.message_id,
            )
            if not message.is_settled:
                await message.nack_async(requeue=True)
            return

        semaphore = self._semaphore

        async def _run() -> None:
            async with semaphore:
                await callback(message)

        task = asyncio.create_task(_run())
        self._tasks.add(task)
        self._task_messages[task] = message
        task.add_done_callback(self._on_task_done)

    def _on_task_done(self, task: asyncio.Task[None]) -> None:
        self._tasks.discard(task)
        self._task_messages.pop(task, None)

    @property
    def pending_count(self) -> int:
        """Number of tasks currently pending/running."""
        return len(self._tasks)

Attributes

worker_count: int property

Return the configured worker count.

pending_count: int property

Number of tasks currently pending/running.

Methods:

start() -> None

Start the worker pool.

Source code in src/rabbitkit/concurrency.py

def start(self) -> None:
    """Start the worker pool."""
    if self._config.worker_count <= 1:
        return
    self._semaphore = asyncio.Semaphore(self._config.worker_count)
    self._running = True
    logger.info("AsyncWorkerPool started with %d workers", self._config.worker_count)

stop(timeout: float | None = None) -> None async

Stop the worker pool, waiting for in-flight tasks.

R-TaskGroup: rather than asyncio.wait + a manual task.cancel() loop, we gather(*tasks, return_exceptions=True) bounded by asyncio.timeout (the 3.11+ idiom). Tasks that don't finish before the deadline are cancelled and awaited once more so their CancelledError is consumed rather than leaking as "Task was destroyed but it is pending" warnings.

H12: a task cancelled by the deadline is not guaranteed to have reached its own ack/nack — CancelledError is a BaseException, so it skips right past the pipeline's except Exception handling and the message is left unsettled. Rather than leave that to the implicit requeue-on-unacked-channel-close behavior when the transport eventually disconnects, any message still unsettled after a task is abandoned is nacked (requeue) here and logged by delivery tag — an explicit, immediate, observable redelivery instead of an implicit one.

Source code in src/rabbitkit/concurrency.py

async def stop(self, timeout: float | None = None) -> None:
    """Stop the worker pool, waiting for in-flight tasks.

    R-TaskGroup: rather than ``asyncio.wait`` + a manual ``task.cancel()``
    loop, we ``gather(*tasks, return_exceptions=True)`` bounded by
    ``asyncio.timeout`` (the 3.11+ idiom). Tasks that don't finish before
    the deadline are cancelled and awaited once more so their
    ``CancelledError`` is consumed rather than leaking as "Task was
    destroyed but it is pending" warnings.

    H12: a task cancelled by the deadline is *not* guaranteed to have
    reached its own ``ack``/``nack`` — ``CancelledError`` is a
    ``BaseException``, so it skips right past the pipeline's
    ``except Exception`` handling and the message is left unsettled.
    Rather than leave that to the implicit requeue-on-unacked-channel-close
    behavior when the transport eventually disconnects, any message still
    unsettled after a task is abandoned is nacked (requeue) here and
    logged by delivery tag — an explicit, immediate, observable
    redelivery instead of an implicit one.
    """
    self._running = False
    if not self._tasks:
        return
    effective = timeout if timeout is not None else self._config.stop_timeout
    tasks = list(self._tasks)
    task_messages = dict(self._task_messages)
    try:
        async with asyncio.timeout(effective):
            await asyncio.gather(*tasks, return_exceptions=True)
    except TimeoutError:
        # Deadline elapsed: cancel any still-running tasks and drain them
        # so cancellation propagates cleanly (no pending-task warnings).
        # In practice `asyncio.timeout` firing cancels this method's own
        # await of gather(), and gather's cancellation handling already
        # cancels + drains every task before re-raising -- so `not_done`
        # is normally empty by the time we get here; this loop is a
        # defensive backstop, not the primary abandonment path.
        not_done = [t for t in tasks if not t.done()]
        for task in not_done:  # pragma: no cover — gather already drains before raising
            task.cancel()  # pragma: no cover
        if not_done:  # pragma: no cover
            await asyncio.gather(*not_done, return_exceptions=True)  # pragma: no cover
        for message in task_messages.values():
            if message.is_settled:
                continue  # handler reached its own ack/nack before being cut off
            logger.warning(
                "AsyncWorkerPool: handler for delivery_tag=%s message_id=%s did not "
                "complete within stop_timeout; abandoning (task cancelled) and nacking "
                "for redelivery — ensure handlers are idempotent under at-least-once delivery",
                message.delivery_tag,
                message.message_id,
            )
            try:
                await message.nack_async(requeue=True)
            except Exception:
                logger.warning(
                    "nack on abandoned handler's message raised", exc_info=True
                )
    finally:
        self._tasks.clear()
        self._task_messages.clear()
    logger.info("AsyncWorkerPool stopped")

submit(callback: Callable[[RabbitMessage], Awaitable[None]], message: RabbitMessage) -> None async

Submit a message for concurrent processing.

If worker_count=1, runs directly (no semaphore). Otherwise, acquires semaphore slot before executing.

H12: refuses (nacks for redelivery) instead of scheduling an unawaited task when the pool isn't running — e.g. a delivery callback firing after stop() already cleared _tasks. Nothing would ever await that task, so it would silently race the event loop's own shutdown instead of the message being cleanly settled.

Source code in src/rabbitkit/concurrency.py

async def submit(
    self,
    callback: Callable[[RabbitMessage], Awaitable[None]],
    message: RabbitMessage,
) -> None:
    """Submit a message for concurrent processing.

    If worker_count=1, runs directly (no semaphore).
    Otherwise, acquires semaphore slot before executing.

    H12: refuses (nacks for redelivery) instead of scheduling an unawaited
    task when the pool isn't running — e.g. a delivery callback firing
    after ``stop()`` already cleared ``_tasks``. Nothing would ever await
    that task, so it would silently race the event loop's own shutdown
    instead of the message being cleanly settled.
    """
    if self._semaphore is None:
        await callback(message)
        return

    if not self._running:
        logger.warning(
            "AsyncWorkerPool.submit() called while stopped (delivery_tag=%s, "
            "message_id=%s); nacking for redelivery instead of orphaning an "
            "unawaited task",
            message.delivery_tag,
            message.message_id,
        )
        if not message.is_settled:
            await message.nack_async(requeue=True)
        return

    semaphore = self._semaphore

    async def _run() -> None:
        async with semaphore:
            await callback(message)

    task = asyncio.create_task(_run())
    self._tasks.add(task)
    self._task_messages[task] = message
    task.add_done_callback(self._on_task_done)

WorkerConfig dataclass

Consumer concurrency configuration.

Accepted by broker.start(), NOT part of RabbitConfig. Added in 0.2.0.

stop_timeout (H12): a FALLBACK drain deadline for the worker pool's stop() — consulted only when no explicit timeout is passed. On the standard shutdown path (broker.stop()) the pool drain budget is ConsumerConfig.graceful_timeout, NOT this field — tune THAT (and keep terminationGracePeriodSeconds a few seconds above it) for Kubernetes; setting only stop_timeout there has no effect (architect review M2 — the old text here said the opposite). stop_timeout matters when you drive a worker pool directly. A handler still running past this deadline is abandoned, not killed: the sync pool's daemon thread keeps running in the background (it is never forcibly stopped — Python cannot interrupt an arbitrary thread), and the async pool cancels the task (which does not guarantee the handler reaches its own ack/nack — CancelledError is a BaseException and is not caught by the pipeline's exception handling). Either way the abandoned delivery is logged by delivery tag/message id, and — for the async pool — nacked for redelivery immediately rather than relying on the implicit requeue that happens when the connection eventually closes. Because the original handler may still complete its side effects after abandonment, handlers must be idempotent under at-least-once delivery regardless of stop_timeout.

Source code in src/rabbitkit/core/config.py

@dataclass(frozen=True, slots=True)
class WorkerConfig:
    """Consumer concurrency configuration.

    Accepted by broker.start(), NOT part of RabbitConfig. Added in 0.2.0.

    ``stop_timeout`` (H12): a FALLBACK drain deadline for the worker pool's
    ``stop()`` — consulted only when no explicit timeout is passed. On the
    standard shutdown path (``broker.stop()``) the pool drain budget is
    **``ConsumerConfig.graceful_timeout``**, NOT this field — tune THAT (and
    keep ``terminationGracePeriodSeconds`` a few seconds above it) for
    Kubernetes; setting only ``stop_timeout`` there has no effect
    (architect review M2 — the old text here said the opposite).
    ``stop_timeout`` matters when you drive a worker pool directly. A
    handler still running past this deadline is **abandoned, not killed**:
    the sync pool's daemon thread keeps running in the background (it is
    never forcibly stopped — Python cannot interrupt an arbitrary thread),
    and the async pool cancels the task (which does not guarantee the
    handler reaches its own ack/nack — ``CancelledError`` is a
    ``BaseException`` and is not caught by the pipeline's exception
    handling). Either way the abandoned delivery is logged by delivery
    tag/message id, and — for the async pool — nacked for redelivery
    immediately rather than relying on the implicit requeue that happens
    when the connection eventually closes. Because the original handler may
    still complete its side effects after abandonment, **handlers must be
    idempotent under at-least-once delivery** regardless of ``stop_timeout``.
    """

    worker_count: int = 1
    prefetch_per_worker: int | None = None
    stop_timeout: float = 30.0
    # M11: bound the sync worker pool's internal work queue. 0 = unbounded
    # (default, unchanged). In practice the broker's prefetch already caps
    # in-flight messages (effective prefetch = worker_count x
    # prefetch_per_worker), so this is a defensive ceiling — set it >= your
    # effective prefetch so it only trips if prefetch isn't being honored.
    # When the queue is full, submit() blocks (backpressure); keep it above
    # prefetch so that never happens on the I/O thread.
    max_queue_size: int = 0

    def __post_init__(self) -> None:
        if self.worker_count < 1:
            raise ConfigValidationError(f"WorkerConfig.worker_count must be >= 1, got {self.worker_count}")
        if self.max_queue_size < 0:
            raise ConfigValidationError(f"WorkerConfig.max_queue_size must be >= 0, got {self.max_queue_size}")
        # Architect review M1: prefetch_per_worker=0 made the effective
        # prefetch worker_count * 0 = 0 = AMQP "unlimited" — see
        # ConsumerConfig.prefetch_count for why that is never intended.
        if self.prefetch_per_worker is not None and self.prefetch_per_worker < 1:
            raise ConfigValidationError(
                f"WorkerConfig.prefetch_per_worker must be >= 1 when set, got "
                f"{self.prefetch_per_worker} (0 would make the effective prefetch "
                "UNLIMITED, not disabled)."
            )

SyncBatchPublisher

Pipelined publisher confirms for sync code on a dedicated SelectConnection I/O thread — raises the ~0.9k msg/s blocking-confirm ceiling for callers who adopt it. Standalone by design (not wired into SyncBroker.publish).

SyncBatchPublisher

Pipelined-confirm publisher on a dedicated pika.SelectConnection.

Thread-safe: any number of caller threads may publish() concurrently. Each call blocks only for ITS OWN confirm (bounded by confirm_timeout), while the I/O thread keeps the channel's confirm window full — confirms for many in-flight messages are serviced concurrently instead of one blocking round-trip per message.

Standalone-only (see module docstring): not wired into SyncBroker.

Source code in src/rabbitkit/sync/batch.py

class SyncBatchPublisher:
    """Pipelined-confirm publisher on a dedicated ``pika.SelectConnection``.

    Thread-safe: any number of caller threads may ``publish()`` concurrently.
    Each call blocks only for ITS OWN confirm (bounded by *confirm_timeout*),
    while the I/O thread keeps the channel's confirm window full — confirms
    for many in-flight messages are serviced concurrently instead of one
    blocking round-trip per message.

    Standalone-only (see module docstring): not wired into ``SyncBroker``.
    """

    def __init__(
        self,
        connection_config: ConnectionConfig | None = None,
        socket_config: SocketConfig | None = None,
        security_config: SecurityConfig | None = None,
        confirm_timeout: float = 5.0,
    ) -> None:
        self._connection_config = connection_config or ConnectionConfig()
        self._socket_config = socket_config or SocketConfig()
        self._security_config = security_config or SecurityConfig()
        self._confirm_timeout = float(confirm_timeout)

        # RLock: settlement helpers are called both standalone and from
        # within larger locked sections (fail-all, return matching).
        self._lock = threading.RLock()
        self._queue: deque[tuple[MessageEnvelope, _Slot]] = deque()
        self._pending: dict[int, _Slot] = {}  # delivery_tag → slot (publish order)
        self._next_tag = 0  # mirrors pika's per-channel delivery-tag counter

        self._connection: Any = None  # pika.SelectConnection (I/O thread's)
        self._channel: Any = None
        self._pika: Any = None  # the imported pika module (set on the I/O thread)

        self._ready = threading.Event()  # connected + channel in confirm mode
        self._closing = threading.Event()
        self._io_dead = threading.Event()  # I/O thread has exited
        self._closed = False
        self._thread: threading.Thread | None = None
        self._start_error: BaseException | None = None
        self._connected_once = False  # current SelectConnection reached confirm mode

        # Bounded reconnect (same spirit as SyncTransport._ensure_connected:
        # bounded attempts, full jitter, exponential backoff).
        self.max_reconnect_attempts: int = 5

    # ── lifecycle ─────────────────────────────────────────────────────────

    def start(self, ready_timeout: float = 30.0) -> None:
        """Spawn the I/O thread and block until the confirm channel is ready.

        Raises ``TimeoutError`` after *ready_timeout* if the broker never
        becomes reachable, or ``RuntimeError`` if the I/O thread gave up
        (connect attempts exhausted). Idempotent while running.
        """
        with self._lock:
            if self._closed:
                raise RuntimeError("SyncBatchPublisher is closed")
            if self._thread is not None:
                return  # already started
            self._thread = threading.Thread(
                target=self._io_loop, name=IO_THREAD_NAME, daemon=True
            )
        self._thread.start()

        deadline = time.monotonic() + ready_timeout
        while not self._ready.wait(timeout=0.02):
            if self._io_dead.is_set():
                err = self._start_error
                self.close(timeout=1.0)
                raise RuntimeError(
                    "SyncBatchPublisher failed to connect (attempts exhausted)"
                ) from err
            if time.monotonic() >= deadline:
                self.close(timeout=1.0)
                raise TimeoutError(
                    f"SyncBatchPublisher not ready within {ready_timeout}s"
                )

    def close(self, timeout: float = 10.0) -> None:
        """Stop accepting publishes, drain briefly, fail stragglers, shut down.

        Waits (bounded by *timeout*) for in-flight confirms to settle, then
        fails any stragglers with ``PublishStatus.ERROR`` (M17: never silent),
        stops the ioloop and joins the I/O thread. Idempotent.
        """
        with self._lock:
            if self._closed:
                return
            self._closing.set()

        deadline = time.monotonic() + timeout
        # Bounded wait for unsettled confirms (skip if nothing is in flight).
        while time.monotonic() < deadline:
            with self._lock:
                unsettled = any(s.outcome is None for s in self._pending.values()) or any(
                    s.outcome is None for _, s in self._queue
                )
            if not unsettled or self._io_dead.is_set():
                break
            time.sleep(0.005)

        # Fail stragglers — every remaining slot gets a terminal outcome.
        self._fail_all(RuntimeError("SyncBatchPublisher closed"))

        connection = self._connection
        if connection is not None:
            try:
                connection.ioloop.add_callback_threadsafe(self._shutdown_io)
            except Exception:
                # ioloop already stopped / connection already dead — the I/O
                # thread exits via its _closing check.
                logger.debug("close(): ioloop wake-up failed (already stopped)")

        thread = self._thread
        if thread is not None and thread is not threading.current_thread():
            thread.join(timeout=max(0.0, deadline - time.monotonic()) + 1.0)

        with self._lock:
            self._closed = True
            self._ready.clear()

    def __enter__(self) -> SyncBatchPublisher:
        self.start()
        return self

    def __exit__(self, *args: Any) -> None:
        self.close()

    # ── publish (caller threads) ──────────────────────────────────────────

    def publish(
        self, envelope: MessageEnvelope, timeout: float | None = None
    ) -> PublishOutcome:
        """Publish *envelope* and block until ITS confirm settles.

        Returns CONFIRMED / NACKED / RETURNED per the broker's verdict,
        TIMEOUT if no verdict arrived within *timeout* (default
        ``confirm_timeout``), or ERROR if the publisher is closed,
        disconnected, or the connection died with this message in flight.
        Never raises for transport-level failures; never hangs.
        """
        wait = self._confirm_timeout if timeout is None else float(timeout)

        with self._lock:
            if self._closed or self._closing.is_set() or not self._ready.is_set():
                return PublishOutcome(
                    status=PublishStatus.ERROR,
                    exchange=envelope.exchange,
                    routing_key=envelope.routing_key,
                    error=RuntimeError("SyncBatchPublisher is not running/connected"),
                )
            slot = _Slot(envelope)
            self._queue.append((envelope, slot))
            connection = self._connection

        try:
            # The ONLY thread-safe way to poke the SelectConnection: the
            # drain itself runs on the I/O thread (invariant 1).
            connection.ioloop.add_callback_threadsafe(self._drain)
        except Exception as exc:
            # Connection died between the ready-check and the wake-up — the
            # close callback's fail-all may already have settled the slot;
            # _settle is first-wins either way (invariant 2).
            with self._lock:
                try:
                    self._queue.remove((envelope, slot))
                except ValueError:
                    pass  # already drained/failed elsewhere
            self._settle(slot, PublishStatus.ERROR, error=exc)

        if not slot.event.wait(timeout=wait):
            with self._lock:
                if slot.outcome is None:
                    # Invariant 3: caller never hangs. Mark abandoned so the
                    # late confirm (if it ever arrives) is a no-op.
                    slot.abandoned = True
                    slot.outcome = PublishOutcome(
                        status=PublishStatus.TIMEOUT,
                        exchange=envelope.exchange,
                        routing_key=envelope.routing_key,
                        error=TimeoutError(
                            f"No publisher confirm within {wait}s"
                        ),
                    )
                    slot.event.set()

        outcome = slot.outcome
        if outcome is None:  # pragma: no cover — event is only set after outcome
            outcome = PublishOutcome(
                status=PublishStatus.ERROR,
                exchange=envelope.exchange,
                routing_key=envelope.routing_key,
                error=RuntimeError("publish slot settled without an outcome"),
            )
        return outcome

    # ── settlement (any thread, lock-guarded) ─────────────────────────────

    def _settle(
        self,
        slot: _Slot,
        status: PublishStatus,
        *,
        error: BaseException | None = None,
        delivery_tag: int | None = None,
    ) -> None:
        """Settle *slot* exactly once (first settlement wins)."""
        with self._lock:
            if slot.outcome is not None or slot.abandoned:
                return  # already settled, or caller timed out (late confirm no-op)
            slot.outcome = PublishOutcome(
                status=status,
                delivery_tag=delivery_tag,
                exchange=slot.envelope.exchange,
                routing_key=slot.envelope.routing_key,
                error=error,
            )
        slot.event.set()

    def _fail_all(self, error: BaseException) -> None:
        """Fail EVERY unsettled slot — in flight and still queued (M17)."""
        with self._lock:
            pending = list(self._pending.items())
            self._pending.clear()
            queued = [slot for _, slot in self._queue]
            self._queue.clear()
        for tag, slot in pending:
            self._settle(slot, PublishStatus.ERROR, error=error, delivery_tag=tag)
        for slot in queued:
            self._settle(slot, PublishStatus.ERROR, error=error)

    # ── I/O thread ────────────────────────────────────────────────────────

    def _io_loop(self) -> None:
        """Thread body: run the SelectConnection ioloop, reconnect bounded."""
        try:
            try:
                import pika
            except ImportError as exc:
                self._start_error = ImportError(
                    "pika is required for SyncBatchPublisher. "
                    "Install it with: pip install rabbitkit[sync]"
                )
                self._start_error.__cause__ = exc
                return

            self._pika = pika
            attempts = 0
            backoff = self._connection_config.reconnect_backoff_base

            while not self._closing.is_set():
                self._connected_once = False
                try:
                    params = make_pika_connection_params(
                        self._connection_config,
                        self._socket_config,
                        self._security_config,
                    )
                    connection = pika.SelectConnection(
                        parameters=params,
                        on_open_callback=self._on_connection_open,
                        on_open_error_callback=self._on_connection_open_error,
                        on_close_callback=self._on_connection_closed,
                    )
                    self._connection = connection
                    connection.ioloop.start()  # returns after ioloop.stop()
                except Exception as exc:
                    logger.warning("SyncBatchPublisher connection attempt failed: %s", exc)
                    self._start_error = exc

                if self._closing.is_set():
                    break
                if self._connected_once:
                    # This connection reached confirm mode — reset the budget.
                    attempts = 0
                    backoff = self._connection_config.reconnect_backoff_base
                attempts += 1
                if attempts > self.max_reconnect_attempts:
                    logger.critical(
                        "SyncBatchPublisher reconnect attempts exhausted after %d tries; giving up",
                        attempts - 1,
                    )
                    break
                # Full jitter (H-SRE3 spirit): spread reconnects across clients.
                sleep_for = random.uniform(0.0, backoff)  # noqa: S311
                logger.warning(
                    "SyncBatchPublisher reconnecting in %.2fs (attempt %d/%d)",
                    sleep_for,
                    attempts,
                    self.max_reconnect_attempts,
                )
                self._closing.wait(timeout=sleep_for)  # interruptible by close()
                backoff = min(backoff * 2, self._connection_config.reconnect_backoff_max)
        finally:
            # Thread exiting for ANY reason: nothing will ever settle these
            # slots again — fail them now (invariant 2).
            self._ready.clear()
            self._fail_all(
                self._start_error
                if self._start_error is not None and not self._connected_once
                else RuntimeError("SyncBatchPublisher I/O thread exited")
            )
            self._io_dead.set()

    def _shutdown_io(self) -> None:
        """Graceful shutdown, on the I/O thread (scheduled by close())."""
        try:
            if self._channel is not None and self._channel.is_open:
                self._channel.close()
            if self._connection is not None and self._connection.is_open:
                # Triggers _on_connection_closed → ioloop.stop().
                self._connection.close()
            else:
                self._connection.ioloop.stop()
        except Exception:  # pragma: no cover — best effort during shutdown
            try:
                self._connection.ioloop.stop()
            except Exception:
                logger.debug("shutdown: ioloop.stop() failed", exc_info=True)

    # ── pika callbacks (I/O thread) ───────────────────────────────────────

    def _on_connection_open(self, connection: Any) -> None:
        connection.channel(on_open_callback=self._on_channel_open)

    def _on_connection_open_error(self, connection: Any, error: Any) -> None:
        logger.warning("SyncBatchPublisher failed to open connection: %s", error)
        self._start_error = (
            error if isinstance(error, BaseException) else RuntimeError(str(error))
        )
        connection.ioloop.stop()

    def _on_connection_closed(self, connection: Any, reason: Any) -> None:
        self._ready.clear()
        self._channel = None
        if not self._closing.is_set():
            logger.warning("SyncBatchPublisher connection closed unexpectedly: %s", reason)
        # M17: every unsettled outcome (in flight AND queued) gets ERROR —
        # the confirms for these delivery tags will never arrive.
        err = reason if isinstance(reason, BaseException) else RuntimeError(str(reason))
        self._fail_all(err)
        connection.ioloop.stop()  # _io_loop decides whether to reconnect

    def _on_channel_open(self, channel: Any) -> None:
        self._channel = channel
        channel.add_on_close_callback(self._on_channel_closed)
        channel.add_on_return_callback(self._on_return)
        # ack_nack_callback: Basic.Ack / Basic.Nack frames (pipelined
        # confirms); callback: Confirm.SelectOk — confirm mode is active.
        channel.confirm_delivery(
            ack_nack_callback=self._on_delivery_confirmation,
            callback=self._on_confirm_select_ok,
        )

    def _on_confirm_select_ok(self, _frame: Any) -> None:
        with self._lock:
            self._next_tag = 0  # delivery tags are per-channel
        self._connected_once = True
        self._ready.set()
        logger.info("SyncBatchPublisher ready (confirm mode active)")
        self._drain()  # anything enqueued in the ready/connected race window

    def _on_channel_closed(self, channel: Any, reason: Any) -> None:
        """Channel died out from under us (connection may still be open)."""
        if self._closing.is_set() or channel is not self._channel:
            return  # shutdown path / stale channel — handled elsewhere
        logger.warning("SyncBatchPublisher channel closed unexpectedly: %s", reason)
        self._ready.clear()
        self._channel = None
        err = reason if isinstance(reason, BaseException) else RuntimeError(str(reason))
        self._fail_all(err)
        # Simplest correct recovery: recycle the whole connection (tag
        # counters and confirm mode are per-channel; a fresh connection via
        # the reconnect loop re-establishes both).
        try:
            if self._connection is not None and self._connection.is_open:
                self._connection.close()
        except Exception:  # pragma: no cover — best effort
            logger.debug("channel-closed: connection.close() failed", exc_info=True)

    def _on_delivery_confirmation(self, method_frame: Any) -> None:
        """Basic.Ack / Basic.Nack — settle one tag, or all ≤ tag if multiple."""
        method = method_frame.method
        tag = int(method.delivery_tag)
        multiple = bool(getattr(method, "multiple", False))
        acked = isinstance(method, self._pika.spec.Basic.Ack)
        status = PublishStatus.CONFIRMED if acked else PublishStatus.NACKED
        error = (
            None
            if acked
            else RuntimeError(f"Broker nacked delivery_tag={tag} (multiple={multiple})")
        )

        with self._lock:
            if multiple:
                tags = sorted(t for t in self._pending if t <= tag)
            else:
                tags = [tag] if tag in self._pending else []
            settled = [(t, self._pending.pop(t)) for t in tags]

        if not settled:
            # Late confirm for an abandoned-and-reaped tag, or unknown tag.
            logger.debug("Confirm for unknown delivery_tag=%s (late/reaped) — ignored", tag)
            return
        for t, slot in settled:
            self._settle(slot, status, error=error, delivery_tag=t)

    def _on_return(self, _channel: Any, method: Any, properties: Any, _body: bytes) -> None:
        """Basic.Return — unroutable mandatory publish bounced by the broker.

        pika delivers the Return BEFORE the corresponding Basic.Ack, so we
        settle the slot RETURNED here and first-settlement-wins makes the
        following Ack a no-op. Matched to the MOST RECENT unsettled publish
        by (exchange, routing_key) and, when the broker echoed one, message_id.
        """
        msg_id = getattr(properties, "message_id", None)
        with self._lock:
            candidate: _Slot | None = None
            candidate_tag: int | None = None
            for t, slot in self._pending.items():  # insertion order = publish order
                if slot.outcome is not None or slot.abandoned:
                    continue
                env = slot.envelope
                if env.exchange != method.exchange or env.routing_key != method.routing_key:
                    continue
                if msg_id is not None and env.message_id != msg_id:
                    continue
                candidate, candidate_tag = slot, t  # keep last match = most recent
            if candidate is None:
                logger.warning(
                    "Basic.Return with no matching unsettled publish "
                    "(exchange=%r routing_key=%r message_id=%r) — ignored",
                    method.exchange,
                    method.routing_key,
                    msg_id,
                )
                return
            # Leave the tag in _pending: the broker still sends the Ack for a
            # returned message; _on_delivery_confirmation pops it (and its
            # settle attempt no-ops — first settlement wins).
            self._settle(
                candidate,
                PublishStatus.RETURNED,
                error=RuntimeError(
                    f"Unroutable: reply_code={getattr(method, 'reply_code', None)} "
                    f"reply_text={getattr(method, 'reply_text', None)}"
                ),
                delivery_tag=candidate_tag,
            )

    # ── publishing (I/O thread) ───────────────────────────────────────────

    def _drain(self) -> None:
        """Drain the caller queue onto the channel (runs on the I/O thread)."""
        while True:
            with self._lock:
                if not self._queue:
                    return
                envelope, slot = self._queue.popleft()
                channel = self._channel

            if channel is None or not self._ready.is_set() or not channel.is_open:
                self._settle(
                    slot,
                    PublishStatus.ERROR,
                    error=RuntimeError("SyncBatchPublisher is not connected"),
                )
                continue

            try:
                properties = self._build_properties(envelope)
            except Exception as exc:
                self._settle(slot, PublishStatus.ERROR, error=exc)
                continue

            with self._lock:
                self._next_tag += 1
                tag = self._next_tag
                self._pending[tag] = slot

            try:
                channel.basic_publish(
                    exchange=envelope.exchange,
                    routing_key=envelope.routing_key,
                    body=envelope.body,
                    properties=properties,
                    mandatory=envelope.mandatory,
                )
            except Exception as exc:
                with self._lock:
                    self._pending.pop(tag, None)
                self._settle(slot, PublishStatus.ERROR, error=exc)
                # Our tag counter may now disagree with pika's — never keep
                # publishing on a desynced channel. Recycle it (the channel
                # close callback fails any siblings and triggers reconnect).
                try:
                    if channel.is_open:
                        channel.close()
                except Exception:  # pragma: no cover — best effort
                    logger.debug("drain: channel.close() failed", exc_info=True)
                return

    def _build_properties(self, envelope: MessageEnvelope) -> Any:
        """Build pika.BasicProperties exactly like SyncTransport._publish_on_channel."""
        properties = self._pika.BasicProperties(
            message_id=envelope.message_id,
            correlation_id=envelope.correlation_id,
            reply_to=envelope.reply_to,
            content_type=envelope.content_type,
            content_encoding=envelope.content_encoding,
            headers=envelope.headers or None,
            delivery_mode=envelope.delivery_mode,
            priority=envelope.priority,
            expiration=envelope.expiration,
            type=envelope.type,
            user_id=envelope.user_id,
            app_id=envelope.app_id,
        )
        if envelope.timestamp:
            properties.timestamp = int(envelope.timestamp.timestamp())
        return properties

Methods:

start(ready_timeout: float = 30.0) -> None

Spawn the I/O thread and block until the confirm channel is ready.

Raises TimeoutError after ready_timeout if the broker never becomes reachable, or RuntimeError if the I/O thread gave up (connect attempts exhausted). Idempotent while running.

Source code in src/rabbitkit/sync/batch.py

def start(self, ready_timeout: float = 30.0) -> None:
    """Spawn the I/O thread and block until the confirm channel is ready.

    Raises ``TimeoutError`` after *ready_timeout* if the broker never
    becomes reachable, or ``RuntimeError`` if the I/O thread gave up
    (connect attempts exhausted). Idempotent while running.
    """
    with self._lock:
        if self._closed:
            raise RuntimeError("SyncBatchPublisher is closed")
        if self._thread is not None:
            return  # already started
        self._thread = threading.Thread(
            target=self._io_loop, name=IO_THREAD_NAME, daemon=True
        )
    self._thread.start()

    deadline = time.monotonic() + ready_timeout
    while not self._ready.wait(timeout=0.02):
        if self._io_dead.is_set():
            err = self._start_error
            self.close(timeout=1.0)
            raise RuntimeError(
                "SyncBatchPublisher failed to connect (attempts exhausted)"
            ) from err
        if time.monotonic() >= deadline:
            self.close(timeout=1.0)
            raise TimeoutError(
                f"SyncBatchPublisher not ready within {ready_timeout}s"
            )

close(timeout: float = 10.0) -> None

Stop accepting publishes, drain briefly, fail stragglers, shut down.

Waits (bounded by timeout) for in-flight confirms to settle, then fails any stragglers with PublishStatus.ERROR (M17: never silent), stops the ioloop and joins the I/O thread. Idempotent.

Source code in src/rabbitkit/sync/batch.py

def close(self, timeout: float = 10.0) -> None:
    """Stop accepting publishes, drain briefly, fail stragglers, shut down.

    Waits (bounded by *timeout*) for in-flight confirms to settle, then
    fails any stragglers with ``PublishStatus.ERROR`` (M17: never silent),
    stops the ioloop and joins the I/O thread. Idempotent.
    """
    with self._lock:
        if self._closed:
            return
        self._closing.set()

    deadline = time.monotonic() + timeout
    # Bounded wait for unsettled confirms (skip if nothing is in flight).
    while time.monotonic() < deadline:
        with self._lock:
            unsettled = any(s.outcome is None for s in self._pending.values()) or any(
                s.outcome is None for _, s in self._queue
            )
        if not unsettled or self._io_dead.is_set():
            break
        time.sleep(0.005)

    # Fail stragglers — every remaining slot gets a terminal outcome.
    self._fail_all(RuntimeError("SyncBatchPublisher closed"))

    connection = self._connection
    if connection is not None:
        try:
            connection.ioloop.add_callback_threadsafe(self._shutdown_io)
        except Exception:
            # ioloop already stopped / connection already dead — the I/O
            # thread exits via its _closing check.
            logger.debug("close(): ioloop wake-up failed (already stopped)")

    thread = self._thread
    if thread is not None and thread is not threading.current_thread():
        thread.join(timeout=max(0.0, deadline - time.monotonic()) + 1.0)

    with self._lock:
        self._closed = True
        self._ready.clear()

publish(envelope: MessageEnvelope, timeout: float | None = None) -> PublishOutcome

Publish envelope and block until ITS confirm settles.

Returns CONFIRMED / NACKED / RETURNED per the broker's verdict, TIMEOUT if no verdict arrived within timeout (default confirm_timeout), or ERROR if the publisher is closed, disconnected, or the connection died with this message in flight. Never raises for transport-level failures; never hangs.

Source code in src/rabbitkit/sync/batch.py

def publish(
    self, envelope: MessageEnvelope, timeout: float | None = None
) -> PublishOutcome:
    """Publish *envelope* and block until ITS confirm settles.

    Returns CONFIRMED / NACKED / RETURNED per the broker's verdict,
    TIMEOUT if no verdict arrived within *timeout* (default
    ``confirm_timeout``), or ERROR if the publisher is closed,
    disconnected, or the connection died with this message in flight.
    Never raises for transport-level failures; never hangs.
    """
    wait = self._confirm_timeout if timeout is None else float(timeout)

    with self._lock:
        if self._closed or self._closing.is_set() or not self._ready.is_set():
            return PublishOutcome(
                status=PublishStatus.ERROR,
                exchange=envelope.exchange,
                routing_key=envelope.routing_key,
                error=RuntimeError("SyncBatchPublisher is not running/connected"),
            )
        slot = _Slot(envelope)
        self._queue.append((envelope, slot))
        connection = self._connection

    try:
        # The ONLY thread-safe way to poke the SelectConnection: the
        # drain itself runs on the I/O thread (invariant 1).
        connection.ioloop.add_callback_threadsafe(self._drain)
    except Exception as exc:
        # Connection died between the ready-check and the wake-up — the
        # close callback's fail-all may already have settled the slot;
        # _settle is first-wins either way (invariant 2).
        with self._lock:
            try:
                self._queue.remove((envelope, slot))
            except ValueError:
                pass  # already drained/failed elsewhere
        self._settle(slot, PublishStatus.ERROR, error=exc)

    if not slot.event.wait(timeout=wait):
        with self._lock:
            if slot.outcome is None:
                # Invariant 3: caller never hangs. Mark abandoned so the
                # late confirm (if it ever arrives) is a no-op.
                slot.abandoned = True
                slot.outcome = PublishOutcome(
                    status=PublishStatus.TIMEOUT,
                    exchange=envelope.exchange,
                    routing_key=envelope.routing_key,
                    error=TimeoutError(
                        f"No publisher confirm within {wait}s"
                    ),
                )
                slot.event.set()

    outcome = slot.outcome
    if outcome is None:  # pragma: no cover — event is only set after outcome
        outcome = PublishOutcome(
            status=PublishStatus.ERROR,
            exchange=envelope.exchange,
            routing_key=envelope.routing_key,
            error=RuntimeError("publish slot settled without an outcome"),
        )
    return outcome