gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-11 21:39:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(heal): select writable recovery intent owner (#7668)

Browse Source
This commit is contained in:
cxymds
2026-09-11 22:38:23 +08:00
committed by GitHub
parent c8ccc1e198
commit c478a392e7
4 changed files with 233 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/error.rs
+54
View File
@@ -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<DiskError> {
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::<ConditionalFileNotCommittedError>().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<tokio::task::JoinError> 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);
crates/ecstore/src/disk/local.rs
+14 -4
View File
@@ -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
crates/heal/src/heal/manager/root_recovery.rs
+39 -21
View File
@@ -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())),
}
}
crates/heal/src/heal/manager/tests/root_recovery.rs
+126
View File
@@ -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<HealTask> {
let task = Arc::new(HealTask::from_request(request, manager.storage.clone()));
*task.status.write().await = HealTaskStatus::Running;