Streaming callbacks + shared storage (multiplexed streaming transport) by T4rk1n · Pull Request #3888 · plotly/dash

T4rk1n · 2026-07-09T18:19:34Z

Summary

This PR adds streaming callbacks and, to support them well across worker processes, a new backend-agnostic shared storage primitive.

Streaming callbacks — decorate an async def generator; each yield is pushed to the browser as it is produced (same shape as a normal return; dash.Patch yields apply incrementally, e.g. LLM token streaming). No opt-in keyword — a generator streams by definition. Sync generators are rejected at registration.
Shared storage (dash.ctx.shared_storage / app.shared_storage) — a cross-process key/value store + ordered publish/subscribe, usable by apps for cross-callback/session state, and used internally as the broker for streaming.
Multiplexed HTTP streaming — all of a page's streams share one downlink connection instead of one per callback, so they no longer hit the browser's ~6-connections-per-host ceiling. Works on Flask, Quart, and FastAPI, no WebSocket required.

Streaming callbacks

@callback(Output("out", "children"), Input("go", "n_clicks"))
async def stream(n):
    async for token in llm.stream(prompt):
        patch = Patch()
        patch += token
        yield patch

Shared storage

app = Dash(__name__)                     # shared_storage=LocalSharedStorage by default
# inside a callback:
dash.ctx.shared_storage.set("k", value)  # cross-process KV (JSON-compatible values)
dash.ctx.shared_storage.publish("topic", msg)

The default LocalSharedStorage elects a single owner process per machine (AF_UNIX socket on POSIX, TCP loopback on Windows — the bind is the lease, re-elected on owner death) and serves the others. A single-process deployment is its own owner and pays no socket overhead. Subscriptions are ordered and replayable: a reconnecting consumer resumes from its last-seen sequence out of a bounded buffer, and a buffer overrun surfaces as an explicit gap rather than a silent loss. Wire codec is msgspec (never pickle); pass shared_storage=None to disable, or a BaseSharedStorage subclass/instance to swap the backend.

Architecture — multiplexed streaming

A streaming callback no longer holds its own HTTP connection. Its POST returns a fast ack; its frames are pumped onto a shared-storage topic and relayed over the page's single downlink, routed back to the right callback by requestId. Because the frames travel through shared storage, the worker that runs a callback and the worker that holds the downlink need not be the same process — shared storage is the broker.

flowchart TB
    subgraph browser["Browser — one page"]
        cbs["streaming callbacks<br/>(async def generators)"]
        sc["StreamClient<br/>single downlink per page"]
        cbs --> sc
    end

    subgraph server["Dash server — any number of worker processes"]
        wa["worker A<br/>runs callback, pumps frames"]
        wb["worker B<br/>serves the downlink"]
    end

    store[("Shared storage owner process<br/>KV + ordered pub/sub<br/>topic per connection")]

    sc -->|"1 · uplink POST, fast ack<br/>streamConnection = conn + requestId"| wa
    wa -->|"2 · publish frames, tagged requestId + seq"| store
    sc -->|"3 · single downlink<br/>streamDownlink = conn, from = seq"| wb
    store -->|"4 · subscribe, replay from seq"| wb
    wb -->|"5 · NDJSON frames"| sc
    sc -->|"6 · route by requestId, apply"| cbs

Transport selection (renderer): a streaming callback rides the WebSocket transport when websocket callbacks are enabled; otherwise the multiplexed HTTP transport when shared storage is available; otherwise falls back to today's one NDJSON connection per callback. So nothing changes for apps that don't opt into shared storage.

Scheduler: long-lived streams no longer consume the renderer's concurrent-request budget, and clientside callbacks are exempt from it — a page full of streams no longer starves other callbacks.

Testing

Shared storage: engine semantics + cross-process KV, pub/sub, reconnect-replay, owner re-election.
Streaming transport: uplink/downlink end-to-end over real HTTP on Flask, Quart, and FastAPI.
Renderer: StreamClient routing, multiplexing, reconnect-from-cursor, keepalive (karma).
Browser validation (Selenium): the streaming integration suite runs over the multiplexed transport, plus an 8-stream demo where all streams and a clientside clock run simultaneously with a single downlink.

Notes

Adds msgspec to install requirements (wire codec; JSON fallback).
Shared-storage values must be JSON-compatible (like dcc.Store).
The in-memory backend is not durable: if the owner process dies, a survivor re-elects with an empty store (durable backends can follow).

Follow-up (not in this PR)

Host the downlink in a SharedWorker so it's one connection per browser (shared across tabs) rather than per page. StreamClient is written host-agnostic for this; the per-page transport already solves the connection-limit problem.

A callback registered with stream=True is a generator (or async generator) whose yields are pushed to the browser as they are produced. Each yielded value has the same shape as a regular return value and replaces the outputs; yielding dash.Patch gives incremental updates (e.g. LLM token streaming). The last yield is the final value. Transport follows the callback's normal selection: over the WebSocket callback transport, frames ride the open connection as callback_response messages with stream: true; otherwise the HTTP response streams NDJSON (application/x-ndjson), one frame per line with a terminal {"done": true} frame. Works on Flask, Quart, and FastAPI; sync generators warn at registration since they occupy a server worker for the whole stream. - dash/_streaming.py: StreamedCallbackResponse marker, context-safe iteration helpers (callback context travels in a contextvars snapshot so set_props/ctx work after dispatch returns), NDJSON serialization, and a thread bridge for async generators on Flask. - dash/_callback.py: stream wrappers building one frame per yield via _prepare_response; per-yield no_update/PreventUpdate handling, on_error support, error frames; registration validation (mutually exclusive with background/clientside/mcp_enabled/api_endpoint). - backends: streaming response branches in each serve_callback; ws.py frame emitter and stream consumers for both the event-loop and threadpool dispatch paths. - renderer: applyStreamFrame applies frames on arrival through the sideUpdate path (Patch applies exactly once); NDJSON reader in handleServerside; stream-aware callback_response handling keeps the request pending until the terminal frame; loading states span the whole stream.

fastapi.testclient imports starlette.testclient, which requires httpx. Skip the protocol-level stream tests when httpx is not installed and add httpx to the CI requirements so the websocket job actually runs them.

KoolADE85

The implementation here will easily starve the server of resources.
Run the sample app with gunicorn -w 1 and observe:

It will terminate streams that take longer than 30s (by default)
The number of simultaneous requests is limited to the number of workers you spawn. After that limit, the server becomes unresponsive and/or terminates worker threads.

KoolADE85 · 2026-07-22T22:04:33Z

+        stream = _kwargs.get("stream", False)
+        is_gen_func = inspect.isgeneratorfunction(func)
+        is_async_gen_func = inspect.isasyncgenfunction(func)
+        if stream:
+            if not (is_gen_func or is_async_gen_func):
+                raise StreamCallbackError(
+                    f"stream=True callback '{callback_id}' must be a generator "
+                    "function (or async generator function) that yields output "
+                    "updates."
+                )


This looks like two "sources of truth" that must agree with each other (stream=True and async/generator).
What do you think about removing the stream flag and just detect streaming automatically based on isasyncgenfunction/ isgeneratorfunction? That would allow devs to just start yielding values without needing to manage flags or understand our API (because there would be no API at all).

Good catch, only need a yield, stream keyword is redundant.

A stream=True callback that goes quiet between yields sends no bytes, so every proxy in a typical deployment eventually closes the connection on its own idle timeout (nginx's proxy_read_timeout defaults to 60s). The renderer treats a stream that ends without a terminal frame as a dropped connection, so this looked like silent truncation. The NDJSON transports now emit a blank line every stream_keepalive_interval milliseconds (new Dash argument, default 15000, None or 0 disables) that the callback spends between yields. The renderer already skips blank lines, so no client change is needed. The WebSocket transport is unaffected -- it has websocket_heartbeat_interval. The two paths need different mechanics. The async path holds the pending __anext__ across timeouts; asyncio.wait_for would cancel the user generator mid-step every time a keepalive came due. The sync path drives the frame generator on a pump thread, since a blocking next() cannot be interrupted on a timer -- one consequence is that a client disconnect no longer raises GeneratorExit into the user generator at its current yield, which is one more reason sync generators warn at registration. Note this does not address gunicorn's worker timeout, which kills the worker rather than the connection and is only fixable via --worker-class.

T4rk1n · 2026-07-27T14:24:31Z

The implementation here will easily starve the server of resources. Run the sample app with gunicorn -w 1 and observe:
1. It will terminate streams that take longer than 30s (by default)

2. The number of simultaneous requests is limited to the number of workers you spawn. After that limit, the server becomes unresponsive and/or terminates worker threads.

With a sync flask backend, it's not really possible to fix at the app level. Need to run with --worker-class gthread --threads 8 --timeout 0 for a true fix. Will add support in cloud and DE.

There is also a warning recommending a different backend when streaming, but maybe we should not allow it?

chgiesse · 2026-07-28T07:12:59Z

@T4rk1n I solved that issue with my event streaming package by limiting the event stream to the normal 30 sec server timeout. With that, the stream stays in the scope of a normal http request while still providing its advantages.

T4rk1n · 2026-07-28T13:16:26Z

@T4rk1n I solved that issue with my event streaming package by limiting the event stream to the normal 30 sec server timeout. With that, the stream stays in the scope of a normal http request while still providing its advantages.

This is too limiting for LLM purposes, conversations can last a long time.

It's entirely out of the application scope since we now support multiple backends and there is a big difference in how uvicorn and gunicorn handle this by default:

gunicorn: default is a flat 30s timeout, no reset on messages.
uvicorn: 60s seconds resets on every message, a keep alive pattern is all you need.

Then there is also reverse proxy (nginx & such) settings that might get in the way.

We'll configure plotly-cloud/enterprise to accept the streaming properly without user configuration and have some documentation for flask apps and gunicorn.

A generator callback streams by definition, so the opt-in keyword was redundant: the only two states it could express beyond the function's own shape were both already errors (stream=True on a non-generator, and a generator function registered without stream=True). Decorating a generator or async generator function is now enough. register_callback detects the generator and flips callback_map[...]["stream"], and the combination checks move with it: _validate_stream_callback rejects background=True, mcp_enabled=True and api_endpoint at registration, keyed off the detected function shape rather than the keyword. The clientside guard goes away with the keyword it rejected. Also drops `stream` from the renderer's ICallbackDefinition, which was never read -- the client keys off the NDJSON content type on HTTP and the stream / done markers on WebSocket responses, both of which live on the response types.

KoolADE85 · 2026-07-29T17:25:54Z

+                warnings.warn(
+                    f"Streaming callback '{callback_id}' is a synchronous "
+                    "generator; it will occupy a server worker for the whole "
+                    "stream. Define it with 'async def' so it runs on the "
+                    "event loop instead.",


I think this warning understates the potential problem.

Suggested change

warnings.warn(

f"Streaming callback '{callback_id}' is a synchronous "

"generator; it will occupy a server worker for the whole "

"stream. Define it with 'async def' so it runs on the "

"event loop instead.",

warnings.warn(

f"Callback '{callback_id}' is a synchronous generator; "

"it can slow down or stall the entire app under heavy load. "

"Consider 'async def' instead for better performance.",

KoolADE85 · 2026-07-29T17:49:37Z

+            if ws_callback.is_shutdown:
+                return {"status": "prevent_update"}


2 bugs around a similar root cause here (also applies to the async version below):

hot-reloading is broken while any stream is alive

streams are left running orphaned after a server shutdown

Here we handle client disconnect - maybe it's also a good place to handle server shutdowns?

Sync generator streaming is too limited/buggy (it occupies a server worker for the whole stream), so reject it at registration with a StreamCallbackError instead of warning. Async generators are now the only supported streaming path; remove the dead sync wrapper machinery. Fix a pre-existing bug on that async path: _astream_frames held one callback-context token across all yields, but the keepalive driver resumes each __anext__ in a freshly copied context (via ensure_future), so the reset ran in a different context and raised. Set/reset the context var around each generator step (and the on_error handler) instead. With stream_keepalive_interval enabled by default, this broke every async streaming callback that used dash.ctx/set_props.

The scheduler caps concurrent callbacks at 12 (prioritizedCallbacks). That cap was only ever meant to bound in-flight HTTP requests to the server (browsers allow ~6 connections per host), but it counted every callback, including ones that hold no HTTP connection. A long-lived streaming callback sits in `watched` for its whole life, so a handful of streams permanently exhaust the budget and starve everything else -- most visibly clientside callbacks, which run in-browser and make no request at all. Enabling websocket callbacks did not help: those callbacks were still counted. Extract the accounting into requestSlot.ts and only count callbacks that actually occupy an HTTP request slot (usesRequestSlot): clientside, streaming and websocket-routed callbacks are exempt and always dispatch. Background callbacks still count, since they poll over HTTP. The 12 is now a named constant, MAX_CONCURRENT_HTTP_CALLBACKS, with a comment on what it bounds. The renderer needs to know which callbacks stream to exempt them, so register_callback now sets a server-inferred `stream` flag on the client callback spec (distinct from the removed stream=True keyword; read only by the scheduler). Adds requestSlot unit tests for every exemption.

… job The streaming callback tests exercise async generators on the Flask backend, which requires flask[async] (the `async` extra). They were living in tests/unit and tests/integration, whose CI jobs (lint-unit, test-main) don't install that extra -- so they had no home that could actually run them. Move both into tests/streaming and add a dedicated streaming-tests job, gated on a streaming_changed path filter, that installs the async extra and runs pytest tests/streaming headless. Mirrors the websocket-tests job. The FastAPI-based tests/websocket/test_ws_stream.py stays put: it needs no flask[async] and is already covered by websocket-tests.

A cross-process key/value store and ordered publish/subscribe available on every app via dash.ctx.shared_storage / app.shared_storage, for sharing state between callbacks or across worker processes without an external service. The default LocalSharedStorage elects a single owner process per machine (binding an AF_UNIX socket on POSIX, TCP loopback on Windows -- the bind is the lease, re-elected on owner death) and serves the rest; a single-process deployment is its own owner and pays no socket overhead. Started lazily on first use, so it costs nothing until touched and never binds in a gunicorn preload master or the Flask reloader parent. Pass shared_storage=None to disable, or a BaseSharedStorage subclass/instance to swap the backend. Subscriptions are ordered and replayable: a consumer that reconnects resumes from its last-seen message out of a bounded buffer, and a buffer overrun surfaces as an explicit gap rather than a silent loss -- the property streaming will rely on. The wire codec is msgspec (msgpack), never pickle, so bytes off the socket cannot execute code; values must be JSON-compatible like dcc.Store. This is phase 1; streaming callbacks move onto it next.

Rebuilds HTTP streaming on the shared-storage pub/sub instead of one NDJSON connection per streaming callback. A browser will hold a single downlink keyed by a connection_id; every streaming callback publishes its frames -- tagged with a request_id -- to that connection's topic, and the downlink relays them, demultiplexed by the client. The store is the cross-process broker, so the worker running a callback and the worker holding the downlink need not be the same process. _stream_hub: topic naming, frame publishing, downlink relay (sync + async), and the pump that drives a callback's frame generator onto the topic (a sync driver for WSGI, an async task for ASGI). Flask dispatch reuses /_dash-update-component -- no new route: a streamDownlink request returns a StreamedCallbackResponse of the connection's envelopes through the existing NDJSON path; a streaming callback carrying a streamConnection pumps its frames on a background thread and returns a fast ack instead of holding the connection. Without a connection it falls back to inline NDJSON, so nothing changes for apps not using the multiplexed transport. Also: StoreEngine.closed lets subscriptions exit cleanly on owner shutdown instead of busy-looping. Quart/FastAPI backends and the client stream worker are next.

Port the downlink/uplink dispatch to the two ASGI backends, so all three backends now serve the multiplexed streaming transport by reusing /_dash-update-component: a streamDownlink request returns the connection's frames as NDJSON; a streaming callback carrying a streamConnection pumps its frames onto the topic and returns a fast ack. On ASGI the pump runs as a fire-and-forget task on the event loop (spawn_async_pump) rather than a thread, and the downlink is an async-generator marker (async_downlink_marker) -- both added to _stream_hub alongside the shared STREAM_ACK. WSGI (Flask) keeps the thread-based pump. Uplink is covered end-to-end over real HTTP on all three backends: callback -> pump -> topic -> drain, tagged by request id including the done terminal.

StreamClient owns a single downlink NDJSON connection for the whole page: every streaming callback POSTs its request (fast ack) tagged with a connection id + request id, and the one downlink relays all their frames, which the client routes back to each callback by request id. The downlink opens on the first streaming callback and closes once none remain in flight (dones match runnings); if it drops with work outstanding it reconnects, resuming from the last sequence it applied so buffered frames replay instead of being lost. To make that lossless resume possible, the shared-storage Subscription now exposes (sequence, message) pairs (iter_with_seq / aiter_with_seq, with the plain message iterators built on them), and the downlink envelope carries its seq. Browser-tested: uplink tagging, frame routing, resolve/reject on the terminal frame, multiplexing two callbacks over one connection, reconnect-from-cursor, and keepalive skipping. Not yet wired into handleServerside -- that is next.

The renderer now routes a streaming callback through the single downlink (StreamClient) when the server advertises it -- config.stream.enabled, set from whether the app has a shared-storage backend -- and it is not already on the WebSocket transport or a background job. handleStreamCallback applies running props, relays frames through applyStreamFrame as they arrive, and resolves empty like the WebSocket streaming path. When shared storage is disabled the old per-callback NDJSON path is used unchanged. This completes the multiplexed HTTP streaming transport end to end: one downlink per page instead of one connection per stream, brokered across worker processes by shared storage, Flask/Quart/FastAPI. Hosting the downlink in a SharedWorker (one connection per browser, shared across tabs) can follow.

Two bugs the unit tests missed (they injected a mock fetch and ran the server in-process): - StreamClient called native fetch as a method of the client object, which throws "Illegal invocation" -- fetch needs this===window. Bind it to globalThis (window on a page, self in a worker). - The Flask downlink used stream_with_context unconditionally, corrupting the request-context teardown on the async dispatch path (ValueError: token created in a different Context). Pass with_request_ctx per dispatch: True on the sync view, False on the async view. Also isolate each streaming integration test with its own shared-storage namespace (a single process runs many apps in the suite; production is one app per process). All six streaming integration tests now pass in a real browser over the multiplexed transport: progressive render, Patch, set_props, downstream triggering, error handling, and loading state.

Note the single-downlink multiplexed HTTP streaming transport: a page's streams share one connection instead of one per callback, so they no longer count against the browser's ~6-connections-per-host limit, and the connection resumes from its last sequence on reconnect without dropping frames.

An async subscription drives its blocking poll via loop.run_in_executor; when the event loop / executor is torn down (a client disconnects, or the app stops) that raises "cannot schedule new futures after shutdown" out of a streaming callback. Catch it and end the subscription cleanly instead of logging a traceback -- notably for long-lived pub/sub subscriptions (e.g. a live feed).

A streaming frame can carry dash.Patch objects (and components) that only Dash's JSON encoder understands. On a single process the frame was only serialized later by the downlink's to_json, so it worked; but in a multi-process deployment the publish crosses the shared-storage socket, whose data-only codec (msgspec) cannot encode a Patch -- raising "<write-only dash.Patch>... is not a dict" from a background pump thread. publish_frame now reduces the frame with Dash's to_json before it enters shared storage, which also matches exactly what the single-connection NDJSON path emits, so the client applies frames identically either way. Covered by a unit test (Patch -> plain, and the msgspec encode that used to raise) and a cross-process test that publishes a Patch frame over the socket.

sonarqubecloud · 2026-07-30T18:22:16Z

Quality Gate passed

Issues
21 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.6% Duplication on New Code

See analysis details on SonarQube Cloud

T4rk1n requested review from KoolADE85, camdecoster and ndrezn as code owners July 9, 2026 18:19

Fix CI: guard WS stream protocol tests on httpx

65c72b9

fastapi.testclient imports starlette.testclient, which requires httpx. Skip the protocol-level stream tests when httpx is not installed and add httpx to the CI requirements so the websocket job actually runs them.

KoolADE85 requested changes Jul 22, 2026

View reviewed changes

T4rk1n added 2 commits July 27, 2026 13:18

Merge branch 'dev' into feat/stream-callbacks

b2c340c

camdecoster assigned KoolADE85 Jul 28, 2026

KoolADE85 reviewed Jul 29, 2026

View reviewed changes

T4rk1n changed the title ~~Add streaming callbacks: @callback(..., stream=True)~~ Add streaming callbacks Jul 30, 2026

T4rk1n force-pushed the feat/stream-callbacks branch from e1d367a to 84551e5 Compare July 30, 2026 13:17

T4rk1n added 10 commits July 30, 2026 09:47

CHANGELOG: fill in PR number for shared storage entry

2fc54fd

T4rk1n changed the title ~~Add streaming callbacks~~ Streaming callbacks + shared storage (multiplexed streaming transport) Jul 30, 2026

T4rk1n added 2 commits July 30, 2026 14:08

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

Streaming callbacks + shared storage (multiplexed streaming transport) - #3888

Streaming callbacks + shared storage (multiplexed streaming transport)#3888
T4rk1n wants to merge 18 commits into
devfrom
feat/stream-callbacks

T4rk1n commented Jul 9, 2026 •

edited

Loading

Uh oh!

KoolADE85 left a comment

Uh oh!

KoolADE85 Jul 22, 2026

Uh oh!

T4rk1n Jul 24, 2026

Uh oh!

T4rk1n commented Jul 27, 2026

Uh oh!

chgiesse commented Jul 28, 2026

Uh oh!

T4rk1n commented Jul 28, 2026

Uh oh!

KoolADE85 Jul 29, 2026

Uh oh!

KoolADE85 Jul 29, 2026

Uh oh!

sonarqubecloud Bot commented Jul 30, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		if ws_callback.is_shutdown:
		return {"status": "prevent_update"}

Uh oh!

Uh oh!

Conversation

T4rk1n commented Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Summary

Streaming callbacks

Shared storage

Architecture — multiplexed streaming

Testing

Notes

Follow-up (not in this PR)

Uh oh!

KoolADE85 left a comment

Choose a reason for hiding this comment

Uh oh!

KoolADE85 Jul 22, 2026

Choose a reason for hiding this comment

Uh oh!

T4rk1n Jul 24, 2026

Choose a reason for hiding this comment

Uh oh!

T4rk1n commented Jul 27, 2026

Uh oh!

chgiesse commented Jul 28, 2026

Uh oh!

T4rk1n commented Jul 28, 2026

Uh oh!

KoolADE85 Jul 29, 2026

Choose a reason for hiding this comment

Uh oh!

KoolADE85 Jul 29, 2026

Choose a reason for hiding this comment

Uh oh!

sonarqubecloud Bot commented Jul 30, 2026

Quality Gate passed

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

T4rk1n commented Jul 9, 2026 •

edited

Loading