diff --git a/Cargo.lock b/Cargo.lock index 1cd26f28..1a32b483 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3944,6 +3944,7 @@ version = "0.1.0" dependencies = [ "clap", "etcd-client", + "lance-context-api", "serde", "serde_json", "tokio", diff --git a/crates/lance-context-api/src/lib.rs b/crates/lance-context-api/src/lib.rs index 3393e2c2..43320846 100644 --- a/crates/lance-context-api/src/lib.rs +++ b/crates/lance-context-api/src/lib.rs @@ -1494,6 +1494,46 @@ pub enum CompactJobStatus { // Unified task scheduler (master control-plane) // --------------------------------------------------------------------------- +/// Stable, machine-readable codes embedded in maintenance error messages. +/// +/// Errors cross worker HTTP responses and durable etcd task records as +/// strings, and Lance wraps our validation failures in its own error +/// variants. A code lets the failure classifier recognise the error +/// without matching on prose, file paths or wrapper formats. Codes are +/// rendered as `[LC_…]` and must never be renamed once released. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum MaintenanceErrorCode { + /// A prepared compaction, index or merge was validated against a + /// snapshot that another writer has since changed. Rebuilding against + /// the latest snapshot can succeed; retry promptly. + StalePreparation, +} + +impl MaintenanceErrorCode { + pub const fn as_str(self) -> &'static str { + match self { + Self::StalePreparation => "LC_STALE_PREPARATION", + } + } + + /// Render `message` with this code attached. + pub fn tag(self, message: &str) -> String { + format!("[{}] {message}", self.as_str()) + } + + /// Find the first code tag in an error string, regardless of wrapping. + pub fn parse(error: &str) -> Option { + let start = error.find("[LC_")?; + let rest = &error[start + 1..]; + let end = rest.find(']')?; + match &rest[..end] { + "LC_STALE_PREPARATION" => Some(Self::StalePreparation), + _ => None, + } + } +} + /// The kind of work a scheduled task performs. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] @@ -1790,4 +1830,22 @@ mod tests { serde_json::from_str(r#"{"kind":"index_id","target":"exp-9"}"#).unwrap(); assert_eq!(req.kind, TaskKind::IndexId); } + + #[test] + fn maintenance_error_code_round_trips_through_wrappers() { + let code = MaintenanceErrorCode::StalePreparation; + let tagged = code.tag("index metadata changed; reprepare"); + assert_eq!( + tagged, + "[LC_STALE_PREPARATION] index metadata changed; reprepare" + ); + let wrapped = format!("Invalid user input: {tagged}, crates/x/src/y.rs:1:2"); + assert_eq!(MaintenanceErrorCode::parse(&wrapped), Some(code)); + assert_eq!( + MaintenanceErrorCode::parse(&wrapped.to_ascii_lowercase()), + None + ); + assert_eq!(MaintenanceErrorCode::parse("[LC_UNKNOWN] x"), None); + assert_eq!(MaintenanceErrorCode::parse("no code"), None); + } } diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs index 12af8f04..3513cdf6 100644 --- a/crates/lance-context-core/src/store_base.rs +++ b/crates/lance-context-core/src/store_base.rs @@ -60,6 +60,7 @@ use lance::index::DatasetIndexExt; use lance::io::{ObjectStoreParams, StorageOptionsAccessor}; use lance::session::Session; use lance::{Error as LanceError, Result as LanceResult}; +use lance_context_api::MaintenanceErrorCode; use lance_index::mem_wal::{ShardManifest, MEM_WAL_INDEX_NAME}; use lance_index::scalar::ScalarIndexParams; use lance_index::IndexType; @@ -1109,9 +1110,9 @@ impl StorageBase { .get(&self.write_shard) .is_some_and(|high| merged_generations.iter().any(|g| g <= high)) { - return Err(LanceError::io( + return Err(LanceError::io(MaintenanceErrorCode::StalePreparation.tag( "prepared merge overlaps committed rollout watermark; reprepare", - )); + ))); } let watermark = if crate::rollout_append::has_cutover(&self.dataset, self.write_shard) { @@ -1500,7 +1501,8 @@ impl StorageBase { || prepared.schema != Schema::from(self.dataset.schema()) { return Err(LanceError::invalid_input( - "index dataset or schema changed; reprepare", + MaintenanceErrorCode::StalePreparation + .tag("index dataset or schema changed; reprepare"), )); } let sources: HashMap<_, _> = self @@ -1516,7 +1518,8 @@ impl StorageBase { .any(|old| sources.get(&old.id).copied() != Some(old)) { return Err(LanceError::invalid_input( - "index source fragment changed; reprepare", + MaintenanceErrorCode::StalePreparation + .tag("index source fragment changed; reprepare"), )); } let current = self.dataset.load_indices().await?; @@ -1527,7 +1530,7 @@ impl StorageBase { .collect(); if replaced != prepared.replaced { return Err(LanceError::invalid_input( - "index metadata changed; reprepare", + MaintenanceErrorCode::StalePreparation.tag("index metadata changed; reprepare"), )); } let transaction = Transaction::new( @@ -1656,7 +1659,8 @@ impl StorageBase { || prepared.schema != Schema::from(self.dataset.schema()) { return Err(LanceError::invalid_input( - "compaction dataset or schema changed; reprepare", + MaintenanceErrorCode::StalePreparation + .tag("compaction dataset or schema changed; reprepare"), )); } // Reject a stale rewrite before any ReserveFragments/index commit. @@ -1672,7 +1676,8 @@ impl StorageBase { for old in &task.original_fragments { if sources.get(&old.id).copied() != Some(old) { return Err(LanceError::invalid_input( - "compaction source fragment changed; reprepare", + MaintenanceErrorCode::StalePreparation + .tag("compaction source fragment changed; reprepare"), )); } } diff --git a/crates/lance-context-merge/Cargo.toml b/crates/lance-context-merge/Cargo.toml index efeda501..0a529604 100644 --- a/crates/lance-context-merge/Cargo.toml +++ b/crates/lance-context-merge/Cargo.toml @@ -8,6 +8,7 @@ description = "Durable ownership for cancellable WAL merge execution" [dependencies] clap = { version = "4", features = ["derive", "env"] } etcd-client = { version = "0.19", features = ["tls"] } +lance-context-api = { version = "0.6.7", path = "../lance-context-api" } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["macros", "rt-multi-thread", "sync", "time"] } diff --git a/crates/lance-context-merge/src/failure.rs b/crates/lance-context-merge/src/failure.rs index c7f2ca9b..7306a7a9 100644 --- a/crates/lance-context-merge/src/failure.rs +++ b/crates/lance-context-merge/src/failure.rs @@ -1,6 +1,7 @@ //! Durable shard failures: task recreation must not reset a retry budget. use crate::{ClaimProof, Coordinator, Execution, Result}; use etcd_client::{Compare, CompareOp, GetOptions, TxnOp}; +use lance_context_api::MaintenanceErrorCode; use serde::{Deserialize, Serialize}; #[derive(Clone, Debug, Serialize, Deserialize, PartialEq, Eq)] @@ -32,10 +33,20 @@ pub fn now_ms() -> u64 { } pub fn classify(error: &str) -> FailureClass { - let error = error.to_ascii_lowercase(); - if error.contains("ownership unresolved") { - FailureClass::OwnershipUnresolved - } else if is_preparation_conflict(&error) { + // A stable code wins over any prose, path or wrapper format. Ownership + // is checked first because an unresolved owner must never be retried + // as an ordinary conflict. + let lowered = error.to_ascii_lowercase(); + if lowered.contains("ownership unresolved") { + return FailureClass::OwnershipUnresolved; + } + if let Some(code) = MaintenanceErrorCode::parse(error) { + return match code { + MaintenanceErrorCode::StalePreparation => FailureClass::Retryable, + }; + } + let error = lowered; + if is_preparation_conflict(&error) { FailureClass::Retryable } else if [ "not found", @@ -66,9 +77,10 @@ pub fn classify(error: &str) -> FailureClass { } } -// These exact validation failures mean the prepared snapshot became stale. -// Lance wraps them in InvalidInput, but rebuilding against the latest snapshot -// can succeed. Do not exempt arbitrary errors merely containing "reprepare". +// Legacy text matcher for failure records written by binaries that predate +// `MaintenanceErrorCode`. New errors carry `[LC_STALE_PREPARATION]` and are +// classified before this runs. Remove once no stored record lacks a code. +// Do not exempt arbitrary errors merely containing "reprepare". fn is_preparation_conflict(error: &str) -> bool { let message = error.strip_prefix("invalid user input: ").unwrap_or(error); let message = match message.split_once(", crates/lance-context-core/src/store_base.rs:") { @@ -158,7 +170,9 @@ impl ShardFailure { // The next fenced completion persists the normal retry result. fn corrected(mut self) -> Self { if self.class == FailureClass::DataOrConfiguration - && is_preparation_conflict(&self.last_error.to_ascii_lowercase()) + && (MaintenanceErrorCode::parse(&self.last_error) + == Some(MaintenanceErrorCode::StalePreparation) + || is_preparation_conflict(&self.last_error.to_ascii_lowercase())) { self.class = FailureClass::Retryable; let (attention, delay) = Self::retry_policy(&self.class, self.consecutive_attempts); @@ -401,6 +415,43 @@ mod tests { } } + #[test] + fn coded_preparation_conflicts_classify_regardless_of_wrapper_or_path() { + let tagged = MaintenanceErrorCode::StalePreparation + .tag("compaction source fragment changed; reprepare"); + for error in [ + format!( + "Invalid user input: {tagged}, crates/lance-context-core/src/store_base.rs:1529:24" + ), + format!("Invalid user input: {tagged}, /moved/elsewhere/writer.rs:7:3"), + format!("Some future Lance wrapper → {tagged}"), + format!("IO error: {tagged}"), + tagged.clone(), + ] { + assert_eq!(classify(&error), FailureClass::Retryable, "{error}"); + } + // Ownership still wins, and the code does not launder a 403. + assert_eq!( + classify(&format!("ownership unresolved: {tagged}")), + FailureClass::OwnershipUnresolved + ); + // An old DataOrConfiguration record carrying a code is corrected too. + let old = ShardFailure { + target: "table".into(), + endpoint: "master:compact".into(), + class: FailureClass::DataOrConfiguration, + consecutive_attempts: 4, + last_failure_ms: 2, + next_retry_ms: 3_602_000, + needs_attention: true, + last_error: format!("Invalid user input: {tagged}"), + }; + let corrected = old.corrected(); + assert_eq!(corrected.class, FailureClass::Retryable); + assert_eq!(corrected.consecutive_attempts, 4); + assert_eq!(corrected.last_failure_ms, 2); + } + #[test] fn old_conflict_records_recover_without_resetting_attempts_or_clock() { for attempts in [1, 4, 15, u32::MAX] { diff --git a/docs/concurrent-indexing.md b/docs/concurrent-indexing.md index 1cf74a99..f0955094 100644 --- a/docs/concurrent-indexing.md +++ b/docs/concurrent-indexing.md @@ -36,8 +36,12 @@ the original snapshot read version, allowing Lance to check intervening commits as well. Newly appended fragments remain unindexed and query planning scans them; index publication must preserve merge watermarks and newly appended rows. -These exact stale-preparation validation failures use the ordinary bounded retry -budget, even when Lance wraps them as `Invalid user input`. The first two failed +Stale-preparation validation failures carry the stable code +`[LC_STALE_PREPARATION]` in their message. The failure classifier recognises the +code regardless of how Lance wraps the error or where the check lives in the +source tree; the older exact-text matcher remains only for failure records +written by binaries that predate the code. These failures use the ordinary +bounded retry budget. The first two failed attempts wait two seconds; repeated failures back off and eventually require attention. Existing records incorrectly classified as data/configuration errors are interpreted with the corrected policy on read, retaining their original