diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index dec94ea..0afaf42 100644 --- a/backend/app/celery_signals.py +++ b/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") diff --git a/backend/app/services/platform_lock.py b/backend/app/services/platform_lock.py index d910016..89f36ca 100644 --- a/backend/app/services/platform_lock.py +++ b/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 diff --git a/backend/app/tasks/maintenance.py b/backend/app/tasks/maintenance.py index 31c0fda..87c7cb8 100644 --- a/backend/app/tasks/maintenance.py +++ b/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 diff --git a/tests/test_download_lane_boot.py b/tests/test_download_lane_boot.py new file mode 100644 index 0000000..e98dd4a --- /dev/null +++ b/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 diff --git a/tests/test_maintenance.py b/tests/test_maintenance.py index 7bf6ab5..4a1712b 100644 --- a/tests/test_maintenance.py +++ b/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