From 1e63f70bf71843dab0c40eb41a28e8da4960cba3 Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Sat, 7 Nov 2026 14:53:57 -0500 Subject: [PATCH] fix: a media backup or restore cut off by a restart carries on where it stopped A media backup and a restore's file phase run as chains of 45-minute chunks, each queuing the next when it ends. A restart killed the chunk in flight, so nothing queued the next one; the run's row stayed "running", "Back up now" skipped it as already running, and only the 7-hour stall sweep ended it. - When the long-maintenance lane starts, it queues the next chunk of every media run still running and every restore fetching its files. - Each chunk writes its token to the run and hands it to the chunk it queues. A chunk queued by any other chunk is dropped (a killed chunk's message redelivered by acks_late, say), so a resume never doubles the chain; a chunk taken over stops at its next file. - The integration job installs postgresql-client when the CI image lacks it: the backup round trip in 375e7e2 runs the real pg_dump and pg_restore. Co-Authored-By: Claude Opus 5.5 --- .forgejo/workflows/build.yml | 3 + README.md | 3 +- backend/app/celery_signals.py | 33 +++++-- backend/app/tasks/backup.py | 165 +++++++++++++++++++++++++------ tests/test_backup_media.py | 64 +++++++++++- tests/test_backup_restore.py | 31 +++++- tests/test_download_lane_boot.py | 27 ++++- 7 files changed, 281 insertions(+), 45 deletions(-) diff --git a/.forgejo/workflows/build.yml b/.forgejo/workflows/build.yml index 8c049f7..07010bc 100644 --- a/.forgejo/workflows/build.yml +++ b/.forgejo/workflows/build.yml @@ -579,6 +579,9 @@ jobs: else pip install -r requirements.txt pytest pytest-asyncio fi + # The backup round trip runs the real pg_dump and pg_restore, from the + # same Debian client package the app image installs. + command -v pg_dump >/dev/null 2>&1 || { apt-get update -qq && apt-get install -y -qq postgresql-client; } # Relax durability on the throwaway CI Postgres so the per-test # TRUNCATE's commit-fsync — the integration teardown's dominant cost # (~1.5-2s/test, which collapsed the suite from ~13min to ~45s) — is diff --git a/README.md b/README.md index 699baf7..e8204d4 100644 --- a/README.md +++ b/README.md @@ -189,7 +189,8 @@ screen: any S3 bucket (Fabled Archives, MinIO, AWS) and a key that may add and read objects. Each night FabledCurator then sends the database as a dated dump under `database/`, and every image and attachment once, named by its fingerprint, under `media/`. It never deletes from the bucket; the bucket's own -retention thins old dumps. +retention thins old dumps. The first copy of a large library takes hours; a +restart along the way carries on where it stopped rather than starting over. To rebuild after losing the server, start a fresh install (first boot as above), enter the same destination, and under **Restore from the bucket** choose diff --git a/backend/app/celery_signals.py b/backend/app/celery_signals.py index f30c492..e15ef2f 100644 --- a/backend/app/celery_signals.py +++ b/backend/app/celery_signals.py @@ -232,17 +232,38 @@ def _consumed_queues(consumer) -> set[str]: @worker_ready.connect def _on_worker_ready(sender=None, **_): - """The download lane clears what its previous process left behind (#4433). + """A lane picks up what its previous process left behind. Only the lane + that runs the work does it; the other lanes booting beside it must not, + or a lane restarting on its own would disturb work still running. + Best-effort: a failure is logged and the worker still starts.""" + queues = _consumed_queues(sender) + if "download" in queues: + _clear_download_leftovers() + if "maintenance_long" in queues: + _resume_offsite_runs() + + +def _resume_offsite_runs() -> None: + """The long-maintenance lane carries on an off-site backup or restore that + a restart cut off mid-chunk. With nothing queuing its next chunk, the run + would read "running" for hours, and block every new run, until the stall + sweep ended it.""" + try: + from .tasks.backup import resume_offsite_runs + + resume_offsite_runs() + except Exception: # noqa: BLE001 — never block the worker from starting + log.exception("long-maintenance lane boot: resuming off-site runs failed") + + +def _clear_download_leftovers() -> None: + """The download lane clears what its previous process left behind. 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. + lock TTL that stalled every other source on the platform. """ - if "download" not in _consumed_queues(sender): - return booted_at = datetime.now(UTC) try: from .services.download_recovery import interrupt_orphaned_download_events diff --git a/backend/app/tasks/backup.py b/backend/app/tasks/backup.py index 25a2998..33d0866 100644 --- a/backend/app/tasks/backup.py +++ b/backend/app/tasks/backup.py @@ -9,6 +9,7 @@ from __future__ import annotations import logging import time +import uuid from datetime import UTC, datetime, timedelta from pathlib import Path @@ -371,6 +372,79 @@ def _offsite_manifest(target) -> dict: "key": backup_media.PREFIX} +# ── One chunk at a time ────────────────────────────────────────────────────── +# +# A media backup and a restore's file phase run as a chain of chunks, each +# queuing the next. The chunk carrying a run writes its token to the run's +# manifest as "chunk", and hands that token to the chunk it queues as `after`. +# A chunk queued to resume a run after a restart carries `resume` instead and +# takes the run over. Any other message is stale (a killed chunk's own message +# delivered again, say) and is dropped, so a run never goes on twice and the +# chain never doubles. + + +def _chunk_token(task) -> str: + return task.request.id or uuid.uuid4().hex + + +def _claim_run(session, run_id: int, status: str, token: str, *, after: str | None, + resume: bool) -> str | None: + """Take the run for this chunk. Returns None when taken, or why not.""" + row = session.get(BackupRun, run_id, with_for_update=True) + if row is None or row.status != status: + session.rollback() + return "the run is no longer going" + held = (row.manifest or {}).get("chunk") + if not resume and held is not None and held != after: + session.rollback() + return "another chunk carries this run on" + row.manifest = {**(row.manifest or {}), "chunk": token} + session.commit() + return None + + +def _still_ours(session, run_id: int, token: str) -> bool: + """Lock the run's row and say whether this chunk still carries it. A chunk + that was taken over stops at its next file rather than write over the + newer chunk's place; on False the transaction is already rolled back.""" + held = session.execute( + select(BackupRun.manifest).where(BackupRun.id == run_id).with_for_update() + ).scalar_one_or_none() + if (held or {}).get("chunk") == token: + return True + session.rollback() + return False + + +def resume_offsite_runs() -> dict: + """Queue the next chunk of every off-site run a restart cut short: a media + backup still running, and a restore fetching its files. Called when the + long-maintenance lane starts, where those chunks run, so none of theirs is + alive then; the queued chunk carries on from the run's saved place. A + restore stopped while loading its database isn't resumed: the stall sweep + ends it, and the screen offers the restore again.""" + with _sync_session_factory()() as session: + media = list(session.execute( + select(BackupRun.id).where(BackupRun.kind == "media") + .where(BackupRun.status == "running") + ).scalars()) + restores = [ + r.id for r in session.execute( + select(BackupRun).where(BackupRun.kind == "restore") + .where(BackupRun.status == "restoring") + ).scalars() + if (r.manifest or {}).get("phase") == "files" + ] + for run_id in media: + backup_media_task.delay(run_id=run_id, resume=True) + for run_id in restores: + restore_offsite_task.delay(run_id=run_id, resume=True) + if media or restores: + log.info("resumed off-site runs after a restart: media %s, restore %s", + media, restores) + return {"media": media, "restore": restores} + + @celery.task( name="backend.app.tasks.backup.backup_media_task", bind=True, @@ -381,15 +455,17 @@ def _offsite_manifest(target) -> dict: # limit included. soft_time_limit=21600, time_limit=23400, ) -def backup_media_task(self, *, run_id: int | None = None, - triggered_by: str = "manual") -> dict: +def backup_media_task(self, *, run_id: int | None = None, triggered_by: str = "manual", + after: str | None = None, resume: bool = False) -> dict: """Send the library's images and attachments that aren't in the off-site bucket yet, one chunk of a run. Called without `run_id` it starts a run (unless one is already going); each chunk queues the next until nothing is - left. A file that can't be sent is listed on the run and tried again by - the next run. Returns {'backup_run_id': N, ...}.""" + left, and a restart resumes the run where it stopped. A file that can't be + sent is listed on the run and tried again by the next run. Returns + {'backup_run_id': N, ...}.""" SessionLocal = _sync_session_factory() now = datetime.now(UTC) + token = _chunk_token(self) with SessionLocal() as session: if run_id is None: going = session.execute( @@ -403,14 +479,15 @@ def backup_media_task(self, *, run_id: int | None = None, return {"skipped": "a restore is going"} row = BackupRun(kind="media", status="running", triggered_by=triggered_by, started_at=now, last_progress_at=now, size_bytes=0, - manifest={"sent": 0, "failed": 0, "failures": [], "cursor": ""}) + manifest={"sent": 0, "failed": 0, "failures": [], "cursor": "", + "chunk": token}) session.add(row) session.commit() run_id = row.id else: - row = session.get(BackupRun, run_id) - if row is None or row.status != "running": - return {"backup_run_id": run_id, "skipped": "run is no longer running"} + why = _claim_run(session, run_id, "running", token, after=after, resume=resume) + if why: + return {"backup_run_id": run_id, "skipped": why} try: with SessionLocal() as session: @@ -419,7 +496,8 @@ def backup_media_task(self, *, run_id: int | None = None, limit_mbps = settings.backup_upload_limit_mbps if target is None: raise backup_s3.S3Error("No off-site destination is set up.") - more = _send_media_chunk(SessionLocal, run_id, target, backup_s3.Pacer(limit_mbps)) + more = _send_media_chunk(SessionLocal, run_id, target, backup_s3.Pacer(limit_mbps), + token) except (SoftTimeLimitExceeded, Exception) as exc: with SessionLocal() as session: row = session.get(BackupRun, run_id) @@ -427,8 +505,10 @@ def backup_media_task(self, *, run_id: int | None = None, _mark_failed(session, row, exc) raise + if more is None: + return {"backup_run_id": run_id, "skipped": "another chunk took the run over"} if more: - backup_media_task.delay(run_id=run_id) + backup_media_task.delay(run_id=run_id, after=token) return {"backup_run_id": run_id, "continues": True} with SessionLocal() as session: row = session.get(BackupRun, run_id) @@ -443,10 +523,11 @@ def backup_media_task(self, *, run_id: int | None = None, return {"backup_run_id": run_id, "continues": False} -def _send_media_chunk(SessionLocal, run_id: int, target, pacer) -> bool: +def _send_media_chunk(SessionLocal, run_id: int, target, pacer, token: str) -> bool | None: """Send files until the chunk's time is up or none are left. Progress is committed after every file, so a chunk that dies loses at most the file - in flight. Returns True when files remain for another chunk.""" + in flight. Returns True when files remain for another chunk, and None when + another chunk has taken the run over.""" tid = backup_media.target_id(target) deadline = time.monotonic() + MEDIA_CHUNK_SECONDS with SessionLocal() as session: @@ -460,6 +541,8 @@ def _send_media_chunk(SessionLocal, run_id: int, target, pacer) -> bool: # across a long transfer holds back vacuum for its whole length. session.commit() if not batch: + if not _still_ours(session, run_id, token): + return None m["cursor"] = cursor row.manifest = dict(m) row.last_progress_at = datetime.now(UTC) @@ -467,10 +550,13 @@ def _send_media_chunk(SessionLocal, run_id: int, target, pacer) -> bool: return False for blob in batch: if time.monotonic() > deadline: + if not _still_ours(session, run_id, token): + return None m["cursor"] = cursor row.manifest = dict(m) session.commit() return True + key = None try: key = backup_media.send(target, blob, IMAGES_ROOT, pacer) except (backup_s3.S3Error, OSError) as exc: @@ -478,7 +564,9 @@ def _send_media_chunk(SessionLocal, run_id: int, target, pacer) -> bool: if len(m.get("failures") or []) < MEDIA_FAILURES_LISTED: m["failures"] = [*(m.get("failures") or []), {"path": blob.path, "error": str(exc)[:300]}] - else: + if not _still_ours(session, run_id, token): + return None + if key is not None: backup_media.record(session, tid, [(blob, key)]) m["sent"] = m.get("sent", 0) + 1 row.size_bytes = (row.size_bytes or 0) + blob.size_bytes @@ -548,10 +636,15 @@ def _library_is_empty(session) -> bool: )).scalar() -def _note(session, row: BackupRun, m: dict) -> None: +def _note(session, row: BackupRun, m: dict, token: str | None = None) -> bool: + """Write the run's progress. Given a chunk's token, writes only while that + chunk still carries the run, and returns False when it doesn't.""" + if token is not None and not _still_ours(session, row.id, token): + return False row.manifest = dict(m) row.last_progress_at = datetime.now(UTC) session.commit() + return True @celery.task( @@ -561,19 +654,22 @@ def _note(session, row: BackupRun, m: dict) -> None: soft_time_limit=21600, time_limit=23400, ) def restore_offsite_task(self, *, dump_key: str | None = None, files_only: bool = False, - run_id: int | None = None) -> dict: + run_id: int | None = None, after: str | None = None, + resume: bool = False) -> dict: """Restore this install from the off-site bucket, one chunk at a time. Started without `run_id`, it replaces the database with the dump at `dump_key` (the newest when None), then fetches every library file the restored rows name. With `files_only`, the database stays and only the missing files are fetched, which also repairs a library that lost files. - Each chunk queues the next; the run's row counts files restored from the - bucket, copied from a duplicate already on disk, already present, missing - from the bucket, and failed. Returns {'backup_run_id': N, ...}.""" + Each chunk queues the next, and a restart resumes the file phase where it + stopped; the run's row counts files restored from the bucket, copied from + a duplicate already on disk, already present, missing from the bucket, + and failed. Returns {'backup_run_id': N, ...}.""" SessionLocal = _sync_session_factory() + token = _chunk_token(self) if run_id is not None: - return _restore_files(SessionLocal, run_id) + return _restore_files(SessionLocal, run_id, token, after=after, resume=resume) now = datetime.now(UTC) with SessionLocal() as session: @@ -601,7 +697,7 @@ def restore_offsite_task(self, *, dump_key: str | None = None, files_only: bool session.commit() if files_only: - return _restore_files(SessionLocal, run_id) + return _restore_files(SessionLocal, run_id, token, after=None, resume=False) run_id = _restore_database(SessionLocal, run_id, target, limit_mbps, dump_key) # The files follow as chunks of their own, on the worker. restore_offsite_task.delay(run_id=run_id) @@ -706,12 +802,14 @@ def _leave_empty_install(SessionLocal, cfg, target, limit_mbps, started_at, mani log.exception("restore: the database couldn't be reset after a failed load") -def _restore_files(SessionLocal, run_id: int) -> dict: +def _restore_files(SessionLocal, run_id: int, token: str, *, after: str | None, + resume: bool) -> dict: """One chunk of a restore's file phase; queues the next while files remain.""" with SessionLocal() as session: + why = _claim_run(session, run_id, "restoring", token, after=after, resume=resume) + if why: + return {"backup_run_id": run_id, "skipped": why} row = session.get(BackupRun, run_id) - if row is None or row.status != "restoring": - return {"backup_run_id": run_id, "skipped": "the restore is no longer going"} settings = ImportSettings.load_sync(session) limit_mbps = settings.backup_upload_limit_mbps try: @@ -722,7 +820,8 @@ def _restore_files(SessionLocal, run_id: int) -> dict: _mark_failed(session, row, exc) raise try: - more = _fetch_files_chunk(SessionLocal, run_id, target, backup_s3.Pacer(limit_mbps)) + more = _fetch_files_chunk(SessionLocal, run_id, target, backup_s3.Pacer(limit_mbps), + token) except (SoftTimeLimitExceeded, Exception) as exc: with SessionLocal() as session: row = session.get(BackupRun, run_id) @@ -730,8 +829,10 @@ def _restore_files(SessionLocal, run_id: int) -> dict: _mark_failed(session, row, exc) raise + if more is None: + return {"backup_run_id": run_id, "skipped": "another chunk took the run over"} if more: - restore_offsite_task.delay(run_id=run_id) + restore_offsite_task.delay(run_id=run_id, after=token) return {"backup_run_id": run_id, "continues": True} with SessionLocal() as session: row = session.get(BackupRun, run_id) @@ -751,9 +852,10 @@ def _restore_files(SessionLocal, run_id: int) -> dict: return {"backup_run_id": run_id, "continues": False} -def _fetch_files_chunk(SessionLocal, run_id: int, target, pacer) -> bool: +def _fetch_files_chunk(SessionLocal, run_id: int, target, pacer, token: str) -> bool | None: """Put library files back until the chunk's time is up or none are left. - Returns True when files remain for another chunk.""" + Returns True when files remain for another chunk, and None when another + chunk has taken the run over.""" tid = backup_media.target_id(target) deadline = time.monotonic() + RESTORE_CHUNK_SECONDS with SessionLocal() as session: @@ -768,13 +870,11 @@ def _fetch_files_chunk(SessionLocal, run_id: int, target, pacer) -> bool: session.commit() if not batch: m["cursor"] = cursor - _note(session, row, m) - return False + return False if _note(session, row, m, token) else None for f in batch: if time.monotonic() > deadline: m["cursor"] = cursor - _note(session, row, m) - return True + return True if _note(session, row, m, token) else None try: outcome = backup_restore.restore_file( target, f, keys.get(f.sha256), IMAGES_ROOT, pacer) @@ -789,5 +889,6 @@ def _fetch_files_chunk(SessionLocal, run_id: int, target, pacer) -> bool: cursor = f.sha256 m["cursor"] = cursor if outcome != "present" or time.monotonic() - noted > _RESTORE_NOTE_SECONDS: - _note(session, row, m) + if not _note(session, row, m, token): + return None noted = time.monotonic() diff --git a/tests/test_backup_media.py b/tests/test_backup_media.py index 940e60b..3949165 100644 --- a/tests/test_backup_media.py +++ b/tests/test_backup_media.py @@ -197,10 +197,72 @@ def test_a_run_out_of_time_queues_its_next_chunk(db_sync, tmp_path, run_env, mon result = tasks.backup_media_task(triggered_by="manual") assert result["continues"] is True - assert run_env == [{"run_id": result["backup_run_id"]}] row = db_sync.get(BackupRun, result["backup_run_id"]) db_sync.refresh(row) assert row.status == "running" and row.last_progress_at is not None + # The next chunk carries the token of the chunk that queued it. + assert run_env == [{"run_id": row.id, "after": row.manifest["chunk"]}] + + +def _going(db_sync, chunk): + row = BackupRun(kind="media", status="running", triggered_by="manual", + started_at=datetime.now(UTC), size_bytes=0, + manifest={"sent": 0, "failed": 0, "failures": [], "cursor": "", + "chunk": chunk}) + db_sync.add(row) + db_sync.commit() + return row + + +def test_only_the_chunk_queued_by_the_one_carrying_the_run_goes_on( + db_sync, tmp_path, run_env, monkeypatch, +): + from backend.app.tasks.backup import backup_media_task + _image(db_sync, tmp_path, "a") + row = _going(db_sync, "live") + monkeypatch.setattr(backup_media, "send", _send_all_but(None)) + + # A killed chunk's own message, delivered again, is dropped. + stale = backup_media_task(run_id=row.id, after="earlier") + assert "another chunk" in stale["skipped"] + assert backup_media_task(run_id=row.id, after="live")["continues"] is False + db_sync.refresh(row) + assert row.status == "ok" and row.manifest["sent"] == 1 + + +def test_a_resume_takes_the_run_and_the_chunk_it_replaced_stops( + db_sync, tmp_path, run_env, monkeypatch, +): + from backend.app.tasks import backup as tasks + for c in "ab": + _image(db_sync, tmp_path, c) + row = _going(db_sync, "killed") + + def send_then_lose_the_run(target, blob, images_root, pacer): + # While this chunk sends its first file, a newer one takes the run. + with tasks._sync_session_factory()() as s: + other = s.get(BackupRun, row.id) + other.manifest = {**other.manifest, "chunk": "newer"} + s.commit() + return backup_media.blob_key(blob.sha256, blob.path) + monkeypatch.setattr(backup_media, "send", send_then_lose_the_run) + + result = tasks.backup_media_task(run_id=row.id, resume=True) + assert "took the run over" in result["skipped"] + assert run_env == [] + db_sync.refresh(row) + assert row.status == "running" and row.manifest["chunk"] == "newer" + assert row.manifest["sent"] == 0 and row.manifest["cursor"] == "" + + +def test_a_restart_resumes_every_media_run_still_going(db_sync, run_env): + from backend.app.tasks.backup import resume_offsite_runs + going = _going(db_sync, "killed") + db_sync.add(BackupRun(kind="media", status="ok", triggered_by="manual", + started_at=datetime.now(UTC), manifest={})) + db_sync.commit() + assert resume_offsite_runs()["media"] == [going.id] + assert run_env == [{"run_id": going.id, "resume": True}] def test_a_second_run_waits_for_the_one_going(db_sync, run_env): diff --git a/tests/test_backup_restore.py b/tests/test_backup_restore.py index 255deef..6968aaf 100644 --- a/tests/test_backup_restore.py +++ b/tests/test_backup_restore.py @@ -188,10 +188,39 @@ def test_a_restore_out_of_time_queues_its_next_chunk(db_sync, tmp_path, run_env, result = tasks.restore_offsite_task(files_only=True) queued, _ = run_env assert result["continues"] is True - assert queued == [{"run_id": result["backup_run_id"]}] row = db_sync.get(BackupRun, result["backup_run_id"]) db_sync.refresh(row) assert row.status == "restoring" and row.last_progress_at is not None + # The next chunk carries the token of the chunk that queued it. + assert queued == [{"run_id": row.id, "after": row.manifest["chunk"]}] + + +def test_a_restart_resumes_the_file_phase_and_drops_the_stale_chunk( + db_sync, tmp_path, run_env, +): + from backend.app.tasks.backup import resume_offsite_runs, restore_offsite_task + _file(tmp_path, "a.jpg", b"aa") + _image(db_sync, tmp_path / "a.jpg", b"aa") + now = datetime.now(UTC) + files = BackupRun(kind="restore", status="restoring", triggered_by="restore", + started_at=now, manifest={"phase": "files", "cursor": "", + "failures": [], "chunk": "killed"}) + loading = BackupRun(kind="restore", status="restoring", triggered_by="restore", + started_at=now, manifest={"phase": "database"}) + db_sync.add_all([files, loading]) + db_sync.commit() + queued, _ = run_env + + # A restore stopped while loading its database isn't resumed. + resume_offsite_runs() + assert queued == [{"run_id": files.id, "resume": True}] + + # The killed chunk's own message, delivered again, is dropped. + assert "another chunk" in restore_offsite_task(run_id=files.id, after="earlier")["skipped"] + done = restore_offsite_task(run_id=files.id, resume=True) + assert done["continues"] is False + db_sync.refresh(files) + assert files.status == "restored" and files.manifest["present"] == 1 def test_a_restore_waits_for_a_backup_that_is_running(db_sync, run_env): diff --git a/tests/test_download_lane_boot.py b/tests/test_download_lane_boot.py index ff52b65..1ab489a 100644 --- a/tests/test_download_lane_boot.py +++ b/tests/test_download_lane_boot.py @@ -1,7 +1,9 @@ -"""#4433: only the process consuming `download` clears a restart's leftovers. +"""Each lane picks up only its own restart's leftovers: the process consuming +`download` clears orphaned walks and platform locks, and the one consuming +`maintenance_long` resumes off-site backups and restores. 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 +ml lane ran these 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 @@ -42,6 +44,10 @@ def calls(monkeypatch): "backend.app.services.platform_lock.release_all_platform_locks", lambda: seen.append("release") or 0, ) + monkeypatch.setattr( + "backend.app.tasks.backup.resume_offsite_runs", + lambda: seen.append("resume") or {}, + ) return seen @@ -50,12 +56,25 @@ def test_the_download_lane_clears_orphans_and_locks(calls): assert calls == ["interrupt", "commit", "release"] -@pytest.mark.parametrize("queues", [("maintenance", "scan"), ("ml",), ("maintenance_long",)]) -def test_other_lanes_leave_downloads_alone(calls, queues): +@pytest.mark.parametrize("queues", [("maintenance", "scan"), ("ml",)]) +def test_other_lanes_leave_downloads_and_backups_alone(calls, queues): celery_signals._on_worker_ready(sender=_consumer(*queues)) assert calls == [] +def test_the_long_maintenance_lane_resumes_offsite_runs_and_nothing_else(calls): + celery_signals._on_worker_ready(sender=_consumer("maintenance_long")) + assert calls == ["resume"] + + +def test_a_failed_resume_does_not_stop_the_worker_starting(monkeypatch, calls): + def boom(): + raise RuntimeError("db down") + + monkeypatch.setattr("backend.app.tasks.backup.resume_offsite_runs", boom) + celery_signals._on_worker_ready(sender=_consumer("maintenance_long")) # must not raise + + def test_unreadable_queues_do_nothing_rather_than_guess(calls): celery_signals._on_worker_ready(sender=SimpleNamespace()) assert calls == []