engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

#000061: SPV pack split — metadata pack + N chunk packs (v3)

Browse source 
Bidirectional sync: producer always emits both kinds; consumer chooses
how much to pull.

  packs/<hash>.metadata.tar.zst       — one per shard (~0.6 GB compressed)
  packs/<hash>.metadata.manifest.ndjson
  packs/<hash>.chunks.tar.zst         — N per shard (each ≤ 4.4 GB cap)
  packs/<hash>.chunks.manifest.ndjson

Each artifact is independently content-addressed by its own manifest
hash. The metadata pack's manifest carries _chunk_pack_hashes — every
chunk pack covering this shard's content — so a "full" consumer can
iterate them. Chunks packs are anonymous from the consumer side
(reachable only via the metadata pack's reference list).

Consumer sync modes:

  arborist cold unpack <metadata_hash>            # default: just-enough
  arborist cold unpack <metadata_hash> --full     # also pulls chunks

just-enough: pull only the metadata pack. Schema fully restored; every
chunks row has content=NULL. Node is immediately queryable for
metadata operations (documents, edges, audit chain, Merkle); chunk-body
queries return null until a future JIT-fetch path fills them on cache
miss.

full: after the metadata pack lands, iterate _chunk_pack_hashes and pull
every chunk pack. Final state: full corpus offline-queryable.

This is fox's original SPV-wallet framing — was right from day one.

Key API changes:

  pack_key(hash, *, kind="chunks", manifest=False)  — kind in bucket path
  stream_packs(chunks, *, max_compressed_bytes, ...) -> Iterator[FilePack]
      now emits chunk-only packs (no extra_members)
  build_metadata_pack(table_files, *, snapshot_root, chunk_pack_hashes, ...)
      single FilePack with v3 metadata manifest
  pull_metadata_pack(conn, backend, hash) -> dict
  pull_chunk_pack(conn, backend, hash) -> dict
  hydrate_from_metadata_pack(conn, backend, metadata_hash, *, mode) -> dict

push_pack orchestrates: dump tables → stream chunk packs (collect hashes)
→ build metadata pack with chunk_pack_hashes → upload all. Returns
{metadata_pack_hash, chunk_pack_hashes, packs: [...]}.

Manifest format v3:
  metadata pack:
    {"_format_version": 3}
    {"_kind": "metadata"}
    {"_snapshot_root": "..."}
    {"_chunk_pack_hashes": [...]}
    {"table_file": "tables/X.jsonl", "hash": "...", "size": N}  per table
  chunks pack:
    {"_format_version": 3}
    {"_kind": "chunks"}
    {"leaf_hash": "...", "size": N}  per chunk

pack_hash for each = hash_leaf(manifest_bytes); content-addressed at
both layers. Two writers with the same shard state produce identical
metadata_pack_hash AND identical chunk_pack_hashes.

Cap-and-split applies only to chunk packs (chunks are bounded N).
Metadata pack is one file per shard; if a single table file is larger
than the cap, that's noted as future row-level split work.

cold list now surfaces kind per artifact (metadata vs chunks), plus
table_count + chunk_pack_hashes (for metadata packs) or chunk_count
(for chunks packs). Operators can quickly find the metadata pack hash
to feed `cold unpack`.

28 cold-object + evict + boto3 tests pass (3 new SPV-shape tests + 1
v3 manifest-shape test + tampered-metadata-pack test).
This commit is contained in:
russell@unturf.com 2026-05-26 09:27:02 -04:00
parent a193394ea8
commit de705c89a6
No known key found for this signature in database
4 changed files with 613 additions and 250 deletions
Download patch file Download diff file Expand all files Collapse all files

83
arborist/cli.py
View file

@ -2868,7 +2868,21 @@ class _LocalOnlyBackend:
def _cmd_cold_unpack(args: argparse.Namespace) -> int:
from arborist.evict import pull_pack
"""Hydrate a shard from a metadata pack hash. SPV-style consumer:
- default mode "just-enough": pull only the metadata pack. Schema
restored, every chunks row has content=NULL. Queryable immediately
for metadata; chunk-body queries return null until a future JIT-
fetch path fills them on cache miss.
- "--full": after the metadata pack lands, iterate its
`_chunk_pack_hashes` and pull every chunk pack. Final state: full
corpus offline-queryable.
Either way, `args.pack_hash` must be a metadata pack hash. Chunk
packs are anonymous from the consumer's perspective — accessed via
the metadata pack's reference list, not by name.
"""
from arborist.evict import hydrate_from_metadata_pack
backend = _make_cold_backend()
conn = (
@ -2877,7 +2891,9 @@ def _cmd_cold_unpack(args: argparse.Namespace) -> int:
else connect(args.db)
)
try:
result = pull_pack(conn, backend, args.pack_hash)
result = hydrate_from_metadata_pack(
conn, backend, args.pack_hash, mode=args.mode,
)
finally:
conn.close()
print(json.dumps(result, indent=2, ensure_ascii=False))
@ -2906,32 +2922,47 @@ def _cmd_cold_stats(args: argparse.Namespace) -> int:
def _cmd_cold_list(args: argparse.Namespace) -> int:
"""Enumerate packs in the bucket for new-peer hydration.
"""Enumerate packs in the bucket for new-peer hydration. SPV-aware.
For each `<pack>.tar.zst`, HEAD the object to get its compressed size,
and (unless --no-manifest) fetch the small manifest sidecar to count
chunks. Output is a JSON array — pipe through jq or feed to a
hydration script that calls `cold unpack` for each entry.
For each `<hash>.{metadata,chunks}.tar.zst`, HEAD for size and
(unless --no-manifest) fetch the small manifest sidecar to surface
kind + chunk_count / table_count + chunk_pack_hashes. Output
enables a consumer to:
- find the metadata pack(s) — the entry points
- know what chunk packs each metadata pack references (full sync)
"""
from arborist.cold_object import PACK_PREFIX, pack_key, parse_manifest
backend = _make_cold_backend()
packs = []
for key in backend.list_keys(PACK_PREFIX):
if not key.endswith(".tar.zst"):
# v3 keys: packs/<hash>.metadata.tar.zst or packs/<hash>.chunks.tar.zst
if key.endswith(".metadata.tar.zst"):
kind = "metadata"
pack_hash = key[len(PACK_PREFIX):-len(".metadata.tar.zst")]
elif key.endswith(".chunks.tar.zst"):
kind = "chunks"
pack_hash = key[len(PACK_PREFIX):-len(".chunks.tar.zst")]
else:
continue
# key = "packs/<pack_hash>.tar.zst"
pack_hash = key[len(PACK_PREFIX):-len(".tar.zst")]
entry = {
"pack_hash": pack_hash,
"kind": kind,
"compressed_bytes": backend.object_size(key),
}
if args.fetch_manifest:
try:
manifest_bytes = backend.get(pack_key(pack_hash, manifest=True))
entry["chunk_count"] = len(parse_manifest(manifest_bytes).chunks)
manifest_bytes = backend.get(
pack_key(pack_hash, kind=kind, manifest=True)
)
parsed = parse_manifest(manifest_bytes)
if kind == "metadata":
entry["table_count"] = len(parsed.tables)
entry["snapshot_root"] = parsed.snapshot_root
entry["chunk_pack_hashes"] = list(parsed.chunk_pack_hashes)
else:
entry["chunk_count"] = len(parsed.chunks)
except Exception as e: # noqa: BLE001 — surface but don't fail
entry["chunk_count"] = None
entry["manifest_error"] = repr(e)
packs.append(entry)
out = {
@ -5707,9 +5738,31 @@ def build_parser() -> argparse.ArgumentParser:
cold_unpack = cold_sub.add_parser(
"unpack",
help="pull one pack from bucket and restore chunks locally (hash-verified)",
help=(
"hydrate a shard from a metadata pack hash. default is "
"--just-enough (metadata only, chunks NULL); --full also "
"pulls every referenced chunk pack."
),
)
cold_unpack.add_argument(
"pack_hash",
help="metadata pack hash (NOT a chunk pack hash). Find via `cold list`.",
)
cold_unpack.add_argument(
"--full",
dest="mode",
action="store_const",
const="full",
default="just-enough",
help="after metadata, pull every chunk pack the metadata references",
)
cold_unpack.add_argument(
"--just-enough",
dest="mode",
action="store_const",
const="just-enough",
help="(default) metadata only — chunks remain NULL for JIT fetch",
)
cold_unpack.add_argument("pack_hash", help="pack_hash = sha256(manifest)")
cold_unpack.set_defaults(func=_cmd_cold_unpack)
cold_stats = cold_sub.add_parser(

294
arborist/cold_object.py
View file

@ -77,11 +77,32 @@ def hash_file_leaf(path: Path, *, chunk_size: int = 1024 * 1024) -> str:
return h.hexdigest()
def pack_key(pack_hash: str, *, manifest: bool = False) -> str:
"""Bucket key for a pack body (default) or its manifest sidecar."""
def pack_key(
pack_hash: str,
*,
kind: str = "chunks",
manifest: bool = False,
) -> str:
"""Bucket key for a pack body (default) or its manifest sidecar.
`kind` distinguishes the two artifact shapes in the v3 SPV layout:
- "chunks" — `packs/<hash>.chunks.tar.zst` (chunk bodies, N per shard)
- "metadata" — `packs/<hash>.metadata.tar.zst` (one per shard, references chunk_pack_hashes)
Both can be pulled independently; a "just-enough" sync only needs the
metadata pack, a "full" sync also pulls every chunk pack in
`_chunk_pack_hashes`. Kind goes into the bucket key so a fresh peer
can list-by-prefix and prioritize metadata downloads.
"""
if kind not in ("metadata", "chunks"):
raise ValueError(f"unknown pack kind: {kind!r}")
if manifest:
return f"{PACK_PREFIX}{pack_hash}.manifest.ndjson"
return f"{PACK_PREFIX}{pack_hash}.tar.zst"
return f"{PACK_PREFIX}{pack_hash}.{kind}.manifest.ndjson"
return f"{PACK_PREFIX}{pack_hash}.{kind}.tar.zst"
PACK_KIND_METADATA = "metadata"
PACK_KIND_CHUNKS = "chunks"
@dataclass(frozen=True)
@ -157,26 +178,37 @@ class ObjectStoreBackend(abc.ABC):
...
# Convenience wrappers — keep call sites short and consistent.
def put_pack(self, pack_hash: str, body: bytes) -> None:
self.put(pack_key(pack_hash), body, content_type="application/zstd")
# `kind` distinguishes metadata packs from chunk packs in the v3 SPV
# layout. Default kind="chunks" preserves the call-site shape for
# chunk-pack producers; metadata-pack producers pass kind="metadata".
def put_pack(self, pack_hash: str, body: bytes, *, kind: str = "chunks") -> None:
self.put(pack_key(pack_hash, kind=kind), body, content_type="application/zstd")
def put_pack_file(self, pack_hash: str, path: Path) -> None:
def put_pack_file(self, pack_hash: str, path: Path, *, kind: str = "chunks") -> None:
"""File-backed pack upload — `stream_packs` produces these.
boto3 `upload_file` reads parts from disk for the multipart upload
instead of loading the whole pack into memory. Required for any
parallelism above ~2 workers at the 4.4 GB cap.
"""
self.put_file(pack_key(pack_hash), path, content_type="application/zstd")
self.put_file(
pack_key(pack_hash, kind=kind), path, content_type="application/zstd"
)
def get_pack(self, pack_hash: str) -> bytes:
return self.get(pack_key(pack_hash))
def get_pack(self, pack_hash: str, *, kind: str = "chunks") -> bytes:
return self.get(pack_key(pack_hash, kind=kind))
def put_pack_manifest(self, pack_hash: str, body: bytes) -> None:
self.put(pack_key(pack_hash, manifest=True), body, content_type="application/x-ndjson")
def put_pack_manifest(
self, pack_hash: str, body: bytes, *, kind: str = "chunks"
) -> None:
self.put(
pack_key(pack_hash, kind=kind, manifest=True),
body,
content_type="application/x-ndjson",
)
def get_pack_manifest(self, pack_hash: str) -> bytes:
return self.get(pack_key(pack_hash, manifest=True))
def get_pack_manifest(self, pack_hash: str, *, kind: str = "chunks") -> bytes:
return self.get(pack_key(pack_hash, kind=kind, manifest=True))
class S3CompatibleBackend(ObjectStoreBackend):
@ -444,7 +476,11 @@ def build_pack(chunks: list[tuple[str, bytes]]) -> Pack:
raise ValueError(f"bad leaf_hash length: {leaf_hash}")
by_hash[leaf_hash] = body
manifest_lines = []
# v3 chunks-only manifest (matches stream_packs output shape).
manifest_lines = [
json.dumps({"_format_version": 3}, sort_keys=True),
json.dumps({"_kind": PACK_KIND_CHUNKS}, sort_keys=True),
]
entries = []
for leaf_hash in sorted(by_hash):
body = by_hash[leaf_hash]
@ -486,7 +522,6 @@ def stream_packs(
max_compressed_bytes: int,
level: int = 3,
work_dir: Path | str | None = None,
extra_members: list[tuple[str, Path, str]] | None = None,
) -> Iterator[FilePack]:
"""Stream (leaf_hash, raw_utf8_body) chunks into FILE-BACKED packs that
each fit in `max_compressed_bytes` compressed bytes.
@ -521,22 +556,16 @@ 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,
content_hash) for files to include AT THE START OF THE FIRST PACK.
Used by pack format v2 to ship shard-metadata table dumps
(`tables/<name>.jsonl`) ahead of the chunk bodies. `content_hash`
must be `hash_leaf(source_file_contents).hex()` — it goes into the
pack manifest so `pack_hash` content-addresses the *full* pack
(tables + chunks), not just chunks. 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).
Pack v3 (SPV split): this emits CHUNK-ONLY packs. Metadata lives in
a separate artifact built by `build_metadata_pack` so consumers
pulling metadata-only can skip the chunk bodies entirely. The
metadata pack carries the list of every chunk_pack_hash for its
shard so a "full" consumer knows what to fetch.
"""
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
@ -558,36 +587,20 @@ def stream_packs(
"writer": writer,
"tar": tar,
"entries_by_hash": {},
"table_refs": [], # (member_name, content_hash, size) — populated for first pack only
}
def _finalize_pack(state: dict) -> FilePack:
entries_sorted = sorted(state["entries_by_hash"].keys())
manifest_lines = []
# Pack format v2: tag explicitly. Consumers branch on this to
# know they should expect + verify `table_file` entries before
# the chunk entries. Absence of the header line = legacy v1
# (chunks-only) — accepted by `parse_manifest` for read but no
# longer produced by `stream_packs`.
has_tables = bool(state["table_refs"])
if has_tables:
manifest_lines.append(
json.dumps({"_format_version": 2}, sort_keys=True)
)
# Table refs first, sorted by tar member name for determinism.
# content_hash binds pack_hash to the file bytes; two writers
# with same tables produce same hashes.
for member_name, content_hash, table_size in sorted(state["table_refs"]):
manifest_lines.append(json.dumps(
{
"table_file": member_name,
"hash": content_hash,
"size": table_size,
},
sort_keys=True,
))
# Pack v3 chunk-pack manifest:
# header line : {"_format_version": 3}
# kind line : {"_kind": "chunks"}
# chunk lines : {"leaf_hash": "...", "size": N}
# The chunk pack is anonymous w.r.t. its metadata pack — the
# backlink lives in the metadata pack's _chunk_pack_hashes list.
manifest_lines = [
json.dumps({"_format_version": 3}, sort_keys=True),
json.dumps({"_kind": PACK_KIND_CHUNKS}, sort_keys=True),
]
entries = []
for leaf_hash in entries_sorted:
size = state["entries_by_hash"][leaf_hash]
@ -614,32 +627,6 @@ def stream_packs(
state = _start_pack()
try:
# Pack format v2: ship the metadata table dumps as the FIRST tar
# members of the first pack. Stream from disk — tarfile.addfile
# reads the file handle in chunks, never loading the whole thing
# into RAM. A 4 GB edges.jsonl thus uses ~tens of KB peak RAM
# instead of 4 GB per worker. TarInfo built by hand (NOT via
# gettarinfo) so mtime/uid/gid are zeroed — two writers' packs
# converge byte-for-byte regardless of filesystem timestamps.
#
# Each extra_member carries (tar_name, source_path, content_hash).
# The content_hash goes into the manifest so pack_hash binds the
# full pack (tables + chunks). Caller computed the hash via
# hash_file_leaf() at dump time.
for member_name, source_path, content_hash in pending_extras:
info = tarfile.TarInfo(name=member_name)
file_size = source_path.stat().st_size
info.size = file_size
# mtime, uid, gid, mode default to 0 / "" — deterministic.
with open(source_path, "rb") as f:
state["tar"].addfile(info, f)
state["table_refs"].append((member_name, content_hash, file_size))
# 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}")
@ -693,6 +680,112 @@ def stream_packs(
raise
def build_metadata_pack(
table_files: list[tuple[str, Path, str]],
*,
snapshot_root: str,
chunk_pack_hashes: list[str],
work_dir: Path | str | None = None,
level: int = 3,
) -> FilePack:
"""Build ONE metadata pack from the shard's table dumps.
`table_files` is a list of (tar_member_name, source_file_path,
content_hash) — the same shape as the old extra_members. Caller has
already dumped each shipped table to a JSONL file on disk.
The metadata pack's manifest carries:
{"_format_version": 3}
{"_kind": "metadata"}
{"_snapshot_root": "..."} # corpus state at pack time
{"_chunk_pack_hashes": [...]} # every chunk pack covering this shard
{"table_file": ..., "hash": ..., "size": ...} per table
`pack_hash = hash_leaf(manifest_bytes).hex()` — content-addressed,
so two writers producing the same shard state + same chunk packs
converge on the same metadata_pack_hash.
Streaming write to disk (tempfile in work_dir): worker RAM stays
bounded by zstd internal buffers regardless of metadata size.
"""
work_path = Path(work_dir) if work_dir is not None else Path(tempfile.gettempdir())
work_path.mkdir(parents=True, exist_ok=True)
fd, path_str = tempfile.mkstemp(
dir=work_path,
prefix="arborist-meta-pack-",
suffix=".tar.zst.tmp",
)
file_obj = os.fdopen(fd, "wb")
cctx = zstandard.ZstdCompressor(level=level)
writer = cctx.stream_writer(file_obj, closefd=False)
tar = tarfile.open(fileobj=writer, mode="w|")
try:
# Tar entries: tables/<name>.jsonl per shipped table.
# Stream from disk (handle iteration), zeroed TarInfo metadata for
# determinism.
table_refs_sorted: list[tuple[str, str, int]] = []
for member_name, source_path, content_hash in sorted(table_files):
info = tarfile.TarInfo(name=member_name)
file_size = source_path.stat().st_size
info.size = file_size
with open(source_path, "rb") as f:
tar.addfile(info, f)
table_refs_sorted.append((member_name, content_hash, file_size))
# Build manifest. Sort chunk_pack_hashes for determinism.
chunk_pack_hashes_sorted = sorted(set(chunk_pack_hashes))
manifest_lines = [
json.dumps({"_format_version": 3}, sort_keys=True),
json.dumps({"_kind": PACK_KIND_METADATA}, sort_keys=True),
json.dumps({"_snapshot_root": snapshot_root}, sort_keys=True),
json.dumps(
{"_chunk_pack_hashes": chunk_pack_hashes_sorted},
sort_keys=True,
),
]
for member_name, content_hash, table_size in table_refs_sorted:
manifest_lines.append(json.dumps(
{
"table_file": member_name,
"hash": content_hash,
"size": table_size,
},
sort_keys=True,
))
manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8")
# Manifest goes LAST in the tar — open_pack iterates by name and
# ignores order, and writing-last lets us include chunk_pack_hashes
# (which weren't known when tar started).
info = tarfile.TarInfo(name="manifest.ndjson")
info.size = len(manifest_bytes)
tar.addfile(info, io.BytesIO(manifest_bytes))
tar.close()
writer.close()
file_obj.close()
pack_hash = hash_leaf(manifest_bytes).hex()
body_size = Path(path_str).stat().st_size
return FilePack(
pack_hash=pack_hash,
manifest_bytes=manifest_bytes,
body_path=Path(path_str),
body_size=body_size,
entries=tuple(), # metadata pack has no PackEntry-style chunks
)
except BaseException:
try: tar.close()
except Exception: pass
try: writer.close()
except Exception: pass
try: file_obj.close()
except Exception: pass
try: Path(path_str).unlink()
except FileNotFoundError: pass
raise
def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]:
"""Stream (leaf_hash, body) pairs from a pack body.
@ -748,21 +841,39 @@ class TableRef:
@dataclass(frozen=True)
class ParsedManifest:
"""Result of parsing a pack manifest. Distinguishes v1 (chunks-only)
from v2 (tables + chunks) so consumers can branch on which
invariants to verify."""
format_version: int # 1 (legacy) or 2 (table-bound)
tables: tuple[TableRef, ...] # empty for v1
"""Result of parsing a pack manifest.
- format_version: 1 (legacy chunks-only), 2 (legacy tables + chunks),
or 3 (SPV-split: pure metadata or pure chunks).
- kind: "metadata" | "chunks" | None (for v1/v2 mixed).
- snapshot_root: corpus state the pack covers (v3 only).
- chunk_pack_hashes: every chunk pack a v3 metadata pack references.
- tables: TableRef list (empty for chunks-only packs).
- chunks: PackEntry list (empty for metadata-only packs).
"""
format_version: int
kind: str | None
snapshot_root: str | None
chunk_pack_hashes: tuple[str, ...]
tables: tuple[TableRef, ...]
chunks: tuple[PackEntry, ...]
def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
"""Parse a manifest.ndjson stream into table refs + chunk entries.
"""Parse a manifest.ndjson stream.
Format detection: a line `{"_format_version": N}` at the top means
v2. Absent means legacy v1 (chunks-only). Lenient on blank lines.
Format detection:
- line `{"_format_version": N}` at top → that version.
- absence → legacy v1.
- v3 packs additionally carry `_kind` (metadata/chunks),
`_snapshot_root`, and (metadata only) `_chunk_pack_hashes`.
Lenient on blank lines and unknown record shapes (forward-compat).
"""
fmt = 1
kind: str | None = None
snapshot_root: str | None = None
chunk_pack_hashes: list[str] = []
tables: list[TableRef] = []
chunks: list[PackEntry] = []
for line in manifest_bytes.decode("utf-8").splitlines():
@ -772,6 +883,12 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
rec = json.loads(line)
if "_format_version" in rec:
fmt = int(rec["_format_version"])
elif "_kind" in rec:
kind = str(rec["_kind"])
elif "_snapshot_root" in rec:
snapshot_root = str(rec["_snapshot_root"])
elif "_chunk_pack_hashes" in rec:
chunk_pack_hashes = [str(h) for h in rec["_chunk_pack_hashes"]]
elif "table_file" in rec:
tables.append(TableRef(
member_name=rec["table_file"],
@ -785,6 +902,9 @@ def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
# silently ignore unknown record shapes (forward compat)
return ParsedManifest(
format_version=fmt,
kind=kind,
snapshot_root=snapshot_root,
chunk_pack_hashes=tuple(chunk_pack_hashes),
tables=tuple(tables),
chunks=tuple(chunks),
)

218
arborist/evict.py
View file

@ -314,7 +314,11 @@ def push_pack(
import tempfile
from pathlib import Path
from arborist.cold_object import hash_file_leaf, stream_packs
from arborist.cold_object import (
build_metadata_pack,
hash_file_leaf,
stream_packs,
)
from arborist.cold_pack_metadata import dump_shard_metadata
from arborist.snapshot import compute_snapshot_root
@ -402,75 +406,114 @@ def push_pack(
# upload path is also memory-bounded.
import shutil
# Pack format v2: dump every shipped table to JSONL in a temp dir, then
# include those files as `tables/<name>.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.
# SPV-split (pack format v3): every shard produces TWO artifact kinds.
#
# Each table file's content_hash is computed via hash_file_leaf (stream
# the file through sha256 in 1 MB chunks) and passed into stream_packs
# so the pack manifest binds pack_hash to (tables + chunks), not just
# chunks. Two writers with the same shard state → same pack_hash.
# 1. N chunk packs (each ≤ cap, content-addressed by chunk-only
# manifest) — heavy, optional. A "just-enough" consumer skips.
# 2. ONE metadata pack (small, content-addressed by table+chunk-
# pack-refs manifest) — light, always pulled by any consumer.
#
# The metadata pack's manifest carries `_chunk_pack_hashes`, so a
# "full" consumer iterates that list and pulls the chunk packs.
# Producer always emits both; consumer chooses sync mode.
# Phase A: dump every shipped table to JSONL on disk.
metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-"))
metadata_summary = dump_shard_metadata(conn, metadata_dir)
extra_members = [
table_files = [
(
f"tables/{name}.jsonl",
info["path"],
hash_file_leaf(info["path"]),
)
for name, info in metadata_summary.items()
if info["rows"] > 0 # skip tables that exist but have no data rows;
# avoids shipping bare-header jsonl files in a
# metadata pack for a freshly-initialized DB
]
# Phase B: stream chunk packs. Each chunk pack is anonymous (no
# backlink to the metadata pack — the metadata pack is the entry
# point). Collect each chunk pack's hash for the metadata manifest.
chunk_pack_hashes_in_order: list[str] = []
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:
if push_to_bucket:
backend.put_pack_file(pack.pack_hash, pack.body_path)
backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes)
backend.put_pack_file(pack.pack_hash, pack.body_path, kind="chunks")
backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes, kind="chunks")
if out_dir is not None:
short = pack.pack_hash[:16]
final_path = out_dir / f"arborist-pack-{short}.tar.zst"
# Move (cheap if on same filesystem) instead of read+write.
final_path = out_dir / f"arborist-pack-{short}.chunks.tar.zst"
shutil.move(str(pack.body_path), str(final_path))
(out_dir / f"arborist-pack-{short}.manifest.ndjson").write_bytes(
(out_dir / f"arborist-pack-{short}.chunks.manifest.ndjson").write_bytes(
pack.manifest_bytes
)
# 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.
chunk_pack_hashes_in_order.append(pack.pack_hash)
pack_results.append({
"pack_hash": pack.pack_hash,
"kind": "chunks",
"chunk_count": len(pack.entries),
"uncompressed_bytes": pack_uncompressed,
"compressed_bytes": pack.body_size,
"snapshot_root": snapshot_root,
})
finally:
# Clean up the temp file when neither local_dir nor an exception
# claimed it. `shutil.move` already removed it on the local_dir
# path; this handles the push-only path and the audit-failure
# case.
if pack.body_path.exists():
try:
pack.body_path.unlink()
except OSError:
pass
# Phase C: build + push the metadata pack now that we know every
# chunk pack's hash. The metadata pack's manifest includes
# _chunk_pack_hashes so the consumer can iterate them for full-sync.
metadata_pack_hash: str | None = None
metadata_pack_bytes: int = 0
if table_files:
meta_pack = build_metadata_pack(
table_files,
snapshot_root=snapshot_root,
chunk_pack_hashes=chunk_pack_hashes_in_order,
)
try:
if push_to_bucket:
backend.put_pack_file(
meta_pack.pack_hash, meta_pack.body_path, kind="metadata"
)
backend.put_pack_manifest(
meta_pack.pack_hash, meta_pack.manifest_bytes, kind="metadata"
)
if out_dir is not None:
short = meta_pack.pack_hash[:16]
final_path = out_dir / f"arborist-pack-{short}.metadata.tar.zst"
shutil.move(str(meta_pack.body_path), str(final_path))
(out_dir / f"arborist-pack-{short}.metadata.manifest.ndjson").write_bytes(
meta_pack.manifest_bytes
)
metadata_pack_hash = meta_pack.pack_hash
metadata_pack_bytes = meta_pack.body_size
pack_results.append({
"pack_hash": meta_pack.pack_hash,
"kind": "metadata",
"table_count": len(table_files),
"chunk_pack_hashes": list(chunk_pack_hashes_in_order),
"compressed_bytes": meta_pack.body_size,
"snapshot_root": snapshot_root,
})
finally:
if meta_pack.body_path.exists():
try:
meta_pack.body_path.unlink()
except OSError:
pass
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:
@ -483,9 +526,16 @@ def push_pack(
"status": "pushed" if push_to_bucket else "written",
"packs": pack_results,
"pack_count": len(pack_results),
"total_chunks": sum(p["chunk_count"] for p in pack_results),
"total_uncompressed_bytes": sum(p["uncompressed_bytes"] for p in pack_results),
"metadata_pack_hash": metadata_pack_hash,
"chunk_pack_hashes": chunk_pack_hashes_in_order,
"total_chunks": sum(
p.get("chunk_count", 0) for p in pack_results if p["kind"] == "chunks"
),
"total_uncompressed_bytes": sum(
p.get("uncompressed_bytes", 0) for p in pack_results if p["kind"] == "chunks"
),
"total_compressed_bytes": sum(p["compressed_bytes"] for p in pack_results),
"metadata_pack_bytes": metadata_pack_bytes,
"snapshot_root": snapshot_root,
"snapshot_doc_count": doc_count_at_pack,
"snapshot_ts": pack_ts,
@ -495,21 +545,94 @@ def push_pack(
}
def pull_pack(
def pull_metadata_pack(
conn: sqlite3.Connection,
backend: "ObjectStoreBackend",
pack_hash: str,
) -> dict:
"""Pull one pack from the bucket and restore everything it contains:
metadata tables (pack v2) + chunk bodies.
"""Pull a metadata pack and restore every shipped table into the
local schema. Chunks rows land with `content=NULL` and `tier='cold'`
— the metadata view is fully populated but chunk bodies are empty.
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.
This is the "just-enough" hydration path: schema is queryable
immediately for documents / audit chain / edges / Merkle structure;
chunk-body queries fall through to JIT fetch (future) or to a
follow-up `pull_chunk_pack` against each `_chunk_pack_hashes` entry.
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.
Returns the parsed manifest's chunk_pack_hashes so a "full" consumer
can iterate them.
"""
return _pull_pack_inner(conn, backend, pack_hash, kind="metadata")
def pull_chunk_pack(
conn: sqlite3.Connection,
backend: "ObjectStoreBackend",
pack_hash: str,
) -> dict:
"""Pull a chunk pack and fill `chunks.content` + `chunks_fts` for
every leaf_hash whose row already exists locally. Idempotent against
a populated DB (skips rows already content-filled).
Assumes `pull_metadata_pack` has already run (or schema otherwise
populated) — chunks without a matching row are reported as
`skipped_unknown` and skipped.
"""
return _pull_pack_inner(conn, backend, pack_hash, kind="chunks")
def hydrate_from_metadata_pack(
conn: sqlite3.Connection,
backend: "ObjectStoreBackend",
metadata_pack_hash: str,
*,
mode: str = "just-enough",
) -> dict:
"""Higher-level consumer entry point. Modes:
- `"just-enough"`: pull only the metadata pack. Schema fully
restored, every chunks row has `content=NULL`. Node is
immediately queryable for the metadata graph (documents, edges,
audit chain, etc.). Chunk-body queries return null until a future
JIT-fetch path (cache miss → CDN/mesh) fills them.
- `"full"`: pull metadata pack, then iterate
`_chunk_pack_hashes` from its manifest and pull every chunk pack.
Final state: full corpus offline-queryable.
"""
if mode not in ("just-enough", "full"):
raise ValueError(f"unknown sync mode: {mode!r}")
meta_result = pull_metadata_pack(conn, backend, metadata_pack_hash)
chunk_results = []
if mode == "full":
for ch_hash in meta_result.get("chunk_pack_hashes_referenced", []):
chunk_results.append(pull_chunk_pack(conn, backend, ch_hash))
return {
"status": "hydrated",
"mode": mode,
"metadata_pack_hash": metadata_pack_hash,
"metadata_pull": meta_result,
"chunk_pulls": chunk_results,
"chunk_packs_pulled": len(chunk_results),
}
def _pull_pack_inner(
conn: sqlite3.Connection,
backend: "ObjectStoreBackend",
pack_hash: str,
*,
kind: str,
) -> dict:
"""Shared pull-and-restore body for metadata + chunks packs.
`kind` ∈ {"metadata", "chunks"} drives both the bucket key suffix
and which restore phase runs. Manifest's `_kind` field is checked
on extraction; a kind mismatch raises ValueError (defends against
a bucket where a chunks-key has been overwritten with metadata
contents).
"""
import io
import shutil
@ -524,7 +647,7 @@ def pull_pack(
from arborist.merkle import hash_leaf
backend_id = backend.identity.to_audit_body()
body = backend.get_pack(pack_hash)
body = backend.get_pack(pack_hash, kind=kind)
# Streaming decompression — packs built by stream_packs lack the
# content-size frame header so the one-shot ZstdDecompressor.decompress
@ -657,17 +780,23 @@ def pull_pack(
)
restored += 1
# Parse manifest a second time for chunk_pack_hashes surface (for the
# metadata-pack-pull caller — it needs the list to drive full-sync).
chunk_pack_hashes_referenced: tuple[str, ...] = ()
if manifest_bytes is not None:
chunk_pack_hashes_referenced = parse_manifest(manifest_bytes).chunk_pack_hashes
with transaction(conn):
append_audit(
conn,
event_type="cold_pack_pulled",
body={
"pack_hash": pack_hash,
"pack_kind": kind,
"chunks_restored": restored,
"chunks_skipped_already_hot": skipped_already_hot,
"chunks_skipped_unknown_locally": skipped_unknown,
"tables_restored": tables_restored,
"pack_format": "v2" if table_files_seen else "v1",
"backend": backend_id,
},
)
@ -675,7 +804,8 @@ def pull_pack(
return {
"status": "pulled",
"pack_hash": pack_hash,
"pack_format": "v2" if table_files_seen else "v1",
"pack_kind": kind,
"chunk_pack_hashes_referenced": list(chunk_pack_hashes_referenced),
"tables_restored": tables_restored,
"chunks_restored": restored,
"chunks_skipped_already_hot": skipped_already_hot,

268
tests/test_cold_object.py
View file

@ -23,7 +23,9 @@ from arborist.cold_object import (
from arborist.compress import unpack_chunk
from arborist.document import Document
from arborist.evict import (
pull_pack,
hydrate_from_metadata_pack,
pull_chunk_pack,
pull_metadata_pack,
push_pack,
)
from arborist.ingest import ingest_source
@ -60,10 +62,13 @@ LONG = (
def test_pack_key_layout():
h = "b" * 64
assert pack_key(h).endswith(".tar.zst")
assert pack_key(h, manifest=True).endswith(".manifest.ndjson")
# Different scheme so a `list_keys("packs/")` can distinguish.
# Default kind="chunks". v3 SPV-split keys carry the kind in the path.
assert pack_key(h).endswith(".chunks.tar.zst")
assert pack_key(h, manifest=True).endswith(".chunks.manifest.ndjson")
assert pack_key(h, kind="metadata").endswith(".metadata.tar.zst")
assert pack_key(h, kind="metadata", manifest=True).endswith(".metadata.manifest.ndjson")
assert pack_key(h) != pack_key(h, manifest=True)
assert pack_key(h, kind="chunks") != pack_key(h, kind="metadata")
def test_build_pack_is_content_addressed():
@ -101,10 +106,9 @@ def test_parse_manifest_round_trip():
leaf = hash_leaf(body).hex()
pack = build_pack([(leaf, body)])
parsed = parse_manifest(pack.manifest_bytes)
# build_pack produces a chunks-only manifest (no tables) — that's the
# v1 shape because build_pack doesn't accept extra_members. Tests of
# the v2 shape live in test_push_pack_v2_hydrates_fresh_empty_db.
assert parsed.format_version == 1
# build_pack produces a chunks-only v3 manifest.
assert parsed.format_version == 3
assert parsed.kind == "chunks"
assert parsed.tables == ()
assert len(parsed.chunks) == 1
assert parsed.chunks[0].leaf_hash == leaf
@ -151,7 +155,9 @@ def test_memory_backend_identity_carries_no_credentials():
# ---------------------------------------------------------------------------
def test_push_pack_then_pull_pack_restores_chunks(tmp_path):
def test_push_pack_then_pull_full_restores_chunks(tmp_path):
"""Push produces 1 metadata pack + N chunk packs. Full hydration
(mode="full") pulls metadata then iterates chunk_pack_hashes."""
db = tmp_path / "pack.db"
conn = connect(db)
backend = MemoryBackend()
@ -161,41 +167,85 @@ def test_push_pack_then_pull_pack_restores_chunks(tmp_path):
"SELECT document_root FROM documents WHERE document_uri='html://p'"
).fetchone()["document_root"]
# Pack while content is still local. Tiny corpus → one pack.
push_result = push_pack(conn, backend, document_root=root)
assert push_result["status"] == "pushed"
assert push_result["pack_count"] == 1
# Pack is bound to a snapshot_root — packs are delayed point-in-time
# snapshots; the audit row pins which corpus state this pack covers
# so falsifications between repacks produce a different pack_hash.
assert len(push_result["snapshot_root"]) == 64
first = push_result["packs"][0]
assert first["chunk_count"] > 0
assert first["compressed_bytes"] <= first["uncompressed_bytes"] + 200
assert first["snapshot_root"] == push_result["snapshot_root"]
pack_hash = first["pack_hash"]
assert backend.head(pack_key(pack_hash))
assert backend.head(pack_key(pack_hash, manifest=True))
# SPV split: at least 1 metadata pack + at least 1 chunks pack.
assert push_result["metadata_pack_hash"] is not None
assert len(push_result["chunk_pack_hashes"]) >= 1
# Now NULL local content so pull_pack has something to do.
conn.execute("UPDATE chunks SET content = NULL, tier='cold'")
conn.execute("DELETE FROM chunks_fts")
conn.commit()
metadata_hash = push_result["metadata_pack_hash"]
# Both pack kinds are in the bucket at their kind-specific keys.
assert backend.head(pack_key(metadata_hash, kind="metadata"))
assert backend.head(pack_key(metadata_hash, kind="metadata", manifest=True))
for ch in push_result["chunk_pack_hashes"]:
assert backend.head(pack_key(ch, kind="chunks"))
pull_result = pull_pack(conn, backend, pack_hash)
assert pull_result["status"] == "pulled"
assert pull_result["chunks_restored"] == first["chunk_count"]
# NULL local content + delete documents so a full pull has to
# restore the whole schema, not just chunks.
with conn:
conn.execute("DELETE FROM chunks_fts")
conn.execute("DELETE FROM chunks")
conn.execute("DELETE FROM documents")
result = hydrate_from_metadata_pack(
conn, backend, metadata_hash, mode="full",
)
assert result["status"] == "hydrated"
assert result["mode"] == "full"
assert result["chunk_packs_pulled"] == len(push_result["chunk_pack_hashes"])
# Schema restored.
assert conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0] == 1
# All chunks restored to hot tier with content.
cold = conn.execute(
"SELECT COUNT(*) FROM chunks WHERE tier='cold'"
"SELECT COUNT(*) FROM chunks WHERE tier='cold' OR content IS NULL"
).fetchone()[0]
assert cold == 0
finally:
conn.close()
def test_push_pack_then_just_enough_hydrate_leaves_chunks_null(tmp_path):
"""just-enough mode: pull only metadata pack. Schema fully restored
but chunks rows have content=NULL — ready for JIT-fetch later."""
db = tmp_path / "je.db"
conn = connect(db)
backend = MemoryBackend()
try:
ingest_source(conn, FakeSource([_doc("html://je", LONG)]))
push_result = push_pack(conn, backend)
metadata_hash = push_result["metadata_pack_hash"]
expected_chunks = sum(p.get("chunk_count", 0) for p in push_result["packs"]
if p.get("kind") == "chunks")
finally:
conn.close()
rx_db = tmp_path / "rx.db"
conn_rx = connect(rx_db)
try:
result = hydrate_from_metadata_pack(
conn_rx, backend, metadata_hash, mode="just-enough",
)
assert result["status"] == "hydrated"
assert result["mode"] == "just-enough"
assert result["chunk_packs_pulled"] == 0
# Documents row exists.
assert conn_rx.execute("SELECT COUNT(*) FROM documents").fetchone()[0] == 1
# Chunks rows exist (metadata restored) but content is NULL.
n_chunks = conn_rx.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
n_null = conn_rx.execute(
"SELECT COUNT(*) FROM chunks WHERE content IS NULL"
).fetchone()[0]
assert n_chunks == expected_chunks
assert n_null == n_chunks
finally:
conn_rx.close()
def test_push_pack_idempotent(tmp_path):
"""Same chunk set → same pack_hash(es) → bucket overwrite is a no-op."""
"""Same shard state on two runs produces the same pack_hashes:
metadata_pack_hash + every chunk_pack_hash. Bucket overwrite is a
no-op."""
db = tmp_path / "pack-id.db"
conn = connect(db)
backend = MemoryBackend()
@ -203,9 +253,8 @@ def test_push_pack_idempotent(tmp_path):
ingest_source(conn, FakeSource([_doc("html://p", LONG)]))
a = push_pack(conn, backend)
b = push_pack(conn, backend)
assert [p["pack_hash"] for p in a["packs"]] == [
p["pack_hash"] for p in b["packs"]
]
assert a["metadata_pack_hash"] == b["metadata_pack_hash"]
assert a["chunk_pack_hashes"] == b["chunk_pack_hashes"]
finally:
conn.close()
@ -258,31 +307,38 @@ def test_push_pack_splits_to_fit_dvdr(tmp_path):
# Tar trailer (~1 KB padding) + zstd frame footer get emitted after
# the last in-loop size check, so overshoot is bounded by trailer
# size + one chunk's worth of compressed bytes.
# size + one chunk's worth of compressed bytes. Only the chunks
# packs are subject to the cap; the metadata pack is its own
# artifact and can be any size.
OVERSHOOT_SLACK = 4_096
for p in result["packs"]:
chunk_packs = [p for p in result["packs"] if p["kind"] == "chunks"]
assert len(chunk_packs) > 1, "small cap should force multi-chunks-pack split"
for p in chunk_packs:
assert p["compressed_bytes"] <= CAP + OVERSHOOT_SLACK, (
f"pack {p['pack_hash'][:8]} compressed={p['compressed_bytes']} "
f"exceeds cap {CAP}+slack {OVERSHOOT_SLACK}"
f"chunks pack {p['pack_hash'][:8]} compressed="
f"{p['compressed_bytes']} exceeds cap {CAP}+slack"
)
assert p["chunk_count"] >= 1
# Round-trip: NULL local content, pull every pack back.
# Round-trip: NULL local content, hydrate-full from metadata pack.
conn.execute("UPDATE chunks SET content = NULL, tier='cold'")
conn.execute("DELETE FROM chunks_fts")
conn.commit()
total_restored = 0
for p in result["packs"]:
r = pull_pack(conn, backend, p["pack_hash"])
total_restored += r["chunks_restored"]
hyd = hydrate_from_metadata_pack(
conn, backend, result["metadata_pack_hash"], mode="full",
)
assert hyd["status"] == "hydrated"
# Total chunks restored == total chunks across all chunk packs.
total_restored = sum(r["chunks_restored"] for r in hyd["chunk_pulls"])
assert total_restored == result["total_chunks"]
finally:
conn.close()
def test_push_pack_local_dir_writes_files(tmp_path):
"""--local-dir writes pack + manifest files for burning to physical media."""
"""--local-dir writes pack + manifest files for burning to physical media.
v3: both metadata pack and chunk packs land in local-dir, with kind
suffix in the filename."""
db = tmp_path / "burn.db"
conn = connect(db)
backend = MemoryBackend()
@ -293,13 +349,17 @@ def test_push_pack_local_dir_writes_files(tmp_path):
assert result["status"] == "pushed"
assert result["local_dir"] == str(burn_dir)
assert burn_dir.exists()
pack_files = sorted(burn_dir.glob("arborist-pack-*.tar.zst"))
manifest_files = sorted(burn_dir.glob("arborist-pack-*.manifest.ndjson"))
assert len(pack_files) == result["pack_count"]
assert len(manifest_files) == result["pack_count"]
# Local files are byte-identical to what we'd burn.
for pack_file, summary in zip(pack_files, result["packs"]):
assert pack_file.stat().st_size == summary["compressed_bytes"]
# v3 filenames: arborist-pack-<short>.{metadata,chunks}.tar.zst
metadata_files = sorted(burn_dir.glob("arborist-pack-*.metadata.tar.zst"))
chunks_files = sorted(burn_dir.glob("arborist-pack-*.chunks.tar.zst"))
assert len(metadata_files) == 1
assert len(chunks_files) == len(result["chunk_pack_hashes"])
# Manifest sidecars present too.
assert len(sorted(burn_dir.glob("arborist-pack-*.metadata.manifest.ndjson"))) == 1
assert (
len(sorted(burn_dir.glob("arborist-pack-*.chunks.manifest.ndjson")))
== len(result["chunk_pack_hashes"])
)
finally:
conn.close()
@ -373,44 +433,44 @@ def test_edges_fan_in_batches_huge_destinations(tmp_path):
conn.close()
def test_v2_pack_hash_binds_table_contents(tmp_path):
"""Two packs with the same chunk set but different table content
must NOT collide on pack_hash.
The v1 bug: pack_hash was hash_leaf(chunk-only-manifest), so
same-chunks + different-tables produced same pack_hash. Bucket
overwrites or mesh-peer disagreement → silent table swap. v2
fixes this by including content_hash for each tables/<name>.jsonl
in the manifest before chunk entries.
This test forces the issue by ingesting two different document
sets that happen to share a chunk — wait, that's impossible by
construction. Instead we check the manifest format directly: a
v2 pack's manifest has a `_format_version: 2` header and
`table_file` entries before `leaf_hash` entries.
"""
db = tmp_path / "v2bind.db"
def test_v3_metadata_pack_manifest_shape(tmp_path):
"""v3 metadata pack manifest carries: _format_version=3, _kind="metadata",
_snapshot_root, _chunk_pack_hashes list, and one table_file entry per
non-empty shipped table. Chunks pack manifest separately carries
_kind="chunks" and leaf_hash entries."""
db = tmp_path / "v3shape.db"
conn = connect(db)
backend = MemoryBackend()
try:
ingest_source(conn, FakeSource([_doc("html://a", LONG)]))
result = push_pack(conn, backend)
manifest = backend.get_pack_manifest(result["packs"][0]["pack_hash"])
parsed = parse_manifest(manifest)
metadata_hash = result["metadata_pack_hash"]
chunk_hashes = result["chunk_pack_hashes"]
assert metadata_hash is not None
assert len(chunk_hashes) >= 1
assert parsed.format_version == 2
# All shipped tables that had rows should appear in the manifest.
# documents + chunks + merkle_nodes + audit_events minimum.
table_names = {t.member_name for t in parsed.tables}
meta_manifest = backend.get_pack_manifest(metadata_hash, kind="metadata")
meta = parse_manifest(meta_manifest)
assert meta.format_version == 3
assert meta.kind == "metadata"
assert meta.snapshot_root and len(meta.snapshot_root) == 64
assert list(meta.chunk_pack_hashes) == chunk_hashes
assert meta.chunks == ()
table_names = {t.member_name for t in meta.tables}
assert "tables/documents.jsonl" in table_names
assert "tables/chunks.jsonl" in table_names
assert "tables/audit_events.jsonl" in table_names
# Every table_file ref carries a hash and size.
for t in parsed.tables:
assert len(t.content_hash) == 64 # sha256 hex
for t in meta.tables:
assert len(t.content_hash) == 64
assert t.size > 0
# Chunks are also there.
assert len(parsed.chunks) > 0
# First chunks pack: chunks only, no tables.
first_chunks_manifest = backend.get_pack_manifest(chunk_hashes[0], kind="chunks")
ch = parse_manifest(first_chunks_manifest)
assert ch.format_version == 3
assert ch.kind == "chunks"
assert ch.tables == ()
assert len(ch.chunks) > 0
finally:
conn.close()
@ -433,14 +493,14 @@ def test_pull_pack_rejects_tampered_table(tmp_path):
try:
ingest_source(conn, FakeSource([_doc("html://t", LONG)]))
result = push_pack(conn, backend)
pack_hash = result["packs"][0]["pack_hash"]
metadata_hash = result["metadata_pack_hash"]
finally:
conn.close()
# Decompress the pack, rewrite documents.jsonl with corrupted bytes,
# repack, replace the pack in the bucket. Manifest stays unchanged
# so the recorded table content_hash no longer matches.
body = backend.get_pack(pack_hash)
# Decompress the metadata pack, rewrite documents.jsonl with corrupted
# bytes, repack, replace in the bucket. Manifest stays unchanged so
# the recorded table content_hash no longer matches.
body = backend.get_pack(metadata_hash, kind="metadata")
raw_tar = _zstd.ZstdDecompressor().stream_reader(_io.BytesIO(body)).read()
out_buf = _io.BytesIO()
cctx = _zstd.ZstdCompressor(level=3)
@ -454,20 +514,20 @@ def test_pull_pack_rejects_tampered_table(tmp_path):
data = f.read()
if m.name == "tables/documents.jsonl":
data = data + b'\n{"injected": "tamper"}\n'
info = _tarfile.TarInfo(name=m.name)
info.size = len(data)
else:
info = _tarfile.TarInfo(name=m.name)
info.size = len(data)
info = _tarfile.TarInfo(name=m.name)
info.size = len(data)
out_tar.addfile(info, _io.BytesIO(data))
backend._tamper(f"packs/{pack_hash}.tar.zst", out_buf.getvalue())
backend._tamper(
f"packs/{metadata_hash}.metadata.tar.zst", out_buf.getvalue()
)
# Fresh DB; expect ValueError on pull.
# Fresh DB; expect ValueError on hydrate (metadata pack tamper fires
# the hash-check during pull_metadata_pack).
rx_db = tmp_path / "rx.db"
conn_rx = connect(rx_db)
try:
with pytest.raises(ValueError, match="table-hash mismatch"):
pull_pack(conn_rx, backend, pack_hash)
pull_metadata_pack(conn_rx, backend, metadata_hash)
finally:
conn_rx.close()
@ -507,7 +567,8 @@ def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path):
result = push_pack(conn_src, backend)
assert result["status"] == "pushed"
pack_hash = result["packs"][0]["pack_hash"]
metadata_hash = result["metadata_pack_hash"]
assert metadata_hash is not None
finally:
conn_src.close()
@ -520,20 +581,19 @@ def test_push_pack_v2_hydrates_fresh_empty_db(tmp_path):
"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.
hyd = hydrate_from_metadata_pack(
conn_rx, backend, metadata_hash, mode="full",
)
assert hyd["status"] == "hydrated"
assert hyd["mode"] == "full"
assert conn_rx.execute(
"SELECT COUNT(*) FROM documents"
).fetchone()[0] == src_docs
# Chunks rows + content 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.