mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-01 19:06:01 +00:00
c38d291505
Replaces the pre-hexagonal scheduled-task implementation with a slice built to the layering spec: a pure domain (two aggregates, two state machines, the policy value object), output ports it declares itself, SQL/launcher/thread adapters implementing them under `app/adapters/`, and a composition root that is the one place any of them is instantiated. The old implementation mixed all of that into `app/scheduler/service.py` and a router that reached straight into repositories, so the rules that matter -- overlap policy, lease handling, which write owns which timestamp -- were only reachable through a live database. They are now unit-assertable on in-memory fakes, with the contract suite running each port against both the fake and real sqlite, and the concurrency invariants pinned by dedicated race tests. Two bugs the old shape hid are fixed on the way: a completion hook that replayed a stale snapshot and rolled back the launch write, and a corrupt stored row surfacing to the client as a 4xx. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
"""Driving adapter -- the clock that asks the schedule service to work.
|
|
|
|
This is the only part of the scheduler that knows about time passing. It owns
|
|
*when* `ScheduleService.run_once` is called and what happens when a poll fails;
|
|
it owns nothing about what a poll means. Everything the old
|
|
`app/scheduler/service.py` mixed into its loop -- overlap policy, lease
|
|
semantics, budget accounting -- now lives in the domain and reaches this file
|
|
only as one awaited call.
|
|
|
|
Two behaviours here are load-bearing:
|
|
|
|
- **A failing poll must not end the loop.** A transient error (SQLite's
|
|
"database is locked" is the realistic one) would otherwise stop every
|
|
scheduled task for the rest of the process life, silently.
|
|
- **Startup reconciliation must not block startup.** The service lets
|
|
reconcile failures propagate on purpose -- whether they are fatal is the
|
|
caller's policy -- and this caller's policy is to log and keep scheduling.
|
|
A gateway refusing to start over leftover rows is worse than one running
|
|
with them.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from datetime import UTC, datetime
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from deerflow.domain.schedule.service import ScheduleService
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
RESTART_ERROR = "interrupted: gateway restarted before the run reached a terminal state"
|
|
|
|
|
|
class SchedulePoller:
|
|
"""Runs `ScheduleService.run_once` on an interval until stopped."""
|
|
|
|
def __init__(self, service: ScheduleService, *, poll_interval_seconds: float) -> None:
|
|
self._service = service
|
|
self._poll_interval_seconds = poll_interval_seconds
|
|
self._task: asyncio.Task | None = None
|
|
self._stop = asyncio.Event()
|
|
|
|
async def start(self) -> None:
|
|
"""Reconcile what a crash left behind, then begin polling.
|
|
|
|
Idempotent: a second call while running is a no-op, so a caller cannot
|
|
end up with two loops claiming the same tasks.
|
|
"""
|
|
if self._task is not None:
|
|
return
|
|
try:
|
|
stale_runs, stuck_tasks = await self._service.reconcile_on_startup(error=RESTART_ERROR)
|
|
if stale_runs:
|
|
logger.warning("Marked %d stale scheduled task run(s) as interrupted after restart", stale_runs)
|
|
if stuck_tasks:
|
|
logger.warning("Cancelled %d stuck once task(s) after restart", stuck_tasks)
|
|
except Exception:
|
|
logger.exception("Failed to reconcile scheduled tasks at startup; scheduling anyway")
|
|
self._stop.clear()
|
|
self._task = asyncio.create_task(self._run_loop())
|
|
|
|
async def stop(self) -> None:
|
|
"""Signal the loop and wait for the in-flight poll to finish."""
|
|
if self._task is None:
|
|
return
|
|
self._stop.set()
|
|
await self._task
|
|
self._task = None
|
|
|
|
async def _run_loop(self) -> None:
|
|
while not self._stop.is_set():
|
|
try:
|
|
await self._service.run_once(now=datetime.now(UTC))
|
|
except Exception:
|
|
logger.exception("Scheduled task poll failed; retrying next interval")
|
|
try:
|
|
# Waiting on the stop event rather than sleeping keeps shutdown
|
|
# prompt at a production-sized interval.
|
|
await asyncio.wait_for(self._stop.wait(), timeout=self._poll_interval_seconds)
|
|
except TimeoutError:
|
|
continue
|