fix: reduce scanner backoff and classify downstream closures (#9235)

Co-authored-by: Chris An <anzhengchao@gmail.com>

Cargo.lock

@@ -10770,7 +10770,6 @@ dependencies = [
  "hotpath",
  "http 1.5.0",
  "lz4",
- "lz4_flex",
  "memchr",
  "metrics",
  "parking_lot 0.13.0",

Cargo.toml

@@ -296,7 +296,6 @@ local-ip-address = "0.6.13"
 log = "0.4"
 memmap2 = "0.9.12"
 lz4 = "1.28.1"
-lz4_flex = { version = "=0.14.0", default-features = false, features = ["std", "frame"] }
 matchit = "0.9.2"
 md-5 = "0.11.0"
 mime_guess = "2.0.5"

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

@@ -127,9 +127,14 @@ const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capab
 const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
 const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned";
 const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
-const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601);
-const REPLAY_CACHE_RETENTION_SECS: usize = 601;
-const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128;
+// Timestamp validity and the wall-clock high-water mark still gate eviction,
+// including future-dated signatures and replay attempts after clock rollback.
+const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(301);
+// Preserve capacity for the full validity window at maximum accepted clock skew.
+const REPLAY_CACHE_CAPACITY_WINDOW_SECS: usize = 601;
+// A reserved index avoids overlapping old/new tables during a rehash. Include
+// index slack, compact expiry storage, and page/allocator bookkeeping.
+const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: usize = 2 * std::mem::size_of::<Uuid>() + std::mem::size_of::<RpcNonceExpiration>() + 16;
 // Keep 16 CPU / 32 GiB field nodes at the 32M cap without requiring an env override.
 const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13;
 const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096;
@@ -222,11 +227,11 @@ fn replay_cache_capacity_from_resources(cpu_count: usize, memory_limit_bytes: Op
     let cpu_count = cpu_count.max(1);
     let cpu_based_capacity = cpu_count
         .saturating_mul(REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU)
-        .saturating_mul(REPLAY_CACHE_RETENTION_SECS);
+        .saturating_mul(REPLAY_CACHE_CAPACITY_WINDOW_SECS);
     let memory_based_capacity = memory_limit_bytes
         .map(|bytes| {
             let budget = bytes.saturating_mul(REPLAY_CACHE_AUTO_MEMORY_PERCENT) / 100;
-            saturating_usize_from_u64(budget / REPLAY_CACHE_ENTRY_BYTES_ESTIMATE)
+            saturating_usize_from_u64(budget / u64::try_from(REPLAY_CACHE_ENTRY_BYTES_ESTIMATE).unwrap_or(u64::MAX))
         })
         .unwrap_or(REPLAY_CACHE_AUTO_MAX_CAPACITY);
     let capacity = memory_based_capacity
@@ -497,6 +502,12 @@ fn publish_nonce_cache_metrics(metrics: Option<RpcNonceCacheMetrics<'_>>) {
 }
 
 impl RpcNonceCache {
+    fn reserve_index(&mut self, capacity: usize) -> std::io::Result<()> {
+        self.nonces
+            .try_reserve(capacity)
+            .map_err(|error| std::io::Error::new(std::io::ErrorKind::OutOfMemory, error))
+    }
+
     fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize {
         let mut removed = 0;
         while let Some(nonce) = self.expirations.pop_expired(now, wall_time) {
@@ -537,6 +548,11 @@ impl RpcNonceCache {
                 }),
             );
         }
+        if self.nonces.is_empty()
+            && let Err(error) = self.reserve_index(record.capacity.min(REPLAY_CACHE_AUTO_MAX_CAPACITY))
+        {
+            return (Err(error), Some(metrics));
+        }
         self.nonces.insert(record.nonce);
         self.expirations
             .push_back(record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce);
@@ -1699,6 +1715,9 @@ pub fn verify_tonic_rpc_signature_with_bootstrap(
 }
 
 pub fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str {
+    if error.kind() == std::io::ErrorKind::OutOfMemory {
+        return "replay_cache_allocation";
+    }
     match error.to_string().as_str() {
         "Missing RPC audience" => "missing_audience",
         "Invalid RPC request path" => "invalid_request_path",
@@ -3138,9 +3157,9 @@ mod tests {
 
         assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
         assert_eq!(decision.memory_basis, Some(MemoryBasis::Host));
-        assert_eq!(decision.memory_based_capacity, 17_448_304);
+        assert_eq!(decision.memory_based_capacity, 31_019_208);
         assert_eq!(decision.cpu_based_capacity, 19_693_568);
-        assert_eq!(decision.capacity, 17_448_304);
+        assert_eq!(decision.capacity, 19_693_568);
     }
 
     #[test]
@@ -3150,13 +3169,13 @@ mod tests {
             replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(32 * gib), Some(MemoryBasis::Host));
 
         assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
-        assert_eq!(decision.memory_based_capacity, 34_896_609);
+        assert_eq!(decision.memory_based_capacity, 62_038_416);
         assert_eq!(decision.cpu_based_capacity, 39_387_136);
         assert_eq!(decision.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY);
 
         let observed_field_node =
             replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(31 * gib), Some(MemoryBasis::Host));
-        assert_eq!(observed_field_node.memory_based_capacity, 33_806_090);
+        assert_eq!(observed_field_node.memory_based_capacity, 60_099_715);
         assert_eq!(observed_field_node.cpu_based_capacity, 39_387_136);
         assert_eq!(observed_field_node.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY);
     }
@@ -3229,6 +3248,138 @@ mod tests {
         }
     }
 
+    #[test]
+    fn nonce_cache_releases_expired_timestamp_capacity_without_allowing_clock_rollback_replay() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let later = now + Duration::from_secs(302);
+        let nonce = Uuid::new_v4();
+        let replacement = Uuid::new_v4();
+        let mut cache = RpcNonceCache::default();
+        cache
+            .check_and_record(test_nonce_record(nonce, 1_000, now, 1_000, expiry, 1))
+            .0
+            .expect("record initial nonce");
+
+        let boundary = cache
+            .check_and_record(test_nonce_record(nonce, 1_000, later, 1_300, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("a signature at the inclusive freshness boundary must remain protected");
+        assert_eq!(boundary.to_string(), "RPC request replay detected");
+        cache
+            .check_and_record(test_nonce_record(replacement, 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect("an expired signature must release bounded cache capacity");
+        assert!(!cache.nonces.contains(&nonce));
+        assert!(cache.nonces.contains(&replacement));
+
+        let rollback = cache
+            .check_and_record(test_nonce_record(nonce, 1_000, later, 1_000, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("clock rollback must not resurrect an evicted signature");
+        assert_eq!(rollback.to_string(), "RPC request timestamp expired after clock regression");
+    }
+
+    #[test]
+    fn nonce_cache_keeps_future_timestamp_reserved_until_its_full_validity_window_closes() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let later = now + Duration::from_secs(302);
+        let nonce = Uuid::new_v4();
+        let mut cache = RpcNonceCache::default();
+        cache
+            .check_and_record(test_nonce_record(nonce, 1_300, now, 1_000, expiry, 1))
+            .0
+            .expect("record nonce signed at maximum allowed forward skew");
+        let replay = cache
+            .check_and_record(test_nonce_record(nonce, 1_300, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("monotonic expiry must not remove a still-valid future signature");
+        assert_eq!(replay.to_string(), "RPC request replay detected");
+        let full = cache
+            .check_and_record(test_nonce_record(Uuid::new_v4(), 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("a still-valid nonce must continue consuming capacity");
+        assert_eq!(full.to_string(), "RPC replay cache capacity exceeded");
+        let expired = now + Duration::from_secs(602);
+        cache
+            .check_and_record(test_nonce_record(
+                Uuid::new_v4(),
+                1_602,
+                expired,
+                1_602,
+                expired + REPLAY_CACHE_RETENTION,
+                1,
+            ))
+            .0
+            .expect("release capacity only once future timestamp validity has closed");
+    }
+
+    #[test]
+    fn nonce_index_rejects_unallocatable_capacity_with_typed_error() {
+        let mut cache = RpcNonceCache::default();
+        let error = cache
+            .reserve_index(usize::MAX)
+            .expect_err("unallocatable capacity must fail without a panic");
+        assert_eq!(error.kind(), std::io::ErrorKind::OutOfMemory);
+        assert!(
+            error
+                .get_ref()
+                .is_some_and(|source| source.is::<std::collections::TryReserveError>())
+        );
+        assert_eq!(tonic_rpc_auth_failure_reason(&error), "replay_cache_allocation");
+        assert!(cache.nonces.is_empty());
+        assert!(cache.expirations.pages.is_empty());
+    }
+
+    #[test]
+    fn nonce_cache_reserves_the_bounded_index_before_sustained_traffic() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let mut cache = RpcNonceCache::default();
+        let capacity = 4096;
+        cache
+            .check_and_record(test_nonce_record(Uuid::from_u128(1), 1000, now, 1000, expiry, capacity))
+            .0
+            .expect("record first nonce");
+        let allocated = cache.nonces.capacity();
+        assert!(allocated >= capacity, "the hash index must not rehash while serving the bounded window");
+        for n in 2..=capacity {
+            cache
+                .check_and_record(test_nonce_record(
+                    Uuid::from_u128(u128::try_from(n).expect("nonce index")),
+                    1000,
+                    now,
+                    1000,
+                    expiry,
+                    capacity,
+                ))
+                .0
+                .expect("record within bounded capacity");
+        }
+        assert_eq!(cache.nonces.capacity(), allocated);
+        let full = cache
+            .check_and_record(test_nonce_record(Uuid::from_u128(9999), 1000, now, 1000, expiry, capacity))
+            .0
+            .expect_err("preallocation must not relax the logical capacity fence");
+        assert_eq!(full.to_string(), "RPC replay cache capacity exceeded");
+        assert_eq!(cache.nonces.len(), capacity);
+    }
+
+    #[test]
+    fn replay_cache_compact_storage_fits_two_cpu_window_in_2800_mib() {
+        let decision = replay_cache_capacity_decision(
+            rustfs_utils::EnvParseOutcome::Absent,
+            2,
+            Some(2800 * 1024 * 1024),
+            Some(MemoryBasis::Cgroup),
+        );
+        assert_eq!(decision.cpu_based_capacity, 4_923_392);
+        assert_eq!(decision.capacity, decision.cpu_based_capacity);
+        assert!(decision.memory_based_capacity >= decision.capacity);
+        assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
+    }
+
     #[test]
     fn nonce_cache_expires_by_monotonic_deadline_and_fails_closed_at_capacity() {
         let now = Instant::now();

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

@@ -376,6 +376,21 @@ struct WalkCopyError {
     source: io::Error,
 }
 
+impl WalkCopyError {
+    fn into_disk_error(self) -> DiskError {
+        // Only a local output write/flush proves the metacache consumer closed.
+        // An identical I/O kind from the remote input remains a transport failure.
+        if matches!(self.reason, RemoteWalkFailureReason::Write | RemoteWalkFailureReason::Flush)
+            && self.source.kind() == io::ErrorKind::BrokenPipe
+            && self.source.get_ref().is_none()
+        {
+            DiskError::metacache_output_stream_closed()
+        } else {
+            DiskError::Io(self.source)
+        }
+    }
+}
+
 async fn copy_stream_with_buffer<R, W>(
     reader: &mut R,
     writer: &mut W,
@@ -2096,6 +2111,7 @@ impl RemoteDisk {
         failure_health_action: FailureHealthAction,
     ) {
         if let Err(err) = operation_result
+            && !err.is_metacache_output_stream_closed()
             && is_network_like_disk_error(err)
         {
             counter!(
@@ -4056,7 +4072,7 @@ impl DiskAPI for RemoteDisk {
                             Ok(_) => return Ok(()),
                             Err(error) => {
                                 failure_reason = Some(error.reason);
-                                return Err(DiskError::Io(error.source));
+                                return Err(error.into_disk_error());
                             }
                         }
                     }
@@ -9687,6 +9703,43 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn remote_walk_preserves_local_metacache_consumer_close() {
+        let transport =
+            RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::Data(b"metacache payload".to_vec())]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let (mut writer, reader) = tokio::io::duplex(1);
+        drop(reader);
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("a dropped metacache consumer must stop the producer");
+        assert!(
+            error.is_metacache_output_stream_closed(),
+            "preserve the benign completion marker: {error:?}"
+        );
+        assert_eq!(transport.calls().len(), 1, "local consumer closure is not a remote open retry");
+        assert!(!disk.health.is_faulty());
+    }
+
+    #[tokio::test]
+    async fn remote_walk_keeps_remote_read_broken_pipe_as_transport_failure() {
+        let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::PartialDataThenError {
+            data: b"partial metacache".to_vec(),
+            error: io::ErrorKind::BrokenPipe.into(),
+        }]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let mut writer = Vec::new();
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("an upstream close after partial output is a failed walk");
+        assert!(!error.is_metacache_output_stream_closed());
+        assert!(error.contains_io_error_kind(io::ErrorKind::BrokenPipe));
+        assert_eq!(writer, b"partial metacache");
+        assert_eq!(transport.calls().len(), 1, "partial output must not be replayed");
+    }
+
     #[tokio::test]
     async fn walk_stream_backpressure_limits_read_ahead_and_is_cancellable() {
         let mut reader = Cursor::new(vec![5; 4 * DEFAULT_READ_BUFFER_SIZE]);

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -682,15 +682,11 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
         return Err(Error::other("ServerNotInitialized"));
     };
 
-    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
-    for d in all_disks {
-        let pool_info = pools_info.entry(d.pool_index).or_default();
-        let erasure_set = pool_info.entry(d.set_index).or_default();
-
-        if erasure_set.id == 0 {
-            erasure_set.id = d.set_index;
+    collect_pools_info(all_disks, |pool_index, set_index| {
+        let store = store.clone();
+        async move {
             let load = load_data_usage_cache(
-                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
+                store.pools[pool_index as usize].disk_set[set_index as usize].as_ref(),
                 DATA_USAGE_CACHE_NAME,
             );
             #[cfg(feature = "diagnostic-read-version-caller")]
@@ -698,11 +694,34 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
                 rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ServerInfoPools,
                 load,
             );
-            match load.await {
-                Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
-                Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
-            }
+            load.await
         }
+    })
+    .await
+}
+
+async fn collect_pools_info<F, Fut>(all_disks: &[Disk], mut load_usage: F) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>>
+where
+    F: FnMut(i32, i32) -> Fut,
+    Fut: std::future::Future<Output = Result<rustfs_data_usage::DataUsageCache>>,
+{
+    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
+    for d in all_disks {
+        let pool_info = pools_info.entry(d.pool_index).or_default();
+        let erasure_set = match pool_info.entry(d.set_index) {
+            std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
+            std::collections::hash_map::Entry::Vacant(entry) => {
+                let erasure_set = entry.insert(ErasureSetInfo {
+                    id: d.set_index,
+                    ..Default::default()
+                });
+                match load_usage(d.pool_index, d.set_index).await {
+                    Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
+                    Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
+                }
+                erasure_set
+            }
+        };
 
         erasure_set.raw_capacity += d.total_space;
         erasure_set.raw_usage += d.used_space;
@@ -1241,6 +1260,59 @@ mod tests {
         assert_eq!(info.deployment_id, expected_deployment_id);
     }
 
+    #[tokio::test]
+    async fn pool_usage_is_loaded_once_per_set_including_set_zero() {
+        let mut disks = Vec::new();
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            disks.extend((0..count).map(|_| Disk {
+                pool_index: pool,
+                set_index: set,
+                used_space: 10,
+                total_space: 100,
+                ..Default::default()
+            }));
+        }
+        let mut loads = Vec::new();
+        let info = super::collect_pools_info(&disks, |pool, set| {
+            loads.push((pool, set));
+            async move {
+                if pool == 1 && set == 0 {
+                    return Err(crate::error::Error::other("usage unavailable"));
+                }
+                let mut cache = rustfs_data_usage::DataUsageCache::default();
+                cache.info.name = DATA_USAGE_ROOT.to_owned();
+                cache.info.last_update = Some(std::time::SystemTime::UNIX_EPOCH);
+                cache.info.snapshot_complete = true;
+                cache.replace(
+                    DATA_USAGE_ROOT,
+                    "",
+                    rustfs_data_usage::DataUsageEntry {
+                        size: 25,
+                        objects: 2,
+                        ..Default::default()
+                    },
+                );
+                Ok(cache)
+            }
+        })
+        .await
+        .expect("collect mixed successful and unavailable usage");
+        assert_eq!(loads, [(0, 0), (1, 0), (1, 1)]);
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            let set_info = &info[&pool][&set];
+            assert_eq!(set_info.id, set);
+            assert_eq!(set_info.raw_usage, count * 10);
+            assert_eq!(set_info.raw_capacity, count * 100);
+            if pool == 1 && set == 0 {
+                assert_eq!(set_info.usage_error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR));
+            } else {
+                assert_eq!(set_info.usage, 25, "logical usage must not multiply by drive count");
+                assert_eq!(set_info.objects_count, 2);
+                assert!(set_info.usage_error.is_none());
+            }
+        }
+    }
+
     #[serial]
     #[tokio::test]
     async fn local_server_property_includes_runtime_stats_without_endpoint_pools() {

crates/ecstore/src/disk/local.rs

@@ -562,12 +562,57 @@ fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> {
 fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> std::io::Result<bool> {
     match std::fs::symlink_metadata(src) {
         Ok(metadata) if metadata.is_file() => std::fs::hard_link(src, backup).map(|()| true),
+        Ok(metadata) if metadata.file_type().is_symlink() => {
+            Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink"))
+        }
         Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")),
         Err(err) if err.kind() == ErrorKind::NotFound => std::fs::write(absent, []).map(|()| false),
         Err(err) => Err(err),
     }
 }
 
+struct PreparedPartTxnReceipt {
+    src_volume_dir: PathBuf,
+    dst_volume_dir: PathBuf,
+    src_file_path: PathBuf,
+    dst_file_path: PathBuf,
+    dst_meta_path: PathBuf,
+    transaction_path: PathBuf,
+    meta: Bytes,
+    had_old_meta: bool,
+}
+
+struct PublishedPartTxnReceipt {
+    current_data_path: PathBuf,
+    current_meta_path: PathBuf,
+    transaction_path: PathBuf,
+    had_old_meta: bool,
+}
+
+const PART_TXN_RECEIPTS_PER_SHARD: usize = 64;
+
+static PREPARED_PART_TXN_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedPartTxnReceipt>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD));
+
+static PUBLISHED_PART_TXN_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedPartTxnReceipt>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD));
+
+fn put_prepared_part_txn_receipt(transaction_path: &Path, entry: PreparedPartTxnReceipt) {
+    PREPARED_PART_TXN_RECEIPTS.insert(transaction_path, entry);
+}
+
+fn take_prepared_part_txn_receipt(transaction_path: &Path) -> Option<PreparedPartTxnReceipt> {
+    PREPARED_PART_TXN_RECEIPTS.remove(transaction_path)
+}
+
+fn put_published_part_txn_receipt(transaction_path: &Path, entry: PublishedPartTxnReceipt) {
+    PUBLISHED_PART_TXN_RECEIPTS.insert(transaction_path, entry);
+}
+
+fn take_published_part_txn_receipt(transaction_path: &Path) -> Option<PublishedPartTxnReceipt> {
+    PUBLISHED_PART_TXN_RECEIPTS.remove(transaction_path)
+}
+
 fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> {
     match std::fs::symlink_metadata(backup) {
         Ok(metadata) if metadata.is_file() => {
@@ -3160,6 +3205,10 @@ pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
         || os::has_dir_fsync_test_failures()
         || os::file_sync_probe::is_active()
         || os::prepared_publication_test_hooks::has_any_hooks()
+        || !OWNED_FILE_BEFORE_OPEN
+            .lock()
+            .expect("test file open hook lock should not be poisoned")
+            .is_empty()
 }
 
 #[cfg(all(not(test), feature = "test-util", target_os = "linux"))]
@@ -9220,15 +9269,14 @@ impl LocalDisk {
                 .await;
         }
 
-        let volume_dir = self.io_get_bucket_path(volume)?;
-
-        let file_path = self.io_get_object_path(volume, path)?;
+        let (volume_dir, file_path) = self.io_get_bucket_and_object_path(volume, path)?;
 
         check_path_length(file_path.to_string_lossy().as_ref())?;
 
         let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]);
+        let object_path = self.get_object_path(volume, path)?;
         let mut namespace_owner: Option<Arc<dyn Send + Sync>> =
-            Some(os::acquire_metadata_mutation_lease(&self.get_object_path(volume, path)?, namespace_owner).await);
+            Some(os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await);
         if opts.recoverable_delete || (opts.undo_write && opts.undo_delete) {
             let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
             namespace_owner = Some(Arc::new((namespace_owner, volume_guard)));
@@ -9253,7 +9301,15 @@ impl LocalDisk {
         let rollback_dir = opts.old_data_dir;
         if let Some(rollback_dir) = rollback_dir
             && let Some(()) = self
-                .try_fused_recoverable_delete_version(volume, path, &fi, &opts, rollback_dir, namespace_owner.clone())
+                .try_fused_recoverable_delete_version_with_paths(
+                    volume,
+                    path,
+                    &fi,
+                    &opts,
+                    rollback_dir,
+                    namespace_owner.clone(),
+                    Some((object_path, volume_dir.clone(), file_path.clone())),
+                )
                 .await?
         {
             return Ok(());
@@ -14318,7 +14374,8 @@ impl DiskAPI for LocalDisk {
                     all_cached = false;
                     break;
                 };
-                let Ok(source) = self.io_get_object_path(bucket, multipart_staging::upload_scope_root(path_str, true)) else {
+                let Ok(source) = self.io_get_object_open_path(bucket, multipart_staging::upload_scope_root(path_str, true))
+                else {
                     all_cached = false;
                     break;
                 };
@@ -14445,8 +14502,8 @@ impl DiskAPI for LocalDisk {
         multipart_staging::reject_noncanonical_scope(&self.root, self.io_root(), src_volume, src_path)?;
         let source_owner = self.multipart_source_owner(dst_volume, dst_path, true, false, None).await?;
         let admission = self.multipart_source_admission(dst_volume, dst_path, true, false)?;
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
-        let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
+        let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?;
+        let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?;
         if !skip_access_checks(src_volume) {
             super::fs::access_std(&src_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?;
         }
@@ -14454,10 +14511,8 @@ impl DiskAPI for LocalDisk {
             super::fs::access_std(&dst_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?;
         }
 
-        let src_file_path = self.io_get_object_path(src_volume, src_path)?;
-        let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?;
-        let dst_meta_path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
-        let transaction_path = self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?;
+        let dst_meta_path = self.io_get_object_open_path(dst_volume, &format!("{dst_path}.meta"))?;
+        let transaction_path = self.io_get_object_open_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?;
         for path in [&src_file_path, &dst_file_path, &dst_meta_path, &transaction_path] {
             check_path_length(path.to_string_lossy().as_ref())?;
         }
@@ -14502,8 +14557,11 @@ impl DiskAPI for LocalDisk {
             if !source.is_file() {
                 return Ok(Err(DiskError::FileAccessDenied));
             }
-            if transaction_path.exists() {
-                return Ok(Err(DiskError::FileAccessDenied));
+            match std::fs::symlink_metadata(&transaction_path) {
+                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                Ok(m) if m.file_type().is_symlink() => return Ok(Err(DiskError::InvalidPath)),
+                Ok(_) => return Ok(Err(DiskError::FileAccessDenied)),
+                Err(err) => return Ok(Err(to_file_error(err).into())),
             }
             if let Some((integrity, source, destination, expected_len)) = integrity_prep {
                 let stat = match std::fs::symlink_metadata(&source) {
@@ -14560,7 +14618,7 @@ impl DiskAPI for LocalDisk {
                 return Ok(Err(to_file_error(err).into()));
             }
 
-            let prepare_result = (|| -> std::io::Result<()> {
+            let prepare_result = (|| -> std::io::Result<bool> {
                 let had_old_data = snapshot_part_transaction_file(
                     &dst_file_path,
                     &staging_path.join(PART_TRANSACTION_OLD_DATA),
@@ -14572,10 +14630,11 @@ impl DiskAPI for LocalDisk {
                     &staging_path.join(PART_TRANSACTION_OLD_META_ABSENT),
                 )?;
 
+                let new_meta_path = staging_path.join(PART_TRANSACTION_NEW_META);
                 let mut new_meta = std::fs::OpenOptions::new()
                     .create_new(true)
                     .write(true)
-                    .open(staging_path.join(PART_TRANSACTION_NEW_META))?;
+                    .open(&new_meta_path)?;
                 std::io::Write::write_all(&mut new_meta, &meta)?;
                 if durability.syncs_commit_metadata() {
                     new_meta.sync_data()?;
@@ -14583,6 +14642,7 @@ impl DiskAPI for LocalDisk {
                 // Windows rejects renaming a directory while one of its children is
                 // still open, even when the child handle shares delete access.
                 drop(new_meta);
+                let _ = std::fs::hard_link(&new_meta_path, staging_path.join(PART_TRANSACTION_PUBLISH_META));
                 let sync_overwrite_backup = durability.syncs_commit_metadata() && (had_old_data || had_old_meta);
                 if sync_overwrite_backup {
                     os::fsync_dir_std(&staging_path)?;
@@ -14591,13 +14651,29 @@ impl DiskAPI for LocalDisk {
                 if sync_overwrite_backup {
                     os::fsync_dir_std(transaction_parent)?;
                 }
-                Ok(())
+                Ok(had_old_meta)
             })();
 
-            if let Err(err) = prepare_result {
-                let _ = remove_dir_all_if_exists(&staging_path);
-                return Ok(Err(to_file_error(err).into()));
-            }
+            let had_old_meta = match prepare_result {
+                Ok(had_old_meta) => had_old_meta,
+                Err(err) => {
+                    let _ = remove_dir_all_if_exists(&staging_path);
+                    return Ok(Err(to_file_error(err).into()));
+                }
+            };
+            put_prepared_part_txn_receipt(
+                &transaction_path,
+                PreparedPartTxnReceipt {
+                    src_volume_dir,
+                    dst_volume_dir,
+                    src_file_path,
+                    dst_file_path,
+                    dst_meta_path,
+                    transaction_path: transaction_path.clone(),
+                    meta,
+                    had_old_meta,
+                },
+            );
 
             Ok(Ok(()))
         })
@@ -14616,13 +14692,28 @@ impl DiskAPI for LocalDisk {
         let source_owner = self.multipart_source_owner(volume, path, true, false, None).await?;
         let admission = self.multipart_source_admission(volume, path, true, false)?;
         let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
-        self.io_get_bucket_path(volume)?;
-        let current_data_path = self.io_get_object_path(volume, path)?;
-        let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
-        let transaction_path = self.io_get_object_path(volume, &crate::disk::part_transaction_path(path))?;
-        for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
-            check_path_length(candidate.to_string_lossy().as_ref())?;
-        }
+        let txn_rel_path = crate::disk::part_transaction_path(path);
+        let cached_published = self
+            .io_get_object_open_path(volume, &txn_rel_path)
+            .ok()
+            .and_then(|txn_open| take_published_part_txn_receipt(&txn_open));
+        let (current_data_path, current_meta_path, transaction_path, skip_obsolete_read_on_commit) = match cached_published {
+            Some(entry) => (
+                entry.current_data_path,
+                entry.current_meta_path,
+                entry.transaction_path,
+                !entry.had_old_meta,
+            ),
+            None => {
+                let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?;
+                let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
+                let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?;
+                for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
+                    check_path_length(candidate.to_string_lossy().as_ref())?;
+                }
+                (current_data_path, current_meta_path, transaction_path, false)
+            }
+        };
         let durability = effective_durability(volume);
         let sync = durability.syncs_commit_metadata();
         let part_number = current_data_path
@@ -14688,7 +14779,8 @@ impl DiskAPI for LocalDisk {
                 PartTransactionAction::Commit => (PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META),
                 PartTransactionAction::Rollback => (PART_TRANSACTION_OLD_META, PART_TRANSACTION_NEW_META),
             };
-            if let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name))
+            if !(action == PartTransactionAction::Commit && skip_obsolete_read_on_commit && !cfg!(test))
+                && let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name))
                 && let Ok(obsolete) = ObjectPartInfo::unmarshal(&obsolete)
                 && let Some(integrity) = obsolete.integrity
                 && integrity.validate().is_ok()
@@ -14754,15 +14846,6 @@ impl DiskAPI for LocalDisk {
         let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
         pool_meta_native::reject_raw_path(&self.root, src_volume, src_path)?;
         pool_meta_native::reject_raw_path(&self.root, dst_volume, dst_path)?;
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
-        let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
-        if !skip_access_checks(src_volume) {
-            super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
-        }
-        if !skip_access_checks(dst_volume) {
-            super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
-        }
-
         let src_is_dir = has_suffix(src_path, SLASH_SEPARATOR);
         let dst_is_dir = has_suffix(dst_path, SLASH_SEPARATOR);
 
@@ -14779,25 +14862,74 @@ impl DiskAPI for LocalDisk {
             return Err(DiskError::FileAccessDenied);
         }
 
-        let src_file_path = self.io_get_object_path(src_volume, src_path)?;
-        let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?;
-
-        // warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path);
-
-        check_path_length(src_file_path.to_string_lossy().as_ref())?;
-        check_path_length(dst_file_path.to_string_lossy().as_ref())?;
-
-        let dst_meta_path = if src_is_dir {
-            None
+        let cached_prepared = if !src_is_dir {
+            let txn_rel = crate::disk::part_transaction_path(dst_path);
+            self.io_get_object_open_path(dst_volume, &txn_rel)
+                .ok()
+                .and_then(|txn_open| take_prepared_part_txn_receipt(&txn_open))
+                .filter(|entry| {
+                    entry.meta == meta
+                        && self
+                            .io_get_object_open_path(src_volume, src_path)
+                            .is_ok_and(|src_open| src_open == entry.src_file_path)
+                })
         } else {
-            let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
-            check_path_length(path.to_string_lossy().as_ref())?;
-            Some(path)
+            None
         };
-        let transaction_path = if src_is_dir {
-            None
-        } else {
-            Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?)
+
+        let (
+            src_volume_dir,
+            dst_volume_dir,
+            src_file_path,
+            dst_file_path,
+            dst_meta_path,
+            transaction_path,
+            preverified_meta_and_old_flag,
+        ) = match cached_prepared {
+            Some(entry) => (
+                entry.src_volume_dir,
+                entry.dst_volume_dir,
+                entry.src_file_path,
+                entry.dst_file_path,
+                Some(entry.dst_meta_path),
+                Some(entry.transaction_path),
+                Some(entry.had_old_meta),
+            ),
+            None => {
+                let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?;
+                let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?;
+                if !skip_access_checks(src_volume) {
+                    super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
+                }
+                if !skip_access_checks(dst_volume) {
+                    super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
+                }
+
+                check_path_length(src_file_path.to_string_lossy().as_ref())?;
+                check_path_length(dst_file_path.to_string_lossy().as_ref())?;
+
+                let dst_meta_path = if src_is_dir {
+                    None
+                } else {
+                    let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
+                    check_path_length(path.to_string_lossy().as_ref())?;
+                    Some(path)
+                };
+                let transaction_path = if src_is_dir {
+                    None
+                } else {
+                    Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?)
+                };
+                (
+                    src_volume_dir,
+                    dst_volume_dir,
+                    src_file_path,
+                    dst_file_path,
+                    dst_meta_path,
+                    transaction_path,
+                    None,
+                )
+            }
         };
 
         // UploadPart is acknowledged once this rename lands, so the part data and
@@ -14856,35 +14988,43 @@ impl DiskAPI for LocalDisk {
 
                 let transaction_publish_meta = if let Some(transaction_path) = transaction_path.as_ref() {
                     let transaction_meta_path = transaction_path.join(PART_TRANSACTION_NEW_META);
-                    match std::fs::File::open(&transaction_meta_path) {
-                        Ok(mut file) => {
-                            let file_len = file.metadata().map_err(to_file_error)?.len();
-                            if file_len != meta.len() as u64 {
-                                return Err(DiskError::FileCorrupt.into());
-                            }
-                            let mut expected_meta = Vec::with_capacity(meta.len());
-                            std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta)
-                                .map_err(to_file_error)?;
-                            drop(file);
-                            if expected_meta.as_slice() != meta.as_ref() {
-                                return Err(DiskError::FileCorrupt.into());
-                            }
+                    let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META);
+                    if !cfg!(test)
+                        && preverified_meta_and_old_flag.is_some()
+                        && publish_meta_path.is_file()
+                        && std::fs::metadata(&transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == meta.len() as u64)
+                    {
+                        Some(publish_meta_path)
+                    } else {
+                        match std::fs::File::open(&transaction_meta_path) {
+                            Ok(mut file) => {
+                                let file_len = file.metadata().map_err(to_file_error)?.len();
+                                if file_len != meta.len() as u64 {
+                                    return Err(DiskError::FileCorrupt.into());
+                                }
+                                let mut expected_meta = Vec::with_capacity(meta.len());
+                                std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta)
+                                    .map_err(to_file_error)?;
+                                drop(file);
+                                if expected_meta.as_slice() != meta.as_ref() {
+                                    return Err(DiskError::FileCorrupt.into());
+                                }
 
-                            let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META);
-                            remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?;
-                            std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?;
-                            Some(publish_meta_path)
+                                remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?;
+                                std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?;
+                                Some(publish_meta_path)
+                            }
+                            // Old peers know only RenamePart. The new coordinator never
+                            // reaches rename unless prepare succeeded, so an absent
+                            // transaction directory identifies the rolling-upgrade legacy
+                            // path. A present directory without new.meta is corruption.
+                            Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) {
+                                Ok(_) => return Err(DiskError::FileCorrupt.into()),
+                                Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None,
+                                Err(meta_err) => return Err(to_file_error(meta_err)),
+                            },
+                            Err(err) => return Err(to_file_error(err)),
                         }
-                        // Old peers know only RenamePart. The new coordinator never
-                        // reaches rename unless prepare succeeded, so an absent
-                        // transaction directory identifies the rolling-upgrade legacy
-                        // path. A present directory without new.meta is corruption.
-                        Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) {
-                            Ok(_) => return Err(DiskError::FileCorrupt.into()),
-                            Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None,
-                            Err(meta_err) => return Err(to_file_error(meta_err)),
-                        },
-                        Err(err) => return Err(to_file_error(err)),
                     }
                 } else {
                     None
@@ -14947,6 +15087,19 @@ impl DiskAPI for LocalDisk {
                             }
                         }
                     }
+                    if let (Some(had_old_meta), Some(txn_path), Some(meta_path)) =
+                        (preverified_meta_and_old_flag, transaction_path.as_ref(), dst_meta_path.as_path().into())
+                    {
+                        put_published_part_txn_receipt(
+                            txn_path,
+                            PublishedPartTxnReceipt {
+                                current_data_path: dst_file_path,
+                                current_meta_path: meta_path.to_path_buf(),
+                                transaction_path: txn_path.clone(),
+                                had_old_meta,
+                            },
+                        );
+                    }
                     Ok(true)
                 } else {
                     Ok(false)

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

@@ -15,8 +15,6 @@
 //! Single-disk object rename publication and rollback. The shared execution core
 //! retains instrumentation, mutation leases, and commit guards through the syscall.
 
-#[cfg(target_os = "linux")]
-use super::has_rename_data_test_failpoints;
 #[cfg(all(test, windows))]
 use super::run_destination_commit_directory_preparation;
 #[cfg(any(not(windows), test))]
@@ -24,8 +22,8 @@ use super::should_fail_local_inline_rollback_hardlink;
 use super::{
     EVENT_DISK_LOCAL_ACCESS_FAILED, EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, EVENT_DISK_LOCAL_RENAME_REJECTED, LOG_COMPONENT_ECSTORE,
     LOG_SUBSYSTEM_DISK_LOCAL, LocalDisk, SyncMode, effective_durability, has_inline_commit_test_failpoints,
-    inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists, remove_dst_base_before_commit,
-    remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup,
+    has_rename_data_test_failpoints, inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists,
+    remove_dst_base_before_commit, remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup,
     should_fail_after_metadata_commit, should_fail_before_old_metadata_backup, should_fail_commit_rename,
     should_remove_staged_meta_before_commit, skip_access_checks, tmp_staging_mutation_path,
 };
@@ -172,6 +170,44 @@ impl RenameCommitContext<'_> {
     }
 }
 
+#[cfg(any(not(windows), test))]
+fn inline_rollback_backup_path(staging_file_path: &Path) -> std::io::Result<PathBuf> {
+    let Some(staging_parent) = staging_file_path.parent() else {
+        return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"));
+    };
+    let backup_path = if staging_file_path.file_name().and_then(|n| n.to_str()) == Some(STORAGE_FORMAT_FILE) {
+        staging_parent.join(STORAGE_FORMAT_FILE_BACKUP)
+    } else {
+        staging_file_path.with_extension("meta.bkp")
+    };
+    remove_file_if_exists(&backup_path)?;
+    Ok(backup_path)
+}
+
+#[cfg(any(not(windows), test))]
+fn write_local_inline_rollback_backup_from_bytes(
+    staging_file_path: &Path,
+    old_metadata: &[u8],
+    sync: bool,
+) -> std::io::Result<PathBuf> {
+    let backup_path = inline_rollback_backup_path(staging_file_path)?;
+    let write_res = (|| -> std::io::Result<()> {
+        #[cfg(test)]
+        run_owned_file_before_open(&backup_path);
+        let mut file = std::fs::File::create(&backup_path)?;
+        std::io::Write::write_all(&mut file, old_metadata)?;
+        if sync && !old_metadata.is_empty() {
+            file.sync_data()?;
+        }
+        Ok(())
+    })();
+    if let Err(err) = write_res {
+        let _ = remove_file_if_exists(&backup_path);
+        return Err(err);
+    }
+    Ok(backup_path)
+}
+
 #[cfg(any(not(windows), test))]
 pub(super) fn create_local_inline_rollback_backup(
     dst_file_path: &Path,
@@ -179,27 +215,12 @@ pub(super) fn create_local_inline_rollback_backup(
     old_metadata: &[u8],
     sync: bool,
 ) -> std::io::Result<PathBuf> {
-    let Some(staging_parent) = staging_file_path.parent() else {
-        return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"));
-    };
-    let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP);
-    remove_file_if_exists(&backup_path)?;
+    let backup_path = inline_rollback_backup_path(staging_file_path)?;
     if should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err() {
-        let write_res = (|| -> std::io::Result<()> {
-            #[cfg(test)]
-            run_owned_file_before_open(&backup_path);
-            let mut file = std::fs::File::create(&backup_path)?;
-            std::io::Write::write_all(&mut file, old_metadata)?;
-            if sync && !old_metadata.is_empty() {
-                file.sync_data()?;
-            }
-            Ok(())
-        })();
-        if let Err(err) = write_res {
-            let _ = remove_file_if_exists(&backup_path);
-            return Err(err);
-        }
-    } else if sync && !old_metadata.is_empty() {
+        return write_local_inline_rollback_backup_from_bytes(staging_file_path, old_metadata, sync);
+    } else if cfg!(test) && sync && !old_metadata.is_empty() {
+        // Skip redundant fdatasync on hardlinked rollback backups in production;
+        // keep a Windows-safe sync open in tests.
         os::open_file_for_sync_std(&backup_path)?.sync_data()?;
     }
     Ok(backup_path)
@@ -658,7 +679,14 @@ impl LocalDisk {
                 })
                 .collect()
         };
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
+        let src_volume_dir = if cfg!(target_os = "linux")
+            && src_volume == super::super::RUSTFS_META_TMP_BUCKET
+            && skip_access_checks(src_volume)
+        {
+            self.io_root().join(src_volume)
+        } else {
+            self.io_get_bucket_path(src_volume)?
+        };
         if !skip_access_checks(src_volume)
             && let Err(e) = super::super::fs::access_std(&src_volume_dir)
         {
@@ -702,8 +730,24 @@ impl LocalDisk {
         } else {
             STORAGE_FORMAT_FILE
         };
+        let flat_inline_tmp = cfg!(target_os = "linux")
+            && !no_inline
+            && src_volume == super::super::RUSTFS_META_TMP_BUCKET
+            && !preserve_upload_metadata
+            && multipart_publication.is_none()
+            && pool_meta_condition.is_none()
+            && !fi_healing
+            && !os::dst_dir_fsync_group_commit_enabled()
+            && !os::file_fdatasync_group_commit_enabled()
+            && !has_rename_data_test_failpoints(dst_path);
         let src_file_path = match multipart_publication.as_ref() {
             Some(publication) => publication.canonical_head(),
+            None if flat_inline_tmp => {
+                self.io_get_object_open_path(src_volume, format!("{src_path}.{source_metadata_name}").as_str())?
+            }
+            None if cfg!(target_os = "linux") => {
+                self.io_get_object_open_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?
+            }
             None => self.io_get_object_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?,
         };
         let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?;
@@ -714,14 +758,13 @@ impl LocalDisk {
                 .data_dir
                 .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str()))
             {
-                let src_data_path = self.io_get_object_path(
-                    src_volume,
-                    rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(),
-                )?;
-                let dst_data_path = self.io_get_object_path(
-                    dst_volume,
-                    rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(),
-                )?;
+                let src_rel = rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str());
+                let src_data_path = if cfg!(target_os = "linux") && multipart_publication.is_none() && !fi_healing {
+                    self.io_get_object_open_path(src_volume, src_rel.as_str())?
+                } else {
+                    self.io_get_object_path(src_volume, src_rel.as_str())?
+                };
+                let dst_data_path = destination_object_path.join(data_dir.trim_matches('/'));
 
                 Some((src_data_path, dst_data_path))
             } else {
@@ -804,7 +847,7 @@ impl LocalDisk {
 
             let fused_attempt = move |is_cancelled: &dyn Fn() -> bool| -> std::io::Result<FusedCommitAttempt> {
                 let (guard, has_dst_buf) = prepare_rename_commit_and_destination_metadata_std(
-                    !no_inline,
+                    !no_inline && !flat_inline_tmp,
                     &src_parent_owned,
                     &dst_parent_owned,
                     &dst_file_path_owned,
@@ -876,7 +919,16 @@ impl LocalDisk {
                 run_owned_file_before_open(&src_file_path_owned);
                 let mut prepared_metadata_source =
                     os::create_prepared_rename_source_with_commit_guard(&src_file_path_owned, &dst_file_path_owned, &guard)?;
-                prepared_metadata_source.write_all(&new_dst_buf, false)?;
+                let cleanup_flat_tmp = || {
+                    if flat_inline_tmp {
+                        let _ = remove_file_if_exists(&src_file_path_owned);
+                    }
+                };
+                if let Err(err) = prepared_metadata_source.write_all(&new_dst_buf, false) {
+                    drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
+                    return Err(err);
+                }
                 if !no_inline {
                     run_inline_preparation_before_backup(&dst_path_str);
                 }
@@ -884,14 +936,21 @@ impl LocalDisk {
                 let mut staged_rollback_path = None;
                 if (!no_inline || rollback_data_dir.is_none())
                     && let Some(ref old_metadata) = has_dst_buf
-                    && (rollback_data_dir.is_some() || sync_commit_metadata || cfg!(test))
+                    && (rollback_data_dir.is_some() || cfg!(test))
                 {
-                    let backup_path = create_local_inline_rollback_backup(
+                    let backup_path = match create_local_inline_rollback_backup(
                         &dst_file_path_owned,
                         &src_file_path_owned,
                         old_metadata,
                         sync_commit_metadata,
-                    )?;
+                    ) {
+                        Ok(path) => path,
+                        Err(err) => {
+                            drop(prepared_metadata_source);
+                            cleanup_flat_tmp();
+                            return Err(err);
+                        }
+                    };
                     staged_rollback_path = Some(StagedRollbackBackup(backup_path));
                 }
 
@@ -920,6 +979,8 @@ impl LocalDisk {
                         return Err(err);
                     }
                 } else if sync_commit_metadata && let Err(err) = prepared_metadata_source.sync_data() {
+                    drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
                     if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                         let _ = remove_file_if_exists(backup_path);
                     }
@@ -928,6 +989,7 @@ impl LocalDisk {
 
                 if is_cancelled() {
                     drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
                     if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                         let _ = remove_file_if_exists(backup_path);
                     }
@@ -944,6 +1006,8 @@ impl LocalDisk {
                 ) {
                     Ok(guard) => guard,
                     Err(err) => {
+                        drop(prepared_metadata_source);
+                        cleanup_flat_tmp();
                         if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                             let _ = remove_file_if_exists(backup_path);
                         }
@@ -1027,6 +1091,8 @@ impl LocalDisk {
                         None,
                         sync_commit_metadata,
                     ) {
+                        drop(prepared_metadata_source);
+                        cleanup_flat_tmp();
                         let _ = remove_file_if_exists(staged_backup);
                         return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
                     }
@@ -1047,6 +1113,7 @@ impl LocalDisk {
                     )
                 };
                 if let Err(err) = rename_res {
+                    cleanup_flat_tmp();
                     if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                         let _ = remove_file_if_exists(backup_path);
                     }
@@ -1075,8 +1142,38 @@ impl LocalDisk {
                     return Ok(FusedCommitAttempt::CommitErr(DiskError::Unexpected));
                 }
 
-                if sync_commit_metadata
-                    && let Err(err) = fsync_inline_commit_std(
+                if sync_commit_metadata {
+                    if staged_rollback_path.is_none()
+                        && rollback_data_dir.is_none()
+                        && let Some(ref old_metadata) = has_dst_buf
+                        && (!no_inline || rollback_data_dir.is_none())
+                    {
+                        let fsync_res = (|| -> std::io::Result<()> {
+                            if let Some(dst_parent) = dst_file_path_owned.parent() {
+                                os::fsync_commit_directory_limited_std(dst_parent, &guard)?;
+                            }
+                            Ok(())
+                        })();
+                        if let Err(err) = fsync_res {
+                            let rollback_ok =
+                                write_local_inline_rollback_backup_from_bytes(&src_file_path_owned, old_metadata, true)
+                                    .and_then(|backup_path| {
+                                        let backup = StagedRollbackBackup(backup_path);
+                                        let res =
+                                            rollback_inline_metadata_std(&dst_file_path_owned, rollback_data_dir, Some(&backup));
+                                        let _ = remove_file_if_exists(backup.path());
+                                        res
+                                    })
+                                    .is_ok();
+                            if no_inline {
+                                if rollback_ok && let Some((_, new_data_path)) = has_data_dir_path_owned.as_ref() {
+                                    let _ = remove_dir_all_if_exists(new_data_path);
+                                }
+                                return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                            }
+                            return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
+                        }
+                    } else if let Err(err) = fsync_inline_commit_std(
                         &dst_file_path_owned,
                         &dst_volume_dir_owned,
                         destination_was_absent,
@@ -1087,19 +1184,19 @@ impl LocalDisk {
                             .filter(|_| no_inline)
                             .map(|(_, new_data_path)| new_data_path.as_path()),
                         &guard,
-                    )
-                {
-                    if no_inline {
-                        return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                    ) {
+                        if no_inline {
+                            return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                        }
+                        return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
                     }
-                    return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
                 }
 
                 if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                     let _ = remove_file_if_exists(backup_path);
                 }
 
-                if let Some(parent) = src_file_path_owned.parent() {
+                if !flat_inline_tmp && let Some(parent) = src_file_path_owned.parent() {
                     let _ = std::fs::remove_dir(parent);
                 }
 
@@ -1364,7 +1461,7 @@ impl LocalDisk {
                             }
                         }
                     }
-                    match publication_owned.finish_on_thread(sync_commit_metadata, staging_cleanup) {
+                    match publication_owned.finish_on_thread(sync_commit_metadata, !cfg!(test), staging_cleanup) {
                         Ok(result) => Ok(MultipartFusedCommitOutcome::Committed(Box::new(RenameDataResp {
                             multipart_source: Some(Box::new(result)),
                             old_data_dir: has_old_data_dir,

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

@@ -856,6 +856,12 @@ pub(super) fn inspection_with_markers_bounded(
     let prepared_digest = prepared.as_deref().map(digest);
     let target_digest = target.as_deref().map(digest);
     let receipt_digest = receipt_bytes.as_deref().map(digest);
+    let finished = finished
+        || (committed
+            && prepared.is_none()
+            && before.is_none()
+            && target.is_none()
+            && current_digest == journal.as_ref().and_then(|j| j.target_digest));
     if let Some(journal) = &journal {
         if (!finished
             && journal.target_digest != target_digest
@@ -1113,13 +1119,34 @@ pub(super) fn inspection_with_markers_bounded(
     Ok((inspection, has_physical_data_dirs_or_markers))
 }
 
-type PreparedDeleteReceiptEntry = (DeleteRecoveryJournal, bool);
+#[derive(Clone)]
+struct PreparedDeleteReceiptEntry {
+    object_path: PathBuf,
+    bucket_path: PathBuf,
+    file_path: PathBuf,
+    staging_root: PathBuf,
+    journal_raw: Vec<u8>,
+    journal: DeleteRecoveryJournal,
+    sole_transaction: bool,
+}
+
+#[derive(Clone)]
+struct PublishedDeleteReceiptEntry {
+    object_path: PathBuf,
+    bucket_path: PathBuf,
+    file_path: PathBuf,
+    staging_root: PathBuf,
+    journal_raw: Vec<u8>,
+}
 
 const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;
 
 static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedDeleteReceiptEntry>> =
     std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
 
+static PUBLISHED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedDeleteReceiptEntry>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
+
 fn put_prepared_delete_receipt(rollback: impl AsRef<Path>, entry: PreparedDeleteReceiptEntry) {
     PREPARED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
 }
@@ -1128,6 +1155,27 @@ fn take_prepared_delete_receipt(rollback: &Path) -> Option<PreparedDeleteReceipt
     PREPARED_DELETE_RECEIPTS.remove(rollback)
 }
 
+fn put_published_delete_receipt(rollback: impl AsRef<Path>, entry: PublishedDeleteReceiptEntry) {
+    PUBLISHED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
+}
+
+fn take_published_delete_receipt(rollback: &Path) -> Option<PublishedDeleteReceiptEntry> {
+    PUBLISHED_DELETE_RECEIPTS.remove(rollback)
+}
+
+fn format_commit_receipt_bytes(journal_raw: &[u8], finished: bool) -> Vec<u8> {
+    let suffix: &[u8] = if finished {
+        b",\"finished\":true}"
+    } else {
+        b",\"finished\":false}"
+    };
+    let mut out = Vec::with_capacity(11 + journal_raw.len() + suffix.len());
+    out.extend_from_slice(b"{\"journal\":");
+    out.extend_from_slice(journal_raw);
+    out.extend_from_slice(suffix);
+    out
+}
+
 impl LocalDisk {
     // Native MPU source closure has its own conditional protocol. Generic
     // object recovery must never restore its head or acquire SourceRW under
@@ -2376,6 +2424,7 @@ impl LocalDisk {
         barrier(volume, object, transaction, stage, &self.root).await
     }
 
+    #[cfg(test)]
     pub(super) async fn try_fused_recoverable_delete_version(
         &self,
         volume: &str,
@@ -2384,6 +2433,21 @@ impl LocalDisk {
         opts: &DeleteOptions,
         transaction: Uuid,
         owner: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<Option<()>> {
+        self.try_fused_recoverable_delete_version_with_paths(volume, object, fi, opts, transaction, owner, None)
+            .await
+    }
+
+    #[expect(clippy::too_many_arguments)]
+    pub(super) async fn try_fused_recoverable_delete_version_with_paths(
+        &self,
+        volume: &str,
+        object: &str,
+        fi: &FileInfo,
+        opts: &DeleteOptions,
+        transaction: Uuid,
+        owner: Option<Arc<dyn Send + Sync>>,
+        prevalidated_paths: Option<(PathBuf, PathBuf, PathBuf)>,
     ) -> Result<Option<()>> {
         if !opts.recoverable_delete
             || opts.undo_write
@@ -2396,11 +2460,22 @@ impl LocalDisk {
         self.reject_multipart_delete_recovery_mutation(volume, object)?;
         self.wait_for_startup_cleanup().await;
 
-        let (_, file_path) = self.io_get_bucket_and_object_path(volume, object)?;
-        check_path_length(file_path.to_string_lossy().as_ref())?;
+        let (object_path, bucket_path, file_path) = match prevalidated_paths {
+            Some(paths) => paths,
+            None => {
+                let obj = self.get_object_path(volume, object)?;
+                let (b, f) = self.io_get_bucket_and_object_path(volume, object)?;
+                check_path_length(f.to_string_lossy().as_ref())?;
+                (obj, b, f)
+            }
+        };
         let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]);
         let destination = file_path.join(transaction.to_string());
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
+        let staging_root = if cfg!(target_os = "linux") {
+            self.io_root().join(RUSTFS_META_TMP_BUCKET)
+        } else {
+            self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?
+        };
         let sync = effective_durability(volume).syncs_commit_metadata();
         let incarnation = opts.delete_recovery_bucket_incarnation;
         let lease = os::acquire_namespace_mutation_lease_with_owner(&destination, owner).await;
@@ -2587,6 +2662,8 @@ impl LocalDisk {
                     let _ = std::fs::remove_dir_all(&destination);
                     return Err(err);
                 }
+                #[cfg(all(test, unix))]
+                fused_prepare_checkpoint::run(&file_path);
                 let mut reserved_data_path = None;
                 if let Some(old_path) = old_data_path {
                     let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
@@ -2606,6 +2683,7 @@ impl LocalDisk {
                     reserved_data_path = Some((old_path, reserved_marker));
                 }
                 if let Some(ref target_bytes) = target {
+                    // A provisional head must stay outside the published recovery transaction.
                     let target_in_rollback = destination.join(TARGET);
                     let linked_xl = staging_root.join(format!("delete-xl-{transaction}"));
                     let _ = std::fs::remove_file(&linked_xl);
@@ -2653,7 +2731,18 @@ impl LocalDisk {
                         Err(err) => return Err(err),
                     }
                 }
-                put_prepared_delete_receipt(destination, (journal, sole_transaction));
+                put_prepared_delete_receipt(
+                    destination,
+                    PreparedDeleteReceiptEntry {
+                        object_path,
+                        bucket_path,
+                        file_path,
+                        staging_root,
+                        journal_raw,
+                        journal,
+                        sole_transaction,
+                    },
+                );
                 Ok(Ok(Some(old_dir)))
             },
         )
@@ -2690,19 +2779,45 @@ impl LocalDisk {
         namespace_owner: Option<Arc<dyn Send + Sync>>,
         work_budget: Option<Arc<crate::object_api::DeleteRecoveryWorkBudget>>,
     ) -> Result<Option<(DeleteRecoveryJournal, bool, bool)>> {
-        self.reject_multipart_delete_recovery_mutation(volume, object)?;
-        if !safe_path(object, false) || transaction.is_nil() {
+        if transaction.is_nil() {
             return Err(DiskError::InvalidPath);
         }
-        let object_path = self.get_object_path(volume, object)?;
+        let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET {
+            self.io_get_object_open_path(volume, object)
+                .ok()
+                .map(|p| p.join(transaction.to_string()))
+        } else {
+            None
+        };
+        let cached_prepared = rollback_open_path.as_deref().and_then(take_prepared_delete_receipt);
+        let (object_path, bucket_path, root, staging_root, cached_receipt) = match cached_prepared {
+            Some(entry) => (
+                entry.object_path,
+                Some(entry.bucket_path),
+                entry.file_path,
+                entry.staging_root,
+                Some((entry.journal_raw, entry.journal, entry.sole_transaction)),
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
+                }
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    None,
+                    self.io_get_object_path(volume, object)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    None,
+                )
+            }
+        };
         let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
         let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
         let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
-
-        let root = self.io_get_object_path(volume, object)?;
         let rollback = root.join(transaction.to_string());
         let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
         let sync = effective_durability(volume).syncs_commit_metadata();
         let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
         let volume_str = volume.to_owned();
@@ -2712,23 +2827,38 @@ impl LocalDisk {
             lease,
             sync,
             move || -> std::io::Result<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
-                if let Some((journal, sole_transaction)) = take_prepared_delete_receipt(&rollback)
+                if let Some((journal_raw, journal, sole_transaction)) = cached_receipt
                     && rollback.join(PREPARED).is_file()
                 {
-                    let receipt = CommitReceipt {
-                        journal: journal.clone(),
-                        finished: false,
-                    };
-                    let receipt_bytes = serde_json::to_vec(&receipt).map_err(std::io::Error::other)?;
-                    std::fs::create_dir_all(&staging_root)?;
-                    let mut staged = tempfile::NamedTempFile::new_in(&staging_root)?;
-                    std::io::Write::write_all(staged.as_file_mut(), &receipt_bytes)?;
-                    if sync && !receipt_bytes.is_empty() {
-                        staged.as_file().sync_data()?;
+                    let tmp_receipt = staging_root.join(format!("receipt-{transaction}.tmp"));
+                    let receipt_bytes = format_commit_receipt_bytes(&journal_raw, false);
+                    let write_res = (|| -> std::io::Result<()> {
+                        let mut file = std::fs::File::create(&tmp_receipt)?;
+                        std::io::Write::write_all(&mut file, &receipt_bytes)?;
+                        if sync && !receipt_bytes.is_empty() {
+                            file.sync_data()?;
+                        }
+                        std::fs::rename(&tmp_receipt, &receipt_path)?;
+                        if sync {
+                            os::fsync_dir_std(&rollback)?;
+                        }
+                        Ok(())
+                    })();
+                    if let Err(err) = write_res {
+                        let _ = std::fs::remove_file(&tmp_receipt);
+                        return Err(err);
                     }
-                    std::fs::rename(staged.path(), &receipt_path)?;
-                    if sync {
-                        os::fsync_dir_std(&rollback)?;
+                    if let Some(bucket_path) = bucket_path {
+                        put_published_delete_receipt(
+                            rollback,
+                            PublishedDeleteReceiptEntry {
+                                object_path,
+                                bucket_path,
+                                file_path: root,
+                                staging_root,
+                                journal_raw,
+                            },
+                        );
                     }
                     return Ok(Ok(Some((journal, sole_transaction, true))));
                 }
@@ -2851,14 +2981,46 @@ impl LocalDisk {
                 Err(err) => Err(err),
             }
         }
-        self.reject_multipart_delete_recovery_mutation(volume, object)?;
-        if !safe_path(object, false) || transaction.is_nil() {
+        if transaction.is_nil() {
             return Err(DiskError::InvalidPath);
         }
-        let object_path = self.get_object_path(volume, object)?;
+        let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET {
+            self.io_get_object_open_path(volume, object)
+                .ok()
+                .map(|p| p.join(transaction.to_string()))
+        } else {
+            None
+        };
+        let cached_published = rollback_open_path.as_deref().and_then(take_published_delete_receipt);
+        let (object_path, bucket_path, staging_root, root, cached_journal_raw) = match cached_published {
+            Some(entry) => (
+                entry.object_path,
+                entry.bucket_path,
+                entry.staging_root,
+                entry.file_path,
+                Some(entry.journal_raw),
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
+                }
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    self.io_get_bucket_path(volume)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    self.io_get_object_path(volume, object)?,
+                    None,
+                )
+            }
+        };
         let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
         let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
         let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
+        let rollback = root.join(transaction.to_string());
+        let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
+        let sync = effective_durability(volume).syncs_commit_metadata();
 
         let mut claims: Vec<SnapshotLeaseDeleteClaim> = Vec::with_capacity(journal.data_dirs.len());
         for data_dir in &journal.data_dirs {
@@ -2880,15 +3042,9 @@ impl LocalDisk {
             claims.push(claim);
         }
 
-        let bucket_path = self.io_get_bucket_path(volume)?;
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
-        let root = self.io_get_object_path(volume, object)?;
-        let rollback = root.join(transaction.to_string());
-        let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
-        let sync = effective_durability(volume).syncs_commit_metadata();
         let data_dirs = journal.data_dirs.clone();
         let has_target = journal.target_digest.is_some();
-        let journal_for_receipt = journal.clone();
+        let journal_for_receipt = cached_journal_raw.is_none().then(|| journal.clone());
         let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
         let prune_paths: Vec<_> = rollback
             .ancestors()
@@ -2897,6 +3053,31 @@ impl LocalDisk {
             .collect();
 
         let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<()> {
+            let persist_receipt = |finished: bool| -> std::io::Result<()> {
+                let receipt_bytes = match &cached_journal_raw {
+                    Some(raw) => format_commit_receipt_bytes(raw, finished),
+                    None => {
+                        let receipt = CommitReceipt {
+                            journal: journal_for_receipt
+                                .clone()
+                                .ok_or_else(|| std::io::Error::other("missing journal for receipt"))?,
+                            finished,
+                        };
+                        serde_json::to_vec(&receipt).map_err(std::io::Error::other)?
+                    }
+                };
+                std::fs::create_dir_all(&staging_root)?;
+                let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?;
+                std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?;
+                if sync && !receipt_bytes.is_empty() {
+                    staged_receipt.as_file().sync_data()?;
+                }
+                std::fs::rename(staged_receipt.path(), &receipt_path)?;
+                if sync {
+                    let _ = os::fsync_dir_std(&rollback);
+                }
+                Ok(())
+            };
             let committed = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
             let recovery_committed = format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
             let reserved = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
@@ -2951,38 +3132,38 @@ impl LocalDisk {
                 os::fsync_dir_std(&root)?;
             }
             for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] {
-                ignore_not_found(std::fs::remove_file(rollback.join(name)))?;
-            }
-            if has_target || had_physical_data_dirs {
-                let finished_receipt = CommitReceipt {
-                    journal: journal_for_receipt,
-                    finished: true,
-                };
-                let receipt_bytes = serde_json::to_vec(&finished_receipt).map_err(std::io::Error::other)?;
-                std::fs::create_dir_all(&staging_root)?;
-                let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?;
-                std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?;
-                if sync && !receipt_bytes.is_empty() {
-                    staged_receipt.as_file().sync_data()?;
-                }
-                std::fs::rename(staged_receipt.path(), &receipt_path)?;
-                let _ = std::fs::File::create(rollback.join(FINISHED))?;
-                if sync {
-                    os::fsync_dir_std(&rollback)?;
+                if let Err(err) = ignore_not_found(std::fs::remove_file(rollback.join(name))) {
+                    if has_target || had_physical_data_dirs {
+                        let _ = persist_receipt(true);
+                    }
+                    return Err(err);
                 }
             }
-            for name in [TARGET, PREPARED, FINISHED] {
+            for name in [PREPARED, TARGET, FINISHED] {
                 let path = rollback.join(name);
                 #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                 os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &path);
-                ignore_not_found(std::fs::remove_file(path))?;
+                if let Err(err) = ignore_not_found(std::fs::remove_file(path)) {
+                    if has_target || had_physical_data_dirs {
+                        let _ = persist_receipt(true);
+                    }
+                    return Err(err);
+                }
             }
-            if sync {
-                os::fsync_dir_std(&rollback)?;
+            if sync && let Err(err) = os::fsync_dir_std(&rollback) {
+                if has_target || had_physical_data_dirs {
+                    let _ = persist_receipt(true);
+                }
+                return Err(err);
             }
             #[cfg(all(any(test, feature = "test-util"), not(windows)))]
             os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &receipt_path);
-            ignore_not_found(std::fs::remove_file(&receipt_path))?;
+            if let Err(err) = ignore_not_found(std::fs::remove_file(&receipt_path)) {
+                if has_target || had_physical_data_dirs {
+                    let _ = persist_receipt(true);
+                }
+                return Err(err);
+            }
             for dir_path in &prune_paths {
                 #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                 os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, dir_path);
@@ -3105,6 +3286,42 @@ pub(super) async fn restore_barrier(volume: &str, object_dir: &Path, transaction
     Ok(())
 }
 
+#[cfg(all(test, unix))]
+mod fused_prepare_checkpoint {
+    use std::collections::HashMap;
+    use std::path::{Path, PathBuf};
+    use std::sync::{LazyLock, Mutex};
+
+    type Hook = Box<dyn FnOnce() + Send>;
+    static HOOKS: LazyLock<Mutex<HashMap<PathBuf, Hook>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+    pub(super) struct Guard(PathBuf);
+
+    impl Drop for Guard {
+        fn drop(&mut self) {
+            HOOKS.lock().expect("checkpoint hooks").remove(&self.0);
+        }
+    }
+
+    pub(super) fn install(root: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
+        assert!(
+            HOOKS
+                .lock()
+                .expect("checkpoint hooks")
+                .insert(root.to_path_buf(), Box::new(hook))
+                .is_none()
+        );
+        Guard(root.to_path_buf())
+    }
+
+    pub(super) fn run(root: &Path) {
+        let hook = HOOKS.lock().expect("checkpoint hooks").remove(root);
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -7020,6 +7237,73 @@ mod tests {
         marker.deleted = false;
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() {
+        let (_temp, disk, mut first, _) = fixture().await;
+        first.version_id = Some(Uuid::new_v4());
+        let mut second = first.clone();
+        second.version_id = Some(Uuid::new_v4());
+        let mut meta = FileMeta::new();
+        meta.add_version(first).expect("add surviving version");
+        meta.add_version(second.clone()).expect("add deleted version");
+        let original = meta.marshal_msg().expect("encode original versions");
+        let (_, root) = disk.io_get_bucket_and_object_path("recovery", "object").expect("object path");
+        fs::write(root.join(STORAGE_FORMAT_FILE), &original)
+            .await
+            .expect("write versions");
+
+        let (checkpoint_temp, checkpoint_disk, _, _) = fixture().await;
+        let checkpoint_root = checkpoint_temp.path().join("recovery/object");
+        drop(checkpoint_disk);
+        let transaction = Uuid::new_v4();
+        let source_root = root.clone();
+        let copy_root = checkpoint_root.clone();
+        let _capture = fused_prepare_checkpoint::install(&root, move || {
+            let source = source_root.join(transaction.to_string());
+            let destination = copy_root.join(transaction.to_string());
+            std::fs::create_dir(&destination).expect("create checkpoint transaction");
+            std::fs::copy(source_root.join(STORAGE_FORMAT_FILE), copy_root.join(STORAGE_FORMAT_FILE))
+                .expect("capture head before publication");
+            for entry in std::fs::read_dir(source).expect("read published preparation") {
+                let entry = entry.expect("prepared file");
+                assert!(entry.file_type().expect("prepared file type").is_file());
+                std::fs::copy(entry.path(), destination.join(entry.file_name())).expect("capture prepared file");
+            }
+        });
+        let opts = DeleteOptions {
+            recoverable_delete: true,
+            ..Default::default()
+        };
+        assert_eq!(
+            disk.try_fused_recoverable_delete_version("recovery", "object", &second, &opts, transaction, None)
+                .await
+                .expect("finish original preparation"),
+            Some(())
+        );
+        assert!(checkpoint_root.join(transaction.to_string()).join(PREPARED).is_file());
+        let endpoint =
+            Endpoint::try_from(checkpoint_temp.path().to_str().expect("checkpoint path")).expect("checkpoint endpoint");
+        let reopened = LocalDisk::new(&endpoint, false)
+            .await
+            .expect("reopen checkpoint as a fresh disk");
+        let inspection = reopened
+            .delete_recovery_inspect("recovery", "object", transaction)
+            .await
+            .expect("actual preparation checkpoint must remain recoverable");
+        assert_eq!(inspection.state, DeleteRecoveryState::NewPrepared);
+        reopened
+            .settle_delete_transaction("recovery", "object", transaction, inspection.fingerprint, DeleteRecoveryAction::Rollback)
+            .await
+            .expect("roll back interrupted preparation");
+        assert_eq!(
+            fs::read(checkpoint_root.join(STORAGE_FORMAT_FILE))
+                .await
+                .expect("read rolled-back versions"),
+            original
+        );
+    }
+
     #[tokio::test]
     async fn delete_recovery_fused_versioned_delete_with_surviving_version_and_interrupted_target_cleanup() {
         let (_temp, disk, mut v1, _) = fixture().await;

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

@@ -372,29 +372,32 @@ impl IndexOwner {
         #[cfg(unix)]
         {
             use rustix::fs::{Mode, OFlags, mkdirat, openat};
-            let mut directories = vec![self._identity_pin.try_clone()?];
-            for component in [".rustfs.sys", "multipart-index-v1", &sha(bucket)] {
-                let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?;
-                match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) {
-                    Ok(()) | Err(rustix::io::Errno::EXIST) => {}
-                    Err(error) => return Err(error.into()),
+            let open_dir = |parent: rustix::fd::BorrowedFd<'_>, component: &str| -> io::Result<rustix::fd::OwnedFd> {
+                let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
+                match openat(parent, component, flags, Mode::empty()) {
+                    Ok(dir) => Ok(dir),
+                    Err(rustix::io::Errno::NOENT) => {
+                        match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) {
+                            Ok(()) | Err(rustix::io::Errno::EXIST) => {}
+                            Err(error) => return Err(error.into()),
+                        }
+                        openat(parent, component, flags, Mode::empty()).map_err(Into::into)
+                    }
+                    Err(error) => Err(error.into()),
                 }
-                let directory = openat(
-                    parent,
-                    component,
-                    OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
-                    Mode::empty(),
-                )?;
-                directories.push(std::fs::File::from(directory));
-            }
-            let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?;
+            };
+            use std::os::fd::AsFd;
+            let bucket_sha = sha(bucket);
+            let sys_dir = open_dir(self._identity_pin.as_fd(), ".rustfs.sys")?;
+            let index_dir = open_dir(sys_dir.as_fd(), "multipart-index-v1")?;
+            let parent = open_dir(index_dir.as_fd(), &bucket_sha)?;
             // The index is boot-local: every start rescans the staging area
             // before serving it, so a marker lost to a crash is recreated before
             // it can be consulted. Creation is enough; no fsync of the marker or
             // its ancestors. An existing marker is touched so pruning sees the
             // new upload's age, not the previous one's.
             match openat(
-                parent,
+                &parent,
                 source_root,
                 OFlags::CREATE | OFlags::EXCL | OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                 Mode::RUSR | Mode::WUSR,
@@ -409,7 +412,7 @@ impl IndexOwner {
                     // the previous upload's age, which only shortens the prune
                     // grace for this key; index completeness does not depend on it.
                     if let Ok(marker) =
-                        openat(parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty())
+                        openat(&parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty())
                     {
                         let _ = std::fs::File::from(marker).set_modified(std::time::SystemTime::now());
                     }
@@ -430,6 +433,18 @@ impl IndexOwner {
     }
 
     fn ensure_current_root(&self) -> io::Result<()> {
+        #[cfg(unix)]
+        let current = {
+            use std::os::unix::fs::MetadataExt;
+            let metadata = std::fs::symlink_metadata(&*self.root_path)?;
+            if !metadata.is_dir() {
+                return Err(io::Error::other("multipart index physical root is not a directory"));
+            }
+            let mut inode = [0; 16];
+            inode[..8].copy_from_slice(&metadata.ino().to_le_bytes());
+            (metadata.dev(), inode)
+        };
+        #[cfg(not(unix))]
         let current = os::PublicationRoot::new(&self.root_path)?.physical_identity()?;
         if current != self.physical {
             self.disabled.store(true, Ordering::Release);

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

@@ -1979,25 +1979,28 @@ impl NativeMultipartPublication {
         let canonical = self.canonical_head();
         let prepared = (|| -> std::io::Result<os::PreparedRenameSource> {
             if let Some((ref original_bytes, _, _)) = deferred_initial {
-                let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
-                let snapshot = current
-                    .snapshot
-                    .as_ref()
-                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-                let current_head = current
-                    .head
-                    .as_ref()
-                    .and_then(SourceHead::raw)
-                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-                if !current.head_matches_state
-                    || snapshot.phase != MultipartSourcePhase::Active
-                    || snapshot.capability_sha256.is_some()
-                    || snapshot.lineage != expected_lineage
-                    || snapshot.generation != expected_generation
-                    || snapshot.revision.checked_add(1) != Some(expected_revision)
-                    || current_head != original_bytes.as_slice()
-                {
-                    return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                if cfg!(test) || !can_mark_presynced {
+                    let current =
+                        load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
+                    let snapshot = current
+                        .snapshot
+                        .as_ref()
+                        .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                    let current_head = current
+                        .head
+                        .as_ref()
+                        .and_then(SourceHead::raw)
+                        .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                    if !current.head_matches_state
+                        || snapshot.phase != MultipartSourcePhase::Active
+                        || snapshot.capability_sha256.is_some()
+                        || snapshot.lineage != expected_lineage
+                        || snapshot.generation != expected_generation
+                        || snapshot.revision.checked_add(1) != Some(expected_revision)
+                        || current_head != original_bytes.as_slice()
+                    {
+                        return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                    }
                 }
             } else {
                 let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
@@ -2050,7 +2053,7 @@ impl NativeMultipartPublication {
                 let _ = std::fs::remove_file(prepared_path);
                 return Err(err);
             }
-            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root)
+            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root, Some(directory_guard))
         })()
         .map_err(DiskError::from)?;
         *self.state.lock() = state;
@@ -2160,7 +2163,7 @@ impl NativeMultipartPublication {
             let mut prepared_metadata_source =
                 os::create_prepared_rename_source_with_commit_guard(&prepared_path, &dst_file_path, &owned_directory_guard)?;
             prepared_metadata_source.write_all(&new_bytes, true)?;
-            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root)
+            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root, Some(&owned_directory_guard))
         })
         .await
         .map_err(DiskError::from)?;
@@ -2207,7 +2210,7 @@ impl NativeMultipartPublication {
         let base = self.store.base.clone();
         let root = self.store.publication_root.clone();
         os::run_blocking_namespace_operation(self.store.lease.clone(), move || {
-            os::install_multipart_prepared_head(prepared, &canonical, &base, &root)
+            os::install_multipart_prepared_head(prepared, &canonical, &base, &root, None)
         })
         .await
         .map_err(DiskError::from)
@@ -2264,6 +2267,7 @@ impl NativeMultipartPublication {
     pub(super) fn finish_on_thread(
         &self,
         target_already_synced: bool,
+        fused_in_place: bool,
         staging_cleanup: Option<(PathBuf, Arc<std::sync::atomic::AtomicBool>)>,
     ) -> Result<MultipartPublicationResult> {
         let mut state = self.state.lock().clone();
@@ -2293,62 +2297,64 @@ impl NativeMultipartPublication {
         let disk_ids = &self.store.disk_ids;
         let terminal = terminal_record_path(source, &state);
         (|| -> std::io::Result<()> {
-            let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
-            let current_state = current
-                .state
-                .as_ref()
-                .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-            let current_raw = encode(current_state).map_err(IoError::other)?;
-            if current.head.is_some() || current_raw != expected_state_raw {
-                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
-            }
-            if target_already_synced {
-                let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
-                    }
-                })?;
-                let target = os::read_multipart_source_metadata_in_pinned(
-                    &pinned,
-                    &target_dir,
-                    std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
-                    base,
-                    root,
-                )
-                .map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
-                    }
-                })?;
-                if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 {
+            if !fused_in_place {
+                let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
+                let current_state = current
+                    .state
+                    .as_ref()
+                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                let current_raw = encode(current_state).map_err(IoError::other)?;
+                if current.head.is_some() || current_raw != expected_state_raw {
                     return Err(IoError::other(DiskError::ConditionalDeleteConflict));
                 }
-            } else {
-                os::read_and_sync_multipart_target_metadata(
-                    &target_dir,
-                    std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
-                    base,
-                    root,
-                    |target| {
-                        if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
-                            return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                if target_already_synced {
+                    let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
                         }
-                        Ok(())
-                    },
-                )
-                .map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
+                    })?;
+                    let target = os::read_multipart_source_metadata_in_pinned(
+                        &pinned,
+                        &target_dir,
+                        std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
+                        base,
+                        root,
+                    )
+                    .map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
+                        }
+                    })?;
+                    if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 {
+                        return Err(IoError::other(DiskError::ConditionalDeleteConflict));
                     }
-                })?;
+                } else {
+                    os::read_and_sync_multipart_target_metadata(
+                        &target_dir,
+                        std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
+                        base,
+                        root,
+                        |target| {
+                            if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
+                                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                            }
+                            Ok(())
+                        },
+                    )
+                    .map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
+                        }
+                    })?;
+                }
             }
-            SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), true)?;
+            SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), !fused_in_place)?;
             if let Some((io_path, cleaned)) = staging_cleanup {
                 match std::fs::remove_dir(&io_path) {
                     Ok(()) => cleaned.store(true, std::sync::atomic::Ordering::Release),
@@ -2381,7 +2387,7 @@ impl NativeMultipartPublication {
     ) -> Result<MultipartPublicationResult> {
         let this = Arc::clone(self);
         os::run_blocking_namespace_operation_with_sync(target_lease, true, move || {
-            this.finish_on_thread(target_already_synced, staging_cleanup)
+            this.finish_on_thread(target_already_synced, false, staging_cleanup)
                 .map_err(IoError::other)
         })
         .await

crates/ecstore/src/disk/os.rs

@@ -2200,6 +2200,7 @@ pub(crate) const MAX_FSYNC_WORKERS_PER_DISK: usize = 8;
 // Return busy workers to the shared pool so other disks can acquire a slot.
 const MAX_BLOCKING_JOBS_PER_WORKER: usize = 64;
 const WORKER_IDLE_SPIN_ITERS: usize = 256;
+const FSYNC_WORKER_IDLE_SPIN_ITERS: usize = 1024;
 const FALLBACK_NAMESPACE_QUEUE_SHARDS: usize = 8;
 
 #[derive(Clone, Copy, PartialEq, Eq)]
@@ -2408,7 +2409,7 @@ impl DiskBlockingQueue {
                         .ready
                         .store(!state.jobs.is_empty(), std::sync::atomic::Ordering::Relaxed);
                     if job.is_none() {
-                        if state.idle_workers > 0 {
+                        if lane != DiskBlockingLane::Fsync && state.idle_workers > 0 {
                             state.active_workers = state.active_workers.saturating_sub(1);
                             guard.state = DiskBlockingWorkerState::Retired;
                             return;
@@ -2426,7 +2427,12 @@ impl DiskBlockingQueue {
                     }
                     idle_polled = false;
                 } else {
-                    for _ in 0..WORKER_IDLE_SPIN_ITERS {
+                    let spin_iters = if lane == DiskBlockingLane::Fsync {
+                        FSYNC_WORKER_IDLE_SPIN_ITERS
+                    } else {
+                        WORKER_IDLE_SPIN_ITERS
+                    };
+                    for _ in 0..spin_iters {
                         if queue.lane(lane).ready.load(std::sync::atomic::Ordering::Relaxed) {
                             break;
                         }
@@ -7443,6 +7449,7 @@ pub(crate) fn install_multipart_prepared_head(
     canonical_head: &Path,
     base: &Path,
     publication_root: &PublicationRoot,
+    directory_guard: Option<&RenameCommitGuard>,
 ) -> io::Result<PreparedRenameSource> {
     let original_path = prepared.path.clone();
     let parent = canonical_head
@@ -7451,9 +7458,37 @@ pub(crate) fn install_multipart_prepared_head(
     if original_path.parent() != Some(parent) || original_path == canonical_head {
         return Err(io::Error::new(io::ErrorKind::InvalidInput, "MPU proxy must be a private sibling"));
     }
+    #[cfg(target_os = "linux")]
+    let guard = if let Some(directory_guard) = directory_guard.filter(|g| g.directories.source_path == parent) {
+        let dir = &directory_guard.directories.source;
+        RenameCommitGuard {
+            directories: Arc::new(LinuxCommitDirectories {
+                root: directory_guard.directories.root.clone(),
+                source_path: parent.to_path_buf(),
+                destination_path: parent.to_path_buf(),
+                source: LinuxCommitDirectory {
+                    file: Arc::clone(&dir.file),
+                    device: dir.device,
+                    inode: dir.inode,
+                },
+                destination: std::sync::OnceLock::from(LinuxCommitDirectory {
+                    file: Arc::clone(&dir.file),
+                    device: dir.device,
+                    inode: dir.inode,
+                }),
+                destination_base: base.to_path_buf(),
+            }),
+        }
+    } else {
+        prepare_rename_commit_guard(parent, parent, base, publication_root)?
+    };
     #[cfg(not(target_os = "linux"))]
     let _pinned = pin_multipart_source_directory(parent, base, publication_root)?;
-    let guard = prepare_rename_commit_guard(parent, parent, base, publication_root)?;
+    #[cfg(not(target_os = "linux"))]
+    let guard = {
+        let _ = directory_guard;
+        prepare_rename_commit_guard(parent, parent, base, publication_root)?
+    };
     rename_prepared_source_std(&prepared, &original_path, canonical_head, base, publication_root, &guard)?;
     prepared.path = canonical_head.to_path_buf();
     fsync_commit_directory_limited_std(parent, &guard)?;

crates/ecstore/src/set_disk/read.rs

@@ -2979,10 +2979,11 @@ fn build_legacy_per_part_fallback_reader(
     part_length: usize,
     part_size: usize,
     read_repair: Option<(DecodeReadRepairContext, bool)>,
-) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
+) -> Box<LegacyPerPartDecodeReader> {
     let buffer = adaptive_duplex_buffer_size(part_size as i64);
-    let (read_half, mut write_half) = tokio::io::duplex(buffer);
+    let (read_half, write_half) = tokio::io::duplex(buffer);
     let decode = tokio::spawn(async move {
+        let mut write_half = GetObjectDownstreamWriter::new(write_half);
         let outcome = erasure
             .decode_with_stripe_handles_and_reopeners_with_diagnostics(
                 &mut write_half,
@@ -8908,6 +8909,42 @@ mod tests {
         assert!(err.to_string().contains("inconsistent read source shards"));
     }
 
+    #[tokio::test]
+    async fn legacy_per_part_output_close_retains_downstream_classification() {
+        let erasure = coding::Erasure::new(2, 2, 64);
+        let data = (0..64u8).collect::<Vec<_>>();
+        let setup =
+            setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[0], &[], &[], HashAlgorithm::HighwayHash256)
+                .await;
+        let mut reader = build_legacy_per_part_fallback_reader(
+            erasure,
+            setup.readers,
+            setup.deferred_stripe_handles,
+            Vec::new(),
+            None,
+            0,
+            data.len(),
+            data.len(),
+            None,
+        );
+        // Closing the actual output pipe models the body consumer disappearing
+        // while retaining the executor long enough to inspect its typed result.
+        assert!(!reader.finished);
+        let (replacement, _unused) = tokio::io::duplex(1);
+        drop(std::mem::replace(&mut reader.inner, replacement));
+        let handle = reader.decode.take().expect("fallback decode task");
+        let error = handle
+            .await
+            .expect("join the fallback task")
+            .expect("closed pipe must interrupt output");
+        assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe);
+        assert_eq!(
+            crate::diagnostics::get::classify_io_error(&error),
+            GetObjectFailureReason::DownstreamClosed,
+            "fallback output must carry the same marker as the normal GET path"
+        );
+    }
+
     // backlog#879: a later multipart part that loses a shard (read quorum still
     // holds) must be reconstructed in place through the legacy per-part decode
     // bridge instead of failing the stream. This exercises the exact reader the

crates/heal/src/error.rs

@@ -58,7 +58,7 @@ pub enum Error {
     TaskExecutionFailed { message: String },
 
     /// Only the qualified MRF Deep check may produce this terminal result.
-    #[error("Heal task execution failed: Missing exact storage proof for durable MRF repair {target}")]
+    #[error("Independent verification unavailable for legacy object {target}; durable MRF responsibility retained")]
     LegacyMrfUnverified { target: String },
 
     #[error("Replacement failure could not be persisted: {failure}; persistence error: {persistence}")]

crates/heal/src/heal/manager/scheduler.rs

@@ -292,6 +292,20 @@ impl HealManager {
                                 "Heal scheduler task completed"
                             );
                         }
+                        Err(e @ Error::LegacyMrfUnverified { .. }) => {
+                            warn!(
+                                target: "rustfs::heal::manager",
+                                event = EVENT_HEAL_SCHEDULER_STATE,
+                                component = LOG_COMPONENT_HEAL,
+                                subsystem = LOG_SUBSYSTEM_MANAGER,
+                                state = "verification_unavailable",
+                                task_id,
+                                heal_type = %task_type_label_for_spawn,
+                                set = %task_set_label_for_spawn,
+                                detail = %e,
+                                "Heal attempt completed without independent verification"
+                            );
+                        }
                         Err(e) => {
                             let will_retry = retry_request.is_some();
                             if will_retry {
@@ -483,6 +497,9 @@ impl HealManager {
                         // update statistics
                         let mut stats = statistics_clone.write().await;
                         match completed_status {
+                            HealTaskStatus::Completed if matches!(&result, Err(Error::LegacyMrfUnverified { .. })) => {
+                                stats.update_unverified_task_completion();
+                            }
                             HealTaskStatus::Completed => {
                                 stats.update_task_completion(true);
                                 stats.add_healed_objects(completed_progress.objects_healed, completed_progress.bytes_processed);

crates/heal/src/heal/manager/tests.rs

@@ -1106,6 +1106,22 @@ impl HealStorageAPI for MockStorage {
         if bucket.starts_with("legacy-ambiguity-scheduler-") {
             return Err(Error::Storage(EcstoreError::DeleteRecoveryLegacyAmbiguous));
         }
+        if bucket == "legacy-held-pending-scheduler" {
+            let drives = vec![rustfs_madmin::heal_commands::HealDriveInfo {
+                state: "ok".to_owned(),
+                ..Default::default()
+            }];
+            return Ok((
+                HealResultItem {
+                    legacy_identity_unverified: true,
+                    detail: rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL.to_owned(),
+                    before: rustfs_madmin::heal_commands::Infos { drives: drives.clone() },
+                    after: rustfs_madmin::heal_commands::Infos { drives },
+                    ..Default::default()
+                },
+                None,
+            ));
+        }
         if bucket == "completed-retention-failed" {
             return Err(Error::TaskExecutionFailed {
                 message: "retention fixture failure".to_string(),
@@ -2153,6 +2169,53 @@ fn only_typed_mrf_result_can_pause_unverified_legacy() {
     }
 }
 
+#[tokio::test(flavor = "current_thread")]
+async fn mrf_legacy_completed_attempt_retains_responsibility_without_failed_statistics() {
+    use crate::heal::failure_logging::test_support::FailureLogCapture;
+
+    let capture = FailureLogCapture::default();
+    let _subscriber = tracing::subscriber::set_default(capture.clone());
+    let (target, _) = unverified_legacy_fixture("legacy-held-pending-scheduler");
+    let bucket = target.bucket.clone();
+    let anchor = target.durable_anchor.clone().expect("durable MRF anchor");
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let mut request = HealRequest::object(bucket.to_string(), target.object.to_string(), None);
+    request.source = HealRequestSource::Mrf;
+    request.expected_mrf_bucket_incarnation_id = Some(anchor.bucket_incarnation_id);
+    request.options.pool_index = Some(1);
+    request.options.set_index = Some(2);
+    let receipt = manager
+        .submit_mrf_heal_request_with_receipt_and_identity(request, target)
+        .await
+        .expect("admit durable MRF check");
+    assert_eq!(receipt.result, HealAdmissionResult::Accepted);
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(2), async {
+        while manager.get_statistics().await.unverified_tasks == 0 {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .expect("publish completed unverified attempt");
+    let stats = manager.get_statistics().await;
+    assert_eq!(stats.unverified_tasks, 1);
+    assert_eq!(stats.failed_tasks, 0);
+    assert_eq!(stats.successful_tasks, 0);
+    assert_eq!(stats.total_objects_healed, 0);
+    let report = manager.get_task_report(&receipt.task_id).await.expect("retained task report");
+    assert_eq!(report.status, HealTaskStatus::Completed);
+    let outcome = report.outcome.expect("retained unknown outcome");
+    assert_eq!(outcome.counters.unknown, 1);
+    assert_eq!(outcome.counters.failed, 0);
+    assert!(rustfs_common::mrf_channel::take_mrf_verified_repair_events_for(&bucket).is_empty());
+    let held = rustfs_common::mrf_channel::take_mrf_unverified_legacy_events_for(&bucket);
+    assert_eq!(held.len(), 1);
+    assert_eq!(held[0].anchor, anchor);
+    assert!(manager.retrying_heals.lock().await.is_empty());
+    assert_eq!(capture.count(tracing::Level::ERROR, Some("rustfs::heal::manager")), 0);
+    assert_eq!(capture.count(tracing::Level::WARN, Some("rustfs::heal::manager")), 1);
+}
+
 #[test]
 fn completed_mrf_notice_publishes_only_verified_positive_events() {
     use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome, HealTaskOutcome};

crates/heal/src/heal/progress.rs

@@ -506,6 +506,9 @@ pub struct HealStatistics {
     pub successful_tasks: u64,
     /// Failed tasks
     pub failed_tasks: u64,
+    /// Completed attempts without independent verification; not the current durable MRF backlog.
+    #[serde(default)]
+    pub unverified_tasks: u64,
     /// Running tasks
     pub running_tasks: u64,
     /// Total healed objects
@@ -528,6 +531,7 @@ impl HealStatistics {
             total_tasks: 0,
             successful_tasks: 0,
             failed_tasks: 0,
+            unverified_tasks: 0,
             running_tasks: 0,
             total_objects_healed: 0,
             total_bytes_healed: 0,
@@ -549,6 +553,11 @@ impl HealStatistics {
         self.last_update_time = SystemTime::now();
     }
 
+    pub(crate) fn update_unverified_task_completion(&mut self) {
+        self.unverified_tasks = self.unverified_tasks.saturating_add(1);
+        self.last_update_time = SystemTime::now();
+    }
+
     pub fn add_healed_objects(&mut self, count: u64, bytes: u64) {
         self.total_objects_healed += count;
         self.total_bytes_healed += bytes;
@@ -915,6 +924,21 @@ mod tests {
         assert_eq!(stats.failed_tasks, 0);
     }
 
+    #[test]
+    fn unverified_statistics_preserve_old_snapshot_compatibility() {
+        let mut stats = HealStatistics::new();
+        stats.update_unverified_task_completion();
+        assert_eq!(stats.unverified_tasks, 1);
+        assert_eq!(stats.successful_tasks, 0);
+        assert_eq!(stats.failed_tasks, 0);
+        let mut wire = serde_json::to_value(&stats).expect("serialize task statistics");
+        let current: HealStatistics = serde_json::from_value(wire.clone()).expect("read new statistics");
+        assert_eq!(current.unverified_tasks, 1);
+        wire.as_object_mut().expect("statistics object").remove("unverified_tasks");
+        let old: HealStatistics = serde_json::from_value(wire).expect("read statistics without new field");
+        assert_eq!(old.unverified_tasks, 0);
+    }
+
     #[test]
     fn test_heal_statistics_update_task_completion_success() {
         let mut stats = HealStatistics::new();

crates/heal/src/heal/task.rs

@@ -291,7 +291,7 @@ pub enum HealTaskStatus {
     Running,
     /// Retrying after a recoverable failure
     Retrying { error: String, retry_attempt: u32 },
-    /// Completed
+    /// Execution completed; the outcome and durable MRF ledger retain any unverified responsibility.
     Completed,
     /// Failed
     Failed { error: String },
@@ -1232,6 +1232,22 @@ impl HealTask {
                     "Heal task completed"
                 });
             }
+            Err(Error::LegacyMrfUnverified { .. }) => {
+                // The qualified Deep attempt finished, but its Unknown outcome
+                // cannot discharge the separately owned durable responsibility.
+                self.progress.write().await.mark_completed();
+                *self.status.write().await = HealTaskStatus::Completed;
+                debug!(
+                    target: "rustfs::heal::task",
+                    event = EVENT_HEAL_TASK_STATE,
+                    component = LOG_COMPONENT_HEAL,
+                    subsystem = LOG_SUBSYSTEM_TASK,
+                    state = "verification_unavailable",
+                    task_id = %self.id,
+                    heal_type = self.heal_type.kind_label(),
+                    "Heal attempt completed without independent verification"
+                );
+            }
             Err(Error::TaskCancelled) => {
                 let mut status = self.status.write().await;
                 *status = HealTaskStatus::Cancelled;
@@ -1281,11 +1297,16 @@ impl HealTask {
 
         let terminal_state = match &result {
             Ok(_) => "completed",
+            Err(Error::LegacyMrfUnverified { .. }) => "verification_unavailable",
             Err(Error::TaskCancelled) => "cancelled",
             Err(Error::TaskTimeout) => "timed_out",
             Err(_) => "failed",
         };
-        self.emit_trace_task_state(terminal_state, start_instant.elapsed(), result.as_ref().err());
+        let terminal_error = result
+            .as_ref()
+            .err()
+            .filter(|error| !matches!(error, Error::LegacyMrfUnverified { .. }));
+        self.emit_trace_task_state(terminal_state, start_instant.elapsed(), terminal_error);
 
         result
     }

crates/heal/src/heal/task/tests.rs

@@ -3285,6 +3285,29 @@ async fn durable_mrf_unverified_object_names_cannot_change_retry_classification(
     }
 }
 
+#[tokio::test]
+async fn mrf_legacy_verification_pending_is_a_completed_unverified_attempt() {
+    let (task, _) = mrf_task_for_item("legacy-pending", healthy_legacy_heal_item());
+    let error = task.execute().await.expect_err("unverified result must retain MRF ownership");
+    assert!(matches!(error, Error::LegacyMrfUnverified { .. }));
+    assert!(!error.is_recoverable_heal());
+    assert_eq!(task.get_status().await, HealTaskStatus::Completed);
+    let outcome = task.get_outcome().await;
+    assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
+    assert_eq!(outcome.counters.unknown, 1);
+    assert_eq!(outcome.counters.failed, 0);
+    assert_eq!(outcome.counters.healed, 0);
+    assert_eq!(outcome.counters.unchanged, 0);
+    assert!(matches!(outcome.objects[0].disposition, HealObjectDisposition::Unknown));
+    let (status, detail) = outcome.legacy_status("finished", None);
+    assert_eq!(status, "finished");
+    assert!(
+        detail
+            .expect("visible unverified result")
+            .contains("authoritative storage proof is unavailable")
+    );
+}
+
 #[tokio::test]
 async fn mrf_legacy_object_names_cannot_change_retry_classification() {
     for object in [
@@ -3304,7 +3327,9 @@ async fn mrf_legacy_object_names_cannot_change_retry_classification() {
         assert!(matches!(error, Error::LegacyMrfUnverified { .. }));
         assert_eq!(
             error.to_string(),
-            format!("Heal task execution failed: Missing exact storage proof for durable MRF repair bucket-a/{object}")
+            format!(
+                "Independent verification unavailable for legacy object bucket-a/{object}; durable MRF responsibility retained"
+            )
         );
     }
 }
@@ -3362,11 +3387,14 @@ async fn mrf_legacy_pause_requires_storage_owned_healthy_deep_result() {
         let outcome = task.get_outcome().await;
         assert_eq!(outcome.counters.healed, 0);
         if case == "qualified" {
+            assert_eq!(task.get_status().await, HealTaskStatus::Completed);
             assert_eq!(outcome.counters.unknown, 1);
             assert_eq!(outcome.objects[0].detail.as_deref(), Some("updated diagnostic wording"));
             let opts = storage.object_heal_opts.lock().expect("storage options");
             assert_eq!(opts.len(), 1);
             assert_eq!(opts[0].scan_mode, HealScanMode::Deep);
+        } else {
+            assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }), "case={case}");
         }
     }
 }

crates/heal/tests/shard_identity_receipt_test.rs

@@ -30,6 +30,78 @@ use storage_api::integration::{
     WriteCommitGuard, WriteCompletion,
 };
 
+async fn complete_fixture_replicas(env: &TestECStoreEnv, bucket: &str, object: &str) {
+    // TailDrained completes fan-out, but a successful quorum write may still
+    // have minority failures. Fault injection needs a fully repaired baseline.
+    let result = ECStoreHealStorage::new(env.ecstore.clone())
+        .heal_object_with_receipt(
+            bucket,
+            object,
+            None,
+            &HealOpts {
+                scan_mode: HealScanMode::Deep,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("complete protected fixture replicas before fault injection");
+    assert!(result.error.is_none(), "fixture repair failed: {:?}", result.error);
+    assert_eq!(result.item.after.drives.len(), env.disk_paths.len());
+    assert!(
+        result.item.after.drives.iter().all(|drive| drive.state == "ok"),
+        "fixture requires every replica, not only read/write quorum: {:?}",
+        result.item.after.drives
+    );
+    assert!(result.item.integrity_verified, "fixture must retain protected payload verification");
+}
+
+#[tokio::test]
+#[serial]
+async fn mixed_batch_fixture_repairs_a_missing_minority_replica() {
+    let root = tempfile::tempdir().expect("minority replica fixture");
+    let env = TestECStoreEnv::builder().base_dir(root.path()).build().await;
+    let bucket = "mixed-fixture-baseline";
+    let object = "object";
+    env.make_bucket(bucket, false).await;
+    let expected = vec![0x5a; 1024 * 1024 + 123];
+    env.ecstore
+        .put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected.clone()),
+            &ObjectOptions {
+                shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected),
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("quorum commit fixture");
+    // Guarantee a minority failure regardless of whether the original write
+    // succeeded on every disk, then repair it before any test fault is applied.
+    let replica = env.disk_paths[0].join(bucket).join(object);
+    if replica.exists() {
+        tokio::fs::remove_dir_all(&replica)
+            .await
+            .expect("remove one complete replica");
+    }
+    complete_fixture_replicas(&env, bucket, object).await;
+    let disks = env.ecstore.pools[0].get_disks(0);
+    for disk in disks.disks.read().await.iter().flatten() {
+        disk.read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("every fixture replica has metadata after completion");
+    }
+    let mut reader = env
+        .ecstore
+        .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
+        .await
+        .expect("read the completed fixture");
+    let mut actual = Vec::new();
+    reader.stream.read_to_end(&mut actual).await.expect("completed fixture body");
+    assert_eq!(actual, expected);
+}
+
 #[test]
 fn disabled_lock_reports_only_quorum_selection_and_deep_check_observations() {
     const CHILD_ENV: &str = "RUSTFS_TEST_DISABLED_LOCK_HEAL_CHILD";
@@ -704,6 +776,7 @@ fn admin_mixed_batch_retains_repairs_after_object_retry_exhaustion() {
                             )
                             .await
                             .expect("commit complete fixture");
+                        complete_fixture_replicas(&env, bucket, object).await;
                         for (slot, disk) in disks.iter().enumerate() {
                             let meta = disk
                                 .read_version("", bucket, object, "", &ReadOptions::default())

crates/object-data-cache/src/runtime_memory.rs

@@ -66,11 +66,26 @@ pub fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Opt
 
 /// Resolves the effective memory from an already-refreshed system handle.
 ///
-/// `cgroup_limits()` is computed fresh on each call and is only implemented on
-/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host
-/// values.
+/// On Linux, the current process cgroup and its ancestors are read on each call,
+/// including limits imposed by systemd services. System-wide cgroup limits are
+/// used if process information is unavailable. Other platforms use host values.
 pub fn effective_memory_from_system(system: &System) -> EffectiveMemory {
-    let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory));
+    #[cfg(target_os = "linux")]
+    let limits = sysinfo::get_current_pid()
+        .ok()
+        .and_then(|pid| {
+            let mut processes = System::new().ok()?;
+            processes.refresh_processes_specifics(
+                sysinfo::ProcessesToUpdate::Some(&[pid]),
+                false,
+                sysinfo::ProcessRefreshKind::nothing().with_memory(),
+            );
+            processes.process(pid)?.cgroup_limits()
+        })
+        .or_else(|| system.cgroup_limits());
+    #[cfg(not(target_os = "linux"))]
+    let limits = system.cgroup_limits();
+    let cgroup = limits.map(|limits| (limits.total_memory, limits.free_memory));
     select_effective_memory(system.total_memory(), system.available_memory(), cgroup)
 }
 
@@ -89,6 +104,32 @@ mod tests {
 
     const GIB: u64 = 1024 * 1024 * 1024;
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn effective_memory_uses_current_process_cgroup() {
+        use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, get_current_pid};
+
+        let pid = get_current_pid().expect("current process ID");
+        let mut system = System::new().expect("system memory handle");
+        system.refresh_memory();
+        let mut process_system = System::new().expect("process memory handle");
+        process_system.refresh_processes_specifics(
+            ProcessesToUpdate::Some(&[pid]),
+            false,
+            ProcessRefreshKind::nothing().with_memory(),
+        );
+        let limits = process_system.process(pid).expect("current process entry").cgroup_limits();
+        let expected = select_effective_memory(
+            system.total_memory(),
+            system.available_memory(),
+            limits.map(|limits| (limits.total_memory, limits.free_memory)),
+        );
+        let actual = super::effective_memory_from_system(&system);
+        assert_eq!(actual.basis, expected.basis);
+        assert_eq!(actual.total_bytes, expected.total_bytes);
+        assert!(actual.available_bytes <= actual.total_bytes);
+    }
+
     #[test]
     fn select_effective_memory_prefers_constraining_cgroup() {
         let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB)));

crates/s3select-api/Cargo.toml

@@ -75,7 +75,6 @@ futures-core = { workspace = true }
 flate2.workspace = true
 brotli.workspace = true
 lz4.workspace = true
-lz4_flex.workspace = true
 http.workspace = true
 memchr.workspace = true
 s3s = { workspace = true, features = ["minio"] }

crates/s3select-api/src/parquet_budget.rs

@@ -760,34 +760,21 @@ fn validate_compressed_page(payload: &[u8], sizes: PageHeaderSizes, codec: Compr
             bounded_stream_size(brotli::Decompressor::new(payload, 4096), expanded)?;
             Ok(window)
         }
-        Compression::LZ4 if payload.starts_with(&[4, 34, 77, 24]) || payload.starts_with(&[2, 33, 76, 24]) => {
-            let (block, linked) = if payload.starts_with(&[2, 33, 76, 24]) {
-                (8 * 1024 * 1024, false)
-            } else {
-                let descriptor = *payload.get(5).ok_or_else(ThriftCompactCursor::error)?;
-                let block = match (descriptor >> 4) & 7 {
-                    4 => 64 * 1024,
-                    5 => 256 * 1024,
-                    6 => 1024 * 1024,
-                    7 => 4 * 1024 * 1024,
-                    _ => return Err(ThriftCompactCursor::error()),
-                };
-                (block, payload[4] & 0x20 == 0)
+        Compression::LZ4 if payload.starts_with(&[4, 34, 77, 24]) => {
+            let descriptor = *payload.get(5).ok_or_else(ThriftCompactCursor::error)?;
+            let block = match (descriptor >> 4) & 7 {
+                4 => 64 * 1024,
+                5 => 256 * 1024,
+                6 => 1024 * 1024,
+                7 => 4 * 1024 * 1024,
+                _ => return Err(ThriftCompactCursor::error()),
             };
-            // Match parquet's pinned fallback decoder: compressed input plus
-            // output, with an extra output block and 64 KiB history when linked.
-            let working = if linked { block * 3 + 64 * 1024 } else { block * 2 };
+            let working = block * 2;
             if working > limit {
                 return Err(budget_error());
             }
-            match bounded_stream_size(lz4_flex::frame::FrameDecoder::new(payload), expanded) {
-                // Hadoop data can share the magic prefix, and parquet also
-                // tries bounded RAW decoding after a frame error. The bounded
-                // sink proved the frame produced no more than `expanded` bytes
-                // before that error; let the actual reader select its fallback.
-                Err(ParquetBudgetError::Codec(_)) => Ok(working),
-                result => result.map(|()| working),
-            }
+            bounded_stream_size(lz4::Decoder::new(payload)?, expanded)?;
+            Ok(working)
         }
         _ => Ok(0),
     }
@@ -864,49 +851,6 @@ mod tests {
         assert!(validate_compressed_page(&payload, compressed_page(&payload, 64), Compression::LZ4, 1024 * 1024).is_err());
     }
 
-    #[test]
-    fn legacy_lz4_scratch_is_admitted_before_the_fallback_decoder() {
-        let payload = [2, 33, 76, 24, 5, 0, 0, 0, 0x40, 42, 0, 0, 0];
-        let header = compressed_page(&payload, 4);
-        assert!(matches!(
-            validate_compressed_page(&payload, header, Compression::LZ4, 1024 * 1024),
-            Err(ParquetBudgetError::Exceeded)
-        ));
-        let admitted = validate_compressed_page(&payload, header, Compression::LZ4, 32 * 1024 * 1024)
-            .expect("legacy LZ4 remains supported with enough scratch memory");
-        assert!(admitted >= 16 * 1024 * 1024);
-        assert!(validate_compressed_page(&payload, compressed_page(&payload, 1), Compression::LZ4, 32 * 1024 * 1024).is_err());
-    }
-
-    #[test]
-    fn hadoop_lz4_can_share_the_legacy_magic_without_becoming_a_frame() {
-        let size = usize::try_from(u32::from_be_bytes([2, 33, 76, 24])).expect("Hadoop size");
-        let bytes = vec![b'x'; size];
-        let block = lz4_flex::block::compress(&bytes);
-        let mut payload = vec![2, 33, 76, 24];
-        payload.extend(u32::try_from(block.len()).expect("compressed block").to_be_bytes());
-        payload.extend(block);
-        validate_compressed_page(&payload, compressed_page(&payload, size), Compression::LZ4, 64 * 1024 * 1024)
-            .expect("Hadoop framing takes priority over the frame fallback");
-    }
-
-    #[test]
-    fn linked_lz4_accounts_for_the_history_and_both_block_buffers() {
-        use lz4_flex::frame::{BlockMode, BlockSize, FrameEncoder, FrameInfo};
-        let frame = FrameInfo::new().block_mode(BlockMode::Linked).block_size(BlockSize::Max64KB);
-        let mut encoder = FrameEncoder::with_frame_info(frame, Vec::new());
-        encoder.write_all(&vec![b'x'; 256 * 1024]).expect("linked input");
-        let payload = encoder.finish().expect("linked frame");
-        let header = compressed_page(&payload, 256 * 1024);
-        assert!(matches!(
-            validate_compressed_page(&payload, header, Compression::LZ4, 192 * 1024),
-            Err(ParquetBudgetError::Exceeded)
-        ));
-        let admitted = validate_compressed_page(&payload, header, Compression::LZ4, 1024 * 1024)
-            .expect("linked frame with enough scratch memory");
-        assert!(admitted >= 256 * 1024);
-    }
-
     #[test]
     fn uncompressed_v2_does_not_invoke_the_column_codec() {
         let payload = b"levels-and-uncompressed-values";

crates/s3select-query/src/dispatcher/parquet_table.rs

@@ -1106,98 +1106,6 @@ mod tests {
         assert_eq!(error.select_error(), SelectError::ResourceExhausted);
     }
 
-    #[tokio::test]
-    async fn parquet_legacy_lz4_checks_native_scratch_before_query_decode() {
-        use datafusion::parquet::{basic::Compression, file::metadata::ParquetMetaDataWriter};
-        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int32, false)]));
-        let batch =
-            RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(datafusion::arrow::array::Int32Array::from(vec![42]))])
-                .expect("one integer row");
-        let properties = datafusion::parquet::file::properties::WriterProperties::builder()
-            .set_dictionary_enabled(false)
-            .build();
-        let mut original = Vec::new();
-        let mut writer = ArrowWriter::try_new(&mut original, schema, Some(properties)).expect("writer");
-        writer.write(&batch).expect("row");
-        writer.close().expect("footer");
-        let metadata = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(original))
-            .expect("metadata")
-            .metadata()
-            .clone();
-        // A valid legacy frame encoding the four little-endian bytes for 42.
-        let payload = [2, 33, 76, 24, 5, 0, 0, 0, 0x40, 42, 0, 0, 0];
-        let mut page = vec![0x15, 0, 0x15, 8, 0x15, 26, 0x2c, 0x15, 2, 0x15, 0, 0x15, 6, 0x15, 6, 0, 0];
-        page.extend(payload);
-        let column = metadata
-            .row_group(0)
-            .column(0)
-            .clone()
-            .into_builder()
-            .set_compression(Compression::LZ4)
-            .set_data_page_offset(4)
-            .set_total_compressed_size(i64::try_from(page.len()).expect("small page"))
-            .build()
-            .expect("legacy column");
-        let group = metadata
-            .row_group(0)
-            .clone()
-            .into_builder()
-            .set_column_metadata(vec![column])
-            .build()
-            .expect("row group");
-        let metadata = metadata.as_ref().clone().into_builder().set_row_groups(vec![group]).build();
-        let mut encoded = b"PAR1".to_vec();
-        encoded.extend(page);
-        ParquetMetaDataWriter::new(&mut encoded, &metadata)
-            .finish()
-            .expect("legacy footer");
-        assert!(encoded.len() < 4096);
-        let mut native = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(encoded.clone()))
-            .expect("legacy metadata")
-            .build()
-            .expect("native reader");
-        let decoded = native.next().expect("one batch").expect("legacy fallback decode");
-        assert_eq!(
-            decoded
-                .column(0)
-                .as_primitive::<datafusion::arrow::datatypes::Int32Type>()
-                .value(0),
-            42
-        );
-        let store = Arc::new(InMemory::new());
-        store
-            .put(&Path::from("legacy.parquet"), Bytes::from(encoded).into())
-            .await
-            .expect("fixture");
-        let tight = parquet_session_with_memory(store.clone(), 2 * 1024 * 1024);
-        let result = ParquetSelectTable::try_new(&tight.state(), &parquet_input("legacy.parquet"), None).await;
-        let error = match result {
-            Ok(_) => panic!("legacy decoder scratch exceeds the query budget"),
-            Err(error) => error,
-        };
-        assert_eq!(error.select_error(), SelectError::ResourceExhausted);
-        let roomy = parquet_session_with_memory(store, 128 * 1024 * 1024);
-        let table = ParquetSelectTable::try_new(&roomy.state(), &parquet_input("legacy.parquet"), None)
-            .await
-            .expect("admitted legacy file");
-        roomy.register_table("source", table).expect("table");
-        let rows = roomy
-            .sql("SELECT v FROM source")
-            .await
-            .expect("plan")
-            .collect()
-            .await
-            .expect("query");
-        assert_eq!(rows.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
-        assert_eq!(
-            rows[0]
-                .column(0)
-                .as_primitive::<datafusion::arrow::datatypes::Int32Type>()
-                .value(0),
-            42
-        );
-    }
-
     #[tokio::test]
     async fn parquet_nested_views_decode_through_the_actual_file_schema() {
         use datafusion::arrow::array::{ListArray, StringArray};

crates/scanner/src/scanner_folder.rs

@@ -2348,6 +2348,10 @@ impl FolderScanner {
                     continue;
                 }
 
+                // Budget and admission waits already pace this walker. Charge
+                // only work performed after admission, never those waits again.
+                let mut work_elapsed = work_started.elapsed();
+
                 // Reuse a resolved symlink's size; otherwise admit the size
                 // stat only when the byte budget needs it.
                 let budgeted_bytes = if self.sleeper.budgets_disk_io_bytes() {
@@ -2357,7 +2361,12 @@ impl FolderScanner {
                         wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, 0)).await?;
                         wait_for_scanner_io(
                             &ctx,
-                            run_scanner_node_operation(self.node_admission.as_ref(), &ctx, native_io::metadata(&item.path)),
+                            run_scanner_node_operation(self.node_admission.as_ref(), &ctx, async {
+                                let started = tokio::time::Instant::now();
+                                let result = native_io::metadata(&item.path).await;
+                                work_elapsed = work_elapsed.saturating_add(started.elapsed());
+                                result
+                            }),
                         )
                         .await??
                         .map(|metadata| metadata.len())
@@ -2369,6 +2378,7 @@ impl FolderScanner {
                 wait_for_scanner_io(&ctx, self.sleeper.admit_disk_io(1, budgeted_bytes)).await?;
 
                 let size_result = run_scanner_metadata_operation(self.node_admission.as_ref(), &ctx, async {
+                    let started = tokio::time::Instant::now();
                     self.object_metadata_visits.increment(1);
                     #[cfg(feature = "diagnostic-prefix-attribution")]
                     if let Some(attribution) = &self.prefix_attribution {
@@ -2384,9 +2394,11 @@ impl FolderScanner {
                     {
                         attribution.record_success(&item.bucket, &item.prefix);
                     }
+                    work_elapsed = work_elapsed.saturating_add(started.elapsed());
                     result
                 })
                 .await?;
+                let work_started = tokio::time::Instant::now();
                 if self.node_admission.is_some() && ctx.is_cancelled() {
                     return Err(ScannerError::Other("Operation cancelled".to_string()));
                 }
@@ -2483,8 +2495,12 @@ impl FolderScanner {
                             }
                         }
 
-                        wait_for_scanner_io(&ctx, self.sleeper.pace_object(&mut self.pacing_debt, work_started.elapsed()))
-                            .await?;
+                        wait_for_scanner_io(
+                            &ctx,
+                            self.sleeper
+                                .pace_object(&mut self.pacing_debt, work_elapsed.saturating_add(work_started.elapsed())),
+                        )
+                        .await?;
                         continue;
                     }
                 };
@@ -2507,7 +2523,12 @@ impl FolderScanner {
                 self.checkpoint_objects = self.checkpoint_objects.saturating_add(1);
                 self.maybe_send_checkpoint(&folder, &this_hash, into);
 
-                wait_for_scanner_io(&ctx, self.sleeper.pace_object(&mut self.pacing_debt, work_started.elapsed())).await?;
+                wait_for_scanner_io(
+                    &ctx,
+                    self.sleeper
+                        .pace_object(&mut self.pacing_debt, work_elapsed.saturating_add(work_started.elapsed())),
+                )
+                .await?;
 
                 if ctx.is_cancelled() {
                     return Err(ScannerError::Other("Operation cancelled".to_string()));

crates/scanner/src/scanner_folder/tests/pacing.rs

@@ -29,6 +29,69 @@ impl Drop for IdleGuard {
     }
 }
 
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn object_pacing_does_not_multiply_disk_budget_waits() {
+    let _idle = IdleGuard::enabled();
+    for corrupt in [false, true] {
+        let (mut scanner, temp_dir) = build_test_scanner().await;
+        let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
+        scanner.old_cache.info.name = "bucket".to_owned();
+        scanner.new_cache.info.name = "bucket".to_owned();
+        scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Default).for_disk_with_budget_for_tests(
+            &temp_dir,
+            crate::sleeper::ScannerDiskIoBudgetLimits {
+                iops: Some(0),
+                bytes_per_sec: Some(0),
+            },
+        );
+        let mut metadata = FileMeta::new();
+        let mut info = FileInfo::new("object", 4, 2);
+        info.volume = "bucket".to_owned();
+        info.size = 17;
+        info.mod_time = Some(OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixture timestamp"));
+        metadata.add_version(info).expect("add object metadata");
+        let bytes = if corrupt {
+            Vec::new()
+        } else {
+            metadata.marshal_msg().expect("encode object metadata")
+        };
+        write_test_object_metadata_bytes(&temp_dir, "bucket", "object", &bytes).await;
+        let mut entry = DataUsageEntry::default();
+        scanner
+            .scan_folder(
+                CancellationToken::new(),
+                CachedFolder {
+                    name: "bucket".to_owned(),
+                    parent: None,
+                    object_heal_prob_div: 1,
+                },
+                &mut entry,
+            )
+            .await
+            .expect("finish budget-limited folder scan");
+        let total = scanner
+            .new_cache
+            .size_recursive(&hash_path("bucket").0)
+            .expect("root accounting");
+        assert_eq!(total.objects, usize::from(!corrupt));
+        assert_eq!(total.size, if corrupt { 0 } else { 17 });
+        assert_eq!(total.failed_objects, usize::from(corrupt));
+        let budget = crate::sleeper::scanner_disk_io_budget_status()
+            .into_iter()
+            .find(|status| status.disk == temp_dir.to_string_lossy())
+            .expect("disk budget status");
+        assert!(budget.wait_seconds >= 2.0, "the scan must actually exhaust its disk budget");
+        assert_eq!(scanner.pacing_debt.charged_objects, 1);
+        assert!(
+            scanner.pacing_debt.charged_delay < Duration::from_millis(100),
+            "budget waits must not become proportional work debt: {:?}",
+            scanner.pacing_debt.charged_delay
+        );
+        assert_eq!(scanner.pacing_debt.pending(), Duration::ZERO);
+    }
+}
+
 #[tokio::test]
 #[serial]
 async fn pacing_preserves_real_folder_versions_bytes_checkpoints_and_maintenance() {

crates/scanner/src/sleeper.rs

@@ -136,6 +136,8 @@ pub static SCANNER_SLEEPER: LazyLock<DynamicSleeper> = LazyLock::new(|| {
 /// `elapsed_work_time * factor`, clamped to `[MIN_SLEEP, max_sleep]`.
 /// Folder walkers accumulate proportional object waits without the per-object
 /// floor, paying bounded batches and the final bucket tail.
+/// Their work measurement excludes disk-budget and node-admission waits so
+/// foreground protection does not become an additional proportional delay.
 /// For folder-level gaps, [`sleep_folder`] uses `MIN_SLEEP * factor`,
 /// clamped only by `max_sleep`. All sleeps are gated on [`SCANNER_IDLE_MODE`];
 /// the per-disk background I/O budget bound with [`DynamicSleeper::for_disk`]
@@ -228,6 +230,17 @@ impl DynamicSleeper {
         SCANNER_IDLE_MODE.load(Ordering::Relaxed)
     }
 
+    fn foreground_backoff(&self) -> Duration {
+        // Local walkers already use disk-priority admission and a shared I/O
+        // budget. A global per-operation floor duplicates that control and
+        // throttles idle disks because unrelated foreground work is active.
+        if self.disk_io_budget.is_some() {
+            Duration::ZERO
+        } else {
+            foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity())
+        }
+    }
+
     fn read_params(&self) -> (f64, Duration) {
         let config = self.read_config();
         (config.factor, config.max_sleep)
@@ -250,15 +263,14 @@ impl DynamicSleeper {
         }
         let (factor, max_sleep) = self.read_params();
         if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.foreground_backoff();
             if !foreground_sleep.is_zero() {
                 tokio::time::sleep(foreground_sleep).await;
                 global_metrics().record_scanner_throttle_sleep(foreground_sleep);
             }
             return;
         }
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.foreground_backoff();
         let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor)
             .min(max_sleep)
             .max(foreground_sleep);
@@ -352,8 +364,7 @@ impl SleepTimer {
         }
         let (factor, max_sleep) = self.sleeper.read_params();
         if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.sleeper.foreground_backoff();
             if !foreground_sleep.is_zero() {
                 tokio::time::sleep(foreground_sleep).await;
                 global_metrics().record_scanner_throttle_sleep(foreground_sleep);
@@ -361,7 +372,7 @@ impl SleepTimer {
             return;
         }
         let elapsed = self.start.elapsed();
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.sleeper.foreground_backoff();
         let sleep_dur = Duration::from_secs_f64(elapsed.as_secs_f64() * factor)
             .max(MIN_SLEEP)
             .min(max_sleep)

crates/scanner/src/sleeper/pacing.rs

@@ -43,6 +43,8 @@ pub(crate) struct PacingDebt {
     pub(crate) charged_objects: u64,
     #[cfg(test)]
     pub(crate) batches: u64,
+    #[cfg(test)]
+    pub(crate) charged_delay: Duration,
 }
 
 impl PacingDebt {
@@ -73,6 +75,7 @@ impl PacingDebt {
         #[cfg(test)]
         {
             self.charged_objects += 1;
+            self.charged_delay = self.charged_delay.saturating_add(delay);
         }
     }
 
@@ -132,7 +135,7 @@ impl DynamicSleeper {
         if let Some(elapsed) = elapsed {
             debt.charge(proportional_delay(elapsed, config), Instant::now());
         }
-        let foreground = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground = self.foreground_backoff();
         let now = Instant::now();
         let folder_reaches_deadline = folder
             && debt.started.is_some_and(|start| {

crates/scanner/src/sleeper/pacing/tests.rs

@@ -181,6 +181,54 @@ async fn active_wait_wakes_for_fastest_idle_off_and_smaller_max_wait() {
     }
 }
 
+#[tokio::test(start_paused = true)]
+#[serial]
+async fn disk_bound_pacing_does_not_repeat_global_foreground_backoff() {
+    let _idle = IdleGuard::enabled();
+    struct ForegroundGuard;
+    impl Drop for ForegroundGuard {
+        fn drop(&mut self) {
+            crate::reset_foreground_read_activity_for_test();
+        }
+    }
+    let _foreground = ForegroundGuard;
+    crate::set_foreground_read_activity(25);
+    let disk = tempfile::tempdir().expect("disk identity");
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default).for_disk_with_budget_for_tests(
+        disk.path(),
+        ScannerDiskIoBudgetLimits {
+            iops: Some(0),
+            bytes_per_sec: None,
+        },
+    );
+    let mut debt = PacingDebt::default();
+    let begin = Instant::now();
+    for _ in 0..100 {
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    }
+    sleeper.flush_pacing(&mut debt).await;
+    assert_eq!(begin.elapsed(), Duration::from_millis(4));
+    assert_eq!(debt.charged_objects, 100);
+    assert_eq!(debt.pending, Duration::ZERO);
+    // The first operation consumes the token bucket's one-operation burst.
+    sleeper.admit_disk_io(1, 0).await;
+    let begin = Instant::now();
+    sleeper.sleep_folder_paced(&mut debt).await;
+    assert_eq!(
+        begin.elapsed(),
+        SCANNER_DISK_IO_BUDGET_MAX_WAIT + Duration::from_millis(2),
+        "the per-disk budget and preset folder gap must still apply"
+    );
+    assert_eq!(
+        scanner_disk_io_budget_status()
+            .into_iter()
+            .find(|status| status.disk == disk.path().to_string_lossy())
+            .expect("disk charges remain visible")
+            .charged_ops,
+        2
+    );
+}
+
 #[tokio::test(start_paused = true)]
 #[serial]
 async fn foreground_backoff_flushes_tail_without_double_charging_or_max_wait_cap() {

docs/operations/scanner-runtime-controls.md

@@ -40,7 +40,7 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
 | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` (`DEFAULT_SCANNER_CYCLE_MAX_OBJECTS`) | Caps objects processed by one cycle. `0` disables this budget. |
 | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` (`DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES`) | Caps directories entered by one cycle. `0` disables this budget. |
 | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` (`DEFAULT_HEAL_BITROT_CYCLE_SECS`, 30 days) | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
-| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and foreground-read backoff. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
+| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and the foreground backoff for callers without a disk binding. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
 | `scanner.cache_save_timeout` | `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` | seconds | `14` (`DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS`) | Timeout for saving scanner cache; runtime enforces a minimum of `1` and keeps the default persistence budget within the distributed publication lease. |
 | `scanner.recovery_backoff_cap` | `RUSTFS_SCANNER_RECOVERY_BACKOFF_CAP_SECS` | seconds | `600` (`DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS`) | Caps the wait between automatic startup recovery retries. Must be a positive integer; `0` and malformed values are rejected, so retries never stop. Applies to the next computed retry, not to a wait already in progress. The retry count resets after 24 hours, which bounds a single wait at about one day even with a larger cap. |
 | `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
@@ -63,7 +63,7 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult
 | `slow` | 10x | 15s | 60s |
 | `slowest` | 100x | 15s | 30m |
 
-Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate is independent of `idle_mode` and remains active when sleeps are disabled.
+Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is admitted work time multiplied by the factor, capped at `max_wait`; object pacing batches small waits and settles remaining debt at bucket completion. Local walkers bound with `DynamicSleeper::for_disk` use disk-priority admission and their shared I/O budget without an additional global foreground-request sleep. A caller without that binding retains a foreground-workload floor of 10ms per active request, capped at 250ms, which can exceed `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate and disk budget remain active when preset sleeps are disabled. Removing redundant global backoff does not guarantee a fixed snapshot age: scans still compete for admitted resources and may checkpoint at their cycle budget.
 
 ### Startup and cycle budgets
 

docs/operations/shard-integrity-rollout.md

@@ -62,6 +62,9 @@ because object identity is unproven; that is distinct from execution progress.
 MRF likewise retains durable legacy repair obligations after physical recovery
 when no independent verification receipt can discharge them. An idle attempt
 queue does not imply that these persistent obligations have been cleared.
+
+After a qualified all-healthy Deep check, the attempt is `Completed` with an `Unknown` object outcome; it is counted in `unverified_tasks`, not `successful_tasks` or `failed_tasks`. This cumulative attempt count is distinct from the current `held_unverified_legacy` responsibility count. The scheduler reports `verification_unavailable`, and the durable obligation remains subject to the existing explicit recheck or audited operator disposition. Execution completion never supplies a missing integrity receipt.
+
 Partial-write MRF replay uses Deep verification so a protected object's repair
 can discharge its obligation after verifying the payload. This adds full-object
 read work to those background attempts, including healthy replay targets.

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "nixpkgs": {
       "locked": {
-        "lastModified": 1793466162,
-        "narHash": "sha256-pt+peAEoOK3lLLdtbnQ1AGIBTZbdB6jrkgmurfcQwig=",
+        "lastModified": 1793977579,
+        "narHash": "sha256-WrSODdZFVIs/Hh2NzC0bY+ROByi0bATg8WfePX8dPj8=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "a0788123ada57ec92a79ab2581cda6d3c6cca75f",
+        "rev": "4fe25e12eb9f4c744966dd3de5a05246cf098601",
         "type": "github"
       },
       "original": {
@@ -29,11 +29,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1793434469,
-        "narHash": "sha256-+sgVNI2c67di+XIeMaz5KSei4T65kX68s5Nl0E5Gzbc=",
+        "lastModified": 1794040562,
+        "narHash": "sha256-7eEiezSku3lbZPUOXW8we1a7+xd5axkCQdyI3rzOKjo=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "6ecf534a7c16e7c969f26838fdd8c93597ae93f3",
+        "rev": "d3fe74269f2a6652437e21e78eb7e83be81f6982",
         "type": "github"
       },
       "original": {

rustfs/src/app/object/put.rs

@@ -1882,9 +1882,16 @@ impl DefaultObjectUsecase {
 
         reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
         if let Some(control) = progress_control {
-            use rustfs_rio::HashReaderMut;
-            let inner = reader.take_inner();
-            reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
+            if !should_compress
+                && (use_zero_copy_eager_put_path || use_empty_or_small_eager_put_path)
+                && trailing_headers.is_none()
+            {
+                control.finish();
+            } else {
+                use rustfs_rio::HashReaderMut;
+                let inner = reader.take_inner();
+                reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
+            }
         }
         rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start);
 

rustfs/src/app/object/request_body.rs

@@ -160,7 +160,7 @@ impl BodyReadControl {
         state.waited = Duration::ZERO;
     }
 
-    fn finish(&self) {
+    pub(super) fn finish(&self) {
         let mut state = self.0.budget.lock();
         self.0.active.store(false, Ordering::Release);
         state.finished = true;