fix(storage): re-issue and read back conditional control-object writes

changelog.d/control-object-writes-reissue-lost-answers.fixed.md

@@ -0,0 +1,12 @@
+- On S3 and Azure, a conditional write of a cluster control object (the lock,
+  cluster state, a deployment resource, a lease heartbeat or claim) whose answer
+  was lost is re-issued and read back instead of failing on the first error. A
+  request the store refused, such as a permission error, no longer leaves its
+  key refused for the rest of the server's life. See
+  [storage failures][control-object-reissue].
+- A graph commit whose manifest request was re-issued after a lost answer, and
+  whose re-issue then met a conflict, is no longer reported as failed while the
+  first request may still land on S3: its outcome stays unknown until a read
+  shows it.
+
+[control-object-reissue]: ../docs/user/operations/storage-failures.md#retries

crates/omnigraph-catalog/src/tests.rs

@@ -15201,7 +15201,7 @@ async fn manifest_create_without_a_definite_answer_stays_in_doubt_issue_1067() {
         fault
             .manifest_puts
             .load(std::sync::atomic::Ordering::SeqCst),
-        1 + crate::manifest_request::MANIFEST_REQUEST_REISSUES as usize,
+        1 + crate::manifest_request::CONDITIONAL_REQUEST_REISSUES as usize,
         "the original request and every re-issue"
     );
 }

crates/omnigraph-core/src/lance_clone.rs

@@ -15,9 +15,13 @@ use lance_table::io::commit::{
 use object_store::ObjectStore as RawObjectStore;
 use object_store::path::Path;
 
+use omnigraph_storage::conditional::{
+    Answer as RequestAnswer, ConditionalRequest, Resolution, Step,
+};
+
 use crate::manifest_request::{
-    FailedRequest, MANIFEST_REQUEST_REISSUES, ManifestRequest, ManifestRequestAttempt,
-    ManifestRequestProbe, classify_failed_request, reissue_pause, request_error_class,
+    FailedRequest, ManifestRequest, ManifestRequestAttempt, ManifestRequestProbe,
+    classify_failed_request, request_error_class,
 };
 
 #[cfg(test)]
@@ -212,16 +216,17 @@ pub async fn create_branch(
 }
 
 impl ManifestLocationCommitHandler {
-    /// Send the commit's manifest put-if-absent until the store answers it.
+    /// Send the commit's manifest put-if-absent until the store answers it,
+    /// by the shared conditional-request policy
+    /// ([`omnigraph_storage::conditional`]).
     ///
     /// A version the store holds goes back to Lance as a conflict, which Lance
     /// reads back by its complete transaction: this commit's own earlier
     /// request is then a success, another commit's a conflict. An ambiguous
     /// failure is re-issued with the same manifest and the same precondition,
-    /// so at most one copy applies, and the key's next definite answer
-    /// settles every earlier request for it. A request still ambiguous after
-    /// the last re-issue goes back to Lance as it failed, recorded as
-    /// unresolved for the committer.
+    /// so at most one copy applies. A request that a re-issue's conflict or
+    /// the last lost answer ends goes back to Lance as it failed, recorded as
+    /// unresolved for the committer: the earlier copy may still land.
     #[allow(clippy::too_many_arguments)]
     async fn resolve_manifest_request(
         &self,
@@ -239,9 +244,11 @@ impl ManifestLocationCommitHandler {
                 attempt.finish(outcome);
             }
         };
-        let mut reissues = 0;
+        // Lance reports a held version as a conflict and reads it back, so
+        // the first copy's conflict ends the request.
+        let mut request = ConditionalRequest::new(true);
         loop {
-            let result = self
+            let answer = match self
                 .inner
                 .commit(
                     manifest,
@@ -252,9 +259,42 @@ impl ManifestLocationCommitHandler {
                     naming_scheme,
                     transaction.clone(),
                 )
-                .await;
-            let error = match result {
-                Ok(location) => {
+                .await
+            {
+                Ok(location) => RequestAnswer::Applied(location),
+                Err(CommitError::CommitConflict) => {
+                    RequestAnswer::Conflict(CommitError::CommitConflict)
+                }
+                Err(CommitError::OtherError(error)) => match classify_failed_request(&error) {
+                    FailedRequest::Refused => {
+                        RequestAnswer::Refused(CommitError::OtherError(error))
+                    }
+                    FailedRequest::Ambiguous => RequestAnswer::Lost(CommitError::OtherError(error)),
+                },
+            };
+            let resolution = match request.answer(answer) {
+                Step::Done(resolution) => resolution,
+                Step::Reissue {
+                    error,
+                    reissue,
+                    pause,
+                } => {
+                    tracing::warn!(
+                        version = manifest.version,
+                        reissue,
+                        error_class = commit_error_class(&error),
+                        "manifest request outcome unknown; re-issuing the identical conditional \
+                         request"
+                    );
+                    tokio::time::sleep(pause).await;
+                    continue;
+                }
+            };
+            return match resolution {
+                Resolution::Applied {
+                    value: location,
+                    reissues,
+                } => {
                     if reissues > 0 {
                         tracing::info!(
                             version = manifest.version,
@@ -263,41 +303,53 @@ impl ManifestLocationCommitHandler {
                         );
                     }
                     finish(attempt, ManifestRequest::Landed);
-                    return Ok(location);
+                    Ok(location)
                 }
-                Err(CommitError::CommitConflict) => {
+                Resolution::Conflict(error) => {
                     finish(attempt, ManifestRequest::Held);
-                    return Err(CommitError::CommitConflict);
+                    Err(error)
                 }
-                Err(CommitError::OtherError(error)) => error,
-            };
-            if reissues == 0 && classify_failed_request(&error) == FailedRequest::Refused {
-                finish(attempt, ManifestRequest::Refused);
-                return Err(CommitError::OtherError(error));
-            }
-            if reissues == MANIFEST_REQUEST_REISSUES {
-                tracing::error!(
-                    version = manifest.version,
+                Resolution::Refused(error) => {
+                    finish(attempt, ManifestRequest::Refused);
+                    Err(error)
+                }
+                Resolution::Unsettled {
+                    error,
+                    conflict,
                     reissues,
-                    error_class = request_error_class(&error),
-                    "manifest request outcome unknown after every re-issue; the version may still \
-                     hold this commit"
-                );
-                finish(attempt, ManifestRequest::Unresolved);
-                return Err(CommitError::OtherError(error));
-            }
-            reissues += 1;
-            tracing::warn!(
-                version = manifest.version,
-                reissue = reissues,
-                error_class = request_error_class(&error),
-                "manifest request outcome unknown; re-issuing the identical conditional request"
-            );
-            tokio::time::sleep(reissue_pause(reissues)).await;
+                } => {
+                    if conflict {
+                        tracing::warn!(
+                            version = manifest.version,
+                            reissues,
+                            "manifest version held after a re-issue; the earlier copy may hold \
+                             it or still be in flight, so its absence on read-back proves nothing"
+                        );
+                    } else {
+                        tracing::error!(
+                            version = manifest.version,
+                            reissues,
+                            error_class = commit_error_class(&error),
+                            "manifest request outcome unknown after every re-issue; the version \
+                             may still hold this commit"
+                        );
+                    }
+                    finish(attempt, ManifestRequest::Unresolved);
+                    Err(error)
+                }
+            };
         }
     }
 }
 
+/// A log-safe class of a failed manifest commit.
+fn commit_error_class(error: &CommitError) -> &'static str {
+    match error {
+        CommitError::CommitConflict => "commit_conflict",
+        CommitError::OtherError(error) => request_error_class(error),
+    }
+}
+
 #[async_trait]
 impl CommitHandler for ManifestLocationCommitHandler {
     /// Whether a version read back as missing after a failed commit proves
@@ -779,14 +831,20 @@ mod tests {
 
     /// The handler re-issues an ambiguous manifest request until the store
     /// answers it; a definite refusal, a held version and a landed manifest
-    /// end it at once; and a request no answer settles stays unresolved.
+    /// end it at once; and a request no answer settles stays unresolved, as
+    /// does one whose re-issue meets a held version.
     #[tokio::test]
     async fn manifest_request_is_re_issued_until_the_store_answers_issue_1067() {
         use Answer::*;
         assert_eq!(resolve(&[Landed]).await, (true, 1, ManifestRequest::Landed));
+        assert_eq!(
+            resolve(&[Held]).await,
+            (false, 1, ManifestRequest::Held),
+            "a held version goes to Lance's read-back, which tells this commit from another"
+        );
         assert_eq!(
             resolve(&[LandedWithLocalError, NativeConditionalPut]).await,
-            (false, 2, ManifestRequest::Held),
+            (false, 2, ManifestRequest::Unresolved),
             "a local error after manifest creation must not be called a refusal"
         );
         assert_eq!(
@@ -796,8 +854,9 @@ mod tests {
         );
         assert_eq!(
             resolve(&[Ambiguous, Held]).await,
-            (false, 2, ManifestRequest::Held),
-            "a held version goes to Lance's read-back, which tells this commit from another"
+            (false, 2, ManifestRequest::Unresolved),
+            "a re-issue's conflict can answer the earlier copy still in flight (S3's 409): \
+             Lance's read-back decides, and a missing version is not proof of absence"
         );
         assert_eq!(
             resolve(&[Refused]).await,
@@ -809,7 +868,7 @@ mod tests {
             (true, 3, ManifestRequest::Landed),
             "after an ambiguous request, a refusal does not settle the key"
         );
-        let unanswered = [Ambiguous; 1 + MANIFEST_REQUEST_REISSUES as usize];
+        let unanswered = [Ambiguous; 1 + omnigraph_storage::CONDITIONAL_REQUEST_REISSUES as usize];
         assert_eq!(
             resolve(&unanswered).await,
             (false, unanswered.len(), ManifestRequest::Unresolved)
@@ -822,7 +881,8 @@ mod tests {
     /// still land keeps the transaction file its manifest would name.
     #[tokio::test]
     async fn a_shared_handler_never_calls_a_missing_version_definitive_issue_1067() {
-        let unanswered = [Answer::Ambiguous; 1 + MANIFEST_REQUEST_REISSUES as usize];
+        let unanswered =
+            [Answer::Ambiguous; 1 + omnigraph_storage::CONDITIONAL_REQUEST_REISSUES as usize];
         for answers in [&[Answer::Landed][..], &[Answer::Refused], &unanswered] {
             let handler = wrapped(Arc::new(ScriptedBackend::new(answers)), None);
             assert!(!handler.is_version_not_found_definitive(), "{answers:?}");

crates/omnigraph-core/src/manifest_request.rs

@@ -19,21 +19,15 @@
 //! conflicting key by comparing the complete transaction it records.
 
 use std::sync::{Arc, Mutex};
-use std::time::Duration;
 
 use lance_core::Error;
+use omnigraph_storage::conditional::{self, RequestClient};
 
-/// How many times one manifest request is re-issued after an ambiguous
-/// answer, and the pause before each re-issue: 0.2 s doubling to 1.6 s, 3 s
-/// in all, beside each request's own client timeout. A store that answers
-/// nothing for that long is failing, not throttling one request.
-pub const MANIFEST_REQUEST_REISSUES: u32 = 4;
-const FIRST_REISSUE_PAUSE: Duration = Duration::from_millis(200);
-
-/// The pause before re-issue `attempt` (1-based).
-pub fn reissue_pause(attempt: u32) -> Duration {
-    FIRST_REISSUE_PAUSE * 2u32.saturating_pow(attempt.saturating_sub(1))
-}
+/// The re-issue budget, pauses and failure classes are the storage crate's
+/// one policy for conditional requests, shared with control-object writes.
+pub use omnigraph_storage::conditional::{
+    CONDITIONAL_REQUEST_REISSUES, FailedRequest, reissue_pause,
+};
 
 /// What a commit's manifest requests established by the time the commit
 /// returned. Ordered by what a caller may conclude: everything above
@@ -54,7 +48,9 @@ pub enum ManifestRequest {
     Landed,
     /// A request may have applied, or may still, and no later request for
     /// the key got an answer that settles it. The version may hold this
-    /// commit, now or later.
+    /// commit, now or later. A re-issue that meets a held version ends here
+    /// too: on S3 the conflict can answer the earlier copy, still in flight,
+    /// so Lance's read-back decides and a missing version proves nothing.
     Unresolved,
 }
 
@@ -126,17 +122,6 @@ impl Drop for ManifestRequestAttempt {
     }
 }
 
-/// How the store answered a manifest put that failed other than by holding
-/// the version.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum FailedRequest {
-    /// The request was refused before it could apply, or never left the
-    /// process: no copy of it can land.
-    Refused,
-    /// The request may have applied, or may still.
-    Ambiguous,
-}
-
 /// A log-safe name for a failed request's error: its object-store variant,
 /// never its message, which can carry object paths.
 pub fn request_error_class(error: &Error) -> &'static str {
@@ -160,37 +145,23 @@ pub fn request_error_class(error: &Error) -> &'static str {
     }
 }
 
-/// Classify a manifest put's failure by the typed error the store returned.
+/// Classify a manifest put's failure by the typed error the store returned,
+/// with the storage crate's classification on a repeating client.
 ///
-/// A remote store's generic failure (a 5xx, a throttle, a timeout, a lost
-/// connection) is ambiguous, and so are an authorization failure and a join
-/// error, which can follow an earlier internal attempt of the same call that
-/// the server applied. A local filesystem failure is ambiguous too: Lance's
-/// Windows rename handler can create the manifest before deleting its staging
-/// file fails. A missing key or container, an invalid path, an unsupported
-/// operation, and a refusal by the owning storage scope, which sends nothing,
-/// are definite. Anything this function cannot recognise is ambiguous.
+/// Lance's own "not found" is a refusal. The handler cannot tell Lance's
+/// provider from a storage scope's single-attempt client, so it takes the
+/// client to repeat requests: an authorization failure and a join error can
+/// then follow an earlier internal attempt that the server applied. A local
+/// filesystem failure is ambiguous too, by the shared rule: Lance's Windows
+/// rename handler can create the manifest before deleting its staging file
+/// fails. An error that is not the store's is ambiguous.
 pub fn classify_failed_request(error: &Error) -> FailedRequest {
-    use object_store::Error as Store;
-
-    let store = match error {
-        Error::NotFound { .. } => return FailedRequest::Refused,
-        Error::IO { source, .. } => source.downcast_ref::<Store>(),
-        _ => None,
-    };
-    match store {
-        Some(Store::Generic { store, .. })
-            if *store == "InMemory" || *store == omnigraph_storage::OWNED_STORE_NAME =>
-        {
-            FailedRequest::Refused
-        }
-        Some(
-            Store::NotFound { .. }
-            | Store::InvalidPath { .. }
-            | Store::NotSupported { .. }
-            | Store::NotImplemented { .. }
-            | Store::UnknownConfigurationKey { .. },
-        ) => FailedRequest::Refused,
+    match error {
+        Error::NotFound { .. } => FailedRequest::Refused,
+        Error::IO { source, .. } => match source.downcast_ref::<object_store::Error>() {
+            Some(store) => conditional::classify_failed_request(store, RequestClient::Repeating),
+            None => FailedRequest::Ambiguous,
+        },
         _ => FailedRequest::Ambiguous,
     }
 }
@@ -287,13 +258,4 @@ mod tests {
             assert!(!request_error_class(&error).contains("bucket"));
         }
     }
-
-    #[test]
-    fn reissue_pauses_double_and_stay_bounded() {
-        let pauses: Vec<_> = (1..=MANIFEST_REQUEST_REISSUES).map(reissue_pause).collect();
-        assert_eq!(
-            pauses,
-            [200, 400, 800, 1600].map(Duration::from_millis).to_vec()
-        );
-    }
 }

crates/omnigraph-storage/Cargo.toml

@@ -21,6 +21,7 @@ url = { workspace = true }
 
 [dev-dependencies]
 tempfile = { workspace = true }
+tokio = { workspace = true, features = ["test-util"] }
 
 [lints]
 workspace = true