cold pack: --jit-blobs mode for online JIT consumer flow
Replaces the batched chunk-pack phase with per-chunk content-addressed blob uploads to `blobs/<hash[:2]>/<hash[2:]>`. 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.
This commit is contained in:
parent
9c747ad862
commit
d43714a503
6 changed files with 273 additions and 7 deletions
|
|
@ -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/<hash[:2]>/<hash[2:]>` 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(
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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/<hash[:2]>/<hash[2:]>`. 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/<hash>` 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,
|
||||
}
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
|
||||
```
|
||||
<bucket>/packs/<pack_hash>.tar.zst # pack body
|
||||
<bucket>/packs/<pack_hash>.manifest.ndjson # pack contents sidecar
|
||||
<bucket>/blobs/<hash[:2]>/<hash[2:]> # per-chunk body — opt-in,
|
||||
# populated only by
|
||||
# `cold pack --jit-blobs`
|
||||
# (online-JIT consumer flow)
|
||||
<bucket>/clones/<snapshot_id>/<NNN>.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/<hash[:2]>/<hash[2:]>`
|
||||
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 <metadata_pack_hash> \
|
||||
--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/<hash> from the bucket and
|
||||
# caches it back into the local row; subsequent reads are free.
|
||||
```
|
||||
|
||||
## DO Spaces quickstart
|
||||
|
||||
```bash
|
||||
|
|
|
|||
|
|
@ -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/<hash[:2]>/<hash[2:]>`. 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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue