fix(merge): classify stale preparations by a stable error code

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<Self> {
+        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);
+    }
 }

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"),
                     ));
                 }
             }

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"] }

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] {

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