SC ReconciliationReconciliation Automation
  1. Home
  2. Ingestion
  3. Async Batch Processing for High-Volume Feeds
  4. Backpressure and Rate Limiting for Supplier Feed Ingestion

Backpressure and Rate Limiting for Supplier Feed Ingestion

↑ Part of Async Batch Processing for High-Volume Feeds.

A bounded asyncio.Semaphore stops an ingester from opening more sockets than the connection pool can hold, but it does not stop the ingester from opening those sockets faster than a supplier’s API quota allows, and it does nothing at all if the process producing records outruns the coroutines consuming them. Overwhelming downstream is rarely one failure — it is three related ones: too many concurrent requests, too many requests per second against a single trading partner, and an unbounded buffer of not-yet-processed records sitting in memory while the broker, database, or matching engine falls behind. This page is narrowly about the mechanics that prevent all three: capping in-flight work with a semaphore, capping per-supplier throughput with a token bucket, and using a bounded queue that blocks the producer rather than growing forever, plus the decision to shed load instead of buffering it when even blocking is not enough.

Operational Trigger Signals

Add explicit backpressure and rate limiting when the pipeline’s own telemetry shows one or more of these conditions, not preemptively on every ingester regardless of load:

  1. Consumer lag rising faster than producer throughput: if the reader task is enqueuing records faster than the worker pool drains them, queue depth climbs monotonically across a run instead of oscillating around a steady state. A queue that never returns to its baseline depth between polling cycles is the first sign the consumer side needs more concurrency, fewer producers, or an explicit cap.
  2. Worker OOM-kills mid-run: an unbounded asyncio.gather() over every record in a feed, or an unbounded list accumulating parsed payloads before a single bulk write, holds the entire feed in memory at once. A process that dies with OSError: [Errno 12] Cannot allocate memory or is OOM-killed by the kernel partway through a large feed is buffering instead of streaming.
  3. 429 Too Many Requests or 503 Service Unavailable from a specific supplier: this is the clearest signal of all — the remote API is telling you its quota, and continuing to retry at the same rate only compounds the problem. A per-supplier 429 rate above a few percent of that supplier’s requests means the token bucket for that supplier is undersized or missing entirely.
  4. Database connection pool saturation downstream of the ingester: if the matcher or staging writer reports TimeoutError: connection pool exhausted while the ingester itself reports healthy throughput, the ingester is producing faster than the database can absorb — a queue-depth or semaphore problem one hop downstream of where it appears.

These four signals point at three different bottlenecks — concurrent load, per-supplier rate, and buffer growth — so the fix is never a single knob. The Async Batch Processing for High-Volume Feeds cluster covers connector and semaphore sizing broadly; this page implements the three specific primitives that keep a concurrent ingester from tipping any of them over.

Bounded-queue backpressure loop with concurrency and rate gating A supplier feed producer writes records into a bounded asyncio Queue with a fixed maxsize. When the queue reaches capacity, the pipeline follows one of two configured policies: buffer, which suspends the producer inside await queue.put() until a worker frees space and then resumes it, or shed, which rejects the incoming request immediately with a 429 or 503 and never enters the queue. Workers pull from the queue only after acquiring a semaphore slot that bounds system-wide in-flight concurrency and a token from a per-supplier token bucket that bounds sustained request rate, then dispatch the record to the downstream broker, database, or matching engine. Feed producer async reader, one record at a time Bounded queue Queue(maxsize=Q) qsize / maxsize Concurrency + rate gate Semaphore max_inflight = N Token bucket rate r, burst C per supplier Downstream broker · DB matcher Queue full: buffer or shed? BLOCK producer await queue.put() suspends until capacity frees SHED request reject now: 429/503 no queue entry put() get() dispatch buffer shed resumes producer

The token bucket and the semaphore answer different questions and both must pass before a request is dispatched. The bucket governs the sustained rate allowed against one supplier’s endpoint; the semaphore governs how many requests, across every supplier, are open on the system at once:

B(t)=min⁡(C,  B(t−Δt)+r⋅Δt),dispatch  ⟺  (B(t)≥c)  ∧  (inflight<N)B(t) = \min\bigl(C,\; B(t - \Delta t) + r \cdot \Delta t\bigr), \qquad \text{dispatch} \iff \bigl(B(t) \ge c\bigr) \;\land\; \bigl(\text{inflight} < N\bigr)

where B(t)B(t) is the bucket’s token level at time tt, CC is the bucket capacity (burst allowance), rr is the refill rate in requests per second, cc is the token cost of the pending request (usually 1), and NN is the semaphore’s max_inflight bound. A record is only dispatched when the supplier-specific bucket has enough tokens and fewer than NN requests are in flight system-wide — passing one gate without the other is not sufficient.

Step-by-Step Implementation

Build the gate as a small composable unit so every producer, regardless of feed format, goes through the same admission control:

  1. Size the semaphore to the tightest downstream resource — the smaller of the connection pool, the database pool, or the matcher’s worker count — never to an arbitrary “feels fast” number.
  2. Give every supplier its own token bucket — a shared bucket lets one high-volume supplier starve a low-volume one, and a single global rate cannot represent ten different quota agreements.
  3. Bound the queue and choose a policy for “full” — decide up front whether a full queue blocks the producer (safe, adds latency) or sheds new work (fast, drops data unless the caller retries).
  4. Wire acquisition order — pull from the queue, acquire a token from the record’s supplier bucket, then acquire a semaphore slot, so a request never holds a semaphore slot while waiting on a rate limit.
  5. Log every gate decision — queue-full events, token waits, and semaphore waits are the signals that later tell you which resource is actually constrained.
  6. Fail closed on rate-limit responses — a 429 from the supplier should shrink that supplier’s effective rate immediately, not just trigger a retry at the same speed.

The implementation below composes a TokenBucket, an asyncio.Semaphore, and a bounded asyncio.Queue into a single SupplierIngestGate. The queue’s put() is the backpressure point: once it is full, the producer’s await suspends until a worker frees a slot, which is what stops an unbounded buildup in memory:

PYTHON
import asyncio
import logging
import time
from dataclasses import dataclass, field
from typing import Awaitable, Callable, Dict

logger = logging.getLogger("supply_chain.ingest.backpressure")

FeedRecord = Dict[str, str]
DownstreamFn = Callable[[FeedRecord], Awaitable[None]]


@dataclass
class TokenBucket:
    """Per-supplier rate limiter: bounds sustained throughput, allows short bursts."""

    capacity: float
    refill_rate: float  # tokens (requests) per second
    tokens: float = field(init=False)
    last_refill: float = field(init=False)

    def __post_init__(self) -> None:
        self.tokens = self.capacity
        self.last_refill = time.monotonic()

    def _refill(self) -> None:
        now = time.monotonic()
        elapsed = now - self.last_refill
        self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
        self.last_refill = now

    async def acquire(self, supplier_id: str, cost: float = 1.0) -> None:
        """Block until `cost` tokens are available for this supplier's bucket."""
        while True:
            self._refill()
            if self.tokens >= cost:
                self.tokens -= cost
                return
            wait_s = (cost - self.tokens) / self.refill_rate
            logger.debug(
                "token_bucket_wait supplier=%s cost=%.1f wait_s=%.3f",
                supplier_id, cost, wait_s,
            )
            await asyncio.sleep(min(wait_s, 0.25))

    def penalize(self, factor: float = 0.5) -> None:
        """Shrink the effective rate after a 429; call from the caller's error handler."""
        self.refill_rate *= factor
        self.tokens = min(self.tokens, self.capacity)


class SupplierIngestGate:
    """Bounds in-flight concurrency, per-supplier rate, and buffered queue depth."""

    def __init__(
        self,
        max_inflight: int,
        supplier_buckets: Dict[str, TokenBucket],
        queue_maxsize: int = 200,
        shed_when_full: bool = False,
    ) -> None:
        self.semaphore = asyncio.Semaphore(max_inflight)
        self.supplier_buckets = supplier_buckets
        self.queue: asyncio.Queue[FeedRecord] = asyncio.Queue(maxsize=queue_maxsize)
        self.shed_when_full = shed_when_full
        self.shed_count = 0

    async def submit(self, record: FeedRecord) -> bool:
        """Producer entrypoint. Returns False if the record was shed, not enqueued."""
        if self.shed_when_full and self.queue.full():
            self.shed_count += 1
            logger.warning(
                "record_shed supplier=%s qsize=%d maxsize=%d",
                record["supplier_id"], self.queue.qsize(), self.queue.maxsize,
            )
            return False
        if self.queue.full():
            logger.info("queue_full_blocking supplier=%s", record["supplier_id"])
        await self.queue.put(record)  # blocks the producer instead of buffering unbounded
        return True

    async def _worker(self, worker_id: int, downstream: DownstreamFn) -> None:
        while True:
            record = await self.queue.get()
            supplier_id = record["supplier_id"]
            bucket = self.supplier_buckets.get(supplier_id)
            try:
                if bucket is not None:
                    await bucket.acquire(supplier_id)
                async with self.semaphore:
                    await downstream(record)
            except Exception as exc:
                logger.error(
                    "record_failed worker=%d supplier=%s err=%s", worker_id, supplier_id, exc
                )
            finally:
                self.queue.task_done()

    async def run(self, downstream: DownstreamFn, worker_count: int = 8) -> None:
        """Start a bounded worker pool that drains the queue under both gates."""
        workers = [asyncio.create_task(self._worker(i, downstream)) for i in range(worker_count)]
        await self.queue.join()
        for w in workers:
            w.cancel()
        logger.info("gate_run_complete workers=%d shed_count=%d", worker_count, self.shed_count)

submit() is the whole backpressure story: when shed_when_full is False (the default), a full queue means await self.queue.put(record) simply does not return until a worker calls task_done() and frees a slot, so the producer’s own coroutine stalls rather than the process growing without bound. Acquiring the token bucket before the semaphore also matters — a request that is waiting on a rate limit should never hold a scarce concurrency slot idle, which is why bucket.acquire() runs outside async with self.semaphore. For the underlying event-loop and task-scheduling mechanics this builds on, see Implementing Asyncio for Concurrent Batch Ingestion.

Configuration Reference

Parameter Accepted values Default Notes
max_inflight (N) 5–200 20 Semaphore bound; set to the smallest downstream pool, not a guess.
TokenBucket.capacity © 5–500 50 Burst allowance per supplier above the sustained rate.
TokenBucket.refill_rate ® 0.5–100 req/s supplier-tier specific Match the supplier’s published or observed quota, not a shared default.
queue_maxsize (Q) 50–2000 200 Bounds worst-case buffered memory; size to chunk_bytes × Q under budget.
shed_when_full True / False False False blocks producers (safe); True rejects new work (fast, lossy unless retried upstream).
worker_count 4–64 8 Consumers draining the queue; independent of max_inflight.
penalize factor 0.3–0.7 0.5 Multiplier applied to refill_rate after a 429 from that supplier.

Set max_inflight first, from the tightest downstream constraint, then size each supplier’s refill_rate from its documented or observed quota — never the reverse, or the semaphore ends up compensating for a rate problem it cannot see. queue_maxsize should be small enough that a full queue represents seconds, not minutes, of backlog; a queue sized to “never fill” defeats the purpose of bounding it.

Debugging & Recovery

  • Producer coroutine appears hung. If submit() never returns, the queue is full and shed_when_full is False — this is the backpressure working as designed, not a bug. Confirm by logging queue.qsize() on an interval; if it is pinned at maxsize, the fix is more workers, a larger max_inflight, or addressing whatever is slow downstream, not raising queue_maxsize, which only delays the same stall.
  • Rising token_bucket_wait durations for one supplier. The bucket is correctly throttling, but if wait times keep growing, either refill_rate is set below what the pipeline actually needs, or a previous penalize() call shrank it after a 429 and it was never restored. Log the bucket’s current refill_rate alongside the wait so you can distinguish a permanent misconfiguration from a temporary penalty.
  • 429 responses continue after adding a token bucket. The bucket caps your outbound rate, but a burst at the start of a run — capacity tokens all spent in the first second — can still exceed a strict supplier quota. Lower capacity closer to refill_rate to flatten the burst, and confirm the supplier’s quota window (per-second vs. per-minute) matches how refill_rate is expressed.
  • Shed records disappearing silently. shed_when_full=True returns False from submit() and increments shed_count, but if the caller does not check that return value, records vanish without a trace. Route every shed record to the same recovery path as a processing failure — the dead-letter queues and retry orchestration pattern applies equally to “rejected before processing” and “failed during processing.”
  • You can see the queue is full but not why. Queue depth, per-supplier token wait time, semaphore wait time, and shed count are the four numbers that localize the bottleneck; expose them as gauges rather than only log lines. The monitoring and alerting for reconciliation pipelines cluster covers wiring these into dashboards and alert thresholds so a growing queue pages someone before it becomes an OOM.

FAQ

Should I block the producer or shed requests when the queue is full?

Block by default. A blocked producer only adds latency, and latency is recoverable; a shed record is data loss unless the upstream caller retries, and most supplier feed sources are not built to retry a shed batch cleanly. Reach for shed_when_full=True only on a path with a cheap, correct retry story upstream — for example, an HTTP endpoint that can return 503 and let the caller’s own retry logic re-send the payload — never for a one-shot poll of a supplier feed with no re-fetch mechanism.

Why do I need both a semaphore and a token bucket instead of just one?

They bound different things. The semaphore caps how many requests are open on the system at any instant, protecting shared resources like the connection pool and database. The token bucket caps how fast requests go out to one specific supplier over time, protecting that supplier’s API quota. A semaphore alone lets you burst against a single slow supplier as fast as the pool allows; a token bucket alone lets you open unlimited concurrent connections as long as the aggregate rate looks fine. Overwhelming downstream usually requires both failure modes to be closed at once.

How do I choose queue_maxsize without just guessing?

Work backward from an acceptable backlog window, not memory alone. Decide how many seconds of buffered work you are willing to tolerate before backpressure kicks in — a few seconds for a latency-sensitive feed, longer for an overnight batch — then multiply by your steady-state processing rate to get queue_maxsize. Cross-check the memory implication (queue_maxsize × average_record_bytes) against the worker’s memory budget; if the two disagree, the backlog-window number should win, since a queue that is memory-safe but represents ten minutes of invisible lag is still an operational problem.