From e704c70f32d0268f437e9ddad05b6e01a2f2f60e Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 25 Sep 2026 10:32:41 -0400 Subject: [PATCH] fix: the download boot hook imports a model-only module, keeping the membership roster off the fetch path (#4433) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR --- backend/app/celery_signals.py | 2 +- backend/app/services/download_recovery.py | 57 +++++++++++++++++++++++ backend/app/tasks/maintenance.py | 45 +----------------- tests/test_download_lane_boot.py | 4 +- tests/test_maintenance.py | 2 +- 5 files changed, 62 insertions(+), 48 deletions(-) create mode 100644 backend/app/services/download_recovery.py diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index 0afaf42..1b9e7b2 100644 --- a/backend/app/celery_signals.py +++ b/backend/app/celery_signals.py @@ -246,7 +246,7 @@ def _on_worker_ready(sender=None, **_): booted_at = datetime.now(UTC) try: from .services.platform_lock import release_all_platform_locks - from .tasks.maintenance import interrupt_orphaned_download_events + from .services.download_recovery import interrupt_orphaned_download_events with sync_session_factory()() as session: closed = interrupt_orphaned_download_events(session, booted_at=booted_at) diff --git a/backend/app/services/download_recovery.py b/backend/app/services/download_recovery.py new file mode 100644 index 0000000..e62fc75 --- /dev/null +++ b/backend/app/services/download_recovery.py @@ -0,0 +1,57 @@ +"""Closing the download runs a worker restart orphaned (#4433). + +Its own module, importing nothing but the model, because its caller is the +worker boot hook in `celery_signals` — which every download task imports via +`celery_app`. Living in `tasks.maintenance` put the whole maintenance import +graph, the membership roster included, on the fetch path, which +`test_gated_reason` forbids. +""" +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import literal, update +from sqlalchemy.dialects.postgresql import JSONB + +from ..models import DownloadEvent + +DOWNLOAD_INTERRUPTED_MESSAGE = ( + "interrupted by a worker restart — the next check picks it up where it left off" +) + + +def interrupt_orphaned_download_events(session, *, booted_at: datetime) -> int: + """Close the download events a restart orphaned, without blaming the source. + + Called when the download lane comes up (#4433). Anything still + pending/running from before this boot belongs to the previous process: + a walk that outlived the 90s stop grace was SIGKILLed, and a queued or + serialize-deferred task is held unacked until Redis redelivers it about an + hour later. Left alone, the 30-min stall sweep would error each one and + bump `consecutive_failures`, backing the source off as if the platform had + failed. + + Instead they end as `skipped` (terminal, not a failure) and the source is + not touched: `last_checked_at` keeps its old value, so the next tick finds + it due and the walk resumes from its checkpoint. A redelivered message + that arrives later finds no pending event and opens a fresh one. + + An event promoted to running after the boot has `started_at` reset to its + real start (download_service), so it is never caught here. Does NOT commit. + """ + now = datetime.now(UTC) + result = session.execute( + update(DownloadEvent) + .where(DownloadEvent.status.in_(["pending", "running"])) + .where(DownloadEvent.started_at < booted_at) + .values( + status="skipped", + finished_at=now, + error=DOWNLOAD_INTERRUPTED_MESSAGE, + metadata_=DownloadEvent.metadata_.op("||")( + literal({"error_type": "interrupted"}, JSONB) + ), + ) + .returning(DownloadEvent.id) + ) + return len(result.all()) diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 87c7cb8..31c0fda 100644 --- a/backend/app/tasks/maintenance.py +++ b/backend/app/tasks/maintenance.py @@ -7,8 +7,7 @@ from datetime import UTC, datetime, timedelta from pathlib import Path from PIL import Image -from sqlalchemy import Integer, and_, cast, delete, func, literal, or_, select, update -from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update from ..celery_app import celery from ..models import ( @@ -732,48 +731,6 @@ def recover_stalled_download_events() -> int: return events_recovered -DOWNLOAD_INTERRUPTED_MESSAGE = ( - "interrupted by a worker restart — the next check picks it up where it left off" -) - - -def interrupt_orphaned_download_events(session, *, booted_at: datetime) -> int: - """Close the download events a restart orphaned, without blaming the source. - - Called when the download lane comes up (#4433). Anything still - pending/running from before this boot belongs to the previous process: - a walk that outlived the 90s stop grace was SIGKILLed, and a queued or - serialize-deferred task is held unacked until Redis redelivers it about an - hour later. Left alone, the 30-min stall sweep would error each one and - bump `consecutive_failures`, backing the source off as if the platform had - failed. - - Instead they end as `skipped` (terminal, not a failure) and the source is - not touched: `last_checked_at` keeps its old value, so the next tick finds - it due and the walk resumes from its checkpoint. A redelivered message - that arrives later finds no pending event and opens a fresh one. - - An event promoted to running after the boot has `started_at` reset to its - real start (download_service), so it is never caught here. Does NOT commit. - """ - now = datetime.now(UTC) - result = session.execute( - update(DownloadEvent) - .where(DownloadEvent.status.in_(["pending", "running"])) - .where(DownloadEvent.started_at < booted_at) - .values( - status="skipped", - finished_at=now, - error=DOWNLOAD_INTERRUPTED_MESSAGE, - metadata_=DownloadEvent.metadata_.op("||")( - literal({"error_type": "interrupted"}, JSONB) - ), - ) - .returning(DownloadEvent.id) - ) - return len(result.all()) - - @celery.task(name="backend.app.tasks.maintenance.recover_stalled_backup_runs") def recover_stalled_backup_runs() -> int: """Flip BackupRun rows stuck in running/restoring past the hard limit diff --git a/tests/test_download_lane_boot.py b/tests/test_download_lane_boot.py index e98dd4a..ff52b65 100644 --- a/tests/test_download_lane_boot.py +++ b/tests/test_download_lane_boot.py @@ -35,7 +35,7 @@ def calls(monkeypatch): monkeypatch.setattr(celery_signals, "sync_session_factory", lambda: _Session) monkeypatch.setattr( - "backend.app.tasks.maintenance.interrupt_orphaned_download_events", + "backend.app.services.download_recovery.interrupt_orphaned_download_events", lambda session, *, booted_at: seen.append("interrupt") or 0, ) monkeypatch.setattr( @@ -66,6 +66,6 @@ def test_a_failure_does_not_stop_the_worker_starting(monkeypatch, calls): raise RuntimeError("db down") monkeypatch.setattr( - "backend.app.tasks.maintenance.interrupt_orphaned_download_events", boom + "backend.app.services.download_recovery.interrupt_orphaned_download_events", boom ) celery_signals._on_worker_ready(sender=_consumer("download")) # must not raise diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 4a1712b..c9df368 100644 --- a/tests/test_maintenance.py +++ b/tests/test_maintenance.py @@ -726,7 +726,7 @@ def test_download_lane_boot_closes_pre_boot_events_without_blaming_the_source(db from sqlalchemy import select from backend.app.models import DownloadEvent, Source - from backend.app.tasks.maintenance import ( + from backend.app.services.download_recovery import ( DOWNLOAD_INTERRUPTED_MESSAGE, interrupt_orphaned_download_events, )