fix(ecstore): let Windows directory pins share writes on every component

A Windows directory pin shared only reads on ancestors, so a directory
pinned as one object's ancestor refused the write open the kernel takes on
a rename target parent. With nested keys (`a/b` beside `a/b/c`) a parent
object's data-directory publication failed with ERROR_SHARING_VIOLATION
while a child write held the pin. Every pinned component now shares reads
and writes and still excludes delete sharing, which is what pins identity.

Also gate the two grouped dst dir fsync rollback tests to Unix, where the
failure hook exists, and make the rebalance dangling-copy fixture wait for
every drive before it removes copies.

.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

@@ -4083,7 +4083,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 +4505,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 +4903,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 +5369,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 +5406,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 +5460,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 +5542,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 +5571,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 +5779,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,17 +5799,7 @@ 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))
@@ -6109,7 +6117,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,28 +6181,29 @@ pub(crate) fn compare_and_update_control_file(
     Ok(ConditionalFileUpdate::Updated)
 }
 
+/// 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.
 #[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,
+            FILE_ATTRIBUTE_DIRECTORY, FILE_ATTRIBUTE_REPARSE_POINT, FILE_READ_ATTRIBUTES, FILE_SHARE_READ, FILE_SHARE_WRITE,
+            FILE_TRAVERSE,
         },
     };
 
@@ -6202,7 +6211,7 @@ fn open_windows_relative_directory_component(
         &parent.handle,
         component,
         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,
@@ -6511,7 +6520,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 +6610,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 +7040,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 +7224,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 +7259,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 +7335,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 +7592,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 +7602,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 +7618,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 +9033,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 +9075,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 +9098,51 @@ 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)]
     #[tokio::test]
     async fn windows_rename_all_supports_same_parent_publication() {
@@ -9717,7 +9840,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"),
@@ -9862,7 +9985,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");