fix(ecstore): complete and abort native multipart uploads on Windows

Native multipart completion publishes its payload from a TMP staging
parent that differs from the canonical head's parent. Linux pins that
parent with a dedicated payload commit guard, but Windows reused the
metadata guard, so the data-dir rename failed its parent check with
InvalidInput, which surfaced as FileNotFound/NoSuchKey.

- Add a Windows multipart payload commit guard that pins the staging
  tree and reuses the metadata transaction's destination parent handle.
- Run the owned-source rename precondition on Windows before opening the
  exclusive DELETE handle, under a non-exclusive identity anchor, and
  require the prepared handle to name the same file. The precondition's
  own opens previously hit ERROR_SHARING_VIOLATION during abort and
  stale-upload cleanup.
- Recognize ERROR_ACCESS_DENIED on a directory as a directory in the
  remove fallback, so remove/remove_std can delete directories on
  Windows; run each fallback in one blocking dispatch.
- Run the complete/abort and directory-removal tests in the Windows
  filesystem workflow, and add a Windows payload guard regression test.

.github/workflows/windows-filesystem.yml

@@ -107,6 +107,21 @@ 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
+
+      - 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/commit.rs

@@ -1594,12 +1594,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 +1609,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/os.rs

@@ -5479,18 +5479,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, Some(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(
@@ -6601,12 +6630,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)?;
@@ -7193,7 +7241,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 +7276,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 +7352,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 };