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

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