Async Processing for Heavy Geometries

Q: When should I use ProcessPoolExecutor instead of ThreadPoolExecutor for geometry work?

Use ProcessPoolExecutor whenever geometry operations are CPU-bound — topology repair with make_valid, coordinate projection loops, or spatial joins over thousands of features. These operations are blocked by Python's GIL and need true OS-level parallelism. ThreadPoolExecutor is suitable only for I/O-bound steps like writing results to PostGIS or publishing to Redis.

Q: How do I prevent memory spikes when deserializing large GeoJSON payloads?

Use ijson for iterative streaming deserialization — it emits geometry objects one at a time without loading the full document into RAM. Pair this with chunked dispatch: split feature collections into coordinate-bounded batches before sending to the worker pool.

Q: How do I handle mixed CRS payloads in an async pipeline?

Detect the source CRS from the payload's crs property or a vendor-specific header, then normalize to EPSG:4326 (WGS 84) inside the worker function before any topology checks. Refusing to dispatch until CRS is resolved prevents silent coordinate corruption in downstream spatial indexes.

Q: What is a safe timeout for a geometry worker process?

Set a per-task timeout via asyncio.wait_for wrapping run_in_executor. A ceiling of 30 seconds works for most polygon simplification and validation workloads; complex union/intersection operations over dense point clouds may need up to 120 seconds. Route timed-out tasks to a dead-letter queue rather than retrying synchronously.

Use async dispatch to a ProcessPoolExecutor to acknowledge a spatial webhook in milliseconds while heavy geometry parsing, validation, and projection happen off the event loop in a separate OS process.

This implementation guide is part of Spatial Payload Routing & Parsing, the section covering how GeoJSON and binary geometry payloads are ingested, classified, validated, and forwarded to spatial consumers.

Prerequisites

Before building an asynchronous geometry pipeline, confirm your environment meets these baseline requirements:

Python 3.10+ — required for asyncio.to_thread, structural pattern matching, and stable ExceptionGroup handling.
Shapely 2.0+ — built on GEOS 3.10+, which provides thread-safe geometry objects and the make_valid function for topology repair.
pyproj 3.4+ — for CRS detection and coordinate transformation via the PROJ 9 bindings.
FastAPI 0.100+ or aiohttp 3.9+ — non-blocking HTTP framework for the ingestion endpoint.
Pydantic v2 — for schema-validated payload parsing before dispatch.
ijson 3.2+ — for streaming JSON deserialization of large feature collections without loading the full document into RAM.
Redis 7+ or RabbitMQ 3.12+ — optional but recommended for durable task queuing and backpressure management between the HTTP layer and the worker pool.

Architecture Overview

The pipeline separates concerns into four layers. The webhook endpoint only performs lightweight I/O; all CPU work happens downstream in isolated processes.

Layer 1 — Ingestion: The FastAPI endpoint receives raw bytes, validates the HMAC signature and Content-Length, publishes the payload to the queue, and responds 202 Accepted in under 100 ms.

Layer 2 — Queue: An async consumer reads from Redis Streams or RabbitMQ. This buffer absorbs traffic spikes and decouples ingestion throughput from worker capacity. Tasks that exceed the retry ceiling move to a dead-letter queue for inspection.

Layer 3 — Worker Pool: A ProcessPoolExecutor runs topology repair (make_valid), CRS normalization to EPSG:4326 (WGS 84), Pydantic schema validation, and coordinate transformation in separate OS processes, bypassing the GIL entirely.

Layer 4 — Persistence: Validated geometries are written to a spatially indexed store. A completion event notifies downstream consumers and can trigger a status callback to the originating client.

Step-by-Step Implementation

Step 1 — Non-Blocking Ingestion and Immediate Acknowledgment

The endpoint streams raw bytes without deserializing them. Validation is limited to the HTTP layer: check the Content-Type, enforce a Content-Length ceiling, and verify the HMAC-SHA256 signature. For details on constructing the signature check, see Securing Webhook Endpoints with Spatial Token Validation.

import asyncio
import hashlib
import hmac
import logging
import os
from concurrent.futures import ProcessPoolExecutor

from fastapi import FastAPI, HTTPException, Request, BackgroundTasks

app = FastAPI()
logger = logging.getLogger(__name__)

WEBHOOK_SECRET = os.environ["WEBHOOK_SECRET"].encode()
MAX_PAYLOAD_BYTES = 100 * 1024 * 1024  # 100 MB ceiling

# Pre-allocate the pool at startup to avoid cold-start latency per request.
WORKER_POOL = ProcessPoolExecutor(max_workers=os.cpu_count() or 4)


def _verify_hmac(body: bytes, signature_header: str | None) -> None:
    """Raise HTTP 401 if the HMAC-SHA256 signature does not match."""
    if not signature_header:
        raise HTTPException(status_code=401, detail="Missing X-Signature header")
    expected = "sha256=" + hmac.new(WEBHOOK_SECRET, body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, signature_header):
        raise HTTPException(status_code=401, detail="Invalid signature")


@app.post("/webhook/spatial")
async def receive_spatial_payload(
    request: Request,
    background_tasks: BackgroundTasks,
) -> dict:
    content_length = int(request.headers.get("content-length", 0))
    if content_length > MAX_PAYLOAD_BYTES:
        raise HTTPException(status_code=413, detail="Payload exceeds 100 MB limit")

    raw_bytes = await request.body()
    _verify_hmac(raw_bytes, request.headers.get("x-signature"))

    # Offload heavy geometry work; respond immediately.
    background_tasks.add_task(_dispatch_to_pool, raw_bytes)
    return {"status": "queued"}

Step 2 — Dispatch to the Process Pool

loop.run_in_executor submits the CPU-bound worker to the pre-allocated ProcessPoolExecutor. The await suspends only this coroutine — the event loop remains free to handle other incoming requests.

async def _dispatch_to_pool(payload: bytes) -> None:
    """Submit geometry processing to a separate OS process."""
    loop = asyncio.get_running_loop()
    try:
        result = await asyncio.wait_for(
            loop.run_in_executor(WORKER_POOL, _process_geometry, payload),
            timeout=30.0,  # seconds; adjust for expected geometry complexity
        )
        logger.info("Worker finished: status=%s", result.get("status"))
    except TimeoutError:
        logger.error("Worker timed out — routing to dead-letter queue")
        await _send_to_dlq(payload)
    except Exception as exc:
        logger.exception("Worker raised unexpected exception: %s", exc)
        await _send_to_dlq(payload)

Step 3 — Topology Validation and CRS Normalization

Each worker function is a plain top-level function (required by pickle serialization between processes). It validates the GeoJSON structure against RFC 7946, repairs invalid topologies, and normalizes coordinates to EPSG:4326 (WGS 84) using pyproj. This mirrors the CRS Normalization Strategies pattern for enforcing a canonical projection before any downstream operation.

import json
from typing import Any

from pyproj import CRS, Transformer
from shapely.geometry import mapping, shape
from shapely.ops import transform as shapely_transform
from shapely.validation import make_valid


def _reproject(geom, source_epsg: int):
    """Reproject a whole Shapely geometry to EPSG:4326 (WGS 84)."""
    transformer = Transformer.from_crs(
        CRS.from_epsg(source_epsg),
        CRS.from_epsg(4326),
        always_xy=True,  # GeoJSON is lon/lat, so force x=lon, y=lat ordering
    )
    # shapely.ops.transform walks every coordinate ring for us, so MultiPolygons
    # and GeometryCollections are reprojected in full — not just the bounds.
    return shapely_transform(transformer.transform, geom)


def _process_geometry(payload: bytes) -> dict[str, Any]:
    """CPU-bound worker: validate, repair, and normalize a GeoJSON feature."""
    try:
        data = json.loads(payload)
        if data.get("type") != "Feature" or "geometry" not in data:
            return {"status": "error", "message": "Payload is not a GeoJSON Feature"}

        geom = shape(data["geometry"])

        # Topology repair — make_valid returns a valid geometry without data loss.
        if not geom.is_valid:
            geom = make_valid(geom)

        # Detect a non-WGS84 source CRS from an optional vendor property and
        # reproject before persistence so every downstream index is EPSG:4326.
        source_epsg = data.get("properties", {}).get("source_epsg", 4326)
        if source_epsg != 4326:
            geom = _reproject(geom, source_epsg)

        return {
            "status": "success",
            "result": {
                "type": "Feature",
                "geometry": mapping(geom),
                "properties": {
                    **data.get("properties", {}),
                    "crs": "EPSG:4326",
                    "bbox": list(geom.bounds),
                    "is_valid": geom.is_valid,
                },
            },
        }
    except Exception as exc:  # noqa: BLE001
        return {"status": "error", "message": str(exc)}

Step 4 — Pydantic Schema Validation

Before the worker passes results to the persistence layer, validate the output schema with Pydantic v2. This catches malformed geometry objects that passed Shapely’s topology check but violate the application-level contract — for example, MultiPolygon features that should have been split, or missing required property fields. This pairs naturally with Geometry Validation Pipelines for a defense-in-depth validation strategy.

from pydantic import BaseModel, Field, model_validator


class SpatialFeatureOut(BaseModel):
    type: str = Field(pattern="^Feature$")
    geometry: dict
    properties: dict

    @model_validator(mode="after")
    def check_geometry_type(self) -> "SpatialFeatureOut":
        allowed = {"Point", "LineString", "Polygon", "MultiPoint",
                   "MultiLineString", "MultiPolygon", "GeometryCollection"}
        if self.geometry.get("type") not in allowed:
            raise ValueError(f"Unsupported geometry type: {self.geometry.get('type')}")
        return self


def validate_output(result: dict) -> SpatialFeatureOut:
    """Raise ValidationError if the processed feature does not meet schema."""
    return SpatialFeatureOut.model_validate(result["result"])

Step 5 — Streaming Deserialization for Large Feature Collections

For payloads exceeding 10 MB, replace json.loads with ijson to avoid heap spikes. The iterator emits one feature at a time, letting the worker dispatch each feature to a separate sub-task rather than holding the entire collection in memory.

import io
import ijson


def _iter_features(payload: bytes):
    """Yield individual GeoJSON features without loading the full collection."""
    f = io.BytesIO(payload)
    parser = ijson.items(f, "features.item")
    for feature in parser:
        yield feature


def _process_feature_collection(payload: bytes) -> list[dict]:
    results = []
    for feature in _iter_features(payload):
        feature_bytes = json.dumps(feature).encode()
        results.append(_process_geometry(feature_bytes))
    return results

Spatial Validation and Error Handling

Shapely’s is_valid flag catches self-intersections, unclosed rings, and duplicate vertices — but it does not validate coordinate range or ring orientation under RFC 7946. Add explicit guards:

from shapely.geometry import shape
from shapely.validation import explain_validity


def full_validation_report(raw_geometry: dict) -> dict:
    """Return a structured validity report for a GeoJSON geometry object."""
    geom = shape(raw_geometry)
    coords = list(geom.geoms) if hasattr(geom, "geoms") else [geom]

    min_x, min_y, max_x, max_y = geom.bounds
    bbox_valid = (
        -180 <= min_x <= max_x <= 180  # longitude range, west <= east
        and -90 <= min_y <= max_y <= 90  # latitude range, south <= north
    )

    report = {
        "is_valid": geom.is_valid,
        "explanation": explain_validity(geom) if not geom.is_valid else None,
        "bbox_valid": bbox_valid,
        "geometry_type": geom.geom_type,
        "coordinate_count": sum(
            len(list(g.exterior.coords)) if hasattr(g, "exterior") else 0
            for g in coords
        ),
    }
    return report

Route features that fail bbox_valid directly to the dead-letter queue — they indicate corrupt coordinate data that topology repair cannot fix. Log the explanation field for every invalid geometry to enable rapid root-cause analysis without replaying the entire payload. This complements the at-least-once delivery pattern where retries must not reprocess already-persisted features.

Retry, Backoff, and Delivery Guarantees

Spatial workloads fail transiently (GEOS internal errors, database timeouts) and permanently (corrupt coordinate rings). Distinguish between these cases in the retry policy:

import random
import asyncio


async def retry_with_backoff(
    payload: bytes,
    max_attempts: int = 4,
    base_delay: float = 1.0,
    max_delay: float = 30.0,
) -> dict | None:
    """
    Retry geometry processing with exponential backoff and jitter.
    Returns None and routes to DLQ after max_attempts.
    """
    for attempt in range(1, max_attempts + 1):
        result = await _dispatch_to_pool(payload)
        if result and result.get("status") == "success":
            return result

        if attempt == max_attempts:
            logger.error(
                "Max retries reached after %d attempts — sending to DLQ", attempt
            )
            await _send_to_dlq(payload)
            return None

        delay = min(base_delay * (2 ** (attempt - 1)), max_delay)
        jitter = random.uniform(0, delay * 0.2)
        logger.warning(
            "Attempt %d failed, retrying in %.2f s", attempt, delay + jitter
        )
        await asyncio.sleep(delay + jitter)

    return None

At-least-once vs exactly-once: The async pipeline above is at-least-once by default — a worker crash after processing but before acknowledgment will cause a re-delivery. To upgrade to effectively-once semantics, assign each payload a deterministic idempotency key (based on a hash of the raw bytes or a vendor-supplied event ID) and store it in Redis before writing to PostGIS. The Event Key Generation for Spatial Data and Cache-Backed Idempotency Checks pages cover this in detail.

Verification

Run this integration test against a local FastAPI instance to confirm the end-to-end pipeline behaves correctly:

import json
import time
import httpx
import pytest

SAMPLE_FEATURE = {
    "type": "Feature",
    "geometry": {
        "type": "Polygon",
        "coordinates": [[
            [-122.4194, 37.7749],
            [-122.4094, 37.7749],
            [-122.4094, 37.7849],
            [-122.4194, 37.7849],
            [-122.4194, 37.7749],
        ]]
    },
    "properties": {"source_epsg": 4326, "name": "test-polygon"}
}


@pytest.mark.asyncio
async def test_webhook_returns_202_immediately():
    payload = json.dumps(SAMPLE_FEATURE).encode()
    async with httpx.AsyncClient(base_url="http://localhost:8000") as client:
        t0 = time.monotonic()
        resp = await client.post(
            "/webhook/spatial",
            content=payload,
            headers={
                "content-type": "application/json",
                "x-signature": _compute_test_hmac(payload),
            },
        )
        elapsed = time.monotonic() - t0

    assert resp.status_code == 202
    assert resp.json()["status"] == "queued"
    # Endpoint must acknowledge before any geometry processing begins.
    assert elapsed < 0.5, f"Acknowledgment too slow: {elapsed:.3f} s"


def _compute_test_hmac(body: bytes) -> str:
    import hashlib, hmac, os
    secret = os.environ.get("WEBHOOK_SECRET", "test-secret").encode()
    digest = hmac.new(secret, body, hashlib.sha256).hexdigest()
    return f"sha256={digest}"

Check the application logs after sending a test payload to confirm the worker completed successfully:

INFO     Worker finished: status=success

If you see Worker timed out, reduce max_workers or lower the Content-Length ceiling — a flooded process pool starves individual workers of CPU time.

Troubleshooting

Symptom	Likely spatial cause	Fix
Endpoint latency spikes above 500 ms	Synchronous JSON parsing blocking the event loop	Replace `json.loads` in the handler with `await request.body()` only; defer all parsing to the worker
Worker returns `TopologicalError`	Self-intersecting polygon rings in source data	Call `make_valid(geom)` before any spatial operation; log `explain_validity(geom)` to identify the ring
CRS mismatch after transformation	Source payload uses EPSG:3857 (Web Mercator) but worker assumes EPSG:4326	Read the `source_epsg` property or a `X-Source-CRS` vendor header; reject payloads without explicit CRS declaration
`PicklingError` in ProcessPoolExecutor	Lambda or closure passed as worker function	Worker function must be a module-level `def`; closures cannot be serialized across process boundaries
Memory grows unbounded under load	Full feature collection loaded into RAM per request	Switch to `ijson` streaming deserialization; enforce the `MAX_PAYLOAD_BYTES` ceiling in the handler
Dead-letter queue fills faster than expected	Retry policy not distinguishing transient from permanent errors	Inspect the `message` field in failed results; route `ValidationError` outcomes directly to DLQ without retry
Worker pool exhausted under burst traffic	`max_workers` set too low for concurrent payload volume	Scale `max_workers` to `os.cpu_count()` and add a semaphore to cap concurrent in-flight dispatch calls

FAQ

When should I use ProcessPoolExecutor instead of ThreadPoolExecutor for geometry work?

Use ProcessPoolExecutor whenever geometry operations are CPU-bound — topology repair with make_valid, coordinate projection loops, or spatial joins over thousands of features. These operations are blocked by Python’s GIL and need true OS-level parallelism. ThreadPoolExecutor is suitable only for I/O-bound steps like writing results to PostGIS or publishing to Redis Streams.

How do I prevent memory spikes when deserializing large GeoJSON payloads?

Use ijson for iterative streaming deserialization — it emits geometry objects one at a time without loading the full document into RAM. Pair this with chunked dispatch: split feature collections into coordinate-bounded batches before sending to the worker pool. For the GeoJSON-to-Protobuf mapping path, binary serialization also shrinks wire payload size by 60–80% before it reaches the ingestion endpoint.

How do I handle mixed CRS payloads in an async pipeline?

Detect the source CRS from the payload’s crs property or a X-Source-CRS vendor header, then normalize to EPSG:4326 (WGS 84) inside the worker function before any topology checks. Refusing to dispatch until CRS is resolved prevents silent coordinate corruption in downstream spatial indexes. The full normalization workflow is covered in CRS Normalization Strategies.

What is a safe timeout for a geometry worker process?

Set a per-task timeout via asyncio.wait_for wrapping run_in_executor. A ceiling of 30 seconds works for most polygon simplification and validation workloads; complex union/intersection operations over dense point clouds may need up to 120 seconds. Route timed-out tasks to a dead-letter queue rather than retrying synchronously — a timeout usually signals a data pathology, not a transient failure.

Spatial Payload Routing & Parsing — parent section covering the full ingestion and routing architecture
Optimizing Async Geometry Parsing with asyncio — deep dive into asyncio integration patterns for this pipeline
CRS Normalization Strategies — enforcing a canonical projection across mixed-CRS spatial event streams
Geometry Validation Pipelines — defense-in-depth topology and schema checks before spatial indexing
Event Key Generation for Spatial Data — deriving deterministic idempotency keys from GeoJSON feature hashes to prevent duplicate writes

Async Processing for Heavy Geometries

Explore this section

Optimizing Async Geometry Parsing with asyncio