#!/usr/bin/env python3 """Parquet mirror sync (Stage-0 warehouse, replaces the nightly bulk ingest). Pulls NEW parquet files the recorder's fold sidecar produced (/data/parquet on the wwf-recorder volume) into live/parquet/, row-verifies each against the box manifest, appends its rows into live/rtds.duckdb's NATIVE tables (same tables as ever — research and tape_sharps keep native-table speed; views-over-parquet was rejected: sim's per-asset point queries would turn minutes into hours), marks it in `ingested`, and ACKs it back (touch /data/parquet/acks/.ok — the fold disk-guard may only ever delete ACKED parquet). The mirror itself is the durable warehouse layer Stage 1 (MotherDuck/ClickHouse) would consume. Transport is the battle-tested flyctl sftp get (+ ssh console for acks); no ingress on the recorder, by design. Quiet no-op when the box is unreachable or another process holds the db write lock — files are immutable and manifest-driven, so the next run catches up. Runs every 15 min via com.jaxperro.tape-sync and from daily.sh (old bulk ingest.py kept as fallback for a fold-less recorder). """ import json import os import shutil import subprocess import sys import time HERE = os.path.dirname(os.path.abspath(__file__)) LIVE = os.path.join(HERE, "..", "live") MIRROR = os.path.join(LIVE, "parquet") LOCAL_MANIFEST = os.path.join(MIRROR, ".mirror_manifest.jsonl") DB = os.path.join(LIVE, "rtds.duckdb") APP, PQ = "wwf-recorder", "/data/parquet" FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl" def box(cmd, timeout=120): r = subprocess.run([FLYCTL, "ssh", "console", "-a", APP, "-C", f"bash -c '{cmd}'"], capture_output=True, text=True, timeout=timeout, stdin=subprocess.DEVNULL) return r.stdout def sftp_get(remote, local, timeout=240): """sftp first (a healthy multi-MB fetch takes <60s — the old 900s window just burned the 15-min cadence when it hung), then the base64-over- console fallback that carried ingest.py through the same flakiness (2026-07-22: sftp timeouts left magic-byte-less partial parquets while console commands stayed fast). Integrity is the caller's manifest row-count verify either way; partial files are removed here.""" timed_out = False try: subprocess.run([FLYCTL, "ssh", "sftp", "get", remote, local, "-a", APP], capture_output=True, timeout=timeout, stdin=subprocess.DEVNULL) except subprocess.TimeoutExpired: timed_out = True # a timed-out get leaves a size>0 PARTIAL — must # not be blessed, or the base64 fallback never # runs and the file loops unreadable forever # (2026-07-23: 8h of exactly that) if not timed_out and os.path.exists(local) and os.path.getsize(local) > 0: return True try: os.remove(local) except OSError: pass try: import base64 raw = box(f"base64 {remote}", timeout=600) blob = base64.b64decode(raw) if not blob: return False with open(local, "wb") as fh: fh.write(blob) print(f"[sync] {os.path.basename(remote)}: sftp failed — " f"base64 fallback carried {len(blob)//1024}KB") return True except Exception as e: print(f"[sync] {os.path.basename(remote)}: both transports failed " f"({type(e).__name__})") try: os.remove(local) except OSError: pass return False def load_manifest(path): out = {} try: with open(path) as fh: for ln in fh: try: d = json.loads(ln) out[d["segment"]] = d except Exception: pass except FileNotFoundError: pass return out def connect_rw(duckdb, tries=3, wait=10): """The 15-min cadence can collide with a long research read (one-writer- OR-many-readers). Retry briefly, then yield to the next run.""" for i in range(tries): try: return duckdb.connect(DB) except Exception as e: if i == tries - 1: print(f"[sync] db locked ({str(e)[:60]}) — yielding to next run") return None time.sleep(wait) def main(): os.makedirs(MIRROR, exist_ok=True) tmp_man = os.path.join(MIRROR, ".remote_manifest.tmp") if not sftp_get(f"{PQ}/manifest.jsonl", tmp_man, timeout=120): print("[sync] box unreachable or no manifest yet — nothing to do") return 0 remote = load_manifest(tmp_man) local = load_manifest(LOCAL_MANIFEST) new = [d for s, d in sorted(remote.items()) if s not in local] os.remove(tmp_man) if not new: print(f"[sync] mirror current ({len(local)} files)") return 0 import duckdb con = connect_rw(duckdb) if con is None: return 0 have = {r[0] for r in con.execute("SELECT segment FROM ingested").fetchall()} got = 0 pending_acks = [] for d in new: rel, seg = d["path"], d["segment"] dst = os.path.join(MIRROR, rel) os.makedirs(os.path.dirname(dst), exist_ok=True) if not os.path.exists(dst) and not sftp_get(f"{PQ}/{rel}", dst): print(f"[sync] {rel}: fetch failed — retry next run") continue try: n, = con.execute( f"SELECT count(*) FROM read_parquet('{dst}')").fetchone() except Exception as e: print(f"[sync] {rel}: unreadable ({e}) — dropped, retry next run") try: os.remove(dst) except OSError: pass continue if n != d["rows"]: print(f"[sync] {rel}: rows {n} != manifest {d['rows']} — dropped") os.remove(dst) continue # append into the native tables exactly once (segment-keyed), then # mirror-manifest + ack. A crash between COMMIT and manifest write # re-runs into the `have` guard — no double insert. if seg not in have: con.execute("BEGIN") if d["family"] == "aux": con.execute(f"""INSERT INTO aux SELECT ts, topic, type, payload FROM read_parquet('{dst}')""") else: con.execute(f"""INSERT INTO trades SELECT ts, wallet, asset, cond, side, price, size, tx, title FROM read_parquet('{dst}')""") con.execute("INSERT INTO ingested VALUES (?,?,?)", [seg, d["rows"], int(time.time())]) con.execute("COMMIT") with open(LOCAL_MANIFEST, "a") as fh: fh.write(json.dumps(d) + "\n") # ACK IS DEFERRED AND NON-FATAL (2026-07-27): one ssh round-trip per # file rides WireGuard, which travel networks throttle to the point # of timing out — and a raised TimeoutExpired used to kill the whole # sync mid-backlog (52 files in, 12 to go). The invariant is # one-directional: the box may delete ONLY what the Mac acked, so a # MISSING ack is always safe (box keeps the file; 25GB ≈ 3+ weeks). # Batch them at the end, best-effort. pending_acks.append(os.path.basename(rel)) got += 1 print(f"[sync] {rel}: {n} rows -> db + mirror (ack queued)") # batched ack: ONE ssh round-trip for the whole run (was one per file). # Never fatal — unacked files simply wait for a healthier network; the # next run re-acks them (touch is idempotent, and `have` keeps the DB # from re-ingesting). if pending_acks: try: box("touch " + " ".join(f"{PQ}/acks/{a}.ok" for a in pending_acks), timeout=180) print(f"[sync] acked {len(pending_acks)} file(s) in one call") except Exception as e: print(f"[sync] ⚠ ack deferred ({type(e).__name__}) — box keeps " f"{len(pending_acks)} file(s) until a healthier run; " f"data already in db + mirror") n, = con.execute("SELECT count(*) FROM trades").fetchone() print(f"[sync] +{got}/{len(new)} files · rtds.duckdb now {n:,} trades") return 0 if __name__ == "__main__": sys.exit(main())