1"""Encrypted-at-rest-by-FileVault-free policy: backups are rsync'ed over SSH (encrypted in transit) to a second2LAN node, plus a pg_dump of the metadata database. Restore drill: `findata backup restore-drill`."""3from __future__ import annotations45import os6import subprocess7from datetime import UTC, datetime8from pathlib import Path9from typing import Any1011from findata_core.config import get_settings12from findata_core.db import transaction1314from findata_ingest.lake import Lake151617def _env() -> dict[str, str]:18 env = dict(os.environ)19 env["PATH"] = "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"20 return env212223def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:24 s = get_settings()25 target = (job.get("params") or {}).get("target") or s.backup_target26 if not target:27 raise RuntimeError("FINDATA_BACKUP_TARGET not configured")28 host, _, remote_dir = target.partition(":")29 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")30 dumps = lake.root / "_pgdump"31 dumps.mkdir(exist_ok=True)32 dump = dumps / f"findata-{stamp}.sql.zst"33 url = s.database_url.get_secret_value()34 cmd = f"pg_dump --no-owner --no-privileges '{url}' | zstd -q -T0 -o '{dump}'"35 r = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=_env(), timeout=3600)36 if r.returncode != 0:37 raise RuntimeError(f"pg_dump failed: {r.stderr[-500:]}")38 for old in sorted(dumps.glob("findata-*.sql.zst"))[:-14]:39 old.unlink()40 subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"mkdir -p {remote_dir}/lake {remote_dir}/pgdump"], check=True, timeout=60, env=_env())41 excludes = ["--exclude", "_tmp/", "--exclude", "_retired/", "--exclude", "exports/", "--exclude", ".writer.lock"]42 rs = subprocess.run(["rsync", "-a", "--delete", "--stats", *excludes, f"{lake.root}/", f"{host}:{remote_dir}/lake/"],43 capture_output=True, text=True, env=_env(), timeout=6 * 3600)44 if rs.returncode != 0:45 raise RuntimeError(f"rsync failed: {rs.stderr[-500:]}")46 rs2 = subprocess.run(["rsync", "-a", f"{dumps}/", f"{host}:{remote_dir}/pgdump/"], capture_output=True, text=True, env=_env(), timeout=3600)47 if rs2.returncode != 0:48 raise RuntimeError(f"rsync pgdump failed: {rs2.stderr[-500:]}")49 # verification: remote file count + bytes of curated/50 v = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"find {remote_dir}/lake/curated -type f | wc -l; du -sk {remote_dir}/lake/curated | cut -f1"],51 capture_output=True, text=True, env=_env(), timeout=600)52 remote_files, remote_kb = [x.strip() for x in v.stdout.strip().splitlines()[:2]] if v.returncode == 0 else ("?", "?")53 local_files = sum(1 for p in (lake.root / "curated").rglob("*") if p.is_file())54 stats = {"at": stamp, "target": target, "pg_dump": dump.name, "pg_dump_bytes": dump.stat().st_size,55 "rsync_summary": [ln for ln in rs.stdout.splitlines() if "Number of files" in ln or "Total transferred" in ln or "Total file size" in ln],56 "remote_curated_files": remote_files, "local_curated_files": local_files, "remote_curated_kb": remote_kb,57 "verified": str(local_files) == str(remote_files)}58 with transaction() as conn:59 conn.execute("INSERT INTO settings_kv(key, value) VALUES ('last_backup', %s) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",60 (__import__("json").dumps(stats, default=str),))61 log.info("backup.done", **{k: v for k, v in stats.items() if k != "rsync_summary"})62 return stats636465def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw") -> dict[str, Any]:66 """Restore the latest pg_dump into a scratch database and the lake into `dest`, then verify manifest hashes."""67 import hashlib68 import json69 s = get_settings()70 host, _, remote_dir = target.partition(":")71 dest.mkdir(parents=True, exist_ok=True)72 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/curated/{dataset_id}/", f"{dest}/curated/{dataset_id}/"], check=True, env=_env(), timeout=3600)73 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/manifests/{dataset_id}/", f"{dest}/manifests/{dataset_id}/"], check=True, env=_env(), timeout=600)74 latest = subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"ls -1 {remote_dir}/pgdump | sort | tail -1"], capture_output=True, text=True, env=_env(), timeout=60).stdout.strip()75 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600)76 base_url = s.database_url.get_secret_value().rsplit("/", 1)[0]77 scratch = f"findata_restore_{datetime.now(UTC).strftime('%H%M%S')}"78 subprocess.run(["bash", "-c", f"psql '{base_url}/postgres' -qc 'CREATE DATABASE {scratch}' && zstd -dc '{dest}/{latest}' | psql -q '{base_url}/{scratch}' >/dev/null"],79 check=True, env=_env(), timeout=3600)80 out = subprocess.run(["bash", "-c", f"psql -tA '{base_url}/{scratch}' -c \"SELECT count(*) FROM lake_files WHERE dataset_id='{dataset_id}' AND version_removed IS NULL\""],81 capture_output=True, text=True, env=_env(), timeout=120).stdout.strip()82 manifests = sorted((dest / "manifests" / dataset_id).glob("v*.json"), key=lambda p: int(p.stem[1:]))83 verified = 084 missing = 085 if manifests:86 m = json.loads(manifests[-1].read_text())87 for f in m["files"]:88 p = dest / f["path"]89 if not p.exists():90 missing += 191 continue92 h = hashlib.sha256(p.read_bytes()).hexdigest()93 verified += int(h == f["sha256"])94 subprocess.run(["bash", "-c", f"psql '{base_url}/postgres' -qc 'DROP DATABASE {scratch}'"], env=_env(), timeout=120)95 return {"pg_dump": latest, "restored_lake_files_in_db": out, "manifest_version": manifests[-1].stem if manifests else None,96 "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)}97