From d43714a50385159af4c4b61b94a4c0cf182d5038 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sat, 30 May 2026 07:19:01 -0400 Subject: [PATCH] cold pack: --jit-blobs mode for online JIT consumer flow MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replaces the batched chunk-pack phase with per-chunk content-addressed blob uploads to `blobs//`. The metadata pack still ships (small, fast to restore), but consumers no longer have to pull multi-GB chunk packs to get queryable: `cold unpack --mode just-enough` + `ARBORIST_JIT_CHUNKS=1` fetches single chunks on cache miss. Producer (`_stream_jit_blobs` in evict.py): - ThreadPoolExecutor with bounded queue (workers*4) keeps memory flat across millions of chunks - HEAD-checks object_size for idempotent re-upload - Mutually exclusive with chunk packs — manifest's `chunk_pack_hashes` is empty in JIT mode (consumer reads that as "JIT-only") Consumer (`hydrate_doc_jit` in cold_clone.py + `_maybe_jit_hydrate` in qa/query.py): - Detects both content shapes that need JIT: NULL (Tier B raw-clone) and zeroblob placeholders (just-enough pack restore, per #53). Discriminator is first-byte = NUL — zstd-framed bodies start with 0x28, plain UTF-8 prose never has leading NUL. - Same placeholder filter applied to chunk-read sites in qa/query.py so partial hydrate doesn't surface zero-bytes content into the LLM context. Test (`TestJitBlobsPackMode` in tests/test_cold_unpack_routed.py): - End-to-end push → just-enough hydrate → JIT-fetch → content matches original byte-for-byte through `unpack_chunk`. Docs (cold-object-store.md): - Hard-invariant #1 updated: bucket holds packs by default; `blobs/` and `clones/` are opt-in prefixes for the JIT and Tier-A flows. - New "Three consumer modes" section: full-pack vs JIT-blobs vs raw-clone comparison table + operator decision tree. --- arborist/cli.py | 17 ++++++ arborist/cold_clone.py | 14 ++++- arborist/evict.py | 85 +++++++++++++++++++++++++++- arborist/qa/query.py | 9 ++- docs/cold-object-store.md | 59 +++++++++++++++++++- tests/test_cold_unpack_routed.py | 96 ++++++++++++++++++++++++++++++++ 6 files changed, 273 insertions(+), 7 deletions(-) diff --git a/arborist/cli.py b/arborist/cli.py index 942a46f..2038688 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -3013,6 +3013,8 @@ def _cmd_cold_pack(args: argparse.Namespace) -> int: push_to_bucket=args.push_to_bucket, allow_license_class=args.allow_license_class, include_fts=args.include_fts, + jit_blobs=getattr(args, "jit_blobs", False), + jit_blobs_workers=getattr(args, "jit_blobs_workers", 16), ) finally: conn.close() @@ -6208,8 +6210,23 @@ def build_parser() -> argparse.ArgumentParser: help="ship the FTS5 shadow-table pack anyway (bigger bucket; the " "restored index is currently non-functional — not recommended).", ) + cold_pack.add_argument( + "--jit-blobs", dest="jit_blobs", action="store_true", + help="upload each chunk as a content-addressed blob at " + "`blobs//` and skip the batched chunk-pack " + "phase. Targets the online JIT consumer flow: consumer pulls only " + "the small metadata pack (`cold unpack --mode just-enough`), then " + "queries with `ARBORIST_JIT_CHUNKS=1` fetch single chunks on " + "demand. Mutually exclusive with the DVD/burn workflow.", + ) + cold_pack.add_argument( + "--jit-blobs-workers", dest="jit_blobs_workers", type=int, default=16, + help="concurrent blob-upload workers when --jit-blobs is set " + "(default 16; PUT throughput is the bottleneck).", + ) cold_pack.set_defaults( func=_cmd_cold_pack, push_to_bucket=True, include_fts=False, + jit_blobs=False, ) cold_unpack = cold_sub.add_parser( diff --git a/arborist/cold_clone.py b/arborist/cold_clone.py index 081b243..c47424f 100644 --- a/arborist/cold_clone.py +++ b/arborist/cold_clone.py @@ -406,9 +406,21 @@ def hydrate_doc_jit( shape it would for a full clone.""" from arborist.merkle import hash_leaf + # Two shapes count as "needs JIT": + # - content IS NULL (Tier B raw-clone produces this) + # - content starts with NUL byte (the pack-meta `just-enough` + # restore writes zeroblob(_content_size) placeholders per #53, + # so we'd otherwise skip them). Real chunk content is either + # zstd-framed (magic \x28) or plain UTF-8 prose (no leading + # NUL in any natural-language text), so first-byte-zero is a + # safe placeholder discriminator. rows = conn.execute( "SELECT chunk_id, leaf_hash FROM chunks " - "WHERE document_root = ? AND content IS NULL", + "WHERE document_root = ? AND (" + " content IS NULL " + " OR length(content) = 0 " + " OR substr(content, 1, 1) = X'00'" + ")", (document_root,), ).fetchall() if not rows: diff --git a/arborist/evict.py b/arborist/evict.py index 4073e68..5b1a108 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -271,6 +271,68 @@ def rehydrate( DEFAULT_MAX_PACK_BYTES = 4_400_000_000 # 4.4 GB — DVD-R safe-fit (~6.5 % buffer) +def _stream_jit_blobs( + chunk_source: Iterable[tuple[str, bytes]], + backend: "ObjectStoreBackend", + *, + workers: int = 16, + push_to_bucket: bool = True, + queue_depth: int | None = None, +) -> dict: + """Upload each (leaf_hash, body) as a content-addressed bucket blob + at `blobs//`. Idempotent: HEAD-checks object_size + against the local body length before PUT. Bounded queue keeps memory + flat regardless of chunk count. + + Companion to chunk packs: single-chunk-addressable so an online JIT + consumer can fetch one chunk per cache miss instead of pulling a + multi-GB pack. Default mode for `cold pack --jit-blobs`. + """ + from concurrent.futures import ThreadPoolExecutor, FIRST_COMPLETED, wait + + stats = {"put": 0, "skipped_present": 0, "bytes": 0, "skipped_local_only": 0} + if not push_to_bucket: + # Local-only mode: count what would have been uploaded; no + # network calls. Lets `cold pack --no-push --jit-blobs` plan a + # snapshot without bucket credentials. + for leaf_hash, body in chunk_source: + stats["skipped_local_only"] += 1 + stats["bytes"] += len(body) + return stats + + def _upload(item: tuple[str, bytes]) -> tuple[str, int]: + leaf_hash, body = item + key = f"blobs/{leaf_hash[:2]}/{leaf_hash[2:]}" + existing = backend.object_size(key) + if existing == len(body): + return ("skipped_present", len(body)) + backend.put(key, body) + return ("put", len(body)) + + depth = queue_depth or max(workers * 4, 64) + it = iter(chunk_source) + with ThreadPoolExecutor(max_workers=workers) as ex: + futs = set() + # Prime up to `depth` in-flight; refill as each completes. + for _ in range(depth): + try: + futs.add(ex.submit(_upload, next(it))) + except StopIteration: + break + while futs: + done, futs = wait(futs, return_when=FIRST_COMPLETED) + for f in done: + tag, size = f.result() + stats[tag] += 1 + stats["bytes"] += size + for _ in range(len(done)): + try: + futs.add(ex.submit(_upload, next(it))) + except StopIteration: + pass + return stats + + def push_pack( conn: sqlite3.Connection, backend: "ObjectStoreBackend", @@ -283,6 +345,8 @@ def push_pack( push_to_bucket: bool = True, allow_license_class: str = "public_redistributable", include_fts: bool = True, + jit_blobs: bool = False, + jit_blobs_workers: int = 16, ) -> dict: """Bundle local chunks into one or more tar.zst packs. @@ -462,10 +526,25 @@ def push_pack( # backlink to the metadata pack — the metadata pack is the entry # point). Collect each chunk pack's hash for the metadata manifest. chunk_pack_hashes_in_order: list[str] = [] - for pack in stream_packs( + jit_blob_stats = {"put": 0, "skipped_present": 0, "bytes": 0} + if jit_blobs: + # Replace batched chunk packs with per-chunk content-addressed + # blob uploads. Online JIT consumers download the metadata pack + # (small) then fetch individual chunks from `blobs/` on + # demand. Chunk packs are wasteful for that workflow (4.4 GB per + # miss); blob mode is single-chunk-addressable. + jit_blob_stats = _stream_jit_blobs( + _chunk_source(), + backend, + workers=jit_blobs_workers, + push_to_bucket=push_to_bucket, + ) + # Skip Phase B entirely — no chunk packs in JIT mode. + chunk_pack_hashes_in_order = [] + for pack in (() if jit_blobs else 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] try: @@ -732,6 +811,8 @@ def push_pack( "skipped_hash_mismatch": skipped_hash_mismatch, "max_pack_bytes": max_pack_bytes, "local_dir": str(out_dir) if out_dir else None, + "jit_blobs": jit_blobs, + "jit_blob_stats": jit_blob_stats, } diff --git a/arborist/qa/query.py b/arborist/qa/query.py index d8595ff..97418fd 100644 --- a/arborist/qa/query.py +++ b/arborist/qa/query.py @@ -1200,7 +1200,10 @@ def _body_density_passes( if not qtokens: return False rows = conn.execute( - "SELECT content FROM chunks WHERE document_root = ? AND content IS NOT NULL", + "SELECT content FROM chunks " + "WHERE document_root = ? AND content IS NOT NULL " + " AND length(content) > 0 " + " AND substr(content, 1, 1) != X'00'", (document_root,), ).fetchall() if not rows: @@ -1886,6 +1889,8 @@ def _load_doc_text(shard_path: str, document_root: str) -> str | None: rows = conn.execute( "SELECT content FROM chunks " "WHERE document_root = ? AND content IS NOT NULL " + " AND length(content) > 0 " + " AND substr(content, 1, 1) != X'00' " "ORDER BY idx ASC", (document_root,), ).fetchall() @@ -1915,6 +1920,8 @@ def _load_doc_chunks( rows = conn.execute( "SELECT idx, leaf_hash, content FROM chunks " "WHERE document_root = ? AND content IS NOT NULL " + " AND length(content) > 0 " + " AND substr(content, 1, 1) != X'00' " "ORDER BY idx ASC", (document_root,), ).fetchall() diff --git a/docs/cold-object-store.md b/docs/cold-object-store.md index ac1b69a..1f1c119 100644 --- a/docs/cold-object-store.md +++ b/docs/cold-object-store.md @@ -24,15 +24,24 @@ backup consumers download packs whole. ## Hard invariants -1. **Bucket holds packs only.** Layout: +1. **Bucket holds packs by default; `blobs/` is opt-in for JIT.** Layout: ``` /packs/.tar.zst # pack body /packs/.manifest.ndjson # pack contents sidecar + /blobs// # per-chunk body — opt-in, + # populated only by + # `cold pack --jit-blobs` + # (online-JIT consumer flow) + /clones//.db # raw shard clones — opt-in, + # populated only by + # `cold stream-snapshot` + # (Tier A raw-clone flow) ``` - No `blobs/` prefix, no per-chunk objects. (One pack ↔ one disc ↔ one - bucket object.) + The classic pack flow (DVD-burn channel) keeps one pack ↔ one disc ↔ + one bucket object. The two opt-in prefixes (`blobs/`, `clones/`) light + up additional consumer flows; see "Three consumer modes" below. Pack contents (v2 format, self-sufficient for new-peer hydration): ``` @@ -144,6 +153,50 @@ 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`. +## Three consumer modes + +A fresh peer has three ways to hydrate from the bucket; producer mode +decides which the bucket supports. + +| mode | producer | consumer | local disk | bucket bytes pulled | first-query latency | +|------|---------------------|-------------------------------------------|-------------------|---------------------|---------------------| +| **full pack** | `cold pack` (default) | `cold unpack --mode full` | full corpus (~35 GB) | metadata + every chunk pack | seconds (all data local) | +| **JIT-blobs** | `cold pack --jit-blobs` | `cold unpack --mode just-enough` + queries with `ARBORIST_JIT_CHUNKS=1` | metadata-only (~27 GB at FTS+schema floor) | metadata + N×blob per query | one round-trip per chunk on first hit, cached after | +| **raw clone (Tier A)** | `cold stream-snapshot` | `cold clone` | full corpus (~35 GB) | raw `.db` files via SQLite Backup API | seconds (all data local) | + +**full-pack mode** is the canonical path: hardened, fully tested, FTS +restorable or rebuildable. Use for genesis recovery of a self-hosting +peer, DVD-archival workflows, and any consumer that wants the full +corpus offline-queryable. + +**JIT-blobs mode** trades steady-state download for fastest time-to- +queryable. The metadata pack is small; chunk bodies stream in as +queries access them. Good for ephemeral nodes, demo VMs, edge servers +that only serve a subset of the corpus. The `blobs//` +prefix is content-addressed: a chunk that exists in two snapshots costs +one bucket object, and producer reuploads are free idempotent no-ops. +Failure mode is graceful — `hydrate_doc_jit` skips missing blobs and +hash-mismatches rather than crashing the query. + +**raw-clone mode** is the simplest. `cold stream-snapshot` runs the +SQLite Backup API over each shard to a tempfile and PUTs the bytes; the +consumer DOWNLOAD the `.db` files and is queryable immediately. No +compression, no restore phase. Larger bucket footprint than the pack +flow but skips the slow metadata-restore step entirely. + +```bash +# Producer (JIT-blobs mode): +arborist cold pack --jit-blobs --shards-dir ~/.arborist/shards + +# Consumer: +arborist cold unpack --mode just-enough \ + --hydrate-shards-dir ~/.arborist/shards --hydrate-M 4 +export ARBORIST_JIT_CHUNKS=1 +arborist query --shards-dir ~/.arborist/shards "what is X?" +# → first hit on each chunk fetches blobs/ from the bucket and +# caches it back into the local row; subsequent reads are free. +``` + ## DO Spaces quickstart ```bash diff --git a/tests/test_cold_unpack_routed.py b/tests/test_cold_unpack_routed.py index a7e9834..edd74b8 100644 --- a/tests/test_cold_unpack_routed.py +++ b/tests/test_cold_unpack_routed.py @@ -325,6 +325,102 @@ class TestPreSizedChunks: assert sz > 0, f"chunk {cid} on shard {i} has zero-size placeholder" +class TestJitBlobsPackMode: + """`cold pack --jit-blobs` skips chunk packs and uploads per-chunk + content-addressed blobs to `blobs//`. A consumer + pulls the metadata pack via `cold unpack --mode just-enough` and + then JIT-fetches individual chunks from the bucket on demand. + """ + + def test_jit_blobs_skips_chunk_packs_and_uploads_blobs( + self, producer_shard: Path, tmp_path: Path + ): + from arborist.cold_clone import hydrate_doc_jit + from arborist.compress import unpack_chunk + from arborist.document import shard_for_document + + backend = MemoryBackend() + with sqlite3.connect(str(producer_shard)) as src_conn: + src_conn.row_factory = sqlite3.Row + push_result = push_pack( + src_conn, backend, + document_root=None, + max_chunks=None, + max_pack_bytes=10 * 1024 ** 2, + allow_license_class="unknown", + jit_blobs=True, + ) + + kinds = [p.get("kind") for p in push_result.get("packs", [])] + assert "chunks" not in kinds, "jit_blobs should skip chunk packs" + assert kinds.count("metadata") == 1 + assert push_result["jit_blobs"] is True + stats = push_result["jit_blob_stats"] + assert stats["put"] > 0 + + # Every chunk row's leaf_hash must have a blob in the bucket. + with sqlite3.connect(f"file:{producer_shard}?mode=ro", uri=True) as src: + src.row_factory = sqlite3.Row + chunks = src.execute( + "SELECT chunk_id, document_root, leaf_hash, content " + "FROM chunks WHERE content IS NOT NULL" + ).fetchall() + for r in chunks: + key = f"blobs/{r['leaf_hash'][:2]}/{r['leaf_hash'][2:]}" + assert backend.head(key), f"missing blob for chunk {r['chunk_id']}" + + # Hydrate metadata-only into M=4 fresh shards. + meta_hash = next( + p["pack_hash"] for p in push_result["packs"] + if p.get("kind") == "metadata" + ) + target_dir = tmp_path / "tgt-jit" + target_dir.mkdir() + M = 4 + targets = _make_target_shards(target_dir, M) + try: + hydrate_from_metadata_pack_routed( + targets, backend, meta_hash, M=M, mode="just-enough", + ) + finally: + for t in targets: + t.close() + + # Pick one doc; JIT-hydrate it on its owning target. + pick = chunks[0] + doc_root = pick["document_root"] + tidx = shard_for_document(doc_root, M) + tdb = target_dir / f"{tidx:03d}.db" + + tconn = sqlite3.connect(str(tdb)) + tconn.row_factory = sqlite3.Row + try: + n_fetched = hydrate_doc_jit(tconn, doc_root, backend) + assert n_fetched > 0, "JIT should fetch placeholder chunks" + + tgt = { + r["leaf_hash"]: r["content"] + for r in tconn.execute( + "SELECT leaf_hash, content FROM chunks WHERE document_root = ?", + (doc_root,), + ) + } + finally: + tconn.close() + + prod = { + r["leaf_hash"]: r["content"] + for r in chunks + if r["document_root"] == doc_root + } + # Both sides must decode to identical text. The producer side + # may be compressed (BLOB) or plain (TEXT); the target side + # after JIT is always plain UTF-8 from the bucket blob. + for lh, prod_body in prod.items(): + assert lh in tgt + assert unpack_chunk(tgt[lh]) == unpack_chunk(prod_body) + + class TestRoutedHydrateValidation: def test_M_mismatch_rejected(self, tmp_path: Path): target_dir = tmp_path / "tgt"