diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 1150f989f..56b0906f1 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -229,6 +229,13 @@ const ENV_RUSTFS_OBJECT_MMAP_READ_METHOD: &str = "RUSTFS_OBJECT_MMAP_READ_METHOD const RUSTFS_OBJECT_MMAP_READ_METHOD_MMAP_COPY: &str = "mmap_copy"; const RUSTFS_OBJECT_MMAP_READ_METHOD_DIRECT_READ_COPY: &str = "direct_read_copy"; +/// Fsync writes and commit-point renames so acknowledged data survives power loss. +/// Disabling trades durability for latency: data acknowledged with 200 OK may only +/// live in the page cache and be lost on a whole-node power failure. +/// Default: true. +const ENV_RUSTFS_DRIVE_SYNC_ENABLE: &str = "RUSTFS_DRIVE_SYNC_ENABLE"; +const DEFAULT_RUSTFS_DRIVE_SYNC_ENABLE: bool = true; + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum LocalReadCopyMethod { MmapCopy, @@ -240,6 +247,11 @@ fn is_direct_io_read_enabled() -> bool { rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE) } +/// Check if durable (fsync) writes are enabled. +pub(crate) fn drive_sync_enabled() -> bool { + rustfs_utils::get_env_bool(ENV_RUSTFS_DRIVE_SYNC_ENABLE, DEFAULT_RUSTFS_DRIVE_SYNC_ENABLE) +} + /// Get the O_DIRECT read threshold size. fn get_direct_io_read_threshold() -> usize { rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD) @@ -1646,6 +1658,13 @@ impl LocalDisk { rename_all(tmp_file_path, &file_path, volume_dir).await?; + if sync + && drive_sync_enabled() + && let Some(parent) = file_path.parent() + { + os::fsync_dir(parent).await.map_err(to_file_error)?; + } + Ok(()) } @@ -1682,10 +1701,20 @@ impl LocalDisk { skip_parent }; + let sync = sync && drive_sync_enabled(); + match data { InternalBuf::Ref(buf) => { let mut f = self.open_file(file_path, O_CREATE | O_WRONLY | O_TRUNC, skip_parent).await?; f.write_all(buf).await.map_err(to_file_error)?; + if sync { + f.sync_data().await.map_err(to_file_error)?; + // Persist the directory entry too, so a freshly created file + // survives power loss along with its contents. + if let Some(parent) = file_path.parent() { + os::fsync_dir(parent).await.map_err(to_file_error)?; + } + } } InternalBuf::Owned(buf) => { let path = file_path.to_path_buf(); @@ -1700,10 +1729,16 @@ impl LocalDisk { .create(true) .write(true) .truncate(true) - .open(path) + .open(&path) .map_err(to_file_error)?; std::io::Write::write_all(&mut f, buf.as_ref()).map_err(to_file_error)?; + if sync { + f.sync_data().map_err(to_file_error)?; + if let Some(parent) = path.parent() { + os::fsync_dir_std(parent).map_err(to_file_error)?; + } + } Ok::<(), std::io::Error>(()) }) @@ -1712,9 +1747,6 @@ impl LocalDisk { } } - // Keep existing durability contract: this path intentionally ignores `sync`. - let _ = sync; - Ok(()) } @@ -2770,8 +2802,23 @@ impl DiskAPI for LocalDisk { } } + // UploadPart is acknowledged once this rename lands, so the part data and + // its directory entry must be durable before we return. + let sync = drive_sync_enabled(); + if sync && !src_is_dir { + let src = src_file_path.clone(); + tokio::task::spawn_blocking(move || std::fs::File::open(&src)?.sync_data()) + .await + .map_err(DiskError::from)? + .map_err(to_file_error)?; + } + rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?; + if sync && let Some(parent) = dst_file_path.parent() { + os::fsync_dir(parent).await.map_err(to_file_error)?; + } + let dst_meta = lstat_std(&dst_file_path).map_err(|e| -> DiskError { to_file_error(e).into() })?; if src_is_dir != dst_meta.is_dir() { warn!( @@ -3593,6 +3640,19 @@ impl DiskAPI for LocalDisk { ) .await?; + // Make shard files durable before the commit rename: once rename_data + // succeeds the write is acknowledged, so data must not live only in the + // page cache. Multipart parts were already synced during rename_part, so + // their fdatasync here is a cheap no-op. A missing source dir is left for + // the rename below to report through the existing rollback path. + if drive_sync_enabled() + && let Some((src_data_path, _)) = has_data_dir_path.as_ref() + && let Err(err) = os::sync_dir_files(src_data_path).await + && err.kind() != ErrorKind::NotFound + { + return Err(to_file_error(err).into()); + } + if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() && let Err(err) = rename_all(src_data_path, dst_data_path, &skip_parent).await { @@ -3667,6 +3727,15 @@ impl DiskAPI for LocalDisk { return Err(err); } + // Persist the directory entries for both the data dir and xl.meta renames; + // without this the commit itself can vanish on power loss. + if drive_sync_enabled() + && let Some(parent) = dst_file_path.parent() + && let Err(err) = os::fsync_dir(parent).await + { + return Err(to_file_error(err).into()); + } + if let Some(src_file_path_parent) = src_file_path.parent() { if src_volume != super::RUSTFS_META_MULTIPART_BUCKET { let _ = remove_std(src_file_path_parent); @@ -3719,12 +3788,16 @@ impl DiskAPI for LocalDisk { if let Some(parent) = src.parent() { std::fs::create_dir_all(parent)?; } + let sync = drive_sync_enabled(); let mut f = std::fs::OpenOptions::new() .create(true) .write(true) .truncate(true) .open(&src)?; std::io::Write::write_all(&mut f, &new_buf)?; + if sync { + f.sync_data()?; + } if let Some(old_dir) = old_data_dir.as_ref() && let Some(ref buf) = has_dst_buf && let Some(dst_parent) = dst.parent() @@ -3749,6 +3822,11 @@ impl DiskAPI for LocalDisk { Err(err) => Err(to_file_error(err)), }?; + // Persist the commit rename's directory entry across power loss. + if sync && let Some(dst_parent) = dst.parent() { + os::fsync_dir_std(dst_parent)?; + } + Ok::<(Option, Option), std::io::Error>((old_data_dir, has_dst_buf)) }) .await diff --git a/crates/ecstore/src/disk/os.rs b/crates/ecstore/src/disk/os.rs index 793295bfd..b8b19541d 100644 --- a/crates/ecstore/src/disk/os.rs +++ b/crates/ecstore/src/disk/os.rs @@ -63,6 +63,43 @@ pub fn check_path_length(path_name: &str) -> Result<()> { Ok(()) } +/// Fsync a directory so recently created or renamed entries survive power loss. +/// No-op on non-Unix platforms where directories cannot be opened for syncing. +pub fn fsync_dir_std(dir: impl AsRef) -> io::Result<()> { + #[cfg(unix)] + { + std::fs::File::open(dir.as_ref())?.sync_all()?; + } + #[cfg(not(unix))] + let _ = dir; + Ok(()) +} + +/// Async wrapper around [`fsync_dir_std`]; runs the blocking fsync off the runtime. +pub async fn fsync_dir(dir: impl AsRef) -> io::Result<()> { + let dir = dir.as_ref().to_path_buf(); + tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await? +} + +/// Fdatasync every regular file directly inside `dir`, then fsync the directory +/// itself. Used at commit points so erasure shard files written through the page +/// cache are durable before their directory is renamed into its final location. +pub fn sync_dir_files_std(dir: impl AsRef) -> io::Result<()> { + for entry in std::fs::read_dir(dir.as_ref())? { + let entry = entry?; + if entry.file_type()?.is_file() { + std::fs::File::open(entry.path())?.sync_data()?; + } + } + fsync_dir_std(dir) +} + +/// Async wrapper around [`sync_dir_files_std`]; runs the blocking syncs off the runtime. +pub async fn sync_dir_files(dir: impl AsRef) -> io::Result<()> { + let dir = dir.as_ref().to_path_buf(); + tokio::task::spawn_blocking(move || sync_dir_files_std(dir)).await? +} + /// Check if the given disk path is the root disk. /// On Windows, always return false. /// On Unix, compare the disk paths. @@ -304,4 +341,30 @@ mod tests { assert!(!dst.exists()); } + + #[tokio::test] + async fn fsync_dir_succeeds_on_directory() { + let temp_dir = tempdir().expect("create temp dir"); + + fsync_dir(temp_dir.path()).await.expect("fsync dir must succeed"); + } + + #[tokio::test] + async fn sync_dir_files_syncs_regular_files_and_dir() { + let temp_dir = tempdir().expect("create temp dir"); + std::fs::write(temp_dir.path().join("part.1"), b"shard-one").expect("write part.1"); + std::fs::write(temp_dir.path().join("part.2"), b"shard-two").expect("write part.2"); + std::fs::create_dir(temp_dir.path().join("subdir")).expect("create subdir"); + + sync_dir_files(temp_dir.path()).await.expect("sync dir files must succeed"); + } + + #[tokio::test] + async fn sync_dir_files_missing_dir_returns_not_found() { + let temp_dir = tempdir().expect("create temp dir"); + let missing = temp_dir.path().join("missing"); + + let err = sync_dir_files(&missing).await.expect_err("missing dir must fail"); + assert_eq!(err.kind(), io::ErrorKind::NotFound); + } } diff --git a/crates/ecstore/src/set_disk/write.rs b/crates/ecstore/src/set_disk/write.rs index b4803c4a9..365c199b1 100644 --- a/crates/ecstore/src/set_disk/write.rs +++ b/crates/ecstore/src/set_disk/write.rs @@ -14,6 +14,16 @@ use super::*; +/// Grace window during which a recently modified object is never deleted as +/// dangling. 0 disables the grace window. +const ENV_HEAL_DANGLING_DELETE_GRACE_SECS: &str = "RUSTFS_HEAL_DANGLING_DELETE_GRACE_SECS"; +const DEFAULT_HEAL_DANGLING_DELETE_GRACE_SECS: u64 = 3600; + +fn dangling_delete_grace() -> time::Duration { + let secs = rustfs_utils::get_env_u64(ENV_HEAL_DANGLING_DELETE_GRACE_SECS, DEFAULT_HEAL_DANGLING_DELETE_GRACE_SECS); + time::Duration::seconds(i64::try_from(secs).unwrap_or(i64::MAX)) +} + /// Result of scanning one disk's copy of a directory prefix while deciding /// whether an orphan (metadata-less) directory tree can be safely purged. enum OrphanDirScan { @@ -582,6 +592,27 @@ impl SetDisks { return Err(DiskError::ErasureReadQuorum); } + // Recently written objects get a grace window before dangling cleanup: after + // an unclean shutdown some disks may still be catching up (or carry writes + // that were never made durable), and deleting the surviving shards right away + // turns a partial loss into a total one. Skip deletion and leave the object + // for a later heal/scanner pass to re-evaluate. + if m.is_valid() + && let Some(mod_time) = m.mod_time + { + let grace = dangling_delete_grace(); + if !grace.is_zero() && OffsetDateTime::now_utc() - mod_time < grace { + info!( + bucket = bucket, + object = object, + mod_time = %mod_time, + grace_secs = grace.whole_seconds(), + "skipping dangling-object deletion within grace window" + ); + return Err(DiskError::ErasureReadQuorum); + } + } + let mut tags: HashMap = HashMap::new(); tags.insert("set".to_string(), self.set_index.to_string()); tags.insert("pool".to_string(), self.pool_index.to_string()); @@ -886,3 +917,25 @@ impl SetDisks { None } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn dangling_delete_grace_defaults_to_one_hour() { + temp_env::with_var(ENV_HEAL_DANGLING_DELETE_GRACE_SECS, None::<&str>, || { + assert_eq!(dangling_delete_grace(), time::Duration::seconds(3600)); + }); + } + + #[test] + fn dangling_delete_grace_env_override_and_disable() { + temp_env::with_var(ENV_HEAL_DANGLING_DELETE_GRACE_SECS, Some("120"), || { + assert_eq!(dangling_delete_grace(), time::Duration::seconds(120)); + }); + temp_env::with_var(ENV_HEAL_DANGLING_DELETE_GRACE_SECS, Some("0"), || { + assert!(dangling_delete_grace().is_zero()); + }); + } +} diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index b972316b5..47bc6fdae 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -48,6 +48,7 @@ const EVENT_HEAL_QUEUE_ADMISSION: &str = "heal_queue_admission"; const EVENT_HEAL_MAINLINE_THROTTLE: &str = "heal_mainline_throttle"; const EVENT_HEAL_SCHEDULER_STATE: &str = "heal_scheduler_state"; const EVENT_HEAL_QUEUE_STATE: &str = "heal_queue_state"; +const EVENT_HEAL_UNCLEAN_SHUTDOWN: &str = "heal_unclean_shutdown"; const MAX_RECOVERABLE_HEAL_RETRIES: u32 = 3; const MAX_RECOVERABLE_HEAL_RETRY_DELAY: Duration = Duration::from_secs(30); @@ -1251,6 +1252,10 @@ impl HealManager { ); } + // Detect a previous unclean shutdown (crash/power loss) and proactively + // verify all local erasure sets instead of waiting for the periodic scanner. + self.process_unclean_shutdown().await; + info!( target: "rustfs::heal::manager", event = EVENT_HEAL_MANAGER_STATE, @@ -1262,6 +1267,117 @@ impl HealManager { Ok(()) } + /// Detect whether the previous run ended without a clean shutdown and, if so, + /// enqueue a full erasure-set heal for every local set. Also (re)writes the + /// marker for the current run; [`super::clear_unclean_shutdown_markers`] + /// removes it again during graceful shutdown. Best-effort: failures only log. + async fn process_unclean_shutdown(&self) { + let mut unclean = false; + let mut set_disk_ids = HashSet::new(); + + { + let local_disk_map = local_disk_map_read().await; + for disk in local_disk_map.values().flatten() { + let endpoint = disk.endpoint(); + match disk + .read_all(super::RUSTFS_META_BUCKET, super::UNCLEAN_SHUTDOWN_MARKER_PATH) + .await + { + Ok(_) => unclean = true, + Err(DiskError::FileNotFound) | Err(DiskError::VolumeNotFound) => {} + Err(err) => { + debug!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + endpoint = %endpoint, + error = ?err, + "Unclean-shutdown marker check failed" + ); + } + } + + let marker = SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs().to_string()) + .unwrap_or_default(); + if let Err(err) = disk + .write_all(super::RUSTFS_META_BUCKET, super::UNCLEAN_SHUTDOWN_MARKER_PATH, marker.into()) + .await + { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + endpoint = %endpoint, + error = ?err, + "Unclean-shutdown marker write failed" + ); + } + + if let Some(set_disk_id) = crate::heal::utils::format_set_disk_id_from_i32(endpoint.pool_idx, endpoint.set_idx) { + set_disk_ids.insert(set_disk_id); + } + } + } + + if !unclean || set_disk_ids.is_empty() { + return; + } + + info!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + set_count = set_disk_ids.len(), + "Unclean shutdown detected; scheduling erasure-set heal for local sets" + ); + + let buckets = match self.storage.list_buckets().await { + Ok(buckets) => buckets.iter().map(|b| b.name.clone()).collect::>(), + Err(err) => { + error!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + error = %err, + "Unclean-shutdown heal skipped: bucket listing failed" + ); + return; + } + }; + + for set_disk_id in set_disk_ids { + let mut req = HealRequest::new( + HealType::ErasureSet { + buckets: buckets.clone(), + set_disk_id: set_disk_id.clone(), + }, + HealOptions { + timeout: None, + ..HealOptions::default() + }, + HealPriority::Low, + ); + req.source = HealRequestSource::AutoHeal; + if let Err(err) = self.submit_heal_request(req).await { + warn!( + target: "rustfs::heal::manager", + event = EVENT_HEAL_UNCLEAN_SHUTDOWN, + component = LOG_COMPONENT_HEAL, + subsystem = LOG_SUBSYSTEM_MANAGER, + set_disk_id, + error = %err, + "Unclean-shutdown heal enqueue failed" + ); + } + } + } + /// Stop HealManager pub async fn stop(&self) -> Result<()> { info!( diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 0be677d2b..90f4da057 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -40,6 +40,35 @@ pub(crate) const DATA_USAGE_CACHE_NAME: &str = ECSTORE_DATA_USAGE_CACHE_NAME; pub(crate) const BUCKET_META_PREFIX: &str = ECSTORE_BUCKET_META_PREFIX; pub(crate) const RUSTFS_META_BUCKET: &str = ECSTORE_RUSTFS_META_BUCKET; +/// Marker written to every local disk while the process runs; removed by +/// [`clear_unclean_shutdown_markers`] on graceful shutdown. Finding it at +/// startup means the previous run crashed or lost power, so the heal manager +/// proactively re-verifies all local erasure sets. +pub(crate) const UNCLEAN_SHUTDOWN_MARKER_PATH: &str = "unclean-shutdown"; + +/// Remove the unclean-shutdown markers from all local disks. Call at the end of +/// a graceful shutdown, after the data plane has stopped accepting writes. +pub async fn clear_unclean_shutdown_markers() { + let local_disk_map = local_disk_map_read().await; + for disk in local_disk_map.values().flatten() { + if let Err(err) = EcstoreDiskAPI::delete( + disk.as_ref(), + RUSTFS_META_BUCKET, + UNCLEAN_SHUTDOWN_MARKER_PATH, + EcstoreDeleteOptions::default(), + ) + .await + && err != DiskError::FileNotFound + { + tracing::warn!( + endpoint = %EcstoreDiskAPI::endpoint(disk.as_ref()), + error = ?err, + "failed to clear unclean-shutdown marker" + ); + } + } +} + pub(crate) type DiskError = EcstoreDiskError; pub(crate) type DiskResult = EcstoreDiskResult; pub(crate) type DiskStore = EcstoreDiskStore; diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index 741304560..051ce9c91 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/rustfs/src/startup_shutdown.rs @@ -201,6 +201,9 @@ pub(crate) async fn run_startup_shutdown_sequence( console_shutdown_handle.shutdown().await; } shutdown_optional_runtime_services(optional_runtime_shutdowns).await; + // The data plane is drained: record this shutdown as clean so the next + // startup skips the unclean-restart erasure-set heal. + rustfs_heal::heal::clear_unclean_shutdown_markers().await; state_manager.update(ServiceState::Stopped); info!( target: "rustfs::main::handle_shutdown",