From 619132da5b7f9108c0adddc301a7744f1c1b99e0 Mon Sep 17 00:00:00 2001 From: ordinary being Date: Wed, 21 Oct 2026 22:34:41 +0800 Subject: [PATCH] fix(scheduler): treat only busy-thread 409s as retryable (#7196) * fix(scheduler): treat only busy-thread 409s as retryable start_run exposes the retryable run-manager admission conflict as BusyThreadConflict (an HTTP 409), but also uses plain 409s for permanent idempotency-key and thread/assistant mismatches. The scheduler treated every HTTP 409 as an overlap conflict and requeued the occurrence, so a permanent launch error was retried until scheduler.queue_timeout_seconds elapsed. Other launch errors are documented to fail immediately, and the MCP notification launcher already distinguishes these types. Match ConflictError and BusyThreadConflict only; a plain 409 now follows the normal failure path. The type is imported lazily so the scheduler module does not take a load-time dependency on app.gateway.services. Tests: a plain idempotency 409 terminalizes the occurrence as failed, while BusyThreadConflict remains queued; existing ConflictError coverage is unchanged. * fix(scheduler): document the busy-thread import boundary and pin the failure detail Keep the BusyThreadConflict import local to the classifier and document why it is guaranteed to resolve: the scheduler is wired by the Gateway, where app.gateway.services is already loaded, and this must not become an optional dependency. Import the same type locally in the test so collection does not initialize gateway services for the rest of the scheduler suite. Also pin the persisted failure detail so the 409 reason stays operator-visible. --------- Co-authored-by: Tong-bit-art --- backend/app/scheduler/service.py | 15 +++- backend/tests/test_scheduled_task_service.py | 80 ++++++++++++++++++++ 2 files changed, 93 insertions(+), 2 deletions(-) diff --git a/backend/app/scheduler/service.py b/backend/app/scheduler/service.py index 76a21df9b4..5ede6915a9 100644 --- a/backend/app/scheduler/service.py +++ b/backend/app/scheduler/service.py @@ -11,7 +11,6 @@ from datetime import UTC, datetime, timedelta from typing import Any, Literal from zoneinfo import ZoneInfo -from fastapi import HTTPException from sqlalchemy import select from app.channels.capabilities import supports_proactive_group_notifications, supports_proactive_notifications @@ -473,7 +472,19 @@ class ScheduledTaskService: def _is_overlap_conflict(exc: Exception) -> bool: if isinstance(exc, ConflictError): return True - return isinstance(exc, HTTPException) and exc.status_code == 409 + # ``start_run`` converts a run-manager admission conflict into this + # typed HTTP 409. Other 409s (idempotency-key mismatch, thread or + # assistant drift) are permanent launch failures; retrying them would + # keep the occurrence queued until the queue timeout. + # + # Keep the import local to avoid changing the scheduler module's + # load-time import graph. The scheduler is wired by the Gateway, where + # ``app.gateway.services`` is already loaded, so this lookup is + # guaranteed; do not make it optional without moving the + # classification to a dependency that is always importable here. + from app.gateway.services import BusyThreadConflict + + return isinstance(exc, BusyThreadConflict) @staticmethod def _task_status_for_failure(task: dict[str, Any], *, trigger: str) -> str: diff --git a/backend/tests/test_scheduled_task_service.py b/backend/tests/test_scheduled_task_service.py index 0bc2766c5a..34ea9b3403 100644 --- a/backend/tests/test_scheduled_task_service.py +++ b/backend/tests/test_scheduled_task_service.py @@ -3,6 +3,7 @@ from datetime import UTC, datetime, timedelta import pytest import pytest_asyncio +from fastapi import HTTPException from app.scheduler.service import ScheduledTaskService from deerflow.runtime import ConflictError, RunStatus @@ -351,6 +352,85 @@ async def test_manual_overlap_conflict_is_kept_in_queue(): assert task_repo.release_calls == [] +@pytest.mark.asyncio +async def test_plain_409_from_launch_is_a_permanent_failure(): + async def fake_launch(**_kwargs): + raise HTTPException(status_code=409, detail="Idempotency-Key already used with a different request") + + task_repo = DummyTaskRepo( + [ + { + "id": "task-permanent-409", + "user_id": "user-1", + "thread_id": "thread-1", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + "status": "enabled", + "overlap_policy": "enqueue", + } + ] + ) + run_repo = DummyRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + result = await service.dispatch_task(task_repo.rows[0], now=datetime.now(UTC), trigger="scheduled") + + assert result["outcome"] == "failed" + assert run_repo.updated[-1][1]["status"] == "failed" + assert "Idempotency-Key already used" in run_repo.updated[-1][1]["error"] + + +@pytest.mark.asyncio +async def test_busy_thread_http_conflict_is_kept_in_queue(): + from app.gateway.services import BusyThreadConflict + + async def fake_launch(**_kwargs): + raise BusyThreadConflict("Thread thread-1 already has an active run") + + task_repo = DummyTaskRepo( + [ + { + "id": "task-busy-thread", + "user_id": "user-1", + "thread_id": "thread-1", + "context_mode": "reuse_thread", + "assistant_id": "lead_agent", + "prompt": "Summarize thread", + "schedule_type": "cron", + "schedule_spec": {"cron": "0 9 * * *"}, + "timezone": "UTC", + "status": "enabled", + "overlap_policy": "enqueue", + } + ] + ) + run_repo = DummyRunRepo() + service = ScheduledTaskService( + task_repo=task_repo, + task_run_repo=run_repo, + launch_run=fake_launch, + poll_interval_seconds=5, + lease_seconds=120, + max_concurrent_runs=3, + ) + + result = await service.dispatch_task(task_repo.rows[0], now=datetime.now(UTC), trigger="scheduled") + + assert result["outcome"] == "queued" + assert run_repo.updated[-1][1]["status"] == "queued" + + @pytest.mark.asyncio async def test_dispatch_task_records_failure_for_legacy_invalid_thread_id(): """Rows persisted before the thread-id contract was centralized may store