From dee977f5a0cd8386685169ac7d8bdad686cf3965 Mon Sep 17 00:00:00 2001 From: Muse Sweep Date: Mon, 26 Oct 2026 12:39:08 +0000 Subject: [PATCH 1/5] feat(cold-store-reaper): prune-backups, get-backup, backup-d1 - prune-backups [--db] [--apply]: retention 30 daily / 12 weekly / 12 monthly as a union of newest-per-period over filename timestamps (pure function of the folder listing); dry-run by default; --apply trashes via files.update {trashed: true} (never deletes) and writes one transfer_ledger row per file (direction 'trash', kind 'backup-prune'). Unparseable names are never pruned. - get-backup : exact-name lookup + full download, reports size + sha256. - backup-d1: deterministic SQL export of the D1 cold-store-manifest database via the REST query API (shared _d1_endpoint/_d1_select helpers extracted from sync-manifest), filed through put-backup. - schema.sql: transfer_ledger CHECK gains the 'trash' direction; existing manifests are migrated in db() by rebuilding the table from the current definition. The D1 copy of the schema needs the same one-line change before prune rows can sync. --- pkgs/cold-store-reaper/cold_store.py | 376 +++++++++++++++++++++++++-- pkgs/cold-store-reaper/schema.sql | 5 +- 2 files changed, 360 insertions(+), 21 deletions(-) diff --git a/pkgs/cold-store-reaper/cold_store.py b/pkgs/cold-store-reaper/cold_store.py index 592c97fda..07da6a482 100755 --- a/pkgs/cold-store-reaper/cold_store.py +++ b/pkgs/cold-store-reaper/cold_store.py @@ -21,6 +21,7 @@ import json import mimetypes import os +import re import sqlite3 import subprocess import sys @@ -236,8 +237,36 @@ def db(self) -> sqlite3.Connection: if not self.manifest_path.exists() or self._schema_missing(conn): conn.executescript(SCHEMA_PATH.read_text()) conn.commit() + self._migrate_ledger_trash(conn) return conn + @staticmethod + def _migrate_ledger_trash(conn) -> None: + """Let pre-v3 manifests record prune rows: rebuild transfer_ledger + with the current schema.sql definition if its CHECK predates the + 'trash' direction. No-op for fresh or already-migrated databases.""" + row = conn.execute( + "SELECT sql FROM sqlite_master WHERE type='table' AND name='transfer_ledger'" + ).fetchone() + if row is None or "trash" in (row["sql"] or ""): + return + text = SCHEMA_PATH.read_text() + create = re.search( + r"CREATE TABLE IF NOT EXISTS transfer_ledger \(.*?\);", text, re.S + ) + index = re.search(r"CREATE INDEX IF NOT EXISTS idx_ledger_ts[^;]*;", text) + if create is None or index is None: + raise ColdStoreError("schema.sql is missing the transfer_ledger definition") + conn.execute("ALTER TABLE transfer_ledger RENAME TO transfer_ledger_pre_trash") + conn.execute(create.group(0)) + conn.execute( + "INSERT INTO transfer_ledger (id, ts, direction, bytes, kind, ref)" + " SELECT id, ts, direction, bytes, kind, ref FROM transfer_ledger_pre_trash" + ) + conn.execute("DROP TABLE transfer_ledger_pre_trash") + conn.execute(index.group(0)) + conn.commit() + @staticmethod def _schema_missing(conn) -> bool: row = conn.execute( @@ -674,8 +703,8 @@ def put_backup(self, db_name: str, file: str) -> dict: The Drive name is dated (--) so a folder listing is already a chronological catalogue. Retention - enforcement (30 daily / 12 weekly / 12 monthly, trash-not-delete) is - NOT implemented yet — see README. + (30 daily / 12 weekly / 12 monthly, trash-not-delete) is enforced + separately by `prune-backups`. """ if db_name not in BACKUP_DBS: raise ColdStoreError( @@ -740,7 +769,220 @@ def list_backups(self, db_name: str | None = None) -> list[dict]: out.sort(key=lambda r: r["created_time"], reverse=True) return out + # ------------------------------------------------- backup retention + @staticmethod + def _parse_backup_stamp(db_name: str, name: str) -> datetime | None: + """Timestamp embedded in a dated backup name, or None if the name + does not follow -- for this db.""" + m = re.fullmatch( + re.escape(db_name) + r"-(\d{8}T\d{6}Z)-.+", name + ) + if not m: + return None + try: + return datetime.strptime(m.group(1), "%Y%m%dT%H%M%SZ").replace( + tzinfo=timezone.utc + ) + except ValueError: + return None + + @staticmethod + def _retention_plan(files: list[dict]) -> tuple[list[dict], list[dict], list[dict]]: + """Split one db's backup listing into (keep, prune, unparseable). + + `files` are list_backups rows for a single db, each with a parsed + `stamp` (datetime) or None. The keep set is the union of: the + newest backup of each of the last 30 distinct calendar days + present, the newest of each of the last 12 distinct ISO weeks + present, and the newest of each of the last 12 distinct calendar + months present — "present" and "newest" are read off the file set + itself (filename timestamps, newest-first), never the wall clock, + so the plan is a pure function of the folder contents. Kept rows + gain a `reasons` list (daily/weekly/monthly); pruned rows are the + parseable remainder (reason: excess); unparseable names are never + pruned, only reported. + """ + parseable = [f for f in files if f["stamp"] is not None] + unparseable = [f for f in files if f["stamp"] is None] + parseable.sort(key=lambda f: (f["stamp"], f["name"]), reverse=True) + + def newest_per(keyfn, limit): + seen: dict = {} + for f in parseable: + key = keyfn(f["stamp"]) + if key not in seen and len(seen) < limit: + seen[key] = f + return seen.values() + + reasons: dict[str, set] = {} + + def mark(kept, reason): + for f in kept: + reasons.setdefault(f["id"], set()).add(reason) + + mark(newest_per(lambda ts: ts.date(), 30), "daily") + mark(newest_per(lambda ts: ts.isocalendar()[:2], 12), "weekly") + mark(newest_per(lambda ts: (ts.year, ts.month), 12), "monthly") + keep, prune = [], [] + for f in parseable: + if f["id"] in reasons: + keep.append({**f, "reasons": sorted(reasons[f["id"]])}) + else: + prune.append({**f, "reasons": ["excess"]}) + return keep, prune, unparseable + + def prune_backups( + self, db_name: str | None = None, apply: bool = False + ) -> list[dict]: + """Enforce backup retention (DESIGN section 5): 30 daily / + 12 weekly / 12 monthly, trash-not-delete. + + Dry-run by default: compute and return the plan without touching + Drive. With apply=True, each prune candidate is trashed via + files.update {trashed: true} (never permanently deleted) and one + transfer_ledger row per file is written (direction 'trash', + kind 'backup-prune', ref = filename). + """ + if db_name is not None and db_name not in BACKUP_DBS: + raise ColdStoreError( + f"unknown backup db {db_name!r} — expected one of {', '.join(BACKUP_DBS)}" + ) + dbs = [db_name] if db_name else list(BACKUP_DBS) + plans = [] + for db in dbs: + rows = self.list_backups(db) + for r in rows: + r["stamp"] = self._parse_backup_stamp(db, r["name"]) + keep, prune, unparseable = self._retention_plan(rows) + plans.append( + { + "db": db, + "keep": keep, + "prune": prune, + "unparseable": unparseable, + "applied": apply, + } + ) + if apply: + conn = self.db() + with conn: + for plan in plans: + for f in plan["prune"]: + status, _h, _resp = http_request( + "PATCH", + self.drive_base + f"/files/{f['id']}", + headers=self._auth_headers( + {"Content-Type": "application/json"} + ), + data=json.dumps({"trashed": True}).encode(), + ) + if status != 200: + raise ColdStoreError( + f"trash failed for backup {f['name']} (HTTP {status})" + ) + self.ledger_add( + conn, "trash", f["size"], "backup-prune", f["name"] + ) + conn.close() + for plan in plans: + for f in plan["keep"] + plan["prune"]: + del f["stamp"] + for f in plan["unparseable"]: + del f["stamp"] + return plans + + def get_backup(self, db_name: str, name: str, out_path: str) -> dict: + """Download one backup file, in full, by its exact Drive name.""" + if db_name not in BACKUP_DBS: + raise ColdStoreError( + f"unknown backup db {db_name!r} — expected one of {', '.join(BACKUP_DBS)}" + ) + if "'" in name: + raise ColdStoreError("backup name must not contain a single quote") + folder_id = self._folder_id(f"db-backups/{db_name}") + q = f"'{folder_id}' in parents and name = '{name}' and trashed = false" + url = ( + self.drive_base + + "/files?" + + urllib.parse.urlencode({"q": q, "fields": "files(id,name,size)"}) + ) + status, _h, resp = http_request("GET", url, headers=self._auth_headers()) + if status != 200: + raise ColdStoreError(f"files.list for get-backup failed (HTTP {status})") + files = json.loads(resp).get("files", []) + if not files: + raise ColdStoreError(f"backup not found: {db_name}/{name}") + file_id = files[0]["id"] + status, _h, resp = http_request( + "GET", + self.drive_base + f"/files/{file_id}?alt=media", + headers=self._auth_headers(), + ) + if status != 200: + raise ColdStoreError(f"backup download failed (HTTP {status})") + Path(out_path).write_bytes(resp) + digest = hashlib.sha256(resp).hexdigest() + return { + "db": db_name, + "name": name, + "drive_file_id": file_id, + "out": out_path, + "bytes": len(resp), + "sha256": digest, + } + # --------------------------------------------------------- manifest sync + def _d1_endpoint(self) -> tuple[str, str]: + """Resolve (bearer token, query URL) for the D1 manifest database. + + The token comes from the environment only; account/database IDs + fall back env -> state.json -> the cold-store defaults. + """ + token = os.environ.get("CLOUDFLARE_API_TOKEN") + if not token: + raise ColdStoreError( + "CLOUDFLARE_API_TOKEN is not set — materialize it from " + "Bitwarden into the environment first (it is never stored)" + ) + state = self.load_state() + account_id = ( + os.environ.get("COLD_STORE_CF_ACCOUNT_ID") + or state.get("cf_account_id") + or DEFAULT_CF_ACCOUNT_ID + ) + database_id = ( + os.environ.get("COLD_STORE_CF_D1_DATABASE_ID") + or state.get("cf_d1_database_id") + or DEFAULT_CF_D1_DATABASE_ID + ) + url = f"{self.d1_base}/accounts/{account_id}/d1/database/{database_id}/query" + return token, url + + def _d1_select(self, sql: str) -> list[dict]: + """Run one read query against D1 and return its result rows.""" + token, url = self._d1_endpoint() + status, _h, resp = http_request( + "POST", + url, + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": "application/json", + }, + data=json.dumps({"sql": sql}).encode(), + ) + if status != 200: + raise ColdStoreError( + f"D1 query failed (HTTP {status}): {d1_error_message(resp)}" + ) + payload = json.loads(resp) if resp else {} + if payload and not payload.get("success", True): + errors = payload.get("errors") or [{}] + raise ColdStoreError( + f"D1 query rejected: {errors[0].get('message', 'unknown error')}" + ) + results = (payload.get("result") or [{}])[0].get("results") or [] + return results + @staticmethod def _table_shape(conn, table: str) -> tuple[list[str], list[str]]: rows = conn.execute(f"PRAGMA table_info({table})").fetchall() @@ -814,24 +1056,7 @@ def sync_manifest(self, dry_run: bool = False) -> dict: } if dry_run: return total - token = os.environ.get("CLOUDFLARE_API_TOKEN") - if not token: - raise ColdStoreError( - "CLOUDFLARE_API_TOKEN is not set — materialize it from " - "Bitwarden into the environment first (it is never stored)" - ) - state = self.load_state() - account_id = ( - os.environ.get("COLD_STORE_CF_ACCOUNT_ID") - or state.get("cf_account_id") - or DEFAULT_CF_ACCOUNT_ID - ) - database_id = ( - os.environ.get("COLD_STORE_CF_D1_DATABASE_ID") - or state.get("cf_d1_database_id") - or DEFAULT_CF_D1_DATABASE_ID - ) - url = f"{self.d1_base}/accounts/{account_id}/d1/database/{database_id}/query" + token, url = self._d1_endpoint() for stmt in statements: status, _h, resp = http_request( "POST", @@ -859,6 +1084,68 @@ def sync_manifest(self, dry_run: bool = False) -> dict: ) return total + # ------------------------------------------------------------- backup-d1 + @staticmethod + def _sql_literal(value) -> str: + """Render one D1 JSON value as a deterministic SQL literal: + NULL bare, numbers bare, strings single-quoted with '' escaping.""" + if value is None: + return "NULL" + if isinstance(value, bool): + return "1" if value else "0" + if isinstance(value, int): + return str(value) + if isinstance(value, float): + return repr(value) + return "'" + str(value).replace("'", "''") + "'" + + def render_d1_export(self) -> tuple[str, list[str], int]: + """Export the whole D1 manifest database as deterministic SQL text. + + User tables are enumerated via sqlite_master (sqlite_% and _cf_% + excluded), columns via pragma_table_info, rows in rowid order. + Returns (sql_text, table_names, total_rows). + """ + tables = [ + r["name"] + for r in self._d1_select( + "SELECT name FROM sqlite_master WHERE type='table'" + " AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '_cf_%'" + " ORDER BY name" + ) + ] + lines = [ + "-- cold-store-manifest D1 export (via REST query API)", + f"-- tables: {', '.join(tables)}", + ] + total = 0 + for t in tables: + cols = [ + r["name"] + for r in self._d1_select(f"SELECT name FROM pragma_table_info('{t}')") + ] + rows = self._d1_select(f'SELECT * FROM "{t}" ORDER BY rowid') + for row in rows: + vals = ", ".join(self._sql_literal(row[c]) for c in cols) + lines.append( + f'INSERT INTO "{t}" ({", ".join(cols)}) VALUES ({vals});' + ) + total += len(rows) + return "\n".join(lines) + "\n", tables, total + + def backup_d1(self) -> dict: + """Export the D1 cold-store-manifest database and file it as a + d1 backup (DESIGN section 5), via the put-backup machinery.""" + text, tables, total = self.render_d1_export() + staging = self.home / "staging" + staging.mkdir(parents=True, exist_ok=True) + path = staging / "cold-store-manifest-export.sql" + path.write_text(text) + result = self.put_backup("d1", str(path)) + result["tables"] = tables + result["rows"] = total + return result + # ---------------------------------------------------------------- status def status_data(self) -> dict: conn = self.db() @@ -970,6 +1257,26 @@ def main(argv=None) -> int: action="store_true", help="print statement counts only; contact nothing", ) + p_prune = sub.add_parser( + "prune-backups", + help="enforce backup retention (30 daily / 12 weekly / 12 monthly, trash-not-delete); dry-run unless --apply", + ) + p_prune.add_argument("--db", choices=BACKUP_DBS, default=None) + p_prune.add_argument( + "--apply", + action="store_true", + help="actually trash the prune candidates (default: print the plan only)", + ) + p_get_backup = sub.add_parser( + "get-backup", help="download one backup file in full by its exact name" + ) + p_get_backup.add_argument("db", choices=BACKUP_DBS) + p_get_backup.add_argument("name") + p_get_backup.add_argument("out_path") + sub.add_parser( + "backup-d1", + help="export the D1 cold-store-manifest database and file it as a d1 backup", + ) args = parser.parse_args(argv) store = build_store(args) try: @@ -1055,6 +1362,35 @@ def main(argv=None) -> int: f"total: {verb} {result['statements']} statement(s), " f"{result['rows']} row(s)" ) + elif args.cmd == "prune-backups": + plans = store.prune_backups(args.db, apply=args.apply) + for plan in plans: + mode = "applied" if plan["applied"] else "dry-run" + print( + f"{plan['db']}: keep {len(plan['keep'])}, " + f"prune {len(plan['prune'])}, " + f"skipped-unparseable {len(plan['unparseable'])} ({mode})" + ) + for f in plan["keep"]: + print(f" keep {f['name']} [{', '.join(f['reasons'])}]") + for f in plan["prune"]: + verb = "trashed" if plan["applied"] else "prune " + print(f" {verb} {f['name']} [excess]") + for f in plan["unparseable"]: + print(f" skip {f['name']} [unparseable]") + elif args.cmd == "get-backup": + result = store.get_backup(args.db, args.name, args.out_path) + print( + f"backup {result['db']}/{result['name']} -> {result['out']}: " + f"{_human(result['bytes'])}, sha256 {result['sha256']}" + ) + elif args.cmd == "backup-d1": + result = store.backup_d1() + print( + f"backup {result['name']}: {_human(result['bytes'])}, " + f"uploaded to Drive as {result['drive_file_id']} " + f"({result['rows']} row(s) across {len(result['tables'])} table(s))" + ) except ColdStoreError as e: print(f"error: {e}", file=sys.stderr) return 1 diff --git a/pkgs/cold-store-reaper/schema.sql b/pkgs/cold-store-reaper/schema.sql index 33a587f64..9d657731e 100644 --- a/pkgs/cold-store-reaper/schema.sql +++ b/pkgs/cold-store-reaper/schema.sql @@ -38,10 +38,13 @@ CREATE TABLE IF NOT EXISTS members ( CREATE INDEX IF NOT EXISTS idx_members_hash ON members(object_hash); -- Reaper/rehydrator budget ledger: one row per action, 24h rolling window computed from ts. +-- v1.1 (reaper v3): direction gains 'trash' for backup-prune rows (kind='backup-prune'). +-- The D1 copy of this schema needs the same one-line CHECK change before prune +-- rows can sync; sync-manifest surfaces D1's own error text until then. CREATE TABLE IF NOT EXISTS transfer_ledger ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL DEFAULT (datetime('now')), - direction TEXT NOT NULL CHECK (direction IN ('upload','download')), + direction TEXT NOT NULL CHECK (direction IN ('upload','download','trash')), bytes INTEGER NOT NULL, kind TEXT, -- bundle | db-backup | manifest | rehydrate | compact ref TEXT -- bundle_id or file name From 35263104520bb6109bfc3a9dd219a3477528bd88 Mon Sep 17 00:00:00 2001 From: Muse Sweep Date: Mon, 26 Oct 2026 12:44:49 +0000 Subject: [PATCH 2/5] test(cold-store-reaper): prune/get-backup/backup-d1 + ledger migration Fake Drive gains files.update PATCH (trash), trashed-aware listings, name+parent file matching and full-media downloads; Fake D1 can answer SELECTs from a backing sqlite database. New coverage: - prune: hand-computed 37-file retention scenario (daily/weekly/ monthly union, reasons, unparseable untouched, already-trashed invisible), --apply trash calls + 'trash' ledger rows, 400-day history bounds, unknown-db rejection - get-backup: round-trip bytes + sha256, missing/unknown-db errors - backup-d1: export against the stubbed D1 loads back into sqlite byte-identical (incl. quote escaping and NULLs), token required - migration: a pre-v3 manifest (CHECK without 'trash') gains the trash direction with its rows preserved --- .../tests/test_cold_store.py | 344 +++++++++++++++++- 1 file changed, 331 insertions(+), 13 deletions(-) diff --git a/pkgs/cold-store-reaper/tests/test_cold_store.py b/pkgs/cold-store-reaper/tests/test_cold_store.py index 115959fd3..264201420 100644 --- a/pkgs/cold-store-reaper/tests/test_cold_store.py +++ b/pkgs/cold-store-reaper/tests/test_cold_store.py @@ -10,6 +10,7 @@ import os import re import shutil +import sqlite3 import stat import subprocess import sys @@ -17,6 +18,7 @@ import threading import unittest import urllib.parse +from datetime import date, timedelta from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from unittest import mock @@ -33,8 +35,9 @@ class FakeDrive: def __init__(self): self.folders = [] # {id, name, parents} self.files = {} # id -> bytes - self.file_meta = {} # id -> {name, parents, createdTime} + self.file_meta = {} # id -> {name, parents, createdTime, trashed} self.sessions = {} # sid -> {meta, chunks: {start: bytes}, total} + self.trashed_calls = [] # [{id, body}] from files.update PATCHes self.counter = 0 self.clock = 0 @@ -123,18 +126,25 @@ def do_GET(self): if parent_m and parent_m.group(1) not in folder["parents"]: continue out.append({"id": folder["id"], "name": folder["name"]}) - if parent_m and not name_m: - # parent-only listing (list-backups): include real files + if parent_m: + # file listing under a parent (list-backups with no + # name filter, get-backup with one); trashed files + # are invisible, exactly as `trashed = false` asks. for fid, meta in fake.file_meta.items(): - if parent_m.group(1) in meta.get("parents", []): - out.append( - { - "id": fid, - "name": meta["name"], - "size": str(len(fake.files[fid])), - "createdTime": meta.get("createdTime", ""), - } - ) + if meta.get("trashed"): + continue + if parent_m.group(1) not in meta.get("parents", []): + continue + if name_m and meta["name"] != name_m.group(1): + continue + out.append( + { + "id": fid, + "name": meta["name"], + "size": str(len(fake.files[fid])), + "createdTime": meta.get("createdTime", ""), + } + ) self._send(200, {"files": out}) return m = re.fullmatch(r"/drive/v3/files/([^/]+)", parsed.path) @@ -145,6 +155,10 @@ def do_GET(self): data = fake.files[fid] range_hdr = self.headers.get("Range", "") rm = re.fullmatch(r"bytes=(\d+)-(\d+)", range_hdr) + if rm is None: + # full download (get-backup): no Range header + self._send(200, data, ctype="application/octet-stream") + return start, end = int(rm.group(1)), int(rm.group(2)) chunk = data[start : end + 1] self._send( @@ -194,6 +208,20 @@ def do_PUT(self): else: self._send(308, b"", headers={"Range": f"bytes=0-{received - 1}"}) + # ------------------------------------------------- files.update + def do_PATCH(self): + parsed = urllib.parse.urlparse(self.path) + m = re.fullmatch(r"/drive/v3/files/([^/]+)", parsed.path) + if not m: + self._send(404, {"error": "not found"}) + return + body = json.loads(self._body() or b"{}") + fid = m.group(1) + fake.trashed_calls.append({"id": fid, "body": body}) + if fid in fake.file_meta and body.get("trashed") is True: + fake.file_meta[fid]["trashed"] = True + self._send(200, {"id": fid, "trashed": bool(body.get("trashed"))}) + return Handler @@ -230,6 +258,20 @@ def write_src(self, name, data: bytes) -> Path: p.write_bytes(data) return p + def seed_drive_file( + self, name, parent_id, data=b"x", trashed=False, created="2026-01-01T00:00:00.000Z" + ): + """Place a file directly in the fake Drive (no upload round-trip).""" + fid = self.fake.next_id("file") + self.fake.files[fid] = data + self.fake.file_meta[fid] = { + "name": name, + "parents": [parent_id], + "createdTime": created, + "trashed": trashed, + } + return fid + class TestAuth(ColdStoreTestBase): def test_auth_url_shape(self): @@ -375,11 +417,17 @@ def test_status_arithmetic_and_rolling_window(self): # ------------------------------------------------------------------ fake D1 class FakeD1: - """Captures D1 REST query calls; can be told to fail.""" + """Captures D1 REST query calls; can be told to fail. + + If `db` (a sqlite3 connection) is set, SELECT statements are executed + against it for real and their rows returned — enough of D1 for the + backup-d1 export path. Non-SELECT statements are only recorded. + """ def __init__(self): self.requests = [] # {path, authorization, sql, params} self.fail = False + self.db = None def make_d1_handler(fake: FakeD1): @@ -414,6 +462,21 @@ def do_POST(self): 500, {"success": False, "errors": [{"message": "fake failure"}]} ) return + sql = payload.get("sql", "") + if fake.db is not None and sql.lstrip().upper().startswith("SELECT"): + cur = fake.db.execute(sql) + cols = [d[0] for d in cur.description] + rows = [dict(zip(cols, r)) for r in cur.fetchall()] + self._send( + 200, + { + "success": True, + "errors": [], + "messages": [], + "result": [{"success": True, "results": rows}], + }, + ) + return self._send( 200, { @@ -614,6 +677,261 @@ def test_sync_requires_token(self): self.store.sync_manifest() +class TestPruneBackups(ColdStoreTestBase): + """Retention scenario, hand-computed (all dates 2026): + + 35 consecutive days of d1 backups, one per day at 02:00, ending + Mar 31, plus same-day dupes on Mar 31 at 01:00 and 03:00 (37 files). + Feb 25..Mar 1 all fall in ISO week 9; months present are Feb + Mar. + + - daily (newest 30 distinct days): Mar 2..Mar 31 newest-per-day + - weekly (newest 12 distinct ISO weeks; only 6 present): newest of + W9..W14 = Mar 1, Mar 8, Mar 15, Mar 22, Mar 29, Mar 31 03:00 + - monthly (newest 12 distinct months; only 2 present): Feb 28, + Mar 31 03:00 + Union keep = 32. Prune = Feb 25, Feb 26, Feb 27, Mar 31 01:00 and + Mar 31 02:00 (the day's newest, 03:00, is the keeper). + """ + + EXPECTED_PRUNE = { + "d1-20260225T020000Z-export.sql", + "d1-20260226T020000Z-export.sql", + "d1-20260227T020000Z-export.sql", + "d1-20260331T010000Z-export.sql", + "d1-20260331T020000Z-export.sql", + } + + def setUp(self): + super().setUp() + folders = self.store.ensure_folders() + self.d1_folder = folders["db-backups/d1"] + + def _seed_series(self): + end = date(2026, 3, 31) + ids = {} + for i in range(35): + d = end - timedelta(days=i) + name = f"d1-{d:%Y%m%d}T020000Z-export.sql" + ids[name] = self.seed_drive_file(name, self.d1_folder, data=b"dump") + for hh in ("010000", "030000"): + name = f"d1-20260331T{hh}Z-export.sql" + ids[name] = self.seed_drive_file(name, self.d1_folder, data=b"dump") + return ids + + def test_prune_dry_run_plan(self): + ids = self._seed_series() + self.seed_drive_file("d1-random.sql", self.d1_folder, data=b"?") + self.seed_drive_file("notes.txt", self.d1_folder, data=b"?") + trashed_id = self.seed_drive_file( + "d1-20200101T020000Z-export.sql", self.d1_folder, data=b"?", trashed=True + ) + plans = self.store.prune_backups("d1") + self.assertEqual(len(plans), 1) + plan = plans[0] + self.assertFalse(plan["applied"]) + keep = {f["name"]: f for f in plan["keep"]} + prune = {f["name"] for f in plan["prune"]} + self.assertEqual(prune, self.EXPECTED_PRUNE) + self.assertEqual(len(keep), 32) + self.assertEqual(keep["d1-20260301T020000Z-export.sql"]["reasons"], ["weekly"]) + self.assertEqual(keep["d1-20260228T020000Z-export.sql"]["reasons"], ["monthly"]) + self.assertEqual( + keep["d1-20260331T030000Z-export.sql"]["reasons"], + ["daily", "monthly", "weekly"], + ) + self.assertEqual(keep["d1-20260302T020000Z-export.sql"]["reasons"], ["daily"]) + skipped = {f["name"] for f in plan["unparseable"]} + self.assertEqual(skipped, {"d1-random.sql", "notes.txt"}) + # the already-trashed file is invisible to the plan entirely + all_ids = {f["id"] for f in plan["keep"] + plan["prune"] + plan["unparseable"]} + self.assertNotIn(trashed_id, all_ids) + self.assertEqual(len(all_ids), 39) + # dry-run touches nothing: no PATCH, no ledger rows + self.assertEqual(self.fake.trashed_calls, []) + conn = self.store.db() + n = conn.execute( + "SELECT count(*) n FROM transfer_ledger WHERE kind='backup-prune'" + ).fetchone()["n"] + conn.close() + self.assertEqual(n, 0) + self.assertEqual(set(ids) - set(keep) - prune, set()) + + def test_prune_apply_trashes_and_ledgers(self): + ids = self._seed_series() + plans = self.store.prune_backups("d1", apply=True) + plan = plans[0] + self.assertTrue(plan["applied"]) + prune_ids = {ids[name] for name in self.EXPECTED_PRUNE} + self.assertEqual({c["id"] for c in self.fake.trashed_calls}, prune_ids) + for call in self.fake.trashed_calls: + self.assertEqual(call["body"], {"trashed": True}) + for fid in prune_ids: + self.assertTrue(self.fake.file_meta[fid]["trashed"]) + conn = self.store.db() + rows = conn.execute( + "SELECT * FROM transfer_ledger WHERE kind='backup-prune'" + ).fetchall() + conn.close() + self.assertEqual(len(rows), len(self.EXPECTED_PRUNE)) + self.assertEqual({r["ref"] for r in rows}, self.EXPECTED_PRUNE) + for r in rows: + self.assertEqual(r["direction"], "trash") + # after applying, the folder's visible set is exactly the keepers + self.assertEqual(len(self.store.list_backups("d1")), 32) + + def test_prune_long_history_oldest_pruned(self): + end = date(2026, 3, 31) + for i in range(400): + d = end - timedelta(days=i) + self.seed_drive_file( + f"d1-{d:%Y%m%d}T020000Z-export.sql", self.d1_folder, data=b"x" + ) + plan = self.store.prune_backups("d1")[0] + keep = {f["name"] for f in plan["keep"]} + prune = {f["name"] for f in plan["prune"]} + self.assertEqual(len(keep) + len(prune), 400) + oldest = f"d1-{end - timedelta(days=399):%Y%m%d}T020000Z-export.sql" + self.assertIn(oldest, prune) + for i in range(30): # newest-per-day of the last 30 days survives + d = end - timedelta(days=i) + self.assertIn(f"d1-{d:%Y%m%d}T020000Z-export.sql", keep) + # the union can never exceed 30 daily + 12 weekly + 12 monthly + self.assertLessEqual(len(keep), 30 + 12 + 12) + + def test_prune_all_dbs_and_unknown_db(self): + plans = self.store.prune_backups() + self.assertEqual([p["db"] for p in plans], list(cold_store.BACKUP_DBS)) + with self.assertRaises(cold_store.ColdStoreError): + self.store.prune_backups("oracle") + + +class TestGetBackup(ColdStoreTestBase): + def test_get_backup_roundtrip(self): + data = b"-- turso dump\n" + bytes(range(256)) * 4 + src = self.write_src("libsql.sql", data) + put = self.store.put_backup("turso", str(src)) + out = Path(self.tmp.name) / "restored.sql" + result = self.store.get_backup("turso", put["name"], str(out)) + self.assertEqual(out.read_bytes(), data) + self.assertEqual(result["bytes"], len(data)) + self.assertEqual(result["sha256"], hashlib.sha256(data).hexdigest()) + self.assertEqual(result["drive_file_id"], put["drive_file_id"]) + + def test_get_backup_missing_and_unknown_db(self): + with self.assertRaises(cold_store.ColdStoreError): + self.store.get_backup( + "d1", "d1-20260101T000000Z-nope.sql", str(Path(self.tmp.name) / "x") + ) + with self.assertRaises(cold_store.ColdStoreError): + self.store.get_backup("oracle", "x", str(Path(self.tmp.name) / "x")) + + +class TestBackupD1(ColdStoreTestBase): + def setUp(self): + super().setUp() + self.d1 = FakeD1() + # check_same_thread=False: the HTTP handler thread runs the queries. + self.d1.db = sqlite3.connect(":memory:", check_same_thread=False) + self.d1.db.executescript( + """ + CREATE TABLE objects (hash TEXT PRIMARY KEY, size INTEGER, note TEXT); + CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT); + INSERT INTO objects VALUES ('aaa', 10, 'plain'); + INSERT INTO objects VALUES ('bbb', 20, 'o''brien'); + INSERT INTO objects VALUES ('ccc', 30, NULL); + INSERT INTO meta VALUES ('schema_version', '1'); + INSERT INTO meta VALUES ('project_budget_bytes', '3298534883328'); + """ + ) + self.d1_server = ThreadingHTTPServer(("127.0.0.1", 0), make_d1_handler(self.d1)) + self.d1_thread = threading.Thread( + target=self.d1_server.serve_forever, daemon=True + ) + self.d1_thread.start() + self.store.d1_base = f"http://127.0.0.1:{self.d1_server.server_address[1]}" + + def tearDown(self): + self.d1_server.shutdown() + self.d1_server.server_close() + super().tearDown() + + def test_backup_d1_export_loads_back_identically(self): + with mock.patch.dict(os.environ, {"CLOUDFLARE_API_TOKEN": "fake-cf-token"}): + result = self.store.backup_d1() + self.assertRegex( + result["name"], r"^d1-\d{8}T\d{6}Z-cold-store-manifest-export\.sql$" + ) + self.assertEqual(result["tables"], ["meta", "objects"]) + self.assertEqual(result["rows"], 5) + folders = self.store.ensure_folders() + meta = self.fake.file_meta[result["drive_file_id"]] + self.assertEqual(meta["parents"], [folders["db-backups/d1"]]) + sql_text = self.fake.files[result["drive_file_id"]].decode() + self.assertTrue( + sql_text.startswith("-- cold-store-manifest D1 export (via REST query API)") + ) + self.assertIn("-- tables: meta, objects", sql_text) + self.assertIn("'o''brien'", sql_text) # quote escaping survived + fresh = sqlite3.connect(":memory:") + fresh.executescript( + "CREATE TABLE objects (hash TEXT PRIMARY KEY, size INTEGER, note TEXT);" + "CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT);" + ) + fresh.executescript(sql_text) + self.assertEqual( + fresh.execute("SELECT hash, size, note FROM objects ORDER BY hash").fetchall(), + [("aaa", 10, "plain"), ("bbb", 20, "o'brien"), ("ccc", 30, None)], + ) + self.assertEqual( + fresh.execute("SELECT key, value FROM meta ORDER BY key").fetchall(), + [("project_budget_bytes", "3298534883328"), ("schema_version", "1")], + ) + + def test_backup_d1_requires_token(self): + env = {k: v for k, v in os.environ.items() if k != "CLOUDFLARE_API_TOKEN"} + with mock.patch.dict(os.environ, env, clear=True): + with self.assertRaises(cold_store.ColdStoreError): + self.store.backup_d1() + + +class TestLedgerTrashMigration(ColdStoreTestBase): + def test_pre_v3_manifest_gains_trash_direction(self): + # Rebuild a manifest exactly as a pre-v3 home has it (CHECK + # without 'trash'), with one live row, then open it via a store: + # the migration must preserve the row and admit 'trash' rows. + legacy_home = Path(self.tmp.name) / "legacy-home" + legacy_home.mkdir() + legacy_schema = cold_store.SCHEMA_PATH.read_text().replace( + "('upload','download','trash')", "('upload','download')" + ) + self.assertIn("IN ('upload','download')", legacy_schema) + conn = sqlite3.connect(legacy_home / "manifest.sqlite") + conn.executescript(legacy_schema) + conn.execute( + "INSERT INTO transfer_ledger (direction, bytes, kind, ref)" + " VALUES ('upload', 42, 'bundle', 'b_legacy')" + ) + conn.commit() + conn.close() + base = self.store.drive_base.rsplit("/drive/v3", 1)[0] + legacy = ColdStore( + legacy_home, + token_url=base + "/token", + drive_base=self.store.drive_base, + upload_base=base, + ) + conn = legacy.db() + rows = conn.execute("SELECT direction, ref FROM transfer_ledger").fetchall() + self.assertEqual([(r["direction"], r["ref"]) for r in rows], [("upload", "b_legacy")]) + with conn: + legacy.ledger_add(conn, "trash", 7, "backup-prune", "d1-old.sql") + n = conn.execute( + "SELECT count(*) n FROM transfer_ledger WHERE direction='trash'" + ).fetchone()["n"] + conn.close() + self.assertEqual(n, 1) + + class TestWorkerDecoder(unittest.TestCase): """Local proof for the rehydrator Worker: the exact decoder module the Worker bundles (worker/src/decoder.js, fzstd) decodes one member frame From deb70afa4db50e478553b3183d9aa3128c7b66cc Mon Sep 17 00:00:00 2001 From: Muse Sweep Date: Mon, 26 Oct 2026 12:46:34 +0000 Subject: [PATCH 3/5] feat(cold-store-reaper): nightly-backup + restore-drill scripts nightly-backup.sh: backup-d1 -> prune-backups --apply -> sync-manifest -> status, aborting loudly on any failure. restore-drill.sh (DESIGN section 8): three pseudo-random cold objects rehydrated through the production Worker with sha256 verification, plus the newest d1 backup downloaded, loaded into in-memory sqlite and row-count-compared against live D1 (local manifest as fallback oracle). Per-item DRILL PASS/FAIL lines and a final verdict; non-zero exit on any failure. Both shellcheck-clean, chmod +x, secrets only via documented environment variables. --- .../scripts/nightly-backup.sh | 37 ++++ .../scripts/restore-drill.sh | 175 ++++++++++++++++++ 2 files changed, 212 insertions(+) create mode 100755 pkgs/cold-store-reaper/scripts/nightly-backup.sh create mode 100755 pkgs/cold-store-reaper/scripts/restore-drill.sh diff --git a/pkgs/cold-store-reaper/scripts/nightly-backup.sh b/pkgs/cold-store-reaper/scripts/nightly-backup.sh new file mode 100755 index 000000000..0b340be7d --- /dev/null +++ b/pkgs/cold-store-reaper/scripts/nightly-backup.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +# nightly-backup.sh — the cold store's nightly DB-backup run +# (DESIGN.md section 5: nightly dumps into Drive db-backups//, +# retention enforced by trashing, never deleting). +# +# Steps, in order: +# 1. cold-store backup-d1 export the D1 cold-store-manifest +# database and file it in db-backups/d1/ +# 2. cold-store prune-backups --apply +# enforce 30 daily / 12 weekly / +# 12 monthly retention across all DBs +# 3. cold-store sync-manifest push the local manifest (including +# tonight's ledger rows) to D1 +# 4. cold-store status one-screen summary (head) +# +# Environment — the operator/cron supplies these; the script itself +# stores nothing and echoes no secret values: +# CLOUDFLARE_API_TOKEN required. Materialize it from Bitwarden +# (cloudflare.api-token-day-to-day) into the +# cron environment; backup-d1 and sync-manifest +# read it from the environment only. +# COLD_STORE_HOME optional; the reaper home (default +# ~/.config/cold-store). That directory's +# credentials.json holds the Drive grant. +# +# Fails loudly: any failing step aborts the run non-zero (set -e). +set -euo pipefail + +: "${CLOUDFLARE_API_TOKEN:?CLOUDFLARE_API_TOKEN must be set — materialize it from Bitwarden into the environment first}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +COLD_STORE="${SCRIPT_DIR}/../cold-store" + +"${COLD_STORE}" backup-d1 +"${COLD_STORE}" prune-backups --apply +"${COLD_STORE}" sync-manifest +"${COLD_STORE}" status | head -n 20 diff --git a/pkgs/cold-store-reaper/scripts/restore-drill.sh b/pkgs/cold-store-reaper/scripts/restore-drill.sh new file mode 100755 index 000000000..6e151aa1b --- /dev/null +++ b/pkgs/cold-store-reaper/scripts/restore-drill.sh @@ -0,0 +1,175 @@ +#!/usr/bin/env bash +# restore-drill.sh — the cold store's quarterly restore drill +# (DESIGN.md section 8: an untested archive is a rumour, not a backup). +# +# What it proves, end to end: +# 1. Three pseudo-random cold objects from the local manifest are +# fetched back through the rehydrator Worker +# (GET https://cold-store-rehydrator.developing-today.workers.dev/fetch/) +# and each response body's sha256 must equal the object hash — +# the same check the reaper's own `fetch` performs, but through +# the production read path (D1 lookup -> Durable Object -> Drive +# Range pull -> zstd decode -> hash verify). +# 2. The newest d1 backup in Drive is downloaded with get-backup, +# loaded into an in-memory SQLite database, and its per-table row +# counts must equal the live D1 counts (queried over the same REST +# API backup-d1 uses). If D1 is unreachable, the fallback oracle is +# the latest local manifest's row counts. +# +# Prints "DRILL PASS:" / "DRILL FAIL:" per item and a final verdict; +# exits non-zero if anything failed. +# +# Environment — the operator supplies these; the script stores nothing +# and echoes no secret values: +# REHYDRATOR_API_KEY required. Bearer key for the Worker +# (Bitwarden: cold-store.rehydrator-api-key). +# CLOUDFLARE_API_TOKEN required. For the live-D1 count check +# (Bitwarden: cloudflare.api-token-day-to-day). +# COLD_STORE_HOME optional; the reaper home (default +# ~/.config/cold-store), home of manifest.sqlite. +set -euo pipefail + +: "${REHYDRATOR_API_KEY:?REHYDRATOR_API_KEY must be set — materialize it from Bitwarden into the environment first}" +: "${CLOUDFLARE_API_TOKEN:?CLOUDFLARE_API_TOKEN must be set — materialize it from Bitwarden into the environment first}" + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +COLD_STORE="${SCRIPT_DIR}/../cold-store" +STORE_HOME="${COLD_STORE_HOME:-${HOME}/.config/cold-store}" +MANIFEST="${STORE_HOME}/manifest.sqlite" +WORKER_BASE="https://cold-store-rehydrator.developing-today.workers.dev" + +WORK="$(mktemp -d)" +trap 'rm -rf "${WORK}"' EXIT + +FAILURES=0 +pass() { echo "DRILL PASS: $1"; } +fail() { + echo "DRILL FAIL: $1" + FAILURES=$((FAILURES + 1)) +} + +# ------------------------------------------------------------ cold objects +python3 - "${MANIFEST}" >"${WORK}/cold-hashes.txt" <<'PY' +import sqlite3 +import sys + +try: + conn = sqlite3.connect(f"file:{sys.argv[1]}?mode=ro", uri=True) + rows = conn.execute( + "SELECT hash FROM objects WHERE state='cold' ORDER BY hash" + ).fetchall() +except sqlite3.Error: + rows = [] +for (h,) in rows: + print(h) +PY +mapfile -t COLD_HASHES <"${WORK}/cold-hashes.txt" + +if [[ "${#COLD_HASHES[@]}" -eq 0 ]]; then + fail "no cold objects in the local manifest — nothing to rehydrate" +else + PICKED="$(printf '%s\n' "${COLD_HASHES[@]}" | shuf | head -n 3)" + while IFS= read -r obj_hash; do + [[ -n "${obj_hash}" ]] || continue + out="${WORK}/object.bin" + if curl -fsS --max-time 120 \ + -H "Authorization: Bearer ${REHYDRATOR_API_KEY}" \ + "${WORKER_BASE}/fetch/${obj_hash}" -o "${out}"; then + actual="$(sha256sum "${out}" | awk '{print $1}')" + if [[ "${actual}" = "${obj_hash}" ]]; then + pass "object ${obj_hash} rehydrated via Worker, sha256 verified" + else + fail "object ${obj_hash} sha256 mismatch (Worker returned ${actual})" + fi + else + fail "object ${obj_hash} fetch via Worker failed" + fi + done <<<"${PICKED}" +fi + +# ------------------------------------------------------------ newest d1 backup +BACKUP_NAME="$("${COLD_STORE}" list-backups --db d1 2>/dev/null \ + | sed -n 's/.*\(d1-[0-9]*T[0-9]*Z-[^ ]*\).*/\1/p' | head -n 1)" || true +if [[ -z "${BACKUP_NAME}" ]]; then + fail "no d1 backup found in Drive" +elif ! "${COLD_STORE}" get-backup d1 "${BACKUP_NAME}" "${WORK}/backup.sql" >/dev/null; then + fail "get-backup could not download ${BACKUP_NAME}" +elif python3 - "${WORK}/backup.sql" "${MANIFEST}" <<'PY' +import json +import os +import sqlite3 +import sys +import urllib.request + +export_path, manifest_path = sys.argv[1], sys.argv[2] +ACCOUNT = "234b866b99ce2b142d7963d8d139cbfa" +DATABASE = "15d5f460-835c-4f24-b229-222908a5b30c" +TABLES_SQL = ( + "SELECT name, sql FROM sqlite_master WHERE type='table'" + " AND name NOT LIKE 'sqlite_%' AND name NOT LIKE '_cf_%'" +) + + +def d1(sql): + req = urllib.request.Request( + f"https://api.cloudflare.com/client/v4/accounts/{ACCOUNT}" + f"/d1/database/{DATABASE}/query", + data=json.dumps({"sql": sql}).encode(), + method="POST", + headers={ + "Authorization": "Bearer " + os.environ["CLOUDFLARE_API_TOKEN"], + "Content-Type": "application/json", + }, + ) + with urllib.request.urlopen(req, timeout=60) as resp: + payload = json.loads(resp.read()) + if not payload.get("success"): + raise RuntimeError("D1 query failed") + return payload["result"][0]["results"] + + +try: + meta = d1(TABLES_SQL) + creates = [(r["name"], r["sql"]) for r in meta] + expected = { + r["name"]: d1(f'SELECT count(*) AS n FROM "{r["name"]}"')[0]["n"] + for r in meta + } + source = "live D1" +except Exception: + conn = sqlite3.connect(f"file:{manifest_path}?mode=ro", uri=True) + creates = conn.execute(TABLES_SQL).fetchall() + expected = { + name: conn.execute(f'SELECT count(*) FROM "{name}"').fetchone()[0] + for name, _sql in creates + } + conn.close() + source = "local manifest (D1 unreachable)" + +mem = sqlite3.connect(":memory:") +for _name, sql in creates: + if sql: + mem.execute(sql) +with open(export_path) as f: + mem.executescript(f.read()) +ok = True +for name, want in sorted(expected.items()): + got = mem.execute(f'SELECT count(*) FROM "{name}"').fetchone()[0] + print(f" {name}: backup has {got} row(s), {source} has {want}") + if got != want: + ok = False +sys.exit(0 if ok else 1) +PY +then + pass "d1 backup ${BACKUP_NAME} loads and row counts match" +else + fail "d1 backup ${BACKUP_NAME} row counts do not match" +fi + +# ------------------------------------------------------------ verdict +if [[ "${FAILURES}" -eq 0 ]]; then + echo "DRILL VERDICT: PASS" +else + echo "DRILL VERDICT: FAIL (${FAILURES} failure(s))" + exit 1 +fi From c3f9a3c8cf9baaf75beec021d6ee737394ec24c5 Mon Sep 17 00:00:00 2001 From: Muse Sweep Date: Mon, 26 Oct 2026 12:47:44 +0000 Subject: [PATCH 4/5] docs(cold-store-reaper): retention rule, new commands, scripts Documents prune-backups (30 daily / 12 weekly / 12 monthly union, trash-not-delete, unparseable names never pruned), get-backup, backup-d1, both operator scripts with their environment prerequisites, and the schema v1.1 'trash' caveat for the D1 copy. --- pkgs/cold-store-reaper/README.md | 60 ++++++++++++++++++++++++++++++-- 1 file changed, 57 insertions(+), 3 deletions(-) diff --git a/pkgs/cold-store-reaper/README.md b/pkgs/cold-store-reaper/README.md index e82a2a861..377b70b9b 100644 --- a/pkgs/cold-store-reaper/README.md +++ b/pkgs/cold-store-reaper/README.md @@ -78,6 +78,10 @@ cold-store status # counts, cold bytes, last # 700 GB/day hard-stop + 3 TiB budget checks cold-store sync-manifest # push the local manifest into D1 cold-store sync-manifest --dry-run # ... or just print the statement counts +cold-store backup-d1 # export the D1 manifest DB into db-backups/d1/ +cold-store prune-backups # retention plan (dry-run) +cold-store prune-backups --apply # ... and trash the excess +cold-store get-backup d1 ./restored.sql # download one backup in full ``` ## DB backups (design section 5) @@ -93,11 +97,61 @@ idempotently). Files land with dated names — cold-store put-backup supabase pgdump-2026-10-26.sql.zst # prints the Drive file id cold-store list-backups # all DBs, newest first cold-store list-backups --db turso +cold-store backup-d1 # export the D1 manifest DB itself + # (REST query API) and file it under d1/ +cold-store get-backup d1 d1-<...>-cold-store-manifest-export.sql ./restored.sql ``` -**Retention is NOT enforced yet**: the design's 30 daily / 12 weekly / -12 monthly policy (trash, not delete) is pending a later version. Until -then, backups accumulate and pruning is manual. +### Retention (design section 5) + +`prune-backups [--db ] [--apply]` enforces **30 daily / 12 weekly / +12 monthly**, per DB folder, as a pure function of the folder listing — +never of the wall clock. For each `db-backups//` folder it lists the +files (already-trashed files are excluded), parses the timestamp out of +each dated filename (`--`; names that do +not parse are **never** pruned, only reported as skipped-unparseable), +and computes the keep set as the union of: + +- the newest backup of each of the **last 30 distinct calendar days** + present in the folder, +- the newest backup of each of the **last 12 distinct ISO weeks** + present, +- the newest backup of each of the **last 12 distinct calendar months** + present. + +Everything else is a prune candidate. The default is a dry run: the +plan is printed with a reason per file (`daily` / `weekly` / `monthly` +for keepers, `excess` for candidates). With `--apply`, each candidate +is **trashed** via `files.update {trashed: true}` — never permanently +deleted, so Drive's own trash grace applies — and one `transfer_ledger` +row per file is written (`direction='trash'`, `kind='backup-prune'`, +`ref=`). + +One migration caveat: the `'trash'` ledger direction is a schema v1.1 +extension (see `schema.sql`). Local manifests are migrated in place on +first open, but the **D1 copy of the schema needs the same one-line +CHECK change** before prune rows can sync — until then `sync-manifest` +fails on the `transfer_ledger` statement with D1's own error text. + +### Scripts + +`scripts/` holds the two operator entry points (bash, `set -euo +pipefail`, no secrets stored or echoed — credentials arrive via the +environment only): + +- `scripts/nightly-backup.sh` — the nightly run: `backup-d1`, then + `prune-backups --apply`, then `sync-manifest`, then `status`. Any + failing step aborts the run non-zero. Requires `CLOUDFLARE_API_TOKEN` + in the environment (materialize from Bitwarden into the cron env). +- `scripts/restore-drill.sh` — the quarterly restore drill (design + section 8): picks 3 pseudo-random cold objects from the local + manifest, fetches each through the rehydrator Worker and verifies its + sha256, then downloads the newest d1 backup with `get-backup`, loads + it into in-memory SQLite, and compares per-table row counts against + live D1 (falling back to the local manifest if D1 is unreachable). + Prints `DRILL PASS`/`DRILL FAIL` per item and a final verdict; exits + non-zero on any failure. Requires `REHYDRATOR_API_KEY` and + `CLOUDFLARE_API_TOKEN` in the environment. ## Manifest sync → D1 From ce44d1e9e5c2803525a568c975bc4bd39937434f Mon Sep 17 00:00:00 2001 From: Muse Sweep Date: Mon, 26 Oct 2026 12:47:44 +0000 Subject: [PATCH 5/5] style(cold-store-reaper): treefmt (ruff format + shfmt) --- pkgs/cold-store-reaper/cold_store.py | 8 ++------ pkgs/cold-store-reaper/scripts/restore-drill.sh | 14 +++++++------- pkgs/cold-store-reaper/tests/test_cold_store.py | 15 ++++++++++++--- 3 files changed, 21 insertions(+), 16 deletions(-) diff --git a/pkgs/cold-store-reaper/cold_store.py b/pkgs/cold-store-reaper/cold_store.py index 07da6a482..69e51d7b5 100755 --- a/pkgs/cold-store-reaper/cold_store.py +++ b/pkgs/cold-store-reaper/cold_store.py @@ -774,9 +774,7 @@ def list_backups(self, db_name: str | None = None) -> list[dict]: def _parse_backup_stamp(db_name: str, name: str) -> datetime | None: """Timestamp embedded in a dated backup name, or None if the name does not follow -- for this db.""" - m = re.fullmatch( - re.escape(db_name) + r"-(\d{8}T\d{6}Z)-.+", name - ) + m = re.fullmatch(re.escape(db_name) + r"-(\d{8}T\d{6}Z)-.+", name) if not m: return None try: @@ -1127,9 +1125,7 @@ def render_d1_export(self) -> tuple[str, list[str], int]: rows = self._d1_select(f'SELECT * FROM "{t}" ORDER BY rowid') for row in rows: vals = ", ".join(self._sql_literal(row[c]) for c in cols) - lines.append( - f'INSERT INTO "{t}" ({", ".join(cols)}) VALUES ({vals});' - ) + lines.append(f'INSERT INTO "{t}" ({", ".join(cols)}) VALUES ({vals});') total += len(rows) return "\n".join(lines) + "\n", tables, total diff --git a/pkgs/cold-store-reaper/scripts/restore-drill.sh b/pkgs/cold-store-reaper/scripts/restore-drill.sh index 6e151aa1b..d0fd60da7 100755 --- a/pkgs/cold-store-reaper/scripts/restore-drill.sh +++ b/pkgs/cold-store-reaper/scripts/restore-drill.sh @@ -65,18 +65,18 @@ for (h,) in rows: PY mapfile -t COLD_HASHES <"${WORK}/cold-hashes.txt" -if [[ "${#COLD_HASHES[@]}" -eq 0 ]]; then +if [[ ${#COLD_HASHES[@]} -eq 0 ]]; then fail "no cold objects in the local manifest — nothing to rehydrate" else PICKED="$(printf '%s\n' "${COLD_HASHES[@]}" | shuf | head -n 3)" while IFS= read -r obj_hash; do - [[ -n "${obj_hash}" ]] || continue + [[ -n ${obj_hash} ]] || continue out="${WORK}/object.bin" if curl -fsS --max-time 120 \ -H "Authorization: Bearer ${REHYDRATOR_API_KEY}" \ "${WORKER_BASE}/fetch/${obj_hash}" -o "${out}"; then actual="$(sha256sum "${out}" | awk '{print $1}')" - if [[ "${actual}" = "${obj_hash}" ]]; then + if [[ ${actual} == "${obj_hash}" ]]; then pass "object ${obj_hash} rehydrated via Worker, sha256 verified" else fail "object ${obj_hash} sha256 mismatch (Worker returned ${actual})" @@ -88,9 +88,9 @@ else fi # ------------------------------------------------------------ newest d1 backup -BACKUP_NAME="$("${COLD_STORE}" list-backups --db d1 2>/dev/null \ - | sed -n 's/.*\(d1-[0-9]*T[0-9]*Z-[^ ]*\).*/\1/p' | head -n 1)" || true -if [[ -z "${BACKUP_NAME}" ]]; then +BACKUP_NAME="$("${COLD_STORE}" list-backups --db d1 2>/dev/null | + sed -n 's/.*\(d1-[0-9]*T[0-9]*Z-[^ ]*\).*/\1/p' | head -n 1)" || true +if [[ -z ${BACKUP_NAME} ]]; then fail "no d1 backup found in Drive" elif ! "${COLD_STORE}" get-backup d1 "${BACKUP_NAME}" "${WORK}/backup.sql" >/dev/null; then fail "get-backup could not download ${BACKUP_NAME}" @@ -167,7 +167,7 @@ else fi # ------------------------------------------------------------ verdict -if [[ "${FAILURES}" -eq 0 ]]; then +if [[ ${FAILURES} -eq 0 ]]; then echo "DRILL VERDICT: PASS" else echo "DRILL VERDICT: FAIL (${FAILURES} failure(s))" diff --git a/pkgs/cold-store-reaper/tests/test_cold_store.py b/pkgs/cold-store-reaper/tests/test_cold_store.py index 264201420..e4733e3b0 100644 --- a/pkgs/cold-store-reaper/tests/test_cold_store.py +++ b/pkgs/cold-store-reaper/tests/test_cold_store.py @@ -259,7 +259,12 @@ def write_src(self, name, data: bytes) -> Path: return p def seed_drive_file( - self, name, parent_id, data=b"x", trashed=False, created="2026-01-01T00:00:00.000Z" + self, + name, + parent_id, + data=b"x", + trashed=False, + created="2026-01-01T00:00:00.000Z", ): """Place a file directly in the fake Drive (no upload round-trip).""" fid = self.fake.next_id("file") @@ -879,7 +884,9 @@ def test_backup_d1_export_loads_back_identically(self): ) fresh.executescript(sql_text) self.assertEqual( - fresh.execute("SELECT hash, size, note FROM objects ORDER BY hash").fetchall(), + fresh.execute( + "SELECT hash, size, note FROM objects ORDER BY hash" + ).fetchall(), [("aaa", 10, "plain"), ("bbb", 20, "o'brien"), ("ccc", 30, None)], ) self.assertEqual( @@ -922,7 +929,9 @@ def test_pre_v3_manifest_gains_trash_direction(self): ) conn = legacy.db() rows = conn.execute("SELECT direction, ref FROM transfer_ledger").fetchall() - self.assertEqual([(r["direction"], r["ref"]) for r in rows], [("upload", "b_legacy")]) + self.assertEqual( + [(r["direction"], r["ref"]) for r in rows], [("upload", "b_legacy")] + ) with conn: legacy.ledger_add(conn, "trash", 7, "backup-prune", "d1-old.sql") n = conn.execute(