fix(storage): reclaim warm-tier staging directories instead of leaking them (#435)

CHANGELOG.md

@@ -254,6 +254,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   never showed `flags=b` — and would have been treated as idle by the new
   timeout sweep. The flag is now set for the duration of a blocking command
   on both runtimes.
+- **Warm-tier transitions leaked their staging directory on every failure
+  (#435).** `transition_to_warm` has ~10 fallible steps between creating
+  `.segment-{id}.staging` and renaming it to its final name, and every early
+  return in that stretch abandoned the whole directory. Found on a live
+  instance: a 27 GB data directory against 2.53 GB of `used_memory`, 20 GB of
+  which was 14,499 orphaned staging directories versus 174 MB in the 94 real
+  segments — 99% of the vector store was abandoned scratch, written in a single
+  three-minute window and already survived a restart. The dot prefix kept them
+  out of `ls` and out of every `vectors/*` glob, so nothing surfaced them. A
+  `Drop` guard now covers every early return, and `sweep_orphan_staging` runs
+  at recovery for orphans no in-process guard can catch (a `kill -9` between
+  the manifest commit and the rename, or anything left by an older build). The
+  sweep is safe by construction: staging paths are produced in exactly one
+  place and read by nothing — every reader opens the final `segment-{id}` name.
+  `transition_to_warm` also removes a stale staging directory for the same id
+  before creating it, so a retry cannot inherit a previous attempt's partial
+  files.
 
 ## [0.8.4] — 2026-07-29
 

src/persistence/recovery.rs

@@ -263,6 +263,15 @@ pub fn recover_shard_v3_pitr(
     if manifest_path.exists() {
         if let Ok(manifest) = ShardManifest::open(&manifest_path) {
             let vectors_dir = shard_dir.join("vectors");
+            // #435: reclaim `.segment-*.staging` leftovers before scanning.
+            // A warm transition that died between its manifest commit and its
+            // rename leaves a fully-written staging dir that nothing reads and
+            // nothing removed — a live instance accumulated 14,499 of them
+            // holding 20 GB against 174 MB of real segments, invisible to `ls`
+            // and to any `vectors/*` glob because of the dot prefix. The
+            // in-process guard covers new failures; this covers orphans from a
+            // kill -9 or an older build.
+            crate::storage::tiered::warm_tier::sweep_orphan_staging(&vectors_dir);
             for entry in manifest.files() {
                 if entry.tier == StorageTier::Warm
                     && entry.status == FileStatus::Active

src/storage/tiered/warm_tier.rs

@@ -17,6 +17,77 @@ use crate::vector::persistence::warm_segment::{
     write_codes_mpf, write_graph_mpf, write_mvcc_mpf, write_vectors_mpf,
 };
 
+/// Removes a staging directory unless the transition disarms it.
+///
+/// `transition_to_warm` has ~10 fallible steps between creating the staging
+/// directory and renaming it away. Every `?` in that stretch used to leak the
+/// whole directory: a live instance accumulated 14,499 orphans holding 20 GB
+/// against 174 MB of real segments (issue #435). A guard is used rather than
+/// cleanup at each `?` precisely because the leak came from the paths nobody
+/// remembered to annotate.
+struct StagingGuard<'a> {
+    path: &'a Path,
+    armed: bool,
+}
+
+impl<'a> StagingGuard<'a> {
+    fn new(path: &'a Path) -> Self {
+        Self { path, armed: true }
+    }
+
+    /// Called once the directory has been renamed away and must NOT be removed.
+    fn disarm(&mut self) {
+        self.armed = false;
+    }
+}
+
+impl Drop for StagingGuard<'_> {
+    fn drop(&mut self) {
+        if self.armed {
+            // Best-effort: a failure to clean up must not mask the error that
+            // caused the unwind, and the startup sweep is the backstop.
+            let _ = std::fs::remove_dir_all(self.path);
+        }
+    }
+}
+
+/// Remove `.segment-*.staging` leftovers from a shard's `vectors/` directory.
+///
+/// The backstop for orphans a guard could not handle: a `kill -9` between the
+/// manifest commit and the rename, or anything written by a build that predates
+/// the guard. Returns how many directories were removed.
+///
+/// Safe by construction: `.staging` paths are produced in exactly one place
+/// (`transition_to_warm`) and consumed by nothing — every reader and recovery
+/// path opens the final `segment-{id}` name — so a staging directory is
+/// unreachable the moment it is not mid-write. Real `segment-*` directories do
+/// not match the pattern and are never touched.
+pub fn sweep_orphan_staging(vectors_dir: &Path) -> usize {
+    let Ok(entries) = std::fs::read_dir(vectors_dir) else {
+        return 0; // fresh shard, or no vector data — nothing to sweep
+    };
+    let mut removed = 0;
+    for entry in entries.flatten() {
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else { continue };
+        if name.starts_with(".segment-")
+            && name.ends_with(".staging")
+            && entry.path().is_dir()
+            && std::fs::remove_dir_all(entry.path()).is_ok()
+        {
+            removed += 1;
+        }
+    }
+    if removed > 0 {
+        tracing::info!(
+            removed,
+            dir = %vectors_dir.display(),
+            "swept orphaned warm-tier staging directories"
+        );
+    }
+    removed
+}
+
 /// Transition a HOT vector segment to WARM (mmap-backed on disk).
 ///
 /// Protocol:
@@ -48,8 +119,16 @@ pub fn transition_to_warm(
     let staging = vectors_dir.join(format!(".segment-{segment_id}.staging"));
     let final_dir = vectors_dir.join(format!("segment-{segment_id}"));
 
-    // Step 1: Create staging directory
+    // Step 1: Create staging directory. Remove a stale leftover for this same
+    // id first — the sibling writer (`vector/persistence/segment_io.rs`) has
+    // always done this; this path did not, so a retry could inherit a previous
+    // attempt's partial files.
+    if staging.exists() {
+        std::fs::remove_dir_all(&staging)?;
+    }
     std::fs::create_dir_all(&staging)?;
+    // Every early return from here to the rename removes the directory (#435).
+    let mut staging_guard = StagingGuard::new(&staging);
 
     // Step 2: Write .mpf files to staging
     write_codes_mpf(&staging.join("codes.mpf"), file_id, codes_data)?;
@@ -114,8 +193,10 @@ pub fn transition_to_warm(
     manifest.add_file(entry);
     manifest.commit()?;
 
-    // Step 6: Rename staging -> final
+    // Step 6: Rename staging -> final. The directory now lives under its final
+    // name, so the guard must not remove it.
     std::fs::rename(&staging, &final_dir)?;
+    staging_guard.disarm();
 
     // Step 7: Fsync parent directory
     fsync_directory(&vectors_dir)?;
@@ -129,6 +210,108 @@ mod tests {
     use super::*;
     use crate::persistence::manifest::ShardManifest;
 
+    /// A failed transition must not leave its staging directory behind.
+    ///
+    /// Found in production: 14,499 orphaned `.segment-*.staging` dirs holding
+    /// 20 GB, against 174 MB of real segments — 99% of the vector store was
+    /// abandoned scratch, all written in one 3-minute window and never
+    /// reclaimed across a restart. Every early return between
+    /// `create_dir_all(&staging)` and the rename used to leak the directory.
+    /// See issue #435.
+    ///
+    /// The failure is forced at the RENAME, which is where production died
+    /// (every orphan was fully written and fsynced). A non-empty `segment-{id}`
+    /// makes `fs::rename` fail with ENOTEMPTY — guaranteed, unlike unlinking
+    /// the manifest, whose already-open fd keeps accepting writes.
+    ///
+    /// Asserting `is_err()` is load-bearing: "no staging dir remains" is also
+    /// true of a SUCCESSFUL transition (the rename moves it), so without
+    /// pinning the failure path this test would pass with the guard deleted.
+    #[test]
+    fn failed_transition_leaves_no_staging_dir() {
+        let tmp = tempfile::tempdir().unwrap();
+        let shard_dir = tmp.path().join("shard-0");
+        let vectors = shard_dir.join("vectors");
+        std::fs::create_dir_all(&vectors).unwrap();
+
+        // Block the rename: a non-empty destination cannot be replaced.
+        std::fs::create_dir_all(vectors.join("segment-7")).unwrap();
+        std::fs::write(vectors.join("segment-7/occupied"), b"in the way").unwrap();
+
+        let manifest_path = shard_dir.join("shard-0.manifest");
+        let mut manifest = ShardManifest::create(&manifest_path).unwrap();
+
+        let result = transition_to_warm(
+            &shard_dir,
+            7,
+            7,
+            b"codes",
+            b"graph",
+            Some(b"vectors"),
+            b"mvcc",
+            &mut manifest,
+            None,
+        );
+
+        assert!(
+            result.is_err(),
+            "test must exercise the FAILURE path — a success would remove the \
+             staging dir by renaming it, making the assertion below vacuous"
+        );
+        let leaked: Vec<_> = std::fs::read_dir(&vectors)
+            .unwrap()
+            .filter_map(|e| e.ok())
+            .map(|e| e.file_name().to_string_lossy().into_owned())
+            .filter(|n| n.ends_with(".staging"))
+            .collect();
+        assert!(
+            leaked.is_empty(),
+            "a failed warm transition must clean up after itself; leaked {leaked:?}"
+        );
+        assert_eq!(
+            std::fs::read(vectors.join("segment-7/occupied")).unwrap(),
+            b"in the way",
+            "cleanup must not touch the pre-existing segment directory"
+        );
+    }
+
+    /// Orphans from older builds (or a kill -9 between commit and rename) must
+    /// be reclaimable at startup — otherwise 20 GB sits there until someone
+    /// notices `du` disagreeing with a glob.
+    #[test]
+    fn startup_sweep_removes_orphan_staging_dirs() {
+        let tmp = tempfile::tempdir().unwrap();
+        let vectors = tmp.path().join("vectors");
+        std::fs::create_dir_all(vectors.join(".segment-1.staging")).unwrap();
+        std::fs::create_dir_all(vectors.join(".segment-2.staging")).unwrap();
+        std::fs::write(vectors.join(".segment-1.staging/codes.mpf"), b"x").unwrap();
+        // A real segment must be left strictly alone.
+        std::fs::create_dir_all(vectors.join("segment-1")).unwrap();
+        std::fs::write(vectors.join("segment-1/codes.mpf"), b"keep").unwrap();
+
+        let removed = sweep_orphan_staging(&vectors);
+
+        assert_eq!(removed, 2, "both orphans must be swept");
+        assert!(
+            vectors.join("segment-1").is_dir(),
+            "the sweep must never touch a real segment"
+        );
+        assert_eq!(
+            std::fs::read(vectors.join("segment-1/codes.mpf")).unwrap(),
+            b"keep",
+            "real segment contents must be untouched"
+        );
+        assert!(!vectors.join(".segment-1.staging").exists());
+        assert!(!vectors.join(".segment-2.staging").exists());
+    }
+
+    /// The sweep must be safe to call on a fresh or missing directory.
+    #[test]
+    fn startup_sweep_tolerates_missing_dir() {
+        let tmp = tempfile::tempdir().unwrap();
+        assert_eq!(sweep_orphan_staging(&tmp.path().join("nope")), 0);
+    }
+
     #[test]
     fn test_transition_to_warm_creates_mpf_files() {
         let tmp = tempfile::tempdir().unwrap();