fix: retry stale maintenance preparations without one-hour cooldown

crates/lance-context-merge/src/failure.rs

@@ -35,6 +35,8 @@ 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) {
+        FailureClass::Retryable
     } else if [
         "not found",
         "notfound",
@@ -64,6 +66,33 @@ 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".
+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:") {
+        Some((message, location)) => {
+            let Some((line, column)) = location.split_once(':') else {
+                return false;
+            };
+            if line.parse::<u32>().is_err() || column.parse::<u32>().is_err() {
+                return false;
+            }
+            message
+        }
+        None => message,
+    };
+    matches!(
+        message,
+        "index dataset or schema changed; reprepare"
+            | "index source fragment changed; reprepare"
+            | "index metadata changed; reprepare"
+            | "compaction dataset or schema changed; reprepare"
+            | "compaction source fragment changed; reprepare"
+    )
+}
+
 // Errors cross worker HTTP and durable task boundaries as strings. Recognize
 // explicit HTTP status syntax, never digits in a request ID, path or timestamp.
 fn has_http_status(error: &str, codes: &[&str]) -> bool {
@@ -106,23 +135,13 @@ fn has_http_status(error: &str, codes: &[&str]) -> bool {
 }
 
 impl ShardFailure {
-    fn advance(
-        target: &str,
-        endpoint: &str,
-        previous: Option<Self>,
-        error: &str,
-        now: u64,
-    ) -> Self {
-        let attempts = previous.map_or(1, |f| f.consecutive_attempts.saturating_add(1));
-        let class = classify(error);
+    fn retry_policy(class: &FailureClass, attempts: u32) -> (bool, u64) {
         let needs_attention = attempts >= 15
             || matches!(
                 class,
                 FailureClass::DataOrConfiguration | FailureClass::OwnershipUnresolved
             );
-        // First task gets three attempts. Later tasks get one half-open probe,
-        // never a fresh budget of three. Cap probing at once per hour.
-        let delay_secs = if class == FailureClass::OwnershipUnresolved {
+        let delay_secs = if *class == FailureClass::OwnershipUnresolved {
             (30u64.saturating_mul(1 << attempts.saturating_sub(1).min(4))).min(300)
         } else if needs_attention {
             3600
@@ -131,6 +150,36 @@ impl ShardFailure {
         } else {
             (30u64.saturating_mul(1 << (attempts - 3).min(5))).min(900)
         };
+        (needs_attention, delay_secs)
+    }
+
+    // Read old records with corrected semantics without resetting attempts,
+    // moving the failure timestamp, or mutating ownership/durable records.
+    // 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())
+        {
+            self.class = FailureClass::Retryable;
+            let (attention, delay) = Self::retry_policy(&self.class, self.consecutive_attempts);
+            self.needs_attention = attention;
+            self.next_retry_ms = self.last_failure_ms.saturating_add(delay * 1000);
+        }
+        self
+    }
+
+    fn advance(
+        target: &str,
+        endpoint: &str,
+        previous: Option<Self>,
+        error: &str,
+        now: u64,
+    ) -> Self {
+        let attempts = previous.map_or(1, |f| f.consecutive_attempts.saturating_add(1));
+        let class = classify(error);
+        // First task gets three attempts. Later tasks get one half-open probe,
+        // never a fresh budget of three. Cap probing at once per hour.
+        let (needs_attention, delay_secs) = Self::retry_policy(&class, attempts);
         Self {
             target: target.into(),
             endpoint: endpoint.into(),
@@ -172,7 +221,11 @@ impl Coordinator {
         response
             .kvs()
             .first()
-            .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string()))
+            .map(|kv| {
+                serde_json::from_slice::<ShardFailure>(kv.value())
+                    .map(ShardFailure::corrected)
+                    .map_err(|e| e.to_string())
+            })
             .transpose()
     }
     /// Commit the terminal outcome and its retry budget in the same etcd
@@ -288,7 +341,11 @@ impl Coordinator {
         let rows = response
             .kvs()
             .iter()
-            .map(|kv| serde_json::from_slice(kv.value()).map_err(|e| e.to_string()))
+            .map(|kv| {
+                serde_json::from_slice::<ShardFailure>(kv.value())
+                    .map(ShardFailure::corrected)
+                    .map_err(|e| e.to_string())
+            })
             .collect::<Result<Vec<_>>>()?;
         let next = if response.more() {
             response
@@ -305,6 +362,82 @@ impl Coordinator {
 #[cfg(test)]
 mod tests {
     use super::*;
+    const STALE_INDEX: &str = "Invalid user input: index metadata changed; reprepare, crates/lance-context-core/src/store_base.rs:1529:24";
+
+    #[test]
+    fn optimistic_preparation_conflicts_keep_the_bounded_retry_budget() {
+        for message in [
+            "index metadata changed; reprepare",
+            "index source fragment changed; reprepare",
+            "index dataset or schema changed; reprepare",
+            "compaction source fragment changed; reprepare",
+            "compaction dataset or schema changed; reprepare",
+        ] {
+            let error = format!("Invalid user input: {message}, crates/lance-context-core/src/store_base.rs:1529:24");
+            let mut previous = None;
+            for attempt in 1..=16 {
+                let next =
+                    ShardFailure::advance("table", "master:index_id", previous, &error, 1000);
+                assert_eq!(next.class, FailureClass::Retryable, "{error}");
+                assert_eq!(next.consecutive_attempts, attempt);
+                assert_eq!(next.needs_attention, attempt >= 15);
+                if attempt < 3 {
+                    assert_eq!(next.next_retry_ms, 3000);
+                } else if attempt >= 15 {
+                    assert_eq!(next.next_retry_ms, 3_601_000);
+                }
+                previous = Some(next);
+            }
+        }
+        for error in [
+            "Invalid user input: schema mismatch; reprepare",
+            "Invalid user input: corrupt index metadata changed; reprepare",
+            "Invalid user input: index metadata changed; reprepare: permission denied",
+            "Invalid user input: index metadata changed; reprepare, crates/lance-context-core/src/store_base.rs:1529:24; HTTP 403",
+            "Invalid user input: index metadata changed; reprepare, /unrelated/path.rs:1:2",
+            "ownership unresolved: index metadata changed; reprepare",
+        ] {
+            assert_ne!(classify(error), FailureClass::Retryable, "{error}");
+        }
+    }
+
+    #[test]
+    fn old_conflict_records_recover_without_resetting_attempts_or_clock() {
+        for attempts in [1, 4, 15, u32::MAX] {
+            let old = ShardFailure {
+                target: "table".into(),
+                endpoint: "master:index_id".into(),
+                consecutive_attempts: attempts,
+                class: FailureClass::DataOrConfiguration,
+                last_error: STALE_INDEX.into(),
+                last_failure_ms: 1000,
+                next_retry_ms: 3_601_000,
+                needs_attention: true,
+            };
+            let corrected = old.clone().corrected();
+            assert_eq!(corrected.class, FailureClass::Retryable);
+            assert_eq!(corrected.consecutive_attempts, attempts);
+            assert_eq!(corrected.last_failure_ms, 1000);
+            assert_eq!(corrected.last_error, old.last_error);
+            assert_eq!(corrected.needs_attention, attempts >= 15);
+            assert_eq!(
+                corrected.next_retry_ms,
+                match attempts {
+                    1 => 3000,
+                    4 => 61_000,
+                    _ => 3_601_000,
+                }
+            );
+            let retry =
+                ShardFailure::advance("table", "master:index_id", Some(old), STALE_INDEX, 5000);
+            assert_eq!(retry.consecutive_attempts, attempts.saturating_add(1));
+            assert_eq!(retry.class, FailureClass::Retryable);
+        }
+        let permanent = ShardFailure::advance("table", "worker", None, "schema mismatch", 1000);
+        let before = serde_json::to_value(&permanent).unwrap();
+        assert_eq!(serde_json::to_value(permanent.corrected()).unwrap(), before);
+    }
+
     #[test]
     fn explicit_http_status_controls_backoff_not_request_id_digits() {
         let azure = include_str!("../tests/fixtures/azure-server-busy.txt");

crates/lance-context-merge/src/lib.rs

@@ -839,6 +839,94 @@ mod tests {
             .unwrap();
     }
 
+    #[tokio::test]
+    #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
+    async fn persisted_preparation_conflict_is_corrected_without_clearing_budget() {
+        let (coordinator, mut client, proof, lease) = fixture().await;
+        let error = "Invalid user input: index metadata changed; reprepare, crates/lance-context-core/src/store_base.rs:1529:24";
+        let endpoint = "master:index_id";
+        // Reproduce the previous binary's durable one-hour permanent-error record.
+        let mut old = coordinator
+            .record_failure(&proof, "table", endpoint, "schema mismatch")
+            .await
+            .unwrap();
+        old.last_error = error.into();
+        old.consecutive_attempts = 4;
+        let records = client
+            .get(
+                format!("{}/merge-failures/", coordinator.prefix),
+                Some(etcd_client::GetOptions::new().with_prefix()),
+            )
+            .await
+            .unwrap();
+        assert_eq!(records.kvs().len(), 1);
+        let key = records.kvs()[0].key().to_vec();
+        let encoded = serde_json::to_vec(&old).unwrap();
+        client
+            .put(key.clone(), encoded.clone(), None)
+            .await
+            .unwrap();
+
+        let reconnected = Coordinator::new(client.clone(), coordinator.prefix.clone());
+        let corrected = reconnected
+            .failure("table", endpoint)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(corrected.class, failure::FailureClass::Retryable);
+        assert_eq!(corrected.consecutive_attempts, 4);
+        assert_eq!(corrected.next_retry_ms, old.last_failure_ms + 60_000);
+        assert!(!corrected.needs_attention);
+        let (page, next) = reconnected.failure_page(None, 1).await.unwrap();
+        assert!(next.is_none());
+        assert_eq!(
+            serde_json::to_value(&page[0]).unwrap(),
+            serde_json::to_value(&corrected).unwrap()
+        );
+        // Reads neither change persisted data nor release ownership.
+        assert_eq!(
+            client.get(key.clone(), None).await.unwrap().kvs()[0].value(),
+            encoded
+        );
+        assert_eq!(
+            client.get(proof.key.clone(), None).await.unwrap().kvs()[0].value(),
+            proof.token.as_bytes()
+        );
+
+        let next = reconnected
+            .record_failure(&proof, "table", endpoint, error)
+            .await
+            .unwrap();
+        assert_eq!(next.consecutive_attempts, 5);
+        assert_eq!(next.class, failure::FailureClass::Retryable);
+        assert_eq!(next.next_retry_ms, next.last_failure_ms + 120_000);
+        client.lease_revoke(lease).await.unwrap();
+        assert!(reconnected
+            .record_failure(&proof, "table", endpoint, error)
+            .await
+            .is_err());
+        assert!(reconnected
+            .clear_failure(&proof, "table", endpoint)
+            .await
+            .is_err());
+        assert_eq!(
+            reconnected
+                .failure("table", endpoint)
+                .await
+                .unwrap()
+                .unwrap()
+                .consecutive_attempts,
+            5
+        );
+        client
+            .delete(
+                coordinator.prefix.clone(),
+                Some(etcd_client::DeleteOptions::new().with_prefix()),
+            )
+            .await
+            .unwrap();
+    }
+
     #[tokio::test]
     #[ignore = "requires isolated local ETCD_TEST_ENDPOINTS"]
     async fn failure_budget_survives_reconnect_and_lost_claim_cannot_clear_it() {

docs/concurrent-indexing.md

@@ -36,6 +36,16 @@ 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
+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
+failure time and attempt count. This does not clear durable records or change
+ownership. Arbitrary invalid input, corruption, and permission errors retain
+their longer cooldown. During a rolling upgrade, older masters may still honor
+the previously stored cooldown until the table is handled by a new master.
+
 The progress watchdog counts completed data/index reads and upload parts,
 including IO from child tasks. Response headers, metadata polling and failed IO
 do not count. Lance 9 does not report fine-grained progress from its basic BTree