diff --git a/crates/tranquil-pds/src/state.rs b/crates/tranquil-pds/src/state.rs index 6ba45c1..6a66128 100644 --- a/crates/tranquil-pds/src/state.rs +++ b/crates/tranquil-pds/src/state.rs @@ -580,6 +580,18 @@ fn wire_tranquil_store( "repaired orphan data files" ); } + if repair.orphan_hints_removed > 0 { + tracing::info!( + removed = repair.orphan_hints_removed, + "repaired orphan hint files" + ); + } + if repair.phantom_index_entries_purged > 0 { + tracing::info!( + purged = repair.phantom_index_entries_purged, + "purged phantom index entries pointing at missing data files" + ); + } if repair.had_errors() { tracing::warn!(errors = repair.repair_errors, "some repairs failed"); } diff --git a/crates/tranquil-store/src/consistency.rs b/crates/tranquil-store/src/consistency.rs index eb3d2c6..9abf945 100644 --- a/crates/tranquil-store/src/consistency.rs +++ b/crates/tranquil-store/src/consistency.rs @@ -28,6 +28,8 @@ pub struct ConsistencyReport { pub orphaned_user_repos: Vec, pub inconsistent_handles: Vec, pub orphan_data_files: Vec, + pub orphan_hint_files: Vec, + pub missing_indexed_files: Vec, pub deserialization_failures: u64, pub eventlog_contiguity: Option, pub cursor_ahead_of_eventlog: bool, @@ -74,6 +76,8 @@ impl ConsistencyReport { && self.orphaned_user_repos.is_empty() && self.inconsistent_handles.is_empty() && self.orphan_data_files.is_empty() + && self.orphan_hint_files.is_empty() + && self.missing_indexed_files.is_empty() && self.deserialization_failures == 0 && self .eventlog_contiguity @@ -84,6 +88,8 @@ impl ConsistencyReport { pub fn has_repairable_issues(&self) -> bool { !self.orphan_data_files.is_empty() + || !self.orphan_hint_files.is_empty() + || !self.missing_indexed_files.is_empty() } pub fn has_unrecoverable_issues(&self) -> bool { @@ -136,6 +142,20 @@ impl ConsistencyReport { "orphan data files with no index references" ); } + if !self.orphan_hint_files.is_empty() { + tracing::warn!( + count = self.orphan_hint_files.len(), + files = ?self.orphan_hint_files, + "orphan hint files with no matching data file" + ); + } + if !self.missing_indexed_files.is_empty() { + tracing::warn!( + count = self.missing_indexed_files.len(), + files = ?self.missing_indexed_files, + "index references data files that are missing on disk" + ); + } if self.deserialization_failures > 0 { tracing::error!( count = self.deserialization_failures, @@ -181,13 +201,15 @@ impl fmt::Display for ConsistencyReport { write!( f, "INCONSISTENT: dangling_roots={}, dangling_records={}, orphaned_repos={}, \ - inconsistent_handles={}, orphan_files={}, deserialize_failures={}, \ - eventlog_gaps={}, cursor_ahead={}", + inconsistent_handles={}, orphan_files={}, orphan_hints={}, missing_indexed_files={}, \ + deserialize_failures={}, eventlog_gaps={}, cursor_ahead={}", self.dangling_root_cids.len(), self.dangling_record_cids.len(), self.orphaned_user_repos.len(), self.inconsistent_handles.len(), self.orphan_data_files.len(), + self.orphan_hint_files.len(), + self.missing_indexed_files.len(), self.deserialization_failures, self.eventlog_contiguity .as_ref() @@ -204,6 +226,8 @@ pub struct ConsistencyCheckOptions { pub check_user_blocks: bool, pub check_eventlog: bool, pub check_orphan_files: bool, + pub check_missing_indexed_files: bool, + pub check_orphan_hint_files: bool, } impl Default for ConsistencyCheckOptions { @@ -214,6 +238,8 @@ impl Default for ConsistencyCheckOptions { check_user_blocks: true, check_eventlog: true, check_orphan_files: true, + check_missing_indexed_files: true, + check_orphan_hint_files: true, } } } @@ -269,6 +295,14 @@ pub fn verify_store_consistency_with_options( check_orphan_data_files(blockstore, block_index, &mut report); } + if options.check_missing_indexed_files { + check_missing_indexed_files(blockstore, block_index, &mut report); + } + + if options.check_orphan_hint_files { + check_orphan_hint_files(blockstore, &mut report); + } + report } @@ -565,6 +599,52 @@ fn check_orphan_data_files( }); } +fn check_missing_indexed_files( + blockstore: &TranquilBlockStore, + block_index: &BlockIndex, + report: &mut ConsistencyReport, +) { + let disk_files: HashSet = match blockstore.list_data_files() { + Ok(files) => files.into_iter().collect(), + Err(e) => { + tracing::warn!(error = %e, "failed to list data files for missing-file check"); + return; + } + }; + + let epoch = blockstore.epoch().current(); + let now = crate::wall_clock_ms(); + let indexed_files = block_index.liveness_by_file(epoch, now, 0); + + indexed_files + .iter() + .filter(|(fid, _)| !disk_files.contains(fid)) + .for_each(|(fid, _)| report.missing_indexed_files.push(*fid)); +} + +fn check_orphan_hint_files(blockstore: &TranquilBlockStore, report: &mut ConsistencyReport) { + let data_files: HashSet = match blockstore.list_data_files() { + Ok(files) => files.into_iter().collect(), + Err(e) => { + tracing::warn!(error = %e, "failed to list data files for orphan-hint check"); + return; + } + }; + + let hint_files = match blockstore.list_hint_files() { + Ok(files) => files, + Err(e) => { + tracing::warn!(error = %e, "failed to list hint files for orphan-hint check"); + return; + } + }; + + hint_files + .iter() + .filter(|fid| !data_files.contains(fid)) + .for_each(|fid| report.orphan_hint_files.push(*fid)); +} + fn try_cid_bytes_to_fixed(bytes: &[u8]) -> Option<[u8; CID_SIZE]> { bytes.try_into().ok() } @@ -621,12 +701,39 @@ pub fn repair_known_issues( } }); + report.orphan_hint_files.iter().for_each(|&file_id| { + let path = blockstore.hint_file_path(file_id); + match std::fs::remove_file(&path) { + Ok(()) => { + tracing::info!(%file_id, "removed orphan hint file"); + result.orphan_hints_removed = result.orphan_hints_removed.saturating_add(1); + } + Err(e) => { + tracing::warn!(%file_id, error = %e, "failed to remove orphan hint file"); + result.repair_errors = result.repair_errors.saturating_add(1); + } + } + }); + + report.missing_indexed_files.iter().for_each(|&file_id| { + let purged = blockstore.block_index().purge_by_file_id(file_id); + tracing::info!( + %file_id, + purged, + "purged phantom index entries for missing data file" + ); + result.phantom_index_entries_purged = + result.phantom_index_entries_purged.saturating_add(purged); + }); + result } #[derive(Debug, Default)] pub struct RepairResult { pub orphan_files_removed: u64, + pub orphan_hints_removed: u64, + pub phantom_index_entries_purged: u64, pub repair_errors: u64, } diff --git a/crates/tranquil-store/tests/external_corruption_recovery.rs b/crates/tranquil-store/tests/external_corruption_recovery.rs new file mode 100644 index 0000000..a773dd0 --- /dev/null +++ b/crates/tranquil-store/tests/external_corruption_recovery.rs @@ -0,0 +1,235 @@ +mod common; + +use std::fs; + +use common::{block_data, test_cid, tiny_blockstore_config, with_runtime}; +use tranquil_store::blockstore::{ + CompactionResult, DataFileId, TranquilBlockStore, hint_file_path, +}; + +fn data_file_path(dir: &std::path::Path, file_id: DataFileId) -> std::path::PathBuf { + dir.join(format!("{file_id}.tqb")) +} + +fn populate_with_compaction_history(store: &TranquilBlockStore, live_cids: &[u32]) { + live_cids.iter().for_each(|&seed| { + store + .put_blocks_blocking(vec![(test_cid(seed), block_data(seed))]) + .unwrap(); + }); + + (0..200u32).for_each(|round| { + let churn = test_cid(50_000 + round); + store + .put_blocks_blocking(vec![(churn, block_data(50_000 + round))]) + .unwrap(); + store.apply_commit_blocking(vec![], vec![churn]).unwrap(); + + if round % 4 == 0 { + common::compact_lowest_liveness(store); + } + }); +} + +#[test] +fn deleting_indexed_data_file_externally_self_heals_on_compaction() { + with_runtime(|| { + let dir = tempfile::TempDir::new().unwrap(); + let live_cids: Vec = (0..12u32).collect(); + + { + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + populate_with_compaction_history(&store, &live_cids); + drop(store); + } + + let data_dir = dir.path().join("data"); + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + + let liveness = store.compaction_liveness(0).unwrap(); + let victim_fid = liveness + .iter() + .filter(|(_, info)| info.live_blocks > 0) + .map(|(&fid, _)| fid) + .next() + .expect("expected at least one file with live blocks"); + + drop(store); + + let victim_path = data_file_path(&data_dir, victim_fid); + assert!( + victim_path.exists(), + "victim data file should exist before deletion" + ); + fs::remove_file(&victim_path).unwrap(); + + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + + let liveness_before = store.compaction_liveness(0).unwrap(); + assert!( + liveness_before.contains_key(&victim_fid), + "index should still claim the deleted file before compaction self-heal" + ); + + let result = store.compact_file(victim_fid, 0).unwrap(); + match result { + CompactionResult::Purged { + file_id, + phantom_blocks, + } => { + assert_eq!(file_id, victim_fid); + assert!( + phantom_blocks > 0, + "expected to purge non-zero phantom entries" + ); + } + CompactionResult::Compacted(stats) => { + panic!( + "expected purge for missing source file, got compaction with {stats:?} live={} dead={}", + stats.live_blocks, stats.dead_blocks + ); + } + } + + let liveness_after = store.compaction_liveness(0).unwrap(); + assert!( + !liveness_after.contains_key(&victim_fid), + "compaction-purge must remove all index entries pointing at the deleted file" + ); + }); +} + +#[test] +fn external_hint_orphan_cleaned_by_consistency_repair() { + with_runtime(|| { + let dir = tempfile::TempDir::new().unwrap(); + + { + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + populate_with_compaction_history(&store, &(0..6u32).collect::>()); + drop(store); + } + + let data_dir = dir.path().join("data"); + + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + let any_existing_fid = store + .list_data_files() + .unwrap() + .into_iter() + .next() + .expect("expected at least one data file after populate"); + drop(store); + + let orphan_hint_fid = DataFileId::new(any_existing_fid.raw().saturating_add(10_000)); + let orphan_path = hint_file_path(&data_dir, orphan_hint_fid); + fs::write(&orphan_path, b"\x00").unwrap(); + + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + + let metastore_dir = dir.path().join("metastore"); + std::fs::create_dir_all(&metastore_dir).unwrap(); + let metastore = tranquil_store::metastore::Metastore::open( + &metastore_dir, + tranquil_store::metastore::MetastoreConfig::default(), + ) + .unwrap(); + + let segments_dir = dir.path().join("eventlog").join("segments"); + std::fs::create_dir_all(&segments_dir).unwrap(); + let eventlog = tranquil_store::eventlog::EventLog::open( + tranquil_store::eventlog::EventLogConfig { + segments_dir, + ..tranquil_store::eventlog::EventLogConfig::default() + }, + tranquil_store::RealIO::new(), + ) + .unwrap(); + + let report = + tranquil_store::consistency::verify_store_consistency(&store, &metastore, &eventlog); + + assert!( + report.orphan_hint_files.contains(&orphan_hint_fid), + "consistency check should flag the synthetic orphan hint file" + ); + + let repair = tranquil_store::consistency::repair_known_issues(&store, &report); + assert!( + repair.orphan_hints_removed >= 1, + "repair should remove the orphan hint file" + ); + assert!( + !orphan_path.exists(), + "orphan hint file should be unlinked after repair" + ); + }); +} + +#[test] +fn consistency_check_flags_and_repairs_missing_indexed_file() { + with_runtime(|| { + let dir = tempfile::TempDir::new().unwrap(); + let live_cids: Vec = (0..10u32).collect(); + + { + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + populate_with_compaction_history(&store, &live_cids); + drop(store); + } + + let data_dir = dir.path().join("data"); + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + let victim_fid = store + .compaction_liveness(0) + .unwrap() + .iter() + .filter(|(_, info)| info.live_blocks > 0) + .map(|(&fid, _)| fid) + .next() + .expect("expected at least one indexed file"); + drop(store); + + fs::remove_file(data_file_path(&data_dir, victim_fid)).unwrap(); + let _ = fs::remove_file(hint_file_path(&data_dir, victim_fid)); + + let store = TranquilBlockStore::open(tiny_blockstore_config(dir.path())).unwrap(); + let metastore_dir = dir.path().join("metastore"); + std::fs::create_dir_all(&metastore_dir).unwrap(); + let metastore = tranquil_store::metastore::Metastore::open( + &metastore_dir, + tranquil_store::metastore::MetastoreConfig::default(), + ) + .unwrap(); + + let segments_dir = dir.path().join("eventlog").join("segments"); + std::fs::create_dir_all(&segments_dir).unwrap(); + let eventlog = tranquil_store::eventlog::EventLog::open( + tranquil_store::eventlog::EventLogConfig { + segments_dir, + ..tranquil_store::eventlog::EventLogConfig::default() + }, + tranquil_store::RealIO::new(), + ) + .unwrap(); + + let report = + tranquil_store::consistency::verify_store_consistency(&store, &metastore, &eventlog); + assert!( + report.missing_indexed_files.contains(&victim_fid), + "consistency check should flag the missing indexed file" + ); + + let repair = tranquil_store::consistency::repair_known_issues(&store, &report); + assert!( + repair.phantom_index_entries_purged > 0, + "repair should purge phantom entries" + ); + + let post_repair_liveness = store.compaction_liveness(0).unwrap(); + assert!( + !post_repair_liveness.contains_key(&victim_fid), + "no index entries should remain for the missing file after repair" + ); + }); +}