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

[KoolADE85]

The loop here ends on client disconnect, but not always on server shutdown. As a result, streams continue even after a Ctrl+C (see the repro app below).
We should respond to a shutdown event the way ws.py does.

import asyncio
from dash import Dash, Input, Output, callback, html

app = Dash(__name__, backend="fastapi")
server = app.server
app.layout = html.Div([
    html.H3("If you click the button, Ctrl+C won't stop this server"),
    html.Button("stream", id="btn", n_clicks=0),
    html.Div(id="out"),
])

@callback(
    Output("out", "children"),
    Input("btn", "n_clicks"),
    prevent_initial_call=True
)
async def long_stream(n):
    for i in range(600):
        await asyncio.sleep(0.5)
        print(f"tick {i}")
        yield f"tick {i}"

if __name__ == "__main__":
    app.run(debug=True)

[KoolADE85]

Here we are implicitly trusting the connectionId sent by a client.
If the ID is somehow leaked or derived, then anyone can use the ID to join someone else's stream and receive its data. Suggest generating the ID server-side and tying it to a cookie or session.

See also _serve_uplink where a connectionId could also be used to inject frames into someone else's stream.

This also applies to the fastapi & quart backends.

[KoolADE85]

This cursor is per-page, monotonic, and is never reset.
On reconnect, the client sends from: this.cursor (line 160 below).
This seems correct as long as the server is running, but a server restart introduces a flaw: the server's cursor must catch up to the client before the stream can resume.

See this example app (works with gunicorn as well as the dev server)

import time

from dash import Dash, Input, Output, callback, dcc, html

app = Dash(__name__, backend="flask")
server = app.server

app.layout = html.Div(
    children=[
        html.H3("Streamed clock stalls silently after a server restart"),
        html.P("Watch it tick a few seconds, then restart the server."),
        html.P("Observe that it stays frozen for as many seconds as the previous server was alive."),
        html.Pre(id="clock", children="waiting for first tick…"),
        dcc.Interval(id="kick", interval=1000),
    ],
)


@callback(
    Output("clock", "children"),
    Input("kick", "n_intervals"),
    prevent_initial_call=True,
)
async def stream_clock(_):
    yield f"serverside clock: {time.strftime('%H:%M:%S')}"


if __name__ == "__main__":
    app.run(debug=True, port=8050)