refactor(coordination): reliable Redis Streams await/notify (replace at-most-once pub/sub)

docs/admin_docs/configuration/cache.mdx

@@ -295,14 +295,25 @@ high-performance distributed operations. This configuration enables:
 
 - **Distributed locking**: Moves lock operations from the metadata database to Redis, improving
   performance and reducing metastore load
-- **Real-time event notifications**: Enables instant pub/sub messaging for task abort signals and
-  completion notifications instead of polling-based approaches
+- **Event-driven notifications**: Task completion and abort signals are delivered over Redis
+  **Streams**, so waiters (sync join-and-wait, task-dependency DAGs, abort listeners) wake when a
+  signal lands instead of polling the metadata database. Because stream entries are persisted, a
+  waiter that reads slightly late, reconnects, or fails over still receives the signal. Without this
+  backend, these operations poll the metadata database instead.
 
 :::note
-This requires Redis or Valkey specifically—it uses Redis-specific features (pub/sub, `SET NX EX`)
-that are not available in general Flask-Caching backends.
+This requires Redis or Valkey specifically—it uses Redis-specific features (Streams, pub/sub,
+`SET NX EX`) that are not available in general Flask-Caching backends.
 :::
 
+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
+`DISTRIBUTED_COORDINATION_SIGNAL_TTL` (seconds, default 24 hours):
+
+```python
+DISTRIBUTED_COORDINATION_SIGNAL_TTL = 24 * 60 * 60
+```
+
 ### Configuration
 
 The distributed coordination uses Flask-Caching style configuration for consistency with other cache

docs/developer_docs/extensions/tasks.md

@@ -403,7 +403,7 @@ The prune job only removes tasks in terminal states (`SUCCESS`, `FAILURE`, `ABOR
 See `superset/config.py` for a complete example configuration.
 
 :::tip Distributed Coordination for Faster Notifications
-By default, abort detection and sync join-and-wait use database polling. Configure `DISTRIBUTED_COORDINATION_CONFIG` to enable Redis pub/sub for real-time notifications. See [Distributed Coordination Backend](/admin-docs/configuration/cache#signal-cache-backend) for configuration details.
+By default, abort detection and sync join-and-wait poll the task row in the metadata database. Configure `DISTRIBUTED_COORDINATION_CONFIG` (Redis/Valkey) and these become event-driven: completion and abort are signalled over Redis **Streams**, so a waiter wakes when the signal lands instead of polling the database. Because stream entries are persisted, a waiter that reads slightly late, reconnects, or fails over still receives the signal. Each signal stream keeps only its latest entry and is given a TTL, so streams for tasks that are never awaited do not accumulate; set the retention window with `DISTRIBUTED_COORDINATION_SIGNAL_TTL` (default 24h). See [Distributed Coordination Backend](/admin-docs/configuration/cache#signal-cache-backend) for configuration details.
 :::
 
 ## API Reference

superset/async_events/cache_backend.py

@@ -154,6 +154,47 @@ def xrange(
         count = count or self.MAX_EVENT_COUNT
         return self._cache.xrange(stream_name, start, end, count)
 
+    def xread(
+        self,
+        streams: dict[str, str],
+        count: int | None = None,
+        block_ms: int | None = None,
+    ) -> list[Any]:
+        """
+        Read new entries from one or more streams, optionally blocking.
+
+        Reliable, event-driven delivery: pass the last id already seen per stream
+        and this returns only entries added *after* it — so a signal is never
+        missed, even across reconnects (unlike pub/sub). ``block_ms`` blocks up to
+        that many milliseconds waiting for a new entry (``None``/``0`` returns
+        immediately).
+
+        :param streams: mapping of ``{stream_name: last_id_seen}``
+        :param count: max entries to return
+        :param block_ms: milliseconds to block for a new entry (``None`` = no block)
+        :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 []
+
+    def stream_last_id(self, stream_name: str) -> str:
+        """
+        Return the id of the last entry in a stream, or ``"0-0"`` if it is empty.
+
+        Used to capture a baseline before waiting so a subsequent blocking
+        :meth:`xread` from that id catches every entry added afterwards — closing
+        the publish-before-subscribe race that pub/sub cannot.
+        """
+        entries = self._cache.xrevrange(stream_name, count=1)
+        if not entries:
+            return "0-0"
+        last_id = entries[0][0]
+        return last_id.decode() if isinstance(last_id, bytes) else last_id
+
+    def expire(self, name: str, seconds: int) -> bool:
+        """Set a TTL (seconds) on a key; used to bound signal-stream growth."""
+        return bool(self._cache.expire(name, seconds))
+
     @classmethod
     def from_config(cls, config: dict[str, Any]) -> RedisCacheBackend:
         kwargs = {
@@ -347,6 +388,29 @@ def xrange(
         count = count or self.MAX_EVENT_COUNT
         return self._cache.xrange(stream_name, start, end, count)
 
+    def xread(
+        self,
+        streams: dict[str, str],
+        count: int | None = None,
+        block_ms: int | None = None,
+    ) -> list[Any]:
+        """Reliable, optionally-blocking stream read (see
+        :meth:`RedisCacheBackend.xread`)."""
+        return self._cache.xread(streams, count=count, block=block_ms) or []
+
+    def stream_last_id(self, stream_name: str) -> str:
+        """Return the last entry id, or ``"0-0"`` if empty (see
+        :meth:`RedisCacheBackend.stream_last_id`)."""
+        entries = self._cache.xrevrange(stream_name, count=1)
+        if not entries:
+            return "0-0"
+        last_id = entries[0][0]
+        return last_id.decode() if isinstance(last_id, bytes) else last_id
+
+    def expire(self, name: str, seconds: int) -> bool:
+        """Set a TTL (seconds) on a key; used to bound signal-stream growth."""
+        return bool(self._cache.expire(name, seconds))
+
     @classmethod
     def from_config(cls, config: dict[str, Any]) -> RedisSentinelCacheBackend:
         kwargs = {

superset/config.py

@@ -3289,6 +3289,13 @@ class ExtraAccessQueryFilters(TypedDict, total=False):
 # }
 DISTRIBUTED_COORDINATION_CONFIG: CacheConfig | None = None
 
+# Retention (seconds) for the Redis Streams the coordination service uses to deliver
+# signals (e.g. task completion/abort). Each signal is one short-lived stream entry
+# that a waiter consumes almost immediately; the TTL is a safety net so signal
+# streams for tasks that never get awaited cannot accumulate in Redis/Valkey
+# indefinitely. Defaults to 24 hours.
+DISTRIBUTED_COORDINATION_SIGNAL_TTL = int(timedelta(hours=24).total_seconds())
+
 # Default lock TTL (time-to-live) in seconds for distributed locks.
 # Can be overridden per-call via the `ttl_seconds` parameter.
 # After TTL expires, the lock is automatically released to prevent deadlocks.