From 2f214202e8deaf24e3d293eff6b31f2b3e5fa260 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Thu, 6 Aug 2026 14:33:13 -0700 Subject: [PATCH 01/10] =?UTF-8?q?perf(fullmap):=20upgrade=20redb=202?= =?UTF-8?q?=E2=86=924=20and=20open=20fullmap=20readers=20with=20shared=20l?= =?UTF-8?q?ocks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fullmap is the sole redb user. redb 2.x takes an EXCLUSIVE flock on every open — even pure reads — so only one opener per fullmap file could exist system-wide: the agent supervisor, its code-executor subprocesses, and parallel agent runs serialized on that lock (masked by the Python-side lock-retry backoff), and a killed-mid-build subprocess could strand it. redb ≥ 3 added ReadOnlyDatabase (shared lock); 4.1 additionally speeds up concurrent multi-threaded reads (~15% on upstream benchmarks) and general write performance (~1.5x on upstream write benchmarks), which benefits the build's redb write phase. This change: - Bumps redb 2.6 → 4.1 (deps + dev-deps). - Switches the ENTIRE read path (DB cache, open_cached/open_cached_shard, schema validation, shard fan-out lookups, CURIE/dim hydration, and the post-build cache prime) to ReadOnlyDatabase — shared locks, so any number of processes read concurrently; only a build-fullmap rebuild (writer) briefly blocks readers. - Maps UpgradeRequired/RepairAborted at read-only open to the actionable rebuild hint (a read-only open can neither upgrade the old v2 file format nor repair a crash-damaged file). - Write path keeps the exclusive-lock Database; set_durability now returns Result (redb ≥ 3). The 16-shard concurrent-writer scheme is unchanged — it remains redb's maximum write parallelism (single WriteTransaction per file by design). - Bumps the schema tag to tablassert.fullmap.v5 (layout unchanged) so the rebuild is explicit and a downgraded extension rejects new files loudly. BREAKING: redb ≥ 3 dropped the v2 file format; existing fullmap files must be rebuilt once via 'tablassert build-fullmap' (BABEL downloads stay cached, so the rebuild is cheap). --- CHANGELOG.md | 6 + docs/fullmap.md | 11 +- rust/Cargo.lock | 4 +- rust/Cargo.toml | 4 +- rust/examples/count_tables.rs | 3 +- rust/src/fullmap.rs | 224 ++++++++++++++++++++++++++++------ rust/tests/build_golden.rs | 4 +- 7 files changed, 209 insertions(+), 47 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d4f45560..12e03105 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,12 @@ All notable changes to this project are documented in this file. ## Unreleased +### Breaking Changes +- **Fullmap databases built by older releases must be rebuilt.** The Rust extension upgraded its embedded database engine from redb 2.6 to redb 4.1, and redb ≥ 3 dropped the old v2 file format. Existing `fullmap.redb` (and sibling `fullmap.s*.redb`) files fail to open with `fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'`. Run `tablassert build-fullmap` once after upgrading (BABEL downloads stay cached, so the rebuild is cheap). The on-disk fullmap schema is now `tablassert.fullmap.v5` (the table layout is unchanged; the bump makes the redb-4 rebuild explicit and lets an older extension reject new files loudly). + +### Changed +- **Fullmap reads no longer serialize across processes.** The lookup path (`lookup_fullmap_terms` and the `hydrate_*` helpers) now opens the fullmap redb files READ-ONLY with a SHARED file lock (redb ≥ 3 `ReadOnlyDatabase`) instead of an exclusive lock: concurrent readers — the agent supervisor, its code-executor subprocesses, and parallel `agent run` processes — no longer contend on the fullmap lock ("Database already open"); only a running `build-fullmap` rebuild can briefly block readers. Read-only opens also never touch the file mtime, making the mtime-keyed Python lookup caches fully stable. The redb 4.1 upgrade additionally speeds up multi-threaded shard reads (~15% on upstream benchmarks) and the fullmap build's redb write phase (~1.5x on upstream write benchmarks). + ## 8.1.0 - 2026-08-03 ### Breaking Changes diff --git a/docs/fullmap.md b/docs/fullmap.md index 071b55d5..8b0dc472 100644 --- a/docs/fullmap.md +++ b/docs/fullmap.md @@ -109,7 +109,7 @@ they hold six tables (see `rust/src/fullmap.rs`): | `categories` | Compact `u16` id → Biolink category string (primary file) | | `sources` | Compact `u8` id → source metadata (name/version) (primary file) | | `curies` | Compact `u32` id → CURIE record (CURIE, preferred name, category, taxon, source) (primary file) | -| `meta` | Schema version tag (`tablassert.fullmap.v4`), the shard count (`shards`), and the BABEL `source_version` used to build the file (primary file) | +| `meta` | Schema version tag (`tablassert.fullmap.v5`), the shard count (`shards`), and the BABEL `source_version` used to build the file (primary file) | The shard files must remain alongside the primary file — lookups discover them as siblings of the resolved primary path. @@ -118,8 +118,13 @@ Lookups (`lookup_fullmap_terms`) check the primary's `meta` schema tag before re `shards` count to open exactly that many shard files, and fan the query terms out across the shards in parallel (releasing the GIL, one reader per non-empty shard, re-merged into input order); a mismatched or missing tag raises rather than silently reading incompatible data. Databases built under the older -`v1`/`v2`/`v3` schemas are rejected — there is no automatic schema migration, so a schema bump (including -the v3→v4 move to sharded files) requires rebuilding via `tablassert build-fullmap`. +`v1`/`v2`/`v3`/`v4` schemas are rejected — there is no automatic schema migration, so a schema bump +(including the v3→v4 move to sharded files and the v4→v5 move to the redb 4 engine) requires rebuilding +via `tablassert build-fullmap`. + +Readers open every fullmap file READ-ONLY with a SHARED file lock (redb ≥ 3 `ReadOnlyDatabase`), so any +number of processes can run lookups against the same fullmap concurrently; only a `build-fullmap` rebuild +(an exclusive-lock writer) briefly blocks readers. ## Usage in Graph Config diff --git a/rust/Cargo.lock b/rust/Cargo.lock index a5813842..f4fbcebd 100644 --- a/rust/Cargo.lock +++ b/rust/Cargo.lock @@ -422,9 +422,9 @@ dependencies = [ [[package]] name = "redb" -version = "2.6.3" +version = "4.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8eca1e9d98d5a7e9002d0013e18d5a9b000aee942eb134883a82f06ebffb6c01" +checksum = "8e925444704b5f17d32bf42f5b6e2df050bceebc3dcd6e71cc73dafe8092e839" dependencies = [ "libc", ] diff --git a/rust/Cargo.toml b/rust/Cargo.toml index c60464e7..8ed17d7d 100644 --- a/rust/Cargo.toml +++ b/rust/Cargo.toml @@ -18,7 +18,7 @@ memmap2 = "0.9" mimalloc = { version = "0.1", default-features = false } pyo3 = "0.29" rayon = "1" -redb = "2" +redb = "4" rlimit = "0.10" rustc-hash = "1" serde = { version = "1", features = ["derive"] } @@ -29,7 +29,7 @@ xxhash-rust = { version = "0.8", features = ["xxh64", "xxh3"] } [dev-dependencies] bincode = "1" flate2 = "1" -redb = "2" +redb = "4" serde = { version = "1", features = ["derive"] } serde_json = "1" tempfile = "3" diff --git a/rust/examples/count_tables.rs b/rust/examples/count_tables.rs index 49a952e5..2afb0c07 100644 --- a/rust/examples/count_tables.rs +++ b/rust/examples/count_tables.rs @@ -1,5 +1,6 @@ //! Count rows in each table of one or more fullmap redb files. //! Usage: cargo run --release --example count_tables -- [db2 ...] +use redb::ReadableDatabase; use redb::ReadableTableMetadata; use redb::TableDefinition; @@ -11,7 +12,7 @@ const CURIES: TableDefinition = TableDefinition::new("curies"); fn main() { for arg in std::env::args().skip(1) { - let db = redb::Database::open(&arg).expect("open db"); + let db = redb::ReadOnlyDatabase::open(&arg).expect("open db"); let read = db.begin_read().expect("begin read"); let records = read .open_table(RECORDS) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index cc98042e..180a27d5 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -3,7 +3,9 @@ use pyo3::exceptions::{PyRuntimeError, PyValueError}; use pyo3::prelude::*; use pyo3::types::{PyAny, PyDict, PyList}; use rayon::prelude::*; -use redb::{Database, Durability, ReadableTable, TableDefinition}; +use redb::{ + Database, Durability, ReadOnlyDatabase, ReadableDatabase, ReadableTable, TableDefinition, +}; use rustc_hash::FxHasher; use serde::{Deserialize, Serialize}; use std::borrow::Cow; @@ -26,7 +28,8 @@ const CATEGORIES: TableDefinition = TableDefinition::new("categories" const SOURCES: TableDefinition = TableDefinition::new("sources"); const CURIES: TableDefinition = TableDefinition::new("curies"); const META: TableDefinition<&str, &str> = TableDefinition::new("meta"); -const SCHEMA_VERSION: &str = "tablassert.fullmap.v4"; +const SCHEMA_VERSION: &str = "tablassert.fullmap.v5"; +const SCHEMA_VERSION_V4: &str = "tablassert.fullmap.v4"; const SCHEMA_VERSION_V3: &str = "tablassert.fullmap.v3"; const SCHEMA_VERSION_V2: &str = "tablassert.fullmap.v2"; const SCHEMA_VERSION_V1: &str = "tablassert.fullmap.v1"; @@ -51,7 +54,7 @@ type MergeItem = (Reverse, Reverse, usize, Vec<(u32, u8)>); /// A read-path fan-out job: the `(input_index, term)` bucket routed to one shard /// plus a clone of that shard's handle, so a worker thread owns both outright /// (no shared receiver or borrow). -type ShardJob = (Vec<(usize, String)>, Arc); +type ShardJob = (Vec<(usize, String)>, Arc); /// Fast non-cryptographic hash map for the build-hot paths (per-worker term /// aggregation, dimension/CURIE interning). `FxHasher` (rustc's own hasher) is /// a single multiply-XOR pass — ~3-5x faster than std's SipHash for these short @@ -64,16 +67,18 @@ type FxHashMap = HashMap>; /// ever allocated on the read path. type LineChunk = (u8, Vec>); -/// Database cache keyed by canonical path only. +/// Read-path database cache keyed by canonical path only. /// -/// redb's `Database::open` updates the file mtime, so keying on `(path, mtime)` -/// made every lookup after the first miss the cache and try to re-open the file, -/// which fails because the first handle still holds redb's exclusive `flock` -/// ("Database already open. Cannot acquire lock."). Keying on the path alone is -/// safe: within a process the DB is only rebuilt via `build_fullmap_db`, which -/// evicts the cache explicitly, and redb's exclusive lock prevents an external -/// rebuild while we hold a handle. -static DB_CACHE: OnceLock>>> = OnceLock::new(); +/// Handles are `ReadOnlyDatabase` (redb ≥ 3), which take a SHARED file lock: +/// any number of processes/threads may hold them concurrently, and they only +/// conflict with a writer (`build-fullmap`'s exclusive lock). Keying on the +/// path alone is safe: within a process the DB is only rebuilt via +/// `build_fullmap_db`, which evicts the cache explicitly; externally, redb's +/// exclusive writer lock fails against our shared lock (and the rebuild also +/// removes the files before recreating them), so no cached handle can outlive +/// a rebuild on this path. Read-only opens never write the file, so the +/// mtime-keyed Python-side caches stay stable under lookups. +static DB_CACHE: OnceLock>>> = OnceLock::new(); /// Number of shards for concurrent maps (power of two for mask routing). const SHARD_COUNT: usize = 64; @@ -142,6 +147,25 @@ fn py_err(err: E) -> PyErr { PyRuntimeError::new_err(err.to_string()) } +/// Open a fullmap file with a SHARED (read-only) lock and map open errors to +/// actionable fullmap errors. +/// +/// A read-only open can neither upgrade an outdated file format (redb ≥ 3 +/// dropped the v2 format that redb 2.x builds wrote) nor repair a crash-damaged +/// file — by design — so `UpgradeRequired`/`RepairAborted` surface as the same +/// rebuild hint `validate_schema` produces for an outdated schema tag, instead +/// of a cryptic redb error. Every other error passes through verbatim. +fn open_read_only(path: &Path) -> PyResult { + ReadOnlyDatabase::open(path).map_err(|err| match err { + redb::DatabaseError::UpgradeRequired(_) | redb::DatabaseError::RepairAborted => { + PyRuntimeError::new_err( + "fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'", + ) + } + other => py_err(other), + }) +} + /// Narrow `value` to its cleaned fixed point — repeated trim + matching / /// duplicate quote stripping — returning a sub-slice of `value` (zero /// allocation). Cleaning only ever narrows to a sub-slice, so the result always @@ -1885,7 +1909,9 @@ fn write_shard_records( progress: Option<&Arc>, ) -> PyResult { let mut write = database.begin_write().map_err(py_err)?; - write.set_durability(Durability::None); + // redb ≥ 3 returns `Result` (fails once the transaction has been used); + // this is the first call on a fresh transaction, so it cannot fail here. + write.set_durability(Durability::None).map_err(py_err)?; let mut table = write.open_table(RECORDS).map_err(py_err)?; let mut merge = MergeHeap::new(run_paths).map_err(py_err)?; // Pre-size the batch to the flush threshold so it never regrows (each @@ -2053,7 +2079,7 @@ fn evict_cached_path(path: &Path) -> PyResult<()> { Ok(()) } -fn cache_database(path: &Path, database: Arc) -> PyResult<()> { +fn cache_database(path: &Path, database: Arc) -> PyResult<()> { let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); cache.write().map_err(py_err)?.insert(canonical, database); @@ -2271,8 +2297,10 @@ pub fn build_fullmap_db( ) })?; - // Cache the freshly-built database for the read path. - let database = Arc::new(Database::open(&output).map_err(py_err)?); + // Cache the freshly-built database for the read path (shared read-only + // lock, so concurrent lookups — including from other processes — never + // contend with it). + let database = Arc::new(open_read_only(&output)?); cache_database(&output, database)?; Ok(()) } @@ -2281,7 +2309,7 @@ pub fn build_fullmap_db( // Read path // --------------------------------------------------------------------------- -fn validate_schema(database: &Database) -> PyResult<()> { +fn validate_schema(database: &ReadOnlyDatabase) -> PyResult<()> { let read = database.begin_read().map_err(py_err)?; let meta = read.open_table(META).map_err(py_err)?; let schema = meta @@ -2290,7 +2318,7 @@ fn validate_schema(database: &Database) -> PyResult<()> { .map(|x| x.value().to_string()); match schema.as_deref() { Some(SCHEMA_VERSION) => Ok(()), - Some(SCHEMA_VERSION_V3 | SCHEMA_VERSION_V2 | SCHEMA_VERSION_V1) => { + Some(SCHEMA_VERSION_V4 | SCHEMA_VERSION_V3 | SCHEMA_VERSION_V2 | SCHEMA_VERSION_V1) => { Err(PyRuntimeError::new_err( "fullmap DB is outdated; rebuild with 'tablassert build-fullmap'", )) @@ -2299,13 +2327,13 @@ fn validate_schema(database: &Database) -> PyResult<()> { } } -fn open_cached(db: PathBuf) -> PyResult> { +fn open_cached(db: PathBuf) -> PyResult> { let canonical = std::fs::canonicalize(&db).unwrap_or(db); let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); if let Some(database) = cache.read().map_err(py_err)?.get(&canonical) { return Ok(Arc::clone(database)); } - let database = Arc::new(Database::open(&canonical).map_err(py_err)?); + let database = Arc::new(open_read_only(&canonical)?); validate_schema(&database)?; let cached = Arc::clone(&database); cache.write().map_err(py_err)?.insert(canonical, database); @@ -2315,14 +2343,14 @@ fn open_cached(db: PathBuf) -> PyResult> { /// Open (and cache) one RECORDS shard by index, deriving its path from the /// primary DB path. Shards hold only RECORDS (no META), so they are not /// schema-validated here — the primary's `validate_schema` gates the layout. -fn open_cached_shard(primary: &Path, index: usize) -> PyResult> { +fn open_cached_shard(primary: &Path, index: usize) -> PyResult> { let path = shard_path(primary, index); let canonical = std::fs::canonicalize(&path).unwrap_or(path); let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); if let Some(database) = cache.read().map_err(py_err)?.get(&canonical) { return Ok(Arc::clone(database)); } - let database = Arc::new(Database::open(&canonical).map_err(py_err)?); + let database = Arc::new(open_read_only(&canonical)?); let cached = Arc::clone(&database); cache.write().map_err(py_err)?.insert(canonical, database); Ok(cached) @@ -2333,7 +2361,7 @@ fn open_cached_shard(primary: &Path, index: usize) -> PyResult> { /// `SHARD_COUNT_SHARDS`; reading the value back keeps the read path compatible /// with databases built before the count was pinned, opening exactly the shards /// that exist and routing with the matching mask. -fn shard_count_of(database: &Database) -> PyResult { +fn shard_count_of(database: &ReadOnlyDatabase) -> PyResult { let read = database.begin_read().map_err(py_err)?; let meta = read.open_table(META).map_err(py_err)?; let count = meta @@ -2353,7 +2381,7 @@ fn shard_count_of(database: &Database) -> PyResult { /// Open (and cache) all RECORDS shard handles for a primary DB path. The shard /// count is read from the primary's META so the read path opens exactly the /// shards the build wrote. -fn open_cached_shards(primary: &Path) -> PyResult>> { +fn open_cached_shards(primary: &Path) -> PyResult>> { let database = open_cached(primary.to_path_buf())?; let shard_count = shard_count_of(&database)?; (0..shard_count) @@ -2361,7 +2389,7 @@ fn open_cached_shards(primary: &Path) -> PyResult>> { .collect() } -fn lookup_pair_chunk(shards: &[Arc], terms: &[String]) -> PyResult { +fn lookup_pair_chunk(shards: &[Arc], terms: &[String]) -> PyResult { // One read transaction + RECORDS table per shard, opened once; each query // term is routed to its shard via `term_shard`. let reads: Vec<_> = shards @@ -2395,7 +2423,7 @@ fn lookup_pair_chunk(shards: &[Arc], terms: &[String]) -> PyResult PyResult> { let read = shard.begin_read().map_err(py_err)?; @@ -2425,7 +2453,7 @@ fn lookup_shard_bucket( /// cap are read on the calling thread, which still overlaps with the spawned /// readers. Pure Rust end-to-end (no `Python`). fn lookup_pair_terms_db( - shards: &[Arc], + shards: &[Arc], terms: &[String], workers: usize, ) -> PyResult { @@ -2523,7 +2551,7 @@ fn lookup_pair_terms( } fn load_string_table( - database: &Database, + database: &ReadOnlyDatabase, table_definition: TableDefinition, ) -> PyResult> { let read = database.begin_read().map_err(py_err)?; @@ -2536,7 +2564,7 @@ fn load_string_table( Ok(out) } -fn load_sources(database: &Database) -> PyResult> { +fn load_sources(database: &ReadOnlyDatabase) -> PyResult> { let read = database.begin_read().map_err(py_err)?; let table = read.open_table(SOURCES).map_err(py_err)?; let mut out = HashMap::new(); @@ -2548,7 +2576,7 @@ fn load_sources(database: &Database) -> PyResult> { Ok(out) } -fn hydrate_curie_rows(database: &Database, curie_ids: &[u32]) -> PyResult> { +fn hydrate_curie_rows(database: &ReadOnlyDatabase, curie_ids: &[u32]) -> PyResult> { let read = database.begin_read().map_err(py_err)?; let table = read.open_table(CURIES).map_err(py_err)?; let mut out = Vec::with_capacity(curie_ids.len()); @@ -2876,7 +2904,7 @@ mod tests { assert_eq!(rows[0].1[0].source_version, FULLMAP_SOURCE_VERSION); } - /// The v4 layout keeps dims+CURIES+META in the primary and moves RECORDS + /// The sharded (v4+) layout keeps dims+CURIES+META in the primary and moves RECORDS /// into sibling shard files. The primary must NOT carry a RECORDS table, /// META must advertise both the schema and the shard count, and every shard /// file must exist (even empty) holding a RECORDS table — this is the @@ -2914,7 +2942,7 @@ mod tests { let _curies = read.open_table(CURIES).unwrap(); assert!( read.open_table(RECORDS).is_err(), - "primary must not hold a RECORDS table in the v4 layout" + "primary must not hold a RECORDS table in the sharded (v4+) layout" ); drop(read); drop(database); @@ -2923,7 +2951,7 @@ mod tests { for index in 0..SHARD_COUNT_SHARDS { let shard = shard_path(&output, index); assert!(shard.exists(), "missing shard file {shard:?}"); - let db = Database::open(&shard).unwrap(); + let db = ReadOnlyDatabase::open(&shard).unwrap(); let read = db.begin_read().unwrap(); let _records = read.open_table(RECORDS).unwrap(); } @@ -2996,7 +3024,7 @@ mod tests { write.commit().unwrap(); drop(database); - let database = Database::open(&output).unwrap(); + let database = ReadOnlyDatabase::open(&output).unwrap(); let count = shard_count_of(&database).unwrap(); drop(database); count @@ -3523,12 +3551,13 @@ mod tests { drop(read); // Exactly s0+s1 exist, each holding a RECORDS table; higher shard files - // up to the compile-time cap must NOT exist. (Direct opens are scoped so - // their flocks drop before the cached opens.) + // up to the compile-time cap must NOT exist. (Read-only opens take a + // SHARED lock, so they coexist even with the cached read-only opens + // that follow.) for index in 0..2 { let shard = shard_path(&output, index); assert!(shard.exists(), "missing shard file {shard:?}"); - let db = Database::open(&shard).unwrap(); + let db = ReadOnlyDatabase::open(&shard).unwrap(); let read = db.begin_read().unwrap(); let _records = read.open_table(RECORDS).unwrap(); } @@ -3720,6 +3749,127 @@ mod tests { .contains("fullmap DB is outdated; rebuild with 'tablassert build-fullmap'")); } + /// Two `ReadOnlyDatabase` handles on one file coexist (shared locks) and + /// both read — the in-process analogue of the cross-process reader + /// guarantee that motivated the redb 4 read-only switch. + #[test] + fn two_read_only_handles_coexist_on_one_file() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + let database = Database::create(&output).unwrap(); + let write = database.begin_write().unwrap(); + { + let mut meta = write.open_table(META).unwrap(); + meta.insert("schema", SCHEMA_VERSION).unwrap(); + } + write.commit().unwrap(); + drop(database); + + let first = ReadOnlyDatabase::open(&output).unwrap(); + let second = ReadOnlyDatabase::open(&output).unwrap(); + for db in [&first, &second] { + let read = db.begin_read().unwrap(); + let meta = read.open_table(META).unwrap(); + assert_eq!(meta.get("schema").unwrap().unwrap().value(), SCHEMA_VERSION); + } + } + + /// A live writer holds the exclusive lock, so a read-only open fails with + /// `DatabaseAlreadyOpen` — and succeeds once the writer drops. This is the + /// reader-vs-rebuild window that `_call_with_lock_retry` (Python side) + /// still covers after the shared-lock switch. + #[test] + fn writer_blocks_read_only_open_until_dropped() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + let writer = Database::create(&output).unwrap(); + + let err = match ReadOnlyDatabase::open(&output) { + Ok(_) => panic!("expected read-only open to be blocked by the writer"), + Err(err) => err, + }; + assert!(matches!(err, redb::DatabaseError::DatabaseAlreadyOpen)); + + drop(writer); + ReadOnlyDatabase::open(&output).unwrap(); + } + + /// A file in the old (v2) redb file format must map to the actionable + /// rebuild hint, not a cryptic redb error: redb ≥ 3 dropped the v2 format + /// that pre-upgrade builds wrote, and a rebuild is the only path forward. + /// Simulate the old format by patching the primary commit slot's + /// format-version byte (file offset 64) to 2 — exactly the condition redb + /// 4 detects as `UpgradeRequired`. + #[test] + fn read_only_open_maps_outdated_file_format_to_rebuild_hint() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + let database = Database::create(&output).unwrap(); + let write = database.begin_write().unwrap(); + { + let mut meta = write.open_table(META).unwrap(); + meta.insert("schema", SCHEMA_VERSION).unwrap(); + } + write.commit().unwrap(); + drop(database); + + patch_db_byte(&output, 64, |version| { + assert_eq!(version, 3, "expected the v3 file format version"); + 2 // v2 format version -> UpgradeRequired + }); + + let err = match open_read_only(&output) { + Ok(_) => panic!("expected open_read_only to fail"), + Err(err) => err, + }; + assert!(err.to_string().contains( + "fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'" + )); + } + + /// A crash-damaged file (recovery required) cannot be repaired through a + /// read-only open — redb aborts the repair — so it must surface the same + /// rebuild hint. Simulate the crash state by setting the RECOVERY_REQUIRED + /// flag (bit 2 of the god byte at file offset 9), as redb's own repair + /// tests do. + #[test] + fn read_only_open_maps_recovery_required_to_rebuild_hint() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + let database = Database::create(&output).unwrap(); + drop(database); + + patch_db_byte(&output, 9, |god| god | 2); + + let err = match open_read_only(&output) { + Ok(_) => panic!("expected open_read_only to fail"), + Err(err) => err, + }; + assert!(err.to_string().contains( + "fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'" + )); + } + + /// Test helper: read one byte at `offset` in `path`, transform it through + /// `f`, and write the result back (no other bytes touched). + fn patch_db_byte(path: &Path, offset: u64, f: impl FnOnce(u8) -> u8) { + use std::io::{Read, Seek, SeekFrom, Write}; + let mut file = std::fs::OpenOptions::new() + .read(true) + .write(true) + .open(path) + .unwrap(); + file.seek(SeekFrom::Start(offset)).unwrap(); + let mut byte = [0u8; 1]; + file.read_exact(&mut byte).unwrap(); + file.seek(SeekFrom::Start(offset)).unwrap(); + file.write_all(&[f(byte[0])]).unwrap(); + } + /// `evict_cached_path` must drop the primary AND all default shard handles so /// a rebuild never serves a stale file. Verified by caching the primary plus /// all shard handles, evicting, and confirming none of the canonical paths diff --git a/rust/tests/build_golden.rs b/rust/tests/build_golden.rs index f2f1215c..99fa6ebb 100644 --- a/rust/tests/build_golden.rs +++ b/rust/tests/build_golden.rs @@ -22,7 +22,7 @@ use flate2::write::GzEncoder; use flate2::Compression; -use redb::{Database, ReadableTable, TableDefinition}; +use redb::{Database, ReadableDatabase, ReadableTable, TableDefinition}; use serde::Deserialize; use std::collections::{BTreeMap, HashMap}; use std::fs::File; @@ -37,7 +37,7 @@ const SOURCES: TableDefinition = TableDefinition::new("sources"); const CURIES: TableDefinition = TableDefinition::new("curies"); const META: TableDefinition<&str, &str> = TableDefinition::new("meta"); -const SCHEMA_VERSION: &str = "tablassert.fullmap.v4"; +const SCHEMA_VERSION: &str = "tablassert.fullmap.v5"; const SHARD_COUNT: usize = 16; /// bincode layout MUST match `CurieRow` in `src/fullmap.rs` (field order + types). From eacd88486fbd6b147a7fe103ea97ba29924cb9e1 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Thu, 6 Aug 2026 14:33:33 -0700 Subject: [PATCH 02/10] docs(fullmap): update lock-semantics comments for shared-lock readers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Follow-up comment/docstring updates for the redb 4 shared-lock read path (no behavior change): - fullmap.py: the lock-retry comment now describes the reader-vs-writer (rebuild) window — readers no longer contend with each other under the shared lock; note in _db_cache_key that read-only opens never touch the mtime, so the mtime-keyed lookup caches are stable across lookups. - agent.py: derive_coverage derivations no longer 'serialize on the fullmap lock across processes' (shared locks run concurrently; only a rebuild blocks them); a killed executor no longer strands an EXCLUSIVE lock (at most a shared one, released on process death) — the execution_timeout rationale is reworded accordingly. --- src/tablassert/agent.py | 10 ++++++---- src/tablassert/fullmap.py | 14 ++++++++++---- 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/src/tablassert/agent.py b/src/tablassert/agent.py index 556dfefc..aa373823 100644 --- a/src/tablassert/agent.py +++ b/src/tablassert/agent.py @@ -2107,8 +2107,9 @@ def build_agent( ``execution_timeout`` (seconds, default 600; ``None`` disables) is the local executor's per-step code timeout. The smolagents default is 30s, which KILLS a ``build_and_audit`` on a large table - (e.g. a 37k-row sheet takes ~60s) MID-BUILD -- stranding the fullmap redb lock and failing every - subsequent build -- so it is raised here to let large-table builds complete. + (e.g. a 37k-row sheet takes ~60s) MID-BUILD so it is raised here to let large-table builds complete. + (Readers only hold a SHARED fullmap lock now, so a killed executor no longer strands an exclusive + lock -- but a mid-build kill still wastes the partial work.) """ _require("smolagents") from smolagents import CodeAgent # local import keeps module import lazy # pyright: ignore[reportMissingImports] @@ -2129,7 +2130,7 @@ def build_agent( "final_answer_checks": checks, "executor_type": "local", # Raise the local executor's 30s default so a large-table build_and_audit is not killed mid-build - # (which would also strand the fullmap redb lock and fail every later build in the loop). + # (which would waste the partial build and churn the coverage loop). "executor_kwargs": {"timeout_seconds": execution_timeout}, } if verbosity_level is not None: @@ -2314,7 +2315,8 @@ def make_tools( best (suboptimal for multi-sheet tables). - ``"derive_coverage"``: ``[read_table, pmc_article_context, derive_config, map_coverage]`` — coverage feedback WITHOUT the KGX build, so the agent can pick the best sheet/columns. map_coverage reads the - fullmap, so these derivations serialize on the fullmap lock across processes. + fullmap with a SHARED lock, so these derivations run concurrently across processes (only a concurrent + fullmap REBUILD blocks them). """ def get_fullmap() -> Path: diff --git a/src/tablassert/fullmap.py b/src/tablassert/fullmap.py index 46615d77..b88ab706 100644 --- a/src/tablassert/fullmap.py +++ b/src/tablassert/fullmap.py @@ -22,10 +22,11 @@ # degraded-mode warning is logged once, not on every lookup against a stale extension. _LEGACY_COMPAT_WARNED: bool = False -# redb opens the fullmap with an exclusive file lock. A concurrent or just-finishing holder (e.g. the -# agent's inner code-executor thread completing a build) can momentarily strand that lock; a lookup that -# lands in that window would otherwise raise ``Database already open`` and surface as a false 0.0 coverage. -# Retry briefly on that contention so transient lock overlap does not corrupt a build/coverage result. +# Fullmap readers open the redb DB with a SHARED lock (redb >= 3 ``ReadOnlyDatabase``), so concurrent +# lookups -- even across processes -- never contend with each other. Only a WRITER (a ``build-fullmap`` +# rebuild, exclusive lock) conflicts with readers; a lookup that lands in that brief rebuild window would +# otherwise raise ``Database already open`` and surface as a false 0.0 coverage. Retry briefly on that +# contention so a transient reader-vs-writer overlap does not corrupt a build/coverage result. _LOCK_RETRY_TOKENS: tuple[str, ...] = ("already open", "acquire lock", "cannot acquire") _LOCK_ATTEMPTS: int = 10 _LOCK_DELAY: float = 0.5 @@ -120,6 +121,11 @@ def _db_cache_key(db: Path) -> tuple[Path, float]: Returns: Canonical path and mtime seconds. + + Note: + Since fullmap readers open read-only (shared lock, no file writes), only + a ``build-fullmap`` rebuild touches the mtime -- so this key is stable + across lookups and flips exactly when the DB is rebuilt. """ resolved: Path = db.resolve() try: From b175c55845640199ce78f178e1d2237d5d1cd7f6 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Thu, 6 Aug 2026 14:50:33 -0700 Subject: [PATCH 03/10] fix(fullmap): address code review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CODE_REVIEWER findings on the redb 4 shared-lock change (behavior was already correct; these fix invariant-bearing comments and harden tests): - DB_CACHE doc: stop claiming a shared lock blocks an external rebuild. Rebuilds remove+recreate the files on fresh inodes (unlink needs no lock), so a stale cross-process handle reads the unlinked old file as a consistent snapshot — state that as the real invariant. - lookup_terms doc: opening the primary once is a cache choice, not a lock constraint (read-only opens coexist now). - set_durability doc: it fails only on PersistentSavepointModified; this crate never uses savepoints, so the Result is unreachable (not 'once the transaction has been used'). - Rename build_fullmap_db_writes_schema_v4_sharded_layout → build_fullmap_db_writes_sharded_layout (it writes/asserts v5 now). - Byte-patch tests: assert the raw DatabaseError variant (UpgradeRequired(2) / RepairAborted) BEFORE the mapped rebuild hint, so they pin which variant fired instead of both passing on the shared match arm. --- rust/src/fullmap.rs | 42 ++++++++++++++++++++++++++++++++---------- 1 file changed, 32 insertions(+), 10 deletions(-) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index 180a27d5..4ce6d155 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -73,11 +73,13 @@ type LineChunk = (u8, Vec>); /// any number of processes/threads may hold them concurrently, and they only /// conflict with a writer (`build-fullmap`'s exclusive lock). Keying on the /// path alone is safe: within a process the DB is only rebuilt via -/// `build_fullmap_db`, which evicts the cache explicitly; externally, redb's -/// exclusive writer lock fails against our shared lock (and the rebuild also -/// removes the files before recreating them), so no cached handle can outlive -/// a rebuild on this path. Read-only opens never write the file, so the -/// mtime-keyed Python-side caches stay stable under lookups. +/// `build_fullmap_db`, which evicts the cache explicitly. Cross-process, a +/// rebuild removes the files and recreates them on fresh inodes — an `unlink` +/// does not need the lock — so a stale cached handle in ANOTHER process keeps +/// reading the unlinked old file as a consistent snapshot until that process +/// reopens; it can never block the rebuild or observe a torn file. Read-only +/// opens never write the file, so the mtime-keyed Python-side caches stay +/// stable under lookups. static DB_CACHE: OnceLock>>> = OnceLock::new(); /// Number of shards for concurrent maps (power of two for mask routing). @@ -1909,8 +1911,9 @@ fn write_shard_records( progress: Option<&Arc>, ) -> PyResult { let mut write = database.begin_write().map_err(py_err)?; - // redb ≥ 3 returns `Result` (fails once the transaction has been used); - // this is the first call on a fresh transaction, so it cannot fail here. + // redb ≥ 3 returns `Result`; it only fails if a persistent savepoint was + // modified in the transaction. This crate never uses savepoints (and this + // is a fresh transaction), so the error is unreachable here. write.set_durability(Durability::None).map_err(py_err)?; let mut table = write.open_table(RECORDS).map_err(py_err)?; let mut merge = MergeHeap::new(run_paths).map_err(py_err)?; @@ -2594,8 +2597,9 @@ fn lookup_terms( terms: Vec, threads: Option, ) -> PyResult)>> { - // Open the primary ONCE for dims/CURIES hydration; pair lookups route to the - // shard files (a second open of any one file would fail on redb's flock). + // Open the primary ONCE (the cached shared-lock handle) for dims/CURIES + // hydration; pair lookups route to the shard files. One handle per file is + // a cache choice, not a lock constraint — read-only opens coexist. let database = open_cached(db.clone())?; let prefix_map = load_string_table(&database, PREFIXES)?; let category_map = load_string_table(&database, CATEGORIES)?; @@ -2910,7 +2914,7 @@ mod tests { /// file must exist (even empty) holding a RECORDS table — this is the /// on-disk contract the read path relies on. #[test] - fn build_fullmap_db_writes_schema_v4_sharded_layout() { + fn build_fullmap_db_writes_sharded_layout() { pyo3::Python::initialize(); let dir = tempfile::tempdir().unwrap(); let synonyms = dir.path().join("HGNC.ndjson"); @@ -3821,6 +3825,15 @@ mod tests { 2 // v2 format version -> UpgradeRequired }); + // Pin the RAW variant first (both UpgradeRequired and RepairAborted map + // to the same rebuild hint in open_read_only, so only this asserts the + // patch hit the format-version check and not some unrelated failure). + let raw = match ReadOnlyDatabase::open(&output) { + Ok(_) => panic!("expected ReadOnlyDatabase::open to fail"), + Err(err) => err, + }; + assert!(matches!(raw, redb::DatabaseError::UpgradeRequired(2))); + let err = match open_read_only(&output) { Ok(_) => panic!("expected open_read_only to fail"), Err(err) => err, @@ -3845,6 +3858,15 @@ mod tests { patch_db_byte(&output, 9, |god| god | 2); + // Pin the RAW variant (see the outdated-format test: both variants map + // to the same rebuild hint, so only this asserts the god-byte patch + // actually put the file into the recovery-required state). + let raw = match ReadOnlyDatabase::open(&output) { + Ok(_) => panic!("expected ReadOnlyDatabase::open to fail"), + Err(err) => err, + }; + assert!(matches!(raw, redb::DatabaseError::RepairAborted)); + let err = match open_read_only(&output) { Ok(_) => panic!("expected open_read_only to fail"), Err(err) => err, From 566c57d6ad2e47a8af6f9ef9f56da6e483d74734 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Thu, 6 Aug 2026 15:01:56 -0700 Subject: [PATCH 04/10] chore: retrigger CI (opened event did not schedule) From d464291a2f6f5fdd14c9657845f2889014111c68 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 12:18:01 -0700 Subject: [PATCH 05/10] docs: stop describing the fullmap rebuild as cheap in the changelog The rebuild still reprocesses the input and rewrites every fullmap file; only the BABEL downloads stay cached. CodeRabbit wording fix on PR #68. --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 12e03105..44aba8b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to this project are documented in this file. ## Unreleased ### Breaking Changes -- **Fullmap databases built by older releases must be rebuilt.** The Rust extension upgraded its embedded database engine from redb 2.6 to redb 4.1, and redb ≥ 3 dropped the old v2 file format. Existing `fullmap.redb` (and sibling `fullmap.s*.redb`) files fail to open with `fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'`. Run `tablassert build-fullmap` once after upgrading (BABEL downloads stay cached, so the rebuild is cheap). The on-disk fullmap schema is now `tablassert.fullmap.v5` (the table layout is unchanged; the bump makes the redb-4 rebuild explicit and lets an older extension reject new files loudly). +- **Fullmap databases built by older releases must be rebuilt.** The Rust extension upgraded its embedded database engine from redb 2.6 to redb 4.1, and redb ≥ 3 dropped the old v2 file format. Existing `fullmap.redb` (and sibling `fullmap.s*.redb`) files fail to open with `fullmap DB is outdated or needs repair; rebuild with 'tablassert build-fullmap'`. Run `tablassert build-fullmap` once after upgrading. BABEL downloads stay cached, but the command rebuilds the fullmap files. The on-disk fullmap schema is now `tablassert.fullmap.v5` (the table layout is unchanged; the bump makes the redb-4 rebuild explicit and lets an older extension reject new files loudly). ### Changed - **Fullmap reads no longer serialize across processes.** The lookup path (`lookup_fullmap_terms` and the `hydrate_*` helpers) now opens the fullmap redb files READ-ONLY with a SHARED file lock (redb ≥ 3 `ReadOnlyDatabase`) instead of an exclusive lock: concurrent readers — the agent supervisor, its code-executor subprocesses, and parallel `agent run` processes — no longer contend on the fullmap lock ("Database already open"); only a running `build-fullmap` rebuild can briefly block readers. Read-only opens also never touch the file mtime, making the mtime-keyed Python lookup caches fully stable. The redb 4.1 upgrade additionally speeds up multi-threaded shard reads (~15% on upstream benchmarks) and the fullmap build's redb write phase (~1.5x on upstream write benchmarks). From 2df04a915522f3193f73fec2f2be12ae4bebf19a Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 12:39:10 -0700 Subject: [PATCH 06/10] fix(fullmap): pin cached read handles to (dev,ino) file generations MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rebuild replaces the fullmap files on fresh inodes (unlink + recreate), but DB_CACHE keyed handles by canonical path only — so a cache hit could serve an old-generation handle next to new-generation ones (in-process: cache_database swapped only the primary entry; cross-process: stale handles read the unlinked files forever). Record (st_dev, st_ino) via MetadataExt for every cached handle (cache_database included) and validate it on every cache hit in open_cached / open_cached_shard: a mismatch evicts the entry and reopens the replacement (the primary re-runs validate_schema). A hit whose path is absent (mid-rebuild window) keeps serving the old snapshot, preserving the cross-process reader behavior; on a miss the generation is stat'ed before the open so a racing rebuild costs at most one extra reopen. open_cached_shards now pins one primary-plus-shards generation: it captures the primary's (dev,ino) before opening shards and re-stats after the last one, retrying the whole bundle up to 5 times before raising PyRuntimeError, so a lookup never mixes files from two builds. --- rust/src/fullmap.rs | 169 +++++++++++++++++++++++++++++++++++--------- 1 file changed, 137 insertions(+), 32 deletions(-) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index 4ce6d155..767681ea 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -14,6 +14,7 @@ use std::collections::{BinaryHeap, HashMap, HashSet}; use std::fs::File; use std::hash::{BuildHasher, BuildHasherDefault}; use std::io::{BufRead, BufReader, BufWriter, Read, Write}; +use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; use std::sync::atomic::{AtomicU32, AtomicU64, AtomicUsize, Ordering}; use std::sync::mpsc::{sync_channel, Receiver, SyncSender}; @@ -67,20 +68,43 @@ type FxHashMap = HashMap>; /// ever allocated on the read path. type LineChunk = (u8, Vec>); -/// Read-path database cache keyed by canonical path only. +/// Read-path database cache keyed by canonical path, generation-checked. /// /// Handles are `ReadOnlyDatabase` (redb ≥ 3), which take a SHARED file lock: /// any number of processes/threads may hold them concurrently, and they only -/// conflict with a writer (`build-fullmap`'s exclusive lock). Keying on the -/// path alone is safe: within a process the DB is only rebuilt via -/// `build_fullmap_db`, which evicts the cache explicitly. Cross-process, a -/// rebuild removes the files and recreates them on fresh inodes — an `unlink` -/// does not need the lock — so a stale cached handle in ANOTHER process keeps -/// reading the unlinked old file as a consistent snapshot until that process -/// reopens; it can never block the rebuild or observe a torn file. Read-only -/// opens never write the file, so the mtime-keyed Python-side caches stay -/// stable under lookups. -static DB_CACHE: OnceLock>>> = OnceLock::new(); +/// conflict with a writer (`build-fullmap`'s exclusive lock). Each entry also +/// records the `(st_dev, st_ino)` generation of the file it was opened from: +/// a rebuild replaces the files via unlink + recreate on FRESH inodes (an +/// `unlink` does not need the lock), so the path alone cannot distinguish the +/// old data from the new. Every cache hit therefore re-stats the canonical +/// path and evicts the entry when `(dev, ino)` no longer matches, so a lookup +/// always reads the generation currently living at the path — in-process after +/// `build_fullmap_db`, or cross-process after another process rebuilt. A +/// stale evicted handle simply keeps reading the unlinked old file as a +/// consistent snapshot; it can never block the rebuild or observe a torn file. +/// Each lookup pins ONE primary-plus-shards generation (`open_cached_shards` +/// re-stats the primary around the shard fan-out and retries the whole bundle +/// on change), and the next lookup after a rebuild opens the replacement. +/// Read-only opens never write the file, so the mtime-keyed Python-side caches +/// stay stable under lookups. +static DB_CACHE: OnceLock>> = OnceLock::new(); + +/// One cached read-only handle plus the `(st_dev, st_ino)` file generation it +/// was opened from, so every cache hit can detect an in-place rebuild. +struct CachedDatabase { + database: Arc, + generation: FileGeneration, +} + +/// On-disk identity of the file at a path: its `(st_dev, st_ino)` pair. A +/// rebuild replaces each file on a fresh inode, so this pair distinguishes the +/// generations living at one path over time. +type FileGeneration = (u64, u64); + +fn generation_of(path: &Path) -> PyResult { + let meta = std::fs::metadata(path).map_err(py_err)?; + Ok((meta.dev(), meta.ino())) +} /// Number of shards for concurrent maps (power of two for mask routing). const SHARD_COUNT: usize = 64; @@ -2084,8 +2108,15 @@ fn evict_cached_path(path: &Path) -> PyResult<()> { fn cache_database(path: &Path, database: Arc) -> PyResult<()> { let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); + let generation = generation_of(&canonical)?; let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); - cache.write().map_err(py_err)?.insert(canonical, database); + cache.write().map_err(py_err)?.insert( + canonical, + CachedDatabase { + database, + generation, + }, + ); Ok(()) } @@ -2330,17 +2361,10 @@ fn validate_schema(database: &ReadOnlyDatabase) -> PyResult<()> { } } +/// Open (and cache) the primary DB, schema-validating it on (re)open. fn open_cached(db: PathBuf) -> PyResult> { let canonical = std::fs::canonicalize(&db).unwrap_or(db); - let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); - if let Some(database) = cache.read().map_err(py_err)?.get(&canonical) { - return Ok(Arc::clone(database)); - } - let database = Arc::new(open_read_only(&canonical)?); - validate_schema(&database)?; - let cached = Arc::clone(&database); - cache.write().map_err(py_err)?.insert(canonical, database); - Ok(cached) + open_cached_path(canonical, true) } /// Open (and cache) one RECORDS shard by index, deriving its path from the @@ -2349,13 +2373,68 @@ fn open_cached(db: PathBuf) -> PyResult> { fn open_cached_shard(primary: &Path, index: usize) -> PyResult> { let path = shard_path(primary, index); let canonical = std::fs::canonicalize(&path).unwrap_or(path); + open_cached_path(canonical, false) +} + +/// Shared cache lookup behind `open_cached` / `open_cached_shard`: serve the +/// cached handle only while the canonical path still points at the recorded +/// `(dev, ino)` generation; a mismatch evicts the stale entry and reopens the +/// replacement (re-running `validate_schema` when `validate`). A hit whose +/// path is absent (mid-rebuild window) keeps serving the old snapshot, as +/// readers did before generation pinning. On a miss the generation is stat'ed +/// BEFORE the open, so a rebuild racing the open costs at most one extra +/// reopen on the next hit — never a stale serve. +fn open_cached_path(canonical: PathBuf, validate: bool) -> PyResult> { let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); - if let Some(database) = cache.read().map_err(py_err)?.get(&canonical) { - return Ok(Arc::clone(database)); + // Copy the hit out of the read lock before any write-lock upgrade attempt. + let hit = { + let map = cache.read().map_err(py_err)?; + map.get(&canonical) + .map(|entry| (Arc::clone(&entry.database), entry.generation)) + }; + if let Some((database, generation)) = hit { + match generation_of(&canonical) { + // Same inode generation: the handle is still the file at the path. + Ok(current) if current == generation => return Ok(database), + // Path gone (rebuild in flight or deleted): keep serving the old + // snapshot; the next lookup after a replacement appears follows it. + Err(_) => return Ok(database), + // A newer generation lives at the path: evict (only the entry we + // observed; another thread may already have refreshed it) and + // reopen below. + Ok(_) => { + let mut map = cache.write().map_err(py_err)?; + if map + .get(&canonical) + .is_some_and(|entry| entry.generation == generation) + { + map.remove(&canonical); + } + } + } } + // Prefer the generation stat'ed BEFORE the open: a rebuild racing the open + // can then cost at most one extra reopen later, never a stale serve. If + // the stat failed (e.g. missing file), `open_read_only` below raises the + // canonical redb error for it; the re-stat then records the generation of + // the file actually opened. + let pre = generation_of(&canonical); let database = Arc::new(open_read_only(&canonical)?); + if validate { + validate_schema(&database)?; + } + let generation = match pre { + Ok(generation) => generation, + Err(_) => generation_of(&canonical)?, + }; let cached = Arc::clone(&database); - cache.write().map_err(py_err)?.insert(canonical, database); + cache.write().map_err(py_err)?.insert( + canonical, + CachedDatabase { + database, + generation, + }, + ); Ok(cached) } @@ -2381,15 +2460,41 @@ fn shard_count_of(database: &ReadOnlyDatabase) -> PyResult { Ok(round_down_pow2(count).clamp(1, SHARD_COUNT_SHARDS)) } -/// Open (and cache) all RECORDS shard handles for a primary DB path. The shard -/// count is read from the primary's META so the read path opens exactly the -/// shards the build wrote. +/// Attempts `open_cached_shards` makes to open one consistent generation +/// before concluding the DB is being rebuilt in a tight loop. +const BUNDLE_OPEN_ATTEMPTS: usize = 5; + +/// Open (and cache) all RECORDS shard handles for a primary DB path, pinned to +/// ONE file generation: the primary's `(dev, ino)` is captured before opening +/// anything and re-checked after the last shard; if a rebuild replaced the +/// files mid-bundle, the whole bundle is retried, so a lookup never mixes an +/// old-generation file with a new-generation one. The shard count is read +/// from the primary's META so the read path opens exactly the shards the +/// build wrote. fn open_cached_shards(primary: &Path) -> PyResult>> { - let database = open_cached(primary.to_path_buf())?; - let shard_count = shard_count_of(&database)?; - (0..shard_count) - .map(|index| open_cached_shard(primary, index)) - .collect() + let canonical = std::fs::canonicalize(primary).unwrap_or_else(|_| primary.to_path_buf()); + for _ in 0..BUNDLE_OPEN_ATTEMPTS { + let pinned = generation_of(&canonical).ok(); + let database = open_cached(primary.to_path_buf())?; + let shard_count = shard_count_of(&database)?; + let shards: Vec> = (0..shard_count) + .map(|index| open_cached_shard(primary, index)) + .collect::>()?; + // Re-stat the primary. Matching generations mean every handle above + // belongs to one build. Absent on BOTH sides means the rebuild window + // straddled the whole bundle, so every handle came from the cache's + // old generation — also consistent. Anything else (a generation flip + // or an appearance mid-bundle) may have mixed two builds: retry. + match (pinned, generation_of(&canonical).ok()) { + (Some(before), Some(after)) if before == after => return Ok(shards), + (None, None) => return Ok(shards), + _ => {} + } + } + let path = canonical.display(); + Err(PyRuntimeError::new_err(format!( + "fullmap at {path} kept changing generation; could not pin one primary-plus-shards bundle after {BUNDLE_OPEN_ATTEMPTS} attempts" + ))) } fn lookup_pair_chunk(shards: &[Arc], terms: &[String]) -> PyResult { From f0d7fdcab3a09b6c818d5d41ef32a8f715d59338 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 12:39:11 -0700 Subject: [PATCH 07/10] test(fullmap): cover generation-pinned cache and rebuild-at-same-path lookups Deterministic rename-over swaps (no sleeps/races): replacing a cached primary or shard file makes the next open see the new content; a full lookup after a rebuild-at-same-path returns only the new generation (old term gone, new term hydrated against the new primary); an absent path serves the cached old snapshot until the replacement appears. --- rust/src/fullmap.rs | 180 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 180 insertions(+) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index 767681ea..0ece0f2e 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -3884,6 +3884,186 @@ mod tests { } } + /// Write a minimal primary-shaped DB at `path` whose META carries `marker` + /// alongside the current schema tag, so tests can tell generations apart. + /// The writer is dropped before returning, so the file is unlocked. + fn write_marker_db(path: &Path, marker: &str) { + let database = Database::create(path).unwrap(); + let write = database.begin_write().unwrap(); + { + let mut meta = write.open_table(META).unwrap(); + meta.insert("schema", SCHEMA_VERSION).unwrap(); + meta.insert("marker", marker).unwrap(); + } + write.commit().unwrap(); + drop(database); + } + + fn marker_of(database: &ReadOnlyDatabase) -> String { + let read = database.begin_read().unwrap(); + let meta = read.open_table(META).unwrap(); + meta.get("marker").unwrap().unwrap().value().to_string() + } + + /// Write a shard-shaped DB (RECORDS only) holding one record `1 -> payload`. + fn write_shard_db(path: &Path, payload: &[u8]) { + let database = Database::create(path).unwrap(); + let write = database.begin_write().unwrap(); + { + let mut records = write.open_table(RECORDS).unwrap(); + records.insert(1u64, payload).unwrap(); + } + write.commit().unwrap(); + drop(database); + } + + fn shard_payload(database: &ReadOnlyDatabase) -> Vec { + let read = database.begin_read().unwrap(); + let table = read.open_table(RECORDS).unwrap(); + table.get(1u64).unwrap().unwrap().value().to_vec() + } + + /// Rename the full sharded build at `src` (primary + every shard) over the + /// same-named files at `dst` — the multi-file generation swap a rebuild + /// performs, atomic per file and deterministic (no sleeps/races). + fn swap_build_over(src: &Path, dst: &Path) { + std::fs::rename(src, dst).unwrap(); + for index in 0..SHARD_COUNT_SHARDS { + std::fs::rename(shard_path(src, index), shard_path(dst, index)).unwrap(); + } + } + + /// While the path is absent (mid-rebuild window), a cached hit keeps + /// serving the old snapshot instead of erroring — the cross-process reader + /// guarantee from before generation pinning. The lookup after the + /// replacement appears follows the new generation (covered above). + #[test] + fn cached_handle_serves_old_snapshot_while_path_absent() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + write_marker_db(&output, "gen1"); + let handle = open_cached(output.clone()).unwrap(); + assert_eq!(marker_of(&handle), "gen1"); + + std::fs::remove_file(&output).unwrap(); + let again = open_cached(output).unwrap(); + assert_eq!( + marker_of(&again), + "gen1", + "an absent path must serve the cached old snapshot, not error" + ); + } + + /// A rebuild replaces the file at a path on a FRESH inode (unlink + + /// recreate); the cache must follow the path: after renaming a new primary + /// over a cached one, the next `open_cached` returns the replacement's + /// content even though the stale handle is still alive. + #[test] + fn cached_primary_follows_replacement_generation() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + write_marker_db(&output, "gen1"); + let stale = open_cached(output.clone()).unwrap(); + assert_eq!(marker_of(&stale), "gen1"); + + // Stage the replacement at a sibling path, then rename over — one + // atomic, deterministic generation swap (fresh inode). + let staging = dir.path().join("fullmap.next.redb"); + write_marker_db(&staging, "gen2"); + std::fs::rename(&staging, &output).unwrap(); + + let fresh = open_cached(output).unwrap(); + assert_eq!( + marker_of(&fresh), + "gen2", + "the next open must see the replacement generation" + ); + assert_eq!( + marker_of(&stale), + "gen1", + "the evicted handle still reads its own old snapshot" + ); + } + + /// Same generation-follow guarantee for shard handles: replacing a shard + /// file under a cached handle makes the next `open_cached_shard` read the + /// replacement's records. + #[test] + fn cached_shard_follows_replacement_generation() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let primary = dir.path().join("fullmap.redb"); + let shard = shard_path(&primary, 3); + write_shard_db(&shard, b"gen1"); + let stale = open_cached_shard(&primary, 3).unwrap(); + assert_eq!(shard_payload(&stale), b"gen1"); + + let staging = dir.path().join("fullmap.s3.next.redb"); + write_shard_db(&staging, b"gen2"); + std::fs::rename(&staging, &shard).unwrap(); + + let fresh = open_cached_shard(&primary, 3).unwrap(); + assert_eq!( + shard_payload(&fresh), + b"gen2", + "the next shard open must see the replacement generation" + ); + } + + /// Rebuild-at-same-path: after replacing the whole cached bundle, the next + /// lookup reads ONLY the new generation — the old term is gone, the new + /// term resolves, and its hydrated CURIE row comes from the NEW primary. + /// A mixed-generation lookup would resurrect the old term or hydrate the + /// new records against the old primary's CURIES and surface the wrong + /// curie/preferred_name. + #[test] + fn lookup_after_rebuild_reads_one_consistent_generation() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let output = dir.path().join("fullmap.redb"); + + // Generation 1: indexes BRCA1 under HGNC:1100; warm the cache with a lookup. + let synonyms1 = dir.path().join("gen1.ndjson"); + let mut file = File::create(&synonyms1).unwrap(); + writeln!( + file, + r#"{{"curie":"HGNC:1100","preferred_name":"BRCA1","names":["BRCA1"],"types":["Gene"],"taxa":["NCBITaxon:9606"]}}"# + ) + .unwrap(); + build_test(output.clone(), Vec::new(), vec![synonyms1], 1, 4_000_000).unwrap(); + let rows = lookup_terms(output.clone(), vec!["brca1".to_string()], Some(1)).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].1[0].curie, "HGNC:1100"); + + // Generation 2 (built elsewhere, then renamed over): indexes TP53. + let dir2 = tempfile::tempdir().unwrap(); + let built = dir2.path().join("fullmap.redb"); + let synonyms2 = dir2.path().join("gen2.ndjson"); + let mut file = File::create(&synonyms2).unwrap(); + writeln!( + file, + r#"{{"curie":"NCBIGene:7157","preferred_name":"TP53","names":["TP53"],"types":["Gene"],"taxa":["NCBITaxon:9606"]}}"# + ) + .unwrap(); + build_test(built.clone(), Vec::new(), vec![synonyms2], 1, 4_000_000).unwrap(); + swap_build_over(&built, &output); + + // Old-generation term must be gone (stale shards would resurrect it). + let rows = lookup_terms(output.clone(), vec!["brca1".to_string()], Some(1)).unwrap(); + assert!( + rows.is_empty(), + "stale shard generation resurrected an old term: {rows:?}" + ); + // New term resolves against the NEW primary's dims/CURIES (a stale + // primary would hydrate the wrong curie/preferred_name). + let rows = lookup_terms(output, vec!["tp53".to_string()], Some(1)).unwrap(); + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].1[0].curie, "NCBIGene:7157"); + assert_eq!(rows[0].1[0].preferred_name, "TP53"); + } + /// A live writer holds the exclusive lock, so a read-only open fails with /// `DatabaseAlreadyOpen` — and succeeds once the writer drops. This is the /// reader-vs-rebuild window that `_call_with_lock_retry` (Python side) From 4159ec7cd45cee404a087fd795887b2300ddb381 Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 12:39:11 -0700 Subject: [PATCH 08/10] docs: document get_fullmap and qualify concurrent-reader generation boundary MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add the missing docstring to the nested get_fullmap in make_tools (the only undocumented changed function behind CodeRabbit's docstring-coverage warning) and qualify the concurrent-reader claims with the generation boundary — lookups pin one primary-plus-shards generation; readers follow a rebuild on the next lookup — in the make_tools docstring, the fullmap lock-comment block, the _db_cache_key Note, and docs/fullmap.md. --- docs/fullmap.md | 4 +++- src/tablassert/agent.py | 4 +++- src/tablassert/fullmap.py | 6 +++++- 3 files changed, 11 insertions(+), 3 deletions(-) diff --git a/docs/fullmap.md b/docs/fullmap.md index 8b0dc472..d2e4185f 100644 --- a/docs/fullmap.md +++ b/docs/fullmap.md @@ -124,7 +124,9 @@ via `tablassert build-fullmap`. Readers open every fullmap file READ-ONLY with a SHARED file lock (redb ≥ 3 `ReadOnlyDatabase`), so any number of processes can run lookups against the same fullmap concurrently; only a `build-fullmap` rebuild -(an exclusive-lock writer) briefly blocks readers. +(an exclusive-lock writer) briefly blocks readers. Each lookup pins one primary-plus-shards file +generation — cached handles are validated against the file's `(dev, ino)` on every use — so a reader +follows a rebuild on the next lookup. ## Usage in Graph Config diff --git a/src/tablassert/agent.py b/src/tablassert/agent.py index aa373823..7309eef0 100644 --- a/src/tablassert/agent.py +++ b/src/tablassert/agent.py @@ -2316,10 +2316,12 @@ def make_tools( - ``"derive_coverage"``: ``[read_table, pmc_article_context, derive_config, map_coverage]`` — coverage feedback WITHOUT the KGX build, so the agent can pick the best sheet/columns. map_coverage reads the fullmap with a SHARED lock, so these derivations run concurrently across processes (only a concurrent - fullmap REBUILD blocks them). + fullmap REBUILD blocks them). A lookup pins one primary-plus-shards generation; readers follow a + rebuild on the next lookup. """ def get_fullmap() -> Path: + """Return the bound fullmap redb path the tools read.""" return fullmap if derive_mode == "derive_only": diff --git a/src/tablassert/fullmap.py b/src/tablassert/fullmap.py index b88ab706..082cfe0b 100644 --- a/src/tablassert/fullmap.py +++ b/src/tablassert/fullmap.py @@ -27,6 +27,8 @@ # rebuild, exclusive lock) conflicts with readers; a lookup that lands in that brief rebuild window would # otherwise raise ``Database already open`` and surface as a false 0.0 coverage. Retry briefly on that # contention so a transient reader-vs-writer overlap does not corrupt a build/coverage result. +# Each lookup pins one primary-plus-shards file generation (cached handles are validated against the +# file's inode on every use); a reader follows a rebuild on the NEXT lookup. _LOCK_RETRY_TOKENS: tuple[str, ...] = ("already open", "acquire lock", "cannot acquire") _LOCK_ATTEMPTS: int = 10 _LOCK_DELAY: float = 0.5 @@ -125,7 +127,9 @@ def _db_cache_key(db: Path) -> tuple[Path, float]: Note: Since fullmap readers open read-only (shared lock, no file writes), only a ``build-fullmap`` rebuild touches the mtime -- so this key is stable - across lookups and flips exactly when the DB is rebuilt. + across lookups and flips exactly when the DB is rebuilt. This mirrors + the Rust-side generation boundary: a lookup pins one primary-plus-shards + generation, and readers follow a rebuild on the next lookup. """ resolved: Path = db.resolve() try: From f76fd3737517c5c0f6149cbd4678ca51a0102c5d Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 13:23:10 -0700 Subject: [PATCH 09/10] fix(fullmap): never serve a mixed-generation bundle in the mid-build window MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rebuild commits the new primary BEFORE the shard files are created, so mid-build the primary path holds the new inode while every shard path is absent. A warm-cache reader could then mix the new primary with old shard snapshots: each shard hit took the absent-path arm while the stable new primary passed the post-stat generation check. - open_cached_path now reports HOW it served: Served::Current vs Served::StaleAbsent, plumbed through open_cached_shard to open_cached_shards; only io::ErrorKind::NotFound serves the stale snapshot — any other metadata error propagates. - open_cached_shards accepts a bundle only when generationally consistent: ALL Current with a stable primary generation, or ALL StaleAbsent (the documented old-snapshot arm); MIXED retries the whole bundle, then raises the "changing generation" exhaustion error. - "changing generation" joins Python's _LOCK_RETRY_TOKENS so _call_with_lock_retry retries the exhaustion; the primary-appears-first invariant is cross-referenced at the up-front unlink and drop(database). - Tests: mixed window exhausts then follows the new generation; all-absent serves the old snapshot; exhaustion message carries the retry token. --- rust/src/fullmap.rs | 260 +++++++++++++++++++++++++++++++------- src/tablassert/fullmap.py | 8 +- tests/test_fullmap.py | 10 +- 3 files changed, 225 insertions(+), 53 deletions(-) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index 0ece0f2e..81327c14 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -74,17 +74,22 @@ type LineChunk = (u8, Vec>); /// any number of processes/threads may hold them concurrently, and they only /// conflict with a writer (`build-fullmap`'s exclusive lock). Each entry also /// records the `(st_dev, st_ino)` generation of the file it was opened from: -/// a rebuild replaces the files via unlink + recreate on FRESH inodes (an -/// `unlink` does not need the lock), so the path alone cannot distinguish the -/// old data from the new. Every cache hit therefore re-stats the canonical -/// path and evicts the entry when `(dev, ino)` no longer matches, so a lookup -/// always reads the generation currently living at the path — in-process after -/// `build_fullmap_db`, or cross-process after another process rebuilt. A -/// stale evicted handle simply keeps reading the unlinked old file as a -/// consistent snapshot; it can never block the rebuild or observe a torn file. -/// Each lookup pins ONE primary-plus-shards generation (`open_cached_shards` -/// re-stats the primary around the shard fan-out and retries the whole bundle -/// on change), and the next lookup after a rebuild opens the replacement. +/// a rebuild replaces the files via unlink + recreate — in practice on FRESH +/// inodes (an `unlink` does not need the lock) — so the path alone cannot +/// distinguish the old data from the new. Every cache hit therefore re-stats +/// the canonical path and evicts the entry when `(dev, ino)` no longer +/// matches, so a lookup always reads the generation currently living at the +/// path — in-process after `build_fullmap_db`, or cross-process after another +/// process rebuilt. A stale evicted handle simply keeps reading the unlinked +/// old file as a consistent snapshot; it can never block the rebuild or +/// observe a torn file. Each lookup pins ONE primary-plus-shards generation: +/// `open_cached_shards` records how every bundle member was served (`Served`) +/// and accepts the bundle only when it is generationally consistent — ALL +/// current with a stable primary generation, or ALL stale because every path +/// is absent (the documented mid-rebuild snapshot); a MIXED bundle (e.g. the +/// new primary committed while the shard files are still absent) is retried +/// whole and finally raises an exhaustion error the Python side retries like +/// lock contention. The next lookup after a rebuild opens the replacement. /// Read-only opens never write the file, so the mtime-keyed Python-side caches /// stay stable under lookups. static DB_CACHE: OnceLock>> = OnceLock::new(); @@ -101,8 +106,8 @@ struct CachedDatabase { /// generations living at one path over time. type FileGeneration = (u64, u64); -fn generation_of(path: &Path) -> PyResult { - let meta = std::fs::metadata(path).map_err(py_err)?; +fn generation_of(path: &Path) -> std::io::Result { + let meta = std::fs::metadata(path)?; Ok((meta.dev(), meta.ino())) } @@ -1816,6 +1821,10 @@ fn write_final_database( // is never touched again (Phase 4 writes only the separate shard DBs below). // Drop it here to release its redb cache and file lock during the long // parallel RECORDS write — a free memory win while the shards are built. + // NOTE: the primary is committed BEFORE the shard files below exist, so on + // rebuild the primary appears first — primary-absent implies shards-absent, + // the invariant the all-stale arm of `open_cached_shards` relies on (see + // the up-front unlink in `build_fullmap_db`). drop(database); // Phase 4: one INDEPENDENT k-way merge per shard, run in parallel — one @@ -2108,7 +2117,7 @@ fn evict_cached_path(path: &Path) -> PyResult<()> { fn cache_database(path: &Path, database: Arc) -> PyResult<()> { let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); - let generation = generation_of(&canonical)?; + let generation = generation_of(&canonical).map_err(py_err)?; let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); cache.write().map_err(py_err)?.insert( canonical, @@ -2255,6 +2264,10 @@ pub fn build_fullmap_db( std::fs::create_dir_all(parent).map_err(py_err)?; } evict_cached_path(&output)?; + // Everything is unlinked BEFORE the new primary is written, and the + // primary is written BEFORE the shards (`write_final_database`): primary + // appears first on rebuild, so primary-absent implies shards-absent — the + // invariant the all-stale arm of `open_cached_shards` relies on. if output.exists() { std::fs::remove_file(&output).map_err(py_err)?; } @@ -2361,16 +2374,29 @@ fn validate_schema(database: &ReadOnlyDatabase) -> PyResult<()> { } } +/// How `open_cached_path` served a handle: as the generation currently living +/// at the path, or as the old snapshot cached for a path that is absent. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +enum Served { + /// The path still points at the handle's `(dev, ino)` generation. + Current, + /// The path is absent (`NotFound`): the handle is the old snapshot the + /// cache keeps alive across the mid-rebuild window. + StaleAbsent, +} + /// Open (and cache) the primary DB, schema-validating it on (re)open. fn open_cached(db: PathBuf) -> PyResult> { let canonical = std::fs::canonicalize(&db).unwrap_or(db); - open_cached_path(canonical, true) + Ok(open_cached_path(canonical, true)?.0) } /// Open (and cache) one RECORDS shard by index, deriving its path from the /// primary DB path. Shards hold only RECORDS (no META), so they are not /// schema-validated here — the primary's `validate_schema` gates the layout. -fn open_cached_shard(primary: &Path, index: usize) -> PyResult> { +/// Returns the handle plus HOW it was served (`Served`) so +/// `open_cached_shards` can pin one bundle generation. +fn open_cached_shard(primary: &Path, index: usize) -> PyResult<(Arc, Served)> { let path = shard_path(primary, index); let canonical = std::fs::canonicalize(&path).unwrap_or(path); open_cached_path(canonical, false) @@ -2378,13 +2404,19 @@ fn open_cached_shard(primary: &Path, index: usize) -> PyResult PyResult> { +/// `(dev, ino)` generation (`Served::Current`); a mismatch evicts the stale +/// entry and reopens the replacement (re-running `validate_schema` when +/// `validate`). A hit whose path is ABSENT with `io::ErrorKind::NotFound` +/// (the mid-rebuild window) keeps serving the old snapshot as +/// `Served::StaleAbsent`, as readers did before generation pinning; any other +/// metadata error propagates instead of silently pinning readers to stale +/// data. On a miss the generation is stat'ed BEFORE the open, so a rebuild +/// racing the open costs at most one extra reopen on the next hit — never a +/// stale serve. +fn open_cached_path( + canonical: PathBuf, + validate: bool, +) -> PyResult<(Arc, Served)> { let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); // Copy the hit out of the read lock before any write-lock upgrade attempt. let hit = { @@ -2395,10 +2427,15 @@ fn open_cached_path(canonical: PathBuf, validate: bool) -> PyResult return Ok(database), + Ok(current) if current == generation => return Ok((database, Served::Current)), // Path gone (rebuild in flight or deleted): keep serving the old // snapshot; the next lookup after a replacement appears follows it. - Err(_) => return Ok(database), + Err(err) if err.kind() == std::io::ErrorKind::NotFound => { + return Ok((database, Served::StaleAbsent)); + } + // Any other metadata failure (permissions, I/O): propagate it — + // serving stale data here would silently pin readers to old state. + Err(err) => return Err(py_err(err)), // A newer generation lives at the path: evict (only the entry we // observed; another thread may already have refreshed it) and // reopen below. @@ -2425,7 +2462,7 @@ fn open_cached_path(canonical: PathBuf, validate: bool) -> PyResult generation, - Err(_) => generation_of(&canonical)?, + Err(_) => generation_of(&canonical).map_err(py_err)?, }; let cached = Arc::clone(&database); cache.write().map_err(py_err)?.insert( @@ -2435,7 +2472,7 @@ fn open_cached_path(canonical: PathBuf, validate: bool) -> PyResult PyResult { const BUNDLE_OPEN_ATTEMPTS: usize = 5; /// Open (and cache) all RECORDS shard handles for a primary DB path, pinned to -/// ONE file generation: the primary's `(dev, ino)` is captured before opening -/// anything and re-checked after the last shard; if a rebuild replaced the -/// files mid-bundle, the whole bundle is retried, so a lookup never mixes an -/// old-generation file with a new-generation one. The shard count is read -/// from the primary's META so the read path opens exactly the shards the -/// build wrote. +/// ONE file generation. The primary's `(dev, ino)` is captured before opening +/// anything and re-stat'ed after the last shard, and every bundle member +/// reports HOW it was served (`Served`). The bundle is accepted only when it +/// is generationally consistent: +/// +/// - ALL `Current` with the primary generation unchanged across the fan-out: +/// every handle belongs to one build — return it. +/// - ALL `StaleAbsent`: the rebuild window straddled the whole bundle. The +/// rebuild unlinks every file up front and recreates the primary BEFORE the +/// shards (primary appears first on rebuild, so primary-absent implies +/// shards-absent), so every handle is the same old snapshot — return the old +/// bundle (the documented snapshot semantics; the next lookup after the +/// replacement appears follows it). +/// - MIXED (any `StaleAbsent` alongside any `Current`): the bundle could pair +/// the NEW primary with OLD shards — exactly the torn state generation +/// pinning exists to prevent — so retry the WHOLE bundle, and after +/// `BUNDLE_OPEN_ATTEMPTS` raise the exhaustion error below (its +/// "changing generation" text is one of the Python-side +/// `_LOCK_RETRY_TOKENS`). +/// +/// The shard count is read from the primary's META so the read path opens +/// exactly the shards the build wrote. fn open_cached_shards(primary: &Path) -> PyResult>> { let canonical = std::fs::canonicalize(primary).unwrap_or_else(|_| primary.to_path_buf()); for _ in 0..BUNDLE_OPEN_ATTEMPTS { let pinned = generation_of(&canonical).ok(); - let database = open_cached(primary.to_path_buf())?; + let (database, primary_served) = open_cached_path(canonical.clone(), true)?; let shard_count = shard_count_of(&database)?; - let shards: Vec> = (0..shard_count) - .map(|index| open_cached_shard(primary, index)) - .collect::>()?; - // Re-stat the primary. Matching generations mean every handle above - // belongs to one build. Absent on BOTH sides means the rebuild window - // straddled the whole bundle, so every handle came from the cache's - // old generation — also consistent. Anything else (a generation flip - // or an appearance mid-bundle) may have mixed two builds: retry. - match (pinned, generation_of(&canonical).ok()) { - (Some(before), Some(after)) if before == after => return Ok(shards), - (None, None) => return Ok(shards), - _ => {} + let mut all_current = primary_served == Served::Current; + let mut all_stale_absent = primary_served == Served::StaleAbsent; + let mut shards: Vec> = Vec::with_capacity(shard_count); + for index in 0..shard_count { + let (database, served) = open_cached_shard(primary, index)?; + all_current &= served == Served::Current; + all_stale_absent &= served == Served::StaleAbsent; + shards.push(database); + } + // ALL stale: every path is absent, so every handle is the same old + // snapshot — a consistent bundle (see the invariant above). + if all_stale_absent { + return Ok(shards); } + // ALL current: re-stat the primary; an unchanged generation means + // every handle above belongs to one build. + if all_current + && matches!( + (pinned, generation_of(&canonical).ok()), + (Some(before), Some(after)) if before == after + ) + { + return Ok(shards); + } + // MIXED serve kinds, or the primary generation moved during the + // fan-out: the handles may span two builds — retry the whole bundle. } let path = canonical.display(); Err(PyRuntimeError::new_err(format!( @@ -3079,7 +3145,7 @@ mod tests { let shard_count = shard_count_of(&database).unwrap(); (0..shard_count) .map(|index| { - let db = open_cached_shard(primary, index).unwrap(); + let db = open_cached_shard(primary, index).unwrap().0; let read = db.begin_read().unwrap(); let table = read.open_table(RECORDS).unwrap(); table.iter().unwrap().count() @@ -3933,6 +3999,102 @@ mod tests { } } + /// Write a minimal consistent bundle at `primary`: a marker primary + /// advertising `shard_count` shards in META plus that many shard DBs, all + /// carrying `marker`, so tests can warm `open_cached_shards` without a + /// full build. + fn write_marker_bundle(primary: &Path, marker: &str, shard_count: usize) { + let database = Database::create(primary).unwrap(); + let write = database.begin_write().unwrap(); + { + let mut meta = write.open_table(META).unwrap(); + meta.insert("schema", SCHEMA_VERSION).unwrap(); + meta.insert("marker", marker).unwrap(); + let count = shard_count.to_string(); + meta.insert("shards", count.as_str()).unwrap(); + } + write.commit().unwrap(); + drop(database); + for index in 0..shard_count { + write_shard_db(&shard_path(primary, index), marker.as_bytes()); + } + } + + /// Mid-rebuild window: the build commits the new primary BEFORE the shard + /// files exist, so a warm-cache reader sees a NEW primary with ABSENT + /// shards. The bundle must never pair them (new primary + old shards): + /// the MIXED serve kinds retry until exhaustion — raising with the stable + /// "changing generation" token the Python-side `_LOCK_RETRY_TOKENS` + /// retries on — and once the gen2 shards appear the same lookup pins the + /// consistent gen2 bundle. + #[test] + fn bundle_open_mixed_window_exhausts_then_follows_new_generation() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let primary = dir.path().join("fullmap.redb"); + + // Warm the cache with a consistent gen1 bundle. + write_marker_bundle(&primary, "gen1", 2); + let shards = open_cached_shards(&primary).unwrap(); + assert!(shards.iter().all(|db| shard_payload(db) == b"gen1")); + + // Mid-window: shard files absent, gen2 primary already at the path. + std::fs::remove_file(shard_path(&primary, 0)).unwrap(); + std::fs::remove_file(shard_path(&primary, 1)).unwrap(); + let staging = dir.path().join("fullmap.next.redb"); + write_marker_bundle(&staging, "gen2", 2); + std::fs::rename(&staging, &primary).unwrap(); + + // No bundle may mix the new primary with the old shards: retries + // exhaust and raise (the exhaustion message carries the Python retry + // token "changing generation"). + let err = match open_cached_shards(&primary) { + Ok(_) => panic!("mixed new-primary + old-shard bundle must not be served"), + Err(err) => err, + }; + assert!( + err.to_string().contains("changing generation"), + "exhaustion message must carry the Python retry token: {err}" + ); + + // Once the gen2 shards appear, the same lookup pins the consistent + // gen2 bundle. + write_shard_db(&shard_path(&primary, 0), b"gen2"); + write_shard_db(&shard_path(&primary, 1), b"gen2"); + let shards = open_cached_shards(&primary).unwrap(); + assert!( + shards.iter().all(|db| shard_payload(db) == b"gen2"), + "the recovered bundle must be the NEW generation" + ); + let database = open_cached(primary).unwrap(); + assert_eq!(marker_of(&database), "gen2"); + } + + /// ALL paths absent with a warm cache: the all-stale arm returns the old + /// bundle as one consistent snapshot — the documented snapshot semantics + /// (generalizing the old (None, None) generation check). + #[test] + fn bundle_open_all_absent_serves_old_snapshot() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let primary = dir.path().join("fullmap.redb"); + write_marker_bundle(&primary, "gen1", 2); + let shards = open_cached_shards(&primary).unwrap(); + assert!(shards.iter().all(|db| shard_payload(db) == b"gen1")); + + std::fs::remove_file(&primary).unwrap(); + std::fs::remove_file(shard_path(&primary, 0)).unwrap(); + std::fs::remove_file(shard_path(&primary, 1)).unwrap(); + + let shards = open_cached_shards(&primary).unwrap(); + assert!( + shards.iter().all(|db| shard_payload(db) == b"gen1"), + "an all-absent bundle must serve the old snapshot" + ); + let database = open_cached(primary).unwrap(); + assert_eq!(marker_of(&database), "gen1"); + } + /// While the path is absent (mid-rebuild window), a cached hit keeps /// serving the old snapshot instead of erroring — the cross-process reader /// guarantee from before generation pinning. The lookup after the @@ -3997,14 +4159,14 @@ mod tests { let primary = dir.path().join("fullmap.redb"); let shard = shard_path(&primary, 3); write_shard_db(&shard, b"gen1"); - let stale = open_cached_shard(&primary, 3).unwrap(); + let stale = open_cached_shard(&primary, 3).unwrap().0; assert_eq!(shard_payload(&stale), b"gen1"); let staging = dir.path().join("fullmap.s3.next.redb"); write_shard_db(&staging, b"gen2"); std::fs::rename(&staging, &shard).unwrap(); - let fresh = open_cached_shard(&primary, 3).unwrap(); + let fresh = open_cached_shard(&primary, 3).unwrap().0; assert_eq!( shard_payload(&fresh), b"gen2", diff --git a/src/tablassert/fullmap.py b/src/tablassert/fullmap.py index 082cfe0b..a97f7a80 100644 --- a/src/tablassert/fullmap.py +++ b/src/tablassert/fullmap.py @@ -28,14 +28,16 @@ # otherwise raise ``Database already open`` and surface as a false 0.0 coverage. Retry briefly on that # contention so a transient reader-vs-writer overlap does not corrupt a build/coverage result. # Each lookup pins one primary-plus-shards file generation (cached handles are validated against the -# file's inode on every use); a reader follows a rebuild on the NEXT lookup. -_LOCK_RETRY_TOKENS: tuple[str, ...] = ("already open", "acquire lock", "cannot acquire") +# file's (dev, ino) on every use); a reader follows a rebuild on the NEXT lookup. "changing generation" +# is the Rust-side exhaustion token raised when no consistent bundle can be pinned mid-rebuild. +_LOCK_RETRY_TOKENS: tuple[str, ...] = ("already open", "acquire lock", "cannot acquire", "changing generation") _LOCK_ATTEMPTS: int = 10 _LOCK_DELAY: float = 0.5 def is_lock_contention(error: BaseException) -> bool: - """Whether ``error`` is the transient redb ``Database already open`` lock contention. + """Whether ``error`` is transient fullmap contention (a redb lock error or the mid-rebuild + "changing generation" bundle-pinning exhaustion), safe to retry via :func:`_call_with_lock_retry`. Callers that wrap a lookup in their OWN retry loop (e.g. ``build_and_audit``'s coverage retry) must NOT retry on this error: :func:`_call_with_lock_retry` already exhausted its diff --git a/tests/test_fullmap.py b/tests/test_fullmap.py index 3e0c5e9a..7ed05c69 100644 --- a/tests/test_fullmap.py +++ b/tests/test_fullmap.py @@ -898,7 +898,15 @@ def test_resolve_batch_on_phase_fires_per_column_in_order(fullmap_db: Path) -> N assert with_cb.to_dicts() == without_cb.to_dicts() -@pytest.mark.parametrize("message", ["Database already open", "failed to acquire lock on fullmap", "Cannot Acquire lock"]) +@pytest.mark.parametrize( + "message", + [ + "Database already open", + "failed to acquire lock on fullmap", + "Cannot Acquire lock", + "fullmap at /data/fullmap.redb kept changing generation; could not pin one primary-plus-shards bundle after 5 attempts", + ], +) def test_is_lock_contention_matches_redb_lock_errors(message: str) -> None: assert is_lock_contention(RuntimeError(message)) From 64fc2f122e8f38025fd58be13850e2d596e382ae Mon Sep 17 00:00:00 2001 From: SkyeAv Date: Fri, 7 Aug 2026 14:14:03 -0700 Subject: [PATCH 10/10] fix(fullmap): pin bundles by build id Generate one build_id per fullmap build and write it to the primary META plus every shard META. Shard META is safe here because v5 already mandates a rebuild, so no shard files without META exist in the wild. Cache each handle's build_id and require primary/shard build_id equality before serving any bundle, including all-stale cached snapshots. Retry mixed or diverged bundles with a short delay so slow shard creation does not burn all attempts instantly. --- rust/src/fullmap.rs | 331 ++++++++++++++++++++++++++----------- rust/tests/build_golden.rs | 23 ++- src/tablassert/fullmap.py | 2 +- 3 files changed, 252 insertions(+), 104 deletions(-) diff --git a/rust/src/fullmap.rs b/rust/src/fullmap.rs index 81327c14..fe135000 100644 --- a/rust/src/fullmap.rs +++ b/rust/src/fullmap.rs @@ -73,32 +73,35 @@ type LineChunk = (u8, Vec>); /// Handles are `ReadOnlyDatabase` (redb ≥ 3), which take a SHARED file lock: /// any number of processes/threads may hold them concurrently, and they only /// conflict with a writer (`build-fullmap`'s exclusive lock). Each entry also -/// records the `(st_dev, st_ino)` generation of the file it was opened from: -/// a rebuild replaces the files via unlink + recreate — in practice on FRESH -/// inodes (an `unlink` does not need the lock) — so the path alone cannot -/// distinguish the old data from the new. Every cache hit therefore re-stats -/// the canonical path and evicts the entry when `(dev, ino)` no longer -/// matches, so a lookup always reads the generation currently living at the -/// path — in-process after `build_fullmap_db`, or cross-process after another -/// process rebuilt. A stale evicted handle simply keeps reading the unlinked -/// old file as a consistent snapshot; it can never block the rebuild or -/// observe a torn file. Each lookup pins ONE primary-plus-shards generation: +/// records the `(st_dev, st_ino)` generation and META `build_id` of the file it +/// was opened from: a rebuild replaces the files via unlink + recreate — in +/// practice on FRESH inodes (an `unlink` does not need the lock) — so the path +/// alone cannot distinguish the old data from the new. Every cache hit +/// therefore re-stats the canonical path and evicts the entry when `(dev, ino)` +/// no longer matches, so a lookup always reads the generation currently living +/// at the path — in-process after `build_fullmap_db`, or cross-process after +/// another process rebuilt. A stale evicted handle simply keeps reading the +/// unlinked old file as a consistent snapshot; it can never block the rebuild +/// or observe a torn file. Each lookup pins ONE primary-plus-shards build: /// `open_cached_shards` records how every bundle member was served (`Served`) -/// and accepts the bundle only when it is generationally consistent — ALL -/// current with a stable primary generation, or ALL stale because every path -/// is absent (the documented mid-rebuild snapshot); a MIXED bundle (e.g. the -/// new primary committed while the shard files are still absent) is retried -/// whole and finally raises an exhaustion error the Python side retries like -/// lock contention. The next lookup after a rebuild opens the replacement. -/// Read-only opens never write the file, so the mtime-keyed Python-side caches -/// stay stable under lookups. +/// and accepts the bundle only when every member has the same `build_id` AND is +/// either ALL current with a stable primary generation or ALL stale because +/// every path is absent (the documented mid-rebuild snapshot). A MIXED bundle +/// (e.g. the new primary committed while the shard files are still absent), or +/// an all-stale cache with diverged build IDs, is retried whole and finally +/// raises an exhaustion error the Python side retries like lock contention. The +/// next lookup after a rebuild opens the replacement. Read-only opens never +/// write the file, so the mtime-keyed Python-side caches stay stable under +/// lookups. static DB_CACHE: OnceLock>> = OnceLock::new(); -/// One cached read-only handle plus the `(st_dev, st_ino)` file generation it -/// was opened from, so every cache hit can detect an in-place rebuild. +/// One cached read-only handle plus the file generation and build token it was +/// opened from, so every cache hit can detect an in-place rebuild and bundle +/// opens can reject torn primary/shard sets. struct CachedDatabase { database: Arc, generation: FileGeneration, + build_id: u64, } /// On-disk identity of the file at a path: its `(st_dev, st_ino)` pair. A @@ -111,6 +114,39 @@ fn generation_of(path: &Path) -> std::io::Result { Ok((meta.dev(), meta.ino())) } +fn new_build_id() -> u64 { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default(); + let seq = BUILD_ID_COUNTER.fetch_add(1, Ordering::Relaxed) + 1; + let id = now.as_secs().rotate_left(32) + ^ u64::from(now.subsec_nanos()) + ^ u64::from(std::process::id()).rotate_left(17) + ^ seq; + if id == 0 { + seq + } else { + id + } +} + +fn read_build_id(database: &ReadOnlyDatabase) -> PyResult { + let read = database.begin_read().map_err(py_err)?; + let meta = read.open_table(META).map_err(py_err)?; + let raw = meta.get("build_id").map_err(py_err)?.ok_or_else(|| { + PyRuntimeError::new_err( + "fullmap DB is missing META.build_id; rebuild with 'tablassert build-fullmap'", + ) + })?; + raw.value().parse::().map_err(|_| { + PyRuntimeError::new_err( + "fullmap DB has invalid META.build_id; rebuild with 'tablassert build-fullmap'", + ) + }) +} + +static BUILD_ID_COUNTER: AtomicU64 = AtomicU64::new(0); + /// Number of shards for concurrent maps (power of two for mask routing). const SHARD_COUNT: usize = 64; const SHARD_MASK: usize = SHARD_COUNT - 1; @@ -1763,6 +1799,8 @@ fn write_final_database( shard_count: usize, progress: Option<&Arc>, ) -> PyResult<()> { + let build_id = new_build_id(); + let build_id_str = build_id.to_string(); let database = redb::Builder::new() .set_cache_size(cache_bytes) .create(output) @@ -1811,6 +1849,8 @@ fn write_final_database( let mut meta = write.open_table(META).map_err(py_err)?; meta.insert("schema", SCHEMA_VERSION).map_err(py_err)?; + meta.insert("build_id", build_id_str.as_str()) + .map_err(py_err)?; let shard_count_str = shard_count.to_string(); meta.insert("shards", shard_count_str.as_str()) .map_err(py_err)?; @@ -1821,10 +1861,10 @@ fn write_final_database( // is never touched again (Phase 4 writes only the separate shard DBs below). // Drop it here to release its redb cache and file lock during the long // parallel RECORDS write — a free memory win while the shards are built. - // NOTE: the primary is committed BEFORE the shard files below exist, so on - // rebuild the primary appears first — primary-absent implies shards-absent, - // the invariant the all-stale arm of `open_cached_shards` relies on (see - // the up-front unlink in `build_fullmap_db`). + // NOTE: `build_fullmap_db` unlinks all old files before this new primary is + // committed, and this primary is committed before the new shard files below. + // During unlink, the primary may be absent while old shards still exist; + // after the primary appears, new shards may still be absent. drop(database); // Phase 4: one INDEPENDENT k-way merge per shard, run in parallel — one @@ -1888,6 +1928,7 @@ fn write_final_database( global, total_estimate, progress.as_ref(), + build_id, ) })); } @@ -1934,6 +1975,7 @@ fn write_final_database( /// /// A shard with zero runs still opens and commits an empty RECORDS table, so an /// empty shard DB is produced and the `shard_count`-file layout stays stable. +#[allow(clippy::too_many_arguments)] fn write_shard_records( database: &Database, run_paths: &[PathBuf], @@ -1942,12 +1984,18 @@ fn write_shard_records( global_written: &AtomicU64, total: u64, progress: Option<&Arc>, + build_id: u64, ) -> PyResult { let mut write = database.begin_write().map_err(py_err)?; // redb ≥ 3 returns `Result`; it only fails if a persistent savepoint was // modified in the transaction. This crate never uses savepoints (and this // is a fresh transaction), so the error is unreachable here. write.set_durability(Durability::None).map_err(py_err)?; + let build_id_str = build_id.to_string(); + let mut meta = write.open_table(META).map_err(py_err)?; + meta.insert("build_id", build_id_str.as_str()) + .map_err(py_err)?; + drop(meta); let mut table = write.open_table(RECORDS).map_err(py_err)?; let mut merge = MergeHeap::new(run_paths).map_err(py_err)?; // Pre-size the batch to the flush threshold so it never regrows (each @@ -2118,12 +2166,14 @@ fn evict_cached_path(path: &Path) -> PyResult<()> { fn cache_database(path: &Path, database: Arc) -> PyResult<()> { let canonical = std::fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); let generation = generation_of(&canonical).map_err(py_err)?; + let build_id = read_build_id(&database)?; let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); cache.write().map_err(py_err)?.insert( canonical, CachedDatabase { database, generation, + build_id, }, ); Ok(()) @@ -2264,10 +2314,9 @@ pub fn build_fullmap_db( std::fs::create_dir_all(parent).map_err(py_err)?; } evict_cached_path(&output)?; - // Everything is unlinked BEFORE the new primary is written, and the - // primary is written BEFORE the shards (`write_final_database`): primary - // appears first on rebuild, so primary-absent implies shards-absent — the - // invariant the all-stale arm of `open_cached_shards` relies on. + // All old files are unlinked before the new primary is written, and the new + // primary is written before the new shards (`write_final_database`). During + // unlink, the primary may be absent while old shards still exist. if output.exists() { std::fs::remove_file(&output).map_err(py_err)?; } @@ -2392,11 +2441,14 @@ fn open_cached(db: PathBuf) -> PyResult> { } /// Open (and cache) one RECORDS shard by index, deriving its path from the -/// primary DB path. Shards hold only RECORDS (no META), so they are not +/// primary DB path. Shards carry RECORDS plus META.build_id; they are not /// schema-validated here — the primary's `validate_schema` gates the layout. -/// Returns the handle plus HOW it was served (`Served`) so -/// `open_cached_shards` can pin one bundle generation. -fn open_cached_shard(primary: &Path, index: usize) -> PyResult<(Arc, Served)> { +/// Returns the handle plus HOW it was served (`Served`) and its build token so +/// `open_cached_shards` can pin one bundle build. +fn open_cached_shard( + primary: &Path, + index: usize, +) -> PyResult<(Arc, Served, u64)> { let path = shard_path(primary, index); let canonical = std::fs::canonicalize(&path).unwrap_or(path); open_cached_path(canonical, false) @@ -2416,22 +2468,29 @@ fn open_cached_shard(primary: &Path, index: usize) -> PyResult<(Arc PyResult<(Arc, Served)> { +) -> PyResult<(Arc, Served, u64)> { let cache = DB_CACHE.get_or_init(|| RwLock::new(HashMap::new())); // Copy the hit out of the read lock before any write-lock upgrade attempt. let hit = { let map = cache.read().map_err(py_err)?; - map.get(&canonical) - .map(|entry| (Arc::clone(&entry.database), entry.generation)) + map.get(&canonical).map(|entry| { + ( + Arc::clone(&entry.database), + entry.generation, + entry.build_id, + ) + }) }; - if let Some((database, generation)) = hit { + if let Some((database, generation, build_id)) = hit { match generation_of(&canonical) { // Same inode generation: the handle is still the file at the path. - Ok(current) if current == generation => return Ok((database, Served::Current)), + Ok(current) if current == generation => { + return Ok((database, Served::Current, build_id)) + } // Path gone (rebuild in flight or deleted): keep serving the old // snapshot; the next lookup after a replacement appears follows it. Err(err) if err.kind() == std::io::ErrorKind::NotFound => { - return Ok((database, Served::StaleAbsent)); + return Ok((database, Served::StaleAbsent, build_id)); } // Any other metadata failure (permissions, I/O): propagate it — // serving stale data here would silently pin readers to old state. @@ -2460,6 +2519,7 @@ fn open_cached_path( if validate { validate_schema(&database)?; } + let build_id = read_build_id(&database)?; let generation = match pre { Ok(generation) => generation, Err(_) => generation_of(&canonical).map_err(py_err)?, @@ -2470,9 +2530,10 @@ fn open_cached_path( CachedDatabase { database, generation, + build_id, }, ); - Ok((cached, Served::Current)) + Ok((cached, Served::Current, build_id)) } /// Read the RECORDS shard count advertised in the primary's META, defaulting to @@ -2500,25 +2561,24 @@ fn shard_count_of(database: &ReadOnlyDatabase) -> PyResult { /// Attempts `open_cached_shards` makes to open one consistent generation /// before concluding the DB is being rebuilt in a tight loop. const BUNDLE_OPEN_ATTEMPTS: usize = 5; +const BUNDLE_OPEN_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(50); /// Open (and cache) all RECORDS shard handles for a primary DB path, pinned to -/// ONE file generation. The primary's `(dev, ino)` is captured before opening -/// anything and re-stat'ed after the last shard, and every bundle member -/// reports HOW it was served (`Served`). The bundle is accepted only when it -/// is generationally consistent: +/// ONE build. The primary's `(dev, ino)` is captured before opening anything +/// and re-stat'ed after the last shard, and every bundle member reports HOW it +/// was served (`Served`) plus the META `build_id` read when its handle opened. +/// The bundle is accepted only when every member has the same `build_id` and is +/// generationally consistent: /// /// - ALL `Current` with the primary generation unchanged across the fan-out: -/// every handle belongs to one build — return it. -/// - ALL `StaleAbsent`: the rebuild window straddled the whole bundle. The -/// rebuild unlinks every file up front and recreates the primary BEFORE the -/// shards (primary appears first on rebuild, so primary-absent implies -/// shards-absent), so every handle is the same old snapshot — return the old -/// bundle (the documented snapshot semantics; the next lookup after the -/// replacement appears follows it). -/// - MIXED (any `StaleAbsent` alongside any `Current`): the bundle could pair -/// the NEW primary with OLD shards — exactly the torn state generation -/// pinning exists to prevent — so retry the WHOLE bundle, and after -/// `BUNDLE_OPEN_ATTEMPTS` raise the exhaustion error below (its +/// every handle belongs to the current build — return it. +/// - ALL `StaleAbsent`: the rebuild window straddled the whole bundle. All old +/// files were unlinked before the new primary was written, and the new primary +/// is written before the new shards; if every cached handle shares one +/// `build_id`, return the old snapshot (the documented snapshot semantics). +/// - MIXED (any `StaleAbsent` alongside any `Current`), unequal `build_id`s, or +/// primary generation movement during fan-out: retry the WHOLE bundle, and +/// after `BUNDLE_OPEN_ATTEMPTS` raise the exhaustion error below (its /// "changing generation" text is one of the Python-side /// `_LOCK_RETRY_TOKENS`). /// @@ -2526,27 +2586,31 @@ const BUNDLE_OPEN_ATTEMPTS: usize = 5; /// exactly the shards the build wrote. fn open_cached_shards(primary: &Path) -> PyResult>> { let canonical = std::fs::canonicalize(primary).unwrap_or_else(|_| primary.to_path_buf()); - for _ in 0..BUNDLE_OPEN_ATTEMPTS { + for attempt in 0..BUNDLE_OPEN_ATTEMPTS { let pinned = generation_of(&canonical).ok(); - let (database, primary_served) = open_cached_path(canonical.clone(), true)?; + let (database, primary_served, primary_build_id) = + open_cached_path(canonical.clone(), true)?; let shard_count = shard_count_of(&database)?; let mut all_current = primary_served == Served::Current; let mut all_stale_absent = primary_served == Served::StaleAbsent; + let mut same_build_id = true; let mut shards: Vec> = Vec::with_capacity(shard_count); for index in 0..shard_count { - let (database, served) = open_cached_shard(primary, index)?; + let (database, served, build_id) = open_cached_shard(primary, index)?; all_current &= served == Served::Current; all_stale_absent &= served == Served::StaleAbsent; + same_build_id &= build_id == primary_build_id; shards.push(database); } - // ALL stale: every path is absent, so every handle is the same old - // snapshot — a consistent bundle (see the invariant above). - if all_stale_absent { + // ALL stale: every path is absent; equal build IDs prove the cached + // handles still form one old snapshot. + if all_stale_absent && same_build_id { return Ok(shards); } - // ALL current: re-stat the primary; an unchanged generation means - // every handle above belongs to one build. - if all_current + // ALL current: re-stat the primary; an unchanged generation plus equal + // build IDs means every handle above belongs to one build. + if same_build_id + && all_current && matches!( (pinned, generation_of(&canonical).ok()), (Some(before), Some(after)) if before == after @@ -2554,8 +2618,11 @@ fn open_cached_shards(primary: &Path) -> PyResult>> { { return Ok(shards); } - // MIXED serve kinds, or the primary generation moved during the + // MIXED serve kinds, unequal build IDs, or primary movement during // fan-out: the handles may span two builds — retry the whole bundle. + if attempt + 1 < BUNDLE_OPEN_ATTEMPTS { + std::thread::sleep(BUNDLE_OPEN_RETRY_DELAY); + } } let path = canonical.display(); Err(PyRuntimeError::new_err(format!( @@ -3079,11 +3146,11 @@ mod tests { assert_eq!(rows[0].1[0].source_version, FULLMAP_SOURCE_VERSION); } - /// The sharded (v4+) layout keeps dims+CURIES+META in the primary and moves RECORDS - /// into sibling shard files. The primary must NOT carry a RECORDS table, - /// META must advertise both the schema and the shard count, and every shard - /// file must exist (even empty) holding a RECORDS table — this is the - /// on-disk contract the read path relies on. + /// The sharded (v5) layout keeps dims+CURIES+META in the primary and moves + /// RECORDS into sibling shard files. The primary must NOT carry a RECORDS + /// table, META must advertise schema/shards/build_id, and every shard file + /// must exist (even empty) holding RECORDS plus the same build_id — this is + /// the on-disk contract the read path relies on. #[test] fn build_fullmap_db_writes_sharded_layout() { pyo3::Python::initialize(); @@ -3100,8 +3167,7 @@ mod tests { build_test(output.clone(), Vec::new(), vec![synonyms], 1, 4_000_000).unwrap(); - // Primary: META (schema=v4, shards=SHARD_COUNT_SHARDS) + dims + CURIES, - // but NO RECORDS. + // Primary: META (schema, shards, build_id) + dims + CURIES, but NO RECORDS. let database = open_cached(output.clone()).unwrap(); let read = database.begin_read().unwrap(); let meta = read.open_table(META).unwrap(); @@ -3110,6 +3176,8 @@ mod tests { meta.get("shards").unwrap().unwrap().value(), SHARD_COUNT_SHARDS.to_string() ); + let build_id = meta.get("build_id").unwrap().unwrap().value().to_string(); + build_id.parse::().unwrap(); drop(meta); let _prefixes = read.open_table(PREFIXES).unwrap(); let _categories = read.open_table(CATEGORIES).unwrap(); @@ -3117,18 +3185,20 @@ mod tests { let _curies = read.open_table(CURIES).unwrap(); assert!( read.open_table(RECORDS).is_err(), - "primary must not hold a RECORDS table in the sharded (v4+) layout" + "primary must not hold a RECORDS table in the sharded (v5) layout" ); drop(read); drop(database); - // All default shard files exist and each holds a RECORDS table. + // All default shard files exist and each holds RECORDS + matching build_id. for index in 0..SHARD_COUNT_SHARDS { let shard = shard_path(&output, index); assert!(shard.exists(), "missing shard file {shard:?}"); let db = ReadOnlyDatabase::open(&shard).unwrap(); let read = db.begin_read().unwrap(); let _records = read.open_table(RECORDS).unwrap(); + let meta = read.open_table(META).unwrap(); + assert_eq!(meta.get("build_id").unwrap().unwrap().value(), build_id); } // The single indexed term still resolves (routed through its shard). @@ -3671,8 +3741,8 @@ mod tests { } /// The build and read paths agree on a parameterized shard count. A 2-shard - /// build must write META.shards="2", create exactly s0+s1 (each with a RECORDS - /// table, even the one that receives no terms), create no higher shard files up + /// build must write META.shards="2", create exactly s0+s1 (each with RECORDS + /// plus META.build_id, even the one that receives no terms), create no higher shard files up /// to the compile-time cap, and the read path must open exactly 2 shards (from /// META) and still resolve every term. The public entry point pins the count /// to `SHARD_COUNT_SHARDS`, but this exercises the same parameterized path the @@ -3950,34 +4020,50 @@ mod tests { } } + fn test_build_id(marker: &str) -> u64 { + xxh64(marker.as_bytes(), 0) + } + /// Write a minimal primary-shaped DB at `path` whose META carries `marker` - /// alongside the current schema tag, so tests can tell generations apart. + /// alongside the current schema tag and build token. /// The writer is dropped before returning, so the file is unlocked. - fn write_marker_db(path: &Path, marker: &str) { + fn write_marker_primary(path: &Path, marker: &str, shard_count: usize) { let database = Database::create(path).unwrap(); let write = database.begin_write().unwrap(); { let mut meta = write.open_table(META).unwrap(); meta.insert("schema", SCHEMA_VERSION).unwrap(); + let build_id = test_build_id(marker).to_string(); + meta.insert("build_id", build_id.as_str()).unwrap(); meta.insert("marker", marker).unwrap(); + let shards = shard_count.to_string(); + meta.insert("shards", shards.as_str()).unwrap(); } write.commit().unwrap(); drop(database); } + fn write_marker_db(path: &Path, marker: &str) { + write_marker_primary(path, marker, SHARD_COUNT_SHARDS); + } + fn marker_of(database: &ReadOnlyDatabase) -> String { let read = database.begin_read().unwrap(); let meta = read.open_table(META).unwrap(); meta.get("marker").unwrap().unwrap().value().to_string() } - /// Write a shard-shaped DB (RECORDS only) holding one record `1 -> payload`. - fn write_shard_db(path: &Path, payload: &[u8]) { + /// Write a shard-shaped DB holding one record `1 -> marker` plus META.build_id. + fn write_shard_db(path: &Path, marker: &str) { let database = Database::create(path).unwrap(); let write = database.begin_write().unwrap(); { + let build_id = test_build_id(marker).to_string(); + let mut meta = write.open_table(META).unwrap(); + meta.insert("build_id", build_id.as_str()).unwrap(); + drop(meta); let mut records = write.open_table(RECORDS).unwrap(); - records.insert(1u64, payload).unwrap(); + records.insert(1u64, marker.as_bytes()).unwrap(); } write.commit().unwrap(); drop(database); @@ -4004,19 +4090,9 @@ mod tests { /// carrying `marker`, so tests can warm `open_cached_shards` without a /// full build. fn write_marker_bundle(primary: &Path, marker: &str, shard_count: usize) { - let database = Database::create(primary).unwrap(); - let write = database.begin_write().unwrap(); - { - let mut meta = write.open_table(META).unwrap(); - meta.insert("schema", SCHEMA_VERSION).unwrap(); - meta.insert("marker", marker).unwrap(); - let count = shard_count.to_string(); - meta.insert("shards", count.as_str()).unwrap(); - } - write.commit().unwrap(); - drop(database); + write_marker_primary(primary, marker, shard_count); for index in 0..shard_count { - write_shard_db(&shard_path(primary, index), marker.as_bytes()); + write_shard_db(&shard_path(primary, index), marker); } } @@ -4059,8 +4135,8 @@ mod tests { // Once the gen2 shards appear, the same lookup pins the consistent // gen2 bundle. - write_shard_db(&shard_path(&primary, 0), b"gen2"); - write_shard_db(&shard_path(&primary, 1), b"gen2"); + write_shard_db(&shard_path(&primary, 0), "gen2"); + write_shard_db(&shard_path(&primary, 1), "gen2"); let shards = open_cached_shards(&primary).unwrap(); assert!( shards.iter().all(|db| shard_payload(db) == b"gen2"), @@ -4070,6 +4146,69 @@ mod tests { assert_eq!(marker_of(&database), "gen2"); } + /// Unlink-phase shape: the primary can be absent while old shard paths are + /// still present. Mixed stale/current serve kinds must exhaust, then a + /// reappearing primary with the shards' build_id forms a consistent bundle. + #[test] + fn bundle_open_primary_absent_shards_current_exhausts_then_follows_primary() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let primary = dir.path().join("fullmap.redb"); + + write_marker_bundle(&primary, "gen1", 2); + let shards = open_cached_shards(&primary).unwrap(); + assert!(shards.iter().all(|db| shard_payload(db) == b"gen1")); + + std::fs::remove_file(&primary).unwrap(); + let err = match open_cached_shards(&primary) { + Ok(_) => panic!("primary-absent + shard-current bundle must not be served"), + Err(err) => err, + }; + assert!( + err.to_string().contains("changing generation"), + "exhaustion message must carry the Python retry token: {err}" + ); + + write_marker_primary(&primary, "gen1", 2); + let shards = open_cached_shards(&primary).unwrap(); + assert!(shards.iter().all(|db| shard_payload(db) == b"gen1")); + let database = open_cached(primary).unwrap(); + assert_eq!(marker_of(&database), "gen1"); + } + + /// Primary-only readers can advance just the primary cache entry. If all + /// paths then go absent, the all-stale arm must reject primary=gen2 plus + /// shards=gen1 rather than serving a torn cached bundle. + #[test] + fn bundle_open_all_absent_rejects_diverged_cached_build_ids() { + pyo3::Python::initialize(); + let dir = tempfile::tempdir().unwrap(); + let primary = dir.path().join("fullmap.redb"); + + write_marker_bundle(&primary, "gen1", 2); + let shards = open_cached_shards(&primary).unwrap(); + assert!(shards.iter().all(|db| shard_payload(db) == b"gen1")); + + let staging = dir.path().join("fullmap.next.redb"); + write_marker_primary(&staging, "gen2", 2); + std::fs::rename(&staging, &primary).unwrap(); + let database = open_cached(primary.clone()).unwrap(); + assert_eq!(marker_of(&database), "gen2"); + + std::fs::remove_file(&primary).unwrap(); + std::fs::remove_file(shard_path(&primary, 0)).unwrap(); + std::fs::remove_file(shard_path(&primary, 1)).unwrap(); + + let err = match open_cached_shards(&primary) { + Ok(_) => panic!("all-stale cache with diverged build IDs must not be served"), + Err(err) => err, + }; + assert!( + err.to_string().contains("changing generation"), + "exhaustion message must carry the Python retry token: {err}" + ); + } + /// ALL paths absent with a warm cache: the all-stale arm returns the old /// bundle as one consistent snapshot — the documented snapshot semantics /// (generalizing the old (None, None) generation check). @@ -4158,12 +4297,12 @@ mod tests { let dir = tempfile::tempdir().unwrap(); let primary = dir.path().join("fullmap.redb"); let shard = shard_path(&primary, 3); - write_shard_db(&shard, b"gen1"); + write_shard_db(&shard, "gen1"); let stale = open_cached_shard(&primary, 3).unwrap().0; assert_eq!(shard_payload(&stale), b"gen1"); let staging = dir.path().join("fullmap.s3.next.redb"); - write_shard_db(&staging, b"gen2"); + write_shard_db(&staging, "gen2"); std::fs::rename(&staging, &shard).unwrap(); let fresh = open_cached_shard(&primary, 3).unwrap().0; diff --git a/rust/tests/build_golden.rs b/rust/tests/build_golden.rs index 99fa6ebb..2050dde6 100644 --- a/rust/tests/build_golden.rs +++ b/rust/tests/build_golden.rs @@ -7,8 +7,9 @@ //! the build through the PUBLIC `build_fullmap_db` re-exported at the crate root, //! then inspect the resulting redb files DIRECTLY: //! -//! * RECORDS live in the 16 sibling shard files (`fullmap.s{0..15}.redb`), which -//! the build does NOT cache, so they open cleanly with `Database::open`. +//! * RECORDS plus META.build_id live in the 16 sibling shard files +//! (`fullmap.s{0..15}.redb`), which the build does NOT cache, so they open +//! cleanly with `Database::open`. //! * dims/CURIES/META live in the primary (`fullmap.redb`), which `build_fullmap_db` //! caches and holds under redb's exclusive flock. To read it directly we COPY //! the committed primary file to a fresh inode (no lock) and open the copy. @@ -419,7 +420,7 @@ fn dimension_tables_are_complete_and_consistent() { } // --------------------------------------------------------------------------- -// (e) SCHEMA PIN — META advertises the v4 schema and 16 shards. +// (e) SCHEMA PIN — META advertises the v5 schema, build_id, and 16 shards. // --------------------------------------------------------------------------- #[test] @@ -431,15 +432,23 @@ fn schema_and_shard_count_are_pinned() { let meta = read.open_table(META).unwrap(); assert_eq!(meta.get("schema").unwrap().unwrap().value(), SCHEMA_VERSION); assert_eq!(meta.get("shards").unwrap().unwrap().value(), "16"); + let build_id = meta.get("build_id").unwrap().unwrap().value().to_string(); + build_id.parse::().unwrap(); drop(meta); drop(read); drop(db); - // Exactly 16 shard files exist on disk (s0..s15), and no s16. + // Exactly 16 shard files exist on disk (s0..s15), each carries the same + // build_id, and no s16 exists. for index in 0..SHARD_COUNT { - assert!( - shard_path(&output, index).exists(), - "missing shard file {index}" + let shard = shard_path(&output, index); + assert!(shard.exists(), "missing shard file {index}"); + let shard_db = Database::open(shard).unwrap(); + let shard_read = shard_db.begin_read().unwrap(); + let shard_meta = shard_read.open_table(META).unwrap(); + assert_eq!( + shard_meta.get("build_id").unwrap().unwrap().value(), + build_id ); } assert!( diff --git a/src/tablassert/fullmap.py b/src/tablassert/fullmap.py index a97f7a80..46d9c536 100644 --- a/src/tablassert/fullmap.py +++ b/src/tablassert/fullmap.py @@ -49,7 +49,7 @@ def is_lock_contention(error: BaseException) -> bool: def _call_with_lock_retry(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any: - """Call a redb-backed ``rs`` function, retrying on transient ``Database already open`` lock contention.""" + """Call a redb-backed ``rs`` function, retrying transient lock or ``changing generation`` contention.""" for attempt in range(_LOCK_ATTEMPTS): try: return fn(*args, **kwargs)