From 50324b4d7a5b8c8e019fd21753f0063bcb88cea7 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 25 May 2026 22:21:45 -0400 Subject: [PATCH] =?UTF-8?q?#000061:=20pack=20format=20v2=20=E2=80=94=20sel?= =?UTF-8?q?f-sufficient=20new-peer=20hydration?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit v1 packs (chunks-only) were under-engineered: a new peer landing on v1 packs would have chunk bodies indexed by leaf_hash but no documents table, no audit chain, no merkle interior, no edges — couldn't actually hydrate. fox: "isn't what I wanted you under engineered..." v2 packs ship every load-bearing shard table alongside chunk bodies in the same tar.zst: manifest.jsonl # chunk catalog (unchanged) tables/documents.jsonl # array-per-line columnar JSONL tables/chunks.jsonl # without content column tables/merkle_nodes.jsonl tables/edges.jsonl # FAN-IN restructured tables/audit_events.jsonl tables/derivations.jsonl tables/concept_relations.jsonl tables/concept_token_idf.jsonl tables/providence_cache.jsonl tables/citation_aliases.jsonl tables/term_aliases.jsonl tables/snapshots.jsonl tables/document_http_meta.jsonl blobs// # raw UTF-8 chunk bodies Two compression strategies inside the pack: 1. Array-per-line JSONL ({"_columns": [...]} header line + ["v1","v2",...] data lines) drops ~30% of uncompressed bytes vs object-per-row JSONL. zstd recovers most of that on its own, but smaller uncompressed footprint also speeds up stream-restore. 2. Edges fan-in restructure at pack-build time: 22M rows of (src_root, edge_type, dst_root, dst_uri, anchor) → ~500k unique (dst_uri, edge_type, anchor, dst_root) groups with src_roots as an array. ~5-10x compressed savings on the dominant table. Reverses on unpack into the per-edge live schema. Live queries unchanged. NOT shipped (per-peer state): mesh_*, selfmodel_*, capital_ledger, memory_*, controller_events, fork_score_branches, adapter_loss_reports, falsifications, schema_meta, meta. NOT shipped (rebuildable): chunks_fts*, documents_fts* — restored from chunks.content + documents.title on unpack. push_pack no longer appends `cold_pack_pushed` to the audit chain. That event leaked into the next push's audit_events.jsonl dump and broke the "two writers at the same corpus state produce identical pack_hash" determinism property. The bucket/disc file IS the receipt; the snapshot_root pinned inside the pack metadata binds it to a corpus state. No load-bearing consumer of the audit row. pull_pack restored to handle both v1 (chunks-only) and v2 (tables + chunks) packs. For v2 it extracts tables/*.jsonl to a temp dir, calls restore_shard_metadata (which INSERT OR IGNOREs into the live schema and expands edges back to per-edge rows), then fills chunk content for every leaf_hash in blobs/. Idempotent against populated DBs (INSERT OR IGNORE all the way down). Self-cleaning temp dir. Sizing measured 2026-05-26: ~2.1 GB per shard pack compressed (chunk content 1.78 GB + metadata ~0.3 GB), ~8.5 GB total across 4 shards. ~20% more than v1 chunks-only for self-sufficient hydration. 24 cold-object + evict tests pass (+1 new test_push_pack_v2_hydrates_fresh_empty_db that builds a pack from a populated DB and unpacks into a completely empty DB to verify all tables restored). Full suite: 2558 passed, 28 skipped, 1 xfailed. --- arborist/cold_object.py | 27 ++ arborist/cold_pack_metadata.py | 355 ++++++++++++++++++ arborist/evict.py | 168 +++++++-- docs/cold-object-store.md | 51 ++- .../ticket-000061-cold-object-store-tier.md | 31 +- tests/test_cold_object.py | 75 ++++ 6 files changed, 654 insertions(+), 53 deletions(-) create mode 100644 arborist/cold_pack_metadata.py diff --git a/arborist/cold_object.py b/arborist/cold_object.py index e9054b3..f198499 100644 --- a/arborist/cold_object.py +++ b/arborist/cold_object.py @@ -470,6 +470,7 @@ def stream_packs( max_compressed_bytes: int, level: int = 3, work_dir: Path | str | None = None, + extra_members: list[tuple[str, Path]] | None = None, ) -> Iterator[FilePack]: """Stream (leaf_hash, raw_utf8_body) chunks into FILE-BACKED packs that each fit in `max_compressed_bytes` compressed bytes. @@ -503,11 +504,20 @@ def stream_packs( Single chunks larger than the cap still get their own pack (we don't drop data — the alternative is silently losing it). + + `extra_members` is a list of (tar_member_name, source_file_path) for + files to include AT THE START OF THE FIRST PACK. Used by pack format + v2 to ship shard-metadata table dumps (`tables/.jsonl`) ahead + of the chunk bodies. Subsequent packs (if the chunk set overflows the + cap) do not repeat the extra members — they're conceptually shard- + level state, not chunk-level. If the v1 chunks-only pack shape is + needed, pass `extra_members=None` (default). """ if max_compressed_bytes <= 0: raise ValueError("max_compressed_bytes must be positive") work_path = Path(work_dir) if work_dir is not None else Path(tempfile.gettempdir()) work_path.mkdir(parents=True, exist_ok=True) + pending_extras = list(extra_members) if extra_members else [] def _start_pack() -> dict: # Tempfile in work_dir, opened for binary write. We hand the raw @@ -560,6 +570,23 @@ def stream_packs( state = _start_pack() try: + # Pack format v2: ship the metadata table dumps as the FIRST tar + # members of the first pack. Add them up front so they're tarred + # before chunks; if the pack overflows the cap mid-chunk, the + # metadata is already safely in pack 1 and pack 2 onward holds + # only the chunk overflow. + for member_name, source_path in pending_extras: + with open(source_path, "rb") as f: + source_bytes = f.read() + info = tarfile.TarInfo(name=member_name) + info.size = len(source_bytes) + state["tar"].addfile(info, io.BytesIO(source_bytes)) + # Force the compressor to flush so size accounting reflects the + # metadata write before we start streaming chunks. + if pending_extras: + state["writer"].flush(zstandard.FLUSH_BLOCK) + state["file_obj"].flush() + for leaf_hash, body in chunks: if len(leaf_hash) != 64: raise ValueError(f"bad leaf_hash length: {leaf_hash}") diff --git a/arborist/cold_pack_metadata.py b/arborist/cold_pack_metadata.py new file mode 100644 index 0000000..5334720 --- /dev/null +++ b/arborist/cold_pack_metadata.py @@ -0,0 +1,355 @@ +"""Shard metadata dump/restore for pack format v2 (#000061 redesign). + +Pack v2 ships every load-bearing table from a shard so a new peer can +hydrate from packs alone. Two compression strategies: + +1. **Array-per-line JSONL** for all wide tables: column names declared + once at the top of the file, each subsequent line is a JSON array + of values in column order. Drops ~30% of uncompressed bytes vs + `{"col": "val"}` objects (zstd recovers most of that on its own, but + the smaller uncompressed footprint also helps stream-restore speed). + +2. **Fan-in restructure for `edges`** at pack-build time. The live + schema is one row per (src_root, dst_root, dst_uri, edge_type, anchor) + — a fan-out view. At pack time we group by (dst_uri, edge_type, + anchor, dst_root) and ship the list of `src_roots` per destination. + 22M rows per shard → ~500k unique destinations → ~5-10x compressed + savings. Reverses on unpack: each fan-in row expands back into N + live-schema edges. Live SQLite queries unchanged. + +Per-peer state (`mesh_*`, `selfmodel_*`, `capital_ledger`, `memory_*`, +`controller_events`, `fork_score_branches`, `adapter_loss_reports`, +`falsifications`, `schema_meta`, `meta`) is NOT shipped — those are local +to each peer and the new peer initializes them on first run. + +FTS5 tables (`chunks_fts*`, `documents_fts*`) are NOT shipped either — +they're rebuilt from `chunks.content` + `documents.title` on unpack. +""" + +from __future__ import annotations + +import itertools +import json +import sqlite3 +from pathlib import Path +from typing import Iterable, Iterator + +# Tables that go in the pack. Order matters for restore (foreign-key +# dependencies); list them in topological order so chunks insert after +# documents, edges after documents, etc. derivations references both +# core_root and src_root → after documents. +SHIPPED_TABLES: tuple[str, ...] = ( + "documents", + "document_http_meta", + "chunks", # content column dropped at dump time + "merkle_nodes", + "edges", # fan-in restructured at dump, expanded at restore + "derivations", + "concept_relations", + "concept_token_idf", + "providence_cache", + "citation_aliases", + "term_aliases", + "audit_events", # last — the chain integrity check happens after rows land + "snapshots", +) + +# Per-peer / rebuildable / not shipped. +EXCLUDED_TABLES: frozenset[str] = frozenset({ + "chunks_fts", "chunks_fts_data", "chunks_fts_idx", + "chunks_fts_content", "chunks_fts_docsize", "chunks_fts_config", + "documents_fts", "documents_fts_data", "documents_fts_idx", + "documents_fts_content", "documents_fts_docsize", "documents_fts_config", + "mesh_identity", "mesh_roster", "mesh_epochs", "mesh_peer_chains", + "selfmodel_records", "selfmodel_capability_claims", + "capital_ledger", "controller_events", "fork_score_branches", + "memory_records", "memory_branch_summaries", + "adapter_loss_reports", "falsifications", + "schema_meta", "meta", "sqlite_sequence", +}) + +# Columns dropped at dump time (lives elsewhere in the pack or rebuildable). +COLUMN_FILTER: dict[str, frozenset[str]] = { + "chunks": frozenset({"content"}), # chunk bodies live in blobs/ tar members +} + + +# --------------------------------------------------------------------------- +# Generic array-per-line JSONL helpers +# --------------------------------------------------------------------------- + + +def _encode_value(v: object) -> object: + """Coerce sqlite3 cell values to JSON-safe primitives. + + Binary blobs (audit body, mesh envelopes) → base64 with a 'b64:' tag + so the decoder knows to reverse. Plain str / int / float / None pass + through. + """ + if isinstance(v, (bytes, bytearray, memoryview)): + import base64 + return "b64:" + base64.b64encode(bytes(v)).decode("ascii") + return v + + +def _decode_value(v: object) -> object: + """Reverse `_encode_value`: 'b64:...' strings → raw bytes.""" + if isinstance(v, str) and v.startswith("b64:"): + import base64 + return base64.b64decode(v[4:]) + return v + + +def write_columnar_jsonl( + out_path: Path, + columns: list[str], + rows: Iterable[tuple], +) -> int: + """Write rows as array-per-line JSONL with a header declaring columns. + + Format (every line is independently parseable JSON): + {"_columns": ["col1", "col2", "col3"]} + ["val1", "val2", 42] + ["val4", "val5", 99] + ... + + Returns the number of data rows written (not counting the header). + """ + n = 0 + with open(out_path, "w", encoding="utf-8") as f: + f.write(json.dumps({"_columns": columns}, sort_keys=True) + "\n") + for row in rows: + encoded = [_encode_value(v) for v in row] + f.write(json.dumps(encoded, ensure_ascii=False, separators=(",", ":")) + "\n") + n += 1 + return n + + +def read_columnar_jsonl(in_path: Path) -> Iterator[dict]: + """Stream rows as dicts. First line is the column header; subsequent + lines are arrays. Returns one dict per data row, bytes-decoded.""" + with open(in_path, "r", encoding="utf-8") as f: + header = json.loads(f.readline()) + if not isinstance(header, dict) or "_columns" not in header: + raise ValueError(f"missing or malformed columnar header in {in_path}") + cols = header["_columns"] + for line in f: + line = line.rstrip("\n") + if not line: + continue + values = json.loads(line) + yield {cols[i]: _decode_value(values[i]) for i in range(len(cols))} + + +# --------------------------------------------------------------------------- +# Per-table dump (sqlite → JSONL on disk) +# --------------------------------------------------------------------------- + + +def _table_columns(conn: sqlite3.Connection, table: str) -> list[str]: + return [r[1] for r in conn.execute(f"PRAGMA table_info({table})").fetchall()] + + +def _select_cols(conn: sqlite3.Connection, table: str) -> list[str]: + cols = _table_columns(conn, table) + drop = COLUMN_FILTER.get(table, frozenset()) + return [c for c in cols if c not in drop] + + +def _dump_generic_table( + conn: sqlite3.Connection, + table: str, + out_path: Path, +) -> int: + """Stream every row of `table` to `out_path` as array-per-line JSONL. + + Stable ORDER BY rowid (or PK if WITHOUT ROWID) so two writers at the + same snapshot produce byte-identical output. Memory bounded by + cursor iteration — never fetchall(). + """ + select_cols = _select_cols(conn, table) + col_list = ", ".join(f'"{c}"' for c in select_cols) + # ORDER BY all selected columns guarantees determinism for tables + # without a single-column PK (composite-PK tables already encoded + # in column order). Cheap for SQLite's planner since PKs are usually + # in the natural sort order. + order_by = ", ".join(f'"{c}"' for c in select_cols) + cursor = conn.execute(f"SELECT {col_list} FROM {table} ORDER BY {order_by}") + return write_columnar_jsonl(out_path, select_cols, cursor) + + +# --------------------------------------------------------------------------- +# Edges fan-in restructure (the architectural fix) +# --------------------------------------------------------------------------- + + +# The live schema has one row per (src_root, edge_type, dst_root, dst_uri, anchor). +# At pack time we group by (dst_uri, edge_type, anchor, dst_root) and store +# src_roots as an array — typical 5-10x reduction on uncompressed bytes +# because many src documents link to the same dst URI. +_EDGES_FAN_IN_COLUMNS = ["dst_uri", "edge_type", "anchor", "dst_root", "src_roots"] + + +def _dump_edges_fan_in(conn: sqlite3.Connection, out_path: Path) -> int: + """Group edges by destination, store src_roots as an array per row. + + Streams rows from SQLite in (dst_uri, edge_type, anchor, dst_root) + sorted order, then groups in Python with itertools.groupby. Both + sides are streaming, so memory stays bounded regardless of edge count. + + Returns the number of fan-in rows written. + """ + # ORDER BY the group key first so itertools.groupby works correctly. + cursor = conn.execute( + "SELECT dst_uri, edge_type, anchor, dst_root, src_root " + "FROM edges " + "ORDER BY dst_uri, edge_type, anchor, dst_root, src_root" + ) + n = 0 + with open(out_path, "w", encoding="utf-8") as f: + f.write(json.dumps({"_columns": _EDGES_FAN_IN_COLUMNS}, sort_keys=True) + "\n") + for key, group in itertools.groupby( + cursor, key=lambda r: (r[0], r[1], r[2], r[3]) + ): + dst_uri, edge_type, anchor, dst_root = key + src_roots = [row[4] for row in group] + row_arr = [dst_uri, edge_type, anchor, dst_root, src_roots] + f.write(json.dumps(row_arr, ensure_ascii=False, separators=(",", ":")) + "\n") + n += 1 + return n + + +def _restore_edges_fan_out(conn: sqlite3.Connection, in_path: Path) -> int: + """Expand fan-in rows back into one edges row per src_root. + + INSERT OR IGNORE so re-running over an existing edges table doesn't + error on duplicate-PK collisions. Batched executemany() for + throughput. + """ + BATCH = 5000 + batch: list[tuple] = [] + n = 0 + for fan_in in read_columnar_jsonl(in_path): + dst_uri = fan_in["dst_uri"] + edge_type = fan_in["edge_type"] + anchor = fan_in["anchor"] + dst_root = fan_in["dst_root"] + for src_root in fan_in["src_roots"]: + batch.append((src_root, dst_root, dst_uri, edge_type, anchor)) + n += 1 + if len(batch) >= BATCH: + conn.executemany( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) " + "VALUES (?, ?, ?, ?, ?)", + batch, + ) + batch.clear() + if batch: + conn.executemany( + "INSERT OR IGNORE INTO edges " + "(src_root, dst_root, dst_uri, edge_type, anchor) " + "VALUES (?, ?, ?, ?, ?)", + batch, + ) + return n + + +# --------------------------------------------------------------------------- +# Public dump / restore entry points +# --------------------------------------------------------------------------- + + +def dump_shard_metadata( + conn: sqlite3.Connection, + out_dir: Path, +) -> dict[str, dict]: + """Dump every shipped table to `out_dir/.jsonl`. + + Returns a per-table summary {table: {rows: N, bytes: M, path: Path}} + that the pack builder uses to construct the manifest. Tables with + zero rows are still written (just the header line) so the consumer + can detect "empty by design" vs "missing from pack." + """ + out_dir.mkdir(parents=True, exist_ok=True) + summary: dict[str, dict] = {} + + # Discover which shipped tables actually exist in this schema (older + # shards may not have all of them). + existing = { + r[0] + for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ).fetchall() + } + + for table in SHIPPED_TABLES: + if table not in existing: + continue + out_path = out_dir / f"{table}.jsonl" + if table == "edges": + rows = _dump_edges_fan_in(conn, out_path) + else: + rows = _dump_generic_table(conn, table, out_path) + summary[table] = { + "rows": rows, + "bytes": out_path.stat().st_size, + "path": out_path, + } + return summary + + +def restore_shard_metadata( + conn: sqlite3.Connection, + table_dir: Path, +) -> dict[str, int]: + """Restore every shipped-table file under `table_dir` into the + connection's schema. Tables are inserted in `SHIPPED_TABLES` order + so foreign-key dependencies are satisfied. + + The schema must already be initialized (via `arborist.store.connect`) + — this function only inserts rows, doesn't CREATE TABLE. + + Returns {table: rows_inserted}. + """ + result: dict[str, int] = {} + for table in SHIPPED_TABLES: + in_path = table_dir / f"{table}.jsonl" + if not in_path.exists(): + continue + if table == "edges": + n = _restore_edges_fan_out(conn, in_path) + else: + n = _restore_generic_table(conn, table, in_path) + result[table] = n + conn.commit() + return result + + +def _restore_generic_table( + conn: sqlite3.Connection, + table: str, + in_path: Path, +) -> int: + """Read array-per-line JSONL → INSERT OR IGNORE batches into `table`.""" + BATCH = 5000 + batch: list[tuple] = [] + cols: list[str] | None = None + n = 0 + insert_sql: str | None = None + for row in read_columnar_jsonl(in_path): + if cols is None: + cols = list(row.keys()) + placeholders = ", ".join("?" for _ in cols) + col_list = ", ".join(f'"{c}"' for c in cols) + insert_sql = ( + f"INSERT OR IGNORE INTO {table} ({col_list}) VALUES ({placeholders})" + ) + batch.append(tuple(row[c] for c in cols)) + n += 1 + if len(batch) >= BATCH: + conn.executemany(insert_sql, batch) + batch.clear() + if batch and insert_sql is not None: + conn.executemany(insert_sql, batch) + return n diff --git a/arborist/evict.py b/arborist/evict.py index 01c1764..700bdf8 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -311,9 +311,11 @@ def push_pack( uncompressed_bytes, compressed_bytes, ...}]`. Idempotent: same chunk grouping → same pack_hashes → bucket overwrite is a no-op. """ + import tempfile from pathlib import Path from arborist.cold_object import stream_packs + from arborist.cold_pack_metadata import dump_shard_metadata from arborist.snapshot import compute_snapshot_root # Default selection = every chunk with local content (surfaces AND @@ -400,7 +402,23 @@ def push_pack( # upload path is also memory-bounded. import shutil - for pack in stream_packs(_chunk_source(), max_compressed_bytes=max_pack_bytes): + # Pack format v2: dump every shipped table to JSONL in a temp dir, then + # include those files as `tables/.jsonl` tar members at the front + # of the first pack. Edges goes through fan-in restructure inside + # `dump_shard_metadata` for ~5-10x compression vs the live-schema row + # layout. Lives only long enough to be tarred + uploaded, then deleted. + metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-")) + metadata_summary = dump_shard_metadata(conn, metadata_dir) + extra_members = [ + (f"tables/{name}.jsonl", info["path"]) + for name, info in metadata_summary.items() + ] + + for pack in stream_packs( + _chunk_source(), + max_compressed_bytes=max_pack_bytes, + extra_members=extra_members, + ): pack_uncompressed = uncompressed_running[0] - last_finalized_uncompressed[0] last_finalized_uncompressed[0] = uncompressed_running[0] try: @@ -415,31 +433,19 @@ def push_pack( (out_dir / f"arborist-pack-{short}.manifest.ndjson").write_bytes( pack.manifest_bytes ) - with transaction(conn): - event_hash = append_audit( - conn, - event_type="cold_pack_pushed", - subject_root=document_root, - body={ - "pack_hash": pack.pack_hash, - "chunk_count": len(pack.entries), - "uncompressed_bytes": pack_uncompressed, - "compressed_bytes": pack.body_size, - "snapshot_root": snapshot_root, - "snapshot_doc_count": doc_count_at_pack, - "snapshot_ts": pack_ts, - "pushed_to_bucket": push_to_bucket, - "local_dir": str(out_dir) if out_dir else None, - "backend": backend_id if push_to_bucket else None, - }, - ) + # Deliberately NO audit_append for the push itself. The pack is + # its own receipt (content-addressed file + snapshot_root binding + # inside the pack's metadata). An audit row here would leak into + # the next push's audit_events.jsonl dump and break the + # "two writers at the same corpus state produce the same pack_hash" + # determinism property. The bucket / disc IS the audit trail for + # the push. pack_results.append({ "pack_hash": pack.pack_hash, "chunk_count": len(pack.entries), "uncompressed_bytes": pack_uncompressed, "compressed_bytes": pack.body_size, "snapshot_root": snapshot_root, - "audit_event_hash": event_hash, }) finally: # Clean up the temp file when neither local_dir nor an exception @@ -454,6 +460,13 @@ def push_pack( skipped_hash_mismatch = nonlocal_skipped[0] + # Clean up the temp dir that held the table JSONL dumps. The tar + # already inlined their bytes; the on-disk copies are no longer needed. + try: + shutil.rmtree(metadata_dir, ignore_errors=True) + except Exception: + pass + if not pack_results: return {"status": "nothing_to_pack", "packs": []} @@ -478,47 +491,120 @@ def pull_pack( backend: "ObjectStoreBackend", pack_hash: str, ) -> dict: - """Pull one pack from the bucket, verify each chunk's hash, restore any - cold chunks whose leaf_hash appears in the pack. + """Pull one pack from the bucket and restore everything it contains: + metadata tables (pack v2) + chunk bodies. - Chunks not currently in the local DB are NOT inserted — packs only - rehydrate previously-ingested chunks. Use the corpus snapshot + ingest - pipeline to bring in genuinely-new documents. + Order matters: tables go first (so `chunks` rows exist before we try + to fill their content), then chunk bodies fill `chunks.content` and + populate `chunks_fts`. INSERT OR IGNORE on table restore means + re-pulling a pack against a populated DB is idempotent. - Returns count restored and count skipped (already hot / unknown locally). + v1 packs (chunks-only, no `tables/` entries) still work — the metadata + restore step becomes a no-op when no table files are found in the tar. """ - from arborist.cold_object import open_pack + import io + import shutil + import tarfile + import tempfile + from pathlib import Path + import zstandard + + from arborist.cold_pack_metadata import restore_shard_metadata + from arborist.merkle import hash_leaf + + backend_id = backend.identity.to_audit_body() body = backend.get_pack(pack_hash) + + # Streaming decompression — packs built by stream_packs lack the + # content-size frame header so the one-shot ZstdDecompressor.decompress + # path errors. stream_reader works with either frame header presence. + dctx = zstandard.ZstdDecompressor() + raw_tar = dctx.stream_reader(io.BytesIO(body)).read() + + # Extract metadata table files to a temp dir; collect chunk bodies as + # we go. The tar can contain both kinds; we branch on member name. + tables_dir = Path(tempfile.mkdtemp(prefix="arborist-restore-")) + chunks_to_restore: list[tuple[str, bytes]] = [] + table_files_seen: list[str] = [] + try: + with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar: + for member in tar: + if member.name == "manifest.ndjson" or member.name == "manifest.jsonl": + continue # informational header, parsed elsewhere if needed + if member.name.startswith("tables/"): + # tables/.jsonl — extract to temp dir for + # restore_shard_metadata to read line-by-line. + f = tar.extractfile(member) + if f is None: + continue + dest = tables_dir / Path(member.name).name + with open(dest, "wb") as out: + out.write(f.read()) + table_files_seen.append(member.name) + elif member.name.startswith("blobs/"): + tail = member.name[len("blobs/"):] + if "/" not in tail: + continue + prefix, rest = tail.split("/", 1) + if len(prefix) != 2 or len(rest) != 62: + continue + expected_hash = prefix + rest + f = tar.extractfile(member) + if f is None: + continue + chunk_body = f.read() + actual_hash = hash_leaf(chunk_body).hex() + if actual_hash != expected_hash: + raise ValueError( + f"pack chunk hash mismatch: declared " + f"{expected_hash}, actual {actual_hash}" + ) + chunks_to_restore.append((expected_hash, chunk_body)) + + # Phase 1: restore metadata tables (no-op for v1 packs). + tables_restored: dict[str, int] = {} + if table_files_seen: + tables_restored = restore_shard_metadata(conn, tables_dir) + finally: + shutil.rmtree(tables_dir, ignore_errors=True) + + # Phase 2: fill chunk content for every leaf_hash in the pack. restored = 0 skipped_unknown = 0 skipped_already_hot = 0 - drifted = 0 - drift_details: list[dict] = [] - backend_id = backend.identity.to_audit_body() - - for leaf_hash, chunk_body in open_pack(body): - # open_pack already verifies sha256(body) == declared leaf_hash; - # tampering raises ValueError before we get here. + for leaf_hash, chunk_body in chunks_to_restore: rows = conn.execute( "SELECT chunk_id, tier FROM chunks WHERE leaf_hash = ?", (leaf_hash,), ).fetchall() if not rows: + # chunks_meta.jsonl restore should have created these rows; + # if we still don't see them, the pack is internally + # inconsistent (chunk body without metadata claim). skipped_unknown += 1 continue text = chunk_body.decode("utf-8") for row in rows: - if row["tier"] == "hot": - skipped_already_hot += 1 - continue + if row["tier"] == "hot" and row["chunk_id"] is not None: + # Already populated — but content may still be NULL if + # this row came from chunks_meta.jsonl, which carries no + # content. Check explicitly. + existing = conn.execute( + "SELECT content IS NOT NULL FROM chunks WHERE chunk_id = ?", + (row["chunk_id"],), + ).fetchone() + if existing and existing[0]: + skipped_already_hot += 1 + continue with transaction(conn): conn.execute( "UPDATE chunks SET content = ?, tier = 'hot' WHERE chunk_id = ?", (pack_chunk(text), row["chunk_id"]), ) conn.execute( - "INSERT INTO chunks_fts (rowid, content) VALUES (?, ?)", + "INSERT OR REPLACE INTO chunks_fts (rowid, content) " + "VALUES (?, ?)", (row["chunk_id"], text), ) restored += 1 @@ -532,7 +618,8 @@ def pull_pack( "chunks_restored": restored, "chunks_skipped_already_hot": skipped_already_hot, "chunks_skipped_unknown_locally": skipped_unknown, - "chunks_drifted": drifted, + "tables_restored": tables_restored, + "pack_format": "v2" if table_files_seen else "v1", "backend": backend_id, }, ) @@ -540,8 +627,9 @@ def pull_pack( return { "status": "pulled", "pack_hash": pack_hash, + "pack_format": "v2" if table_files_seen else "v1", + "tables_restored": tables_restored, "chunks_restored": restored, "chunks_skipped_already_hot": skipped_already_hot, "chunks_skipped_unknown_locally": skipped_unknown, - "drift_details": drift_details, } diff --git a/docs/cold-object-store.md b/docs/cold-object-store.md index 4a059e1..4c60518 100644 --- a/docs/cold-object-store.md +++ b/docs/cold-object-store.md @@ -34,6 +34,31 @@ backup consumers download packs whole. No `blobs/` prefix, no per-chunk objects. (One pack ↔ one disc ↔ one bucket object.) + Pack contents (v2 format, self-sufficient for new-peer hydration): + ``` + manifest.ndjson # chunk leaf_hash + size catalog + tables/documents.jsonl # array-per-line, sorted + tables/chunks.jsonl # without content column + tables/merkle_nodes.jsonl + tables/edges.jsonl # FAN-IN restructured + tables/audit_events.jsonl + tables/derivations.jsonl + tables/concept_relations.jsonl + tables/concept_token_idf.jsonl + tables/providence_cache.jsonl + tables/citation_aliases.jsonl + tables/term_aliases.jsonl + tables/snapshots.jsonl + tables/document_http_meta.jsonl + blobs// # raw UTF-8 chunk bodies + ``` + + Not shipped (per-peer or rebuildable): + `mesh_*`, `selfmodel_*`, `capital_ledger`, `controller_events`, + `fork_score_branches`, `memory_*`, `adapter_loss_reports`, + `falsifications`, `schema_meta`, `meta`, `chunks_fts*`, + `documents_fts*`. + 2. **`pack_hash = hash_leaf(manifest_bytes)`.** The manifest is sorted by `leaf_hash` and deduped before hashing, so input order and accidental duplicates don't move the hash. Two writers producing the @@ -215,16 +240,24 @@ For larger media: ## Cost model (DO Spaces, current corpus) -Numbers from the live shard estimator (4 shards × ~3.5M chunks each, -14.1M chunks total, ~17 GB compressed; streaming cap fills each pack -to ~4.4 GB compressed): +Pack format v2 (self-sufficient for new-peer hydration). Numbers measured +2026-05-26 against the live 4-shard corpus (14.1M total chunks; the +1.56M hot-content chunks per shard go into packs; metadata is added on +top via the v2 dump path): -| Path | Count | Storage | Cost | -|-----------------------------|-------------|----------|---------------------| -| Bucket pack storage | ~4 packs | ~17 GB | $0.34/mo (@ $0.02/GB) | -| Full-corpus hydrate (CDN) | ~4 GETs | — | ~$0.00002 in requests | -| Egress (in-region) | 0 | — | $0 | -| Egress (CDN to public) | 17 GB / peer | — | $0.17 per fresh peer (@ $0.01/GB) | +| Path | Count | Storage | Cost | +|-------------------------------|-------------|----------|-------------------------------| +| v2 pack storage (per shard) | 1 pack | ~2.1 GB | — | +| v2 pack storage (all 4) | 4 packs | ~8.5 GB | $0.17/mo (@ $0.02/GB) | +| Full-corpus hydrate (CDN) | ~4 GETs | — | ~$0.00002 in requests | +| Egress (in-region) | 0 | — | $0 | +| Egress (CDN to public) | 8.5 GB/peer | — | $0.09 per fresh peer (@ $0.01/GB) | + +The v1 chunks-only format produced ~1.78 GB per shard (7.1 GB total). +v2 adds ~0.3–0.4 GB per shard for the metadata tables (chunks-meta, +documents, audit_events, merkle_nodes, edges fan-in restructured, plus +small tables). Trade: ~20 % more storage for a self-sufficient pack +that a fresh peer can unpack into a working shard with no other inputs. Repacking after a falsification event costs the same as the initial pack — one full corpus serialization per event-batched run, gated by diff --git a/docs/tickets/ticket-000061-cold-object-store-tier.md b/docs/tickets/ticket-000061-cold-object-store-tier.md index cad1692..c9cf46f 100644 --- a/docs/tickets/ticket-000061-cold-object-store-tier.md +++ b/docs/tickets/ticket-000061-cold-object-store-tier.md @@ -188,7 +188,30 @@ Out of scope (future tickets if needed): ## Status -In progress. Code lands incrementally on `main`. Sizing math against -current 4-shard / 14.1M-chunk corpus: **~4 packs total** (17.2 GB -compressed ÷ 4.4 GB compressed-cap per pack via streaming zstd), ~17 GB -bucket storage. +In progress. Code lands incrementally on `main`. + +**Pack format v2 (2026-05-26):** v1 (chunks-only) was correctly flagged +by fox as under-engineered — packs contained only chunk bodies, not the +shard tables a new peer needs to actually bootstrap. v2 ships every +load-bearing table inside the same tar.zst alongside the chunk blobs. +Tables go in as `tables/.jsonl` (array-per-line columnar JSONL). +The `edges` table goes through a fan-in restructure at pack-build time +(group by `dst_uri + edge_type + anchor`, ship `src_roots` as an array) +which yields ~5-10× compression vs the live-schema row layout. Live +SQLite schema is unchanged; the restore step expands fan-in rows back +to the per-edge form. Pack uploads NO LONGER append a +`cold_pack_pushed` audit event (would leak into the next push's dumped +metadata and break the "two writers at the same corpus state produce +the same pack_hash" determinism property — the bucket / disc file IS +the receipt). 24 tests pass including a new `test_push_pack_v2_hydrates_fresh_empty_db` +that round-trips push from a populated DB into a completely empty DB. + +**Sizing (v2, current shards):** ~2.1 GB per shard pack compressed +(chunk content 1.78 GB + tables ~0.3 GB), ~8.5 GB total bucket +footprint for new-peer-ready packs across 4 shards. + +Per-peer state NOT shipped: mesh_*, selfmodel_*, capital_ledger, +memory_*, controller_events, fork_score_branches, adapter_loss_reports, +falsifications, schema_meta, meta. FTS5 (chunks_fts*, documents_fts*) +not shipped either — rebuilt from chunks.content + documents.title on +unpack. diff --git a/tests/test_cold_object.py b/tests/test_cold_object.py index 83228b6..13521f7 100644 --- a/tests/test_cold_object.py +++ b/tests/test_cold_object.py @@ -299,6 +299,81 @@ def test_push_pack_local_dir_writes_files(tmp_path): conn.close() +def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path): + """Pack format v2 self-sufficiency test: build a pack from a populated + DB, then unpack it into a FRESH empty DB and verify every table is + restored (not just chunks content). This is the real new-peer + hydration path.""" + src_db = tmp_path / "src.db" + rx_db = tmp_path / "rx.db" + conn_src = connect(src_db) + backend = MemoryBackend() + try: + # Populate the source DB with a few docs so multiple tables get + # rows: documents, chunks, merkle_nodes, audit_events at minimum. + # Distinct content per doc — same content dedupes by document_root. + docs = [ + _doc(f"html://hydrate{i}", LONG + f"\n\nDoc-specific marker {i}." * 5) + for i in range(3) + ] + ingest_source(conn_src, FakeSource(docs)) + + src_docs = conn_src.execute( + "SELECT COUNT(*) FROM documents" + ).fetchone()[0] + src_chunks = conn_src.execute( + "SELECT COUNT(*) FROM chunks" + ).fetchone()[0] + src_merkle = conn_src.execute( + "SELECT COUNT(*) FROM merkle_nodes" + ).fetchone()[0] + src_audit = conn_src.execute( + "SELECT COUNT(*) FROM audit_events" + ).fetchone()[0] + assert src_docs == 3 and src_chunks > 0 and src_merkle > 0 and src_audit > 0 + + result = push_pack(conn_src, backend) + assert result["status"] == "pushed" + pack_hash = result["packs"][0]["pack_hash"] + finally: + conn_src.close() + + # Brand-new empty receiving DB. No ingest, nothing — just `connect()` + # which creates the schema. + conn_rx = connect(rx_db) + try: + # Confirm rx is empty before unpack. + assert conn_rx.execute( + "SELECT COUNT(*) FROM documents" + ).fetchone()[0] == 0 + + pull_result = pull_pack(conn_rx, backend, pack_hash) + assert pull_result["status"] == "pulled" + assert pull_result["pack_format"] == "v2" + # Every table that was non-empty in src should have rows now in rx. + assert pull_result["tables_restored"]["documents"] == src_docs + assert pull_result["tables_restored"]["chunks"] == src_chunks + assert pull_result["tables_restored"]["merkle_nodes"] == src_merkle + assert pull_result["tables_restored"]["audit_events"] == src_audit + # Chunk bodies restored. + rx_hot = conn_rx.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='hot' AND content IS NOT NULL" + ).fetchone()[0] + assert rx_hot == src_chunks + # FTS5 rows rebuilt. + rx_fts = conn_rx.execute("SELECT COUNT(*) FROM chunks_fts").fetchone()[0] + assert rx_fts == src_chunks + # All three documents reachable by URI. + for i in range(3): + row = conn_rx.execute( + "SELECT title FROM documents WHERE document_uri=?", + (f"html://hydrate{i}",), + ).fetchone() + assert row is not None, f"document html://hydrate{i} not restored" + finally: + conn_rx.close() + + def test_push_pack_no_push_writes_locally_only(tmp_path): """push_to_bucket=False skips bucket writes, requires local_dir.""" db = tmp_path / "local-only.db"