fix(storage): make acknowledged writes durable across power loss (#4221)

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<uuid::Uuid>, Option<Bytes>), std::io::Error>((old_data_dir, has_dst_buf))
             })
             .await

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<Path>) -> 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<Path>) -> 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<Path>) -> 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<Path>) -> 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);
+    }
 }

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<String, String> = 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());
+        });
+    }
+}

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::<Vec<String>>(),
+            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!(

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<T> = EcstoreDiskResult<T>;
 pub(crate) type DiskStore = EcstoreDiskStore;

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",