diff --git a/crates/tranquil-store/src/blockstore/compaction.rs b/crates/tranquil-store/src/blockstore/compaction.rs index e5cccac..80392da 100644 --- a/crates/tranquil-store/src/blockstore/compaction.rs +++ b/crates/tranquil-store/src/blockstore/compaction.rs @@ -153,7 +153,7 @@ fn stream_compact( 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( 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(); diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index 413ce46..2e7b54d 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -1054,6 +1054,7 @@ fn drain_and_process_remaining( struct RotationState { file_id: DataFileId, fd: FileId, + hint_fd: FileId, } fn process_batch( @@ -1071,7 +1072,7 @@ fn process_batch( let mut all_decrements: Vec<[u8; CID_SIZE]> = Vec::new(); let mut current_hint_fd = state.hint_fd; - let mut rotation: Option = None; + let mut rotations: Vec = Vec::new(); let mut data_writer = DataFileWriter::resume(manager.io(), state.fd, state.file_id, state.position); @@ -1124,9 +1125,10 @@ fn process_batch( 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( }); 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( 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(); diff --git a/crates/tranquil-store/tests/fd_lifecycle.rs b/crates/tranquil-store/tests/fd_lifecycle.rs new file mode 100644 index 0000000..5d36a5c --- /dev/null +++ b/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 { + 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}" + ); + } +}