chore(ecstore): fold the ListObjects forwarders into the ECStore impl (#6079)

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/erasure/coding/erasure.rs

@@ -71,10 +71,16 @@ impl EncodedBlock {
 
 const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;
 const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
+const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16;
+// Vec growth may retain twice the requested logical length. Keeping the logical
+// workspace at half the budget bounds each cached workspace's shard allocation to 1 MiB.
+const LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE: usize = 512 * 1024;
 
 type ModernReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<ReedSolomon>>>;
+type LegacyReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<LegacyReedSolomonEncoder>>>;
 
 static MODERN_REED_SOLOMON_CACHE: OnceLock<ModernReedSolomonCache> = OnceLock::new();
+static LEGACY_REED_SOLOMON_CACHE: OnceLock<LegacyReedSolomonCache> = OnceLock::new();
 
 /// Errors returned when constructing an [`Erasure`] codec.
 #[derive(Debug, thiserror::Error)]
@@ -141,43 +147,61 @@ pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize {
 struct LegacyReedSolomonEncoder {
     data_shards: usize,
     parity_shards: usize,
-    encoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
-    decoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
-}
-
-impl Clone for LegacyReedSolomonEncoder {
-    fn clone(&self) -> Self {
-        Self {
-            data_shards: self.data_shards,
-            parity_shards: self.parity_shards,
-            encoder_cache: std::sync::RwLock::new(None),
-            decoder_cache: std::sync::RwLock::new(None),
-        }
-    }
+    cache_workspaces: bool,
+    encoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
+    decoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
 }
 
 impl LegacyReedSolomonEncoder {
-    fn new(_data_shards: usize, _parity_shards: usize) -> io::Result<Self> {
+    fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
+        Self::with_workspace_cache(data_shards, parity_shards, false)
+    }
+
+    fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result<Self> {
         Ok(Self {
-            data_shards: _data_shards,
-            parity_shards: _parity_shards,
-            encoder_cache: std::sync::RwLock::new(None),
-            decoder_cache: std::sync::RwLock::new(None),
+            data_shards,
+            parity_shards,
+            cache_workspaces,
+            encoder_cache: RwLock::new(None),
+            decoder_cache: RwLock::new(None),
         })
     }
 
+    fn logical_shard_bytes_upper_bound(&self, shard_len: usize) -> Option<usize> {
+        let aligned_shard_len = shard_len.checked_add(63)?.checked_div(64)?.checked_mul(64)?;
+        let high_rate_decoder_work_count = self
+            .parity_shards
+            .checked_next_power_of_two()?
+            .checked_add(self.data_shards)?
+            .checked_next_power_of_two()?;
+        let low_rate_decoder_work_count = self
+            .data_shards
+            .checked_next_power_of_two()?
+            .checked_add(self.parity_shards)?
+            .checked_next_power_of_two()?;
+        aligned_shard_len.checked_mul(high_rate_decoder_work_count.max(low_rate_decoder_work_count))
+    }
+
+    fn should_cache_workspace(&self, shard_len: usize) -> bool {
+        self.cache_workspaces
+            && self
+                .logical_shard_bytes_upper_bound(shard_len)
+                .is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE)
+    }
+
     fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
         let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
         if shards_vec.is_empty() {
             return Ok(());
         }
         let shard_len = shards_vec[0].len();
+        let cached_encoder = self
+            .encoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?
+            .take();
         let mut encoder = {
-            let mut cache_guard = self
-                .encoder_cache
-                .write()
-                .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?;
-            match cache_guard.take() {
+            match cached_encoder {
                 Some(mut cached) => {
                     if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() {
                         reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
@@ -204,10 +228,15 @@ impl LegacyReedSolomonEncoder {
             }
         }
         drop(result);
-        *self
-            .encoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .encoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return encoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(encoder);
+            }
+        }
         Ok(())
     }
 
@@ -221,13 +250,13 @@ impl LegacyReedSolomonEncoder {
             .find_map(|s| s.as_ref().map(|v| v.len()))
             .ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?;
 
+        let cached_decoder = self
+            .decoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?
+            .take();
         let mut decoder = {
-            let mut cache_guard = self
-                .decoder_cache
-                .write()
-                .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?;
-
-            match cache_guard.take() {
+            match cached_decoder {
                 Some(mut cached_decoder) => {
                     if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) {
                         warn!("Failed to reset SIMD decoder: {:?}, creating new one", e);
@@ -274,10 +303,15 @@ impl LegacyReedSolomonEncoder {
 
         drop(result);
 
-        *self
-            .decoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .decoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return decoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(decoder);
+            }
+        }
 
         Ok(())
     }
@@ -435,6 +469,39 @@ fn cached_modern_reed_solomon(data_shards: usize, parity_shards: usize) -> Resul
     Ok(encoder)
 }
 
+fn cached_legacy_reed_solomon(data_shards: usize, parity_shards: usize) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
+    let cache = LEGACY_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new()));
+    cached_legacy_reed_solomon_in(cache, data_shards, parity_shards)
+}
+
+fn cached_legacy_reed_solomon_in(
+    cache: &LegacyReedSolomonCache,
+    data_shards: usize,
+    parity_shards: usize,
+) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
+    let key = (data_shards, parity_shards);
+    if let Some(encoder) = cache
+        .read()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .get(&key)
+        .cloned()
+    {
+        return Ok(encoder);
+    }
+
+    let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if let Some(existing) = cache.get(&key) {
+        return Ok(Arc::clone(existing));
+    }
+    if cache.len() < LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
+        let encoder = Arc::new(LegacyReedSolomonEncoder::with_workspace_cache(data_shards, parity_shards, true)?);
+        cache.insert(key, Arc::clone(&encoder));
+        return Ok(encoder);
+    }
+    drop(cache);
+    Ok(Arc::new(LegacyReedSolomonEncoder::new(data_shards, parity_shards)?))
+}
+
 fn encode_parity_shards<F>(shards: &mut [Option<Vec<u8>>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()>
 where
     F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>,
@@ -551,7 +618,7 @@ pub struct Erasure {
     pub data_shards: usize,
     pub parity_shards: usize,
     encoder: Option<ReedSolomonEncoder>,
-    legacy_encoder: Option<LegacyReedSolomonEncoder>,
+    legacy_encoder: Option<Arc<LegacyReedSolomonEncoder>>,
     pub block_size: usize,
     uses_legacy: bool,
     _id: Uuid,
@@ -687,7 +754,7 @@ impl Erasure {
 
         let legacy_encoder = if uses_legacy && parity_shards > 0 {
             Some(
-                LegacyReedSolomonEncoder::new(data_shards, parity_shards)
+                cached_legacy_reed_solomon(data_shards, parity_shards)
                     .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?,
             )
         } else {
@@ -1405,7 +1472,7 @@ mod tests {
         assert_eq!(cloned.block_size, legacy.block_size);
         assert!(cloned.uses_legacy);
 
-        let data = b"legacy clone should keep independent SIMD caches";
+        let data = b"legacy clone should preserve SIMD codec behavior";
         let encoded = cloned.encode_data(data).expect("legacy clone should encode");
         let mut shards = optional_shards(&encoded);
         shards[0] = None;
@@ -1413,6 +1480,93 @@ mod tests {
         assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data);
     }
 
+    #[test]
+    fn legacy_codecs_share_process_cache_across_erasure_instances() {
+        let first = Erasure::new_with_options(6, 3, 64, true)
+            .legacy_encoder
+            .expect("legacy codec should be initialized");
+        let second = Erasure::new_with_options(6, 3, 128, true)
+            .legacy_encoder
+            .expect("same legacy shard layout should be initialized");
+
+        assert!(Arc::ptr_eq(&first, &second));
+    }
+
+    #[test]
+    fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() {
+        let four_plus_two = Erasure::new_with_options(4, 2, 64, true)
+            .legacy_encoder
+            .expect("legacy codec should be initialized");
+        let four_plus_one = Erasure::new_with_options(4, 1, 64, true)
+            .legacy_encoder
+            .expect("distinct parity layout should be initialized");
+        let three_plus_two = Erasure::new_with_options(3, 2, 64, true)
+            .legacy_encoder
+            .expect("distinct data layout should be initialized");
+
+        assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one));
+        assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two));
+        assert_eq!(four_plus_two.logical_shard_bytes_upper_bound(64 * 1024), Some(512 * 1024));
+        assert!(four_plus_two.should_cache_workspace(64 * 1024));
+        assert!(!four_plus_two.should_cache_workspace(64 * 1024 + 1));
+
+        let nine_plus_seven =
+            LegacyReedSolomonEncoder::with_workspace_cache(9, 7, true).expect("9+7 legacy codec should construct");
+        assert_eq!(nine_plus_seven.logical_shard_bytes_upper_bound(16 * 1024), Some(512 * 1024));
+        assert!(nine_plus_seven.should_cache_workspace(16 * 1024));
+        assert!(!nine_plus_seven.should_cache_workspace(16 * 1024 + 1));
+
+        let uncached = LegacyReedSolomonEncoder::new(4, 2).expect("uncached legacy codec should construct");
+        assert!(!uncached.should_cache_workspace(64));
+    }
+
+    #[test]
+    fn saturated_legacy_codec_cache_does_not_retain_more_workspaces() {
+        let cache = RwLock::new(HashMap::new());
+        for parity_shards in 1..=LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
+            let cached =
+                cached_legacy_reed_solomon_in(&cache, 32, parity_shards).expect("cacheable legacy codec should construct");
+            assert!(cached.cache_workspaces);
+        }
+
+        let uncached =
+            cached_legacy_reed_solomon_in(&cache, 31, 1).expect("uncached legacy codec should construct after saturation");
+        assert!(!uncached.cache_workspaces);
+        assert_eq!(
+            cache.read().expect("cache lock should remain healthy").len(),
+            LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES
+        );
+    }
+
+    #[test]
+    fn concurrent_legacy_codecs_preserve_byte_exact_results() {
+        let barrier = Arc::new(std::sync::Barrier::new(2));
+        let payloads = [vec![0x35; 257], vec![0xca; 1025]];
+
+        std::thread::scope(|scope| {
+            let handles = payloads.each_ref().map(|payload| {
+                let barrier = Arc::clone(&barrier);
+                scope.spawn(move || {
+                    let erasure = Erasure::new_with_options(6, 3, 2048, true);
+                    barrier.wait();
+                    let encoded = erasure.encode_data(payload).expect("concurrent legacy encode should succeed");
+                    barrier.wait();
+
+                    let mut shards = optional_shards(&encoded);
+                    shards[0] = None;
+                    erasure
+                        .decode_data(&mut shards)
+                        .expect("concurrent legacy decode should reconstruct the missing shard");
+                    recover_data(&shards, erasure.data_shards, payload.len())
+                })
+            });
+
+            for (handle, payload) in handles.into_iter().zip(payloads.iter()) {
+                assert_eq!(handle.join().expect("concurrent legacy codec worker should not panic"), *payload);
+            }
+        });
+    }
+
     #[test]
     fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() {
         let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct");

crates/ecstore/src/object_api/readers.rs

@@ -15,6 +15,7 @@
 use super::*;
 
 use crate::io_support::rio::Index;
+use std::mem::MaybeUninit;
 
 #[cfg(feature = "rio-v2")]
 const DARE_PAYLOAD_SIZE: i64 = 64 * 1024;
@@ -922,7 +923,7 @@ struct SkipReader<R> {
     inner: R,
     bytes_to_skip: usize,
     bytes_skipped: usize,
-    scratch: Vec<u8>,
+    scratch: Box<[MaybeUninit<u8>]>,
 }
 
 impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
@@ -931,7 +932,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
             inner,
             bytes_to_skip,
             bytes_skipped: 0,
-            scratch: vec![0u8; 8192],
+            scratch: Box::<[u8]>::new_uninit_slice(8192),
         }
     }
 }
@@ -943,7 +944,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> AsyncRead for SkipReader<R> {
         while this.bytes_skipped < this.bytes_to_skip {
             let remaining = this.bytes_to_skip - this.bytes_skipped;
             let scratch_len = remaining.min(this.scratch.len());
-            let mut scratch_buf = ReadBuf::new(&mut this.scratch[..scratch_len]);
+            let mut scratch_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]);
             match Pin::new(&mut this.inner).poll_read(cx, &mut scratch_buf) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
@@ -974,7 +975,7 @@ pub struct RangedDecompressReader<R: AsyncRead + Unpin + Send + Sync + 'static>
     target_length: usize,
     current_offset: usize,
     bytes_returned: usize,
-    scratch: Vec<u8>,
+    scratch: Box<[MaybeUninit<u8>]>,
     drain_on_done: bool,
     drain_task: Option<tokio::task::JoinHandle<()>>,
 }
@@ -1012,7 +1013,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
             target_length: actual_length,
             current_offset: 0,
             bytes_returned: 0,
-            scratch: vec![0u8; 8192],
+            scratch: Box::<[u8]>::new_uninit_slice(8192),
             drain_on_done,
             drain_task: None,
         })
@@ -1062,7 +1063,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
             }
 
             let scratch_len = std::cmp::min(this.scratch.len(), std::cmp::max(buf_capacity, 1));
-            let mut temp_read_buf = ReadBuf::new(&mut this.scratch[..scratch_len]);
+            let mut temp_read_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]);
 
             let Some(inner) = this.inner.as_mut() else {
                 return Poll::Ready(Ok(()));
@@ -1114,7 +1115,8 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
                             );
 
                             if bytes_to_return > 0 {
-                                let data_slice = &this.scratch[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
+                                let data_slice =
+                                    &temp_read_buf.filled()[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
                                 buf.put_slice(data_slice);
                                 this.bytes_returned += bytes_to_return;
 
@@ -1133,7 +1135,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
                             std::cmp::min(n, std::cmp::min(buf.remaining(), this.target_length - this.bytes_returned));
 
                         if bytes_to_return > 0 {
-                            buf.put_slice(&this.scratch[..bytes_to_return]);
+                            buf.put_slice(&temp_read_buf.filled()[..bytes_to_return]);
                             this.bytes_returned += bytes_to_return;
 
                             tracing::trace!("Returned {} bytes at offset {}", bytes_to_return, old_offset);
@@ -1263,6 +1265,43 @@ mod tests {
     use temp_env::async_with_vars;
     use tokio::io::AsyncReadExt;
 
+    #[derive(Debug)]
+    struct PendingPartialReader {
+        data: &'static [u8],
+        position: usize,
+        pending: bool,
+    }
+
+    impl PendingPartialReader {
+        fn new(data: &'static [u8]) -> Self {
+            Self {
+                data,
+                position: 0,
+                pending: true,
+            }
+        }
+    }
+
+    impl AsyncRead for PendingPartialReader {
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if self.pending {
+                self.pending = false;
+                cx.waker().wake_by_ref();
+                return Poll::Pending;
+            }
+            if self.position == self.data.len() {
+                return Poll::Ready(Ok(()));
+            }
+
+            let length = buf.remaining().min(3).min(self.data.len() - self.position);
+            let end = self.position + length;
+            buf.put_slice(&self.data[self.position..end]);
+            self.position = end;
+            self.pending = true;
+            Poll::Ready(Ok(()))
+        }
+    }
+
     const TEST_DIRECT_KEY_HEADER: &str = "x-rustfs-test-direct-key";
     const TEST_OBJECT_KEY_HEADER: &str = "x-rustfs-test-object-key";
     const TEST_NONCE_HEADER: &str = "x-rustfs-test-nonce";
@@ -1400,6 +1439,36 @@ mod tests {
         assert_eq!(result, b"World");
     }
 
+    #[tokio::test]
+    async fn uninitialized_scratch_preserves_partial_pending_and_eof_reads() {
+        let mut skipped = SkipReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5);
+        let mut skipped_output = Vec::new();
+        skipped
+            .read_to_end(&mut skipped_output)
+            .await
+            .expect("skip reader should survive partial pending reads through EOF");
+        assert_eq!(skipped_output, b"56789abcdef");
+
+        let mut ranged = RangedDecompressReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5, 7, 16)
+            .expect("valid range should construct");
+        let mut ranged_output = Vec::new();
+        ranged
+            .read_to_end(&mut ranged_output)
+            .await
+            .expect("range reader should survive partial pending reads through EOF");
+        assert_eq!(ranged_output, b"56789ab");
+    }
+
+    #[tokio::test]
+    async fn uninitialized_skip_scratch_reports_early_eof() {
+        let mut reader = SkipReader::new(PendingPartialReader::new(b"short"), 6);
+        let error = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("EOF before the skip boundary must remain visible");
+        assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof);
+    }
+
     #[tokio::test]
     async fn test_ranged_decompress_reader_from_start() {
         let original_data = b"Hello, World! This is a test.";

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/mod.rs

@@ -2792,6 +2792,9 @@ pub struct SetDisks {
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
     get_object_metadata_cache_generations: Arc<[AtomicU64]>,
+    /// GET codecs keyed by every persisted layout dimension that affects
+    /// decoding. Clones of a set share the memoized shells.
+    erasure_cache: Arc<ErasureCache>,
     pub lockers: Vec<Arc<dyn LockClient>>,
     shared_lockers: Arc<[Arc<dyn LockClient>]>,
     local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
@@ -2814,6 +2817,137 @@ pub struct SetDisks {
     storage_class_config_override: Arc<std::sync::RwLock<Option<Arc<storageclass::Config>>>>,
 }
 
+const ERASURE_CACHE_MAX_ENTRIES: usize = 32;
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+struct ErasureCacheKey {
+    data_shards: usize,
+    parity_shards: usize,
+    block_size: usize,
+    uses_legacy: bool,
+}
+
+struct ErasureCache {
+    entries: parking_lot::RwLock<HashMap<ErasureCacheKey, Arc<coding::Erasure>>>,
+}
+
+impl std::fmt::Debug for ErasureCache {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("ErasureCache")
+            .field("entries", &self.entries.read().len())
+            .finish()
+    }
+}
+
+impl ErasureCache {
+    fn new() -> Self {
+        Self {
+            entries: parking_lot::RwLock::new(HashMap::new()),
+        }
+    }
+
+    fn get_or_try_insert(
+        &self,
+        key: ErasureCacheKey,
+    ) -> std::result::Result<Arc<coding::Erasure>, coding::ErasureConstructionError> {
+        if let Some(erasure) = self.entries.read().get(&key) {
+            return Ok(Arc::clone(erasure));
+        }
+
+        // Serialize first construction for a key so concurrent cold GETs still
+        // create exactly one shell. Codec construction never awaits.
+        let mut entries = self.entries.write();
+        if let Some(erasure) = entries.get(&key) {
+            return Ok(Arc::clone(erasure));
+        }
+        let erasure = Arc::new(coding::Erasure::try_new_with_options(
+            key.data_shards,
+            key.parity_shards,
+            key.block_size,
+            key.uses_legacy,
+        )?);
+        if entries.len() < ERASURE_CACHE_MAX_ENTRIES {
+            entries.insert(key, Arc::clone(&erasure));
+        }
+        Ok(erasure)
+    }
+
+    fn get_for_file_info(&self, fi: &FileInfo) -> Result<Arc<coding::Erasure>> {
+        self.get_or_try_insert(ErasureCacheKey {
+            data_shards: fi.erasure.data_blocks,
+            parity_shards: fi.erasure.parity_blocks,
+            block_size: fi.erasure.block_size,
+            uses_legacy: fi.uses_legacy_checksum,
+        })
+        .map_err(Error::from)
+    }
+}
+
+#[cfg(test)]
+mod erasure_cache_tests {
+    use super::*;
+
+    #[test]
+    fn reuses_shells_and_keeps_every_layout_dimension_in_the_key() {
+        let cache = ErasureCache::new();
+        let base = ErasureCacheKey {
+            data_shards: 4,
+            parity_shards: 2,
+            block_size: 1_048_576,
+            uses_legacy: false,
+        };
+        let first = cache.get_or_try_insert(base).expect("modern shell should construct");
+        let reused = cache.get_or_try_insert(base).expect("same modern shell should be cached");
+        assert!(Arc::ptr_eq(&first, &reused));
+
+        for distinct in [
+            ErasureCacheKey { data_shards: 3, ..base },
+            ErasureCacheKey {
+                parity_shards: 1,
+                ..base
+            },
+            ErasureCacheKey {
+                block_size: 524_288,
+                ..base
+            },
+            ErasureCacheKey {
+                uses_legacy: true,
+                ..base
+            },
+        ] {
+            let shell = cache.get_or_try_insert(distinct).expect("distinct shell should construct");
+            assert!(!Arc::ptr_eq(&first, &shell));
+        }
+        assert_eq!(cache.entries.read().len(), 5);
+    }
+
+    #[test]
+    fn does_not_cache_invalid_layouts_or_grow_past_the_bound() {
+        let cache = ErasureCache::new();
+        let invalid = ErasureCacheKey {
+            data_shards: 4,
+            parity_shards: 2,
+            block_size: 0,
+            uses_legacy: false,
+        };
+        assert!(cache.get_or_try_insert(invalid).is_err());
+        assert!(cache.entries.read().is_empty());
+
+        for block_size in 1..=(ERASURE_CACHE_MAX_ENTRIES + 1) {
+            cache
+                .get_or_try_insert(ErasureCacheKey {
+                    data_shards: 4,
+                    parity_shards: 2,
+                    block_size,
+                    uses_legacy: false,
+                })
+                .expect("bounded cache fixture should construct");
+        }
+        assert_eq!(cache.entries.read().len(), ERASURE_CACHE_MAX_ENTRIES);
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 struct GetObjectMetadataCacheKey {
     bucket: Arc<str>,
@@ -3212,6 +3346,7 @@ impl SetDisks {
                     .map(|_| AtomicU64::new(0))
                     .collect::<Vec<_>>(),
             ),
+            erasure_cache: Arc::new(ErasureCache::new()),
             lockers,
             shared_lockers,
             // Sourced from the instance context so each instance owns its lock
@@ -9816,6 +9951,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
             &disk_files,
             &disks,
@@ -9879,6 +10015,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
             &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
@@ -9959,6 +10096,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             &fi,
             &files,
             &disks,
@@ -10044,6 +10182,7 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
+                Arc::new(ErasureCache::new()),
                 range_offset,
                 range_length as i64,
                 &mut writer,
@@ -10155,6 +10294,7 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
+                Arc::new(ErasureCache::new()),
                 0,
                 total_size as i64,
                 &mut writer,

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
-        .get_or_init(|| std::sync::Mutex::new(None))
-        .lock()
-        .expect("multipart commit barrier mutex should not poison")
-        .as_ref()
-        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
-        .cloned();
+    let barrier = {
+        let mut slot = MULTIPART_COMMIT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("multipart commit barrier mutex should not poison");
+        if slot
+            .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/read.rs

@@ -482,6 +482,7 @@ impl SetDisks {
     pub(super) async fn try_get_object_direct_data_shards_with_fileinfo(
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         fi: &FileInfo,
         files: &[FileInfo],
         disks: &[Option<DiskStore>],
@@ -502,13 +503,7 @@ impl SetDisks {
             return Ok(None);
         }
 
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
+        let erasure = erasure_cache.get_for_file_info(fi)?;
 
         let checksum_info = fi.erasure.get_checksum_info(part.number);
         let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -636,6 +631,7 @@ impl SetDisks {
         // &self,
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         offset: usize,
         length: i64,
         writer: &mut W,
@@ -730,13 +726,7 @@ impl SetDisks {
             object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
         );
 
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
+        let erasure = erasure_cache.get_for_file_info(&fi)?;
 
         let part_indices: Vec<usize> = (part_index..=last_part_index).collect();
         debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
@@ -1170,6 +1160,7 @@ impl SetDisks {
     pub(super) async fn get_object_decode_reader_with_fileinfo(
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         fi: &FileInfo,
         files: &[FileInfo],
         disks: &[Option<DiskStore>],
@@ -1180,14 +1171,7 @@ impl SetDisks {
         metrics_size_bucket: &'static str,
         prefer_data_blocks_first_reader_setup: bool,
     ) -> Result<GetCodecStreamingReaderBuildOutcome> {
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
-
+        let erasure = erasure_cache.get_for_file_info(fi)?;
         let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
 
         if fi.parts.len() == 1 {
@@ -1574,7 +1558,7 @@ struct LazyCodecPartContext {
     fi: FileInfo,
     files: Vec<FileInfo>,
     disks: Vec<Option<DiskStore>>,
-    erasure: coding::Erasure,
+    erasure: Arc<coding::Erasure>,
     skip_verify_bitrot: bool,
     metrics_object_class: &'static str,
     metrics_size_bucket: &'static str,
@@ -2058,6 +2042,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2088,6 +2073,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             2,
             1,
             &mut output,
@@ -2111,6 +2097,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             usize::MAX,
             1,
             &mut output,
@@ -2132,6 +2119,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             1,
             1,
             &mut output,
@@ -2155,6 +2143,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2192,6 +2181,7 @@ mod metadata_cache_tests {
         SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             0,
             &mut output,
@@ -2224,6 +2214,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -4128,6 +4119,7 @@ mod tests {
         let result = SetDisks::get_object_decode_reader_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             &fi,
             &[],
             &[],
@@ -4150,6 +4142,7 @@ mod tests {
         let invalid_size = SetDisks::get_object_decode_reader_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             &single_part,
             &[],
             &[],
@@ -4170,6 +4163,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4194,6 +4188,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4222,6 +4217,7 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
+                    Arc::new(ErasureCache::new()),
                     &multipart,
                     &[],
                     &[],
@@ -4275,6 +4271,7 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
+                    Arc::new(ErasureCache::new()),
                     &fi,
                     &files,
                     &disks,
@@ -4328,6 +4325,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &fi,
                 &files,
                 &disks,
@@ -4372,6 +4370,7 @@ mod tests {
         SetDisks::get_object_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             0,
             part_data.len() as i64,
             &mut output,

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/ecstore/src/store/list.rs

@@ -1,81 +0,0 @@
-// Copyright 2024 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 super::*;
-
-impl ECStore {
-    #[instrument(level = "trace", skip(self))]
-    #[allow(clippy::too_many_arguments)]
-    pub(super) async fn handle_list_objects_v2(
-        self: Arc<Self>,
-        bucket: &str,
-        prefix: &str,
-        continuation_token: Option<String>,
-        delimiter: Option<String>,
-        max_keys: i32,
-        fetch_owner: bool,
-        start_after: Option<String>,
-        incl_deleted: bool,
-    ) -> Result<ListObjectsV2Info> {
-        self.inner_list_objects_v2(
-            bucket,
-            prefix,
-            continuation_token,
-            delimiter,
-            max_keys,
-            fetch_owner,
-            start_after,
-            incl_deleted,
-        )
-        .await
-    }
-
-    #[instrument(skip(self))]
-    pub(super) async fn handle_list_object_versions(
-        self: Arc<Self>,
-        bucket: &str,
-        prefix: &str,
-        marker: Option<String>,
-        version_marker: Option<String>,
-        delimiter: Option<String>,
-        max_keys: i32,
-    ) -> Result<ListObjectVersionsInfo> {
-        self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
-            .await
-    }
-
-    pub(crate) async fn list_object_versions_for_lifecycle(
-        self: Arc<Self>,
-        bucket: &str,
-        prefix: &str,
-        marker: Option<String>,
-        version_marker: Option<String>,
-        delimiter: Option<String>,
-        max_keys: i32,
-    ) -> Result<ListObjectVersionsInfo> {
-        self.inner_list_object_versions_for_lifecycle(bucket, prefix, marker, version_marker, delimiter, max_keys)
-            .await
-    }
-
-    pub(super) async fn handle_walk(
-        self: Arc<Self>,
-        rx: CancellationToken,
-        bucket: &str,
-        prefix: &str,
-        result: tokio::sync::mpsc::Sender<ObjectInfoOrErr>,
-        opts: WalkOptions,
-    ) -> Result<()> {
-        self.walk_internal(rx, bucket, prefix, result, opts).await
-    }
-}

crates/ecstore/src/store/list_objects.rs

@@ -3844,7 +3844,7 @@ impl ECStore {
         .await
     }
 
-    pub(crate) async fn inner_list_object_versions_for_lifecycle(
+    pub(crate) async fn list_object_versions_for_lifecycle(
         self: Arc<Self>,
         bucket: &str,
         prefix: &str,

crates/ecstore/src/store/mod.rs

@@ -148,7 +148,6 @@ mod heal_walk;
 pub use heal_walk::HealWalkVersion;
 mod init;
 pub(crate) mod init_format;
-mod list;
 pub(crate) mod list_objects;
 mod multipart;
 mod object;
@@ -601,7 +600,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore {
         start_after: Option<String>,
         incl_deleted: bool,
     ) -> Result<ListObjectsV2Info> {
-        self.handle_list_objects_v2(
+        self.inner_list_objects_v2(
             bucket,
             prefix,
             continuation_token,
@@ -624,7 +623,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore {
         delimiter: Option<String>,
         max_keys: i32,
     ) -> Result<ListObjectVersionsInfo> {
-        self.handle_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
+        self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys)
             .await
     }
 
@@ -636,7 +635,7 @@ impl crate::storage_api_contracts::list::ListOperations for ECStore {
         result: tokio::sync::mpsc::Sender<ObjectInfoOrErr>,
         opts: WalkOptions,
     ) -> Result<()> {
-        self.handle_walk(rx, bucket, prefix, result, opts).await
+        self.walk_internal(rx, bucket, prefix, result, opts).await
     }
 }
 

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,
-    has_internal_suffix, insert_str, is_encryption_metadata_key, starts_with_ignore_ascii_case,
+    SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID,
+    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
-            .clone()
-            .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts)
-            .await
-            .map_err(ApiError::from)?;
-        let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
-        record_capacity_write(Some(capacity_scope_token)).await;
-
-        if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
-            if opts.replication_request {
-                let quota_checker = QuotaChecker::new(metadata_sys.clone());
-                match quota_checker
-                    .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64)
+        let complete_commit = spawn_traced_join({
+            let store = Arc::clone(&store);
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let upload_id = upload_id.clone();
+            let opts = opts.clone();
+            let quota_metadata_sys = quota_metadata_sys.clone();
+            async move {
+                let obj_info = store
+                    .clone()
+                    .complete_multipart_upload(&bucket, &key, &upload_id, uploaded_parts, &opts)
                     .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)
-                            ),
-                        ));
+                    .map_err(ApiError::from)?;
+                let _ = invalidate_object_data_cache_after_complete_multipart_success(&cache_adapter, &bucket, &key).await;
+                record_capacity_write(Some(capacity_scope_token)).await;
+
+                if let Some(metadata_sys) = quota_metadata_sys.as_ref() {
+                    if opts.replication_request {
+                        let quota_checker = QuotaChecker::new(metadata_sys.clone());
+                        match quota_checker
+                            .check_quota(&bucket, QuotaOperation::PutObject, obj_info.size.max(0) as u64)
+                            .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 {
-                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;
-            }
-        }
-
-        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 = if versioned {
+            obj_info.version_id.map(|v| v.to_string())
+        } 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_crc32c: checksum_crc32c.clone(),
-            checksum_sha1: checksum_sha1.clone(),
-            checksum_sha256: checksum_sha256.clone(),
-            checksum_crc64nvme: checksum_crc64nvme.clone(),
-            checksum_type: checksum_type.clone(),
+            checksum_crc32,
+            checksum_crc32c,
+            checksum_sha1,
+            checksum_sha256,
+            checksum_crc64nvme,
+            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
-            .as_deref()
-            .and_then(ObjectTrafficHealth::track_write_storage);
-        let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        let (obj_info, backfilled_old_current_size) = match store
-            .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
-            .await
-            .map_err(ApiError::from)
-        {
-            Ok(obj_info) => {
-                store_put_watchdog.cancel();
-                debug!(
-                    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 = "success",
-                    "PutObject store write returned"
-                );
-                obj_info
+        let put_commit = spawn_traced_join({
+            let store = Arc::clone(&store);
+            let bucket = bucket.clone();
+            let key = key.clone();
+            let opts = opts.clone();
+            let cache_adapter = cache_adapter.clone();
+            let request_id = request_id.clone();
+            let put_path = put_path.to_string();
+            async move {
+                let object_traffic_progress = object_traffic_health
+                    .as_deref()
+                    .and_then(ObjectTrafficHealth::track_write_storage);
+                let mut reader = reader;
+                let store_put_stage_start = put_stage_metrics_enabled.then(Instant::now);
+                let (obj_info, backfilled_old_current_size) = match store
+                    .put_object_with_old_current_size(&bucket, &key, &mut reader, &opts)
+                    .await
+                    .map_err(ApiError::from)
+                {
+                    Ok(obj_info) => {
+                        store_put_watchdog.cancel();
+                        debug!(
+                            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 = "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();
-                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"
-                );
-                let result: S3Result<S3Response<PutObjectOutput>> = Err(err.into());
+                let result: S3Result<S3Response<PutObjectOutput>> = Err(S3Error::with_message(
+                    S3ErrorCode::InternalError,
+                    format!("put object commit owner task failed: {err}"),
+                ));
                 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};
 }
 

scripts/check_logging_guardrails.sh

@@ -984,7 +984,9 @@ trace_hot_spans=(
   "crates/ecstore/src/store/object.rs:handle_get_object_info"
   "crates/ecstore/src/set_disk/ops/object.rs:get_object_info"
   "crates/ecstore/src/store/mod.rs:list_objects_v2"
-  "crates/ecstore/src/store/list.rs:handle_list_objects_v2"
+  # The ECStore handle_list_objects_v2 forwarder was folded into the trait impl
+  # above, so store/mod.rs now carries this hot path's TRACE requirement
+  # directly (backlog#1821).
   # The pool-level Sets::list_objects_v2 wrapper was removed with its duplicate
   # pagination pipeline (backlog#1821); the remaining ECStore and SetDisks
   # wrappers below still carry the TRACE requirement.