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

feat(store): consistency check & repair for orphan hints etc

Browse Source 
Lewis: May this revision serve well! <lu5a@proton.me>
This commit is contained in:
Lewis
2026-05-17 11:50:32 +03:00
parent d07d702dd4
commit a7517ed5c9
3 changed files with 356 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/tranquil-pds/src/state.rs
+12
View File
@@ -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");
}
crates/tranquil-store/src/consistency.rs
+109 -2
View File
@@ -28,6 +28,8 @@ pub struct ConsistencyReport {
pub orphaned_user_repos: Vec<OrphanedUserRepo>,
pub inconsistent_handles: Vec<InconsistentHandle>,
pub orphan_data_files: Vec<DataFileId>,
pub orphan_hint_files: Vec<DataFileId>,
pub missing_indexed_files: Vec<DataFileId>,
pub deserialization_failures: u64,
pub eventlog_contiguity: Option<SequenceContiguityResult>,
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<S: StorageIO + 'static>(
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<DataFileId> = 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<DataFileId> = 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,
}
crates/tranquil-store/tests/external_corruption_recovery.rs
+235
View File
@@ -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<u32> = (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::<Vec<_>>());
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<u32> = (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"
);
});
}