fix: a restart no longer strands downloads — the download lane closes its predecessor's runs as interrupted and frees the platform locks at boot (#4433)

A walk that outlives the 90s stop grace is SIGKILLed: its event never
finalizes and its platform lock is held for the 27-min TTL. The 30-min
sweep then errored every stranded event and bumped consecutive_failures,
backing sources off (and blocking backfills) as if the platform failed.

On worker_ready, the process consuming 'download' ends pre-boot
pending/running events as skipped with error_type 'interrupted', leaves
the source untouched so the next tick resumes it, and releases the
serialized platforms' locks.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LVjrnpQjRgHdvq95rASoiR

backend/app/celery_signals.py

@@ -23,7 +23,13 @@ import logging
 from datetime import UTC, datetime
 
 from celery.exceptions import SoftTimeLimitExceeded
-from celery.signals import task_failure, task_postrun, task_prerun, task_retry
+from celery.signals import (
+    task_failure,
+    task_postrun,
+    task_prerun,
+    task_retry,
+    worker_ready,
+)
 
 from .models import TaskRun
 from .tasks._sync_engine import sync_session_factory
@@ -213,3 +219,43 @@ def _on_retry(sender=None, request=None, reason=None, einfo=None, **_):
         error_message=str(reason) if reason else None,
         retry_count=getattr(request, "retries", 0),
     )
+
+
+def _consumed_queues(consumer) -> set[str]:
+    """The queue names this worker process consumes (its `-Q`), or empty when
+    they can't be read — which makes the boot hook below a no-op, not a guess."""
+    try:
+        return {q.name for q in consumer.task_consumer.queues}
+    except Exception:  # noqa: BLE001 — shape varies across celery versions
+        return set()
+
+
+@worker_ready.connect
+def _on_worker_ready(sender=None, **_):
+    """The download lane clears what its previous process left behind (#4433).
+
+    A restart SIGKILLs any walk that outlives the stop grace, so its event is
+    never finalized and its platform lock is never released. Both used to wait
+    out timers — a 30-min stall sweep that then blamed the source, and a 27-min
+    lock TTL that stalled every other source on the platform. Only the process
+    consuming `download` does this; the other lanes booting beside it must not.
+    Best-effort: a failure here is logged and the worker still starts.
+    """
+    if "download" not in _consumed_queues(sender):
+        return
+    booted_at = datetime.now(UTC)
+    try:
+        from .services.platform_lock import release_all_platform_locks
+        from .tasks.maintenance import interrupt_orphaned_download_events
+
+        with sync_session_factory()() as session:
+            closed = interrupt_orphaned_download_events(session, booted_at=booted_at)
+            session.commit()
+        released = release_all_platform_locks()
+        if closed or released:
+            log.info(
+                "download lane boot: closed %d orphaned download event(s) as "
+                "interrupted, released %d platform lock(s)", closed, released,
+            )
+    except Exception:  # noqa: BLE001 — never block the worker from starting
+        log.exception("download lane boot recovery failed")

backend/app/services/platform_lock.py

@@ -60,3 +60,22 @@ def platform_lock(platform: str, *, ttl_seconds: int):
     except redis.RedisError as exc:  # pragma: no cover - broker outage
         log.warning("platform_lock unavailable for %s: %s", platform, exc)
         return None
+
+
+def release_all_platform_locks() -> int:
+    """Drop every serialized platform's lock. Returns how many were held.
+
+    Only for the download lane's boot (#4433). A worker that restarts mid-walk
+    is SIGKILLed past its stop grace, so its `finally` never releases the lock,
+    and the TTL keeps every other source on that platform bouncing for up to
+    27 minutes after the new worker is ready. At boot no walk of ours can be
+    running, so a held lock names a dead one. Assumes one download consumer —
+    the only shape FC deploys; a second replica booting would free a live
+    walk's lock (not corrupt it: the walk runs on, a second walk may overlap it).
+    """
+    try:
+        client = _redis()
+        return int(client.delete(*(f"{_LOCK_PREFIX}{p}" for p in SERIALIZED_PLATFORMS)))
+    except redis.RedisError as exc:  # pragma: no cover - broker outage
+        log.warning("could not release platform locks at boot: %s", exc)
+        return 0

backend/app/tasks/maintenance.py

@@ -7,7 +7,8 @@ from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 from PIL import Image
-from sqlalchemy import Integer, and_, cast, delete, func, or_, select, update
+from sqlalchemy import Integer, and_, cast, delete, func, literal, or_, select, update
+from sqlalchemy.dialects.postgresql import JSONB
 
 from ..celery_app import celery
 from ..models import (
@@ -731,6 +732,48 @@ 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

@@ -0,0 +1,71 @@
+"""#4433: only the process consuming `download` clears a restart's leftovers.
+
+The consolidated container boots four lanes side by side. If the scheduler or
+ml lane ran this too, a lane restarting on its own (supervisord restarts a
+crashed program) would close the events of walks that are still running.
+"""
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from backend.app import celery_signals
+
+
+def _consumer(*queues: str):
+    return SimpleNamespace(
+        task_consumer=SimpleNamespace(queues=[SimpleNamespace(name=q) for q in queues])
+    )
+
+
+@pytest.fixture
+def calls(monkeypatch):
+    seen: list[str] = []
+
+    class _Session:
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+        def commit(self):
+            seen.append("commit")
+
+    monkeypatch.setattr(celery_signals, "sync_session_factory", lambda: _Session)
+    monkeypatch.setattr(
+        "backend.app.tasks.maintenance.interrupt_orphaned_download_events",
+        lambda session, *, booted_at: seen.append("interrupt") or 0,
+    )
+    monkeypatch.setattr(
+        "backend.app.services.platform_lock.release_all_platform_locks",
+        lambda: seen.append("release") or 0,
+    )
+    return seen
+
+
+def test_the_download_lane_clears_orphans_and_locks(calls):
+    celery_signals._on_worker_ready(sender=_consumer("default", "download", "import"))
+    assert calls == ["interrupt", "commit", "release"]
+
+
+@pytest.mark.parametrize("queues", [("maintenance", "scan"), ("ml",), ("maintenance_long",)])
+def test_other_lanes_leave_downloads_alone(calls, queues):
+    celery_signals._on_worker_ready(sender=_consumer(*queues))
+    assert calls == []
+
+
+def test_unreadable_queues_do_nothing_rather_than_guess(calls):
+    celery_signals._on_worker_ready(sender=SimpleNamespace())
+    assert calls == []
+
+
+def test_a_failure_does_not_stop_the_worker_starting(monkeypatch, calls):
+    def boom(session, *, booted_at):
+        raise RuntimeError("db down")
+
+    monkeypatch.setattr(
+        "backend.app.tasks.maintenance.interrupt_orphaned_download_events", boom
+    )
+    celery_signals._on_worker_ready(sender=_consumer("download"))  # must not raise

tests/test_maintenance.py

@@ -718,6 +718,57 @@ def test_recover_stalled_download_skips_fresh(db_sync):
     assert failures == 0
 
 
+def test_download_lane_boot_closes_pre_boot_events_without_blaming_the_source(db_sync):
+    """#4433: a restart SIGKILLs a walk past its stop grace and strands queued
+    ones. At the download lane's boot, everything pending/running from before
+    it ends as `skipped`/interrupted — and the source is left as it was, so it
+    is due on the next tick instead of backed off as a failure."""
+    from sqlalchemy import select
+
+    from backend.app.models import DownloadEvent, Source
+    from backend.app.tasks.maintenance import (
+        DOWNLOAD_INTERRUPTED_MESSAGE,
+        interrupt_orphaned_download_events,
+    )
+
+    sid = _make_source(db_sync, slug="rebooted")
+    booted_at = datetime.now(UTC)
+    before = booted_at - timedelta(minutes=5)
+    walking = DownloadEvent(
+        source_id=sid, status="running", started_at=before,
+        metadata_={"live": {"downloaded": 3}},
+    )
+    queued = DownloadEvent(source_id=sid, status="pending", started_at=before)
+    finished = DownloadEvent(source_id=sid, status="ok", started_at=before)
+    # Promoted to running after the boot: download_service resets started_at.
+    fresh = DownloadEvent(
+        source_id=sid, status="running", started_at=booted_at + timedelta(seconds=5),
+    )
+    db_sync.add_all([walking, queued, finished, fresh])
+    db_sync.commit()
+
+    closed = interrupt_orphaned_download_events(db_sync, booted_at=booted_at)
+    db_sync.commit()
+
+    assert closed == 2
+    db_sync.expire_all()
+    for ev in (walking, queued):
+        assert ev.status == "skipped"
+        assert ev.error == DOWNLOAD_INTERRUPTED_MESSAGE
+        assert ev.finished_at is not None
+        assert ev.metadata_["error_type"] == "interrupted"
+    assert walking.metadata_["live"] == {"downloaded": 3}
+    assert finished.status == "ok"
+    assert fresh.status == "running"
+    src = db_sync.execute(
+        select(Source.consecutive_failures, Source.last_error, Source.last_checked_at)
+        .where(Source.id == sid)
+    ).one()
+    assert src.consecutive_failures == 0
+    assert src.last_error is None
+    assert src.last_checked_at is None
+
+
 def test_recover_stalled_download_flips_stale_pending(db_sync):
     """A 2-hour-old pending event flips to error AND the source is bumped
     (consecutive_failures, last_error, last_checked_at) so the next scan