fix(ecstore): retry Windows directory pins past transient delete holders

A directory pin excludes delete sharing to fix the directory's identity,
so it fails while another handle holds delete access. Pruning an empty
object prefix takes that access briefly: RemoveDirectoryW opens the
directory for delete and closes it again even when the directory is not
empty. A concurrent PUT of a nested key pinning the same prefix then
failed with ERROR_SHARING_VIOLATION on some drives and lost write
quorum.

Keep the NTSTATUS of a relative open and let a directory pin retry
STATUS_SHARING_VIOLATION and STATUS_DELETE_PENDING with six doubling
delays from 1 ms. A retry after a finished removal follows the caller's
create disposition, and a lasting delete holder still fails the pin.

Replace the source-parent assertion that a writer cannot open the
pinned directory, which no longer holds now that pins share writes, with
the guarantees that matter: the non-empty parent cannot become a mount
point and cannot be deleted or renamed.

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

@@ -9344,7 +9344,12 @@ 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)?
+                        // Windows only flushes a handle opened for writing.
+                        #[cfg(windows)]
+                        let receipt = std::fs::OpenOptions::new().write(true).open(receipt_path)?;
+                        #[cfg(not(windows))]
+                        let receipt = std::fs::File::open(receipt_path)?;
+                        receipt
                     }
                     Err(err) => return Err(err),
                 };
@@ -13695,6 +13700,10 @@ impl DiskAPI for LocalDisk {
                     return Ok(Err(DiskError::FileCorrupt));
                 }
                 let verify_and_sync = (|| -> std::io::Result<bool> {
+                    // Windows only flushes a handle opened for writing.
+                    #[cfg(windows)]
+                    let mut file = std::fs::OpenOptions::new().read(true).write(true).open(&source)?;
+                    #[cfg(not(windows))]
                     let mut file = std::fs::File::open(&source)?;
                     let mut header = [0u8; rustfs_filemeta::shard_integrity::INDEX_HEADER_SIZE];
                     std::io::Read::read_exact(&mut file, &mut header)?;
@@ -18943,6 +18952,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() {
@@ -19031,6 +19043,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,12 @@ 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()?;
+        // Windows only flushes a handle opened for writing.
+        #[cfg(windows)]
+        let backup = std::fs::OpenOptions::new().write(true).open(&backup_path)?;
+        #[cfg(not(windows))]
+        let backup = std::fs::File::open(&backup_path)?;
+        backup.sync_data()?;
     }
     Ok(backup_path)
 }
@@ -1594,12 +1599,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
@@ -1608,9 +1614,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

@@ -1527,7 +1527,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)?;
@@ -1560,7 +1560,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.
@@ -4083,7 +4103,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,
     };
@@ -4505,7 +4525,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()?;
@@ -4903,7 +4923,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)
     }
 
@@ -5367,14 +5389,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(
@@ -5407,35 +5426,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(),
@@ -5479,18 +5480,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(
@@ -5532,20 +5562,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)
@@ -5565,6 +5591,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;
@@ -5751,22 +5799,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 {
@@ -5781,26 +5819,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,
@@ -5811,6 +5836,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},
@@ -5825,10 +5908,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()
@@ -5868,13 +5948,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) })
@@ -6109,7 +6193,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"),
@@ -6173,41 +6257,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));
@@ -6511,7 +6632,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)))]
     {
@@ -6601,12 +6722,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)?;
@@ -7012,12 +7152,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,
@@ -7193,7 +7336,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!(
@@ -7228,6 +7371,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() {
@@ -7249,7 +7447,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 };
@@ -7506,7 +7704,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
@@ -7517,13 +7714,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,
@@ -7534,22 +7730,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)
@@ -8955,7 +9145,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,
@@ -8997,13 +9187,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"))
@@ -9020,6 +9210,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() {
@@ -9717,7 +10016,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"),
@@ -9751,30 +10050,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)
@@ -9783,6 +10103,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)]
@@ -9862,7 +10183,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

@@ -1899,6 +1899,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");