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 transaction13from findata_ingest.lake import Lake141516def _env() -> dict[str, str]:17 env = dict(os.environ)18 env["PATH"] = "/opt/homebrew/opt/postgresql@17/bin:/opt/homebrew/bin:/usr/bin:/bin:/usr/sbin:/sbin"19 return env202122def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:23 s = get_settings()24 target = (job.get("params") or {}).get("target") or s.backup_target25 if not target:26 raise RuntimeError("FINDATA_BACKUP_TARGET not configured")27 host, _, remote_dir = target.partition(":")28 stamp = datetime.now(UTC).strftime("%Y%m%dT%H%M%SZ")29 dumps = lake.root / "_pgdump"30 dumps.mkdir(exist_ok=True)31 dump = dumps / f"findata-{stamp}.sql.zst"32 url = s.database_url.get_secret_value()33 cmd = f"pg_dump --no-owner --no-privileges '{url}' | zstd -q -T0 -o '{dump}'"34 r = subprocess.run(["bash", "-c", cmd], capture_output=True, text=True, env=_env(), timeout=3600)35 if r.returncode != 0:36 raise RuntimeError(f"pg_dump failed: {r.stderr[-500:]}")37 for old in sorted(dumps.glob("findata-*.sql.zst"))[:-14]:38 old.unlink()39 subprocess.run(["ssh", "-o", "BatchMode=yes", host, f"mkdir -p {remote_dir}/lake {remote_dir}/pgdump"], check=True, timeout=60, env=_env())40 excludes = ["--exclude", "_tmp/", "--exclude", "_retired/", "--exclude", "exports/", "--exclude", ".writer.lock"]41 rs = subprocess.run(["rsync", "-a", "--delete", "--stats", *excludes, f"{lake.root}/", f"{host}:{remote_dir}/lake/"],42 capture_output=True, text=True, env=_env(), timeout=6 * 3600)43 if rs.returncode != 0:44 raise RuntimeError(f"rsync failed: {rs.stderr[-500:]}")45 rs2 = subprocess.run(["rsync", "-a", f"{dumps}/", f"{host}:{remote_dir}/pgdump/"], capture_output=True, text=True, env=_env(), timeout=3600)46 if rs2.returncode != 0:47 raise RuntimeError(f"rsync pgdump failed: {rs2.stderr[-500:]}")48 # verification: remote file count + bytes of curated/49 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"],50 capture_output=True, text=True, env=_env(), timeout=600)51 remote_files, remote_kb = [x.strip() for x in v.stdout.strip().splitlines()[:2]] if v.returncode == 0 else ("?", "?")52 local_files = sum(1 for p in (lake.root / "curated").rglob("*") if p.is_file())53 stats = {"at": stamp, "target": target, "pg_dump": dump.name, "pg_dump_bytes": dump.stat().st_size,54 "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],55 "remote_curated_files": remote_files, "local_curated_files": local_files, "remote_curated_kb": remote_kb,56 "verified": str(local_files) == str(remote_files)}57 with transaction() as conn:58 conn.execute("INSERT INTO settings_kv(key, value) VALUES ('last_backup', %s) ON CONFLICT (key) DO UPDATE SET value=EXCLUDED.value, updated_at=now()",59 (__import__("json").dumps(stats, default=str),))60 log.info("backup.done", **{k: v for k, v in stats.items() if k != "rsync_summary"})61 return stats626364def restore_drill(target: str, dest: Path, dataset_id: str = "market_bars.stock.1m.raw") -> dict[str, Any]:65 """Restore the latest pg_dump into a scratch database and the lake into `dest`, then verify manifest hashes."""66 import hashlib67 import json68 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 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/lake/manifests/{dataset_id}/", f"{dest}/manifests/{dataset_id}/"], check=True, env=_env(), timeout=600)73 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()74 subprocess.run(["rsync", "-a", f"{host}:{remote_dir}/pgdump/{latest}", f"{dest}/"], check=True, env=_env(), timeout=600)75 base_url = s.database_url.get_secret_value().rsplit("/", 1)[0]76 scratch = f"findata_restore_{datetime.now(UTC).strftime('%H%M%S')}"77 subprocess.run(["bash", "-c", f"psql '{base_url}/postgres' -qc 'CREATE DATABASE {scratch}' && zstd -dc '{dest}/{latest}' | psql -q '{base_url}/{scratch}' >/dev/null"],78 check=True, env=_env(), timeout=3600)79 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\""],80 capture_output=True, text=True, env=_env(), timeout=120).stdout.strip()81 manifests = sorted((dest / "manifests" / dataset_id).glob("v*.json"), key=lambda p: int(p.stem[1:]))82 verified = 083 missing = 084 if manifests:85 m = json.loads(manifests[-1].read_text())86 for f in m["files"]:87 p = dest / f["path"]88 if not p.exists():89 missing += 190 continue91 h = hashlib.sha256(p.read_bytes()).hexdigest()92 verified += int(h == f["sha256"])93 subprocess.run(["bash", "-c", f"psql '{base_url}/postgres' -qc 'DROP DATABASE {scratch}'"], env=_env(), timeout=120)94 return {"pg_dump": latest, "restored_lake_files_in_db": out, "manifest_version": manifests[-1].stem if manifests else None,95 "files_hash_verified": verified, "files_missing": missing, "dest": str(dest)}96