mirrors/tranquil-pds
1
0
Fork 0
mirror of https://tangled.org/tranquil.farm/tranquil-pds synced 2026-09-12 05:06:04 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(tranquil-store): no more orphan data files, recover torn-hint tails, header-safe resume

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-04-21 09:59:37 +03:00
parent 2afd075496
commit 9d81e58803
7 changed files with 193 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-store/Cargo.toml
+2 -1
View File
@@ -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"
crates/tranquil-store/src/bin/tranquil_gauntlet.rs
+7
View File
@@ -365,6 +365,13 @@ fn install_interrupt(rt: &Runtime) -> Arc<AtomicBool> {
}
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 {
crates/tranquil-store/src/blockstore/compaction.rs
+7
View File
@@ -102,6 +102,13 @@ pub(super) fn compact_on_writer_thread<S: StorageIO>(
.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);
crates/tranquil-store/src/blockstore/group_commit.rs
+59 -38
View File
@@ -563,6 +563,16 @@ fn initialize_active_state<S: StorageIO>(
)));
}
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<S: StorageIO>(
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<S: StorageIO>(
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<S: StorageIO>(
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));
crates/tranquil-store/src/blockstore/hash_index.rs
+94 -47
View File
@@ -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::<Slot>();
@@ -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<u64> {
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<u64> {
}
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<u64> {
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<CheckpointPositions>,
loaded_checkpoint_epoch: Option<CommitEpoch>,
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<Self> {
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);
}
crates/tranquil-store/src/blockstore/store.rs
+3 -2
View File
@@ -393,11 +393,12 @@ impl<S: StorageIO + Send + Sync + 'static> TranquilBlockStore<S> {
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,
crates/tranquil-store/src/gauntlet/invariants.rs
+21 -8
View File
@@ -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<S: StorageIO + Send + Sync + 'static> Invariant<S> 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<String> = 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<S: StorageIO + Send + Sync + 'static> Invariant<S> 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<String> = Vec::new();
listed.iter().for_each(|fid| {
@@ -487,21 +498,23 @@ impl<S: StorageIO + Send + Sync + 'static> Invariant<S> 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(_) => {}