fix(ecstore): preserve remote preparation corruption

crates/capture-authority/src/application.rs

@@ -1142,3 +1142,85 @@ fn validate_application_tables(
     }
     Ok(first_operations)
 }
+
+impl Store {
+    pub(crate) fn authorize_created(
+        &self,
+        expected: &DecideCreated,
+        decision: LogId,
+    ) -> Result<crate::CommittedCreatedAuthorization, StoreError> {
+        if !self.writable {
+            return Err(StoreError::WriteDisabled);
+        }
+        expected.encode()?;
+        let tx = self.database.begin_read()?;
+        let meta = tx.open_table(META)?;
+        let schema = number(required(meta.get("schema")?)?.value())?;
+        if !matches!(schema, 6 | 7) {
+            return Err(StoreError::UnsupportedSchema);
+        }
+        let (applied, _) = native_application_state(&tx)?;
+        let committed = crate::committed::validate_committed(&meta, &tx.open_table(LOGS)?)?;
+        let identity = crate::decode_identity(required(meta.get("identity")?)?.value())?;
+        let binding = read_binding(&meta)?;
+        let operations = tx.open_table(OPERATIONS)?;
+        let record = operations
+            .get(expected.operation_id.as_slice())?
+            .ok_or(StoreError::CreatedAuthorizationMismatch)?;
+        let provenance = decode_provenance(record.value())?;
+        let command = DecideCreated::decode(&provenance.operation.canonical_command)?;
+        if command.encode()? != provenance.operation.canonical_command {
+            return Err(StoreError::CorruptRecord);
+        }
+        if command.operation_id != expected.operation_id
+            || command.object != expected.object
+            || command.expected_head != expected.expected_head
+            || command.binding_revision != expected.binding_revision
+        {
+            return Err(StoreError::CreatedAuthorizationMismatch);
+        }
+        if command.prepared != expected.prepared {
+            return Err(StoreError::CreatedAuthorizationMismatch);
+        }
+        let CreatedResult::Created {
+            version_id,
+            event_id,
+            decision_log_id,
+        } = provenance.operation.result
+        else {
+            return Err(StoreError::CreatedAuthorizationMismatch);
+        };
+        if decision_log_id != decision
+            || provenance.first_decision_log_id != decision
+            || version_id != command.prepared.version_id
+            || event_id != command.operation_id
+            || binding.revision != command.binding_revision
+        {
+            return Err(StoreError::CreatedAuthorizationMismatch);
+        }
+        for boundary in [applied, committed] {
+            if boundary.is_none_or(|boundary| {
+                boundary.index < decision.index || (boundary.index == decision.index && boundary != decision)
+            }) {
+                return Err(StoreError::CreatedAuthorizationMismatch);
+            }
+        }
+        let events = tx.open_table(EVENTS)?;
+        let event = decode_event(required(events.get(command.operation_id.as_slice())?)?.value())?;
+        let versions = tx.open_table(VERSIONS)?;
+        let version = decode_version(
+            required(versions.get(version_key(&command.object, command.prepared.version_id)?.as_slice())?)?.value(),
+        )?;
+        if event.object != command.object
+            || event.prepared != command.prepared
+            || event.decision_log_id != decision
+            || event.binding != binding
+            || version.prepared != command.prepared
+            || version.decision_log_id != decision
+            || version.binding_revision != binding.revision
+        {
+            return Err(StoreError::CorruptRecord);
+        }
+        Ok(crate::CommittedCreatedAuthorization::verified(identity, command, binding, decision))
+    }
+}

crates/capture-authority/src/committed_created_authorization_tests.rs

@@ -0,0 +1,278 @@
+//! Reuse the existing actual Raft owner fixture, never a caller-constructed result proof.
+use super::*;
+
+async fn decide(cluster: &Cluster, value: u8) -> (DecideCreated, LogId) {
+    let input = command(value);
+    let expected = input.created().clone();
+    let response = cluster
+        .node(0)
+        .owner()
+        .client_write(input)
+        .await
+        .expect("actual Raft Created decision");
+    let Some(CreatedResult::Created { decision_log_id, .. }) = response.data.created_result() else {
+        panic!("successful Created result required")
+    };
+    (expected, *decision_log_id)
+}
+
+#[tokio::test]
+async fn committed_created_authorization_exact() {
+    let cluster = Cluster::start(&[0], 100).await;
+    let (expected, first) = decide(&cluster, 1).await;
+    let authorization = cluster
+        .node(0)
+        .owner()
+        .authorize_created(expected.clone(), first)
+        .await
+        .expect("durable exact authorization");
+    assert_eq!(authorization.identity(), identity(0));
+    assert_eq!(authorization.operation_id(), expected.operation_id);
+    assert_eq!(authorization.object(), &expected.object);
+    assert_eq!(authorization.prepared(), &expected.prepared);
+    assert_eq!(authorization.binding(), &binding());
+    assert_eq!(authorization.decision_log_id(), first);
+    assert_eq!(cluster.node(0).page().await.events[0].captured.decision_log_id, first);
+    cluster.stop().await;
+}
+
+#[tokio::test]
+async fn committed_created_authorization_retry_first_decision() {
+    let cluster = Cluster::start(&[0], 100).await;
+    let (expected, first) = decide(&cluster, 2).await;
+    let before = cluster
+        .node(0)
+        .owner()
+        .authorize_created(expected.clone(), first)
+        .await
+        .expect("first authorization");
+    let response = cluster
+        .node(0)
+        .owner()
+        .client_write(CaptureCommand::try_created(expected.clone()).expect("retry command"))
+        .await
+        .expect("real retry");
+    assert_ne!(LogId::from(response.log_id), first);
+    assert!(
+        matches!(response.data.created_result(), Some(CreatedResult::Created{decision_log_id,..}) if *decision_log_id==first)
+    );
+    let after = cluster
+        .node(0)
+        .owner()
+        .authorize_created(expected.clone(), first)
+        .await
+        .expect("first decision retained");
+    assert_eq!(before, after);
+    assert!(
+        cluster
+            .node(0)
+            .owner()
+            .authorize_created(expected, response.log_id.into())
+            .await
+            .is_err()
+    );
+    assert_eq!(cluster.node(0).page().await.events.len(), 1);
+    cluster.stop().await;
+}
+
+#[tokio::test]
+async fn committed_created_authorization_rejects_mismatch() {
+    let cluster = Cluster::start(&[0], 100).await;
+    let (expected, first) = decide(&cluster, 3).await;
+    let before = cluster.node(0).page().await;
+    let durable_before = cluster
+        .node(0)
+        .writer
+        .write(|store| {
+            Ok((
+                store.read_vote()?,
+                store.read_committed()?,
+                store.read_entries(0, u64::MAX)?,
+                store.native_application_state()?,
+                store.created_events_page(None, 256)?,
+            ))
+        })
+        .await
+        .expect("durable state before rejected authorization");
+    let mut variants = Vec::new();
+    let mut altered = expected.clone();
+    altered.operation_id = [4; 16];
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.object.key.push(9);
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.object.bucket_incarnation = [9; 16];
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.expected_head = Some([9; 16]);
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.prepared.preparation_id = [9; 16];
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.prepared.version_id = [9; 16];
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.prepared.content_digest = [9; 32];
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.prepared.content_length += 1;
+    variants.push(altered);
+    let mut altered = expected.clone();
+    altered.binding_revision += 1;
+    variants.push(altered);
+    for altered in variants {
+        assert!(
+            cluster.node(0).owner().authorize_created(altered, first).await.is_err(),
+            "AUTHORIZATION_REQUIRES_EXACT_PREPARED_IDENTITY_AND_COMMAND"
+        );
+    }
+    for wrong in [
+        LogId {
+            term: first.term + 1,
+            ..first
+        },
+        LogId {
+            leader_node: first.leader_node + 1,
+            ..first
+        },
+        LogId {
+            index: first.index + 1,
+            ..first
+        },
+    ] {
+        assert!(
+            cluster
+                .node(0)
+                .owner()
+                .authorize_created(expected.clone(), wrong)
+                .await
+                .is_err()
+        );
+    }
+    let durable_after = cluster
+        .node(0)
+        .writer
+        .write(|store| {
+            Ok((
+                store.read_vote()?,
+                store.read_committed()?,
+                store.read_entries(0, u64::MAX)?,
+                store.native_application_state()?,
+                store.created_events_page(None, 256)?,
+            ))
+        })
+        .await
+        .expect("durable state after rejected authorization");
+    assert_eq!(durable_before, durable_after, "REJECTED_AUTHORIZATION_DOES_NOT_WRITE");
+    let mut failed = expected.clone();
+    failed.operation_id = [5; 16];
+    failed.prepared.version_id = [5; 16];
+    let response = cluster
+        .node(0)
+        .owner()
+        .client_write(CaptureCommand::try_created(failed.clone()).expect("failed command"))
+        .await
+        .expect("real head mismatch decision");
+    assert!(matches!(response.data.created_result(), Some(CreatedResult::HeadMismatch { .. })));
+    assert!(
+        cluster
+            .node(0)
+            .owner()
+            .authorize_created(failed, response.log_id.into())
+            .await
+            .is_err()
+    );
+    assert_eq!(cluster.node(0).page().await, before);
+    cluster.stop().await;
+}
+
+#[tokio::test]
+async fn committed_created_authorization_restart() {
+    let routes = Arc::new(Routes::default());
+    let mut node = Node::start(0, routes.clone(), config(100)).await;
+    node.owner().bootstrap(members(&[0])).await.expect("bootstrap actual owner");
+    node.owner()
+        .raft()
+        .wait(Some(Duration::from_secs(15)))
+        .state(ServerState::Leader, "leader")
+        .await
+        .expect("leader ready");
+    let expected = command(6).created().clone();
+    let response = node.owner().client_write(command(6)).await.expect("real first write");
+    let Some(CreatedResult::Created {
+        decision_log_id: first, ..
+    }) = response.data.created_result()
+    else {
+        panic!("Created")
+    };
+    let first = *first;
+    let before = node
+        .owner()
+        .authorize_created(expected.clone(), first)
+        .await
+        .expect("before restart");
+    node.stop(&routes).await;
+    let Node {
+        directory,
+        writer,
+        owner: _,
+        id,
+    } = node;
+    drop(writer);
+    let writer = NativeLogWriter::new(Store::open(directory.path().join("db"), identity(id)).expect("retained redb"));
+    // A retained commit-before-apply crash fixture is not a client-provided proof.
+    // Only real owner initialization may replay it before authorization can succeed.
+    use openraft::storage::{RaftLogStorage, RaftLogStorageExt};
+    let pending = command(7).created().clone();
+    let mut logs = NativeRaftLogStore::new(writer.clone());
+    let last = logs.get_log_state().await.expect("retained tail").last_log_id.expect("tail");
+    let pending_id = LogId {
+        term: last.leader_id.term,
+        leader_node: last.leader_id.node_id,
+        index: last.index + 1,
+    };
+    logs.blocking_append(vec![CaptureRaftEntry {
+        log_id: pending_id.into(),
+        payload: openraft::EntryPayload::Normal(CaptureCommand::try_created(pending.clone()).expect("pending command")),
+    }])
+    .await
+    .expect("retained crash suffix");
+    logs.save_committed(Some(pending_id.into()))
+        .await
+        .expect("retained commit marker");
+    assert!(
+        writer.authorize_created(pending.clone(), pending_id).await.is_err(),
+        "UNAPPLIED_COMMIT_IS_NOT_AUTHORIZATION"
+    );
+    drop(logs);
+    let owner = NativeRaftOwner::start(
+        writer.clone(),
+        id,
+        config(100),
+        CheckpointLimits::default(),
+        DirectFactory {
+            source: id,
+            routes: routes.clone(),
+        },
+    )
+    .await
+    .expect("real recovery");
+    routes.register(id, owner.raft().clone());
+    let after = owner
+        .authorize_created(expected, first)
+        .await
+        .expect("durable authorization after restart");
+    assert_eq!(before, after);
+    let replayed = owner
+        .authorize_created(pending.clone(), pending_id)
+        .await
+        .expect("actual owner committed recovery authorizes");
+    assert_eq!(replayed.prepared(), &pending.prepared);
+    assert_eq!(replayed.decision_log_id(), pending_id);
+    owner.shutdown().await.expect("drain recovered owner");
+    routes.remove(id);
+    drop(owner);
+    drop(writer);
+}

crates/capture-authority/src/created.rs

@@ -74,6 +74,52 @@ pub struct CreatedApplyResult {
     pub result: CreatedResult,
 }
 
+/// Local publication authority re-read from the actual owner's durable Created decision.
+/// This is not a serializable remote proof and does not attest physical preparation durability.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct CommittedCreatedAuthorization {
+    identity: crate::NodeIdentity,
+    command: DecideCreated,
+    binding: CaptureBinding,
+    decision_log_id: LogId,
+}
+impl CommittedCreatedAuthorization {
+    pub(crate) fn verified(
+        identity: crate::NodeIdentity,
+        command: DecideCreated,
+        binding: CaptureBinding,
+        decision_log_id: LogId,
+    ) -> Self {
+        Self {
+            identity,
+            command,
+            binding,
+            decision_log_id,
+        }
+    }
+    pub fn identity(&self) -> crate::NodeIdentity {
+        self.identity
+    }
+    pub fn operation_id(&self) -> [u8; 16] {
+        self.command.operation_id
+    }
+    pub fn object(&self) -> &ObjectIdentity {
+        &self.command.object
+    }
+    pub fn expected_head(&self) -> Option<[u8; 16]> {
+        self.command.expected_head
+    }
+    pub fn prepared(&self) -> &PreparedIdentity {
+        &self.command.prepared
+    }
+    pub fn binding(&self) -> &CaptureBinding {
+        &self.binding
+    }
+    pub fn decision_log_id(&self) -> LogId {
+        self.decision_log_id
+    }
+}
+
 pub(crate) struct Decoder<'a> {
     bytes: &'a [u8],
 }

crates/capture-authority/src/lib.rs

@@ -43,8 +43,9 @@ pub use native_snapshot::{NativeSnapshotError, NativeSnapshotView};
 mod raft_types;
 pub mod snapshot;
 pub use created::{
-    CaptureBinding, CreatedApplyResult, CreatedEvent, CreatedOperation, CreatedOperationProvenance, CreatedResult,
-    CreatedVersion, DecideCreated, MAX_OBJECT_KEY_LENGTH, ObjectIdentity, PreparedIdentity,
+    CaptureBinding, CommittedCreatedAuthorization, CreatedApplyResult, CreatedEvent, CreatedOperation,
+    CreatedOperationProvenance, CreatedResult, CreatedVersion, DecideCreated, MAX_OBJECT_KEY_LENGTH, ObjectIdentity,
+    PreparedIdentity,
 };
 pub use node::{CaptureNode, CaptureNodeError};
 pub use raft_types::{
@@ -105,6 +106,14 @@ impl NativeLogWriter {
         }
     }
 
+    async fn authorize_created(
+        &self,
+        expected: DecideCreated,
+        decision: LogId,
+    ) -> Result<CommittedCreatedAuthorization, NativeWriteError> {
+        self.write(move |store| store.authorize_created(&expected, decision)).await
+    }
+
     pub async fn snapshot_view(&self) -> Result<NativeSnapshotView, NativeSnapshotError> {
         let store = self.store.clone();
         tokio::task::spawn_blocking(move || NativeSnapshotView::acquire(store))
@@ -309,6 +318,8 @@ pub enum StoreError {
     UnsupportedSchema,
     #[error("corrupt record")]
     CorruptRecord,
+    #[error("Created publication authorization does not match a committed successful decision")]
+    CreatedAuthorizationMismatch,
     #[error("invalid event page limit")]
     InvalidEventPage,
     #[error("invalid membership")]
@@ -360,6 +371,7 @@ impl StoreError {
     pub fn code(&self) -> &'static str {
         match self {
             Self::IdentityMismatch => "identity_mismatch",
+            Self::CreatedAuthorizationMismatch => "created_authorization_mismatch",
             Self::InvalidEventPage => "invalid_event_page",
             Self::UnsupportedSchema => "unsupported_schema",
             Self::CorruptRecord => "corrupt_record",

crates/capture-authority/src/raft_owner.rs

@@ -205,6 +205,17 @@ impl NativeRaftOwner {
             .map_err(|error| NativeRaftOwnerError::Write(Box::new(error)))
     }
 
+    /// Re-read an exact successful first decision from this owner's durable view.
+    /// Use the Created result's decision LogId, not an idempotent retry entry's outer LogId.
+    pub async fn authorize_created(
+        &self,
+        expected: crate::DecideCreated,
+        expected_decision: crate::LogId,
+    ) -> Result<crate::CommittedCreatedAuthorization, NativeRaftOwnerError> {
+        self.require_running()?;
+        Ok(self.writer.authorize_created(expected, expected_decision).await?)
+    }
+
     pub fn metrics(&self) -> watch::Receiver<RaftMetrics<u64, CaptureNode>> {
         self.raft.metrics()
     }

crates/capture-authority/src/raft_owner_tests.rs

@@ -880,3 +880,6 @@ async fn raft_owner_shutdown_inflight_replication() {
     drop(store);
     cluster.stop().await;
 }
+
+#[path = "committed_created_authorization_tests.rs"]
+mod authorization;

crates/ecstore/src/api/mod.rs

@@ -644,7 +644,13 @@ pub mod store_list {
 
 pub mod storage {
     pub use crate::core::pools::{HealLifecycleExpiryContext, POOL_META_NAME};
+    pub use crate::disk::local::native_prepare::{
+        NativePreparationDescription, NativePreparationProbe, NativePreparationReceipt,
+    };
     pub use crate::store::HealWalkVersion;
+    pub use crate::store::native_prepare::{
+        NativePreparationFinalizeRequest, NativePreparationRecovery, NativePreparationTarget,
+    };
     pub use crate::store::{
         BootstrapLocalTarget, ECStore, SCANNER_PUBLICATION_LEASE_TTL_MS, ScannerDataMovementPauseStatus, all_local_disk,
         all_local_disk_path, find_local_disk_by_ref, init_local_disks, init_local_disks_with_instance_ctx, init_lock_clients,

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -58,13 +58,15 @@ use rustfs_protos::proto_gen::node_service::{
     BatchReadVersionRequest, BatchReadVersionResponse, CheckPartsRequest, CompareAndUpdateFileOutcome,
     CompareAndUpdateFileRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest,
     DeleteVersionsResponse, DeleteVolumeRequest, DiskInfoRequest, ListDirRequest, ListVolumesRequest, MakeVolumeRequest,
-    MakeVolumesRequest, PreparePartTransactionRequest, ReadAllRequest, ReadMetadataRequest, ReadMultipleRequest,
-    ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest, RenameDataRequest, RenameFileRequest,
-    SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest, SnapshotLeaseRequest,
-    SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest, WriteMetadataRequest,
-    node_service_client::NodeServiceClient,
+    MakeVolumesRequest, NativePreparationBeginRequest, NativePreparationFinalizeRequest, NativePreparationLocateRequest,
+    NativePreparationProbeRequest, NativePreparationRecoverRequest, PreparePartTransactionRequest, ReadAllRequest,
+    ReadMetadataRequest, ReadMultipleRequest, ReadMultipleResponse, ReadPartsRequest, ReadVersionRequest, ReadXlRequest,
+    RenameDataRequest, RenameFileRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest, SnapshotLeaseRenewRequest,
+    SnapshotLeaseRequest, SnapshotLeaseResponse, StatVolumeRequest, UpdateMetadataRequest, VerifyFileRequest, WriteAllRequest,
+    WriteMetadataRequest, node_service_client::NodeServiceClient,
 };
 use serde::{Serialize, de::DeserializeOwned};
+use sha2::{Digest, Sha256};
 use std::{
     future::Future,
     io::Cursor,
@@ -176,6 +178,172 @@ fn attach_mutation_body_digest<T>(
     crate::cluster::rpc::set_tonic_rolling_canonical_body_digest(request, &canonical_body).map_err(Error::other)
 }
 
+fn native_response_proof(body: std::result::Result<Vec<u8>, std::num::TryFromIntError>, proof: &[u8]) -> Result<()> {
+    let body = body.map_err(|_| Error::other("native preparation response is too large to authenticate"))?;
+    if proof.len() != 32 {
+        return Err(Error::other("native preparation response proof has invalid length"));
+    }
+    crate::cluster::rpc::verify_tonic_rpc_response_proof(&body, proof)
+        .map_err(|_| Error::other("native preparation response proof is invalid"))
+}
+
+fn native_response_error(error: Option<rustfs_protos::proto_gen::node_service::Error>) -> DiskError {
+    let error = DiskError::from(error.unwrap_or_default());
+    match error {
+        DiskError::FileCorrupt => DiskError::Io(std::io::Error::new(std::io::ErrorKind::InvalidData, error)),
+        error => error,
+    }
+}
+
+fn native_uuid_field(value: &[u8], field: &str) -> Result<Uuid> {
+    let id = Uuid::from_slice(value).map_err(|_| Error::other(format!("native preparation returned invalid {field}")))?;
+    if id.is_nil() {
+        return Err(Error::other(format!("native preparation returned nil {field}")));
+    }
+    Ok(id)
+}
+
+fn native_version_field(value: &[u8]) -> Result<Uuid> {
+    Uuid::from_slice(value).map_err(|_| Error::other("native preparation returned invalid version id"))
+}
+
+fn native_object_length_from_metadata(input: &crate::disk::local::native_prepare::NativePrepareInput) -> Result<u64> {
+    let raw = rustfs_filemeta::FileMeta::load(&input.metadata).map_err(Error::other)?;
+    if raw.versions.len() != 1 {
+        return Err(Error::other("native preparation metadata must have one version"));
+    }
+    let info = rustfs_filemeta::get_file_info(
+        &input.metadata,
+        &input.bucket,
+        &input.key,
+        &input.version.to_string(),
+        rustfs_filemeta::FileInfoOpts {
+            data: true,
+            include_free_versions: false,
+            include_part_checksums: true,
+        },
+    )
+    .map_err(Error::other)?;
+    if info.version_id != Some(input.version) || info.deleted || info.mark_deleted || info.size < 0 {
+        return Err(Error::other("native preparation metadata identity mismatch"));
+    }
+    u64::try_from(info.size).map_err(Error::other)
+}
+
+fn validate_native_metadata_for_version(
+    metadata: &[u8],
+    expected_digest: Option<&[u8; 32]>,
+    bucket: &str,
+    key: &str,
+    version: Uuid,
+) -> Result<()> {
+    validate_native_metadata(metadata, expected_digest)?;
+    let raw = rustfs_filemeta::FileMeta::load(metadata).map_err(Error::other)?;
+    if raw.versions.len() != 1 {
+        return Err(Error::other("native preparation metadata must have one version"));
+    }
+    let info = rustfs_filemeta::get_file_info(
+        metadata,
+        bucket,
+        key,
+        &version.to_string(),
+        rustfs_filemeta::FileInfoOpts {
+            data: true,
+            include_free_versions: false,
+            include_part_checksums: true,
+        },
+    )
+    .map_err(Error::other)?;
+    if info.version_id != Some(version) || info.deleted || info.mark_deleted || info.size < 0 {
+        return Err(Error::other("native preparation metadata identity mismatch"));
+    }
+    Ok(())
+}
+
+fn native_digest_field(value: &[u8], field: &str) -> Result<[u8; 32]> {
+    value
+        .try_into()
+        .map_err(|_| Error::other(format!("native preparation returned invalid {field}")))
+}
+
+fn native_receipt_from_wire(
+    receipt: rustfs_protos::proto_gen::node_service::NativePreparationReceipt,
+    expected_disk: Uuid,
+    expected_store: Uuid,
+    expected_preparation: Uuid,
+    expected_bucket: &str,
+    expected_key: &str,
+    expected_incarnation: Uuid,
+    expected_object: Option<(&[u8], u64)>,
+    expected_version: Option<Uuid>,
+) -> Result<crate::disk::local::native_prepare::NativePreparationReceipt> {
+    let disk = native_uuid_field(&receipt.disk_id, "disk id")?;
+    let store = native_uuid_field(&receipt.store_id, "store id")?;
+    let deployment = native_uuid_field(&receipt.deployment_id, "deployment id")?;
+    let preparation = native_uuid_field(&receipt.preparation_id, "preparation id")?;
+    let version = native_version_field(&receipt.version_id)?;
+    let incarnation = native_uuid_field(&receipt.bucket_incarnation_id, "bucket incarnation")?;
+    if disk != expected_disk
+        || store != expected_store
+        || deployment != expected_store
+        || preparation != expected_preparation
+        || incarnation != expected_incarnation
+        || receipt.bucket != expected_bucket
+        || receipt.key != expected_key
+        || expected_version.is_some_and(|expected| version != expected)
+    {
+        return Err(Error::other("native preparation receipt identity mismatch"));
+    }
+    let object_digest = native_digest_field(&receipt.object_digest, "object digest")?;
+    if expected_object.is_some_and(|(digest, length)| digest != object_digest.as_slice() || length != receipt.object_length) {
+        return Err(Error::other("native preparation receipt object mismatch"));
+    }
+    let set_digest = native_digest_field(&receipt.set_digest, "set digest")?;
+    let distribution_digest = native_digest_field(&receipt.distribution_digest, "distribution digest")?;
+    let metadata_digest = native_digest_field(&receipt.metadata_digest, "metadata digest")?;
+    let shard_digest = native_digest_field(&receipt.shard_digest, "shard digest")?;
+    let data_blocks = usize::try_from(receipt.data_blocks).map_err(Error::other)?;
+    let parity_blocks = usize::try_from(receipt.parity_blocks).map_err(Error::other)?;
+    let shard_index = usize::try_from(receipt.shard_index).map_err(Error::other)?;
+    if data_blocks == 0
+        || parity_blocks == 0
+        || data_blocks
+            .checked_add(parity_blocks)
+            .is_none_or(|total| shard_index == 0 || shard_index > total)
+    {
+        return Err(Error::other("native preparation receipt has invalid erasure geometry"));
+    }
+    Ok(crate::disk::local::native_prepare::NativePreparationReceipt {
+        preparation,
+        version,
+        disk,
+        set_digest,
+        distribution_digest,
+        metadata_digest,
+        shard_digest,
+        shard_length: receipt.shard_length,
+        bucket: receipt.bucket,
+        key: receipt.key,
+        deployment,
+        incarnation,
+        object_digest,
+        object_length: receipt.object_length,
+        data_blocks,
+        parity_blocks,
+        shard_index,
+    })
+}
+
+fn validate_native_metadata(metadata: &[u8], expected_digest: Option<&[u8; 32]>) -> Result<()> {
+    if metadata.len() > rustfs_protos::NATIVE_PREPARATION_METADATA_MAX_BYTES {
+        return Err(Error::other("native preparation metadata exceeds the protocol limit"));
+    }
+    if expected_digest.is_some_and(|expected| Sha256::digest(metadata).as_slice() != expected.as_slice()) {
+        return Err(Error::other("native preparation metadata digest mismatch"));
+    }
+    Ok(())
+}
+
 fn decode_volume_infos(volume_infos: Vec<String>) -> Result<Vec<VolumeInfo>> {
     volume_infos
         .into_iter()
@@ -896,6 +1064,339 @@ impl RemoteDisk {
         .await
     }
 
+    pub(crate) async fn native_begin_stream(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+    ) -> Result<()> {
+        let disk_id = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        if preparation.is_nil() || store_id.is_nil() || incarnation.is_nil() || bucket.is_empty() || key.is_empty() {
+            return Err(DiskError::InvalidPath);
+        }
+        self.execute_with_timeout_for_op(
+            "native_preparation_begin",
+            || async {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(NativePreparationBeginRequest {
+                    disk: self.disk_ref().await,
+                    disk_id: disk_id.as_bytes().to_vec().into(),
+                    store_id: store_id.as_bytes().to_vec().into(),
+                    preparation_id: preparation.as_bytes().to_vec().into(),
+                    bucket: bucket.to_owned(),
+                    key: key.to_owned(),
+                    bucket_incarnation_id: incarnation.as_bytes().to_vec().into(),
+                });
+                let canonical = rustfs_protos::canonical_native_preparation_begin_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical, "native_preparation_begin")?;
+                let canonical_request = request.get_ref().clone();
+                let response = client.native_preparation_begin(request).await?.into_inner();
+                native_response_proof(
+                    rustfs_protos::canonical_native_preparation_begin_response_body(&canonical_request, &response),
+                    &response.response_proof,
+                )?;
+                if !response.success {
+                    return Err(native_response_error(response.error));
+                }
+                Ok(())
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    pub(crate) async fn native_finalize(
+        &self,
+        input: crate::disk::local::native_prepare::NativePrepareInput,
+    ) -> Result<crate::disk::local::native_prepare::NativePreparationReceipt> {
+        let disk_id = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        if input.preparation.is_nil() || input.store_id.is_nil() || input.bucket_incarnation.is_nil() {
+            return Err(DiskError::InvalidPath);
+        }
+        validate_native_metadata(&input.metadata, None)?;
+        let expected_object_digest = input.object_digest;
+        let expected_object_length = native_object_length_from_metadata(&input)?;
+        let expected_version = input.version;
+        self.execute_with_timeout_for_op(
+            "native_preparation_finalize",
+            || async move {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(NativePreparationFinalizeRequest {
+                    disk: self.disk_ref().await,
+                    disk_id: disk_id.as_bytes().to_vec().into(),
+                    store_id: input.store_id.as_bytes().to_vec().into(),
+                    preparation_id: input.preparation.as_bytes().to_vec().into(),
+                    bucket: input.bucket.clone(),
+                    key: input.key.clone(),
+                    bucket_incarnation_id: input.bucket_incarnation.as_bytes().to_vec().into(),
+                    object_digest: expected_object_digest.to_vec().into(),
+                    object_length: expected_object_length,
+                    version_id: expected_version.as_bytes().to_vec().into(),
+                    metadata: input.metadata.clone(),
+                });
+                let canonical = rustfs_protos::canonical_native_preparation_finalize_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical, "native_preparation_finalize")?;
+                let canonical_request = request.get_ref().clone();
+                let response = client.native_preparation_finalize(request).await?.into_inner();
+                native_response_proof(
+                    rustfs_protos::canonical_native_preparation_finalize_response_body(&canonical_request, &response),
+                    &response.response_proof,
+                )?;
+                if !response.success {
+                    return Err(native_response_error(response.error));
+                }
+                let receipt = response
+                    .receipt
+                    .ok_or_else(|| Error::other("native finalize response omitted its receipt"))?;
+                let receipt = native_receipt_from_wire(
+                    receipt,
+                    disk_id,
+                    input.store_id,
+                    input.preparation,
+                    &input.bucket,
+                    &input.key,
+                    input.bucket_incarnation,
+                    Some((&expected_object_digest, expected_object_length)),
+                    Some(expected_version),
+                )?;
+                validate_native_metadata_for_version(
+                    &input.metadata,
+                    Some(&receipt.metadata_digest),
+                    &input.bucket,
+                    &input.key,
+                    expected_version,
+                )?;
+                Ok(receipt)
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    pub(crate) async fn native_probe(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+    ) -> Result<crate::disk::local::native_prepare::NativePreparationProbe> {
+        let disk_id = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        if preparation.is_nil() || store_id.is_nil() || incarnation.is_nil() || bucket.is_empty() || key.is_empty() {
+            return Err(DiskError::InvalidPath);
+        }
+        self.execute_with_timeout_for_op(
+            "native_preparation_probe",
+            || async {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(NativePreparationProbeRequest {
+                    disk: self.disk_ref().await,
+                    disk_id: disk_id.as_bytes().to_vec().into(),
+                    store_id: store_id.as_bytes().to_vec().into(),
+                    preparation_id: preparation.as_bytes().to_vec().into(),
+                    bucket: bucket.to_owned(),
+                    key: key.to_owned(),
+                    bucket_incarnation_id: incarnation.as_bytes().to_vec().into(),
+                });
+                let canonical = rustfs_protos::canonical_native_preparation_probe_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical, "native_preparation_probe")?;
+                let canonical_request = request.get_ref().clone();
+                let response = client.native_preparation_probe(request).await?.into_inner();
+                native_response_proof(
+                    rustfs_protos::canonical_native_preparation_probe_response_body(&canonical_request, &response),
+                    &response.response_proof,
+                )?;
+                if !response.success {
+                    return Err(native_response_error(response.error));
+                }
+                validate_native_metadata(&response.metadata, None)?;
+                if !response.exists {
+                    if response.prepared
+                        || !response.metadata.is_empty()
+                        || !response.version_id.is_empty()
+                        || !response.object_digest.is_empty()
+                        || response.object_length != 0
+                        || !response.set_digest.is_empty()
+                        || !response.distribution_digest.is_empty()
+                    {
+                        return Err(Error::other("native probe absent response carries object state"));
+                    }
+                    return Ok(crate::disk::local::native_prepare::NativePreparationProbe::Absent);
+                }
+                let version = native_version_field(&response.version_id)?;
+                let object_digest = native_digest_field(&response.object_digest, "object digest")?;
+                let set_digest = native_digest_field(&response.set_digest, "set digest")?;
+                let distribution_digest = native_digest_field(&response.distribution_digest, "distribution digest")?;
+                if !response.metadata.is_empty() {
+                    validate_native_metadata_for_version(&response.metadata, None, bucket, key, version)?;
+                }
+                Ok(crate::disk::local::native_prepare::NativePreparationProbe::Existing {
+                    version,
+                    prepared: response.prepared,
+                    object_digest,
+                    object_length: response.object_length,
+                    set_digest,
+                    distribution_digest,
+                    metadata: (!response.metadata.is_empty()).then_some(response.metadata),
+                })
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    pub(crate) async fn native_recover(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+    ) -> Result<(crate::disk::local::native_prepare::NativePreparationReceipt, Bytes)> {
+        let disk_id = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        if preparation.is_nil() || store_id.is_nil() || incarnation.is_nil() || bucket.is_empty() || key.is_empty() {
+            return Err(DiskError::InvalidPath);
+        }
+        self.execute_with_timeout_for_op(
+            "native_preparation_recover",
+            || async {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(NativePreparationRecoverRequest {
+                    disk: self.disk_ref().await,
+                    disk_id: disk_id.as_bytes().to_vec().into(),
+                    store_id: store_id.as_bytes().to_vec().into(),
+                    preparation_id: preparation.as_bytes().to_vec().into(),
+                    bucket: bucket.to_owned(),
+                    key: key.to_owned(),
+                    bucket_incarnation_id: incarnation.as_bytes().to_vec().into(),
+                });
+                let canonical = rustfs_protos::canonical_native_preparation_recover_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical, "native_preparation_recover")?;
+                let canonical_request = request.get_ref().clone();
+                let response = client.native_preparation_recover(request).await?.into_inner();
+                native_response_proof(
+                    rustfs_protos::canonical_native_preparation_recover_response_body(&canonical_request, &response),
+                    &response.response_proof,
+                )?;
+                if !response.success {
+                    return Err(native_response_error(response.error));
+                }
+                let receipt = response
+                    .receipt
+                    .ok_or_else(|| Error::other("native recover response omitted its receipt"))?;
+                let receipt =
+                    native_receipt_from_wire(receipt, disk_id, store_id, preparation, bucket, key, incarnation, None, None)?;
+                validate_native_metadata_for_version(
+                    &response.metadata,
+                    Some(&receipt.metadata_digest),
+                    bucket,
+                    key,
+                    receipt.version,
+                )?;
+                Ok((receipt, response.metadata))
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    pub(crate) async fn native_locate(
+        &self,
+        store_id: Uuid,
+        preparation: Uuid,
+    ) -> Result<Option<crate::disk::local::native_prepare::NativePreparationDescription>> {
+        let disk_id = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        if preparation.is_nil() || store_id.is_nil() {
+            return Err(DiskError::InvalidPath);
+        }
+        self.execute_with_timeout_for_op(
+            "native_preparation_locate",
+            || async {
+                let mut client = self.get_client().await?;
+                let mut request = Request::new(NativePreparationLocateRequest {
+                    disk: self.disk_ref().await,
+                    disk_id: disk_id.as_bytes().to_vec().into(),
+                    store_id: store_id.as_bytes().to_vec().into(),
+                    preparation_id: preparation.as_bytes().to_vec().into(),
+                });
+                let canonical = rustfs_protos::canonical_native_preparation_locate_request_body(request.get_ref());
+                attach_mutation_body_digest(&mut request, canonical, "native_preparation_locate")?;
+                let canonical_request = request.get_ref().clone();
+                let response = client.native_preparation_locate(request).await?.into_inner();
+                native_response_proof(
+                    rustfs_protos::canonical_native_preparation_locate_response_body(&canonical_request, &response),
+                    &response.response_proof,
+                )?;
+                if !response.success {
+                    return Err(native_response_error(response.error));
+                }
+                if !response.found {
+                    if !response.bucket.is_empty()
+                        || !response.key.is_empty()
+                        || !response.bucket_incarnation_id.is_empty()
+                        || !response.metadata.is_empty()
+                        || response.prepared
+                    {
+                        return Err(Error::other("native locate absent response carries preparation state"));
+                    }
+                    return Ok(None);
+                }
+                if response.bucket.is_empty()
+                    || response.bucket.len() > 255
+                    || response.key.is_empty()
+                    || response.key.len() > 4096
+                    || preparation.is_nil()
+                {
+                    return Err(Error::other("native locate response has invalid description"));
+                }
+                let incarnation = native_uuid_field(&response.bucket_incarnation_id, "bucket incarnation")?;
+                validate_native_metadata(&response.metadata, None)?;
+                if response.metadata.is_empty() {
+                    return Err(Error::other("native locate response omitted metadata"));
+                }
+                let raw_metadata = rustfs_filemeta::FileMeta::load(&response.metadata)
+                    .map_err(|_| Error::other("native locate response metadata is invalid"))?;
+                if raw_metadata.versions.len() != 1 {
+                    return Err(Error::other("native locate metadata must have one version"));
+                }
+                let version = raw_metadata.versions[0].header.version_id.unwrap_or_else(Uuid::nil);
+                Ok(Some(crate::disk::local::native_prepare::NativePreparationDescription {
+                    preparation,
+                    bucket: response.bucket,
+                    key: response.key,
+                    incarnation,
+                    version,
+                    metadata: response.metadata,
+                    prepared: response.prepared,
+                }))
+            },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
     pub(crate) async fn ns_scanner_server_epoch(&self) -> Result<Option<Uuid>> {
         if self.health.is_faulty() {
             return Err(DiskError::FaultyDisk);

crates/ecstore/src/disk/local.rs

@@ -6867,6 +6867,28 @@ impl LocalDisk {
         opt: &DeleteOptions,
         namespace_owner: Option<Arc<dyn Send + Sync>>,
     ) -> Result<()> {
+        #[cfg(unix)]
+        if let Some(native_path) = crate::disk::native_prepare_rpc::parse_native_stream_path(volume, path)? {
+            return match native_path.file {
+                crate::disk::native_prepare_rpc::NativeStreamFile::Staging(staging_id) => {
+                    let root = self.io_root().to_path_buf();
+                    tokio::task::spawn_blocking(move || {
+                        native_prepare::remove_stream_staging_file(
+                            &root,
+                            native_path.preparation,
+                            native_path.data_dir,
+                            staging_id,
+                        )
+                    })
+                    .await
+                    .map_err(DiskError::from)?
+                    .map_err(to_file_error)?;
+                    Ok(())
+                }
+                crate::disk::native_prepare_rpc::NativeStreamFile::Part(_) => Err(DiskError::FileAccessDenied.into()),
+            };
+        }
+
         let volume_dir = self.io_get_bucket_path(volume)?;
         if !skip_access_checks(volume)
             && let Err(e) = cached_access(&volume_dir).await
@@ -10457,6 +10479,33 @@ impl DiskAPI for LocalDisk {
     #[tracing::instrument(level = "trace", skip_all)]
     async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()> {
         crate::hp_guard!("LocalDisk::rename_file");
+        #[cfg(unix)]
+        {
+            use crate::disk::native_prepare_rpc::{NativeStreamFile, parse_native_stream_path};
+
+            let src_native = parse_native_stream_path(src_volume, src_path)?;
+            let dst_native = parse_native_stream_path(dst_volume, dst_path)?;
+            if src_native.is_some() || dst_native.is_some() {
+                let (Some(src), Some(dst)) = (src_native, dst_native) else {
+                    return Err(DiskError::InvalidPath.into());
+                };
+                let (NativeStreamFile::Staging(staging_id), NativeStreamFile::Part(part)) = (src.file, dst.file) else {
+                    return Err(DiskError::InvalidPath.into());
+                };
+                if src_volume != dst_volume || src.preparation != dst.preparation || src.data_dir != dst.data_dir {
+                    return Err(DiskError::InvalidPath.into());
+                }
+                let root = self.io_root().to_path_buf();
+                tokio::task::spawn_blocking(move || {
+                    native_prepare::publish_stream_part(&root, src.preparation, src.data_dir, staging_id, part)
+                })
+                .await
+                .map_err(DiskError::from)?
+                .map_err(to_file_error)?;
+                return Ok(());
+            }
+        }
+
         let src_volume_dir = self.io_get_bucket_path(src_volume)?;
         let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
         if !skip_access_checks(src_volume) {
@@ -10538,6 +10587,28 @@ impl DiskAPI for LocalDisk {
             }
         }
 
+        if let Some(native_path) = crate::disk::native_prepare_rpc::parse_native_stream_path(volume, path)? {
+            #[cfg(unix)]
+            {
+                use crate::disk::native_prepare_rpc::NativeStreamFile;
+
+                let root = self.io_root().to_path_buf();
+                let file = tokio::task::spawn_blocking(move || match native_path.file {
+                    NativeStreamFile::Part(part) => {
+                        native_prepare::open_stream_part(&root, native_path.preparation, native_path.data_dir, part)
+                    }
+                    NativeStreamFile::Staging(id) => {
+                        native_prepare::open_stream_staging_file(&root, native_path.preparation, native_path.data_dir, id)
+                    }
+                })
+                .await
+                .map_err(DiskError::from)?
+                .map_err(to_file_error)?;
+                return Ok(Box::new(tokio::fs::File::from_std(file)));
+            }
+            #[cfg(not(unix))]
+            return Err(DiskError::UnsupportedDisk);
+        }
         self.io_backend
             .open_write(volume, path, WriteMode::Truncate { size_hint: _file_size })
             .await
@@ -11553,6 +11624,124 @@ mod test {
     use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf};
     use tracing_subscriber::fmt::MakeWriter;
 
+    #[cfg(unix)]
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn native_stream_reserved_split_never_truncates_attested_shard() {
+        use crate::bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks};
+        use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = "native-split-regression";
+        let key = "object";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("real bucket");
+        let preparation = Uuid::new_v4();
+        drop(
+            store
+                .prepare_native_object(
+                    bucket,
+                    key,
+                    preparation,
+                    &mut crate::object_api::PutObjReader::from_vec(vec![51; 2 * 1024 * 1024 + 7]),
+                )
+                .await
+                .expect("actual erasure/bitrot preparation"),
+        );
+        let mut endpoint = Endpoint::try_from(dirs[0].path().to_str().expect("root")).expect("endpoint");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let disk = LocalDisk::new(&endpoint, false).await.expect("actual LocalDisk");
+        let path = format!("native-preparations/{preparation}/{preparation}/part.1");
+        let full = dirs[0].path().join(RUSTFS_META_BUCKET).join(&path);
+        let before = std::fs::read(&full).expect("actual durable shard");
+        for (volume, candidate) in [
+            (
+                format!("{RUSTFS_META_BUCKET}/native-preparations/{preparation}"),
+                format!("{preparation}/part.1"),
+            ),
+            (
+                format!("{RUSTFS_META_BUCKET}/native-preparations"),
+                format!("{preparation}/{preparation}/part.1"),
+            ),
+            (RUSTFS_META_BUCKET.to_owned(), format!("./{path}")),
+            (RUSTFS_META_BUCKET.to_owned(), format!("junk/../{path}")),
+            (
+                format!("{RUSTFS_META_BUCKET}/junk/../native-preparations/{preparation}"),
+                format!("{preparation}/part.1"),
+            ),
+            (
+                format!("junk/../{RUSTFS_META_BUCKET}"),
+                format!("native-preparations/{preparation}/{preparation}/part.1"),
+            ),
+            (
+                RUSTFS_META_BUCKET.to_owned(),
+                format!("native-preparations//{preparation}/{preparation}/part.1"),
+            ),
+        ] {
+            assert!(
+                matches!(disk.create_file("", &volume, &candidate, 1).await, Err(DiskError::InvalidPath)),
+                "NATIVE_RESERVED_SPLIT_MUST_REJECT_BEFORE_TRUNCATE: {volume}/{candidate}"
+            );
+            assert_eq!(
+                std::fs::read(&full).expect("retained actual shard"),
+                before,
+                "NATIVE_RESERVED_SPLIT_MUST_PRESERVE_ATTESTED_SHARD"
+            );
+        }
+
+        let staged_id = Uuid::new_v4();
+        let staged_path = format!("native-preparations/{preparation}/{preparation}/.rustfs-put-{staged_id}");
+        let mut staged = disk
+            .create_file("", RUSTFS_META_BUCKET, &staged_path, 7)
+            .await
+            .expect("create private native staging file");
+        staged.write_all(b"staged!").await.expect("write staged payload");
+        staged.shutdown().await.expect("close staged payload");
+        disk.rename_file(
+            RUSTFS_META_BUCKET,
+            &staged_path,
+            RUSTFS_META_BUCKET,
+            &format!("native-preparations/{preparation}/{preparation}/part.2"),
+        )
+        .await
+        .expect("publish private staging file without ordinary volume rename");
+        assert_eq!(
+            std::fs::read(
+                dirs[0]
+                    .path()
+                    .join(RUSTFS_META_BUCKET)
+                    .join(format!("native-preparations/{preparation}/{preparation}/part.2"))
+            )
+            .expect("read published private part"),
+            b"staged!"
+        );
+
+        let abandoned_id = Uuid::new_v4();
+        let abandoned_path = format!("native-preparations/{preparation}/{preparation}/.rustfs-put-{abandoned_id}");
+        let mut abandoned = disk
+            .create_file("", RUSTFS_META_BUCKET, &abandoned_path, 3)
+            .await
+            .expect("create private cleanup staging file");
+        abandoned.write_all(b"tmp").await.expect("write cleanup staging file");
+        abandoned.shutdown().await.expect("close cleanup staging file");
+        disk.delete(
+            RUSTFS_META_BUCKET,
+            &abandoned_path,
+            DeleteOptions {
+                immediate: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("delete private staging file after failed publication");
+        assert!(!dirs[0].path().join(RUSTFS_META_BUCKET).join(&abandoned_path).exists());
+        assert!(store.recover_native_object(bucket, key, preparation).await.is_ok());
+    }
+
     #[cfg(unix)]
     #[tokio::test]
     async fn read_dir_entries_skip_symlinks_without_truncating_listing() {

crates/ecstore/src/disk/local/native_prepare.rs

@@ -25,24 +25,25 @@ pub struct NativePrepareInput {
     pub metadata: bytes::Bytes,
     pub bucket_fence: Arc<ObjectLockConfigSnapshot>,
 }
-/// Only a successful local durability boundary can construct this value.
+/// Only a successful disk owner durability boundary can construct this value.
 pub struct NativePreparationReceipt {
-    preparation: Uuid,
-    version: Uuid,
-    disk: Uuid,
-    set_digest: [u8; 32],
-    distribution_digest: [u8; 32],
-    metadata_digest: [u8; 32],
-    shard_digest: [u8; 32],
-    bucket: String,
-    key: String,
-    deployment: Uuid,
-    incarnation: Uuid,
-    object_digest: [u8; 32],
-    object_length: u64,
-    data_blocks: usize,
-    parity_blocks: usize,
-    shard_index: usize,
+    pub(crate) preparation: Uuid,
+    pub(crate) version: Uuid,
+    pub(crate) disk: Uuid,
+    pub(crate) set_digest: [u8; 32],
+    pub(crate) distribution_digest: [u8; 32],
+    pub(crate) metadata_digest: [u8; 32],
+    pub(crate) shard_digest: [u8; 32],
+    pub(crate) shard_length: u64,
+    pub(crate) bucket: String,
+    pub(crate) key: String,
+    pub(crate) deployment: Uuid,
+    pub(crate) incarnation: Uuid,
+    pub(crate) object_digest: [u8; 32],
+    pub(crate) object_length: u64,
+    pub(crate) data_blocks: usize,
+    pub(crate) parity_blocks: usize,
+    pub(crate) shard_index: usize,
 }
 impl NativePreparationReceipt {
     pub fn bucket(&self) -> &str {
@@ -81,6 +82,9 @@ impl NativePreparationReceipt {
     pub fn distribution_digest(&self) -> &[u8; 32] {
         &self.distribution_digest
     }
+    pub fn shard_length(&self) -> u64 {
+        self.shard_length
+    }
     pub fn metadata_digest(&self) -> &[u8; 32] {
         &self.metadata_digest
     }
@@ -89,7 +93,7 @@ impl NativePreparationReceipt {
     }
 }
 /// A routing observation, never a durability or publication receipt.
-pub(crate) enum NativePreparationProbe {
+pub enum NativePreparationProbe {
     Absent,
     Existing {
         version: Uuid,
@@ -102,7 +106,69 @@ pub(crate) enum NativePreparationProbe {
     },
 }
 #[cfg(all(test, unix))]
-pub(crate) use tests::{clear_native_preparation_probe, pause_native_preparation};
+pub(crate) use tests::{clear_native_preparation_probe, fail_native_preparation, pause_native_preparation};
+#[cfg(all(test, unix))]
+pub(crate) fn canonical_test_record(value: &serde_json::Value) -> Vec<u8> {
+    serde_json::to_vec(&serde_json::from_value::<durable::Record>(value.clone()).expect("typed durable record"))
+        .expect("canonical durable record")
+}
+
+#[cfg(unix)]
+pub(super) fn open_stream_part(
+    root: &std::path::Path,
+    preparation: Uuid,
+    data_dir: Uuid,
+    part: usize,
+) -> std::io::Result<std::fs::File> {
+    durable::open_stream_file(root, preparation, data_dir, format!("part.{part}"))
+}
+
+#[cfg(unix)]
+pub(super) fn open_stream_staging_file(
+    root: &std::path::Path,
+    preparation: Uuid,
+    data_dir: Uuid,
+    staging_id: Uuid,
+) -> std::io::Result<std::fs::File> {
+    durable::open_stream_file(root, preparation, data_dir, format!(".rustfs-put-{staging_id}"))
+}
+
+#[cfg(unix)]
+pub(super) fn publish_stream_part(
+    root: &std::path::Path,
+    preparation: Uuid,
+    data_dir: Uuid,
+    staging_id: Uuid,
+    part: usize,
+) -> std::io::Result<()> {
+    durable::publish_stream_part(root, preparation, data_dir, staging_id, part)
+}
+
+#[cfg(unix)]
+pub(super) fn remove_stream_staging_file(
+    root: &std::path::Path,
+    preparation: Uuid,
+    data_dir: Uuid,
+    staging_id: Uuid,
+) -> std::io::Result<()> {
+    durable::remove_stream_staging_file(root, preparation, data_dir, staging_id)
+}
+
+pub struct NativePreparationDescription {
+    pub preparation: Uuid,
+    pub bucket: String,
+    pub key: String,
+    pub incarnation: Uuid,
+    pub version: Uuid,
+    pub metadata: bytes::Bytes,
+    pub prepared: bool,
+}
+pub(crate) struct NativePreparationListing {
+    pub records: Vec<NativePreparationDescription>,
+    pub unfinished: Vec<Uuid>,
+    pub next: Option<Uuid>,
+}
+
 impl LocalDisk {
     #[cfg(unix)]
     pub(crate) async fn probe_native(
@@ -242,6 +308,90 @@ impl LocalDisk {
         .map_err(Into::into)
     }
     /// Retain the real bucket and object owners through every blocking write.
+    #[cfg(unix)]
+    pub(crate) async fn list_native_preparations(
+        &self,
+        store_id: Uuid,
+        cursor: Option<Uuid>,
+        limit: usize,
+    ) -> Result<NativePreparationListing> {
+        if !(1..=64).contains(&limit) || cursor.is_some_and(|id| id.is_nil()) {
+            return Err(DiskError::InvalidPath);
+        }
+        let disk = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        let root = self.io_root().to_path_buf();
+        tokio::task::spawn_blocking(move || durable::list(&root, disk, store_id, cursor, limit))
+            .await
+            .map_err(DiskError::from)?
+            .map_err(Into::into)
+    }
+    #[cfg(unix)]
+    pub(crate) async fn locate_native_preparation(
+        &self,
+        store_id: Uuid,
+        preparation: Uuid,
+    ) -> Result<Option<NativePreparationDescription>> {
+        if preparation.is_nil() {
+            return Err(DiskError::InvalidPath);
+        }
+        let disk = self
+            .get_disk_id()
+            .await?
+            .filter(|id| !id.is_nil())
+            .ok_or(DiskError::InvalidPath)?;
+        let root = self.io_root().to_path_buf();
+        tokio::task::spawn_blocking(move || durable::locate(&root, disk, store_id, preparation))
+            .await
+            .map_err(DiskError::from)?
+            .map_err(Into::into)
+    }
+
+    /// Create the private staging owner before the ordinary streaming disk writer opens parts.
+    #[cfg(unix)]
+    pub(crate) async fn begin_native_stream(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        incarnation: Uuid,
+        fence: Arc<ObjectLockConfigSnapshot>,
+    ) -> Result<()> {
+        if preparation.is_nil() || !fence.is_valid_for_destructive_put(store_id, bucket, incarnation) {
+            return Err(DiskError::InvalidPath);
+        }
+        let root = self.io_root().to_path_buf();
+        let bucket = bucket.to_owned();
+        let path = self
+            .root
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string());
+        let owner = crate::disk::os::acquire_namespace_mutation_lease_with_owner(&path, None).await;
+        crate::disk::os::run_blocking_namespace_operation(owner, move || {
+            if !fence.is_valid_for_destructive_put(store_id, &bucket, incarnation) {
+                return Err(std::io::Error::other("native stream fence changed"));
+            }
+            durable::begin_stream(&root, preparation)
+        })
+        .await
+        .map_err(Into::into)
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn begin_native_stream(
+        &self,
+        _preparation: Uuid,
+        _store_id: Uuid,
+        _bucket: &str,
+        _incarnation: Uuid,
+        _fence: Arc<ObjectLockConfigSnapshot>,
+    ) -> Result<()> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
     pub async fn prepare_native(&self, input: NativePrepareInput) -> Result<NativePreparationReceipt> {
         self.prepare_native_inner(input, None).await
     }
@@ -291,40 +441,44 @@ impl LocalDisk {
             .validate(rustfs_filemeta::ValidationMode::RequireErasure)
             .map_err(|_| DiskError::InvalidPath)?
             .ok_or(DiskError::InvalidPath)?;
-        if !metadata.inline_data()
-            || metadata.parts.len() != 1
-            || metadata.size < 0
-            || metadata.erasure.get_checksum_info(1).algorithm.size() == 0
-        {
+        if metadata.parts.is_empty() || metadata.size < 0 || metadata.erasure.get_checksum_info(1).algorithm.size() == 0 {
             return Err(DiskError::UnsupportedDisk);
         }
-        let shard = metadata.data.as_ref().ok_or(DiskError::InvalidPath)?;
-        let part = &metadata.parts[0];
-        if part.number != 1 || part.size != usize::try_from(metadata.size).map_err(|_| DiskError::InvalidPath)? {
+        if metadata.inline_data() {
+            let shard = metadata.data.as_ref().ok_or(DiskError::InvalidPath)?;
+            if metadata.parts.len() != 1
+                || metadata.parts[0].number != 1
+                || metadata.parts[0].size != usize::try_from(metadata.size).map_err(|_| DiskError::InvalidPath)?
+            {
+                return Err(DiskError::InvalidPath);
+            }
+            let codec = crate::erasure::coding::Erasure::try_new_with_options(
+                metadata.erasure.data_blocks,
+                metadata.erasure.parity_blocks,
+                metadata.erasure.block_size,
+                metadata.uses_legacy_checksum,
+            )?;
+            let checksum = metadata.erasure.get_checksum_info(1).algorithm;
+            let checksum = if metadata.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
+                rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+            } else {
+                checksum
+            };
+            let shard_length = usize::try_from(
+                codec.shard_file_size(i64::try_from(metadata.parts[0].size).map_err(|_| DiskError::InvalidPath)?),
+            )
+            .map_err(|_| DiskError::InvalidPath)?;
+            crate::erasure::coding::bitrot_verify(
+                std::io::Cursor::new(shard),
+                shard.len(),
+                shard_length,
+                checksum,
+                codec.shard_size(),
+            )
+            .await?;
+        } else if metadata.data_dir.is_none_or(|id| id.is_nil()) {
             return Err(DiskError::InvalidPath);
         }
-        let codec = crate::erasure::coding::Erasure::try_new_with_options(
-            metadata.erasure.data_blocks,
-            metadata.erasure.parity_blocks,
-            metadata.erasure.block_size,
-            metadata.uses_legacy_checksum,
-        )?;
-        let checksum = metadata.erasure.get_checksum_info(1).algorithm;
-        let checksum = if metadata.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
-            rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
-        } else {
-            checksum
-        };
-        let shard_length = usize::try_from(codec.shard_file_size(i64::try_from(part.size).map_err(|_| DiskError::InvalidPath)?))
-            .map_err(|_| DiskError::InvalidPath)?;
-        crate::erasure::coding::bitrot_verify(
-            std::io::Cursor::new(shard),
-            shard.len(),
-            shard_length,
-            checksum,
-            codec.shard_size(),
-        )
-        .await?;
         let disk = self
             .get_disk_id()
             .await?
@@ -348,23 +502,24 @@ impl LocalDisk {
         let owner = crate::disk::os::acquire_namespace_mutation_lease_with_owner(&prep_path, Some(object_owner)).await;
         let admission = crate::disk::os::acquire_file_sync_admission(self.file_sync_permits.clone()).await?;
         let root = self.io_root().to_path_buf();
+        let runtime = tokio::runtime::Handle::current();
         #[cfg(unix)]
         return crate::disk::os::run_blocking_namespace_operation(owner, move || {
             let _admission = admission;
             let _operation_owner = operation_owner;
-            durable::prepare(&root, disk, version, input, metadata)
+            durable::prepare(&root, disk, version, input, metadata, runtime)
         })
         .await
         .map_err(Into::into);
         #[cfg(not(unix))]
         {
-            let _ = (owner, admission, root, disk, version, input, metadata, operation_owner);
+            let _ = (owner, admission, root, disk, version, input, metadata, operation_owner, runtime);
             Err(DiskError::UnsupportedDisk)
         }
     }
 }
 #[cfg(unix)]
-mod durable {
+pub(super) mod durable {
     use super::*;
     use rustix::fs::{self, AtFlags, Mode, OFlags};
     use serde::{Deserialize, Serialize};
@@ -401,6 +556,10 @@ mod durable {
         shard_length: u64,
         #[serde(default, skip_serializing_if = "Option::is_none")]
         recovery_metadata: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        set_disks: Option<Vec<Uuid>>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        data_dir: Option<Uuid>,
     }
     impl Record {
         pub(super) fn new(
@@ -436,6 +595,8 @@ mod durable {
                 shard_digest: Sha256::digest(shard).into(),
                 shard_length: u64::try_from(shard.len()).map_err(invalid)?,
                 recovery_metadata: Some(base64_simd::STANDARD.encode_to_string(metadata)),
+                set_disks: None,
+                data_dir: None,
             })
         }
         fn receipt(&self) -> NativePreparationReceipt {
@@ -447,6 +608,7 @@ mod durable {
                 distribution_digest: self.distribution_digest,
                 metadata_digest: self.metadata_digest,
                 shard_digest: self.shard_digest,
+                shard_length: self.shard_length,
                 bucket: self.bucket.clone(),
                 key: self.key.clone(),
                 deployment: self.deployment,
@@ -468,7 +630,7 @@ mod durable {
         device: u64,
         inode: u64,
     }
-    struct Directory {
+    pub(in crate::disk::local) struct Directory {
         file: File,
         links: Vec<Link>,
         preparation: Uuid,
@@ -477,7 +639,7 @@ mod durable {
         root_inode: u64,
     }
     impl Directory {
-        fn open(root: &Path, preparation: Uuid, create: bool) -> io::Result<Self> {
+        pub(in crate::disk::local) fn open(root: &Path, preparation: Uuid, create: bool) -> io::Result<Self> {
             let mut file = File::from(fs::open(
                 root,
                 OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
@@ -530,7 +692,7 @@ mod durable {
             let sys = self.links.get(1).ok_or_else(|| invalid("missing system directory"))?;
             read_file(&sys.parent, "format.json", METADATA_LIMIT)?.ok_or_else(|| invalid("missing disk format"))
         }
-        fn check(&self) -> io::Result<()> {
+        pub(in crate::disk::local) fn check(&self) -> io::Result<()> {
             let root = fs::stat(&self.root)?;
             if u64::try_from(root.st_dev).map_err(invalid)? != self.root_device
                 || u64::try_from(root.st_ino).map_err(invalid)? != self.root_inode
@@ -547,10 +709,10 @@ mod durable {
             }
             Ok(())
         }
-        fn read(&self, name: &str, limit: usize) -> io::Result<Option<Vec<u8>>> {
+        pub(in crate::disk::local) fn read(&self, name: &str, limit: usize) -> io::Result<Option<Vec<u8>>> {
             read_file(&self.file, name, limit)
         }
-        fn write(&self, name: &str, bytes: &[u8]) -> io::Result<()> {
+        pub(in crate::disk::local) fn write(&self, name: &str, bytes: &[u8]) -> io::Result<()> {
             self.check()?;
             let tmp = format!(".stage-{}", Uuid::new_v4());
             let fd = fs::openat(
@@ -785,6 +947,385 @@ mod durable {
         }
         Ok(result)
     }
+    fn description(root: &Path, disk: Uuid, store_id: Uuid, preparation: Uuid) -> io::Result<NativePreparationDescription> {
+        let record = inspect(root, preparation)?;
+        if record.disk != disk || record.deployment != store_id || record.bucket.len() > 255 || record.key.len() > 4096 {
+            return Err(invalid("native locator identity mismatch"));
+        }
+        let dir = Directory::open(root, preparation, false)?;
+        let metadata = match dir.read("xl.meta", METADATA_LIMIT)? {
+            Some(bytes) => bytes,
+            None => record
+                .recovery_metadata
+                .as_ref()
+                .map(|bytes| base64_simd::STANDARD.decode_to_vec(bytes).map_err(invalid))
+                .transpose()?
+                .ok_or_else(|| invalid("native locator metadata missing"))?,
+        };
+        if metadata.len() > METADATA_LIMIT
+            || u64::try_from(metadata.len()).map_err(invalid)? != record.metadata_length
+            || Sha256::digest(&metadata).as_slice() != record.metadata_digest
+            || Sha256::digest(dir.format_bytes()?).as_slice() != record.disk_format_digest
+        {
+            return Err(invalid("native locator metadata/format mismatch"));
+        }
+        let encoded = serde_json::to_vec(&record).map_err(invalid)?;
+        let prepared = match dir.read("PREPARED", 32)? {
+            None => false,
+            Some(bytes) if bytes.as_slice() == Sha256::digest(&encoded).as_slice() => true,
+            Some(_) => return Err(invalid("native locator marker mismatch")),
+        };
+        dir.check()?;
+        if dir.read("intent.json", RECORD_LIMIT)?.as_deref() != Some(encoded.as_slice()) {
+            return Err(invalid("native locator record changed"));
+        }
+        Ok(NativePreparationDescription {
+            preparation,
+            bucket: record.bucket,
+            key: record.key,
+            incarnation: record.incarnation,
+            version: record.version,
+            metadata: metadata.into(),
+            prepared,
+        })
+    }
+
+    fn listing_root(root: &Path, disk: Uuid, store_id: Uuid) -> io::Result<(File, File)> {
+        let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
+        let root = File::from(fs::open(root, flags, Mode::empty())?);
+        let sys = File::from(fs::openat(&root, ".rustfs.sys", flags, Mode::empty())?);
+        let format: crate::layout::format::FormatV3 = serde_json::from_slice(
+            &read_file(&sys, "format.json", METADATA_LIMIT)?.ok_or_else(|| invalid("native listing disk format missing"))?,
+        )
+        .map_err(invalid)?;
+        if store_id.is_nil() || disk.is_nil() || format.id != store_id || format.erasure.this != disk {
+            return Err(invalid("native listing physical disk identity mismatch"));
+        }
+        Ok((root, sys))
+    }
+    fn check_listing_root(root_path: &Path, root: &File, sys: &File) -> io::Result<()> {
+        let expected = fs::fstat(root)?;
+        let actual = fs::stat(root_path)?;
+        let expected_sys = fs::fstat(sys)?;
+        let actual_sys = fs::statat(root, ".rustfs.sys", AtFlags::SYMLINK_NOFOLLOW)?;
+        if expected.st_dev != actual.st_dev
+            || expected.st_ino != actual.st_ino
+            || expected_sys.st_dev != actual_sys.st_dev
+            || expected_sys.st_ino != actual_sys.st_ino
+        {
+            return Err(invalid("native listing root changed"));
+        }
+        Ok(())
+    }
+    pub(super) fn locate(
+        root: &Path,
+        disk: Uuid,
+        store_id: Uuid,
+        preparation: Uuid,
+    ) -> io::Result<Option<NativePreparationDescription>> {
+        let (root_fd, sys) = listing_root(root, disk, store_id)?;
+        let value = match Directory::open(root, preparation, false) {
+            Ok(_) => Some(description(root, disk, store_id, preparation)?),
+            Err(error) if error.kind() == io::ErrorKind::NotFound => None,
+            Err(error) => return Err(error),
+        };
+        check_listing_root(root, &root_fd, &sys)?;
+        Ok(value)
+    }
+    pub(super) fn list(
+        root: &Path,
+        disk: Uuid,
+        store_id: Uuid,
+        cursor: Option<Uuid>,
+        limit: usize,
+    ) -> io::Result<NativePreparationListing> {
+        let (root_fd, sys) = listing_root(root, disk, store_id)?;
+        let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
+        let container = match fs::openat(&sys, "native-preparations", flags, Mode::empty()) {
+            Ok(fd) => File::from(fd),
+            Err(rustix::io::Errno::NOENT) => {
+                check_listing_root(root, &root_fd, &sys)?;
+                return Ok(NativePreparationListing {
+                    records: Vec::new(),
+                    unfinished: Vec::new(),
+                    next: None,
+                });
+            }
+            Err(error) => return Err(error.into()),
+        };
+        let identity = fs::fstat(&container)?;
+        if identity.st_mode & 0o077 != 0 {
+            return Err(invalid("native listing directory is not private"));
+        }
+        let mut names = std::collections::BTreeSet::new();
+        for entry in fs::Dir::read_from(&container)? {
+            let entry = entry?;
+            let name = entry.file_name().to_str().map_err(invalid)?;
+            if name == "." || name == ".." {
+                continue;
+            }
+            let id = Uuid::parse_str(name).map_err(invalid)?;
+            if id.is_nil() || name != id.to_string() {
+                return Err(invalid("invalid native preparation directory name"));
+            }
+            if cursor.is_some_and(|after| id <= after) {
+                continue;
+            }
+            names.insert(id);
+            if names.len() > limit + 1 {
+                names.pop_last();
+            }
+        }
+        let mut more = names.len() > limit;
+        let mut result = NativePreparationListing {
+            records: Vec::new(),
+            unfinished: Vec::new(),
+            next: None,
+        };
+        let mut bytes = 0_usize;
+        for id in names.into_iter().take(limit) {
+            let dir = Directory::open(root, id, false)?;
+            if dir.read("intent.json", RECORD_LIMIT)?.is_none() {
+                result.unfinished.push(id);
+            } else {
+                let record = description(root, disk, store_id, id)?;
+                let length = record
+                    .metadata
+                    .len()
+                    .checked_add(record.bucket.len())
+                    .and_then(|size| size.checked_add(record.key.len()))
+                    .and_then(|size| size.checked_add(256))
+                    .ok_or_else(|| invalid("native listing size overflow"))?;
+                if bytes.checked_add(length).is_none_or(|size| size > 2 * METADATA_LIMIT) {
+                    if result.records.is_empty() && result.unfinished.is_empty() {
+                        return Err(invalid("native listing record exceeds bound"));
+                    }
+                    more = true;
+                    break;
+                }
+                bytes += length;
+                result.records.push(record);
+            }
+            result.next = Some(id);
+        }
+        let current = fs::statat(&sys, "native-preparations", AtFlags::SYMLINK_NOFOLLOW)?;
+        if identity.st_dev != current.st_dev || identity.st_ino != current.st_ino {
+            return Err(invalid("native listing container changed"));
+        }
+        check_listing_root(root, &root_fd, &sys)?;
+        if !more {
+            result.next = None;
+        }
+        Ok(result)
+    }
+
+    pub(super) fn begin_stream(root: &Path, preparation: Uuid) -> io::Result<()> {
+        Directory::open(root, preparation, true)?.check()
+    }
+
+    pub(super) fn open_stream_file(root: &Path, preparation: Uuid, data_dir: Uuid, file_name: String) -> io::Result<File> {
+        if preparation.is_nil() || data_dir.is_nil() || !valid_stream_file_name(&file_name) {
+            return Err(invalid("invalid native stream path"));
+        }
+        let dir = Directory::open(root, preparation, false)?;
+        let (data, identity) = open_stream_data_dir(&dir, data_dir, true)?;
+        // An existing candidate is immutable. Concurrent retries cannot truncate an attested shard.
+        let file = File::from(fs::openat(
+            &data,
+            file_name,
+            OFlags::WRONLY | OFlags::CREATE | OFlags::EXCL | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::RUSR | Mode::WUSR,
+        )?);
+        dir.check()?;
+        let current = fs::statat(&dir.file, data_dir.to_string(), AtFlags::SYMLINK_NOFOLLOW)?;
+        if identity.st_dev != current.st_dev || identity.st_ino != current.st_ino {
+            return Err(invalid("native stream directory changed"));
+        }
+        Ok(file)
+    }
+
+    fn valid_stream_file_name(file_name: &str) -> bool {
+        file_name
+            .strip_prefix("part.")
+            .and_then(|value| value.parse::<usize>().ok())
+            .filter(|part| *part > 0)
+            .is_some_and(|part| file_name == format!("part.{part}"))
+            || file_name
+                .strip_prefix(".rustfs-put-")
+                .and_then(|value| Uuid::parse_str(value).ok())
+                .filter(|id| !id.is_nil())
+                .is_some_and(|id| file_name == format!(".rustfs-put-{id}"))
+    }
+
+    fn open_stream_data_dir(dir: &Directory, data_dir: Uuid, create: bool) -> io::Result<(File, rustix::fs::Stat)> {
+        let name = data_dir.to_string();
+        if create {
+            match fs::mkdirat(&dir.file, &name, Mode::RUSR | Mode::WUSR | Mode::XUSR) {
+                Ok(()) => {}
+                Err(rustix::io::Errno::EXIST) => {}
+                Err(error) => return Err(error.into()),
+            }
+            // Redo this barrier when the directory survived an earlier unknown outcome.
+            sync(&dir.file, dir.preparation, "stream:data-dir-parent")?;
+        }
+        let data = File::from(fs::openat(
+            &dir.file,
+            &name,
+            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?);
+        let identity = fs::fstat(&data)?;
+        if identity.st_mode & 0o077 != 0 {
+            return Err(invalid("native streaming directory is not private"));
+        }
+        Ok((data, identity))
+    }
+
+    pub(super) fn publish_stream_part(
+        root: &Path,
+        preparation: Uuid,
+        data_dir: Uuid,
+        staging_id: Uuid,
+        part: usize,
+    ) -> io::Result<()> {
+        if preparation.is_nil() || data_dir.is_nil() || staging_id.is_nil() || part == 0 {
+            return Err(invalid("invalid native stream publication path"));
+        }
+        let dir = Directory::open(root, preparation, false)?;
+        let (data, identity) = open_stream_data_dir(&dir, data_dir, false)?;
+        let staging_name = format!(".rustfs-put-{staging_id}");
+        let staged = File::from(fs::openat(
+            &data,
+            &staging_name,
+            OFlags::RDONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?);
+        let metadata = staged.metadata()?;
+        let staged_identity = fs::fstat(&staged)?;
+        if !metadata.is_file() || staged_identity.st_nlink != 1 {
+            return Err(invalid("native streaming stage is not a private file"));
+        }
+        sync(&staged, preparation, "stream:publish-file")?;
+        fs::renameat_with(&data, &staging_name, &data, format!("part.{part}"), rustix::fs::RenameFlags::NOREPLACE)?;
+        sync(&data, preparation, "stream:publish-directory")?;
+        dir.check()?;
+        let current = fs::statat(&dir.file, data_dir.to_string(), AtFlags::SYMLINK_NOFOLLOW)?;
+        if identity.st_dev != current.st_dev || identity.st_ino != current.st_ino {
+            return Err(invalid("native streaming directory changed"));
+        }
+        Ok(())
+    }
+
+    pub(super) fn remove_stream_staging_file(root: &Path, preparation: Uuid, data_dir: Uuid, staging_id: Uuid) -> io::Result<()> {
+        if preparation.is_nil() || data_dir.is_nil() || staging_id.is_nil() {
+            return Err(invalid("invalid native stream cleanup path"));
+        }
+        let dir = Directory::open(root, preparation, false)?;
+        let (data, identity) = open_stream_data_dir(&dir, data_dir, false)?;
+        let staging_name = format!(".rustfs-put-{staging_id}");
+        fs::unlinkat(&data, &staging_name, AtFlags::empty())?;
+        sync(&data, preparation, "stream:cleanup-directory")?;
+        dir.check()?;
+        let current = fs::statat(&dir.file, data_dir.to_string(), AtFlags::SYMLINK_NOFOLLOW)?;
+        if identity.st_dev != current.st_dev || identity.st_ino != current.st_ino {
+            return Err(invalid("native streaming directory changed"));
+        }
+        Ok(())
+    }
+
+    fn stream_parts(dir: &Directory, info: &FileInfo, runtime: &tokio::runtime::Handle) -> io::Result<([u8; 32], u64)> {
+        let data_dir = info
+            .data_dir
+            .filter(|id| !id.is_nil())
+            .ok_or_else(|| invalid("missing streaming data dir"))?;
+        let name = data_dir.to_string();
+        let data = File::from(fs::openat(
+            &dir.file,
+            &name,
+            OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+            Mode::empty(),
+        )?);
+        let identity = fs::fstat(&data)?;
+        let codec = crate::erasure::coding::Erasure::try_new_with_options(
+            info.erasure.data_blocks,
+            info.erasure.parity_blocks,
+            info.erasure.block_size,
+            info.uses_legacy_checksum,
+        )
+        .map_err(|error| error.into_io_error())?;
+        let mut digest = Sha256::new();
+        let mut total = 0_u64;
+        let mut object_size = 0_usize;
+        let mut previous = 0;
+        for part in &info.parts {
+            if part.number <= previous {
+                return Err(invalid("invalid streaming part order"));
+            }
+            previous = part.number;
+            object_size = object_size
+                .checked_add(part.size)
+                .ok_or_else(|| invalid("part length overflow"))?;
+            let part_name = format!("part.{}", part.number);
+            let mut file = File::from(fs::openat(
+                &data,
+                &part_name,
+                OFlags::RDONLY | OFlags::NONBLOCK | OFlags::NOFOLLOW | OFlags::CLOEXEC,
+                Mode::empty(),
+            )?);
+            let before = fs::fstat(&file)?;
+            if before.st_nlink != 1 || !file.metadata()?.is_file() {
+                return Err(invalid("invalid streaming part file"));
+            }
+            let physical = usize::try_from(file.metadata()?.len()).map_err(invalid)?;
+            let algorithm = info.erasure.get_checksum_info(part.number).algorithm;
+            if algorithm.size() == 0 {
+                return Err(invalid("missing streaming checksum"));
+            }
+            let algorithm = if info.uses_legacy_checksum && algorithm == rustfs_utils::HashAlgorithm::HighwayHash256S {
+                rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+            } else {
+                algorithm
+            };
+            let expected = usize::try_from(codec.shard_file_size(i64::try_from(part.size).map_err(invalid)?)).map_err(invalid)?;
+            runtime.block_on(crate::erasure::coding::bitrot_verify(
+                tokio::fs::File::from_std(file.try_clone()?),
+                physical,
+                expected,
+                algorithm,
+                codec.shard_size(),
+            ))?;
+            // try_clone shares the file offset; rewind before the aggregate digest.
+            use std::io::{Seek, SeekFrom};
+            file.seek(SeekFrom::Start(0))?;
+            let mut buffer = [0_u8; 64 * 1024];
+            loop {
+                let size = file.read(&mut buffer)?;
+                if size == 0 {
+                    break;
+                }
+                digest.update(&buffer[..size]);
+                total = total
+                    .checked_add(u64::try_from(size).map_err(invalid)?)
+                    .ok_or_else(|| invalid("shard length overflow"))?;
+            }
+            sync(&file, dir.preparation, &format!("{part_name}:file"))?;
+            let after = fs::statat(&data, &part_name, AtFlags::SYMLINK_NOFOLLOW)?;
+            if before.st_dev != after.st_dev || before.st_ino != after.st_ino || before.st_size != after.st_size {
+                return Err(invalid("streaming part changed"));
+            }
+        }
+        if i64::try_from(object_size).map_err(invalid)? != info.size {
+            return Err(invalid("streaming object length mismatch"));
+        }
+        sync(&data, dir.preparation, "parts:directory")?;
+        let current = fs::statat(&dir.file, &name, AtFlags::SYMLINK_NOFOLLOW)?;
+        if identity.st_dev != current.st_dev || identity.st_ino != current.st_ino {
+            return Err(invalid("streaming directory changed"));
+        }
+        dir.check()?;
+        Ok((digest.finalize().into(), total))
+    }
+
     pub(super) fn inspect(root: &Path, preparation: Uuid) -> io::Result<Record> {
         let dir = Directory::open(root, preparation, false)?;
         let record: Record = serde_json::from_slice(
@@ -792,11 +1333,14 @@ mod durable {
                 .ok_or_else(|| invalid("missing native intent"))?,
         )
         .map_err(invalid)?;
-        if !matches!((record.format, record.recovery_metadata.is_some()), (1, false) | (2, true))
+        if !matches!((record.format, record.recovery_metadata.is_some()), (1, false) | (2, true) | (3, true))
             || record.preparation != preparation
             || record.version.is_nil()
             || record.disk.is_nil()
             || record.incarnation.is_nil()
+            || (record.format == 3
+                && (record.set_disks.as_ref().is_none_or(|set| set.is_empty()) || record.data_dir.is_none_or(|id| id.is_nil())))
+            || (record.format < 3 && (record.set_disks.is_some() || record.data_dir.is_some()))
         {
             return Err(invalid("invalid native intent"));
         }
@@ -846,8 +1390,9 @@ mod durable {
             || decoded.version_id != Some(record.version)
             || decoded.deleted
             || decoded.mark_deleted
-            || !decoded.inline_data()
-            || decoded.parts.len() != 1
+            || decoded.parts.is_empty()
+            || (record.format < 3 && (!decoded.inline_data() || decoded.parts.len() != 1))
+            || (record.format == 3 && (decoded.inline_data() || decoded.data_dir != record.data_dir))
             || decoded.size != i64::try_from(record.object_length).map_err(invalid)?
             || decoded.erasure.data_blocks != record.data_blocks
             || decoded.erasure.parity_blocks != record.parity_blocks
@@ -855,7 +1400,9 @@ mod durable {
         {
             return Err(invalid("native metadata identity mismatch"));
         }
-        if decoded.parts[0].number != 1 || decoded.parts[0].size != usize::try_from(decoded.size).map_err(invalid)? {
+        if record.format < 3
+            && (decoded.parts[0].number != 1 || decoded.parts[0].size != usize::try_from(decoded.size).map_err(invalid)?)
+        {
             return Err(invalid("native part identity mismatch"));
         }
         let (deployment, format_digest, set) = verify_disk_layout(&dir, disk, store_id, &decoded)?;
@@ -866,37 +1413,46 @@ mod durable {
             || record.disk_format_digest != format_digest
             || record.set_digest != set_digest
             || record.distribution_digest != distribution_digest
+            || (record.format == 3 && record.set_disks.as_ref() != Some(&set))
         {
             return Err(invalid("native preparation physical layout mismatch"));
         }
-        let shard = decoded.data.as_ref().ok_or_else(|| invalid("missing native shard"))?;
-        if u64::try_from(shard.len()).map_err(invalid)? != record.shard_length
-            || Sha256::digest(shard).as_slice() != record.shard_digest
-        {
-            return Err(invalid("corrupt native shard"));
-        }
-        let part = &decoded.parts[0];
-        let codec = crate::erasure::coding::Erasure::try_new_with_options(
-            decoded.erasure.data_blocks,
-            decoded.erasure.parity_blocks,
-            decoded.erasure.block_size,
-            decoded.uses_legacy_checksum,
-        )
-        .map_err(|error| error.into_io_error())?;
-        let checksum = decoded.erasure.get_checksum_info(1).algorithm;
-        let checksum = if decoded.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
-            rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+        if record.format == 3 {
+            let (digest, length) = stream_parts(&dir, &decoded, &runtime)?;
+            if digest != record.shard_digest || length != record.shard_length {
+                return Err(invalid("corrupt native streaming shards"));
+            }
         } else {
-            checksum
-        };
-        let shard_size = usize::try_from(codec.shard_file_size(i64::try_from(part.size).map_err(invalid)?)).map_err(invalid)?;
-        runtime.block_on(crate::erasure::coding::bitrot_verify(
-            std::io::Cursor::new(shard),
-            shard.len(),
-            shard_size,
-            checksum,
-            codec.shard_size(),
-        ))?;
+            let shard = decoded.data.as_ref().ok_or_else(|| invalid("missing native shard"))?;
+            if u64::try_from(shard.len()).map_err(invalid)? != record.shard_length
+                || Sha256::digest(shard).as_slice() != record.shard_digest
+            {
+                return Err(invalid("corrupt native shard"));
+            }
+            let part = &decoded.parts[0];
+            let codec = crate::erasure::coding::Erasure::try_new_with_options(
+                decoded.erasure.data_blocks,
+                decoded.erasure.parity_blocks,
+                decoded.erasure.block_size,
+                decoded.uses_legacy_checksum,
+            )
+            .map_err(|error| error.into_io_error())?;
+            let checksum = decoded.erasure.get_checksum_info(1).algorithm;
+            let checksum = if decoded.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
+                rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+            } else {
+                checksum
+            };
+            let shard_size =
+                usize::try_from(codec.shard_file_size(i64::try_from(part.size).map_err(invalid)?)).map_err(invalid)?;
+            runtime.block_on(crate::erasure::coding::bitrot_verify(
+                std::io::Cursor::new(shard),
+                shard.len(),
+                shard_size,
+                checksum,
+                codec.shard_size(),
+            ))?;
+        }
         let marker = dir
             .read("PREPARED", 32)?
             .ok_or_else(|| invalid("native preparation not committed"))?;
@@ -978,9 +1534,10 @@ mod durable {
         version: Uuid,
         input: NativePrepareInput,
         info: FileInfo,
+        runtime: tokio::runtime::Handle,
     ) -> io::Result<NativePreparationReceipt> {
         let metadata = &input.metadata;
-        let shard = info.data.as_ref().ok_or_else(|| invalid("missing native shard"))?;
+        let shard = info.data.as_deref().unwrap_or(&[]);
 
         if metadata.len() > METADATA_LIMIT {
             return Err(invalid("native preparation metadata too large"));
@@ -988,6 +1545,14 @@ mod durable {
         let dir = Directory::open(root, input.preparation, true)?;
         let (deployment, format_digest, set) = verify_disk_layout(&dir, disk, input.store_id, &info)?;
         let mut record = Record::new(&input, disk, &info, metadata, shard, version, deployment, format_digest, &set)?;
+        if !info.inline_data() {
+            let (digest, length) = stream_parts(&dir, &info, &runtime)?;
+            record.format = 3;
+            record.set_disks = Some(set);
+            record.data_dir = info.data_dir;
+            record.shard_digest = digest;
+            record.shard_length = length;
+        }
         let existing = dir.read("intent.json", RECORD_LIMIT)?;
         if let Some(bytes) = &existing {
             let saved: Record = serde_json::from_slice(bytes).map_err(invalid)?;
@@ -1083,6 +1648,7 @@ mod tests {
         seen: Vec<String>,
         fail_before: Option<String>,
         fail_after: Option<String>,
+        repeat_failure: bool,
         pause: Option<(String, std::sync::mpsc::SyncSender<()>, std::sync::mpsc::Receiver<()>)>,
     }
     static SYNC_PROBES: std::sync::LazyLock<std::sync::Mutex<std::collections::HashMap<Uuid, SyncProbe>>> =
@@ -1102,6 +1668,15 @@ mod tests {
         );
         (entered_rx, release_tx)
     }
+    pub(crate) fn fail_native_preparation(preparation: Uuid, stage: &str, after: bool) {
+        let probe = SyncProbe {
+            fail_before: (!after).then(|| stage.to_owned()),
+            fail_after: after.then(|| stage.to_owned()),
+            repeat_failure: true,
+            ..Default::default()
+        };
+        SYNC_PROBES.lock().expect("probe mutex").insert(preparation, probe);
+    }
     pub(crate) fn clear_native_preparation_probe(preparation: Uuid) {
         SYNC_PROBES.lock().expect("probe mutex").remove(&preparation);
     }
@@ -1111,7 +1686,9 @@ mod tests {
             if let Some(p) = probes.get_mut(&id) {
                 p.seen.push(stage.into());
                 if p.fail_before.as_deref() == Some(stage) {
-                    p.fail_before = None;
+                    if !p.repeat_failure {
+                        p.fail_before = None;
+                    }
                     return Err(std::io::Error::other("injected before native fsync"));
                 }
                 if p.pause.as_ref().is_some_and(|(name, _, _)| name == stage) {
@@ -1133,7 +1710,9 @@ mod tests {
         let mut probes = SYNC_PROBES.lock().expect("probe mutex");
         if let Some(p) = probes.get_mut(&id) {
             if p.fail_after.as_deref() == Some(stage) {
-                p.fail_after = None;
+                if !p.repeat_failure {
+                    p.fail_after = None;
+                }
                 return Err(std::io::Error::other("injected unknown native fsync outcome"));
             }
         }
@@ -1540,6 +2119,43 @@ mod tests {
             );
         }
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn native_stream_retries_data_dir_parent_barrier_after_failure() {
+        let (dirs, _store, disk, input) = fixture().await;
+        disk.prepare_native(input.clone()).await.expect("prepare durable root");
+        let data_dir = Uuid::new_v4();
+        let stage = format!(".rustfs-put-{}", Uuid::new_v4());
+        SYNC_PROBES.lock().expect("probe").insert(
+            input.preparation,
+            SyncProbe {
+                fail_before: Some("stream:data-dir-parent".into()),
+                ..Default::default()
+            },
+        );
+
+        assert!(super::durable::open_stream_file(dirs[0].path(), input.preparation, data_dir, stage.clone()).is_err());
+        let file =
+            super::durable::open_stream_file(dirs[0].path(), input.preparation, data_dir, stage).expect("redo parent barrier");
+        drop(file);
+
+        let stages = SYNC_PROBES
+            .lock()
+            .expect("probe")
+            .remove(&input.preparation)
+            .expect("recorded fsync calls")
+            .seen;
+        assert_eq!(
+            stages
+                .iter()
+                .filter(|stage| stage.as_str() == "stream:data-dir-parent")
+                .count(),
+            2,
+            "NATIVE_STREAM_PARENT_BARRIER_RETRY_MUST_FSYNC_EXISTING_DIRECTORY"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn native_prepare_restart_recovers_exact_identity() {

crates/ecstore/src/disk/mod.rs

@@ -24,6 +24,7 @@ pub mod fs;
 pub mod health_state;
 mod io_admission;
 pub mod local;
+pub(crate) mod native_prepare_rpc;
 pub mod os;
 
 #[cfg(test)]
@@ -230,6 +231,152 @@ pub enum Disk {
 }
 
 impl Disk {
+    #[cfg(unix)]
+    pub(crate) async fn native_begin_stream(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+        fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<()> {
+        match self {
+            Disk::Local(local) => {
+                local
+                    .get_disk()
+                    .begin_native_stream(preparation, store_id, bucket, incarnation, fence)
+                    .await
+            }
+            Disk::Remote(remote) => {
+                remote
+                    .native_begin_stream(preparation, store_id, bucket, key, incarnation)
+                    .await
+            }
+        }
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn native_begin_stream(
+        &self,
+        _preparation: Uuid,
+        _store_id: Uuid,
+        _bucket: &str,
+        _key: &str,
+        _incarnation: Uuid,
+        _fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<()> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
+    #[cfg(unix)]
+    pub(crate) async fn native_finalize(
+        &self,
+        input: local::native_prepare::NativePrepareInput,
+    ) -> Result<local::native_prepare::NativePreparationReceipt> {
+        match self {
+            Disk::Local(local) => local.get_disk().prepare_native(input).await,
+            Disk::Remote(remote) => remote.native_finalize(input).await,
+        }
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn native_finalize(
+        &self,
+        _input: local::native_prepare::NativePrepareInput,
+    ) -> Result<local::native_prepare::NativePreparationReceipt> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
+    #[cfg(unix)]
+    pub(crate) async fn native_recover(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+        fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<(local::native_prepare::NativePreparationReceipt, Bytes)> {
+        match self {
+            Disk::Local(local) => {
+                local
+                    .get_disk()
+                    .recover_native_with_metadata(preparation, store_id, bucket, incarnation, fence)
+                    .await
+            }
+            Disk::Remote(remote) => remote.native_recover(preparation, store_id, bucket, key, incarnation).await,
+        }
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn native_recover(
+        &self,
+        _preparation: Uuid,
+        _store_id: Uuid,
+        _bucket: &str,
+        _key: &str,
+        _incarnation: Uuid,
+        _fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<(local::native_prepare::NativePreparationReceipt, Bytes)> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
+    #[cfg(unix)]
+    pub(crate) async fn native_probe(
+        &self,
+        preparation: Uuid,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        incarnation: Uuid,
+        fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<local::native_prepare::NativePreparationProbe> {
+        match self {
+            Disk::Local(local) => {
+                local
+                    .get_disk()
+                    .probe_native(preparation, store_id, bucket, key, incarnation, fence)
+                    .await
+            }
+            Disk::Remote(remote) => remote.native_probe(preparation, store_id, bucket, key, incarnation).await,
+        }
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn native_probe(
+        &self,
+        _preparation: Uuid,
+        _store_id: Uuid,
+        _bucket: &str,
+        _key: &str,
+        _incarnation: Uuid,
+        _fence: Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<local::native_prepare::NativePreparationProbe> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
+    #[cfg(unix)]
+    pub(crate) async fn native_locate(
+        &self,
+        store_id: Uuid,
+        preparation: Uuid,
+    ) -> Result<Option<local::native_prepare::NativePreparationDescription>> {
+        match self {
+            Disk::Local(local) => local.get_disk().locate_native_preparation(store_id, preparation).await,
+            Disk::Remote(remote) => remote.native_locate(store_id, preparation).await,
+        }
+    }
+
+    #[cfg(not(unix))]
+    pub(crate) async fn native_locate(
+        &self,
+        _store_id: Uuid,
+        _preparation: Uuid,
+    ) -> Result<Option<local::native_prepare::NativePreparationDescription>> {
+        Err(DiskError::UnsupportedDisk)
+    }
+
     pub(crate) async fn set_disk_id_state(&self, id: Option<Uuid>) -> Result<()> {
         match self {
             Disk::Local(local_disk) => {

crates/ecstore/src/disk/native_prepare_rpc.rs

@@ -0,0 +1,145 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::disk::{
+    RUSTFS_META_BUCKET,
+    error::{DiskError, Result},
+};
+use rustfs_utils::path::path_join_buf;
+use uuid::Uuid;
+
+const NATIVE_STREAM_PREFIX: &str = "native-preparations/";
+const NATIVE_STREAM_ROOT: &str = "native-preparations";
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) enum NativeStreamFile {
+    Part(usize),
+    Staging(Uuid),
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub(crate) struct NativeStreamPath {
+    pub(crate) preparation: Uuid,
+    pub(crate) data_dir: Uuid,
+    pub(crate) file: NativeStreamFile,
+}
+
+fn targets_native_stream_namespace(volume: &str, path: &str) -> bool {
+    if !volume.contains(RUSTFS_META_BUCKET) && !path.contains(NATIVE_STREAM_ROOT) {
+        return false;
+    }
+    let normalized = path_join_buf(&[volume, path]);
+    let prefix = format!("{RUSTFS_META_BUCKET}/{NATIVE_STREAM_PREFIX}");
+    normalized == format!("{RUSTFS_META_BUCKET}/{NATIVE_STREAM_ROOT}") || normalized.starts_with(&prefix)
+}
+
+/// Recognize only the reserved per-preparation shard path used by the native
+/// stream writer. Malformed paths under that namespace must not fall through to
+/// the general disk writer.
+pub(crate) fn parse_native_stream_path(volume: &str, path: &str) -> Result<Option<NativeStreamPath>> {
+    if !targets_native_stream_namespace(volume, path) {
+        return Ok(None);
+    }
+    if volume != RUSTFS_META_BUCKET {
+        return Err(DiskError::InvalidPath);
+    }
+    if !path.starts_with(NATIVE_STREAM_PREFIX) {
+        return Err(DiskError::InvalidPath);
+    }
+
+    let mut components = path.split('/');
+    if components.next() != Some("native-preparations") {
+        return Err(DiskError::InvalidPath);
+    }
+    let preparation = components
+        .next()
+        .and_then(|value| Uuid::parse_str(value).ok())
+        .filter(|value| !value.is_nil())
+        .ok_or(DiskError::InvalidPath)?;
+    let data_dir = components
+        .next()
+        .and_then(|value| Uuid::parse_str(value).ok())
+        .filter(|value| !value.is_nil())
+        .ok_or(DiskError::InvalidPath)?;
+    let file_name = components.next().ok_or(DiskError::InvalidPath)?;
+    let file = file_name
+        .strip_prefix("part.")
+        .and_then(|value| value.parse::<usize>().ok())
+        .filter(|value| *value > 0)
+        .filter(|part| file_name == format!("part.{part}"))
+        .map(NativeStreamFile::Part)
+        .or_else(|| {
+            file_name
+                .strip_prefix(".rustfs-put-")
+                .and_then(|value| Uuid::parse_str(value).ok())
+                .filter(|value| !value.is_nil())
+                .filter(|id| file_name == format!(".rustfs-put-{id}"))
+                .map(NativeStreamFile::Staging)
+        });
+
+    if components.next().is_some()
+        || file.is_none()
+        || path != format!("native-preparations/{preparation}/{data_dir}/{file_name}")
+    {
+        return Err(DiskError::InvalidPath);
+    }
+    Ok(Some(NativeStreamPath {
+        preparation,
+        data_dir,
+        file: file.expect("validated native stream filename"),
+    }))
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn native_stream_path_requires_canonical_identity_and_part() {
+        let preparation = Uuid::from_u128(1);
+        let data_dir = Uuid::from_u128(2);
+        let valid = format!("native-preparations/{preparation}/{data_dir}/part.1");
+        assert!(
+            parse_native_stream_path(RUSTFS_META_BUCKET, &valid)
+                .expect("valid path")
+                .is_some()
+        );
+        let staging = format!("native-preparations/{preparation}/{data_dir}/.rustfs-put-{data_dir}");
+        assert!(
+            parse_native_stream_path(RUSTFS_META_BUCKET, &staging)
+                .expect("valid staging path")
+                .is_some()
+        );
+        assert!(parse_native_stream_path("bucket", &valid).expect("other volume").is_none());
+        assert!(
+            parse_native_stream_path(RUSTFS_META_BUCKET, "ordinary/object")
+                .expect("ordinary path")
+                .is_none()
+        );
+        for invalid in [
+            "native-preparations/00000000-0000-0000-0000-000000000000/00000000-0000-0000-0000-000000000002/part.1".to_owned(),
+            format!("native-preparations/{preparation}/{data_dir}/part.0"),
+            format!("native-preparations/{preparation}/{data_dir}/part.01"),
+            format!("native-preparations/{preparation}/{data_dir}/.rustfs-put-{}", Uuid::nil()),
+            format!("native-preparations/{preparation}/{data_dir}/.rustfs-put-not-a-uuid"),
+            format!("native-preparations/{preparation}/{data_dir}/part.1/extra"),
+            format!("native-preparations/{preparation}/../part.1"),
+        ] {
+            assert!(matches!(
+                parse_native_stream_path(RUSTFS_META_BUCKET, &invalid),
+                Err(DiskError::InvalidPath)
+            ));
+        }
+    }
+}

crates/ecstore/src/set_disk/ops/native_prepare.rs

@@ -32,7 +32,322 @@ pub(crate) fn same_identity(a: &NativePreparationReceipt, b: &NativePreparationR
         && a.erasure_geometry().1 == b.erasure_geometry().1
 }
 
+pub(super) struct DigestReader<R> {
+    inner: R,
+    digest: Sha256,
+}
+impl<R> DigestReader<R> {
+    pub(super) fn new(inner: R) -> Self {
+        Self {
+            inner,
+            digest: Sha256::new(),
+        }
+    }
+    pub(super) fn into_inner_and_digest(self) -> (R, [u8; 32]) {
+        (self.inner, self.digest.finalize().into())
+    }
+}
+impl<R: tokio::io::AsyncRead + Unpin> tokio::io::AsyncRead for DigestReader<R> {
+    fn poll_read(
+        mut self: std::pin::Pin<&mut Self>,
+        cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        let before = buf.filled().len();
+        let result = std::pin::Pin::new(&mut self.inner).poll_read(cx, buf);
+        if matches!(result, std::task::Poll::Ready(Ok(()))) {
+            self.digest.update(&buf.filled()[before..]);
+        }
+        result
+    }
+}
+
+struct DigestSink {
+    digest: Sha256,
+    length: usize,
+}
+impl tokio::io::AsyncWrite for DigestSink {
+    fn poll_write(
+        mut self: std::pin::Pin<&mut Self>,
+        _: &mut std::task::Context<'_>,
+        bytes: &[u8],
+    ) -> std::task::Poll<std::io::Result<usize>> {
+        match self.length.checked_add(bytes.len()) {
+            Some(length) => {
+                self.length = length;
+                self.digest.update(bytes);
+                std::task::Poll::Ready(Ok(bytes.len()))
+            }
+            None => std::task::Poll::Ready(Err(std::io::Error::other("native content length overflow"))),
+        }
+    }
+    fn poll_flush(self: std::pin::Pin<&mut Self>, _: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+    fn poll_shutdown(self: std::pin::Pin<&mut Self>, _: &mut std::task::Context<'_>) -> std::task::Poll<std::io::Result<()>> {
+        std::task::Poll::Ready(Ok(()))
+    }
+}
+
 impl SetDisks {
+    fn native_expected_set_digest(&self, store_id: Uuid) -> Result<[u8; 32]> {
+        let set = self
+            .format
+            .erasure
+            .sets
+            .get(self.set_index)
+            .ok_or_else(|| Error::other("native target set missing"))?;
+        if self.format.id != store_id
+            || set.len() != self.set_drive_count
+            || set.iter().any(Uuid::is_nil)
+            || set.iter().enumerate().any(|(slot, id)| set[slot + 1..].contains(id))
+        {
+            return Err(Error::other("invalid native target physical set"));
+        }
+        Ok(Sha256::digest(serde_json::to_vec(set)?).into())
+    }
+
+    /// Attest already encoded private shards while the caller owns the actual PUT commit guards.
+    pub(crate) async fn attest_native_stream(
+        &self,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        preparation: Uuid,
+        incarnation: Uuid,
+        candidates: Vec<(crate::disk::DiskStore, FileInfo)>,
+        object_digest: [u8; 32],
+        object_length: u64,
+        fence: Arc<ObjectLockConfigSnapshot>,
+    ) -> Result<crate::store::native_prepare::NativeQuorumPreparation> {
+        self.attest_native_stream_owned(
+            store_id,
+            bucket,
+            key,
+            preparation,
+            incarnation,
+            candidates,
+            object_digest,
+            object_length,
+            fence,
+            None,
+        )
+        .await
+    }
+    async fn attest_native_stream_owned(
+        &self,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        preparation: Uuid,
+        incarnation: Uuid,
+        candidates: Vec<(crate::disk::DiskStore, FileInfo)>,
+        object_digest: [u8; 32],
+        object_length: u64,
+        fence: Arc<ObjectLockConfigSnapshot>,
+        operation_owner: Option<Arc<rustfs_lock::NamespaceLockGuard>>,
+    ) -> Result<crate::store::native_prepare::NativeQuorumPreparation> {
+        let expected_set = self.native_expected_set_digest(store_id)?;
+        if !fence.is_valid_for_destructive_put(store_id, bucket, incarnation) {
+            return Err(Error::other("native adoption bucket fence mismatch"));
+        }
+        let layout = resolve_write_layout(
+            &self.storage_class_config_snapshot(),
+            self.pool_index,
+            self.set_drive_count,
+            self.default_parity_count,
+            None,
+            false,
+        )?;
+        let (send, mut receive) = mpsc::unbounded_channel();
+        for (disk, info) in candidates {
+            if info.size != i64::try_from(object_length).map_err(Error::other)?
+                || (info.name != key && info.name != format!("{bucket}/{key}"))
+                || info.inline_data()
+                || info.deleted
+                || info.mark_deleted
+            {
+                return Err(Error::other("native adoption metadata identity mismatch"));
+            }
+            let version = info
+                .version_id
+                .filter(|id| !id.is_nil())
+                .ok_or_else(|| Error::other("native adoption version missing"))?;
+            let mut metadata = FileMeta::new();
+            metadata.add_version(info)?;
+            let input = NativePrepareInput {
+                preparation,
+                store_id,
+                bucket: bucket.into(),
+                bucket_incarnation: incarnation,
+                object_digest,
+                key: key.into(),
+                version,
+                metadata: metadata.marshal_msg()?.into(),
+                bucket_fence: fence.clone(),
+            };
+            let send = send.clone();
+            let owner = operation_owner.clone();
+            tokio::spawn(async move {
+                let result = match (disk.as_ref(), owner) {
+                    (Disk::Local(local), Some(owner)) => local.get_disk().prepare_native_with_owner(input, owner).await,
+                    (_, owner) => {
+                        let _owner = owner;
+                        disk.native_finalize(input).await
+                    }
+                };
+                let _ = send.send(result);
+            });
+        }
+        drop(send);
+        let mut receipts = Vec::new();
+        while let Some(result) = receive.recv().await {
+            let receipt = match result {
+                Ok(receipt) => receipt,
+                Err(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::InvalidData => return Err(error.into()),
+                Err(_) => continue,
+            };
+            if receipt.set_digest() != &expected_set
+                || receipts.first().is_some_and(|old| !same_identity(old, &receipt))
+                || receipts
+                    .iter()
+                    .any(|old: &NativePreparationReceipt| old.disk_id() == receipt.disk_id())
+            {
+                return Err(Error::other("native adoption quorum identity mismatch"));
+            }
+            receipts.push(receipt);
+        }
+        crate::store::native_prepare::NativeQuorumPreparation::from_receipts(layout.write_quorum, receipts)
+    }
+
+    pub(crate) async fn prepare_native_stream(
+        &self,
+        store_id: Uuid,
+        bucket: &str,
+        key: &str,
+        preparation: Uuid,
+        incarnation: Uuid,
+        data: &mut crate::object_api::PutObjReader,
+        fence: Arc<ObjectLockConfigSnapshot>,
+        operation_owner: Arc<rustfs_lock::NamespaceLockGuard>,
+    ) -> Result<(usize, Vec<NativePreparationReceipt>)> {
+        use crate::disk::RUSTFS_META_BUCKET;
+        let disks = self.get_disks_internal().await;
+        let layout = resolve_write_layout(
+            &self.storage_class_config_snapshot(),
+            self.pool_index,
+            disks.len(),
+            self.default_parity_count,
+            None,
+            false,
+        )?;
+        let mut info = FileInfo::new(key, layout.data_drives, layout.parity_drives);
+        info.name = key.to_owned();
+        info.version_id = Some(preparation);
+        info.data_dir = Some(preparation);
+        info.size = data.size();
+        info.versioned = true;
+        info.mod_time = Some(time::OffsetDateTime::now_utc());
+        let erasure = Arc::new(Erasure::try_new_with_options(
+            info.erasure.data_blocks,
+            info.erasure.parity_blocks,
+            info.erasure.block_size,
+            false,
+        )?);
+        let mut shuffled = Self::shuffle_disks_owned(disks.clone(), &info.erasure.distribution);
+        let path = format!("native-preparations/{preparation}/{preparation}/part.1");
+        let mut writers = Vec::with_capacity(shuffled.len());
+        for disk in &mut shuffled {
+            let result = match disk {
+                Some(disk) => {
+                    if let Err(error) = disk
+                        .native_begin_stream(preparation, store_id, bucket, key, incarnation, fence.clone())
+                        .await
+                    {
+                        if matches!(&error, DiskError::Io(error) if error.kind() == std::io::ErrorKind::InvalidData) {
+                            return Err(error.into());
+                        }
+                        writers.push(None);
+                        continue;
+                    }
+                    crate::io_support::bitrot::create_bitrot_writer(
+                        false,
+                        Some(disk),
+                        RUSTFS_META_BUCKET,
+                        &path,
+                        erasure.shard_file_size(info.size),
+                        erasure.shard_size(),
+                        rustfs_utils::HashAlgorithm::HighwayHash256S,
+                    )
+                    .await
+                }
+                None => Err(DiskError::DiskNotFound),
+            };
+            match result {
+                Ok(writer) => writers.push(Some(writer)),
+                Err(_) => {
+                    *disk = None;
+                    writers.push(None);
+                }
+            }
+        }
+        let stream = std::mem::replace(
+            &mut data.stream,
+            rustfs_rio::HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
+        );
+        let reader = DigestReader::new(stream);
+        let (reader, size, _, _) = erasure
+            .encode_with_shard_integrity(reader, &mut writers, layout.write_quorum, 1, IntegrityEncodeMode::Streaming, false)
+            .await?;
+        let (stream, digest) = reader.into_inner_and_digest();
+        data.stream = stream;
+        if info.size >= 0 && i64::try_from(size).map_err(Error::other)? != info.size {
+            return Err(Error::other("native stream length mismatch"));
+        }
+        info.size = i64::try_from(size).map_err(Error::other)?;
+        info.add_object_part(1, String::new(), size, None, info.size, None, None);
+        let mut candidates = Vec::new();
+        for (slot, disk) in disks.into_iter().enumerate() {
+            let index = *info
+                .erasure
+                .distribution
+                .get(slot)
+                .ok_or_else(|| Error::other("native stream distribution missing"))?;
+            if writers
+                .get(
+                    index
+                        .checked_sub(1)
+                        .ok_or_else(|| Error::other("invalid native stream index"))?,
+                )
+                .and_then(Option::as_ref)
+                .is_none()
+            {
+                continue;
+            }
+            let Some(disk) = disk else {
+                continue;
+            };
+            let mut per_disk = info.clone();
+            per_disk.erasure.index = index;
+            candidates.push((disk, per_disk));
+        }
+        let owner = self
+            .attest_native_stream_owned(
+                store_id,
+                bucket,
+                key,
+                preparation,
+                incarnation,
+                candidates,
+                digest,
+                u64::try_from(size).map_err(Error::other)?,
+                fence,
+                Some(operation_owner),
+            )
+            .await?;
+        Ok((layout.write_quorum, owner.into_receipts()))
+    }
+
     pub(crate) async fn prepare_native_bytes(
         &self,
         store_id: Uuid,
@@ -237,8 +552,9 @@ impl SetDisks {
         incarnation: Uuid,
         fence: Arc<ObjectLockConfigSnapshot>,
         prepared_slots: Vec<bool>,
-    ) -> Result<(usize, Vec<NativePreparationReceipt>)> {
+    ) -> Result<(usize, Vec<NativePreparationReceipt>, Vec<(Uuid, Bytes)>)> {
         let disks = self.get_disks_internal().await;
+        let expected_set = self.native_expected_set_digest(store_id)?;
         let layout = resolve_write_layout(
             &self.storage_class_config_snapshot(),
             self.pool_index,
@@ -248,7 +564,9 @@ impl SetDisks {
             false,
         )?;
         let mut receipts: Vec<NativePreparationReceipt> = Vec::new();
-        let mut readers = std::iter::repeat_with(|| None).take(disks.len()).collect::<Vec<_>>();
+        let mut verified_metadata = Vec::new();
+        let mut sources: Vec<Option<(crate::disk::DiskStore, FileInfo)>> =
+            std::iter::repeat_with(|| None).take(disks.len()).collect();
         let mut erasure = None;
         let mut modification_time = None;
         let mut legacy_codec = None;
@@ -259,20 +577,18 @@ impl SetDisks {
                 continue;
             }
             let disk = disk.ok_or_else(|| Error::other("prepared native disk unavailable"))?;
-            let Disk::Local(wrapper) = disk.as_ref() else {
-                return Err(Error::other("native remote recovery unsupported"));
-            };
-            let local = wrapper.get_disk();
-            let (receipt, metadata) = match local
-                .recover_native_with_metadata(preparation, store_id, bucket, incarnation, fence.clone())
+            let (receipt, metadata) = match disk
+                .native_recover(preparation, store_id, bucket, key, incarnation, fence.clone())
                 .await
             {
                 Ok(value) => value,
                 Err(error) => return Err(error.into()),
             };
+            if receipt.set_digest() != &expected_set {
+                return Err(Error::other("native recovery receipt target physical set mismatch"));
+            }
             if receipt.key() != key
                 || receipt.preparation_id() != preparation
-                || receipt.version_id() != preparation
                 || receipt.bucket() != bucket
                 || receipt.bucket_incarnation() != incarnation
                 || receipt.deployment_id() != store_id
@@ -287,11 +603,12 @@ impl SetDisks {
             if receipts.iter().any(|r| r.disk_id() == receipt.disk_id()) {
                 return Err(Error::other("duplicate native recovery disk"));
             }
+            verified_metadata.push((receipt.disk_id(), metadata.clone()));
             let info = rustfs_filemeta::get_file_info(
                 &metadata,
                 bucket,
                 key,
-                &preparation.to_string(),
+                &receipt.version_id().to_string(),
                 FileInfoOpts {
                     data: true,
                     include_free_versions: false,
@@ -327,24 +644,18 @@ impl SetDisks {
                 .index
                 .checked_sub(1)
                 .ok_or_else(|| Error::other("invalid recovered native index"))?;
-            let slot = readers
+            let slot = sources
                 .get_mut(index)
                 .ok_or_else(|| Error::other("recovered native index outside set"))?;
             if slot.is_some() {
                 return Err(Error::other("duplicate recovered native shard"));
             }
-            let checksum = info.erasure.get_checksum_info(1).algorithm;
-            let checksum = if info.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
-                rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
-            } else {
-                checksum
-            };
-            *slot = Some(BitrotReader::new(
-                Cursor::new(info.data.ok_or_else(|| Error::other("native recovery shard missing"))?),
-                selected.shard_size(),
-                checksum,
-                false,
-            ));
+            if let Some((_, previous)) = sources.iter().flatten().next() {
+                if previous.parts != info.parts || previous.data_dir != info.data_dir {
+                    return Err(Error::other("conflicting native recovered part manifest"));
+                }
+            }
+            sources[index] = Some((disk.clone(), info));
             erasure = Some(selected);
             receipts.push(receipt);
         }
@@ -353,21 +664,81 @@ impl SetDisks {
         }
         let first = receipts.first().ok_or(Error::ErasureWriteQuorum)?;
         let length = usize::try_from(first.object_length()).map_err(Error::other)?;
-        if length > 1024 * 1024 {
-            return Err(Error::other("native recovery length exceeds inline limit"));
-        }
-        let mut output = Vec::with_capacity(length);
-        let (size, failure) = erasure
+        let mut output = DigestSink {
+            digest: Sha256::new(),
+            length: 0,
+        };
+        let codec = erasure.ok_or(Error::ErasureWriteQuorum)?;
+        let parts = sources
+            .iter()
+            .flatten()
+            .next()
             .ok_or(Error::ErasureWriteQuorum)?
-            .decode(&mut output, readers, 0, length, length)
-            .await;
-        if let Some(error) = failure {
-            return Err(Error::from(error));
+            .1
+            .parts
+            .clone();
+        let mut size = 0_usize;
+        for part in parts {
+            let mut readers = Vec::with_capacity(sources.len());
+            for source in &sources {
+                let Some((disk, info)) = source else {
+                    readers.push(None);
+                    continue;
+                };
+                let checksum = info.erasure.get_checksum_info(part.number).algorithm;
+                let checksum = if info.uses_legacy_checksum && checksum == rustfs_utils::HashAlgorithm::HighwayHash256S {
+                    rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+                } else {
+                    checksum
+                };
+                let reader = if info.inline_data() {
+                    crate::io_support::bitrot::ShardReader::InMemory(Cursor::new(
+                        info.data.clone().ok_or_else(|| Error::other("native inline shard missing"))?,
+                    ))
+                } else {
+                    use crate::disk::DiskAPI;
+                    let data_dir = info.data_dir.ok_or_else(|| Error::other("native stream data dir missing"))?;
+                    let shard_size = usize::try_from(codec.shard_file_size(i64::try_from(part.size).map_err(Error::other)?))
+                        .map_err(Error::other)?;
+                    let physical = if matches!(
+                        checksum,
+                        rustfs_utils::HashAlgorithm::HighwayHash256S | rustfs_utils::HashAlgorithm::HighwayHash256SLegacy
+                    ) {
+                        shard_size
+                            .div_ceil(codec.shard_size())
+                            .checked_mul(checksum.size())
+                            .and_then(|hashes| shard_size.checked_add(hashes))
+                            .ok_or_else(|| Error::other("native physical shard length overflow"))?
+                    } else {
+                        shard_size
+                    };
+                    crate::io_support::bitrot::ShardReader::Stream(
+                        disk.read_file_stream(
+                            crate::disk::RUSTFS_META_BUCKET,
+                            &format!("native-preparations/{preparation}/{data_dir}/part.{}", part.number),
+                            0,
+                            physical,
+                        )
+                        .await?,
+                    )
+                };
+                readers.push(Some(BitrotReader::new(reader, codec.shard_size(), checksum, false)));
+            }
+            let (decoded, failure) = codec.decode(&mut output, readers, 0, part.size, part.size).await;
+            if let Some(error) = failure {
+                return Err(error.into());
+            }
+            if decoded != part.size {
+                return Err(Error::other("native recovered part length mismatch"));
+            }
+            size = size
+                .checked_add(decoded)
+                .ok_or_else(|| Error::other("native decoded length overflow"))?;
         }
-        if size != length || output.len() != length || Sha256::digest(&output).as_slice() != first.object_digest() {
+        if size != length || output.length != length || output.digest.finalize().as_slice() != first.object_digest() {
             return Err(Error::other("native recovered content digest mismatch"));
         }
-        Ok((layout.write_quorum, receipts))
+        Ok((layout.write_quorum, receipts, verified_metadata))
     }
 }
 
@@ -380,19 +751,30 @@ impl SetDisks {
         preparation: Uuid,
         incarnation: Uuid,
         fence: Arc<ObjectLockConfigSnapshot>,
+        match_object: bool,
     ) -> Vec<std::result::Result<NativePreparationProbe, DiskError>> {
         let mut observations = Vec::new();
         for disk in self.get_disks_internal().await {
             observations.push(match disk {
-                Some(disk) => match disk.as_ref() {
-                    Disk::Local(local) => {
-                        local
-                            .get_disk()
-                            .probe_native(preparation, store_id, bucket, key, incarnation, fence.clone())
-                            .await
+                Some(disk) => {
+                    if match_object {
+                        match disk.native_locate(store_id, preparation).await {
+                            Ok(Some(record))
+                                if record.bucket != bucket || record.key != key || record.incarnation != incarnation =>
+                            {
+                                observations.push(Ok(NativePreparationProbe::Absent));
+                                continue;
+                            }
+                            Err(error) => {
+                                observations.push(Err(error));
+                                continue;
+                            }
+                            _ => {}
+                        }
                     }
-                    Disk::Remote(_) => Err(DiskError::InvalidPath),
-                },
+                    disk.native_probe(preparation, store_id, bucket, key, incarnation, fence.clone())
+                        .await
+                }
                 None => Err(DiskError::InvalidPath),
             });
         }

crates/ecstore/src/store/native_prepare.rs

@@ -1,17 +1,56 @@
 //! Private native object preparation; a data quorum is not publication authority.
 use super::ECStore;
 use crate::bucket::utils::check_put_object_args;
+use crate::disk::DiskAPI;
 use crate::{
-    disk::local::native_prepare::NativePreparationReceipt,
+    disk::local::native_prepare::{
+        NativePreparationDescription, NativePreparationProbe, NativePreparationReceipt, NativePrepareInput,
+    },
     error::{Error, Result},
     object_api::PutObjReader,
 };
+use bytes::Bytes;
 use uuid::Uuid;
 
+/// Identity supplied to a node-local native preparation operation.
+///
+/// `store_id` and `bucket_incarnation` are expectations to validate, not
+/// authorization tokens. The receiving ECStore establishes its own fence.
+#[derive(Clone, Debug)]
+pub struct NativePreparationTarget {
+    pub store_id: Uuid,
+    pub disk_id: Uuid,
+    pub bucket: String,
+    pub key: String,
+    pub preparation: Uuid,
+    pub bucket_incarnation: Uuid,
+}
+
+/// Metadata and content identity for finalizing a shard already streamed to a disk.
+pub struct NativePreparationFinalizeRequest {
+    pub target: NativePreparationTarget,
+    pub version: Uuid,
+    pub object_digest: [u8; 32],
+    pub metadata: Bytes,
+}
+
+/// Verified local recovery data returned by a target owner.
+pub struct NativePreparationRecovery {
+    pub receipt: NativePreparationReceipt,
+    pub metadata: Bytes,
+}
+
 pub struct NativeQuorumPreparation {
     receipts: Vec<NativePreparationReceipt>,
 }
 impl NativeQuorumPreparation {
+    pub(crate) fn into_receipts(self) -> Vec<NativePreparationReceipt> {
+        self.receipts
+    }
+    pub(crate) fn receipts(&self) -> &[NativePreparationReceipt] {
+        &self.receipts
+    }
+
     pub fn preparation_id(&self) -> Uuid {
         self.receipts[0].preparation_id()
     }
@@ -25,7 +64,7 @@ impl NativeQuorumPreparation {
         self.receipts[0].object_length()
     }
     pub fn receipt_count(&self) -> usize {
-        self.receipts.len()
+        self.receipts().len()
     }
     pub fn disk_ids(&self) -> impl Iterator<Item = Uuid> + '_ {
         self.receipts.iter().map(NativePreparationReceipt::disk_id)
@@ -40,13 +79,14 @@ struct Identity {
     distribution: [u8; 32],
 }
 struct Observation {
+    set: std::sync::Arc<crate::set_disk::SetDisks>,
     identity: Option<Identity>,
     metadata: Vec<Option<bytes::Bytes>>,
     prepared: Vec<bool>,
     unknown: bool,
 }
 impl NativeQuorumPreparation {
-    fn from_receipts(quorum: usize, receipts: Vec<NativePreparationReceipt>) -> Result<Self> {
+    pub(crate) fn from_receipts(quorum: usize, receipts: Vec<NativePreparationReceipt>) -> Result<Self> {
         let first = receipts.first().ok_or_else(|| Error::other("empty native quorum"))?;
         let distinct: std::collections::HashSet<_> = receipts.iter().map(NativePreparationReceipt::disk_id).collect();
         if receipts.len() < quorum
@@ -59,6 +99,150 @@ impl NativeQuorumPreparation {
     }
 }
 impl ECStore {
+    /// Locate a durable preparation on the exact local disk named by the caller.
+    /// The description is local locator data and is not a preparation receipt.
+    #[cfg(unix)]
+    pub async fn locate_native_preparation_on_disk(
+        &self,
+        disk_id: Uuid,
+        store_id: Uuid,
+        preparation: Uuid,
+    ) -> Result<Option<NativePreparationDescription>> {
+        if store_id != self.id || store_id.is_nil() || disk_id.is_nil() || preparation.is_nil() {
+            return Err(Error::other("native preparation locator identity mismatch"));
+        }
+        let disk = self.resolve_native_local_disk(disk_id).await?;
+        disk.locate_native_preparation(store_id, preparation)
+            .await
+            .map_err(Into::into)
+    }
+
+    #[cfg(not(unix))]
+    pub async fn locate_native_preparation_on_disk(
+        &self,
+        _disk_id: Uuid,
+        _store_id: Uuid,
+        _preparation: Uuid,
+    ) -> Result<Option<NativePreparationDescription>> {
+        Err(Error::UnsupportedDisk)
+    }
+
+    async fn validate_native_target(
+        &self,
+        target: &NativePreparationTarget,
+    ) -> Result<std::sync::Arc<crate::object_api::ObjectLockConfigSnapshot>> {
+        if target.store_id != self.id
+            || self.id.is_nil()
+            || target.disk_id.is_nil()
+            || target.preparation.is_nil()
+            || target.bucket_incarnation.is_nil()
+        {
+            return Err(Error::other("native preparation target identity mismatch"));
+        }
+        check_put_object_args(&target.bucket, &target.key)?;
+
+        let fence = self.object_lock_config_snapshot(&target.bucket).await?;
+        let incarnation = self.bucket_incarnation_id_from_disk(&target.bucket).await?;
+        if incarnation != target.bucket_incarnation || !fence.is_valid_for_destructive_put(self.id, &target.bucket, incarnation) {
+            return Err(Error::other("native preparation target bucket fence mismatch"));
+        }
+        Ok(fence)
+    }
+
+    async fn resolve_native_local_disk(&self, disk_id: Uuid) -> Result<std::sync::Arc<crate::disk::local::LocalDisk>> {
+        use crate::disk::{Disk, DiskAPI};
+
+        let mut matches = Vec::new();
+        let mut lookup_error = None;
+        for pool in &self.pools {
+            for set in &pool.disk_set {
+                let disks = set.disks.read().await.clone();
+                for disk in disks.into_iter().flatten() {
+                    let Disk::Local(local) = disk.as_ref() else {
+                        continue;
+                    };
+                    let local_disk = local.get_disk();
+                    match local_disk.get_disk_id().await {
+                        Ok(Some(id)) if id == disk_id => matches.push(local_disk),
+                        Ok(_) => {}
+                        Err(error) => {
+                            lookup_error.get_or_insert(error);
+                        }
+                    }
+                }
+            }
+        }
+        if matches.len() != 1 {
+            if matches.is_empty()
+                && let Some(error) = lookup_error
+            {
+                return Err(error.into());
+            }
+            return Err(Error::other("native preparation target disk is not uniquely local"));
+        }
+        matches
+            .pop()
+            .ok_or_else(|| Error::other("native preparation target disk disappeared"))
+    }
+
+    /// Create private stream staging on the exact local disk named by the caller.
+    /// The caller holds the distributed object lock across all disk RPCs; this owner must not
+    /// reacquire that non-reentrant lock. Store and bucket identities are checked against fresh
+    /// local state, while disk operations take their own physical mutation locks.
+    pub async fn begin_native_preparation_on_disk(&self, target: &NativePreparationTarget) -> Result<()> {
+        let fence = self.validate_native_target(target).await?;
+        let disk = self.resolve_native_local_disk(target.disk_id).await?;
+        disk.begin_native_stream(target.preparation, self.id, &target.bucket, target.bucket_incarnation, fence)
+            .await?;
+        Ok(())
+    }
+
+    /// Verify and durably attest the stream already written to the named local disk.
+    /// This does not select placement or encode the object.
+    pub async fn finalize_native_preparation_on_disk(
+        &self,
+        request: NativePreparationFinalizeRequest,
+    ) -> Result<NativePreparationReceipt> {
+        let target = &request.target;
+        let fence = self.validate_native_target(target).await?;
+        let disk = self.resolve_native_local_disk(target.disk_id).await?;
+        disk.prepare_native(NativePrepareInput {
+            preparation: target.preparation,
+            store_id: self.id,
+            bucket: target.bucket.clone(),
+            bucket_incarnation: target.bucket_incarnation,
+            object_digest: request.object_digest,
+            key: target.key.clone(),
+            version: request.version,
+            metadata: request.metadata,
+            bucket_fence: fence,
+        })
+        .await
+        .map_err(Into::into)
+    }
+
+    /// Inspect the named local disk for a preparation without trusting a remote fence.
+    pub async fn probe_native_preparation_on_disk(&self, target: &NativePreparationTarget) -> Result<NativePreparationProbe> {
+        let fence = self.validate_native_target(target).await?;
+        let disk = self.resolve_native_local_disk(target.disk_id).await?;
+        disk.probe_native(target.preparation, self.id, &target.bucket, &target.key, target.bucket_incarnation, fence)
+            .await
+            .map_err(Into::into)
+    }
+
+    /// Recover and revalidate durable preparation state from the named local disk.
+    pub async fn recover_native_preparation_on_disk(
+        &self,
+        target: &NativePreparationTarget,
+    ) -> Result<NativePreparationRecovery> {
+        let fence = self.validate_native_target(target).await?;
+        let disk = self.resolve_native_local_disk(target.disk_id).await?;
+        let (receipt, metadata) = disk
+            .recover_native_with_metadata(target.preparation, self.id, &target.bucket, target.bucket_incarnation, fence)
+            .await?;
+        Ok(NativePreparationRecovery { receipt, metadata })
+    }
+
     async fn native_observations(
         &self,
         bucket: &str,
@@ -66,56 +250,75 @@ impl ECStore {
         preparation: Uuid,
         incarnation: Uuid,
         fence: std::sync::Arc<crate::object_api::ObjectLockConfigSnapshot>,
+    ) -> Result<Vec<Observation>> {
+        self.native_observations_scan(bucket, key, preparation, incarnation, fence, false)
+            .await
+    }
+    async fn native_observations_scan(
+        &self,
+        bucket: &str,
+        key: &str,
+        preparation: Uuid,
+        incarnation: Uuid,
+        fence: std::sync::Arc<crate::object_api::ObjectLockConfigSnapshot>,
+        all_sets: bool,
     ) -> Result<Vec<Observation>> {
         let mut pools = Vec::new();
         for pool in &self.pools {
-            let candidates = pool
-                .get_disks_by_key(key)
-                .probe_native_candidates(self.id, bucket, key, preparation, incarnation, fence.clone())
-                .await;
-            let mut observed = Observation {
-                identity: None,
-                metadata: Vec::new(),
-                prepared: Vec::new(),
-                unknown: false,
+            let sets = if all_sets {
+                pool.disk_set.clone()
+            } else {
+                vec![pool.get_disks_by_key(key)]
             };
-            for candidate in candidates {
-                let mut bytes = None;
-                let mut prepared = false;
-                match candidate {
-                    Ok(crate::disk::local::native_prepare::NativePreparationProbe::Absent) => {}
-                    Ok(crate::disk::local::native_prepare::NativePreparationProbe::Existing {
-                        version,
-                        prepared: ready,
-                        object_digest,
-                        object_length,
-                        set_digest,
-                        distribution_digest,
-                        metadata,
-                    }) => {
-                        let identity = Identity {
+            for set in sets {
+                let candidates = set
+                    .probe_native_candidates(self.id, bucket, key, preparation, incarnation, fence.clone(), all_sets)
+                    .await;
+                let mut observed = Observation {
+                    set,
+                    identity: None,
+                    metadata: Vec::new(),
+                    prepared: Vec::new(),
+                    unknown: false,
+                };
+                for candidate in candidates {
+                    let mut bytes = None;
+                    let mut prepared = false;
+                    match candidate {
+                        Ok(crate::disk::local::native_prepare::NativePreparationProbe::Absent) => {}
+                        Ok(crate::disk::local::native_prepare::NativePreparationProbe::Existing {
                             version,
-                            digest: object_digest,
-                            length: object_length,
-                            set: set_digest,
-                            distribution: distribution_digest,
-                        };
-                        if version != preparation || observed.identity.as_ref().is_some_and(|old| old != &identity) {
-                            return Err(Error::other("conflicting native preparation identity"));
+                            prepared: ready,
+                            object_digest,
+                            object_length,
+                            set_digest,
+                            distribution_digest,
+                            metadata,
+                        }) => {
+                            let identity = Identity {
+                                version,
+                                digest: object_digest,
+                                length: object_length,
+                                set: set_digest,
+                                distribution: distribution_digest,
+                            };
+                            if observed.identity.as_ref().is_some_and(|old| old != &identity) {
+                                return Err(Error::other("conflicting native preparation identity"));
+                            }
+                            observed.identity = Some(identity);
+                            bytes = metadata;
+                            prepared = ready;
                         }
-                        observed.identity = Some(identity);
-                        bytes = metadata;
-                        prepared = ready;
+                        Err(crate::disk::error::Error::Io(error)) if error.kind() == std::io::ErrorKind::InvalidData => {
+                            return Err(Error::from(error));
+                        }
+                        Err(_) => observed.unknown = true,
                     }
-                    Err(crate::disk::error::Error::Io(error)) if error.kind() == std::io::ErrorKind::InvalidData => {
-                        return Err(Error::from(error));
-                    }
-                    Err(_) => observed.unknown = true,
+                    observed.metadata.push(bytes);
+                    observed.prepared.push(prepared);
                 }
-                observed.metadata.push(bytes);
-                observed.prepared.push(prepared);
+                pools.push(observed);
             }
-            pools.push(observed);
         }
         Ok(pools)
     }
@@ -145,9 +348,70 @@ impl ECStore {
         use sha2::{Digest, Sha256};
         use tokio::io::AsyncReadExt;
         check_put_object_args(bucket, key)?;
-        if preparation.is_nil() || self.pools.is_empty() || data.size() < 0 || data.size() > 1024 * 1024 {
+        if preparation.is_nil() || self.pools.is_empty() || data.size() < -1 {
             return Err(Error::other("unsupported native preparation"));
         }
+        let mut remote = false;
+        for pool in &self.pools {
+            if pool
+                .get_disks_by_key(key)
+                .disks
+                .read()
+                .await
+                .iter()
+                .flatten()
+                .any(|disk| !disk.is_local())
+            {
+                remote = true;
+                break;
+            }
+        }
+        if remote || data.size() < 0 || data.size() > 1024 * 1024 {
+            let fence = self.object_lock_config_snapshot(bucket).await?;
+            let incarnation = self.bucket_incarnation_id(bucket).await?;
+            let owner = std::sync::Arc::new(
+                self.handle_new_ns_lock(bucket, key)
+                    .await?
+                    .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+                    .await?,
+            );
+            let observations = self
+                .native_observations_scan(bucket, key, preparation, incarnation, fence.clone(), true)
+                .await?;
+            if let Some(selected) = Self::existing_native_pool(&observations)? {
+                let identity = observations[selected]
+                    .identity
+                    .as_ref()
+                    .ok_or_else(|| Error::other("missing native identity"))?;
+                let mut hash = Sha256::new();
+                let mut length = 0_u64;
+                let mut buffer = vec![0_u8; 64 * 1024];
+                loop {
+                    let size = data.stream.read(&mut buffer).await?;
+                    if size == 0 {
+                        break;
+                    }
+                    length = length
+                        .checked_add(u64::try_from(size).map_err(Error::other)?)
+                        .ok_or_else(|| Error::other("native retry length overflow"))?;
+                    hash.update(&buffer[..size]);
+                }
+                if hash.finalize().as_slice() != identity.digest || length != identity.length {
+                    return Err(Error::other("native streaming retry content conflict"));
+                }
+                drop(owner);
+                return self.recover_native_object(bucket, key, preparation).await;
+            }
+            let selected = if self.pools.len() == 1 {
+                0
+            } else {
+                self.get_pool_idx_no_lock(bucket, key, data.size().max(0)).await?
+            };
+            let set = self.pools[selected].get_disks_by_key(key);
+            let (quorum, receipts) =
+                Box::pin(set.prepare_native_stream(self.id, bucket, key, preparation, incarnation, data, fence, owner)).await?;
+            return NativeQuorumPreparation::from_receipts(quorum, receipts);
+        }
         let length = data.size();
         let mut body = Vec::new();
         (&mut data.stream).take(1024 * 1024 + 1).read_to_end(&mut body).await?;
@@ -200,19 +464,34 @@ impl ECStore {
         if preparation.is_nil() || self.pools.is_empty() {
             return Err(Error::other("invalid native preparation"));
         }
-        let fence = self.object_lock_config_snapshot(bucket).await?;
-        let incarnation = self.bucket_incarnation_id(bucket).await?;
         let _owner = self
             .handle_new_ns_lock(bucket, key)
             .await?
             .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
             .await?;
+        self.recover_native_object_unlocked(bucket, key, preparation)
+            .await
+            .map(|(owner, _)| owner)
+    }
+    /// Caller owns the distributed namespace lock; each disk still acquires its actual physical recovery owners.
+    pub(crate) async fn recover_native_object_unlocked(
+        &self,
+        bucket: &str,
+        key: &str,
+        preparation: Uuid,
+    ) -> Result<(NativeQuorumPreparation, Vec<(Uuid, bytes::Bytes)>)> {
+        check_put_object_args(bucket, key)?;
+        if preparation.is_nil() || self.pools.is_empty() {
+            return Err(Error::other("invalid native preparation"));
+        }
+        let fence = self.object_lock_config_snapshot(bucket).await?;
+        let incarnation = self.bucket_incarnation_id(bucket).await?;
         let observations = self
-            .native_observations(bucket, key, preparation, incarnation, fence.clone())
+            .native_observations_scan(bucket, key, preparation, incarnation, fence.clone(), true)
             .await?;
         let selected = Self::existing_native_pool(&observations)?.ok_or_else(|| Error::other("native preparation absent"))?;
-        let (quorum, receipts) = self.pools[selected]
-            .get_disks_by_key(key)
+        let (quorum, receipts, metadata) = observations[selected]
+            .set
             .recover_native_bytes(
                 self.id,
                 bucket,
@@ -223,6 +502,108 @@ impl ECStore {
                 observations[selected].prepared.clone(),
             )
             .await?;
-        NativeQuorumPreparation::from_receipts(quorum, receipts)
+        Ok((NativeQuorumPreparation::from_receipts(quorum, receipts)?, metadata))
+    }
+}
+
+#[cfg(all(test, unix))]
+mod target_owner_tests {
+    use super::*;
+    use crate::{
+        bucket::metadata_sys::{init_bucket_metadata_sys, test_support::isolated_store_over_temp_disks},
+        disk::DiskAPI,
+        object_api::PutObjReader,
+        storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions},
+    };
+    use uuid::Uuid;
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn target_owner_apis_use_fresh_local_identity_and_exact_disk() {
+        let (dirs, store) = isolated_store_over_temp_disks().await;
+        init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let bucket = "native-target-owner";
+        let key = "object.bin";
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create target bucket");
+
+        let preparation = Uuid::new_v4();
+        let body = b"target-owner facade identity";
+        let prepared = store
+            .prepare_native_object(bucket, key, preparation, &mut PutObjReader::from_vec(body.to_vec()))
+            .await
+            .expect("prepare native object");
+        let disk = store.pools[0].get_disks_by_key(key).disks.read().await[0]
+            .clone()
+            .expect("first local disk");
+        let disk_id = disk.get_disk_id().await.expect("disk identity").expect("formatted disk");
+        assert!(prepared.disk_ids().any(|id| id == disk_id));
+        let target = NativePreparationTarget {
+            store_id: store.id,
+            disk_id,
+            bucket: bucket.into(),
+            key: key.into(),
+            preparation,
+            bucket_incarnation: store
+                .bucket_incarnation_id_from_disk(bucket)
+                .await
+                .expect("durable bucket incarnation"),
+        };
+
+        let metadata_path = dirs[0]
+            .path()
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string())
+            .join("xl.meta");
+        let metadata = Bytes::from(std::fs::read(metadata_path).expect("persisted shard metadata"));
+        let unrelated_format = dirs[1].path().join(".rustfs.sys/format.json");
+        let saved_unrelated_format = std::fs::read(&unrelated_format).expect("unrelated disk format");
+        std::fs::write(&unrelated_format, b"invalid format").expect("damage unrelated disk format");
+        let caller_owner = store
+            .handle_new_ns_lock(bucket, key)
+            .await
+            .expect("caller namespace lock")
+            .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+            .await
+            .expect("caller owns namespace lock");
+        let receipt = store
+            .finalize_native_preparation_on_disk(NativePreparationFinalizeRequest {
+                target: target.clone(),
+                version: preparation,
+                object_digest: prepared.object_digest(),
+                metadata: metadata.clone(),
+            })
+            .await
+            .expect("target-owner idempotent finalize");
+        assert_eq!(receipt.disk_id(), disk_id);
+        assert_eq!(receipt.object_digest(), &prepared.object_digest());
+
+        assert!(matches!(
+            tokio::time::timeout(
+                std::time::Duration::from_secs(2),
+                store.probe_native_preparation_on_disk(&target),
+            )
+            .await
+            .expect("target-owner must not reacquire the caller's non-reentrant lock")
+            .expect("target-owner probe"),
+            NativePreparationProbe::Existing { prepared: true, metadata: Some(found), .. } if found == metadata
+        ));
+        let recovered = store
+            .recover_native_preparation_on_disk(&target)
+            .await
+            .expect("target-owner recovery");
+        assert_eq!(recovered.receipt.disk_id(), disk_id);
+        assert_eq!(recovered.metadata, metadata);
+
+        let mut wrong_incarnation = target.clone();
+        wrong_incarnation.bucket_incarnation = Uuid::new_v4();
+        assert!(store.begin_native_preparation_on_disk(&wrong_incarnation).await.is_err());
+        let mut wrong_disk = target;
+        wrong_disk.disk_id = Uuid::new_v4();
+        assert!(store.begin_native_preparation_on_disk(&wrong_disk).await.is_err());
+        drop(caller_owner);
+        std::fs::write(unrelated_format, saved_unrelated_format).expect("restore unrelated disk format");
     }
 }

crates/ecstore/src/store/native_prepare_tests.rs

@@ -31,6 +31,71 @@ async fn fixture() -> (Vec<TempDir>, Arc<ECStore>) {
     (dirs, store)
 }
 
+async fn multi_pool_fixture() -> (Vec<TempDir>, Arc<ECStore>) {
+    let dirs = (0..8)
+        .map(|_| tempfile::tempdir().expect("real local disk directory"))
+        .collect::<Vec<_>>();
+    let pools = EndpointServerPools(
+        (0..2)
+            .map(|pool_index| {
+                let endpoints = (0..4)
+                    .map(|disk_index| {
+                        let dir = &dirs[pool_index * 4 + disk_index];
+                        let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("disk path")).expect("endpoint");
+                        endpoint.set_pool_index(pool_index);
+                        endpoint.set_set_index(0);
+                        endpoint.set_disk_index(disk_index);
+                        endpoint
+                    })
+                    .collect::<Vec<_>>();
+                PoolEndpoints {
+                    legacy: false,
+                    set_count: 1,
+                    drives_per_set: 4,
+                    endpoints: Endpoints::from(endpoints),
+                    cmd_line: "native-multi-pool-test".into(),
+                    platform: "test".into(),
+                }
+            })
+            .collect(),
+    );
+    let instance = Arc::new(InstanceContext::new());
+    init_local_disks_with_instance_ctx(&instance, pools.clone())
+        .await
+        .expect("initialize both physical pools");
+    let store =
+        ECStore::new_with_instance_ctx("127.0.0.1:0".parse().expect("test address"), pools, CancellationToken::new(), instance)
+            .await
+            .expect("initialize multi-pool store");
+    init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+    store
+        .make_bucket(BUCKET, &MakeBucketOptions::default())
+        .await
+        .expect("create bucket across both pools");
+    (dirs, store)
+}
+
+async fn prepare_stream_in_pool(store: &Arc<ECStore>, pool_index: usize, preparation: Uuid, body: Vec<u8>) {
+    let incarnation = store.bucket_incarnation_id(BUCKET).await.expect("bucket incarnation");
+    let fence = store.object_lock_config_snapshot(BUCKET).await.expect("bucket fence");
+    let owner = Arc::new(
+        store
+            .handle_new_ns_lock(BUCKET, KEY)
+            .await
+            .expect("object namespace lock")
+            .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+            .await
+            .expect("object write lock"),
+    );
+    let mut reader = PutObjReader::from_vec(body);
+    let (quorum, receipts) = store.pools[pool_index]
+        .get_disks_by_key(KEY)
+        .prepare_native_stream(store.id, BUCKET, KEY, preparation, incarnation, &mut reader, fence, owner)
+        .await
+        .expect("prepare on the specified physical pool");
+    assert!(receipts.len() >= quorum, "NATIVE_CROSS_POOL_PREPARE_REQUIRES_WRITE_QUORUM");
+}
+
 async fn assert_not_published(store: &ECStore) {
     let result = store.get_object_info(BUCKET, KEY, &ObjectOptions::default()).await;
     assert!(
@@ -428,3 +493,489 @@ async fn native_object_prepare_nonzero_set_reopens_the_routed_identity() {
         "NP02_OPERATION_IDENTITY_BOUND: select actual key/set facts"
     );
 }
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_large_restart() {
+    let (dirs, store) = fixture().await;
+    let body = (0..(5 * 1024 * 1024 + 37))
+        .map(|index| u8::try_from(index % 251).expect("bounded byte"))
+        .collect::<Vec<_>>();
+    let expected: [u8; 32] = Sha256::digest(&body).into();
+    let preparation = Uuid::new_v4();
+    let prepared = store
+        .prepare_native_object(BUCKET, KEY, preparation, &mut PutObjReader::from_vec(body.clone()))
+        .await
+        .expect("actual multi-block streaming preparation");
+    assert_eq!(prepared.object_digest(), expected);
+    assert_eq!(prepared.object_length(), u64::try_from(body.len()).expect("bounded content length"));
+    assert!(prepared.receipt_count() >= 3);
+    assert_not_published(&store).await;
+    let mut records = Vec::new();
+    for disk in &dirs {
+        let path = disk
+            .path()
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string());
+        let metadata = std::fs::read(path.join("xl.meta")).expect("exact streaming metadata");
+        let record: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(path.join("intent.json")).expect("durable locator"))
+                .expect("strict streaming intent JSON");
+        assert_eq!(record["format"], 3);
+        let disk_format: crate::layout::format::FormatV3 =
+            serde_json::from_slice(&std::fs::read(disk.path().join(".rustfs.sys/format.json")).expect("actual disk format"))
+                .expect("typed disk format");
+        let disk_id: Uuid = serde_json::from_value(record["disk"].clone()).expect("record disk identity");
+        let expected_set = disk_format
+            .erasure
+            .sets
+            .iter()
+            .find(|set| set.contains(&disk_id))
+            .expect("disk belongs to one ordered set");
+        let saved_set: Vec<Uuid> = serde_json::from_value(record["set_disks"].clone()).expect("persistent disk set");
+        assert_eq!(&saved_set, expected_set, "persist the full ordered physical set");
+        assert_eq!(record["data_dir"], preparation.to_string());
+        assert!(
+            std::fs::metadata(path.join(preparation.to_string()).join("part.1"))
+                .expect("real non-inline shard")
+                .len()
+                > 0
+        );
+        records.push((path, metadata));
+    }
+    drop(prepared);
+    drop(store);
+    let restarted = reopen(&dirs).await;
+    let recovered = restarted
+        .recover_native_object(BUCKET, KEY, preparation)
+        .await
+        .expect("actual streaming reopen recovery");
+    assert_eq!(recovered.object_digest(), expected);
+    assert_eq!(recovered.object_length(), u64::try_from(body.len()).expect("length"));
+    assert_eq!(recovered.version_id(), preparation);
+    for (path, bytes) in records {
+        assert_eq!(std::fs::read(path.join("xl.meta")).expect("preserved metadata"), bytes);
+    }
+    assert_not_published(&restarted).await;
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_locator_conflict() {
+    let (dirs, store) = fixture().await;
+    let preparation = Uuid::new_v4();
+    let body = vec![17; 2 * 1024 * 1024 + 1];
+    drop(
+        store
+            .prepare_native_object(BUCKET, KEY, preparation, &mut PutObjReader::from_vec(body))
+            .await
+            .expect("real durable streaming preparation"),
+    );
+    for disk in &dirs {
+        let path = disk
+            .path()
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string());
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(path.join("intent.json")).expect("record")).expect("JSON");
+        record["set_disks"]
+            .as_array_mut()
+            .expect("saved physical disk order")
+            .swap(0, 1);
+        // Preserve the actual record serialization order; receipt marker signs exact canonical record bytes.
+        let bytes = crate::disk::local::native_prepare::canonical_test_record(&record);
+        std::fs::write(path.join("intent.json"), &bytes).expect("install conflicting durable locator");
+        std::fs::write(path.join("PREPARED"), Sha256::digest(&bytes)).expect("matching record marker");
+    }
+    assert!(
+        store.recover_native_object(BUCKET, KEY, preparation).await.is_err(),
+        "NATIVE_STREAM_REQUIRES_SAVED_PHYSICAL_SET"
+    );
+    assert_not_published(&store).await;
+}
+
+#[test]
+#[serial_test::serial]
+fn native_stream_prepare_locator_conflict_across_pools_and_unknown_disk() {
+    std::thread::Builder::new()
+        .stack_size(32 * 1024 * 1024)
+        .spawn(|| {
+            let runtime = tokio::runtime::Builder::new_multi_thread()
+                .enable_all()
+                .worker_threads(2)
+                .thread_stack_size(32 * 1024 * 1024)
+                .build()
+                .expect("large-stack test runtime");
+            runtime.block_on(async {
+                use crate::disk::DiskAPI as _;
+
+                let (dirs, store) = multi_pool_fixture().await;
+                let preparation = Uuid::new_v4();
+                prepare_stream_in_pool(&store, 0, preparation, vec![41; 2 * 1024 * 1024 + 3]).await;
+                prepare_stream_in_pool(&store, 1, preparation, vec![42; 2 * 1024 * 1024 + 3]).await;
+
+                let pool_zero: serde_json::Value = serde_json::from_slice(
+                    &std::fs::read(
+                        dirs[0]
+                            .path()
+                            .join(".rustfs.sys/native-preparations")
+                            .join(preparation.to_string())
+                            .join("intent.json"),
+                    )
+                    .expect("pool zero durable locator"),
+                )
+                .expect("pool zero locator JSON");
+                let pool_one: serde_json::Value = serde_json::from_slice(
+                    &std::fs::read(
+                        dirs[4]
+                            .path()
+                            .join(".rustfs.sys/native-preparations")
+                            .join(preparation.to_string())
+                            .join("intent.json"),
+                    )
+                    .expect("pool one durable locator"),
+                )
+                .expect("pool one locator JSON");
+                assert_ne!(
+                    pool_zero["object_digest"], pool_one["object_digest"],
+                    "NATIVE_CROSS_POOL_CONFLICT_FIXTURE_MUST_HAVE_DIFFERENT_CONTENT"
+                );
+                assert!(
+                    store.recover_native_object(BUCKET, KEY, preparation).await.is_err(),
+                    "NATIVE_STREAM_MUST_REJECT_PREPARATION_IN_MULTIPLE_POOLS"
+                );
+                assert_not_published(&store).await;
+
+                let unknown_preparation = Uuid::new_v4();
+                prepare_stream_in_pool(&store, 0, unknown_preparation, vec![51; 2 * 1024 * 1024 + 5]).await;
+                let unknown_disk = store.pools[0].get_disks_by_key(KEY).disks.read().await[0]
+                    .as_ref()
+                    .expect("pool zero physical disk")
+                    .get_disk_id()
+                    .await
+                    .expect("read pool zero physical identity")
+                    .expect("known pool zero physical identity");
+                let disk_root = dirs
+                    .iter()
+                    .find(|dir| {
+                        serde_json::from_slice::<crate::layout::format::FormatV3>(
+                            &std::fs::read(dir.path().join(".rustfs.sys/format.json")).expect("read physical format"),
+                        )
+                        .is_ok_and(|format| format.erasure.this == unknown_disk)
+                    })
+                    .expect("resolve pool zero physical root");
+                let prep_dir = disk_root
+                    .path()
+                    .join(".rustfs.sys/native-preparations")
+                    .join(unknown_preparation.to_string());
+                let record_path = prep_dir.join("intent.json");
+                let mut record: serde_json::Value =
+                    serde_json::from_slice(&std::fs::read(&record_path).expect("durable locator")).expect("locator JSON");
+                let unknown_identity = Uuid::new_v4();
+                assert_ne!(unknown_disk, unknown_identity);
+                record["disk"] = serde_json::Value::String(unknown_identity.to_string());
+                let bytes = crate::disk::local::native_prepare::canonical_test_record(&record);
+                std::fs::write(&record_path, &bytes).expect("install unknown disk identity");
+                std::fs::write(prep_dir.join("PREPARED"), Sha256::digest(&bytes))
+                    .expect("retain valid marker for unknown identity");
+
+                assert!(
+                    store.recover_native_object(BUCKET, KEY, unknown_preparation).await.is_err(),
+                    "NATIVE_STREAM_MUST_FAIL_CLOSED_ON_UNKNOWN_DISK_IDENTITY"
+                );
+                assert!(
+                    store
+                        .prepare_native_object(
+                            BUCKET,
+                            KEY,
+                            unknown_preparation,
+                            &mut PutObjReader::from_vec(vec![51; 2 * 1024 * 1024 + 5]),
+                        )
+                        .await
+                        .is_err(),
+                    "NATIVE_STREAM_MUST_NOT_REALLOCATE_WHEN_DISK_IDENTITY_IS_UNKNOWN"
+                );
+                assert!(
+                    dirs[4..].iter().all(|dir| !dir
+                        .path()
+                        .join(".rustfs.sys/native-preparations")
+                        .join(unknown_preparation.to_string())
+                        .exists()),
+                    "NATIVE_STREAM_UNKNOWN_DISK_MUST_NOT_RECEIVE_REASSIGNED_PREPARATION"
+                );
+                assert_not_published(&store).await;
+            });
+        })
+        .expect("spawn large-stack native preparation test")
+        .join()
+        .expect("native preparation test thread");
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_crash_barriers() {
+    use crate::disk::local::native_prepare::{clear_native_preparation_probe, fail_native_preparation};
+    for (stage, after, recoverable) in [
+        ("part.1:file", false, false),
+        ("intent.json:file", true, false),
+        ("PREPARED:directory", true, true),
+    ] {
+        let (dirs, store) = fixture().await;
+        let preparation = Uuid::new_v4();
+        fail_native_preparation(preparation, stage, after);
+        let result = store
+            .prepare_native_object(BUCKET, KEY, preparation, &mut PutObjReader::from_vec(vec![23; 2 * 1024 * 1024 + 3]))
+            .await;
+        assert!(result.is_err(), "injected disk barrier must not return a receipt: {stage}");
+        clear_native_preparation_probe(preparation);
+        assert_not_published(&store).await;
+        drop(store);
+        let restarted = reopen(&dirs).await;
+        let recovered = restarted.recover_native_object(BUCKET, KEY, preparation).await;
+        assert_eq!(recovered.is_ok(), recoverable, "durable boundary: {stage}");
+        assert_not_published(&restarted).await;
+    }
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_multi_part_restart() {
+    use crate::disk::Disk;
+    use rustfs_filemeta::FileInfoOpts;
+    let (dirs, store) = fixture().await;
+    let body = vec![29; 3 * 1024 * 1024 + 13];
+    let encoded_preparation = Uuid::new_v4();
+    drop(
+        store
+            .prepare_native_object(BUCKET, KEY, encoded_preparation, &mut PutObjReader::from_vec(body.clone()))
+            .await
+            .expect("existing ordinary streaming encoder fixture"),
+    );
+    let preparation = Uuid::new_v4();
+    let incarnation = store.bucket_incarnation_id(BUCKET).await.expect("actual incarnation");
+    let fence = store.object_lock_config_snapshot(BUCKET).await.expect("actual bucket fence");
+    let set = store.pools[0].get_disks_by_key(KEY);
+    let disks = set.disks.read().await.clone();
+    let guard = store
+        .handle_new_ns_lock(BUCKET, KEY)
+        .await
+        .expect("actual commit namespace")
+        .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await
+        .expect("owned commit lock");
+    let mut candidates = Vec::new();
+    for (dir, disk) in dirs.iter().zip(disks) {
+        let original = dir
+            .path()
+            .join(".rustfs.sys/native-preparations")
+            .join(encoded_preparation.to_string());
+        let metadata = std::fs::read(original.join("xl.meta")).expect("encoder metadata");
+        let mut info = rustfs_filemeta::get_file_info(
+            &metadata,
+            BUCKET,
+            KEY,
+            &encoded_preparation.to_string(),
+            FileInfoOpts {
+                data: false,
+                include_free_versions: false,
+                include_part_checksums: true,
+            },
+        )
+        .expect("exact FileInfo");
+        let first = info.erasure.block_size;
+        let codec = crate::erasure::coding::Erasure::try_new_with_options(
+            info.erasure.data_blocks,
+            info.erasure.parity_blocks,
+            first,
+            info.uses_legacy_checksum,
+        )
+        .expect("existing erasure geometry");
+        let split = crate::erasure::coding::bitrot_shard_file_size(
+            usize::try_from(codec.shard_file_size(i64::try_from(first).expect("part size"))).expect("shard size"),
+            codec.shard_size(),
+            info.erasure.get_checksum_info(1).algorithm,
+        );
+        let shard = std::fs::read(original.join(encoded_preparation.to_string()).join("part.1")).expect("actual encoded shard");
+        let disk = disk.expect("actual disk");
+        let Disk::Local(local) = disk.as_ref() else {
+            panic!("local fixture");
+        };
+        local
+            .get_disk()
+            .begin_native_stream(preparation, store.id, BUCKET, incarnation, fence.clone())
+            .await
+            .expect("actual private owner");
+        let destination = dir
+            .path()
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string())
+            .join(preparation.to_string());
+        std::fs::create_dir(&destination).expect("private part fixture directory");
+        std::fs::write(destination.join("part.1"), &shard[..split]).expect("first actual encoded blocks");
+        std::fs::write(destination.join("part.2"), &shard[split..]).expect("remaining actual encoded blocks");
+        info.version_id = Some(preparation);
+        info.data_dir = Some(preparation);
+        info.parts.clear();
+        info.add_object_part(1, String::new(), first, None, i64::try_from(first).expect("size"), None, None);
+        info.add_object_part(
+            2,
+            String::new(),
+            body.len() - first,
+            None,
+            i64::try_from(body.len() - first).expect("size"),
+            None,
+            None,
+        );
+        candidates.push((disk.clone(), info));
+    }
+    let owner = set
+        .attest_native_stream(
+            store.id,
+            BUCKET,
+            KEY,
+            preparation,
+            incarnation,
+            candidates,
+            Sha256::digest(&body).into(),
+            u64::try_from(body.len()).expect("length"),
+            fence,
+        )
+        .await
+        .expect("actual adoption verifies/fsyncs both parts without rereading upload or taking namespace");
+    assert!(owner.receipt_count() >= 3);
+    drop(owner);
+    drop(guard);
+    drop(set);
+    drop(store);
+    let restarted = reopen(&dirs).await;
+    let recovered = restarted
+        .recover_native_object(BUCKET, KEY, preparation)
+        .await
+        .expect("actual multi-part recovery decode");
+    assert_eq!(recovered.object_digest(), <[u8; 32]>::from(Sha256::digest(&body)));
+    assert_eq!(recovered.object_length(), u64::try_from(body.len()).expect("length"));
+    assert_not_published(&restarted).await;
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_rejects_overwrite() {
+    use crate::disk::{DiskAPI, RUSTFS_META_BUCKET};
+    let (dirs, store) = fixture().await;
+    let preparation = Uuid::new_v4();
+    drop(
+        store
+            .prepare_native_object(BUCKET, KEY, preparation, &mut PutObjReader::from_vec(vec![31; 2 * 1024 * 1024 + 11]))
+            .await
+            .expect("attested private stream"),
+    );
+    let disk = store.pools[0].get_disks_by_key(KEY).disks.read().await[0]
+        .clone()
+        .expect("actual disk");
+    let path = format!("native-preparations/{preparation}/{preparation}/part.1");
+    let full = dirs[0].path().join(RUSTFS_META_BUCKET).join(&path);
+    let before = std::fs::read(&full).expect("attested shard bytes");
+    assert!(
+        disk.create_file("", RUSTFS_META_BUCKET, &path, 1).await.is_err(),
+        "NATIVE_STREAM_ATTESTED_PART_MUST_NOT_BE_TRUNCATED"
+    );
+    assert_eq!(std::fs::read(&full).expect("retained shard"), before);
+    assert!(store.recover_native_object(BUCKET, KEY, preparation).await.is_ok());
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn native_stream_prepare_record_locator() {
+    use crate::disk::Disk;
+    let (dirs, store) = fixture().await;
+    let preparation = Uuid::new_v4();
+    drop(
+        store
+            .prepare_native_object(BUCKET, KEY, preparation, &mut PutObjReader::from_vec(vec![37; 2 * 1024 * 1024 + 9]))
+            .await
+            .expect("actual prepared locator"),
+    );
+    let disk = store.pools[0].get_disks_by_key(KEY).disks.read().await[0]
+        .clone()
+        .expect("actual disk");
+    let Disk::Local(local) = disk.as_ref() else {
+        panic!("local fixture");
+    };
+    let listed = local
+        .get_disk()
+        .list_native_preparations(store.id, None, 1)
+        .await
+        .expect("descriptor-relative list");
+    assert_eq!(listed.records.len(), 1);
+    assert!(listed.unfinished.is_empty());
+    assert_eq!(listed.next, None);
+    let record = &listed.records[0];
+    assert_eq!(record.preparation, preparation);
+    assert_eq!(record.bucket, BUCKET);
+    assert_eq!(record.key, KEY);
+    assert_eq!(record.version, preparation);
+    assert!(record.prepared);
+    let path = dirs[0]
+        .path()
+        .join(".rustfs.sys/native-preparations")
+        .join(preparation.to_string());
+    assert_eq!(
+        record.metadata.as_ref(),
+        std::fs::read(path.join("xl.meta")).expect("retained exact metadata")
+    );
+    let direct = local
+        .get_disk()
+        .locate_native_preparation(store.id, preparation)
+        .await
+        .expect("direct locator")
+        .expect("present record");
+    assert_eq!(direct.metadata, record.metadata);
+    assert!(
+        local
+            .get_disk()
+            .locate_native_preparation(store.id, Uuid::new_v4())
+            .await
+            .expect("healthy absence")
+            .is_none()
+    );
+    assert!(local.get_disk().list_native_preparations(store.id, None, 65).await.is_err());
+    // Sender-owned namespace guard must not be acquired again by the recovery receiver.
+    let guard = store
+        .handle_new_ns_lock(BUCKET, KEY)
+        .await
+        .expect("namespace")
+        .get_write_lock(crate::set_disk::get_lock_acquire_timeout())
+        .await
+        .expect("actual sender lock");
+    let (owner, metadata) = tokio::time::timeout(
+        std::time::Duration::from_secs(10),
+        store.recover_native_object_unlocked(BUCKET, KEY, preparation),
+    )
+    .await
+    .expect("no recursive namespace lock")
+    .expect("actual disk recovery under sender lock");
+    assert!(metadata.iter().any(|(id, bytes)| {
+        owner
+            .receipts()
+            .iter()
+            .any(|receipt| receipt.disk_id() == *id && receipt.metadata_digest() == &<[u8; 32]>::from(Sha256::digest(bytes)))
+    }));
+    drop(guard);
+    // Moving a published part must not erase its durable location facts; locate grants no readiness.
+    std::fs::rename(
+        path.join(preparation.to_string()).join("part.1"),
+        dirs[0].path().join("published-part-test"),
+    )
+    .expect("publication progress fixture");
+    let located = local
+        .get_disk()
+        .locate_native_preparation(store.id, preparation)
+        .await
+        .expect("retained locator after part move")
+        .expect("retained record");
+    assert_eq!(located.metadata, record.metadata);
+    assert!(
+        store.recover_native_object(BUCKET, KEY, preparation).await.is_err(),
+        "prepared marker is not publication authority"
+    );
+}

crates/protos/src/generated/proto_gen/node_service.rs

@@ -156,6 +156,217 @@ pub struct CompareAndUpdateFileResponse {
     #[prost(message, optional, tag = "3")]
     pub error: ::core::option::Option<Error>,
 }
+/// Native preparation carries only bounded object metadata over the control RPC.
+/// Shard bytes use the authenticated data stream and remain private until finalized.
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationReceipt {
+    #[prost(bytes = "bytes", tag = "1")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub deployment_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "5")]
+    pub version_id: ::prost::bytes::Bytes,
+    #[prost(string, tag = "6")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "7")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "8")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "9")]
+    pub object_digest: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "10")]
+    pub object_length: u64,
+    #[prost(bytes = "bytes", tag = "11")]
+    pub set_digest: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "12")]
+    pub distribution_digest: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "13")]
+    pub metadata_digest: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "14")]
+    pub shard_digest: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "15")]
+    pub shard_length: u64,
+    #[prost(uint32, tag = "16")]
+    pub data_blocks: u32,
+    #[prost(uint32, tag = "17")]
+    pub parity_blocks: u32,
+    #[prost(uint32, tag = "18")]
+    pub shard_index: u32,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationBeginRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+    #[prost(string, tag = "5")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "6")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationBeginResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(message, optional, tag = "2")]
+    pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationFinalizeRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+    #[prost(string, tag = "5")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "6")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "8")]
+    pub object_digest: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "9")]
+    pub object_length: u64,
+    #[prost(bytes = "bytes", tag = "10")]
+    pub version_id: ::prost::bytes::Bytes,
+    /// Exact serialized FileMeta; maximum 1 MiB. Peers reject larger values.
+    #[prost(bytes = "bytes", tag = "11")]
+    pub metadata: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationFinalizeResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(message, optional, tag = "2")]
+    pub receipt: ::core::option::Option<NativePreparationReceipt>,
+    #[prost(message, optional, tag = "3")]
+    pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationProbeRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+    #[prost(string, tag = "5")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "6")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationProbeResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bool, tag = "2")]
+    pub exists: bool,
+    #[prost(bool, tag = "3")]
+    pub prepared: bool,
+    /// Exact serialized FileMeta when it exists and can be read; maximum 1 MiB.
+    #[prost(bytes = "bytes", tag = "5")]
+    pub metadata: ::prost::bytes::Bytes,
+    #[prost(message, optional, tag = "6")]
+    pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub response_proof: ::prost::bytes::Bytes,
+    /// Observation only; this is not a durability receipt.
+    #[prost(bytes = "bytes", tag = "8")]
+    pub version_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "9")]
+    pub object_digest: ::prost::bytes::Bytes,
+    #[prost(uint64, tag = "10")]
+    pub object_length: u64,
+    #[prost(bytes = "bytes", tag = "11")]
+    pub set_digest: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "12")]
+    pub distribution_digest: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationRecoverRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+    #[prost(string, tag = "5")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "6")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "7")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationRecoverResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(message, optional, tag = "2")]
+    pub receipt: ::core::option::Option<NativePreparationReceipt>,
+    /// Exact persisted FileMeta; maximum 1 MiB. Peers reject larger values.
+    #[prost(bytes = "bytes", tag = "3")]
+    pub metadata: ::prost::bytes::Bytes,
+    #[prost(message, optional, tag = "4")]
+    pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "bytes", tag = "5")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationLocateRequest {
+    #[prost(string, tag = "1")]
+    pub disk: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "2")]
+    pub disk_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "3")]
+    pub store_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "4")]
+    pub preparation_id: ::prost::bytes::Bytes,
+}
+#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
+pub struct NativePreparationLocateResponse {
+    #[prost(bool, tag = "1")]
+    pub success: bool,
+    #[prost(bool, tag = "2")]
+    pub found: bool,
+    #[prost(string, tag = "3")]
+    pub bucket: ::prost::alloc::string::String,
+    #[prost(string, tag = "4")]
+    pub key: ::prost::alloc::string::String,
+    #[prost(bytes = "bytes", tag = "5")]
+    pub bucket_incarnation_id: ::prost::bytes::Bytes,
+    #[prost(bytes = "bytes", tag = "6")]
+    pub metadata: ::prost::bytes::Bytes,
+    #[prost(bool, tag = "7")]
+    pub prepared: bool,
+    #[prost(message, optional, tag = "8")]
+    pub error: ::core::option::Option<Error>,
+    #[prost(bytes = "bytes", tag = "9")]
+    pub response_proof: ::prost::bytes::Bytes,
+}
 #[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
 pub struct DeleteRequest {
     /// indicate which one in the disks
@@ -2065,6 +2276,81 @@ pub mod node_service_client {
                 .insert(GrpcMethod::new("node_service.NodeService", "CompareAndUpdateFile"));
             self.inner.unary(req, path, codec).await
         }
+        pub async fn native_preparation_begin(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NativePreparationBeginRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationBeginResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/NativePreparationBegin");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "NativePreparationBegin"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn native_preparation_finalize(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NativePreparationFinalizeRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationFinalizeResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/NativePreparationFinalize");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "NativePreparationFinalize"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn native_preparation_probe(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NativePreparationProbeRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationProbeResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/NativePreparationProbe");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "NativePreparationProbe"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn native_preparation_recover(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NativePreparationRecoverRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationRecoverResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/NativePreparationRecover");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "NativePreparationRecover"));
+            self.inner.unary(req, path, codec).await
+        }
+        pub async fn native_preparation_locate(
+            &mut self,
+            request: impl tonic::IntoRequest<super::NativePreparationLocateRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationLocateResponse>, tonic::Status> {
+            self.inner
+                .ready()
+                .await
+                .map_err(|e| tonic::Status::unknown(format!("Service was not ready: {}", e.into())))?;
+            let codec = tonic_prost::ProstCodec::default();
+            let path = http::uri::PathAndQuery::from_static("/node_service.NodeService/NativePreparationLocate");
+            let mut req = request.into_request();
+            req.extensions_mut()
+                .insert(GrpcMethod::new("node_service.NodeService", "NativePreparationLocate"));
+            self.inner.unary(req, path, codec).await
+        }
         pub async fn delete(
             &mut self,
             request: impl tonic::IntoRequest<super::DeleteRequest>,
@@ -3468,6 +3754,26 @@ pub mod node_service_server {
             &self,
             request: tonic::Request<super::CompareAndUpdateFileRequest>,
         ) -> std::result::Result<tonic::Response<super::CompareAndUpdateFileResponse>, tonic::Status>;
+        async fn native_preparation_begin(
+            &self,
+            request: tonic::Request<super::NativePreparationBeginRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationBeginResponse>, tonic::Status>;
+        async fn native_preparation_finalize(
+            &self,
+            request: tonic::Request<super::NativePreparationFinalizeRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationFinalizeResponse>, tonic::Status>;
+        async fn native_preparation_probe(
+            &self,
+            request: tonic::Request<super::NativePreparationProbeRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationProbeResponse>, tonic::Status>;
+        async fn native_preparation_recover(
+            &self,
+            request: tonic::Request<super::NativePreparationRecoverRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationRecoverResponse>, tonic::Status>;
+        async fn native_preparation_locate(
+            &self,
+            request: tonic::Request<super::NativePreparationLocateRequest>,
+        ) -> std::result::Result<tonic::Response<super::NativePreparationLocateResponse>, tonic::Status>;
         async fn delete(
             &self,
             request: tonic::Request<super::DeleteRequest>,
@@ -4194,6 +4500,146 @@ pub mod node_service_server {
                     };
                     Box::pin(fut)
                 }
+                "/node_service.NodeService/NativePreparationBegin" => {
+                    #[allow(non_camel_case_types)]
+                    struct NativePreparationBeginSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::NativePreparationBeginRequest> for NativePreparationBeginSvc<T> {
+                        type Response = super::NativePreparationBeginResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::NativePreparationBeginRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::native_preparation_begin(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NativePreparationBeginSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/NativePreparationFinalize" => {
+                    #[allow(non_camel_case_types)]
+                    struct NativePreparationFinalizeSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::NativePreparationFinalizeRequest> for NativePreparationFinalizeSvc<T> {
+                        type Response = super::NativePreparationFinalizeResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::NativePreparationFinalizeRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::native_preparation_finalize(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NativePreparationFinalizeSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/NativePreparationProbe" => {
+                    #[allow(non_camel_case_types)]
+                    struct NativePreparationProbeSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::NativePreparationProbeRequest> for NativePreparationProbeSvc<T> {
+                        type Response = super::NativePreparationProbeResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::NativePreparationProbeRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::native_preparation_probe(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NativePreparationProbeSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/NativePreparationRecover" => {
+                    #[allow(non_camel_case_types)]
+                    struct NativePreparationRecoverSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::NativePreparationRecoverRequest> for NativePreparationRecoverSvc<T> {
+                        type Response = super::NativePreparationRecoverResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::NativePreparationRecoverRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::native_preparation_recover(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NativePreparationRecoverSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
+                "/node_service.NodeService/NativePreparationLocate" => {
+                    #[allow(non_camel_case_types)]
+                    struct NativePreparationLocateSvc<T: NodeService>(pub Arc<T>);
+                    impl<T: NodeService> tonic::server::UnaryService<super::NativePreparationLocateRequest> for NativePreparationLocateSvc<T> {
+                        type Response = super::NativePreparationLocateResponse;
+                        type Future = BoxFuture<tonic::Response<Self::Response>, tonic::Status>;
+                        fn call(&mut self, request: tonic::Request<super::NativePreparationLocateRequest>) -> Self::Future {
+                            let inner = Arc::clone(&self.0);
+                            let fut = async move { <T as NodeService>::native_preparation_locate(&inner, request).await };
+                            Box::pin(fut)
+                        }
+                    }
+                    let accept_compression_encodings = self.accept_compression_encodings;
+                    let send_compression_encodings = self.send_compression_encodings;
+                    let max_decoding_message_size = self.max_decoding_message_size;
+                    let max_encoding_message_size = self.max_encoding_message_size;
+                    let inner = self.inner.clone();
+                    let fut = async move {
+                        let method = NativePreparationLocateSvc(inner);
+                        let codec = tonic_prost::ProstCodec::default();
+                        let mut grpc = tonic::server::Grpc::new(codec)
+                            .apply_compression_config(accept_compression_encodings, send_compression_encodings)
+                            .apply_max_message_size_config(max_decoding_message_size, max_encoding_message_size);
+                        let res = grpc.unary(method, req).await;
+                        Ok(res)
+                    };
+                    Box::pin(fut)
+                }
                 "/node_service.NodeService/Delete" => {
                     #[allow(non_camel_case_types)]
                     struct DeleteSvc<T: NodeService>(pub Arc<T>);

crates/protos/src/lib.rs

@@ -1099,6 +1099,198 @@ pub fn canonical_write_all_request_body(
     Ok(body.finish())
 }
 
+pub const NATIVE_PREPARATION_METADATA_MAX_BYTES: usize = 1024 * 1024;
+
+pub fn canonical_native_preparation_begin_request_body(
+    request: &proto_gen::node_service::NativePreparationBeginRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-begin-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_bytes(&request.disk_id)?;
+    body.push_bytes(&request.store_id)?;
+    body.push_bytes(&request.preparation_id)?;
+    body.push_str(&request.bucket)?;
+    body.push_str(&request.key)?;
+    body.push_bytes(&request.bucket_incarnation_id)?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_finalize_request_body(
+    request: &proto_gen::node_service::NativePreparationFinalizeRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-finalize-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_bytes(&request.disk_id)?;
+    body.push_bytes(&request.store_id)?;
+    body.push_bytes(&request.preparation_id)?;
+    body.push_str(&request.bucket)?;
+    body.push_str(&request.key)?;
+    body.push_bytes(&request.bucket_incarnation_id)?;
+    body.push_bytes(&request.object_digest)?;
+    body.push_u64(request.object_length);
+    body.push_bytes(&request.version_id)?;
+    body.push_bytes(&request.metadata)?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_probe_request_body(
+    request: &proto_gen::node_service::NativePreparationProbeRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-probe-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_bytes(&request.disk_id)?;
+    body.push_bytes(&request.store_id)?;
+    body.push_bytes(&request.preparation_id)?;
+    body.push_str(&request.bucket)?;
+    body.push_str(&request.key)?;
+    body.push_bytes(&request.bucket_incarnation_id)?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_recover_request_body(
+    request: &proto_gen::node_service::NativePreparationRecoverRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-recover-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_bytes(&request.disk_id)?;
+    body.push_bytes(&request.store_id)?;
+    body.push_bytes(&request.preparation_id)?;
+    body.push_str(&request.bucket)?;
+    body.push_str(&request.key)?;
+    body.push_bytes(&request.bucket_incarnation_id)?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_locate_request_body(
+    request: &proto_gen::node_service::NativePreparationLocateRequest,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-locate-request-v1\0");
+    body.push_str(&request.disk)?;
+    body.push_bytes(&request.disk_id)?;
+    body.push_bytes(&request.store_id)?;
+    body.push_bytes(&request.preparation_id)?;
+    Ok(body.finish())
+}
+
+fn push_native_preparation_error(
+    body: &mut CanonicalBodyBuilder,
+    error: Option<&proto_gen::node_service::Error>,
+) -> Result<(), std::num::TryFromIntError> {
+    body.push_bool(error.is_some());
+    if let Some(error) = error {
+        body.push_u32(error.code);
+        body.push_str(&error.error_info)?;
+    }
+    Ok(())
+}
+
+fn push_native_preparation_receipt(
+    body: &mut CanonicalBodyBuilder,
+    receipt: Option<&proto_gen::node_service::NativePreparationReceipt>,
+) -> Result<(), std::num::TryFromIntError> {
+    body.push_bool(receipt.is_some());
+    let Some(receipt) = receipt else {
+        return Ok(());
+    };
+
+    body.push_bytes(&receipt.disk_id)?;
+    body.push_bytes(&receipt.store_id)?;
+    body.push_bytes(&receipt.deployment_id)?;
+    body.push_bytes(&receipt.preparation_id)?;
+    body.push_bytes(&receipt.version_id)?;
+    body.push_str(&receipt.bucket)?;
+    body.push_str(&receipt.key)?;
+    body.push_bytes(&receipt.bucket_incarnation_id)?;
+    body.push_bytes(&receipt.object_digest)?;
+    body.push_u64(receipt.object_length);
+    body.push_bytes(&receipt.set_digest)?;
+    body.push_bytes(&receipt.distribution_digest)?;
+    body.push_bytes(&receipt.metadata_digest)?;
+    body.push_bytes(&receipt.shard_digest)?;
+    body.push_u64(receipt.shard_length);
+    body.push_u32(receipt.data_blocks);
+    body.push_u32(receipt.parity_blocks);
+    body.push_u32(receipt.shard_index);
+    Ok(())
+}
+
+pub fn canonical_native_preparation_begin_response_body(
+    request: &proto_gen::node_service::NativePreparationBeginRequest,
+    response: &proto_gen::node_service::NativePreparationBeginResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let request_body = canonical_native_preparation_begin_request_body(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-begin-response-v1\0");
+    body.push_bytes(&request_body)?;
+    body.push_bool(response.success);
+    push_native_preparation_error(&mut body, response.error.as_ref())?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_finalize_response_body(
+    request: &proto_gen::node_service::NativePreparationFinalizeRequest,
+    response: &proto_gen::node_service::NativePreparationFinalizeResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let request_body = canonical_native_preparation_finalize_request_body(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-finalize-response-v1\0");
+    body.push_bytes(&request_body)?;
+    body.push_bool(response.success);
+    push_native_preparation_receipt(&mut body, response.receipt.as_ref())?;
+    push_native_preparation_error(&mut body, response.error.as_ref())?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_probe_response_body(
+    request: &proto_gen::node_service::NativePreparationProbeRequest,
+    response: &proto_gen::node_service::NativePreparationProbeResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let request_body = canonical_native_preparation_probe_request_body(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-probe-response-v1\0");
+    body.push_bytes(&request_body)?;
+    body.push_bool(response.success);
+    body.push_bool(response.exists);
+    body.push_bool(response.prepared);
+    body.push_bytes(&response.metadata)?;
+    push_native_preparation_error(&mut body, response.error.as_ref())?;
+    body.push_bytes(&response.version_id)?;
+    body.push_bytes(&response.object_digest)?;
+    body.push_u64(response.object_length);
+    body.push_bytes(&response.set_digest)?;
+    body.push_bytes(&response.distribution_digest)?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_recover_response_body(
+    request: &proto_gen::node_service::NativePreparationRecoverRequest,
+    response: &proto_gen::node_service::NativePreparationRecoverResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let request_body = canonical_native_preparation_recover_request_body(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-recover-response-v1\0");
+    body.push_bytes(&request_body)?;
+    body.push_bool(response.success);
+    push_native_preparation_receipt(&mut body, response.receipt.as_ref())?;
+    body.push_bytes(&response.metadata)?;
+    push_native_preparation_error(&mut body, response.error.as_ref())?;
+    Ok(body.finish())
+}
+
+pub fn canonical_native_preparation_locate_response_body(
+    request: &proto_gen::node_service::NativePreparationLocateRequest,
+    response: &proto_gen::node_service::NativePreparationLocateResponse,
+) -> Result<Vec<u8>, std::num::TryFromIntError> {
+    let request_body = canonical_native_preparation_locate_request_body(request)?;
+    let mut body = CanonicalBodyBuilder::new(b"rustfs-native-preparation-locate-response-v1\0");
+    body.push_bytes(&request_body)?;
+    body.push_bool(response.success);
+    body.push_bool(response.found);
+    body.push_str(&response.bucket)?;
+    body.push_str(&response.key)?;
+    body.push_bytes(&response.bucket_incarnation_id)?;
+    body.push_bytes(&response.metadata)?;
+    body.push_bool(response.prepared);
+    push_native_preparation_error(&mut body, response.error.as_ref())?;
+    Ok(body.finish())
+}
+
 pub fn canonical_compare_and_update_file_request_body(
     request: &proto_gen::node_service::CompareAndUpdateFileRequest,
 ) -> Result<Vec<u8>, std::num::TryFromIntError> {
@@ -1269,7 +1461,10 @@ pub fn canonical_make_volumes_request_body(
 mod disk_mutation_canonical_tests {
     use super::proto_gen::node_service::{
         CompareAndUpdateFileRequest, DeletePathsRequest, DeleteRequest, DeleteVersionRequest, DeleteVersionsRequest,
-        DeleteVolumeRequest, MakeVolumeRequest, MakeVolumesRequest, PreparePartTransactionRequest, RenameDataRequest,
+        DeleteVolumeRequest, MakeVolumeRequest, MakeVolumesRequest, NativePreparationBeginRequest,
+        NativePreparationFinalizeRequest, NativePreparationFinalizeResponse, NativePreparationLocateRequest,
+        NativePreparationLocateResponse, NativePreparationProbeRequest, NativePreparationProbeResponse, NativePreparationReceipt,
+        NativePreparationRecoverRequest, NativePreparationRecoverResponse, PreparePartTransactionRequest, RenameDataRequest,
         RenameFileRequest, RenamePartRequest, SettlePartTransactionRequest, SnapshotLeaseReleaseRequest,
         SnapshotLeaseRenewRequest, SnapshotLeaseRequest, UpdateMetadataRequest, WriteAllRequest, WriteMetadataRequest,
     };
@@ -1318,6 +1513,500 @@ mod disk_mutation_canonical_tests {
         assert_all_distinct(&bodies);
     }
 
+    #[test]
+    fn native_preparation_begin_body_binds_every_field() {
+        let baseline = NativePreparationBeginRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let mut bodies = vec![canonical_native_preparation_begin_request_body(&baseline).unwrap()];
+        for mutate in [
+            |r: &mut NativePreparationBeginRequest| r.disk = "node-b".into(),
+            |r: &mut NativePreparationBeginRequest| r.disk_id = vec![5; 16].into(),
+            |r: &mut NativePreparationBeginRequest| r.store_id = vec![5; 16].into(),
+            |r: &mut NativePreparationBeginRequest| r.preparation_id = vec![5; 16].into(),
+            |r: &mut NativePreparationBeginRequest| r.bucket = "other".into(),
+            |r: &mut NativePreparationBeginRequest| r.key = "other.jpg".into(),
+            |r: &mut NativePreparationBeginRequest| r.bucket_incarnation_id = vec![5; 16].into(),
+        ] {
+            let mut request = baseline.clone();
+            mutate(&mut request);
+            bodies.push(canonical_native_preparation_begin_request_body(&request).unwrap());
+        }
+        assert_all_distinct(&bodies);
+    }
+
+    #[test]
+    fn native_preparation_finalize_body_binds_every_field() {
+        let baseline = NativePreparationFinalizeRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+            object_digest: vec![5; 32].into(),
+            object_length: 1234,
+            version_id: vec![6; 16].into(),
+            metadata: vec![7, 8, 9].into(),
+        };
+        let mut bodies = vec![canonical_native_preparation_finalize_request_body(&baseline).unwrap()];
+        for mutate in [
+            |r: &mut NativePreparationFinalizeRequest| r.disk = "node-b".into(),
+            |r: &mut NativePreparationFinalizeRequest| r.disk_id = vec![9; 16].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.store_id = vec![9; 16].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.preparation_id = vec![9; 16].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.bucket = "other".into(),
+            |r: &mut NativePreparationFinalizeRequest| r.key = "other.jpg".into(),
+            |r: &mut NativePreparationFinalizeRequest| r.bucket_incarnation_id = vec![9; 16].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.object_digest = vec![9; 32].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.object_length = 4321,
+            |r: &mut NativePreparationFinalizeRequest| r.version_id = vec![9; 16].into(),
+            |r: &mut NativePreparationFinalizeRequest| r.metadata = vec![7, 8, 10].into(),
+        ] {
+            let mut request = baseline.clone();
+            mutate(&mut request);
+            bodies.push(canonical_native_preparation_finalize_request_body(&request).unwrap());
+        }
+        assert_all_distinct(&bodies);
+    }
+
+    #[test]
+    fn native_preparation_probe_body_binds_every_field() {
+        let baseline = NativePreparationProbeRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let mut bodies = vec![canonical_native_preparation_probe_request_body(&baseline).unwrap()];
+        for mutate in [
+            |r: &mut NativePreparationProbeRequest| r.disk = "node-b".into(),
+            |r: &mut NativePreparationProbeRequest| r.disk_id = vec![5; 16].into(),
+            |r: &mut NativePreparationProbeRequest| r.store_id = vec![5; 16].into(),
+            |r: &mut NativePreparationProbeRequest| r.preparation_id = vec![5; 16].into(),
+            |r: &mut NativePreparationProbeRequest| r.bucket = "other".into(),
+            |r: &mut NativePreparationProbeRequest| r.key = "other.jpg".into(),
+            |r: &mut NativePreparationProbeRequest| r.bucket_incarnation_id = vec![5; 16].into(),
+        ] {
+            let mut request = baseline.clone();
+            mutate(&mut request);
+            bodies.push(canonical_native_preparation_probe_request_body(&request).unwrap());
+        }
+        assert_all_distinct(&bodies);
+    }
+
+    #[test]
+    fn native_preparation_recover_body_binds_every_field() {
+        let baseline = NativePreparationRecoverRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let mut bodies = vec![canonical_native_preparation_recover_request_body(&baseline).unwrap()];
+        for mutate in [
+            |r: &mut NativePreparationRecoverRequest| r.disk = "node-b".into(),
+            |r: &mut NativePreparationRecoverRequest| r.disk_id = vec![5; 16].into(),
+            |r: &mut NativePreparationRecoverRequest| r.store_id = vec![5; 16].into(),
+            |r: &mut NativePreparationRecoverRequest| r.preparation_id = vec![5; 16].into(),
+            |r: &mut NativePreparationRecoverRequest| r.bucket = "other".into(),
+            |r: &mut NativePreparationRecoverRequest| r.key = "other.jpg".into(),
+            |r: &mut NativePreparationRecoverRequest| r.bucket_incarnation_id = vec![5; 16].into(),
+        ] {
+            let mut request = baseline.clone();
+            mutate(&mut request);
+            bodies.push(canonical_native_preparation_recover_request_body(&request).unwrap());
+        }
+        assert_all_distinct(&bodies);
+    }
+
+    fn native_preparation_receipt() -> NativePreparationReceipt {
+        NativePreparationReceipt {
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            deployment_id: vec![3; 16].into(),
+            preparation_id: vec![4; 16].into(),
+            version_id: vec![5; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![6; 16].into(),
+            object_digest: vec![7; 32].into(),
+            object_length: 1234,
+            set_digest: vec![8; 32].into(),
+            distribution_digest: vec![9; 32].into(),
+            metadata_digest: vec![10; 32].into(),
+            shard_digest: vec![11; 32].into(),
+            shard_length: 512,
+            data_blocks: 4,
+            parity_blocks: 2,
+            shard_index: 3,
+        }
+    }
+
+    #[test]
+    fn native_preparation_begin_response_binds_request_status_and_error() {
+        use super::proto_gen::node_service::{Error, NativePreparationBeginResponse};
+
+        let request = NativePreparationBeginRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let response = NativePreparationBeginResponse {
+            success: true,
+            error: None,
+            response_proof: vec![1; 32].into(),
+        };
+        let baseline = canonical_native_preparation_begin_response_body(&request, &response).unwrap();
+
+        let failed = NativePreparationBeginResponse {
+            success: false,
+            error: Some(Error {
+                code: 17,
+                error_info: "disk unavailable".into(),
+            }),
+            ..response.clone()
+        };
+        assert_ne!(baseline, canonical_native_preparation_begin_response_body(&request, &failed).unwrap());
+
+        let mut changed_request = request.clone();
+        changed_request.disk_id = vec![9; 16].into();
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_begin_response_body(&changed_request, &response).unwrap()
+        );
+
+        let mut proof_only = response.clone();
+        proof_only.response_proof = vec![2; 32].into();
+        assert_eq!(baseline, canonical_native_preparation_begin_response_body(&request, &proof_only).unwrap());
+    }
+
+    #[test]
+    fn native_preparation_finalize_response_binds_request_status_error_and_receipt() {
+        use super::proto_gen::node_service::Error;
+
+        let request = NativePreparationFinalizeRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+            object_digest: vec![5; 32].into(),
+            object_length: 1234,
+            version_id: vec![6; 16].into(),
+            metadata: vec![7, 8].into(),
+        };
+        let response = NativePreparationFinalizeResponse {
+            success: true,
+            receipt: Some(native_preparation_receipt()),
+            error: None,
+            response_proof: vec![1; 32].into(),
+        };
+        let baseline = canonical_native_preparation_finalize_response_body(&request, &response).unwrap();
+        let mut bodies = vec![baseline.clone()];
+        for mutate in [
+            |r: &mut NativePreparationReceipt| r.disk_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.store_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.deployment_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.preparation_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.version_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.bucket = "other".into(),
+            |r: &mut NativePreparationReceipt| r.key = "other.jpg".into(),
+            |r: &mut NativePreparationReceipt| r.bucket_incarnation_id = vec![12; 16].into(),
+            |r: &mut NativePreparationReceipt| r.object_digest = vec![12; 32].into(),
+            |r: &mut NativePreparationReceipt| r.object_length = 4321,
+            |r: &mut NativePreparationReceipt| r.set_digest = vec![12; 32].into(),
+            |r: &mut NativePreparationReceipt| r.distribution_digest = vec![12; 32].into(),
+            |r: &mut NativePreparationReceipt| r.metadata_digest = vec![12; 32].into(),
+            |r: &mut NativePreparationReceipt| r.shard_digest = vec![12; 32].into(),
+            |r: &mut NativePreparationReceipt| r.shard_length = 1024,
+            |r: &mut NativePreparationReceipt| r.data_blocks = 8,
+            |r: &mut NativePreparationReceipt| r.parity_blocks = 4,
+            |r: &mut NativePreparationReceipt| r.shard_index = 5,
+        ] {
+            let mut changed = response.clone();
+            mutate(changed.receipt.as_mut().unwrap());
+            bodies.push(canonical_native_preparation_finalize_response_body(&request, &changed).unwrap());
+        }
+        assert_all_distinct(&bodies);
+
+        let failed = NativePreparationFinalizeResponse {
+            success: false,
+            error: Some(Error {
+                code: 19,
+                error_info: "finalize failed".into(),
+            }),
+            ..response.clone()
+        };
+        assert_ne!(baseline, canonical_native_preparation_finalize_response_body(&request, &failed).unwrap());
+
+        let mut changed_request = request.clone();
+        changed_request.metadata = vec![7, 9].into();
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_finalize_response_body(&changed_request, &response).unwrap()
+        );
+
+        let mut proof_only = response.clone();
+        proof_only.response_proof = vec![2; 32].into();
+        assert_eq!(
+            baseline,
+            canonical_native_preparation_finalize_response_body(&request, &proof_only).unwrap()
+        );
+    }
+
+    #[test]
+    fn native_preparation_probe_response_binds_state_receipt_metadata_and_error() {
+        use super::proto_gen::node_service::Error;
+
+        let request = NativePreparationProbeRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let response = NativePreparationProbeResponse {
+            success: true,
+            exists: true,
+            prepared: true,
+            metadata: vec![1, 2, 3].into(),
+            error: None,
+            response_proof: vec![1; 32].into(),
+            version_id: vec![4; 16].into(),
+            object_digest: vec![5; 32].into(),
+            object_length: 4096,
+            set_digest: vec![6; 32].into(),
+            distribution_digest: vec![7; 32].into(),
+        };
+        let baseline = canonical_native_preparation_probe_response_body(&request, &response).unwrap();
+        let mut bodies = vec![baseline.clone()];
+        for (exists, prepared) in [(false, true), (true, false), (false, false)] {
+            let changed = NativePreparationProbeResponse {
+                exists,
+                prepared,
+                ..response.clone()
+            };
+            bodies.push(canonical_native_preparation_probe_response_body(&request, &changed).unwrap());
+        }
+        let metadata_changed = NativePreparationProbeResponse {
+            metadata: vec![1, 2, 4].into(),
+            ..response.clone()
+        };
+        bodies.push(canonical_native_preparation_probe_response_body(&request, &metadata_changed).unwrap());
+        for changed in [
+            NativePreparationProbeResponse {
+                version_id: vec![8; 16].into(),
+                ..response.clone()
+            },
+            NativePreparationProbeResponse {
+                object_digest: vec![8; 32].into(),
+                ..response.clone()
+            },
+            NativePreparationProbeResponse {
+                object_length: 8192,
+                ..response.clone()
+            },
+            NativePreparationProbeResponse {
+                set_digest: vec![8; 32].into(),
+                ..response.clone()
+            },
+            NativePreparationProbeResponse {
+                distribution_digest: vec![8; 32].into(),
+                ..response.clone()
+            },
+        ] {
+            bodies.push(canonical_native_preparation_probe_response_body(&request, &changed).unwrap());
+        }
+        assert_all_distinct(&bodies);
+
+        let failed = NativePreparationProbeResponse {
+            success: false,
+            error: Some(Error {
+                code: 21,
+                error_info: "probe failed".into(),
+            }),
+            ..response.clone()
+        };
+        assert_ne!(baseline, canonical_native_preparation_probe_response_body(&request, &failed).unwrap());
+
+        let mut changed_request = request.clone();
+        changed_request.preparation_id = vec![9; 16].into();
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_probe_response_body(&changed_request, &response).unwrap()
+        );
+    }
+
+    #[test]
+    fn native_preparation_recover_response_binds_request_receipt_metadata_and_error() {
+        use super::proto_gen::node_service::Error;
+
+        let request = NativePreparationRecoverRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+        };
+        let response = NativePreparationRecoverResponse {
+            success: true,
+            receipt: Some(native_preparation_receipt()),
+            metadata: vec![1, 2, 3].into(),
+            error: None,
+            response_proof: vec![1; 32].into(),
+        };
+        let baseline = canonical_native_preparation_recover_response_body(&request, &response).unwrap();
+
+        let metadata_changed = NativePreparationRecoverResponse {
+            metadata: vec![1, 2, 4].into(),
+            ..response.clone()
+        };
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_recover_response_body(&request, &metadata_changed).unwrap()
+        );
+
+        let failed = NativePreparationRecoverResponse {
+            success: false,
+            error: Some(Error {
+                code: 23,
+                error_info: "recover failed".into(),
+            }),
+            ..response.clone()
+        };
+        assert_ne!(baseline, canonical_native_preparation_recover_response_body(&request, &failed).unwrap());
+
+        let mut changed_request = request.clone();
+        changed_request.key = "other.jpg".into();
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_recover_response_body(&changed_request, &response).unwrap()
+        );
+
+        let mut proof_only = response.clone();
+        proof_only.response_proof = vec![2; 32].into();
+        assert_eq!(
+            baseline,
+            canonical_native_preparation_recover_response_body(&request, &proof_only).unwrap()
+        );
+    }
+
+    #[test]
+    fn native_preparation_locate_bodies_bind_route_and_description() {
+        use super::proto_gen::node_service::Error;
+
+        let request = NativePreparationLocateRequest {
+            disk: "node-a".into(),
+            disk_id: vec![1; 16].into(),
+            store_id: vec![2; 16].into(),
+            preparation_id: vec![3; 16].into(),
+        };
+        let response = NativePreparationLocateResponse {
+            success: true,
+            found: true,
+            bucket: "photos".into(),
+            key: "image.jpg".into(),
+            bucket_incarnation_id: vec![4; 16].into(),
+            metadata: vec![5, 6, 7].into(),
+            prepared: true,
+            error: None,
+            response_proof: vec![1; 32].into(),
+        };
+        let baseline = canonical_native_preparation_locate_response_body(&request, &response).unwrap();
+        let mut bodies = vec![baseline.clone()];
+        for changed in [
+            NativePreparationLocateResponse {
+                success: false,
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                found: false,
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                bucket: "other".into(),
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                key: "other.jpg".into(),
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                bucket_incarnation_id: vec![8; 16].into(),
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                metadata: vec![5, 6, 8].into(),
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                prepared: false,
+                ..response.clone()
+            },
+            NativePreparationLocateResponse {
+                error: Some(Error {
+                    code: 25,
+                    error_info: "locate failed".into(),
+                }),
+                ..response.clone()
+            },
+        ] {
+            bodies.push(canonical_native_preparation_locate_response_body(&request, &changed).unwrap());
+        }
+        assert_all_distinct(&bodies);
+
+        let mut changed_request = request.clone();
+        changed_request.store_id = vec![9; 16].into();
+        assert_ne!(
+            baseline,
+            canonical_native_preparation_locate_response_body(&changed_request, &response).unwrap()
+        );
+
+        let mut proof_only = response.clone();
+        proof_only.response_proof = vec![2; 32].into();
+        assert_eq!(
+            baseline,
+            canonical_native_preparation_locate_response_body(&request, &proof_only).unwrap()
+        );
+
+        let mut request_bodies = vec![canonical_native_preparation_locate_request_body(&request).unwrap()];
+        for mutate in [
+            |r: &mut NativePreparationLocateRequest| r.disk = "node-b".into(),
+            |r: &mut NativePreparationLocateRequest| r.disk_id = vec![9; 16].into(),
+            |r: &mut NativePreparationLocateRequest| r.store_id = vec![9; 16].into(),
+            |r: &mut NativePreparationLocateRequest| r.preparation_id = vec![9; 16].into(),
+        ] {
+            let mut changed = request.clone();
+            mutate(&mut changed);
+            request_bodies.push(canonical_native_preparation_locate_request_body(&changed).unwrap());
+        }
+        assert_all_distinct(&request_bodies);
+    }
+
     #[test]
     fn rename_data_canonical_body_is_injective_across_field_boundaries() {
         // Length prefixes must prevent moving bytes between adjacent fields from colliding.

crates/protos/src/node.proto

@@ -142,6 +142,132 @@ message CompareAndUpdateFileResponse {
   optional Error error = 3;
 }
 
+// Native preparation carries only bounded object metadata over the control RPC.
+// Shard bytes use the authenticated data stream and remain private until finalized.
+message NativePreparationReceipt {
+  bytes disk_id = 1; // 16-byte UUID.
+  bytes store_id = 2; // 16-byte UUID.
+  bytes deployment_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+  bytes version_id = 5; // 16-byte UUID.
+  string bucket = 6;
+  string key = 7;
+  bytes bucket_incarnation_id = 8; // 16-byte UUID.
+  bytes object_digest = 9; // 32-byte SHA-256.
+  uint64 object_length = 10;
+  bytes set_digest = 11;
+  bytes distribution_digest = 12;
+  bytes metadata_digest = 13;
+  bytes shard_digest = 14;
+  uint64 shard_length = 15;
+  uint32 data_blocks = 16;
+  uint32 parity_blocks = 17;
+  uint32 shard_index = 18;
+  reserved 19, 20;
+}
+
+message NativePreparationBeginRequest {
+  string disk = 1;
+  bytes disk_id = 2; // 16-byte UUID.
+  bytes store_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+  string bucket = 5;
+  string key = 6;
+  bytes bucket_incarnation_id = 7; // 16-byte UUID.
+}
+
+message NativePreparationBeginResponse {
+  bool success = 1;
+  optional Error error = 2;
+  bytes response_proof = 3;
+}
+
+message NativePreparationFinalizeRequest {
+  string disk = 1;
+  bytes disk_id = 2; // 16-byte UUID.
+  bytes store_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+  string bucket = 5;
+  string key = 6;
+  bytes bucket_incarnation_id = 7; // 16-byte UUID.
+  bytes object_digest = 8; // 32-byte SHA-256.
+  uint64 object_length = 9;
+  bytes version_id = 10; // 16-byte UUID.
+  // Exact serialized FileMeta; maximum 1 MiB. Peers reject larger values.
+  bytes metadata = 11;
+}
+
+message NativePreparationFinalizeResponse {
+  bool success = 1;
+  NativePreparationReceipt receipt = 2;
+  optional Error error = 3;
+  bytes response_proof = 4;
+}
+
+message NativePreparationProbeRequest {
+  string disk = 1;
+  bytes disk_id = 2; // 16-byte UUID.
+  bytes store_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+  string bucket = 5;
+  string key = 6;
+  bytes bucket_incarnation_id = 7; // 16-byte UUID.
+}
+
+message NativePreparationProbeResponse {
+  bool success = 1;
+  bool exists = 2;
+  bool prepared = 3;
+  reserved 4;
+  // Exact serialized FileMeta when it exists and can be read; maximum 1 MiB.
+  bytes metadata = 5;
+  optional Error error = 6;
+  bytes response_proof = 7;
+  bytes version_id = 8; // 16-byte UUID; observation only, not a durability receipt.
+  bytes object_digest = 9; // 32-byte SHA-256.
+  uint64 object_length = 10;
+  bytes set_digest = 11;
+  bytes distribution_digest = 12;
+}
+
+message NativePreparationRecoverRequest {
+  string disk = 1;
+  bytes disk_id = 2; // 16-byte UUID.
+  bytes store_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+  string bucket = 5;
+  string key = 6;
+  bytes bucket_incarnation_id = 7; // 16-byte UUID.
+}
+
+message NativePreparationRecoverResponse {
+  bool success = 1;
+  NativePreparationReceipt receipt = 2;
+  // Exact persisted FileMeta; maximum 1 MiB. Peers reject larger values.
+  bytes metadata = 3;
+  optional Error error = 4;
+  bytes response_proof = 5;
+}
+
+message NativePreparationLocateRequest {
+  string disk = 1;
+  bytes disk_id = 2; // 16-byte UUID.
+  bytes store_id = 3; // 16-byte UUID.
+  bytes preparation_id = 4; // 16-byte UUID.
+}
+
+message NativePreparationLocateResponse {
+  bool success = 1;
+  bool found = 2;
+  string bucket = 3;
+  string key = 4;
+  bytes bucket_incarnation_id = 5; // 16-byte UUID.
+  bytes metadata = 6; // Exact serialized FileMeta; maximum 1 MiB.
+  bool prepared = 7;
+  optional Error error = 8;
+  bytes response_proof = 9;
+}
+
 message DeleteRequest {
   string disk = 1;  // indicate which one in the disks
   string volume = 2;
@@ -1216,6 +1342,11 @@ service NodeService {
   rpc ReadAll(ReadAllRequest) returns (ReadAllResponse) {}; // auth-policy: read-only
   rpc WriteAll(WriteAllRequest) returns (WriteAllResponse) {}; // auth-policy: body-bound
   rpc CompareAndUpdateFile(CompareAndUpdateFileRequest) returns (CompareAndUpdateFileResponse) {}; // auth-policy: body-bound
+  rpc NativePreparationBegin(NativePreparationBeginRequest) returns (NativePreparationBeginResponse) {}; // auth-policy: body-bound
+  rpc NativePreparationFinalize(NativePreparationFinalizeRequest) returns (NativePreparationFinalizeResponse) {}; // auth-policy: body-bound
+  rpc NativePreparationProbe(NativePreparationProbeRequest) returns (NativePreparationProbeResponse) {}; // auth-policy: body-bound
+  rpc NativePreparationRecover(NativePreparationRecoverRequest) returns (NativePreparationRecoverResponse) {}; // auth-policy: body-bound
+  rpc NativePreparationLocate(NativePreparationLocateRequest) returns (NativePreparationLocateResponse) {}; // auth-policy: body-bound
   rpc Delete(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound
   rpc DeleteAtIncarnation(DeleteRequest) returns (DeleteResponse) {}; // auth-policy: body-bound
   rpc AcquireSnapshotLease(SnapshotLeaseRequest) returns (SnapshotLeaseResponse) {}; // auth-policy: body-bound

docs/architecture/native-object-preparation.md

@@ -24,16 +24,20 @@ flowchart LR
 
 ## Encoding and placement
 
-The supported profile is bounded, single-part, single-block inline data on local
-disks, subject to the configured inline admission limit. Unsupported sizes or
-remote owners fail explicitly; they do not fall back to public PUT.
+Small local candidates retain the bounded inline preparation path. Larger or
+unknown-length candidates use the ordinary streaming erasure/bitrot encoder to
+write private part files, with a fixed-size SHA-256 read buffer rather than
+buffering the object. Unsupported owners fail explicitly; preparation never
+falls back to a visible public PUT.
 `SetDisks::prepare_native_bytes` in
 `crates/ecstore/src/set_disk/ops/native_prepare.rs` uses the existing write layout
 and encoder. Verification uses the codec indicated by the decoded metadata,
 including its shard width and checksum compatibility mode.
 
-The object key selects a set in each pool. Existing private intents fix the pool
-for retries. Absence is established only against a healthy disk identity;
+New candidates use the existing pool/set selection. Recovery searches actual
+sets for the preparation UUID, then verifies the retained physical disk set and
+distribution rather than recomputing placement from the current key hash.
+Existing private intents fix placement for retries. Absence is established only against a healthy disk identity;
 unreadable state in another pool prevents assigning a new pool. Conflicting
 identities are errors, even if another subset could form a quorum.
 
@@ -46,7 +50,11 @@ Detached disk writes retain the parent object owner after early quorum return or
 caller cancellation. An unobserved response does not authorize rollback.
 
 Version 2 intents retain the original raw metadata so an intent-only interruption
-can be repaired without reserializing metadata maps. Existing version 1 intents
+can be repaired without reserializing metadata maps. Version 3 streaming intents
+also retain the ordered physical disk UUID set and data directory UUID. The exact
+FileInfo supplies the part manifest; each file's bitrot data and the aggregate
+shard digest/length are verified before the record and prepared marker barriers.
+Part files and their directory are synced before a disk can attest preparation. Existing version 1 intents
 keep their original canonical representation and marker. A missing shard with an
 existing prepared marker is corruption, not a successful preparation.
 
@@ -55,7 +63,8 @@ existing prepared marker is corruption, not a successful preparation.
 Recovery acquires fresh lifecycle and namespace owners, verifies persisted disk
 facts, decodes the actual shards, and checks the complete object length and SHA-256.
 It requires the write quorum and consistent identity, codec, modification time,
-and block geometry. Retry preserves original metadata bytes and rejects changed
+part manifest and block geometry. It decodes each retained part into a streaming
+digest sink, so recovery does not allocate a buffer proportional to object size. Retry preserves original metadata bytes and rejects changed
 content. Fewer receipts or unknown I/O outcomes leave durable facts available for
 recovery without granting publication authority.
 

docs/architecture/native-raft-owner.md

@@ -27,3 +27,11 @@ A `RaftNetworkFactory<CaptureTypeConfig>` is mandatory. This module has no defau
 `crates/capture-authority/src/raft_owner_tests.rs` uses real Raft instances and real redb log/state-machine adapters. Its direct RPC factory is only a test transport: it forwards append, vote and snapshot calls to actual peer Raft handles. Tests cover explicit bootstrap, restart without initialize, retry identity, runtime identity rejection, live shared reader, shutdown, three-node majority, leader failover and snapshot catch-up through actual Raft RPC. Readiness comes from metrics and authoritative events, rather than a fixed sleep.
 
 These tests establish the internal owner and consensus path under that test transport. They do not prove production network trust or wire native capture into S3 PUT, Copy or multipart publication. S06 remains blocked, R12 remains frozen, and production S3 automatic capture remains a later integration task. Removing this lifecycle adapter does not change the accepted native storage/snapshot formats; retain the selected durable baseline after prefix purge and follow its existing empty-baseline rollback restrictions.
+
+## Exact Created publication authorization
+
+`NativeRaftOwner::authorize_created(expected, expected_decision)` reads the canonical operation, its successful first result, event, version, capture binding, native provenance and applied/committed boundaries under one ordered Store read transaction. Supply the complete `DecideCreated` and the successful result's first full decision LogId. An idempotent retry's outer log index does not replace that identity. Rejected or unapplied operations cannot mint authority.
+
+The returned `CommittedCreatedAuthorization` has private fields, read-only accessors, no public constructor and no deserialization. It binds the durable cluster, operation, object incarnation/key, expected head, preparation/version/digest/length, binding and first decision. Its node identity identifies the current issuing owner, not a requirement to publish on the original leader. Recovery reads retained provenance and returns the same decision facts after actual Raft recovery.
+
+This local type is not a remote signature or proof of physical object durability. Publication must separately validate the real durable preparation and current bucket incarnation, namespace and head before making bytes visible. It does not revoke historical authorization when a later head is created, and it provides no external command endpoint. Streaming preparation and S3 publication consume this boundary separately.

rustfs/src/server/http.rs

@@ -4159,6 +4159,291 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn native_stream_prepare_remote_restart() {
+        use crate::storage::rpc::node_service::make_server_for_context;
+        use crate::storage::storage_api::ecstore_disk::DiskAPI as _;
+        use crate::storage::storage_api::{
+            ECStore, InstanceContext, bootstrap_instance_ctx, ecstore_storage::init_local_disks_with_instance_ctx,
+        };
+        use rustfs_ecstore::api::{
+            error::is_err_object_not_found,
+            object::{ObjectOptions, PutObjReader},
+        };
+        use rustfs_iam::store::Store as _;
+        use rustfs_storage_api::object::{ListOperations, ObjectOperations};
+        use sha2::Digest as _;
+        use tokio_util::sync::CancellationToken;
+
+        let listener = match TcpListener::bind("127.0.0.1:0").await {
+            Ok(listener) => listener,
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
+            Err(err) => panic!("bind native preparation peer: {err}"),
+        };
+        let addr = listener.local_addr().expect("peer listener address");
+        let previous_node_name = rustfs_common::get_global_local_node_name().await;
+        rustfs_common::set_global_local_node_name(&addr.to_string()).await;
+        let _ = rustfs_credentials::set_global_rpc_secret("native-stream-remote-restart-secret".to_string());
+
+        let root = tempfile::tempdir().expect("native preparation root");
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .base_dir(root.path())
+            .init_bucket_metadata(true)
+            .build()
+            .await;
+        let bucket = "native-remote-restart";
+        let key = "large/private-object.bin";
+        env.make_bucket(bucket, false).await;
+
+        let mut local_pool = env.endpoint_pools.as_ref()[0].clone();
+        let local_endpoints = env
+            .disk_paths
+            .iter()
+            .enumerate()
+            .map(|(index, path)| {
+                let mut endpoint =
+                    Endpoint::try_from(format!("http://{addr}{}", path.display()).as_str()).expect("local peer endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(index);
+                endpoint.is_local = true;
+                endpoint
+            })
+            .collect::<Vec<_>>();
+        local_pool.endpoints = Endpoints::from(local_endpoints);
+        let owner_pools = EndpointServerPools::from(vec![local_pool]);
+        let owner_instance = bootstrap_instance_ctx();
+        init_local_disks_with_instance_ctx(&owner_instance, owner_pools.clone())
+            .await
+            .expect("register the production local disk owners");
+
+        let object_store = rustfs_iam::store::object::ObjectStore::new(env.ecstore.clone());
+        object_store
+            .save_iam_config(
+                serde_json::json!({"version": 1}),
+                format!("{}/format.json", *rustfs_iam::store::object::IAM_CONFIG_PREFIX),
+            )
+            .await
+            .expect("seed IAM metadata for the node service");
+        let iam = rustfs_iam::build_iam_sys(env.ecstore.clone())
+            .await
+            .expect("build fixture IAM");
+        let app_context = Arc::new(crate::runtime_sources::AppContext::with_default_interfaces(
+            env.ecstore.clone(),
+            iam,
+            Arc::new(rustfs_kms::KmsServiceManager::new()),
+        ));
+
+        let node_service = make_server_for_context(Some(app_context));
+        for endpoint in owner_pools.as_ref()[0].endpoints.as_ref() {
+            let disk = crate::storage::storage_api::find_local_disk_by_ref(&endpoint.to_string())
+                .await
+                .expect("resolve the same registered disk used by production HTTP RPC");
+            if let Err(error) = disk
+                .make_volume(crate::storage::storage_api::ecstore_disk::RUSTFS_META_BUCKET)
+                .await
+            {
+                assert_eq!(
+                    error,
+                    crate::storage::storage_api::DiskError::VolumeExists,
+                    "create owner metadata volume"
+                );
+            }
+        }
+        let node = InterceptedService::new(
+            NodeServiceServer::new(node_service)
+                .max_decoding_message_size(rustfs_protos::internode_rpc_max_message_size())
+                .max_encoding_message_size(rustfs_protos::internode_rpc_max_message_size()),
+            check_auth,
+        );
+        let grpc = RpcRequestPathService::new(Routes::new(node).prepare());
+        let http =
+            InternodeRpcService::new(tower::service_fn(|_| async { Ok::<_, Infallible>(Response::new(s3s::Body::empty())) }));
+        let service = hybrid(http, grpc);
+        let shutdown = tokio_util::sync::CancellationToken::new();
+        let shutdown_for_task = shutdown.clone();
+        let server = tokio::spawn(async move {
+            loop {
+                tokio::select! {
+                    _ = shutdown_for_task.cancelled() => break,
+                    accepted = listener.accept() => {
+                        let (socket, _) = accepted.expect("accept native preparation peer connection");
+                        let service = service.clone();
+                        tokio::spawn(async move {
+                            let _ = ConnBuilder::new(TokioExecutor::new())
+                                .serve_connection(TokioIo::new(socket), TowerToHyperService::new(service))
+                                .await;
+                        });
+                    }
+                }
+            }
+        });
+
+        let mut remote_pool = env.endpoint_pools.as_ref()[0].clone();
+        let remote_endpoints = env
+            .disk_paths
+            .iter()
+            .enumerate()
+            .map(|(index, path)| {
+                let mut endpoint =
+                    Endpoint::try_from(format!("http://{addr}{}", path.display()).as_str()).expect("remote peer endpoint");
+                endpoint.set_pool_index(0);
+                endpoint.set_set_index(0);
+                endpoint.set_disk_index(index);
+                endpoint.is_local = false;
+                endpoint
+            })
+            .collect::<Vec<_>>();
+        remote_pool.endpoints = Endpoints::from(remote_endpoints);
+        let remote_pools = EndpointServerPools::from(vec![remote_pool]);
+        let client_instance = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&client_instance, remote_pools.clone())
+            .await
+            .expect("connect production RemoteDisk clients");
+        let client_cancel = CancellationToken::new();
+        let client = ECStore::new_with_instance_ctx(addr, remote_pools.clone(), client_cancel.clone(), client_instance)
+            .await
+            .expect("open coordinator over real RemoteDisk clients");
+        assert_eq!(client.id, env.ecstore.id, "remote coordinator must use the same deployment identity");
+
+        let preparation = uuid::Uuid::new_v4();
+        let body = (0..(5 * 1024 * 1024 + 37))
+            .map(|index| u8::try_from(index % 251).expect("bounded byte"))
+            .collect::<Vec<_>>();
+        let expected_digest: [u8; 32] = sha2::Sha256::digest(&body).into();
+        let prepared = client
+            .prepare_native_object(bucket, key, preparation, &mut PutObjReader::from_vec(body.clone()))
+            .await
+            .expect("authenticated TCP streaming preparation over remote disks");
+        assert_eq!(prepared.object_digest(), expected_digest);
+        assert!(prepared.receipt_count() >= 3, "remote write quorum must be distinct and durable");
+        assert_eq!(
+            prepared.disk_ids().collect::<std::collections::HashSet<_>>().len(),
+            prepared.receipt_count(),
+            "remote quorum must name distinct physical disks"
+        );
+        let expected_receipt_count = prepared.receipt_count();
+        let expected_disk_ids = prepared.disk_ids().collect::<std::collections::HashSet<_>>();
+        let mut durable_disk_count = 0;
+        for path in &env.disk_paths {
+            let record = path.join(".rustfs.sys/native-preparations").join(preparation.to_string());
+            if record.exists() {
+                assert!(record.join("PREPARED").is_file(), "owner disk persisted the prepared marker");
+                assert!(record.join(preparation.to_string()).join("part.1").is_file());
+                durable_disk_count += 1;
+            }
+        }
+        assert!(durable_disk_count >= 3, "real remote quorum must exist on owner disks");
+        drop(prepared);
+        assert!(
+            client
+                .get_object_info(bucket, key, &ObjectOptions::default())
+                .await
+                .is_err_and(|error| is_err_object_not_found(&error))
+        );
+        let listed = client
+            .clone()
+            .list_objects_v2(bucket, "large/", None, None, 100, false, None, false)
+            .await
+            .expect("ordinary LIST remains available");
+        assert!(
+            listed.objects.is_empty() && listed.prefixes.is_empty(),
+            "private preparation must stay out of ordinary LIST"
+        );
+        client_cancel.cancel();
+        drop(client);
+        tokio::task::yield_now().await;
+
+        let restarted_instance = Arc::new(InstanceContext::new());
+        init_local_disks_with_instance_ctx(&restarted_instance, remote_pools.clone())
+            .await
+            .expect("reconnect RemoteDisk clients after coordinator restart");
+        let restarted_cancel = CancellationToken::new();
+        let restarted = ECStore::new_with_instance_ctx(addr, remote_pools, restarted_cancel.clone(), restarted_instance)
+            .await
+            .expect("restart coordinator over durable remote disks");
+        let recovered = restarted
+            .recover_native_object(bucket, key, preparation)
+            .await
+            .expect("recover exact remote preparation after restart");
+        assert_eq!(recovered.object_digest(), expected_digest);
+        assert_eq!(recovered.object_length(), u64::try_from(body.len()).expect("bounded length"));
+        assert_eq!(recovered.receipt_count(), expected_receipt_count);
+        assert_eq!(recovered.disk_ids().collect::<std::collections::HashSet<_>>(), expected_disk_ids);
+        assert!(
+            restarted
+                .get_object_info(bucket, key, &ObjectOptions::default())
+                .await
+                .is_err_and(|error| is_err_object_not_found(&error))
+        );
+        let listed = restarted
+            .clone()
+            .list_objects_v2(bucket, "large/", None, None, 100, false, None, false)
+            .await
+            .expect("ordinary LIST remains available after restart");
+        assert!(
+            listed.objects.is_empty() && listed.prefixes.is_empty(),
+            "recovered preparation remains private"
+        );
+        drop(recovered);
+
+        assert_eq!(durable_disk_count, 4, "failure cases require all four owner disks to be prepared");
+        let record_dir = env.disk_paths[0]
+            .join(".rustfs.sys/native-preparations")
+            .join(preparation.to_string());
+        let intent_path = record_dir.join("intent.json");
+        let marker_path = record_dir.join("PREPARED");
+        let original_intent = std::fs::read(&intent_path).expect("read durable remote intent");
+        let original_marker = std::fs::read(&marker_path).expect("read durable remote marker");
+        let mut invalid_marker = original_marker.clone();
+        invalid_marker[0] ^= 1;
+        let mut intent: serde_json::Value = serde_json::from_slice(&original_intent).expect("decode remote intent");
+        let original_set_digest = serde_json::to_string(&intent["set_digest"]).expect("serialize retained set digest");
+        let byte = intent["set_digest"][0].as_u64().expect("set digest byte");
+        intent["set_digest"][0] = serde_json::json!(byte ^ 1);
+        let changed_set_digest = serde_json::to_string(&intent["set_digest"]).expect("serialize conflicting set digest");
+        // Preserve the typed record's canonical field order while changing its physical identity.
+        let conflicting_intent = String::from_utf8(original_intent.clone())
+            .expect("UTF-8 intent")
+            .replacen(
+                &format!("\"set_digest\":{original_set_digest}"),
+                &format!("\"set_digest\":{changed_set_digest}"),
+                1,
+            )
+            .into_bytes();
+        assert_ne!(conflicting_intent, original_intent, "inject a real physical set identity conflict");
+        let conflicting_marker = sha2::Sha256::digest(&conflicting_intent).to_vec();
+        for (intent, marker) in [
+            (&original_intent, &invalid_marker),
+            (&conflicting_intent, &conflicting_marker),
+        ] {
+            std::fs::write(&intent_path, intent).expect("inject remote intent state");
+            std::fs::write(&marker_path, marker).expect("inject remote marker state");
+            let result = restarted.recover_native_object(bucket, key, preparation).await;
+            std::fs::write(&intent_path, &original_intent).expect("restore remote intent");
+            std::fs::write(&marker_path, &original_marker).expect("restore remote marker");
+            assert!(
+                result.is_err_and(|error| matches!(error, rustfs_ecstore::api::error::StorageError::FileCorrupt)),
+                "one corrupt remote owner must fail recovery even when three healthy disks form a quorum"
+            );
+        }
+
+        let unavailable_path = root.path().join("unavailable-native-disk");
+        std::fs::rename(&env.disk_paths[0], &unavailable_path).expect("take one owner disk offline");
+        let degraded = restarted.recover_native_object(bucket, key, preparation).await;
+        std::fs::rename(&unavailable_path, &env.disk_paths[0]).expect("restore offline owner disk");
+        let degraded = degraded.expect("three healthy remote owners must retain the write quorum");
+        assert_eq!(degraded.receipt_count(), 3);
+        assert_eq!(degraded.object_digest(), expected_digest);
+        assert_eq!(degraded.object_length(), u64::try_from(body.len()).expect("bounded length"));
+        drop(degraded);
+        restarted_cancel.cancel();
+        shutdown.cancel();
+        let _ = server.await;
+        rustfs_common::set_global_local_node_name(&previous_node_name).await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn peer_rest_heal_control_uses_production_auth_and_keeps_validation_errors_online() {

rustfs/src/storage/rpc/node_service.rs

@@ -1741,6 +1741,36 @@ impl Node for NodeService {
     async fn delete_paths(&self, request: Request<DeletePathsRequest>) -> Result<Response<DeletePathsResponse>, Status> {
         self.handle_delete_paths(request).await
     }
+    async fn native_preparation_begin(
+        &self,
+        request: Request<NativePreparationBeginRequest>,
+    ) -> Result<Response<NativePreparationBeginResponse>, Status> {
+        self.handle_native_preparation_begin(request).await
+    }
+    async fn native_preparation_finalize(
+        &self,
+        request: Request<NativePreparationFinalizeRequest>,
+    ) -> Result<Response<NativePreparationFinalizeResponse>, Status> {
+        self.handle_native_preparation_finalize(request).await
+    }
+    async fn native_preparation_probe(
+        &self,
+        request: Request<NativePreparationProbeRequest>,
+    ) -> Result<Response<NativePreparationProbeResponse>, Status> {
+        self.handle_native_preparation_probe(request).await
+    }
+    async fn native_preparation_recover(
+        &self,
+        request: Request<NativePreparationRecoverRequest>,
+    ) -> Result<Response<NativePreparationRecoverResponse>, Status> {
+        self.handle_native_preparation_recover(request).await
+    }
+    async fn native_preparation_locate(
+        &self,
+        request: Request<NativePreparationLocateRequest>,
+    ) -> Result<Response<NativePreparationLocateResponse>, Status> {
+        self.handle_native_preparation_locate(request).await
+    }
     async fn acquire_snapshot_lease(
         &self,
         request: Request<SnapshotLeaseRequest>,
@@ -3097,7 +3127,7 @@ mod tests {
     use tonic::{Request, Response, Status};
     use uuid::Uuid;
 
-    const DISK_MUTATION_RPC_METHODS: [&str; 24] = [
+    const DISK_MUTATION_RPC_METHODS: [&str; 29] = [
         "renamedata",
         "renamedataatincarnation",
         "writemetadataatincarnation",
@@ -3122,6 +3152,11 @@ mod tests {
         "acquiresnapshotlease",
         "renewsnapshotlease",
         "releasesnapshotlease",
+        "nativepreparationbegin",
+        "nativepreparationfinalize",
+        "nativepreparationprobe",
+        "nativepreparationrecover",
+        "nativepreparationlocate",
     ];
 
     fn normalized_rpc_method(method: &str) -> String {

rustfs/src/storage/rpc/node_service/disk.rs

@@ -19,7 +19,10 @@ use crate::storage::storage_api::rpc_consumer::node_service::{
     validate_batch_read_version_item_count,
 };
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
-use crate::storage::storage_api::{PartTransactionAction, RenameDataResp, SnapshotLeaseToken, verify_tonic_mutation_body_digest};
+use crate::storage::storage_api::{
+    PartTransactionAction, RenameDataResp, SnapshotLeaseToken, StorageError, sign_tonic_rpc_response_proof,
+    verify_tonic_mutation_body_digest,
+};
 use bytes::Bytes;
 use rustfs_filemeta::FileInfo;
 use rustfs_io_metrics::internode_metrics::{
@@ -118,6 +121,107 @@ const MSGPACK_ENCODE_CAPACITY_HINT: usize = 512;
 const FILE_INFO_MSGPACK_ENCODE_CAPACITY_HINT: usize = 1024;
 const SNAPSHOT_LEASE_PROTOCOL_VERSION: u32 = 1;
 
+fn parse_native_uuid(bytes: &[u8], field: &str) -> std::result::Result<Uuid, StorageError> {
+    Uuid::from_slice(bytes)
+        .ok()
+        .filter(|id| !id.is_nil())
+        .ok_or_else(|| StorageError::other(format!("{field} must be a non-nil UUID")))
+}
+
+fn native_target(
+    disk: &str,
+    disk_id: &[u8],
+    store_id: &[u8],
+    preparation_id: &[u8],
+    bucket: &str,
+    key: &str,
+    bucket_incarnation_id: &[u8],
+) -> std::result::Result<rustfs_ecstore::api::storage::NativePreparationTarget, StorageError> {
+    if disk.is_empty() {
+        return Err(StorageError::other("disk reference is required"));
+    }
+    Ok(rustfs_ecstore::api::storage::NativePreparationTarget {
+        disk_id: parse_native_uuid(disk_id, "disk_id")?,
+        store_id: parse_native_uuid(store_id, "store_id")?,
+        preparation: parse_native_uuid(preparation_id, "preparation_id")?,
+        bucket: bucket.to_owned(),
+        key: key.to_owned(),
+        bucket_incarnation: parse_native_uuid(bucket_incarnation_id, "bucket_incarnation_id")?,
+    })
+}
+
+fn native_receipt(
+    receipt: rustfs_ecstore::api::storage::NativePreparationReceipt,
+) -> std::result::Result<NativePreparationReceipt, StorageError> {
+    let (data_blocks, parity_blocks, shard_index) = receipt.erasure_geometry();
+    Ok(NativePreparationReceipt {
+        disk_id: receipt.disk_id().as_bytes().to_vec().into(),
+        store_id: receipt.deployment_id().as_bytes().to_vec().into(),
+        deployment_id: receipt.deployment_id().as_bytes().to_vec().into(),
+        preparation_id: receipt.preparation_id().as_bytes().to_vec().into(),
+        version_id: receipt.version_id().as_bytes().to_vec().into(),
+        bucket: receipt.bucket().to_owned(),
+        key: receipt.key().to_owned(),
+        bucket_incarnation_id: receipt.bucket_incarnation().as_bytes().to_vec().into(),
+        object_digest: receipt.object_digest().to_vec().into(),
+        object_length: receipt.object_length(),
+        set_digest: receipt.set_digest().to_vec().into(),
+        distribution_digest: receipt.distribution_digest().to_vec().into(),
+        metadata_digest: receipt.metadata_digest().to_vec().into(),
+        shard_digest: receipt.shard_digest().to_vec().into(),
+        shard_length: receipt.shard_length(),
+        data_blocks: u32::try_from(data_blocks).map_err(|_| StorageError::other("data_blocks exceeds protocol range"))?,
+        parity_blocks: u32::try_from(parity_blocks).map_err(|_| StorageError::other("parity_blocks exceeds protocol range"))?,
+        shard_index: u32::try_from(shard_index).map_err(|_| StorageError::other("shard_index exceeds protocol range"))?,
+    })
+}
+
+fn native_error(error: StorageError) -> Option<Error> {
+    let error = error.narrow_to_disk().unwrap_or_else(DiskError::other);
+    // The generic I/O wire code loses ErrorKind; corruption must remain fatal to quorum callers.
+    let error = match error {
+        DiskError::Io(ref error) if error.kind() == std::io::ErrorKind::InvalidData => DiskError::FileCorrupt,
+        error => error,
+    };
+    Some(error.into())
+}
+
+fn validate_native_metadata_length(
+    metadata: &[u8],
+    bucket: &str,
+    key: &str,
+    version: Uuid,
+    object_length: u64,
+) -> std::result::Result<(), StorageError> {
+    let mut file_meta = rustfs_filemeta::FileMeta::new();
+    let consumed = file_meta
+        .unmarshal_msg(metadata)
+        .map_err(|err| StorageError::other(format!("decode native preparation metadata: {err}")))?;
+    if usize::try_from(consumed).map_err(|_| StorageError::other("native preparation metadata length is invalid"))?
+        != metadata.len()
+    {
+        return Err(StorageError::other("native preparation metadata has trailing bytes"));
+    }
+    let (_, version_meta) = file_meta
+        .find_version(Some(version))
+        .map_err(|err| StorageError::other(format!("native preparation version is absent from metadata: {err}")))?;
+    let info = version_meta
+        .into_fileinfo(bucket, key, true)
+        .map_err(|err| StorageError::other(format!("decode native preparation file info: {err}")))?;
+    if u64::try_from(info.size).map_err(|_| StorageError::other("native preparation object length is invalid"))? != object_length
+    {
+        return Err(StorageError::other("object_length does not match persisted metadata"));
+    }
+    Ok(())
+}
+
+fn sign_native_response(body: std::result::Result<Vec<u8>, std::num::TryFromIntError>) -> std::result::Result<Bytes, Status> {
+    let body = body.map_err(|_| Status::internal("native preparation response length cannot be represented"))?;
+    sign_tonic_rpc_response_proof(&body)
+        .map(Bytes::from)
+        .map_err(|_| Status::unavailable("native preparation response proof is unavailable"))
+}
+
 fn snapshot_lease_response(result: Result<SnapshotLeaseToken, DiskError>) -> Response<SnapshotLeaseResponse> {
     match result {
         Ok(token) => Response::new(SnapshotLeaseResponse {
@@ -378,6 +482,314 @@ fn encode_rename_data_response_payloads(
 }
 
 impl NodeService {
+    pub(super) async fn handle_native_preparation_begin(
+        &self,
+        request: Request<NativePreparationBeginRequest>,
+    ) -> Result<Response<NativePreparationBeginResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_native_preparation_begin_request_body(request.get_ref()),
+            "native_preparation_begin",
+        )?;
+        let local_target = self.local_mutation_target();
+        let request = request.into_inner();
+        let result = async {
+            let target = native_target(
+                &request.disk,
+                &request.disk_id,
+                &request.store_id,
+                &request.preparation_id,
+                &request.bucket,
+                &request.key,
+                &request.bucket_incarnation_id,
+            )?;
+            let LocalMutationTarget::Ready(store) = local_target else {
+                return Err(StorageError::other("native preparation requires a ready storage instance"));
+            };
+            store.begin_native_preparation_on_disk(&target).await
+        }
+        .await;
+        let mut response = NativePreparationBeginResponse {
+            success: result.is_ok(),
+            error: result.err().and_then(native_error),
+            response_proof: Bytes::new(),
+        };
+        response.response_proof =
+            sign_native_response(rustfs_protos::canonical_native_preparation_begin_response_body(&request, &response))?;
+        Ok(Response::new(response))
+    }
+
+    pub(super) async fn handle_native_preparation_finalize(
+        &self,
+        request: Request<NativePreparationFinalizeRequest>,
+    ) -> Result<Response<NativePreparationFinalizeResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_native_preparation_finalize_request_body(request.get_ref()),
+            "native_preparation_finalize",
+        )?;
+        let local_target = self.local_mutation_target();
+        let request = request.into_inner();
+        let result = async {
+            let target = native_target(
+                &request.disk,
+                &request.disk_id,
+                &request.store_id,
+                &request.preparation_id,
+                &request.bucket,
+                &request.key,
+                &request.bucket_incarnation_id,
+            )?;
+            let LocalMutationTarget::Ready(store) = local_target else {
+                return Err(StorageError::other("native preparation requires a ready storage instance"));
+            };
+            if request.metadata.len() > rustfs_protos::NATIVE_PREPARATION_METADATA_MAX_BYTES {
+                return Err(StorageError::other("native preparation metadata exceeds size limit"));
+            }
+            let object_digest: [u8; 32] = request
+                .object_digest
+                .as_ref()
+                .try_into()
+                .map_err(|_| StorageError::other("object_digest must contain 32 bytes"))?;
+            let version = parse_native_uuid(&request.version_id, "version_id")?;
+            validate_native_metadata_length(&request.metadata, &target.bucket, &target.key, version, request.object_length)?;
+            let receipt = store
+                .finalize_native_preparation_on_disk(rustfs_ecstore::api::storage::NativePreparationFinalizeRequest {
+                    target,
+                    version,
+                    object_digest,
+                    metadata: request.metadata.clone(),
+                })
+                .await?;
+            native_receipt(receipt)
+        }
+        .await;
+        let (success, receipt, error) = match result {
+            Ok(receipt) => (true, Some(receipt), None),
+            Err(error) => (false, None, native_error(error)),
+        };
+        let mut response = NativePreparationFinalizeResponse {
+            success,
+            receipt,
+            error,
+            response_proof: Bytes::new(),
+        };
+        response.response_proof =
+            sign_native_response(rustfs_protos::canonical_native_preparation_finalize_response_body(&request, &response))?;
+        Ok(Response::new(response))
+    }
+
+    pub(super) async fn handle_native_preparation_probe(
+        &self,
+        request: Request<NativePreparationProbeRequest>,
+    ) -> Result<Response<NativePreparationProbeResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_native_preparation_probe_request_body(request.get_ref()),
+            "native_preparation_probe",
+        )?;
+        let local_target = self.local_mutation_target();
+        let request = request.into_inner();
+        let result = async {
+            let target = native_target(
+                &request.disk,
+                &request.disk_id,
+                &request.store_id,
+                &request.preparation_id,
+                &request.bucket,
+                &request.key,
+                &request.bucket_incarnation_id,
+            )?;
+            let LocalMutationTarget::Ready(store) = local_target else {
+                return Err(StorageError::other("native preparation requires a ready storage instance"));
+            };
+            store.probe_native_preparation_on_disk(&target).await
+        }
+        .await;
+        let (
+            success,
+            exists,
+            prepared,
+            version_id,
+            object_digest,
+            object_length,
+            set_digest,
+            distribution_digest,
+            metadata,
+            error,
+        ) = match result {
+            Ok(rustfs_ecstore::api::storage::NativePreparationProbe::Absent) => (
+                true,
+                false,
+                false,
+                Bytes::new(),
+                Bytes::new(),
+                0,
+                Bytes::new(),
+                Bytes::new(),
+                Bytes::new(),
+                None,
+            ),
+            Ok(rustfs_ecstore::api::storage::NativePreparationProbe::Existing {
+                version,
+                prepared,
+                object_digest,
+                object_length,
+                set_digest,
+                distribution_digest,
+                metadata,
+            }) => (
+                true,
+                true,
+                prepared,
+                version.as_bytes().to_vec().into(),
+                object_digest.to_vec().into(),
+                object_length,
+                set_digest.to_vec().into(),
+                distribution_digest.to_vec().into(),
+                metadata.unwrap_or_default(),
+                None,
+            ),
+            Err(error) => (
+                false,
+                false,
+                false,
+                Bytes::new(),
+                Bytes::new(),
+                0,
+                Bytes::new(),
+                Bytes::new(),
+                Bytes::new(),
+                native_error(error),
+            ),
+        };
+        let mut response = NativePreparationProbeResponse {
+            success,
+            exists,
+            prepared,
+            metadata,
+            error,
+            response_proof: Bytes::new(),
+            version_id,
+            object_digest,
+            object_length,
+            set_digest,
+            distribution_digest,
+        };
+        response.response_proof =
+            sign_native_response(rustfs_protos::canonical_native_preparation_probe_response_body(&request, &response))?;
+        Ok(Response::new(response))
+    }
+
+    pub(super) async fn handle_native_preparation_recover(
+        &self,
+        request: Request<NativePreparationRecoverRequest>,
+    ) -> Result<Response<NativePreparationRecoverResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_native_preparation_recover_request_body(request.get_ref()),
+            "native_preparation_recover",
+        )?;
+        let local_target = self.local_mutation_target();
+        let request = request.into_inner();
+        let result = async {
+            let target = native_target(
+                &request.disk,
+                &request.disk_id,
+                &request.store_id,
+                &request.preparation_id,
+                &request.bucket,
+                &request.key,
+                &request.bucket_incarnation_id,
+            )?;
+            let LocalMutationTarget::Ready(store) = local_target else {
+                return Err(StorageError::other("native preparation requires a ready storage instance"));
+            };
+            let recovered = store.recover_native_preparation_on_disk(&target).await?;
+            Ok::<_, StorageError>((native_receipt(recovered.receipt)?, recovered.metadata))
+        }
+        .await;
+        let (success, receipt, metadata, error) = match result {
+            Ok((receipt, metadata)) => (true, Some(receipt), metadata, None),
+            Err(error) => (false, None, Bytes::new(), native_error(error)),
+        };
+        let mut response = NativePreparationRecoverResponse {
+            success,
+            receipt,
+            metadata,
+            error,
+            response_proof: Bytes::new(),
+        };
+        response.response_proof =
+            sign_native_response(rustfs_protos::canonical_native_preparation_recover_response_body(&request, &response))?;
+        Ok(Response::new(response))
+    }
+
+    pub(super) async fn handle_native_preparation_locate(
+        &self,
+        request: Request<NativePreparationLocateRequest>,
+    ) -> Result<Response<NativePreparationLocateResponse>, Status> {
+        verify_disk_mutation_digest(
+            &request,
+            rustfs_protos::canonical_native_preparation_locate_request_body(request.get_ref()),
+            "native_preparation_locate",
+        )?;
+        let local_target = self.local_mutation_target();
+        let request = request.into_inner();
+        let result = async {
+            if request.disk.is_empty() {
+                return Err(StorageError::other("disk reference is required"));
+            }
+            let disk_id = parse_native_uuid(&request.disk_id, "disk_id")?;
+            let store_id = parse_native_uuid(&request.store_id, "store_id")?;
+            let preparation_id = parse_native_uuid(&request.preparation_id, "preparation_id")?;
+            let LocalMutationTarget::Ready(store) = local_target else {
+                return Err(StorageError::other("native preparation requires a ready storage instance"));
+            };
+            store
+                .locate_native_preparation_on_disk(disk_id, store_id, preparation_id)
+                .await
+        }
+        .await;
+        let (success, found, bucket, key, incarnation, metadata, prepared, error) = match result {
+            Ok(Some(description)) => (
+                true,
+                true,
+                description.bucket,
+                description.key,
+                description.incarnation.as_bytes().to_vec().into(),
+                description.metadata,
+                description.prepared,
+                None,
+            ),
+            Ok(None) => (true, false, String::new(), String::new(), Bytes::new(), Bytes::new(), false, None),
+            Err(error) => (
+                false,
+                false,
+                String::new(),
+                String::new(),
+                Bytes::new(),
+                Bytes::new(),
+                false,
+                native_error(error),
+            ),
+        };
+        let mut response = NativePreparationLocateResponse {
+            success,
+            found,
+            bucket,
+            key,
+            bucket_incarnation_id: incarnation,
+            metadata,
+            prepared,
+            error,
+            response_proof: Bytes::new(),
+        };
+        response.response_proof =
+            sign_native_response(rustfs_protos::canonical_native_preparation_locate_response_body(&request, &response))?;
+        Ok(Response::new(response))
+    }
+
     pub(super) async fn handle_acquire_snapshot_lease(
         &self,
         request: Request<SnapshotLeaseRequest>,