From c58accabe746c9566da57ab7f1c8f1b4c4950435 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 23 Aug 2026 07:58:22 +0800 Subject: [PATCH] fix(ecstore): support Windows checkpoint CAS --- crates/ecstore/src/disk/local.rs | 70 ++++++++++++++++++++++++-- crates/ecstore/src/disk/os.rs | 85 ++++++++++++++++++++++++++++++++ 2 files changed, 151 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index e1da1429f..5217e819b 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -8073,7 +8073,25 @@ impl DiskAPI for LocalDisk { .map_err(DiskError::from)??); } - #[cfg(not(unix))] + #[cfg(windows)] + { + let file_path = self.io_get_object_path(volume, path)?; + let sync_metadata = effective_durability(volume).syncs_commit_metadata(); + let publication_root = self.publication_root.clone(); + return Ok(tokio::task::spawn_blocking(move || { + os::compare_and_update_control_file( + &file_path, + expected.as_deref(), + replacement.as_deref(), + sync_metadata, + &publication_root, + ) + }) + .await + .map_err(DiskError::from)??); + } + + #[cfg(not(any(unix, windows)))] { let _ = (volume, path, expected, replacement); Err(DiskError::MethodNotAllowed) @@ -21828,9 +21846,9 @@ mod test { assert!(matches!(results[1].as_ref().unwrap_err(), DiskError::Io(_))); } - #[cfg(unix)] + #[cfg(any(unix, windows))] #[tokio::test] - async fn conditional_file_update_never_deletes_a_new_owner() { + async fn windows_and_unix_conditional_file_update_never_deletes_a_new_owner() { use tempfile::tempdir; let dir = tempdir().expect("temp dir should be created"); @@ -21861,8 +21879,18 @@ mod test { disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH) .await .expect("new owner marker should remain"), - owner_b + owner_b.clone() ); + assert_eq!( + disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Some(owner_b), None) + .await + .expect("current owner should remove marker"), + ConditionalFileUpdate::Updated + ); + assert!(matches!( + disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH).await, + Err(DiskError::FileNotFound) + )); } #[cfg(unix)] @@ -21901,6 +21929,40 @@ mod test { assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::WouldBlock)); } + #[cfg(windows)] + #[tokio::test] + async fn windows_conditional_file_update_returns_would_block_when_marker_lock_is_contended() { + let dir = tempfile::tempdir().expect("temp dir should be created"); + let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse"); + let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created"); + ensure_test_volume(&disk, RUSTFS_META_BUCKET).await; + let marker_path = disk + .get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH) + .expect("marker path should resolve"); + let lock_path = marker_path + .parent() + .expect("marker path should have a parent") + .join(".rustfs-cas.lock"); + let lock = std::fs::OpenOptions::new() + .create(true) + .truncate(false) + .read(true) + .write(true) + .open(lock_path) + .expect("marker lock should open"); + lock.try_lock().expect("marker lock should be held"); + + let err = tokio::time::timeout( + Duration::from_secs(1), + disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, None, Some(Bytes::from_static(b"owner"))), + ) + .await + .expect("contended conditional update must not block") + .expect_err("contended conditional update must retry"); + + assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::WouldBlock)); + } + #[cfg(target_os = "linux")] #[tokio::test] async fn replacement_io_paths_stay_under_the_mount_lease() { diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 77ea01991..7e6a92557 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(windows)] +use crate::disk::ConditionalFileUpdate; use crate::disk::error::DiskError; use crate::disk::error::Result; use crate::disk::error_conv::to_file_error; @@ -3458,6 +3460,89 @@ fn read_windows_relative_file(file_path: &Path, parent_guard: &ExistingBaseDirec Ok(Some(data)) } +#[cfg(windows)] +pub(crate) fn compare_and_update_control_file( + file_path: &Path, + expected: Option<&[u8]>, + replacement: Option<&[u8]>, + sync_metadata: bool, + publication_root: &PublicationRoot, +) -> io::Result { + use windows_sys::{ + Wdk::Storage::FileSystem::{ + FILE_NON_DIRECTORY_FILE, FILE_OPEN, FILE_OPEN_IF, FILE_OPEN_REPARSE_POINT, FILE_SYNCHRONOUS_IO_NONALERT, + }, + Win32::Storage::FileSystem::{ + DELETE, FILE_ATTRIBUTE_NORMAL, FILE_READ_ATTRIBUTES, FILE_SHARE_READ, FILE_SHARE_WRITE, FILE_WRITE_DATA, SYNCHRONIZE, + }, + }; + + 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 lock = open_windows_relative( + parent_guard.last_handle()?, + std::ffi::OsStr::new(".rustfs-cas.lock"), + SYNCHRONIZE | FILE_READ_ATTRIBUTES | FILE_WRITE_DATA, + FILE_SHARE_READ | FILE_SHARE_WRITE, + FILE_OPEN_IF, + FILE_NON_DIRECTORY_FILE | FILE_OPEN_REPARSE_POINT | FILE_SYNCHRONOUS_IO_NONALERT, + FILE_ATTRIBUTE_NORMAL, + true, + )?; + validate_windows_owned_file(&lock)?; + match lock.as_file().try_lock() { + Ok(()) => {} + Err(std::fs::TryLockError::WouldBlock) => return Err(io::Error::from(io::ErrorKind::WouldBlock)), + Err(std::fs::TryLockError::Error(err)) => return Err(err), + } + + let current = read_windows_relative_file(file_path, &parent_guard)?; + let matches = match (¤t, expected) { + (None, None) => true, + (Some(current), Some(expected)) => current.as_slice() == expected, + _ => false, + }; + if !matches { + return Ok(match current { + None => ConditionalFileUpdate::Missing, + Some(_) => ConditionalFileUpdate::Mismatch, + }); + } + + match replacement { + Some(replacement) => RenameDestinationPathGuard { + directory: parent.to_path_buf(), + _directory_guard: parent_guard, + } + .write_file_for_path_access(file_path, replacement, sync_metadata, sync_metadata)?, + None => { + let file_name = file_path + .file_name() + .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "conditional file must have a name"))?; + let file = open_windows_relative( + parent_guard.last_handle()?, + file_name, + DELETE | SYNCHRONIZE | FILE_READ_ATTRIBUTES, + FILE_SHARE_READ, + FILE_OPEN, + FILE_NON_DIRECTORY_FILE | FILE_OPEN_REPARSE_POINT | FILE_SYNCHRONOUS_IO_NONALERT, + 0, + true, + )?; + validate_windows_owned_file(&file)?; + set_windows_file_delete_on_close(&file, true)?; + drop(file); + if sync_metadata { + fsync_dir_std(parent)?; + } + } + } + + Ok(ConditionalFileUpdate::Updated) +} + #[cfg(windows)] fn open_windows_directory_component( parent: &WindowsDirectoryHandle,