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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

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)

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())

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

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

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,
     )