diff --git a/crates/tranquil-store/Cargo.toml b/crates/tranquil-store/Cargo.toml index cc7d30e..5dddd60 100644 --- a/crates/tranquil-store/Cargo.toml +++ b/crates/tranquil-store/Cargo.toml @@ -37,10 +37,11 @@ uuid = { workspace = true } tempfile = { version = "3", optional = true } clap = { workspace = true, optional = true } toml = { version = "0.8", optional = true } +tracing-subscriber = { workspace = true, features = ["env-filter"], optional = true } [features] test-harness = ["dep:tempfile"] -gauntlet-cli = ["test-harness", "dep:clap", "dep:toml"] +gauntlet-cli = ["test-harness", "dep:clap", "dep:toml", "dep:tracing-subscriber"] [[bin]] name = "tranquil-gauntlet" diff --git a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs index 4e0822e..75a98b0 100644 --- a/crates/tranquil-store/src/bin/tranquil_gauntlet.rs +++ b/crates/tranquil-store/src/bin/tranquil_gauntlet.rs @@ -365,6 +365,13 @@ fn install_interrupt(rt: &Runtime) -> Arc { } fn main() -> ExitCode { + let _ = tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("off")), + ) + .with_writer(io::stderr) + .try_init(); let cli = Cli::parse(); match cli.cmd { Cmd::Farm { diff --git a/crates/tranquil-store/src/blockstore/compaction.rs b/crates/tranquil-store/src/blockstore/compaction.rs index 80392da..054c66b 100644 --- a/crates/tranquil-store/src/blockstore/compaction.rs +++ b/crates/tranquil-store/src/blockstore/compaction.rs @@ -102,6 +102,13 @@ pub(super) fn compact_on_writer_thread( .io() .delete(&hint_file_path(manager.data_dir(), source_file_id)) .ok(); + if live_count == 0 { + manager.delete_data_file(new_file_id).ok(); + manager + .io() + .delete(&hint_file_path(manager.data_dir(), new_file_id)) + .ok(); + } manager.io().sync_dir(manager.data_dir())?; let reclaimed_bytes = source_size.saturating_sub(new_size); diff --git a/crates/tranquil-store/src/blockstore/group_commit.rs b/crates/tranquil-store/src/blockstore/group_commit.rs index 407fefc..92d0e36 100644 --- a/crates/tranquil-store/src/blockstore/group_commit.rs +++ b/crates/tranquil-store/src/blockstore/group_commit.rs @@ -563,6 +563,16 @@ fn initialize_active_state( ))); } + let header_end = super::data_file::BLOCK_HEADER_SIZE as u64; + let position = match file_size < header_end { + true => { + let writer = DataFileWriter::new(manager.io(), fd, wc.file_id)?; + writer.sync()?; + writer.position() + } + false => BlockOffset::new(file_size), + }; + let hint_path = hint_file_path(data_dir, wc.file_id); let hint_fd = manager.io().open(&hint_path, OpenOptions::read_write())?; let hint_size = manager.io().file_size(hint_fd)?; @@ -570,7 +580,7 @@ fn initialize_active_state( Ok(ActiveState { file_id: wc.file_id, fd, - position: BlockOffset::new(file_size), + position, hint_fd, hint_position: HintOffset::new(hint_size), }) @@ -1170,6 +1180,8 @@ fn rollback_batch( state: &ActiveState, rotations: &[RotationState], ) { + let _ = manager.io().truncate(state.fd, state.position.raw()); + let _ = manager.io().sync(state.fd); let _ = manager .io() .truncate(state.hint_fd, state.hint_position.raw()); @@ -1226,46 +1238,55 @@ fn process_batch( hint_writer.append_hint(cid_bytes, &loc)?; loc } - None => { - if manager.should_rotate(data_writer.position()) { - data_writer.sync()?; - hint_writer.sync()?; - - let next_id = ctx.file_ids.allocate(); - let next_fd = manager.open_for_append(next_id)?; - - tracing::info!( - from = %data_writer.file_id(), - to = %next_id, - "data file rotation" - ); - - data_writer = DataFileWriter::new(manager.io(), next_fd, next_id)?; - - let new_hint_path = hint_file_path(manager.data_dir(), next_id); - let new_hint_fd = manager - .io() - .open(&new_hint_path, OpenOptions::read_write())?; - - manager.io().sync_dir(manager.data_dir())?; - - current_hint_fd = new_hint_fd; - hint_writer = HintFileWriter::new(manager.io(), new_hint_fd); - rotations.push(RotationState { - file_id: next_id, - fd: next_fd, - hint_fd: new_hint_fd, - }); + None => match index.get(cid_bytes) { + Some(existing) => { + dedup_hits = dedup_hits.saturating_add(1); + let loc = existing.location; + hint_writer.append_hint(cid_bytes, &loc)?; + dedup.insert(*cid_bytes, loc); + loc } + None => { + if manager.should_rotate(data_writer.position()) { + data_writer.sync()?; + hint_writer.sync()?; - let loc = data_writer.append_block(cid_bytes, data)?; - hint_writer.append_hint(cid_bytes, &loc)?; + let next_id = ctx.file_ids.allocate(); + let next_fd = manager.open_for_append(next_id)?; - block_bytes = block_bytes.saturating_add(data.len() as u64); - block_count = block_count.saturating_add(1); - dedup.insert(*cid_bytes, loc); - loc - } + tracing::info!( + from = %data_writer.file_id(), + to = %next_id, + "data file rotation" + ); + + data_writer = DataFileWriter::new(manager.io(), next_fd, next_id)?; + + let new_hint_path = hint_file_path(manager.data_dir(), next_id); + let new_hint_fd = manager + .io() + .open(&new_hint_path, OpenOptions::read_write())?; + + manager.io().sync_dir(manager.data_dir())?; + + current_hint_fd = new_hint_fd; + hint_writer = HintFileWriter::new(manager.io(), new_hint_fd); + rotations.push(RotationState { + file_id: next_id, + fd: next_fd, + hint_fd: new_hint_fd, + }); + } + + let loc = data_writer.append_block(cid_bytes, data)?; + hint_writer.append_hint(cid_bytes, &loc)?; + + block_bytes = block_bytes.saturating_add(data.len() as u64); + block_count = block_count.saturating_add(1); + dedup.insert(*cid_bytes, loc); + loc + } + }, }; index_entries.push((*cid_bytes, location)); diff --git a/crates/tranquil-store/src/blockstore/hash_index.rs b/crates/tranquil-store/src/blockstore/hash_index.rs index a0e3be7..bfebba8 100644 --- a/crates/tranquil-store/src/blockstore/hash_index.rs +++ b/crates/tranquil-store/src/blockstore/hash_index.rs @@ -704,6 +704,7 @@ impl HashTable { const CHECKPOINT_MAGIC: [u8; 8] = *b"TQCKPT01"; const CHECKPOINT_VERSION_V1: u32 = 1; const CHECKPOINT_VERSION_V2: u32 = 2; +const CHECKPOINT_VERSION_V3: u32 = 3; const CHECKPOINT_HEADER_SIZE: usize = 128; const TRAILER_MAGIC: u64 = 0xDEAD_BEEF_CAFE_F00D; const SLOT_SIZE: usize = std::mem::size_of::(); @@ -733,6 +734,7 @@ const H_CHECKPOINT_EPOCH: usize = 56; const H_HINT_FILE_ID: usize = 64; const H_HINT_OFFSET: usize = 72; const H_HEADER_CHECKSUM: usize = 80; +const H_GENERATION: usize = 88; fn header_checksum(buf: &[u8; CHECKPOINT_HEADER_SIZE]) -> u64 { xxhash_rust::xxh3::xxh3_64(&buf[..H_HEADER_CHECKSUM]) @@ -745,10 +747,11 @@ fn serialize_header( cursor_offset: u64, checkpoint_epoch: u64, shard_count: u16, + generation: u64, ) -> [u8; CHECKPOINT_HEADER_SIZE] { let mut buf = [0u8; CHECKPOINT_HEADER_SIZE]; buf[H_MAGIC..H_MAGIC + 8].copy_from_slice(&CHECKPOINT_MAGIC); - buf[H_VERSION..H_VERSION + 4].copy_from_slice(&CHECKPOINT_VERSION_V2.to_le_bytes()); + buf[H_VERSION..H_VERSION + 4].copy_from_slice(&CHECKPOINT_VERSION_V3.to_le_bytes()); buf[H_SHARD_COUNT..H_SHARD_COUNT + 2].copy_from_slice(&shard_count.to_le_bytes()); buf[H_SLOT_COUNT..H_SLOT_COUNT + 8].copy_from_slice(&slot_count.to_le_bytes()); buf[H_ENTRY_COUNT..H_ENTRY_COUNT + 8].copy_from_slice(&entry_count.to_le_bytes()); @@ -756,6 +759,7 @@ fn serialize_header( buf[H_CURSOR_OFFSET..H_CURSOR_OFFSET + 8].copy_from_slice(&cursor_offset.to_le_bytes()); buf[H_CHECKPOINT_EPOCH..H_CHECKPOINT_EPOCH + 8] .copy_from_slice(&checkpoint_epoch.to_le_bytes()); + buf[H_GENERATION..H_GENERATION + 8].copy_from_slice(&generation.to_le_bytes()); let checksum = header_checksum(&buf); buf[H_HEADER_CHECKSUM..H_HEADER_CHECKSUM + 8].copy_from_slice(&checksum.to_le_bytes()); buf @@ -796,6 +800,7 @@ pub fn write_checkpoint( table: &HashTable, path: &Path, epoch: CommitEpoch, + generation: u64, positions: &CheckpointPositions, ) -> io::Result<()> { use std::io::Write; @@ -817,6 +822,7 @@ pub fn write_checkpoint( cursor_offset, epoch.raw(), shard_count, + generation, ); let slot_bytes = slots_as_bytes(&table.slots); @@ -844,7 +850,7 @@ pub fn write_checkpoint( Ok(()) } -fn parse_checkpoint_header(data: &[u8]) -> io::Result<(usize, usize, u32, u64, u64, u16)> { +fn parse_checkpoint_header(data: &[u8]) -> io::Result<(usize, usize, u32, u64, u64, u16, u64)> { if data.len() < CHECKPOINT_HEADER_SIZE + 16 { return Err(io::Error::new( io::ErrorKind::InvalidData, @@ -863,7 +869,10 @@ fn parse_checkpoint_header(data: &[u8]) -> io::Result<(usize, usize, u32, u64, u } let version = u32::from_le_bytes(hdr[H_VERSION..H_VERSION + 4].try_into().unwrap()); - if version != CHECKPOINT_VERSION_V1 && version != CHECKPOINT_VERSION_V2 { + if version != CHECKPOINT_VERSION_V1 + && version != CHECKPOINT_VERSION_V2 + && version != CHECKPOINT_VERSION_V3 + { return Err(io::Error::new( io::ErrorKind::InvalidData, format!("checkpoint version {version} unsupported"), @@ -921,12 +930,19 @@ fn parse_checkpoint_header(data: &[u8]) -> io::Result<(usize, usize, u32, u64, u ); let shard_count = match version { - CHECKPOINT_VERSION_V2 => { + CHECKPOINT_VERSION_V2 | CHECKPOINT_VERSION_V3 => { u16::from_le_bytes(hdr[H_SHARD_COUNT..H_SHARD_COUNT + 2].try_into().unwrap()) } _ => 0, }; + let generation = match version { + CHECKPOINT_VERSION_V3 => { + u64::from_le_bytes(hdr[H_GENERATION..H_GENERATION + 8].try_into().unwrap()) + } + _ => 0, + }; + Ok(( slot_count, entry_count, @@ -934,6 +950,7 @@ fn parse_checkpoint_header(data: &[u8]) -> io::Result<(usize, usize, u32, u64, u cursor_offset, checkpoint_epoch, shard_count, + generation, )) } @@ -948,11 +965,20 @@ fn deserialize_shard_positions(data: &[u8], count: usize) -> Vec<(DataFileId, Hi .collect() } -pub fn read_checkpoint(path: &Path) -> io::Result<(HashTable, CommitEpoch, CheckpointPositions)> { +pub fn read_checkpoint( + path: &Path, +) -> io::Result<(HashTable, CommitEpoch, CheckpointPositions, u64)> { let data = std::fs::read(path)?; - let (slot_count, entry_count, cursor_file_id, cursor_offset, checkpoint_epoch, shard_count) = - parse_checkpoint_header(&data)?; + let ( + slot_count, + entry_count, + cursor_file_id, + cursor_offset, + checkpoint_epoch, + shard_count, + generation, + ) = parse_checkpoint_header(&data)?; let hdr: &[u8; CHECKPOINT_HEADER_SIZE] = data[..CHECKPOINT_HEADER_SIZE].try_into().unwrap(); let version = u32::from_le_bytes(hdr[H_VERSION..H_VERSION + 4].try_into().unwrap()); @@ -972,7 +998,7 @@ pub fn read_checkpoint(path: &Path) -> io::Result<(HashTable, CommitEpoch, Check let shard_pos_region = &data[shard_pos_start..shard_pos_start + shard_pos_size]; let data_checksum = match version { - CHECKPOINT_VERSION_V2 => { + CHECKPOINT_VERSION_V2 | CHECKPOINT_VERSION_V3 => { let mut hasher = xxhash_rust::xxh3::Xxh3::new(); hasher.update(slot_region); hasher.update(shard_pos_region); @@ -1035,7 +1061,7 @@ pub fn read_checkpoint(path: &Path) -> io::Result<(HashTable, CommitEpoch, Check let epoch = CommitEpoch::new(checkpoint_epoch); let positions = match version { - CHECKPOINT_VERSION_V2 if shard_count > 0 => CheckpointPositions( + CHECKPOINT_VERSION_V2 | CHECKPOINT_VERSION_V3 if shard_count > 0 => CheckpointPositions( deserialize_shard_positions(shard_pos_region, shard_count as usize), ), _ => { @@ -1047,12 +1073,12 @@ pub fn read_checkpoint(path: &Path) -> io::Result<(HashTable, CommitEpoch, Check } }; - Ok((table, epoch, positions)) + Ok((table, epoch, positions, generation)) } pub fn load_best_checkpoint( index_dir: &Path, -) -> Option<(HashTable, CommitEpoch, CheckpointPositions)> { +) -> Option<(HashTable, CommitEpoch, CheckpointPositions, u64)> { let path_a = index_dir.join("checkpoint_a.tqc"); let path_b = index_dir.join("checkpoint_b.tqc"); @@ -1060,7 +1086,7 @@ pub fn load_best_checkpoint( let result_b = read_checkpoint(&path_b).ok(); match (result_a, result_b) { - (Some(a), Some(b)) => match a.1.raw() >= b.1.raw() { + (Some(a), Some(b)) => match (a.3, a.1.raw()) >= (b.3, b.1.raw()) { true => Some(a), false => Some(b), }, @@ -1070,7 +1096,7 @@ pub fn load_best_checkpoint( } } -fn read_checkpoint_epoch(path: &Path) -> Option { +fn read_checkpoint_meta(path: &Path) -> Option<(u64, u64)> { let mut file = std::fs::File::open(path).ok()?; let mut buf = [0u8; CHECKPOINT_HEADER_SIZE]; std::io::Read::read_exact(&mut file, &mut buf).ok()?; @@ -1081,7 +1107,10 @@ fn read_checkpoint_epoch(path: &Path) -> Option { } let version = u32::from_le_bytes(buf[H_VERSION..H_VERSION + 4].try_into().ok()?); - if version != CHECKPOINT_VERSION_V1 && version != CHECKPOINT_VERSION_V2 { + if version != CHECKPOINT_VERSION_V1 + && version != CHECKPOINT_VERSION_V2 + && version != CHECKPOINT_VERSION_V3 + { return None; } @@ -1094,33 +1123,41 @@ fn read_checkpoint_epoch(path: &Path) -> Option { return None; } - Some(u64::from_le_bytes( + let epoch = u64::from_le_bytes( buf[H_CHECKPOINT_EPOCH..H_CHECKPOINT_EPOCH + 8] .try_into() .ok()?, - )) + ); + let generation = match version { + CHECKPOINT_VERSION_V3 => { + u64::from_le_bytes(buf[H_GENERATION..H_GENERATION + 8].try_into().ok()?) + } + _ => 0, + }; + Some((epoch, generation)) } pub fn write_checkpoint_ab( table: &HashTable, index_dir: &Path, epoch: CommitEpoch, + generation: u64, positions: &CheckpointPositions, ) -> io::Result<()> { let path_a = index_dir.join("checkpoint_a.tqc"); let path_b = index_dir.join("checkpoint_b.tqc"); - let epoch_a = read_checkpoint_epoch(&path_a); - let epoch_b = read_checkpoint_epoch(&path_b); + let meta_a = read_checkpoint_meta(&path_a); + let meta_b = read_checkpoint_meta(&path_b); - let target_path = match (epoch_a, epoch_b) { - (Some(a), Some(b)) if a >= b => path_b, + let target_path = match (meta_a, meta_b) { + (Some(a), Some(b)) if (a.1, a.0) >= (b.1, b.0) => path_b, (Some(_), Some(_)) => path_a, (Some(_), None) => path_b, (None, _) => path_a, }; - write_checkpoint(table, &target_path, epoch, positions) + write_checkpoint(table, &target_path, epoch, generation, positions) } #[derive(Debug)] @@ -1146,6 +1183,7 @@ pub struct BlockIndex { checkpoint_lock: parking_lot::Mutex<()>, loaded_checkpoint_positions: Option, loaded_checkpoint_epoch: Option, + next_generation: std::sync::atomic::AtomicU64, } impl BlockIndex { @@ -1156,33 +1194,36 @@ impl BlockIndex { checkpoint_lock: parking_lot::Mutex::new(()), loaded_checkpoint_positions: None, loaded_checkpoint_epoch: None, + next_generation: std::sync::atomic::AtomicU64::new(1), } } pub fn open(index_dir: &Path) -> io::Result { std::fs::create_dir_all(index_dir)?; - let (table, checkpoint_positions, checkpoint_epoch) = match load_best_checkpoint(index_dir) - { - Some((table, epoch, positions)) => { - tracing::info!( - blocks = table.len(), - epoch = epoch.raw(), - shard_positions = positions.0.len(), - "loaded block index from checkpoint" - ); - (table, Some(positions), Some(epoch)) - } - None => { - tracing::info!("no valid checkpoint found, starting with empty index"); - (HashTable::with_capacity(64), None, None) - } - }; + let (table, checkpoint_positions, checkpoint_epoch, loaded_generation) = + match load_best_checkpoint(index_dir) { + Some((table, epoch, positions, gen_value)) => { + tracing::info!( + blocks = table.len(), + epoch = epoch.raw(), + shard_positions = positions.0.len(), + generation = gen_value, + "loaded block index from checkpoint" + ); + (table, Some(positions), Some(epoch), gen_value) + } + None => { + tracing::info!("no valid checkpoint found, starting with empty index"); + (HashTable::with_capacity(64), None, None, 0) + } + }; Ok(Self { table: RwLock::new(table), index_dir: index_dir.to_path_buf(), checkpoint_lock: parking_lot::Mutex::new(()), loaded_checkpoint_positions: checkpoint_positions, loaded_checkpoint_epoch: checkpoint_epoch, + next_generation: std::sync::atomic::AtomicU64::new(loaded_generation + 1), }) } @@ -1481,9 +1522,12 @@ impl BlockIndex { hint_positions: &ShardHintPositions, ) -> io::Result<()> { let _guard = self.checkpoint_lock.lock(); + let generation = self + .next_generation + .fetch_add(1, std::sync::atomic::Ordering::AcqRel); let table = self.table.read(); let positions = hint_positions.snapshot(); - write_checkpoint_ab(&table, &self.index_dir, epoch, &positions) + write_checkpoint_ab(&table, &self.index_dir, epoch, generation, &positions) } pub fn write_checkpoint_with_positions( @@ -1492,8 +1536,11 @@ impl BlockIndex { positions: &CheckpointPositions, ) -> io::Result<()> { let _guard = self.checkpoint_lock.lock(); + let generation = self + .next_generation + .fetch_add(1, std::sync::atomic::Ordering::AcqRel); let table = self.table.read(); - write_checkpoint_ab(&table, &self.index_dir, epoch, positions) + write_checkpoint_ab(&table, &self.index_dir, epoch, generation, positions) } pub fn index_dir(&self) -> &Path { @@ -2045,8 +2092,8 @@ mod tests { let epoch = CommitEpoch::new(42); let positions = CheckpointPositions::single(DataFileId::new(5), HintOffset::new(12345)); - write_checkpoint(&table, &path, epoch, &positions).unwrap(); - let (restored, restored_epoch, restored_pos) = read_checkpoint(&path).unwrap(); + write_checkpoint(&table, &path, epoch, 7, &positions).unwrap(); + let (restored, restored_epoch, restored_pos, _gen) = read_checkpoint(&path).unwrap(); assert_eq!(restored.len(), 10); assert_eq!(restored_epoch.raw(), 42); @@ -2078,14 +2125,14 @@ mod tests { table .insert_or_increment(&test_cid(1), test_loc(0, 0, 10)) .unwrap(); - write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(1), &pos).unwrap(); + write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(1), 1, &pos).unwrap(); table .insert_or_increment(&test_cid(2), test_loc(0, 100, 10)) .unwrap(); - write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(2), &pos).unwrap(); + write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(2), 2, &pos).unwrap(); - let (best, epoch, _) = load_best_checkpoint(dir.path()).unwrap(); + let (best, epoch, _, _) = load_best_checkpoint(dir.path()).unwrap(); assert_eq!(epoch.raw(), 2); assert_eq!(best.len(), 2); } @@ -2099,16 +2146,16 @@ mod tests { table .insert_or_increment(&test_cid(1), test_loc(0, 0, 10)) .unwrap(); - write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(1), &pos).unwrap(); + write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(1), 1, &pos).unwrap(); table .insert_or_increment(&test_cid(2), test_loc(0, 100, 10)) .unwrap(); - write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(2), &pos).unwrap(); + write_checkpoint_ab(&table, dir.path(), CommitEpoch::new(2), 2, &pos).unwrap(); std::fs::write(dir.path().join("checkpoint_b.tqc"), b"corrupt").unwrap(); - let (best, epoch, _) = load_best_checkpoint(dir.path()).unwrap(); + let (best, epoch, _, _) = load_best_checkpoint(dir.path()).unwrap(); assert_eq!(epoch.raw(), 1); assert_eq!(best.len(), 1); } diff --git a/crates/tranquil-store/src/blockstore/store.rs b/crates/tranquil-store/src/blockstore/store.rs index 8928822..b44fd15 100644 --- a/crates/tranquil-store/src/blockstore/store.rs +++ b/crates/tranquil-store/src/blockstore/store.rs @@ -393,11 +393,12 @@ impl TranquilBlockStore { io.sync(fd).map_err(RepoError::storage)?; } - if !hint_exists && !scanned_entries.is_empty() { + if !scanned_entries.is_empty() { tracing::info!( file_id = %file_id, scanned = scanned_entries.len(), - "rebuilding index from data file (no hint file, treating as restored backup)" + hint_exists, + "reindexing blocks past hint coverage" ); let cursor = super::types::WriteCursor { file_id, diff --git a/crates/tranquil-store/src/gauntlet/invariants.rs b/crates/tranquil-store/src/gauntlet/invariants.rs index 08036c5..2f91c7d 100644 --- a/crates/tranquil-store/src/gauntlet/invariants.rs +++ b/crates/tranquil-store/src/gauntlet/invariants.rs @@ -7,7 +7,9 @@ use cid::Cid; use jacquard_repo::mst::Mst; use super::oracle::{Oracle, hex_short, try_cid_to_fixed}; -use crate::blockstore::{CidBytes, CompactionError, TranquilBlockStore, hash_to_cid_bytes}; +use crate::blockstore::{ + BLOCK_HEADER_SIZE, CidBytes, CompactionError, TranquilBlockStore, hash_to_cid_bytes, +}; use crate::eventlog::{EventSequence, SegmentId}; use crate::io::{RealIO, StorageIO}; @@ -388,10 +390,18 @@ impl Invariant for NoOrphanFiles { let result = tokio::task::spawn_blocking(move || { let disk = store_c.list_data_files().map_err(|e| e.to_string())?; let liveness = store_c.compaction_liveness(0).map_err(|e| e.to_string())?; + let header = BLOCK_HEADER_SIZE as u64; let orphans: Vec = disk .iter() .filter(|fid| !liveness.contains_key(fid)) - .map(|fid| format!("{fid}")) + .filter_map(|fid| { + let path = store_c.data_file_path(*fid); + let size = std::fs::metadata(&path).map(|m| m.len()).unwrap_or(0); + match size > header { + true => Some(format!("{fid} ({size} B)")), + false => None, + } + }) .collect(); Ok::<_, String>(orphans) }) @@ -479,6 +489,7 @@ impl Invariant for ManifestEqualsRealit tokio::task::spawn_blocking(move || { let listed = store.list_data_files().map_err(|e| e.to_string())?; let liveness = store.compaction_liveness(0).map_err(|e| e.to_string())?; + let header = BLOCK_HEADER_SIZE as u64; let mut violations: Vec = Vec::new(); listed.iter().for_each(|fid| { @@ -487,21 +498,23 @@ impl Invariant for ManifestEqualsRealit Err(e) => violations.push(format!("{fid}: metadata {e}")), Ok(meta) => { let on_disk = meta.len(); + let content = on_disk.saturating_sub(header); match liveness.get(fid) { - None => violations.push(format!( + None if on_disk > header => violations.push(format!( "{fid}: listed on disk at {on_disk} B but not in index liveness" )), - Some(info) if on_disk < info.total_bytes => { + None => {} + Some(info) if content < info.total_bytes => { violations.push(format!( - "{fid}: on-disk {on_disk} B < index total_bytes {}", + "{fid}: on-disk {on_disk} B (content {content}) < index total_bytes {}", info.total_bytes )); } - Some(info) if on_disk > info.total_bytes => { + Some(info) if content > info.total_bytes => { violations.push(format!( - "{fid}: on-disk {on_disk} B > index total_bytes {}, {} B unaccounted", + "{fid}: on-disk {on_disk} B (content {content}) > index total_bytes {}, {} B unaccounted", info.total_bytes, - on_disk - info.total_bytes + content - info.total_bytes )); } Some(_) => {}