Repository navigation
Conversation
Adapt tishun#2 to the current metadata heartbeat field and extend API. Renew owned running payload, metadata, liveness and active-slot retention under WATCH without shortening longer extensions or reviving missing history. Cover idle producers and takeover, completion and history-loss races against real Redis. Preserve existing public APIs and document retention timing.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## feature/resumable-streams #98 +/- ##
=============================================================
+ Coverage 96.14% 96.17% +0.02%
=============================================================
Files 17 17
Lines 3737 3761 +24
=============================================================
+ Hits 3593 3617 +24
Misses 144 144
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
jruaux
marked this pull request as ready for review
October 28, 2026 19:29
Author
|
Superseded by #99. #99 replaces the per-stream heartbeat that this PR fixes: each run is one Redis Stream whose lifetime is set once at open(), so a live producer's history can no longer expire during an idle wait, and extend() is gone. This PR's stream_case fixture and gated_exec helper are carried over into #99 as tests/stream_support.py, and the stream tests run on standalone and OSS Cluster (and Enterprise when STREAMS_TEST_ENTERPRISE_URL is set). Reopen if #99 is not adopted. |
tishun
added a commit
that referenced
this pull request
Oct 30, 2026
* refactor(streams): simplify the write path with creation-time TTLs and a process lease An append is now XADD plus one HMGET of status and cancel in a plain pipeline: 2 server commands instead of 8 in MULTI. open() writes a stream.start entry and gives every key a stream_max_run TTL, so nothing slides and a live producer's keys never lapse during an idle wait. An append past half the lifetime extends it. The per-stream heartbeat, keepalive(), attach(), writer fencing, StreamFenced and the started and alive meta fields are gone. Each process keeps one lease: a set of the runs it still has a writer for, rewritten every stream_lease_interval, plus a cancel set other processes write to. Readers end a running run missing from its owner's lease as producer_lost; open() takes over a slot held by such a run. suspend() and resume() hand a run to another process; a resume of a run its owner still vouches for raises StreamBusy. append() and finish() raise StreamFinished when a reader already ended the run, cancel() ends a suspended run, readers stop on a run whose entries expired, and each lifespan gets its own process ID so forked workers hold separate leases. Docs still describe the old API. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): release transactional pipelines with async with On redis-py 8.0.1, an async RedisCluster whose transactional pipeline was executed outside async with can fail to close afterwards with 'Event loop is closed'. Every pipeline the backend runs now releases its connection on exit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): run the write-path tests on every Redis topology Port the stream_case fixture and gated_exec from #98. stream_case runs a test against standalone Redis, and against an OSS Cluster or Redis Enterprise when STREAMS_TEST_CLUSTER_URL or STREAMS_TEST_ENTERPRISE_URL is set, with its own key prefix and cleanup. test_stream_topologies covers appends that write no TTLs, a crashed process reported lost and its scope freed, a dropped writer reported lost, a cancel reaching another process during a tool call, suspend and a once-only resume, cancelling a suspended run, a reader whose end aborts when the run is resumed before its EXEC, and concurrent starts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(streams): share the retry loop, final entry and vouching One _retry() helper replaces the three retry loops in _pipeline, _guarded and _create. _final_xadd() builds the final entry for finish, a cancel of a suspended run and a reader ending a lost run. _vouching() holds a run in the lease while open() or resume() decides, and _check_live() is the one guard for append, extend and suspend. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): keep each cross-process scenario in one place Drop the standalone-only copies of eight scenarios that test_stream_topologies runs on every topology, after moving their extra assertions there: suspended status, TTL and sequence numbers, a second cancel, and no stray meta after concurrent starts. collect(), crash() and configure() move to tests/stream_support.py, replacing per-file copies. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): test every Redis call against a real server Remove the hand-written client and pipeline fakes and the MagicMock/AsyncMock clients from the stream unit tests. What they checked now runs on real Redis, on every topology stream_case is given: the support probe on RESP2 and RESP3 and against an unreachable server, detection caching, the commands one append runs (read from INFO commandstats: one XADD, one HMGET), lease renewal and cancel collection, a dropped writer leaving the lease, every producer state a quiet reader can find, and resume() and reads through RESP2 and RESP3 clients. The unit tests left make no Redis call; a backend there gets a client factory that fails the test if called. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(streams): keep each run in one Redis key A run is now one Redis Stream, <prefix>:stream:<scope>:<id>, and nothing else: the meta hash, the scope's active slot, the process lease and the cancel sets are gone, and so are every WATCH transaction and the per-stream heartbeat. open() writes the start entry and sets the run's lifetime in one MULTI; created is the reply of EXPIRE ... NX, and a creating open widens the stream's IDMP window to the run's life. append() is one XADD. finish(), cancel() and a reader's producer_lost end share one IDMP pair, so a run has one final entry, and EXPIRE ... LT lets ending only shorten it. A reader that sees nothing for stream_silence_timeout ends the run; keepalive(), which spawn() runs, pings a quiet producer. Removed: active(), exclusive, StreamBusy, suspend(), resume(), and the settings stream_gone_ttl, stream_lease_interval and stream_lease_ttl. Added: stream_silence_timeout. A cancel from another process ends the run for readers but no longer reaches its producer. The incident triage example keeps its own current-run pointer per engineer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): check the cancel route on a server without streams The test app no longer has an active route, so TestUnsupportedServer hit a 404 instead of the 503 it asserts; it runs only against a server without stream support, the redis 7.4 CI job. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): keep a run's end through late writes, retries and restarts No write trims a run any more: XADD MAXLEN could remove the final entry once a producer that missed a cancel wrote past it. stream_maxlen now caps each writer's events instead, and append() raises past it. Appends from one writer go out one at a time, so a retry is never overtaken by enough appends to push its IDMP pair out of Redis's 100-pair history. open() sets the stream's IDMP window (a constant day) inside its creating transaction, so no cancel can land before it and lose its pair; a retried start sends the same value, which clears nothing. extend() is gone: it could lengthen a run another process had already ended. stream_done_ttl must be shorter than stream_max_run. The incident triage example tells an ended run by its TTL rather than its last 20 entries, and uses the renamed settings; its claim has tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ci: run the integration suite on Redis 8.6.2 8.6.2 is the oldest server resumable streams support, and the IDMP behaviour the stream tests rely on was only checked on 8.8. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(streams): describe one key per run The guide, the API reference and the configuration page still described the meta hash, the scope's active slot, the producer heartbeat, active(), exclusive, StreamBusy, attach(), StreamFenced, extend(), trimming by stream_maxlen and the removed settings. They now describe the run as one stream, the fixed run lifetime, the per-writer event cap, silence detection with keepalive(), a cancel that does not reach a producer in another process, and how an application keeps its own current-run pointer for one run at a time. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): allow any stream_done_ttl Requiring stream_done_ttl shorter than stream_max_run only served the incident triage example, which tells an ended run by its TTL; it rejected settings that work, such as REDIS_STREAM_DONE_TTL=3600 with the default lifetime. Ending only ever shortens a run, so a longer stream_done_ttl means a finished run keeps what lifetime it has left. The example now checks the assumption itself at startup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Nitpicking : address documentation changes # Conflicts: # docs/api/configuration.md # docs/api/reference.md # docs/guide/streams.md --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Tihomir Mateev <tihomir.mateev@gmail.com>
tishun
added a commit
that referenced
this pull request
Oct 30, 2026
* refactor(streams): simplify the write path with creation-time TTLs and a process lease An append is now XADD plus one HMGET of status and cancel in a plain pipeline: 2 server commands instead of 8 in MULTI. open() writes a stream.start entry and gives every key a stream_max_run TTL, so nothing slides and a live producer's keys never lapse during an idle wait. An append past half the lifetime extends it. The per-stream heartbeat, keepalive(), attach(), writer fencing, StreamFenced and the started and alive meta fields are gone. Each process keeps one lease: a set of the runs it still has a writer for, rewritten every stream_lease_interval, plus a cancel set other processes write to. Readers end a running run missing from its owner's lease as producer_lost; open() takes over a slot held by such a run. suspend() and resume() hand a run to another process; a resume of a run its owner still vouches for raises StreamBusy. append() and finish() raise StreamFinished when a reader already ended the run, cancel() ends a suspended run, readers stop on a run whose entries expired, and each lifespan gets its own process ID so forked workers hold separate leases. Docs still describe the old API. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): release transactional pipelines with async with On redis-py 8.0.1, an async RedisCluster whose transactional pipeline was executed outside async with can fail to close afterwards with 'Event loop is closed'. Every pipeline the backend runs now releases its connection on exit. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): run the write-path tests on every Redis topology Port the stream_case fixture and gated_exec from #98. stream_case runs a test against standalone Redis, and against an OSS Cluster or Redis Enterprise when STREAMS_TEST_CLUSTER_URL or STREAMS_TEST_ENTERPRISE_URL is set, with its own key prefix and cleanup. test_stream_topologies covers appends that write no TTLs, a crashed process reported lost and its scope freed, a dropped writer reported lost, a cancel reaching another process during a tool call, suspend and a once-only resume, cancelling a suspended run, a reader whose end aborts when the run is resumed before its EXEC, and concurrent starts. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(streams): share the retry loop, final entry and vouching One _retry() helper replaces the three retry loops in _pipeline, _guarded and _create. _final_xadd() builds the final entry for finish, a cancel of a suspended run and a reader ending a lost run. _vouching() holds a run in the lease while open() or resume() decides, and _check_live() is the one guard for append, extend and suspend. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): keep each cross-process scenario in one place Drop the standalone-only copies of eight scenarios that test_stream_topologies runs on every topology, after moving their extra assertions there: suspended status, TTL and sequence numbers, a second cancel, and no stray meta after concurrent starts. collect(), crash() and configure() move to tests/stream_support.py, replacing per-file copies. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): test every Redis call against a real server Remove the hand-written client and pipeline fakes and the MagicMock/AsyncMock clients from the stream unit tests. What they checked now runs on real Redis, on every topology stream_case is given: the support probe on RESP2 and RESP3 and against an unreachable server, detection caching, the commands one append runs (read from INFO commandstats: one XADD, one HMGET), lease renewal and cancel collection, a dropped writer leaving the lease, every producer state a quiet reader can find, and resume() and reads through RESP2 and RESP3 clients. The unit tests left make no Redis call; a backend there gets a client factory that fails the test if called. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * refactor(streams): keep each run in one Redis key A run is now one Redis Stream, <prefix>:stream:<scope>:<id>, and nothing else: the meta hash, the scope's active slot, the process lease and the cancel sets are gone, and so are every WATCH transaction and the per-stream heartbeat. open() writes the start entry and sets the run's lifetime in one MULTI; created is the reply of EXPIRE ... NX, and a creating open widens the stream's IDMP window to the run's life. append() is one XADD. finish(), cancel() and a reader's producer_lost end share one IDMP pair, so a run has one final entry, and EXPIRE ... LT lets ending only shorten it. A reader that sees nothing for stream_silence_timeout ends the run; keepalive(), which spawn() runs, pings a quiet producer. Removed: active(), exclusive, StreamBusy, suspend(), resume(), and the settings stream_gone_ttl, stream_lease_interval and stream_lease_ttl. Added: stream_silence_timeout. A cancel from another process ends the run for readers but no longer reaches its producer. The incident triage example keeps its own current-run pointer per engineer. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * test(streams): check the cancel route on a server without streams The test app no longer has an active route, so TestUnsupportedServer hit a 404 instead of the 503 it asserts; it runs only against a server without stream support, the redis 7.4 CI job. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): keep a run's end through late writes, retries and restarts No write trims a run any more: XADD MAXLEN could remove the final entry once a producer that missed a cancel wrote past it. stream_maxlen now caps each writer's events instead, and append() raises past it. Appends from one writer go out one at a time, so a retry is never overtaken by enough appends to push its IDMP pair out of Redis's 100-pair history. open() sets the stream's IDMP window (a constant day) inside its creating transaction, so no cancel can land before it and lose its pair; a retried start sends the same value, which clears nothing. extend() is gone: it could lengthen a run another process had already ended. stream_done_ttl must be shorter than stream_max_run. The incident triage example tells an ended run by its TTL rather than its last 20 entries, and uses the renamed settings; its claim has tests. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * ci: run the integration suite on Redis 8.6.2 8.6.2 is the oldest server resumable streams support, and the IDMP behaviour the stream tests rely on was only checked on 8.8. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * docs(streams): describe one key per run The guide, the API reference and the configuration page still described the meta hash, the scope's active slot, the producer heartbeat, active(), exclusive, StreamBusy, attach(), StreamFenced, extend(), trimming by stream_maxlen and the removed settings. They now describe the run as one stream, the fixed run lifetime, the per-writer event cap, silence detection with keepalive(), a cancel that does not reach a producer in another process, and how an application keeps its own current-run pointer for one run at a time. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * fix(streams): allow any stream_done_ttl Requiring stream_done_ttl shorter than stream_max_run only served the incident triage example, which tells an ended run by its TTL; it rejected settings that work, such as REDIS_STREAM_DONE_TTL=3600 with the default lifetime. Ending only ever shortens a run, so a longer stream_done_ttl means a finished run keeps what lifetime it has left. The example now checks the assumption itself at startup. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> * Nitpicking : address documentation changes # Conflicts: # docs/api/configuration.md # docs/api/reference.md # docs/guide/streams.md --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com> Co-authored-by: Tihomir Mateev <tihomir.mateev@gmail.com>
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
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.
A producer can remain alive during a long tool call while its stream history expires. Its next append then fails with StreamExpired. Heartbeats now renew active payload and metadata retention together with liveness and the run's owned active slot.
Renewal uses a bounded same-slot WATCH/MULTI/EXEC transaction to check ownership, running state and retained history before changing TTLs. It preserves longer explicit
extend()deadlines, leaves finished retention alone and never recreates missing history. Active-slot renewal is included in that transaction and only updates the slot while it still names this run. An ordinary append continues to reset normal retention.Adapts tishun/fastapi-redis-sdk#2 to the current metadata
alivefield, integer-second settings andextend()API. No new public APIs or dependency changes. This targetsfeature/resumable-streamsindependently of #96 and #97.Validation
Limits
Leave this PR in draft. Enterprise/Cloud, Sentinel failover, resharding, sustained capacity and real-network HTTP disconnects remain unqualified. The change does not fix concurrent creation or general append fencing. Normal Cluster operation does not qualify redis-py's watched-transaction recovery under connection loss.
The guide documents that heartbeat intervals must remain comfortably shorter than active retention and liveness TTLs; renewal cannot recover keys that expire before the next beat.