cold-recovery: fix pathological restore speed + dead FTS-pack rebuild
Genesis hydrate from cold packs was broken end-to-end (the #46 SPV-wallet test never passed): - Phase-1 metadata restore (90M-edge fan-out + zeroblob chunk placeholders) ran with SQLite's ~2 MB default cache — the 512 MB / MEMORY-journal tuning was applied AFTER Phase 1, so INSERT OR IGNORE into the indexed edges table thrashed: ~4 h/pack, never finishing. Move the bulk PRAGMAs before Phase 1 (evict.py). - The routed restores held one unbounded transaction; add bounded incremental commits (edges every 1M rows, routed tables every 200k) so the txn and in-RAM MEMORY journal stay small (cold_pack_metadata.py). - FTS packs restore a non-functional index (shadow rows present, MATCH returns 0). _rebuild_fts_on_target now clears-first so a rebuild from content overwrites the dead pack-FTS idempotently (migrate.py). Validated on the 3090: full corpus recovered (3,468,226 docs / 6,235,588 chunks / 90,592,990 edges, exact match to source), search + STRICT Q&A working. Restore ~4 h/pack -> ~8-20 min/pack; FTS rebuilt from content in ~5 min (parallel).
This commit is contained in:
parent
e497a9501a
commit
6eada6dc89
3 changed files with 54 additions and 23 deletions
|
|
@ -570,7 +570,13 @@ def _restore_routed_table(
|
||||||
csize_idx: int | None = None # position of _content_size in JSONL row
|
csize_idx: int | None = None # position of _content_size in JSONL row
|
||||||
target_content_pos: int | None = None # position of `content` in cols
|
target_content_pos: int | None = None # position of `content` in cols
|
||||||
insert_sql: str | None = None
|
insert_sql: str | None = None
|
||||||
|
# Commit periodically so the open transaction — and, under
|
||||||
|
# journal_mode=MEMORY, the in-RAM rollback journal — stays bounded.
|
||||||
|
# Chunks pad `content` with a full-size zeroblob, so a whole pack's
|
||||||
|
# worth held uncommitted is GBs of RAM. Commit every COMMIT_EVERY rows.
|
||||||
|
COMMIT_EVERY = 200_000
|
||||||
n = 0
|
n = 0
|
||||||
|
since_commit = 0
|
||||||
for row in read_columnar_jsonl(in_path):
|
for row in read_columnar_jsonl(in_path):
|
||||||
if cols is None:
|
if cols is None:
|
||||||
jsonl_cols = list(row.keys())
|
jsonl_cols = list(row.keys())
|
||||||
|
|
@ -611,10 +617,15 @@ def _restore_routed_table(
|
||||||
else:
|
else:
|
||||||
buckets[idx].append(tuple(row[c] for c in cols))
|
buckets[idx].append(tuple(row[c] for c in cols))
|
||||||
n += 1
|
n += 1
|
||||||
|
since_commit += 1
|
||||||
for i, bucket in enumerate(buckets):
|
for i, bucket in enumerate(buckets):
|
||||||
if len(bucket) >= BATCH and insert_sql is not None:
|
if len(bucket) >= BATCH and insert_sql is not None:
|
||||||
targets[i].executemany(insert_sql, bucket)
|
targets[i].executemany(insert_sql, bucket)
|
||||||
bucket.clear()
|
bucket.clear()
|
||||||
|
if since_commit >= COMMIT_EVERY:
|
||||||
|
for t in targets:
|
||||||
|
t.commit()
|
||||||
|
since_commit = 0
|
||||||
if insert_sql is not None:
|
if insert_sql is not None:
|
||||||
for i, bucket in enumerate(buckets):
|
for i, bucket in enumerate(buckets):
|
||||||
if bucket:
|
if bucket:
|
||||||
|
|
@ -631,6 +642,7 @@ def _restore_edges_fan_out_routed(
|
||||||
edge to the shard its src_root hashes to. Mirrors
|
edge to the shard its src_root hashes to. Mirrors
|
||||||
``_restore_edges_fan_out`` but multi-shard."""
|
``_restore_edges_fan_out`` but multi-shard."""
|
||||||
BATCH = 5000
|
BATCH = 5000
|
||||||
|
COMMIT_EVERY = 1_000_000 # bound the open txn / in-RAM MEMORY journal
|
||||||
buckets: list[list[tuple]] = [[] for _ in range(M)]
|
buckets: list[list[tuple]] = [[] for _ in range(M)]
|
||||||
sql = (
|
sql = (
|
||||||
"INSERT OR IGNORE INTO edges "
|
"INSERT OR IGNORE INTO edges "
|
||||||
|
|
@ -638,6 +650,7 @@ def _restore_edges_fan_out_routed(
|
||||||
"VALUES (?, ?, ?, ?, ?)"
|
"VALUES (?, ?, ?, ?, ?)"
|
||||||
)
|
)
|
||||||
n = 0
|
n = 0
|
||||||
|
since_commit = 0
|
||||||
for fan_in in read_columnar_jsonl(in_path):
|
for fan_in in read_columnar_jsonl(in_path):
|
||||||
dst_uri = fan_in["dst_uri"]
|
dst_uri = fan_in["dst_uri"]
|
||||||
edge_type = fan_in["edge_type"]
|
edge_type = fan_in["edge_type"]
|
||||||
|
|
@ -649,9 +662,14 @@ def _restore_edges_fan_out_routed(
|
||||||
idx = _shard_idx_for_root(src_root, M)
|
idx = _shard_idx_for_root(src_root, M)
|
||||||
buckets[idx].append((src_root, dst_root, dst_uri, edge_type, anchor))
|
buckets[idx].append((src_root, dst_root, dst_uri, edge_type, anchor))
|
||||||
n += 1
|
n += 1
|
||||||
|
since_commit += 1
|
||||||
if len(buckets[idx]) >= BATCH:
|
if len(buckets[idx]) >= BATCH:
|
||||||
targets[idx].executemany(sql, buckets[idx])
|
targets[idx].executemany(sql, buckets[idx])
|
||||||
buckets[idx].clear()
|
buckets[idx].clear()
|
||||||
|
if since_commit >= COMMIT_EVERY:
|
||||||
|
for t in targets:
|
||||||
|
t.commit()
|
||||||
|
since_commit = 0
|
||||||
for i, bucket in enumerate(buckets):
|
for i, bucket in enumerate(buckets):
|
||||||
if bucket:
|
if bucket:
|
||||||
targets[i].executemany(sql, bucket)
|
targets[i].executemany(sql, bucket)
|
||||||
|
|
|
||||||
|
|
@ -1127,6 +1127,30 @@ def _pull_pack_inner_routed(
|
||||||
f"{tampered_tables}"
|
f"{tampered_tables}"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# Pre-tune every target writer for bulk genesis BEFORE the heavy
|
||||||
|
# Phase-1 restore. The 512 MB page cache is the dominant lever:
|
||||||
|
# Phase 1 fans 90M edges out with INSERT OR IGNORE into the
|
||||||
|
# *indexed* edges table; with SQLite's ~2 MB default cache the
|
||||||
|
# random index-page access thrashes and the restore degrades to
|
||||||
|
# HOURS (2026-05-29: 4 h for one pack, only 3 GB in). Setting the
|
||||||
|
# tuning here — not after Phase 1, where it used to live — gives
|
||||||
|
# the edge fan-out the big cache too. None of these apply outside
|
||||||
|
# a hydrate; a crash mid-hydrate is recoverable (re-pull), so
|
||||||
|
# durability of intermediate state has no value.
|
||||||
|
#
|
||||||
|
# busy_timeout first: in parallel-hydrate (xargs -P N) multiple
|
||||||
|
# processes change journal_mode on the same file at once and
|
||||||
|
# serialize on a brief exclusive lock; without busy_timeout they
|
||||||
|
# throw `database is locked` (Fix #54). Serial hydrate sidesteps
|
||||||
|
# that contention entirely and is the supported genesis path.
|
||||||
|
for t in targets:
|
||||||
|
t.execute("PRAGMA busy_timeout = 30000") # wait, don't throw
|
||||||
|
t.execute("PRAGMA synchronous = OFF")
|
||||||
|
t.execute("PRAGMA journal_mode = MEMORY") # no on-disk WAL
|
||||||
|
t.execute("PRAGMA temp_store = MEMORY")
|
||||||
|
t.execute("PRAGMA cache_size = -524288") # 512 MB page cache
|
||||||
|
t.execute("PRAGMA mmap_size = 536870912") # 512 MB read mmap
|
||||||
|
|
||||||
# Phase 1: metadata tables → routed across M targets.
|
# Phase 1: metadata tables → routed across M targets.
|
||||||
tables_restored: dict[str, int] = {}
|
tables_restored: dict[str, int] = {}
|
||||||
if table_files_seen:
|
if table_files_seen:
|
||||||
|
|
@ -1136,29 +1160,6 @@ def _pull_pack_inner_routed(
|
||||||
finally:
|
finally:
|
||||||
shutil.rmtree(tables_dir, ignore_errors=True)
|
shutil.rmtree(tables_dir, ignore_errors=True)
|
||||||
|
|
||||||
# Pre-tune every target writer for bulk genesis. None of these
|
|
||||||
# apply outside a hydrate context — they're set per-connection,
|
|
||||||
# not at the schema level, so they affect only this writer's
|
|
||||||
# behaviour. Crash mid-hydrate is recoverable (re-pull from
|
|
||||||
# bucket), so durability of intermediate state has no value.
|
|
||||||
#
|
|
||||||
# busy_timeout MUST come FIRST: in parallel-hydrate (xargs -P N),
|
|
||||||
# N workers in separate processes each open connections to all M
|
|
||||||
# target shards and apply PRAGMAs. Multiple processes attempting
|
|
||||||
# PRAGMA journal_mode change on the same file at the same instant
|
|
||||||
# serialize on a brief exclusive lock; without busy_timeout they
|
|
||||||
# throw `database is locked` and abort. With busy_timeout SQLite
|
|
||||||
# waits up to 30s and retries. 2026-05-27 v4-fixed bench: 3 of 4
|
|
||||||
# workers crashed at this exact PRAGMA; only shard 000 ended up
|
|
||||||
# fully hydrated. Fix #54.
|
|
||||||
for t in targets:
|
|
||||||
t.execute("PRAGMA busy_timeout = 30000") # 30 sec wait, not throw
|
|
||||||
t.execute("PRAGMA synchronous = OFF")
|
|
||||||
t.execute("PRAGMA journal_mode = MEMORY") # no on-disk WAL
|
|
||||||
t.execute("PRAGMA temp_store = MEMORY")
|
|
||||||
t.execute("PRAGMA cache_size = -524288") # 512 MB page cache
|
|
||||||
t.execute("PRAGMA mmap_size = 536870912") # 512 MB read mmap
|
|
||||||
|
|
||||||
# Phase 2: fill chunk content. Bench-max'd 2026-05-26 after the
|
# Phase 2: fill chunk content. Bench-max'd 2026-05-26 after the
|
||||||
# initial naive impl took ~2h wall on a 6M-chunk corpus (same speed
|
# initial naive impl took ~2h wall on a 6M-chunk corpus (same speed
|
||||||
# as a fresh ingest — the whole point of a pack restore is to be
|
# as a fresh ingest — the whole point of a pack restore is to be
|
||||||
|
|
|
||||||
|
|
@ -540,6 +540,18 @@ def _rebuild_fts_on_target(conn: sqlite3.Connection) -> None:
|
||||||
"""
|
"""
|
||||||
from arborist.compress import unpack_chunk
|
from arborist.compress import unpack_chunk
|
||||||
|
|
||||||
|
# Idempotent: clear any existing FTS rows first. A cold-pack FTS
|
||||||
|
# restore leaves a BROKEN index (shadow rows present, MATCH returns
|
||||||
|
# nothing — 2026-05-29); rebuilding over it without clearing would
|
||||||
|
# double-insert. 'delete-all' resets a contentless fts5 cleanly and
|
||||||
|
# is a no-op on a fresh/empty index (reshard targets).
|
||||||
|
for _ft in ("chunks_fts", "documents_fts"):
|
||||||
|
try:
|
||||||
|
with conn:
|
||||||
|
conn.execute(f"INSERT INTO {_ft}({_ft}) VALUES('delete-all')")
|
||||||
|
except sqlite3.OperationalError:
|
||||||
|
pass
|
||||||
|
|
||||||
batch_size = 5_000
|
batch_size = 5_000
|
||||||
fts_rows: list[tuple[int, str]] = []
|
fts_rows: list[tuple[int, str]] = []
|
||||||
cur = conn.execute(
|
cur = conn.execute(
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue