Reactive Streams Concepts, Demand, Backpressure, and When Reactive Access Helps

Learning outcomes

An AtlasMart export endpoint may return a very large graph-shaped result to a downstream analytics service. An async API can avoid blocking threads, but it does not automatically express how many records the downstream consumer is prepared to accept. Reactive Streams adds a demand contract: producers should not overwhelm consumers that have requested less work.

1. Async and reactive answer different questions

Reactive access helps most when the result is large, consumers are variably paced, or a broader reactive stack already exists. For a small bounded API response, adding reactive operators can increase complexity without improving the database plan.

2. Current driver support boundary

3. Python: bounded pull without a reactive API

async def process_in_windows(driver, customer_id, window=100):    async with driver.session(database="neo4j", fetch_size=window) as session:        result = await session.run(            """            CYPHER 25            MATCH (:Customer {customerId:$id})-[:PLACED]->(o:Order)            RETURN o.orderId AS orderId, o.status AS status            ORDER BY o.orderId            """,            id=customer_id,        )        while True:            records = await result.fetch(window)            if not records:                break            await send_downstream([r.data() for r in records])        await result.consume()

The session fetch_size controls how many records the driver requests from Neo4j per batch (default 1000 in Python 6.3), but it is not a server-side LIMIT, nor does it cap the database memory required by an earlier Sort or aggregation operator. It controls record transfer/buffering behavior.

4. Optional JavaScript Reactive API example

mkdir js-reactivecd js-reactivenpm init -ynpm install neo4j-driver@6.2.0 rxjs

import neo4j from 'neo4j-driver'import { bufferCount, concatMap } from 'rxjs/operators'const driver = neo4j.driver(  'bolt://127.0.0.1:7687',  neo4j.auth.basic('neo4j', 'atlasmart-course-2026'))const session = driver.rxSession({ database: 'neo4j' })const subscription = session.run(`  CYPHER 25  MATCH (:Customer {customerId:$id})-[:PLACED]->(o:Order)  RETURN o.orderId AS orderId, o.status AS status  ORDER BY o.orderId`, { id: 'C-1001' }).records().pipe(  bufferCount(100),  // concatMap preserves one downstream batch at a time.  concatMap(records => persistBatch(records.map(r => r.toObject())))).subscribe({  error: async err => { console.error(err); await session.close(); await driver.close() },  complete: async () => { await session.close(); await driver.close() }})

This is an optional learning path. Exact RxJS operator behavior is part of the application stack, not a Neo4j server guarantee. Backpressure cannot repair an unbounded Cypher traversal that already generates an enormous server-side intermediate result; query cardinality still matters.

5. Wrong: reactive syntax with unbounded work behind it

6. Evidence: demand, bytes and completion

Production judgment

Summary and next step

Reactive demand is a client flow-control contract, not a substitute for bounded Cypher or capacity planning. Lesson 4 attacks another common overload source: N+1 and chatty graph access.

Authoritative references

• Current Neo4j versions — Current server and 5.26 LTS release snapshot.
• Neo4j Python Driver Manual — Official Python driver guide used by the mandatory async lab.
• Python Driver 6.3 API — Current synchronous and asynchronous API contracts and configuration.
• Python async API — Async driver/session/result lifecycle, cancellation and concurrency rules.
• Python concurrency guide — AsyncGraphDatabase and concurrent workflow guidance.
• Python performance recommendations — Driver performance, routing and batching guidance.
• Cypher UNWIND — List-to-row semantics and ordering boundary.
• Transaction management — Server-side transaction limits and timeout concepts.
• JavaScript driver installation — Current full vs lite driver and Reactive API boundary.
• JavaScript driver npm package — Current 6.2.0 package and RxJS dependency.