Add remote offload of completed backups to Hetzner Storage Box

Ships every completed backup off-box over WebDAV via rclone, verified by
exact byte-size match before local retention or schedule rotation ever
touches it. Fixes prune_orphans to skip confirmed-offloaded backups whose
local copy was already purged (it previously hard-deleted their DB row,
discarding the only pointer to the remote copy). Installs rclone in the
Docker image and gitignores the container's rclone.conf location, which
lives inside the bind-mounted repo root and holds live credentials.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01XjW4qocnaJxhugUi5ca8Wo

.gitignore

@@ -16,6 +16,10 @@ devplace-init.lock
 notification-private.pem
 notification-private.pkcs8.pem
 notification-public.pem
+# HOME=/app in the Docker app container, so an rclone.conf created via
+# `rclone config` or DEVPLACE_RCLONE_CONFIG's default lands here on the
+# bind-mounted host tree - it holds live remote-storage credentials.
+/.config/
 .pytest_cache/
 .ruff_cache/
 .opencode

CLAUDE.md

@@ -146,7 +146,7 @@ Nested `CLAUDE.md` files (loaded automatically by Claude Code only when a file i
 | `devplacepy/services/moderation/CLAUDE.md` | Trust and safety: the reportable-target registry, the content filter and its five choke points, the report queue and its atomic resolution, enforcement, consent, maturity, account deletion |
 | `devplacepy/services/acceptance/CLAUDE.md` | Acceptance convergence: the opt-in service that grants every policy agreement to every account that has not declined it, its invisibility contract and the ledger-as-decline-register rule |
 | `devplacepy/services/audit/CLAUDE.md` | Audit log: recorders, categories, retention |
-| `devplacepy/services/backup/CLAUDE.md` | Backup service: targets, worker, schedules, primary-admin-only download |
+| `devplacepy/services/backup/CLAUDE.md` | Backup service: targets, worker, schedules, primary-admin-only download, remote offload to Hetzner Storage Box |
 | `devplacepy/services/telegram/CLAUDE.md` | Telegram bot bridge |
 | `devplacepy/services/email/CLAUDE.md` | Devii IMAP/SMTP email tools |
 | `devplacepy/services/gitea/CLAUDE.md` | Issue tracker (Gitea-backed, no local issue store) |

Dockerfile

@@ -3,7 +3,7 @@ FROM python:3.13-slim
 WORKDIR /app
 
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    curl ca-certificates \
+    curl ca-certificates rclone \
     libglib2.0-0 libpango-1.0-0 libpangoft2-1.0-0 libharfbuzz0b libfontconfig1 \
     fonts-dejavu-core \
     && rm -rf /var/lib/apt/lists/*

devplacepy/config.py

@@ -22,6 +22,15 @@ ZIP_STAGING_DIR = DATA_DIR / "zip_staging"
 FORK_STAGING_DIR = DATA_DIR / "fork_staging"
 BACKUPS_DIR = DATA_DIR / "backups"
 BACKUP_STAGING_DIR = DATA_DIR / "backup_staging"
+
+RCLONE_BIN = environ.get("DEVPLACE_RCLONE_BIN", "rclone")
+RCLONE_CONFIG_FILE = environ.get(
+    "DEVPLACE_RCLONE_CONFIG",
+    str(Path(environ.get("HOME", "/root")) / ".config" / "rclone" / "rclone.conf"),
+)
+BACKUP_OFFLOAD_REMOTE = environ.get(
+    "DEVPLACE_BACKUP_OFFLOAD_REMOTE", "storagebox:devplacepy-backups"
+)
 SEO_REPORTS_DIR = DATA_DIR / "seo_reports"
 PLANNING_REPORTS_DIR = DATA_DIR / "planning_reports"
 DBAPI_DIR = DATA_DIR / "dbapi"

devplacepy/services/backup/CLAUDE.md

@@ -15,6 +15,20 @@ Admin-only, enterprise-grade backups built on the **same async-job pattern as zi
 - **Data model** (`store.py`, ensured in `database.init_db` via `backup_store.ensure_tables()`): `backups` (NOT soft-deletable - an archive is a reclaimable operational artifact, hard-deleted like zips) and `backup_schedules` (in `SOFT_DELETE_TABLES`, born-live `deleted_at:None`). `store` holds all CRUD plus `compute_storage_stats()` (du of every major data area + `shutil.disk_usage`, run in `asyncio.to_thread` from the route, 30s in-process TTL cache so the walk never blocks).
 - **Permanent artifact:** `cleanup(job)` only removes leftover staging, NEVER the archive. Job retention prunes the `jobs` row; the archive and `backups` row persist until an admin deletes it, a schedule rotates it out (`keep_last`), or `devplace backups clear`. Deleting a backup is a HARD delete (unlink file + delete row) - correct because backups are GC artifacts, the documented exception to the soft-delete rule.
 
+## Remote offload
+
+`devplacepy/services/backup/offload.py` ships completed archives to a Hetzner Storage Box over WebDAV via `rclone` (`config.RCLONE_BIN`/`config.RCLONE_CONFIG_FILE`, remote name `config.BACKUP_OFFLOAD_REMOTE`, default `storagebox:devplacepy-backups`) - deliberately **not** the `/backup` davfs2 mount, whose FUSE metadata cache lives on the root filesystem and breaks exactly when disk fills (the original outage cause). `BackupService._run_offload_cycle` (throttled to `backup_offload_interval_seconds`, default 300s, via `ConfigField`s in the `Offload` group) runs each cycle after `_fire_due_schedules`:
+
+1. `upload_pending` - every `done` backup with `remote_uploaded_at` unset and a live `local_path` is `rclone copyto`'d to `<remote>/<target>/<filename>`, then verified by exact byte-size match (`rclone size --json`) against `size_bytes` recorded at finalize time. Only on a verified match does `store.mark_remote_uploaded` set `remote_path`/`remote_uploaded_at`. A failed or unverified upload is silently retried next cycle - `remote_uploaded_at` is the only source of truth for "is this backup actually safe off-box."
+2. `enforce_local_retention` (`backup_offload_keep_local`, default 1) - per target, keeps the newest N **offloaded** local copies and unlinks the rest (`store.mark_local_purged`: clears `local_path`, sets `local_purged_at`, row and `remote_path` persist). A backup with no confirmed remote copy is never touched, no matter how old.
+3. `enforce_remote_retention` (`backup_offload_keep_remote`, default 30) - per target, `rclone lsjson` the remote dir and `deletefile` anything beyond the newest N, sorted by filename (safe because the `{target}-YYYYMMDD-HHMMSS-*` name is lexicographically chronological, same property `schedule.to_iso` relies on).
+
+**`rotate_schedule` is offload-aware:** it now skips any row with an empty `remote_uploaded_at` - a schedule's `keep_last` can never hard-delete a backup that was never confirmed off-box, even if offload is disabled entirely (rotation then simply stops happening, which is the safe failure direction).
+
+**`prune_orphans` (`devplace backups prune`) is offload-aware for the same reason:** a `done` row with a confirmed `remote_uploaded_at` is skipped even when its `local_path` is empty/missing - that is the normal steady state after `enforce_local_retention` purges the local copy, not an orphan. Only a `done` row with no confirmed remote copy AND a missing local file counts as truly orphaned and gets hard-deleted. Without this check, running the CLI prune after offload has done its job would delete the DB record for every successfully offloaded backup, discarding the only pointer to its `remote_path`.
+
+**Operational prerequisite (production, not automatic):** the `rclone` binary is installed in the shipped Docker image, but a working WebDAV remote still needs to exist at `config.RCLONE_CONFIG_FILE` (default `$HOME/.config/rclone/rclone.conf` inside the app container, overridable via `DEVPLACE_RCLONE_CONFIG`) with a remote named to match `config.BACKUP_OFFLOAD_REMOTE`'s prefix (default `storagebox`) pointing at the Hetzner Storage Box's WebDAV endpoint and credentials - `rclone config` (interactive) or a hand-written `rclone.conf` generates it. Until that file exists, every `upload_pending` attempt fails fast (`rclone` errors "didn't find section") and is logged and retried next cycle; local retention and rotation both stay disabled the whole time (see above), so backups simply accumulate locally with no data loss, they just never leave the box. **In Docker, `HOME=/app` (the bind-mounted repo root, `docker-compose.yml`), so the default config path resolves to `<repo>/.config/rclone/rclone.conf` on the host - `.gitignore` excludes `/.config/` precisely because this file holds live remote-storage credentials; never force-add it.**
+
 ## Schedules
 
 `backup_schedules` carry `kind` (`interval`|`cron`), `every_seconds`/`cron`, `enabled`, `keep_last`, `next_run_at`, run bookkeeping. `_fire_due_schedules` (lock-owner only, so each fires once) compares `next_run_at <= to_iso(now_utc())` and enqueues a `backup` job + a `backups` record, then advances `next_run_at` via `schedule.next_run`. **Timestamp format is load-bearing:** schedule `next_run_at` uses the devii `schedule.to_iso` format (`%Y-%m-%dT%H:%M:%S`, no tz/micros) on BOTH sides of the comparison so lexicographic compare equals chronological - do not mix it with `datetime.isoformat()`.

devplacepy/services/backup/offload.py

@@ -0,0 +1,112 @@
+# retoor <retoor@molodetz.nl>
+
+import asyncio
+import json
+from pathlib import Path
+
+from devplacepy import config
+from devplacepy.services.backup import store
+
+
+async def _run_rclone(*args: str, timeout: float = 1800.0) -> tuple[int, str, str]:
+    proc = await asyncio.create_subprocess_exec(
+        config.RCLONE_BIN,
+        "--config",
+        config.RCLONE_CONFIG_FILE,
+        *args,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    try:
+        out, err = await asyncio.wait_for(proc.communicate(), timeout=timeout)
+    except asyncio.TimeoutError:
+        proc.kill()
+        await proc.wait()
+        return 124, "", "rclone timed out"
+    return proc.returncode, out.decode("utf-8", "replace"), err.decode("utf-8", "replace")
+
+
+def _remote_dir(target: str) -> str:
+    return f"{config.BACKUP_OFFLOAD_REMOTE}/{target}"
+
+
+async def upload_pending(log=lambda message: None) -> int:
+    uploaded = 0
+    for row in store.list_pending_offload():
+        local_path = Path(row["local_path"])
+        remote_path = f"{_remote_dir(row['target'])}/{row['filename']}"
+        code, _, err = await _run_rclone("copyto", str(local_path), remote_path)
+        if code != 0:
+            log(f"Offload failed for {row['filename']}: {err.strip()[:300]}")
+            continue
+        size_code, size_out, size_err = await _run_rclone("size", remote_path, "--json")
+        if size_code != 0:
+            log(f"Offload verify failed for {row['filename']}: {size_err.strip()[:300]}")
+            continue
+        try:
+            remote_bytes = json.loads(size_out).get("bytes", -1)
+        except (ValueError, TypeError):
+            remote_bytes = -1
+        expected_bytes = int(row.get("size_bytes") or 0)
+        if remote_bytes != expected_bytes:
+            log(
+                f"Offload size mismatch for {row['filename']} "
+                f"(local {expected_bytes}, remote {remote_bytes}), will retry"
+            )
+            continue
+        store.mark_remote_uploaded(row["uid"], remote_path)
+        uploaded += 1
+        log(f"Offloaded {row['filename']} ({store.human_bytes(expected_bytes)}) to {remote_path}")
+    return uploaded
+
+
+async def enforce_local_retention(keep_local: int, log=lambda message: None) -> int:
+    keep_local = max(1, keep_local)
+    purged = 0
+    for target in store.BACKUP_TARGETS:
+        rows = store.list_offloaded_local_by_target(target)
+        for row in rows[keep_local:]:
+            store.mark_local_purged(row["uid"])
+            purged += 1
+            log(f"Purged local copy of {row['filename']} (kept remotely at {row['remote_path']})")
+    return purged
+
+
+async def enforce_remote_retention(keep_remote: int, log=lambda message: None) -> int:
+    if keep_remote < 1:
+        return 0
+    removed = 0
+    for target in store.BACKUP_TARGETS:
+        remote_dir = _remote_dir(target)
+        code, out, _ = await _run_rclone("lsjson", remote_dir)
+        if code != 0:
+            continue
+        try:
+            entries = json.loads(out)
+        except (ValueError, TypeError):
+            continue
+        files = sorted(
+            (entry for entry in entries if not entry.get("IsDir")),
+            key=lambda entry: entry.get("Name", ""),
+            reverse=True,
+        )
+        for entry in files[keep_remote:]:
+            remote_path = f"{remote_dir}/{entry['Name']}"
+            del_code, _, del_err = await _run_rclone("deletefile", remote_path)
+            if del_code == 0:
+                removed += 1
+                log(f"Removed old remote backup {remote_path}")
+            else:
+                log(f"Could not remove remote backup {remote_path}: {del_err.strip()[:200]}")
+    return removed
+
+
+async def run_cycle(*, keep_local: int, keep_remote: int, log=lambda message: None) -> dict:
+    uploaded = await upload_pending(log=log)
+    purged_local = await enforce_local_retention(keep_local, log=log)
+    removed_remote = await enforce_remote_retention(keep_remote, log=log)
+    return {
+        "uploaded": uploaded,
+        "purged_local": purged_local,
+        "removed_remote": removed_remote,
+    }

devplacepy/services/backup/service.py

@@ -6,12 +6,14 @@ import logging
 import shutil
 import sqlite3
 import sys
+import time
 from datetime import datetime, timezone
 from pathlib import Path
 
 from devplacepy import config
 from devplacepy.attachments import _directory_for
-from devplacepy.services.backup import store
+from devplacepy.services.backup import offload, store
+from devplacepy.services.base import ConfigField
 from devplacepy.services.devii.tasks.schedule import next_run as schedule_next_run
 from devplacepy.services.devii.tasks.schedule import now_utc, to_iso
 from devplacepy.services.jobs import queue
@@ -22,6 +24,10 @@ logger = logging.getLogger(__name__)
 
 WORKER_MODULE = "devplacepy.services.jobs.backup_worker"
 
+DEFAULT_OFFLOAD_INTERVAL_SECONDS = 300
+DEFAULT_OFFLOAD_KEEP_LOCAL = 1
+DEFAULT_OFFLOAD_KEEP_REMOTE = 30
+
 
 class BackupService(JobService):
     kind = "backup"
@@ -34,6 +40,53 @@ class BackupService(JobService):
 
     def __init__(self):
         super().__init__(name="backup", interval_seconds=15)
+        self._last_offload_at = 0.0
+        self.offload_enabled_field = ConfigField(
+            "backup_offload_enabled",
+            "Offload to remote storage",
+            type="bool",
+            default=True,
+            help=(
+                "Ship completed backups to the configured remote storage and purge "
+                "older local copies once a remote copy is confirmed."
+            ),
+            group="Offload",
+        )
+        self.offload_interval_field = ConfigField(
+            "backup_offload_interval_seconds",
+            "Offload check interval (seconds)",
+            type="int",
+            default=DEFAULT_OFFLOAD_INTERVAL_SECONDS,
+            minimum=60,
+            help="How often to check for backups to offload and rotate.",
+            group="Offload",
+        )
+        self.offload_keep_local_field = ConfigField(
+            "backup_offload_keep_local",
+            "Local copies to keep per target",
+            type="int",
+            default=DEFAULT_OFFLOAD_KEEP_LOCAL,
+            minimum=1,
+            maximum=10,
+            help="Local archives beyond this count are purged per target, once offloaded.",
+            group="Offload",
+        )
+        self.offload_keep_remote_field = ConfigField(
+            "backup_offload_keep_remote",
+            "Remote copies to keep per target",
+            type="int",
+            default=DEFAULT_OFFLOAD_KEEP_REMOTE,
+            minimum=1,
+            maximum=1000,
+            help="Remote archives beyond this count are deleted per target.",
+            group="Offload",
+        )
+        self.config_fields += [
+            self.offload_enabled_field,
+            self.offload_interval_field,
+            self.offload_keep_local_field,
+            self.offload_keep_remote_field,
+        ]
 
     async def run_once(self) -> None:
         await super().run_once()
@@ -41,6 +94,30 @@ class BackupService(JobService):
             self._fire_due_schedules()
         except Exception as exc:
             self.log(f"Schedule pass failed: {exc}")
+        await self._run_offload_cycle()
+
+    async def _run_offload_cycle(self) -> None:
+        if not self.offload_enabled_field.read():
+            return
+        interval = max(60, int(self.offload_interval_field.read()))
+        now = time.monotonic()
+        if now - self._last_offload_at < interval:
+            return
+        self._last_offload_at = now
+        try:
+            result = await offload.run_cycle(
+                keep_local=int(self.offload_keep_local_field.read()),
+                keep_remote=int(self.offload_keep_remote_field.read()),
+                log=self.log,
+            )
+            if result["uploaded"] or result["purged_local"] or result["removed_remote"]:
+                self.log(
+                    f"Offload cycle: {result['uploaded']} uploaded, "
+                    f"{result['purged_local']} local purged, "
+                    f"{result['removed_remote']} remote pruned"
+                )
+        except Exception as exc:
+            self.log(f"Offload cycle failed: {exc}")
 
     async def process(self, job: dict) -> dict:
         from devplacepy.services.audit import record as audit

devplacepy/services/backup/store.py

@@ -79,6 +79,9 @@ def ensure_tables() -> None:
         ("created_at", ""),
         ("completed_at", ""),
         ("error", ""),
+        ("remote_path", ""),
+        ("remote_uploaded_at", ""),
+        ("local_purged_at", ""),
     ):
         if not backups.has_column(column):
             backups.create_column_by_example(column, example)
@@ -216,16 +219,67 @@ def rotate_schedule(schedule_uid: str, keep_last: int) -> int:
     )
     removed = 0
     for row in rows[keep_last:]:
+        if not row.get("remote_uploaded_at"):
+            continue
         delete_backup(row["uid"])
         removed += 1
     return removed
 
 
+def mark_remote_uploaded(uid: str, remote_path: str) -> None:
+    get_table("backups").update(
+        {"uid": uid, "remote_path": remote_path, "remote_uploaded_at": now_iso()},
+        ["uid"],
+    )
+
+
+def mark_local_purged(uid: str) -> None:
+    row = get_backup(uid)
+    if row:
+        _unlink_archive(row)
+    get_table("backups").update(
+        {"uid": uid, "local_path": "", "local_purged_at": now_iso()},
+        ["uid"],
+    )
+
+
+def list_pending_offload(limit: int = 50) -> list[dict]:
+    if "backups" not in db.tables:
+        return []
+    rows = get_table("backups").find(
+        status=STATUS_DONE, order_by=["created_at"], _limit=limit
+    )
+    return [
+        row
+        for row in rows
+        if not row.get("remote_uploaded_at")
+        and row.get("local_path")
+        and Path(row["local_path"]).is_file()
+    ]
+
+
+def list_offloaded_local_by_target(target: str) -> list[dict]:
+    if "backups" not in db.tables:
+        return []
+    rows = get_table("backups").find(
+        target=target, status=STATUS_DONE, order_by=["-created_at"]
+    )
+    return [
+        row
+        for row in rows
+        if row.get("remote_uploaded_at")
+        and row.get("local_path")
+        and Path(row["local_path"]).is_file()
+    ]
+
+
 def prune_orphans() -> int:
     removed = 0
     for row in list_backups(limit=100000):
         if row.get("status") != STATUS_DONE:
             continue
+        if row.get("remote_uploaded_at"):
+            continue
         local_path = row.get("local_path") or ""
         if not local_path or not Path(local_path).is_file():
             get_table("backups").delete(uid=row["uid"])

devplacepy/templates/docs/backups.html

@@ -61,6 +61,47 @@ Schedules are evaluated by the backup service, which runs only on the worker hol
 lock, so each scheduled backup fires exactly once. Create, edit, enable, disable, run, and delete
 schedules from the dashboard.
 
+## Remote offload
+
+Every completed backup is shipped off-box to a Hetzner Storage Box over WebDAV (via `rclone`,
+independently of the `/backup` mount used for manual browsing) so a full local disk can never take
+the only copy of a backup with it. This runs as part of the backup service's own cycle, throttled to
+an interval separate from schedule triggers, and is configured from **Admin -> Services -> Backup**
+under the **Offload** group:
+
+- **Offload to remote storage** - master on/off switch. When off, nothing is uploaded and local
+  copies are never purged (see below).
+- **Offload check interval** - how often the service looks for newly completed backups to upload and
+  rotates old copies, in seconds.
+- **Local copies to keep per target** - across the whole target (database, uploads, keys, full,
+  regardless of which schedule produced them), local archive files beyond this count are removed once
+  a remote copy is confirmed. Default 1: the newest local archive per type stays on disk, everything
+  older lives only on the Storage Box.
+- **Remote copies to keep per target** - archives on the Storage Box beyond this count are deleted,
+  oldest first, per target.
+
+A backup is only ever considered "offloaded" after its upload is verified by an exact byte-size match
+against the size recorded when the archive was built - a failed or unverified upload is retried on
+the next cycle and never counted as safe. **Local retention only ever touches archives with a
+confirmed remote copy.** This also changes what schedule `keep_last` rotation does: it now skips any
+backup that has not yet been confirmed offloaded, so an aggressive `keep_last` can no longer delete
+the only surviving copy of a backup before it ever reached remote storage - if offload is turned off,
+schedule rotation simply stops deleting anything, which is the safe direction to fail in.
+
+Purging a local copy is different from deleting a backup: the archive's `backups` row and its
+`remote_uploaded_at` / `remote_path` persist, only the local file is unlinked. The dashboard and
+`devplace backups list` still show the backup; only its local download would need to come from the
+Storage Box copy first. A full delete (admin action, `keep_last` rotation, or `devplace backups
+clear`) still removes the record entirely and does not currently remove the matching remote copy -
+treat the remote side as the durable archive and the admin delete action as "no longer needed
+locally," not as "erase everywhere."
+
+Offload requires a one-time server-side setup step outside this dashboard: an `rclone` remote
+configured on the host (or inside the app container) pointing at the Hetzner Storage Box's WebDAV
+endpoint, with credentials. Until that remote exists, uploads fail every cycle (logged, retried
+automatically) and archives simply accumulate locally with no data loss - they just never leave the
+box, and local retention stays inactive the whole time since it never touches an unconfirmed backup.
+
 ## Storage visibility
 
 The dashboard reports the size and file count of every major data area (database, uploads,

tests/unit/services/backup/store.py

@@ -0,0 +1,67 @@
+# retoor <retoor@molodetz.nl>
+
+import tempfile
+from pathlib import Path
+
+from devplacepy.services.backup import store
+
+
+def _make_backup(target, *, offloaded, missing_local=False, schedule_uid=""):
+    job_uid = f"job-{store.generate_uid()}"
+    uid = store.create_backup(
+        target=target, created_by="test", job_uid=job_uid, schedule_uid=schedule_uid
+    )
+    path = Path(tempfile.gettempdir()) / f"backup-store-test-{uid}.tar.gz"
+    if not missing_local:
+        path.write_bytes(b"x" * 10)
+    store.finalize_backup(
+        uid,
+        filename=path.name,
+        local_path=str(path),
+        stats={"bytes_out": 10, "bytes_in": 10, "file_count": 1, "dir_count": 0, "sha256": "abc"},
+    )
+    if offloaded:
+        store.mark_remote_uploaded(uid, f"remote:{target}/{path.name}")
+    return uid, path
+
+
+def test_prune_orphans_spares_offloaded_backups_with_purged_local_copies(local_db):
+    offloaded_uid, _ = _make_backup("database", offloaded=True)
+    store.mark_local_purged(offloaded_uid)
+    orphan_uid, _ = _make_backup("database", offloaded=False, missing_local=True)
+
+    try:
+        removed = store.prune_orphans()
+
+        remaining = {row["uid"] for row in store.list_backups(limit=1000)}
+        assert offloaded_uid in remaining
+        assert orphan_uid not in remaining
+        assert removed >= 1
+    finally:
+        store.delete_backup(offloaded_uid)
+
+
+def test_rotate_schedule_never_removes_a_backup_without_a_confirmed_remote_copy(local_db):
+    schedule_uid = store.create_schedule(
+        name="test-rotate",
+        target="database",
+        kind="interval",
+        every_seconds=86400,
+        cron="",
+        keep_last=1,
+        created_by="test",
+        next_run_at="2026-01-01T00:00:00",
+    )
+    older_uid, _ = _make_backup("database", offloaded=False, schedule_uid=schedule_uid)
+    newer_uid, _ = _make_backup("database", offloaded=True, schedule_uid=schedule_uid)
+
+    try:
+        removed = store.rotate_schedule(schedule_uid, keep_last=1)
+
+        remaining = {row["uid"] for row in store.list_backups(limit=1000)}
+        assert removed == 0
+        assert older_uid in remaining
+        assert newer_uid in remaining
+    finally:
+        store.delete_backup(older_uid)
+        store.delete_backup(newer_uid)