fix(gtf): harden reaper/self-fence design + doc gaps (review follow-ups)

Addresses a full design + docs re-review of the GTF resilience mechanisms.

Design:
- R1 (reaper): claim the FAILURE transition via CAS *before* firing destructive
  cleanup. Revoke and out-of-band query cancellation now run only after the
  reaper wins the CAS, so a briefly-stalled-but-healthy worker that revives and
  commits its own terminal status keeps its warehouse query (no false-positive
  kill, which would also cascade FAILURE to dependents).
- R2 (context): elect the abort/timeout/self-fence winner under a lock so abort
  handlers run exactly once across the three daemon threads (listener, timeout
  timer, heartbeat) that can race to abort.
- R3 (DAO): anchor the heartbeat write and the orphan scan on the *database*
  clock (dialect-aware naive-UTC now, app-side fallback) so worker/reaper host
  clock skew can't reap a live task early.
- R5 (heartbeat): only self-fence once the fence callback is armed (task is
  executing); a stalled heartbeat during the pre-execution DAG wait keeps
  retrying so recovered connectivity resumes the task instead of forfeiting it.

Docs:
- Document get_dependency_payloads() (+ the immediate=True pairing), add a
  reap_orphaned_tasks beat-schedule snippet, name the
  TASK_PROGRESS_UPDATE_THROTTLE_INTERVAL / TASK_ABORT_POLLING_DEFAULT_INTERVAL
  tunables, and note orphan-reap/self-fence as FAILURE causes.
- Fix stale 'prune cron reaps' comments (reap.py, models/tasks.py, migration).

Tests: reaper no longer cancels/revokes on a lost CAS; concurrent abort election
runs handlers once; heartbeat does not fence before armed.

Author: villebro

docs/developer_docs/extensions/tasks.md

@@ -107,7 +107,7 @@ PENDING ──→ IN_PROGRESS ────→ SUCCESS
 | `IN_PROGRESS` | Executing                                            |
 | `ABORTING`    | Abort/timeout triggered, abort handlers running      |
 | `SUCCESS`     | Completed successfully                               |
-| `FAILURE`     | Failed with error or abort/cleanup handler exception |
+| `FAILURE`     | Failed with error, abort/cleanup handler exception, orphan reaping, or worker self-fence |
 | `ABORTED`     | Cancelled by user/admin                              |
 | `TIMED_OUT`   | Exceeded configured timeout                          |
 
@@ -158,7 +158,7 @@ In the Task List UI, when a payload is defined, an info icon appears in the **De
 
 #### Forcing an Immediate Write
 
-By default `update_task()` throttles database writes (batching frequent updates to limit metastore load). Pass `immediate=True` to bypass throttling and write synchronously:
+By default `update_task()` throttles database writes (batching frequent updates to limit metastore load, at most one write per `TASK_PROGRESS_UPDATE_THROTTLE_INTERVAL` seconds, default 2). Pass `immediate=True` to bypass throttling and write synchronously:
 
 ```python
 ctx.update_task(payload={"result_cache_key": key}, immediate=True)
@@ -370,6 +370,21 @@ Because dependents hold a worker slot while awaiting their prerequisites, a deep
 
 Cycles (including self-dependencies) are rejected at schedule time. Dependency edges are removed automatically when either endpoint task is pruned.
 
+**Reading a prerequisite's output.** A dependent reads the payloads its prerequisites published via `ctx.get_dependency_payloads()`, which returns the prerequisites' payloads in dependency-edge order. Pair it with the prerequisite writing its result with `ctx.update_task(payload=..., immediate=True)` so the value is flushed (not held in the write-throttle buffer) by the time the dependency gate releases the dependent:
+
+```python
+@task
+def totals_task() -> None:
+    ctx = get_context()
+    # immediate=True so the dependent observes this the moment the gate releases.
+    ctx.update_task(payload={"result_cache_key": key}, immediate=True)
+
+@task
+def dependent_task() -> None:
+    ctx = get_context()
+    upstream = ctx.get_dependency_payloads()  # [{"result_cache_key": ...}, ...]
+```
+
 ## Task Scopes
 
 ```python
@@ -422,12 +437,22 @@ A task whose worker dies mid-execution (OOM kill, crash, lost broker message) wo
 
 Enable the `reap_orphaned_tasks` beat schedule on a short interval (e.g. every minute) so orphaned tasks — and their warehouse queries — do not linger; it is separate from `prune_tasks` (a heavier retention delete that runs infrequently). Keep `GTF_ORPHAN_TASK_TIMEOUT` comfortably larger than the heartbeat interval (≥ ~3×) so a brief pause or CPU-bound stretch is not mistaken for a dead worker.
 
+```python
+# In your superset_config.py, add to your Celery beat schedule:
+CELERY_CONFIG.beat_schedule["reap_orphaned_tasks"] = {
+    "task": "reap_orphaned_tasks",
+    "schedule": crontab(minute="*", hour="*"),  # Run every minute
+}
+```
+
+Unlike `prune_tasks`, the reaper takes no kwargs — it reads `GTF_ORPHAN_TASK_TIMEOUT` from config.
+
 :::note Cancelling the underlying query
 For long-running work backed by an external query, register an `on_abort` handler that cancels it (this is how async chart-data query tasks cancel the warehouse query on engines that support cancellation). Without such a handler an abort/timeout frees the task but cannot stop the external work.
 :::
 
 :::tip Distributed Coordination for Faster Notifications
-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.
+By default, abort detection and sync join-and-wait poll the task row in the metadata database (every `TASK_ABORT_POLLING_DEFAULT_INTERVAL` seconds, default 10). 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
@@ -451,6 +476,7 @@ By default, abort detection and sync join-and-wait poll the task row in the meta
 | Method                           | Description                                   |
 | -------------------------------- | --------------------------------------------- |
 | `update_task(progress, payload, immediate=False)` | Update progress and/or custom payload (`immediate=True` bypasses write throttling) |
+| `get_dependency_payloads()`      | Return prerequisite tasks' payloads, in dependency-edge order |
 | `on_cleanup(handler)`            | Register cleanup handler                      |
 | `on_abort(handler)`              | Register abort handler (makes task abortable) |
 

superset/commands/tasks/reap.py

@@ -40,7 +40,7 @@
 class ReapOrphanedTasksCommand(BaseCommand):
     """Recover tasks abandoned by a worker that stopped refreshing its heartbeat.
 
-    Run from the prune cron before row deletion. For each orphan (an active task
+    Run from the ``reap_orphaned_tasks`` beat job. For each orphan (an active task
     whose liveness heartbeat has gone stale — see ``TaskDAO.find_orphaned``) the
     command transitions the row to ``FAILURE`` and publishes completion so waiters
     (sync joiners, DAG dependents, chart-data pollers) unblock, then revokes the
@@ -90,12 +90,15 @@ def _reap(self, task_uuid: UUID, stats_logger: BaseStatsLogger) -> bool:
                 )
             ),
         )
-        self._revoke(properties.get("celery_task_id"), stats_logger)
-        self._cancel_orphaned_query(properties)
-
         properties["error_message"] = ORPHAN_ERROR_MESSAGE
         properties["exception_type"] = "OrphanedTaskError"
 
+        # Claim the terminal transition FIRST, before any destructive side effect.
+        # The CAS is atomic (row-locked) against a worker that merely stalled and
+        # then revived to commit its own terminal status: if that worker wins, this
+        # returns False and we must NOT cancel its (healthy) warehouse query or
+        # revoke its job. Only once we own the FAILURE transition is the task
+        # genuinely orphaned and its query safe to cancel out-of-band.
         if not TaskDAO.conditional_status_update(
             task_uuid,
             TaskStatus.FAILURE,
@@ -108,6 +111,11 @@ def _reap(self, task_uuid: UUID, stats_logger: BaseStatsLogger) -> bool:
 
         db.session.commit()  # pylint: disable=consider-using-transaction
 
+        # We own the terminal transition — the worker is gone (or lost the race),
+        # so its Celery job and warehouse query are safe to clean up.
+        self._revoke(properties.get("celery_task_id"), stats_logger)
+        self._cancel_orphaned_query(properties)
+
         from superset.tasks.manager import TaskManager
 
         TaskManager.publish_completion(task_uuid, TaskStatus.FAILURE.value)

superset/daos/tasks.py

@@ -100,23 +100,59 @@ def get_id(cls, task_uuid: UUID) -> int | None:
         """
         return db.session.query(Task.id).filter(Task.uuid == task_uuid).scalar()
 
+    # Dialect SQL for the database's current time as a naive UTC timestamp. Used
+    # so the heartbeat write and the orphan scan share one clock (the DB's) rather
+    # than the worker's and the reaper's host clocks, which can skew and cause a
+    # live task to be reaped early. None -> fall back to an app-side timestamp on
+    # unrecognized dialects (correct if hosts are NTP-synced).
+    _DB_UTCNOW_SQL = {
+        "postgresql": "timezone('UTC', now())",
+        "mysql": "UTC_TIMESTAMP()",
+        "sqlite": "CURRENT_TIMESTAMP",
+    }
+
+    @classmethod
+    def _db_utcnow_sql(cls) -> str | None:
+        """Return the dialect's naive-UTC-now SQL, or None to use an app clock."""
+        return cls._DB_UTCNOW_SQL.get(db.session.get_bind().dialect.name)
+
+    @classmethod
+    def _db_utcnow(cls) -> datetime:
+        """Read the database's current time as a naive UTC datetime."""
+        if (expr := cls._db_utcnow_sql()) is None:
+            return naive_utcnow()
+        # type_coerce applies DateTime result processing so SQLite's text value is
+        # parsed into a datetime like the other drivers already return.
+        return db.session.scalar(
+            sa.select(sa.type_coerce(sa.text(expr), sa.DateTime()))
+        )
+
     @classmethod
     def touch_heartbeat(cls, task_id: int) -> None:
         """Bump a task's ``last_heartbeat`` to now without touching ``changed_on``.
 
         Called on an interval by the executing worker's heartbeat thread so the
-        prune cron can tell a live task from an orphaned one. Uses a raw textual
+        reaper can tell a live task from an orphaned one. Uses a raw textual
         UPDATE on the single column: ``changed_on`` carries a client-side
         ``onupdate`` default that ORM/Core updates would fire, and any bump to
         ``changed_on`` would resurface the task in ``get_statuses_changed_since``
         every heartbeat (resetting client polling backoff). A textual statement
         bypasses that default processing entirely. Bound by integer ``id`` to
         sidestep ``UUIDType`` dialect handling.
+
+        Stamps the *database* clock (via ``_db_utcnow_sql``) rather than the
+        worker's host clock so the value is comparable to the reaper's scan
+        without cross-host skew (see ``find_orphaned``).
         """
-        db.session.execute(
-            sa.text("UPDATE tasks SET last_heartbeat = :ts WHERE id = :id"),
-            {"ts": naive_utcnow(), "id": task_id},
-        )
+        if (expr := cls._db_utcnow_sql()) is not None:
+            # expr is a fixed per-dialect literal, not user input.
+            sql = f"UPDATE tasks SET last_heartbeat = {expr} WHERE id = :id"  # noqa: S608
+            db.session.execute(sa.text(sql), {"id": task_id})
+        else:
+            db.session.execute(
+                sa.text("UPDATE tasks SET last_heartbeat = :ts WHERE id = :id"),
+                {"ts": naive_utcnow(), "id": task_id},
+            )
         # Deliberate standalone commit: this is a single out-of-band column write,
         # not a unit of work, and must not be wrapped in the ORM transaction flow.
         db.session.commit()  # pylint: disable=consider-using-transaction
@@ -132,10 +168,14 @@ def find_orphaned(cls, orphan_timeout_seconds: int) -> list[UUID]:
         still being worked on (fresh heartbeat) is never returned, so this never
         interferes with a live worker's cooperative abort/cleanup.
 
+        The staleness cutoff is anchored on the *database* clock (the same clock
+        the heartbeat is stamped with) so it does not depend on this host's clock
+        agreeing with the workers'.
+
         Skips the base filter: internal maintenance over all tasks, not a
         user-facing listing.
         """
-        stale_before = naive_utcnow() - timedelta(seconds=orphan_timeout_seconds)
+        stale_before = cls._db_utcnow() - timedelta(seconds=orphan_timeout_seconds)
         rows = (
             db.session.query(Task.uuid)
             .filter(

superset/migrations/versions/2026-08-21_12-00_7e2c9a4f1b83_create_task_dependencies_table.py

@@ -149,9 +149,9 @@ def upgrade():
         )
 
     # Liveness marker for orphan detection: the executing worker bumps
-    # ``tasks.last_heartbeat`` on a background thread, and the prune cron reaps
-    # ACTIVE tasks whose heartbeat has gone stale (a dead/orphaned worker). The
-    # index backs the reaper's ``last_heartbeat < now - timeout`` scan.
+    # ``tasks.last_heartbeat`` on a background thread, and the reap_orphaned_tasks
+    # beat job reaps ACTIVE tasks whose heartbeat has gone stale (a dead/orphaned
+    # worker). The index backs the reaper's ``last_heartbeat < now - timeout`` scan.
     add_columns(
         TASKS_TABLE,
         Column("last_heartbeat", DateTime, nullable=True),

superset/models/tasks.py

@@ -88,10 +88,10 @@ class Task(CoreTask, AuditMixinNullable, Model):
     ended_at = Column(DateTime, nullable=True)
 
     # Liveness marker bumped by the executing worker's heartbeat thread while it
-    # holds the task. The prune cron reaps ACTIVE tasks whose heartbeat has gone
-    # stale (a dead/orphaned worker). Written out-of-band via a raw UPDATE (see
-    # TaskDAO.touch_heartbeat) so a heartbeat never advances changed_on and thus
-    # never resurfaces the task in the status-change poll.
+    # holds the task. The reap_orphaned_tasks beat job reaps ACTIVE tasks whose
+    # heartbeat has gone stale (a dead/orphaned worker). Written out-of-band via a
+    # raw UPDATE (see TaskDAO.touch_heartbeat) so a heartbeat never advances
+    # changed_on and thus never resurfaces the task in the status-change poll.
     last_heartbeat = Column(DateTime, nullable=True, index=True)
 
     # User context for execution

superset/tasks/context.py

@@ -71,6 +71,9 @@ def __init__(self, task: "Task") -> None:
         self._abort_detected = False
         self._abort_handlers_completed = False  # Track if all abort handlers finished
         self._execution_completed = False  # Set by executor after task work completes
+        # Elects one abort trigger across the listener/timeout/fence threads so
+        # abort handlers run exactly once.
+        self._abort_lock = threading.Lock()
 
         # Collected handler failures for unified reporting
         self._handler_failures: list[TaskContext.HandlerFailure] = []
@@ -416,21 +419,25 @@ def _on_abort_detected(self) -> None:
         """
         Callback invoked by TaskManager when abort is detected.
 
-        Triggers all registered abort handlers.
+        Triggers all registered abort handlers exactly once, even when the abort
+        listener, timeout timer, and heartbeat self-fence race to call this
+        concurrently: the winner is elected atomically under ``_abort_lock``.
         """
-        if self._abort_detected:
-            return  # Already handled
+        with self._abort_lock:
+            if self._abort_detected:
+                return  # Another thread already won the election
 
-        # Check if task execution has already completed (late abort race).
-        # Executor sets _execution_completed after task work finishes.
-        if self._execution_completed:
-            logger.info(
-                "Abort detected for task %s but execution already completed",
-                self._task_uuid,
-            )
-            return
+            # Check if task execution has already completed (late abort race).
+            # Executor sets _execution_completed after task work finishes.
+            if self._execution_completed:
+                logger.info(
+                    "Abort detected for task %s but execution already completed",
+                    self._task_uuid,
+                )
+                return
+
+            self._abort_detected = True
 
-        self._abort_detected = True
         logger.info("Abort detected for task %s", self._task_uuid)
         self._trigger_abort_handlers()
 

superset/tasks/heartbeat.py

@@ -43,9 +43,10 @@ class HeartbeatController:
 
     The heartbeat thread starts before the ``TaskContext`` exists (it spans the
     whole time the worker holds the task, including the DAG wait), so the
-    executor registers the fence callback once the context is built. Until then
-    a fence can only stop the heartbeat and log — there is no running query to
-    cancel yet.
+    executor registers the fence callback once the context is built. The worker
+    only self-fences once it is ``armed`` (a callback is registered); before that
+    there is no running query to cancel, so a stalled heartbeat is simply left
+    for the reaper and the loop keeps trying in case connectivity recovers.
     """
 
     def __init__(self) -> None:
@@ -55,6 +56,11 @@ def on_fence(self, callback: Callable[[], None]) -> None:
         """Register the callback invoked when the worker self-fences."""
         self._fence_callback = callback
 
+    @property
+    def armed(self) -> bool:
+        """True once a fence callback is registered (task is executing)."""
+        return self._fence_callback is not None
+
     def invoke_fence(self) -> None:
         """Invoke the registered fence callback, if one has been registered."""
         if self._fence_callback is not None:
@@ -80,7 +86,10 @@ def task_heartbeat(  # noqa: C901
     reaper has already (or will shortly) mark FAILURE, the worker fails itself
     via the registered fence callback. A single failed write is tolerated; only
     a sustained outage spanning the orphan window fences, so a transient blip
-    never kills a healthy task.
+    never kills a healthy task. Fencing only kicks in once the controller is
+    armed (the task is executing and has a fence callback); a stalled heartbeat
+    during the pre-execution DAG wait keeps retrying so recovered connectivity
+    resumes the task rather than forfeiting it.
 
     The thread is a daemon so it never blocks worker shutdown (matching the
     existing timeout/abort threads) and relies on DB drivers releasing the GIL
@@ -142,7 +151,9 @@ def _beat() -> None:
             with app.app_context():
                 if _write():
                     deadline = _new_deadline()
-                elif naive_utcnow() >= deadline:
+                elif controller.armed and naive_utcnow() >= deadline:
+                    # Only fence an executing task; a pre-execution stall is left
+                    # to the reaper and keeps retrying in case the outage clears.
                     _fence()
                     return
 

tests/unit_tests/tasks/test_heartbeat.py

@@ -116,6 +116,24 @@ def test_heartbeat_does_not_fence_while_writes_succeed() -> None:
     fence_callback.assert_not_called()
 
 
+def test_heartbeat_does_not_fence_before_armed() -> None:
+    """During the pre-execution DAG wait (no callback yet) a stalled heartbeat
+    must not fence — it keeps retrying so recovered connectivity resumes the task."""
+    stats = MagicMock()
+    app = _mock_app(interval=0.02, stats=stats, orphan_timeout=0)
+
+    with patch(
+        "superset.daos.tasks.TaskDAO.touch_heartbeat",
+        side_effect=RuntimeError("db down"),
+    ):
+        with task_heartbeat(11, app):  # never call on_fence -> controller not armed
+            time.sleep(0.1)  # deadline passes and several beats fail
+
+    # Writes kept failing and being counted, but no fence fired (not armed).
+    stats.incr.assert_any_call("gtf.task.heartbeat_failure")
+    assert ("gtf.task.self_fenced",) not in [c.args for c in stats.incr.call_args_list]
+
+
 def test_heartbeat_tolerates_intermittent_failures() -> None:
     """A transient blip (fail then success) resets the deadline and does not fence."""
     stats = MagicMock()