Backpressure control for EDI ingestion workers #

When a container vessel berths, its manifests, VERMAS declarations, and COARRI/CODECO status messages arrive as a single correlated burst — thousands of interchanges within minutes of the ATA. Backpressure control for EDI ingestion workers is the discipline that stops that spike from cascading into the systems the workers write to: a customs declaration API with a fixed throttle, a terminal operating system (TOS) with connection limits, and a shared database that degrades under write amplification. The workers themselves scale horizontally, but the slow dependency behind them does not, so the engineering problem is regulating the rate at which a fast, elastic worker pool is allowed to push work into a fixed-capacity downstream. This page specifies the four controls that do it — bounded prefetch, per-queue rate limiting, a token-bucket gate, and load shedding — and draws the sharp line between broker backpressure and downstream backpressure that governs which control you reach for.

Architecture Alignment #

This task refines the throughput contract established by Async Batch Processing Pipelines within the broader Document Ingestion & EDI Parsing Workflows domain. The queue topology and dead-letter wiring come from Building Celery queues for maritime doc ingestion; this page adds the flow-control layer on top of it. Two kinds of backpressure operate here and confusing them is the most common production mistake. Broker backpressure is inbound: the broker (Redis or RabbitMQ) already absorbs the arrival spike by letting queue depth grow, and prefetch settings decide how much of that depth a single worker is allowed to reserve. Downstream backpressure is outbound: it is the pushback a saturated customs API or TOS exerts on the workers, expressed as 429/503 responses, rising latency, or connection refusals — pushback the broker knows nothing about. Prefetch and acks_late tune the first; a rate limiter and a token-bucket gate tune the second. The same non-blocking posture the landside Terminal API Polling Strategies apply when reading a TOS is what the ingestion workers must apply when writing to one.

Prerequisites & Environment Setup #

The controls target Python 3.11+ (for asyncio.timeout, StrEnum, and precise asyncio.Semaphore semantics). Install the worker, broker client, and structured-logging stack:

pip install "celery[redis]==5.4.*" redis structlog

Structured JSON logging is mandatory: every throttle decision, scale signal, and shed event must be queryable in the observability stack and reproducible for a customs audit, so bare print() and unstructured logging strings are unacceptable. Configure structlog once, at worker boot:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.add_log_level,
        structlog.processors.TimeStamper(fmt="iso", utc=True),
        structlog.processors.JSONRenderer(),
    ],
)
log = structlog.get_logger("maritime.backpressure")

The flow-control layer is driven entirely by environment variables so operators can retune a throttle during a berthing surge without redeploying:

Run the write-heavy tasks on their own pool so a customs-submission stall never starves a fast TOS status push: celery -A maritime_ingestion worker -Q edi.customs,edi.tos -c 12 --prefetch-multiplier=1.

Step-by-step Implementation #

Step 1 — Bound prefetch so one worker cannot swallow a burst #

Broker backpressure begins with prefetch. By default a Celery worker reserves a block of messages up front; during a vessel-arrival burst one worker can hoard hundreds of interchanges, so when it stalls on a slow customs call the whole block is stranded and unavailable to idle siblings. Setting worker_prefetch_multiplier = 1 with task_acks_late = True means each worker holds exactly one in-flight task and acknowledges only after success — a node loss redelivers a single message, not a hoarded block. This is the queue-depth-shaping control; it does nothing about downstream saturation, which is why it is only the first layer.

from celery import Celery

app = Celery("maritime_ingestion", broker="redis://broker:6379/0")

app.conf.worker_prefetch_multiplier = 1        # no hoarding under burst
app.conf.task_acks_late = True                 # ack only after success
app.conf.task_reject_on_worker_lost = True     # redeliver on SIGKILL / OOM
app.conf.worker_max_tasks_per_child = 200      # bound memory from large stowage plans

Step 2 — Rate-limit each queue with Celery rate_limit #

The cheapest downstream-backpressure control is Celery’s built-in rate_limit, which paces the dispatch of a named task. A customs authority’s B2B gateway commonly publishes a hard ceiling — say 30 declarations per minute per sender — and exceeding it earns a 429 and, on repeat offence, a temporary suspension. Annotate each task with the ceiling its downstream advertises; Celery smooths 30/m into roughly one task every two seconds rather than a thundering block. Note the important nuance: rate_limit is enforced per worker instance, so divide the fleet-wide budget by your worker count, or centralise the true global limit in the token-bucket gate of Step 3.

from celery import shared_task

app.conf.task_annotations = {
    "maritime_ingestion.submit_customs_entry": {"rate_limit": "30/m"},
    "maritime_ingestion.push_tos_update": {"rate_limit": "120/m"},
}


@shared_task(name="maritime_ingestion.submit_customs_entry",
             rate_limit="30/m", acks_late=True)
def submit_customs_entry(entry_id: str) -> None:
    log.info("customs_entry_dispatched", entry_id=entry_id)
    ...

Step 3 — Gate a slow downstream with an async token bucket #

rate_limit is coarse and per-worker; a shared, precise limit needs a token-bucket gate that every worker consults against the same budget. The bucket refills at the rate the customs API can genuinely sustain and forces a call to await a refill when empty — converting downstream backpressure into a bounded, logged wait instead of a retry storm. Pair it with an asyncio.Semaphore to cap concurrency (open connections) independently of rate (calls per second); the two limits are orthogonal and a saturated TOS usually enforces both. The IMO FAL single-window submissions this gate protects are exactly the calls you cannot afford to have rejected mid-surge.

from __future__ import annotations

import asyncio
import time

import structlog

log = structlog.get_logger("maritime.backpressure")


class AsyncTokenBucket:
    """Paces calls into a slow downstream (customs API / TOS).

    Refills ``rate`` tokens per second up to ``capacity``. Each call must
    acquire a token, awaiting a refill when the bucket runs dry so the
    downstream is never handed more than it can sustain.
    """

    def __init__(self, rate: float, capacity: float) -> None:
        self._rate: float = rate
        self._capacity: float = capacity
        self._tokens: float = capacity
        self._updated: float = time.monotonic()
        self._lock = asyncio.Lock()

    def _replenish(self) -> None:
        now = time.monotonic()
        self._tokens = min(self._capacity, self._tokens + (now - self._updated) * self._rate)
        self._updated = now

    async def acquire(self, tokens: float = 1.0) -> float:
        async with self._lock:
            self._replenish()
            if self._tokens >= tokens:
                self._tokens -= tokens
                return 0.0
            wait_s: float = (tokens - self._tokens) / self._rate
            log.warning("downstream_gate_throttled",
                        wait_s=round(wait_s, 3), tokens_left=round(self._tokens, 2))
            await asyncio.sleep(wait_s)
            self._replenish()
            self._tokens = max(0.0, self._tokens - tokens)
            return wait_s


async def submit_with_gate(
    entry_id: str, bucket: AsyncTokenBucket, sem: asyncio.Semaphore
) -> None:
    waited = await bucket.acquire()            # rate limit
    async with sem:                            # concurrency limit
        log.info("customs_call_admitted", entry_id=entry_id, throttled_s=waited)
        # await customs_client.submit(entry_id)

Step 4 — Monitor queue depth and emit autoscaling signals #

Backpressure without observability is a silent stall. Sample the broker’s queue length on a fixed interval and compare it against the high and low watermarks: crossing the high watermark means arrivals are outpacing drain and the pool should scale out; falling below the low watermark means the surge has cleared and the pool can scale in. Emit the decision as a structured event a Kubernetes HPA (via a custom-metrics adapter) or KEDA Redis scaler consumes — the worker never scales itself, it only reports the signal. Track queue depth as a gauge, not a counter, so an alert can fire on sustained saturation rather than a single spike.

import os

import redis
import structlog

log = structlog.get_logger("maritime.autoscale")

HWM: int = int(os.environ.get("QUEUE_HWM", "5000"))
LWM: int = int(os.environ.get("QUEUE_LWM", "500"))


def emit_scaling_signal(client: redis.Redis, queue: str) -> str:
    depth: int = client.llen(queue)
    if depth >= HWM:
        signal = "scale_out"
    elif depth <= LWM:
        signal = "scale_in"
    else:
        signal = "hold"
    log.info("queue_depth_sampled", queue=queue, depth=depth,
             hwm=HWM, lwm=LWM, signal=signal)
    return signal

Step 5 — Shed to quarantine under sustained overload #

Autoscaling has a ceiling — node quotas, broker connection limits, and the fixed capacity of the customs API mean the pool cannot grow forever. When depth stays above the high watermark for several consecutive intervals, the only remaining control is to shed load: divert the lowest-value documents to the quarantine topic for later replay rather than letting them keep the downstream saturated. Shed by document class, never blindly: a bulk scanned-B/L batch or an archive replay can wait, but a SOLAS VERMAS declaration or a customs-deadline filing must always pass. Quarantine is a deferral, not a drop — the payload is durable and reprocessed once depth falls back under the low watermark.

from __future__ import annotations

from dataclasses import dataclass

import structlog

log = structlog.get_logger("maritime.backpressure")

SHED_AFTER_INTERVALS: int = 3
SHEDDABLE: frozenset[str] = frozenset({"BULK_BL_PDF", "ARCHIVE_REPLAY"})


@dataclass(frozen=True)
class LoadState:
    depth: int
    intervals_over_hwm: int


def should_shed(state: LoadState, document_type: str) -> bool:
    if state.intervals_over_hwm < SHED_AFTER_INTERVALS:
        return False                            # a brief spike is not overload
    if document_type in SHEDDABLE:
        log.warning("load_shed_to_quarantine", document_type=document_type,
                    depth=state.depth, intervals=state.intervals_over_hwm)
        return True
    log.info("shed_exempt", document_type=document_type)   # compliance-critical: never shed
    return False

Edge Cases & Carrier Deviations #

A subtle deviation is the customs gateway that advertises a per-minute quota but enforces it as a per-second token bucket of its own. A naive 30/m dispatcher that releases all thirty calls in the first two seconds of the minute will still be throttled, because the gateway’s internal bucket holds only a few tokens at any instant. Match the downstream’s real shape — a low CUSTOMS_RATE_PER_S with a small burst capacity — rather than trusting the coarse per-minute figure printed in the integration guide.

Verification & Testing #

Assert the two behaviours that carry the most operational risk — the gate serving a burst up to capacity then throttling, and shedding firing only for sheddable classes under sustained overload — with pytest. Drive the async gate with asyncio.run so the throttle path is genuinely exercised.

import asyncio

import pytest

from maritime_ingestion.backpressure import AsyncTokenBucket, LoadState, should_shed


@pytest.fixture
def bucket() -> AsyncTokenBucket:
    return AsyncTokenBucket(rate=5.0, capacity=5.0)


def test_gate_serves_burst_up_to_capacity(bucket: AsyncTokenBucket) -> None:
    async def drain() -> list[float]:
        return [await bucket.acquire() for _ in range(5)]
    assert all(wait == 0.0 for wait in asyncio.run(drain()))   # capacity is free


def test_gate_throttles_beyond_capacity(bucket: AsyncTokenBucket) -> None:
    async def drain() -> float:
        for _ in range(5):
            await bucket.acquire()
        return await bucket.acquire()          # sixth call awaits a refill
    assert asyncio.run(drain()) > 0.0


def test_shed_only_sheddable_classes_after_sustained_overload() -> None:
    hot = LoadState(depth=9000, intervals_over_hwm=4)
    assert should_shed(hot, "BULK_BL_PDF") is True
    assert should_shed(hot, "VERMAS") is False           # compliance-critical, never shed
    brief = LoadState(depth=9000, intervals_over_hwm=1)
    assert should_shed(brief, "BULK_BL_PDF") is False    # a brief spike is not overload

A healthy run emits one JSON line per throttle and shed decision; a gate wait looks like {"event": "downstream_gate_throttled", "wait_s": 0.2, "tokens_left": 0.0, "log_level": "warning", "timestamp": "..."}. Assert on the structured keys, never on a formatted string.

Frequently Asked Questions #

Related #

• Building Celery queues for maritime doc ingestion — the queue, prefetch, and DLQ wiring this flow-control layer sits on top of
• Async Batch Processing Pipelines — the parent topic area defining throughput and backpressure contracts
• Async bulk B/L ingestion with Celery and a Redis DLQ — the bulk path whose low-priority traffic is shed first under overload
• Terminal API Polling Strategies — the read-side counterpart applying the same non-blocking discipline to a TOS
• Document Ingestion & EDI Parsing Workflows — the domain governing ingestion, schema, and resilience contracts

Up: Async Batch Processing Pipelines — the topic area governing asynchronous document throughput.