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 <noreply@anthropic.com>

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

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

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 == []