refactor(coordination): reliable Redis Streams await/notify (replace at-most-once pub/sub) - #43409
Conversation
Add xread (blocking, reliable delivery), stream_last_id (baseline capture for race-free waits), and expire (bound signal-stream growth) to both RedisCacheBackend and RedisSentinelCacheBackend. Foundation for moving GTF wait/notify off at-most-once pub/sub onto Redis Streams.
…ub to nudges Move the coordination service's await/notify off at-most-once pub/sub onto Redis Streams (guaranteed delivery), and migrate GTF task completion/abort onto it. - CoordinationService.notify() appends a signal to the channel's stream (MAXLEN 1 + TTL) so a waiter that reads late, reconnects, or survives a failover still sees it. wait_for_signal/listen_for_signal now block on a stream (event-driven, no polling) when a coordination backend is set, capturing a baseline id first so no signal is missed; without a backend they poll the caller's predicate exactly as before. The predicate stays the source of truth, so duplicate/late signals are harmless. Pub/sub (publish) is retained only as an explicit best-effort, at-most-once nudge — documented as such — and is no longer used for awaiting. - TaskManager.publish_completion/publish_abort emit via notify() (streams). - SignalListener no longer wraps a pub/sub subscription. - New config DISTRIBUTED_COORDINATION_SIGNAL_TTL (default 24h) bounds signal-stream retention so streams for never-awaited tasks cannot accumulate. Unit tests updated (coordination + task manager) and green locally (390 passed).
…livery model Add unit tests for RedisCacheBackend.xread/stream_last_id/expire (delegation, None->[], bytes-id decode, empty->0-0). Document the coordination service's reliable Streams-based signalling (event-driven when a backend is set, DB-poll fallback otherwise) and the DISTRIBUTED_COORDINATION_SIGNAL_TTL retention config in the GTF developer docs and the cache admin docs.
|
Bito Automatic Review Skipped - Branch Excluded |
✅ Deploy Preview for superset-docs-preview ready!
To edit notification comments on pull requests, go to your Netlify project configuration. |
Describe the coordination notifications as event-driven over Redis Streams and explain the mechanism (persisted entries → delivered after a late read/reconnect/ failover) without subjective 'reliable/guaranteed/instant' framing, in the docs and docstrings. Also refresh stale pub/sub references in TaskManager now that signalling rides streams.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## gaq-to-gtf #43409 +/- ##
===========================================
Coverage 78.88% 78.88%
===========================================
Files 2882 2882
Lines 164678 164697 +19
Branches 38024 38027 +3
===========================================
+ Hits 129899 129918 +19
- Misses 32332 32333 +1
+ Partials 2447 2446 -1
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:
|
dosubot
Bot
added
change:backend
|
/review |
Code Review Agent Run #42f078Actionable Suggestions - 0Additional Suggestions - 2
Filtered by Review RulesBito filtered these suggestions based on rules created automatically for your feedback. Manage rules.
Review Details
Bito Usage GuideCommands Type the following command in the pull request comment and save the comment.
Refer to the documentation for additional commands. Configuration This repository uses Documentation & Help AI Code Review powered by |
…h; review nits Address PR apache#43409 review: downgrade best-effort signal-failure logs to warning (listeners/waiters fall back to polling), assert the MAXLEN constant instead of a literal in the notify test, add a Sentinel-backend stream-helpers test, and add a backend-path listen test — closing the codecov gaps on the Sentinel backend and the listener stream loop.
![codeant-ai-for-open-source[bot]](https://avatars.githubusercontent.com/u/151810221?s=60&v=4){kind=small}
| Each signal stream keeps only its latest entry and is given a TTL, so signal streams for tasks that | ||
| are never awaited do not accumulate in Redis/Valkey. Set the retention window with |
There was a problem hiding this comment.
Suggestion: The retention description incorrectly states that each stream keeps only its latest entry. The coordination backend passes MAXLEN without disabling Redis-py's default approximate trimming, so Redis may retain more than one entry. Document this as an approximate bound or change the implementation to request exact trimming if retaining exactly one entry is required. [docstring mismatch]
Severity Level: Minor 🧹
- ⚠️ Redis streams may retain multiple signal entries.
- ⚠️ Documentation overstates stream trimming precision.
- ⚠️ Repeated notifications can use extra temporary Redis memory.
Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** docs/admin_docs/configuration/cache.mdx
**Line:** 309:310
**Comment:**
*Docstring Mismatch: The retention description incorrectly states that each stream keeps only its latest entry. The coordination backend passes `MAXLEN` without disabling Redis-py's default approximate trimming, so Redis may retain more than one entry. Document this as an approximate bound or change the implementation to request exact trimming if retaining exactly one entry is required.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix
|
The flagged issue is correct. The documentation states that each stream keeps only its latest entry, but the implementation uses To resolve this, you can update the documentation to reflect that the retention is approximate, or update the Would you like me to fetch all other comments on this PR to validate and implement fixes for them as well? docs/admin_docs/configuration/cache.mdx |
| :returns: redis-py XREAD reply — ``[[stream, [(id, {field: value}), ...]]]`` | ||
| — or an empty list when nothing arrived before the block elapsed | ||
| """ | ||
| return self._cache.xread(streams, count=count, block=block_ms) or [] |
There was a problem hiding this comment.
Suggestion: The helper forwards a blocking XREAD duration that can exceed the configured Redis socket timeout. In deployments setting CACHE_REDIS_SOCKET_TIMEOUT below the one-second or five-second block interval, redis-py raises a socket timeout before the requested block completes; _read_stream treats that as an empty read, causing repeated predicate/database checks instead of event-driven waiting. Ensure the read connection's socket timeout is compatible with the blocking duration or use a dedicated coordination connection. [performance]
Severity Level: Major ⚠️
- ⚠️ Task completion waits repeatedly query the metastore.
- ⚠️ Abort listeners lose event-driven waiting.
- ⚠️ Redis traffic increases with short socket timeouts.Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** superset/async_events/cache_backend.py
**Line:** 178:178
**Comment:**
*Performance: The helper forwards a blocking `XREAD` duration that can exceed the configured Redis socket timeout. In deployments setting `CACHE_REDIS_SOCKET_TIMEOUT` below the one-second or five-second block interval, redis-py raises a socket timeout before the requested block completes; `_read_stream` treats that as an empty read, causing repeated predicate/database checks instead of event-driven waiting. Ensure the read connection's socket timeout is compatible with the blocking duration or use a dedicated coordination connection.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| last_id = cls._read_stream( | ||
| backend, | ||
| channel, | ||
| last_id, | ||
| cls._bounded_block_ms(_STREAM_BLOCK_MS, remaining), | ||
| ) | ||
| if (result := check()) is not None: | ||
| return result |
There was a problem hiding this comment.
Suggestion: The deadline is checked before _read_stream but not after it returns. If the blocking read consumes the remaining timeout, the code still invokes the database predicate and can return a successful result after the requested timeout has expired; an expensive predicate can also extend the overrun. Check the deadline again before running check() after each read. [logic error]
Severity Level: Minor 🧹
- ⚠️ Completion waits may exceed their configured timeout.
- ⚠️ Callers receive success after timeout expiration.
- ⚠️ Database predicate latency increases the overrun.Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** superset/coordination/base.py
**Line:** 319:326
**Comment:**
*Logic Error: The deadline is checked before `_read_stream` but not after it returns. If the blocking read consumes the remaining timeout, the code still invokes the database predicate and can return a successful result after the requested timeout has expired; an expensive predicate can also extend the overrun. Check the deadline again before running `check()` after each read.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix| cls, | ||
| channel: str, | ||
| check: Callable[[], T | None], | ||
| deadline: float | None, |
There was a problem hiding this comment.
Suggestion: A Redis connection or failover error from stream_last_id occurs before the try block, so the listener thread exits without logging the failure or falling back to database polling. Move backend initialization into the protected error-handling path and preserve the listener's fallback behavior when Redis is temporarily unavailable. [resource leak]
Severity Level: Major ⚠️
- ❌ Background task abort listener exits during Redis startup failure.
- ⚠️ Aborted tasks may continue executing.
- ⚠️ Listener failure is not logged by its handler.Prompt for AI Agent 🤖
This is a comment left during a code review.
**Path:** superset/coordination/base.py
**Line:** 340:343
**Comment:**
*Resource Leak: A Redis connection or failover error from `stream_last_id` occurs before the `try` block, so the listener thread exits without logging the failure or falling back to database polling. Move backend initialization into the protected error-handling path and preserve the listener's fallback behavior when Redis is temporarily unavailable.
Validate the correctness of the flagged issue. If correct, How can I resolve this? If you propose a fix, implement it and please make it concise.
Once fix is implemented, also check other comments on the same PR, and ask user if the user wants to fix the rest of the comments as well. if said yes, then fetch all the comments validate the correctness and implement a minimal fix1 participant
SUMMARY
Moves the coordination service's await/notify off at-most-once pub/sub onto Redis Streams (guaranteed delivery), and migrates GTF task completion/abort onto it.
Why. Redis Pub/Sub is documented at-most-once / fire-and-forget — "if the subscriber is unable to handle the message (for example, due to an error or a network disconnect) the message is forever lost." To stay correct on top of that, the old
wait_for_signaltreated a published message as a mere nudge and re-read the task row every ~1s as a backstop — i.e. it degraded to polling even when Redis was available (and each poll loaded the fullTaskORM). Redis's own recommendation for guaranteed delivery is Streams (persisted, replayable), which lets the wait be genuinely event-driven with no busy-poll.What changed.
CoordinationService.notify(channel)appends a signal to the channel's Redis Stream (MAXLEN 1+ TTL). Guaranteed delivery: a waiter that reads slightly late, reconnects, or survives a failover still sees it.wait_for_signal/listen_for_signalblock on the stream (event-driven, no polling) whenDISTRIBUTED_COORDINATION_CONFIGis set — capturing a baseline id first so no signal is missed (race-free). Without a backend they poll the caller's predicate exactly as before. The predicate stays the source of truth, so a duplicate/late signal is harmless.TaskManager.publish_completion/publish_abortemit vianotify(streams).publish) is retained only as an explicit best-effort, at-most-once nudge — documented as such — and is no longer used for awaiting.SignalListenerno longer wraps a subscription.xread(blocking),stream_last_id(baseline),expire; and configDISTRIBUTED_COORDINATION_SIGNAL_TTL(default 24h) bounds signal-stream retention so streams for never-awaited tasks cannot accumulate.Delivery model (documented). Streams for anything that must be delivered (task completion/abort — the typical case); pub/sub only for loss-tolerant nudges; DB polling as the no-backend fallback. Docs updated in the GTF developer guide and the cache admin guide.
BEFORE / AFTER
TESTING INSTRUCTIONS
tests/unit_tests/coordination/test_service.py(notify + stream wait + socket-timeout retry),tests/unit_tests/tasks/test_manager.py(completion/abort via streams, stream wait),tests/unit_tests/async_events/test_cache_backend.py(xread/stream_last_id/expire).DISTRIBUTED_COORDINATION_CONFIGset, run a sync join-and-wait / a task DAG and confirm dependents wake promptly with no per-second DB reads; confirm signal stream keys (gtf:complete:*,gtf:abort:*) carry a TTL and self-expire.ADDITIONAL INFORMATION
GLOBAL_TASK_FRAMEWORKCoordinationService.notify;DISTRIBUTED_COORDINATION_SIGNAL_TTLconfig)Note: the whole-project frontend type-check has pre-existing failures unrelated to this change.