fix(ecstore): reconcile object cleanup receipts (#6077)

* fix(s3): keep multipart completion publication owned

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(s3): keep put publication owned

Co-Authored-By: heihutu <heihutu@gmail.com>

* chore(app): route multipart context through facade

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): gate object transaction fencing

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): fence object transaction epochs

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(ecstore): reconcile old data cleanup receipts

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/config/src/constants/object.rs

@@ -137,6 +137,22 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE);
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED);
 
+/// Request the object-transaction fencing contract used by storage-owned
+/// cleanup receipts and lock-window optimizations.
+///
+/// This is fail-closed: enabling the writer without a live fleet proof rejects
+/// the commit rather than silently using a legacy-safe path.
+pub const ENV_OBJECT_TRANSACTION_FENCING_WRITE: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE";
+pub const DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE: bool = false;
+
+/// Operator-attested confirmation that every serving node understands the
+/// object transaction fencing contract.
+pub const ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED";
+pub const DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: bool = false;
+
+const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE);
+const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED);
+
 /// Request preserving legacy per-part checksum metadata during data movement.
 ///
 /// This remains ineffective until
@@ -673,4 +689,13 @@ mod remote_version_state_tests {
             "RUSTFS_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED"
         );
     }
+
+    #[test]
+    fn object_transaction_fencing_gate_uses_stable_environment_names() {
+        assert_eq!(super::ENV_OBJECT_TRANSACTION_FENCING_WRITE, "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE");
+        assert_eq!(
+            super::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED,
+            "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED"
+        );
+    }
 }

crates/ecstore/src/services/notification_sys.rs

@@ -206,6 +206,38 @@ pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStat
     })
 }
 
+#[cfg(test)]
+pub(crate) struct RemoteVersionStateFleetProofGuard;
+
+#[cfg(test)]
+impl Drop for RemoteVersionStateFleetProofGuard {
+    fn drop(&mut self) {
+        replace_remote_version_state_fleet_proof(None);
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn install_remote_version_state_fleet_proof_for_test(topology_fingerprint: &str) -> RemoteVersionStateFleetProofGuard {
+    match REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.to_string()) {
+        Ok(()) => {}
+        Err(_)
+            if REMOTE_VERSION_STATE_PROBE_TOPOLOGY
+                .get()
+                .is_some_and(|current| current == topology_fingerprint) => {}
+        Err(_) => panic!("remote version state test topology is already bound to another fingerprint"),
+    }
+    let peer_epochs = BTreeMap::new();
+    if let Some(err) = publish_remote_version_state_probe_result(
+        remote_version_state_fleet_proof_slot(),
+        topology_fingerprint,
+        Ok(peer_epochs),
+        Instant::now(),
+    ) {
+        panic!("test proof installation must not fail: {err}");
+    }
+    RemoteVersionStateFleetProofGuard
+}
+
 fn remote_version_state_fleet_proof_valid_at(
     proof: Option<&RemoteVersionStateFleetProof>,
     expected_topology: &str,

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

@@ -1425,6 +1425,33 @@ impl SetDisks {
     /// post-heal tail — reclaim identically. Never fails the heal: delete errors
     /// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
     async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
+        match self.reconcile_old_data_cleanup_receipts(bucket, object).await {
+            Ok(removed) if removed > 0 => {
+                debug!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket,
+                    object,
+                    removed,
+                    state = "old_data_cleanup_receipt_reconciled",
+                    "Set disk old-data cleanup receipts reconciled"
+                );
+            }
+            Ok(_) => {}
+            Err(e) => {
+                warn!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket,
+                    object,
+                    error = %e,
+                    state = "old_data_cleanup_receipt_reconcile_failed",
+                    "Set disk old-data cleanup receipt reconcile failed"
+                );
+            }
+        }
         match self.reclaim_orphan_data_dirs(bucket, object).await {
             Ok(removed) if removed > 0 => {
                 debug!(

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

@@ -22,6 +22,11 @@
 
 use super::super::*;
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
+use super::object::{
+    assign_object_transaction_epoch, object_transaction_fencing_fleet_proof, object_transaction_fencing_fleet_proof_matches,
+    object_transaction_fencing_requested, old_data_cleanup_receipt_path, read_object_transaction_epoch_fence,
+    verify_object_transaction_epoch_fence,
+};
 use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
@@ -63,6 +68,7 @@ pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
+    BeforeTransactionEpochVerify,
     AfterRename,
 }
 
@@ -153,13 +159,24 @@ impl Drop for MultipartCommitBarrier {
 
 #[cfg(test)]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
-    let barrier = MULTIPART_COMMIT_BARRIER
+    let barrier = {
-        .get_or_init(|| std::sync::Mutex::new(None))
+        let mut slot = MULTIPART_COMMIT_BARRIER
-        .lock()
+            .get_or_init(|| std::sync::Mutex::new(None))
-        .expect("multipart commit barrier mutex should not poison")
+            .lock()
-        .as_ref()
+            .expect("multipart commit barrier mutex should not poison");
-        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
+        if slot
-        .cloned();
+            .as_ref()
+            .is_some_and(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
+        {
+            if pause == MultipartCommitPause::BeforeTransactionEpochVerify {
+                slot.take()
+            } else {
+                slot.clone()
+            }
+        } else {
+            None
+        }
+    };
     if let Some(barrier) = barrier
         && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
             (current < barrier.expected_arrivals).then_some(current + 1)
@@ -2296,6 +2313,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
+        let transaction_fencing_proof = object_transaction_fencing_fleet_proof();
+        if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() {
+            return Err(Error::other("object transaction fencing requires a live fleet capability proof"));
+        }
+        let transaction_epoch_fence = if transaction_fencing_proof.is_some() {
+            Some(read_object_transaction_epoch_fence(self.as_ref(), bucket, object).await?)
+        } else {
+            None
+        };
+        let transaction_epoch =
+            transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas));
+
         let commit_set = self.clone();
         let commit_bucket = bucket.to_owned();
         let commit_object = object.to_owned();
@@ -2323,6 +2352,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             // The trailing `_` drops the rename_data old-size backfill
             // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
             // `get_object_info` lookup, so the backfill has no consumer here yet.
+            if let Some(proof) = transaction_fencing_proof.as_ref()
+                && !object_transaction_fencing_fleet_proof_matches(proof)
+            {
+                return Err(Error::other(
+                    "object transaction fencing fleet capability changed during complete_multipart_upload",
+                ));
+            }
+            if let Some(expected) = transaction_epoch_fence {
+                #[cfg(test)]
+                pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeTransactionEpochVerify).await;
+                verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?;
+            }
             let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
                 &shuffle_disks,
                 RUSTFS_META_MULTIPART_BUCKET,
@@ -2354,6 +2395,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 });
             }
 
+            if let Some(old_dir) = op_old_dir {
+                commit_set
+                    .persist_old_data_cleanup_receipts(
+                        &cleanup_disks,
+                        &commit_bucket,
+                        &commit_object,
+                        old_dir,
+                        fi.data_dir,
+                        transaction_epoch,
+                    )
+                    .await;
+            }
+
             // Crash-consistency injection: hard power loss after the authoritative
             // rename_data commit succeeded but before the stale part.N.meta cleanup.
             // The new version is durably committed and visible, so a crash here must
@@ -2469,9 +2523,10 @@ fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart])
 mod tests {
     use super::*;
     use crate::config::storageclass::lookup_config_for_pools_without_env;
-    use crate::disk::DiskAPI as _;
+    use crate::disk::{DiskAPI as _, ReadOptions};
     use crate::disk::{endpoint::Endpoint, format::FormatV3};
     use crate::layout::endpoints::SetupType;
+    use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test;
     // No-locker helpers resolve to the isolated-context variants (see
     // `hermetic_set_disks_isolated`); the guard-based tests build through
     // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context
@@ -2882,6 +2937,208 @@ mod tests {
         }
     }
 
+    async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec<Option<Uuid>> {
+        let mut epochs = Vec::with_capacity(disks.len());
+        for (disk_index, disk) in disks.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}"));
+            epochs.push(
+                file_info
+                    .object_transaction_epoch()
+                    .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")),
+            );
+        }
+        epochs
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_requires_live_fleet_proof_before_multipart_commit() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-transaction-fencing-no-proof";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(
+            &set_disks,
+            bucket,
+            object,
+            b"must-not-complete-without-proof",
+            &ObjectOptions::default(),
+        )
+        .await;
+
+        let err = temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                set_disks
+                    .clone()
+                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
+                    .await
+            },
+        )
+        .await
+        .expect_err("multipart completion must fail closed without a live fleet proof");
+
+        assert!(
+            err.to_string()
+                .contains("object transaction fencing requires a live fleet capability proof"),
+            "unexpected error: {err:?}"
+        );
+        set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect_err("failed fenced completion must not publish object metadata");
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_persists_epoch_on_multipart_commit() {
+        let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-object-transaction-epoch";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) =
+            stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                set_disks
+                    .clone()
+                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
+                    .await
+                    .expect("fenced multipart completion should commit with a live proof");
+            },
+        )
+        .await;
+
+        let epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
+        let first = epochs[0].expect("fenced multipart completion should persist an epoch");
+        assert!(!first.is_nil());
+        assert!(epochs.into_iter().all(|epoch| epoch == Some(first)));
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_rejects_stale_multipart_epoch() {
+        let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-object-transaction-stale-epoch";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                let mut initial_reader = PutObjReader::from_vec(b"initial fenced object".to_vec());
+                set_disks
+                    .put_object(
+                        bucket,
+                        object,
+                        &mut initial_reader,
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("initial fenced PUT should commit");
+                let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object)
+                    .await
+                    .into_iter()
+                    .next()
+                    .flatten()
+                    .expect("initial fenced PUT should persist an epoch");
+
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, b"stale multipart body", &ObjectOptions::default())
+                        .await;
+                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeTransactionEpochVerify);
+                let stale_set = Arc::clone(&set_disks);
+                let stale = tokio::spawn(async move {
+                    stale_set
+                        .clone()
+                        .complete_multipart_upload(
+                            bucket,
+                            object,
+                            &upload_id,
+                            parts,
+                            &ObjectOptions {
+                                no_lock: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                barrier.wait_until_paused().await;
+
+                let mut winner_reader = PutObjReader::from_vec(b"winning put body".to_vec());
+                set_disks
+                    .put_object(
+                        bucket,
+                        object,
+                        &mut winner_reader,
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("concurrent fenced PUT should advance the epoch");
+                let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object)
+                    .await
+                    .into_iter()
+                    .next()
+                    .flatten()
+                    .expect("winning fenced PUT should persist an epoch");
+                assert_ne!(winning_epoch, initial_epoch);
+
+                barrier.release();
+                let err = stale
+                    .await
+                    .expect("stale multipart task should not panic")
+                    .expect_err("stale epoch multipart completion must be rejected");
+                assert_eq!(err, StorageError::PreconditionFailed);
+
+                let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
+                assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch)));
+                let mut reader = set_disks
+                    .get_object_reader(
+                        bucket,
+                        object,
+                        None,
+                        HeaderMap::new(),
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("winning object should remain readable");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("winning body should stream");
+                assert_eq!(restored, b"winning put body");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn complete_multipart_quota_rejection_preserves_destination_and_upload() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -6193,6 +6450,24 @@ mod tests {
             (body, etag)
         }
 
+        async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid {
+            disk.read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("current object metadata should read")
+                .data_dir
+                .expect("test object should be stored out-of-line")
+        }
+
+        async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool {
+            disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok()
+        }
+
+        async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool {
+            disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir))
+                .await
+                .is_ok()
+        }
+
         async fn upload_is_listed(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, upload_id: &str) -> bool {
             let page = set_disks
                 .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None)
@@ -6313,6 +6588,106 @@ mod tests {
             let (body_after, _) = read_object(&set_disks, bucket, object).await;
             assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object");
         }
+
+        #[tokio::test]
+        #[serial(storage_class_env)]
+        async fn post_commit_crash_receipt_reclaims_old_data_after_restart() {
+            let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+            let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let bucket = "multipart-crash-old-data-receipt";
+            let object = "crash-old-data-object";
+            make_bucket_on_all(&disk_stores, bucket).await;
+
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                    (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+                ],
+                async {
+                    let old = payload(0x51);
+                    let (u_old, parts_old) = stage_upload(&set_disks, bucket, object, &old).await;
+                    complete(&set_disks, bucket, object, &u_old, parts_old)
+                        .await
+                        .expect("the old version should commit");
+                    let old_dir = current_data_dir(&disk_stores[0], bucket, object).await;
+
+                    let new = payload(0x52);
+                    let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
+                    crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
+                    let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await;
+                    assert!(
+                        matches!(crashed, Err(StorageError::Unexpected)),
+                        "the post-commit crash point must surface as unexpected, got {crashed:?}"
+                    );
+                    crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
+
+                    let (body, _) = read_object(&set_disks, bucket, object).await;
+                    assert_eq!(body, new, "the committed replacement must remain readable after the crash");
+                    for disk in &disk_stores {
+                        assert!(
+                            cleanup_receipt_exists(disk, bucket, object, old_dir).await,
+                            "post-commit crash must leave a durable old-data cleanup receipt"
+                        );
+                        assert!(
+                            data_dir_exists(disk, bucket, object, old_dir).await,
+                            "post-commit crash must leave old data for restart reconciliation"
+                        );
+                    }
+
+                    let restarted_endpoints = temp_dirs
+                        .iter()
+                        .enumerate()
+                        .map(|(disk_idx, dir)| {
+                            let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8"))
+                                .expect("endpoint should parse");
+                            endpoint.set_pool_index(0);
+                            endpoint.set_set_index(0);
+                            endpoint.set_disk_index(disk_idx);
+                            endpoint
+                        })
+                        .collect::<Vec<_>>();
+                    let mut reloaded = Vec::with_capacity(restarted_endpoints.len());
+                    for endpoint in &restarted_endpoints {
+                        reloaded.push(
+                            new_disk(
+                                endpoint,
+                                &DiskOption {
+                                    cleanup: false,
+                                    health_check: false,
+                                },
+                            )
+                            .await
+                            .expect("disk should restart"),
+                        );
+                    }
+                    let restarted_set = SetDisks::new_with_instance_ctx(
+                        "restart-cleanup-receipt-test-owner".to_string(),
+                        Arc::new(RwLock::new(reloaded.iter().cloned().map(Some).collect())),
+                        4,
+                        2,
+                        0,
+                        0,
+                        restarted_endpoints,
+                        set_disks.format.clone(),
+                        Vec::new(),
+                        Arc::new(crate::runtime::instance::InstanceContext::new()),
+                    )
+                    .await;
+                    let removed = restarted_set
+                        .reconcile_old_data_cleanup_receipts(bucket, object)
+                        .await
+                        .expect("restart receipt reconciliation should succeed");
+                    assert_eq!(removed, 4, "restart reconciler should delete all receipt targets");
+                    for disk in &reloaded {
+                        assert!(
+                            !data_dir_exists(disk, bucket, object, old_dir).await,
+                            "restart reconciler must reclaim the old data dir"
+                        );
+                    }
+                },
+            )
+            .await;
+        }
     }
 
     #[test]

crates/ecstore/src/set_disk/replication.rs

@@ -48,6 +48,23 @@ impl RestoreCleanupIdentity {
     }
 }
 
+fn ensure_restore_metadata_lock_held(bucket: &str, object: &str, opts: &ObjectOptions, mode: &'static str) -> Result<()> {
+    if opts
+        .namespace_lock_fence
+        .as_ref()
+        .is_some_and(NamespaceLockFence::is_lock_lost)
+    {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode,
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    Ok(())
+}
+
 impl SetDisks {
     pub(super) async fn finalize_restore_metadata(
         &self,
@@ -88,6 +105,7 @@ impl SetDisks {
         if !expected.matches_file_info(&fi, &expected_etag) {
             return Err(Error::other("restored object changed before restore metadata finalization"));
         }
+        ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?;
         let restore_expiry =
             lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1));
         fi.metadata.insert(
@@ -159,6 +177,7 @@ impl SetDisks {
         if !expected.matches_file_info(&fi, &expected_etag) {
             return Ok(());
         }
+        ensure_restore_metadata_lock_held(bucket, object, opts, "restore_cleanup_metadata")?;
         fi.metadata.remove(X_AMZ_RESTORE.as_str());
         fi.metadata.remove(AMZ_RESTORE_EXPIRY_DAYS);
         fi.metadata.remove(AMZ_RESTORE_REQUEST_DATE);

crates/filemeta/src/fileinfo.rs

@@ -18,8 +18,9 @@ use rmp_serde::Serializer;
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::http::{
     AMZ_OBJECT_TAGGING, SUFFIX_COMPRESSION, SUFFIX_DATA_MOVED, SUFFIX_DATA_MOVED_TAGS, SUFFIX_FREE_VERSION, SUFFIX_HEALING,
-    SUFFIX_INLINE_DATA, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, contains_key_str, get_str,
+    SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID,
-    has_internal_suffix, insert_str, is_encryption_metadata_key, starts_with_ignore_ascii_case,
+    contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key,
+    starts_with_ignore_ascii_case,
 };
 use s3s::dto::{RestoreStatus, Timestamp};
 use s3s::header::X_AMZ_RESTORE;
@@ -1172,6 +1173,22 @@ impl FileInfo {
         insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, String::new());
     }
 
+    pub fn set_object_transaction_epoch(&mut self, epoch: Uuid) {
+        insert_str(&mut self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH, epoch.to_string());
+    }
+
+    pub fn object_transaction_epoch(&self) -> Result<Option<Uuid>> {
+        if !contains_key_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH) {
+            return Ok(None);
+        }
+        let value = get_consistent_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH).ok_or(Error::FileCorrupt)?;
+        let epoch = Uuid::parse_str(value).map_err(|_| Error::FileCorrupt)?;
+        if epoch.is_nil() {
+            return Err(Error::FileCorrupt);
+        }
+        Ok(Some(epoch))
+    }
+
     pub fn inline_data(&self) -> bool {
         contains_key_str(&self.metadata, SUFFIX_INLINE_DATA) && !self.is_remote()
     }
@@ -1484,6 +1501,46 @@ mod tests {
         assert_eq!(ei.get_checksum_info(99).algorithm, HashAlgorithm::HighwayHash256S);
     }
 
+    #[test]
+    fn object_transaction_epoch_uses_consistent_dual_internal_metadata() {
+        let mut fi = validation_test_fileinfo();
+        assert_eq!(fi.object_transaction_epoch().expect("absent epoch should decode"), None);
+
+        let epoch = Uuid::new_v4();
+        let epoch_text = epoch.to_string();
+        fi.set_object_transaction_epoch(epoch);
+        assert_eq!(fi.object_transaction_epoch().expect("written epoch should decode"), Some(epoch));
+        assert_eq!(fi.metadata.get("x-rustfs-internal-object-transaction-epoch"), Some(&epoch_text));
+        assert_eq!(fi.metadata.get("x-minio-internal-object-transaction-epoch"), Some(&epoch_text));
+
+        let mut rustfs_only = validation_test_fileinfo();
+        rustfs_only
+            .metadata
+            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), epoch_text);
+        assert_eq!(
+            rustfs_only
+                .object_transaction_epoch()
+                .expect("single compatibility key should decode"),
+            Some(epoch)
+        );
+
+        let mut conflicting = fi.clone();
+        conflicting
+            .metadata
+            .insert("x-minio-internal-object-transaction-epoch".to_string(), Uuid::new_v4().to_string());
+        assert_eq!(conflicting.object_transaction_epoch(), Err(Error::FileCorrupt));
+
+        let mut malformed = validation_test_fileinfo();
+        malformed
+            .metadata
+            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), "not-a-uuid".to_string());
+        assert_eq!(malformed.object_transaction_epoch(), Err(Error::FileCorrupt));
+
+        let mut nil = validation_test_fileinfo();
+        nil.set_object_transaction_epoch(Uuid::nil());
+        assert_eq!(nil.object_transaction_epoch(), Err(Error::FileCorrupt));
+    }
+
     // backlog#949: distribution range/permutation validation.
     #[test]
     fn is_valid_distribution_accepts_permutation() {

crates/utils/src/http/metadata_compat.rs

@@ -59,6 +59,7 @@ pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destina
 pub const SUFFIX_TRANSITION_TRANSACTION_ID: &str = "transition-transaction-id";
 pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id";
 pub const SUFFIX_BUCKET_INCARNATION_ID: &str = "bucket-incarnation-id";
+pub const SUFFIX_OBJECT_TRANSACTION_EPOCH: &str = "object-transaction-epoch";
 pub const SUFFIX_FREE_VERSION: &str = "free-version";
 pub const SUFFIX_PURGESTATUS: &str = "purgestatus";
 pub const SUFFIX_REPLICA_STATUS: &str = "replica-status";

rustfs/src/app/multipart_usecase.rs

@@ -46,6 +46,7 @@ use super::storage_api::multipart_usecase::options::{
     get_content_sha256_with_query, get_opts, namespace_reserved_user_metadata, parse_copy_source_range,
     put_opts_with_replication_authorization, validate_archive_content_encoding,
 };
+use super::storage_api::multipart_usecase::request_context::spawn_traced_join;
 use super::storage_api::multipart_usecase::s3_api::multipart::{
     ListMultipartUploadsParams, build_list_multipart_uploads_output, build_list_parts_output,
     parse_list_multipart_uploads_params, parse_list_parts_params, parse_upload_part_number,
@@ -588,56 +589,94 @@ impl DefaultMultipartUsecase {
             None => None,
         };
 
-        let obj_info = store
+        let complete_commit = spawn_traced_join({
-            .clone()
+            let store = Arc::clone(&store);
-            .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts)
+            let bucket = bucket.clone();
-            .await
+            let key = key.clone();
-            .map_err(ApiError::from)?;
+            let upload_id = upload_id.clone();
-        let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
+            let opts = opts.clone();
-        record_capacity_write(Some(capacity_scope_token)).await;
+            let quota_metadata_sys = quota_metadata_sys.clone();
-
+            async move {
-        if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
+                let obj_info = store
-            if opts.replication_request {
+                    .clone()
-                let quota_checker = QuotaChecker::new(metadata_sys.clone());
+                    .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts)
-                match quota_checker
-                    .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64)
                     .await
-                {
+                    .map_err(ApiError::from)?;
-                    Ok(check_result) if !check_result.allowed => {
+                let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
-                        let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await;
+                record_capacity_write(Some(capacity_scope_token)).await;
-                        let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await;
+
-                        return Err(S3Error::with_message(
+                if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
-                            S3ErrorCode::InvalidRequest,
+                    if opts.replication_request {
-                            format!(
+                        let quota_checker = QuotaChecker::new(metadata_sys.clone());
-                                "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes",
+                        match quota_checker
-                                check_result.current_usage.unwrap_or(0),
+                            .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64)
-                                check_result.quota_limit.unwrap_or(0)
+                            .await
-                            ),
+                        {
-                        ));
+                            Ok(check_result) if !check_result.allowed => {
+                                let _ = store.delete_object(&bucket, &key, ObjectOptions::default()).await;
+                                let _ = invalidate_object_data_cache_after_delete_success(&cache_adapter, &bucket, &key).await;
+                                return Err(S3Error::with_message(
+                                    S3ErrorCode::InvalidRequest,
+                                    format!(
+                                        "Bucket quota exceeded. Current usage: {} bytes, limit: {} bytes",
+                                        check_result.current_usage.unwrap_or(0),
+                                        check_result.quota_limit.unwrap_or(0)
+                                    ),
+                                ));
+                            }
+                            Err(err) => {
+                                warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err);
+                            }
+                            Ok(_) => {}
+                        }
                     }
-                    Err(err) => {
+
-                        warn!("Quota check failed for bucket {} after multipart completion: {}", bucket, err);
+                    let committed_size = if opts.replication_request {
+                        obj_info.size.max(0) as u64
+                    } else {
+                        quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())?
+                    };
+                    if versioned {
+                        record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
+                    } else {
+                        record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
                     }
-                    Ok(_) => {}
                 }
+
+                enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await;
+
+                let mt2 = obj_info.user_defined.clone();
+                let dsc = must_replicate_object(
+                    &bucket,
+                    &key,
+                    &mt2,
+                    "".to_string(),
+                    opts.delete_marker_replication_status(),
+                    opts.clone(),
+                )
+                .await;
+
+                if dsc.replicate_any() {
+                    warn!("need multipart replication");
+                    schedule_object_replication(obj_info.clone(), store, dsc).await;
+                }
+
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                Ok::<_, S3Error>(obj_info)
             }
+        });
+        let obj_info = complete_commit.await.map_err(|err| {
+            S3Error::with_message(
+                S3ErrorCode::InternalError,
+                format!("complete multipart upload commit owner task failed: {err}"),
+            )
+        })??;
 
-            let committed_size = if opts.replication_request {
+        let mpu_version = if versioned {
-                obj_info.size.max(0) as u64
+            obj_info.version_id.map(|v| v.to_string())
-            } else {
+        } else {
-                quota_accounting_object_size(&obj_info, opts.quota_admission.is_some())?
+            None
-            };
+        };
-            if versioned {
-                record_bucket_object_version_write_memory(&bucket, previous_current_size, committed_size).await;
-            } else {
-                record_bucket_object_write_memory(&bucket, previous_current_size, committed_size).await;
-            }
-        }
-
-        enqueue_transition_immediate(&obj_info, LcEventSrc::S3CompleteMultipartUpload).await;
-
-        let raw_mpu_version = obj_info.version_id.map(|v| v.to_string());
-        let mpu_version = if versioned { raw_mpu_version.clone() } else { None };
         let mpu_version_for_event = mpu_version.clone();
         // checksum: stored (decrypted) values take precedence over the request input;
         // additional algorithms (XXHash3/64/128, SHA-512, MD5), which have no typed
@@ -660,28 +699,18 @@ impl DefaultMultipartUsecase {
             bucket: Some(bucket.clone()),
             key: Some(key.clone()),
             e_tag: obj_info.etag.clone().map(|etag| to_s3s_etag(&etag)),
-            location: Some(location.clone()),
+            location: Some(location),
             server_side_encryption: server_side_encryption.clone(),
             ssekms_key_id: ssekms_key_id.clone(),
-            checksum_crc32: checksum_crc32.clone(),
+            checksum_crc32,
-            checksum_crc32c: checksum_crc32c.clone(),
+            checksum_crc32c,
-            checksum_sha1: checksum_sha1.clone(),
+            checksum_sha1,
-            checksum_sha256: checksum_sha256.clone(),
+            checksum_sha256,
-            checksum_crc64nvme: checksum_crc64nvme.clone(),
+            checksum_crc64nvme,
-            checksum_type: checksum_type.clone(),
+            checksum_type,
             version_id: mpu_version,
             ..Default::default()
         };
-        let mt2 = obj_info.user_defined.clone();
-        let dsc =
-            must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone())
-                .await;
-
-        if dsc.replicate_any() {
-            warn!("need multipart replication");
-            schedule_object_replication(obj_info.clone(), store, dsc).await;
-        }
-
         // Set object info for event notification
         helper = helper.object(obj_info);
         if let Some(version_id) = &mpu_version_for_event {
@@ -712,7 +741,6 @@ impl DefaultMultipartUsecase {
         }
         let result = Ok(response);
         let _ = helper.complete(&result);
-        rustfs_scanner::record_dirty_usage_bucket(&bucket);
         result
     }
 

rustfs/src/app/object_usecase.rs

@@ -2989,6 +2989,11 @@ struct PutObjectChecksums {
     crc64nvme: Option<String>,
 }
 
+struct PutObjectCommitResult {
+    obj_info: ObjectInfo,
+    put_versioned: bool,
+}
+
 fn normalize_delete_objects_version_id(
     version_id: Option<String>,
 ) -> std::result::Result<(Option<String>, Option<Uuid>), String> {
@@ -5932,7 +5937,7 @@ impl DefaultObjectUsecase {
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
         rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start);
 
-        let mut reader = PutObjReader::new(reader);
+        let reader = PutObjReader::new(reader);
 
         let mt2 = metadata.clone();
         opts.user_defined.extend(metadata);
@@ -6005,97 +6010,145 @@ impl DefaultObjectUsecase {
         } else {
             None
         };
-        let object_traffic_progress = object_traffic_health
+        let put_commit = spawn_traced_join({
-            .as_deref()
+            let store = Arc::clone(&store);
-            .and_then(ObjectTrafficHealth::track_write_storage);
+            let bucket = bucket.clone();
-        let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now);
+            let key = key.clone();
-        let (obj_info, backfilled_old_current_size) = match store
+            let opts = opts.clone();
-            .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
+            let cache_adapter = cache_adapter.clone();
-            .await
+            let request_id = request_id.clone();
-            .map_err(ApiError::from)
+            let put_path = put_path.to_string();
-        {
+            async move {
-            Ok(obj_info) => {
+                let object_traffic_progress = object_traffic_health
-                store_put_watchdog.cancel();
+                    .as_deref()
-                debug!(
+                    .and_then(ObjectTrafficHealth::track_write_storage);
-                    target: "rustfs::app::object_usecase",
+                let mut reader = reader;
-                    event = EVENT_PUT_OBJECT_STORE_RETURNED,
+                let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now);
-                    component = LOG_COMPONENT_APP,
+                let (obj_info, backfilled_old_current_size) = match store
-                    subsystem = LOG_SUBSYSTEM_OBJECT,
+                    .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
-                    request_id = %request_id,
+                    .await
-                    bucket = %bucket,
+                    .map_err(ApiError::from)
-                    key = %key,
+                {
-                    put_path = put_path,
+                    Ok(obj_info) => {
-                    object_size = actual_size,
+                        store_put_watchdog.cancel();
-                    duration_ms = start_time.elapsed().as_millis() as u64,
+                        debug!(
-                    result = "success",
+                            target: "rustfs::app::object_usecase",
-                    "PutObject store write returned"
+                            event = EVENT_PUT_OBJECT_STORE_RETURNED,
-                );
+                            component = LOG_COMPONENT_APP,
-                obj_info
+                            subsystem = LOG_SUBSYSTEM_OBJECT,
+                            request_id = %request_id,
+                            bucket = %bucket,
+                            key = %key,
+                            put_path = %put_path,
+                            object_size = actual_size,
+                            duration_ms = start_time.elapsed().as_millis() as u64,
+                            result = "success",
+                            "PutObject store write returned"
+                        );
+                        obj_info
+                    }
+                    Err(err) => {
+                        store_put_watchdog.cancel();
+                        rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
+                        warn!(
+                            target: "rustfs::app::object_usecase",
+                            event = EVENT_PUT_OBJECT_STORE_RETURNED,
+                            component = LOG_COMPONENT_APP,
+                            subsystem = LOG_SUBSYSTEM_OBJECT,
+                            request_id = %request_id,
+                            bucket = %bucket,
+                            key = %key,
+                            put_path = %put_path,
+                            object_size = actual_size,
+                            duration_ms = start_time.elapsed().as_millis() as u64,
+                            result = "error",
+                            error = %err,
+                            "PutObject store write returned"
+                        );
+                        return Err(err.into());
+                    }
+                };
+                rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
+                drop(object_traffic_progress);
+                #[cfg(test)]
+                wait_for_put_post_store_test_hook(&bucket).await;
+
+                let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now);
+                maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await;
+                let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await;
+
+                let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
+                // Fast in-memory update for immediate quota and admin usage consistency.
+                // The previous current size comes from the prelookup when it ran,
+                // otherwise from the rename_data backfill (rustfs/backlog#1009); the
+                // backfill reproduces the lookup's observation bit for bit (latest
+                // version's ObjectInfo.size — 0 for a delete-marker latest — or
+                // not-found → None).
+                match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size))
+                {
+                    Some(previous_current_size) => {
+                        if put_versioned {
+                            record_bucket_object_version_write_memory(
+                                &bucket,
+                                previous_current_size,
+                                obj_info.size.max(0) as u64,
+                            )
+                            .await;
+                        } else {
+                            record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
+                        }
+                    }
+                    None => {
+                        // Neither source could determine the previous state (peers
+                        // predating the backfill field during a rolling upgrade, or
+                        // sub-quorum metadata divergence). Record the components that
+                        // are correct regardless; the next authoritative scanner
+                        // refresh replaces the in-memory numbers.
+                        debug!(
+                            target: "rustfs::app::object_usecase",
+                            bucket = %bucket,
+                            key = %key,
+                            put_versioned,
+                            "put_object old-size backfill unknown; recording degraded usage delta"
+                        );
+                        record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned)
+                            .await;
+                    }
+                }
+
+                if dsc.replicate_any() {
+                    schedule_object_replication(obj_info.clone(), store, dsc).await;
+                }
+
+                rustfs_scanner::record_dirty_usage_bucket(&bucket);
+                rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start);
+
+                let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now);
+                let manager = get_capacity_manager();
+                manager.record_write_operation().await;
+                rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start);
+
+                Ok::<_, S3Error>(PutObjectCommitResult { obj_info, put_versioned })
+            }
+        });
+        let PutObjectCommitResult { obj_info, put_versioned } = match put_commit.await {
+            Ok(Ok(result)) => result,
+            Ok(Err(err)) => {
+                let result: S3Result<S3Response<PutObjectOutput>> = Err(err);
+                put_request_guard.finish_err();
+                let _ = helper.complete(&result);
+                return result;
             }
             Err(err) => {
-                store_put_watchdog.cancel();
+                let result: S3Result<S3Response<PutObjectOutput>> = Err(S3Error::with_message(
-                rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
+                    S3ErrorCode::InternalError,
-                warn!(
+                    format!("put object commit owner task failed: {err}"),
-                    target: "rustfs::app::object_usecase",
+                ));
-                    event = EVENT_PUT_OBJECT_STORE_RETURNED,
-                    component = LOG_COMPONENT_APP,
-                    subsystem = LOG_SUBSYSTEM_OBJECT,
-                    request_id = %request_id,
-                    bucket = %bucket,
-                    key = %key,
-                    put_path = put_path,
-                    object_size = actual_size,
-                    duration_ms = start_time.elapsed().as_millis() as u64,
-                    result = "error",
-                    error = %err,
-                    "PutObject store write returned"
-                );
-                let result: S3Result<S3Response<PutObjectOutput>> = Err(err.into());
                 put_request_guard.finish_err();
                 let _ = helper.complete(&result);
                 return result;
             }
         };
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
-        drop(object_traffic_progress);
-        #[cfg(test)]
-        wait_for_put_post_store_test_hook(&bucket).await;
-
-        let post_store_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        maybe_enqueue_transition_immediate(&obj_info, LcEventSrc::S3PutObject).await;
-        let _ = invalidate_object_data_cache_after_put_success(&cache_adapter, &bucket, &key).await;
-
-        let put_versioned = BucketVersioningSys::prefix_enabled(&bucket, &key).await;
-        // Fast in-memory update for immediate quota and admin usage consistency.
-        // The previous current size comes from the prelookup when it ran,
-        // otherwise from the rename_data backfill (rustfs/backlog#1009); the
-        // backfill reproduces the lookup's observation bit for bit (latest
-        // version's ObjectInfo.size — 0 for a delete-marker latest — or
-        // not-found → None).
-        match prelookup_previous_current_size.or_else(|| previous_current_size_from_backfill(backfilled_old_current_size)) {
-            Some(previous_current_size) => {
-                if put_versioned {
-                    record_bucket_object_version_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
-                } else {
-                    record_bucket_object_write_memory(&bucket, previous_current_size, obj_info.size.max(0) as u64).await;
-                }
-            }
-            None => {
-                // Neither source could determine the previous state (peers
-                // predating the backfill field during a rolling upgrade, or
-                // sub-quorum metadata divergence). Record the components that
-                // are correct regardless; the next authoritative scanner
-                // refresh replaces the in-memory numbers.
-                debug!(
-                    target: "rustfs::app::object_usecase",
-                    bucket = %bucket,
-                    key = %key,
-                    put_versioned,
-                    "put_object old-size backfill unknown; recording degraded usage delta"
-                );
-                record_bucket_object_write_unknown_previous_memory(&bucket, obj_info.size.max(0) as u64, put_versioned).await;
-            }
-        }
 
         let raw_version = obj_info.version_id.map(|v| v.to_string());
 
@@ -6110,17 +6163,6 @@ impl DefaultObjectUsecase {
 
         let expiration = resolve_put_object_expiration(&bucket, &obj_info).await;
 
-        // Reuse the single replication decision computed before commit (see `dsc`
-        // above) so the pending metadata persisted with the object and the
-        // post-commit schedule always derive from the same immutable decision.
-        // Recomputing here would repeat the versioning/config/target traversal and,
-        // worse, allow a replication-config hot update between the two phases to
-        // produce a pending-without-schedule or schedule-without-pending divergence
-        // (https://github.com/rustfs/backlog/issues/1320).
-        if dsc.replicate_any() {
-            schedule_object_replication(obj_info.clone(), store, dsc).await;
-        }
-
         let mut checksums = PutObjectChecksums {
             crc32: input.checksum_crc32,
             crc32c: input.checksum_crc32c,
@@ -6159,14 +6201,6 @@ impl DefaultObjectUsecase {
         inject_additional_checksum_headers(&mut response.headers, &put_extra_checksum_headers);
         let result = Ok(response);
         let _ = helper.complete(&result);
-        rustfs_scanner::record_dirty_usage_bucket(&bucket);
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_post_store_bookkeeping", post_store_stage_start);
-
-        // Record write operation for capacity management (inline to avoid per-request tokio::spawn overhead)
-        let capacity_update_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        let manager = get_capacity_manager();
-        manager.record_write_operation().await;
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_capacity_update", capacity_update_stage_start);
 
         // Record PutObject metrics via zero-copy-metrics
         {
@@ -11507,6 +11541,76 @@ mod tests {
         assert!(!recovered.write_stalled);
     }
 
+    #[tokio::test]
+    #[serial_test::serial(body_cache_hook)]
+    async fn cancelled_put_request_completes_post_commit_publication() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_cold_fill_test_context().await;
+        let bucket = format!("put-owner-tail-{}", Uuid::new_v4());
+        let object = "object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("PUT owner-tail bucket must be created");
+
+        let old_body = Bytes::from_static(b"old body that must be invalidated");
+        let old_info = put_real_cold_fill_object(&store, &bucket, object, &old_body).await;
+        let adapter = context.object_data_cache();
+        let old_plan = real_cold_fill_plan(&adapter, &bucket, object, &old_info);
+
+        let post_store_entered = Arc::new(tokio::sync::Barrier::new(2));
+        let post_store_resume = Arc::new(tokio::sync::Barrier::new(2));
+        install_put_post_store_test_hook(bucket.clone(), Arc::clone(&post_store_entered), Arc::clone(&post_store_resume));
+
+        let payload = Bytes::from_static(b"published despite caller cancellation");
+        let put_input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
+            .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
+            .build()
+            .expect("PUT input must build");
+        let put_usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+        let put = tokio::spawn(async move {
+            put_usecase
+                .execute_put_object(&FS::new(), build_request(put_input, Method::PUT))
+                .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(10), post_store_entered.wait())
+            .await
+            .expect("PUT must reach the post-store owner-tail hook");
+        assert_eq!(
+            adapter.fill_body(&old_plan, old_body.clone()).await,
+            rustfs_object_data_cache::ObjectDataCacheFillResult::Inserted,
+            "test must republish the old body while the owner tail is paused"
+        );
+        put.abort();
+        post_store_resume.wait().await;
+        let _ = put.await.expect_err("outer request task must be cancelled");
+
+        tokio::time::timeout(Duration::from_secs(10), async {
+            loop {
+                if matches!(
+                    adapter.lookup_body(&old_plan).await,
+                    rustfs_object_data_cache::ObjectDataCacheLookup::Miss
+                ) {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("post-commit owner tail must invalidate stale body cache after caller cancellation");
+
+        let recovered = store
+            .get_object_info(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("cancelled request's owned commit must still publish the object");
+        assert_eq!(recovered.size, i64::try_from(payload.len()).expect("test payload length must fit i64"));
+    }
+
     #[tokio::test]
     async fn object_progress_tracks_zero_byte_and_zero_copy_put_lock_waits() {
         use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};

rustfs/src/app/storage_api.rs

@@ -1150,7 +1150,9 @@ pub(crate) mod multipart_usecase {
         }
     }
 
-    pub(crate) use super::{access, bucket, data_usage, error, helper, io, object_utils, options, s3_api, set_disk, sse};
+    pub(crate) use super::{
+        access, bucket, data_usage, error, helper, io, object_utils, options, request_context, s3_api, set_disk, sse,
+    };
     pub(crate) use crate::storage::storage_api::{ECStore, StorageObjectInfo, StorageObjectOptions, StoragePutObjReader};
 }