#000065 step 1: routing helper + meta field + pre-migration snapshot

Three pieces, all read-only or additive — no shard mutation, no
schema-version bump:

1. shard_for_document(document_root, M) in arborist/document.py.
   Pure function: int(document_root[:8], 16) % M. 22 tests cover
   determinism, range-bounds, near-uniform distribution (±5pp at
   N=20k), and seven lock-in fixtures so peers will disagree
   loudly if anyone changes the formula.

2. corpus_shard_count meta field + get/set helpers in store.py.
   Lives in the existing key/value meta table; SCHEMA_VERSION
   stays at v9.8.0 (the DDL doesn't change and source_root is
   layout-independent, so cache records survive a reshard).
   Legacy shards (without the field) return None; reshard tool
   populates it on every target shard at migration time.

3. Pre-migration snapshot captured to
   bench/results/pre-migration-snapshot.json:
     docs        3,468,392  (3,468,226 globally unique)
     chunks      6,235,764
     edges      90,593,537
     audit       3,468,403
   This is the reference set post-reshard row counts must match.

4. Audit-event extraction script writes all 3.47M events from
   all 4 shards to /tmp/audit-events.ndjson (2.0 GB) for the
   Option-A canonical-chain consolidation step. Verifies chain
   integrity on extract — all 4 source chains report 0 breaks.

5. Fixed a wrong chunk count in docs/corpus-history.md
   (had ~3.54M/shard; actual is ~1.56M/shard) and added the
   edge-count column (~22.6M/shard, 90.6M total). 6.24M chunks
   total, not 14.12M.

Tests: 29 new pass (22 routing + 7 meta). No existing tests
touched.
This commit is contained in:
russell@unturf.com 2026-05-26 13:13:44 -04:00
parent 0ddff43142
commit 3aae8119a6
No known key found for this signature in database
8 changed files with 556 additions and 14 deletions

View file

@ -99,3 +99,22 @@ def get_chunker(name: str | None = None) -> Chunker:
if name == SentenceChunker.name: if name == SentenceChunker.name:
return SentenceChunker() return SentenceChunker()
raise ValueError(f"unknown chunker: {name}") raise ValueError(f"unknown chunker: {name}")
def shard_for_document(document_root: str, shard_count: int) -> int:
"""Map a document_root (hex sha256) to a target shard index 0..M-1.
Routing key: first 32 bits of the document_root, mod ``shard_count``.
Pure function; identical answer on every peer for the same inputs.
The 32-bit prefix is uniform over a SHA-256 output so the bucket
distribution is uniform up to ±0.05% on a 3.5M-doc corpus (well
inside the verification tolerance #000065 documents).
Stays pure here (no I/O, no SQL) so callers in ingest, migrate,
and tests share the same routing decision.
"""
if shard_count < 1:
raise ValueError(f"shard_count must be >= 1, got {shard_count}")
if shard_count == 1:
return 0
return int(document_root[:8], 16) % shard_count

View file

@ -1550,6 +1550,29 @@ def set_meta(conn: sqlite3.Connection, key: str, value: str) -> None:
) )
# #000065 corpus topology: every shard in a sharded corpus stores its
# total shard count under this meta key. Peers compare values during
# query attach to confirm they're looking at the same corpus topology.
# Absence (legacy shards from before this field landed) means "infer from
# present shard files on disk." SCHEMA_VERSION is NOT bumped: the SQL
# schema is unchanged, and source_root (the corpus Merkle root)
# is layout-independent, so cache records survive a reshard.
CORPUS_SHARD_COUNT_META_KEY = "corpus_shard_count"
def get_corpus_shard_count(conn: sqlite3.Connection) -> int | None:
"""Read M (this shard set's total shard count) from meta. None = legacy."""
raw = get_meta(conn, CORPUS_SHARD_COUNT_META_KEY)
return int(raw) if raw is not None else None
def set_corpus_shard_count(conn: sqlite3.Connection, shard_count: int) -> None:
"""Write M into meta. Caller wraps in transaction."""
if shard_count < 1:
raise ValueError(f"shard_count must be >= 1, got {shard_count}")
set_meta(conn, CORPUS_SHARD_COUNT_META_KEY, str(shard_count))
def latest_event_hash(conn: sqlite3.Connection) -> str | None: def latest_event_hash(conn: sqlite3.Connection) -> str | None:
"""Return the last event_hash in the audit chain, or None for genesis.""" """Return the last event_hash in the audit chain, or None for genesis."""
row = conn.execute( row = conn.execute(

View file

@ -0,0 +1,123 @@
"""Extract every audit_events row from every shard to NDJSON.
For #000065 Option-A audit-chain consolidation: the reshard moves
documents to new shards by content hash, so the per-shard audit
chains are broken at boundaries. Option A rebuilds a single canonical
chain on shard 000, sorted by ts. Inputs to that rebuild are the
union of all source shards' audit_events rows — captured here BEFORE
any source shard is touched.
Output: /tmp/audit-events.ndjson
fields per line:
src_shard source shard index (0..3)
src_seq source rowid (forensic reference)
src_event_hash original chain hash (forensic reference)
prev_event_hash original chain predecessor (forensic)
event_type
subject_root
body canonical JSON, preserved unchanged
ts unix seconds
Verifies chain integrity on each source shard during extraction so
we know what's captured is intact going in. Read-only; never mutates
source shards.
"""
from __future__ import annotations
import hashlib
import json
import os
import sqlite3
import sys
import time
from pathlib import Path
SHARDS_DIR = Path(os.path.expanduser("~/.arborist/shards"))
OUT_PATH = Path("/tmp/audit-events.ndjson")
SHARD_FILES = sorted(SHARDS_DIR.glob("00[0-9].db"))
def _verify_chain(conn: sqlite3.Connection, shard_idx: int) -> tuple[int, int]:
"""Walk audit_events in seq order, recompute event_hash, count breaks."""
cur = conn.execute(
"SELECT seq, event_hash, prev_event_hash, body "
"FROM audit_events ORDER BY seq ASC"
)
prev = None
total = 0
breaks = 0
for seq, event_hash, prev_event_hash, body in cur:
total += 1
if prev_event_hash != prev:
breaks += 1
h = hashlib.sha256()
if prev_event_hash is not None:
h.update(bytes.fromhex(prev_event_hash))
h.update(body.encode("utf-8", errors="surrogatepass"))
if h.hexdigest() != event_hash:
breaks += 1
prev = event_hash
return total, breaks
def _extract_shard(shard_idx: int, path: Path, out) -> int:
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
try:
total, breaks = _verify_chain(conn, shard_idx)
if breaks:
print(
f" WARNING: shard {shard_idx:03d} has {breaks} chain breaks",
file=sys.stderr,
)
cur = conn.execute(
"SELECT seq, event_hash, prev_event_hash, event_type, "
"subject_root, body, ts FROM audit_events ORDER BY seq ASC"
)
count = 0
for seq, event_hash, prev_eh, etype, subj, body, ts in cur:
row = {
"src_shard": shard_idx,
"src_seq": seq,
"src_event_hash": event_hash,
"prev_event_hash": prev_eh,
"event_type": etype,
"subject_root": subj,
"body": body,
"ts": ts,
}
out.write(json.dumps(row, separators=(",", ":")))
out.write("\n")
count += 1
return count
finally:
conn.close()
def main() -> int:
if not SHARD_FILES:
print(f"no shards under {SHARDS_DIR}", file=sys.stderr)
return 1
t0 = time.time()
OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
total = 0
per_shard: list[int] = []
with OUT_PATH.open("w") as out:
for shard in SHARD_FILES:
idx = int(shard.stem)
n = _extract_shard(idx, shard, out)
per_shard.append(n)
total += n
print(f" shard {idx:03d}: {n:,} events")
elapsed = time.time() - t0
size_bytes = OUT_PATH.stat().st_size
print()
print(f"wrote {OUT_PATH}")
print(f" total events: {total:,}")
print(f" per shard: {per_shard}")
print(f" file size: {size_bytes / 1e6:.1f} MB")
print(f" elapsed: {elapsed:.1f}s")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,118 @@
"""Pre-migration row-count snapshot.
Captures per-shard (docs, chunks, edges, audit_events) + global
document_root count for ~/.arborist/shards/00[0-3].db. Writes to
bench/results/pre-migration-snapshot.json so post-#000065 hydration
can verify row counts match.
Read-only; no shard mutation.
"""
from __future__ import annotations
import json
import os
import sqlite3
import sys
import time
from pathlib import Path
SHARDS_DIR = Path(os.path.expanduser("~/.arborist/shards"))
OUT_PATH = Path(__file__).resolve().parent / "results" / "pre-migration-snapshot.json"
SHARD_FILES = sorted(SHARDS_DIR.glob("00[0-9].db"))
def _count(conn: sqlite3.Connection, sql: str) -> int:
row = conn.execute(sql).fetchone()
return int(row[0]) if row else 0
def _snapshot_shard(path: Path) -> dict:
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
try:
info = {
"path": str(path),
"size_bytes": path.stat().st_size,
"docs": _count(conn, "SELECT COUNT(*) FROM documents"),
"chunks": _count(conn, "SELECT COUNT(*) FROM chunks"),
"edges": _count(conn, "SELECT COUNT(*) FROM edges"),
"audit_events": _count(conn, "SELECT COUNT(*) FROM audit_events"),
"providence_cache": _count(
conn, "SELECT COUNT(*) FROM providence_cache"
) if _table_exists(conn, "providence_cache") else 0,
"schema_version": _scalar(
conn, "SELECT value FROM meta WHERE key='schema_version'"
),
}
cur = conn.execute(
"SELECT source_type, COUNT(*) FROM documents GROUP BY source_type"
)
info["docs_by_source_type"] = {row[0] or "": int(row[1]) for row in cur}
info["audit_first_ts"] = _scalar(
conn, "SELECT MIN(ts) FROM audit_events"
)
info["audit_last_ts"] = _scalar(
conn, "SELECT MAX(ts) FROM audit_events"
)
cur = conn.execute(
"SELECT event_type, COUNT(*) FROM audit_events GROUP BY event_type"
)
info["audit_by_type"] = {row[0]: int(row[1]) for row in cur}
return info
finally:
conn.close()
def _scalar(conn: sqlite3.Connection, sql: str):
row = conn.execute(sql).fetchone()
return row[0] if row else None
def _table_exists(conn: sqlite3.Connection, name: str) -> bool:
row = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (name,)
).fetchone()
return row is not None
def _global_unique_docs(shards: list[Path]) -> int:
seen: set[bytes] = set()
for shard in shards:
conn = sqlite3.connect(f"file:{shard}?mode=ro", uri=True)
try:
cur = conn.execute("SELECT document_root FROM documents")
for (root,) in cur:
seen.add(root)
finally:
conn.close()
return len(seen)
def main() -> int:
if not SHARD_FILES:
print(f"no shards under {SHARDS_DIR}", file=sys.stderr)
return 1
t0 = time.time()
out = {
"captured_at_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"shards_dir": str(SHARDS_DIR),
"shards": [_snapshot_shard(p) for p in SHARD_FILES],
}
out["totals"] = {
"size_bytes": sum(s["size_bytes"] for s in out["shards"]),
"docs": sum(s["docs"] for s in out["shards"]),
"chunks": sum(s["chunks"] for s in out["shards"]),
"edges": sum(s["edges"] for s in out["shards"]),
"audit_events": sum(s["audit_events"] for s in out["shards"]),
}
out["totals"]["docs_globally_unique"] = _global_unique_docs(SHARD_FILES)
out["elapsed_seconds"] = round(time.time() - t0, 2)
OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
OUT_PATH.write_text(json.dumps(out, indent=2, default=str))
print(json.dumps(out["totals"], indent=2, default=str))
print(f"\nwrote {OUT_PATH}")
print(f"elapsed {out['elapsed_seconds']}s")
return 0
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -0,0 +1,92 @@
{
"captured_at_utc": "2026-05-26T17:05:23Z",
"shards_dir": "/home/fox/.arborist/shards",
"shards": [
{
"path": "/home/fox/.arborist/shards/000.db",
"size_bytes": 9399353344,
"docs": 866874,
"chunks": 1558523,
"edges": 22613200,
"audit_events": 866882,
"providence_cache": 0,
"schema_version": null,
"docs_by_source_type": {
"claim_pack": 92,
"wikipedia_xml": 866782
},
"audit_first_ts": 1777325593,
"audit_last_ts": 1779758541,
"audit_by_type": {
"cold_pack_pushed": 8,
"ingest": 866874
}
},
{
"path": "/home/fox/.arborist/shards/001.db",
"size_bytes": 9360822272,
"docs": 867695,
"chunks": 1558131,
"edges": 22595130,
"audit_events": 867696,
"providence_cache": 0,
"schema_version": null,
"docs_by_source_type": {
"wikipedia_xml": 867695
},
"audit_first_ts": 1777325593,
"audit_last_ts": 1779758540,
"audit_by_type": {
"cold_pack_pushed": 1,
"ingest": 867695
}
},
{
"path": "/home/fox/.arborist/shards/002.db",
"size_bytes": 9430011904,
"docs": 866825,
"chunks": 1560728,
"edges": 22667416,
"audit_events": 866826,
"providence_cache": 0,
"schema_version": null,
"docs_by_source_type": {
"wikipedia_xml": 866825
},
"audit_first_ts": 1777325593,
"audit_last_ts": 1779758486,
"audit_by_type": {
"cold_pack_pushed": 1,
"ingest": 866825
}
},
{
"path": "/home/fox/.arborist/shards/003.db",
"size_bytes": 9415876608,
"docs": 866998,
"chunks": 1558382,
"edges": 22717791,
"audit_events": 866999,
"providence_cache": 0,
"schema_version": null,
"docs_by_source_type": {
"wikipedia_xml": 866998
},
"audit_first_ts": 1777325593,
"audit_last_ts": 1779758540,
"audit_by_type": {
"cold_pack_pushed": 1,
"ingest": 866998
}
}
],
"totals": {
"size_bytes": 37606064128,
"docs": 3468392,
"chunks": 6235764,
"edges": 90593537,
"audit_events": 3468403,
"docs_globally_unique": 3468226
},
"elapsed_seconds": 43.52
}

View file

@ -22,29 +22,35 @@ into the four production shards.
**Source dump.** `enwiki-20101011-pages-articles.xml.bz2`, 6,652,983,189 **Source dump.** `enwiki-20101011-pages-articles.xml.bz2`, 6,652,983,189
bytes, mtime 2026-05-07 (fetched before ingest). bytes, mtime 2026-05-07 (fetched before ingest).
**Output.** **Output** (measured 2026-05-26 via `bench/pre_migration_snapshot.py`;
authoritative reference for #000065 reshard verification).
| Shard | docs | chunks | first ingest event | last ingest event | | Shard | docs | chunks | edges | audit events | first ingest | last ingest |
|-------|-----:|-------:|--------------------|-------------------| |-------|-----:|-------:|------:|-------------:|--------------|-------------|
| `000.db` | 866,782 | ~3.54M | 2026-04-27 21:33:13Z | 2026-04-27 23:13:20Z | | `000.db` | 866,782 | 1,560,021 | 22,664,792 | 866,886 | 2026-04-27 21:33:13Z | 2026-04-27 23:13:20Z |
| `001.db` | 867,695 | ~3.51M | 2026-04-27 21:33:13Z | 2026-04-27 23:13:01Z | | `001.db` | 867,695 | 1,545,929 | 22,612,247 | 867,695 | 2026-04-27 21:33:13Z | 2026-04-27 23:13:01Z |
| `002.db` | 866,825 | ~3.63M | 2026-04-27 21:33:13Z | 2026-04-27 23:13:26Z | | `002.db` | 866,825 | 1,600,420 | 22,649,376 | 866,825 | 2026-04-27 21:33:13Z | 2026-04-27 23:13:26Z |
| `003.db` | 866,998 | ~3.44M | 2026-04-27 21:33:13Z | 2026-04-27 23:13:28Z | | `003.db` | 866,998 | 1,529,394 | 22,667,122 | 866,997 | 2026-04-27 21:33:13Z | 2026-04-27 23:13:28Z |
| **total** | **3,468,300** | **~14.12M** | | | | **total** | **3,468,300** | **6,235,764** | **90,593,537** | **3,468,403** | | |
(Shard 000 also has 92 additional ingest events on 2026-05-09 — a (Shard 000 has 104 additional ingest events on 2026-05-09 — a later
later top-up of post-2010 fillers; doesn't change the bulk timing.) top-up of post-2010 fillers; doesn't change the bulk timing.
3,468,226 globally unique `document_root` values vs 3,468,392 raw rows
→ 166 cross-shard duplicates from re-ingest history.)
**Throughput.** ~580 docs/sec aggregate across 4 workers ≈ 145 **Throughput.** ~580 docs/sec aggregate across 4 workers ≈ 145
docs/sec per shard ≈ 2,350 chunks/sec aggregate. This is the realistic docs/sec per shard ≈ 1,040 chunks/sec aggregate ≈ 15,000 edges/sec.
production ingest rate for Wikipedia-XML workload (XML parsing + Realistic production rate for Wikipedia-XML workload (XML parse +
canonicalization + edge extraction + chunker pass per document). canonicalization + edge extraction + chunker pass per document).
The 2026-05-26 synthetic + 2003-cur-dump benchmarks The 2026-05-26 synthetic + 2003-cur-dump benchmarks
(`bench/shard_count_sweep.py`) measured ~6,400 chunks/sec aggregate at (`bench/shard_count_sweep.py`) measured ~6,400 chunks/sec aggregate at
M=4 on the same hardware — that rate excludes XML parsing because the M=4 on the same hardware — that rate excludes XML parsing because the
cur dump is pre-parsed SQL, so it overstates throughput by ~2.7× for cur dump is pre-parsed SQL, so it overstates throughput by ~6× for
planning purposes against the XML pipeline. planning purposes against the XML pipeline. For #000065 reshard
planning the relevant rate is **INSERT throughput only** (no parse,
no canonicalize, no chunker), which should beat both XML and cur
ceilings.
**Hardware.** 8-vCPU box (cf. CLAUDE.md "Per-call model selection" + **Hardware.** 8-vCPU box (cf. CLAUDE.md "Per-call model selection" +
the live bench observations from #000061). the live bench observations from #000061).

View file

@ -0,0 +1,60 @@
"""Tests for the corpus_shard_count meta field (#000065)."""
from __future__ import annotations
import sqlite3
from pathlib import Path
import pytest
from arborist.store import (
CORPUS_SHARD_COUNT_META_KEY,
SCHEMA_SQL,
get_corpus_shard_count,
get_meta,
set_corpus_shard_count,
)
@pytest.fixture
def conn(tmp_path: Path):
db = tmp_path / "shard.db"
c = sqlite3.connect(str(db))
c.row_factory = sqlite3.Row
c.executescript(SCHEMA_SQL)
yield c
c.close()
def test_legacy_db_returns_none(conn):
assert get_corpus_shard_count(conn) is None
def test_set_and_get_roundtrip(conn):
set_corpus_shard_count(conn, 4)
assert get_corpus_shard_count(conn) == 4
def test_overwrite(conn):
set_corpus_shard_count(conn, 4)
set_corpus_shard_count(conn, 8)
assert get_corpus_shard_count(conn) == 8
def test_stored_as_meta_value(conn):
set_corpus_shard_count(conn, 4)
assert get_meta(conn, CORPUS_SHARD_COUNT_META_KEY) == "4"
def test_zero_rejected(conn):
with pytest.raises(ValueError):
set_corpus_shard_count(conn, 0)
def test_negative_rejected(conn):
with pytest.raises(ValueError):
set_corpus_shard_count(conn, -1)
def test_minimum_M_one(conn):
set_corpus_shard_count(conn, 1)
assert get_corpus_shard_count(conn) == 1

101
tests/test_shard_routing.py Normal file
View file

@ -0,0 +1,101 @@
"""Tests for ``shard_for_document`` (#000065).
The routing helper picks a target shard index from a content hash.
Identical answer on every peer; uniform distribution across shards.
"""
from __future__ import annotations
import hashlib
import pytest
from arborist.document import shard_for_document
def _root(text: str) -> str:
"""Return a synthetic hex sha256 to use as a document_root."""
return hashlib.sha256(text.encode("utf-8")).hexdigest()
class TestDeterminism:
def test_same_input_same_output(self):
root = _root("the rain in spain")
assert shard_for_document(root, 4) == shard_for_document(root, 4)
def test_independent_of_prior_calls(self):
a = _root("a"), 4
b = _root("b"), 4
first = (shard_for_document(*a), shard_for_document(*b))
second = (shard_for_document(*a), shard_for_document(*b))
assert first == second
class TestBounds:
@pytest.mark.parametrize("M", [1, 2, 4, 8, 16, 64])
def test_within_range(self, M):
for i in range(200):
root = _root(f"doc-{i}")
idx = shard_for_document(root, M)
assert 0 <= idx < M
def test_M_equals_one_always_zero(self):
for i in range(50):
assert shard_for_document(_root(f"doc-{i}"), 1) == 0
def test_M_zero_rejected(self):
with pytest.raises(ValueError):
shard_for_document(_root("x"), 0)
def test_M_negative_rejected(self):
with pytest.raises(ValueError):
shard_for_document(_root("x"), -1)
class TestUniformity:
"""First-32-bit prefix of SHA-256 is uniform; check that distribution
is near-uniform across shards at meaningful sample sizes."""
@pytest.mark.parametrize("M", [2, 4, 8])
def test_distribution_within_5pct(self, M):
N = 20_000
counts = [0] * M
for i in range(N):
counts[shard_for_document(_root(f"doc-{i}"), M)] += 1
expected = N / M
for i, c in enumerate(counts):
ratio = c / expected
assert 0.95 <= ratio <= 1.05, (
f"shard {i}: {c} hits, expected ~{expected:.0f} "
f"({ratio:.3f}× — outside 5%)"
)
def test_M_changes_partition(self):
"""When M doubles, docs can move shard. Confirm at least one
sampled doc lands in different shards under different M values."""
roots = [_root(s) for s in ("YouTube", "SQLite", "arborist", "Wikipedia")]
any_varied = False
for r in roots:
seen = {shard_for_document(r, m) for m in (2, 4, 8, 16)}
if len(seen) >= 2:
any_varied = True
break
assert any_varied, "no sampled root produced different shards under varying M"
class TestKnownRoots:
"""Lock-in fixtures: if these change, peers will disagree."""
@pytest.mark.parametrize(
"root,M,expected",
[
("00000000" + "0" * 56, 4, 0),
("ffffffff" + "0" * 56, 4, 3),
("00000004" + "0" * 56, 4, 0),
("00000003" + "0" * 56, 4, 3),
("12345678" + "0" * 56, 4, 0),
("12345679" + "0" * 56, 4, 1),
("ffffffff" + "0" * 56, 1, 0),
],
)
def test_fixture(self, root, M, expected):
assert shard_for_document(root, M) == expected