diff --git a/Makefile b/Makefile index b943df0..49cedef 100644 --- a/Makefile +++ b/Makefile @@ -965,7 +965,8 @@ cold-hydrate: bootstrap ## genesis a fresh peer from cloud, M-aware: pull every just-enough) MODE_FLAG="--just-enough" ;; \ *) echo "HYDRATE_MODE=$$MODE invalid (full | just-enough)"; exit 2 ;; \ esac; \ - $(COLD_PACK_PICK_JOBS); \ + JOBS="$(COLD_PACK_JOBS)"; [ -z "$$JOBS" ] && JOBS=1; \ + echo ">> COLD_PACK_JOBS=$$JOBS (serial default: M-aware hydrate routes every pack into all M shared target shards, so parallel workers contend on the same files — #54. Override COLD_PACK_JOBS=N at your own risk.)"; \ echo ">> discovering metadata packs in bucket"; \ hashes=$$($(ARBORIST) cold list --no-manifest \ | $(VENV)/bin/python -c \ @@ -978,15 +979,12 @@ cold-hydrate: bootstrap ## genesis a fresh peer from cloud, M-aware: pull every --hydrate-shards-dir $(HYDRATE_DIR) \ --hydrate-M $(HYDRATE_M) \ $$MODE_FLAG - @if [ "$${HYDRATE_MODE:-full}" = "full" ] && [ "$${HYDRATE_REBUILD_FTS:-auto}" != "0" ]; then \ - has_fts=$$(python3 -c "import sqlite3; c=sqlite3.connect('$(HYDRATE_DIR)/000.db'); n=c.execute('SELECT COUNT(*) FROM chunks_fts_data').fetchone()[0]; print(n)" 2>/dev/null); \ - if [ "$${HYDRATE_REBUILD_FTS:-auto}" = "1" ] || [ -z "$$has_fts" ] || [ "$$has_fts" = "0" ]; then \ - echo ">> chunks_fts_data empty on shard 000 → serial FTS5 rebuild"; \ - $(ARBORIST) cold rebuild-fts --shards-dir $(HYDRATE_DIR); \ - else \ - echo ">> chunks_fts_data already populated (fts pack restored) — skipping rebuild"; \ - fi; \ + @if [ "$${HYDRATE_MODE:-full}" = "full" ] && [ "$${HYDRATE_REBUILD_FTS:-1}" != "0" ]; then \ + echo ">> rebuilding FTS from content (parallel; cold-pack FTS restores a dead index, so always rebuild)"; \ + $(ARBORIST) cold rebuild-fts --shards-dir $(HYDRATE_DIR); \ fi + @echo ">> verifying recovery (content materialized + searchable; fails loudly if not)" + @$(ARBORIST) cold verify --shards-dir $(HYDRATE_DIR) @echo ">> hydration complete:" @ls -lh $(HYDRATE_DIR)/*.db 2>/dev/null diff --git a/arborist/cli.py b/arborist/cli.py index 3e4cd27..73e94d5 100644 --- a/arborist/cli.py +++ b/arborist/cli.py @@ -3122,47 +3122,130 @@ def _cmd_cold_unpack(args: argparse.Namespace) -> int: return 0 -def _cmd_cold_rebuild_fts(args: argparse.Namespace) -> int: - """Sequentially rebuild chunks_fts + documents_fts on every shard. - - Pairs with `arborist cold unpack --hydrate-shards-dir` which - deliberately skips FTS5 indexing in its parallel inner loop — - concurrent FTS5 writes across 4 shards serialize on the ext4 - journal and become the throughput floor. This serial post-pass - gives each shard the full filesystem journal in its turn, then - moves on. Total wall time is bounded by the slowest single-shard - FTS rebuild, not by 4× contention. - """ +def _cold_rebuild_fts_one(path_str: str) -> tuple[str, float, int]: + """ProcessPool worker: clear + rebuild FTS on one shard. Each shard is + its own DB file, so workers never contend. ``_rebuild_fts_on_target`` + clears-first, overwriting the dead index a cold-pack FTS restore + leaves behind, then rebuilds from chunk content.""" + import time as _t + from pathlib import Path as _P + from arborist.store import connect as _connect from arborist.migrate import _rebuild_fts_on_target + t = _t.time() + conn = _connect(path_str) + try: + conn.execute("PRAGMA synchronous = OFF") + conn.execute("PRAGMA journal_mode = MEMORY") + conn.execute("PRAGMA temp_store = MEMORY") + conn.execute("PRAGMA cache_size = -524288") + _rebuild_fts_on_target(conn) + cf = conn.execute("SELECT count(*) FROM chunks_fts").fetchone()[0] + finally: + conn.close() + return (_P(path_str).name, _t.time() - t, int(cf)) + + +def _cmd_cold_rebuild_fts(args: argparse.Namespace) -> int: + """Rebuild chunks_fts + documents_fts on every shard, in PARALLEL + (one process per shard — separate DB files, so no lock contention). + + FTS is derived from chunk content, so this is the canonical repair + for a cold-recovered peer: a cold-pack FTS restore leaves a + non-functional index (rows present, MATCH returns 0), and + ``_rebuild_fts_on_target`` clears-first + rebuilds from content, so + this overwrites it cleanly. Run after `cold unpack --hydrate-shards-dir`. + """ + import concurrent.futures as _cf shards_dir = Path(args.shards_dir).expanduser() db_files = sorted(shards_dir.glob("00[0-9].db")) if not db_files: print(f"no shards under {shards_dir}", file=sys.stderr) return 1 + req = int(getattr(args, "workers", 0) or 0) + workers = max(1, min(req if req > 0 else len(db_files), len(db_files))) started = time.time() - for path in db_files: - per_started = time.time() - conn = connect(path) - try: - conn.execute("PRAGMA synchronous = OFF") - conn.execute("PRAGMA journal_mode = MEMORY") - conn.execute("PRAGMA temp_store = MEMORY") - conn.execute("PRAGMA cache_size = -524288") - _rebuild_fts_on_target(conn) - finally: - conn.close() - print( - f" {path.name}: FTS rebuilt in {time.time() - per_started:.1f}s", - file=sys.stderr, - ) + with _cf.ProcessPoolExecutor(max_workers=workers) as ex: + for name, secs, cf in ex.map(_cold_rebuild_fts_one, [str(p) for p in db_files]): + print(f" {name}: FTS rebuilt in {secs:.1f}s (chunks_fts={cf})", file=sys.stderr) print( - f"all {len(db_files)} shards FTS-rebuilt in {time.time() - started:.1f}s", + f"all {len(db_files)} shards FTS-rebuilt in {time.time() - started:.1f}s " + f"({workers}-way parallel)", file=sys.stderr, ) return 0 +def _cmd_cold_verify(args: argparse.Namespace) -> int: + """Self-check a hydrated shard set: chunk content materialized (not + zero-filled placeholders) AND FTS searchable. Exits non-zero with a + summary if any shard fails, so a bad recovery fails LOUDLY instead of + silently serving empty results (the 2026-05-29 genesis trap: 3/4 + shards had zero-filled bodies + dead FTS and nobody noticed).""" + import re as _re + from arborist.compress import unpack_chunk + + shards_dir = Path(args.shards_dir).expanduser() + db_files = sorted(shards_dir.glob("00[0-9].db")) + if not db_files: + print(f"no shards under {shards_dir}", file=sys.stderr) + return 1 + n_sample = max(1, int(getattr(args, "sample", 20) or 20)) + failures: list[str] = [] + checked = 0 + for path in db_files: + conn = connect(path) + try: + nchunks = conn.execute( + "SELECT count(*) FROM chunks WHERE content IS NOT NULL" + ).fetchone()[0] + if not nchunks: + continue # legitimately empty target shard — nothing to verify + checked += 1 + rows = conn.execute( + "SELECT content FROM chunks WHERE content IS NOT NULL LIMIT ?", + (n_sample,), + ).fetchall() + real_words: set[str] = set() + zero = 0 + for (packed,) in rows: + try: + txt = unpack_chunk(packed) + except Exception: + txt = None + if not (txt and str(txt).strip("\x00").strip()): + zero += 1 + else: + for w in _re.findall(r"[a-z]{4,}", str(txt).lower())[:5]: + real_words.add(w) + if zero == len(rows): + failures.append( + f"{path.name}: content zero-filled ({zero}/{len(rows)} sampled " + f"of {nchunks}) — chunk bodies never materialized" + ) + continue + probe = next(iter(real_words), None) + if probe is not None: + hits = conn.execute( + "SELECT count(*) FROM chunks_fts WHERE chunks_fts MATCH ?", + (probe,), + ).fetchone()[0] + if not hits: + failures.append( + f"{path.name}: FTS dead (MATCH {probe!r} = 0 with {nchunks} " + f"chunks) — run `cold rebuild-fts`" + ) + finally: + conn.close() + if failures: + print("COLD VERIFY FAILED:", file=sys.stderr) + for f in failures: + print(" - " + f, file=sys.stderr) + return 1 + print(f"cold verify OK: {checked} non-empty shard(s) — content materialized + searchable") + return 0 + + def _cmd_cold_stats(args: argparse.Namespace) -> int: from arborist.cold_object import PACK_PREFIX @@ -6075,12 +6158,18 @@ def build_parser() -> argparse.ArgumentParser: ) cold_pack.add_argument( "--no-fts", dest="include_fts", action="store_false", - help="skip the FTS5 shadow-table pack (#000067 phase 2). " - "Smaller bucket footprint, but consumer must run " - "`arborist cold rebuild-fts` after unpack to enable body search.", + help="(default) skip the FTS5 shadow-table pack. The cold-pack FTS " + "restore produces a dead index (rows present, MATCH=0; 2026-05-29) " + "and FTS is derivable from content, so consumers rebuild it with " + "`cold rebuild-fts` after unpack.", + ) + cold_pack.add_argument( + "--with-fts", dest="include_fts", action="store_true", + help="ship the FTS5 shadow-table pack anyway (bigger bucket; the " + "restored index is currently non-functional — not recommended).", ) cold_pack.set_defaults( - func=_cmd_cold_pack, push_to_bucket=True, include_fts=True, + func=_cmd_cold_pack, push_to_bucket=True, include_fts=False, ) cold_unpack = cold_sub.add_parser( @@ -6133,18 +6222,39 @@ def build_parser() -> argparse.ArgumentParser: cold_rebuild_fts = cold_sub.add_parser( "rebuild-fts", help=( - "serially rebuild chunks_fts + documents_fts on every shard " - "in SHARDS_DIR. Pairs with `cold unpack --hydrate-shards-dir` " - "which defers FTS to avoid ext4-journal contention in the " - "parallel inner loop." + "rebuild chunks_fts + documents_fts from content on every shard " + "in SHARDS_DIR, in parallel (one process per shard). Clears-first, " + "so it also repairs the dead index a cold-pack FTS restore leaves." ), ) cold_rebuild_fts.add_argument( "--shards-dir", dest="shards_dir", required=True, - help="directory of shards to FTS-rebuild (one shard at a time)", + help="directory of shards to FTS-rebuild", + ) + cold_rebuild_fts.add_argument( + "--workers", dest="workers", type=int, default=0, + help="parallel workers (default 0 = one per shard)", ) cold_rebuild_fts.set_defaults(func=_cmd_cold_rebuild_fts) + cold_verify = cold_sub.add_parser( + "verify", + help=( + "self-check a hydrated shard set: chunk content materialized " + "(not zero-filled) AND FTS searchable. Non-zero exit if any " + "shard fails — so a bad recovery fails loudly." + ), + ) + cold_verify.add_argument( + "--shards-dir", dest="shards_dir", required=True, + help="directory of shards to verify", + ) + cold_verify.add_argument( + "--sample", dest="sample", type=int, default=20, + help="chunks sampled per shard for the content check (default 20)", + ) + cold_verify.set_defaults(func=_cmd_cold_verify) + cold_list = cold_sub.add_parser( "list", help="enumerate packs in the bucket with metadata (pack_hash, size, chunk_count) for new-peer hydration",