Require shared storage for HTTP streaming callbacks

.ai/ARCHITECTURE.md

@@ -926,8 +926,14 @@ Dash(shared_storage=RedisSharedStorage(url="redis://host:6379"))
 ```
 
 When disabled, accessing `app.shared_storage` / `dash.ctx.shared_storage`
-raises `SharedStorageError`, and streaming callbacks still work, each on its own
-connection instead of the multiplexed downlink.
+raises `SharedStorageError`. Streaming callbacks over HTTP need it: server setup
+raises `StreamCallbackError` naming them (`_validate.validate_stream_callbacks`),
+so an app cannot stream on a fallback locally and then break once deployed.
+WebSocket streaming callbacks (`websocket_callbacks=True` or `websocket=True`
+on a backend with `websocket_capability`, i.e. FastAPI or Quart) send frames on
+the socket and do not need it; on Flask those flags change nothing and the check
+still applies. Browsers without `SharedWorker` do not use the WebSocket
+transport and stream over HTTP, one connection per stream.
 
 **`LocalSharedStorage`** — in-memory, no external service. On first use every
 worker races to become the single **owner** for the machine by binding a stable

CHANGELOG.md

@@ -5,6 +5,7 @@ This project adheres to [Semantic Versioning](https://semver.org/).
 ## [Unreleased]
 
 ### Added
+- [#4099](https://github.com/plotly/dash/pull/4099) Streaming callbacks over HTTP now require shared storage and fail at server setup without it, instead of falling back to one connection per stream; WebSocket streaming callbacks do not need it.
 - [#4095](https://github.com/plotly/dash/pull/4095) Make shared storage and the multiplexed streaming downlink opt-in (`shared_storage=` or `DASH_SHARED_STORAGE`), and move `msgspec` to a new `dash[shared_storage]` extra for the persistent `LocalSharedStorage` modes (`dash[diskcache]` and `dash[redis]` include it).
 - [#4026](https://github.com/plotly/dash/pull/4026) Add `DASH_SECRET_KEY` so a hosting platform can give every worker and pod the same signing secret for page, background and stream tokens without editing the app (`server.secret_key` still wins and is not changed).
 - [#4026](https://github.com/plotly/dash/pull/4026) Add `DASH_SHARED_STORAGE` to pick the shared-storage backend when the app does not pass `shared_storage=`: `local`, `none`, `diskcache:///abs/path`, or a `redis://` / `rediss://` URL.

dash/_shared_storage/_codec.py

@@ -1,10 +1,11 @@
 """Codec for the shared-storage transport and the persistent backends.
 
 Uses ``msgspec`` (msgpack: fast, compact) from the ``dash[shared_storage]``
-extra, also part of ``dash[diskcache]`` and ``dash[redis]``. Without it, only the in-memory transport works and it falls back to
-stdlib JSON; the backends that write encoded bytes somewhere lasting (disk
-persistence, diskcache, Redis) call :func:`require_msgspec` so the stored
-format never depends on what happens to be installed. Both formats are
+extra, also part of ``dash[diskcache]`` and ``dash[redis]``. Without it, only
+the in-memory transport works and it falls back to stdlib JSON; the backends
+that write encoded bytes somewhere lasting (disk persistence, diskcache, Redis)
+call :func:`require_msgspec` so the stored format never depends on what
+happens to be installed. Both formats are
 data-only, so bytes read off the socket cannot execute code.
 
 Payloads must be JSON-compatible (dict / list / str / int / float / bool /

dash/_validate.py

@@ -564,6 +564,33 @@ def validate_background_callbacks(callback_map):
             )
 
 
+def validate_stream_callbacks(
+    callback_map, shared_storage_enabled, websocket_callbacks, websocket_capable
+):
+    # Over HTTP a stream needs shared storage to broker its frames. Fail the same
+    # way locally as in a deployment rather than streaming one way here and
+    # breaking there. A websocket flag only counts on a backend that serves
+    # WebSockets; on Flask those callbacks still run over HTTP.
+    if shared_storage_enabled:
+        return
+    streams = [
+        callback_id
+        for callback_id, callback in callback_map.items()
+        if callback.get("stream")
+        and not (
+            websocket_capable and (websocket_callbacks or callback.get("websocket"))
+        )
+    ]
+    if streams:
+        raise exceptions.StreamCallbackError(
+            f"Streaming callbacks {streams} run over HTTP and need shared "
+            "storage. Pass Dash(shared_storage=dash.LocalSharedStorage) (or a "
+            "Diskcache or Redis backend), set DASH_SHARED_STORAGE, or run them "
+            "over WebSocket on a backend that supports it (FastAPI or Quart, "
+            "with websocket_callbacks=True or websocket=True on the callback)."
+        )
+
+
 def validate_duplicate_output(
     output, prevent_initial_call, config_prevent_initial_call
 ):

dash/dash.py

@@ -480,8 +480,9 @@ class Dash(ObsoleteChecker):
     :param shared_storage: Backend for ``dash.ctx.shared_storage`` and the
         streaming-callback transport: a ``BaseSharedStorage`` subclass or
         instance, or ``None`` to disable. Off by default: pass
-        ``LocalSharedStorage`` for one machine or pod. Without it, streaming
-        callbacks still work, each on its own connection. When not passed, the
+        ``LocalSharedStorage`` for one machine or pod. Streaming callbacks
+        over HTTP need it, and server setup raises without it; WebSocket
+        streaming callbacks do not. When not passed, the
         ``DASH_SHARED_STORAGE`` environment variable can choose it: ``local``,
         ``none``, ``diskcache:///absolute/path``, or a ``redis://`` /
         ``rediss://`` URL. An explicit argument, including ``None``, always
@@ -1270,14 +1271,14 @@ def _config(self):
                 "heartbeat_interval": self._websocket_heartbeat_interval,
             }
 
-        # Streaming callbacks use the single multiplexed downlink only when a
-        # shared-storage backend is available to broker frames across workers;
-        # otherwise the client streams each callback on its own connection.
-        # The worker script hosts that downlink in a SharedWorker so every tab
-        # of the browser shares one connection (browsers cap connections per
-        # host); the renderer falls back to a per-tab downlink without it.
-        # "mode" is how the downlink is served: one open connection (ASGI) or
-        # polling (WSGI, where an open response would pin a worker thread).
+        # HTTP streaming callbacks need shared storage to broker frames across
+        # workers onto the single multiplexed downlink (_setup_server refuses
+        # them without it). The worker script hosts that downlink in a
+        # SharedWorker so every tab of the browser shares one connection
+        # (browsers cap connections per host); the renderer falls back to a
+        # per-tab downlink without SharedWorker support. "mode" is how the
+        # downlink is served: one open connection (ASGI) or polling (WSGI,
+        # where an open response would pin a worker thread).
         config["stream"] = {"enabled": self.shared_storage_enabled}
         if self.shared_storage_enabled:
             config["stream"].update(
@@ -1933,6 +1934,12 @@ def _setup_server(self):
             _callback.GLOBAL_CALLBACK_LIST.clear()
 
             _validate.validate_background_callbacks(self.callback_map)
+            _validate.validate_stream_callbacks(
+                self.callback_map,
+                self.shared_storage_enabled,
+                self._websocket_callbacks,
+                self.backend.websocket_capability,
+            )
 
             cancels = {}
 

tests/streaming/conftest.py

@@ -18,6 +18,9 @@
 def _isolate_shared_storage(monkeypatch):
     namespace = f"streamtest-{uuid.uuid4().hex[:12]}"
     monkeypatch.setattr(_local, "_default_namespace", lambda: namespace)
+    # HTTP streaming needs shared storage, so apps here get it unless they pass
+    # their own.
+    monkeypatch.setenv("DASH_SHARED_STORAGE", "local")
     yield
 
 

tests/streaming/test_stream_callbacks_unit.py

@@ -521,3 +521,58 @@ async def scenario():
         },
     ]
     assert all(m["rid"] == "rid" for _, m in published)
+
+
+def _stream_app(**kwargs):
+    callback_kwargs = (
+        {"websocket": kwargs.pop("websocket")} if "websocket" in kwargs else {}
+    )
+    app = Dash(__name__, **kwargs)
+    app.layout = html.Div([html.Button(id="btn"), html.Div(id="out")])
+
+    @app.callback(
+        Output("out", "children"), Input("btn", "n_clicks"), **callback_kwargs
+    )
+    async def stream(n):
+        yield n
+
+    return app
+
+
+def test_stcb025_http_stream_without_shared_storage_fails_at_setup():
+    # The error must come at setup, the same locally and once deployed, instead
+    # of streaming on a fallback that breaks behind a real deployment.
+    app = _stream_app(shared_storage=None)
+    with pytest.raises(StreamCallbackError, match="need shared storage"):
+        app._setup_server()  # pylint: disable=protected-access
+
+
+def test_stcb026_http_stream_without_shared_storage_fails_first_request():
+    app = _stream_app(shared_storage=None)
+    app.server.testing = True
+    client = app.server.test_client()
+    for _ in range(2):
+        with pytest.raises(StreamCallbackError, match="out.children"):
+            client.get("/")
+
+
+def test_stcb027_websocket_streams_do_not_need_shared_storage():
+    pytest.importorskip("fastapi")
+    _stream_app(
+        shared_storage=None, backend="fastapi", websocket_callbacks=True
+    )._setup_server()  # pylint: disable=protected-access
+    _stream_app(
+        shared_storage=None, backend="fastapi", websocket=True
+    )._setup_server()  # pylint: disable=protected-access
+
+
+@pytest.mark.parametrize("kwargs", [{"websocket_callbacks": True}, {"websocket": True}])
+def test_stcb029_websocket_flag_on_flask_still_needs_shared_storage(kwargs):
+    # Flask has no WebSocket route, so these streams would run over HTTP.
+    app = _stream_app(shared_storage=None, **kwargs)
+    with pytest.raises(StreamCallbackError, match="need shared storage"):
+        app._setup_server()  # pylint: disable=protected-access
+
+
+def test_stcb028_http_stream_with_shared_storage_sets_up():
+    _stream_app()._setup_server()  # pylint: disable=protected-access

tests/streaming/test_stream_transport.py

@@ -167,7 +167,7 @@ def test_downlink_without_shared_storage_is_rejected(flavor):
     # request for one gets 403 instead of a SharedStorageError 500.
     body = {"streamDownlink": {"from": 0}}
     if flavor == "flask":
-        app = Dash(__name__)
+        app = Dash(__name__, shared_storage=None)
         app.layout = html.Div()
         resp = app.server.test_client().post(_uplink_url(app), json=body)
         assert resp.status_code == 403
@@ -177,14 +177,14 @@ def test_downlink_without_shared_storage_is_rejected(flavor):
         from fastapi import FastAPI
         from fastapi.testclient import TestClient
 
-        app = Dash(__name__, server=FastAPI())
+        app = Dash(__name__, server=FastAPI(), shared_storage=None)
         app.layout = html.Div()
         app._setup_server()  # pylint: disable=protected-access
         with TestClient(app.server) as client:
             assert client.post(_uplink_url(app), json=body).status_code == 403
         return
     quart = pytest.importorskip("quart")
-    app = Dash(__name__, server=quart.Quart(__name__))
+    app = Dash(__name__, server=quart.Quart(__name__), shared_storage=None)
     app.layout = html.Div()
     app._setup_server()  # pylint: disable=protected-access