From 19b2de4089a1ffd9a835ddcf07d585971aa6b336 Mon Sep 17 00:00:00 2001 From: Yao Lu Date: Mon, 6 Jul 2026 14:26:58 -0700 Subject: [PATCH] fix: don't spend retry budget on undeliverable dispatch When a worker is selected but the task can't be delivered to it (its node has no live dispatch subscriber, or publish raises), the dispatcher requeued with count_retry=True, exhausting the task's max_attempts in seconds and failing with a null-error max_attempts_exceeded that masks the real cause. Route no_dispatch_subscriber and publish_failed through a new _grace_then_fail_undeliverable() keyed on a dedicated TaskRecord.no_dispatch_since clock (so the eligibility path's no_eligible_since reset can't clobber it): retry without consuming the execution budget, then fail after no_worker_grace_sec with a descriptive error naming the orphaned worker+node. Signed-off-by: Yao Lu --- src/server/dispatcher/base.py | 67 +++++++++++++++++-- src/server/task/models.py | 6 ++ .../dispatcher/test_dispatch_undeliverable.py | 67 +++++++++++++++++++ 3 files changed, 135 insertions(+), 5 deletions(-) create mode 100644 tests/server/dispatcher/test_dispatch_undeliverable.py diff --git a/src/server/dispatcher/base.py b/src/server/dispatcher/base.py index 71a6e4e..877e3d1 100644 --- a/src/server/dispatcher/base.py +++ b/src/server/dispatcher/base.py @@ -130,6 +130,43 @@ def _grace_then_fail( self._requeue_task(task_id, reason=reason, count_retry=False) return False + def _grace_then_fail_undeliverable( + self, + task_id: str, + record: TaskRecord, + reason: str, + message: str, + extra_payload: dict[str, Any] | None = None, + ) -> bool: + """A worker was selected but the task could not be delivered to it (its + node has no live dispatch subscriber, or publish failed). This is an + infrastructure fault, not a task-execution failure: keep retrying without + spending the task's ``max_attempts`` budget, then fail with a descriptive + error after ``no_worker_grace_sec``. Uses its own clock so the eligibility + path resetting ``no_eligible_since`` does not clobber the timer.""" + self._runtime.release_merge(task_id) + record.last_error = message + now = time.time() + if record.no_dispatch_since is None: + record.no_dispatch_since = now + waited = now - record.no_dispatch_since + if waited >= self._no_worker_grace_sec: + self._logger.warning( + "Task %s undeliverable after %.0fs (%s); failing", + task_id, + waited, + reason, + ) + payload = {"reason": reason} + if extra_payload: + payload.update(extra_payload) + self._fail_task( + task_id, message, worker_id=record.last_failed_worker, payload=payload + ) + return False + self._requeue_task(task_id, reason=reason, front=True, count_retry=False) + return False + def _grace_then_fail_exhausted( self, task_id: str, record: TaskRecord, failed_ids: set[str] ) -> bool: @@ -447,17 +484,37 @@ def dispatch_once(self, task_id: str) -> bool: receivers = self._worker_registry.publish_task(worker, message) except Exception as exc: self._logger.warning("Failed to publish task %s: %s", task_id, exc) - self._requeue_task(task_id, reason="publish_failed", front=True) - return False + return self._grace_then_fail_undeliverable( + task_id, + record, + reason="publish_failed", + message=f"Failed to publish task to worker {worker.id}: {exc}", + extra_payload={"worker_id": worker.id, "node_id": worker.node_id}, + ) if receivers <= 0: self._logger.info( - "No subscribers on tasks channel; delaying task %s", task_id + "Node %s dispatch channel has no live subscriber; delaying task %s " + "(orphaned worker %s)", + worker.node_id, + task_id, + worker.id, + ) + return self._grace_then_fail_undeliverable( + task_id, + record, + reason="no_dispatch_subscriber", + message=( + f"Selected worker {worker.id} on node {worker.node_id} has no " + "live dispatch subscriber (orphaned/stale worker registration); " + "the worker heartbeats but its supervisor is not consuming the " + "node dispatch channel" + ), + extra_payload={"worker_id": worker.id, "node_id": worker.node_id}, ) - self._requeue_task(task_id, reason="no_subscribers", front=True) - return False # 9. Mark dispatched + record.no_dispatch_since = None self._runtime.mark_dispatched(task_id, worker) if merged_children: try: diff --git a/src/server/task/models.py b/src/server/task/models.py index aee2d1c..cfc2d40 100644 --- a/src/server/task/models.py +++ b/src/server/task/models.py @@ -126,6 +126,12 @@ class TaskRecord(BaseModel): description="Epoch seconds when no eligible worker was first observed.", exclude=True, ) + no_dispatch_since: float | None = Field( + default=None, + description="Epoch seconds when a selected worker was first found " + "undeliverable (no live dispatch subscriber / publish failure).", + exclude=True, + ) local_name: str | None = Field(default=None, description="Workflow stage name.") graph_node_name: str | None = Field(default=None, description="Graph node name.") load: int = Field(default=0, description="Load score.") diff --git a/tests/server/dispatcher/test_dispatch_undeliverable.py b/tests/server/dispatcher/test_dispatch_undeliverable.py new file mode 100644 index 0000000..cb5dc5d --- /dev/null +++ b/tests/server/dispatcher/test_dispatch_undeliverable.py @@ -0,0 +1,67 @@ +"""Undeliverable-dispatch grace behavior. + +When a worker is selected but the task cannot be delivered to it (its node has +no live dispatch subscriber, or publish raises), the dispatcher must treat this +as an infrastructure fault: retry without spending the task's ``max_attempts`` +budget, then fail with a descriptive error after the no-worker grace. +""" + +import time +from types import SimpleNamespace + +from .helpers import make_capturing_dispatcher + + +def _record() -> SimpleNamespace: + return SimpleNamespace( + no_dispatch_since=None, + last_error=None, + last_failed_worker=None, + ) + + +def test_undeliverable_within_grace_requeues_without_retry() -> None: + disp = make_capturing_dispatcher(grace_sec=60) + disp._runtime.release_merge = lambda task_id: None + record = _record() + + result = disp._grace_then_fail_undeliverable( + "tsk-1", record, reason="no_dispatch_subscriber", message="orphaned" + ) + + assert result is False + assert not disp.failed + assert len(disp.requeued) == 1 + _, kwargs = disp.requeued[0] + # Infra fault must NOT consume the execution retry budget. + assert kwargs["count_retry"] is False + assert kwargs["reason"] == "no_dispatch_subscriber" + assert record.no_dispatch_since is not None + assert record.last_error == "orphaned" + + +def test_undeliverable_after_grace_fails_with_message() -> None: + disp = make_capturing_dispatcher(grace_sec=60) + disp._runtime.release_merge = lambda task_id: None + record = _record() + # Pretend the condition has persisted longer than the grace window. + record.no_dispatch_since = time.time() - 120 + + result = disp._grace_then_fail_undeliverable( + "tsk-1", + record, + reason="no_dispatch_subscriber", + message="orphaned worker w1 on node n1", + extra_payload={"worker_id": "w1", "node_id": "n1"}, + ) + + assert result is False + assert not disp.requeued + assert len(disp.failed) == 1 + task_id, error_message, kwargs = disp.failed[0] + assert task_id == "tsk-1" + assert error_message == "orphaned worker w1 on node n1" + payload = kwargs["payload"] + assert payload["reason"] == "no_dispatch_subscriber" + assert payload["worker_id"] == "w1" + assert payload["node_id"] == "n1"