Streaming callbacks with multiplexed transport

[T4rk1n]

Split out of #3888 and stacked on #3930 (shared storage) — this is the streaming half. It targets feat/shared_storage; review it as the diff on top of that PR, and it will be retargeted to dev once #3930 merges.

Summary

This PR adds streaming callbacks and the multiplexed HTTP transport that carries them across worker processes, built on the backend-agnostic shared storage primitive from #3930.

• 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.
• 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.
• Shared storage (dash.ctx.shared_storage / app.shared_storage, from Shared storage: backend-agnostic state manager + pub/sub #3930) — a cross-process key/value store + ordered publish/subscribe, used internally here as the broker for streaming.

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 (base PR #3930)

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. Pass shared_storage=None to disable, or a BaseSharedStorage subclass/instance to swap the backend (DiskcacheSharedStorage, RedisSharedStorage, ... — see #3930).

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

Loading

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

• Streaming transport: uplink/downlink end-to-end over real HTTP on Flask, Quart, and FastAPI.
• Renderer: StreamClient routing, multiplexing, reconnect-from-cursor, keepalive (karma).
• Shared-storage broker: the iter_with_seq / aiter_with_seq sequence-aware subscriptions the downlink resumes from, across all backends (test_stream_hub.py).
• 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

• Shared-storage values must be JSON-compatible (like dcc.Store).
• The default in-memory backend is not durable: if the owner process dies, a survivor re-elects with an empty store. For multi-pod deployments behind a load balancer use RedisSharedStorage (see Shared storage: backend-agnostic state manager + pub/sub #3930).

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.

[sonarqubecloud]

Quality Gate passed

Issues
16 New issues
0 Accepted issues

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

See analysis details on SonarQube Cloud