mirror of
https://github.com/rustfs/rustfs.git
synced 2026-11-03 17:45:05 +00:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| c830af66d5 | |||
| 5345772af4 |
@@ -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"
|
||||
|
||||
@@ -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),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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)?;
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -5479,18 +5481,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(
|
||||
@@ -5565,6 +5596,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;
|
||||
@@ -6601,12 +6654,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 +7265,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 +7300,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 +7376,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 };
|
||||
|
||||
@@ -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");
|
||||
|
||||
Reference in New Issue
Block a user