Backpressure & Flow Control for Spatial Streams

Backpressure for a spatial stream has to be measured in work rather than in messages, because a single partition can carry point pings costing two milliseconds and multipolygons costing nine hundred — so a prefetch window that is safe for one is three orders of magnitude wrong for the other, and which you receive is decided by geography.

This topic sits under Queue Management, Retries & Delivery Guarantees, which covers how spatial events move from producer to consumer without loss or duplication. Backpressure is the mechanism that keeps a consumer inside its own capacity; when it fails, the symptoms show up as the consumer lag described in Consumer Lag & Partition Skew Monitoring, and the retries it triggers are governed by Exponential Backoff & Jitter for Spatial Webhooks.

Prerequisites

• Python 3.11+ — asyncio.TaskGroup and timeout simplify the bounded-concurrency code below
• A broker with pause/resume — aiokafka’s pause()/resume(), or Redis Streams consumer groups where you control the read loop
• Vertex count in the envelope — precomputed at ingest, so the consumer can estimate cost without parsing the geometry
• A process pool for heavy geometry — as described in Async Processing for Geometry-Heavy Payloads
• Per-partition lag metrics — the saturation signal that drives pausing
• A documented shedding policy — which topics may be shed, and which may never be

Why message-count prefetch fails here

Every broker’s flow control is expressed in messages: max.poll.records, prefetch_count, COUNT on XREADGROUP. That unit assumes messages are roughly interchangeable in cost, which is true of order events and false of spatial ones.

The eviction in that third row is the part that turns a slow consumer into an outage. Kafka expects a poll() within max.poll.interval.ms; a consumer that is busy rather than polling is presumed dead, its partitions are reassigned, and the batch it was halfway through is redelivered to another member — which is holding the same prefetch setting and meets the same fate.

Architecture: bound the work, not the count

The fix is to keep the broker’s own window small and enforce a second, work-aware bound inside the consumer. Vertex count, precomputed at ingest and carried in the envelope, is a good cost proxy: it is cheap to read, correlates well with shapely runtime, and does not require deserialising the geometry to obtain.

Layer breakdown:

1. Small broker window — set max.poll.records low (10–25) so the broker never hands you an unbounded amount of unknown work in one call.
2. Vertex budget — a semaphore denominated in vertices rather than tasks. Each message is charged its envelope’s vertex count before dispatch and refunded on completion.
3. Pause on exhaustion — when the budget cannot admit the next message, pause the partition and keep polling. This is the critical detail: a paused consumer still polls, so the session timer is satisfied.
4. Shed on sustained pressure — if the pause has not cleared within a bounded window, drop self-superseding events by geographic priority rather than accumulating a backlog that will never drain.

Step-by-step implementation

Step 1 — Charge work against a budget, not tasks against a counter

import asyncio
from dataclasses import dataclass


class VertexBudget:
    """A semaphore denominated in vertices rather than in tasks.

    A plain asyncio.Semaphore(N) bounds the *number* of in-flight messages,
    which is exactly the unit that tells us nothing here. This bounds the work.
    """

    def __init__(self, capacity: int) -> None:
        self._capacity = capacity
        self._available = capacity
        self._cond = asyncio.Condition()

    async def acquire(self, cost: int) -> None:
        # A single geometry larger than the whole budget would deadlock, so it
        # is clamped: it runs alone, which is the correct behaviour for a
        # payload that genuinely exceeds our capacity estimate.
        cost = min(cost, self._capacity)
        async with self._cond:
            while self._available < cost:
                await self._cond.wait()
            self._available -= cost

    async def release(self, cost: int) -> None:
        cost = min(cost, self._capacity)
        async with self._cond:
            self._available += cost
            self._cond.notify_all()

    @property
    def saturated(self) -> bool:
        return self._available <= 0

The clamp in acquire matters. Without it, a single 41,000-vertex multipolygon arriving at a budget of 40,000 waits forever for capacity that can never exist, and the consumer stops with no error.

Step 2 — Pause the partition instead of blocking the poll loop

from aiokafka import AIOKafkaConsumer
from aiokafka.structs import TopicPartition

BUDGET = VertexBudget(capacity=60_000)


async def consume(consumer: AIOKafkaConsumer, pool) -> None:
    paused: set[TopicPartition] = set()

    while True:
        batches = await consumer.getmany(timeout_ms=500, max_records=25)

        for tp, messages in batches.items():
            for msg in messages:
                cost = msg.headers_dict.get("vertex_count", 40)

                if BUDGET.saturated and tp not in paused:
                    # Pause, but keep polling. getmany() continues to be called
                    # and simply returns nothing for this partition, so the
                    # broker's session timer stays satisfied and we are not
                    # evicted from the group mid-batch.
                    consumer.pause(tp)
                    paused.add(tp)

                await BUDGET.acquire(cost)
                asyncio.create_task(_process(pool, msg, cost, consumer, tp, paused))

        for tp in list(paused):
            if not BUDGET.saturated:
                consumer.resume(tp)
                paused.discard(tp)

Step 3 — Refund on completion, including on failure

async def _process(pool, msg, cost, consumer, tp, paused) -> None:
    try:
        loop = asyncio.get_running_loop()
        await loop.run_in_executor(pool, _repair_geometry, msg.value)
    except Exception:
        # A failed message still returns its budget. Forgetting this is a slow
        # leak: every failure permanently shrinks capacity until the consumer
        # stalls at a budget of zero with no message in flight.
        raise
    finally:
        await BUDGET.release(cost)

The finally is the whole reliability story of this component. A refund that happens only on success leaks capacity on every failure, and the consumer degrades over hours into a stall that looks nothing like the geometry error that caused it.

Step 4 — Shed by geographic priority, and only where it is safe

SHEDDABLE_TOPICS = {"vehicle-positions", "sensor-telemetry"}

async def should_shed(msg, pause_seconds: float) -> bool:
    """Shed only self-superseding events, and only under sustained pressure."""
    if pause_seconds < 30:
        return False
    if msg.topic not in SHEDDABLE_TOPICS:
        # Cadastral edits, boundary changes, tile invalidations: nothing will
        # resend these, so dropping one is data loss rather than degradation.
        return False
    # Within a sheddable topic, drop the lowest-priority regions first.
    return msg.headers_dict.get("region_priority", 0) < 2

Spatial validation and error handling

Vertex count is a proxy, and proxies drift. A 2,000-vertex polygon with many interior rings can cost more to repair than a 5,000-vertex simple one, because topology repair is not linear in vertex count. Treat the budget as approximate and set it against measured p99 rather than against a computed ideal; the point is to bound the error, not to eliminate it.

A missing vertex count must default high, not low. If an envelope arrives without the field — an old schema version, a producer that has not been updated — defaulting to a small cost lets an unbounded payload through the one control designed to stop it. Default to the budget’s clamp value so an unmeasured payload is treated as expensive until proven otherwise.

Shedding must never apply to a stream with ordering guarantees. Dropping one event from a version-guarded stream is survivable, because a later version supersedes it. Dropping one from a stream whose consumer applies deltas is corruption, because the state never converges. Enumerate sheddable topics explicitly, as above, rather than deriving the decision from a priority field alone.

Retry, backoff and delivery guarantees

Backpressure and retries push in opposite directions on the same pipe, and under saturation they can amplify each other. A saturated consumer begins timing out; those timeouts generate retries; the retries arrive as additional load on the consumer that was already saturated. This is the mechanism by which a slow consumer becomes an unavailable one.

Gate retries on the same saturation signal that drives pausing. When the budget is exhausted, the retry admission check should fail first, so a consumer under pressure sheds redelivered work before it sheds fresh work — a retry is by definition something the system has already attempted, while a fresh event is something it has not yet seen at all. The token-bucket budget in Tuning Retry Budgets for Webhook Provider SLAs is the right place to apply that gate.

Verification

The test that matters is a burst of heavy geometry against a running consumer, asserting that the broker session survives.

import asyncio
import pytest


@pytest.mark.asyncio
async def test_heavy_burst_does_not_exceed_poll_interval(consumer, pool, clock):
    """A burst of large geometries must not stall the poll loop.

    The failure this guards against is subtle: the consumer keeps working and
    still gets evicted, because it stopped polling while it worked.
    """
    heavy = [_message(vertex_count=41_000) for _ in range(50)]
    await _produce(heavy)

    gaps = []
    async for gap in _poll_gaps(consumer, pool, duration=60):
        gaps.append(gap)

    assert max(gaps) < 300, f"poll gap {max(gaps):.1f}s exceeds max.poll.interval.ms"


@pytest.mark.asyncio
async def test_budget_is_refunded_on_failure():
    """The gate's negative case: a failing message must return its budget.

    Without this assertion the leak is invisible — capacity shrinks slowly and
    the consumer stalls hours later, long after the failing payload is gone.
    """
    budget = VertexBudget(capacity=1000)
    await budget.acquire(600)
    with pytest.raises(RuntimeError):
        try:
            raise RuntimeError("topology repair failed")
        finally:
            await budget.release(600)
    assert not budget.saturated

Troubleshooting

FAQ

Related

• Queue Management, Retries & Delivery Guarantees — the section this topic belongs to
• Consumer Lag & Partition Skew Monitoring — the saturation signal that drives pausing, and how to tell skew from genuine overload
• Async Processing for Geometry-Heavy Payloads — the process pool the budget dispatches into
• Tuning Retry Budgets for Webhook Provider SLAs — where to gate retries on the saturation signal
• Spatial Partitioning Strategies — why one partition ends up carrying the heavy geometry in the first place

Explore this section