Sizing a Prefetch Window for Mixed Geometry Payloads

Compute the memory bound from measured bytes per vertex and the deadline bound from measured vertices per second, then take the smaller — a budget that fits in memory but cannot be drained inside the poll interval gets the consumer evicted exactly as if there were no backpressure at all.

This guide sits under Backpressure & Flow Control for Spatial Consumers, within Queue Management, Retries & Delivery Guarantees. It produces the VERTEX_BUDGET enforced by Applying Backpressure When a Spatial Consumer Falls Behind.

When to use this pattern

  • The budget in production was a round number somebody chose, which is the usual case.
  • The consumer is either evicted under load or never approaches its memory limit — the two symptoms of the two bounds being wrong in opposite directions.
  • The payload mix has changed since the number was set.

Two bounds, and either can bind

Which constraint binds depends on the payload mix Two independent bounds on the vertex budget are plotted against the proportion of heavy geometry in the stream. The memory bound is flat: a container with two gigabytes available for in-flight geometry, at a measured one hundred and forty bytes per vertex, permits about fourteen million vertices in flight regardless of what shape they are in. The poll-deadline bound falls steeply as the mix gets heavier: with a thirty-second deadline and a target of using half of it, a consumer processing eight hundred thousand vertices per second on point-dominated traffic can hold twelve million, but the same consumer processing complex polygons at one hundred and twenty thousand vertices per second can hold only one point eight million. The two curves cross around a mix of fifteen per cent heavy geometry. Below that crossing the memory bound is the binding one; above it the deadline bound is, and by a wide margin. A budget chosen from memory alone is therefore correct for a light stream and catastrophically wrong for a heavy one, which is exactly the direction in which streams drift as producers add richer data. vertex budget permitted by each constraint, against payload mix 14 M 0 0% heavy 50% heavy memory bound — 2 GB ÷ 140 bytes/vertex ≈ 14 M, flat poll-deadline bound — 15 s of headroom × measured vertices/second crossover ≈ 15% heavy memory binds the deadline binds, by a wide margin A budget chosen from memory alone is correct for a light stream and catastrophically wrong for a heavy one — which is exactly the direction streams drift as producers add richer data. Compute both and take the smaller.
Figure 1. The two constraints are not two views of the same number. They cross, and which one binds changes as the stream's content changes underneath a budget nobody revisits.

Complete runnable implementation

python 
import gc
import json
import statistics
import time
import tracemalloc
from dataclasses import dataclass

from shapely.geometry import shape

# Fraction of the broker's poll interval the in-flight work may occupy.
# The rest is headroom for a slow message, a GC pause and the poll itself.
DEADLINE_HEADROOM = 0.5


@dataclass(frozen=True, slots=True)
class BudgetRecommendation:
    bytes_per_vertex: float
    vertices_per_second: float
    memory_bound: int
    deadline_bound: int
    budget: int
    binding: str


def measure_bytes_per_vertex(payloads: list[bytes]) -> float:
    """Resident bytes per vertex, measured on real payloads.

    Not 16 bytes for two doubles: the geometry exists as JSON text, then as
    Python lists and floats, then as a shapely array, and the intermediate
    representations coexist during parsing.
    """
    gc.collect()
    tracemalloc.start()
    geoms, vertices = [], 0
    for raw in payloads:
        geom = shape(json.loads(raw)["geometry"])
        geoms.append(geom)
        vertices += _count_vertices(geom)
    _, peak = tracemalloc.get_traced_memory()
    tracemalloc.stop()
    del geoms
    return peak / max(1, vertices)


def measure_vertices_per_second(payloads: list[bytes], handler,
                                repeats: int = 3) -> float:
    """Throughput on the SAME payloads the budget will govern.

    A figure taken from point-ping traffic overstates a polygon workload by
    an order of magnitude, and the budget derived from it evicts the consumer.
    """
    rates = []
    for _ in range(repeats):
        vertices = 0
        start = time.perf_counter()
        for raw in payloads:
            geom = shape(json.loads(raw)["geometry"])
            handler(geom)
            vertices += _count_vertices(geom)
        rates.append(vertices / (time.perf_counter() - start))
    # Median, not mean: one GC pause during a run should not set the budget.
    return statistics.median(rates)


def recommend(payloads: list[bytes], handler,
              memory_bytes_available: int,
              max_poll_interval_seconds: float) -> BudgetRecommendation:
    bpv = measure_bytes_per_vertex(payloads)
    vps = measure_vertices_per_second(payloads, handler)

    memory_bound = int(memory_bytes_available / bpv)
    deadline_bound = int(vps * max_poll_interval_seconds * DEADLINE_HEADROOM)

    budget = min(memory_bound, deadline_bound)
    return BudgetRecommendation(
        bytes_per_vertex=bpv,
        vertices_per_second=vps,
        memory_bound=memory_bound,
        deadline_bound=deadline_bound,
        budget=budget,
        binding="memory" if memory_bound < deadline_bound else "poll deadline",
    )


def _count_vertices(geom) -> int:
    if hasattr(geom, "geoms"):
        return sum(_count_vertices(g) for g in geom.geoms)
    if geom.geom_type == "Polygon":
        return len(geom.exterior.coords) + sum(len(r.coords) for r in geom.interiors)
    return len(getattr(geom, "coords", ()))

Reporting which bound binds is the part that makes the recommendation actionable. A budget limited by memory is fixed by a bigger container; one limited by the deadline is fixed by faster processing or a longer max.poll.interval.ms, and those are different tickets for different teams.

Measuring on the wrong traffic gives a confidently wrong number A budget is derived twice for the same consumer. The first measurement uses a sample taken at three in the morning, which is almost entirely point pings from parked vehicles: the handler processes about nine hundred thousand vertices per second because there is nothing to do per geometry, and the resulting deadline bound is thirteen and a half million vertices. Deployed, that budget lets the consumer accept a batch of parcel boundaries whose processing rate is closer to one hundred thousand vertices per second, so draining thirteen and a half million takes over two minutes against a thirty-second deadline — the member is evicted, the batch is redelivered, and the backpressure mechanism has made no difference at all. The second measurement uses a sample from the busiest hour, containing the real mix of point pings and polygons: the measured rate is one hundred and forty thousand vertices per second and the budget is two point one million, which drains in fifteen seconds. The mechanism was correct in both cases; only the input to the sizing differed, and the failure it produced is indistinguishable from having no backpressure at all. same consumer, same mechanism, two sizing samples sample taken at 03:00 — parked-fleet pings measured rate: 900 000 vertices/s deadline bound: 13.5 M vertices deployed against real polygon traffic at 100 000 vertices/s → over 2 minutes to drain evicted at 30 s · batch redelivered indistinguishable from having no backpressure sample taken during the busiest hour the real mix: pings and parcel boundaries measured rate: 140 000 vertices/s deadline bound: 2.1 M vertices drains in 15 s · half the deadline, as designed the mechanism was identical in both cases — only the sizing input differed Measure on the traffic the budget will govern, at the hour it will govern it. A benchmark run on convenient data produces a number with a decimal point and no relationship to the workload.
Figure 2. A wrongly-sized budget does not fail visibly — it produces exactly the eviction the backpressure was added to prevent, which sends the investigation back to the mechanism.

Parameter reference

Name Type Spatial constraint Default
DEADLINE_HEADROOM float Fraction of the poll interval the budget may occupy 0.5
memory_bytes_available int For in-flight geometry only, not the whole container —
max_poll_interval_seconds float The broker’s own setting, read from configuration 30.0
Sample list Real payloads from the busiest hour, not a convenient one —
Rate statistic — Median across repeats; a GC pause must not set the budget median
binding str Reported, because the two bounds need different fixes —

Gotchas and spatial edge cases

  1. tracemalloc measures Python allocations, not the whole process. Shapely’s geometry arrays live partly in native memory that tracemalloc does not see, so the measured bytes per vertex is a lower bound. Cross-check against the container’s resident set under load, and prefer the larger figure — an over-cautious budget costs throughput, an over-confident one costs the container.

  2. The handler must be the real one. Measuring parse rate alone gives a figure several times higher than a handler that also does a spatial join, writes to PostGIS or reprojects. The budget governs the whole pipeline stage, so the measurement has to as well.

  3. A sample of large geometries alone is as wrong as a sample of small ones. The budget governs the mix, and a stream that is ninety per cent pings by count but sixty per cent polygon vertices by volume has a rate somewhere between the two extremes. Sample by taking a contiguous window of real traffic rather than by selecting interesting messages.

  4. Vertex count and processing cost decouple for some handlers. A handler dominated by a per-message database round trip has a cost proportional to message count, not vertices, so a vertex budget lets through thousands of cheap messages that collectively blow the deadline. Where that is the case, charge both and bound both.

  5. Re-measure after any change to the deployment shape. Halving the container’s memory halves the memory bound; moving from four processes to eight halves the memory available to each while leaving the deadline bound unchanged, which can flip which constraint binds.

  6. The measurement must run outside the event loop. Timing a handler that awaits on a shared loop measures scheduling, not processing. Use the same isolation the production path uses, which for geometry work means a process pool — see Optimizing Async Geometry Parsing with asyncio.

Doubling the process count can flip which constraint binds A consumer runs four processes in a container with eight gigabytes available for in-flight geometry, so each process has two gigabytes and a memory bound of about fourteen million vertices — comfortably above the deadline bound of two point one million, which is therefore the binding constraint. Someone doubles the process count to improve throughput. The container's memory is unchanged, so each process now has one gigabyte and a memory bound of about seven million vertices; the deadline bound is unchanged at two point one million per process because each still has a full poll interval. In this case the deadline still binds and nothing breaks. Halve the memory again, to sixteen processes, and the memory bound falls to about one point seven million — below the deadline bound — so memory becomes the binding constraint and a budget still configured at two point one million now permits more in-flight geometry than the process can hold. The scaling change did not touch the budget, the code or the traffic, and it invalidated the sizing. 8 GB container for in-flight geometry, split N ways 4 processes · 2 GB each memory bound ≈ 14 M vertices deadline bound ≈ 2.1 M the deadline binds 8 processes · 1 GB each memory bound ≈ 7 M vertices deadline bound ≈ 2.1 M, unchanged the deadline still binds — nothing breaks 16 processes · 0.5 GB each memory bound ≈ 1.7 M vertices deadline bound ≈ 2.1 M memory binds — the budget is now too large The scaling change touched neither the budget, the code, nor the traffic — and it invalidated the sizing. Recompute both bounds whenever the deployment shape changes, not only when the payload mix does.
Figure 3. Scaling out is usually treated as free. For a memory-bounded budget it is a change to one of the two inputs, and only one of them.

Verification

python 
import pytest


def test_the_smaller_bound_wins():
    rec = recommend(_mixed_sample(), _real_handler,
                    memory_bytes_available=2 * 1024**3,
                    max_poll_interval_seconds=30.0)
    assert rec.budget == min(rec.memory_bound, rec.deadline_bound)
    assert rec.binding in ("memory", "poll deadline")


def test_heavy_sample_produces_a_smaller_budget_than_a_light_one():
    """The property that makes the sample choice matter."""
    light = recommend(_point_pings(), _real_handler, 2 * 1024**3, 30.0)
    heavy = recommend(_parcel_boundaries(), _real_handler, 2 * 1024**3, 30.0)
    assert heavy.budget < light.budget


def test_budget_drains_inside_the_deadline():
    """The assertion the deadline bound exists to guarantee."""
    rec = recommend(_mixed_sample(), _real_handler, 2 * 1024**3, 30.0)
    drain_seconds = rec.budget / rec.vertices_per_second
    assert drain_seconds <= 30.0 * DEADLINE_HEADROOM + 0.5


def test_bytes_per_vertex_is_not_the_naive_figure():
    """16 bytes for two doubles would size the budget an order of magnitude high."""
    bpv = measure_bytes_per_vertex(_mixed_sample())
    assert bpv > 40, f"{bpv:.1f} bytes/vertex is implausibly low — check the measurement"

The last test guards the measurement rather than the result. If measure_bytes_per_vertex is ever changed in a way that reports the serialised size instead of the resident size, every derived budget becomes several times too large and nothing else in the suite notices.