From c478a392e77d7ff31bc90aaeb079bbb5bb7cc24a Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 11 Sep 2026 22:38:23 +0800 Subject: [PATCH] fix(heal): select writable recovery intent owner (#7668) --- crates/ecstore/src/disk/error.rs | 54 ++++++++ crates/ecstore/src/disk/local.rs | 18 ++- crates/heal/src/heal/manager/root_recovery.rs | 60 ++++++--- .../src/heal/manager/tests/root_recovery.rs | 126 ++++++++++++++++++ 4 files changed, 233 insertions(+), 25 deletions(-) diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index 8e551b4af..fad6e561d 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -41,6 +41,14 @@ struct DanglingDeleteGraceError { grace_secs: i64, } +/// Marks a conditional-file write that failed before its publication rename. +/// Callers may choose another owner only while this marker is preserved; every +/// unmarked error remains commit-ambiguous and must fail closed. +#[derive(Debug)] +struct ConditionalFileNotCommittedError { + source: io::Error, +} + // DiskError == StorageErr #[derive(Debug, thiserror::Error)] pub enum DiskError { @@ -220,6 +228,18 @@ impl std::fmt::Display for DanglingDeleteGraceError { impl StdError for DanglingDeleteGraceError {} +impl std::fmt::Display for ConditionalFileNotCommittedError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.source.fmt(f) + } +} + +impl StdError for ConditionalFileNotCommittedError { + fn source(&self) -> Option<&(dyn StdError + 'static)> { + Some(&self.source) + } +} + fn classify_internode_missing_error(error: &InternodeHttpError) -> Option { if error.is_remote_file_not_found() { return Some(DiskError::FileNotFound); @@ -293,6 +313,22 @@ impl DiskError { }) } + pub(crate) fn conditional_file_not_committed(source: io::Error) -> io::Error { + io::Error::new(source.kind(), ConditionalFileNotCommittedError { source }) + } + + /// Whether a local conditional-file replacement failed before the target + /// publication rename and therefore cannot have committed new owner bytes. + pub fn is_conditional_file_not_committed(&self) -> bool { + matches!( + self, + DiskError::Io(io_error) + if io_error + .get_ref() + .is_some_and(|source| source.downcast_ref::().is_some()) + ) + } + pub fn is_dangling_delete_grace(&self) -> bool { matches!(self, DiskError::Io(io_error) if Self::io_error_is_dangling_delete_grace(io_error)) } @@ -627,6 +663,9 @@ impl From for DiskError { impl Clone for DiskError { fn clone(&self) -> Self { match self { + DiskError::Io(io_error) if self.is_conditional_file_not_committed() => DiskError::Io( + DiskError::conditional_file_not_committed(io::Error::new(io_error.kind(), io_error.to_string())), + ), DiskError::Io(io_error) => DiskError::Io( rustfs_rio::clone_internode_http_io_error(io_error) .and_then(std::io::Error::into_inner) @@ -820,6 +859,21 @@ mod tests { use super::*; use std::collections::HashMap; + #[test] + fn conditional_file_not_committed_marker_is_explicit_and_clone_safe() { + let marked = DiskError::from(DiskError::conditional_file_not_committed(io::Error::new( + io::ErrorKind::PermissionDenied, + "staging rejected", + ))); + assert!(marked.is_conditional_file_not_committed()); + assert!(marked.clone().is_conditional_file_not_committed()); + assert!(!DiskError::Timeout.is_conditional_file_not_committed()); + assert!( + !DiskError::Io(io::Error::new(io::ErrorKind::PermissionDenied, "rename rejected")) + .is_conditional_file_not_committed() + ); + } + #[test] fn terminal_read_error_preserves_kind_and_disk_classification() { let timeout = terminal_read_error_to_io(DiskError::Timeout); diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index f62def0a8..d888e3f7d 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -8957,7 +8957,8 @@ impl DiskAPI for LocalDisk { .truncate(false) .read(true) .write(true) - .open(&lock_path)?; + .open(&lock_path) + .map_err(DiskError::conditional_file_not_committed)?; flock(&lock, FlockOperation::NonBlockingLockExclusive).map_err(std::io::Error::from)?; let result = (|| { let current = match std::fs::read(&file_path) { @@ -9012,10 +9013,15 @@ impl DiskAPI for LocalDisk { .ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "conditional file has no parent"))?; let temporary = parent.join(format!(".{}.{}.tmp", path.replace('/', "_"), Uuid::new_v4())); let write_result = (|| -> std::io::Result<()> { - let mut staged = std::fs::OpenOptions::new().create_new(true).write(true).open(&temporary)?; - staged.write_all(&replacement)?; + let not_committed = DiskError::conditional_file_not_committed; + let mut staged = std::fs::OpenOptions::new() + .create_new(true) + .write(true) + .open(&temporary) + .map_err(not_committed)?; + staged.write_all(&replacement).map_err(not_committed)?; if sync_metadata { - staged.sync_all()?; + staged.sync_all().map_err(not_committed)?; } std::fs::rename(&temporary, &file_path)?; Ok(()) @@ -22650,6 +22656,10 @@ mod test { .await .expect_err("directory fsync failure must fail the CAS update"); assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other)); + assert!( + !err.is_conditional_file_not_committed(), + "an error after publication rename must remain commit-ambiguous" + ); assert_eq!( disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH) .await diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index 85d0e3e89..7643908ed 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -520,32 +520,50 @@ impl RootHealRecovery { let _guard = self.mutation.lock().await; let disks = self.disks().await?; let existing = Self::find(&disks, &request.id).await?; - let (disk, expected) = match existing { - Some((disk, bytes)) => (disk, Some(bytes)), - None => { - let disk = disks - .first() - .cloned() - .ok_or_else(|| Error::Other("No local disk available for root heal shutdown recovery".to_string()))?; - (disk, None) - } - }; if request.options.no_lock { return Err(Error::Other("Administrator root heal cannot skip namespace locking".to_string())); } let bytes = serde_json::to_vec(&RootHealIntent::from_request(request)) .map_err(|error| Error::Other(format!("Serialize root heal recovery record: {error}")))?; - match EcstoreDiskAPI::compare_and_update_file( - disk.as_ref(), - RUSTFS_META_BUCKET, - &intent_path(&request.id)?, - expected, - Some(bytes.into()), - ) - .await? - { - EcstoreConditionalFileUpdate::Updated => Ok(()), - _ => Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))), + let path = intent_path(&request.id)?; + if let Some((disk, expected)) = existing { + return match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + &path, + Some(expected), + Some(bytes.into()), + ) + .await? + { + EcstoreConditionalFileUpdate::Updated => Ok(()), + _ => Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))), + }; + } + + if disks.is_empty() { + return Err(Error::Other("No local disk available for root heal shutdown recovery".to_string())); + } + let mut last_not_committed = None; + for disk in &disks { + match EcstoreDiskAPI::compare_and_update_file( + disk.as_ref(), + RUSTFS_META_BUCKET, + &path, + None, + Some(bytes.clone().into()), + ) + .await + { + Ok(EcstoreConditionalFileUpdate::Updated) => return Ok(()), + Ok(_) => return Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))), + Err(error) if error.is_conditional_file_not_committed() => last_not_committed = Some(error), + Err(error) => return Err(Error::Disk(error)), + } + } + match last_not_committed { + Some(error) => Err(Error::Disk(error)), + None => Err(Error::Other("No local disk accepted the root heal recovery record".to_string())), } } diff --git a/crates/heal/src/heal/manager/tests/root_recovery.rs b/crates/heal/src/heal/manager/tests/root_recovery.rs index 0b3e4e680..88bd55c2a 100644 --- a/crates/heal/src/heal/manager/tests/root_recovery.rs +++ b/crates/heal/src/heal/manager/tests/root_recovery.rs @@ -17,6 +17,44 @@ use super::*; use crate::heal::RUSTFS_META_BUCKET; use std::collections::HashSet; +#[cfg(unix)] +struct RestoreDirectoryMode { + path: std::path::PathBuf, + mode: u32, +} + +#[cfg(unix)] +impl RestoreDirectoryMode { + fn read_only(path: std::path::PathBuf) -> Self { + use std::os::unix::fs::PermissionsExt as _; + + let mode = std::fs::metadata(&path) + .expect("metadata directory mode") + .permissions() + .mode(); + std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o555)).expect("make metadata directory read-only"); + Self { path, mode } + } +} + +#[cfg(unix)] +impl Drop for RestoreDirectoryMode { + fn drop(&mut self) { + use std::os::unix::fs::PermissionsExt as _; + + let _ = std::fs::set_permissions(&self.path, std::fs::Permissions::from_mode(self.mode)); + } +} + +#[cfg(unix)] +fn ordered_recovery_disks(first: DiskStore, second: DiskStore) -> (DiskStore, DiskStore) { + if first.endpoint().to_string() <= second.endpoint().to_string() { + (first, second) + } else { + (second, first) + } +} + async fn recovery_disk() -> (TempDir, DiskStore) { let temp = TempDir::new().expect("temporary root recovery disk"); let endpoint = Endpoint::try_from(temp.path().to_string_lossy().as_ref()).expect("disk endpoint"); @@ -78,6 +116,94 @@ fn completed_admin_status(heal_type: &HealType, completed_at: SystemTime) -> Com } } +#[cfg(unix)] +#[tokio::test] +async fn root_recovery_new_intent_skips_prepublication_read_only_owner() { + let (first_temp, first_disk) = recovery_disk().await; + let (second_temp, second_disk) = recovery_disk().await; + let first_endpoint = first_disk.endpoint().to_string(); + let (read_only_disk, writable_disk) = ordered_recovery_disks(first_disk, second_disk); + let read_only_root = if read_only_disk.endpoint().to_string() == first_endpoint { + first_temp.path() + } else { + second_temp.path() + }; + let _restore = RestoreDirectoryMode::read_only(read_only_root.join(RUSTFS_META_BUCKET)); + let manager = recovery_manager(vec![read_only_disk.clone(), writable_disk.clone()]); + let mut request = admin_request(HealType::Object { + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + }); + + let receipt = manager + .submit_heal_request_with_receipt(request.clone()) + .await + .expect("a writable local disk should own the admin heal intent"); + assert_eq!(receipt.result, HealAdmissionResult::Accepted); + let path = format!("root-heal-{}.json", request.id); + assert!(matches!( + read_only_disk.read_all(RUSTFS_META_BUCKET, &path).await, + Err(DiskError::FileNotFound) + )); + assert!(writable_disk.read_all(RUSTFS_META_BUCKET, &path).await.is_ok()); + + request.retry_attempts = 1; + manager + .root_recovery + .persist(&request) + .await + .expect("an existing fallback owner should remain updateable"); + let pending = manager.root_recovery.pending().await.expect("read the single durable owner"); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].id, request.id); + assert_eq!(pending[0].retry_attempts, 1); +} + +#[cfg(unix)] +#[tokio::test] +async fn root_recovery_existing_owner_never_migrates_after_write_rejection() { + let (first_temp, first_disk) = recovery_disk().await; + let (second_temp, second_disk) = recovery_disk().await; + let first_endpoint = first_disk.endpoint().to_string(); + let (owner_disk, alternate_disk) = ordered_recovery_disks(first_disk, second_disk); + let owner_root = if owner_disk.endpoint().to_string() == first_endpoint { + first_temp.path() + } else { + second_temp.path() + }; + let manager = recovery_manager(vec![owner_disk.clone(), alternate_disk.clone()]); + let mut request = root_request(); + manager + .root_recovery + .persist(&request) + .await + .expect("create the canonical owner"); + let path = format!("root-heal-{}.json", request.id); + let committed = owner_disk + .read_all(RUSTFS_META_BUCKET, &path) + .await + .expect("canonical owner bytes"); + let _restore = RestoreDirectoryMode::read_only(owner_root.join(RUSTFS_META_BUCKET)); + + request.retry_attempts = 1; + assert!( + manager.root_recovery.persist(&request).await.is_err(), + "an existing owner write rejection must fail closed" + ); + assert_eq!( + owner_disk + .read_all(RUSTFS_META_BUCKET, &path) + .await + .expect("original owner remains"), + committed + ); + assert!(matches!( + alternate_disk.read_all(RUSTFS_META_BUCKET, &path).await, + Err(DiskError::FileNotFound) + )); +} + async fn active_root(manager: &HealManager, request: HealRequest) -> Arc { let task = Arc::new(HealTask::from_request(request, manager.storage.clone())); *task.status.write().await = HealTaskStatus::Running;