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 s = get_settings()69 host, _, remote_dir = target.partition(":")70 dest.mkdir(parents=True, exist_ok=True)71 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/curated/{dataset_id}/", f"{dest}/curated/{dataset_id}/"], check=True, env=_env(), timeout=3600)72 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()73 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600)74 base_url = s.database_url.get_secret_value().rsplit("/", 1)[0]75 scratch = f"findata_restore_{datetime.now(UTC).strftime('%H%M%S')}"76 env = _env()77 env["FD_PG_ADMIN"] = f"{base_url}/postgres"78 env["FD_PG_SCRATCH"] = f"{base_url}/{scratch}"79 env["FD_DUMP"] = f"{dest}/{latest}"8081 def _sh(cmd: str, timeout: int) -> subprocess.CompletedProcess:82 r = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=env, timeout=timeout)83 if r.returncode != 0:84 from findata_core.redact import redact85 raise RuntimeError(redact(f"restore step failed: {r.stderr.strip()[-400:]}", s.secret_values()))86 return r8788 _sh('psql "$FD_PG_ADMIN" -qc "CREATE DATABASE \"$(basename "$FD_PG_SCRATCH")\"" && zstd -dc "$FD_DUMP" | psql -q "$FD_PG_SCRATCH" >/dev/null', 3600)89 out = _sh(f'psql -tA "$FD_PG_SCRATCH" -c "SELECT count(*) FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 120).stdout.strip()90 listing = _sh(f'psql -tA -F "|" "$FD_PG_SCRATCH" -c "SELECT path, sha256 FROM lake_files WHERE dataset_id=\'{dataset_id}\' AND version_removed IS NULL"', 600).stdout91 verified = 092 missing = 093 files = [ln.split("|") for ln in listing.strip().splitlines() if "|" in ln]94 for path, sha in files:95 p = dest / path96 if not p.exists():97 missing += 198 continue99 h = hashlib.sha256(p.read_bytes()).hexdigest()100 verified += int(h == sha)101 subprocess.run(["bash", "-c", 'psql "$FD_PG_ADMIN" -qc "DROP DATABASE \"$(basename "$FD_PG_SCRATCH")\""'], env=env, timeout=120, capture_output=True)102 return {"pg_dump": latest, "restored_lake_files_in_db": out, "files_listed": len(files),103 "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)}104