feat(ecstore): add read-only multipart completion bindings

Observe active, completed and retained multipart ownership without replay, recovery, acknowledgement or reclamation. Only active uploads supply preflight metadata. Completed and retained records require validated physical evidence; completion still owns publication.

Preserve native inspection errors before reporting absence. Completed quorum retains precedence, and UnsupportedDisk keeps its compatibility behavior.

Validation: 17 binding tests and all 7,453 ecstore tests passed with four workers and no skips. Both default and internode-transport,test-util all-target Clippy passed with warnings denied. Formatting, script tests, test wiring and prepush checks passed.

Combined mutations of inspection error propagation, read quorum selection and UnsupportedDisk handling produced four intended runtime failures while two controls passed. Sources were restored by hash and actually recompiled before the full final run.

Application and CompleteMultipartUpload integration are not included.

crates/ecstore/src/disk/os.rs

@@ -4553,6 +4553,7 @@ pub(crate) mod multipart_source_syscall_failures {
     use super::*;
     #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
     pub(crate) enum Stage {
+        Read,
         Create,
         Write,
         FileSync,
@@ -4560,7 +4561,7 @@ pub(crate) mod multipart_source_syscall_failures {
     }
     type Key = (Stage, PathBuf);
     struct Failure {
-        kind: io::ErrorKind,
+        error: Mutex<Option<io::Error>>,
         triggered: std::sync::atomic::AtomicUsize,
     }
     static FAILURES: LazyLock<Mutex<HashMap<Key, Arc<Failure>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
@@ -4585,9 +4586,12 @@ pub(crate) mod multipart_source_syscall_failures {
         }
     }
     pub(crate) fn fail_once(stage: Stage, path: &Path, kind: io::ErrorKind) -> FailureGuard {
+        fail_once_with_error(stage, path, io::Error::new(kind, "injected native physical syscall failure"))
+    }
+    pub(crate) fn fail_once_with_error(stage: Stage, path: &Path, error: io::Error) -> FailureGuard {
         let key = (stage, path.to_path_buf());
         let failure = Arc::new(Failure {
-            kind,
+            error: Mutex::new(Some(error)),
             triggered: std::sync::atomic::AtomicUsize::new(0),
         });
         assert!(
@@ -4599,7 +4603,7 @@ pub(crate) mod multipart_source_syscall_failures {
     pub(crate) fn check(stage: Stage, path: &Path) -> io::Result<()> {
         if let Some(failure) = FAILURES.lock().remove(&(stage, path.to_path_buf())) {
             failure.triggered.fetch_add(1, std::sync::atomic::Ordering::Release);
-            return Err(io::Error::new(failure.kind, "injected native physical syscall failure"));
+            return Err(failure.error.lock().take().expect("one-shot physical fault owns its error"));
         }
         Ok(())
     }
@@ -8076,6 +8080,8 @@ pub(crate) fn read_multipart_source_metadata_in_pinned(
     base: &Path,
     publication_root: &PublicationRoot,
 ) -> io::Result<Vec<u8>> {
+    #[cfg(test)]
+    multipart_source_syscall_failures::check(multipart_source_syscall_failures::Stage::Read, &parent.join(name))?;
     read_multipart_metadata_in_pinned(pinned, parent, name, base, publication_root, Some(16 * 1024 * 1024))
 }
 

crates/ecstore/src/object_api/mod.rs

@@ -101,6 +101,7 @@ pub(crate) fn legacy_encrypted_range_seek_enabled() -> bool {
 mod body_cache_hook;
 mod encryption;
 mod hook_slot;
+mod multipart_completion_binding;
 mod object_mutation_hook;
 mod read_condition;
 mod readers;
@@ -126,6 +127,7 @@ pub use encryption::{
     EncryptionResolutionError, EncryptionResolutionErrorKind, ObjectEncryptionResolver, ReadEncryptionMaterial,
     ReadEncryptionMode, ReadEncryptionRequest,
 };
+pub use multipart_completion_binding::MultipartCompletionBinding;
 pub(crate) use object_mutation_hook::notify_object_mutation;
 pub use object_mutation_hook::{ObjectMutationHook, register_object_mutation_hook, unregister_object_mutation_hook};
 pub use readers::*;

crates/ecstore/src/object_api/multipart_completion_binding.rs

@@ -0,0 +1,79 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+
+//! Read-only upload ownership evidence for completion callers, not publication authority.
+//! ECStore supplies validated records; completion still rechecks its locked storage state.
+
+use crate::storage_api_contracts::multipart::MultipartInfo;
+
+/// A storage observation of the namespace belonging to an upload.
+///
+/// This does not reserve an upload or authorize publication. Abort, completion,
+/// bucket recreation, and replacement can invalidate it before completion runs.
+#[derive(Clone)]
+pub struct MultipartCompletionBinding {
+    bucket: String,
+    object: String,
+    active_info: Option<MultipartInfo>,
+}
+
+impl MultipartCompletionBinding {
+    pub(crate) fn active(info: MultipartInfo) -> Self {
+        Self {
+            bucket: info.bucket.clone(),
+            object: info.object.clone(),
+            active_info: Some(info),
+        }
+    }
+
+    pub(crate) fn recorded(bucket: &str, object: &str) -> Self {
+        Self {
+            bucket: bucket.to_owned(),
+            object: rustfs_utils::path::decode_dir_object(object),
+            active_info: None,
+        }
+    }
+
+    /// The bucket identified by the storage observation.
+    pub fn bucket(&self) -> &str {
+        &self.bucket
+    }
+
+    /// The decoded object key identified by the storage observation.
+    pub fn object(&self) -> &str {
+        &self.object
+    }
+
+    /// Metadata from an ordinary active lookup, when that lookup succeeded.
+    /// Completed and retained records never supply active preflight metadata.
+    pub fn active_info(&self) -> Option<&MultipartInfo> {
+        self.active_info.as_ref()
+    }
+}
+
+impl std::fmt::Debug for MultipartCompletionBinding {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("MultipartCompletionBinding")
+            .field("bucket", &self.bucket)
+            .field("object", &self.object)
+            .field("has_active_info", &self.active_info.is_some())
+            .finish()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn completion_binding_debug_excludes_preflight_metadata() {
+        let binding = MultipartCompletionBinding::active(MultipartInfo {
+            user_defined: [("test-marker".to_owned(), "private-preflight-sentinel".to_owned())].into(),
+            ..Default::default()
+        });
+        let debug = format!("{binding:?}");
+        assert!(!debug.contains("private-preflight-sentinel"));
+        assert!(debug.contains("has_active_info: true"));
+    }
+}

crates/ecstore/src/set_disk/core/multipart_source.rs

@@ -306,6 +306,7 @@ pub(in crate::set_disk) struct CleanupSourceMetadata {
     pub metadata: Vec<rustfs_filemeta::FileInfo>,
     pub errors: Vec<Option<DiskError>>,
     pub certified_original: Option<rustfs_filemeta::FileInfo>,
+    pub inspection_error: Option<DiskError>,
 }
 
 /// A successful close is useful only to the guarded cleanup caller. Its owned
@@ -478,6 +479,10 @@ pub(in crate::set_disk) async fn cleanup_source_metadata(
     bind_multipart_proof_snapshot(disks, authority)?;
     let ids = disk_ids(disks).await;
     let inspections = cleanup_inspections(disks, path, None).await;
+    let inspection_error = inspections.iter().find_map(|inspection| match inspection {
+        Err(DiskError::UnsupportedDisk) | Ok(_) => None,
+        Err(error) => Some(error.clone()),
+    });
     if metadata.len() != disks.len() || errors.len() != disks.len() {
         return Err(DiskError::FileCorrupt);
     }
@@ -519,6 +524,7 @@ pub(in crate::set_disk) async fn cleanup_source_metadata(
             metadata,
             errors,
             certified_original: None,
+            inspection_error,
         });
     }
     for (slot, original) in originals.iter_mut().enumerate() {
@@ -564,6 +570,7 @@ pub(in crate::set_disk) async fn cleanup_source_metadata(
         metadata,
         errors,
         certified_original,
+        inspection_error,
     })
 }
 

crates/ecstore/src/store/multipart.rs

@@ -845,6 +845,50 @@ impl ECStore {
         .await
     }
 
+    /// Observe an upload's original namespace without recovering or acknowledging it.
+    ///
+    /// Only ordinary active metadata is returned for application preflight. This
+    /// observation grants no right to publish; completion revalidates under its locks.
+    /// Cancellation leaves storage unchanged, and callers may retry this lookup.
+    ///
+    /// # Errors
+    /// Preserves argument, bucket lifetime, access, I/O and quorum errors. Fallback
+    /// discovery runs only after every ordinary probe reports InvalidUploadID.
+    pub async fn lookup_multipart_completion_binding(
+        &self,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        opts: &ObjectOptions,
+    ) -> Result<Option<crate::object_api::MultipartCompletionBinding>> {
+        check_list_parts_args(bucket, object, upload_id)?;
+        let encoded_object = encode_dir_object_ref(object);
+        let object = encoded_object.as_ref();
+        let (opts, _bucket_lifecycle_guard) = self.guard_multipart_bucket_incarnation(bucket, opts).await?;
+        let order = if self.single_pool() {
+            vec![0]
+        } else {
+            self.existing_multipart_pool_order().await
+        };
+        for &index in &order {
+            match self.pools[index].get_multipart_info(bucket, object, upload_id, &opts).await {
+                Ok(info) => return Ok(Some(crate::object_api::MultipartCompletionBinding::active(info))),
+                Err(error) if is_err_invalid_upload_id(&error) => {}
+                Err(error) => return Err(error),
+            }
+        }
+        for index in order {
+            if let Some(binding) = self.pools[index]
+                .get_disks_by_key(object)
+                .lookup_recorded_multipart_completion(bucket, object, upload_id, &opts)
+                .await?
+            {
+                return Ok(Some(binding));
+            }
+        }
+        Ok(None)
+    }
+
     #[instrument(skip(self))]
     pub(super) async fn handle_get_multipart_info(
         &self,