From 73ca7572d8ed974b497ed3f204ff8d585a336f98 Mon Sep 17 00:00:00 2001
From: water <672684719@qq.com>
Date: Thu, 27 Aug 2026 22:13:43 +0800
Subject: [PATCH] fix: remove recycled log files leaked by unclean shutdown
 (#407)

`purge_to` keeps the recycled append files under their original
`.raftlog` name on disk (rename only happens in `SinglePipe::drop`).
If the process exits without running `Drop` (SIGKILL / OOM / panic=abort),
those files are re-scanned as regular append files on restart and form a
hole before the active range. The hole check removed them from the
in-memory list but never deleted the physical files, so they leaked on
disk forever and the leak accumulated across unclean restarts.

Now the invalid files drained by the hole check are also deleted from
disk. Recycled (reserved) files are still skipped, matching the existing
metadata-cleanup behavior.

Signed-off-by: waterWang <waterWang@users.noreply.github.com>
---
 src/engine.rs                     | 71 +++++++++++++++++++++++++++++++
 src/file_pipe_log/pipe_builder.rs | 26 ++++++++++-
 2 files changed, 96 insertions(+), 1 deletion(-)

diff --git a/src/engine.rs b/src/engine.rs
index 1f1d55fd..87564997 100644
--- a/src/engine.rs
+++ b/src/engine.rs
@@ -2265,6 +2265,77 @@ pub(crate) mod tests {
         assert!(start_2 < start_3);
     }
 
+    #[test]
+    fn test_remove_recycled_files_left_by_unclean_shutdown() {
+        let dir = tempfile::Builder::new()
+            .prefix("test_unclean_recycled")
+            .tempdir()
+            .unwrap();
+        let entry_data = vec![b'x'; 16];
+        let cfg = Config {
+            dir: dir.path().to_str().unwrap().to_owned(),
+            target_file_size: ReadableSize(1),
+            purge_threshold: ReadableSize(50),
+            format_version: Version::V2,
+            enable_log_recycle: true,
+            prefill_for_recycle: true,
+            ..Default::default()
+        };
+        let fs = Arc::new(DefaultFileSystem);
+        let engine = RaftLogEngine::open_with_file_system(cfg, fs.clone()).unwrap();
+        for rid in 1..=10 {
+            engine.append(rid, 1, 11, Some(&entry_data));
+        }
+        for rid in 1..=10 {
+            engine.clean(rid);
+        }
+        engine.purge_manager.must_rewrite_append_queue(None, None);
+        // A clean shutdown renames recycled files to `*.raftlog.reserved`. Simulate
+        // an unclean shutdown (SIGKILL / OOM, `SinglePipe::drop` never runs) by
+        // renaming them back to plain append files, reproducing the leaked-files
+        // on-disk state reported in issue #407.
+        let engine = engine.reopen();
+        let dir_path = dir.path();
+        let mut reserved_names = Vec::new();
+        for entry in std::fs::read_dir(dir_path).unwrap() {
+            let name = entry.unwrap().file_name().into_string().unwrap();
+            if name.ends_with(".raftlog.reserved") {
+                reserved_names.push(name);
+            }
+        }
+        assert!(!reserved_names.is_empty());
+        let ghost_names: Vec<String> = reserved_names
+            .iter()
+            .map(|n| n.trim_end_matches(".reserved").to_owned())
+            .collect();
+        for (name, ghost) in reserved_names.iter().zip(ghost_names.iter()) {
+            std::fs::rename(dir_path.join(name), dir_path.join(ghost)).unwrap();
+        }
+        // Reopen: the leaked (ghost) files form a hole before the active range and
+        // must be physically removed instead of leaking on disk forever.
+        let engine = engine.reopen();
+        let disk_append_count = std::fs::read_dir(dir_path)
+            .unwrap()
+            .filter(|e| {
+                e.as_ref()
+                    .unwrap()
+                    .file_name()
+                    .into_string()
+                    .unwrap()
+                    .ends_with(".raftlog")
+            })
+            .count();
+        // Every plain `.raftlog` file on disk must belong to the active span; any
+        // leaked recycled file would show up as a stray file and break the equality.
+        assert_eq!(
+            engine.file_count(Some(LogQueue::Append)),
+            disk_append_count,
+            "leaked recycled files should be removed on restart"
+        );
+        let (start, end) = engine.file_span(LogQueue::Append);
+        assert!(start <= end);
+    }
+
     #[test]
     fn test_simple_write_perf_context() {
         let dir = tempfile::Builder::new()
diff --git a/src/file_pipe_log/pipe_builder.rs b/src/file_pipe_log/pipe_builder.rs
index 96af7dcb..b154afc1 100644
--- a/src/file_pipe_log/pipe_builder.rs
+++ b/src/file_pipe_log/pipe_builder.rs
@@ -177,7 +177,31 @@ impl<F: FileSystem> DualPipesBuilder<F> {
                     invalid_idx = i + 1;
                 }
             }
-            files.drain(..invalid_idx);
+            if invalid_idx > 0 {
+                // Files before the first hole are unreachable and are dropped from the
+                // in-memory list. Physically remove them as well, otherwise they leak
+                // on disk forever. This can happen when a process exits without running
+                // `SinglePipe::drop` (e.g. SIGKILL / OOM), leaving recycled append files
+                // that still use their original `.raftlog` name: on restart they are
+                // re-scanned as regular append files and form a hole before the active
+                // range, but only the metadata (not the log file) was cleaned before.
+                let invalid_files = files.drain(..invalid_idx).collect::<Vec<_>>();
+                if !is_recycled_file {
+                    // Collect the paths first so the file handles (and thus the files)
+                    // can be closed before deletion, which matters on Windows.
+                    let mut paths = Vec::with_capacity(invalid_files.len());
+                    for f in &invalid_files {
+                        let file_id = FileId { queue, seq: f.seq };
+                        paths.push(file_id.build_file_path(&self.dirs[f.path_id]));
+                    }
+                    drop(invalid_files);
+                    for path in paths {
+                        if let Err(e) = self.file_system.delete(&path) {
+                            error!("failed to delete leaked log file {}: {e}.", path.display());
+                        }
+                    }
+                }
+            }
             // Try to cleanup stale metadata left by the previous version.
             if files.is_empty() || is_recycled_file {
                 continue;