Compare commits

...

1 Commits

Author SHA1 Message Date
唐小鸭 5345772af4 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.
2026-11-02 17:16:19 +08:00
4 changed files with 157 additions and 27 deletions
+15
View File
@@ -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"
+31 -20
View File
@@ -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),
}
}
+4 -3
View File
@@ -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,
+107 -4
View File
@@ -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 };