Shared storage: backend-agnostic state manager + pub/sub - #3930
Open
T4rk1n wants to merge 16 commits into
Open
Conversation
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.
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).
LocalSharedStorage only reaches one container, which fragments across pods behind a load balancer (Dash Enterprise scales apps multi-pod). Add two more BaseSharedStorage backends, selected via shared_storage=: - DiskcacheSharedStorage: KV + append-log pub/sub on a diskcache.Cache, shared by every process on one host. Single-machine parity; not for pods. - RedisSharedStorage: KV + Redis Streams pub/sub (atomic INCR+XADD, MAXLEN- bounded, SharedStorageGap past the trimmed floor). Correct across pods. Both reuse a shared PollingSubscription. Tests mirror the local suite; the Redis tests run against the CI redis service and skip when none is reachable.
Give RedisSharedStorage its own extra instead of borrowing redis from the celery extra. Same pin as celery.txt for now; in Dash 5.0 redis moves out of the celery extra and lives here only. Point the ImportError, docs, and CI at dash[redis].
The typing-compliance test runs mypy over an app that imports dash, surfacing 7 errors in the shared-storage code: - redis/diskcache: ignore the optional-dep imports (not installed / unstubbed in the typing job), matching the background managers' pattern - annotate _publish_script and Dash._shared_storage_instance so they aren't inferred as None-only - assert the coordinator's family/token are set before connect_to_owner
Parametrize the Dash-app integration tests over all three backends (Local, Diskcache, Redis-if-reachable) and add scenarios that exercise the real callback-dispatch pipeline: KV round-trip, state shared across two callbacks, a counter persisting across independent dispatches, and two separate Dash apps sharing one backend store. Redis params skip when no server is reachable.
Replace the HTTP-dispatch integration tests with real dash_duo browser tests: start a server, click buttons, assert on the rendered DOM. The counter scenario (one callback writes a shared value, another reads it) runs on Local, Diskcache, and Redis-when-reachable. These need a browser, so move the whole tests/shared_storage suite out of the lint-unit job into a dedicated shared-storage CI job with a Redis service and setup-chrome (modeled on the background-callbacks job), run with --headless.
KoolADE85
reviewed
Jul 31, 2026
| @@ -0,0 +1,37 @@ | |||
| """Codec for the shared-storage socket transport. | |||
|
|
|||
| Prefers ``msgspec`` (msgpack: fast, compact) and falls back to stdlib ``json``. | |||
Contributor
There was a problem hiding this comment.
This looks like a Claude-ism: is there some condition that demands a real reason to "fall back"? I'd argue we should pick 1 implementation and keep it DRY.
Comment on lines
+19
to
+22
| # Per-topic replay buffer size. Sized to cover a reconnect window comfortably; | ||
| # a producer that outruns a disconnected consumer past this raises a gap rather | ||
| # than dropping messages silently. | ||
| DEFAULT_BUFFER = 2048 |
Contributor
There was a problem hiding this comment.
I'm a bit uneasy about this: it looks like this means we will keep old messages around in a buffer? We are going to have to maintain code paths for this, and it has a limited use-case.
In a traditional callback, if there's a network blip, you must re-trigger the callback anyway. It's a design choice that I feel we should not change just because we're streaming now.
As implemented, the feature is "general purpose" (very cool btw!). It will also be useful for publishing larger datasets from a shared location. A callback that publishes a 2MB dataset will silently retain 2048 of them (4GB per topic!). Could OOM or fill up disk space, and it's not obvious that this will happen.
I'm inclined to suggest that old data just be discarded - no buffer. If the client falls behind for some reason, then they need to trigger the callback again.
Alternatively, I suggest making the buffer 0 by default: safe out-of-the-box but can be enabled if you know what you're doing.
Contributor
Author
There was a problem hiding this comment.
The buffer is to ensure the frames are not dropped, there can be a slight delay before a subscribe actually start or burst of events (diskcache has polling, so it need buffer). We can't set the buffer to 0 since that would drop all messages. I think 32 would be a good middle ground.
msgspec is in requirements/install.txt, so the try/except ImportError JSON branch in _codec.py was dead code. Always use msgspec msgpack and remove the unused CODEC constant. Update transport docstring and ARCHITECTURE.md accordingly.
DEFAULT_BUFFER was 2048: each pub/sub topic retained up to 2048 arbitrary user payloads, an unbounded-ish memory hazard given users publish unpredictable data. Drop the default to 32 -- still enough to survive normal publish bursts and brief reconnects, ~64x less memory. Deployments needing a wider reconnect window pass buffer_size explicitly. Note: 0 is not a safe default -- it drops every message (deque(maxlen=0) discards appends, diskcache evicts the just-written seq, Redis XADD MAXLEN 0 trims to empty), since the buffer is the delivery path, not just replay history.
set(key, value, ttl=<seconds>) gives a key a bounded lifetime; reads return the default once it elapses. ttl=None (default) never expires. Expiry is lazy/best-effort -- an expired key is dropped on its next read. Each backend uses its native mechanism: a monotonic deadline on the in-memory owner engine (value stored as (value, deadline), purged lazily on get), diskcache expire=, Redis PX (milliseconds, so sub-second ttl survives). ttl is plumbed through the local socket transport as a 4th element of the set request. Pairs with the smaller default buffer: TTL lets apps bound key/value memory the same way the buffer bounds pub/sub memory.
Add an opt-in `mode` argument to LocalSharedStorage so its key/value store can survive process restarts and owner re-election (pub/sub stays transient): - "memory" (default): unchanged, in-memory only. - "persist": write-through to disk on every set/delete. - "persist-reset": in-memory speed, flushed every `flush_interval` seconds and on clean exit. Only the elected owner persists; recovery runs on every election, so data also survives owner death. The on-disk store is chunked (keys sharded across msgpack chunk files via a key->chunk index, so a mutation rewrites only the affected chunk; atomic temp-file + os.replace), adapted from raposa's BinaryStorage. TTLs are stored as wall-clock deadlines and converted to/from the engine's monotonic deadlines across a restart. Default store location is a per-namespace folder under the user cache dir, overridable via `path`. Persistence hooks into StoreEngine.set/delete (the single mutation chokepoint for both owner-local and remote-client writes). _flush serializes its snapshot+write so concurrent write-through flushers cannot clobber a newer value; the persist-reset flush thread survives a failed cycle; and the final flush is wired through StoreEngine.close().
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Split out from #3888 (Streaming callbacks + shared storage) so it can be reviewed on its own. This PR is phase 1: the backend-agnostic shared storage layer. The streaming work builds on top of it and stays on #3888 for now.
What this adds
Shared storage: a backend-agnostic state manager available on every app via
dash.ctx.shared_storage(andapp.shared_storage).get/set/delete) and ordered publish/subscribe (publish/subscribe) for sharing state between callbacks or across worker processes without an external service.LocalSharedStorageelects a single owner process per machine (binding anAF_UNIXsocket 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.shared_storage=NonetoDash(...)to disable, or aBaseSharedStoragesubclass/instance to swap the backend.dcc.Store).Contributor Checklist
pytest tests/shared_storage— 17 passed).tests/shared_storage/, plus a dedicated CI job).CHANGELOG.md.