From 06dda5045a6ef0eb9e349c589c06170bde21a7b6 Mon Sep 17 00:00:00 2001 From: rayhpeng Date: Wed, 29 Jul 2026 19:46:44 +0800 Subject: [PATCH] fix(schedule): stop a corrupt stored row surfacing as a client error SqlScheduledTaskRepository._to_domain raised InvalidScheduleError for a row whose stored schedule no longer parses -- the same error the aggregate raises for a client-submitted schedule, which the router maps to 422. A stored fault therefore told the client its perfectly fine request was wrong, and made the row unrepairable over HTTP: PATCH reads the task before writing, so the fix path 422'd too. Enum rebuild failures were worse -- a raw ValueError crossed the boundary untranslated. The rebuild is now translated to a dedicated CorruptStoredScheduleError, raised only by the persistence adapter and deliberately absent from the router's status table, so it falls through to the unclassified-500 branch: a server-side fault reported as one. List reads keep skipping and logging the bad row. Pinned by tests/test_schedule_corrupt_rows.py against a real sqlite database. --- .../schedule/scheduled_task_repository.py | 63 +++++---- .../deerflow/domain/schedule/__init__.py | 2 + .../deerflow/domain/schedule/exceptions.py | 11 ++ backend/tests/test_schedule_corrupt_rows.py | 126 ++++++++++++++++++ 4 files changed, 174 insertions(+), 28 deletions(-) create mode 100644 backend/tests/test_schedule_corrupt_rows.py diff --git a/backend/app/adapters/schedule/scheduled_task_repository.py b/backend/app/adapters/schedule/scheduled_task_repository.py index 750cb4107..2513c33be 100644 --- a/backend/app/adapters/schedule/scheduled_task_repository.py +++ b/backend/app/adapters/schedule/scheduled_task_repository.py @@ -26,7 +26,7 @@ from datetime import UTC, datetime, timedelta from sqlalchemy import and_, or_, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker -from deerflow.domain.schedule.exceptions import InvalidScheduleError +from deerflow.domain.schedule.exceptions import CorruptStoredScheduleError, ScheduleError from deerflow.domain.schedule.model import TERMINAL_TASK_STATUSES, ContextMode, ScheduledTask, ScheduleSpec, ScheduleType, TaskStatus from deerflow.domain.schedule.ports import ScheduledTaskRepository @@ -92,39 +92,46 @@ class SqlScheduledTaskRepository(ScheduledTaskRepository): def _to_domain(row: ScheduledTaskRow) -> ScheduledTask: """ORM row -> aggregate. - Raises InvalidScheduleError for a row whose stored schedule can no - longer be parsed. That is a **new** failure mode: the legacy repository - returned bare dicts and a corrupt spec only surfaced at dispatch time. - Single-row reads let it propagate; list reads skip and log, so one bad - row cannot 500 an entire listing. + Raises CorruptStoredScheduleError for a row that can no longer be + rebuilt -- unparseable schedule, unknown enum value. The rebuild goes + through the aggregate's own validation, so a hand-damaged row blows up + here rather than flowing on; the translation to the dedicated error is + what keeps it off ``InvalidScheduleError``'s client-facing 422 mapping + (a stored fault is the server's problem, and PATCH -- which reads + before writing -- must not become unusable for the one row it could + repair). Single-row reads let it propagate; list reads skip and log, + so one bad row cannot take down an entire listing. """ - return ScheduledTask( - task_id=row.id, - user_id=row.user_id, - title=row.title, - prompt=row.prompt, - schedule=SqlScheduledTaskRepository._spec_from_row(row), - context_mode=ContextMode(row.context_mode), - thread_id=row.thread_id, - assistant_id=row.assistant_id, - status=TaskStatus(row.status), - overlap_policy=row.overlap_policy, - next_run_at=_tz_aware(row.next_run_at), - last_run_at=_tz_aware(row.last_run_at), - last_run_id=row.last_run_id, - last_thread_id=row.last_thread_id, - last_error=row.last_error, - run_count=row.run_count, - created_at=_tz_aware(row.created_at), - updated_at=_tz_aware(row.updated_at), - ) + try: + return ScheduledTask( + task_id=row.id, + user_id=row.user_id, + title=row.title, + prompt=row.prompt, + schedule=SqlScheduledTaskRepository._spec_from_row(row), + context_mode=ContextMode(row.context_mode), + thread_id=row.thread_id, + assistant_id=row.assistant_id, + status=TaskStatus(row.status), + overlap_policy=row.overlap_policy, + next_run_at=_tz_aware(row.next_run_at), + last_run_at=_tz_aware(row.last_run_at), + last_run_id=row.last_run_id, + last_thread_id=row.last_thread_id, + last_error=row.last_error, + run_count=row.run_count, + created_at=_tz_aware(row.created_at), + updated_at=_tz_aware(row.updated_at), + ) + except (ScheduleError, ValueError) as exc: + raise CorruptStoredScheduleError(f"stored scheduled task {row.id} cannot be rebuilt: {exc}") from exc @classmethod def _to_domain_or_skip(cls, row: ScheduledTaskRow) -> ScheduledTask | None: try: return cls._to_domain(row) - except InvalidScheduleError: - logger.warning("Skipping scheduled task %s: stored schedule is unparseable", row.id, exc_info=True) + except CorruptStoredScheduleError: + logger.warning("Skipping scheduled task %s: stored row cannot be rebuilt", row.id, exc_info=True) return None @staticmethod diff --git a/backend/packages/harness/deerflow/domain/schedule/__init__.py b/backend/packages/harness/deerflow/domain/schedule/__init__.py index c0bcd05fd..f673e972e 100644 --- a/backend/packages/harness/deerflow/domain/schedule/__init__.py +++ b/backend/packages/harness/deerflow/domain/schedule/__init__.py @@ -18,6 +18,7 @@ from deerflow.domain.schedule.commands import ( ) from deerflow.domain.schedule.exceptions import ( ActiveRunConflictError, + CorruptStoredScheduleError, InvalidContextModeError, InvalidScheduleError, LaunchFailedError, @@ -52,6 +53,7 @@ __all__ = [ "ActiveRunConflictError", "ContextChange", "ContextMode", + "CorruptStoredScheduleError", "CreateScheduledTask", "DeleteTask", "DispatchOutcome", diff --git a/backend/packages/harness/deerflow/domain/schedule/exceptions.py b/backend/packages/harness/deerflow/domain/schedule/exceptions.py index d34d4a815..d07c3eb36 100644 --- a/backend/packages/harness/deerflow/domain/schedule/exceptions.py +++ b/backend/packages/harness/deerflow/domain/schedule/exceptions.py @@ -33,6 +33,17 @@ class ThreadNotFoundError(ScheduleError): """reuse_thread points at a thread the user cannot access.""" +class CorruptStoredScheduleError(ScheduleError): + """A stored task row can no longer be rebuilt into a valid aggregate. + + Raised by the persistence adapter, never by the aggregate: it means the + *storage* is damaged, not that a client submitted something invalid -- + which is why it is deliberately absent from the router's status table and + falls through to the unclassified-500 branch instead of riding + ``InvalidScheduleError``'s 422. + """ + + class ActiveRunConflictError(ScheduleError): """The task already holds its single active run slot. diff --git a/backend/tests/test_schedule_corrupt_rows.py b/backend/tests/test_schedule_corrupt_rows.py new file mode 100644 index 000000000..d3335982f --- /dev/null +++ b/backend/tests/test_schedule_corrupt_rows.py @@ -0,0 +1,126 @@ +"""A corrupt stored schedule is an operator problem, not a client error. + +``SqlScheduledTaskRepository._to_domain`` rebuilds the aggregate on every +read, so a row whose stored schedule no longer parses surfaces at read time. +It used to surface as ``InvalidScheduleError`` -- the same error the aggregate +raises for a *client-submitted* schedule, which the router maps to 422. That +double duty told the client "your request is wrong" about a request that was +perfectly fine, and made the row unrepairable over HTTP: PATCH reads the task +before writing, so the fix path 422'd too. + +``CorruptStoredScheduleError`` splits the vocabulary: it is raised only by the +persistence adapter, and it is deliberately absent from the router's status +table so it falls through to the unclassified-500 branch -- a server-side +fault reported as one. + +SQL-only: the in-memory double stores whole aggregates and cannot hold a +corrupt row, which is exactly why this file exists beside the contract suite +rather than inside it. +""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from datetime import UTC, datetime + +import pytest +import pytest_asyncio + +from app.adapters.schedule.scheduled_task_repository import SqlScheduledTaskRepository +from app.gateway.routers.schedule.router import _STATUS_BY_ERROR +from deerflow.config.database_config import DatabaseConfig +from deerflow.domain.schedule.exceptions import CorruptStoredScheduleError, InvalidScheduleError +from deerflow.domain.schedule.model import ScheduledTask, SchedulePolicy, ScheduleSpec +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow + +pytestmark = pytest.mark.asyncio + +NOW = datetime(2026, 7, 27, 12, 0, tzinfo=UTC) + + +@pytest_asyncio.fixture +async def repo(tmp_path) -> AsyncIterator[SqlScheduledTaskRepository]: + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + try: + sf = get_session_factory() + assert sf is not None + yield SqlScheduledTaskRepository(sf) + finally: + await close_engine() + + +async def _add_task(repo: SqlScheduledTaskRepository, *, title: str = "healthy") -> ScheduledTask: + return await repo.add( + ScheduledTask.create( + user_id="user-1", + title=title, + prompt="p", + schedule=ScheduleSpec.cron_schedule("0 9 * * *", "UTC"), + context_mode="fresh_thread_per_run", + thread_id=None, + now=NOW, + policy=SchedulePolicy(), + ) + ) + + +async def _corrupt(task_id: str, **columns) -> None: + """Damage a stored row directly -- the exact shape a bug or a manual edit + would leave behind, unreachable through the port.""" + sf = get_session_factory() + assert sf is not None + async with sf() as session: + row = await session.get(ScheduledTaskRow, task_id) + assert row is not None + for name, value in columns.items(): + setattr(row, name, value) + await session.commit() + + +class TestSingleRowReads: + async def test_a_corrupt_schedule_raises_the_dedicated_error(self, repo): + task = await _add_task(repo) + await _corrupt(task.task_id, schedule_spec={}) + + with pytest.raises(CorruptStoredScheduleError): + await repo.get(task.task_id, user_id="user-1") + + async def test_the_dedicated_error_is_not_the_client_facing_one(self, repo): + """The router maps InvalidScheduleError to 422; a corrupt row must not + ride that mapping.""" + task = await _add_task(repo) + await _corrupt(task.task_id, schedule_spec={}) + + with pytest.raises(CorruptStoredScheduleError) as exc_info: + await repo.get(task.task_id, user_id="user-1") + assert not isinstance(exc_info.value, InvalidScheduleError) + + async def test_a_corrupt_context_mode_is_reported_the_same_way(self, repo): + """Enum rebuild failures are the same fault as spec parse failures: a + raw ValueError out of the adapter would be a technical exception + crossing the boundary untranslated.""" + task = await _add_task(repo) + await _corrupt(task.task_id, context_mode="not-a-mode") + + with pytest.raises(CorruptStoredScheduleError): + await repo.get(task.task_id, user_id="user-1") + + +class TestListReads: + async def test_a_corrupt_row_does_not_take_down_the_listing(self, repo): + healthy = await _add_task(repo, title="healthy") + broken = await _add_task(repo, title="broken") + await _corrupt(broken.task_id, schedule_spec={}) + + listed = await repo.list_by_user("user-1") + + assert [t.task_id for t in listed] == [healthy.task_id] + + +class TestRouterMapping: + def test_the_corrupt_row_error_is_unclassified_on_purpose(self): + """Absent from the status table means the router's fallthrough turns + it into a 500 -- a new protocol decision would have to add it here + deliberately.""" + assert CorruptStoredScheduleError not in _STATUS_BY_ERROR