cold-recovery hardening: drop FTS packs, parallel rebuild, loud self-check
Make cold recovery correct-by-default and fail-loud, closing the class that silently produced a corrupt, unqueryable corpus (2026-05-29): - cold rebuild-fts now runs in PARALLEL (one process per shard; separate files, no contention) and clears-first, so it also repairs the dead index a cold-pack FTS restore leaves. Replaces the serial post-pass. - new `cold verify`: self-check a hydrated shard set — chunk content materialized (not zero-filled) AND FTS searchable (MATCH a word taken from sampled content). Non-zero exit if any shard fails. Validated: it passes the good recovery and fails the corrupt genesis-test, catching both the zero-filled-bodies and dead-FTS classes. - cold pack defaults to --no-fts (FTS is derived from content and the FTS pack restore is non-functional anyway); --with-fts to opt back in. - make cold-hydrate: serial by default (M-aware hydrate routes every pack into all M shared target shards, so parallel workers contend — #54); always rebuild FTS from content (drop the broken shard-000-only gate); run `cold verify` at the end so a bad recovery fails loudly.
This commit is contained in:
parent
7f7eeefeb9
commit
6d2a75d80b
2 changed files with 153 additions and 45 deletions
16
Makefile
16
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
|
||||
|
||||
|
|
|
|||
182
arborist/cli.py
182
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",
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue