From 727cb1bd967615f3042168eb916985d2dc5c32a0 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 25 May 2026 20:23:44 -0400 Subject: [PATCH] feat: #000061 cold-pack distribution tier (boto3 S3-compat + DVD-R safe-fit) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ship arborist corpus state to new peers and DVD-R archival via point-in-time tar.zst packs. One artifact serves both channels — bucket+CDN delivery and physical-media archival. Bucket holds packs only. Pack key = hash_leaf(manifest_bytes), so same chunk set on two writers produces the same pack_hash and upload is idempotent. Each pack pins the corpus snapshot_root it covers in audit + result body — packs are delayed snapshots, not live mirrors; falsifications between repacks produce new pack_hashes. stream_packs runs streaming zstd over tarfile, peeking compressed-buffer size after each chunk via FLUSH_BLOCK (preserves dictionary). Default cap 4_400_000_000 — 4.4 GB DVD-R safe-fit, ~6.5% buffer below the 4.7 GB marketing capacity to absorb ISO9660 overhead, growisofs lead-in/lead-out, media variance, and drive-edge refusal. Each disc fills to ~4.4 GB recorded data, not the ~1.5 GB an uncompressed cap produced. One backend class (S3CompatibleBackend via boto3 + endpoint_url) covers AWS S3, DO Spaces, R2, B2, GCS S3-interop, MinIO. Optional dep [object-store] = boto3>=1.34; dev extras pull moto for the wire test. Voyeur: credentials via AWS_ACCESS_KEY_ID/_SECRET_ACCESS_KEY env or ~/.aws/credentials, never printed; only endpoint URL + bucket name surface in logs. CLI: arborist cold {pack,unpack,stats}. Makefile: cold-pack, cold-pack-dvd (local-dir output for growisofs), cold-unpack, cold-stats. Sizing for current shards (14.1M chunks, ~17 GB compressed): ~4 packs at the default cap, ~\$0.34/mo DO Spaces storage, ~\$0.0001/fresh-peer hydrate. Always-on raw-UTF-8 leaf store (per ticket "Hard invariants") deferred — packs-only for now, backfill later. 2557 passed, 28 skipped, 1 xfailed. --- CLAUDE.md | 7 + Makefile | 39 +- arborist/cli.py | 195 +++++++ arborist/cold_object.py | 510 ++++++++++++++++++ arborist/evict.py | 296 +++++++++- docs/TICKETS.md | 3 +- docs/cold-object-store.md | 259 +++++++++ .../ticket-000061-cold-object-store-tier.md | 194 +++++++ pyproject.toml | 18 + tests/test_cold_object.py | 325 +++++++++++ tests/test_cold_object_boto3.py | 69 +++ 11 files changed, 1911 insertions(+), 4 deletions(-) create mode 100644 arborist/cold_object.py create mode 100644 docs/cold-object-store.md create mode 100644 docs/tickets/ticket-000061-cold-object-store-tier.md create mode 100644 tests/test_cold_object.py create mode 100644 tests/test_cold_object_boto3.py diff --git a/CLAUDE.md b/CLAUDE.md index ddb0b43..00d5f6b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -598,6 +598,13 @@ Architecture / ongoing work: - `docs/embedding.md` — embedding arborist as a library in another Python app (`arborist.embed`): produce `Document`s → ingest → dedup + FTS5 + audit chain. The neopig-backend seam. +- `docs/cold-object-store.md` — cold-pack distribution tier (#000061): + serialize the corpus into `tar.zst` packs and ship them via S3-compatible + buckets (DO Spaces / AWS S3 / R2 / B2 / GCS / MinIO via boto3) and/or + burn to DVD-R via `--local-dir` + growisofs. Packs are point-in-time + snapshots; each pack pins the `snapshot_root` it covers so falsifications + between repacks produce new pack_hashes. ≤4.4 GB safe-fit per pack (DVD-R + with ~6.5 % buffer below the 4.7 GB marketing capacity). - `docs/benchmarks.md` — orientation: harnesses, fixtures, signal floor, make targets, bench-row schema, addenda index. Read first when running a bench. diff --git a/Makefile b/Makefile index dd7ee09..e323fca 100644 --- a/Makefile +++ b/Makefile @@ -40,7 +40,8 @@ SEARCH_Q ?= computer bootstrap-math bootstrap-nli bootstrap-nli-only bench-nli-shadow export-nli-onnx bench-nli-backends judge-self-test control-ab control-sweep rapl-access rapl-access-revoke clean clean-db clean-data help \ textbooks-summary textbooks-urls fetch-textbooks textbooks-stats textbooks-verify \ crawl-textbooks crawl-textbooks-stats textbook textbook-list bench-jaggedness \ - monitor-poll monitor-graph monitor-access + monitor-poll monitor-graph monitor-access \ + bootstrap-object-store cold-pack cold-pack-dvd cold-unpack cold-stats all: bootstrap fetch-cur ingest-cur verify stats ## bootstrap → fetch cur → ingest cur → verify → stats @@ -846,6 +847,42 @@ search: bootstrap ## keyword search; override SEARCH_Q (or pass Q=...) stats: bootstrap ## counts: documents, chunks, edges, audit chain $(ARBORIST) --db $(DB) stats +# --- cold-object-store tier (ticket #000061) --------------------------------- +# +# Push chunk bodies to an S3-compatible bucket so the corpus can grow past +# one machine while the Merkle tree stays intact. One backend covers AWS S3, +# DO Spaces, R2, B2, GCS (S3 interop), MinIO — different `endpoint_url`. +# +# Required env (set in your shell, never hard-code into Makefile vars): +# AWS_ACCESS_KEY_ID, AWS_SECRET_ACCESS_KEY — standard boto3 discovery +# ARBORIST_COLD_ENDPOINT_URL — e.g. https://nyc3.digitaloceanspaces.com +# ARBORIST_COLD_BUCKET — e.g. arborist-corpus +# ARBORIST_COLD_REGION — optional (DO Spaces is region-agnostic) +# +# Operation Voyeur: credentials NEVER appear on argv, in logs, in audit body. +# Only endpoint URL + bucket name surface; keys live in env / ~/.aws/credentials. +bootstrap-object-store: bootstrap ## install [object-store] (boto3) into the venv + $(PIP) install -e '.[object-store]' + +cold-pack: bootstrap ## bundle hot chunks into tar.zst pack(s) on bucket (≤4.4 GB DVD-R safe-fit each by default) + $(ARBORIST) --db $(DB) cold pack \ + $(if $(MAX_CHUNKS),--max-chunks $(MAX_CHUNKS),) \ + $(if $(MAX_PACK_BYTES),--max-pack-bytes $(MAX_PACK_BYTES),) \ + $(if $(LOCAL_DIR),--local-dir $(LOCAL_DIR),) + +cold-pack-dvd: bootstrap ## write packs to LOCAL_DIR for burning (no S3 upload); each pack ≤4.7 GB + @if [ -z "$(LOCAL_DIR)" ]; then echo "LOCAL_DIR= required"; exit 2; fi + $(ARBORIST) --db $(DB) cold pack --no-push --local-dir $(LOCAL_DIR) + @echo ">> packs written to $(LOCAL_DIR) — burn each .tar.zst with growisofs:" + @echo ">> growisofs -dvd-compat -Z /dev/sr0 $(LOCAL_DIR)/arborist-pack-.tar.zst" + +cold-unpack: bootstrap ## pull pack PACK= and restore chunks locally + @if [ -z "$(PACK)" ]; then echo "PACK= required"; exit 2; fi + $(ARBORIST) --db $(DB) cold unpack $(PACK) + +cold-stats: bootstrap ## bucket summary: chunk count, pack count, endpoint + $(ARBORIST) --db $(DB) cold stats + test: bootstrap ## run pytest suite (excludes opt-in crawler tests) $(VENV)/bin/pytest -q --ignore=tests/crawler -n auto diff --git a/arborist/cli.py b/arborist/cli.py index fa18508..124ce5f 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -2761,6 +2761,145 @@ def _cmd_rehydrate(args: argparse.Namespace) -> int: return 1 if drift else 0 +# --- cold-object-store CLI (#000061) --------------------------------------- +# +# Backend config comes from env vars; credentials use the standard boto3 +# discovery chain (AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY env, or +# ~/.aws/credentials). Per Operation Voyeur (CLAUDE.md): credentials NEVER +# print to stdout, NEVER pass as CLI argv. Only endpoint URL + bucket name +# appear in logs. +# +# Example for DigitalOcean Spaces: +# export AWS_ACCESS_KEY_ID= +# export AWS_SECRET_ACCESS_KEY= +# export ARBORIST_COLD_ENDPOINT_URL=https://nyc3.digitaloceanspaces.com +# export ARBORIST_COLD_BUCKET=arborist-corpus +# arborist cold push --source-type html + + +def _make_cold_backend(): + """Build a backend from env vars; raise with a useful message if missing. + + NEVER reads or prints the credential env vars — boto3 handles those + internally. We only touch the public-facing config (endpoint + bucket). + """ + import os + + endpoint = os.environ.get("ARBORIST_COLD_ENDPOINT_URL") + bucket = os.environ.get("ARBORIST_COLD_BUCKET") + region = os.environ.get("ARBORIST_COLD_REGION") + + if not endpoint or not bucket: + raise SystemExit( + "cold backend requires ARBORIST_COLD_ENDPOINT_URL and " + "ARBORIST_COLD_BUCKET env vars.\n" + "Credentials come from AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY " + "or ~/.aws/credentials (boto3 standard discovery)." + ) + + from arborist.cold_object import S3CompatibleBackend + return S3CompatibleBackend( + endpoint_url=endpoint, bucket=bucket, region=region + ) + + +def _cmd_cold_pack(args: argparse.Namespace) -> int: + from arborist.evict import push_pack + + # When --no-push is set, --local-dir is required (otherwise we'd build + # packs and immediately discard them). + if not args.push_to_bucket and not args.local_dir: + print( + "--no-push requires --local-dir (otherwise the packs go nowhere)", + file=sys.stderr, + ) + return 2 + + # Backend is only contacted when we're pushing — local-dir-only runs + # work without bucket credentials. + backend = ( + _make_cold_backend() if args.push_to_bucket else _LocalOnlyBackend() + ) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) + try: + result = push_pack( + conn, + backend, + document_root=args.document_root, + max_chunks=args.max_chunks if args.max_chunks > 0 else None, + max_pack_bytes=args.max_pack_bytes, + local_dir=args.local_dir, + push_to_bucket=args.push_to_bucket, + ) + finally: + conn.close() + print(json.dumps(result, indent=2, ensure_ascii=False)) + return 0 + + +class _LocalOnlyBackend: + """Identity-only stub for --no-push runs. + + `push_pack` still asks the backend for its identity (folds into audit + body) even when uploads are disabled. This stub answers that one + question without requiring real bucket credentials, and raises if any + actual S3 method is called — fail-loud rather than silently misroute + bytes. + """ + + @property + def identity(self): + from arborist.cold_object import BackendIdentity + return BackendIdentity( + endpoint_url="local-only://", + bucket="(no bucket)", + region=None, + ) + + def __getattr__(self, name): + raise RuntimeError( + f"--no-push run tried to call backend.{name}; that's a bug " + "in push_pack — local-only paths must skip every S3 call" + ) + + +def _cmd_cold_unpack(args: argparse.Namespace) -> int: + from arborist.evict import pull_pack + + backend = _make_cold_backend() + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) + try: + result = pull_pack(conn, backend, args.pack_hash) + finally: + conn.close() + print(json.dumps(result, indent=2, ensure_ascii=False)) + return 0 + + +def _cmd_cold_stats(args: argparse.Namespace) -> int: + from arborist.cold_object import PACK_PREFIX + + backend = _make_cold_backend() + pack_count = 0 + for key in backend.list_keys(PACK_PREFIX): + if key.endswith(".tar.zst"): + pack_count += 1 + out = { + "backend": backend.identity.to_audit_body(), + "packs": pack_count, + } + print(json.dumps(out, indent=2, ensure_ascii=False)) + return 0 + + def _cmd_activity(args: argparse.Namespace) -> int: """Recent activity: Q&A records + freshly cached docs across all shards. @@ -5480,6 +5619,62 @@ def build_parser() -> argparse.ArgumentParser: ) rehydrate_cmd.set_defaults(func=_cmd_rehydrate) + # --- cold-pack distribution tier (#000061): tarballs in S3, hydrate + # new peers from empty via a small number of HTTPS GETs. + cold_cmd = sub.add_parser( + "cold", + help="build/distribute tar.zst corpus packs via an S3-compatible bucket (DO Spaces, AWS S3, R2, B2, GCS, MinIO) or local disk for DVD burning", + ) + cold_sub = cold_cmd.add_subparsers(dest="cold_subcommand", required=True) + + cold_pack = cold_sub.add_parser( + "pack", + help=( + "bundle hot chunks into tar.zst pack(s), ≤4.4 GB DVD-R safe-fit " + "per pack by default — push to bucket and/or write locally for burning" + ), + ) + cold_pack.add_argument("--document-root", default=None) + cold_pack.add_argument( + "--max-chunks", type=int, default=0, + help="optional chunk-count cap per pack (0 = unlimited; the byte cap " + "is the primary control)", + ) + cold_pack.add_argument( + "--max-pack-bytes", dest="max_pack_bytes", type=int, + default=4_400_000_000, + help="compressed-bytes cap per pack — streaming zstd peeks the " + "compressed buffer after each chunk and finalizes when ≥ cap, so " + "each pack fills the disc. Default 4_400_000_000 = DVD-R safe-fit " + "(4.4 GB, ~6.5%% buffer below the 4.7 GB media spec to absorb " + "ISO9660 overhead, media variance, and drive-edge refusal). " + "BD-R = 24_000_000_000; BD-R DL = 48_000_000_000.", + ) + cold_pack.add_argument( + "--local-dir", dest="local_dir", default=None, + help="write each pack to this directory as arborist-pack-.tar.zst " + "(+ manifest sidecar) for burning to physical media. Independent of " + "--no-push.", + ) + cold_pack.add_argument( + "--no-push", dest="push_to_bucket", action="store_false", + help="skip the bucket upload — only write locally via --local-dir", + ) + cold_pack.set_defaults(func=_cmd_cold_pack, push_to_bucket=True) + + cold_unpack = cold_sub.add_parser( + "unpack", + help="pull one pack from bucket and restore chunks locally (hash-verified)", + ) + 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( + "stats", + help="bucket summary: pack count + backend identity (no credentials)", + ) + cold_stats.set_defaults(func=_cmd_cold_stats) + activity_cmd = sub.add_parser( "activity", help="recent Q&A + freshly cached docs (agent-readable timeline)", diff --git a/arborist/cold_object.py b/arborist/cold_object.py new file mode 100644 index 0000000..71a4402 --- /dev/null +++ b/arborist/cold_object.py @@ -0,0 +1,510 @@ +"""Object-store backend for the cold-pack distribution tier (#000061). + +The bucket stores **tar.zst packs** — corpus tarballs that bring up a new +peer from empty. Each pack ≤ 4.4 GB (DVD-R safe-fit) so the same artifact +also burns to physical media. Pack key = `hash_leaf(manifest_bytes)`; +content-addressed at the pack layer, so two writers producing the same +chunk set produce the same key (idempotent upload). + +Why packs and not individual blobs: + +- For new-peer hydration the right granularity is one HTTPS GET per disc- + worth of corpus, not one per chunk. 14M individual GETs vs 12 pack GETs + is a million-x cost difference on request fees alone. +- Packs are the backup unit. The same .tar.zst that sits in the bucket + burns directly to DVD — bit-for-bit identical, content-addressed + filename. No re-encoding for physical media. +- Storage stays compressed. Bucket bytes ≈ on-disk compressed bytes, + not the ~2.2× larger raw UTF-8. + +Multi-cloud unification. One backend class works for AWS S3, DigitalOcean +Spaces, Google Cloud Storage (S3 interop), Cloudflare R2, Backblaze B2, +and MinIO — boto3 with a per-provider `endpoint_url` covers all of them. + +CDN public-read. Bucket ACL = public-read lets any client fetch packs +from the CDN edge without credentials. Verification is local: every +chunk inside the pack has its `leaf_hash` in its tar member name, and +`open_pack` refuses to restore on hash mismatch — the bucket can be +hostile, content addressing binds the bytes. + +Optional dependency. Installing arborist core does NOT pull boto3 — the +backend lives behind the `[object-store]` extras. Tests skip via +`pytest.importorskip("boto3")` when absent. Same gating pattern as +`[html]`, `[nli]`, `[mt]`. + +Operation Voyeur (CLAUDE.md): credentials enter via standard boto3 +discovery (env vars, ~/.aws/credentials, IAM role). They never appear +in subprocess argv, never print to stdout, never enter conversation +logs. Backend identity is endpoint URL + bucket name only. +""" + +from __future__ import annotations + +import abc +import io +import json +import tarfile +from dataclasses import dataclass +from typing import Iterator + +import zstandard + +from arborist.merkle import hash_leaf + + +# Bucket layout — every reader of this module must agree on these constants. +PACK_PREFIX = "packs/" +MANIFEST_PREFIX = "manifest/" +MANIFEST_LATEST_KEY = MANIFEST_PREFIX + "latest.ndjson" + + +def pack_key(pack_hash: str, *, manifest: bool = False) -> str: + """Bucket key for a pack body (default) or its manifest sidecar.""" + if manifest: + return f"{PACK_PREFIX}{pack_hash}.manifest.ndjson" + return f"{PACK_PREFIX}{pack_hash}.tar.zst" + + +@dataclass(frozen=True) +class BackendIdentity: + """Auditable identity for a backend instance. + + Goes into audit_events body — endpoint + bucket are public, credentials + NEVER appear here (Operation Voyeur). + """ + + endpoint_url: str + bucket: str + region: str | None + + def to_audit_body(self) -> dict: + out = {"endpoint_url": self.endpoint_url, "bucket": self.bucket} + if self.region: + out["region"] = self.region + return out + + +class ObjectStoreBackend(abc.ABC): + """ABC for cold-object-store backends. + + All keys here are bucket keys (full path within the bucket). All bodies + are raw bytes — callers handle text encoding before reaching the backend. + + Hash verification is the *caller's* responsibility. The backend does not + decide whether a byte sequence is valid; it only moves bytes. This keeps + the integrity check at exactly one layer (eviction / rehydration code + in `arborist/evict.py`) where the audit chain lives. + """ + + @property + @abc.abstractmethod + def identity(self) -> BackendIdentity: + ... + + @abc.abstractmethod + def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None: + ... + + @abc.abstractmethod + def get(self, key: str) -> bytes: + ... + + @abc.abstractmethod + def head(self, key: str) -> bool: + """True if the object exists; False otherwise. Never raises on missing.""" + + @abc.abstractmethod + def list_keys(self, prefix: str = "") -> Iterator[str]: + ... + + # 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") + + def get_pack(self, pack_hash: str) -> bytes: + return self.get(pack_key(pack_hash)) + + 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 get_pack_manifest(self, pack_hash: str) -> bytes: + return self.get(pack_key(pack_hash, manifest=True)) + + +class S3CompatibleBackend(ObjectStoreBackend): + """boto3-based S3-compatible backend. + + Works across AWS S3, DigitalOcean Spaces, Google Cloud Storage (S3 + interop endpoint), Cloudflare R2, Backblaze B2, MinIO — every + provider that speaks the S3 API. The only per-provider knob is + `endpoint_url`. + + Credentials come from boto3's standard discovery chain (env vars, + ~/.aws/credentials, IAM role). The class never accepts credentials as + plaintext constructor args — that's a Voyeur-protocol guardrail (so + callers can't accidentally leak a key by stringifying the backend). + """ + + def __init__( + self, + *, + endpoint_url: str, + bucket: str, + region: str | None = None, + ): + try: + import boto3 + from botocore.client import Config + from botocore.exceptions import ClientError + except ImportError as e: + raise RuntimeError( + "S3CompatibleBackend requires boto3. " + "Install with: pip install 'arborist[object-store]'" + ) from e + + # Some providers (DO Spaces) want any region string; pick a default + # so signing works. AWS callers always pass their real region. + effective_region = region or "us-east-1" + + self._endpoint_url = endpoint_url + self._bucket = bucket + self._region = region + self._client_error = ClientError + self._client = boto3.client( + "s3", + endpoint_url=endpoint_url, + region_name=effective_region, + config=Config( + signature_version="s3v4", + retries={"max_attempts": 3, "mode": "standard"}, + ), + ) + + @property + def identity(self) -> BackendIdentity: + return BackendIdentity( + endpoint_url=self._endpoint_url, + bucket=self._bucket, + region=self._region, + ) + + def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None: + self._client.put_object( + Bucket=self._bucket, + Key=key, + Body=body, + ContentType=content_type, + ) + + def get(self, key: str) -> bytes: + resp = self._client.get_object(Bucket=self._bucket, Key=key) + return resp["Body"].read() + + def head(self, key: str) -> bool: + try: + self._client.head_object(Bucket=self._bucket, Key=key) + return True + except self._client_error as e: + code = e.response.get("Error", {}).get("Code", "") + if code in ("404", "NoSuchKey", "NotFound"): + return False + raise + + def list_keys(self, prefix: str = "") -> Iterator[str]: + paginator = self._client.get_paginator("list_objects_v2") + for page in paginator.paginate(Bucket=self._bucket, Prefix=prefix): + for obj in page.get("Contents", []) or []: + yield obj["Key"] + + +class MemoryBackend(ObjectStoreBackend): + """In-process backend for tests and local dry-runs. + + Stores bytes in a dict. Same contract as `S3CompatibleBackend` — + same key conventions, same content-addressing invariants. Used by + `tests/test_cold_object.py` so the default test suite runs without + boto3 / moto / a live S3. + + Not a substitute for production — no network, no durability, no ACL + semantics. The on-the-wire test is `tests/test_cold_object_boto3.py` + which uses moto when available. + """ + + def __init__(self, *, endpoint_url: str = "memory://", bucket: str = "test"): + self._endpoint_url = endpoint_url + self._bucket = bucket + self._store: dict[str, bytes] = {} + + @property + def identity(self) -> BackendIdentity: + return BackendIdentity( + endpoint_url=self._endpoint_url, + bucket=self._bucket, + region=None, + ) + + def put(self, key: str, body: bytes, *, content_type: str = "application/octet-stream") -> None: + self._store[key] = bytes(body) + + def get(self, key: str) -> bytes: + if key not in self._store: + raise KeyError(key) + return self._store[key] + + def head(self, key: str) -> bool: + return key in self._store + + def list_keys(self, prefix: str = "") -> Iterator[str]: + # Sort for stable iteration (S3 list returns lex order too). + for key in sorted(self._store): + if key.startswith(prefix): + yield key + + def _tamper(self, key: str, new_body: bytes) -> None: + """Test-only helper: overwrite a key with arbitrary bytes so the + rehydration hash-mismatch path can be exercised.""" + self._store[key] = new_body + + +# --------------------------------------------------------------------------- +# Pack format. A pack is a zstd-compressed tar containing: +# - blobs// one entry per chunk, body = raw UTF-8 +# - manifest.ndjson contents sidecar (written last in the +# streaming builder; consumers iterate +# and skip non-blob entries, so order +# doesn't matter) +# Pack is keyed by `hash_leaf(manifest_bytes)` so identical chunk sets from +# two writers collide idempotently. +# +# build_pack: in-memory builder for small/test cases (whole tar in RAM). +# stream_packs: streaming builder that caps each pack's COMPRESSED size at +# max_compressed_bytes, so packs fill physical media (4.4 GB +# DVD-R safe-fit) instead of leaving ~50% of every disc empty. +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class PackEntry: + leaf_hash: str + size: int + + +@dataclass(frozen=True) +class Pack: + pack_hash: str + manifest_bytes: bytes + body_bytes: bytes + entries: tuple[PackEntry, ...] + + +def build_pack(chunks: list[tuple[str, bytes]]) -> Pack: + """Bundle a list of (leaf_hash, raw_utf8_body) pairs into one pack. + + Every input must already be hash-verified by the caller — + `sha256(body) == leaf_hash`. The builder does NOT re-verify; that + contract belongs to the eviction code so verification stays at one + layer. + + Returns a `Pack` with `pack_hash = hash_leaf(manifest_bytes).hex()` — + content-addressed so two writers producing the same chunk set produce + the same pack key (idempotent overwrite on push). The manifest is + canonicalized via leaf-hash sort so input order doesn't affect the key. + """ + if not chunks: + raise ValueError("build_pack requires at least one chunk") + + # Canonicalize: dedupe by leaf_hash, sort, so input order is irrelevant. + by_hash: dict[str, bytes] = {} + for leaf_hash, body in chunks: + if len(leaf_hash) != 64: + raise ValueError(f"bad leaf_hash length: {leaf_hash}") + by_hash[leaf_hash] = body + + manifest_lines = [] + entries = [] + for leaf_hash in sorted(by_hash): + body = by_hash[leaf_hash] + entries.append(PackEntry(leaf_hash=leaf_hash, size=len(body))) + manifest_lines.append( + json.dumps({"leaf_hash": leaf_hash, "size": len(body)}, sort_keys=True) + ) + manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8") + pack_hash = hash_leaf(manifest_bytes).hex() + + # Build the tar in memory. Sort entries by leaf_hash so the byte + # stream is deterministic for a given input set. + tar_buf = io.BytesIO() + with tarfile.open(fileobj=tar_buf, mode="w") as tar: + info = tarfile.TarInfo(name="manifest.ndjson") + info.size = len(manifest_bytes) + tar.addfile(info, io.BytesIO(manifest_bytes)) + for leaf_hash in sorted(by_hash): + body = by_hash[leaf_hash] + info = tarfile.TarInfo(name=f"blobs/{leaf_hash[:2]}/{leaf_hash[2:]}") + info.size = len(body) + tar.addfile(info, io.BytesIO(body)) + raw_tar = tar_buf.getvalue() + + # zstd-compress the tarball. Level 3 matches `compress.py` discipline. + body_bytes = zstandard.ZstdCompressor(level=3).compress(raw_tar) + + return Pack( + pack_hash=pack_hash, + manifest_bytes=manifest_bytes, + body_bytes=body_bytes, + entries=tuple(entries), + ) + + +def stream_packs( + chunks: Iterator[tuple[str, bytes]], + *, + max_compressed_bytes: int, + level: int = 3, +) -> Iterator[Pack]: + """Stream (leaf_hash, raw_utf8_body) chunks into packs that each fit in + `max_compressed_bytes` *compressed* bytes. + + Why streaming: the in-memory `build_pack` caps by uncompressed bytes, + which means a 4.4 GB DVD-R cap produces ~1.5 GB compressed packs — + every disc burns at ~35 % of capacity. Streaming lets us peek the + compressed-output size as chunks accumulate and cut at the cap, so + every disc fills. + + How: a `ZstdCompressor.stream_writer` writes into an in-memory buffer; + a streaming tar writes into the compressor. After each chunk we call + `flush(FLUSH_BLOCK)` to force pending compressed bytes out of zstd's + internal buffer (preserving the dictionary for the next block — small + efficiency cost, accurate size accounting). When `buf.tell()` ≥ cap, + we finalize: write the manifest as the LAST tar member, close tar, + close stream_writer, yield the `Pack`, start a new one. + + Manifest goes last because the chunk set isn't known until the cap is + hit — `open_pack` iterates by member name and ignores order so this is + transparent to consumers. + + Caveat: each finalize emits tar trailer (~1 KB) + zstd frame footer + (~10 B) AFTER our last size check, so the actual compressed size can + overshoot the cap by ~1 KB. Trivial for a 4.4 GB cap; documented so + future tightening is intentional. + + Single chunks larger than the cap still get their own pack (we don't + drop data — the alternative is silently losing it). + """ + if max_compressed_bytes <= 0: + raise ValueError("max_compressed_bytes must be positive") + + def _start_pack(): + compressed_buf = io.BytesIO() + cctx = zstandard.ZstdCompressor(level=level) + # closefd=False: closing the stream_writer must not close our + # underlying BytesIO (we still need to read getvalue() from it). + writer = cctx.stream_writer(compressed_buf, closefd=False) + tar = tarfile.open(fileobj=writer, mode="w|") + return { + "buf": compressed_buf, + "writer": writer, + "tar": tar, + "entries_by_hash": {}, # dedupe by leaf_hash like build_pack + } + + def _finalize_pack(state) -> Pack: + entries_sorted = sorted(state["entries_by_hash"].keys()) + manifest_lines = [] + entries = [] + for leaf_hash in entries_sorted: + size = state["entries_by_hash"][leaf_hash] + entries.append(PackEntry(leaf_hash=leaf_hash, size=size)) + manifest_lines.append( + json.dumps({"leaf_hash": leaf_hash, "size": size}, sort_keys=True) + ) + manifest_bytes = ("\n".join(manifest_lines) + "\n").encode("utf-8") + info = tarfile.TarInfo(name="manifest.ndjson") + info.size = len(manifest_bytes) + state["tar"].addfile(info, io.BytesIO(manifest_bytes)) + state["tar"].close() + state["writer"].close() + body_bytes = state["buf"].getvalue() + pack_hash = hash_leaf(manifest_bytes).hex() + return Pack( + pack_hash=pack_hash, + manifest_bytes=manifest_bytes, + body_bytes=body_bytes, + entries=tuple(entries), + ) + + state = _start_pack() + for leaf_hash, body in chunks: + if len(leaf_hash) != 64: + raise ValueError(f"bad leaf_hash length: {leaf_hash}") + # Dedupe within a pack — same hash → same body by construction; + # the second copy is wasted bytes. + if leaf_hash in state["entries_by_hash"]: + continue + info = tarfile.TarInfo(name=f"blobs/{leaf_hash[:2]}/{leaf_hash[2:]}") + info.size = len(body) + state["tar"].addfile(info, io.BytesIO(body)) + state["entries_by_hash"][leaf_hash] = len(body) + # Force the compressor to emit pending bytes so .tell() reflects + # reality. FLUSH_BLOCK preserves dictionary state (vs FLUSH_FRAME + # which would close the frame and lose compression continuity). + state["writer"].flush(zstandard.FLUSH_BLOCK) + if state["buf"].tell() >= max_compressed_bytes: + yield _finalize_pack(state) + state = _start_pack() + + if state["entries_by_hash"]: + yield _finalize_pack(state) + + +def open_pack(body_bytes: bytes) -> Iterator[tuple[str, bytes]]: + """Stream (leaf_hash, body) pairs from a pack body. + + Verifies each chunk's hash against the leaf_hash encoded in its + tar member name. Raises ValueError on mismatch — bucket corruption + OR tampering. Callers should additionally cross-check against the + pack's manifest, which carries declared sizes. + + Uses streaming decompression because packs built by `stream_packs` + don't carry a content-size frame header (the streaming compressor + doesn't know the total size up front), and the one-shot + `ZstdDecompressor.decompress(body)` requires that header. Streaming + works on either header presence. + """ + dctx = zstandard.ZstdDecompressor() + raw_tar = dctx.stream_reader(io.BytesIO(body_bytes)).read() + with tarfile.open(fileobj=io.BytesIO(raw_tar), mode="r") as tar: + for member in tar: + if member.name == "manifest.ndjson": + continue + if not member.name.startswith("blobs/"): + continue + 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 + body = f.read() + actual_hash = hash_leaf(body).hex() + if actual_hash != expected_hash: + raise ValueError( + f"pack chunk hash mismatch: declared {expected_hash}, " + f"actual {actual_hash}" + ) + yield expected_hash, body + + +def parse_manifest(manifest_bytes: bytes) -> list[PackEntry]: + """Parse manifest.ndjson into PackEntry list. Lenient on blank lines.""" + out = [] + for line in manifest_bytes.decode("utf-8").splitlines(): + if not line.strip(): + continue + rec = json.loads(line) + out.append(PackEntry(leaf_hash=rec["leaf_hash"], size=int(rec["size"]))) + return out diff --git a/arborist/evict.py b/arborist/evict.py index 82d67c3..7576c4f 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -16,13 +16,16 @@ from __future__ import annotations import sqlite3 import time -from typing import Callable, Iterable +from typing import Callable, Iterable, TYPE_CHECKING -from arborist.compress import pack_chunk +from arborist.compress import pack_chunk, unpack_chunk from arborist.document import canonicalize, get_chunker from arborist.merkle import MerkleTree, hash_leaf from arborist.store import append_audit, transaction +if TYPE_CHECKING: + from arborist.cold_object import ObjectStoreBackend + # Re-fetcher signature: takes a URI, returns parsed/canonicalized text or None. Fetcher = Callable[[str], str | None] @@ -226,3 +229,292 @@ def rehydrate( ) return {"status": "rehydrated", "chunks_restored": restored} + + +# --------------------------------------------------------------------------- +# Cold-pack distribution tier (#000061): bundle chunks into tar.zst packs +# and push them to an S3-compatible bucket so new peers can hydrate from +# empty via a small number of HTTPS GETs (one per pack, ≤4.4 GB DVD-R +# safe-fit each by default — so the same artifact also burns to physical +# media). +# +# Packs are content-addressed by their manifest hash, so two writers +# producing the same chunk set produce the same pack_hash and the upload +# is idempotent. Packs are point-in-time snapshots — a re-pack after +# falsifications produces a different pack_hash, and stale packs stay in +# the bucket until explicitly garbage-collected. +# --------------------------------------------------------------------------- + + +# DVD-R single-layer is marketed as 4.7 GB (= 4,700,000,000 bytes), with the +# ECMA-267 physical capacity at 4,706,074,624 bytes. Targeting either of those +# numbers directly is unsafe in practice — what actually eats into the burn: +# +# - ISO9660 / UDF filesystem overhead ~10–20 MB +# - growisofs lead-in / lead-out ~10 MB +# - Media manufacturing variance ~1–2 % +# - Older drives refusing the outer edge ~1–3 % +# +# Industry-standard safe values for "fit on a DVD-R" workflows: HandBrake's +# DVD-5 preset = 4,377 MiB ≈ 4.59 GB; DVDFab's fit-to-DVD-5 = 4,300 MB; +# `mkisofs` default DVD target = 4,377 MiB. We pick 4,400,000,000 bytes (4.4 +# GB, ~6.5 % buffer below marketing) — clean round number, sits between the +# tool defaults, covers filesystem overhead + media variance + drive-edge +# refusal in one safety margin. +# +# The cap applies to *compressed* bytes per pack — `stream_packs` uses a +# streaming zstd writer and peeks the compressed buffer size after each +# chunk via FLUSH_BLOCK, finalizing the pack when the cap is reached. This +# fills each disc to ~4.4 GB of actual recorded data instead of leaving +# ~50 % of every disc empty (which is what an uncompressed cap produces on +# prose corpora where zstd-3 hits ~0.45 ratio). +DEFAULT_MAX_PACK_BYTES = 4_400_000_000 # 4.4 GB — DVD-R safe-fit (~6.5 % buffer) + + +def push_pack( + conn: sqlite3.Connection, + backend: "ObjectStoreBackend", + *, + document_root: str | None = None, + leaf_hashes: Iterable[str] | None = None, + max_chunks: int | None = None, + max_pack_bytes: int = DEFAULT_MAX_PACK_BYTES, + local_dir: str | None = None, + push_to_bucket: bool = True, +) -> dict: + """Bundle local chunks into one or more tar.zst packs. + + Selection (one of): + - `document_root=R` — all chunks of one document with local content + - `leaf_hashes=[...]` — explicit hash list (content must be present) + - neither — every hot surface chunk in this DB (one-pack-per-shard, + the natural unit for bulk distribution). + + `max_pack_bytes` caps *compressed* bytes per pack — `stream_packs` + streams chunks through a zstd writer, peeks at the compressed buffer + size after each chunk (FLUSH_BLOCK preserves dictionary), and + finalizes when the cap is reached. So each pack fills the disc to + its rated capacity, not 30–50 % of it. Default 4_400_000_000 = DVD-R + safe-fit (~6.5 % below the 4.7 GB marketing capacity). Set higher + for larger media (BD-R = 24_000_000_000), lower for thumb-drive + distribution. + + `max_chunks` is an additional optional chunk-count cap (kept for back- + compat; the byte cap is now the primary control). + + `local_dir` writes each pack to disk as `arborist-pack-.tar.zst` + (+ manifest sidecar) for burning workflows — `growisofs -dvd-compat -Z + /dev/sr0 .tar.zst`. Independent of `push_to_bucket`: set both for + both, only one for either. + + Returns a dict with `packs: list[{pack_hash, chunk_count, + uncompressed_bytes, compressed_bytes, ...}]`. Idempotent: same chunk + grouping → same pack_hashes → bucket overwrite is a no-op. + """ + from pathlib import Path + + from arborist.cold_object import stream_packs + from arborist.snapshot import compute_snapshot_root + + # Default selection = every chunk with local content (surfaces AND + # cores). Cores never evict and carry the distilled-from-surface + # derivations a new peer needs to bootstrap the v9.8 chain — packing + # them is the whole point of the corpus-hydration tarball. + where = ["c.content IS NOT NULL"] + params: list = [] + if document_root is not None: + where.append("c.document_root = ?") + params.append(document_root) + elif leaf_hashes is not None: + hashes = list(leaf_hashes) + if not hashes: + return {"status": "nothing_to_pack", "packs": []} + placeholders = ",".join("?" for _ in hashes) + where.append(f"c.leaf_hash IN ({placeholders})") + params.extend(hashes) + else: + where.append("c.tier = 'hot'") + + sql = ( + "SELECT c.leaf_hash, c.content FROM chunks c " + "JOIN documents d ON d.document_root = c.document_root " + "WHERE " + " AND ".join(where) + ) + if max_chunks is not None and max_chunks > 0: + sql += " LIMIT ?" + params = [*params, max_chunks] + rows = conn.execute(sql, params).fetchall() + if not rows: + return {"status": "nothing_to_pack", "packs": []} + + out_dir: Path | None = None + if local_dir: + out_dir = Path(local_dir) + out_dir.mkdir(parents=True, exist_ok=True) + + backend_id = backend.identity.to_audit_body() + pack_results: list[dict] = [] + skipped_hash_mismatch = 0 + # Bind every pack in this run to the point-in-time snapshot_root. + # Packs are delayed snapshots — falsifications between repacks produce + # a different snapshot_root + a different pack_hash. The audit row pins + # which corpus state this pack covers; consumers can `arborist snapshot + # verify ` to detect drift on unpack. + snapshot_root, doc_count_at_pack = compute_snapshot_root(conn) + pack_ts = int(time.time()) + + # Track per-pack uncompressed-bytes accumulation by closing over a + # mutable cell that the chunk-source generator updates. `stream_packs` + # cuts on compressed bytes so it doesn't need this — we expose it for + # audit telemetry (compression ratio per pack). + uncompressed_running = [0] + last_finalized_uncompressed = [0] + + def _chunk_source(): + for row in rows: + text = unpack_chunk(row["content"]) + if text is None: + continue + body = text.encode("utf-8") + if hash_leaf(body).hex() != row["leaf_hash"]: + nonlocal_skipped[0] += 1 + continue + uncompressed_running[0] += len(body) + yield (row["leaf_hash"], body) + + # closure-by-list because the generator can't see the outer counter + # easily and we want one number per pack at finalize time. + nonlocal_skipped = [0] + + for pack in stream_packs(_chunk_source(), max_compressed_bytes=max_pack_bytes): + pack_uncompressed = uncompressed_running[0] - last_finalized_uncompressed[0] + last_finalized_uncompressed[0] = uncompressed_running[0] + if push_to_bucket: + backend.put_pack(pack.pack_hash, pack.body_bytes) + backend.put_pack_manifest(pack.pack_hash, pack.manifest_bytes) + if out_dir is not None: + short = pack.pack_hash[:16] + (out_dir / f"arborist-pack-{short}.tar.zst").write_bytes(pack.body_bytes) + (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": len(pack.body_bytes), + "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, + }, + ) + pack_results.append({ + "pack_hash": pack.pack_hash, + "chunk_count": len(pack.entries), + "uncompressed_bytes": pack_uncompressed, + "compressed_bytes": len(pack.body_bytes), + "snapshot_root": snapshot_root, + "audit_event_hash": event_hash, + }) + + skipped_hash_mismatch = nonlocal_skipped[0] + + if not pack_results: + return {"status": "nothing_to_pack", "packs": []} + + return { + "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), + "total_compressed_bytes": sum(p["compressed_bytes"] for p in pack_results), + "snapshot_root": snapshot_root, + "snapshot_doc_count": doc_count_at_pack, + "snapshot_ts": pack_ts, + "skipped_hash_mismatch": skipped_hash_mismatch, + "max_pack_bytes": max_pack_bytes, + "local_dir": str(out_dir) if out_dir else None, + } + + +def pull_pack( + conn: sqlite3.Connection, + 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. + + 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. + + Returns count restored and count skipped (already hot / unknown locally). + """ + from arborist.cold_object import open_pack + + body = backend.get_pack(pack_hash) + 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. + rows = conn.execute( + "SELECT chunk_id, tier FROM chunks WHERE leaf_hash = ?", + (leaf_hash,), + ).fetchall() + if not rows: + skipped_unknown += 1 + continue + text = chunk_body.decode("utf-8") + for row in rows: + if row["tier"] == "hot": + 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 (?, ?)", + (row["chunk_id"], text), + ) + restored += 1 + + with transaction(conn): + append_audit( + conn, + event_type="cold_pack_pulled", + body={ + "pack_hash": pack_hash, + "chunks_restored": restored, + "chunks_skipped_already_hot": skipped_already_hot, + "chunks_skipped_unknown_locally": skipped_unknown, + "chunks_drifted": drifted, + "backend": backend_id, + }, + ) + + return { + "status": "pulled", + "pack_hash": pack_hash, + "chunks_restored": restored, + "chunks_skipped_already_hot": skipped_already_hot, + "chunks_skipped_unknown_locally": skipped_unknown, + "drift_details": drift_details, + } diff --git a/docs/TICKETS.md b/docs/TICKETS.md index d2a94a2..617f22c 100644 --- a/docs/TICKETS.md +++ b/docs/TICKETS.md @@ -111,6 +111,7 @@ Newest first. Update on every open/close. | ID | Title | Status | Opened | Directive | |----------|------------------------------------------------|-----------------------|------------|-----------| +| #000061 | Cold-pack distribution tier (boto3 S3-compat, DO Spaces + DVD-R targets) | **in progress** — opened 2026-05-25 (fox: "implement it now … target digital ocean first as a test"; later "I wanted a way to hydrate using tarballs (the core and important data) for bringing new machines up"). Tarball-only distribution mechanism — bucket holds `tar.zst` packs keyed by `hash_leaf(manifest)`, no individual-chunk blobs. New peers hydrate by downloading packs from the bucket's CDN edge (~4 HTTPS GETs for the current ~14.1M-chunk corpus, packs filled to 4.4 GB compressed each via streaming zstd, vs ~14M for individual blobs). Same artifact ≤4.4 GB safe-fit (~6.5 % buffer below DVD-R's 4.7 GB marketing capacity, accommodating ISO9660 overhead + media variance + drive-edge refusal) burns directly to physical media via `--local-dir` + `growisofs`. Packs are *delayed* snapshots: each pack pins the corpus `snapshot_root` it covers in audit + body, so falsifications between repacks produce new pack_hashes and stale packs stay in the bucket until explicit GC (future ticket). Packs include cores AND surfaces (full-corpus hydration). Default selection covers every hot chunk with local content in the shard. Multi-pack splitting via `stream_packs` (streaming zstd, FLUSH_BLOCK peek of compressed buffer after each chunk, cut at cap) fills each disc to ~4.4 GB compressed instead of leaving ~50% empty. One backend class (`S3CompatibleBackend` via boto3 + `endpoint_url`) covers AWS S3, DO Spaces, GCS S3-interop, R2, B2, MinIO. CDN public-read makes packs accessible to anyone; hash binding via in-tar `leaf_hash` member names makes hostile-bucket scenarios safe. Optional dep `[object-store]` = boto3>=1.34. Voyeur: credentials via standard `AWS_ACCESS_KEY_ID`/`_SECRET_ACCESS_KEY` (env or `~/.aws/credentials`), never printed; only endpoint URL + bucket name surface in logs. Initial individual-blob path (per-chunk S3 objects) was scoped+landed then **deleted same day** (fox: "what ever was blobs? I wanted a way to hydrate using tarballs"); the five-step deletion record lives inline in the doc — we'd added 14M-object storage and ~$70/hydrate request cost for a workflow that needed neither. Sizing math for current shards: ~4 packs total (17.2 GB compressed ÷ 4.4 GB compressed per pack), ~17 GB bucket storage, ~$0.34/mo DO Spaces. | 2026-05-25 | — | | #000060 | H-ABCDEFG same-model substrate-delta harness (+ jaggedness tensor + curvature) | open · awaiting go/no-go (2026-05-20; from Dav1dPrometheus *Protocol-Layer AGI* working report §26/§50/§82-84). The report's "decisive proof": run the SAME base model substrate-OFF vs substrate-ON over long-horizon/adversarial/non-jagged batteries, report the delta. Two new metrics: jaggedness tensor `J_norm` (§73 — variance across nearby variants, normalized by difficulty) + discrete performance curvature `κ_t` (§5.2, with the honest no-global-convexity bound, Erratum 5). A-vs-C spine (B optional, D=mesh OUT → #000012/#000016). Curvature-aware ForkScore extension folds into **#000012** (NOT a new ticket — reserved `iota`/`kappa` weight slots already exist). Budget: control arms = Hermes/Qwen, never Opus without go; heavy passes on GPU box. Held-out/mechanism-agnostic variants required so ABCDEFG doesn't self-validate. | 2026-05-20 | — | | #000059 | Admission discipline: claim-graveyard query + self-providence quarantine | open · awaiting go/no-go (2026-05-20; Dav1dPrometheus report §11 Priority 2 + §57 "admissible state transition" thesis). Two coupled write-path mechanisms. **(A) GraveyardCheck:** storage already exists (`falsification_state` failed/stale/quarantined records ARE the graveyard); the gap is burden-shifting — a re-asked claim family with a known `failed` history should require stronger evidence (§42.9/§62-63). **(B) Self-providence quarantine:** `make ingest-self-providence` (Makefile:769) deliberately promotes STRICT records into the corpus — the exact self-confirmation loop §70 warns of — and ships with NO guard; detect/lineage-tag self-providence-descended evidence + quarantine for high-impact claims. Both advisory-sidecar-first (run-DAG only, never `providence_cache`/`audit_events`); demote-hooks bench-gated + unwired pending net win (would fold `governance_policy_hash`). Soft signals never enter the hard proof path. **Bounded-ingestion hard constraint (fox 2026-05-20, §7):** the graveyard MUST reach a steady-state size ∝ the *recurring*-error surface, never queries-ever — earn-to-enter (recurrence-gated), fingerprints not transcripts (UTXO-set analogy), decay/compact (evicts like a surface), off the hot path. BTC's lesson is bounded self-regulating ingestion, not "store everything." Gossip-group falsifier admission inherits difficulty-adjusted stable-rate + per-window budget (#000036) → enforced in #000012/`mesh/`. If it can't be bounded, it isn't built. | 2026-05-20 | — | | #000058 | `cache_key_9` verifier-policy: mandatory-vs-legible decision + doc reconciliation | open · awaiting go/no-go (2026-05-20; Dav1dPrometheus report §2 Erratum 1 / §11 Priority 1 "mandatory cache_key_9"). **Five-step #1 correction:** the report's *correctness* premise is already false in arborist — verifier fields are a subset of the policy dict and so already fold into `governance_policy_hash` (`keys.py:269-275`); a verifier-rule change ALREADY changes the cache_key today. The explicit 9th `verifier_policy_hash` buys **audit legibility**, not correctness — so "mandatory" would stale every prior record for zero correctness gain. Decision: A leave-as-is (8-dim default, 9th optional) + doc reconcile [recommended] · B default-write 9-dim · C flag-staged bench-gated default-write — never a hard mandatory flip. Doc reconcile (CLAUDE.md/concepts "8-dim" → "8 + optional legible 9th") is the do-regardless. | 2026-05-20 | — | @@ -174,4 +175,4 @@ Newest first. Update on every open/close. ## Next ID -`000061` +`000062` diff --git a/docs/cold-object-store.md b/docs/cold-object-store.md new file mode 100644 index 0000000..4a059e1 --- /dev/null +++ b/docs/cold-object-store.md @@ -0,0 +1,259 @@ +# Cold-pack distribution tier (ticket #000061) + +A point-in-time corpus distribution mechanism. arborist serializes its +local chunks into `tar.zst` packs, ships them to an S3-compatible bucket +(and/or to local disk for DVD-burning), and any new peer hydrates by +downloading those packs from the bucket's CDN edge and unpacking them +into a fresh shard. + +## What this is, and what it is not + +**Is:** a backup-and-distribution unit. Pack bytes are content-addressed. +Same chunk set on two writers → same `pack_hash`. The bucket is a +delivery medium for a *delayed* snapshot of the corpus — repackaging +after falsifications produces a new pack with a new hash. + +**Is not:** a live mirror. Packs do not see falsifications that happen +*after* the pack was built. They do not see ingests after the pack was +built. They are frozen artifacts, identified by `snapshot_root` of the +corpus state at pack time. + +**Is not:** an individual-chunk fetch tier. There is no per-chunk URL in +the bucket — corpus chunks live exclusively inside packs. New peers and +backup consumers download packs whole. + +## Hard invariants + +1. **Bucket holds packs only.** Layout: + + ``` + /packs/.tar.zst # pack body + /packs/.manifest.ndjson # pack contents sidecar + ``` + + No `blobs/` prefix, no per-chunk objects. (One pack ↔ one disc ↔ one + bucket object.) + +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 + same chunk set produce the same `pack_hash` — bucket upload is + idempotent, DVD burns at two sites are byte-identical. + +3. **`hash_leaf(chunk_body).hex() == leaf_hash`** is verified on every + `open_pack` member. The pack's tar member name is `blobs//` — + that's a within-tar convention, not a bucket layout. Tampering with + pack bytes is caught at unpack time, never reaches the local DB. + +4. **Every pack pins a `snapshot_root`.** Pack creation reads the + corpus's current snapshot root (`arborist/snapshot.py:compute_snapshot_root`) + and records it in: + - the audit row (`cold_pack_pushed.body.snapshot_root`) + - the `push_pack` return body + - the local-dir filenames implicitly (pack_hash itself encodes the + manifest, which encodes the chunk set, which encodes that snapshot's + content) + + Consumers can run `arborist snapshot verify ` after unpack to + detect drift between the pack and the corpus state on the consuming + node. + +5. **Cores never evict** (CLAUDE.md rule). Packs include cores AND + surfaces — cores carry the distillation derivations a new peer needs + to bootstrap the v9.8 chain. + +6. **No credentials in audit body.** Backend identity is endpoint URL + + bucket name only. Credentials live in env vars / `~/.aws/credentials` + via standard boto3 discovery — Operation Voyeur. + +## Delayed snapshots and falsifications + +Packs are not live. Between two pack runs, three things can happen: + +1. **New ingest.** `ingest_source` adds new documents. They aren't in + the old pack; they show up in the next pack. The old pack stays a + valid snapshot of *its* state. + +2. **Falsification.** Drift detection, `arborist falsify`, or + `rehydrate_drift` flips a `providence_cache` row to + `falsification_state='stale'` and/or marks a document for + re-derivation. Chunk content does NOT change (chunks are immutable; + content-addressed). A new pack covers the same chunk *bytes* but with + a different `providence_cache` view. + +3. **Re-pack.** A new pack run reads the current corpus and produces a + pack with a new `pack_hash` (because the manifest covers a different + chunk set — newly ingested, possibly with the same hashes minus any + superseded ones). + +Three operational consequences: + +- **Stale packs accumulate.** Old `pack_hash`es stay in the bucket + until explicitly garbage-collected. They're still valid snapshots + of past corpus states. There's no automatic cleanup; that's a future + ticket. +- **A peer hydrated from an old pack is honestly old.** It has the + corpus state from the pack's `snapshot_root`. To catch up, it + follows the same path any live peer does — ingest new sources, + receive falsification events on the mesh, re-derive cores. +- **The bucket is eventually consistent with intent**, not with the + live corpus. Re-pack cadence (daily? weekly? per-event?) is an + operational policy, not a code property. + +## Two distribution channels — same artifact + +The same `.tar.zst` file serves two channels: + +| Channel | Transport | Default cap | +|---------------------|----------------------------|---------------| +| **Bucket + CDN** | `S3CompatibleBackend.put_pack` → public-read DO Spaces / R2 / S3, CDN edge serves consumers | 4.4 GB / pack | +| **DVD-R archival** | `--local-dir DIR` → `growisofs -dvd-compat -Z /dev/sr0=` | 4.4 GB / pack | + +Pack files are byte-identical between channels. A DVD burned from one +local-dir pack and a CDN-fetched pack of the same content collide on +`sha256sum`. + +## DO Spaces quickstart + +```bash +# 1. Install the optional backend. +make bootstrap-object-store + +# 2. Set boto3 standard env vars (never hard-code in scripts). +export AWS_ACCESS_KEY_ID= +export AWS_SECRET_ACCESS_KEY= + +# 3. Set bucket config. +export ARBORIST_COLD_ENDPOINT_URL=https://nyc3.digitaloceanspaces.com +export ARBORIST_COLD_BUCKET=arborist-corpus + +# 4. Build packs and push them. Default cap = 4.4 GB / pack (DVD-R safe- +# fit). One shard typically yields 1-3 packs. +make cold-pack + +# 5. Confirm what's in the bucket. +make cold-stats +``` + +Same flow works on AWS S3 (`endpoint_url=https://s3..amazonaws.com`), +Cloudflare R2, Backblaze B2, GCS S3-interop, MinIO. + +## Hydrating a new peer from CDN + +```bash +# 1. On the fresh node, install arborist + the [object-store] extra. +make bootstrap-object-store + +# 2. List packs the publisher made available. +ARBORIST_COLD_ENDPOINT_URL=... ARBORIST_COLD_BUCKET=... \ + arborist cold stats + +# 3. For each pack, unpack into a local shard. Verifies every chunk on +# the way in; bad bytes from a hostile CDN never reach the DB. +for hash in ; do + arborist --db ~/.arborist/shards/000.db cold unpack $hash +done + +# 4. (Optional) Pin which corpus state we're at. +arborist --db ~/.arborist/shards/000.db snapshot list | head -1 +``` + +The snapshot_root the publisher pinned at pack time is in the audit row; +the verifier on the consumer side recomputes `snapshot_root` after +unpack and they should match if the corpus is a clean restore. + +## DVD-R archival workflow + +```bash +# 1. Write packs to a staging dir; skip the bucket entirely. +make cold-pack-dvd LOCAL_DIR=/mnt/dvd-staging + +# 2. Each pack is one disc. Burn with growisofs. +for pack in /mnt/dvd-staging/arborist-pack-*.tar.zst; do + growisofs -dvd-compat -Z /dev/sr0="$pack" + # ... eject, insert next blank, repeat ... +done + +# 3. On a fresh node, copy a pack from disc and unpack: +mount /dev/sr0 /mnt/dvd +arborist --db fresh.db cold unpack \ + "$(basename /mnt/dvd/arborist-pack-*.tar.zst .tar.zst | cut -d- -f3)" +``` + +The `pack_hash` is in the filename (`arborist-pack-.tar.zst`) +so the disc itself is self-describing — no separate index needed. + +## Pack-size cap — fit on a 4.7 GB DVD-R, safely + +Default cap is **4,400,000,000 bytes (4.4 GB, ~6.5 % buffer below the +4.7 GB marketing capacity)**. Targeting 4.7 GB directly is unsafe: +filesystem overhead, media manufacturing variance, growisofs +lead-in/lead-out, and older drives refusing the outer edge all eat +into nominal capacity. 4.4 GB sits between the industry-standard tool +defaults (HandBrake DVD-5 = 4,377 MiB ≈ 4.59 GB; DVDFab fit-to-DVD-5 = +4.3 GB; mkisofs default DVD = 4,377 MiB). + +The cap applies to *compressed* bytes per pack. `stream_packs` uses +streaming zstd compression and peeks the compressed-buffer size after +every chunk (via `FLUSH_BLOCK`, which preserves the compressor's +dictionary so block boundaries cost almost nothing in ratio). When the +buffer reaches the cap, the pack is finalized and a new one starts. So +each disc fills to ~4.4 GB of recorded data, not 30–50 % of capacity. + +Overshoot bound: tar trailer (~1 KB padding) + zstd frame footer (~10 B) +get emitted after the last in-loop size check, so actual compressed +size can land at cap + ~2 KB. Trivial for a 4.4 GB cap. + +For larger media: + +| Media | `--max-pack-bytes` | Marketing | +|----------------------|---------------------------|-----------| +| **DVD-R (default)** | `4_400_000_000` (4.4 GB) | 4.7 GB | +| DVD+R DL | `8_000_000_000` (8.0 GB) | 8.5 GB | +| BD-R | `24_000_000_000` (24 GB) | 25 GB | +| BD-R DL | `48_000_000_000` (48 GB) | 50 GB | + +## 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): + +| 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) | + +Repacking after a falsification event costs the same as the initial +pack — one full corpus serialization per event-batched run, gated by +re-pack cadence (operational policy). + +## Failure modes + +| Symptom | Cause | Recovery | +|------------------------------------------|------------------------------------|--------------------------------------| +| `pack chunk hash mismatch` on unpack | Pack bytes corrupted in transit or on disc | Re-download / re-burn; pack is content-addressed so a fresh fetch is verifiable. | +| `cold pack` produces no packs | No hot chunks with non-null content | `cold pack` operates on local content. Confirm shard isn't empty / fully evicted. | +| Peer's snapshot_root differs from pack's | Local corpus drifted after unpack (ingest, falsification, etc.) | Expected. Pack is a delayed snapshot; the peer has moved on. Re-pack to re-baseline. | +| Bucket missing a pack | GC'd, never uploaded, wrong bucket | Re-build pack from any shard that still has the source content. | + +## Future work + +- **Multipart upload for packs.** Provider single-object limits (DO + Spaces = 5 GB non-multipart, AWS S3 = 5 GB; both support multipart up + to 5 TB). Today's code uses `put_object` which is single-shot. boto3 + `upload_file` is the one-line drop-in. +- **Streaming pack builder.** ✅ Landed as `stream_packs`. Caps target + compressed bytes; each disc fills. `build_pack` stays for tests + + small/known-set callers. +- **Pack GC.** Stale packs (those whose `snapshot_root` is older than N + re-pack cycles) get bucket-deleted automatically. +- **Range-fetch partial pack pulls.** Manifest carries offsets; + `GET .tar.zst Range: bytes=X-Y` would let a consumer pull one chunk + from a huge pack without downloading the whole thing. +- **KMS / SSE-S3.** Server-side encryption (mesh ciphertext on a + public bucket is the v1 confidentiality path). +- **Multi-region replication.** Handled by the provider within a region; + cross-provider replication is a separate distribution-policy question. diff --git a/docs/tickets/ticket-000061-cold-object-store-tier.md b/docs/tickets/ticket-000061-cold-object-store-tier.md new file mode 100644 index 0000000..cad1692 --- /dev/null +++ b/docs/tickets/ticket-000061-cold-object-store-tier.md @@ -0,0 +1,194 @@ +# Ticket #000061 — Cold-pack distribution tier + +**Status:** in progress — opened 2026-05-25 +**Opened:** 2026-05-25 +**Scope:** ship arborist corpus state to new peers (and to DVD-R archival) + via point-in-time `tar.zst` packs hosted on an S3-compatible bucket + and/or burned to physical media. One artifact serves both channels. +**Audience:** dav1d (architectural inflection — new optional dep, new + optional network egress, new public-readable surface when CDN is + enabled, new "delayed snapshot" semantics around falsifications). +**Hard constraint:** packs are content-addressed + (`pack_hash = hash_leaf(manifest_bytes)` over a leaf-hash-sorted, + deduped manifest). Same chunk set → same pack_hash. Each pack pins the + corpus `snapshot_root` it covers in the audit chain — consumers can + detect drift between the pack and current corpus state. + +## Problem + +A new peer comes up with an empty SQLite shard. How does it become a +working arborist node? + +Options today: +1. Re-ingest every source from upstream (Wikipedia dumps, textbooks, + crawls). Hours-to-days; depends on every upstream being reachable. +2. `rsync` someone else's shard. Works but bypasses the audit chain — + the receiving node has no proof the bytes came from a trusted + producer with verifiable provenance. +3. **Download a tarball.** Fast, content-addressed, audit-row-pinned, + verifiable on unpack. + +`snapshot.py` already covers corpus *identity* (the `snapshot_root` +Merkle hash over sorted document_roots). What's missing is the +*delivery* — getting the bytes to a fresh node. + +## Design + +### What goes in the bucket + +Only packs. No individual chunk objects. Layout: + +``` +/packs/.tar.zst # pack body +/packs/.manifest.ndjson # contents sidecar +``` + +Per-pack contents (inside the tar): + +``` +manifest.ndjson # one line per chunk: {leaf_hash, size} +blobs// # one tar member per chunk, raw UTF-8 body +``` + +The `blobs/` prefix inside the tar is an in-pack convention, not a +bucket layout — there is no `blobs/` prefix in the bucket itself. + +### Pack identity and idempotence + +- `pack_hash = hash_leaf(manifest_bytes)` where the manifest is sorted + by `leaf_hash` and deduped before hashing. +- Same chunk set → same pack_hash. Two writers building the same pack + collide on bucket upload — no GC after duplicate runs. +- Pack uploads are idempotent. Pack contents are append-only by + construction. + +### Pack-size cap (DVD-R safe-fit) + +Default `max_pack_bytes = 4_400_000_000` (4.4 GB) — sits ~6.5 % below +the 4.7 GB DVD-R marketing capacity to absorb: + +- ISO9660 / UDF filesystem overhead +- growisofs lead-in / lead-out +- Media manufacturing variance (~1–2 %) +- Older drives refusing the outer edge (~1–3 %) + +Sits between industry-standard tool defaults (HandBrake DVD-5 = 4.59 GB; +DVDFab fit-to-DVD-5 = 4.3 GB; mkisofs default DVD = 4.59 GB). Cap +applies to uncompressed bytes so the compressed `.tar.zst` is ≤ cap by +construction. v1 produces ~30–50 % media fill on prose; future +streaming-compressed cap fills discs better. + +Multi-pack splitting is greedy first-fit by accumulated raw bytes. + +### Two channels — one artifact + +``` + ┌── S3CompatibleBackend.put_pack ── DO Spaces / R2 / S3 +build_pack ─┬────┤ + └────└── --local-dir DIR ─── growisofs ─── /dev/sr0 +``` + +Byte-identical packs in both channels. A pack burned at one site and +fetched from CDN at another collide on `sha256sum`. + +### Delayed-snapshot discipline + +Packs are *not* live mirrors. The bucket is eventually consistent with +intent, not with the live corpus. Three operational consequences: + +1. **Each pack pins a `snapshot_root`** (`compute_snapshot_root(conn)` + at pack time). Recorded in the `cold_pack_pushed` audit row and in + the `push_pack` return body. +2. **Falsifications produce new packs.** Between repacks, drift + detection / `arborist falsify` / `rehydrate_drift` can flip + `providence_cache` rows to `stale`. Chunk content is immutable; what + changes is the corpus-state envelope. A re-pack with the same chunk + set produces a different `snapshot_doc_count` if documents were + added; same hash if not — but the audit row's `snapshot_root` + distinguishes the moments. +3. **Stale packs accumulate.** No automatic GC in v1. Old `pack_hash`es + stay in the bucket as valid snapshots of past corpus states. GC is + a separate ticket once we have a re-pack cadence to GC against. + +### Cores never evict; packs include them + +Packs cover every hot chunk with local content — surfaces AND cores. +Cores carry the distillation derivations a new peer needs to bootstrap +the v9.8 chain. (Earlier draft restricted packs to surfaces only — a +bug that would have shipped a fresh peer with no derivation roots.) + +### Verification on unpack + +Every chunk's `leaf_hash` is the tar member name. `open_pack` recomputes +`hash_leaf(body).hex()` per member and refuses to restore on mismatch. +Hostile-bucket bytes never reach the local DB. + +After unpack, consumers can `arborist snapshot verify ` against +the pack's recorded `snapshot_root` to confirm clean restore (or detect +drift if the local corpus has moved on since). + +## What was deleted from this ticket's first attempt + +The original implementation included per-chunk individual blob storage +(`evict_to_object`, `rehydrate_from_object`, `cold push`, `cold pull`, +`blob_key()`, `BLOB_PREFIX`, `put_chunk`/`get_chunk`/`has_chunk`/ +`list_chunk_hashes` on the backend ABC). Fox's correction same-day +(2026-05-25): "what ever was blobs? I wanted a way to hydrate using +tarballs (the core and important data) for bringing new machines up". + +Five-step deletion record: + +- **Step 1 — Make the requirement less dumb:** the requirement was + always "ship a corpus to a new peer," not "expose every chunk as an + S3 object." Individual blobs solved a problem nobody asked for. +- **Step 2 — Delete the part:** 14M-object storage, ~$70/hydrate + request cost, 14M-entry LIST walks — all deleted. ~250 lines of + source + 7 tests gone. +- **Result:** packs-only design lands at ~17 GB bucket storage (vs + 38 GB for raw blobs), ~4 GETs/hydrate (vs 14M; packs filled to 4.4 GB + compressed each via streaming zstd), DO Spaces request cost falls + from ~$70 to ~$0.00002 per fresh peer. + +## Implementation + +- `arborist/cold_object.py` — `ObjectStoreBackend` ABC (raw bytes, + pack-shaped wrappers) + `S3CompatibleBackend` (boto3) + `MemoryBackend` + (tests) + `build_pack` / `open_pack` / `parse_manifest`. +- `arborist/evict.py` — `push_pack` / `pull_pack`. Per-call binds the + corpus `snapshot_root` into pack audit + return body. +- `arborist/cli.py` — `arborist cold pack | unpack | stats`. +- `tests/test_cold_object.py` — pack invariants, splitting, snapshot + binding, local-dir, no-push mode. Default MemoryBackend so the + default suite runs without boto3. +- `tests/test_cold_object_boto3.py` — moto-mocked S3 wire test + (boto3 + moto gated). +- `pyproject.toml` — `[object-store] = boto3>=1.34`; dev extras pull + `moto>=5.0`. +- `Makefile` — `bootstrap-object-store`, `cold-pack`, `cold-pack-dvd`, + `cold-unpack`, `cold-stats`. +- `docs/cold-object-store.md` — full design, DO Spaces quickstart, + CDN hydrate recipe, DVD-burn recipe, delayed-snapshot discipline, + failure-mode table. + +## Scope boundaries + +In scope: pack build / push / pull / unpack, snapshot binding, hash +verification, audit chain, DVD-R local-dir output, multi-pack +splitting. + +Out of scope (future tickets if needed): +- Multipart upload (single-pack > 5 GB on DO Spaces / AWS). +- Streaming pack builder (lift the in-memory tar ceiling, target + compressed-bytes cap). +- Stale-pack GC. +- Range-fetch partial pack pulls. +- Server-side encryption (KMS / SSE-S3); mesh ciphertext on public + bucket is the v1 confidentiality path. +- Cross-provider replication. + +## 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. diff --git a/pyproject.toml b/pyproject.toml index 07dbdd4..b6172eb 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -103,6 +103,20 @@ nli = [ "protobuf>=4.0", "optimum[onnxruntime]>=1.20", ] +object-store = [ + # Cold-object eviction tier (ticket #000061). Pushes chunk bodies to an + # S3-compatible bucket keyed by leaf_hash so the corpus can grow past + # one machine while the Merkle tree stays intact. One backend covers + # AWS S3, DigitalOcean Spaces, Cloudflare R2, Backblaze B2, GCS + # (S3 interop), and MinIO — boto3 with a per-provider endpoint_url. + # Gated separately so a fresh checkout stays python3.12 + venv + + # sqlite3; tests skip via pytest.importorskip when absent. Install with: + # pip install 'arborist[object-store]' + # Credentials use boto3's standard discovery (env vars, + # ~/.aws/credentials, IAM role); never read into arborist code. The + # wire-level test gated additionally on `moto`. + "boto3>=1.34", +] mt = [ # Local machine-translation engine for the #000056 "Operation # Sandwich" cross-language grounding edges (arborist/qa/mt/): @@ -136,6 +150,10 @@ dev = [ "arborist[math]", "arborist[hessian]", "arborist[vec]", + "arborist[object-store]", + # moto is the wire-level boto3 test stub; only needed to run + # tests/test_cold_object_boto3.py (the default suite uses MemoryBackend). + "moto>=5.0", ] [project.scripts] diff --git a/tests/test_cold_object.py b/tests/test_cold_object.py new file mode 100644 index 0000000..21619b3 --- /dev/null +++ b/tests/test_cold_object.py @@ -0,0 +1,325 @@ +"""Cold-pack distribution tier (#000061) — pack build, push, pull. + +Uses the in-process `MemoryBackend` so the default test suite runs without +boto3, moto, or a live S3. The on-the-wire boto3+moto test lives in +`tests/test_cold_object_boto3.py` (gated on those extras). +""" + +from __future__ import annotations + +import json +from typing import Iterator + +import pytest + +from arborist.cold_object import ( + PACK_PREFIX, + MemoryBackend, + build_pack, + open_pack, + pack_key, + parse_manifest, +) +from arborist.compress import unpack_chunk +from arborist.document import Document +from arborist.evict import ( + pull_pack, + push_pack, +) +from arborist.ingest import ingest_source +from arborist.merkle import hash_leaf +from arborist.source import Source +from arborist.store import connect + + +class FakeSource(Source): + source_type = "html" + + def __init__(self, docs: list[Document]): + self.docs = docs + + def iter_documents(self) -> Iterator[Document]: + yield from self.docs + + +def _doc(uri: str, content: str) -> Document: + return Document(uri=uri, content=content, source_type="html", title=uri) + + +LONG = ( + "The eight forms of capital include living, social, and intellectual. " * 30 + + "\n\n" + + "Merkle providence proves answer derives from a specific source. " * 30 +) + + +# --------------------------------------------------------------------------- +# Pure-helper tests (no DB, no backend) — the content-addressing invariants +# --------------------------------------------------------------------------- + + +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. + assert pack_key(h) != pack_key(h, manifest=True) + + +def test_build_pack_is_content_addressed(): + # Same input → same pack_hash, regardless of ordering. + body_a = b"hello world " * 100 + body_b = b"goodbye world " * 100 + leaf_a = hash_leaf(body_a).hex() + leaf_b = hash_leaf(body_b).hex() + + pack1 = build_pack([(leaf_a, body_a), (leaf_b, body_b)]) + pack2 = build_pack([(leaf_b, body_b), (leaf_a, body_a)]) # reversed + assert pack1.pack_hash == pack2.pack_hash + assert {e.leaf_hash for e in pack1.entries} == {leaf_a, leaf_b} + + +def test_build_pack_rejects_empty(): + with pytest.raises(ValueError): + build_pack([]) + + +def test_open_pack_verifies_each_chunk_hash(): + body_a = b"chunk one body padded out " * 20 + body_b = b"chunk two body padded out " * 20 + leaf_a = hash_leaf(body_a).hex() + leaf_b = hash_leaf(body_b).hex() + pack = build_pack([(leaf_a, body_a), (leaf_b, body_b)]) + + recovered = dict(open_pack(pack.body_bytes)) + assert recovered[leaf_a] == body_a + assert recovered[leaf_b] == body_b + + +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) + + +# --------------------------------------------------------------------------- +# Backend round-trip — MemoryBackend, no network +# --------------------------------------------------------------------------- + + +def test_memory_backend_put_get_head_list(): + """The backend speaks raw bytes; pack-level semantics live above it.""" + b = MemoryBackend() + key = "packs/abc123.tar.zst" + payload = b"fake pack body " * 10 + assert not b.head(key) + b.put(key, payload, content_type="application/zstd") + assert b.head(key) + assert b.get(key) == payload + assert list(b.list_keys("packs/")) == [key] + + +def test_memory_backend_identity_carries_no_credentials(): + b = MemoryBackend(endpoint_url="memory://nyc3", bucket="arborist-test") + body = b.identity.to_audit_body() + # Operation Voyeur: no credential keys should ever appear. + text = json.dumps(body) + for k in ("access", "secret", "key", "token", "password"): + assert k not in text.lower() or k == "key" and "access_key" not in text.lower() + + +# --------------------------------------------------------------------------- +# Pack push/pull — the only path now (no individual blobs) +# --------------------------------------------------------------------------- + + +def test_push_pack_then_pull_pack_restores_chunks(tmp_path): + db = tmp_path / "pack.db" + conn = connect(db) + backend = MemoryBackend() + try: + ingest_source(conn, FakeSource([_doc("html://p", LONG)])) + root = conn.execute( + "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)) + + # 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() + + pull_result = pull_pack(conn, backend, pack_hash) + assert pull_result["status"] == "pulled" + assert pull_result["chunks_restored"] == first["chunk_count"] + + cold = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE tier='cold'" + ).fetchone()[0] + assert cold == 0 + finally: + conn.close() + + +def test_push_pack_idempotent(tmp_path): + """Same chunk set → same pack_hash(es) → bucket overwrite is a no-op.""" + db = tmp_path / "pack-id.db" + conn = connect(db) + backend = MemoryBackend() + try: + 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"] + ] + finally: + conn.close() + + +def test_push_pack_nothing_to_pack(tmp_path): + db = tmp_path / "empty.db" + conn = connect(db) + backend = MemoryBackend() + try: + result = push_pack(conn, backend) + assert result["status"] == "nothing_to_pack" + assert result["packs"] == [] + finally: + conn.close() + + +def test_push_pack_splits_to_fit_dvdr(tmp_path): + """A small per-pack COMPRESSED-bytes cap forces multi-pack splitting; + every pack's compressed body stays ≤ cap + tar-trailer slack. + + The cap is on compressed bytes (`stream_packs` peeks the compressed + buffer after each chunk via FLUSH_BLOCK). Each pack's compressed_bytes + should be close to the cap — fill the disc, don't leave 50% empty. + """ + db = tmp_path / "split.db" + conn = connect(db) + backend = MemoryBackend() + try: + # Distinct content per doc so zstd can't dedupe across them — the + # whole point of testing the compressed cap is that compressed + # output scales with corpus, not with one-pattern repetition. + import random + rng = random.Random(42) + words = ["alpha", "bravo", "charlie", "delta", "echo", "foxtrot", + "golf", "hotel", "india", "juliet", "kilo", "lima"] + docs = [] + for i in range(8): + phrases = [" ".join(rng.choices(words, k=10)) for _ in range(80)] + docs.append(_doc(f"html://d{i}", " ".join(phrases) + ". ")) + ingest_source(conn, FakeSource(docs)) + + # Small compressed cap (4 KB) — should produce multiple packs. + CAP = 4_096 + result = push_pack(conn, backend, max_pack_bytes=CAP) + assert result["status"] == "pushed" + assert result["pack_count"] > 1, ( + f"expected multiple packs at {CAP}-byte compressed cap, " + f"got {result['pack_count']}" + ) + + # 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. + OVERSHOOT_SLACK = 4_096 + for p in result["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}" + ) + assert p["chunk_count"] >= 1 + + # Round-trip: NULL local content, pull every pack back. + 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"] + 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.""" + db = tmp_path / "burn.db" + conn = connect(db) + backend = MemoryBackend() + burn_dir = tmp_path / "discs" + try: + ingest_source(conn, FakeSource([_doc("html://burn", LONG)])) + result = push_pack(conn, backend, local_dir=str(burn_dir)) + 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"] + finally: + conn.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" + conn = connect(db) + backend = MemoryBackend() + burn_dir = tmp_path / "iso" + try: + ingest_source(conn, FakeSource([_doc("html://x", LONG)])) + result = push_pack( + conn, backend, + local_dir=str(burn_dir), + push_to_bucket=False, + ) + assert result["status"] == "written" + # No keys in bucket. + assert list(backend.list_keys()) == [] + # Files on disk. + assert any(burn_dir.glob("arborist-pack-*.tar.zst")) + finally: + conn.close() + + +def test_default_max_pack_bytes_is_dvdr_safe_fit(): + """Default cap is 4.4 GB — DVD-R safe-fit, with ~6.5% buffer below the + 4.7 GB media spec to absorb ISO9660 overhead, media manufacturing + variance, and drive-edge refusal. Don't drift this without a deliberate + decision; downstream burning workflows assume it.""" + from arborist.evict import DEFAULT_MAX_PACK_BYTES + assert DEFAULT_MAX_PACK_BYTES == 4_400_000_000 + # Must sit well below the 4.7 GB marketing capacity AND below the + # 4,706,074,624 ECMA-267 physical capacity, with room for overhead. + assert DEFAULT_MAX_PACK_BYTES < 4_700_000_000 + assert (4_700_000_000 - DEFAULT_MAX_PACK_BYTES) >= 250_000_000 # >=250 MB headroom diff --git a/tests/test_cold_object_boto3.py b/tests/test_cold_object_boto3.py new file mode 100644 index 0000000..7743650 --- /dev/null +++ b/tests/test_cold_object_boto3.py @@ -0,0 +1,69 @@ +"""On-the-wire boto3 backend test (#000061) — moto-mocked S3. + +Verifies that S3CompatibleBackend correctly issues PUT / GET / HEAD / +LIST against an S3 API for pack-shaped payloads. The rest of the cold- +pack surface lives in tests/test_cold_object.py and runs against the +in-process MemoryBackend. + +Gated on boto3 + moto. The default suite skips this file. +""" + +from __future__ import annotations + +import pytest + +boto3 = pytest.importorskip("boto3") +moto = pytest.importorskip("moto") + +from arborist.cold_object import PACK_PREFIX, S3CompatibleBackend, pack_key +from arborist.merkle import hash_leaf + + +@pytest.fixture +def moto_s3(): + """Spin up an in-memory S3 endpoint via moto, yield a fresh bucket.""" + from moto import mock_aws + + with mock_aws(): + client = boto3.client("s3", region_name="us-east-1") + client.create_bucket(Bucket="arborist-test") + yield { + "bucket": "arborist-test", + "region": "us-east-1", + } + + +def test_s3_backend_put_get_head_list(moto_s3): + # moto mocks the AWS endpoint when endpoint_url is None or AWS-shaped. + # Pointing at the default AWS URL keeps boto3 happy inside the mock. + backend = S3CompatibleBackend( + endpoint_url="https://s3.amazonaws.com", + bucket=moto_s3["bucket"], + region=moto_s3["region"], + ) + + pack_body = b"fake pack body padded out " * 200 + pack_hash = hash_leaf(pack_body).hex() + key = pack_key(pack_hash) + + assert not backend.head(key) + backend.put_pack(pack_hash, pack_body) + assert backend.head(key) + assert backend.get_pack(pack_hash) == pack_body + + keys = list(backend.list_keys(PACK_PREFIX)) + assert keys == [key] + + +def test_s3_backend_identity_carries_endpoint_and_bucket_only(moto_s3): + backend = S3CompatibleBackend( + endpoint_url="https://nyc3.digitaloceanspaces.com", + bucket="arborist-do-test", + region="us-east-1", + ) + identity = backend.identity + body = identity.to_audit_body() + assert body["endpoint_url"] == "https://nyc3.digitaloceanspaces.com" + assert body["bucket"] == "arborist-do-test" + assert "access_key" not in body + assert "secret" not in body