FLUSHDB must fail if a data file cannot be unlinked

crates/engine/src/log/mod.rs

@@ -991,13 +991,8 @@ impl NamespaceLog {
     }
 
     async fn flush_inner(&self) -> Result<()> {
-        // Drop file handles inside their cells.
-        self.sealed.borrow_mut().clear();
-        self.index.borrow_mut().clear();
-
-        // Unlink all data-* files (including current active — it's still held
-        // through the Rc until we replace it; on Linux the inode's blocks stay
-        // alive for the open handle and are freed when we drop the Rc).
+        // Unlink first. Clearing the index before a failed unlink returns Ok
+        // leaves RAM empty and disk intact — reopen resurrects the namespace.
         let to_unlink: Vec<PathBuf> = match std::fs::read_dir(&self.dir) {
             Ok(entries) => entries
                 .flatten()
@@ -1015,15 +1010,20 @@ impl NamespaceLog {
             .iter()
             .map(|p| monoio::fs::remove_file(p.clone()))
             .collect();
-        for (path, res) in to_unlink.iter().zip(join_all(unlink_futures).await) {
+        for (_path, res) in to_unlink.iter().zip(join_all(unlink_futures).await) {
             if let Err(e) = res {
-                warn!(path = %path.display(), error = %e, "failed to unlink data file during flush");
+                return Err(EngineError::Io { source: e });
             }
         }
 
+        self.sealed.borrow_mut().clear();
+        self.index.borrow_mut().clear();
+
         let path = self.dir.join(data_filename(0));
         let new_active = Rc::new(LogFile::open_rw(path, 0).await?);
-        sync_dir(&self.dir).await; // make the recreated file's directory entry durable
+        // open_rw does not truncate; a leftover inode would replay on reopen.
+        new_active.truncate_to(0).await?;
+        sync_dir(&self.dir).await;
         *self.active.borrow_mut() = new_active;
         self.unsynced_bytes.set(0);
         Ok(())
@@ -3012,6 +3012,51 @@ mod enospc_recovery_tests {
     }
 }
 
+#[cfg(test)]
+mod flush_unlink_tests {
+    use super::*;
+    use crate::log::config::LogConfig;
+    use bytes::Bytes;
+    use tempfile::TempDir;
+
+    fn run<F: std::future::Future>(f: F) -> F::Output {
+        monoio::RuntimeBuilder::<monoio::FusionDriver>::new()
+            .enable_timer()
+            .build()
+            .expect("monoio runtime")
+            .block_on(f)
+    }
+
+    /// A leftover name that cannot be unlinked (directory named like a
+    /// data file) must fail the flush. Clearing the index first then
+    /// returning Ok leaves RAM empty and disk intact — reopen resurrects.
+    #[test]
+    fn flush_must_fail_if_a_data_file_cannot_be_unlinked() {
+        run(async {
+            let dir = TempDir::new().unwrap();
+            let path = dir.path().to_path_buf();
+            let cfg = LogConfig {
+                rotate_threshold: 1 << 40,
+                fanout: 8,
+                value_sep_threshold: 1 << 20,
+            };
+            let log = NamespaceLog::open(path.clone(), cfg).await.unwrap();
+            log.put_full(Bytes::from_static(b"k"), b"v", &[], None)
+                .await
+                .unwrap();
+
+            std::fs::create_dir(path.join("data-9999999999.log")).unwrap();
+
+            let r = log.flush().await;
+            assert!(r.is_err(), "flush must fail-stop on unlink, got {r:?}");
+            assert!(
+                log.index.borrow().get(b"k").is_some(),
+                "failed flush must not drop the live index"
+            );
+        });
+    }
+}
+
 #[cfg(test)]
 mod fd_footprint_tests {
     use super::*;