#000061: bind pack_hash to full pack content (tables + chunks)
Caught a real defect on the live v2 corpus run: pack_hash was computed
from the chunk-only manifest, so two packs with the same chunk set but
different table content collided on pack_hash. Observed live: v1 packs
and v2 packs for the same shard produced the SAME pack_hash:
v1 (chunks-only) b3c427baaf9b40e3c33ace8438b4254e3abdc7b38bdcb7b1a8012588cb32848a
v2 (tables + chunks) b3c427baaf9b40e3c33ace8438b4254e3abdc7b38bdcb7b1a8012588cb32848a
↑ identical, different bytes
Consequences if shipped: bucket overwrites swap whose tables you get,
mesh peers A and B disagree on what "pack X" is while both claim to
have it, content-addressing story is broken for the metadata half.
Fix: manifest format v2 prepends a `{"_format_version": 2}` header,
then one row per shipped table:
{"table_file": "tables/audit_events.jsonl", "hash": "...", "size": N}
{"table_file": "tables/chunks.jsonl", "hash": "...", "size": N}
...
{"leaf_hash": "...", "size": N} # chunks come after
{"leaf_hash": "...", "size": N}
pack_hash = hash_leaf(manifest_bytes) now binds the full pack content.
Two writers with the same shard state produce the same pack_hash; same
chunks + different tables produce different pack_hashes.
Implementation:
- hash_file_leaf(path) — streaming sha256 with leaf 0x00 prefix, 1 MB
reads, used to fingerprint multi-GB tables/<name>.jsonl files without
loading them into RAM.
- stream_packs' extra_members now takes (member_name, source_path,
content_hash) tuples. push_pack computes the hash with hash_file_leaf
after the dump finishes.
- _finalize_pack writes the format header + table refs (sorted by name
for determinism) before chunk entries.
- parse_manifest returns a ParsedManifest (format_version, tables[],
chunks[]) instead of just list[PackEntry]. Backward-compatible with
v1 manifests (no header → format_version=1, tables=()).
- pull_pack pulls the manifest aside during tar walk, then after
extraction verifies each tables/<name>.jsonl file via hash_file_leaf
against the manifest's content_hash. Mismatch → ValueError, no rows
reach the live schema.
New tests:
- test_v2_pack_hash_binds_table_contents — asserts manifest format +
table refs + per-table hash & size.
- test_pull_pack_rejects_tampered_table — corrupts one table's bytes
in a real pack, confirms pull_pack raises on hash mismatch.
27 cold-object + evict tests pass.
This commit is contained in:
parent
bc7efe4434
commit
a193394ea8
4 changed files with 271 additions and 24 deletions
|
|
@ -2929,7 +2929,7 @@ def _cmd_cold_list(args: argparse.Namespace) -> int:
|
|||
if args.fetch_manifest:
|
||||
try:
|
||||
manifest_bytes = backend.get(pack_key(pack_hash, manifest=True))
|
||||
entry["chunk_count"] = len(parse_manifest(manifest_bytes))
|
||||
entry["chunk_count"] = len(parse_manifest(manifest_bytes).chunks)
|
||||
except Exception as e: # noqa: BLE001 — surface but don't fail
|
||||
entry["chunk_count"] = None
|
||||
entry["manifest_error"] = repr(e)
|
||||
|
|
|
|||
|
|
@ -61,6 +61,22 @@ MANIFEST_PREFIX = "manifest/"
|
|||
MANIFEST_LATEST_KEY = MANIFEST_PREFIX + "latest.ndjson"
|
||||
|
||||
|
||||
def hash_file_leaf(path: Path, *, chunk_size: int = 1024 * 1024) -> str:
|
||||
"""Compute `hash_leaf(file_contents).hex()` by streaming the file in
|
||||
1 MB chunks. Used to fingerprint multi-GB table dumps for the pack
|
||||
manifest without loading the file into RAM."""
|
||||
import hashlib
|
||||
h = hashlib.sha256()
|
||||
h.update(b"\x00") # leaf-domain prefix, matches arborist.merkle.hash_leaf
|
||||
with open(path, "rb") as f:
|
||||
while True:
|
||||
buf = f.read(chunk_size)
|
||||
if not buf:
|
||||
break
|
||||
h.update(buf)
|
||||
return h.hexdigest()
|
||||
|
||||
|
||||
def pack_key(pack_hash: str, *, manifest: bool = False) -> str:
|
||||
"""Bucket key for a pack body (default) or its manifest sidecar."""
|
||||
if manifest:
|
||||
|
|
@ -470,7 +486,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,
|
||||
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.
|
||||
|
|
@ -505,13 +521,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) 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. 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).
|
||||
`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).
|
||||
"""
|
||||
if max_compressed_bytes <= 0:
|
||||
raise ValueError("max_compressed_bytes must be positive")
|
||||
|
|
@ -539,11 +558,36 @@ 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,
|
||||
))
|
||||
|
||||
entries = []
|
||||
for leaf_hash in entries_sorted:
|
||||
size = state["entries_by_hash"][leaf_hash]
|
||||
|
|
@ -577,12 +621,19 @@ def stream_packs(
|
|||
# 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.
|
||||
for member_name, source_path in pending_extras:
|
||||
#
|
||||
# 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)
|
||||
info.size = source_path.stat().st_size
|
||||
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:
|
||||
|
|
@ -684,12 +735,56 @@ def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]:
|
|||
yield expected_hash, body
|
||||
|
||||
|
||||
def parse_manifest(manifest_bytes: bytes) -> list[PackEntry]:
|
||||
"""Parse manifest.ndjson into PackEntry list. Lenient on blank lines."""
|
||||
out = []
|
||||
@dataclass(frozen=True)
|
||||
class TableRef:
|
||||
"""One table file reference in a v2 pack manifest. Hash binds the
|
||||
consumer's verification path: after extracting tar member
|
||||
`tables/<name>.jsonl`, the consumer must compute hash_leaf of the
|
||||
bytes and compare to this `content_hash`."""
|
||||
member_name: str
|
||||
content_hash: str
|
||||
size: int
|
||||
|
||||
|
||||
@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
|
||||
chunks: tuple[PackEntry, ...]
|
||||
|
||||
|
||||
def parse_manifest(manifest_bytes: bytes) -> ParsedManifest:
|
||||
"""Parse a manifest.ndjson stream into table refs + chunk entries.
|
||||
|
||||
Format detection: a line `{"_format_version": N}` at the top means
|
||||
v2. Absent means legacy v1 (chunks-only). Lenient on blank lines.
|
||||
"""
|
||||
fmt = 1
|
||||
tables: list[TableRef] = []
|
||||
chunks: list[PackEntry] = []
|
||||
for line in manifest_bytes.decode("utf-8").splitlines():
|
||||
if not line.strip():
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
rec = json.loads(line)
|
||||
out.append(PackEntry(leaf_hash=rec["leaf_hash"], size=int(rec["size"])))
|
||||
return out
|
||||
if "_format_version" in rec:
|
||||
fmt = int(rec["_format_version"])
|
||||
elif "table_file" in rec:
|
||||
tables.append(TableRef(
|
||||
member_name=rec["table_file"],
|
||||
content_hash=rec["hash"],
|
||||
size=int(rec["size"]),
|
||||
))
|
||||
elif "leaf_hash" in rec:
|
||||
chunks.append(PackEntry(
|
||||
leaf_hash=rec["leaf_hash"], size=int(rec["size"])
|
||||
))
|
||||
# silently ignore unknown record shapes (forward compat)
|
||||
return ParsedManifest(
|
||||
format_version=fmt,
|
||||
tables=tuple(tables),
|
||||
chunks=tuple(chunks),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -314,7 +314,7 @@ def push_pack(
|
|||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from arborist.cold_object import stream_packs
|
||||
from arborist.cold_object import hash_file_leaf, stream_packs
|
||||
from arborist.cold_pack_metadata import dump_shard_metadata
|
||||
from arborist.snapshot import compute_snapshot_root
|
||||
|
||||
|
|
@ -407,10 +407,19 @@ def push_pack(
|
|||
# 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.
|
||||
#
|
||||
# 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.
|
||||
metadata_dir = Path(tempfile.mkdtemp(prefix="arborist-meta-"))
|
||||
metadata_summary = dump_shard_metadata(conn, metadata_dir)
|
||||
extra_members = [
|
||||
(f"tables/{name}.jsonl", info["path"])
|
||||
(
|
||||
f"tables/{name}.jsonl",
|
||||
info["path"],
|
||||
hash_file_leaf(info["path"]),
|
||||
)
|
||||
for name, info in metadata_summary.items()
|
||||
]
|
||||
|
||||
|
|
@ -510,6 +519,7 @@ def pull_pack(
|
|||
|
||||
import zstandard
|
||||
|
||||
from arborist.cold_object import hash_file_leaf, parse_manifest
|
||||
from arborist.cold_pack_metadata import restore_shard_metadata
|
||||
from arborist.merkle import hash_leaf
|
||||
|
||||
|
|
@ -524,14 +534,23 @@ def pull_pack(
|
|||
|
||||
# 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.
|
||||
# The manifest is read separately so we can verify each table file's
|
||||
# bytes against the manifest's content_hash (v2 only) on extraction.
|
||||
tables_dir = Path(tempfile.mkdtemp(prefix="arborist-restore-"))
|
||||
chunks_to_restore: list[tuple[str, bytes]] = []
|
||||
table_files_seen: list[str] = []
|
||||
manifest_bytes: bytes | None = None
|
||||
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
|
||||
# Pull the manifest aside so we can verify table hashes
|
||||
# after extraction. (Manifest itself is the integrity
|
||||
# root: pack_hash = hash_leaf(manifest_bytes).)
|
||||
f = tar.extractfile(member)
|
||||
if f is not None:
|
||||
manifest_bytes = f.read()
|
||||
continue
|
||||
if member.name.startswith("tables/"):
|
||||
# tables/<name>.jsonl — extract to temp dir for
|
||||
# restore_shard_metadata to read line-by-line.
|
||||
|
|
@ -562,6 +581,35 @@ def pull_pack(
|
|||
)
|
||||
chunks_to_restore.append((expected_hash, chunk_body))
|
||||
|
||||
# Pack format v2: cross-check each extracted table file's bytes
|
||||
# against the content_hash recorded in the manifest. Tampered or
|
||||
# truncated tables fail loudly here, never reach the live schema.
|
||||
# v1 packs (no manifest table refs) skip this verification.
|
||||
tampered_tables: list[dict] = []
|
||||
manifest_format = 1
|
||||
if manifest_bytes is not None:
|
||||
parsed = parse_manifest(manifest_bytes)
|
||||
manifest_format = parsed.format_version
|
||||
table_ref_by_name = {t.member_name: t for t in parsed.tables}
|
||||
for member_name in table_files_seen:
|
||||
ref = table_ref_by_name.get(member_name)
|
||||
if ref is None:
|
||||
continue # tar carries a table not in the manifest;
|
||||
# forward-compat: just don't verify it.
|
||||
local_path = tables_dir / Path(member_name).name
|
||||
actual = hash_file_leaf(local_path)
|
||||
if actual != ref.content_hash:
|
||||
tampered_tables.append({
|
||||
"member": member_name,
|
||||
"expected_hash": ref.content_hash,
|
||||
"actual_hash": actual,
|
||||
})
|
||||
if tampered_tables:
|
||||
raise ValueError(
|
||||
f"pack {pack_hash[:12]} table-hash mismatch: "
|
||||
f"{tampered_tables}"
|
||||
)
|
||||
|
||||
# Phase 1: restore metadata tables (no-op for v1 packs).
|
||||
tables_restored: dict[str, int] = {}
|
||||
if table_files_seen:
|
||||
|
|
|
|||
|
|
@ -100,10 +100,15 @@ def test_parse_manifest_round_trip():
|
|||
body = b"abc" * 100
|
||||
leaf = hash_leaf(body).hex()
|
||||
pack = build_pack([(leaf, body)])
|
||||
entries = parse_manifest(pack.manifest_bytes)
|
||||
assert len(entries) == 1
|
||||
assert entries[0].leaf_hash == leaf
|
||||
assert entries[0].size == len(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
|
||||
assert parsed.tables == ()
|
||||
assert len(parsed.chunks) == 1
|
||||
assert parsed.chunks[0].leaf_hash == leaf
|
||||
assert parsed.chunks[0].size == len(body)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -368,6 +373,105 @@ 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"
|
||||
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)
|
||||
|
||||
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}
|
||||
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
|
||||
assert t.size > 0
|
||||
# Chunks are also there.
|
||||
assert len(parsed.chunks) > 0
|
||||
finally:
|
||||
conn.close()
|
||||
|
||||
|
||||
def test_pull_pack_rejects_tampered_table(tmp_path):
|
||||
"""If the manifest says tables/X.jsonl should hash to H but the tar
|
||||
member has different bytes, pull_pack refuses to restore.
|
||||
|
||||
Tampering test: build a real v2 pack, then construct a corrupted
|
||||
copy that swaps one table's bytes but keeps the original manifest.
|
||||
pull_pack must raise on hash mismatch.
|
||||
"""
|
||||
import io as _io
|
||||
import tarfile as _tarfile
|
||||
import zstandard as _zstd
|
||||
|
||||
db = tmp_path / "src.db"
|
||||
conn = connect(db)
|
||||
backend = MemoryBackend()
|
||||
try:
|
||||
ingest_source(conn, FakeSource([_doc("html://t", LONG)]))
|
||||
result = push_pack(conn, backend)
|
||||
pack_hash = result["packs"][0]["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)
|
||||
raw_tar = _zstd.ZstdDecompressor().stream_reader(_io.BytesIO(body)).read()
|
||||
out_buf = _io.BytesIO()
|
||||
cctx = _zstd.ZstdCompressor(level=3)
|
||||
with cctx.stream_writer(out_buf, closefd=False) as writer:
|
||||
with _tarfile.open(fileobj=writer, mode="w|") as out_tar:
|
||||
with _tarfile.open(fileobj=_io.BytesIO(raw_tar), mode="r") as in_tar:
|
||||
for m in in_tar:
|
||||
f = in_tar.extractfile(m)
|
||||
if f is None:
|
||||
continue
|
||||
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)
|
||||
out_tar.addfile(info, _io.BytesIO(data))
|
||||
backend._tamper(f"packs/{pack_hash}.tar.zst", out_buf.getvalue())
|
||||
|
||||
# Fresh DB; expect ValueError on pull.
|
||||
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)
|
||||
finally:
|
||||
conn_rx.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
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue