diff --git a/arborist/cold_pack_metadata.py b/arborist/cold_pack_metadata.py index 237cfa8..e8017f6 100644 --- a/arborist/cold_pack_metadata.py +++ b/arborist/cold_pack_metadata.py @@ -570,7 +570,13 @@ def _restore_routed_table( csize_idx: int | None = None # position of _content_size in JSONL row target_content_pos: int | None = None # position of `content` in cols 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 + since_commit = 0 for row in read_columnar_jsonl(in_path): if cols is None: jsonl_cols = list(row.keys()) @@ -611,10 +617,15 @@ def _restore_routed_table( else: buckets[idx].append(tuple(row[c] for c in cols)) n += 1 + since_commit += 1 for i, bucket in enumerate(buckets): if len(bucket) >= BATCH and insert_sql is not None: targets[i].executemany(insert_sql, bucket) bucket.clear() + if since_commit >= COMMIT_EVERY: + for t in targets: + t.commit() + since_commit = 0 if insert_sql is not None: for i, bucket in enumerate(buckets): if bucket: @@ -631,6 +642,7 @@ def _restore_edges_fan_out_routed( edge to the shard its src_root hashes to. Mirrors ``_restore_edges_fan_out`` but multi-shard.""" BATCH = 5000 + COMMIT_EVERY = 1_000_000 # bound the open txn / in-RAM MEMORY journal buckets: list[list[tuple]] = [[] for _ in range(M)] sql = ( "INSERT OR IGNORE INTO edges " @@ -638,6 +650,7 @@ def _restore_edges_fan_out_routed( "VALUES (?, ?, ?, ?, ?)" ) n = 0 + since_commit = 0 for fan_in in read_columnar_jsonl(in_path): dst_uri = fan_in["dst_uri"] edge_type = fan_in["edge_type"] @@ -649,9 +662,14 @@ def _restore_edges_fan_out_routed( idx = _shard_idx_for_root(src_root, M) buckets[idx].append((src_root, dst_root, dst_uri, edge_type, anchor)) n += 1 + since_commit += 1 if len(buckets[idx]) >= BATCH: targets[idx].executemany(sql, buckets[idx]) buckets[idx].clear() + if since_commit >= COMMIT_EVERY: + for t in targets: + t.commit() + since_commit = 0 for i, bucket in enumerate(buckets): if bucket: targets[i].executemany(sql, bucket) diff --git a/arborist/evict.py b/arborist/evict.py index 91930c3..bbe50c1 100644 --- a/arborist/evict.py +++ b/arborist/evict.py @@ -1127,6 +1127,30 @@ def _pull_pack_inner_routed( 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. tables_restored: dict[str, int] = {} if table_files_seen: @@ -1136,29 +1160,6 @@ def _pull_pack_inner_routed( finally: 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 # 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 diff --git a/arborist/migrate.py b/arborist/migrate.py index 6c127ff..8d67549 100644 --- a/arborist/migrate.py +++ b/arborist/migrate.py @@ -540,6 +540,18 @@ def _rebuild_fts_on_target(conn: sqlite3.Connection) -> None: """ 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 fts_rows: list[tuple[int, str]] = [] cur = conn.execute(