Async Batch Processing for Telemetry Ingestion and Route Optimization
Bounded queues and chunked writes decouple ingestion latency from solver cycles.
A municipal collection fleet emits telemetry faster than any route solver can consume it. During a morning dispatch window, thousands of GPS fixes, bin fill-level readings, and scale events arrive per second, while a single vehicle routing solve for one depot can take hundreds of milliseconds. If you feed each reading straight into the solver as it lands, ingestion latency and computation latency become the same number — the event loop stalls, the queue behind it grows without bound, and the process is eventually killed by the OOM reaper mid-shift. The stops that were in flight when it died are the stops nobody gets a compliance record for.
Async batch processing is the buffer that keeps those two rates independent. It folds a stream of individual readings into time-and-size-bounded batches, validates each one before it can reach the solver, dispatches heavy solves to a worker pool without blocking the ingestion coroutine, and writes results in chunks so a slow database never applies backpressure all the way up to the sensor. This page sits under the Telematics & Sensor Data Ingestion framework and inherits its non-negotiable rule: nothing is dropped silently. Every reading either enters a batch, lands in a dead-letter queue with a reason, or produces an auditable record — there is no fourth outcome.
Prerequisites
The pipeline runs on Python 3.10+ (the code uses X | Y unions and asyncio.Queue generic parameters) and deliberately keeps its dependency surface small — the concurrency primitives are all stdlib asyncio. Only the schema layer and the async database driver are pinned.
python -m pip install \
"pydantic==2.9.2" \
"asyncpg==0.30.0"Before any reading enters the pipeline it must satisfy a strict schema. Coercion is not allowed here for the same reason it is not allowed in the Schema Validation Pipelines reference: a silently coerced latitude is a route sent to the wrong street. The batch stage consumes only records that already carry a validated shape and a content hash for idempotency.
from __future__ import annotations
from pydantic import BaseModel, Field
class SensorReading(BaseModel):
"""One normalized telemetry record entering the batch pipeline."""
sensor_id: str = Field(min_length=1)
vehicle_id: str = Field(min_length=1)
captured_at: float = Field(gt=0) # unix epoch seconds
lat: float = Field(ge=-90, le=90)
lon: float = Field(ge=-180, le=180)
fill_pct: float = Field(ge=0, le=100)
payload_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
class BatchEnvelope(BaseModel):
"""A time- and size-bounded batch handed from ingestion to the solver stage."""
batch_id: str
opened_at: float # unix epoch seconds
readings: list[SensorReading]The payload_hash is the SHA-256 the Bin Sensor API Sync layer computes at the edge from sensor ID, timestamp, and raw metric. Carrying it through the batch means a duplicate reading re-delivered after a reconnect is deduplicated before it can trigger a redundant solve.
Core Implementation
The pipeline is four coroutines connected by two bounded queues. Bounded is the operative word — an unbounded queue is not a buffer, it is a memory leak with a delay.
1. Bounded ingestion with a validation gate
Raw payloads enter through a single gate that validates and either enqueues or dead-letters. The queue’s maxsize is the backpressure mechanism: when the solver falls behind, put blocks the producer instead of letting the queue grow, which is exactly the signal a healthy system should propagate.
import asyncio
from pydantic import ValidationError
INGEST_QUEUE: asyncio.Queue[SensorReading] = asyncio.Queue(maxsize=10_000)
DEAD_LETTER: asyncio.Queue[dict] = asyncio.Queue(maxsize=10_000)
async def ingest_raw(raw: dict) -> None:
"""Validate one raw payload, then enqueue it or route it to the dead-letter queue."""
try:
reading = SensorReading.model_validate(raw)
except ValidationError as exc:
await DEAD_LETTER.put({"raw": raw, "errors": exc.errors()})
return
await INGEST_QUEUE.put(reading) # blocks when full → backpressure upstream2. Time-and-size batch aggregation
The aggregator folds individual readings into a BatchEnvelope, closing the batch when either it reaches a size ceiling or a wall-clock deadline passes. The dual bound matters: size alone starves the solver during quiet periods (a half-full batch never ships), and time alone lets a burst build an oversized matrix that blows the solver’s memory budget.
import time
import uuid
async def aggregate_batches(
source: asyncio.Queue[SensorReading],
sink: asyncio.Queue[BatchEnvelope],
max_size: int = 500,
max_seconds: float = 2.0,
) -> None:
"""Fold a stream of readings into size- and time-bounded batches."""
while True:
batch: list[SensorReading] = []
deadline = time.monotonic() + max_seconds
while len(batch) < max_size:
timeout = deadline - time.monotonic()
if timeout <= 0:
break
try:
reading = await asyncio.wait_for(source.get(), timeout)
except asyncio.TimeoutError:
break # deadline hit → ship what we have
batch.append(reading)
source.task_done()
if batch:
await sink.put(BatchEnvelope(
batch_id=uuid.uuid4().hex,
opened_at=time.time(),
readings=batch,
))3. Solver dispatch without blocking the loop
The solver is CPU-bound and blocking — the constraint model documented in the OR-Tools Implementation reference does not yield to the event loop. Calling it directly from a coroutine would freeze ingestion for the entire solve. It runs in a thread instead, and a semaphore caps how many solves proceed at once so a burst of batches cannot spawn unbounded threads.
import hashlib
import json
import logging
from dataclasses import dataclass
SOLVER_SLOTS = asyncio.Semaphore(4)
@dataclass(frozen=True)
class SolveResult:
batch_id: str
rows: list[tuple] # serialized route-assignment rows
compliance_hash: str
def solve_route_batch(batch: BatchEnvelope) -> SolveResult:
"""Blocking OR-Tools solve. Model construction lives in the OR-Tools Implementation page."""
# ... build distance matrix + register Time/Load dimensions (see OR-Tools Implementation) ...
ordered = sorted(batch.readings, key=lambda r: r.captured_at)
rows = [(batch.batch_id, r.sensor_id, r.vehicle_id, r.lat, r.lon) for r in ordered]
digest = hashlib.sha256(
json.dumps([r[1:] for r in rows], sort_keys=True).encode()
).hexdigest()
return SolveResult(batch_id=batch.batch_id, rows=rows, compliance_hash=digest)
async def solver_worker(sink: asyncio.Queue[BatchEnvelope],
pool, logger: logging.Logger) -> None:
"""Consume batches, solve off the event loop, persist, and log the compliance record."""
while True:
batch = await sink.get()
try:
async with SOLVER_SLOTS:
result = await asyncio.to_thread(solve_route_batch, batch)
await persist_result(pool, result)
except Exception as exc: # isolate one poison batch
logger.error("batch_failed", extra={"extra_fields": {
"batch_id": batch.batch_id, "error": repr(exc)}})
await DEAD_LETTER.put({"batch_id": batch.batch_id, "error": repr(exc)})
else:
logger.info("batch_committed", extra={"extra_fields": {
"batch_id": batch.batch_id,
"reading_count": len(batch.readings),
"compliance_hash": result.compliance_hash}})
finally:
sink.task_done()4. Chunked async writes and structured compliance logging
A slow write is the most common source of backpressure, so results are flushed to the database in chunks over a connection pool rather than one giant statement that holds a connection for seconds. Alongside the write, every committed batch emits a structured record through the site-wide JSONFormatter pattern, so the compliance trail is a first-class output of the pipeline and not an afterthought.
INSERT_SQL = (
"INSERT INTO route_assignments (batch_id, sensor_id, vehicle_id, lat, lon) "
"VALUES ($1, $2, $3, $4, $5)"
)
async def persist_result(pool, result: SolveResult, chunk_size: int = 200) -> None:
"""Write solver output in bounded chunks so one commit never stalls the pipeline."""
for start in range(0, len(result.rows), chunk_size):
chunk = result.rows[start:start + chunk_size]
async with pool.acquire() as conn:
await conn.executemany(INSERT_SQL, chunk)
class JSONFormatter(logging.Formatter):
"""Emit one JSON object per log record for append-only compliance storage."""
def format(self, record: logging.LogRecord) -> str:
from datetime import datetime, timezone
payload = {
"ts": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"event": record.getMessage(),
}
payload.update(getattr(record, "extra_fields", {}))
return json.dumps(payload, separators=(",", ":"), sort_keys=True)
def build_compliance_logger() -> logging.Logger:
logger = logging.getLogger("async_batch")
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)
return loggerWiring the stages together, the whole pipeline is a set of long-lived tasks over two bounded queues, started once and supervised for the life of the process.
import asyncpg
async def main() -> None:
logger = build_compliance_logger()
pool = await asyncpg.create_pool(dsn="postgresql://ops@localhost/routing", max_size=8)
batch_queue: asyncio.Queue[BatchEnvelope] = asyncio.Queue(maxsize=1_000)
tasks = [
asyncio.create_task(aggregate_batches(INGEST_QUEUE, batch_queue)),
*[asyncio.create_task(solver_worker(batch_queue, pool, logger)) for _ in range(4)],
]
try:
await asyncio.gather(*tasks)
finally:
await pool.close()
if __name__ == "__main__":
asyncio.run(main())Regulatory Mapping
Batching is an availability optimization, but the records it produces are compliance artifacts, so each tunable parameter still traces to a specific rule. Cite these in code comments and audit records, not only in prose.
- Batch window duration (
max_seconds) → service-completion timeliness under municipal collection ordinances. The window is the maximum staleness between a bin-empty event and its committed record. It must stay shorter than the shortest reporting interval any local ordinance imposes; a two-second window is safe, a two-minute window can put a proof-of-service timestamp outside its required reporting bucket. - Compliance hash (
compliance_hash) → EPA e-Manifest, 40 CFR Part 262 (Uniform Hazardous Waste Manifest, EPA Form 8700-22). The SHA-256 over each batch’s serialized assignments is the tamper-evident anchor that lets an auditor replay a batch and confirm the manifested route was not altered after commit. - Solver duration and rest parameters carried in each reading → FMCSA Hours of Service, 49 CFR Part 395. The batch does not enforce Hours of Service itself; it carries
vehicle_idandcaptured_atso the solver can bound cumulative drive time. The parameter-by-parameter translation lives in the DOT/FMCSA Rule Mapping reference. - Dead-letter retention → recordkeeping obligations, 40 CFR 262.40 (three-year manifest retention). A rejected payload is not discarded; the dead-letter queue must persist to durable, append-only storage for at least the retention period, because “we dropped a reading and kept no record” is itself a recordkeeping failure.
Access to the queues and the assignment table they write to is scoped by the Security & Access Boundaries model, so a batch worker’s database credential cannot become a path around tenant isolation.
Validation & Verification
You verify a batch pipeline by forcing its boundaries, not by watching it run under normal load where every path looks fine.
First, assert the aggregator honours both bounds — that it ships early on the size ceiling and ships a partial batch on the time deadline.
import pytest
@pytest.mark.asyncio
async def test_aggregator_respects_size_and_time() -> None:
source: asyncio.Queue[SensorReading] = asyncio.Queue()
sink: asyncio.Queue[BatchEnvelope] = asyncio.Queue()
def reading(i: int) -> SensorReading:
return SensorReading(
sensor_id=f"s{i}", vehicle_id="V-1", captured_at=1.0 + i,
lat=40.0, lon=-75.0, fill_pct=80.0, payload_hash="0" * 64)
for i in range(3):
source.put_nowait(reading(i))
task = asyncio.create_task(
aggregate_batches(source, sink, max_size=500, max_seconds=0.2))
batch = await asyncio.wait_for(sink.get(), timeout=1.0) # ships on the deadline
task.cancel()
assert len(batch.readings) == 3 # partial batch shipped
assert batch.readings[0].sensor_id == "s0"Second, assert that an invalid payload is dead-lettered rather than raised into the ingestion path.
@pytest.mark.asyncio
async def test_invalid_payload_is_dead_lettered() -> None:
await ingest_raw({"sensor_id": "s1", "vehicle_id": "V-1", "captured_at": 1.0,
"lat": 999.0, "lon": -75.0, "fill_pct": 80.0,
"payload_hash": "0" * 64}) # lat out of range
dead = await asyncio.wait_for(DEAD_LETTER.get(), timeout=1.0)
assert dead["raw"]["lat"] == 999.0
assert INGEST_QUEUE.empty()Third, confirm the compliance record is actually emitted. In staging, attach the JSONFormatter, run one batch through solver_worker, and assert the batch_committed line carries batch_id, reading_count, and compliance_hash — those are the fields a compliance query filters on. A batch that commits but leaves no structured record is, for audit purposes, a batch that never ran.
Failure Modes & Edge Cases
Each failure mode below has a coded response rather than a bare exception, because the whole point of batching is to survive load that would kill a naive pipeline.
The ingest queue fills up. This is backpressure working, not a fault — but the producer must react rather than block forever. Give the edge-facing put a timeout and shed load deliberately, recording every shed reading so nothing vanishes silently.
async def ingest_with_shedding(reading: SensorReading, timeout: float = 0.5) -> None:
try:
await asyncio.wait_for(INGEST_QUEUE.put(reading), timeout)
except asyncio.TimeoutError:
await DEAD_LETTER.put({"shed": reading.model_dump(),
"reason": "ingest_queue_saturated"})A poison batch throws inside the solver. One malformed batch must never take down the worker or the loop. The try/except in solver_worker already isolates it: the batch is dead-lettered and the worker moves on. The isolation boundary is deliberate — asyncio.gather across workers would otherwise let one exception cancel its siblings.
The solver returns nothing feasible for a batch. A saturated capacity or a closed window mid-batch means no compliant route exists. Do not widen a hard boundary to force output; hand the batch to the deterministic Fallback Routing Logic tier, which defers the least-critical stops via a disjunction penalty and escalates the conflict rather than dispatching an illegal route.
A blocking call sneaks into a coroutine. A synchronous database driver or a time.sleep inside the loop reintroduces exactly the stall batching exists to prevent. Catch it in staging with the event-loop debug hook, which logs any callback that runs longer than the slow-callback threshold.
loop = asyncio.get_event_loop()
loop.set_debug(True)
loop.slow_callback_duration = 0.1 # log any callback exceeding 100 msThe dead-letter queue grows unbounded. A rising dead-letter rate is a symptom, usually of an upstream contract change from a sensor vendor. Alert on its depth, and drain it to durable storage on its own schedule so a reprocessing backlog never competes with live ingestion for memory.
Integration Checklist
Complete every item before enabling the pipeline in production:
Frequently Asked Questions
Why batch at all instead of solving each reading as it arrives?
Because the solver’s cost is dominated by matrix construction and search, which amortize across a batch. Solving one reading at a time repeats that fixed cost for every record and couples ingestion latency to solve latency, so a single slow solve stalls the whole intake. Batching lets thousands of readings share one solve and keeps the two rates independent.
What size and time window should I use for a batch?
Bound both, and tune against real peak-window telemetry. Size caps the distance matrix so the solver’s memory stays predictable; the time window caps staleness so a proof-of-service record never ages past its reporting bucket. A common municipal starting point is a few hundred readings or a two-second deadline, whichever trips first, then adjusted from replayed load.
Won't running the solver in a thread hit the GIL?
OR-Tools releases the GIL during its native solve, so asyncio.to_thread genuinely parallelizes across cores and keeps the event loop responsive for ingestion. For pure-Python CPU work that holds the GIL, switch the same dispatch pattern to a ProcessPoolExecutor — the coroutine structure is identical, only the executor changes.
How do I stop the pipeline from silently dropping readings under load?
Make every exit auditable. A reading either enters a batch, lands in the dead-letter queue with a reason, or produces a committed record — there is no path that drops it without a trace. Load shedding uses a bounded put timeout that writes the shed reading to the dead-letter queue, and that queue is alerted on and drained to durable storage.
What happens to a batch the solver can't route?
It is handed to the deterministic fallback tier rather than force-fitted. The fallback defers the least-critical stops via a disjunction penalty and escalates the conflict to a dispatcher, so a hard boundary like capacity or Hours of Service is never relaxed just to make a batch commit.
Related
- Bin Sensor API Sync — the edge layer that computes the payload hash each batch deduplicates on.
- Schema Validation Pipelines — the strict-schema posture the ingestion gate inherits.
- GPS Polling Strategies — the cadence that determines how fast readings arrive at the ingest queue.
- OR-Tools Implementation — the blocking solver the worker pool dispatches to a thread.
- Fallback Routing Logic — where infeasible batches go instead of relaxing a hard constraint.