Backpressure with Bounded asyncio Queues for Graph Writes

An OpenStreetMap importer that parses features faster than Neo4j can absorb them has exactly two ways to die: it exhausts the connection pool or it exhausts the heap. The symptom is a resident-set graph that climbs steadily through a country-scale run until the process is OOMKilled, or a flood of ConnectionAcquisitionTimeout errors the moment the writer falls behind. The root cause is the same in both cases — the producer and the consumer are decoupled with no shared limit, so a parser reading a memory-mapped extract at hundreds of thousands of features per second keeps buffering work that the database has not yet accepted. This page resolves it with one runnable pattern: a bounded asyncio.Queue sitting between a single fast producer and a fixed pool of Neo4j writer workers, so that when the database slows, await queue.put() blocks the producer and the slow tier sets the pace for the whole pipeline. It is the focused mechanism behind the broader techniques in scaling async graph ingestion with Python asyncio.

Prerequisites & Versions

The queue and shutdown mechanics are pure asyncio; the write side needs the official async driver and a running Neo4j with a uniqueness constraint so the MERGE seeks instead of scans.

This pattern assumes the base topology already follows sound conventions from OSM data ingestion pipelines — features flattened into source/target rows with coordinates — and that a uniqueness constraint on the node key exists so the writer’s MERGE is an index seek rather than a full scan.

Implementation

The module below is self-contained and runnable. One producer coroutine pulls parsed features and pushes them onto a bounded asyncio.Queue; a fixed pool of writer workers pulls from the same queue, accumulates a local batch, and flushes it with a batched UNWIND. Shutdown is explicit: after the producer finishes, one sentinel per worker is enqueued so every worker flushes its partial batch and exits cleanly — no rows are silently dropped on the way out.

import asyncio
import logging
from typing import Any, AsyncIterator

from neo4j import AsyncDriver, AsyncGraphDatabase
from neo4j.exceptions import ServiceUnavailable, TransientError

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("bounded_ingest")

_SHUTDOWN = object()  # sentinel: one per worker triggers a graceful drain

WRITE_QUERY = """
UNWIND $batch AS row
MERGE (u:Node {id: row.source})
  ON CREATE SET u.location =
    point({srid: 4326, longitude: row.source_lon, latitude: row.source_lat})
MERGE (v:Node {id: row.target})
  ON CREATE SET v.location =
    point({srid: 4326, longitude: row.target_lon, latitude: row.target_lat})
MERGE (u)-[e:ROUTE {kind: row.kind}]->(v)
  SET e.length_m = row.length_m
"""


async def ensure_schema(driver: AsyncDriver) -> None:
    async with driver.session() as session:
        await session.run(
            "CREATE CONSTRAINT node_id IF NOT EXISTS "
            "FOR (n:Node) REQUIRE n.id IS UNIQUE"
        )
        await session.run(
            "CREATE POINT INDEX node_location IF NOT EXISTS "
            "FOR (n:Node) ON (n.location)"
        )


async def writer_worker(
    wid: int, driver: AsyncDriver, queue: "asyncio.Queue[Any]", flush_size: int
) -> None:
    """Drain the queue into a local batch; flush on size or on the shutdown sentinel."""
    batch: list[dict[str, Any]] = []

    async def flush() -> None:
        if not batch:
            return
        for attempt in range(1, 4):
            try:
                async with driver.session() as session:
                    result = await session.run(WRITE_QUERY, batch=batch)
                    await result.consume()
                break
            except (TransientError, ServiceUnavailable) as exc:
                backoff = 0.25 * (2 ** (attempt - 1))
                log.warning("worker %d transient (%d/3): %s", wid, attempt, exc)
                await asyncio.sleep(backoff)
        else:
            log.error("worker %d dropped %d rows after retries", wid, len(batch))
        batch.clear()

    while True:
        item = await queue.get()
        try:
            if item is _SHUTDOWN:
                await flush()  # drain what we hold before exiting
                return
            batch.append(item)
            if len(batch) >= flush_size:
                await flush()
        finally:
            queue.task_done()


async def produce(queue: "asyncio.Queue[Any]", features: AsyncIterator[dict]) -> None:
    """Push features onto the queue; put() blocks when full — this is the backpressure."""
    async for feature in features:
        await queue.put(feature)


async def run_pipeline(
    driver: AsyncDriver,
    features: AsyncIterator[dict],
    *,
    workers: int = 8,
    queue_max: int = 2_000,
    flush_size: int = 500,
) -> None:
    await ensure_schema(driver)
    queue: "asyncio.Queue[Any]" = asyncio.Queue(maxsize=queue_max)
    consumers = [
        asyncio.create_task(writer_worker(i, driver, queue, flush_size))
        for i in range(workers)
    ]
    await produce(queue, features)          # blocks whenever the DB falls behind
    for _ in consumers:                     # one sentinel per worker
        await queue.put(_SHUTDOWN)
    await asyncio.gather(*consumers)        # every worker flushes, then returns
    log.info("pipeline drained cleanly")


async def demo_features(n: int) -> AsyncIterator[dict]:
    """Stand-in parser: yields edge records with flat memory footprint."""
    base_lat, base_lon = 37.7749, -122.4194  # San Francisco
    for i in range(n):
        yield {
            "source": f"n{i}", "target": f"n{i + 1}", "kind": "residential",
            "source_lon": base_lon + i * 1e-4, "source_lat": base_lat + i * 1e-4,
            "target_lon": base_lon + (i + 1) * 1e-4, "target_lat": base_lat + (i + 1) * 1e-4,
            "length_m": 14.2,
        }


async def main() -> None:
    driver = AsyncGraphDatabase.driver(
        "bolt://localhost:7687",
        auth=("neo4j", "password"),
        max_connection_pool_size=16,        # >= worker count, with headroom
        connection_acquisition_timeout=30.0,
    )
    try:
        await run_pipeline(driver, demo_features(50_000), workers=8)
    finally:
        await driver.close()


if __name__ == "__main__":
    asyncio.run(main())

How It Works

The design rests on one property of asyncio.Queue: a queue created with maxsize > 0 makes await queue.put() suspend the calling coroutine once the queue is full, and resume it only when a worker calls queue.get(). That single suspension is the entire backpressure mechanism, and three details make it production-safe:

• The bound is the memory ceiling. With maxsize=queue_max, resident memory for buffered work is capped at roughly queue_max × bytes_per_feature plus workers × flush_size for the in-flight batches — a fixed number that does not grow with the size of the extract. Drop the bound (or use maxsize=0) and the producer races ahead unthrottled; that is the classic unbounded-queue out-of-memory failure that this whole pattern exists to prevent.
• A worker pool decouples parse speed from write speed. The producer never opens a session. It only enqueues. The workers writer coroutines are the only tasks that touch the driver, so the number of concurrent transactions is bounded by the pool size, not by how fast features arrive. Sizing max_connection_pool_size at or above the worker count keeps every worker’s session() acquisition immediate; the pool discipline here is the same one described in the async batch processing for graphs overview.
• Sentinel shutdown guarantees the last batch is written. After produce returns, the pipeline enqueues exactly one _SHUTDOWN object per worker. Each worker, on receiving its sentinel, calls flush() one final time before returning — so a worker holding 300 rows below the flush_size threshold still commits them. Ending the run by cancelling tasks instead would discard those partial batches, which is the silent data-loss trap.

Coordinates use Neo4j’s point({longitude, latitude}) convention (WGS84 / EPSG:4326) so the populated location property can later be seeked by the spatial indexing strategies layer rather than scanned.

Common Failure Patterns

1. Unbounded queue silently OOMs the importer. The default asyncio.Queue() has maxsize=0, meaning unbounded. put() never blocks, the producer drains the parser at full speed, and buffered features accumulate until the process is killed — usually deep into a long run, which makes it look intermittent. The fix is to always pass a finite bound and let it throttle:

# WRONG — unbounded: producer outruns the writers, heap grows without limit
queue = asyncio.Queue()
# RIGHT — bounded: put() blocks when full, so the DB paces the producer
queue = asyncio.Queue(maxsize=2_000)

2. Cancelling workers instead of draining loses the tail batch. Tearing down with task.cancel() after the producer finishes interrupts any worker mid-accumulation, and every row below the flush threshold is discarded without error. Enqueue one sentinel per worker and gather them so each flushes first:

# WRONG — cancels workers holding un-flushed rows
for c in consumers:
    c.cancel()
# RIGHT — sentinel-driven drain; every worker flushes its partial batch
for _ in consumers:
    await queue.put(_SHUTDOWN)
await asyncio.gather(*consumers)

3. One slow worker stalls the whole pipeline. With a single worker, or a worker wedged on a query holding a lock, the queue fills, put() blocks indefinitely, and throughput collapses to that one worker’s rate. Run a pool large enough that a transient slowdown on one worker is absorbed by the others, and keep connection_acquisition_timeout finite so a genuinely stuck write fails fast instead of pinning a slot forever. If chunks routinely run slow, the bottleneck is the write plan, not the queue — profile it with the techniques in optimizing Cypher query plans for spatial data.

Performance Notes

The queue depth in steady state follows Little’s Law. With a producer arrival rate $\lambda_{p}$ (features per second) and a total consumer service rate $\mu = W / s$ — $W$ workers each taking $s$ seconds to flush one batch of $F$ rows, so $\mu = W \cdot F / s$ features per second — the mean number of buffered items is:

$$L = \lambda_{\text{eff}} \cdot t_{\text{wait}}, \qquad \lambda_{\text{eff}} = \min!\left(\lambda_{p},; \frac{W \cdot F}{s}\right)$$

The min is the whole point: when $\lambda_{p} > \mu$, the queue saturates at maxsize, put() blocks, and the effective throughput is clamped to the drain rate $\mu$ rather than the parse rate. The system runs at the speed of its slowest tier — the database — with a flat, predictable memory footprint instead of an OOM.

Tuning follows from that. Raise workers (and max_connection_pool_size with it) until $\mu$ stops rising, which is the point where server-side lock contention or index-split cost caps write throughput; past that, more workers only deepen lock queues. Size flush_size to amortize round-trip latency — a few hundred to a few thousand rows per UNWIND — and size queue_max to absorb short parse bursts without letting buffered memory grow unbounded. When you need cold bulk-load throughput rather than continuous ingestion alongside live reads, switch to a server-side import path; the async queue here wins whenever writes must interleave with ongoing attribute synchronization.

Related

• Scaling async graph ingestion with Python asyncio
• Building automated OSM-to-graph ETL pipelines
• Optimizing Cypher query plans for spatial data
• Spatial indexing strategies

This guide is part of Async Batch Processing for Graphs, within the broader Spatial Graph Construction & OSM Ingestion guide.