refactor(ecstore): share the flushable file open for Windows sync sites

Windows flushes only a handle with write access. Three sync sites each carried their own cfg pair for that; route them through one helper that opens read-write on Windows and read-only elsewhere.

.github/workflows/cla.yml

@@ -95,7 +95,7 @@ jobs:
 
       - name: Run CLA Bot
         if: github.event_name != 'merge_group'
-        uses: overtrue/cla-bot@4bd1387c19691168c42fcf5edfa17b15819b2ab3 # v0.1.1
+        uses: overtrue/cla-bot@f054da99d5777e47e1af9933737b9c0c5584be8d # v0.1.0
         with:
           github-token: ${{ github.token }}
           registry-token: ${{ steps.registry-token.outputs.token }}

.github/workflows/windows-filesystem.yml

@@ -107,6 +107,32 @@ jobs:
         shell: pwsh
         run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture
 
+      # Native multipart completion publishes its payload from a different
+      # parent than its metadata, and abort renames under a precondition; both
+      # need their own Windows handle discipline.
+      - name: Test native multipart complete and abort publication
+        shell: pwsh
+        run: >-
+          cargo test -p rustfs-ecstore --lib --
+          native_abort_missing_state_with_actual_owned_records_never_becomes_legacy_or_missing
+          native_abort_pending_intent_is_not_a_delete_certificate_and_raw_cleanup_preserves_exact_evidence
+          --nocapture
+
+      # Owned-undo reclaim removes a directory it pinned, copy heal persists
+      # fault bytes through a handle, and metadata reads meet a directory.
+      - name: Test native multipart reclaim, copy heal, and directory reads
+        shell: pwsh
+        run: >-
+          cargo test -p rustfs-ecstore --lib --
+          native_multipart_owned_undo
+          native_multipart_copy_heal
+          orphan_multipart_cleanup_rejects_metadata_directory_on_one_disk
+          --nocapture
+
+      - name: Test directory removal fallback
+        shell: pwsh
+        run: "cargo test -p rustfs-ecstore --lib disk::fs::tests::test_remove -- --nocapture"
+
       - name: Test Windows path conversion
         shell: pwsh
         run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"

crates/e2e_test/src/protocols/README.md

@@ -77,7 +77,9 @@ core protocol surface plus cross-protocol consistency:
   existing path returns an error, bad-password authentication is rejected
 - Cross-protocol via aws-sdk-s3: SFTP write then S3 read with SHA256 match,
   S3 write then SFTP read with SHA256 match
-- Cross-API directory visibility: SFTP-created sub-directory visible via S3 ListObjectsV2, and an S3 trailing-slash directory object visible via SFTP readdir as a directory entry
+- Cross-API directory visibility: SFTP-created sub-directory visible via S3
+  ListObjectsV2, S3-created `__XLDIR__` marker visible via SFTP readdir as a
+  directory entry
 
 #### sftp_compliance (`test_sftp_compliance_suite`)
 

crates/e2e_test/src/protocols/sftp_core.rs

@@ -394,9 +394,9 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         info!("PASS: S3-written object matches via S3 GetObject and SFTP READ");
 
         // --- Cross-API directory visibility: SFTP mkdir, S3 ListObjectsV2 ---
-        // SFTP mkdir writes a zero-byte object at the trailing-slash key.
-        // Storage encodes that key to the __XLDIR__ marker, and the S3
-        // listing path decodes the marker back to a trailing-slash key.
+        // SFTP mkdir writes a __XLDIR__ marker. The rustfs S3 listing path
+        // decodes that marker back to a trailing-slash key, so the asserted
+        // pattern is "subdir_sftp/".
         info!("Testing SFTP: SFTP-created sub-directory visible via S3 ListObjectsV2");
         let sftp_subdir_name = "subdir_sftp";
         let sftp_subdir_path = format!("/{bucket}/{sftp_subdir_name}");
@@ -422,21 +422,20 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         );
         info!("PASS: SFTP mkdir visible to S3 ListObjectsV2");
 
-        // --- Cross-API directory visibility: S3 directory object, SFTP readdir ---
-        // aws-sdk-s3 PutObject writes a zero-byte object at the trailing-slash
-        // key. The __XLDIR__ suffix is reserved on the client-key boundary, so
-        // a direct put of that suffix is rejected. Storage encodes the slash
-        // key, and SFTP readdir must show a directory entry.
-        info!("Testing SFTP: S3-created directory object visible via SFTP readdir");
+        // --- Cross-API directory visibility: S3 marker, SFTP readdir ---
+        // aws-sdk-s3 PutObject writes a zero-byte marker keyed with the
+        // __XLDIR__ suffix. SFTP readdir must decode the marker back to a
+        // bare directory entry whose file_type reports as a directory.
+        info!("Testing SFTP: S3-created __XLDIR__ marker visible via SFTP readdir");
         let s3_subdir_name = "subdir_s3";
-        let s3_subdir_key = format!("{s3_subdir_name}/");
+        let s3_subdir_marker_key = format!("{s3_subdir_name}{XLDIR_SUFFIX}");
         s3.put_object()
             .bucket(bucket)
-            .key(&s3_subdir_key)
+            .key(&s3_subdir_marker_key)
             .body(ByteStream::from_static(b""))
             .send()
             .await
-            .map_err(|e| anyhow!("S3 PutObject {}/{} failed: {:?}", bucket, s3_subdir_key, e))?;
+            .map_err(|e| anyhow!("S3 PutObject {}/{} failed: {:?}", bucket, s3_subdir_marker_key, e))?;
         let bucket_entries: Vec<(String, bool)> = sftp
             .read_dir(&bucket_path)
             .await?
@@ -445,9 +444,9 @@ pub async fn test_sftp_core_operations() -> Result<()> {
         let visible_via_sftp = bucket_entries.iter().any(|(name, is_dir)| name == s3_subdir_name && *is_dir);
         assert!(
             visible_via_sftp,
-            "S3-created directory object must appear as a directory in SFTP readdir: entries were {bucket_entries:?}"
+            "S3-created marker must appear as a directory in SFTP readdir: entries were {bucket_entries:?}"
         );
-        info!("PASS: S3 directory object visible to SFTP readdir as a directory");
+        info!("PASS: S3 marker visible to SFTP readdir as a directory");
 
         // --- Pre-cleanup of cross-protocol fixtures ---
         // Removes the new files and sub-directories so the existing rmdir

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

@@ -996,32 +996,21 @@ fn internode_idempotent_read_retries() -> usize {
 static PREWARMED_PEERS: std::sync::LazyLock<std::sync::Mutex<std::collections::HashSet<String>>> =
     std::sync::LazyLock::new(|| std::sync::Mutex::new(std::collections::HashSet::new()));
 
-fn claim_control_channel_prewarm(addr: &str) -> bool {
-    let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else {
-        return false;
-    };
-    prewarmed.insert(addr.to_owned())
-}
-
-fn release_control_channel_prewarm(addr: &str) {
-    if let Ok(mut prewarmed) = PREWARMED_PEERS.lock() {
-        prewarmed.remove(addr);
-    }
-}
-
 /// Best-effort background prewarm of a peer's control channel (grpc-optimization P3-1). Deduped per
-/// peer; a dial failure releases the marker so a later disk construction can retry the prewarm.
+/// peer; a dial failure just falls through to the existing lazy connect and recovery monitor.
 fn spawn_control_channel_prewarm(addr: String) {
-    if !claim_control_channel_prewarm(&addr) {
-        return;
+    {
+        let Ok(mut prewarmed) = PREWARMED_PEERS.lock() else {
+            return;
+        };
+        if !prewarmed.insert(addr.clone()) {
+            return;
+        }
     }
     tokio::spawn(async move {
         match node_service_time_out_client_no_auth(&addr).await {
             Ok(_) => debug!(addr = %addr, "internode control channel prewarmed"),
-            Err(err) => {
-                release_control_channel_prewarm(&addr);
-                debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)");
-            }
+            Err(err) => debug!(addr = %addr, error = %err, "internode control channel prewarm failed (best-effort)"),
         }
     });
 }
@@ -3691,13 +3680,7 @@ impl DiskAPI for RemoteDisk {
                         );
                     }
                     let rpc_started = read_version_stage_timer(batch_read_version_attribution_enabled);
-                    let response_result = client.batch_read_version(request).await;
-                    if let Some(rpc_started) = rpc_started {
-                        crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_batch_read_version_duration(
-                            rpc_started.elapsed(),
-                        );
-                    }
-                    let response = match response_result {
+                    let response = match client.batch_read_version(request).await {
                         Ok(response) => {
                             record_batch_read_version_stage(INTERNODE_STAGE_BATCH_READ_VERSION_RPC_ROUNDTRIP, rpc_started);
                             response.into_inner()
@@ -4712,10 +4695,7 @@ impl DiskAPI for RemoteDisk {
                     read_multiple_req_bin: read_multiple_req_bin.into(),
                 });
 
-                let rpc_started = Instant::now();
-                let response = client.read_multiple(request).await;
-                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_multiple_duration(rpc_started.elapsed());
-                let response = response?.into_inner();
+                let response = client.read_multiple(request).await?.into_inner();
 
                 if !response.success {
                     return Err(response.error.unwrap_or_default().into());
@@ -4766,7 +4746,6 @@ impl DiskAPI for RemoteDisk {
                 attach_mutation_body_digest(&mut request, canonical_body, "write_all")?;
 
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_request();
-                let rpc_started = Instant::now();
                 let response = match client.write_all(request).await {
                     Ok(response) => response.into_inner(),
                     Err(err) => {
@@ -4775,7 +4754,6 @@ impl DiskAPI for RemoteDisk {
                     }
                 };
 
-                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_duration(rpc_started.elapsed());
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_write_all_sent_bytes(data_len);
 
                 if !response.success {
@@ -4873,7 +4851,6 @@ impl DiskAPI for RemoteDisk {
                 });
 
                 crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_request();
-                let rpc_started = Instant::now();
                 let response = match client.read_all(request).await {
                     Ok(response) => response.into_inner(),
                     Err(err) => {
@@ -4882,8 +4859,6 @@ impl DiskAPI for RemoteDisk {
                     }
                 };
 
-                crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_duration(rpc_started.elapsed());
-
                 if !response.success {
                     crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_all_error();
                     return Err(response.error.unwrap_or_default().into());
@@ -12000,15 +11975,4 @@ mod tests {
 
         accept_task.abort();
     }
-
-    #[test]
-    fn prewarm_claim_is_deduplicated_and_released_after_failure() {
-        let addr = "http://prewarm-retry-test.invalid:9000";
-        release_control_channel_prewarm(addr);
-        assert!(claim_control_channel_prewarm(addr));
-        assert!(!claim_control_channel_prewarm(addr));
-        release_control_channel_prewarm(addr);
-        assert!(claim_control_channel_prewarm(addr));
-        release_control_channel_prewarm(addr);
-    }
 }

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

@@ -481,14 +481,6 @@ pub(crate) fn record_remote_disk_grpc_write_all_request() {
         .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
 }
 
-pub(crate) fn record_remote_disk_grpc_write_all_duration(duration: Duration) {
-    global_internode_metrics().record_duration_for_operation_and_backend(
-        INTERNODE_OPERATION_GRPC_WRITE_ALL,
-        INTERNODE_TRANSPORT_BACKEND_GRPC,
-        duration,
-    );
-}
-
 pub(crate) fn record_remote_disk_grpc_write_all_sent_bytes(bytes: usize) {
     global_internode_metrics().record_sent_bytes_for_operation_and_backend(
         INTERNODE_OPERATION_GRPC_WRITE_ALL,
@@ -529,14 +521,6 @@ pub(crate) fn record_remote_disk_grpc_read_all_request() {
         .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC);
 }
 
-pub(crate) fn record_remote_disk_grpc_read_all_duration(duration: Duration) {
-    global_internode_metrics().record_duration_for_operation_and_backend(
-        INTERNODE_OPERATION_GRPC_READ_ALL,
-        INTERNODE_TRANSPORT_BACKEND_GRPC,
-        duration,
-    );
-}
-
 #[cfg(feature = "diagnostic-read-version-caller")]
 fn read_version_request_mode(opts: &ReadOptions) -> ReadVersionRequestMode {
     match (opts.read_data, opts.healing, opts.incl_free_versions) {
@@ -614,17 +598,6 @@ pub(crate) fn record_remote_disk_grpc_batch_read_version_stage(stage: &'static s
     );
 }
 
-pub(crate) fn record_remote_disk_grpc_batch_read_version_duration(duration: Duration) {
-    if !rustfs_io_metrics::get_stage_metrics_enabled() {
-        return;
-    }
-    global_internode_metrics().record_duration_for_operation_and_backend(
-        INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION,
-        INTERNODE_TRANSPORT_BACKEND_GRPC,
-        duration,
-    );
-}
-
 pub(crate) fn record_remote_disk_grpc_batch_read_version_error() {
     if !rustfs_io_metrics::get_stage_metrics_enabled() {
         return;
@@ -717,14 +690,6 @@ pub(crate) fn record_remote_disk_grpc_read_multiple_recv_bytes(bytes: usize) {
     record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, bytes);
 }
 
-pub(crate) fn record_remote_disk_grpc_read_multiple_duration(duration: Duration) {
-    global_internode_metrics().record_duration_for_operation_and_backend(
-        INTERNODE_OPERATION_GRPC_READ_MULTIPLE,
-        INTERNODE_TRANSPORT_BACKEND_GRPC,
-        duration,
-    );
-}
-
 /// Payload-size threshold (bytes) above which a unary internode gRPC response is counted as a
 /// "large payload" for alerting. Env-overridable via `RUSTFS_INTERNODE_RPC_LARGE_PAYLOAD_WARN_BYTES`.
 fn internode_rpc_large_payload_warn_bytes() -> usize {

crates/ecstore/src/disk/fs.rs

@@ -148,47 +148,58 @@ pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
     native_io(move || create_dir_all_std(&path)).await
 }
 
-fn is_dir_error(e: &io::Error) -> bool {
+fn is_dir_error(path: &Path, e: &io::Error) -> bool {
     e.raw_os_error() == Some(libc::EISDIR)
         || e.kind() == io::ErrorKind::IsADirectory
         // macOS: remove_file on a directory returns EPERM
         || (cfg!(target_os = "macos") && e.raw_os_error() == Some(libc::EPERM))
+        || is_windows_dir_access_denied(path, e)
+}
+
+/// Windows reports `remove_file` on a directory (or directory symlink) as
+/// ERROR_ACCESS_DENIED, which is indistinguishable from a real permission
+/// failure without inspecting the entry itself.
+#[cfg(windows)]
+fn is_windows_dir_access_denied(path: &Path, e: &io::Error) -> bool {
+    use std::os::windows::fs::FileTypeExt;
+
+    e.kind() == io::ErrorKind::PermissionDenied
+        && std::fs::symlink_metadata(path).is_ok_and(|metadata| {
+            let file_type = metadata.file_type();
+            file_type.is_dir() || file_type.is_symlink_dir()
+        })
+}
+
+#[cfg(not(windows))]
+fn is_windows_dir_access_denied(_path: &Path, _e: &io::Error) -> bool {
+    false
 }
 
 #[tracing::instrument(level = "debug", skip_all)]
 pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
-    // Try remove_file first; fall back to remove_dir if it's a directory
-    let file_path = path.as_ref().to_path_buf();
-    match native_io(move || std::fs::remove_file(file_path)).await {
-        Ok(()) => Ok(()),
-        Err(e) if is_dir_error(&e) => {
-            let path = path.as_ref().to_path_buf();
-            native_io(move || std::fs::remove_dir(path)).await
-        }
-        Err(e) => Err(e),
-    }
+    let path = path.as_ref().to_path_buf();
+    native_io(move || remove_std(path)).await
 }
 
 #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
     // Try remove_file first; fall back to remove_dir_all if it's a directory
-    let file_path = path.as_ref().to_path_buf();
-    match native_io(move || std::fs::remove_file(file_path)).await {
+    let path = path.as_ref().to_path_buf();
+    native_io(move || match std::fs::remove_file(&path) {
         Ok(()) => Ok(()),
-        Err(e) if is_dir_error(&e) => {
-            let path = path.as_ref().to_path_buf();
-            native_io(move || std::fs::remove_dir_all(path)).await
-        }
+        Err(e) if is_dir_error(&path, &e) => std::fs::remove_dir_all(&path),
         Err(e) => Err(e),
-    }
+    })
+    .await
 }
 
 #[tracing::instrument(level = "debug", skip_all)]
 pub fn remove_std(path: impl AsRef<Path>) -> io::Result<()> {
     // Try remove_file first; fall back to remove_dir if it's a directory
-    match std::fs::remove_file(path.as_ref()) {
+    let path = path.as_ref();
+    match std::fs::remove_file(path) {
         Ok(()) => Ok(()),
-        Err(e) if is_dir_error(&e) => std::fs::remove_dir(path.as_ref()),
+        Err(e) if is_dir_error(path, &e) => std::fs::remove_dir(path),
         Err(e) => Err(e),
     }
 }

crates/ecstore/src/disk/local.rs

@@ -9992,7 +9992,7 @@ impl LocalDisk {
                         if !std::fs::symlink_metadata(&receipt_path)?.is_file() {
                             return Err(std::io::Error::new(ErrorKind::InvalidData, "delete receipt is not a regular file"));
                         }
-                        std::fs::File::open(receipt_path)?
+                        os::open_file_for_sync_std(&receipt_path)?
                     }
                     Err(err) => return Err(err),
                 };
@@ -14513,7 +14513,7 @@ impl DiskAPI for LocalDisk {
                     return Ok(Err(DiskError::FileCorrupt));
                 }
                 let verify_and_sync = (|| -> std::io::Result<bool> {
-                    let mut file = std::fs::File::open(&source)?;
+                    let mut file = os::open_file_for_sync_std(&source)?;
                     let mut header = [0u8; rustfs_filemeta::shard_integrity::INDEX_HEADER_SIZE];
                     std::io::Read::read_exact(&mut file, &mut header)?;
                     if header != integrity.index_header() {
@@ -19754,6 +19754,9 @@ mod test {
         );
     }
 
+    // Only the Unix grouped dst dir fsync worker consults the injected
+    // failure; the non-Unix worker syncs by path without the hook.
+    #[cfg(unix)]
     #[tokio::test]
     #[serial_test::serial(dst_dir_fsync_group_commit)]
     async fn rename_data_non_inline_dst_dir_fsync_group_commit_failure_rolls_back_fresh_put() {
@@ -19842,6 +19845,9 @@ mod test {
         );
     }
 
+    // Only the Unix grouped dst dir fsync worker consults the injected
+    // failure; the non-Unix worker syncs by path without the hook.
+    #[cfg(unix)]
     #[tokio::test]
     #[serial_test::serial(dst_dir_fsync_group_commit)]
     async fn rename_data_inline_dst_dir_fsync_group_commit_failure_rolls_back_fresh_put() {

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

@@ -198,7 +198,7 @@ pub(super) fn create_local_inline_rollback_backup(
             return Err(err);
         }
     } else if sync && !old_metadata.is_empty() {
-        std::fs::File::open(&backup_path)?.sync_data()?;
+        os::open_file_for_sync_std(&backup_path)?.sync_data()?;
     }
     Ok(backup_path)
 }
@@ -1597,12 +1597,13 @@ impl LocalDisk {
                 );
             }
             if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() {
-                #[cfg(target_os = "linux")]
+                #[cfg(any(target_os = "linux", windows))]
                 let payload_guard = if multipart_publication.is_some() {
                     Some(
                         os::multipart_payload_commit_guard(
                             &commit.directory_guard,
                             src_data_path.parent().ok_or(DiskError::InvalidPath)?,
+                            commit.publication_root,
                             commit.mutation_lease.clone(),
                         )
                         .await
@@ -1611,9 +1612,9 @@ impl LocalDisk {
                 } else {
                     None
                 };
-                #[cfg(target_os = "linux")]
+                #[cfg(any(target_os = "linux", windows))]
                 let directory_guard = payload_guard.as_ref().unwrap_or(&commit.directory_guard);
-                #[cfg(not(target_os = "linux"))]
+                #[cfg(not(any(target_os = "linux", windows)))]
                 let directory_guard = &commit.directory_guard;
                 if let Err(err) = os::rename_all_with_commit_guard(
                     src_data_path,

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

@@ -1517,7 +1517,7 @@ impl LocalDisk {
                     }
                     Err(error) => return Err(error),
                 };
-                let _directory = os::pin_multipart_source_directory(&directory, base, root)?;
+                let directory_pin = os::pin_multipart_source_directory(&directory, base, root)?;
                 if !parts.is_empty() {
                     let anchor_path = directory.join(RETIREMENT_ANCHOR);
                     let bytes = os::read_multipart_source_metadata(&anchor_path, base, root)?;
@@ -1550,7 +1550,13 @@ impl LocalDisk {
                 }
                 // An anchor-less empty DD is only rmdir'ed. A new child makes
                 // this fail closed; no recursive removal can consume its bytes.
+                // A Windows pin refuses DELETE sharing, so the leaf pin is
+                // released first; the object pin still holds the parent.
+                #[cfg(windows)]
+                drop(directory_pin);
                 std::fs::remove_dir(&directory)?;
+                #[cfg(not(windows))]
+                drop(directory_pin);
                 if sync {
                     os::fsync_dir_std(&object_path)?;
                 }

crates/ecstore/src/disk/os.rs

@@ -610,6 +610,26 @@ pub(crate) mod prepare_rename_test_hooks {
             hook();
         }
     }
+
+    /// Hooks keyed by a directory component name, run before a Windows
+    /// directory pin retries a transient sharing conflict on it.
+    #[cfg(windows)]
+    static BEFORE_COMPONENT_RETRY: LazyLock<Mutex<HashMap<std::ffi::OsString, Hook>>> =
+        LazyLock::new(|| Mutex::new(HashMap::new()));
+
+    #[cfg(windows)]
+    pub(crate) fn install_before_component_retry(component: &std::ffi::OsStr, hook: impl FnOnce() + Send + 'static) {
+        BEFORE_COMPONENT_RETRY.lock().insert(component.to_os_string(), Box::new(hook));
+    }
+
+    #[cfg(windows)]
+    pub(crate) fn run_before_component_retry(component: &std::ffi::OsStr) {
+        let hook = BEFORE_COMPONENT_RETRY.lock().remove(component);
+        if let Some(hook) = hook {
+            let _pause = super::enter_blocking_test_hook();
+            hook();
+        }
+    }
 }
 
 /// Fsync a directory so recently created or renamed entries survive power loss.
@@ -644,6 +664,16 @@ pub(crate) fn fsync_directory_handle_std(dir: &Path, opened: Option<&std::fs::Fi
     Ok(())
 }
 
+/// Open an existing file to read it and flush it. Windows flushes only a
+/// handle with write access, so the file is opened read-write there; other
+/// platforms keep the read-only open that `sync_all` already accepts.
+pub(crate) fn open_file_for_sync_std(path: &Path) -> io::Result<std::fs::File> {
+    #[cfg(windows)]
+    return std::fs::OpenOptions::new().read(true).write(true).open(path);
+    #[cfg(not(windows))]
+    std::fs::File::open(path)
+}
+
 /// Async wrapper around [`fsync_dir_std`]; runs the blocking Unix fsync off the runtime.
 pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     fsync_dir_with_owner(dir.as_ref(), None).await
@@ -4584,7 +4614,7 @@ pub(crate) fn rename_source_is_missing(src_file_path: &Path, publication_root: &
     let Some(source_parent) = src_file_path.parent() else {
         return false;
     };
-    let source_parent_guard = match lock_windows_directory_tree(source_parent, None, publication_root) {
+    let source_parent_guard = match lock_windows_directory_tree(source_parent, publication_root) {
         Ok(guard) => guard,
         Err(err) => return err.kind() == io::ErrorKind::NotFound,
     };
@@ -5006,7 +5036,7 @@ fn prepare_rename_with_retry(
             open_windows_rename_source_identity(src_file_path, &source_parent_guard)?;
         (source_parent_guard, parent_guard, source_identity_anchor, expected_source_identity)
     } else {
-        let source_parent_guard = lock_windows_directory_tree(source_parent, destination_parent, publication_root)?;
+        let source_parent_guard = lock_windows_directory_tree(source_parent, publication_root)?;
         let (source_identity_anchor, expected_source_identity) =
             open_windows_rename_source_identity(src_file_path, &source_parent_guard)?;
         let parent_guard = prepare_destination_parent()?;
@@ -5427,7 +5457,9 @@ impl PublicationRoot {
     pub(crate) fn open_readonly(&self, path: &Path) -> io::Result<std::fs::File> {
         #[cfg(target_os = "linux")]
         return self.open_relative(path, rustix::fs::OFlags::RDONLY, rustix::fs::Mode::empty());
-        #[cfg(not(target_os = "linux"))]
+        #[cfg(windows)]
+        return open_windows_readonly(path);
+        #[cfg(not(any(target_os = "linux", windows)))]
         std::fs::File::open(path)
     }
 
@@ -5891,14 +5923,11 @@ impl RenameCommitGuard {
     }
 
     /// Reopen a destination tree for handle-relative child publication.
-    /// Ancestors remain write-exclusive while the final parent shares writes
-    /// required by the kernel's relative rename. Delete sharing stays omitted
-    /// throughout, so every retained directory identity remains pinned.
+    /// Every component is pinned by `open_windows_directory_component`.
     fn destination_directory_guard(&self, directory: &Path, create_missing: bool) -> io::Result<RenameDestinationPathGuard> {
         #[cfg(windows)]
         {
             use windows_sys::Wdk::Storage::FileSystem::{FILE_OPEN, FILE_OPEN_IF};
-            use windows_sys::Win32::Storage::FileSystem::{FILE_SHARE_READ, FILE_SHARE_WRITE};
 
             let relative = directory.strip_prefix(&self.destination_parent).map_err(|_| {
                 io::Error::new(
@@ -5931,35 +5960,17 @@ impl RenameCommitGuard {
             let parent = handles
                 .last()
                 .ok_or_else(|| io::Error::other("Windows destination guard lost its parent handle"))?;
-            let mut relative_components = relative
-                .components()
-                .filter_map(|component| match component {
-                    Component::Normal(component) => Some(component),
-                    _ => None,
-                })
-                .peekable();
-            let destination_parent_share = if relative_components.peek().is_none() {
-                FILE_SHARE_READ | FILE_SHARE_WRITE
-            } else {
-                FILE_SHARE_READ
-            };
-            handles.push(open_windows_relative_directory_component(
-                parent,
-                component,
-                FILE_OPEN,
-                destination_parent_share,
-            )?);
-            while let Some(component) = relative_components.next() {
+            handles.push(open_windows_directory_component(parent, component, FILE_OPEN)?);
+            let relative_components = relative.components().filter_map(|component| match component {
+                Component::Normal(component) => Some(component),
+                _ => None,
+            });
+            for component in relative_components {
                 let parent = handles
                     .last()
                     .ok_or_else(|| io::Error::other("Windows destination path guard lost its parent handle"))?;
                 let disposition = if create_missing { FILE_OPEN_IF } else { FILE_OPEN };
-                let share_access = if relative_components.peek().is_none() {
-                    FILE_SHARE_READ | FILE_SHARE_WRITE
-                } else {
-                    FILE_SHARE_READ
-                };
-                handles.push(open_windows_relative_directory_component(parent, component, disposition, share_access)?);
+                handles.push(open_windows_directory_component(parent, component, disposition)?);
             }
             Ok(RenameDestinationPathGuard {
                 directory: directory.to_path_buf(),
@@ -6003,18 +6014,47 @@ pub(crate) fn multipart_payload_commit_guard_std(
     })
 }
 
+#[cfg(windows)]
+fn multipart_payload_commit_guard_std(
+    metadata_guard: &RenameCommitGuard,
+    source_parent: &Path,
+    publication_root: &PublicationRoot,
+) -> io::Result<RenameCommitGuard> {
+    // Pin the payload staging tree like a distinct-tree metadata source; the
+    // target stays the exact parent handle the metadata transaction holds.
+    let source_parent_guard = lock_windows_directory_tree(source_parent, publication_root)?;
+    Ok(RenameCommitGuard {
+        source_parent: source_parent.to_path_buf(),
+        destination_parent: metadata_guard.destination_parent.clone(),
+        source_parent_guard,
+        destination_parent_guard: metadata_guard.destination_parent_guard.clone(),
+    })
+}
+
 /// Native MPU metadata and payload have distinct source parents. Reuse the
 /// exact target identity already held by the metadata transaction, while
 /// independently pinning the private payload source under the same root.
-#[cfg(target_os = "linux")]
+#[cfg(any(target_os = "linux", windows))]
 pub(crate) async fn multipart_payload_commit_guard(
     metadata_guard: &RenameCommitGuard,
     source_parent: &Path,
+    publication_root: &PublicationRoot,
     lease: Arc<NamespaceMutationLease>,
 ) -> io::Result<RenameCommitGuard> {
     let metadata_guard = metadata_guard.clone();
     let source_parent = source_parent.to_path_buf();
-    run_blocking_namespace_operation(lease, move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent)).await
+    #[cfg(target_os = "linux")]
+    let operation = {
+        // The Linux guard already carries its publication root.
+        let _ = publication_root;
+        move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent)
+    };
+    #[cfg(windows)]
+    let operation = {
+        let publication_root = publication_root.clone();
+        move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent, &publication_root)
+    };
+    run_blocking_namespace_operation(lease, operation).await
 }
 
 pub(crate) fn prepare_rename_commit_guard(
@@ -6056,20 +6096,16 @@ pub(crate) fn prepare_rename_commit_guard(
     }
     #[cfg(windows)]
     {
-        // A same-directory rename must use one shared-write parent handle:
-        // retaining a second read-only-share handle would block the kernel's
-        // relative target open. Delete sharing stays excluded, so identity is
-        // still pinned. Distinct source trees remain strict.
+        // A same-directory rename reuses one guard for both sides. Every pin
+        // shares writes and excludes delete; see
+        // `open_windows_directory_component`.
         let same_parent = publication_root.relative_path(source_parent)? == publication_root.relative_path(destination_parent)?;
         let (source_parent_guard, destination_parent_guard) = if same_parent {
             let destination_parent_guard =
                 mkdir_all_below_existing_base_std(destination_parent, destination_base, publication_root)?;
             (destination_parent_guard.clone(), destination_parent_guard)
         } else {
-            // The source parent also hosts private rollback staging files.
-            // Their handle-relative publication needs write sharing on this
-            // final directory while delete sharing remains excluded.
-            let source_parent_guard = lock_windows_directory_tree(source_parent, Some(source_parent), publication_root)?;
+            let source_parent_guard = lock_windows_directory_tree(source_parent, publication_root)?;
             let destination_parent_guard =
                 mkdir_all_below_existing_base_std(destination_parent, destination_base, publication_root)?;
             (source_parent_guard, destination_parent_guard)
@@ -6089,6 +6125,28 @@ pub(crate) fn prepare_rename_commit_guard(
     }
 }
 
+/// Unix opens a directory read-only and lets the reader classify it, while
+/// Windows refuses the plain open with ACCESS_DENIED. Reopen only an actual
+/// directory with backup semantics so the same `is_dir` checks apply.
+#[cfg(windows)]
+fn open_windows_readonly(path: &Path) -> io::Result<std::fs::File> {
+    use std::os::windows::fs::OpenOptionsExt;
+    use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_BACKUP_SEMANTICS;
+
+    match std::fs::File::open(path) {
+        Err(error)
+            if error.kind() == io::ErrorKind::PermissionDenied
+                && std::fs::symlink_metadata(path).is_ok_and(|metadata| metadata.is_dir()) =>
+        {
+            std::fs::OpenOptions::new()
+                .read(true)
+                .custom_flags(FILE_FLAG_BACKUP_SEMANTICS)
+                .open(path)
+        }
+        result => result,
+    }
+}
+
 #[cfg(windows)]
 fn open_windows_publication_root(path: &Path) -> io::Result<(PathBuf, WindowsDirectoryHandle)> {
     use std::os::windows::fs::OpenOptionsExt;
@@ -6275,22 +6333,12 @@ fn windows_file_attribute_tag(
 }
 
 #[cfg(windows)]
-fn lock_windows_directory_tree(
-    path: &Path,
-    shared_write_ancestor: Option<&Path>,
-    publication_root: &PublicationRoot,
-) -> io::Result<ExistingBaseDirectoryGuard> {
+fn lock_windows_directory_tree(path: &Path, publication_root: &PublicationRoot) -> io::Result<ExistingBaseDirectoryGuard> {
     use windows_sys::Wdk::Storage::FileSystem::FILE_OPEN;
-    use windows_sys::Win32::Storage::FileSystem::{FILE_SHARE_READ, FILE_SHARE_WRITE};
 
     let relative = publication_root.relative_path(path)?;
-    let shared_write_relative = shared_write_ancestor
-        .map(|ancestor| publication_root.relative_path(ancestor))
-        .transpose()?
-        .filter(|ancestor| relative.starts_with(*ancestor));
     let mut handles = Vec::with_capacity(relative.components().count().saturating_add(1));
     handles.push(publication_root.directory.clone());
-    let mut opened_relative = PathBuf::new();
 
     for component in relative.components() {
         let Component::Normal(component) = component else {
@@ -6305,26 +6353,13 @@ fn lock_windows_directory_tree(
         let parent = handles
             .last()
             .ok_or_else(|| io::Error::other("Windows directory guard lost its root handle"))?;
-        opened_relative.push(component);
-        let child = if shared_write_relative.is_some_and(|ancestor| opened_relative.as_path() == ancestor) {
-            // A handle-relative rename opens the target parent for write. When
-            // that parent is also a source ancestor, this source-side handle
-            // must share write access or the transaction blocks itself. Delete
-            // sharing remains omitted, so the directory identity stays pinned.
-            open_windows_relative_directory_component(parent, component, FILE_OPEN, FILE_SHARE_READ | FILE_SHARE_WRITE)?
-        } else {
-            open_windows_directory_component(parent, component, FILE_OPEN)?
-        };
-        handles.push(child);
+        handles.push(open_windows_directory_component(parent, component, FILE_OPEN)?);
     }
 
     Ok(ExistingBaseDirectoryGuard::new(handles))
 }
 
 #[cfg(windows)]
-// SAFETY: the object attributes borrow a checked UTF-16 component and live
-// parent handle for the duration of the synchronous NtCreateFile call.
-#[allow(unsafe_code)]
 fn open_windows_relative(
     parent: &winapi_util::Handle,
     component: &std::ffi::OsStr,
@@ -6335,6 +6370,64 @@ fn open_windows_relative(
     file_attributes: u32,
     dont_reparse: bool,
 ) -> io::Result<winapi_util::Handle> {
+    open_windows_relative_with_status(
+        parent,
+        component,
+        &WindowsRelativeOpen {
+            desired_access,
+            share_access,
+            create_disposition,
+            create_options,
+            file_attributes,
+            dont_reparse,
+        },
+    )
+    .map_err(|failure| failure.error)
+}
+
+/// `NtCreateFile` arguments for one handle-relative open.
+#[cfg(windows)]
+struct WindowsRelativeOpen {
+    desired_access: u32,
+    share_access: u32,
+    create_disposition: u32,
+    create_options: u32,
+    file_attributes: u32,
+    dont_reparse: bool,
+}
+
+/// A failed relative open that keeps the NTSTATUS, which tells transient
+/// states such as a pending delete apart from errors that share its Win32 code.
+#[cfg(windows)]
+struct WindowsRelativeOpenError {
+    status: Option<windows_sys::Win32::Foundation::NTSTATUS>,
+    error: io::Error,
+}
+
+#[cfg(windows)]
+impl From<io::Error> for WindowsRelativeOpenError {
+    fn from(error: io::Error) -> Self {
+        Self { status: None, error }
+    }
+}
+
+#[cfg(windows)]
+// SAFETY: the object attributes borrow a checked UTF-16 component and live
+// parent handle for the duration of the synchronous NtCreateFile call.
+#[allow(unsafe_code)]
+fn open_windows_relative_with_status(
+    parent: &winapi_util::Handle,
+    component: &std::ffi::OsStr,
+    request: &WindowsRelativeOpen,
+) -> std::result::Result<winapi_util::Handle, WindowsRelativeOpenError> {
+    let &WindowsRelativeOpen {
+        desired_access,
+        share_access,
+        create_disposition,
+        create_options,
+        file_attributes,
+        dont_reparse,
+    } = request;
     use std::{
         mem::size_of,
         os::windows::{ffi::OsStrExt, io::AsRawHandle, io::FromRawHandle},
@@ -6349,10 +6442,7 @@ fn open_windows_relative(
 
     let mut name = component.encode_wide().collect::<Vec<_>>();
     if name.is_empty() || name.contains(&0) {
-        return Err(io::Error::new(
-            io::ErrorKind::InvalidInput,
-            "relative Windows file name is empty or contains a NUL",
-        ));
+        return Err(io::Error::new(io::ErrorKind::InvalidInput, "relative Windows file name is empty or contains a NUL").into());
     }
     let name_bytes = name
         .len()
@@ -6392,13 +6482,17 @@ fn open_windows_relative(
         )
     };
     if status < 0 {
-        return match i32::try_from(unsafe { RtlNtStatusToDosError(status) }) {
-            Ok(code) => Err(io::Error::from_raw_os_error(code)),
-            Err(_) => Err(io::Error::other(format!("Windows relative open failed with NTSTATUS {status:#x}"))),
+        let error = match i32::try_from(unsafe { RtlNtStatusToDosError(status) }) {
+            Ok(code) => io::Error::from_raw_os_error(code),
+            Err(_) => io::Error::other(format!("Windows relative open failed with NTSTATUS {status:#x}")),
         };
+        return Err(WindowsRelativeOpenError {
+            status: Some(status),
+            error,
+        });
     }
     if handle.is_null() {
-        return Err(io::Error::other("Windows relative open returned an invalid handle"));
+        return Err(io::Error::other("Windows relative open returned an invalid handle").into());
     }
 
     Ok(unsafe { winapi_util::Handle::from_raw_handle(handle) })
@@ -6633,7 +6727,7 @@ pub(crate) fn compare_and_update_control_file(
     let parent = file_path
         .parent()
         .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "conditional file has no parent"))?;
-    let parent_guard = lock_windows_directory_tree(parent, Some(parent), publication_root)?;
+    let parent_guard = lock_windows_directory_tree(parent, publication_root)?;
     let lock = open_windows_relative(
         parent_guard.last_handle()?,
         std::ffi::OsStr::new(".rustfs-cas.lock"),
@@ -6697,41 +6791,78 @@ pub(crate) fn compare_and_update_control_file(
     Ok(ConditionalFileUpdate::Updated)
 }
 
+/// How often a directory pin retries a transient delete-access conflict.
+/// Six doubling delays from 1 ms wait about 63 ms in total, which outlasts a
+/// prune whose thread is preempted while it holds delete access.
+#[cfg(windows)]
+const WINDOWS_DIRECTORY_PIN_RETRIES: u32 = 6;
+#[cfg(windows)]
+const WINDOWS_DIRECTORY_PIN_FIRST_RETRY_DELAY: std::time::Duration = std::time::Duration::from_millis(1);
+
+/// Pin one directory component of a guarded path.
+///
+/// The handle shares reads and writes but never delete: renaming or removing
+/// a directory needs delete access to it, so omitting delete sharing pins its
+/// identity. Write sharing must stay granted on every component, ancestors
+/// included, because the kernel opens the target parent for write during a
+/// rename. A directory pinned here as one object's ancestor is also the
+/// publication parent of a nested object (`a/b` holds `a/b/c`), and a
+/// read-only share would make that unrelated rename fail with
+/// ERROR_SHARING_VIOLATION. Write sharing does not unpin an ancestor: a
+/// pinned ancestor always contains its pinned child, and NTFS sets a
+/// name-surrogate reparse point only on an empty directory.
+///
+/// Excluding delete also means the open fails while another handle holds
+/// delete access. Pruning an empty prefix directory takes that access only
+/// briefly: `RemoveDirectoryW` opens the directory for delete and closes it
+/// again, whether or not the directory turns out to be empty. A sharing
+/// violation or a pending delete is therefore retried a bounded number of
+/// times. After a removal finishes, the retry sees the directory as missing
+/// and follows `create_disposition`, so `FILE_OPEN` still reports it missing
+/// and `FILE_OPEN_IF` creates it again.
 #[cfg(windows)]
 fn open_windows_directory_component(
     parent: &WindowsDirectoryHandle,
     component: &std::ffi::OsStr,
     create_disposition: u32,
-) -> io::Result<WindowsDirectoryHandle> {
-    use windows_sys::Win32::Storage::FileSystem::FILE_SHARE_READ;
-
-    open_windows_relative_directory_component(parent, component, create_disposition, FILE_SHARE_READ)
-}
-
-#[cfg(windows)]
-fn open_windows_relative_directory_component(
-    parent: &WindowsDirectoryHandle,
-    component: &std::ffi::OsStr,
-    create_disposition: u32,
-    share_access: u32,
 ) -> io::Result<WindowsDirectoryHandle> {
     use windows_sys::{
         Wdk::Storage::FileSystem::{FILE_DIRECTORY_FILE, FILE_OPEN_REPARSE_POINT},
-        Win32::Storage::FileSystem::{
-            FILE_ATTRIBUTE_DIRECTORY, FILE_ATTRIBUTE_REPARSE_POINT, FILE_READ_ATTRIBUTES, FILE_TRAVERSE,
+        Win32::{
+            Foundation::{STATUS_DELETE_PENDING, STATUS_SHARING_VIOLATION},
+            Storage::FileSystem::{
+                FILE_ATTRIBUTE_DIRECTORY, FILE_ATTRIBUTE_REPARSE_POINT, FILE_READ_ATTRIBUTES, FILE_SHARE_READ, FILE_SHARE_WRITE,
+                FILE_TRAVERSE,
+            },
         },
     };
 
-    let anchor = open_windows_relative(
-        &parent.handle,
-        component,
-        FILE_TRAVERSE | FILE_READ_ATTRIBUTES,
-        share_access,
+    let request = WindowsRelativeOpen {
+        desired_access: FILE_TRAVERSE | FILE_READ_ATTRIBUTES,
+        share_access: FILE_SHARE_READ | FILE_SHARE_WRITE,
         create_disposition,
-        FILE_DIRECTORY_FILE | FILE_OPEN_REPARSE_POINT,
-        FILE_ATTRIBUTE_DIRECTORY,
-        true,
-    )?;
+        create_options: FILE_DIRECTORY_FILE | FILE_OPEN_REPARSE_POINT,
+        file_attributes: FILE_ATTRIBUTE_DIRECTORY,
+        dont_reparse: true,
+    };
+    let mut retry_delay = WINDOWS_DIRECTORY_PIN_FIRST_RETRY_DELAY;
+    let mut retries = 0;
+    let anchor = loop {
+        match open_windows_relative_with_status(&parent.handle, component, &request) {
+            Ok(anchor) => break anchor,
+            Err(WindowsRelativeOpenError {
+                status: Some(STATUS_SHARING_VIOLATION | STATUS_DELETE_PENDING),
+                ..
+            }) if retries < WINDOWS_DIRECTORY_PIN_RETRIES => {
+                #[cfg(test)]
+                prepare_rename_test_hooks::run_before_component_retry(component);
+                std::thread::sleep(retry_delay);
+                retry_delay = retry_delay.saturating_mul(2);
+                retries += 1;
+            }
+            Err(failure) => return Err(failure.error),
+        }
+    };
     let info = windows_file_attribute_tag(&anchor)?;
     if info.FileAttributes & FILE_ATTRIBUTE_DIRECTORY == 0 {
         return Err(io::Error::from(io::ErrorKind::NotADirectory));
@@ -7035,7 +7166,7 @@ pub(crate) fn pin_multipart_source_directory(
     #[cfg(windows)]
     {
         let _ = base;
-        lock_windows_directory_tree(directory, Some(directory), publication_root)
+        lock_windows_directory_tree(directory, publication_root)
     }
     #[cfg(not(any(unix, windows)))]
     {
@@ -7125,12 +7256,31 @@ pub(crate) fn rename_multipart_source_owned_std(
     {
         let _ = &guard;
         let source_guard = pin_multipart_source_directory(source_parent, base, root)?;
+        // The prepared Windows source holds DELETE access and shares only
+        // read, so the precondition's own opens of that entry (or, for a
+        // directory, of its guarded tree) would fail with a sharing violation.
+        // Check it first under a non-exclusive identity anchor, then require
+        // the exclusive handle to name the same file.
+        #[cfg(windows)]
+        let (_source_identity_anchor, expected_source_identity) = {
+            let anchor = open_windows_rename_source_identity(source, &source_guard)?;
+            precondition()?;
+            anchor
+        };
         let preparation = prepare_rename_with_retry(source, destination, base, root)?;
+        #[cfg(windows)]
+        if windows_file_identity(&preparation.source)? != expected_source_identity {
+            return Err(io::Error::new(
+                io::ErrorKind::InvalidData,
+                "rename source identity changed after its precondition",
+            ));
+        }
         #[cfg(all(any(test, feature = "test-util"), not(windows)))]
         {
             prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, source);
             prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, destination);
         }
+        #[cfg(not(windows))]
         precondition()?;
         #[cfg(unix)]
         validate_prepared_rename_source(&pinned_source, source)?;
@@ -7536,12 +7686,15 @@ fn read_multipart_metadata(
         let parent = path
             .parent()
             .ok_or_else(|| io::Error::other("multipart metadata has no parent"))?;
-        let guard = prepare_rename_commit_guard(parent, parent, base, publication_root)?;
+        // A read must not recreate a retired upload: pin the existing parent
+        // instead of taking the write-side guard, which creates missing parents.
+        existing_base_directory_suffix(parent, base)?;
+        let guard = pin_multipart_source_directory(parent, base, publication_root)?;
         let name = path
             .file_name()
             .ok_or_else(|| io::Error::other("multipart metadata has no name"))?;
         let mut file = open_windows_relative(
-            guard.source_parent_guard.last_handle()?,
+            guard.last_handle()?,
             name,
             SYNCHRONIZE | FILE_READ_ATTRIBUTES | FILE_READ_DATA,
             FILE_SHARE_READ,
@@ -7717,7 +7870,7 @@ mod multipart_source_preservation_tests {
         let metadata_guard =
             prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
         let lease = acquire_namespace_mutation_lease(&target).await;
-        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, lease.clone())
+        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
             .await
             .expect("distinct pinned payload source");
         assert!(
@@ -7752,6 +7905,61 @@ mod multipart_source_preservation_tests {
         );
     }
 
+    #[cfg(windows)]
+    #[tokio::test]
+    async fn windows_multipart_payload_guard_publishes_distinct_source_into_metadata_target() {
+        // Build paths from the configured root: the resolved handle path may
+        // be verbatim, where `/` in a joined key is not a separator.
+        let (root, publication, _source, _destination) = fixture();
+        let root = root.path();
+        let canonical = root.join("multipart/upload");
+        let temporary = root.join("tmp/attempt");
+        let bucket = root.join("bucket");
+        let target = bucket.join("object");
+        for directory in [&canonical, &temporary, &target] {
+            std::fs::create_dir_all(directory).expect("two-source guard directories");
+        }
+        let payload = temporary.join("payload");
+        std::fs::create_dir(&payload).expect("private payload DD");
+        std::fs::write(payload.join("part.1"), b"native selected shard").expect("private payload shard");
+        let head = canonical.join("xl.meta");
+        let destination = target.join("xl.meta");
+        let metadata_guard =
+            prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
+        let lease = acquire_namespace_mutation_lease(&target).await;
+        assert!(
+            rename_all_with_commit_guard(&payload, target.join("payload"), &bucket, &publication, &metadata_guard, lease.clone())
+                .await
+                .is_err(),
+            "the metadata guard does not pin the private payload parent"
+        );
+        assert!(
+            payload.join("part.1").exists(),
+            "a rejected payload rename must leave its source in place"
+        );
+        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
+            .await
+            .expect("distinct pinned payload source");
+        let mut prepared = create_prepared_rename_source_with_commit_guard(&head, &destination, &metadata_guard)
+            .expect("canonical prepared metadata");
+        prepared
+            .write_all(b"prepared native metadata", true)
+            .expect("durable canonical metadata");
+        rename_all_with_commit_guard(&payload, target.join("payload"), &bucket, &publication, &payload_guard, lease.clone())
+            .await
+            .expect("payload rename uses its own pinned TMP parent");
+        assert!(head.exists(), "payload publication must not consume the canonical head");
+        rename_all_with_prepared_source(prepared, &head, &destination, &bucket, &publication, &metadata_guard, lease)
+            .await
+            .expect("metadata rename keeps its canonical parent");
+        assert!(!head.exists());
+        assert_eq!(std::fs::read(&destination).expect("published metadata"), b"prepared native metadata");
+        assert_eq!(
+            std::fs::read(target.join("payload/part.1")).expect("published selected shard"),
+            b"native selected shard"
+        );
+    }
+
     #[cfg(target_os = "linux")]
     #[tokio::test]
     async fn multipart_payload_guard_rejects_replaced_private_source_or_shared_destination() {
@@ -7773,7 +7981,7 @@ mod multipart_source_preservation_tests {
             let metadata_guard =
                 prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
             let lease = acquire_namespace_mutation_lease(&target).await;
-            let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, lease.clone())
+            let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
                 .await
                 .expect("pin payload and shared target");
             let replaced = if replace_source { &temporary } else { &target };
@@ -8030,7 +8238,6 @@ pub(crate) fn mkdir_all_below_existing_base_std(
     #[cfg(windows)]
     {
         use windows_sys::Wdk::Storage::FileSystem::{FILE_OPEN, FILE_OPEN_IF};
-        use windows_sys::Win32::Storage::FileSystem::{FILE_SHARE_READ, FILE_SHARE_WRITE};
 
         let base_relative = publication_root.relative_path(base_dir)?;
         let capacity = base_relative
@@ -8041,13 +8248,12 @@ pub(crate) fn mkdir_all_below_existing_base_std(
         let mut handles = Vec::with_capacity(capacity);
         handles.push(publication_root.directory.clone());
         let mut guard = ExistingBaseDirectoryGuard::new(handles);
-        let mut components = base_relative
+        let components = base_relative
             .components()
             .map(|component| (component, FILE_OPEN))
             .chain(relative.components().map(|component| (component, FILE_OPEN_IF)))
-            .filter(|(component, _)| !matches!(component, Component::CurDir))
-            .peekable();
-        while let Some((component, disposition)) = components.next() {
+            .filter(|(component, _)| !matches!(component, Component::CurDir));
+        for (component, disposition) in components {
             let Component::Normal(component) = component else {
                 return Err(io::Error::new(
                     io::ErrorKind::InvalidInput,
@@ -8058,22 +8264,16 @@ pub(crate) fn mkdir_all_below_existing_base_std(
                 .handles
                 .last()
                 .ok_or_else(|| io::Error::other("Windows publication root guard is empty"))?;
-            // The kernel opens the final parent for write during a relative
-            // rename. Share writes from its first open: a temporary read-only
-            // share would block another rename into the same trash directory.
-            // Ancestors stay strict and no handle shares delete access, keeping
-            // every retained directory identity pinned.
-            let share_access = if components.peek().is_none() {
-                FILE_SHARE_READ | FILE_SHARE_WRITE
-            } else {
-                FILE_SHARE_READ
-            };
-            let child = open_windows_relative_directory_component(parent, component, disposition, share_access)?;
+            // Every component, including the final parent the kernel opens
+            // for write during a relative rename, shares writes from its first
+            // open; see `open_windows_directory_component`.
+            let child = open_windows_directory_component(parent, component, disposition)?;
             guard.handles.push(child);
-            #[cfg(test)]
-            if components.peek().is_none() {
-                prepare_rename_test_hooks::run_after_component_opened(dir_path);
-            }
+        }
+        // The hook observes the window right after the final component opened.
+        #[cfg(test)]
+        if guard.handles.len() > 1 {
+            prepare_rename_test_hooks::run_after_component_opened(dir_path);
         }
 
         Ok(guard)
@@ -9479,7 +9679,7 @@ mod tests {
         use std::os::windows::fs::OpenOptionsExt;
         use std::sync::atomic::{AtomicBool, Ordering};
         use windows_sys::Win32::{
-            Foundation::{ERROR_SHARING_VIOLATION, GENERIC_WRITE},
+            Foundation::ERROR_SHARING_VIOLATION,
             Storage::FileSystem::{
                 DELETE, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_DELETE, FILE_SHARE_READ,
                 FILE_SHARE_WRITE,
@@ -9521,13 +9721,13 @@ mod tests {
             let second_preparation = prepare_rename_with_retry(&sources[1], &destinations[1], base, &publication_root)
                 .expect("prepare the second trash rename");
             assert!(interleaved.load(Ordering::Acquire), "the competing parent-open window must be exercised");
-            for (path, access) in [(&tmp, GENERIC_WRITE), (&trash, DELETE)] {
+            for path in [&tmp, &trash] {
                 let err = std::fs::OpenOptions::new()
-                    .access_mode(access)
+                    .access_mode(DELETE)
                     .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
                     .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
                     .open(path)
-                    .expect_err("ancestor writes and final-parent deletion must remain excluded");
+                    .expect_err("pinned ancestors and final parents must exclude deletion");
                 assert_eq!(
                     err.raw_os_error(),
                     Some(i32::try_from(ERROR_SHARING_VIOLATION).expect("Windows error code must fit i32"))
@@ -9544,6 +9744,115 @@ mod tests {
         }
     }
 
+    #[cfg(windows)]
+    #[test]
+    fn windows_rename_into_a_pinned_ancestor_succeeds_while_deletion_stays_excluded() {
+        use std::os::windows::fs::OpenOptionsExt;
+        use windows_sys::Win32::{
+            Foundation::ERROR_SHARING_VIOLATION,
+            Storage::FileSystem::{
+                DELETE, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_DELETE, FILE_SHARE_READ,
+                FILE_SHARE_WRITE,
+            },
+        };
+
+        // `bucket/parent` is an object directory and also the prefix of the
+        // nested object `bucket/parent/child`. A guard on the nested object
+        // pins `parent` as an ancestor while the parent object publishes a new
+        // data directory into it by pathname.
+        let temp_dir = tempdir().expect("create temp dir");
+        let base = temp_dir.path().join("bucket");
+        let parent = base.join("parent");
+        let child = parent.join("child");
+        std::fs::create_dir_all(&child).expect("create nested object directory");
+        let staged = temp_dir.path().join("staged-data-dir");
+        std::fs::create_dir(&staged).expect("create staged data directory");
+        std::fs::write(staged.join("part.1"), b"payload").expect("write staged part");
+        let publication_root = PublicationRoot::new(temp_dir.path()).expect("open publication root");
+        let child_guard =
+            super::mkdir_all_below_existing_base_std(&child, &base, &publication_root).expect("pin the nested object tree");
+
+        let destination = parent.join(uuid::Uuid::new_v4().to_string());
+        std::fs::rename(&staged, &destination).expect("a pinned ancestor must accept a rename into it");
+        assert_eq!(std::fs::read(destination.join("part.1")).expect("read published part"), b"payload");
+
+        let err = std::fs::OpenOptions::new()
+            .access_mode(DELETE)
+            .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
+            .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
+            .open(&parent)
+            .expect_err("the pinned ancestor must still refuse deletion");
+        assert_eq!(
+            err.raw_os_error(),
+            Some(i32::try_from(ERROR_SHARING_VIOLATION).expect("Windows error code must fit i32"))
+        );
+        drop(child_guard);
+    }
+
+    #[cfg(windows)]
+    #[test]
+    fn windows_directory_pin_retries_a_transient_delete_holder_and_fails_closed_on_a_lasting_one() {
+        use std::os::windows::fs::OpenOptionsExt;
+        use windows_sys::Win32::{
+            Foundation::ERROR_SHARING_VIOLATION,
+            Storage::FileSystem::{
+                DELETE, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_READ_ATTRIBUTES, FILE_SHARE_DELETE,
+                FILE_SHARE_READ, FILE_SHARE_WRITE, SYNCHRONIZE,
+            },
+        };
+
+        // Open the prefix the way `RemoveDirectoryW` does while it prunes an
+        // object prefix: delete access, every share mode granted.
+        let open_like_a_pruner = |path: &Path| {
+            std::fs::OpenOptions::new()
+                .access_mode(DELETE | SYNCHRONIZE | FILE_READ_ATTRIBUTES)
+                .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
+                .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
+                .open(path)
+                .expect("open the prefix for delete")
+        };
+        let temp_dir = tempdir().expect("create temp dir");
+        let base = temp_dir.path().join("bucket");
+        let publication_root = PublicationRoot::new(temp_dir.path()).expect("open publication root");
+
+        let transient_name = uuid::Uuid::new_v4().to_string();
+        let transient_prefix = base.join(&transient_name);
+        std::fs::create_dir_all(&transient_prefix).expect("create transient prefix");
+        let holder = Arc::new(std::sync::Mutex::new(Some(open_like_a_pruner(&transient_prefix))));
+        let holder_for_hook = Arc::clone(&holder);
+        let released = Arc::new(std::sync::atomic::AtomicBool::new(false));
+        let released_for_hook = Arc::clone(&released);
+        prepare_rename_test_hooks::install_before_component_retry(std::ffi::OsStr::new(&transient_name), move || {
+            holder_for_hook.lock().expect("holder mutex poisoned").take();
+            released_for_hook.store(true, std::sync::atomic::Ordering::Release);
+        });
+        let child = transient_prefix.join("child");
+        let guard = super::mkdir_all_below_existing_base_std(&child, &base, &publication_root)
+            .expect("a pin must outlast a transient delete holder");
+        assert!(
+            released.load(std::sync::atomic::Ordering::Acquire),
+            "the pin must have met the delete holder"
+        );
+        assert!(child.is_dir(), "the retried walk must still create the child");
+        let err = std::fs::OpenOptions::new()
+            .access_mode(DELETE)
+            .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
+            .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
+            .open(&transient_prefix)
+            .expect_err("the retried pin must still exclude deletion");
+        assert_eq!(err.raw_os_error(), i32::try_from(ERROR_SHARING_VIOLATION).ok());
+        drop(guard);
+
+        let lasting_prefix = base.join(uuid::Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&lasting_prefix).expect("create lasting prefix");
+        let lasting_holder = open_like_a_pruner(&lasting_prefix);
+        let Err(err) = super::mkdir_all_below_existing_base_std(&lasting_prefix.join("child"), &base, &publication_root) else {
+            panic!("a pin must not wait forever on a lasting delete holder");
+        };
+        assert_eq!(err.raw_os_error(), i32::try_from(ERROR_SHARING_VIOLATION).ok());
+        drop(lasting_holder);
+    }
+
     #[cfg(windows)]
     #[tokio::test]
     async fn windows_rename_all_supports_same_parent_publication() {
@@ -10241,7 +10550,7 @@ mod tests {
 
         let publication_root = test_publication_root(&[&src, &dst, &base]);
         let source_parent_guard =
-            lock_windows_directory_tree(src.parent().expect("source reparse entry must have a parent"), None, &publication_root)
+            lock_windows_directory_tree(src.parent().expect("source reparse entry must have a parent"), &publication_root)
                 .expect("pin the source parent tree");
         let identity_error = match open_windows_rename_source_identity(&src, &source_parent_guard) {
             Ok(_) => panic!("a non-dedup source reparse entry must be rejected by its tag"),
@@ -10275,30 +10584,51 @@ mod tests {
 
     #[cfg(windows)]
     #[tokio::test]
-    async fn windows_rename_all_excludes_source_parent_reparse_writers() {
+    async fn windows_rename_all_keeps_the_source_parent_from_becoming_a_reparse_point() {
         use std::os::windows::fs::OpenOptionsExt;
         use windows_sys::Win32::{
-            Foundation::GENERIC_WRITE,
-            Storage::FileSystem::{FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_READ, FILE_SHARE_WRITE},
+            Foundation::{ERROR_DIR_NOT_EMPTY, ERROR_SHARING_VIOLATION, GENERIC_WRITE},
+            Storage::FileSystem::{
+                DELETE, FILE_FLAG_BACKUP_SEMANTICS, FILE_FLAG_OPEN_REPARSE_POINT, FILE_SHARE_DELETE, FILE_SHARE_READ,
+                FILE_SHARE_WRITE,
+            },
         };
 
         let temp_dir = tempdir().expect("create temp dir");
         let base = temp_dir.path().join("bucket");
         let source_parent = temp_dir.path().join("staging");
+        let outside = temp_dir.path().join("outside");
         let src = source_parent.join("staged-object");
         let dst = base.join("published-object");
         std::fs::create_dir(&base).expect("create destination base");
         std::fs::create_dir(&source_parent).expect("create source parent");
+        std::fs::create_dir(&outside).expect("create outside target");
         std::fs::write(&src, b"original").expect("write original staged object");
 
         let source_parent_for_hook = source_parent.clone();
+        let outside_for_hook = outside.clone();
         windows_rename_test_hooks::install_before_publication(&dst, move || {
-            std::fs::OpenOptions::new()
+            // The pin shares writes, so a writer can open the source parent,
+            // but the parent still holds the staged entry and NTFS refuses to
+            // turn a non-empty directory into a mount point.
+            let writable_parent = std::fs::OpenOptions::new()
                 .access_mode(GENERIC_WRITE)
                 .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE)
                 .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
                 .open(&source_parent_for_hook)
-                .expect_err("the source-parent guard must exclude a reparse writer");
+                .map(winapi_util::Handle::from_file)
+                .expect("a write-sharing pin admits a concurrent writer");
+            let err = try_set_windows_mount_point(&writable_parent, &outside_for_hook)
+                .expect_err("the pinned source parent must not become a mount point");
+            assert_eq!(err.raw_os_error(), i32::try_from(ERROR_DIR_NOT_EMPTY).ok());
+
+            let err = std::fs::OpenOptions::new()
+                .access_mode(DELETE)
+                .share_mode(FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE)
+                .custom_flags(FILE_FLAG_BACKUP_SEMANTICS | FILE_FLAG_OPEN_REPARSE_POINT)
+                .open(&source_parent_for_hook)
+                .expect_err("the source-parent guard must exclude deletion and rename");
+            assert_eq!(err.raw_os_error(), i32::try_from(ERROR_SHARING_VIOLATION).ok());
         });
 
         rename_all(&src, &dst, &base)
@@ -10307,6 +10637,7 @@ mod tests {
 
         assert!(!src.exists(), "the original staged entry must be moved");
         assert_eq!(std::fs::read(&dst).expect("read published original object"), b"original");
+        assert!(!outside.join("published-object").exists(), "publication must not escape the base");
     }
 
     #[cfg(windows)]
@@ -10386,7 +10717,7 @@ mod tests {
         let guard = mkdir_all_below_existing_base_std(&base, &base).expect("guard destination base");
         let publication_root = test_publication_root(&[&src, &dst, &base]);
         let source_parent_guard =
-            lock_windows_directory_tree(src.parent().expect("source path must have a parent"), None, &publication_root)
+            lock_windows_directory_tree(src.parent().expect("source path must have a parent"), &publication_root)
                 .expect("anchor source parent");
         let source = open_windows_rename_source(&src, &source_parent_guard).expect("anchor source entry");
 

crates/ecstore/src/services/rebalance/entry.rs

@@ -1910,6 +1910,10 @@ mod tests {
                 &ObjectOptions {
                     version_id: Some(version_id.to_string()),
                     mod_time: earlier(version.mod_time),
+                    // Without the object lock the rename waits for every
+                    // drive, so no early-ACK tail is still writing the copies
+                    // removed below.
+                    no_lock: true,
                     ..versioned
                 },
             ))

crates/ecstore/src/store/init/multipart_copy_heal_tests.rs

@@ -318,7 +318,10 @@ mod multipart_copy_heal_tests {
             let mut corrupt = original.clone();
             corrupt[0] ^= 0x80;
             std::fs::write(&shard, &corrupt).expect("actual bitrot corruption in shared physical shard");
-            std::fs::File::open(&shard)
+            // Windows only flushes a handle opened for writing.
+            std::fs::OpenOptions::new()
+                .write(true)
+                .open(&shard)
                 .expect("open corrupted actual shard")
                 .sync_all()
                 .expect("persist actual fault bytes");

crates/protocols/src/sftp/dir.rs

@@ -56,17 +56,6 @@ pub(super) fn dot_entries() -> Vec<File> {
     ]
 }
 
-/// Client-visible S3 key for an SFTP sub-directory.
-///
-/// `object_key` is the parsed prefix without a trailing slash. The
-/// returned key ends in `/` so the storage layer can encode it to the
-/// reserved directory marker. Callers must not encode the key first:
-/// `validate_object_key` rejects any client key that already contains
-/// the marker suffix.
-fn directory_client_key(object_key: &str) -> String {
-    format!("{object_key}/")
-}
-
 fn directory_listing_input(
     bucket: &str,
     prefix: &str,
@@ -311,19 +300,19 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
         Ok(())
     }
 
-    /// MKDIR for a sub-directory path: write a zero-byte object at the
-    /// client-visible directory key (`foo/`). Storage encodes that key to
-    /// the on-disk marker `foo__XLDIR__` after the S3 client-key boundary.
-    /// The encoded suffix is reserved, so PutObject must not be given it.
+    /// MKDIR for a sub-directory path: write a zero-byte object at
+    /// encode_dir_object(prefix + "/"). The encoding maps "foo/" to
+    /// "foo__XLDIR__", which matches the RustFS marker convention used
+    /// by the S3, Swift, and WebDAV backends.
     pub(super) async fn mkdir_subdir_marker(&self, bucket: &str, object_key: &str) -> Result<(), SftpError> {
-        let dir_key = directory_client_key(object_key);
-        self.authorize(&S3Action::PutObject, bucket, Some(&dir_key)).await?;
+        let marker_key = path::encode_dir_object(&format!("{object_key}/"));
+        self.authorize(&S3Action::PutObject, bucket, Some(&marker_key)).await?;
 
         let body = stream::once(async { Ok::<Bytes, std::io::Error>(Bytes::new()) });
         let streaming = StreamingBlob::wrap(body);
         let input = PutObjectInput::builder()
             .bucket(bucket.to_string())
-            .key(dir_key)
+            .key(marker_key.clone())
             .content_length(Some(0))
             .body(Some(streaming))
             .build()
@@ -344,14 +333,15 @@ impl<S: StorageBackend + Send + Sync + 'static> SftpDriver<S> {
     }
 
     /// RMDIR for a sub-directory path: validate no objects under the
-    /// prefix, then authorise and delete the trailing-slash directory key.
-    /// Storage encodes that key to the on-disk marker.
+    /// prefix, then authorise and delete the __XLDIR__ marker that
+    /// represents the directory.
     pub(super) async fn rmdir_subdir_marker(&self, bucket: &str, object_key: &str) -> Result<(), SftpError> {
-        let dir_key = directory_client_key(object_key);
-        self.validate_directory_empty(bucket, &dir_key).await?;
+        let prefix = format!("{object_key}/");
+        self.validate_directory_empty(bucket, &prefix).await?;
 
-        self.authorize(&S3Action::DeleteObject, bucket, Some(&dir_key)).await?;
-        self.run_backend("delete_object", self.storage.delete_object(bucket, &dir_key, self.credentials()))
+        let marker_key = path::encode_dir_object(&prefix);
+        self.authorize(&S3Action::DeleteObject, bucket, Some(&marker_key)).await?;
+        self.run_backend("delete_object", self.storage.delete_object(bucket, &marker_key, self.credentials()))
             .await?;
         Ok(())
     }
@@ -847,95 +837,4 @@ mod tests {
             "error-level event must carry the SFTP READDIR failed message, captured: {captured}"
         );
     }
-
-    #[tokio::test]
-    async fn mkdir_subdir_puts_trailing_slash_client_key() {
-        let backend = Arc::new(DummyBackend::new());
-        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
-        let authorized = Arc::new(std::sync::Mutex::new(Vec::new()));
-        let record = Arc::clone(&authorized);
-
-        let status = with_test_auth_override(
-            move |action, bucket, key| {
-                record
-                    .lock()
-                    .expect("auth log")
-                    .push((action.as_str(), bucket.to_string(), key.map(str::to_string)));
-                true
-            },
-            driver.mkdir(1, "/bucket/photos/2024".to_string(), russh_sftp::protocol::FileAttributes::default()),
-        )
-        .await
-        .expect("mkdir of an ordinary directory");
-
-        assert!(matches!(status.status_code, StatusCode::Ok));
-        let calls = backend.put_object_calls();
-        assert_eq!(calls.len(), 1, "mkdir must issue one PutObject");
-        assert_eq!(calls[0].bucket, "bucket");
-        assert_eq!(calls[0].key, "photos/2024/");
-        assert!(
-            !calls[0].key.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX),
-            "the client key must not carry the reserved directory-marker suffix"
-        );
-        let auth = authorized.lock().expect("auth log");
-        assert_eq!(
-            auth.as_slice(),
-            [("s3:PutObject", "bucket".to_string(), Some("photos/2024/".to_string()))]
-        );
-    }
-
-    #[tokio::test]
-    async fn rmdir_subdir_deletes_trailing_slash_client_key() {
-        let backend = Arc::new(DummyBackend::new());
-        backend.queue_delete_object_ok();
-        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
-        let authorized = Arc::new(std::sync::Mutex::new(Vec::new()));
-        let record = Arc::clone(&authorized);
-
-        let status = with_test_auth_override(
-            move |action, bucket, key| {
-                record
-                    .lock()
-                    .expect("auth log")
-                    .push((action.as_str(), bucket.to_string(), key.map(str::to_string)));
-                true
-            },
-            driver.rmdir(2, "/bucket/photos/2024".to_string()),
-        )
-        .await
-        .expect("rmdir of an empty directory");
-
-        assert!(matches!(status.status_code, StatusCode::Ok));
-        let calls = backend.delete_object_calls();
-        assert_eq!(calls.len(), 1, "rmdir must issue one DeleteObject");
-        assert_eq!(calls[0].bucket, "bucket");
-        assert_eq!(calls[0].key, "photos/2024/");
-        assert!(
-            !calls[0].key.contains(rustfs_utils::path::GLOBAL_DIR_SUFFIX),
-            "the client key must not carry the reserved directory-marker suffix"
-        );
-        let auth = authorized.lock().expect("auth log");
-        assert_eq!(
-            auth.as_slice(),
-            [
-                ("s3:ListBucket", "bucket".to_string(), Some("photos/2024/".to_string())),
-                ("s3:DeleteObject", "bucket".to_string(), Some("photos/2024/".to_string())),
-            ]
-        );
-    }
-
-    #[tokio::test]
-    async fn mkdir_rejects_reserved_directory_marker_before_put() {
-        let backend = Arc::new(DummyBackend::new());
-        let mut driver = build_driver(Arc::clone(&backend), TEST_PART_SIZE);
-        let path = format!("/bucket/sub{}", rustfs_utils::path::GLOBAL_DIR_SUFFIX);
-
-        let err = driver
-            .mkdir(3, path, russh_sftp::protocol::FileAttributes::default())
-            .await
-            .expect_err("a client path containing the reserved marker must be rejected");
-
-        assert_eq!(StatusCode::from(err), StatusCode::BadMessage);
-        assert!(backend.put_object_calls().is_empty(), "rejection must happen before PutObject");
-    }
 }

crates/protocols/src/sftp/driver.rs

@@ -831,10 +831,14 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
 
     /// SSH_FXP_MKDIR, SFTP Internet Draft section 6.6. Bucket-level path
     /// (only the bucket component is set) issues CreateBucket. Sub-bucket
-    /// path issues PutObject of a zero-byte object at the logical prefix
-    /// plus a trailing slash. Storage encodes that client key into the
-    /// directory marker. MKDIR at the SFTP root returns Failure because
-    /// there is no parent into which a new top-level entity could be added.
+    /// path issues PutObject of a zero-byte object at the encoded
+    /// directory marker key. MKDIR at the SFTP root returns Failure
+    /// because there is no parent into which a new top-level entity could
+    /// be added.
+    ///
+    /// The directory-marker key is built with rustfs_utils::path::
+    /// encode_dir_object so the key format matches the convention used
+    /// by the rest of RustFS (S3, Swift, WebDAV).
     #[tracing::instrument(level = "info", skip(self, _attrs), fields(id, path = %sanitise_control_bytes(&path)), err(Debug))]
     async fn mkdir(&mut self, id: u32, path: String, _attrs: FileAttributes) -> Result<Status, Self::Error> {
         self.enforce_server_readonly()?;
@@ -854,9 +858,8 @@ impl<S: StorageBackend + Send + Sync + 'static> russh_sftp::server::Handler for
     /// SSH_FXP_RMDIR, SFTP Internet Draft section 6.6. Empty check then
     /// delete. Bucket-level path lists the bucket with max_keys=1 and,
     /// on an empty result, calls DeleteBucket. Sub-bucket path lists the
-    /// prefix and, on an empty result, calls DeleteObject on the
-    /// trailing-slash directory key. Storage encodes that key to the
-    /// on-disk marker.
+    /// prefix and, on an empty result, calls DeleteObject on the encoded
+    /// directory marker.
     ///
     /// validate_directory_empty propagates the list error rather than
     /// swallowing it. Without that, a transient backend error during

crates/scanner/src/data_usage_define.rs

@@ -1823,12 +1823,7 @@ impl DataUsageCache {
             return;
         };
 
-        // Wide bucket roots previously only flattened above a hardcoded 250k
-        // fanout, so caches with 10k–250k direct children (common with many
-        // top-level prefixes) never shrank and retained one entry per prefix
-        // after #8971's non-root compaction. Honor `limit` here so completed
-        // sweeps and emergency compaction actually bound root fanout (#8961).
-        if top_e.children.len() > limit {
+        if top_e.children.len() > 250_000 {
             self.reduce_children_of(&hash_path(&self.info.name), limit, true);
         }
         if self.cache.len() <= limit {

crates/scanner/src/data_usage_define/tests.rs

@@ -3223,200 +3223,6 @@ fn test_reduce_children_of_usize_underflow_saturates() {
     assert!(cache.find("bucket/large/b").is_some());
 }
 
-#[test]
-fn test_force_compact_flattens_wide_root_above_limit() {
-    // Regression for #8961: a bucket root with fanout between the ordinary
-    // compact limit and the old hardcoded 250_000 ceiling must shrink.
-    // Before the fix, force_compact kept one entry per prefix forever.
-    const PREFIXES: usize = 64;
-    const LIMIT: usize = 16;
-    const {
-        assert!(PREFIXES > LIMIT);
-    }
-
-    let root = hash_path("bucket");
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: "bucket".to_string(),
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    cache.replace_hashed(&root, &None, &DataUsageEntry::default());
-    for index in 0..PREFIXES {
-        let child = hash_path(&format!("bucket/prefix-{index:04}"));
-        cache.replace_hashed(
-            &child,
-            &Some(root.clone()),
-            &DataUsageEntry {
-                objects: 1,
-                size: 1,
-                compacted: true,
-                ..Default::default()
-            },
-        );
-    }
-    assert_eq!(cache.find("bucket").map(|entry| entry.children.len()), Some(PREFIXES));
-    assert_eq!(cache.cache.len(), PREFIXES + 1);
-
-    cache.force_compact(LIMIT);
-
-    let root_entry = cache.find("bucket").expect("root retained");
-    assert!(root_entry.compacted, "wide root must flatten under force_compact");
-    assert!(root_entry.children.is_empty(), "flattened root must drop per-prefix children");
-    assert_eq!((root_entry.objects, root_entry.size), (PREFIXES, PREFIXES));
-    assert_eq!(cache.cache.len(), 1, "prefix leaves must be released after root flatten");
-}
-
-#[test]
-fn test_force_compact_keeps_narrow_root_children() {
-    const PREFIXES: usize = 4;
-    const LIMIT: usize = 16;
-    const {
-        assert!(PREFIXES < LIMIT);
-    }
-
-    let root = hash_path("bucket");
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: "bucket".to_string(),
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    cache.replace_hashed(
-        &root,
-        &None,
-        &DataUsageEntry {
-            objects: PREFIXES,
-            ..Default::default()
-        },
-    );
-    for index in 0..PREFIXES {
-        let child = hash_path(&format!("bucket/prefix-{index:04}"));
-        cache.replace_hashed(
-            &child,
-            &Some(root.clone()),
-            &DataUsageEntry {
-                objects: 1,
-                compacted: true,
-                ..Default::default()
-            },
-        );
-    }
-
-    cache.force_compact(LIMIT);
-
-    let root_entry = cache.find("bucket").expect("root retained");
-    assert!(!root_entry.compacted);
-    assert_eq!(root_entry.children.len(), PREFIXES);
-    assert_eq!(cache.cache.len(), PREFIXES + 1);
-}
-
-#[test]
-fn test_force_compact_interrupted_ceiling_retains_wide_root_children() {
-    // Interrupted sweeps pass the emergency FORCE ceiling into force_compact so
-    // a resume frontier under a 10k–250k-wide root stays addressable.
-    const PREFIXES: usize = 64;
-    const ORDINARY_LIMIT: usize = 16;
-    const INTERRUPTED_CEILING: usize = 250_000;
-    const {
-        assert!(PREFIXES > ORDINARY_LIMIT);
-        assert!(PREFIXES < INTERRUPTED_CEILING);
-    }
-
-    let root = hash_path("bucket");
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: "bucket".to_string(),
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    cache.replace_hashed(&root, &None, &DataUsageEntry::default());
-    for index in 0..PREFIXES {
-        let child = hash_path(&format!("bucket/prefix-{index:04}"));
-        cache.replace_hashed(
-            &child,
-            &Some(root.clone()),
-            &DataUsageEntry {
-                objects: 1,
-                size: 1,
-                compacted: true,
-                ..Default::default()
-            },
-        );
-    }
-
-    cache.force_compact(INTERRUPTED_CEILING);
-
-    let root_entry = cache.find("bucket").expect("root retained");
-    assert!(!root_entry.compacted, "interrupted ceiling must not flatten a 64-wide root");
-    assert_eq!(root_entry.children.len(), PREFIXES);
-    assert_eq!(cache.cache.len(), PREFIXES + 1);
-    // Sanity: the ordinary completion limit would have flattened this layout.
-    let mut completed = cache.clone();
-    completed.force_compact(ORDINARY_LIMIT);
-    assert!(
-        completed.find("bucket").is_some_and(|entry| entry.compacted),
-        "ordinary limit still flattens the same wide root after completion"
-    );
-}
-
-#[test]
-fn test_force_compact_interrupted_ceiling_seals_reloaded_frontier() {
-    // The interrupted call runs before seal_scan_frontier. Flattening at the
-    // ordinary limit deletes the child that frontier names, so a saved cache
-    // reloads with no resume position and the next cycle replays finished prefixes.
-    const PREFIXES: usize = 64;
-    const ORDINARY_LIMIT: usize = 16;
-    const INTERRUPTED_CEILING: usize = 250_000;
-    const {
-        assert!(PREFIXES > ORDINARY_LIMIT);
-        assert!(PREFIXES < INTERRUPTED_CEILING);
-    }
-    let frontier = format!("bucket/prefix-{:04}", PREFIXES / 2);
-
-    let mut cache = DataUsageCache::default();
-    cache.prepare_bucket_checkpoint("bucket", 8, 1, DataUsageCacheSource::new(0, 0), TEST_PLAN_DIGEST, valid_scan_identity());
-    let root = hash_path("bucket");
-    cache.replace_hashed(&root, &None, &DataUsageEntry::default());
-    for index in 0..PREFIXES {
-        let child = hash_path(&format!("bucket/prefix-{index:04}"));
-        cache.replace_hashed(
-            &child,
-            &Some(root.clone()),
-            &DataUsageEntry {
-                objects: 1,
-                size: 1,
-                compacted: true,
-                ..Default::default()
-            },
-        );
-    }
-
-    let mut interrupted = cache.clone();
-    interrupted.force_compact(INTERRUPTED_CEILING);
-    assert!(interrupted.find(&frontier).is_some(), "interrupted ceiling must keep the frontier child");
-    interrupted
-        .seal_scan_frontier(Some(&frontier))
-        .expect("seal frontier after interrupted compaction");
-    let encoded = rmp_serde::to_vec(&interrupted).expect("partial cache should encode");
-    let reloaded: DataUsageCache = rmp_serde::from_slice(&encoded).expect("partial cache should reload");
-    assert_eq!(reloaded.validated_scan_frontier(), Some(frontier.as_str()));
-    assert!(reloaded.find(&frontier).is_some(), "reloaded frontier child must stay addressable");
-
-    cache.force_compact(ORDINARY_LIMIT);
-    assert!(cache.find(&frontier).is_none(), "ordinary limit removes the frontier child");
-    cache
-        .seal_scan_frontier(Some(&frontier))
-        .expect("seal after a completed flatten");
-    assert!(
-        cache.validated_scan_frontier().is_none(),
-        "a flattened root cannot seal the removed frontier child"
-    );
-}
-
 #[tokio::test]
 async fn test_retry_save_op_times_out_and_returns_error_after_retries() {
     let attempts = Arc::new(AtomicUsize::new(0));

crates/scanner/src/scanner_folder.rs

@@ -433,7 +433,6 @@ enum FolderScanOutcome {
     Namespace,
     ErasureData,
     Vanished,
-    Unreadable,
 }
 
 fn is_erasure_candidate(name: &str) -> bool {
@@ -524,18 +523,6 @@ fn data_usage_entry_has_progress(entry: &DataUsageEntry) -> bool {
     data_usage_root_has_progress(entry)
 }
 
-/// Root-fanout limit for `DataUsageCache::force_compact` at the end of a sweep.
-///
-/// An interrupted sweep still has to seal its resume frontier, so it keeps the
-/// emergency ceiling. A finished walk applies the ordinary child limit.
-fn force_compact_limit_for_sweep(interrupted: bool) -> usize {
-    if interrupted {
-        DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS
-    } else {
-        DATA_SCANNER_COMPACT_AT_CHILDREN
-    }
-}
-
 fn set_scan_checkpoint(cache: &mut DataUsageCache, reason: DataUsageScanCheckpointReason) {
     let resume_after = clone_cache_value(&cache.info.scan_resume_after).or_else(|| {
         cache
@@ -1428,7 +1415,7 @@ impl FolderScanner {
         mode: FolderScanMode,
     ) -> Result<FolderScanOutcome, ScannerError> {
         if self.checkpoint_tx.is_none() || self.old_cache.info.scan_progress.is_none() {
-            return Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode, false)).await;
+            return Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).await;
         }
         let depth = self.checkpoint_depth;
         if depth >= MAX_DATA_USAGE_CACHE_DEPTH {
@@ -1442,7 +1429,7 @@ impl FolderScanner {
         std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
         std::mem::swap(parent_entry, &mut self.checkpoint_ancestors[depth].entry);
         self.checkpoint_depth += 1;
-        let result = Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode, false)).await;
+        let result = Box::pin(self.scan_folder_inner(ctx, folder, child_entry, mode)).await;
         self.checkpoint_depth -= 1;
         // Restore the parent before cancellation/error handling stitches partial work.
         std::mem::swap(parent_hash, &mut self.checkpoint_ancestors[depth].hash);
@@ -1863,7 +1850,7 @@ impl FolderScanner {
         folder: CachedFolder,
         into: &mut DataUsageEntry,
     ) -> Result<(), ScannerError> {
-        self.scan_folder_inner(ctx, folder, into, FolderScanMode::Namespace, false)
+        self.scan_folder_inner(ctx, folder, into, FolderScanMode::Namespace)
             .await
             .map(|_| ())
     }
@@ -1876,7 +1863,6 @@ impl FolderScanner {
         folder: CachedFolder,
         into: &mut DataUsageEntry,
         mode: FolderScanMode,
-        skip_unreadable_directory: bool,
     ) -> Result<FolderScanOutcome, ScannerError> {
         let done_folder = Metrics::time(Metric::ScanFolder);
         let trace_started_at = trace_start_instant();
@@ -1999,20 +1985,6 @@ impl FolderScanner {
                     }
                     None
                 }
-                Err(e) if skip_unreadable_directory && !ctx.is_cancelled() && e.kind() == ErrorKind::PermissionDenied => {
-                    warn!(
-                        target: "rustfs::scanner::folder",
-                        event = EVENT_SCANNER_FOLDER_STATE,
-                        component = LOG_COMPONENT_SCANNER,
-                        subsystem = LOG_SUBSYSTEM_FOLDER,
-                        state = "child_scan_failed",
-                        folder = %folder.name,
-                        error = %e,
-                        "Scanner skipped an unreadable child directory"
-                    );
-                    done_folder();
-                    return Ok(FolderScanOutcome::Unreadable);
-                }
                 Err(e) => return Err(ScannerError::Io(e)),
             };
             let mut pending_entry_progress = 0_u64;
@@ -2561,23 +2533,13 @@ impl FolderScanner {
             }
 
             // Checkpointed sweeps need child entries to validate their resume
-            // frontier (#8292). Outside that, also compact a wide bucket root:
-            // excluding it left #8961's ~170k top-level prefixes unbounded until
-            // the emergency 250k ceiling. Keep FORCE as the absolute backstop.
-            let is_scan_root = folder.name == self.old_cache.info.name;
-            let forward_sweep = self.old_cache.info.scan_progress.is_some();
-            let total_folders = queued_folders.len();
-            let force_compact = total_folders >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS;
-            // Reused root children still have links in the new cache.
-            let should_compact = force_compact
-                || (!forward_sweep
-                    && total_folders >= DATA_SCANNER_COMPACT_AT_FOLDERS
-                    && (!is_scan_root || into.children.is_empty()));
-            // Preserve the ordinary namespace walk's child isolation when the
-            // root starts sharing one accumulator across its descendants.
-            let skip_unreadable_children =
-                skip_unreadable_directory || (is_scan_root && !into.compacted && should_compact && !force_compact);
+            // frontier. Keep the emergency compaction limit for large trees.
+            let should_compact = (self.new_cache.info.name != folder.name
+                && self.old_cache.info.scan_progress.is_none()
+                && queued_folders.len() >= DATA_SCANNER_COMPACT_AT_FOLDERS)
+                || queued_folders.len() >= DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS;
 
+            let total_folders = queued_folders.len();
             self.alert_excessive_folders(&folder.name, total_folders);
 
             if !into.compacted && should_compact {
@@ -2600,6 +2562,7 @@ impl FolderScanner {
                 }
             }
 
+            let is_scan_root = folder.name == self.old_cache.info.name;
             let scan_checkpoint = self.old_cache.info.scan_checkpoint.as_ref();
             let checkpoint_resume_after = scan_checkpoint.and_then(|checkpoint| {
                 if is_scan_root {
@@ -2623,6 +2586,7 @@ impl FolderScanner {
                 .is_some_and(|resume_match| matches!(resume_match, FolderResumeMatch::Descendant));
             let scan_resume_after = checkpoint_resume_after.or(self.old_cache.info.scan_resume_after.as_deref());
             let has_queued_folders = !queued_folders.is_empty();
+            let forward_sweep = self.old_cache.info.scan_progress.is_some();
             let resume_order = if forward_sweep {
                 order_queued_folders_for_resume(&mut queued_folders, None)
             } else {
@@ -2726,14 +2690,7 @@ impl FolderScanner {
                 (self.update_current_path)(&folder_item.name).await;
 
                 let outcome = if into.compacted {
-                    Box::pin(self.scan_folder_inner(
-                        ctx.clone(),
-                        folder_item.clone(),
-                        into,
-                        child_mode,
-                        skip_unreadable_children && child_mode == FolderScanMode::Namespace,
-                    ))
-                    .await?
+                    Box::pin(self.scan_folder_inner(ctx.clone(), folder_item.clone(), into, child_mode)).await?
                 } else {
                     let mut dst = DataUsageEntry::default();
                     match self
@@ -2768,9 +2725,6 @@ impl FolderScanner {
                     }
                 };
                 self.finish_raw_enumeration_parent(&folder_item.name);
-                if outcome == FolderScanOutcome::Unreadable {
-                    continue;
-                }
                 if outcome == FolderScanOutcome::Vanished {
                     self.new_cache.delete_recursive(&h);
                     if let Some(updates) = &mut self.usage_updates {
@@ -3250,15 +3204,13 @@ impl FolderScanner {
         self.compact_completed_folder(&folder, &this_hash, into);
 
         // Compact if too many children...
-        // Allow compact_self on the bucket root when this is not a checkpointed
-        // forward sweep so wide roots can flatten instead of retaining one
-        // entry per top-level prefix (#8961). Checkpointed sweeps keep the
-        // previous root exclusion so resume frontiers stay addressable (#8292).
         if !into.compacted {
             let done_compact = Metrics::time(Metric::CompactFolder);
-            let compact_self = self.new_cache.info.name != folder.name || self.old_cache.info.scan_progress.is_none();
-            self.new_cache
-                .reduce_children_of(&this_hash, DATA_SCANNER_COMPACT_AT_CHILDREN, compact_self);
+            self.new_cache.reduce_children_of(
+                &this_hash,
+                DATA_SCANNER_COMPACT_AT_CHILDREN,
+                self.new_cache.info.name != folder.name,
+            );
             done_compact();
         }
 
@@ -3463,7 +3415,7 @@ pub(crate) async fn scan_data_folder_scoped(
             let coverage_gap = scanner.coverage_gap;
             scanner.reconcile_failed_objects_after_full_scan();
             let new_cache = scanner.as_mut_new_cache();
-            new_cache.force_compact(force_compact_limit_for_sweep(false));
+            new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
             new_cache.info.last_update = Some(SystemTime::now());
             new_cache.info.next_cycle = next_cycle;
             let unresolved_objects = coverage_gap
@@ -3564,13 +3516,7 @@ pub(crate) async fn scan_data_folder_scoped(
                     || !new_cache.info.size_reconciliation.is_empty()
                 {
                     if new_cache.root().is_some() {
-                        // Interrupted sweeps seal a resume frontier after this.
-                        // Flattening the root at the ordinary 10k limit would
-                        // drop frontier-named children and force the next cycle
-                        // to replay completed prefixes (#8292 / review on #9193).
-                        // Keep only the emergency FORCE ceiling here; successful
-                        // completions above apply DATA_SCANNER_COMPACT_AT_CHILDREN.
-                        new_cache.force_compact(force_compact_limit_for_sweep(true));
+                        new_cache.force_compact(DATA_SCANNER_COMPACT_AT_CHILDREN);
                     }
                     new_cache.info.last_update = Some(SystemTime::now());
                     new_cache.info.next_cycle = next_cycle;

crates/scanner/src/scanner_folder/tests.rs

@@ -4475,47 +4475,3 @@ fn raw_enumeration_progress_retains_resume_index_until_unordered_entries_reappea
         "same source identity should keep the resume index even when read_dir order changes"
     );
 }
-
-#[test]
-fn force_compact_limit_for_interrupted_sweep_keeps_frontier_above_ordinary_limit() {
-    const PREFIXES: usize = DATA_SCANNER_COMPACT_AT_CHILDREN + 1;
-    const {
-        assert!(PREFIXES > DATA_SCANNER_COMPACT_AT_CHILDREN);
-        assert!(PREFIXES < DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS);
-    }
-
-    let frontier = format!("bucket/prefix-{:05}", PREFIXES / 2);
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: "bucket".to_string(),
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    let root = hash_path("bucket");
-    cache.replace_hashed(&root, &None, &DataUsageEntry::default());
-    for index in 0..PREFIXES {
-        let child = hash_path(&format!("bucket/prefix-{index:05}"));
-        cache.replace_hashed(
-            &child,
-            &Some(root.clone()),
-            &DataUsageEntry {
-                objects: 1,
-                size: 1,
-                compacted: true,
-                ..Default::default()
-            },
-        );
-    }
-
-    let mut completed = cache.clone();
-    completed.force_compact(force_compact_limit_for_sweep(false));
-    assert!(completed.find("bucket").is_some_and(|entry| entry.compacted));
-    assert!(completed.find(&frontier).is_none());
-
-    cache.force_compact(force_compact_limit_for_sweep(true));
-    assert!(cache.find(&frontier).is_some(), "interrupted ceiling must keep the frontier child");
-    let root_entry = cache.find("bucket").expect("root retained");
-    assert!(!root_entry.compacted);
-    assert_eq!(root_entry.children.len(), PREFIXES);
-}

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/deep_compacted.rs

@@ -293,60 +293,3 @@ async fn check_large_prefix_resumption(mode: HealScanMode) {
     }
     panic!("large prefix must finish within repeated bounded scans");
 }
-
-#[cfg(unix)]
-#[tokio::test]
-#[serial]
-async fn wide_root_legacy_permission_failure_isolated_to_child() {
-    use std::os::unix::fs::PermissionsExt;
-
-    assert!(!crate::scanner_is_erasure().await, "single-drive Namespace scan");
-    let (scanner, root) = build_test_scanner().await;
-    let _cleanup = TestGuard {
-        temp_dir: Some(root.clone()),
-    };
-    write_checkpoint_object(&root, "healthy/object", &[(None, 7)]).await;
-    let bucket = root.join("bucket");
-    let denied = bucket.join("denied");
-    std::fs::create_dir(&denied).expect("create denied root child");
-    for idx in 0..DATA_SCANNER_COMPACT_AT_FOLDERS - 2 {
-        std::fs::create_dir(bucket.join(format!("empty-{idx:05}"))).expect("create wide directory");
-    }
-    assert_eq!(std::fs::read_dir(&bucket).expect("list root").count(), DATA_SCANNER_COMPACT_AT_FOLDERS);
-    std::fs::set_permissions(&denied, std::fs::Permissions::from_mode(0o000)).expect("remove child read access");
-
-    match std::fs::read_dir(&denied) {
-        Err(error) => assert_eq!(error.kind(), std::io::ErrorKind::PermissionDenied),
-        Ok(_) => {
-            std::fs::set_permissions(denied, std::fs::Permissions::from_mode(0o700)).expect("restore child access");
-            eprintln!("skipping permission-denied fixture because this process can read mode-000 directories");
-            return;
-        }
-    }
-    let mut cache = DataUsageCache::default();
-    cache.prepare_for_scan("bucket", 11, 7, SOURCE, PLAN, false);
-    assert!(cache.info.scan_progress.is_none());
-    cache.info.skip_healing = true;
-    let budget = ScannerCycleBudget::new(&CancellationToken::new(), Default::default());
-    let wide = scanner
-        .local_disk
-        .clone()
-        .nsscanner_disk(
-            budget.token(),
-            budget,
-            vec![scanner.local_disk.clone()],
-            cache,
-            None,
-            super::scan_options(HealScanMode::Normal),
-        )
-        .await;
-    std::fs::set_permissions(denied, std::fs::Permissions::from_mode(0o700)).expect("restore child access");
-    let ScannerDiskScanOutcome::Complete(wide) = wide.expect("wide legacy scanner must keep the same child-error isolation")
-    else {
-        panic!("wide root should complete despite one unreadable prefix");
-    };
-    let summary = wide.checked_flatten("bucket").expect("aggregate healthy child");
-    assert!(summary.compacted);
-    assert_eq!((summary.objects, summary.size), (1, 7));
-    assert_eq!(wide.cache.len(), 1);
-}

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

@@ -205,7 +205,6 @@ async fn erasure_directory_probe_reaches_eof_across_budgeted_batches() {
                 folder,
                 &mut DataUsageEntry::default(),
                 FolderScanMode::ErasureCandidate { reuse_namespace: false },
-                false,
             )
             .await
             .expect("classify full UUID namespace");
@@ -272,7 +271,6 @@ async fn scanner_directory_probe_budget_wait_cancels_without_rate_delay() {
             folder,
             &mut DataUsageEntry::default(),
             FolderScanMode::ErasureCandidate { reuse_namespace: false },
-            false,
         )
         .await;
     task.await.expect("open-wait cancellation");
@@ -312,7 +310,6 @@ async fn scanner_directory_probe_cancels_during_entry_budget_wait() {
             folder,
             &mut DataUsageEntry::default(),
             FolderScanMode::ErasureCandidate { reuse_namespace: false },
-            false,
         )
         .await;
     assert!(ctx.is_cancelled());
@@ -382,7 +379,7 @@ async fn erasure_candidate_defers_metadata_until_part_classification_finishes()
     scanner.budget = ScannerCycleBudget::new_with_progress_tracking(&ctx, Default::default());
     let mut into = DataUsageEntry::default();
     let outcome = scanner
-        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false }, false)
+        .scan_folder_inner(ctx, folder, &mut into, FolderScanMode::ErasureCandidate { reuse_namespace: false })
         .await
         .expect("classify candidate");
     assert_eq!(outcome, FolderScanOutcome::ErasureData);
@@ -595,7 +592,6 @@ async fn erasure_candidate_part_symlink_is_not_data_evidence() {
             folder,
             &mut DataUsageEntry::default(),
             FolderScanMode::ErasureCandidate { reuse_namespace: false },
-            false,
         )
         .await
         .expect("classify symlink namespace");

docs/operations/rolling-restart.md

@@ -22,16 +22,6 @@ Each row is a feature whose activation makes older binaries unable to read what
 - Rolling restart (no downtime): restart one node at a time, wait for the restarted node to report `200` on `/health/ready`, and then wait until the other nodes no longer list its drives under `details.storage.unavailableDrives` before touching the next one. The remaining nodes keep serving traffic.
 - Sequential cold start (several nodes down): nodes started before the cluster has quorum come up in degraded mode. The process stays alive, answers `503` with the blocking reason, and recovers automatically as soon as enough peers are online. Do not restart-loop them; keep starting the remaining nodes.
 
-## Returning a recovered node to a load balancer
-
-Probe each backend directly on its S3 listener, using its configured TLS verification when HTTPS is enabled. Require HTTP `200` from `/health/ready` (or `/minio/health/ready`) before routing object requests to that backend. For a workload that also requires write availability, additionally require HTTP `200` from `/minio/health/cluster` on that backend. A read-only workload can use node readiness alone; `/minio/health/cluster/read` has a more conservative lock check and adds no startup-publication guarantee. See the [probe semantics](../architecture/readiness-matrix.md#probe-semantics).
-
-Cluster write/read probes can return `200` while a restarted node still returns `503` for object requests: quorum observations can succeed before startup publishes `FullReady`, including while startup waits for peer re-admission. The node readiness aliases remain `503` during that interval, with `startup_finalization_pending` in `degradedReasons`. Liveness and cluster probes alone are therefore insufficient for backend admission. This also applies when every node's cluster probe succeeds.
-
-Readiness does not certify that healing has completed for every object. The startup peer re-admission wait is bounded and fails open; before stopping another node, also follow the survivor checks in the [rolling restart procedure](#rolling-restart-procedure). No health endpoint is an atomic guarantee that every subsequent object read or write will succeed. The [storage detail contract](../architecture/readiness-matrix.md#storage-detail-contract) describes these sampling limits.
-
-Clients should use bounded retries with backoff and jitter for transient `503` responses. Honor `Retry-After` when present, within the request's retry/deadline budget; it is a suggested delay, not a recovery deadline. The startup admission gate sends `Retry-After: 5` and `x-rustfs-readiness-pending` naming the pending dependency. Other `503` paths can use different headers, so inspect the actual response. Retrying a GET is safe; retrying a write with an ambiguous outcome requires accounting for versioning, concurrent writers, and request preconditions. For recovery validation, retain the first failed response and use distinct payloads or version IDs for overwrites so a successful retry cannot hide a failed assertion or identify the wrong version as durable.
-
 ## Why a single node cannot serve alone
 
 Erasure coding shards every object, including internal metadata such as IAM users, groups, and policies under `.rustfs.sys`, across the drives of a set. Reading an object back needs a read quorum of shards online. With the drives of one set spread over several nodes, one node alone can never satisfy the read quorum; this is a property of erasure coding, not a bug. The cluster becomes readable once enough nodes are up (for internal configuration objects, written with maximum parity, typically about half the nodes of a set).
@@ -69,10 +59,6 @@ On Kubernetes the dial itself is the slow part. A replaced pod gets a new IP, bu
 
 After a distributed node reaches local readiness, it asks the peers in every erasure set it serves how they currently see its drives. It uses the `PeerDriveAdmission` internode RPC, and each peer answers with the same predicate its own `/health/ready` uses for `unavailableDrives`. Startup readiness is published once a strict majority of each set's peers count all of this node's drives in that set as online. In a three-node set that means both peers. While the node waits, `/health/ready` stays `503` with `startup_finalization_pending` and the log shows `event="startup_peer_readmission" state="waiting"`.
 
-Both local dependency checks and peer re-admission rounds are bounded by the startup budget, including in-flight observations and polling delays. An exhausted peer-readmission budget starts no new probe round. Within a round, startup proceeds as soon as every shared set has its required peer majority; pending surplus read-only RPCs are cancelled. Votes are never carried across rounds. The unreachable-peer grace also bounds in-flight rounds: enough reachable replies clear an earlier shortfall before unrelated peers finish, and only a completed round can establish a new unreachable shortfall.
-
-After deferred IAM initialization succeeds, recovery retains that IAM owner while preparing the topology and claiming the application context once. A later local-readiness timeout retries publication with the prepared context; it does not reinitialize IAM or reinstall the context.
-
 The wait always ends in publication:
 
 - `result="admitted"`: the peers re-admitted the node.
@@ -107,7 +93,7 @@ When the whole cluster (or several nodes) went down and nodes are brought back o
 
 | Variable | Default | Effect |
 | --- | --- | --- |
-| `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` | `120` (`DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS`) | Budget per readiness-publication attempt for local readiness and subsequent peer re-admission, including in-flight checks and polling delays. Local timeout returns an error to the startup/recovery caller; peer re-admission timeout publishes readiness with a warning. Raising it can delay S3 admission; the HTTP listeners are already bound. |
+| `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` | `120` (`DEFAULT_STARTUP_READINESS_MAX_WAIT_SECS`) | How long startup waits for full readiness before continuing in degraded mode with background recovery. Raising it delays the listener during genuinely slow starts; lowering it surfaces degraded mode sooner. Recovery retries continue regardless of this limit. |
 | `RUSTFS_STARTUP_PEER_READMISSION_ENABLE` | `true` (`DEFAULT_STARTUP_PEER_READMISSION_ENABLE`) | Hold startup readiness until a strict majority of each erasure set's peers count this node's drives online. The wait is bounded by `RUSTFS_STARTUP_READINESS_MAX_WAIT_SECS` and fails open. Set `false` to publish on this node's own view only. |
 | `RUSTFS_DRIVE_RETURNING_PROBE_INTERVAL_SECS` | `2` (`DEFAULT_DRIVE_RETURNING_PROBE_INTERVAL_SECS`) | Interval between recovery probes to a remote drive that is `offline` or `returning`. |
 | `RUSTFS_DRIVE_RETURNING_SUCCESS_THRESHOLD` | `3` (`DEFAULT_DRIVE_RETURNING_SUCCESS_THRESHOLD`) | Consecutive successful probes before an `offline` remote drive is `online` again. Together with the interval this bounds how long survivors keep a restarted peer out after they can dial it. |

rustfs/src/server/layer.rs

@@ -4574,112 +4574,66 @@ mod tests {
             async {
                 let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO));
                 let readiness = Arc::new(GlobalReadiness::new());
-                readiness.mark_stage(rustfs_common::SystemStage::IamReady);
                 let inner = CountingHybridService::default();
                 let calls = inner.calls();
-                let public_health =
-                    public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness)).await;
-                let mut service =
-                    crate::server::ReadinessGateLayer::new(Arc::clone(&readiness)).layer(public_health.layer(inner));
+                let mut service = public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness))
+                    .await
+                    .layer(inner);
 
-                // Cluster quorum can recover before this node publishes S3 admission.
-                for method in [Method::GET, Method::HEAD] {
-                    for path in [
-                        MINIO_HEALTH_CLUSTER_PATH,
-                        MINIO_HEALTH_CLUSTER_READ_PATH,
-                        HEALTH_COMPAT_LIVE_PATH,
-                    ] {
-                        let response = service
-                            .call(
-                                Request::builder()
-                                    .method(method.clone())
-                                    .uri(path)
-                                    .body(Full::<Bytes>::from(Bytes::new()))
-                                    .expect("probe request before admission publication"),
-                            )
-                            .await
-                            .expect("probe response before admission publication");
-                        assert_eq!(response.status(), StatusCode::OK, "{method} {path}");
-                        let body = BodyExt::collect(response.into_body()).await.expect("probe body").to_bytes();
-                        if method == Method::HEAD {
-                            assert!(body.is_empty(), "HEAD {path}");
-                        } else {
-                            let payload: serde_json::Value = serde_json::from_slice(&body).expect("probe JSON");
-                            if path == HEALTH_COMPAT_LIVE_PATH {
-                                assert_eq!(payload["status"], "ok");
-                                assert!(payload.get("ready").is_none());
-                            } else {
-                                assert_eq!(payload["ready"], true, "{path}");
-                            }
-                        }
-                    }
-                }
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request before admission publication"),
+                    )
+                    .await
+                    .expect("readiness response before admission publication");
+                assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("readiness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
+                assert_eq!(payload["ready"], false);
+                assert_eq!(payload["details"]["storage"]["ready"], true);
+                assert_eq!(payload["details"]["iam"]["ready"], true);
+                assert_eq!(payload["details"]["lock"]["ready"], true);
+                assert_eq!(payload["degradedReasons"], serde_json::json!(["startup_finalization_pending"]));
 
-                for published in [false, true] {
-                    if published {
-                        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
-                    }
-                    for method in [Method::GET, Method::HEAD] {
-                        for path in [HEALTH_READY_PATH, MINIO_HEALTH_READY_PATH] {
-                            let response = service
-                                .call(
-                                    Request::builder()
-                                        .method(method.clone())
-                                        .uri(path)
-                                        .body(Full::<Bytes>::from(Bytes::new()))
-                                        .expect("node readiness request"),
-                                )
-                                .await
-                                .expect("node readiness response");
-                            let expected = if published {
-                                StatusCode::OK
-                            } else {
-                                StatusCode::SERVICE_UNAVAILABLE
-                            };
-                            assert_eq!(response.status(), expected, "{method} {path}, published={published}");
-                            let body = BodyExt::collect(response.into_body())
-                                .await
-                                .expect("readiness body")
-                                .to_bytes();
-                            if method == Method::HEAD {
-                                assert!(body.is_empty(), "HEAD {path}");
-                            } else {
-                                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
-                                assert_eq!(payload["ready"], published);
-                                for component in ["storage", "iam", "lock"] {
-                                    assert_eq!(payload["details"][component]["ready"], true, "{component}");
-                                }
-                                let reasons = if published {
-                                    serde_json::json!([])
-                                } else {
-                                    serde_json::json!(["startup_finalization_pending"])
-                                };
-                                assert_eq!(payload["degradedReasons"], reasons);
-                            }
-                        }
-                    }
-                    assert_eq!(calls.load(Ordering::SeqCst), 0, "health probes bypass the inner service");
-                    for method in [Method::GET, Method::PUT] {
-                        let response = service
-                            .call(
-                                Request::builder()
-                                    .method(method.clone())
-                                    .uri("/bucket/object")
-                                    .body(Full::<Bytes>::from(Bytes::new()))
-                                    .expect("object request"),
-                            )
-                            .await
-                            .expect("object admission response");
-                        if published {
-                            assert_eq!(response.status(), StatusCode::IM_A_TEAPOT, "{method} reaches the inner service");
-                        } else {
-                            assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE, "{method} stays gated");
-                            assert_eq!(response.headers()[http::header::RETRY_AFTER], "5");
-                            assert_eq!(response.headers()["x-rustfs-readiness-pending"], "startup_finalization");
-                        }
-                    }
-                    assert_eq!(calls.load(Ordering::SeqCst), if published { 2 } else { 0 });
-                }
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_COMPAT_LIVE_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("liveness request before admission publication"),
+                    )
+                    .await
+                    .expect("liveness response before admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("liveness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("liveness JSON");
+                assert_eq!(payload["status"], "ok");
+                assert!(payload.get("ready").is_none());
+
+                readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::HEAD)
+                            .uri(MINIO_HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request after admission publication"),
+                    )
+                    .await
+                    .expect("readiness response after admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                assert_eq!(calls.load(Ordering::SeqCst), 0);
             },
         )
         .await;

rustfs/src/server/mod.rs

@@ -92,8 +92,6 @@ pub(crate) use readiness::collect_node_readiness_report;
 pub(crate) use readiness::observe_peer_drive_admission;
 pub use readiness::publish_ready_when_runtime_ready;
 pub(crate) use readiness::snapshot_dependency_readiness_report;
-#[cfg(test)]
-pub(crate) use readiness::wait_for_runtime_readiness_with;
 pub(crate) use readiness::{collect_cluster_read_health_report, collect_cluster_write_health_report};
 pub(crate) use ssec_transport::NativeTlsTransport;
 

rustfs/src/server/readiness.rs

@@ -75,9 +75,6 @@ fn startup_runtime_readiness_max_wait() -> Duration {
         Duration::from_secs(secs)
     }
 }
-const EVENT_STARTUP_RUNTIME_READINESS: &str = "startup_runtime_readiness";
-const LOG_COMPONENT_SERVER: &str = "server";
-const LOG_SUBSYSTEM_READINESS: &str = "readiness";
 const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
 const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
 const METRIC_POOL_METADATA_CHECK_TIMEOUT_TOTAL: &str = "rustfs_pool_metadata_check_timeouts_total";
@@ -1276,15 +1273,6 @@ async fn collect_lock_quorum_observation_uncached() -> LockQuorumObservation {
     }
 }
 
-/// Wait for local runtime dependencies without exceeding the observation budget.
-///
-/// The budget covers both observations and polling sleeps. An immediately ready
-/// observation can succeed with a zero budget. Cancellation drops the in-flight
-/// observation; only success invokes `on_ready`.
-///
-/// # Errors
-/// Returns an error on expiry, retaining the last completed observation when
-/// available. The readiness callback is never invoked after a timeout.
 pub async fn wait_for_runtime_readiness_with<F, Fut, ReadyFn>(
     max_wait: Duration,
     poll_interval: Duration,
@@ -1297,60 +1285,35 @@ where
     ReadyFn: FnMut(DependencyReadiness),
 {
     let startup_deadline = tokio::time::Instant::now() + max_wait;
-    let mut last_readiness = None;
-    let result = tokio::time::timeout_at(startup_deadline, async {
-        loop {
-            // Tokio timeouts poll the observation first. Keep the initial
-            // zero-budget observation, but start no later one at expiry.
-            if last_readiness.is_some() && tokio::time::Instant::now() >= startup_deadline {
-                return None;
-            }
-            let readiness = load_readiness().await;
-            last_readiness = Some(readiness);
-            if readiness.storage_ready
-                && readiness.iam_ready
-                && readiness.lock_quorum_ready
-                && readiness.peer_health_ready
-                && (max_wait.is_zero() || tokio::time::Instant::now() < startup_deadline)
-            {
-                return Some(readiness);
-            }
-            if tokio::time::Instant::now() >= startup_deadline {
-                return None;
-            }
-            info!(
-                event = EVENT_STARTUP_RUNTIME_READINESS,
-                component = LOG_COMPONENT_SERVER,
-                subsystem = LOG_SUBSYSTEM_READINESS,
-                state = "waiting",
-                storage_ready = readiness.storage_ready,
-                iam_ready = readiness.iam_ready,
-                lock_quorum_ready = readiness.lock_quorum_ready,
-                peer_health_ready = readiness.peer_health_ready,
-                "Runtime node readiness has not been reached yet; delaying ready state publication"
-            );
-            tokio::time::sleep(poll_interval).await;
-        }
-    })
-    .await;
-    match result {
-        Ok(Some(readiness)) => {
+
+    loop {
+        let readiness = load_readiness().await;
+        if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready && readiness.peer_health_ready {
             on_ready(readiness);
-            Ok(())
+            return Ok(());
         }
-        Ok(None) | Err(_) => {
-            let detail = match last_readiness {
-                Some(readiness) => format!(
-                    "storage_ready={}, iam_ready={}, lock_quorum_ready={}, peer_health_ready={}",
-                    readiness.storage_ready, readiness.iam_ready, readiness.lock_quorum_ready, readiness.peer_health_ready
-                ),
-                None => "no readiness observation completed".to_string(),
-            };
-            Err(std::io::Error::other(format!(
-                "startup readiness timed out after {}s: {detail}",
-                max_wait.as_secs()
-            )))
+
+        if tokio::time::Instant::now() >= startup_deadline {
+            let reason = format!(
+                "startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}, peer_health_ready={}",
+                max_wait.as_secs(),
+                readiness.storage_ready,
+                readiness.iam_ready,
+                readiness.lock_quorum_ready,
+                readiness.peer_health_ready
+            );
+            return Err(std::io::Error::other(reason));
         }
+
+        info!(
+            target: "rustfs::server::readiness",
+            storage_ready = readiness.storage_ready,
+            iam_ready = readiness.iam_ready,
+            lock_quorum_ready = readiness.lock_quorum_ready,
+            peer_health_ready = readiness.peer_health_ready,
+            "Runtime node readiness has not been reached yet; delaying ready state publication"
+        );
+        tokio::time::sleep(poll_interval).await;
     }
 }
 
@@ -1655,7 +1618,6 @@ mod tests {
     /// predicate counts node-2's drive online, then publish.
     #[tokio::test(start_paused = true)]
     async fn restarted_node_stays_unready_until_survivor_counts_its_drive_online() {
-        use futures::StreamExt;
         use peer_readmission::{
             PeerDriveView, PeerProbe, ReadmissionOutcome, UNREACHABLE_PEER_GRACE, host_drive_observations, readmission_plan,
             wait_for_peer_readmission_with,
@@ -1692,8 +1654,8 @@ mod tests {
         let deadline = tokio::time::Instant::now() + Duration::from_secs(120);
 
         let gate = async {
-            let outcome = wait_for_peer_readmission_with(&plan, deadline, Duration::from_secs(1), UNREACHABLE_PEER_GRACE, || {
-                futures::stream::once(async {
+            let outcome =
+                wait_for_peer_readmission_with(&plan, deadline, Duration::from_secs(1), UNREACHABLE_PEER_GRACE, || async {
                     let mut online_hosts = HashSet::from(["node-0:9000".to_string(), "node-1:9000".to_string()]);
                     if survivor_reaches_restarted.load(Ordering::SeqCst) {
                         online_hosts.insert("node-2:9000".to_string());
@@ -1715,9 +1677,7 @@ mod tests {
                     );
                     HashMap::from([("node-0:9000".to_string(), view.clone()), ("node-1:9000".to_string(), view)])
                 })
-                .flat_map(futures::stream::iter)
-            })
-            .await;
+                .await;
             readiness.mark_stage(rustfs_common::SystemStage::FullReady);
             outcome
         };
@@ -2019,150 +1979,6 @@ mod tests {
         assert_eq!(state_manager.current_state(), ServiceState::Ready);
     }
 
-    #[tokio::test(start_paused = true)]
-    async fn wait_for_runtime_readiness_cancels_a_suspended_observation() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        struct ObservationDrop(Arc<AtomicUsize>);
-        impl Drop for ObservationDrop {
-            fn drop(&mut self) {
-                self.0.fetch_add(1, Ordering::SeqCst);
-            }
-        }
-        let dropped = Arc::new(AtomicUsize::new(0));
-        let published = AtomicUsize::new(0);
-        let started = tokio::time::Instant::now();
-        let wait = wait_for_runtime_readiness_with(
-            Duration::from_secs(2),
-            Duration::from_secs(1),
-            || {
-                let guard = ObservationDrop(Arc::clone(&dropped));
-                async move {
-                    let _guard = guard;
-                    std::future::pending::<DependencyReadiness>().await
-                }
-            },
-            |_| {
-                published.fetch_add(1, Ordering::SeqCst);
-            },
-        );
-        let err = tokio::time::timeout(Duration::from_secs(3), wait)
-            .await
-            .expect("startup budget bounds a suspended dependency observation")
-            .expect_err("no dependency observation completed");
-        assert!(err.to_string().contains("no readiness observation completed"));
-        assert_eq!(started.elapsed(), Duration::from_secs(2));
-        assert_eq!(dropped.load(Ordering::SeqCst), 1);
-        assert_eq!(published.load(Ordering::SeqCst), 0);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn wait_for_runtime_readiness_bounds_the_poll_sleep() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        let polls = AtomicUsize::new(0);
-        let started = tokio::time::Instant::now();
-        let wait = wait_for_runtime_readiness_with(
-            Duration::from_secs(2),
-            Duration::from_secs(30),
-            || {
-                polls.fetch_add(1, Ordering::SeqCst);
-                future::ready(DependencyReadiness::default())
-            },
-            |_| panic!("an unready observation must never publish"),
-        );
-        let err = tokio::time::timeout(Duration::from_secs(3), wait)
-            .await
-            .expect("startup budget bounds the poll sleep")
-            .expect_err("unready observation");
-        assert_eq!(started.elapsed(), Duration::from_secs(2));
-        assert_eq!(polls.load(Ordering::SeqCst), 1, "expiry must not start another observation");
-        assert!(err.to_string().contains("lock_quorum_ready=false"));
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn wait_for_runtime_readiness_publishes_once_before_the_deadline() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        let polls = AtomicUsize::new(0);
-        let published = AtomicUsize::new(0);
-        let started = tokio::time::Instant::now();
-        wait_for_runtime_readiness_with(
-            Duration::from_secs(2),
-            Duration::from_secs(1),
-            || {
-                let ready = polls.fetch_add(1, Ordering::SeqCst) > 0;
-                future::ready(DependencyReadiness {
-                    storage_ready: ready,
-                    iam_ready: ready,
-                    lock_quorum_ready: ready,
-                    peer_health_ready: ready,
-                })
-            },
-            |_| {
-                published.fetch_add(1, Ordering::SeqCst);
-            },
-        )
-        .await
-        .expect("dependencies recovered before expiry");
-        assert_eq!(started.elapsed(), Duration::from_secs(1));
-        assert_eq!(published.load(Ordering::SeqCst), 1);
-        assert_eq!(polls.load(Ordering::SeqCst), 2);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn wait_for_runtime_readiness_starts_no_observation_at_expiry() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        let polls = AtomicUsize::new(0);
-        let published = AtomicUsize::new(0);
-        let started = tokio::time::Instant::now();
-        wait_for_runtime_readiness_with(
-            Duration::from_secs(2),
-            Duration::from_secs(1),
-            || {
-                let ready = polls.fetch_add(1, Ordering::SeqCst) >= 2;
-                future::ready(DependencyReadiness {
-                    storage_ready: ready,
-                    iam_ready: ready,
-                    lock_quorum_ready: ready,
-                    peer_health_ready: ready,
-                })
-            },
-            |_| {
-                published.fetch_add(1, Ordering::SeqCst);
-            },
-        )
-        .await
-        .expect_err("expiry must prevent a third observation even when immediately ready");
-        assert_eq!(started.elapsed(), Duration::from_secs(2));
-        assert_eq!(polls.load(Ordering::SeqCst), 2);
-        assert_eq!(published.load(Ordering::SeqCst), 0);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn wait_for_runtime_readiness_does_not_publish_an_observation_at_expiry() {
-        use std::sync::atomic::{AtomicUsize, Ordering};
-        let published = AtomicUsize::new(0);
-        let started = tokio::time::Instant::now();
-        wait_for_runtime_readiness_with(
-            Duration::from_secs(2),
-            Duration::from_secs(1),
-            || async {
-                tokio::time::sleep(Duration::from_secs(2)).await;
-                DependencyReadiness {
-                    storage_ready: true,
-                    iam_ready: true,
-                    lock_quorum_ready: true,
-                    peer_health_ready: true,
-                }
-            },
-            |_| {
-                published.fetch_add(1, Ordering::SeqCst);
-            },
-        )
-        .await
-        .expect_err("an observation completing at the deadline must not publish");
-        assert_eq!(started.elapsed(), Duration::from_secs(2));
-        assert_eq!(published.load(Ordering::SeqCst), 0);
-    }
-
     #[test]
     fn probe_path_checks_admin_boundaries() {
         assert!(is_probe_path("/minio/admin/v3/info"));

rustfs/src/startup_iam.rs

@@ -85,35 +85,32 @@ where
     publish_ready().await
 }
 
-async fn prepare_iam_startup(
+async fn finalize_iam_recovery(
     iam: Arc<rustfs_iam::sys::IamSys<rustfs_iam::store::object::ObjectStore>>,
     store: Arc<ECStore>,
     kms_interface: Arc<KmsServiceManager>,
     readiness: Arc<GlobalReadiness>,
+    state_manager: Option<Arc<ServiceStateManager>>,
     server_ctx: Arc<ServerContextSlot>,
-) -> Result<HealTopologyReady> {
+) -> Result<()> {
     let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?;
-    publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools, || {
-        AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)
+    AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
+    let topology_ready =
+        publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools).await?;
+    publish_runtime_ready_after_heal_topology(topology_ready, || async {
+        publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
     })
     .await
 }
 
-async fn publish_iam_ready_after_heal_topology<InstallFn>(
+async fn publish_iam_ready_after_heal_topology(
     readiness: &GlobalReadiness,
     cache: Arc<tokio::sync::OnceCell<String>>,
     endpoint_pools: crate::storage_api::cluster::EndpointServerPools,
-    install_context: InstallFn,
-) -> Result<HealTopologyReady>
-where
-    InstallFn: FnOnce() -> Result<()>,
-{
+) -> Result<HealTopologyReady> {
     crate::storage_api::startup::heal_control::initialize_heal_topology_fingerprint(cache, endpoint_pools)
         .await
         .map_err(std::io::Error::other)?;
-    // Finish fallible preparation before claiming the one-time context slot.
-    // No fallible/async step may follow a successful claim before this token.
-    install_context()?;
     readiness.mark_stage(SystemStage::IamReady);
     Ok(HealTopologyReady(()))
 }
@@ -148,7 +145,8 @@ fn spawn_iam_recovery_task(
     let finalize_readiness = readiness;
     let finalize_state_manager = state_manager;
     let finalize_server_ctx = server_ctx;
-    let prepare_readiness = finalize_readiness.clone();
+    let recovered_iam = Arc::new(std::sync::Mutex::new(None));
+    let init_recovered_iam = recovered_iam.clone();
     tokio::spawn(async move {
         run_iam_recovery_loop(
             initial_interval,
@@ -156,23 +154,25 @@ fn spawn_iam_recovery_task(
             shutdown_token,
             move || {
                 let store = init_store.clone();
-                Box::pin(async move { attempt_init_iam_sys(store).await })
+                let recovered_iam = init_recovered_iam.clone();
+                Box::pin(async move {
+                    let iam = attempt_init_iam_sys(store).await?;
+                    *recovered_iam
+                        .lock()
+                        .map_err(|_| std::io::Error::other("IAM recovery state poisoned"))? = Some(iam);
+                    Ok(())
+                })
             },
-            move |iam| {
+            move || {
                 let store = finalize_store.clone();
                 let kms_interface = finalize_kms_interface.clone();
-                let readiness = prepare_readiness.clone();
-                let server_ctx = finalize_server_ctx.clone();
-                Box::pin(prepare_iam_startup(iam, store, kms_interface, readiness, server_ctx))
-            },
-            move |topology_ready| {
                 let readiness = finalize_readiness.clone();
                 let state_manager = finalize_state_manager.clone();
+                let server_ctx = finalize_server_ctx.clone();
+                let iam = recovered_iam.lock().ok().and_then(|mut recovered| recovered.take());
                 Box::pin(async move {
-                    publish_runtime_ready_after_heal_topology(topology_ready, || async {
-                        publish_ready_when_runtime_ready(readiness.as_ref(), state_manager.as_deref()).await
-                    })
-                    .await
+                    let iam = iam.ok_or_else(|| std::io::Error::other("IAM recovered but unavailable"))?;
+                    finalize_iam_recovery(iam, store, kms_interface, readiness, state_manager, server_ctx).await
                 })
             },
         )
@@ -180,27 +180,23 @@ fn spawn_iam_recovery_task(
     });
 }
 
-type RecoveryFuture<T = ()> = Pin<Box<dyn Future<Output = Result<T>> + Send>>;
+type RecoveryFuture = Pin<Box<dyn Future<Output = Result<()>> + Send>>;
 
-async fn run_iam_recovery_loop<T, Prepared, InitFn, PrepareFn, FinalizeFn>(
+async fn run_iam_recovery_loop<InitFn, FinalizeFn>(
     initial_interval: Duration,
     max_interval: Duration,
     shutdown_token: Option<tokio_util::sync::CancellationToken>,
     mut init_fn: InitFn,
-    mut prepare_fn: PrepareFn,
     mut finalize_fn: FinalizeFn,
 ) where
-    T: Clone,
-    Prepared: Clone,
-    InitFn: FnMut() -> RecoveryFuture<T>,
-    PrepareFn: FnMut(T) -> RecoveryFuture<Prepared>,
-    FinalizeFn: FnMut(Prepared) -> RecoveryFuture,
+    InitFn: FnMut() -> RecoveryFuture,
+    FinalizeFn: FnMut() -> RecoveryFuture,
 {
     let mut attempts: u64 = 0;
     let degraded_since = std::time::Instant::now();
 
-    // Keep the initialized IAM owner across preparation/publication retries.
-    let recovered = loop {
+    // Phase 1: retry init until it succeeds
+    loop {
         attempts += 1;
         let sleep_duration = compute_backoff_interval(attempts, initial_interval, max_interval);
         if let Some(token) = shutdown_token.as_ref() {
@@ -213,7 +209,7 @@ async fn run_iam_recovery_loop<T, Prepared, InitFn, PrepareFn, FinalizeFn>(
         }
 
         match init_fn().await {
-            Ok(recovered) => {
+            Ok(()) => {
                 let degraded_secs = degraded_since.elapsed().as_secs();
                 info!(
                     event = EVENT_IAM_BOOTSTRAP_RECOVERED,
@@ -223,7 +219,7 @@ async fn run_iam_recovery_loop<T, Prepared, InitFn, PrepareFn, FinalizeFn>(
                     degraded_duration_secs = degraded_secs,
                     "IAM bootstrap recovered after startup; publishing IAM readiness"
                 );
-                break recovered;
+                break;
             }
             Err(err) => {
                 let next_interval = compute_backoff_interval(attempts + 1, initial_interval, max_interval);
@@ -254,24 +250,13 @@ async fn run_iam_recovery_loop<T, Prepared, InitFn, PrepareFn, FinalizeFn>(
                 }
             }
         }
-    };
+    }
 
-    // Preparation claims the context once. Only publication retries after it.
-    let mut prepared: Option<Prepared> = None;
+    // Phase 2: retry finalize until readiness is published
     let mut finalize_attempts: u64 = 0;
     loop {
-        let result = match prepared.as_ref() {
-            Some(prepared) => finalize_fn(prepared.clone()).await,
-            None => match prepare_fn(recovered.clone()).await {
-                Ok(value) => {
-                    prepared = Some(value);
-                    continue;
-                }
-                Err(error) => Err(error),
-            },
-        };
         finalize_attempts += 1;
-        match result {
+        match finalize_fn().await {
             Ok(()) => {
                 info!(
                     event = EVENT_IAM_READINESS_PUBLISHED,
@@ -399,7 +384,11 @@ pub(crate) async fn bootstrap_or_defer_iam_init(
 ) -> Result<IamBootstrapDisposition> {
     match attempt_init_iam_sys(store.clone()).await {
         Ok(iam) => {
-            let topology_ready = prepare_iam_startup(iam, store, kms_interface, readiness, server_ctx).await?;
+            let endpoint_pools = heal_control_endpoint_pools(store.as_ref())?;
+            AppContext::ensure_startup_after_iam(store, kms_interface, &server_ctx, iam)?;
+            let topology_ready =
+                publish_iam_ready_after_heal_topology(readiness.as_ref(), server_ctx.heal_topology_fingerprint(), endpoint_pools)
+                    .await?;
             return Ok(IamBootstrapDisposition::ReadyInline(topology_ready));
         }
         Err(err) => {
@@ -597,7 +586,7 @@ mod tests {
         let expected = crate::storage_api::startup::heal_control::heal_topology_fingerprint(&endpoint_pools)
             .expect("valid topology should hash");
 
-        publish_iam_ready_after_heal_topology(&readiness, Arc::clone(&cache), endpoint_pools, || Ok(()))
+        publish_iam_ready_after_heal_topology(&readiness, Arc::clone(&cache), endpoint_pools)
             .await
             .expect("topology initialization should publish IAM readiness");
 
@@ -610,7 +599,6 @@ mod tests {
             &failed_readiness,
             Arc::clone(&invalid_cache),
             EndpointServerPools::from(Vec::<PoolEndpoints>::new()),
-            || panic!("invalid topology must not claim the context"),
         )
         .await
         .expect_err("invalid topology must block IAM readiness");
@@ -640,8 +628,7 @@ mod tests {
                     Ok(())
                 })
             },
-            |()| Box::pin(async { Ok(()) }),
-            move |_| {
+            move || {
                 let finalize_calls = finalize_calls.clone();
                 Box::pin(async move {
                     let call = finalize_calls.fetch_add(1, Ordering::SeqCst) + 1;
@@ -684,8 +671,7 @@ mod tests {
                     }
                 })
             },
-            |()| Box::pin(async { Ok(()) }),
-            move |_| {
+            move || {
                 let readiness = readiness_for_finalize.clone();
                 let observed_iam_ready = observed_iam_ready_for_finalize.clone();
                 Box::pin(async move {
@@ -705,197 +691,6 @@ mod tests {
         assert!(readiness.is_ready());
     }
 
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn recovery_publication_timeout_reuses_the_installed_context() {
-        let context = crate::app::gating_test_env::shared_gating_ambient().await;
-        let server_ctx = ServerContextSlot::new();
-        // The legacy fixture exposes its real disks without instance endpoints.
-        let (paths, store) = crate::app::gating_test_env::shared_gating_ecstore_and_disk_paths().await;
-        assert!(Arc::ptr_eq(&store, &context.object_store()));
-        let endpoints = EndpointServerPools::from(vec![PoolEndpoints {
-            legacy: false,
-            set_count: 1,
-            drives_per_set: paths.len(),
-            endpoints: Endpoints::from(
-                paths
-                    .iter()
-                    .enumerate()
-                    .map(|(index, path)| {
-                        let mut endpoint =
-                            Endpoint::try_from(path.to_str().expect("fixture disk path")).expect("fixture endpoint");
-                        endpoint.set_pool_index(0);
-                        endpoint.set_set_index(0);
-                        endpoint.set_disk_index(index);
-                        endpoint
-                    })
-                    .collect::<Vec<_>>(),
-            ),
-            cmd_line: "recovery-publication-test".to_string(),
-            platform: String::new(),
-        }]);
-        crate::storage_api::startup::heal_control::initialize_heal_topology_fingerprint(
-            server_ctx.heal_topology_fingerprint(),
-            endpoints.clone(),
-        )
-        .await
-        .expect("prepare fixture fingerprint before pausing time");
-        // A fresh paused runtime keeps real fixture I/O and its timer-wheel
-        // rounding out of the exact retry/deadline observations below.
-        tokio::task::spawn_blocking(move || {
-            let runtime = tokio::runtime::Builder::new_current_thread()
-                .enable_all()
-                .start_paused(true)
-                .build()
-                .expect("paused recovery runtime");
-            runtime.block_on(async move {
-                let readiness = Arc::new(GlobalReadiness::new());
-                let init_calls = Arc::new(AtomicUsize::new(0));
-                let prepare_calls = Arc::new(AtomicUsize::new(0));
-                let publish_calls = Arc::new(AtomicUsize::new(0));
-                let started = tokio::time::Instant::now();
-                let recovery = run_iam_recovery_loop(
-                    Duration::from_secs(5),
-                    Duration::from_secs(30),
-                    None,
-                    {
-                        let context = Arc::clone(&context);
-                        let calls = Arc::clone(&init_calls);
-                        move || {
-                            calls.fetch_add(1, Ordering::SeqCst);
-                            let context = Arc::clone(&context);
-                            Box::pin(async move { Ok(context) })
-                        }
-                    },
-                    {
-                        let server_ctx = Arc::clone(&server_ctx);
-                        let readiness = Arc::clone(&readiness);
-                        let calls = Arc::clone(&prepare_calls);
-                        move |context| {
-                            calls.fetch_add(1, Ordering::SeqCst);
-                            let server_ctx = Arc::clone(&server_ctx);
-                            let readiness = Arc::clone(&readiness);
-                            let endpoints = endpoints.clone();
-                            Box::pin(async move {
-                                publish_iam_ready_after_heal_topology(
-                                    readiness.as_ref(),
-                                    server_ctx.heal_topology_fingerprint(),
-                                    endpoints,
-                                    || server_ctx.try_install(context),
-                                )
-                                .await
-                            })
-                        }
-                    },
-                    {
-                        let readiness = Arc::clone(&readiness);
-                        let calls = Arc::clone(&publish_calls);
-                        move |prepared| {
-                            let attempt = calls.fetch_add(1, Ordering::SeqCst);
-                            let readiness = Arc::clone(&readiness);
-                            Box::pin(async move {
-                                let result = publish_runtime_ready_after_heal_topology(prepared, || async {
-                                    crate::server::wait_for_runtime_readiness_with(
-                                        Duration::from_secs(2),
-                                        Duration::from_secs(1),
-                                        || async move {
-                                            if attempt == 0 {
-                                                std::future::pending::<crate::shared_types::DependencyReadiness>().await
-                                            } else {
-                                                crate::shared_types::DependencyReadiness {
-                                                    storage_ready: true,
-                                                    iam_ready: true,
-                                                    lock_quorum_ready: true,
-                                                    peer_health_ready: true,
-                                                }
-                                            }
-                                        },
-                                        |_| readiness.mark_stage(SystemStage::FullReady),
-                                    )
-                                    .await
-                                })
-                                .await;
-                                if attempt == 0 {
-                                    assert!(result.is_err(), "the first publication must time out");
-                                    assert!(!readiness.is_ready(), "timeout cannot publish FullReady");
-                                }
-                                result
-                            })
-                        }
-                    },
-                );
-                tokio::time::timeout(Duration::from_secs(20), recovery)
-                    .await
-                    .expect("publication retry must succeed without attempting another context installation");
-                assert_eq!(init_calls.load(Ordering::SeqCst), 1);
-                assert_eq!(prepare_calls.load(Ordering::SeqCst), 1);
-                assert_eq!(publish_calls.load(Ordering::SeqCst), 2);
-                assert_eq!(started.elapsed(), Duration::from_secs(12));
-                assert!(readiness.is_ready());
-                assert!(Arc::ptr_eq(&server_ctx.installed_app_context().expect("installed context"), &context));
-                assert_eq!(
-                    server_ctx
-                        .try_install(context)
-                        .expect_err("duplicate installs remain conflicts")
-                        .kind(),
-                    std::io::ErrorKind::AlreadyExists
-                );
-            });
-        })
-        .await
-        .expect("recovery model completes without panic");
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn recovery_preparation_retries_reuse_the_initialized_owner() {
-        let owner = Arc::new(AtomicUsize::new(0));
-        let init_calls = Arc::new(AtomicUsize::new(0));
-        let publish_calls = Arc::new(AtomicUsize::new(0));
-        run_iam_recovery_loop(
-            Duration::from_secs(1),
-            Duration::from_secs(2),
-            None,
-            {
-                let owner = Arc::clone(&owner);
-                let calls = Arc::clone(&init_calls);
-                move || {
-                    calls.fetch_add(1, Ordering::SeqCst);
-                    let owner = Arc::clone(&owner);
-                    Box::pin(async move { Ok(owner) })
-                }
-            },
-            {
-                let expected = Arc::clone(&owner);
-                move |owner| {
-                    assert!(Arc::ptr_eq(&expected, &owner), "preparation must keep the same initialized IAM owner");
-                    Box::pin(async move {
-                        if owner.fetch_add(1, Ordering::SeqCst) == 0 {
-                            Err(Error::other("topology not ready"))
-                        } else {
-                            Ok(owner)
-                        }
-                    })
-                }
-            },
-            {
-                let calls = Arc::clone(&publish_calls);
-                move |owner| {
-                    let calls = Arc::clone(&calls);
-                    Box::pin(async move {
-                        assert_eq!(owner.load(Ordering::SeqCst), 2, "publication follows successful preparation");
-                        calls.fetch_add(1, Ordering::SeqCst);
-                        Ok(())
-                    })
-                }
-            },
-        )
-        .await;
-        assert_eq!(init_calls.load(Ordering::SeqCst), 1);
-        assert_eq!(owner.load(Ordering::SeqCst), 2);
-        assert_eq!(publish_calls.load(Ordering::SeqCst), 1);
-        assert_eq!(Arc::strong_count(&owner), 1, "loop releases its initialized and prepared owners");
-    }
-
     #[tokio::test(start_paused = true)]
     async fn recovery_loop_stops_after_shutdown_cancellation() {
         let init_calls = Arc::new(AtomicUsize::new(0));
@@ -916,8 +711,7 @@ mod tests {
                             Err(Error::other("keep retrying"))
                         })
                     },
-                    |()| Box::pin(async { Ok(()) }),
-                    |()| Box::pin(async { Ok(()) }),
+                    move || Box::pin(async { Ok(()) }),
                 )
                 .await;
             }