fix(tranquil-store): bound writer fd usage across rotations

Lewis: May this revision serve well! <lu5a@proton.me>

crates/tranquil-store/src/blockstore/compaction.rs

@@ -153,7 +153,7 @@ fn stream_compact<S: StorageIO>(
     let mut live_count: u64 = 0;
     let mut dead_count: u64 = 0;
 
-    reader.try_for_each(|r| {
+    let scan_result = reader.try_for_each(|r| {
         let record = r?;
         match record {
             ReadBlockRecord::Valid {
@@ -192,11 +192,16 @@ fn stream_compact<S: StorageIO>(
             ReadBlockRecord::Corrupted { .. } | ReadBlockRecord::Truncated { .. } => {}
         }
         Ok::<_, CompactionError>(())
-    })?;
+    });
 
-    writer.sync()?;
-    hint_writer.sync()?;
-    manager.io().sync_dir(manager.data_dir())?;
+    let finalize_result = scan_result
+        .and_then(|()| writer.sync().map_err(CompactionError::from))
+        .and_then(|()| hint_writer.sync().map_err(CompactionError::from))
+        .and_then(|()| manager.io().sync_dir(manager.data_dir()).map_err(CompactionError::from));
+
+    let _ = manager.io().close(hint_fd);
+
+    finalize_result?;
 
     let new_size = writer.position().raw();
 

crates/tranquil-store/src/blockstore/group_commit.rs

@@ -1054,6 +1054,7 @@ fn drain_and_process_remaining<S: StorageIO>(
 struct RotationState {
     file_id: DataFileId,
     fd: FileId,
+    hint_fd: FileId,
 }
 
 fn process_batch<S: StorageIO>(
@@ -1071,7 +1072,7 @@ fn process_batch<S: StorageIO>(
     let mut all_decrements: Vec<[u8; CID_SIZE]> = Vec::new();
 
     let mut current_hint_fd = state.hint_fd;
-    let mut rotation: Option<RotationState> = None;
+    let mut rotations: Vec<RotationState> = Vec::new();
 
     let mut data_writer =
         DataFileWriter::resume(manager.io(), state.fd, state.file_id, state.position);
@@ -1124,9 +1125,10 @@ fn process_batch<S: StorageIO>(
 
                         current_hint_fd = new_hint_fd;
                         hint_writer = HintFileWriter::new(manager.io(), new_hint_fd);
-                        rotation = Some(RotationState {
+                        rotations.push(RotationState {
                             file_id: next_id,
                             fd: next_fd,
+                            hint_fd: new_hint_fd,
                         });
                     }
 
@@ -1152,9 +1154,13 @@ fn process_batch<S: StorageIO>(
     });
 
     if let Err(e) = write_result {
-        if let Some(rot) = rotation {
+        rotations.into_iter().for_each(|rot| {
             manager.rollback_rotation(rot.file_id, rot.fd);
-        }
+            let _ = manager.io().close(rot.hint_fd);
+            let _ = manager
+                .io()
+                .delete(&hint_file_path(manager.data_dir(), rot.file_id));
+        });
         return Err(e);
     }
 
@@ -1172,9 +1178,21 @@ fn process_batch<S: StorageIO>(
     hint_writer.sync()?;
     let sync_nanos = t.elapsed().as_nanos() as u64;
 
-    if let Some(ref rot) = rotation {
-        manager.commit_rotation(rot.file_id, rot.fd);
-        ctx.active_files.register(ctx.shard_id, rot.file_id);
+    if !rotations.is_empty() {
+        let old_file_id = state.file_id;
+        let old_hint_fd = state.hint_fd;
+        let last_idx = rotations.len() - 1;
+        rotations.iter().enumerate().for_each(|(i, rot)| {
+            if i == last_idx {
+                manager.commit_rotation(rot.file_id, rot.fd);
+                ctx.active_files.register(ctx.shard_id, rot.file_id);
+            } else {
+                let _ = manager.io().close(rot.hint_fd);
+                manager.evict_handle(rot.file_id);
+            }
+        });
+        manager.evict_handle(old_file_id);
+        let _ = manager.io().close(old_hint_fd);
     }
 
     state.file_id = data_writer.file_id();

crates/tranquil-store/tests/fd_lifecycle.rs

@@ -0,0 +1,107 @@
+use std::io::BufRead;
+use std::path::Path;
+
+use jacquard_repo::storage::BlockStore;
+use tranquil_store::blockstore::{BlockStoreConfig, GroupCommitConfig, TranquilBlockStore};
+
+const POST_DROP_FD_TOLERANCE: i64 = 2;
+
+fn fd_count() -> usize {
+    std::fs::read_dir("/proc/self/fd")
+        .map(|it| it.count())
+        .unwrap_or(0)
+}
+
+fn log_rlimit(label: &str) {
+    if let Ok(f) = std::fs::File::open("/proc/self/limits") {
+        let reader = std::io::BufReader::new(f);
+        reader
+            .lines()
+            .map_while(Result::ok)
+            .find(|l| l.contains("open files"))
+            .into_iter()
+            .for_each(|l| eprintln!("[{label}] {l}"));
+    }
+}
+
+fn config_for(dir: &Path, max_file_size: u64) -> BlockStoreConfig {
+    BlockStoreConfig {
+        data_dir: dir.join("data"),
+        index_dir: dir.join("index"),
+        max_file_size,
+        group_commit: GroupCommitConfig {
+            checkpoint_interval_ms: 100,
+            checkpoint_write_threshold: 10,
+            ..GroupCommitConfig::default()
+        },
+        shard_count: 1,
+    }
+}
+
+fn tiny_block(seed: u64) -> Vec<u8> {
+    let bytes = seed.to_le_bytes();
+    (0..64).map(|i| bytes[i % 8] ^ (i as u8).wrapping_mul(31)).collect()
+}
+
+#[tokio::test]
+async fn fds_stable_within_store_lifetime() {
+    log_rlimit("start");
+    let dir = tempfile::TempDir::new().unwrap();
+    let cfg = config_for(dir.path(), 4096);
+    let base = fd_count() as i64;
+    eprintln!("baseline fds: {base}");
+
+    let store = TranquilBlockStore::open(cfg.clone()).expect("open");
+    let after_open = fd_count() as i64;
+    eprintln!("after open: {after_open} fds, delta {}", after_open - base);
+
+    for i in 0..20_000u64 {
+        let data = tiny_block(i);
+        store.put(&data).await.expect("put");
+        if i.is_multiple_of(2_000) {
+            let fds = fd_count() as i64;
+            eprintln!("after {i} puts: {fds} fds, delta {}", fds - base);
+        }
+    }
+
+    let final_fds = fd_count() as i64;
+    eprintln!("final: {final_fds} fds, delta {}", final_fds - base);
+
+    drop(store);
+    let after_drop = fd_count() as i64;
+    let delta = after_drop - base;
+    eprintln!("after drop: {after_drop} fds, delta {delta}");
+
+    assert!(
+        delta <= POST_DROP_FD_TOLERANCE,
+        "fd leak after store drop: baseline {base}, after_drop {after_drop}, delta {delta}"
+    );
+}
+
+#[tokio::test]
+async fn fds_stable_across_reopens() {
+    log_rlimit("start");
+    let dir = tempfile::TempDir::new().unwrap();
+    let cfg = config_for(dir.path(), 4096);
+    let base = fd_count() as i64;
+    eprintln!("baseline: {base}");
+
+    for cycle in 0..20usize {
+        let store = TranquilBlockStore::open(cfg.clone()).expect("open");
+        for i in 0..2_000u64 {
+            let data = tiny_block((cycle as u64) * 10_000 + i);
+            store.put(&data).await.expect("put");
+        }
+        let before_drop = fd_count() as i64;
+        drop(store);
+        let after_drop = fd_count() as i64;
+        let delta = after_drop - base;
+        eprintln!(
+            "cycle {cycle}: before_drop {before_drop}, after_drop {after_drop}, delta {delta}"
+        );
+        assert!(
+            delta <= POST_DROP_FD_TOLERANCE,
+            "fd leak across reopens at cycle {cycle}: baseline {base}, after_drop {after_drop}, delta {delta}"
+        );
+    }
+}