Event Routing & Backpressure for Real-Time Geofence Triggers

A containment evaluation is only half of a geofencing system. The moment a device crosses a perimeter and the evaluator emits an ENTER, EXIT, or DWELL trigger, that event enters a second pipeline whose failures are quieter but more expensive than a slow point-in-polygon check: a dropped trigger is a compliance event that never fired, a duplicated trigger is a surge fare charged twice, a reordered pair is a phantom transition where a vehicle appears to exit a zone it never entered, and an unbounded buffer under a telemetry burst is an out-of-memory kill that takes the whole node down mid-surge. Routing triggers reliably at 50k+ events/sec is a distributed-systems problem layered on top of a real-time one: you must publish, partition, deduplicate, and acknowledge every trigger inside a tight delivery budget while shedding load safely when downstream consumers fall behind. This page is the reference for the backend and platform engineers who own that downstream half — it fixes the routing topology, the per-stage delivery budget, the throughput math behind producer batching and partition sizing, and the memory, async, and failure-handling invariants that keep trigger semantics intact when the queue fills. It is the routing counterpart to core architecture and latency constraints, which governs everything upstream of the emit boundary.

Routing Topology at a Glance #

The topology has exactly two escape hatches, and both are deliberate. When the routing substrate cannot absorb the publish rate — a broker partition leader election, a slow downstream consumer group building lag — the bounded queue in front of the producer fills, and its depth thresholds convert that pressure into a signaled decision instead of silent heap growth. When a specific trigger cannot be processed no matter how many times it is retried — a malformed payload, a fence id that no longer exists, a schema mismatch — it branches to a dead-letter topic so one poison message never stalls the partition behind it. Every design choice below exists to keep those two paths deterministic.

Pipeline Partitioning & Delivery SLA #

The routing pipeline must be partitioned into measurable stages for the same reason the evaluation pipeline is: end-to-end delivery time hides which stage regressed. If you only measure the gap between trigger emission and downstream acknowledgement, a producer stuck on linger.ms looks identical to a consumer stuck on a rebalance, and you cannot enforce a per-stage contract. Trigger emission — building the event object, computing the idempotency key, serializing to a binary frame — should cost well under 2ms at P95; anything slower usually means JSON serialization on the hot path or a key derivation that hashes more state than it needs to. The idempotency and dedup gate is a single in-memory lookup and must stay under 2ms even at 50k/sec, which is only achievable if the lookup is $O(1)$ (covered below). Broker publish is the stage with real variance: with acknowledgement from the partition leader and one in-sync replica, a healthy Kafka cluster holds P95 broker publish under 6ms, but it spikes hard during leader elections, so the hard timeout must be generous enough to ride out a sub-second election without shedding good traffic.

The allocation below is the contract a 50k/sec routing tier holds itself to. Each stage owns a hard ceiling, and the sum of ceilings sits below the end-to-end delivery target so that broker jitter and consumer rebalances have budget to live in.

The load-bearing number is the composite: end-to-end trigger delivery P99 must stay under 40ms at 50k/sec, because the trigger usually feeds a system with its own deadline — a dynamic-pricing engine that recomputes a surge multiplier, a compliance checker with a regulatory response window, a driver-app push that feels broken if it lags a physical crossing by more than a beat. Enforcement is mechanical: each stage records a time.perf_counter_ns delta into a Prometheus histogram, an aggregator computes rolling P95/P99 per stage, and a stage that breaches its ceiling for three consecutive scrape intervals trips the circuit breaker guarding the next stage. This is the routing analogue of the phase-level error budget described in core architecture and latency constraints, and production monitoring and observability details the exact metric families and alert thresholds that make the table above observable rather than aspirational.

A second, subtler contract is ordering. Trigger semantics are stateful — an EXIT is only meaningful after the matching ENTER — so triggers for a single device must never be reordered even as the system parallelizes across 50k of them. The only reliable mechanism is partition affinity: partition the routing substrate by device_id so that every trigger for one device lands on the same partition and is consumed by one member of the consumer group in offset order. Get the partitioning key wrong and you reintroduce phantom transitions that no amount of downstream logic can fully repair.

Algorithmic Throughput & Routing Math #

Sustained throughput is a batching and partitioning problem before it is a hardware problem. A producer that publishes one record per network round trip is bottlenecked by latency, not bandwidth; batching amortizes the round-trip and TLS framing cost across many triggers. With a batch size $B$ bytes, a mean serialized trigger size $s$ bytes, and a linger window $t_{\ell }$ seconds, a single producer connection sustains

$${\lambda }_{\text{prod}}\approx \frac{1}{t_{\ell }}\cdot \frac{B}{s}\text{records/sec, capped by}\frac{B}{s}\text{ records per batch.}$$

A compact trigger serializes to roughly $s\approx 200$ bytes, so a 16KB batch (batch.size = 16384) carries about 80 triggers, and with linger.ms = 5 a single producer flushes ~80 records every 5ms — around 16k/sec per connection before compression, and materially more once lz4 compression shrinks the on-wire batch. This is why the reference below batches aggressively: dropping linger to 0 for latency actually lowers peak throughput because it defeats batching, so the tuning is a deliberate trade-off explored in tuning Kafka producer batching for geofence triggers.

Partition count sets the ceiling on consumer parallelism. Total sustained throughput is bounded by

$${\lambda }_{\text{total}}=\min ({\lambda }_{\text{prod}}\cdot P_{\text{conn}},;P_{\text{part}}\cdot {\lambda }_{\text{part}}),$$

where $P_{\text{part}}$ is the partition count and ${\lambda }_{\text{part}}$ is the per-partition consume rate. A single partition on commodity hardware sustains ~80k triggers/sec/partition with 16KB linger batches and lz4 compression, so a 50k/sec workload is comfortably served by a handful of partitions — but you over-provision partitions (typically 12–24) not for raw throughput but so the consumer group has room to scale members and rebalance without a single member becoming a hot spot. The choice of substrate itself — a partitioned Kafka log versus a Redis Stream consumer group — trades durability, replay depth, and operational weight, and is the subject of Kafka vs Redis Streams for trigger routing.

Deduplication throughput hinges on the lookup being constant time. The dedup gate answers “have I already emitted this exact transition?” for every trigger, so an $O(n)$ or even $O(\log n)$ structure would dominate the 2ms budget at 50k/sec. A hash set keyed on the idempotency key is $O(1)$ amortized; when the dedup window grows large enough that the set’s memory footprint matters, a Bloom filter trades a tunable false-positive rate for a fixed bit array, with

$$p_{\text{fp}}\approx (1-e^{-kn/m}{)}^k,$$

for $k$ hash functions, $n$ inserted keys, and $m$ bits — so a 1% false-positive target at $n=10^6$ keys needs roughly $m\approx 9.6$ Mbit and $k=7$, which is a few megabytes for a million-trigger window versus tens of megabytes for a full hash set. A false positive here means a possibly duplicate trigger is double-checked against durable state, never silently dropped, so the filter sits in front of an authoritative check rather than replacing it. The exact-once versus at-least-once trade-off this enables is developed in idempotent trigger emission semantics.

Implementation Reference #

The following is a production-shaped async trigger router built on aiokafka, with idempotency keys, a bounded asyncio.Queue as the backpressure valve, dead-letter routing for poison messages, and a circuit breaker between the router and a slow downstream. The inline comments flag the non-obvious decisions; the geometry and evaluation that produced each trigger live upstream, and this component’s only job is to route it safely.

import asyncio
import time
import logging
from dataclasses import dataclass
from enum import Enum
from typing import Final

from aiokafka import AIOKafkaProducer  # aiokafka >= 0.10

logger = logging.getLogger(__name__)

_DEDUP_WINDOW: Final[int] = 1_000_000  # bounded: cap keys before eviction


class TriggerState(Enum):
    ENTER = "enter"
    EXIT = "exit"
    DWELL = "dwell"


@dataclass(slots=True, frozen=True)  # __slots__ + frozen: no per-instance __dict__, hashable
class GeofenceTrigger:
    device_id: str
    fence_id: str
    state: TriggerState
    event_time_ms: int   # device event time -> ordering + idempotency bucket
    emitted_ns: int      # monotonic emit time -> delivery SLA measurement

    def idempotency_key(self) -> str:
        # Deterministic: identical crossings collapse to one key regardless of retries.
        bucket = self.event_time_ms // 1000  # 1s dedup bucket tolerates clock jitter
        return f"{self.device_id}:{self.fence_id}:{self.state.value}:{bucket}"


class RouterShedError(Exception):
    """Raised when the bounded queue is saturated and the trigger must go to the DLQ."""


class _CircuitBreaker:
    __slots__ = ("_fail_threshold", "_reset_after_s", "_failures", "_opened_at")

    def __init__(self, fail_threshold: int = 20, reset_after_s: float = 5.0) -> None:
        self._fail_threshold = fail_threshold
        self._reset_after_s = reset_after_s
        self._failures = 0
        self._opened_at = 0.0

    def allow(self) -> bool:
        if self._failures < self._fail_threshold:
            return True
        # Half-open probe once the cooldown elapses, else stay open and shed.
        if time.monotonic() - self._opened_at >= self._reset_after_s:
            self._failures = self._fail_threshold - 1  # allow a single trial publish
            return True
        return False

    def record(self, ok: bool) -> None:
        if ok:
            self._failures = 0
        else:
            self._failures += 1
            if self._failures == self._fail_threshold:
                self._opened_at = time.monotonic()


class TriggerRouter:
    def __init__(
        self,
        producer: AIOKafkaProducer,
        topic: str,
        dlq_topic: str,
        queue_capacity: int = 100_000,
        pause_ratio: float = 0.70,
        shed_ratio: float = 0.90,
    ) -> None:
        self._producer = producer
        self._topic = topic
        self._dlq_topic = dlq_topic
        # Bounded queue IS the backpressure signal; unbounded only defers OOM.
        self._queue: asyncio.Queue[GeofenceTrigger] = asyncio.Queue(maxsize=queue_capacity)
        self._cap = queue_capacity
        self._pause_ratio = pause_ratio
        self._shed_ratio = shed_ratio
        self._seen: set[str] = set()  # O(1) dedup; bounded by _DEDUP_WINDOW below
        self._breaker = _CircuitBreaker()
        self._metrics: dict[str, int] = {"routed": 0, "deduped": 0, "shed": 0, "dlq": 0}

    async def submit(self, trigger: GeofenceTrigger) -> None:
        # Load-shed at the boundary: never block the producer of good triggers.
        depth = self._queue.qsize()
        if depth >= self._cap * self._shed_ratio:
            self._metrics["shed"] += 1
            raise RouterShedError(trigger.idempotency_key())
        await self._queue.put(trigger)

    def _dedup(self, key: str) -> bool:
        if key in self._seen:  # amortized O(1) membership test
            self._metrics["deduped"] += 1
            return True
        if len(self._seen) >= _DEDUP_WINDOW:
            self._seen.clear()  # coarse eviction; a TTL ring buffer is the refinement
        self._seen.add(key)
        return False

    async def _publish(self, trigger: GeofenceTrigger) -> None:
        key = trigger.idempotency_key()
        if self._dedup(key):
            return  # exact replay suppressed before it ever hits the broker
        partition_key = trigger.device_id.encode()  # affinity: per-device ordering
        payload = self._serialize(trigger)           # binary frame, not JSON
        target = self._topic if self._breaker.allow() else self._dlq_topic
        try:
            # aiokafka batches internally; await resolves on broker ack (acks="all").
            await asyncio.wait_for(
                self._producer.send_and_wait(target, payload, key=partition_key),
                timeout=0.025,  # hard publish timeout from the SLA table
            )
            self._breaker.record(ok=True)
            self._metrics["routed" if target is self._topic else "dlq"] += 1
        except (asyncio.TimeoutError, Exception) as exc:  # noqa: BLE001
            self._breaker.record(ok=False)
            logger.error("publish failed for %s: %s", key, exc)
            await self._to_dlq(payload, reason=str(exc))

    async def _to_dlq(self, payload: bytes, reason: str) -> None:
        try:
            await self._producer.send_and_wait(self._dlq_topic, payload)
            self._metrics["dlq"] += 1
        except Exception:  # noqa: BLE001 - DLQ is best-effort; drop-log only past here
            logger.critical("DLQ publish failed; trigger lost, reason=%s", reason)

    async def run(self) -> None:
        while True:
            trigger = await self._queue.get()
            try:
                await self._publish(trigger)
            finally:
                self._queue.task_done()
                await asyncio.sleep(0)  # yield so the drain loop cannot starve the scheduler

    @staticmethod
    def _serialize(trigger: GeofenceTrigger) -> bytes:
        # Placeholder for a real binary codec (protobuf / msgpack).
        return trigger.idempotency_key().encode()

    def qsize(self) -> int:
        return self._queue.qsize()

    def metrics(self) -> dict[str, int]:
        return dict(self._metrics)

Three decisions carry the architectural weight. First, submit refuses work at the boundary once the queue crosses the shed ratio and raises RouterShedError rather than awaiting an unbounded put — that exception is the backpressure contract, letting the caller divert to a dead-letter path instead of silently inflating the heap. Second, the dedup set is consulted before the broker send, so an exact replay never consumes broker bandwidth or a partition offset. Third, the circuit breaker sits between the router and the broker: when publishes fail repeatedly it flips the target to the dead-letter topic and periodically half-opens to probe recovery, which is expanded in circuit breakers for downstream trigger consumers.

Deterministic Memory & Cache Locality #

The router allocates on the hottest path in the system — once per trigger, 50k times a second — so per-message allocation discipline is not optional. The GeofenceTrigger dataclass is declared slots=True, frozen=True: __slots__ drops the per-instance __dict__, cutting each trigger’s footprint by roughly 40% and, more importantly, removing a dictionary allocation from every emission; frozen=True makes instances hashable so they can key a dedup structure directly. The idempotency key is a short string derived arithmetically, not a serialized blob, so key computation allocates one small string rather than copying the trigger’s state.

The dedup window is the one structure that grows without bound if left alone, and an unbounded set of a million-plus keys is both a memory leak and a source of the exact garbage-collection pauses the delivery budget cannot absorb. The reference caps it at _DEDUP_WINDOW and clears coarsely; the production refinement is a TTL-bounded ring buffer or an LRU keyed on the same idempotency key, so the window covers the plausible replay horizon (a few seconds of retries) and no more. This mirrors the memory-constrained spatial processing discipline on the evaluation side: bound every structure whose size is driven by traffic, and reclaim on a schedule you control rather than one the collector picks for you.

Garbage-collection discipline is explicit. CPython’s collector introduces non-deterministic pauses once object churn exceeds ~10k allocations/sec, and a single multi-millisecond gen-2 pause inside the publish loop blows the P99 ceiling for every trigger queued behind it. During sustained publish, call gc.freeze() after warm-up to move long-lived router state out of the collector’s scan set, and consider gc.disable() with a manual gc.collect() during idle troughs so collection never lands mid-burst. Because the batching producer holds serialized frames until flush, keeping those frames compact and short-lived — serialize late, let the batch own the bytes — keeps the collector’s working set small.

Async Execution & Queue Semantics #

Routing lives entirely on the event loop, so event-loop safety is the governing constraint: one blocking call in the drain loop stalls every concurrent publish. The aiokafka producer is natively async and batches internally, so send_and_wait yields to the loop while a batch fills and while the broker acknowledges; the await is the backpressure point where a slow broker naturally slows the drain rate without any thread ever blocking. The bounded queue in front of the drain loop closes the loop back to the producer of triggers: when the broker slows, the drain rate drops, the queue fills toward its thresholds, and submit begins shedding — a clean, signaled chain from broker pressure to load-shed decision with no unbounded buffer anywhere in between. Sizing that queue is a real trade-off between burst absorption and delivery latency, worked through in backpressure and flow-control strategies.

The circuit breaker is the second async safety mechanism, and it belongs between the router and its slow dependency, not at the edge. When a downstream consumer group builds lag or a broker partition goes unavailable, a breaker that trips after a failure threshold sheds to the dead-letter topic at a defined boundary instead of letting every publish burn its full 25ms timeout and propagate backpressure all the way to ingress. Its half-open state periodically admits a single trial publish so the system recovers automatically once the dependency heals, rather than requiring a manual reset. That decision to fail toward the dead-letter path rather than block is the safest default under load, and the poison-message handling it depends on is detailed in dead-letter topics and poison message handling.

Delivery guarantees are the semantic layer on top of all this. At-least-once delivery — commit consumer offsets only after downstream work succeeds — is the pragmatic default, and it is safe precisely because emission is idempotent: a redelivered ENTER collapses against its idempotency key and never double-charges a surge fare. Exactly-once is achievable with transactional producers and read-committed consumers, but it costs throughput and operational complexity that most geofencing workloads do not need once idempotent emission is in place; the trade-off is quantified in idempotent trigger emission semantics. The invariant is that reordering-sensitive triggers stay partition-affine by device_id, so at-least-once redelivery replays in the original order and the dedup gate absorbs the duplicates.

Operational Debugging Runbook #

When the routing tier breaches its delivery budget or consumer lag climbs, work the stages in order — the fault is almost always isolated to one row of the SLA table.

1. Profile the drain loop. Attach py-spy dump --pid <pid> for a non-intrusive stack snapshot, then py-spy record -o router.svg for a flame graph. A profile dominated by send_and_wait wait time points to broker or downstream slowness, not router CPU; a profile dominated by serialization or key derivation points to the emit stage. Trace time.perf_counter_ns deltas across emit, dedup, and publish to localize the regression against the budget table.
2. Detect queue saturation. Poll router.qsize() on a 1s interval. If depth crosses the 70% pause ratio and keeps climbing toward the 90% shed ratio, the drain loop cannot keep up — confirm the shed metric is rising, which is correct load-shedding, and check whether the bottleneck is broker publish latency or a lagging consumer group before adding partitions.
3. Measure consumer lag directly. Read consumer-group lag from broker metrics (kafka-consumer-groups --describe, or the records-lag-max JMX gauge surfaced through prometheus_client). Rising lag with a healthy producer means downstream consumers are the bottleneck and the circuit breaker should be shedding to the dead-letter topic; flat lag with a full queue means the producer or broker publish is the constraint.
4. Watch the dedup window grow. Run tracemalloc in staging and diff snapshots every 100k triggers. If the top frame is the dedup set and RSS climbs without a matching rise in unique devices, the eviction policy is too coarse — switch the coarse clear() for a TTL ring buffer and re-measure. Unbounded dedup growth is the most common cause of a slow OOM in this tier.
5. Quantify GC pauses. Read gc.get_stats() for per-generation collection counts and correlate gen-2 collections against the P99 delivery histogram. If they line up, apply gc.freeze() after warm-up and target pause times under 2ms; verify against the per-stage histogram, never against averages, since averages hide exactly the tail that breaks the budget.
6. Exercise the failure paths. Force the circuit breaker open by pointing the producer at an unreachable broker and confirm triggers divert to the dead-letter topic rather than blocking; then restore the broker and confirm the half-open probe closes the breaker automatically. Replay a batch of dead-letter triggers and confirm the idempotency gate suppresses the ones already delivered, so replay is safe and never double-fires a compliance alert.

Conclusion #

Routing geofence triggers reliably is a distributed-systems problem with a real-time deadline, and its invariants are consistent across every deployment. Make emission idempotent so retries, redelivery, and dead-letter replay all collapse to a single effect. Keep every traffic-driven buffer bounded — the ingress queue, the dedup window, the producer batch — so pressure becomes a signaled decision instead of an out-of-memory kill. Make backpressure explicit at a defined boundary with concrete depth thresholds rather than letting an unbounded queue defer the failure. Prefer the dead-letter path over blocking whenever a message is poison or a dependency is down, and let a circuit breaker with a half-open probe recover automatically. Keep reordering-sensitive triggers partition-affine by device so at-least-once redelivery replays in order and the dedup gate absorbs the duplicates. Hold those, and a routing tier sustains 50k+ triggers/sec inside a P99 delivery budget under 40ms while shedding load safely under bursts; violate any one, and the failure is a dropped compliance event or a double-charged fare that no downstream logic can fully undo.

Related #

• Kafka vs Redis Streams for trigger routing — choosing the routing substrate by durability, replay, and operational weight.
• Dead-letter topics and poison message handling — bounded retries and safe quarantine for un-routable triggers.
• Idempotent trigger emission semantics — deterministic keys and exactly-once vs at-least-once delivery.
• Backpressure and flow-control strategies — sizing bounded queues and shedding load without dropping high-priority events.
• Production monitoring and observability — the metric families and alert thresholds behind the delivery budget.
• Circuit breakers for downstream trigger consumers — failing to the dead-letter path and half-open recovery.
• Core architecture and latency constraints — the upstream evaluation pipeline that emits every trigger this tier routes.
• Spatial indexing for real-time checks — the index primitives that decide which fence a trigger belongs to.