From 6e9193dca89f18d7fb1f79c0fafff745e654ec6d Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 19:42:35 +0100 Subject: [PATCH 01/23] mason: preserve reservation ledger failure context Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3.rs | 82 +++++++++++++++++++++++++++------ 1 file changed, 67 insertions(+), 15 deletions(-) diff --git a/crates/aft/src/blob_store/v3.rs b/crates/aft/src/blob_store/v3.rs index 8916e5103..52cc4f7a2 100644 --- a/crates/aft/src/blob_store/v3.rs +++ b/crates/aft/src/blob_store/v3.rs @@ -156,6 +156,11 @@ pub enum Error { Sqlite(rusqlite::Error), Incompatible(UnsupportedPersistedFormat), Refused(&'static str), + Ledger { + path: PathBuf, + operation: &'static str, + source: Box, + }, InvalidIdentity, ReservationRegistryMismatch, } @@ -166,12 +171,30 @@ impl fmt::Display for Error { Self::Sqlite(e) => write!(f, "{e}"), Self::Incompatible(e) => write!(f, "{e}"), Self::Refused(reason) => f.write_str(reason), + Self::Ledger { + path, + operation, + source, + } => write!( + f, + "reservation_ledger_unreadable: {operation} {}: {source}", + path.display() + ), Self::InvalidIdentity => f.write_str("invalid v3 family, scope or lock identity"), Self::ReservationRegistryMismatch => f.write_str("reservation_registry_mismatch"), } } } impl std::error::Error for Error {} +impl Error { + fn ledger(path: &Path, operation: &'static str, source: impl Into) -> Self { + Self::Ledger { + path: path.to_path_buf(), + operation, + source: Box::new(source.into()), + } + } +} impl From for Error { fn from(e: std::io::Error) -> Self { Self::Io(e) @@ -739,14 +762,14 @@ fn for_each_registry( let path = entry.path().join("members.sqlite"); if !path .try_exists() - .map_err(|_| Error::Refused("reservation_ledger_unreadable"))? + .map_err(|error| Error::ledger(&path, "stat registry", error))? { continue; } let name = entry.file_name(); body( name.to_str() - .ok_or(Error::Refused("reservation_ledger_unreadable"))?, + .ok_or_else(|| Error::ledger(&path, "decode family", Error::InvalidIdentity))?, &path, )?; } @@ -784,13 +807,13 @@ fn preflight_floor(storage: &Path, request: u64) -> Result<()> { } fn recover_ledgers(storage: &Path, lock: &ReservationLock) -> Result<()> { lock.verify(storage)?; - for_each_registry(storage, |family, _| { + for_each_registry(storage, |family, path| { let recover = LedgerReader::open_existing(storage, family) .and_then(|reader| reader.reservation_owners()) .map_or(true, |(_, abandoned)| abandoned); if recover { RegistryWriter::open_existing(storage, family, lock) - .map_err(|_| Error::Refused("reservation_ledger_unreadable"))?; + .map_err(|error| Error::ledger(path, "recover registry", error))?; } Ok(()) }) @@ -808,14 +831,18 @@ fn check_floor(storage: &Path, request: u64, own: Option<(&Path, &Connection)>) let registry = entry.path().join("members.sqlite"); if !registry .try_exists() - .map_err(|_| Error::Refused("reservation_ledger_unreadable"))? + .map_err(|error| Error::ledger(®istry, "stat registry for admission", error))? { continue; } let name = entry.file_name(); - let name = name - .to_str() - .ok_or(Error::Refused("reservation_ledger_unreadable"))?; + let name = name.to_str().ok_or_else(|| { + Error::ledger( + ®istry, + "decode family for admission", + Error::InvalidIdentity, + ) + })?; let own_connection = own.filter(|(path, _)| { registry == *path || crate::db::file_identity::identity_of(®istry).is_some_and(|identity| { @@ -827,7 +854,9 @@ fn check_floor(storage: &Path, request: u64, own: Option<(&Path, &Connection)>) || LedgerReader::open_existing(storage, name).and_then(|r| r.accounting()), |(_, connection)| accounting(connection), ) - .map_err(|_| Error::Refused("reservation_ledger_unreadable"))?; + .map_err(|error| { + Error::ledger(®istry, "read accounting for admission", error) + })?; reserved = reserved .checked_add(ledger.reserved_bytes) .ok_or(Error::Refused("device_free_space_floor"))?; @@ -1099,7 +1128,11 @@ impl<'a> RegistryWriter<'a> { None => return settle_registry_growth(&mut connection, &self.dir, bytes, counted), } } - Err(Error::Refused("reservation_ledger_unreadable")) + Err(Error::ledger( + &self.dir, + "measure registry", + std::io::Error::other("namespace size did not stabilize after 8 measurements"), + )) } pub fn with_barrier( &mut self, @@ -1142,8 +1175,14 @@ impl<'a> RegistryWriter<'a> { pub fn reserve(&mut self, request: u64) -> Result { let storage = self.lock.storage.clone(); recover_ledgers(&storage, self.lock)?; - let owner_start = crate::root_cache::process_start_time_ms(std::process::id()) - .ok_or(Error::Refused("reservation_ledger_unreadable"))?; + let owner_start = + crate::root_cache::process_start_time_ms(std::process::id()).ok_or_else(|| { + Error::ledger( + &self.dir, + "reserve", + std::io::Error::other("current process start time unavailable"), + ) + })?; let before = allocated_bytes(&self.dir)?; let registry = self.registry_identity()?; let path = self.dir.join("members.sqlite"); @@ -1255,7 +1294,11 @@ fn settle_registry_growth( checkpoint(connection)?; recorded = safe; } - Err(Error::Refused("reservation_ledger_unreadable")) + Err(Error::ledger( + dir, + "settle registry growth", + std::io::Error::other("registry size did not stabilize after 8 corrections"), + )) } #[cfg(test)] @@ -1272,12 +1315,21 @@ pub(crate) fn grow_and_expire_recheck_for_test() { } fn checkpoint(c: &Connection) -> Result<()> { - let (busy, _, _): (i64, i64, i64) = + let (busy, log, checkpointed): (i64, i64, i64) = c.query_row("PRAGMA wal_checkpoint(TRUNCATE)", [], |r| { Ok((r.get(0)?, r.get(1)?, r.get(2)?)) })?; if busy != 0 { - return Err(Error::Refused("reservation_ledger_unreadable")); + return Err(Error::ledger( + Path::new(c.path().unwrap_or("")), + "checkpoint registry", + rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY), + Some(format!( + "wal_checkpoint(TRUNCATE): busy={busy}, log={log}, checkpointed={checkpointed}" + )), + ), + )); } Ok(()) } From 6d8772de6789df967f43d1c121df0b9af252f5a8 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:10:33 +0100 Subject: [PATCH 02/23] mason: preserve transient v3 contention through callgraph views The historical Windows CI failure was not reproduced in native branch-switch runs. This fixes mislabeling found in code: a busy WAL checkpoint and an inventory that changes through eight confirmation attempts are transient contention, not an unreadable ledger. Exercise a real pinned WAL reader through checkout callgraph acquisition and deterministic derived SQLite growth during measurement. Keep corrupt/incomplete peer-ledger admission refused; update those assertions only to require operation/path/source diagnostics rather than a cause-free code. Preserve migration storage gaps for permanent ledger failures without swallowing busy. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3.rs | 46 +++++++- crates/aft/src/blob_store/v3/tests.rs | 124 +++++++++++++++++++- crates/aft/src/migration/v3_import.rs | 1 + crates/aft/src/migration/v3_import_tests.rs | 19 +++ crates/aft/src/views/callgraph.rs | 82 +++++++++++-- crates/aft/src/views/first_load.rs | 6 +- 6 files changed, 258 insertions(+), 20 deletions(-) diff --git a/crates/aft/src/blob_store/v3.rs b/crates/aft/src/blob_store/v3.rs index 52cc4f7a2..f70a628a6 100644 --- a/crates/aft/src/blob_store/v3.rs +++ b/crates/aft/src/blob_store/v3.rs @@ -156,6 +156,7 @@ pub enum Error { Sqlite(rusqlite::Error), Incompatible(UnsupportedPersistedFormat), Refused(&'static str), + Busy(&'static str), Ledger { path: PathBuf, operation: &'static str, @@ -171,13 +172,19 @@ impl fmt::Display for Error { Self::Sqlite(e) => write!(f, "{e}"), Self::Incompatible(e) => write!(f, "{e}"), Self::Refused(reason) => f.write_str(reason), + Self::Busy(reason) => write!(f, "v3 store busy: {reason}"), Self::Ledger { path, operation, source, } => write!( f, - "reservation_ledger_unreadable: {operation} {}: {source}", + "{}: {operation} {}: {source}", + if self.is_busy() { + "reservation_ledger_busy" + } else { + "reservation_ledger_unreadable" + }, path.display() ), Self::InvalidIdentity => f.write_str("invalid v3 family, scope or lock identity"), @@ -185,8 +192,35 @@ impl fmt::Display for Error { } } } -impl std::error::Error for Error {} +impl std::error::Error for Error { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(error) => Some(error), + Self::Sqlite(error) => Some(error), + Self::Ledger { source, .. } => Some(source.as_ref()), + _ => None, + } + } +} impl Error { + /// Contention is retryable; unreadable or incompatible ledgers are not + /// treated as empty accounting or as transient contention. + pub fn is_busy(&self) -> bool { + match self { + Self::Busy(_) => true, + Self::Sqlite(rusqlite::Error::SqliteFailure(error, _)) => matches!( + error.code, + rusqlite::ErrorCode::DatabaseBusy | rusqlite::ErrorCode::DatabaseLocked + ), + Self::Io(error) => { + error.kind() == std::io::ErrorKind::WouldBlock + || cfg!(windows) && matches!(error.raw_os_error(), Some(32 | 33)) + } + Self::Ledger { source, .. } => source.is_busy(), + _ => false, + } + } + fn ledger(path: &Path, operation: &'static str, source: impl Into) -> Self { Self::Ledger { path: path.to_path_buf(), @@ -1122,6 +1156,8 @@ impl<'a> RegistryWriter<'a> { // walk did not count, so only a complete recheck that finds the // same total confirms it. A recheck the deadline stops is not a // confirmation. + #[cfg(test)] + tests::measurement_recheck_hook()?; match allocated_bytes_until(&self.dir, deadline)? { Some(again) if again == bytes => return Ok(Measurement::Recorded(bytes)), Some(_) => {} @@ -1131,7 +1167,7 @@ impl<'a> RegistryWriter<'a> { Err(Error::ledger( &self.dir, "measure registry", - std::io::Error::other("namespace size did not stabilize after 8 measurements"), + Error::Busy("namespace size did not stabilize after 8 measurements"), )) } pub fn with_barrier( @@ -1297,7 +1333,7 @@ fn settle_registry_growth( Err(Error::ledger( dir, "settle registry growth", - std::io::Error::other("registry size did not stabilize after 8 corrections"), + Error::Busy("registry size did not stabilize after 8 corrections"), )) } @@ -1322,7 +1358,7 @@ fn checkpoint(c: &Connection) -> Result<()> { if busy != 0 { return Err(Error::ledger( Path::new(c.path().unwrap_or("")), - "checkpoint registry", + "checkpoint WAL", rusqlite::Error::SqliteFailure( rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY), Some(format!( diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index fbe23a83b..ad1892f38 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -120,6 +120,112 @@ fn db(storage: &Path, family: &str) -> PathBuf { family_dir(storage, family).unwrap().join("members.sqlite") } +type RecheckHook = Box Result<()>>; +thread_local! { + static RECHECK_HOOK: std::cell::RefCell> = const { std::cell::RefCell::new(None) }; +} +pub(super) fn measurement_recheck_hook() -> Result<()> { + RECHECK_HOOK.with(|hook| match &mut *hook.borrow_mut() { + Some(hook) => hook(), + None => Ok(()), + }) +} + +#[test] +fn checkpoint_contention_is_busy_and_preserves_sqlite_cause() { + let _guard = serial(); + let root = tempdir().unwrap(); + seed(root.path(), A); + let path = db(root.path(), A); + let lock = ReservationLock::acquire(root.path()).unwrap(); + let mut writer = RegistryWriter::open_for_barrier(root.path(), A, &lock).unwrap(); + writer + .connection() + .unwrap() + .busy_timeout(std::time::Duration::ZERO) + .unwrap(); + let reader = IdentityConnection::open_with_flags( + &path, + OpenFlags::SQLITE_OPEN_READ_ONLY, + "v3-checkpoint-blocker", + ) + .unwrap(); + reader + .execute_batch("BEGIN; SELECT * FROM accounting;") + .unwrap(); + writer + .with_barrier(|tx| { + tx.execute("UPDATE accounting SET watermark=watermark+1", [])?; + Ok(()) + }) + .unwrap(); + + let error = writer.covering_measurement().unwrap_err(); + assert!(error.is_busy(), "{error}"); + let detail = error.to_string(); + assert!( + detail.starts_with("reservation_ledger_busy: checkpoint WAL"), + "{detail}" + ); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!( + detail.contains("wal_checkpoint(TRUNCATE): busy=1"), + "{detail}" + ); + assert!(matches!(error, Error::Ledger { source, .. } + if matches!(*source, Error::Sqlite(rusqlite::Error::SqliteFailure(code, _)) + if code.code == rusqlite::ErrorCode::DatabaseBusy))); + + reader.execute_batch("ROLLBACK").unwrap(); + writer.covering_measurement().unwrap(); + assert_eq!(writer.accounting().unwrap().reserved_bytes, 0); +} + +#[test] +fn changing_derived_database_reports_busy_then_measurement_recovers() { + let _guard = serial(); + let root = tempdir().unwrap(); + seed(root.path(), A); + let derived_dir = view_dir(root.path(), A, "churning-view").unwrap(); + fs::create_dir_all(&derived_dir).unwrap(); + let derived = IdentityConnection::open( + derived_dir.join("derived-generation.sqlite"), + "v3-measurement-churn", + ) + .unwrap(); + derived + .execute_batch("CREATE TABLE growing(payload BLOB)") + .unwrap(); + let lock = ReservationLock::acquire(root.path()).unwrap(); + let mut writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap(); + let writes = std::rc::Rc::new(std::cell::Cell::new(0)); + let count = writes.clone(); + // A derived graph can commit outside the root reservation lock. Place one + // real SQLite commit between each inventory walk and its confirmation. + RECHECK_HOOK.with(|hook| { + *hook.borrow_mut() = Some(Box::new(move || { + derived.execute("INSERT INTO growing VALUES(zeroblob(16384))", [])?; + count.set(count.get() + 1); + Ok(()) + })) + }); + let result = writer.covering_measurement(); + RECHECK_HOOK.with(|hook| *hook.borrow_mut() = None); + let error = result.unwrap_err(); + assert_eq!(writes.get(), 8); + assert!(error.is_busy(), "{error}"); + let detail = error.to_string(); + assert!( + detail.starts_with("reservation_ledger_busy: measure registry"), + "{detail}" + ); + assert!(detail.contains("8 measurements"), "{detail}"); + + let measured = writer.covering_measurement().unwrap(); + assert_eq!(measured, allocated_bytes(writer.family_dir()).unwrap()); + assert_eq!(writer.accounting().unwrap().reserved_bytes, 0); +} + #[test] fn floor_refusal_before_creation_and_budget_recheck_leave_no_plane_or_row() { let _guard = serial(); @@ -219,7 +325,14 @@ fn unreadable_peer_ledger_refuses_instead_of_ignoring_reserved_bytes() { fs::create_dir_all(&dir).unwrap(); fs::write(dir.join("members.sqlite"), b"not sqlite").unwrap(); let error = publish(root.path(), B, 100, None, |_| Ok(())).unwrap_err(); - assert_eq!(error.to_string(), "reservation_ledger_unreadable"); + let detail = error.to_string(); + assert!(!error.is_busy(), "{detail}"); + assert!( + detail.starts_with("reservation_ledger_unreadable: recover registry"), + "{detail}" + ); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!(detail.contains("file is not a database"), "{detail}"); assert!(!db(root.path(), B).exists()); } @@ -1477,7 +1590,14 @@ fn incomplete_marker_with_application_data_is_not_reset() { let root = tempdir().unwrap(); legacy_bootstrap(root.path(), Kind::Registry, 2, true); let error = publish(root.path(), B, 1, None, |_| Ok(())).unwrap_err(); - assert_eq!(error.to_string(), "reservation_ledger_unreadable"); + let detail = error.to_string(); + assert!(!error.is_busy(), "{detail}"); + assert!( + detail.starts_with("reservation_ledger_unreadable: recover registry"), + "{detail}" + ); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!(detail.contains("schema_drift"), "{detail}"); let path = db(root.path(), A); assert_eq!(crate::db::file_identity::open_connections(&path), 0); let c = IdentityConnection::open_with_flags( diff --git a/crates/aft/src/migration/v3_import.rs b/crates/aft/src/migration/v3_import.rs index 43a06d831..98a6024f4 100644 --- a/crates/aft/src/migration/v3_import.rs +++ b/crates/aft/src/migration/v3_import.rs @@ -1360,6 +1360,7 @@ fn storage_gap(error: &v3::Error) -> Option { v3::Error::Refused(reason) if STORAGE_GAP_REFUSALS.contains(reason) => { Some(format!("storage_gap:{reason}")) } + v3::Error::Ledger { .. } if !error.is_busy() => Some(format!("storage_gap:{error}")), _ => None, } } diff --git a/crates/aft/src/migration/v3_import_tests.rs b/crates/aft/src/migration/v3_import_tests.rs index 3f96c8219..f98240848 100644 --- a/crates/aft/src/migration/v3_import_tests.rs +++ b/crates/aft/src/migration/v3_import_tests.rs @@ -462,3 +462,22 @@ fn a_refused_reservation_is_a_named_storage_gap() { .unwrap() .exists()); } + +#[test] +fn ledger_storage_gap_keeps_diagnostics_but_does_not_swallow_busy() { + let error = |code| v3::Error::Ledger { + path: PathBuf::from("family/members.sqlite"), + operation: "recover registry", + source: Box::new(v3::Error::Sqlite(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(code), + Some("underlying SQLite cause".into()), + ))), + }; + let unreadable = error(rusqlite::ffi::SQLITE_NOTADB); + assert_eq!( + storage_gap(&unreadable), + Some(format!("storage_gap:{unreadable}")) + ); + assert!(unreadable.to_string().contains("underlying SQLite cause")); + assert_eq!(storage_gap(&error(rusqlite::ffi::SQLITE_BUSY)), None); +} diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs index 9e890b83d..0a9395cbc 100644 --- a/crates/aft/src/views/callgraph.rs +++ b/crates/aft/src/views/callgraph.rs @@ -104,6 +104,15 @@ impl ManifestBlobReader for BlobReader { /// Reads callgraph blobs from the family's v3 callgraph plane store. pub struct V3BlobReader(pub crate::blob_store::v3::PlaneReader); + +pub(super) fn v3_error(error: crate::blob_store::v3::Error) -> PlaneError { + if error.is_busy() { + self::error(format!("{V3_STORE_BUSY}: {error}")) + } else { + self::error(error) + } +} + impl ManifestBlobReader for V3BlobReader { fn read_callgraph_blob(&self, key: &str) -> Result>, ManifestJoinError> { let bytes = parse_hex32(key) @@ -358,9 +367,9 @@ impl PlaneAdapter for CallgraphPlane { } } } -/// Why a checkout view could not supply a reader. `busy` marks a transient -/// SQLite lock on the view's database: the graph exists and the caller should -/// retry, which is reported as building rather than unavailable. +/// Why a checkout view could not supply a reader. `busy` marks transient +/// contention on a view database or its family ledger, so callers can retry +/// rather than reporting a permanently unavailable graph. #[derive(Clone, Debug)] pub(crate) struct CallgraphViewError { pub reason: String, @@ -376,6 +385,16 @@ impl From for CallgraphViewError { } } +impl From for CallgraphViewError { + fn from(error: PlaneError) -> Self { + Self { + busy: error.reason.starts_with(DERIVED_STORE_BUSY) + || error.reason.starts_with(V3_STORE_BUSY), + reason: error.to_string(), + } + } +} + impl From<&str> for CallgraphViewError { fn from(reason: &str) -> Self { reason.to_owned().into() @@ -391,6 +410,7 @@ impl std::fmt::Display for CallgraphViewError { /// Prefix of the plane error for a derived database that a SQLite lock kept /// from opening. pub(crate) const DERIVED_STORE_BUSY: &str = "callgraph derived store is busy"; +const V3_STORE_BUSY: &str = "callgraph v3 store is busy"; /// Supplies a checkout's verified call graph to one request, without /// exposing its storage layout. Implementations never publish v1/v2 data and @@ -457,7 +477,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource { let (mut snapshot, revision) = self .driver .verified_bind(&self.access, seed) - .map_err(|error| error.to_string())?; + .map_err(CallgraphViewError::from)?; if has_unpublished_changes(&snapshot) { // Publish the edits as a new worker generation (a donor clone // patched with the diff) before the single install, as a fresh @@ -471,7 +491,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource { &snapshot, super::first_load::DERIVED_CALLGRAPH_SEEDING, ) - .map_err(|error| error.to_string())?; + .map_err(CallgraphViewError::from)?; let cut = snapshot .install_cut() .cloned() @@ -492,7 +512,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource { } self.driver .install(&self.access, &snapshot, revision) - .map_err(|error| error.to_string())?; + .map_err(CallgraphViewError::from)?; // Read the installed cut directly: `installed_snapshot` would mark a // checkout without a steady watcher subscription overflowed at once. // That mark only decides how the next bind reconciles; this cut was @@ -528,10 +548,7 @@ impl CallgraphReaderSource for CheckoutCallgraphSource { self.bridge .adapter .open_generation(&self.access, snapshot.generation()) - .map_err(|error| CallgraphViewError { - busy: error.reason.starts_with(DERIVED_STORE_BUSY), - reason: error.to_string(), - })?; + .map_err(CallgraphViewError::from)?; self.bridge .adapter .reader(&self.access, &snapshot) @@ -761,6 +778,51 @@ mod tests { super::super::donor::events_under(dir) } + #[test] + fn v3_checkpoint_contention_reaches_checkout_reader_as_busy() { + use crate::blob_store::v3; + use crate::db::file_identity::IdentityConnection; + let _serial = v3::serial_for_test(); + let temp = tempfile::tempdir().unwrap(); + let storage = temp.path().join("storage"); + let root = checkout(&temp.path().join("checkout")); + let graph = view(&root, &storage); + assert_eq!(callers(&graph.reader().unwrap(), "a.ts", "target").len(), 1); + let plane = v3::plane_path(&storage, "donor-reuse", FamilyPlane::Callgraph).unwrap(); + let blocker = IdentityConnection::open_with_flags( + &plane, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + "callgraph-v3-checkpoint-blocker", + ) + .unwrap(); + // Pin a real WAL snapshot before the next publication appends frames. + blocker + .execute_batch("BEGIN; SELECT * FROM sqlite_schema;") + .unwrap(); + let added = root.join("c.ts"); + std::fs::write( + &added, + "import { caller } from './b';\nexport function top() { return caller(); }\n", + ) + .unwrap(); + graph.record_change(&added); + let error = match graph.reader() { + Ok(_) => panic!("a pinned WAL reader should prevent the truncating checkpoint"), + Err(error) => error, + }; + assert!(error.busy, "{error}"); + assert!(error.reason.contains(V3_STORE_BUSY), "{error}"); + assert!(error.reason.contains("checkpoint WAL"), "{error}"); + assert!(error.reason.contains("callgraph.sqlite"), "{error}"); + assert!( + error.reason.contains("wal_checkpoint(TRUNCATE): busy=1"), + "{error}" + ); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + assert_eq!(callers(&graph.reader().unwrap(), "b.ts", "caller").len(), 1); + } + #[test] fn sealed_base_donor_serves_checkouts_without_reextraction_or_cold_builds() { use super::super::donor::DonorEvent; diff --git a/crates/aft/src/views/first_load.rs b/crates/aft/src/views/first_load.rs index fa339168b..ea0ff969f 100644 --- a/crates/aft/src/views/first_load.rs +++ b/crates/aft/src/views/first_load.rs @@ -4719,7 +4719,7 @@ impl CompositePlane for CallgraphBridge { } Ok(()) }) - .map_err(|e| error(e.to_string()))?; + .map_err(super::callgraph::v3_error)?; } for (_, key) in &keys { if !v3_readers.contains(key) { @@ -5012,7 +5012,7 @@ impl CompositePlane for CallgraphBridge { } Ok(()) }) - .map_err(|e| error(e.to_string()))?; + .map_err(super::callgraph::v3_error)?; } for (path, key, _) in &payloads { if let Some(super::manifest_v2::EntryV2::Regular { @@ -5039,7 +5039,7 @@ impl CompositePlane for CallgraphBridge { &route.family, FamilyPlane::Callgraph, ) - .map_err(|e| error(e.to_string()))?, + .map_err(super::callgraph::v3_error)?, ); let dir = route.member_dir(); // Open (and so initialize and recover) the member's view store before From 8fa2b982bf2732d63cabb41c383089c4cd58de0c Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:21:49 +0100 Subject: [PATCH 03/23] mason: make ledger diagnostics regressions filesystem independent A corrupt header is refused as schema drift by format validation before SQLite reads it. Assert that preserved cause. Create a distinct derived SQLite generation per measurement handshake so Unix preallocation cannot make a growing database's allocated-byte charge temporarily constant. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/tests.rs | 23 +++++++++++------------ 1 file changed, 11 insertions(+), 12 deletions(-) diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index ad1892f38..c446a96ec 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -188,23 +188,22 @@ fn changing_derived_database_reports_busy_then_measurement_recovers() { seed(root.path(), A); let derived_dir = view_dir(root.path(), A, "churning-view").unwrap(); fs::create_dir_all(&derived_dir).unwrap(); - let derived = IdentityConnection::open( - derived_dir.join("derived-generation.sqlite"), - "v3-measurement-churn", - ) - .unwrap(); - derived - .execute_batch("CREATE TABLE growing(payload BLOB)") - .unwrap(); let lock = ReservationLock::acquire(root.path()).unwrap(); let mut writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap(); let writes = std::rc::Rc::new(std::cell::Cell::new(0)); let count = writes.clone(); - // A derived graph can commit outside the root reservation lock. Place one - // real SQLite commit between each inventory walk and its confirmation. + // A derived graph can commit outside the root reservation lock. Create a + // new generation between each inventory walk and its confirmation, so + // growth does not depend on the filesystem's allocation granularity. RECHECK_HOOK.with(|hook| { *hook.borrow_mut() = Some(Box::new(move || { - derived.execute("INSERT INTO growing VALUES(zeroblob(16384))", [])?; + let derived = IdentityConnection::open( + derived_dir.join(format!("derived-{}.sqlite", count.get())), + "v3-measurement-churn", + )?; + derived.execute_batch( + "CREATE TABLE growing(payload BLOB); INSERT INTO growing VALUES(zeroblob(16384));", + )?; count.set(count.get() + 1); Ok(()) })) @@ -332,7 +331,7 @@ fn unreadable_peer_ledger_refuses_instead_of_ignoring_reserved_bytes() { "{detail}" ); assert!(detail.contains("members.sqlite"), "{detail}"); - assert!(detail.contains("file is not a database"), "{detail}"); + assert!(detail.contains("schema_drift"), "{detail}"); assert!(!db(root.path(), B).exists()); } From 9536adb84cf04c861731680f89e4174215288680 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 20:56:55 +0100 Subject: [PATCH 04/23] mason: retain busy ledgers across callgraph driver setup The second historical Windows failure was not reproduced. Its loader/driver error spelling reveals an additional loss of classification: base-reference errors pass through registry I/O and driver strings, and view construction previously returned String. Preserve a stable busy prefix before the typed v3 error is erased, recognize the driver wrapper, and return CallgraphViewError from construction instead. Test real SQLite contention through the ledger/driver wrapping and view construction, retaining permanent refusal behavior and verifying retry succeeds after the blocker closes. The readiness helper retains its 60-second cooperative deadline; no assertions or deadline were weakened. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3.rs | 3 +- crates/aft/src/views/base_refs.rs | 1 + crates/aft/src/views/callgraph.rs | 116 ++++++++++++++++++++++++++++-- 3 files changed, 115 insertions(+), 5 deletions(-) diff --git a/crates/aft/src/blob_store/v3.rs b/crates/aft/src/blob_store/v3.rs index f70a628a6..315a94cb9 100644 --- a/crates/aft/src/blob_store/v3.rs +++ b/crates/aft/src/blob_store/v3.rs @@ -29,6 +29,7 @@ pub const NAMESPACE_BUDGET: u64 = 4 * 1024 * 1024 * 1024; /// ages them out by epoch instead of by marking. pub const SYMBOL_RETENTION_SWEEPS: i64 = 8; const TIMEOUT_MS: i64 = 5_000; +pub(crate) const BUSY_PREFIX: &str = "v3 store busy: "; mod connections; mod copy; mod initialization; @@ -172,7 +173,7 @@ impl fmt::Display for Error { Self::Sqlite(e) => write!(f, "{e}"), Self::Incompatible(e) => write!(f, "{e}"), Self::Refused(reason) => f.write_str(reason), - Self::Busy(reason) => write!(f, "v3 store busy: {reason}"), + Self::Busy(reason) => write!(f, "{BUSY_PREFIX}{reason}"), Self::Ledger { path, operation, diff --git a/crates/aft/src/views/base_refs.rs b/crates/aft/src/views/base_refs.rs index 244eaf3da..0bf640a1b 100644 --- a/crates/aft/src/views/base_refs.rs +++ b/crates/aft/src/views/base_refs.rs @@ -81,6 +81,7 @@ pub enum BaseRefError { impl fmt::Display for BaseRefError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Self::Store(error) if error.is_busy() => write!(f, "{}{error}", v3::BUSY_PREFIX), Self::Store(error) => write!(f, "v3 store: {error}"), Self::View(error) => write!(f, "{error}"), Self::Io(error) => write!(f, "{error}"), diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs index 0a9395cbc..945d340dd 100644 --- a/crates/aft/src/views/callgraph.rs +++ b/crates/aft/src/views/callgraph.rs @@ -387,9 +387,16 @@ impl From for CallgraphViewError { impl From for CallgraphViewError { fn from(error: PlaneError) -> Self { + // Base-reference errors enter through the loader's driver stage, + // whereas plane materializers return their reason without a stage. + let reason = error + .reason + .strip_prefix("[driver]: ") + .unwrap_or(&error.reason); Self { - busy: error.reason.starts_with(DERIVED_STORE_BUSY) - || error.reason.starts_with(V3_STORE_BUSY), + busy: reason.starts_with(DERIVED_STORE_BUSY) + || reason.starts_with(V3_STORE_BUSY) + || reason.starts_with(crate::blob_store::v3::BUSY_PREFIX), reason: error.to_string(), } } @@ -613,7 +620,7 @@ impl CallgraphView { storage: crate::root_cache::StoragePaths, family: String, policy: crate::blob_store::v2::TrigramPolicy, - ) -> Result { + ) -> Result { let scope = view_scope(&root); let producers = super::manifest_v2::Producers { trigram: policy.fingerprint_hex(), @@ -636,7 +643,7 @@ impl CallgraphView { Arc::new(super::first_load::ConfiguredMembershipWalker), vec![trigram, bridge.clone()], ) - .map_err(|error| error.to_string())? + .map_err(CallgraphViewError::from)? .with_adapters(vec![adapter]), ); driver.register_write_intent(); @@ -823,6 +830,107 @@ mod tests { assert_eq!(callers(&graph.reader().unwrap(), "b.ts", "caller").len(), 1); } + #[test] + fn v3_busy_ledger_survives_driver_error_wrapping() { + use crate::blob_store::v3; + use crate::db::file_identity::IdentityConnection; + let _serial = v3::serial_for_test(); + let temp = tempfile::tempdir().unwrap(); + let family = "driver-busy"; + v3::publish(temp.path(), family, 1024, None, |_| Ok(())).unwrap(); + let path = v3::family_dir(temp.path(), family) + .unwrap() + .join("members.sqlite"); + let lock = v3::ReservationLock::acquire(temp.path()).unwrap(); + let mut writer = v3::RegistryWriter::open_for_barrier(temp.path(), family, &lock).unwrap(); + let blocker = IdentityConnection::open_with_flags( + &path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_ONLY, + "callgraph-v3-ledger-blocker", + ) + .unwrap(); + blocker + .execute_batch("BEGIN; SELECT * FROM accounting;") + .unwrap(); + writer + .with_barrier(|tx| { + tx.execute("UPDATE accounting SET watermark=watermark+1", [])?; + Ok(()) + }) + .unwrap(); + let ledger_error = writer.covering_measurement().unwrap_err(); + let driver_error = |error| { + PlaneError::loader( + super::super::contracts::LoaderStage::Driver, + super::super::base_refs::BaseRefError::from(error).to_string(), + ) + }; + let error = CallgraphViewError::from(driver_error(ledger_error)); + assert!(error.busy, "{error}"); + assert!( + error + .reason + .starts_with("view loader [driver]: v3 store busy: "), + "{error}" + ); + assert!(error.reason.contains("members.sqlite"), "{error}"); + assert!( + error.reason.contains("wal_checkpoint(TRUNCATE): busy=1"), + "{error}" + ); + let permanent = CallgraphViewError::from(driver_error(v3::Error::Refused( + "reservation_ledger_unreadable", + ))); + assert!(!permanent.busy, "{permanent}"); + blocker.execute_batch("ROLLBACK").unwrap(); + writer.covering_measurement().unwrap(); + } + + #[test] + fn v3_busy_ledger_during_view_construction_is_retryable() { + use crate::blob_store::v3; + use crate::db::file_identity::IdentityConnection; + let _serial = v3::serial_for_test(); + let temp = tempfile::tempdir().unwrap(); + let storage = temp.path().join("storage"); + let root = checkout(&temp.path().join("checkout")); + v3::publish(&storage, "donor-reuse", 1024, None, |_| Ok(())).unwrap(); + let path = v3::family_dir(&storage, "donor-reuse") + .unwrap() + .join("members.sqlite"); + let blocker = IdentityConnection::open_with_flags( + &path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE, + "callgraph-v3-open-blocker", + ) + .unwrap(); + // SQLite's exclusive locking mode prevents a second connection from + // opening the ledger until this transaction and connection are gone. + blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap(); + let construct = || { + CallgraphView::new( + root.clone(), + crate::root_cache::StoragePaths::same(&storage), + "donor-reuse".into(), + crate::blob_store::v2::TrigramPolicy { + max_file_size: 1 << 20, + }, + ) + }; + let error = match construct() { + Ok(_) => panic!("view construction must not open an exclusively held ledger"), + Err(error) => error, + }; + assert!(error.busy, "{error}"); + assert!(error.reason.contains("v3 store busy: "), "{error}"); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + assert_eq!( + callers(&construct().unwrap().reader().unwrap(), "a.ts", "target").len(), + 1 + ); + } + #[test] fn sealed_base_donor_serves_checkouts_without_reextraction_or_cold_builds() { use super::super::donor::DonorEvent; From 9ada0ed35c0dd4ed71315894777c6a27b4a37415 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:09:53 +0100 Subject: [PATCH 05/23] mason: preserve SQLite busy during v3 format validation The exclusive-ledger construction regression ran red on both native Windows and Linux: initialization and validation swallowed SQLITE_BUSY and reported schema drift. Preserve contention at each lossy format/seed read while leaving all non-contention malformed/incomplete-schema decisions unchanged. Record the failed read, ledger path and SQLite cause. Also account for the family-registry I/O wrapper before classifying driver errors. Add read-only validation coverage, including a successful reopen after the exclusive connection closes. These deterministic findings do not identify the cause of either historical CI incident. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3.rs | 55 ++++++++++++++----- .../aft/src/blob_store/v3/initialization.rs | 43 +++++++++------ crates/aft/src/blob_store/v3/tests.rs | 37 +++++++++++++ crates/aft/src/views/callgraph.rs | 3 + 4 files changed, 109 insertions(+), 29 deletions(-) diff --git a/crates/aft/src/blob_store/v3.rs b/crates/aft/src/blob_store/v3.rs index 315a94cb9..12f784615 100644 --- a/crates/aft/src/blob_store/v3.rs +++ b/crates/aft/src/blob_store/v3.rs @@ -540,25 +540,50 @@ fn schema_rows(c: &Connection) -> rusqlite::Result> { c.prepare("SELECT type,name,tbl_name,sql FROM sqlite_schema WHERE substr(name,1,7) != 'sqlite_' ORDER BY type,name,tbl_name,sql")? .query_map([], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)))?.collect() } +fn format_read( + result: rusqlite::Result, + path: &Path, + operation: &'static str, +) -> Result> { + match result { + Ok(value) => Ok(Some(value)), + Err(error) => { + let error = Error::Sqlite(error); + // A locked format table is not evidence of a missing or corrupt + // schema. Other read failures retain the format-drift checks. + if error.is_busy() { + Err(Error::ledger(path, operation, error)) + } else { + Ok(None) + } + } + } +} fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> { #[cfg(test)] let _profile = work_profile::span("validate_database"); - let version = c - .query_row( + let version = format_read( + c.query_row( "SELECT format_version FROM format_meta WHERE singleton=1", [], |r| r.get::<_, u64>(0), - ) - .unwrap_or(0); + ), + path, + "read format version", + )? + .unwrap_or(0); let displayed = fs::canonicalize(path).unwrap_or_else(|_| path.to_path_buf()); persisted_format::gate(kind.store(), &displayed, path, Some(version)) .map_err(Error::Incompatible)?; if version == 0 { return Err(incompatible(kind, path, 0, IncompatibleReason::SchemaDrift)); } - let app = c - .query_row("PRAGMA application_id", [], |r| r.get::<_, i64>(0)) - .unwrap_or(0); + let app = format_read( + c.query_row("PRAGMA application_id", [], |r| r.get::<_, i64>(0)), + path, + "read application id", + )? + .unwrap_or(0); if app != kind.app_id() { return Err(incompatible( kind, @@ -567,13 +592,16 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> { IncompatibleReason::ApplicationId, )); } - let (schema, layout, text): (u32, u32, String) = c - .query_row( + let (schema, layout, text): (u32, u32, String) = format_read( + c.query_row( "SELECT schema_version,layout_tag,schema_text FROM format_meta WHERE singleton=1", [], |r| Ok((r.get(0)?, r.get(1)?, r.get(2)?)), - ) - .map_err(|_| incompatible(kind, path, version, IncompatibleReason::SchemaDrift))?; + ), + path, + "read format schema", + )? + .ok_or_else(|| incompatible(kind, path, version, IncompatibleReason::SchemaDrift))?; let compiled_schema = match (kind, version, schema) { // A registry of the base-reference protocol carries its own exact // text; any other version/schema pair falls through to the drift check. @@ -610,8 +638,9 @@ fn validate(c: &Connection, path: &Path, kind: Kind) -> Result<()> { }; if !known_version || text != format!("{META}{compiled_schema}") - || schema_rows(c).ok() != Some(schema_rows(&expected)?) - || !initialization::has_seed(c, kind) + || format_read(schema_rows(c), path, "read schema objects")? + != Some(schema_rows(&expected)?) + || !format_read(initialization::has_seed(c, kind), path, "read seed")?.unwrap_or(false) { return Err(incompatible( kind, diff --git a/crates/aft/src/blob_store/v3/initialization.rs b/crates/aft/src/blob_store/v3/initialization.rs index e72ed6a2a..f8d4f3032 100644 --- a/crates/aft/src/blob_store/v3/initialization.rs +++ b/crates/aft/src/blob_store/v3/initialization.rs @@ -4,32 +4,43 @@ use super::*; -fn has_marker(c: &Connection) -> bool { - c.query_row("SELECT 1 FROM format_meta WHERE singleton=1", [], |r| { - r.get::<_, i64>(0) - }) - .is_ok() +fn has_marker(c: &Connection, path: &Path) -> Result { + Ok(format_read( + c.query_row("SELECT 1 FROM format_meta WHERE singleton=1", [], |r| { + r.get::<_, i64>(0) + }), + path, + "read initialization marker", + )? + .is_some()) } -fn has_any_seed(c: &Connection, kind: Kind) -> bool { +fn has_any_seed(c: &Connection, path: &Path, kind: Kind) -> Result { let sql = match kind { Kind::Registry => "SELECT 1 FROM accounting WHERE singleton=1", Kind::Plane(_) => "SELECT 1 FROM store_meta WHERE singleton=1", }; - c.query_row(sql, [], |r| r.get::<_, i64>(0)).is_ok() + Ok(format_read( + c.query_row(sql, [], |r| r.get::<_, i64>(0)), + path, + "read initialization seed", + )? + .is_some()) } -pub(super) fn has_seed(c: &Connection, kind: Kind) -> bool { +pub(super) fn has_seed(c: &Connection, kind: Kind) -> rusqlite::Result { match kind { - Kind::Registry => has_any_seed(c, kind), + Kind::Registry => c.query_row("SELECT 1 FROM accounting WHERE singleton=1", [], |_| { + Ok(true) + }), Kind::Plane(plane) => c .query_row("SELECT plane FROM store_meta WHERE singleton=1", [], |r| { r.get::<_, String>(0) }) - .is_ok_and(|stored| stored == plane.as_str()), + .map(|stored| stored == plane.as_str()), } } -fn resumable(c: &Connection, kind: Kind) -> Result { - let Ok(rows) = schema_rows(c) else { +fn resumable(c: &Connection, path: &Path, kind: Kind) -> Result { + let Some(rows) = format_read(schema_rows(c), path, "read bootstrap schema")? else { return Ok(false); }; let app: i64 = c.pragma_query_value(None, "application_id", |r| r.get(0))?; @@ -121,10 +132,10 @@ fn resumable(c: &Connection, kind: Kind) -> Result { } pub(super) fn ensure(c: &mut IdentityConnection, path: &Path, kind: Kind) -> Result<()> { - if has_marker(c) && has_any_seed(c, kind) { + if has_marker(c, path)? && has_any_seed(c, path, kind)? { return Ok(()); } - if !resumable(c, kind)? { + if !resumable(c, path, kind)? { return Ok(()); } // SQLite must select pointer-map mode before the first write transaction. @@ -133,10 +144,10 @@ pub(super) fn ensure(c: &mut IdentityConnection, path: &Path, kind: Kind) -> Res #[cfg(test)] tests::initialization_hook(kind, "auto_vacuum"); let tx = c.transaction_with_behavior(TransactionBehavior::Immediate)?; - if has_marker(&tx) && has_any_seed(&tx, kind) { + if has_marker(&tx, path)? && has_any_seed(&tx, path, kind)? { return Ok(()); } - if !resumable(&tx, kind)? { + if !resumable(&tx, path, kind)? { return Ok(()); } // Old interrupted bootstraps can contain a prefix of the schema. Only diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index c446a96ec..6dc598a81 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -181,6 +181,43 @@ fn checkpoint_contention_is_busy_and_preserves_sqlite_cause() { assert_eq!(writer.accounting().unwrap().reserved_bytes, 0); } +#[test] +fn busy_validation_is_not_persisted_as_schema_drift() { + let _guard = serial(); + let root = tempdir().unwrap(); + seed(root.path(), A); + let path = db(root.path(), A); + let blocker = IdentityConnection::open_with_flags( + &path, + OpenFlags::SQLITE_OPEN_READ_WRITE, + "v3-validation-blocker", + ) + .unwrap(); + blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap(); + let error = match LedgerReader::open_existing(root.path(), A) { + Ok(_) => panic!("an exclusively held ledger must not validate through another connection"), + Err(error) => error, + }; + assert!(error.is_busy(), "{error}"); + let detail = error.to_string(); + assert!( + detail.starts_with("reservation_ledger_busy: read format version"), + "{detail}" + ); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!(detail.contains("database is locked"), "{detail}"); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + assert_eq!( + LedgerReader::open_existing(root.path(), A) + .unwrap() + .accounting() + .unwrap() + .reserved_bytes, + 0 + ); +} + #[test] fn changing_derived_database_reports_busy_then_measurement_recovers() { let _guard = serial(); diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs index 945d340dd..6bea9ce28 100644 --- a/crates/aft/src/views/callgraph.rs +++ b/crates/aft/src/views/callgraph.rs @@ -393,6 +393,9 @@ impl From for CallgraphViewError { .reason .strip_prefix("[driver]: ") .unwrap_or(&error.reason); + let reason = reason + .strip_prefix("family registry I/O error: ") + .unwrap_or(reason); Self { busy: reason.starts_with(DERIVED_STORE_BUSY) || reason.starts_with(V3_STORE_BUSY) From 92c1ad1b0b0ff12d74ecfff6af06ae5191511c4e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Wed, 28 Oct 2026 21:38:45 +0100 Subject: [PATCH 06/23] Search descriptor for the v3 ledger busy fix Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- .../slice-descriptors/land-v3-ledger-busy.json | 12 ++++++++++++ 1 file changed, 12 insertions(+) create mode 100644 benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json diff --git a/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json b/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json new file mode 100644 index 000000000..b790f460a --- /dev/null +++ b/benchmarks/aft-search/slice-descriptors/land-v3-ledger-busy.json @@ -0,0 +1,12 @@ +{ + "slice_class": "non_ranking", + "targeted_mechanism": "none", + "kind": "harness", + "fixtures": [ + "harness-goldens" + ], + "slices": [ + "v3-ledger-busy" + ], + "note": "Views v3 ledger: busy contention reported as busy/retry, not unreadable or schema drift. No changed path matches RANKING_FENCE_PREFIXES." +} From 465e989f1c27a069e34eaa6b649c2421f2eba09c Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 01:24:05 +0100 Subject: [PATCH 07/23] mason: treat base completion contention as building Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 49 +++++++------ .../aft/src/inspect/view_projection_tests.rs | 4 ++ crates/aft/src/views/callgraph.rs | 72 ++++++++++++++++++- 3 files changed, 101 insertions(+), 24 deletions(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index 54d3545fd..a86d2f37a 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4951,6 +4951,31 @@ pub enum CallgraphStoreAccess { Error(CallGraphStoreError), } +pub(crate) fn checkout_view_callgraph_access( + reader: Option< + Result, crate::views::callgraph::CallgraphViewError>, + >, +) -> CallgraphStoreAccess { + match reader { + Some(Ok(store)) => CallgraphStoreAccess::Ready(store), + // Transient contention while opening a view—including another driver + // finishing shared artifacts or a derived-database checkpoint—is + // retryable, not evidence that the graph is unavailable. + Some(Err(error)) if error.busy => CallgraphStoreAccess::Building, + Some(Err(error)) + if error + .reason + .contains(crate::views::read::CALLGRAPH_DISABLED) => + { + CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason)) + } + Some(Err(error)) => CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(format!( + "callgraph view unavailable: {error}" + ))), + None => CallgraphStoreAccess::Unavailable, + } +} + #[derive(Clone, Copy)] enum CallgraphBackgroundWork { Ensure, @@ -11055,29 +11080,7 @@ impl AppContext { if !demand { return CallgraphStoreAccess::Building; } - return match self.callgraph_view_reader() { - Some(Ok(store)) => CallgraphStoreAccess::Ready(store), - // A lock held on the view's database says nothing about the - // graph: report SQLite's busy error, which callers present as - // building, retry shortly. - Some(Err(error)) if error.busy => CallgraphStoreAccess::Error( - CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure( - rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY), - Some(error.reason), - )), - ), - Some(Err(error)) - if error - .reason - .contains(crate::views::read::CALLGRAPH_DISABLED) => - { - CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason)) - } - Some(Err(error)) => CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable( - format!("callgraph view unavailable: {error}"), - )), - None => CallgraphStoreAccess::Unavailable, - }; + return checkout_view_callgraph_access(self.callgraph_view_reader()); } if self.config().views.enabled && self.config().indexes.callgraph { let on_demand = self.callgraph_build_on_demand(); diff --git a/crates/aft/src/inspect/view_projection_tests.rs b/crates/aft/src/inspect/view_projection_tests.rs index 14c988110..8db2f9059 100644 --- a/crates/aft/src/inspect/view_projection_tests.rs +++ b/crates/aft/src/inspect/view_projection_tests.rs @@ -770,6 +770,7 @@ fn views_tier2_pending_plane_never_falls_back_to_legacy() { #[test] fn views_tier2_and_legacy_dead_code_verdicts_match() { let (_project, _storage, mut job, request) = view_projection_fixture(); + let view = crate::views::ViewStore::open(&request.storage, &request.scope).unwrap(); for source in [ "export function used() {}\nexport function dead() {}\n", "export function used() { dead(); }\nexport function dead() {}\n", @@ -779,6 +780,9 @@ fn views_tier2_and_legacy_dead_code_verdicts_match() { ] { write_projection_cache_file(&job.project_root.join("target.ts"), source); crate::views::assembly::publish_checkout(&request).unwrap(); + // Publication makes the immutable generation current before its + // detached checkpoint closes the last writer; wait for readability. + crate::views::wait_for_derived_checkpoint_for_test(view.view_dir()); Arc::make_mut(&mut job.config).views.enabled = true; let view = build_tier2_callgraph_snapshot_with_refresh(&job, true, &job.scope_files).unwrap(); diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs index 6bea9ce28..680ce959e 100644 --- a/crates/aft/src/views/callgraph.rs +++ b/crates/aft/src/views/callgraph.rs @@ -399,7 +399,8 @@ impl From for CallgraphViewError { Self { busy: reason.starts_with(DERIVED_STORE_BUSY) || reason.starts_with(V3_STORE_BUSY) - || reason.starts_with(crate::blob_store::v3::BUSY_PREFIX), + || reason.starts_with(crate::blob_store::v3::BUSY_PREFIX) + || reason.starts_with(V3_BASE_COMPLETION_BUSY), reason: error.to_string(), } } @@ -421,6 +422,7 @@ impl std::fmt::Display for CallgraphViewError { /// from opening. pub(crate) const DERIVED_STORE_BUSY: &str = "callgraph derived store is busy"; const V3_STORE_BUSY: &str = "callgraph v3 store is busy"; +const V3_BASE_COMPLETION_BUSY: &str = "v3 base plane completion is owned elsewhere"; /// Supplies a checkout's verified call graph to one request, without /// exposing its storage layout. Implementations never publish v1/v2 data and @@ -934,6 +936,74 @@ mod tests { ); } + #[test] + fn v3_base_completion_owned_by_another_driver_answers_building_then_ready() { + use super::super::base_builder::{build_base, BaseBuild, BaseBuildInput}; + use super::super::eviction::DiskBudget; + + let _serial = crate::blob_store::v3::serial_for_test(); + let temp = tempfile::tempdir().unwrap(); + let storage = temp.path().join("storage"); + let root = checkout(&temp.path().join("checkout")); + let paths: Vec<_> = FILES + .iter() + .map(|(name, _)| RelPath::new(name.as_bytes().to_vec()).unwrap()) + .collect(); + let BaseBuild::Published { claim, .. } = build_base( + BaseBuildInput { + storage: &storage, + family: "donor-reuse", + root: &root, + paths: &paths, + policy: crate::blob_store::v2::TrigramPolicy { + max_file_size: 1 << 20, + }, + semantic_producer: None, + callgraph_producer: PRODUCER.into(), + ignore_fingerprint: None, + history: None, + budget: DiskBudget::default(), + }, + None, + ) + .unwrap() else { + panic!("the fixture must own its fresh base") + }; + let outcome = + super::super::base_builder::claim_base(&storage, "donor-reuse", claim.scope(), None) + .unwrap(); + let owner_pid = match &outcome { + super::super::base_builder::ClaimOutcome::Busy { owner_pid, .. } => *owner_pid, + _ => panic!("the live base owner must fence another completion driver"), + }; + assert_eq!(owner_pid, std::process::id()); + let error = CallgraphViewError::from(PlaneError::loader( + super::super::contracts::LoaderStage::Driver, + format!("{V3_BASE_COMPLETION_BUSY}: {outcome:?}"), + )); + assert!(error.busy, "{error}"); + assert!(error.reason.contains(V3_BASE_COMPLETION_BUSY), "{error}"); + assert!( + error + .reason + .contains(&format!("owner_pid: {}", std::process::id())), + "{error}" + ); + assert!(matches!( + crate::context::checkout_view_callgraph_access(Some(Err(error))), + crate::context::CallgraphStoreAccess::Building + )); + + claim.release().unwrap(); + let graph = view(&root, &storage); + let crate::context::CallgraphStoreAccess::Ready(ready) = + crate::context::checkout_view_callgraph_access(Some(graph.reader())) + else { + panic!("the completed base must serve a ready call graph") + }; + assert_eq!(callers(&ready, "a.ts", "target").len(), 1); + } + #[test] fn sealed_base_donor_serves_checkouts_without_reextraction_or_cold_builds() { use super::super::donor::DonorEvent; From 1e4cd5d85087577245acd655018894770af3b25d Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:04:11 +0100 Subject: [PATCH 08/23] Keep reporting a busy views call graph as SQLite busy Mapping view contention to Building changed agent-visible answers: callers lost "persisted callgraph store is busy" for a building message, and inspect reported the view as not ready instead of pending. The busy flag on base-completion contention is the actual fix; the access keeps its SQLite busy form, which every caller already treats as retryable. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 14 ++++++++++---- crates/aft/src/views/callgraph.rs | 3 ++- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index a86d2f37a..fe64b2af9 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4958,10 +4958,16 @@ pub(crate) fn checkout_view_callgraph_access( ) -> CallgraphStoreAccess { match reader { Some(Ok(store)) => CallgraphStoreAccess::Ready(store), - // Transient contention while opening a view—including another driver - // finishing shared artifacts or a derived-database checkpoint—is - // retryable, not evidence that the graph is unavailable. - Some(Err(error)) if error.busy => CallgraphStoreAccess::Building, + // Transient contention while opening a view (a locked derived + // database, or another driver finishing the shared base) says nothing + // about the graph. Report SQLite's busy error, which every caller + // already presents as busy/pending with a retry. + Some(Err(error)) if error.busy => CallgraphStoreAccess::Error( + CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure( + rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY), + Some(error.reason), + )), + ), Some(Err(error)) if error .reason diff --git a/crates/aft/src/views/callgraph.rs b/crates/aft/src/views/callgraph.rs index 680ce959e..8c5a1188c 100644 --- a/crates/aft/src/views/callgraph.rs +++ b/crates/aft/src/views/callgraph.rs @@ -991,7 +991,8 @@ mod tests { ); assert!(matches!( crate::context::checkout_view_callgraph_access(Some(Err(error))), - crate::context::CallgraphStoreAccess::Building + crate::context::CallgraphStoreAccess::Error(error) + if error.is_transient_lock_contention() )); claim.release().unwrap(); From 7bf5d343e99f69f9141f4d7337f1a81fb4e817e1 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:18:19 +0100 Subject: [PATCH 09/23] Format the busy views call-graph mapping Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index fe64b2af9..d1cfd0e4b 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4962,12 +4962,12 @@ pub(crate) fn checkout_view_callgraph_access( // database, or another driver finishing the shared base) says nothing // about the graph. Report SQLite's busy error, which every caller // already presents as busy/pending with a retry. - Some(Err(error)) if error.busy => CallgraphStoreAccess::Error( - CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure( + Some(Err(error)) if error.busy => CallgraphStoreAccess::Error(CallGraphStoreError::Sqlite( + rusqlite::Error::SqliteFailure( rusqlite::ffi::Error::new(rusqlite::ffi::SQLITE_BUSY), Some(error.reason), - )), - ), + ), + )), Some(Err(error)) if error .reason From 479ab4ba508ba7740721273e56f2ef0b4a3a874d Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 01:57:52 +0100 Subject: [PATCH 10/23] mason: select native Windows test targets and features Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- scripts/windows-gate.sh | 27 ++++++++++++++++++++++++--- 1 file changed, 24 insertions(+), 3 deletions(-) diff --git a/scripts/windows-gate.sh b/scripts/windows-gate.sh index 28788f2b2..1033d22a8 100755 --- a/scripts/windows-gate.sh +++ b/scripts/windows-gate.sh @@ -18,6 +18,8 @@ base=origin/main full=0 filter= filter_set=0 +test_target= +features= timeout_minutes=60 show_status=0 plan_only=0 @@ -30,12 +32,18 @@ while (($#)); do --filter) if (($# < 2)); then echo "--filter requires a test filter" >&2; exit 2; fi filter=$2; filter_set=1; shift 2 ;; + --test-target) + if (($# < 2)) || [[ -z "$2" ]]; then echo "--test-target requires a target" >&2; exit 2; fi + test_target=$2; shift 2 ;; + --features) + if (($# < 2)) || [[ -z "$2" ]]; then echo "--features requires a feature list" >&2; exit 2; fi + features=$2; shift 2 ;; --timeout-minutes) if (($# < 2)); then echo "--timeout-minutes requires a positive integer" >&2; exit 2; fi timeout_minutes=$2; shift 2 ;; --status) show_status=1; shift ;; --plan) plan_only=1; shift ;; - --help|-h) echo 'Run with runon windows; options: --base, --full, --filter, --timeout-minutes, --status, --plan'; exit 0 ;; + --help|-h) echo 'Run with runon windows; options: --base, --full, --filter, --test-target, --features, --timeout-minutes, --status, --plan'; exit 0 ;; *) echo "Unknown argument: $1" >&2; exit 2 ;; esac done @@ -54,6 +62,10 @@ if ((filter_set)) && [[ -z "$filter" ]]; then echo "filter must not be empty" >&2 exit 2 fi +if ((full)) && [[ -n "$test_target" ]]; then + echo '--full and --test-target cannot be combined' >&2 + exit 2 +fi if ((show_status)); then printf 'Windows native test runner\nCargo target directory: %s\n' "${CARGO_TARGET_DIR:-$repo_root/target}" @@ -100,7 +112,9 @@ select_path() { esac } -if [[ -n "$filter" ]]; then +if [[ -n "$test_target" ]]; then + plan_label="explicit test target $test_target" +elif [[ -n "$filter" ]]; then plan_label='explicit filter' else base_sha="$(git rev-parse --verify "$base^{commit}")" @@ -129,6 +143,10 @@ if [[ -n "$filter" ]]; then add_job lib '' "$filter" if ((full)); then add_job test integration "$filter"; fi fi +if [[ -n "$test_target" ]]; then + job_kinds=(); job_targets=(); job_filters=() + add_job test "$test_target" "$filter" +fi selected_kinds=(); selected_targets=(); selected_filters=() for ((i = 0; i < ${#job_kinds[@]}; i++)); do @@ -158,12 +176,14 @@ fi print_job() { local kind=$1 target=$2 test_filter=$3 local -a args=(test --locked -j 4 -p agent-file-tools "--$kind") + [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$target" ]] || args+=("$target") [[ -z "$test_filter" ]] || args+=("$test_filter") args+=(-- --test-threads 4) printf ' cargo'; printf ' %q' "${args[@]}"; printf '\n' } -printf ' storage preflight: cargo test --locked -j 4 -p agent-file-tools --lib gate_hermeticity_tests::gate_resolves_every_user_config_and_state_path_under_the_gate_homes -- --test-threads 4\n' +printf ' storage preflight:\n' +print_job lib '' gate_hermeticity_tests::gate_resolves_every_user_config_and_state_path_under_the_gate_homes for ((i = 0; i < ${#job_kinds[@]}; i++)); do print_job "${job_kinds[$i]}" "${job_targets[$i]}" "${job_filters[$i]}"; done if ((plan_only)); then exit 0; fi @@ -218,6 +238,7 @@ started=$SECONDS run_job() { local kind=$1 target=$2 test_filter=$3 label=$4 remaining output code local -a args=(test --locked -j 4 -p agent-file-tools "--$kind") + [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$target" ]] || args+=("$target") [[ -z "$test_filter" ]] || args+=("$test_filter") args+=(-- --test-threads 4) From 0d2c3941c880f4a8e974082dd29f055d2222f171 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:05:51 +0100 Subject: [PATCH 11/23] mason: pin failed ledger handoff and correct native gate arguments Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/tests.rs | 38 +++++++++++++++++++++++++++ scripts/windows-gate.sh | 4 +-- 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index 6dc598a81..1e6805975 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -1162,6 +1162,44 @@ fn member_kind_constraint_and_v3_namespace_are_not_v2_registration() { .starts_with(family_dir(root.path(), A).unwrap())); } +#[test] +fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { + let _guard = serial(); + let root = tempdir().unwrap(); + seed(root.path(), A); + let path = db(root.path(), A); + let blocker = IdentityConnection::open_with_flags( + &path, + OpenFlags::SQLITE_OPEN_READ_WRITE, + "v3-upgrade-blocker", + ) + .unwrap(); + // Rollback mode permits an exclusive lock while the idle read-only handle + // is still open, making a failed connection handoff deterministic. + blocker.execute_batch("PRAGMA journal_mode=DELETE").unwrap(); + let ledger = LedgerReader::open_existing(root.path(), A).unwrap(); + blocker.execute_batch("BEGIN EXCLUSIVE").unwrap(); + let lock = ReservationLock::acquire(root.path()).unwrap(); + let error = match RegistryWriter::open_existing(root.path(), A, &lock) { + Ok(_) => panic!("writer upgrade must be blocked by the exclusive transaction"), + Err(error) => error, + }; + assert!(error.is_busy(), "{error}"); + let error = ledger.accounting().unwrap_err(); + assert!(error.is_busy(), "existing reader lost the busy cause: {error}"); + let detail = error.to_string(); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!(detail.contains("read format version"), "{detail}"); + assert!(detail.contains("database is locked"), "{detail}"); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + assert_eq!(ledger.accounting().unwrap().reserved_bytes, 0); + assert_eq!(crate::db::file_identity::open_connections(&path), 1); + let writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap(); + assert!(Arc::ptr_eq(&ledger.database, &writer.database)); + assert_eq!(crate::db::file_identity::open_connections(&path), 1); +} + #[test] fn read_only_first_owner_upgrades_without_a_second_live_connection() { let _g = serial(); diff --git a/scripts/windows-gate.sh b/scripts/windows-gate.sh index 1033d22a8..770a13fe0 100755 --- a/scripts/windows-gate.sh +++ b/scripts/windows-gate.sh @@ -176,8 +176,8 @@ fi print_job() { local kind=$1 target=$2 test_filter=$3 local -a args=(test --locked -j 4 -p agent-file-tools "--$kind") - [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$target" ]] || args+=("$target") + [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$test_filter" ]] || args+=("$test_filter") args+=(-- --test-threads 4) printf ' cargo'; printf ' %q' "${args[@]}"; printf '\n' @@ -238,8 +238,8 @@ started=$SECONDS run_job() { local kind=$1 target=$2 test_filter=$3 label=$4 remaining output code local -a args=(test --locked -j 4 -p agent-file-tools "--$kind") - [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$target" ]] || args+=("$target") + [[ -z "$features" ]] || args+=(--features "$features") [[ -z "$test_filter" ]] || args+=("$test_filter") args+=(-- --test-threads 4) remaining=$((timeout_minutes * 60 - (SECONDS - started))) From 565f985dd8e0cf2b8d4e0f494c952ee05d6c3b41 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:17:53 +0100 Subject: [PATCH 12/23] mason: report all view readiness gates and full wait errors Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/tests.rs | 5 +- .../tests/integration/branch_switch_test.rs | 191 ++++++++++++++++-- 2 files changed, 174 insertions(+), 22 deletions(-) diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index 1e6805975..cd501e047 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -1186,7 +1186,10 @@ fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { }; assert!(error.is_busy(), "{error}"); let error = ledger.accounting().unwrap_err(); - assert!(error.is_busy(), "existing reader lost the busy cause: {error}"); + assert!( + error.is_busy(), + "existing reader lost the busy cause: {error}" + ); let detail = error.to_string(); assert!(detail.contains("members.sqlite"), "{detail}"); assert!(detail.contains("read format version"), "{detail}"); diff --git a/crates/aft/tests/integration/branch_switch_test.rs b/crates/aft/tests/integration/branch_switch_test.rs index 6a3afbe73..dc0b75d17 100644 --- a/crates/aft/tests/integration/branch_switch_test.rs +++ b/crates/aft/tests/integration/branch_switch_test.rs @@ -1,4 +1,4 @@ -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::io::{Read, Write}; use std::net::{SocketAddr, TcpListener, TcpStream}; use std::path::{Path, PathBuf}; @@ -355,28 +355,132 @@ fn configure_context_with_views( ctx } +#[derive(Debug, Default)] +struct IndexWaitDiagnostics { + status: BTreeMap<&'static str, String>, + view_status: BTreeMap<&'static str, BTreeMap>, + last_error: BTreeMap<&'static str, String>, +} + +impl IndexWaitDiagnostics { + fn observe_views(&mut self, ctx: &AppContext) { + use aft::blob_store::v2::FamilyPlane; + use aft::views::readiness::PlaneState; + + if let Some(runtime) = ctx.checkout_semantic_runtime() { + if let Some(error) = runtime.unavailable_reason() { + self.last_error.insert("semantic", error); + } + let snapshot = runtime.installed(); + let membership = snapshot.membership(); + for (name, plane) in [ + ("callgraph", FamilyPlane::Callgraph), + ("semantic", FamilyPlane::Semantic), + ("lexical", FamilyPlane::Trigram), + ] { + let mut counts = BTreeMap::new(); + for path in membership.keys() { + let state = snapshot.plane_readiness(path, plane); + let status = match state { + Some(PlaneState::Ready { .. }) => "ready".to_owned(), + Some(PlaneState::Pending { reason }) => format!("pending: {reason:?}"), + Some(PlaneState::Failed { .. }) => { + self.last_error + .insert(name, format!("{path:?}: {state:#?}")); + "failed".to_owned() + } + None => "unset".to_owned(), + }; + *counts.entry(status).or_insert(0) += 1; + } + self.view_status.insert(name, counts); + } + } + } + + fn observe_lexical(&mut self, answer: &Value) { + self.status + .insert("lexical", answer["index_status"].to_string()); + if answer["success"] == false || !answer["index_read_gap"].is_null() { + self.last_error.insert("lexical", format!("{answer:#}")); + } + } + + fn report(&self, ctx: &AppContext) -> String { + // Diagnostics only sample passive state. Running another query here + // could wake a retry and hide the state that exhausted the deadline. + let health = ctx.try_health_snapshot(&ctx.canonical_cache_root()); + let planes: BTreeMap<_, _> = ["callgraph", "semantic", "lexical"] + .into_iter() + .map(|plane| { + ( + plane, + json!({ + "final_status": self.status.get(plane), + "view_readiness": self.view_status.get(plane), + "last_error": self.last_error.get(plane), + }), + ) + }) + .collect(); + format!( + "planes={:#}\nsemantic_slot={}\nlast_callgraph_reconcile={:#?}\nhealth={:#}", + json!(planes), + ctx.checkout_semantic().describe(), + ctx.last_callgraph_reconcile(), + json!(health), + ) + } +} + fn wait_until_ready(ctx: &AppContext) -> Arc { - let deadline = Instant::now() + ROW_DEADLINE; + wait_until_ready_observed(ctx, &mut IndexWaitDiagnostics::default()) +} + +fn wait_until_ready_observed( + ctx: &AppContext, + diagnostics: &mut IndexWaitDiagnostics, +) -> Arc { + let started = Instant::now(); + let deadline = started + ROW_DEADLINE; loop { aft::runtime_drain::drain_watcher_events(ctx); aft::runtime_drain::drain_search_index_events(ctx); aft::runtime_drain::drain_callgraph_store_events(ctx); aft::runtime_drain::drain_semantic_index_events(ctx); aft::runtime_drain::drain_semantic_refresh_events(ctx); - let callgraph = match ctx.callgraph_store_for_ops() { - CallgraphStoreAccess::Ready(store) => Some(store), - CallgraphStoreAccess::Building => None, + let mut failure = None; + let (callgraph_status, callgraph) = match ctx.callgraph_store_for_ops() { + CallgraphStoreAccess::Ready(store) => ("ready", Some(store)), + CallgraphStoreAccess::Building => ("building", None), CallgraphStoreAccess::Suspended(reason) => { - panic!("callgraph suspended while waiting: {reason:?}") + failure = Some(format!("callgraph suspended while waiting: {reason:?}")); + ("suspended", None) + } + CallgraphStoreAccess::Unavailable => { + failure = Some("callgraph unavailable while waiting".to_owned()); + ("unavailable", None) + } + CallgraphStoreAccess::Off => { + failure = Some("callgraph index off while waiting".to_owned()); + ("off", None) } - CallgraphStoreAccess::Unavailable => panic!("callgraph unavailable while waiting"), - CallgraphStoreAccess::Off => panic!("callgraph index off while waiting"), // The view reader opens without a busy wait, so a publication // checkpointing the derived database can briefly lock it. Commands // report that as a retryable `callgraph_building`; wait it out too. - CallgraphStoreAccess::Error(error) if error.is_transient_lock_contention() => None, - CallgraphStoreAccess::Error(error) => panic!("callgraph failed while waiting: {error}"), + CallgraphStoreAccess::Error(error) => { + diagnostics + .last_error + .insert("callgraph", format!("{error}\n{error:#?}")); + if !error.is_transient_lock_contention() { + failure = Some(format!("callgraph failed while waiting: {error}")); + } + ("error", None) + } }; + diagnostics + .status + .insert("callgraph", callgraph_status.to_owned()); // With views on, grep and glob read the checkout's lexical view on // demand and no legacy search index is ever built; their readiness // is checked by the grep answers the callers wait for. @@ -387,17 +491,42 @@ fn wait_until_ready(ctx: &AppContext) -> Arc 0 && absent["total_matches"] == 0 @@ -541,7 +678,7 @@ fn wait_for_views_branch(ctx: &AppContext, root: &Path, branch: char) { } assert!( Instant::now() < deadline, - "views branch {branch} did not converge: present={present:#} absent={absent:#} callers={caller_result:#}" + "views branch {branch} did not converge: present={present:#} absent={absent:#} callers={caller_result:#}\n{}", diagnostics.report(ctx) ); thread::sleep(Duration::from_millis(25)); } @@ -613,13 +750,20 @@ fn views_round_trip_reuses_semantic_blobs_without_embedding() { /// semantic file of the checkout as it is on disk. fn wait_for_semantic_view(ctx: &AppContext) { let deadline = Instant::now() + ROW_DEADLINE; + let mut diagnostics = IndexWaitDiagnostics::default(); loop { aft::runtime_drain::drain_watcher_events(ctx); let runtime = ctx.checkout_semantic_runtime(); let complete = runtime.as_ref().is_some_and(|runtime| { - runtime - .search(&[1.0, 0.0, 0.0], 1, &|_| true) - .is_ok_and(|answer| answer.complete()) + match runtime.search(&[1.0, 0.0, 0.0], 1, &|_| true) { + Ok(answer) => answer.complete(), + Err(error) => { + diagnostics + .last_error + .insert("semantic", format!("{error:#?}")); + false + } + } }); // A switch the watcher has not delivered yet looks complete for the // previous branch; the reconciled snapshot must match the disk too. @@ -640,12 +784,17 @@ fn wait_for_semantic_view(ctx: &AppContext) { .is_some_and(|state| state.is_pending()) }) }); + diagnostics + .status + .insert("semantic", format!("complete={complete} current={current}")); + diagnostics.observe_views(ctx); if complete && current { return; } assert!( Instant::now() < deadline, - "views-on semantic view never caught up" + "views-on semantic view never caught up\n{}", + diagnostics.report(ctx) ); thread::sleep(Duration::from_millis(25)); } From 8f831c305ee6cd7ba0b1e3311f7ddae2adf6db0e Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:32:34 +0100 Subject: [PATCH 13/23] mason: recover shared ledger readers after a failed writer upgrade Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/connections.rs | 32 ++++++++++---- crates/aft/src/blob_store/v3/tests.rs | 48 +++++++++++++++++---- 2 files changed, 63 insertions(+), 17 deletions(-) diff --git a/crates/aft/src/blob_store/v3/connections.rs b/crates/aft/src/blob_store/v3/connections.rs index 19308636c..03db0816f 100644 --- a/crates/aft/src/blob_store/v3/connections.rs +++ b/crates/aft/src/blob_store/v3/connections.rs @@ -12,7 +12,9 @@ enum Slot { connection: IdentityConnection, writable: bool, }, - Closed, + Closed { + route: PathBuf, + }, } pub(super) struct Database { @@ -27,7 +29,7 @@ impl Deref for Guard<'_> { fn deref(&self) -> &Self::Target { match &*self.0 { Slot::Open { connection, .. } => connection, - Slot::Closed => unreachable!("guard is created only for an open slot"), + Slot::Closed { .. } => unreachable!("guard is created only for an open slot"), } } } @@ -35,18 +37,30 @@ impl DerefMut for Guard<'_> { fn deref_mut(&mut self) -> &mut Self::Target { match &mut *self.0 { Slot::Open { connection, .. } => connection, - Slot::Closed => unreachable!("guard is created only for an open slot"), + Slot::Closed { .. } => unreachable!("guard is created only for an open slot"), } } } impl Database { pub(super) fn connection(&self) -> Result> { - let slot = self + let mut slot = self .slot .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if matches!(*slot, Slot::Closed) { - return Err(Error::Refused("reservation_ledger_unreadable")); + if let Slot::Closed { route } = &*slot { + if matches!((self.identity, identity_of(route)), (Some(expected), Some(actual)) if expected != actual) + { + return Err(Error::Refused("live_sqlite_file_replaced")); + } + // If replacing a read-only connection with a writable one fails, + // existing readers can reopen read-only on demand. A still-held + // SQLite lock returns its busy error; a later read retries after + // the lock is released. + let connection = open_db(route, self.kind, false, true)?; + *slot = Slot::Open { + connection, + writable: false, + }; } Ok(Guard(slot)) } @@ -55,7 +69,7 @@ impl Database { .slot .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - let reopen = matches!(*slot, Slot::Closed) + let reopen = matches!(*slot, Slot::Closed { .. }) || (!readonly && matches!( *slot, @@ -69,7 +83,9 @@ impl Database { // the family is exclusively held, before opening its replacement. // Existing read capabilities still refer to this Database object, // so later operations use its replacement connection. - *slot = Slot::Closed; + *slot = Slot::Closed { + route: route.to_path_buf(), + }; let connection = open_db(route, self.kind, create, readonly)?; *slot = Slot::Open { connection, diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index cd501e047..01326d698 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -1162,12 +1162,9 @@ fn member_kind_constraint_and_v3_namespace_are_not_v2_registration() { .starts_with(family_dir(root.path(), A).unwrap())); } -#[test] -fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { - let _guard = serial(); - let root = tempdir().unwrap(); - seed(root.path(), A); - let path = db(root.path(), A); +fn blocked_read_only_upgrade(root: &Path) -> (LedgerReader, IdentityConnection, ReservationLock) { + seed(root, A); + let path = db(root, A); let blocker = IdentityConnection::open_with_flags( &path, OpenFlags::SQLITE_OPEN_READ_WRITE, @@ -1177,14 +1174,23 @@ fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { // Rollback mode permits an exclusive lock while the idle read-only handle // is still open, making a failed connection handoff deterministic. blocker.execute_batch("PRAGMA journal_mode=DELETE").unwrap(); - let ledger = LedgerReader::open_existing(root.path(), A).unwrap(); + let ledger = LedgerReader::open_existing(root, A).unwrap(); blocker.execute_batch("BEGIN EXCLUSIVE").unwrap(); - let lock = ReservationLock::acquire(root.path()).unwrap(); - let error = match RegistryWriter::open_existing(root.path(), A, &lock) { + let lock = ReservationLock::acquire(root).unwrap(); + let error = match RegistryWriter::open_existing(root, A, &lock) { Ok(_) => panic!("writer upgrade must be blocked by the exclusive transaction"), Err(error) => error, }; assert!(error.is_busy(), "{error}"); + (ledger, blocker, lock) +} + +#[test] +fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { + let _guard = serial(); + let root = tempdir().unwrap(); + let path = db(root.path(), A); + let (ledger, blocker, lock) = blocked_read_only_upgrade(root.path()); let error = ledger.accounting().unwrap_err(); assert!( error.is_busy(), @@ -1197,12 +1203,36 @@ fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { blocker.execute_batch("ROLLBACK").unwrap(); drop(blocker); assert_eq!(ledger.accounting().unwrap().reserved_bytes, 0); + assert_eq!( + ledger + .read(|c| Ok(c.pragma_query_value(None, "query_only", |r| r.get::<_, i64>(0))?)) + .unwrap(), + 1 + ); assert_eq!(crate::db::file_identity::open_connections(&path), 1); let writer = RegistryWriter::open_existing(root.path(), A, &lock).unwrap(); assert!(Arc::ptr_eq(&ledger.database, &writer.database)); assert_eq!(crate::db::file_identity::open_connections(&path), 1); } +#[test] +fn failed_upgrade_reader_rejects_a_replaced_database() { + let _guard = serial(); + let root = tempdir().unwrap(); + let path = db(root.path(), A); + let (ledger, blocker, _lock) = blocked_read_only_upgrade(root.path()); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + let original = path.with_extension("original.sqlite"); + fs::rename(&path, &original).unwrap(); + fs::copy(&original, &path).unwrap(); + assert!(matches!( + ledger.accounting(), + Err(Error::Refused("live_sqlite_file_replaced")) + )); + assert_eq!(crate::db::file_identity::open_connections(&path), 0); +} + #[test] fn read_only_first_owner_upgrades_without_a_second_live_connection() { let _g = serial(); From 97dcaa39eb8d395633ca3d438c2e2036a8df87d8 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:37:04 +0100 Subject: [PATCH 14/23] mason: retain operation and path on reader recovery failures Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/connections.rs | 3 ++- crates/aft/src/blob_store/v3/tests.rs | 4 ++++ 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/aft/src/blob_store/v3/connections.rs b/crates/aft/src/blob_store/v3/connections.rs index 03db0816f..d61d496f0 100644 --- a/crates/aft/src/blob_store/v3/connections.rs +++ b/crates/aft/src/blob_store/v3/connections.rs @@ -56,7 +56,8 @@ impl Database { // existing readers can reopen read-only on demand. A still-held // SQLite lock returns its busy error; a later read retries after // the lock is released. - let connection = open_db(route, self.kind, false, true)?; + let connection = open_db(route, self.kind, false, true) + .map_err(|source| Error::ledger(route, "reopen shared reader", source))?; *slot = Slot::Open { connection, writable: false, diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index 01326d698..1416651b2 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -1197,6 +1197,10 @@ fn failed_read_only_upgrade_preserves_busy_and_existing_reader_recovers() { "existing reader lost the busy cause: {error}" ); let detail = error.to_string(); + assert!( + detail.starts_with("reservation_ledger_busy: reopen shared reader"), + "{detail}" + ); assert!(detail.contains("members.sqlite"), "{detail}"); assert!(detail.contains("read format version"), "{detail}"); assert!(detail.contains("database is locked"), "{detail}"); From 0f551c5dedb0ba6474f178d3059c5868bb9c33ba Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 02:52:43 +0100 Subject: [PATCH 15/23] mason: retain callgraph busy causes before the Building verdict Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 106 +++++++++++++++++- .../tests/integration/branch_switch_test.rs | 8 +- 2 files changed, 110 insertions(+), 4 deletions(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index d1cfd0e4b..24812b6ef 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4587,6 +4587,7 @@ pub struct AppContext { lexical_view: parking_lot::Mutex>>, lexical_routes: [AtomicU64; 3], callgraph_view: parking_lot::Mutex>>, + callgraph_view_read_diagnostics: parking_lot::Mutex, callgraph_view_routes: AtomicU64, /// Files whose symbols a views-routed root's prewarm reused from, or /// stored into, the family's v3 symbol plane. @@ -4982,6 +4983,16 @@ pub(crate) fn checkout_view_callgraph_access( } } +/// Acquisition diagnostics for the checkout callgraph view. Busy errors become +/// `Building` at the query boundary, but retain their full cause here. +#[derive(Debug, Clone, Default, Serialize)] +pub struct CallgraphViewReadDiagnostics { + pub attempts: u64, + pub last_attempt_ms: u64, + pub total_attempt_ms: u64, + pub last_error: Option, +} + #[derive(Clone, Copy)] enum CallgraphBackgroundWork { Ensure, @@ -5482,6 +5493,9 @@ impl AppContext { lexical_view: parking_lot::Mutex::new(None), lexical_routes: Default::default(), callgraph_view: parking_lot::Mutex::new(None), + callgraph_view_read_diagnostics: parking_lot::Mutex::new( + CallgraphViewReadDiagnostics::default(), + ), callgraph_view_routes: AtomicU64::new(0), symbol_view_routes: Arc::default(), checkout_semantic: Arc::default(), @@ -11072,7 +11086,13 @@ impl AppContext { crate::views::contracts::WaitOutcome::Installed(snapshot) | crate::views::contracts::WaitOutcome::TimedOut { snapshot, .. } => snapshot, }; - return match runtime.callgraph.reader(&runtime.access, snapshot) { + let started = Instant::now(); + let reader = runtime.callgraph.reader(&runtime.access, snapshot); + self.note_callgraph_view_read( + started.elapsed(), + reader.as_ref().err().map(ToString::to_string), + ); + return match reader { Ok(reader) => CallgraphStoreAccess::Ready(Arc::clone(&reader.store)), Err(error) if error.reason == crate::views::read::CALLGRAPH_DISABLED => { CallgraphStoreAccess::Error(CallGraphStoreError::Unavailable(error.reason)) @@ -11086,7 +11106,16 @@ impl AppContext { if !demand { return CallgraphStoreAccess::Building; } - return checkout_view_callgraph_access(self.callgraph_view_reader()); + let started = Instant::now(); + let reader = self.callgraph_view_reader(); + self.note_callgraph_view_read( + started.elapsed(), + reader + .as_ref() + .and_then(|reader| reader.as_ref().err()) + .map(ToString::to_string), + ); + return checkout_view_callgraph_access(reader); } if self.config().views.enabled && self.config().indexes.callgraph { let on_demand = self.callgraph_build_on_demand(); @@ -12278,6 +12307,23 @@ impl AppContext { )); } + /// Passive diagnostic state, including the last busy error hidden by a + /// `Building` access result. A successful retry does not erase that error. + pub fn callgraph_view_read_diagnostics(&self) -> CallgraphViewReadDiagnostics { + self.callgraph_view_read_diagnostics.lock().clone() + } + + fn note_callgraph_view_read(&self, elapsed: Duration, error: Option) { + let mut diagnostics = self.callgraph_view_read_diagnostics.lock(); + let elapsed_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX); + diagnostics.attempts = diagnostics.attempts.saturating_add(1); + diagnostics.last_attempt_ms = elapsed_ms; + diagnostics.total_attempt_ms = diagnostics.total_attempt_ms.saturating_add(elapsed_ms); + if error.is_some() { + diagnostics.last_error = error; + } + } + /// The call-graph reader of this checkout's view, acquired once per /// pinned request. `None` means views do not route call-graph reads; an /// acquisition error is final and never answered from another store. @@ -18617,6 +18663,62 @@ mod callgraph_store_for_ops_tests { store.sqlite_path().to_path_buf() } + #[test] + fn callgraph_view_diagnostics_preserve_busy_cause_and_retry_on_the_same_context() { + use crate::blob_store::v3; + use crate::db::file_identity::IdentityConnection; + + let _serial = v3::serial_for_test(); + let (_dirs, ctx) = cold_build_context(); + let mut config = (*ctx.config()).clone(); + config.views.enabled = true; + ctx.set_config(config); + let root = ctx.callgraph_project_root().unwrap(); + ctx.set_canonical_cache_root(root.clone()); + assert!(matches!( + ctx.callgraph_store_for_ops(), + CallgraphStoreAccess::Ready(_) + )); + ctx.callgraph_view.lock().take(); + + let storage = ctx.storage_paths().artifacts; + let family = ctx.memoized_artifact_cache_key(&root); + let path = v3::family_dir(&storage, &family) + .unwrap() + .join("members.sqlite"); + let blocker = IdentityConnection::open_with_flags( + &path, + rusqlite::OpenFlags::SQLITE_OPEN_READ_WRITE, + "callgraph-view-diagnostics-blocker", + ) + .unwrap(); + blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap(); + assert!(matches!( + ctx.callgraph_store_for_ops(), + CallgraphStoreAccess::Building + )); + let busy = ctx.callgraph_view_read_diagnostics(); + assert_eq!(busy.attempts, 2); + let detail = busy + .last_error + .as_deref() + .expect("Building must retain its busy cause"); + assert!(detail.contains("reservation_ledger_busy"), "{detail}"); + assert!(detail.contains("read format version"), "{detail}"); + assert!(detail.contains("members.sqlite"), "{detail}"); + assert!(detail.contains("database is locked"), "{detail}"); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + assert!(matches!( + ctx.callgraph_store_for_ops(), + CallgraphStoreAccess::Ready(_) + )); + let ready = ctx.callgraph_view_read_diagnostics(); + assert_eq!(ready.attempts, 3); + assert_eq!(ready.last_error, busy.last_error); + assert!(ready.total_attempt_ms >= ready.last_attempt_ms); + } + #[test] fn callgraph_busy_readiness_query_reports_building() { for read_only in [true, false] { diff --git a/crates/aft/tests/integration/branch_switch_test.rs b/crates/aft/tests/integration/branch_switch_test.rs index dc0b75d17..591a689de 100644 --- a/crates/aft/tests/integration/branch_switch_test.rs +++ b/crates/aft/tests/integration/branch_switch_test.rs @@ -367,6 +367,9 @@ impl IndexWaitDiagnostics { use aft::blob_store::v2::FamilyPlane; use aft::views::readiness::PlaneState; + if let Some(error) = ctx.callgraph_view_read_diagnostics().last_error { + self.last_error.insert("callgraph", error); + } if let Some(runtime) = ctx.checkout_semantic_runtime() { if let Some(error) = runtime.unavailable_reason() { self.last_error.insert("semantic", error); @@ -417,15 +420,16 @@ impl IndexWaitDiagnostics { plane, json!({ "final_status": self.status.get(plane), - "view_readiness": self.view_status.get(plane), + "semantic_member_readiness": self.view_status.get(plane), "last_error": self.last_error.get(plane), }), ) }) .collect(); format!( - "planes={:#}\nsemantic_slot={}\nlast_callgraph_reconcile={:#?}\nhealth={:#}", + "planes={:#}\ncallgraph_view_reads={:#}\nsemantic_slot={}\nlast_callgraph_reconcile={:#?}\nhealth={:#}", json!(planes), + json!(ctx.callgraph_view_read_diagnostics()), ctx.checkout_semantic().describe(), ctx.last_callgraph_reconcile(), json!(health), From ed2f434430bbe706999589dfe805c398df150f64 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:06:14 +0100 Subject: [PATCH 16/23] mason: assert the initialization-stage contention observed by the route The new route-level regression reaches the initialization-marker read, not format validation. Pin the actual SQLite stage while retaining its ledger path and source, and verify that the same context retries successfully after unlock. Final checks on this source tree: Linux diagnostic regression 1 passed, v3 failed_ regressions 3 passed, views 629 passed/14 ignored, watcher branch_switch 2 passed; native Windows diagnostic regression 1 passed, recovery/replacement regressions 2 passed, and three loaded whole-watcher runs each passed 35 tests. Windows GNU test compilation and cargo fmt --check passed. The pre-fix Windows load run reproduced the readiness timeout on attempt 2 at 60.1867 seconds; its busy cause was discarded before the original API exposed Building. The new passive diagnostics retain that cause and acquisition timings. The separate bare unreadable connection bug was reproduced deterministically before its fix. No timeout or existing assertion was weakened. Load-sibling failures are disclosed separately in the delivery: macOS manifest_incremental pin I/O fixtures, and the final Windows large_fill_size_gate_takes_cold_slot_before_source_preparation assertion. They were not silently rewritten or counted as passed. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index 24812b6ef..2692c927b 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -18704,7 +18704,7 @@ mod callgraph_store_for_ops_tests { .as_deref() .expect("Building must retain its busy cause"); assert!(detail.contains("reservation_ledger_busy"), "{detail}"); - assert!(detail.contains("read format version"), "{detail}"); + assert!(detail.contains("read initialization marker"), "{detail}"); assert!(detail.contains("members.sqlite"), "{detail}"); assert!(detail.contains("database is locked"), "{detail}"); blocker.execute_batch("ROLLBACK").unwrap(); From a61483ef52841331286fa894906d8ee4b4ee6881 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 03:45:55 +0100 Subject: [PATCH 17/23] mason: keep acquisition diagnostics on the SQLite busy contract Preserve the busy-to-SQLITE_BUSY mapping from the updated ledger branch. The route regression now asserts the retryable SQLite busy verdict while retaining the full acquisition error and attempt counts across a successful retry; the diagnostic comments no longer imply a Building conversion. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/context.rs | 26 ++++++++++++------- .../tests/integration/branch_switch_test.rs | 2 +- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/crates/aft/src/context.rs b/crates/aft/src/context.rs index 2692c927b..fd978f745 100644 --- a/crates/aft/src/context.rs +++ b/crates/aft/src/context.rs @@ -4983,8 +4983,8 @@ pub(crate) fn checkout_view_callgraph_access( } } -/// Acquisition diagnostics for the checkout callgraph view. Busy errors become -/// `Building` at the query boundary, but retain their full cause here. +/// Acquisition diagnostics for the checkout callgraph view, including the full +/// cause of retryable SQLite contention and time spent acquiring readers. #[derive(Debug, Clone, Default, Serialize)] pub struct CallgraphViewReadDiagnostics { pub attempts: u64, @@ -12307,8 +12307,8 @@ impl AppContext { )); } - /// Passive diagnostic state, including the last busy error hidden by a - /// `Building` access result. A successful retry does not erase that error. + /// Passive diagnostic state, including the last acquisition error. A + /// successful retry does not erase that error. pub fn callgraph_view_read_diagnostics(&self) -> CallgraphViewReadDiagnostics { self.callgraph_view_read_diagnostics.lock().clone() } @@ -18693,16 +18693,24 @@ mod callgraph_store_for_ops_tests { ) .unwrap(); blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap(); - assert!(matches!( - ctx.callgraph_store_for_ops(), - CallgraphStoreAccess::Building - )); + let CallgraphStoreAccess::Error(error) = ctx.callgraph_store_for_ops() else { + panic!("a locked view must retain the SQLite busy verdict"); + }; + assert!( + matches!( + &error, + CallGraphStoreError::Sqlite(rusqlite::Error::SqliteFailure(code, _)) + if code.extended_code == rusqlite::ffi::SQLITE_BUSY + ), + "{error}" + ); + assert!(error.is_transient_lock_contention(), "{error}"); let busy = ctx.callgraph_view_read_diagnostics(); assert_eq!(busy.attempts, 2); let detail = busy .last_error .as_deref() - .expect("Building must retain its busy cause"); + .expect("SQLite busy must retain its acquisition cause"); assert!(detail.contains("reservation_ledger_busy"), "{detail}"); assert!(detail.contains("read initialization marker"), "{detail}"); assert!(detail.contains("members.sqlite"), "{detail}"); diff --git a/crates/aft/tests/integration/branch_switch_test.rs b/crates/aft/tests/integration/branch_switch_test.rs index 591a689de..645a86f86 100644 --- a/crates/aft/tests/integration/branch_switch_test.rs +++ b/crates/aft/tests/integration/branch_switch_test.rs @@ -471,7 +471,7 @@ fn wait_until_ready_observed( } // The view reader opens without a busy wait, so a publication // checkpointing the derived database can briefly lock it. Commands - // report that as a retryable `callgraph_building`; wait it out too. + // expose a retryable SQLite busy error; wait it out too. CallgraphStoreAccess::Error(error) => { diagnostics .last_error From 0ffa792900d4dd4d6c3cfcb8e8ef404bcbecd0ba Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 04:59:00 +0100 Subject: [PATCH 18/23] Build the replaced-database test file through SQLite, not a file copy The v3 source fence forbids copying SQLite file-set bytes outside SQLite, even in tests. VACUUM INTO gives the same content a new file identity, which is what the replaced-database check needs. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/blob_store/v3/tests.rs | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/crates/aft/src/blob_store/v3/tests.rs b/crates/aft/src/blob_store/v3/tests.rs index 9a4a1c4b3..573cb43d7 100644 --- a/crates/aft/src/blob_store/v3/tests.rs +++ b/crates/aft/src/blob_store/v3/tests.rs @@ -1267,7 +1267,12 @@ fn failed_upgrade_reader_rejects_a_replaced_database() { drop(blocker); let original = path.with_extension("original.sqlite"); fs::rename(&path, &original).unwrap(); - fs::copy(&original, &path).unwrap(); + // Rebuild the file through SQLite itself: the source fence forbids copying + // file-set bytes, and VACUUM INTO gives the same content a new identity. + Connection::open(&original) + .unwrap() + .execute("VACUUM INTO ?1", [path.to_str().unwrap()]) + .unwrap(); assert!(matches!( ledger.accounting(), Err(Error::Refused("live_sqlite_file_replaced")) From 6fcb2e43e44a4fc30bf49472d62e840899fab95f Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 05:42:49 +0100 Subject: [PATCH 19/23] mason: pin child lexical publication recovery after busy startup Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/commands/configure.rs | 98 +++++++++++++++++++ .../aft/tests/integration/per_checkout_9.rs | 57 ++++++++++- 2 files changed, 150 insertions(+), 5 deletions(-) diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs index adb66a7dc..ff7de0906 100644 --- a/crates/aft/src/commands/configure.rs +++ b/crates/aft/src/commands/configure.rs @@ -17024,6 +17024,104 @@ mod tests { lock } + #[test] + fn busy_lexical_bind_retries_until_parent_can_read_child_trigram() { + let _git_env = crate::test_env::hermetic_git_env_guard(); + let temp = tempfile::tempdir().unwrap(); + let base = std::fs::canonicalize(temp.path()).unwrap(); + let projects = base.join("projects"); + let root = projects.join("alpha"); + let storage = base.join("storage"); + init_git_fixture(&root); + let ctx = test_context(); + let response = super::handle_configure( + &configure_request_with_params(json!({ + "project_root": root, + "harness": "opencode", + "storage_dir": storage, + "config": [user_tier(json!({ + "indexes": { "trigram": true, "semantic": false, "callgraph": false }, + "views": { "enabled": true }, + }))], + })), + &ctx, + ); + assert!(response.success, "{response:?}"); + assert!(ctx.artifact_identity_provisional_reason().is_none()); + assert!(ctx.views_route_root()); + let family = ctx.cached_artifact_cache_key(&root).unwrap(); + let mut state = super::ConfigureMaintenanceState::default(); + state.absorb_enqueued(&ctx); + while state.jobs.front().unwrap().stage != super::ConfigureMaintenanceStage::ViewLoad { + assert!(super::drain_deferred_configure_maintenance_unit( + &ctx, &mut state + )); + } + + let parent = test_context(); + parent.update_config(|config| { + config.project_root = Some(projects.clone()); + config.views.enabled = true; + }); + let planes = crate::views::parent::RequestedPlanes { + trigram: true, + semantic: false, + callgraph: false, + }; + assert!(crate::views::parent::prepare(&projects, planes)); + assert!(crate::views::parent::activate( + &parent, + &projects, + &storage, + &parent.config().semantic + )); + let session = crate::views::parent::session_for_root(&projects).unwrap(); + assert!(session.wait_rounds(1, Duration::from_secs(30))); + assert!(!session.children()[0].trigram_ready()); + + { + let reservation = crate::blob_store::v3::ReservationLock::acquire(&storage).unwrap(); + crate::blob_store::v3::open_base_refs_registry(&storage, &family, &reservation, None) + .unwrap(); + } + let ledger = crate::blob_store::v3::family_dir(&storage, &family) + .unwrap() + .join("members.sqlite"); + let blocker = rusqlite::Connection::open(ledger).unwrap(); + blocker.execute_batch("PRAGMA locking_mode=EXCLUSIVE; BEGIN EXCLUSIVE; UPDATE accounting SET watermark=watermark+1;").unwrap(); + // Only the configure tail may publish the child; the parent never binds it. + while super::drain_deferred_configure_maintenance_unit(&ctx, &mut state) {} + assert!(!session.children()[0].trigram_ready()); + assert_eq!( + state.jobs.len(), + 1, + "busy lexical bind was dropped instead of parked" + ); + assert_eq!( + state.jobs[0].stage, + super::ConfigureMaintenanceStage::ViewLoad + ); + let retry_at = state.jobs[0] + .retry_at() + .expect("busy bind needs a retry deadline"); + state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1)); + assert!(!state.has_ready_job()); + blocker.execute_batch("ROLLBACK").unwrap(); + drop(blocker); + state.replay_retry_cutoff = Some(retry_at); + assert!(state.has_ready_job()); + while super::drain_deferred_configure_maintenance_unit(&ctx, &mut state) {} + assert!(state.jobs.is_empty()); + let deadline = Instant::now() + Duration::from_secs(30); + while !session.children()[0].trigram_ready() { + assert!( + Instant::now() < deadline, + "parent never saw the retried lexical publication" + ); + std::thread::sleep(Duration::from_millis(20)); + } + } + fn trigram_bind_request(root: &std::path::Path, storage: &std::path::Path) -> RawRequest { configure_request_with_params(json!({ "project_root": root, diff --git a/crates/aft/tests/integration/per_checkout_9.rs b/crates/aft/tests/integration/per_checkout_9.rs index 2b1bd6fd2..4d7c5771d 100644 --- a/crates/aft/tests/integration/per_checkout_9.rs +++ b/crates/aft/tests/integration/per_checkout_9.rs @@ -378,6 +378,9 @@ fn configure(root: &Path, storage: &Path, doc: Value) -> Arc { } fn drain(ctx: &AppContext) { + // Match the runtime maintenance tick: a busy bind can defer identity + // resolution or view publication beyond configure's first drain. + aft::runtime_drain::drain_deferred_configure_maintenance(ctx); aft::runtime_drain::drain_watcher_events(ctx); aft::runtime_drain::drain_search_index_events(ctx); aft::runtime_drain::drain_semantic_index_events(ctx); @@ -390,14 +393,12 @@ fn trigram_artifact(root: &Path, storage: &Path) -> PathBuf { } /// Waits until the child session's trigram index is ready and persisted. -/// Each round also runs the configure-tail drain both runtimes run on their -/// maintenance tick: a child whose bind met a busy aft.db settles its identity -/// there once the database is free, and only then starts its index. +/// The runtime drain also settles a bind whose identity was deferred by busy +/// storage, so its index can start once the database is free. fn wait_trigram(ctx: &AppContext, root: &Path, storage: &Path) { let deadline = Instant::now() + DEADLINE; loop { drain(ctx); - aft::runtime_drain::drain_deferred_configure_maintenance(ctx); let ready = ctx .search_index() .read() @@ -420,6 +421,53 @@ fn wait_trigram(ctx: &AppContext, root: &Path, storage: &Path) { } } +#[test] +fn runtime_drains_retry_child_trigram_after_busy_identity() { + let temp = tempfile::tempdir().unwrap(); + let base = std::fs::canonicalize(temp.path()).unwrap(); + let root = base.join("alpha"); + let storage = base.join("storage"); + init_repo( + &root, + &[("src/lib.rs", "pub fn retry_marker() {}\n".into())], + ); + std::fs::create_dir_all(&storage).unwrap(); + let lock = rusqlite::Connection::open(storage.join("aft.db")).unwrap(); + // Rollback-journal mode blocks identity readers as well as writers. + lock.execute_batch("PRAGMA journal_mode=DELETE; BEGIN EXCLUSIVE;") + .unwrap(); + let ctx = configure(&root, &storage, child_trigram_doc()); + assert!(ctx.artifact_identity_provisional_reason().is_some()); + assert!(ctx.cached_artifact_cache_key(&root).is_none()); + assert!(!ctx.config().indexes.trigram); + lock.execute_batch("ROLLBACK").unwrap(); + drop(lock); + + let deadline = Instant::now() + DEADLINE; + loop { + drain(&ctx); + if ctx + .search_index() + .read() + .unwrap() + .as_ref() + .is_some_and(|index| index.ready) + && trigram_artifact(&root, &storage).is_file() + { + break; + } + assert!( + Instant::now() < deadline, + "runtime drains never retried the child's busy identity: {:?}", + ctx.degraded_reasons() + ); + thread::sleep(Duration::from_millis(20)); + } + assert!(ctx.artifact_identity_provisional_reason().is_none()); + assert!(ctx.config().indexes.trigram); + assert!(ctx.cached_artifact_cache_key(&root).is_some()); +} + /// Asks the child session for its call graph until it answers. The first /// answer binds the child's v3 call-graph view and publishes it; a later one /// publishes the edits made since. @@ -427,7 +475,6 @@ fn child_callgraph_answer(ctx: &AppContext, root: &Path) -> Value { let deadline = Instant::now() + DEADLINE; loop { drain(ctx); - aft::runtime_drain::drain_deferred_configure_maintenance(ctx); let answer = aft::commands::callers::handle_callers( &request(json!({ "id": "child-callers", From 77ecea58afc66582f235562b940d1c8b465a1df1 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 05:51:41 +0100 Subject: [PATCH 20/23] mason: retry busy child lexical binds from configure maintenance Keep ViewLoad parked with a one-second backoff for explicit v3 busy markers. Parent sessions only read child publications, so dropping this first bind leaves them incomplete until a direct child query happens. Terminal ledger errors still follow the unavailable path; the busy contract and all existing parent-search assertions and deadlines are unchanged. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/commands/configure.rs | 47 ++++++++++++++++++- .../aft/tests/integration/per_checkout_9.rs | 10 +++- 2 files changed, 54 insertions(+), 3 deletions(-) diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs index ff7de0906..d47d35ac2 100644 --- a/crates/aft/src/commands/configure.rs +++ b/crates/aft/src/commands/configure.rs @@ -8092,6 +8092,7 @@ struct ConfigureMaintenanceContinuation { semantic_waits_for_callgraph_start: bool, view_load: Option>>, replay_retry_at: Option, + lexical_bind_retry_at: Option, callgraph_upgrade_retry_at: Option, replay_error_logged: bool, } @@ -8105,16 +8106,28 @@ impl ConfigureMaintenanceContinuation { semantic_waits_for_callgraph_start: false, view_load: None, replay_retry_at: None, + lexical_bind_retry_at: None, callgraph_upgrade_retry_at: None, replay_error_logged: false, } } fn retry_at(&self) -> Option { - self.callgraph_upgrade_retry_at.or(self.replay_retry_at) + self.callgraph_upgrade_retry_at + .or(self.lexical_bind_retry_at) + .or(self.replay_retry_at) } } +fn lexical_bind_is_busy(reason: &str) -> bool { + // The lexical snapshot boundary carries strings. Match only the explicit + // retry markers preserved by the v3 registry and base builder, including + // markers nested beneath a driver or registry diagnostic. + reason.contains(crate::blob_store::v3::BUSY_PREFIX) + || reason.contains("v3 store busy: ") + || reason.contains("v3 base build busy; retain direct-scan coverage and retry") +} + #[derive(Debug, Default)] pub(crate) struct ConfigureMaintenanceState { jobs: VecDeque, @@ -8986,6 +8999,7 @@ fn run_configure_maintenance_unit_inner( continuation.stage = ConfigureMaintenanceStage::ViewLoad; } ConfigureMaintenanceStage::ViewLoad => { + let retrying = continuation.lexical_bind_retry_at.take().is_some(); if crate::views::parent::held_by(ctx, &job.canonical_cache_root) { // A parent folder owns no view; its session reads the children's. ctx.clear_view_runtime(); @@ -9007,6 +9021,19 @@ fn run_configure_maintenance_unit_inner( ctx.clear_view_runtime(); } match ctx.lexical_view_snapshot("bind") { + Some(Err(error)) if lexical_bind_is_busy(&error) => { + if !retrying { + slog_warn!( + "lexical checkout view busy; configure maintenance will retry: {}", + error + ); + } + // A parent only reads child publications; it cannot + // recover a dropped child bind by querying that child. + continuation.lexical_bind_retry_at = + Some(Instant::now() + Duration::from_secs(1)); + return ConfigureMaintenanceUnitResult::Continue; + } Some(Err(error)) => { slog_warn!("lexical checkout view unavailable: {}", error); } @@ -17024,6 +17051,24 @@ mod tests { lock } + #[test] + fn lexical_bind_busy_markers_do_not_retry_terminal_failures() { + for reason in [ + "family registry I/O error: reservation_ledger_busy: read format: database is locked", + "[driver]: v3 store busy: reservation lock timeout", + "[driver]: v3 base build busy; retain direct-scan coverage and retry", + ] { + assert!(super::lexical_bind_is_busy(reason), "{reason}"); + } + for reason in [ + "reservation_ledger_unreadable: permission denied", + "live_sqlite_file_replaced", + "a file named busy.rs is unreadable", + ] { + assert!(!super::lexical_bind_is_busy(reason), "{reason}"); + } + } + #[test] fn busy_lexical_bind_retries_until_parent_can_read_child_trigram() { let _git_env = crate::test_env::hermetic_git_env_guard(); diff --git a/crates/aft/tests/integration/per_checkout_9.rs b/crates/aft/tests/integration/per_checkout_9.rs index 4d7c5771d..410650e58 100644 --- a/crates/aft/tests/integration/per_checkout_9.rs +++ b/crates/aft/tests/integration/per_checkout_9.rs @@ -437,7 +437,10 @@ fn runtime_drains_retry_child_trigram_after_busy_identity() { lock.execute_batch("PRAGMA journal_mode=DELETE; BEGIN EXCLUSIVE;") .unwrap(); let ctx = configure(&root, &storage, child_trigram_doc()); - assert!(ctx.artifact_identity_provisional_reason().is_some()); + assert!(ctx + .degraded_reasons() + .iter() + .any(|reason| reason.contains("aft.db was busy at bind"))); assert!(ctx.cached_artifact_cache_key(&root).is_none()); assert!(!ctx.config().indexes.trigram); lock.execute_batch("ROLLBACK").unwrap(); @@ -463,7 +466,10 @@ fn runtime_drains_retry_child_trigram_after_busy_identity() { ); thread::sleep(Duration::from_millis(20)); } - assert!(ctx.artifact_identity_provisional_reason().is_none()); + assert!(!ctx + .degraded_reasons() + .iter() + .any(|reason| reason.contains("aft.db was busy at bind"))); assert!(ctx.config().indexes.trigram); assert!(ctx.cached_artifact_cache_key(&root).is_some()); } From 1f827f77c3a142d09c54dd422f8fb67d16582c9b Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:01:48 +0100 Subject: [PATCH 21/23] mason: back off persistent lexical contention and assert pending coverage Retry busy lexical publication after 1, 2, 4, 8, 16, then at most 30 seconds; reset after the bind succeeds. The forced-lock regression keeps the parent incomplete and checks that its answer names alpha's busy trigram gap before verifying eventual publication without a child query. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/commands/configure.rs | 54 ++++++++++++++++++++++++++-- 1 file changed, 52 insertions(+), 2 deletions(-) diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs index d47d35ac2..80d39da6e 100644 --- a/crates/aft/src/commands/configure.rs +++ b/crates/aft/src/commands/configure.rs @@ -8093,6 +8093,7 @@ struct ConfigureMaintenanceContinuation { view_load: Option>>, replay_retry_at: Option, lexical_bind_retry_at: Option, + lexical_bind_retry_delay: Duration, callgraph_upgrade_retry_at: Option, replay_error_logged: bool, } @@ -8107,6 +8108,7 @@ impl ConfigureMaintenanceContinuation { view_load: None, replay_retry_at: None, lexical_bind_retry_at: None, + lexical_bind_retry_delay: Duration::from_secs(1), callgraph_upgrade_retry_at: None, replay_error_logged: false, } @@ -9031,7 +9033,10 @@ fn run_configure_maintenance_unit_inner( // A parent only reads child publications; it cannot // recover a dropped child bind by querying that child. continuation.lexical_bind_retry_at = - Some(Instant::now() + Duration::from_secs(1)); + Some(Instant::now() + continuation.lexical_bind_retry_delay); + continuation.lexical_bind_retry_delay = + (continuation.lexical_bind_retry_delay * 2) + .min(Duration::from_secs(30)); return ConfigureMaintenanceUnitResult::Continue; } Some(Err(error)) => { @@ -9043,6 +9048,7 @@ fn run_configure_maintenance_unit_inner( } _ => {} } + continuation.lexical_bind_retry_delay = Duration::from_secs(1); } else { ctx.clear_view_runtime(); } @@ -17107,7 +17113,10 @@ mod tests { parent.update_config(|config| { config.project_root = Some(projects.clone()); config.views.enabled = true; + config.indexes.semantic = false; + config.indexes.callgraph = false; }); + parent.set_canonical_cache_root(projects.clone()); let planes = crate::views::parent::RequestedPlanes { trigram: true, semantic: false, @@ -17149,12 +17158,53 @@ mod tests { let retry_at = state.jobs[0] .retry_at() .expect("busy bind needs a retry deadline"); + assert_eq!( + state.jobs[0].lexical_bind_retry_delay, + Duration::from_secs(2) + ); state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1)); assert!(!state.has_ready_job()); + // Long-held locks must not keep taking a maintenance worker each second. + for seconds in [2, 4, 8, 16, 30, 30] { + state.replay_retry_cutoff = state.jobs[0].retry_at(); + let before = Instant::now(); + assert!(!super::drain_deferred_configure_maintenance_unit( + &ctx, &mut state + )); + let retry_at = state.jobs[0].retry_at().unwrap(); + assert!(retry_at >= before + Duration::from_secs(seconds)); + assert!(retry_at <= Instant::now() + Duration::from_secs(seconds)); + assert_eq!( + state.jobs[0].stage, + super::ConfigureMaintenanceStage::ViewLoad + ); + } + assert!(session.wait_rounds(session.rounds() + 1, Duration::from_secs(30))); + let query: RawRequest = serde_json::from_value(json!({ + "id": "parent-busy-lexical", + "command": "semantic_search", + "query": "tracked", + })) + .unwrap(); + let pending = crate::views::parent::route(&query, &parent).unwrap(); + assert_eq!(pending.data["complete"], false, "{pending:?}"); + let text = pending.data["text"].as_str().unwrap(); + assert!( + text.contains("alpha") && text.contains("busy"), + "{pending:?}" + ); blocker.execute_batch("ROLLBACK").unwrap(); drop(blocker); - state.replay_retry_cutoff = Some(retry_at); + state.replay_retry_cutoff = state.jobs[0].retry_at(); assert!(state.has_ready_job()); + assert!(super::drain_deferred_configure_maintenance_unit( + &ctx, &mut state + )); + assert!(state.jobs[0].lexical_bind_retry_at.is_none()); + assert_eq!( + state.jobs[0].lexical_bind_retry_delay, + Duration::from_secs(1) + ); while super::drain_deferred_configure_maintenance_unit(&ctx, &mut state) {} assert!(state.jobs.is_empty()); let deadline = Instant::now() + Duration::from_secs(30); From a16d7f2ef3928cb9b70dd555852c008860bd0662 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:05:39 +0100 Subject: [PATCH 22/23] mason: retain parent busy coverage until child publication arrives Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/commands/configure.rs | 23 +++++++++++------------ crates/aft/src/views/parent/worker.rs | 9 +++++++-- crates/aft/src/views/trigram.rs | 9 +++++++++ 3 files changed, 27 insertions(+), 14 deletions(-) diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs index 80d39da6e..1929d4e34 100644 --- a/crates/aft/src/commands/configure.rs +++ b/crates/aft/src/commands/configure.rs @@ -8121,15 +8121,6 @@ impl ConfigureMaintenanceContinuation { } } -fn lexical_bind_is_busy(reason: &str) -> bool { - // The lexical snapshot boundary carries strings. Match only the explicit - // retry markers preserved by the v3 registry and base builder, including - // markers nested beneath a driver or registry diagnostic. - reason.contains(crate::blob_store::v3::BUSY_PREFIX) - || reason.contains("v3 store busy: ") - || reason.contains("v3 base build busy; retain direct-scan coverage and retry") -} - #[derive(Debug, Default)] pub(crate) struct ConfigureMaintenanceState { jobs: VecDeque, @@ -9023,7 +9014,7 @@ fn run_configure_maintenance_unit_inner( ctx.clear_view_runtime(); } match ctx.lexical_view_snapshot("bind") { - Some(Err(error)) if lexical_bind_is_busy(&error) => { + Some(Err(error)) if crate::views::trigram::bind_is_busy(&error) => { if !retrying { slog_warn!( "lexical checkout view busy; configure maintenance will retry: {}", @@ -17064,14 +17055,14 @@ mod tests { "[driver]: v3 store busy: reservation lock timeout", "[driver]: v3 base build busy; retain direct-scan coverage and retry", ] { - assert!(super::lexical_bind_is_busy(reason), "{reason}"); + assert!(crate::views::trigram::bind_is_busy(reason), "{reason}"); } for reason in [ "reservation_ledger_unreadable: permission denied", "live_sqlite_file_replaced", "a file named busy.rs is unreadable", ] { - assert!(!super::lexical_bind_is_busy(reason), "{reason}"); + assert!(!crate::views::trigram::bind_is_busy(reason), "{reason}"); } } @@ -17195,6 +17186,14 @@ mod tests { ); blocker.execute_batch("ROLLBACK").unwrap(); drop(blocker); + assert!(session.wait_rounds(session.rounds() + 1, Duration::from_secs(30))); + let pending = crate::views::parent::route(&query, &parent).unwrap(); + assert_eq!(pending.data["complete"], false, "{pending:?}"); + let text = pending.data["text"].as_str().unwrap(); + assert!( + text.contains("alpha") && text.contains("busy"), + "{pending:?}" + ); state.replay_retry_cutoff = state.jobs[0].retry_at(); assert!(state.has_ready_job()); assert!(super::drain_deferred_configure_maintenance_unit( diff --git a/crates/aft/src/views/parent/worker.rs b/crates/aft/src/views/parent/worker.rs index 965ad620f..bb0d2b0e8 100644 --- a/crates/aft/src/views/parent/worker.rs +++ b/crates/aft/src/views/parent/worker.rs @@ -245,8 +245,13 @@ fn refresh_trigram(session: &ParentSession, child: &Child) { Some(&session.storage), ); let Some(artifact) = artifact else { - *write(&child.trigram) = - Plane::Gap("no trigram index has been built for this repository yet".into()); + let mut trigram = write(&child.trigram); + // Releasing the ledger lock does not itself publish a usable lexical index. + // Keep its last busy diagnostic until a usable publication arrives. + if matches!(&*trigram, Plane::Gap(reason) if super::super::trigram::bind_is_busy(reason)) { + return; + } + *trigram = Plane::Gap("no trigram index has been built for this repository yet".into()); return; }; if read(&child.trigram).ready().is_some_and( diff --git a/crates/aft/src/views/trigram.rs b/crates/aft/src/views/trigram.rs index c3c42bfd4..eb1ecc1c3 100644 --- a/crates/aft/src/views/trigram.rs +++ b/crates/aft/src/views/trigram.rs @@ -45,6 +45,15 @@ fn absolute_member_path( } Ok(root.join(relative)) } +pub(crate) fn bind_is_busy(reason: &str) -> bool { + // Lexical snapshot errors carry strings. Match only the explicit retry + // markers preserved by the v3 registry and base builder, including markers + // nested beneath a driver or registry diagnostic. + reason.contains(crate::blob_store::v3::BUSY_PREFIX) + || reason.contains("v3 store busy: ") + || reason.contains("v3 base build busy; retain direct-scan coverage and retry") +} + /// Checkout-local lexical authority, independent of the semantic worker's /// lifetime and model availability. Queries serialize reconciliation so an /// installed membership and its payloads always come from the same snapshot. From 5c59b95e6f8fd30c439ab64acab9911d4b682727 Mon Sep 17 00:00:00 2001 From: ualtinok <94532+ualtinok@users.noreply.github.com> Date: Thu, 29 Oct 2026 06:32:36 +0100 Subject: [PATCH 23/23] mason: recognize ledger busy marker and test backoff without SQLite waits The marker regression caught that BUSY_PREFIX already means v3 store busy, not reservation_ledger_busy. Cover both explicit forms. Exercise capped scheduling with an injected Instant after one real locked bind, avoiding repeated SQLite timeout waits in the unit gate while retaining the locked and post-unlock coverage assertions. Co-authored-by: alfonso-aft <289616620+alfonso-aft@users.noreply.github.com> --- crates/aft/src/commands/configure.rs | 27 ++++++++++++++------------- crates/aft/src/views/trigram.rs | 2 +- 2 files changed, 15 insertions(+), 14 deletions(-) diff --git a/crates/aft/src/commands/configure.rs b/crates/aft/src/commands/configure.rs index 1929d4e34..c6fe7c3e2 100644 --- a/crates/aft/src/commands/configure.rs +++ b/crates/aft/src/commands/configure.rs @@ -8119,6 +8119,12 @@ impl ConfigureMaintenanceContinuation { .or(self.lexical_bind_retry_at) .or(self.replay_retry_at) } + + fn defer_lexical_bind(&mut self, now: Instant) { + self.lexical_bind_retry_at = Some(now + self.lexical_bind_retry_delay); + self.lexical_bind_retry_delay = + (self.lexical_bind_retry_delay * 2).min(Duration::from_secs(30)); + } } #[derive(Debug, Default)] @@ -9023,11 +9029,7 @@ fn run_configure_maintenance_unit_inner( } // A parent only reads child publications; it cannot // recover a dropped child bind by querying that child. - continuation.lexical_bind_retry_at = - Some(Instant::now() + continuation.lexical_bind_retry_delay); - continuation.lexical_bind_retry_delay = - (continuation.lexical_bind_retry_delay * 2) - .min(Duration::from_secs(30)); + continuation.defer_lexical_bind(Instant::now()); return ConfigureMaintenanceUnitResult::Continue; } Some(Err(error)) => { @@ -17155,16 +17157,15 @@ mod tests { ); state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1)); assert!(!state.has_ready_job()); - // Long-held locks must not keep taking a maintenance worker each second. + // Check every backoff step without spending SQLite's busy timeout on + // each attempt. The real locked bind above arms the same continuation. for seconds in [2, 4, 8, 16, 30, 30] { - state.replay_retry_cutoff = state.jobs[0].retry_at(); - let before = Instant::now(); - assert!(!super::drain_deferred_configure_maintenance_unit( - &ctx, &mut state - )); + let now = Instant::now(); + state.jobs[0].defer_lexical_bind(now); let retry_at = state.jobs[0].retry_at().unwrap(); - assert!(retry_at >= before + Duration::from_secs(seconds)); - assert!(retry_at <= Instant::now() + Duration::from_secs(seconds)); + assert_eq!(retry_at, now + Duration::from_secs(seconds)); + state.replay_retry_cutoff = Some(retry_at - Duration::from_nanos(1)); + assert!(!state.has_ready_job()); assert_eq!( state.jobs[0].stage, super::ConfigureMaintenanceStage::ViewLoad diff --git a/crates/aft/src/views/trigram.rs b/crates/aft/src/views/trigram.rs index eb1ecc1c3..932afc395 100644 --- a/crates/aft/src/views/trigram.rs +++ b/crates/aft/src/views/trigram.rs @@ -50,7 +50,7 @@ pub(crate) fn bind_is_busy(reason: &str) -> bool { // markers preserved by the v3 registry and base builder, including markers // nested beneath a driver or registry diagnostic. reason.contains(crate::blob_store::v3::BUSY_PREFIX) - || reason.contains("v3 store busy: ") + || reason.contains("reservation_ledger_busy:") || reason.contains("v3 base build busy; retain direct-scan coverage and retry") }