use crate::{
SeqNo, Table, TableId, UserKey,
config::{Config, TreeType},
version::{BlobFileList, Level, Run, Version},
};
use std::{path::PathBuf, sync::Arc};
type UnreadableFiles = Vec<(PathBuf, String)>;
#[derive(Debug)]
pub struct RepairReport {
pub recovered: usize,
pub salvaged: usize,
pub unreadable: usize,
pub unreadable_files: Vec<(PathBuf, String)>,
pub excluded_files: Vec<(PathBuf, String)>,
pub lost_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)>,
pub unknowable_losses: Vec<PathBuf>,
pub blob_files_salvaged: Vec<(PathBuf, String)>,
pub method: &'static str,
pub warnings: Vec<&'static str>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum WalReplayScope {
TailOnly,
LostUpTo(SeqNo),
FullHistory,
}
impl RepairReport {
#[must_use]
pub fn wal_replay_scope(&self) -> WalReplayScope {
if !self.unknowable_losses.is_empty() {
return WalReplayScope::FullHistory;
}
let mut ceiling: Option<SeqNo> = None;
for (_, _, _, bound) in &self.lost_coverage {
match bound {
None => return WalReplayScope::FullHistory,
Some(b) => ceiling = Some(ceiling.map_or(*b, |c| c.max(*b))),
}
}
match ceiling {
None => WalReplayScope::TailOnly,
Some(b) => WalReplayScope::LostUpTo(b),
}
}
}
pub(crate) fn compute_table_checksum_from(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
start: u64,
) -> crate::Result<u128> {
compute_table_checksum_with_overrides(fs, path, start, &[])
}
pub(crate) fn compute_table_checksum(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
) -> crate::Result<u128> {
compute_table_checksum_with_overrides(fs, path, 0, &[])
}
pub(crate) fn compute_table_checksum_with_overrides(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
start: u64,
overrides: &[(u64, Vec<u8>)],
) -> crate::Result<u128> {
crate::file::checksum_from_with_overrides(fs, path, start, overrides)
}
fn highest_existing_version_id(
fs: &dyn crate::fs::Fs,
folder: &std::path::Path,
) -> crate::Result<Option<u64>> {
Ok(fs
.read_dir(folder)?
.into_iter()
.filter_map(|e| {
e.file_name
.strip_prefix('v')
.and_then(|rest| rest.parse::<u64>().ok())
})
.max())
}
fn discard_unreferenced(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
sync_mode: crate::fs::SyncMode,
) -> crate::Result<()> {
let remove = |target: &std::path::Path| -> crate::Result<()> {
match fs.remove_file(target) {
Ok(()) => Ok(()),
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => Ok(()),
Err(e) => Err(e.into()),
}
};
let sidecar = crate::restrict_bound::sidecar_path(path);
if fs.exists(&sidecar)? {
remove(&sidecar)?;
}
remove(path)?;
if let Some(dir) = path.parent() {
fs.sync_directory_with(dir, sync_mode)?;
}
Ok(())
}
pub(crate) use crate::file::{REPAIR_TMP_SUFFIX, table_id_from_repair_tmp_name};
fn repair_tmp_path(table_path: &std::path::Path) -> PathBuf {
let mut name = table_path.file_name().unwrap_or_default().to_os_string();
name.push(REPAIR_TMP_SUFFIX);
table_path.with_file_name(name)
}
pub(crate) fn commit_repair_tmp(
fs: &dyn crate::fs::Fs,
tmp_path: &std::path::Path,
table_path: &std::path::Path,
sync_mode: crate::fs::SyncMode,
manifest_restricted: bool,
) -> crate::Result<()> {
let tmp_sidecar = crate::restrict_bound::sidecar_path(tmp_path);
let dest_sidecar = crate::restrict_bound::sidecar_path(table_path);
if fs.exists(&tmp_sidecar)? {
fs.rename(&tmp_sidecar, &dest_sidecar)?;
} else if manifest_restricted {
} else if fs.exists(&dest_sidecar)? {
match fs.remove_file(&dest_sidecar) {
Ok(()) => {}
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => return Err(e.into()),
}
}
fs.rename(tmp_path, table_path)?;
if let Some(dir) = table_path.parent() {
fs.sync_directory_with(dir, sync_mode)?;
}
Ok(())
}
#[cfg(feature = "std")]
pub(crate) fn repair_tmp_is_published(
config: &Config,
fs: &Arc<dyn crate::fs::Fs>,
tmp_path: &std::path::Path,
table_id: TableId,
manifest_checksum: crate::Checksum,
restriction: Option<&crate::UserKey>,
) -> crate::Result<bool> {
let original_is_authoritative = || -> crate::Result<bool> {
let original = tmp_path.with_file_name(table_id.to_string());
let digest = match restriction {
None => compute_table_checksum(&**fs, &original),
Some(bound) => {
let table = match crate::table::Table::recover(repair_recover_params(
config,
original.clone(),
manifest_checksum,
table_id,
Arc::clone(fs),
None,
)) {
Ok(table) => table,
Err(e) if is_environmental(&e) => return Err(e),
Err(_) => return Ok(false),
};
match table.punch_offset_for(bound.as_ref()) {
Ok(offset) => compute_table_checksum_from(&**fs, &original, offset),
Err(e) if is_environmental(&e) => return Err(e),
Err(_) => return Ok(false),
}
}
};
match digest {
Ok(digest) => Ok(crate::Checksum::from_raw(digest) == manifest_checksum),
Err(e) if is_environmental(&e) => Err(e),
Err(_) => Ok(false),
}
};
let condemn_only_if_proven = |ambiguity: crate::Error| -> crate::Result<bool> {
if original_is_authoritative()? {
Ok(false)
} else {
Err(ambiguity)
}
};
let digest_mismatch = |digest: u128| crate::Error::ChecksumMismatch {
got: crate::Checksum::from_raw(digest),
expected: manifest_checksum,
};
let Some(bound) = restriction else {
return match compute_table_checksum(&**fs, tmp_path) {
Ok(digest) if crate::Checksum::from_raw(digest) == manifest_checksum => Ok(true),
Ok(digest) => condemn_only_if_proven(digest_mismatch(digest)),
Err(e) if is_environmental(&e) => Err(e),
Err(temp_err) => condemn_only_if_proven(temp_err),
};
};
let table = match crate::table::Table::recover(repair_recover_params(
config,
tmp_path.to_path_buf(),
manifest_checksum,
table_id,
Arc::clone(fs),
None,
)) {
Ok(table) => table,
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => return condemn_only_if_proven(e),
};
let punch_offset = match table.punch_offset_for(bound.as_ref()) {
Ok(offset) => offset,
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => return condemn_only_if_proven(e),
};
match compute_table_checksum_from(&**fs, tmp_path, punch_offset) {
Ok(digest) if crate::Checksum::from_raw(digest) == manifest_checksum => Ok(true),
Ok(digest) => condemn_only_if_proven(digest_mismatch(digest)),
Err(e) if is_environmental(&e) => Err(e),
Err(temp_err) => condemn_only_if_proven(temp_err),
}
}
#[cfg(feature = "std")]
fn restricted_suffix_digest(
config: &Config,
fs: &Arc<dyn crate::fs::Fs>,
table_path: &std::path::Path,
table_id: TableId,
bound: &crate::UserKey,
) -> crate::Result<Option<crate::Checksum>> {
let table = match Table::recover(repair_recover_params(
config,
table_path.to_path_buf(),
crate::Checksum::from_raw(0),
table_id,
Arc::clone(fs),
None,
)) {
Ok(table) => table,
Err(e) if is_environmental(&e) => return Err(e),
Err(_) => return Ok(None),
};
match table.suffix_checksum_for(Some(bound)) {
Ok(d) => Ok(Some(d)),
Err(e) if is_environmental(&e) => Err(e),
Err(_) => Ok(None),
}
}
#[cfg(feature = "std")]
fn trustworthy_restriction_bound(
config: &Config,
fs: &dyn crate::fs::Fs,
table_path: &std::path::Path,
table_id: TableId,
manifest_restriction: &ManifestRestriction,
) -> crate::Result<Option<crate::UserKey>> {
match manifest_restriction {
ManifestRestriction::Restricted(bound) => Ok(Some(bound.clone())),
ManifestRestriction::Unrestricted => Ok(None),
ManifestRestriction::Unknown => {
match crate::restrict_bound::read(fs, table_path, config.encryption.as_deref()) {
Ok(crate::restrict_bound::SidecarRead::Present(id, b)) if id == table_id => {
Ok(Some(b.into()))
}
Err(e) if is_environmental(&e) => Err(e),
Ok(_) | Err(_) => Ok(None),
}
}
}
}
#[cfg(feature = "std")]
fn repair_recover_params(
config: &Config,
file_path: PathBuf,
checksum: crate::Checksum,
table_id: TableId,
fs: Arc<dyn crate::fs::Fs>,
global_seqno: Option<SeqNo>,
) -> crate::table::RecoverParams {
let mut params = crate::table::RecoverParams::new(
file_path,
checksum,
table_id,
fs,
config.comparator.clone(),
config.cache.clone(),
);
if let Some(g) = global_seqno {
params.global_seqno = g;
}
params.encryption.clone_from(&config.encryption);
#[cfg(zstd_any)]
{
params.zstd_dictionary.clone_from(&config.zstd_dictionary);
}
params
}
fn is_environmental(e: &crate::Error) -> bool {
e.is_environmental()
}
fn has_unrecoverable_ingest_offset(
bulk_ingested: Option<bool>,
item_count: u64,
max_local_seqno: crate::SeqNo,
) -> bool {
match bulk_ingested {
Some(flagged) => flagged,
None => item_count > 0 && max_local_seqno == 0,
}
}
#[derive(Copy, Clone, PartialEq, Eq)]
enum Fidelity {
Complete,
GeometryRestricted,
Salvaged,
}
impl Fidelity {
fn is_complete(self) -> bool {
matches!(self, Self::Complete)
}
}
struct TableCandidate {
table: Table,
fidelity: Fidelity,
fs: Arc<dyn crate::fs::Fs>,
path: PathBuf,
matches_manifest: bool,
}
#[must_use = "the displaced duplicate's file must be removed from tables/"]
fn keep_best_candidate(
map: &mut crate::HashMap<TableId, TableCandidate>,
id: TableId,
candidate: TableCandidate,
) -> Option<TableCandidate> {
let keeps_existing = match map.get(&id) {
Some(existing) if existing.fidelity.is_complete() && candidate.fidelity.is_complete() => {
existing.matches_manifest || !candidate.matches_manifest
}
Some(existing) if existing.fidelity.is_complete() => true,
Some(_) if candidate.fidelity.is_complete() => false,
Some(existing) => candidate.table.metadata.item_count <= existing.table.metadata.item_count,
None => false,
};
if keeps_existing {
return Some(candidate);
}
map.insert(id, candidate)
}
#[cfg(feature = "std")]
fn remove_temp(config: &Config, temp: &std::path::Path) -> crate::Result<()> {
match config.fs.remove_file(temp) {
Ok(()) => Ok(()),
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => Ok(()),
Err(e) => {
log::error!(
"repair: cannot remove salvage temp {} ({e}); failing the repair — \
left in place, the next open's orphan sweep would hit the same error",
temp.display(),
);
Err(e.into())
}
}
}
#[cfg(feature = "std")]
fn same_physical_file(
fs_a: &dyn crate::fs::Fs,
a: &std::path::Path,
fs_b: &dyn crate::fs::Fs,
b: &std::path::Path,
) -> crate::Result<bool> {
match (fs_a.backend_id(), fs_b.backend_id()) {
(Some(id_a), Some(id_b)) if id_a == id_b => {}
(Some(_), Some(_)) => return Ok(false),
_ => {
if !core::ptr::addr_eq(
core::ptr::from_ref::<dyn crate::fs::Fs>(fs_a),
core::ptr::from_ref::<dyn crate::fs::Fs>(fs_b),
) {
return Err(crate::Error::Io(crate::io::Error::new(
crate::io::ErrorKind::InvalidInput,
"alias identity is inconclusive: the backend reports no namespace id",
)));
}
}
}
Ok(fs_a.same_file(a, b)?)
}
#[cfg(feature = "std")]
fn discard_duplicate(
loser: TableCandidate,
unreadable_files: &mut Vec<(PathBuf, String)>,
redundant_unreadable: &mut crate::HashSet<PathBuf>,
discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
let TableCandidate {
table, fs, path, ..
} = loser;
drop(table); let reason = "duplicate table id superseded by another copy";
discard_after_commit.push((fs, path.clone(), reason.to_string()));
redundant_unreadable.insert(path.clone());
unreadable_files.push((path, reason.to_string()));
}
#[cfg(feature = "std")]
#[expect(
clippy::too_many_arguments,
reason = "location + report threaded through"
)]
fn keep_salvaged_replacement(
map: &mut crate::HashMap<TableId, TableCandidate>,
unreadable_files: &mut Vec<(PathBuf, String)>,
redundant_unreadable: &mut crate::HashSet<PathBuf>,
discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
swap_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
id: TableId,
table: Table,
fs: &Arc<dyn crate::fs::Fs>,
table_path: &std::path::Path,
output_path: PathBuf,
) -> crate::Result<()> {
let restricted = table.restrict_lower_bound().is_some();
swap_after_commit.push((
Arc::clone(fs),
output_path,
table_path.to_path_buf(),
restricted,
));
record_best(
map,
unreadable_files,
redundant_unreadable,
discard_after_commit,
swap_after_commit,
id,
table,
Fidelity::Salvaged,
fs,
table_path,
false,
)
}
#[cfg(feature = "std")]
#[expect(
clippy::too_many_arguments,
reason = "location + report threaded through"
)]
fn record_best(
map: &mut crate::HashMap<TableId, TableCandidate>,
unreadable_files: &mut Vec<(PathBuf, String)>,
redundant_unreadable: &mut crate::HashSet<PathBuf>,
discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
swap_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
id: TableId,
table: Table,
fidelity: Fidelity,
fs: &Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
matches_manifest: bool,
) -> crate::Result<()> {
let candidate = TableCandidate {
table,
fidelity,
fs: Arc::clone(fs),
path: path.to_path_buf(),
matches_manifest,
};
if let Some(loser) = keep_best_candidate(map, id, candidate) {
let is_alias = match map.get(&id) {
Some(kept) => same_physical_file(&*loser.fs, &loser.path, &*kept.fs, &kept.path)?,
None => false,
};
let mut orphaned_temps = Vec::new();
swap_after_commit.retain(|(fs, temp, dest, _)| {
let publishes_loser = dest == &loser.path;
if publishes_loser {
orphaned_temps.push((Arc::clone(fs), temp.clone()));
}
!publishes_loser
});
for (fs, temp) in orphaned_temps {
discard_after_commit.push((
fs,
temp,
"unpublished replacement of a displaced duplicate".to_string(),
));
}
if is_alias {
return Ok(());
}
discard_duplicate(
loser,
unreadable_files,
redundant_unreadable,
discard_after_commit,
);
}
Ok(())
}
fn is_corruption(res: crate::Result<()>) -> crate::Result<bool> {
match res {
Ok(()) => Ok(false),
Err(e) if is_environmental(&e) => Err(e),
Err(_) => Ok(true),
}
}
fn block_verify_verdict(
config: &Config,
folder_fs: &Arc<dyn crate::fs::Fs>,
table_path: &std::path::Path,
table: &Table,
) -> crate::Result<BlockVerifyVerdict> {
let data_start = table.punch_offset()?;
let report = crate::verify::verify_sst_file_with_context(
folder_fs,
table_path,
config.encryption.as_ref(),
Some(table.metadata.id),
data_start,
);
for e in &report.errors {
if let crate::verify::BlockVerifyError::SstFileUnreadable { error, .. }
| crate::verify::BlockVerifyError::DataReadError { error, .. } = e
&& error.kind().is_environmental()
{
return Err(crate::Error::Io(crate::io::Error::new(
error.kind(),
error.to_string(),
)));
}
}
let verdict = if !report
.errors
.iter()
.all(|e| matches!(e, crate::verify::BlockVerifyError::EccParityMismatch { .. }))
{
BlockVerifyVerdict::Corrupt
} else if report
.warnings
.iter()
.any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. }))
{
BlockVerifyVerdict::DegradedUnscanned
} else if is_corruption(table.verify_blob_links())? {
BlockVerifyVerdict::Corrupt
} else if is_corruption(table.verify_tli_mirrors())? {
BlockVerifyVerdict::Corrupt
} else if is_corruption(table.verify_block_layout())? {
BlockVerifyVerdict::Corrupt
} else if is_corruption(
table
.verify_reconcile_gates(config.prefix_extractor.as_ref(), false)
.map_err(|(_, e)| e),
)? {
BlockVerifyVerdict::Corrupt
} else if !report.is_ok() {
BlockVerifyVerdict::DegradedButReadable
} else if report.has_warnings() {
BlockVerifyVerdict::DegradedButReadable
} else {
BlockVerifyVerdict::Clean
};
Ok(verdict)
}
enum BlockVerifyVerdict {
Clean,
DegradedButReadable,
DegradedUnscanned,
Corrupt,
}
#[derive(Debug)]
enum RepairKeepDecision {
Keep,
Salvage,
Drop(&'static str),
}
pub(crate) fn toc_may_hide_deletions(
folder_fs: &Arc<dyn crate::fs::Fs>,
table_path: &std::path::Path,
) -> crate::Result<bool> {
let mut file = match folder_fs.open(table_path, &crate::fs::FsOpenOptions::new().read(true)) {
Ok(file) => file,
Err(e) => {
let err = crate::Error::Io(e);
return if is_environmental(&err) {
Err(err)
} else {
Ok(true)
};
}
};
match crate::sfa::Reader::from_reader(&mut file) {
Ok(reader) => Ok(crate::verify::toc_may_hide_deletion_section(
reader.toc(),
reader.toc_pos(),
)),
Err(crate::sfa::Error::Io(e)) => {
let err = crate::Error::Io(e);
if is_environmental(&err) {
Err(err)
} else {
Ok(true)
}
}
Err(_) => Ok(true),
}
}
fn verify_keep_decision(
config: &Config,
folder_fs: &Arc<dyn crate::fs::Fs>,
table_path: &std::path::Path,
table: &Table,
allow_resurrection: bool,
salvage: bool,
) -> crate::Result<RepairKeepDecision> {
Ok(
match block_verify_verdict(config, folder_fs, table_path, table)? {
BlockVerifyVerdict::Clean => RepairKeepDecision::Keep,
BlockVerifyVerdict::Corrupt => {
if toc_may_hide_deletions(folder_fs, table_path)? && !allow_resurrection {
RepairKeepDecision::Drop(
"TOC corruption may hide deletion metadata (range tombstones \
/ delete bitmap); its visibility is unrecoverable, so the table \
is excluded to avoid resurrecting masked rows. Enable \
resurrection to salvage it, accepting that suppressed rows \
reappear",
)
} else if salvage {
RepairKeepDecision::Salvage
} else {
RepairKeepDecision::Drop(
"verification found corrupt data blocks; run a salvage-enabled \
repair to rewrite the readable blocks",
)
}
}
BlockVerifyVerdict::DegradedButReadable => {
if salvage && table.range_tombstones().is_empty() {
RepairKeepDecision::Salvage
} else {
log::warn!(
"table {} at {}: every payload verified clean but its ECC is \
partially uncheckable or rotted, and this repair cannot rewrite it \
(salvage off, or range tombstones it cannot re-emit) — keeping the \
table as-is; a patrol heal or recompaction re-stamps it under \
fresh, verifiable parity",
table.metadata.id,
table_path.display(),
);
RepairKeepDecision::Keep
}
}
BlockVerifyVerdict::DegradedUnscanned => {
if salvage && table.range_tombstones().is_empty() {
RepairKeepDecision::Salvage
} else if salvage {
RepairKeepDecision::Drop(
"ECC descriptor unrecognized (the block walk cannot verify the \
table) and salvage cannot re-emit its range tombstones; the table \
is excluded (recompact it under a supported scheme to re-admit it)",
)
} else {
RepairKeepDecision::Drop(
"ECC descriptor unrecognized (the block walk cannot verify the \
table); run a salvage-enabled repair to rewrite it under fresh, \
verifiable parity",
)
}
}
},
)
}
enum SalvageOutcome {
Salvaged(Table),
Unusable,
PunchedBoundLost,
}
#[cfg(feature = "std")]
struct TableSalvage<'a> {
source: &'a std::path::Path,
table_path: &'a std::path::Path,
table_id: TableId,
reject_punched_without_bound: bool,
blob_rewrite:
Option<Arc<crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>>>,
recovered_global_seqno: Option<SeqNo>,
}
fn try_salvage_table(
config: &Config,
fs: &Arc<dyn crate::fs::Fs>,
allow_resurrection: bool,
salvage: TableSalvage<'_>,
) -> crate::Result<SalvageOutcome> {
let TableSalvage {
source,
table_path,
table_id,
reject_punched_without_bound,
blob_rewrite,
recovered_global_seqno,
} = salvage;
let report = crate::salvage::salvage_with_context(
source,
table_path.to_path_buf(),
fs,
&config.comparator,
&crate::salvage::SalvageOptions {
encryption: config.encryption.clone(),
#[cfg(zstd_any)]
zstd_dictionary: config.zstd_dictionary.clone(),
table_id,
expected_stored_id: Some(table_id),
allow_delete_resurrection: allow_resurrection,
sync_mode: config.sync_mode,
prefix_extractor: config.prefix_extractor.clone(),
blob_rewrite,
output_id: None,
progress: config.recovery_progress.clone(),
},
)?;
if report.salvaged_path.is_none() {
return Ok(SalvageOutcome::Unusable);
}
if !report.dropped.is_empty() {
log::warn!(
"salvaged table {table_id}: recovered {} block(s), dropped {} corrupt block(s)",
report.blocks_salvaged,
report.dropped.len(),
);
}
if reject_punched_without_bound
&& dropped_data_extent_is_zeroed(&**fs, source, &report.dropped)?
{
discard_unreferenced(&**fs, table_path, config.sync_mode)?;
return Ok(SalvageOutcome::PunchedBoundLost);
}
let checksum = crate::Checksum::from_raw(compute_table_checksum(&**fs, table_path)?);
let table = match Table::recover(repair_recover_params(
config,
table_path.to_path_buf(),
checksum,
table_id,
Arc::clone(fs),
recovered_global_seqno,
)) {
Ok(table) => table,
Err(e) => {
if let Err(rm) = discard_unreferenced(&**fs, table_path, config.sync_mode) {
log::error!(
"salvaged copy {} could not be removed after its reopen failed ({rm}); \
it stays until the next orphan sweep",
table_path.display(),
);
}
return Err(e);
}
};
if recovered_global_seqno.is_none()
&& has_unrecoverable_ingest_offset(
table.metadata.bulk_ingested,
table.metadata.item_count,
table.max_local_seqno(),
)
{
drop(table);
discard_unreferenced(&**fs, table_path, config.sync_mode)?;
return Ok(SalvageOutcome::Unusable);
}
Ok(SalvageOutcome::Salvaged(table))
}
#[cfg(feature = "std")]
fn dropped_data_extent_is_zeroed(
fs: &dyn crate::fs::Fs,
source: &std::path::Path,
dropped: &[crate::salvage::DroppedBlock],
) -> crate::Result<bool> {
let scan = excised_extents(fs, source, dropped)?;
Ok(!scan.proven.is_empty() || scan.unattributed)
}
#[cfg(feature = "std")]
struct ExcisedScan {
proven: Vec<(u64, u64)>,
unattributed: bool,
}
#[cfg(feature = "std")]
fn excised_extents(
fs: &dyn crate::fs::Fs,
source: &std::path::Path,
dropped: &[crate::salvage::DroppedBlock],
) -> crate::Result<ExcisedScan> {
const MIN_RUN: u64 = crate::table::block::Header::MIN_LEN as u64;
let mut excised: Vec<(u64, u64)> = Vec::new();
if dropped.is_empty() {
return Ok(ExcisedScan {
proven: excised,
unattributed: false,
});
}
let mut file = fs.open(source, &crate::fs::FsOpenOptions::new().read(true))?;
let file_len = crate::fs::FsFile::metadata(&*file)?.len;
let data_end = match crate::sfa::Reader::from_reader(&mut file) {
Ok(reader) => reader
.toc()
.iter()
.find(|e| e.name() == b"data")
.map_or(file_len, |e| e.pos().saturating_add(e.len()).min(file_len)),
Err(_) => file_len,
};
let mut starts: Vec<u64> = dropped
.iter()
.filter(|d| d.section == b"data" && d.offset < data_end)
.map(|d| d.offset)
.collect();
starts.sort_unstable();
starts.dedup();
let header_decodes_at = |pos: u64| -> crate::Result<bool> {
use crate::coding::Decode;
let max = crate::table::block::Header::MAX_LEN as u64;
let want = usize::try_from(file_len.saturating_sub(pos).min(max)).unwrap_or(0);
if want < crate::table::block::Header::MIN_LEN {
return Ok(false);
}
let bytes = crate::file::read_exact(&*file, pos, want)?;
Ok(crate::table::block::Header::decode_from(&mut &bytes[..]).is_ok())
};
const CHUNK: usize = 64 * 1024;
for (i, &start) in starts.iter().enumerate() {
let end = starts.get(i + 1).copied().unwrap_or(data_end).min(data_end);
let mut offset = start;
let mut run: u64 = 0;
while offset < end {
let want = usize::try_from(end - offset).unwrap_or(CHUNK).min(CHUNK);
let bytes = crate::file::read_exact(&*file, offset, want)?;
for (j, &b) in bytes.iter().enumerate() {
if b == 0 {
run += 1;
} else {
let run_end = offset + j as u64;
if run >= MIN_RUN && header_decodes_at(run_end)? {
excised.push((run_end - run, run_end));
}
run = 0;
}
}
offset += want as u64;
}
if run >= MIN_RUN {
excised.push((end - run, end));
}
}
excised.sort_unstable();
let mut merged: Vec<(u64, u64)> = Vec::with_capacity(excised.len());
for (start, end) in excised {
match merged.last_mut() {
Some(last) if start <= last.1 => last.1 = last.1.max(end),
_ => merged.push((start, end)),
}
}
let mut proven = Vec::with_capacity(merged.len());
let mut unattributed = false;
for (start, end) in merged {
match fs.extent_contains_hole(source, start, end - start)? {
Some(true) => proven.push((start, end)),
Some(false) => {}
None => unattributed = true,
}
}
if unattributed {
unattributed = fs.capabilities(source).punch_hole;
}
Ok(ExcisedScan {
proven,
unattributed,
})
}
#[cfg(feature = "std")]
fn source_prefix_is_punched(
fs: &dyn crate::fs::Fs,
table_path: &std::path::Path,
) -> crate::Result<bool> {
const PROBE: usize = 64;
let file = fs.open(table_path, &crate::fs::FsOpenOptions::new().read(true))?;
let len = crate::fs::FsFile::metadata(&*file)?.len;
if len == 0 {
return Ok(false);
}
let n = usize::try_from(len).unwrap_or(PROBE).min(PROBE);
let bytes = crate::file::read_exact(&*file, 0, n)?;
if !bytes.iter().all(|&b| b == 0) {
return Ok(false);
}
Ok(fs.extent_contains_hole(table_path, 0, n as u64)? == Some(true))
}
#[cfg(feature = "std")]
fn restrict_salvaged_output(
folder_fs: &dyn crate::fs::Fs,
config: &Config,
table_path: &std::path::Path,
salvaged: Table,
restrict_bound: Option<crate::UserKey>,
allow_resurrection: bool,
) -> crate::Result<Table> {
match restrict_bound {
Some(bound) if !allow_resurrection => {
let table_id = salvaged.metadata.id;
let restricted = crate::restrict_bound::write(
folder_fs,
table_path,
config.encryption.as_deref(),
table_id,
&bound,
config.sync_mode,
)
.and_then(|()| salvaged.reopen_restricted(bound));
match restricted {
Ok(table) => Ok(table),
Err(e) => {
drop(salvaged);
discard_unreferenced(folder_fs, table_path, config.sync_mode)?;
Err(e)
}
}
}
_ => {
crate::restrict_bound::remove(folder_fs, table_path, config.sync_mode);
Ok(salvaged)
}
}
}
#[cfg(feature = "std")]
struct BlobLiveTotals {
items: u64,
uncompressed_bytes: u64,
compressed_bytes: u64,
}
#[cfg(feature = "std")]
#[derive(Debug)]
enum BlobFrontier {
Whole,
Punched(u64),
FullyConsumed,
}
fn derive_blob_frontier(
fs: &Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
blob_id: crate::vlog::BlobFileId,
) -> crate::Result<BlobFrontier> {
let mut file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
let (data_start, data_end) = {
let reader = crate::sfa::Reader::from_reader(&mut file)?;
let data = reader
.toc()
.section(b"data")
.ok_or(crate::Error::InvalidHeader("BlobFile"))?;
let end = data
.pos()
.checked_add(data.len())
.ok_or(crate::Error::InvalidHeader("BlobFile"))?;
(data.pos(), end)
};
if data_start >= data_end {
return Ok(BlobFrontier::Whole);
}
let skip_zeros = |from: u64| -> crate::Result<u64> {
const CHUNK: u64 = 64 * 1024;
let mut pos = from;
while pos < data_end {
#[allow(
clippy::cast_possible_truncation,
reason = "min() bounds the window by CHUNK, which fits usize"
)]
let want = (data_end - pos).min(CHUNK) as usize;
let chunk = crate::file::read_exact(&*file, pos, want)?;
match chunk.iter().position(|b| *b != 0) {
Some(hit) => return Ok(pos + hit as u64),
None => pos += want as u64,
}
}
Ok(data_end)
};
let is_reclaimed = |from: u64, to: u64| -> crate::Result<bool> {
if to <= from {
return Ok(false);
}
Ok(fs.extent_contains_hole(path, from, to - from)? == Some(true))
};
if skip_zeros(data_start)? == data_start {
return Ok(BlobFrontier::Whole);
}
let committed = |c: u64| {
if c == 0 {
BlobFrontier::Whole
} else {
BlobFrontier::Punched(c)
}
};
let mut pos = data_start;
let mut anchored: u64 = 0;
loop {
let run_end = skip_zeros(pos)?;
if !is_reclaimed(pos, run_end)? {
return Ok(committed(anchored));
}
if run_end >= data_end {
return Ok(if anchored == 0 {
BlobFrontier::FullyConsumed
} else {
committed(anchored)
});
}
let mut scanner = crate::vlog::BlobFileScanner::resume(path, &**fs, blob_id, run_end)?;
match scanner.next() {
Some(Ok(entry)) if !entry.resynced => {
anchored = run_end;
pos = entry.frame_end;
}
Some(Err(e)) if is_environmental(&e) => return Err(e),
_ => return Ok(committed(anchored)),
}
loop {
match scanner.next() {
None => return Ok(committed(anchored)),
Some(Ok(entry)) if !entry.resynced => pos = entry.frame_end,
Some(Err(e)) if is_environmental(&e) => return Err(e),
Some(Ok(_) | Err(_)) => {
if skip_zeros(pos)? == pos {
return Ok(committed(anchored));
}
break;
}
}
}
}
}
#[cfg(feature = "std")]
fn set_aside_table(
table: Table,
reason: &str,
unreadable_files: &mut Vec<(PathBuf, String)>,
discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
let path = (*table.path).clone();
let fs = table.fs.clone();
drop(table); discard_after_commit.push((fs, path.clone(), reason.to_string()));
unreadable_files.push((path, reason.to_string()));
}
#[cfg(feature = "std")]
fn set_aside_path(
fs: &Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
reason: &str,
unreadable_files: &mut Vec<(PathBuf, String)>,
discard_after_commit: &mut Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
) {
discard_after_commit.push((Arc::clone(fs), path.to_path_buf(), reason.to_string()));
unreadable_files.push((path.to_path_buf(), reason.to_string()));
}
#[cfg(feature = "std")]
fn validate_blob_frames(
config: &Config,
path: &std::path::Path,
blob_id: crate::vlog::BlobFileId,
live_data_start: u64,
handle: &crate::vlog::BlobFile,
) -> crate::Result<Option<BlobLiveTotals>> {
let fs = &config.fs;
let compression = handle.compression();
let comparator = &config.comparator;
let scanner = if live_data_start > 0 {
crate::vlog::BlobFileScanner::resume(path, &**fs, blob_id, live_data_start)?
} else {
crate::vlog::BlobFileScanner::new(path, &**fs, blob_id)?
};
let mut count: u64 = 0;
let mut uncompressed_total: u64 = 0;
let mut compressed_total: u64 = 0;
let mut first_key: Option<crate::UserKey> = None;
let mut prev: Option<(crate::UserKey, crate::SeqNo)> = None;
for item in scanner {
match item {
Ok(entry) if !entry.resynced => {
if crate::salvage::blob_key_regresses(comparator, prev.as_ref(), &entry) {
log::warn!(
"blob file {blob_id} at {}: frame at {} regresses below the \
previous frame's key — the frames were reordered",
path.display(),
entry.offset,
);
return Ok(None);
}
if crate::salvage::decompress_blob_value(
compression,
&entry.value,
entry.uncompressed_len as usize,
#[cfg(zstd_any)]
config.zstd_dictionary.as_deref(),
)
.is_err()
{
log::warn!(
"blob file {blob_id} at {}: frame at {} does not decompress \
despite a clean checksum",
path.display(),
entry.offset,
);
return Ok(None);
}
count += 1;
uncompressed_total += u64::from(entry.uncompressed_len);
compressed_total += entry.value.len() as u64;
if first_key.is_none() {
first_key = Some(entry.key.clone());
}
prev = Some((entry.key.clone(), entry.seqno));
}
Err(e) if is_environmental(&e) => return Err(e),
Ok(_) | Err(_) => return Ok(None),
}
}
let meta = handle.meta();
if live_data_start == 0 {
let range_matches = match (&first_key, &prev) {
(Some(first), Some((last, _))) => {
meta.key_range.min().as_ref() == first.as_ref()
&& meta.key_range.max().as_ref() == last.as_ref()
}
_ => count == 0,
};
if meta.item_count != count
|| meta.total_uncompressed_bytes != uncompressed_total
|| meta.total_compressed_bytes != compressed_total
|| !range_matches
{
log::warn!(
"blob file {blob_id} at {}: metadata disagrees with the scanned frames \
(meta: {} items / {} uncompressed bytes; scanned: {count} / \
{uncompressed_total})",
path.display(),
meta.item_count,
meta.total_uncompressed_bytes,
);
return Ok(None);
}
} else {
let range_contains = match (&first_key, &prev) {
(Some(first), Some((last, _))) => {
comparator.compare(meta.key_range.min().as_ref(), first.as_ref())
!= core::cmp::Ordering::Greater
&& comparator.compare(last.as_ref(), meta.key_range.max().as_ref())
!= core::cmp::Ordering::Greater
}
_ => true,
};
if meta.item_count < count
|| meta.total_uncompressed_bytes < uncompressed_total
|| meta.total_compressed_bytes < compressed_total
|| !range_contains
{
log::warn!(
"blob file {blob_id} at {}: metadata understates the scanned live \
suffix (meta: {} items / {} uncompressed bytes; suffix: {count} / \
{uncompressed_total})",
path.display(),
meta.item_count,
meta.total_uncompressed_bytes,
);
return Ok(None);
}
}
Ok(Some(BlobLiveTotals {
items: count,
uncompressed_bytes: uncompressed_total,
compressed_bytes: compressed_total,
}))
}
#[cfg(feature = "std")]
fn handle_below_blob_frontier(
table: &Table,
frontiers: &crate::HashMap<crate::vlog::BlobFileId, u64>,
) -> crate::Result<Option<String>> {
use crate::coding::Decode;
for entry in table.scan()? {
let entry = entry?;
if entry.key.value_type != crate::ValueType::Indirection {
continue;
}
let mut cursor = &entry.value[..];
let ind = crate::blob_tree::handle::BlobIndirection::decode_from(&mut cursor)?;
if let Some(&frontier) = frontiers.get(&ind.vhandle.blob_file_id)
&& ind.vhandle.offset < frontier
{
return Ok(Some(format!(
"blob handle into file {} at offset {} lies below its recovered \
live-data frontier {frontier}",
ind.vhandle.blob_file_id, ind.vhandle.offset,
)));
}
}
Ok(None)
}
#[cfg(feature = "std")]
struct BlobRecovery {
files: Vec<crate::vlog::BlobFile>,
unreadable: UnreadableFiles,
rewrites: crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite>,
frag: crate::blob_tree::FragmentationMap,
stale: Vec<(PathBuf, String)>,
discard: Vec<(PathBuf, String)>,
excluded: Vec<(PathBuf, String)>,
}
fn recover_blob_files(
config: &Config,
published: &mut PublishedBlobReplacements<'_>,
referenced: &crate::HashSet<crate::vlog::BlobFileId>,
committed_frontiers: Option<&crate::HashMap<crate::vlog::BlobFileId, u64>>,
committed_checksums: Option<&crate::HashMap<crate::vlog::BlobFileId, crate::Checksum>>,
) -> crate::Result<BlobRecovery> {
let blobs_folder = config.path.join(crate::file::BLOBS_FOLDER);
let mut blob_files: Vec<crate::vlog::BlobFile> = Vec::new();
let mut unreadable: UnreadableFiles = Vec::new();
let mut discard: Vec<(PathBuf, String)> = Vec::new();
let mut excluded: Vec<(PathBuf, String)> = Vec::new();
let mut rewrites: crate::HashMap<crate::vlog::BlobFileId, crate::salvage::BlobFileRewrite> =
crate::HashMap::default();
let mut stale_originals: Vec<(PathBuf, String)> = Vec::new();
let mut frag = crate::blob_tree::FragmentationMap::default();
if !config.fs.exists(&blobs_folder)? {
return Ok(BlobRecovery {
files: blob_files,
unreadable,
rewrites,
frag,
stale: stale_originals,
discard,
excluded,
});
}
let mut candidates: Vec<(crate::vlog::BlobFileId, PathBuf, String)> = Vec::new();
for dirent in config.fs.read_dir(&blobs_folder)? {
let crate::fs::FsDirEntry {
path: blob_path,
file_name,
is_dir,
} = dirent;
if is_dir {
continue;
}
if let Some(p) = &config.recovery_progress
&& let Ok(meta) = config.fs.metadata(&blob_path)
{
p.add_bytes_processed(meta.len);
}
let blob_id = match crate::file::BlobDirEntry::classify(&file_name) {
crate::file::BlobDirEntry::Blob(id) => id,
crate::file::BlobDirEntry::SalvageTmp(_) => {
remove_temp(config, &blob_path)?;
continue;
}
crate::file::BlobDirEntry::Foreign => {
log::debug!(
"repair: ignoring {} in the blobs folder: not an engine file",
blob_path.display(),
);
continue;
}
};
candidates.push((blob_id, blob_path, file_name));
}
let mut next_blob_id: Option<crate::vlog::BlobFileId> = candidates
.iter()
.map(|(id, _, _)| *id)
.chain(referenced.iter().copied())
.max()
.map_or(Some(0), |max| max.checked_add(1));
candidates.sort_by(|a, b| {
a.0.cmp(&b.0)
.then_with(|| {
(a.1 != blobs_folder.join(a.0.to_string()))
.cmp(&(b.1 != blobs_folder.join(b.0.to_string())))
})
.then_with(|| a.2.cmp(&b.2))
});
let mut kept_paths: crate::HashMap<crate::vlog::BlobFileId, PathBuf> =
crate::HashMap::default();
let resolve_frontier = |blob_id, blob_path: &std::path::Path| -> crate::Result<BlobFrontier> {
match committed_frontiers.and_then(|m| m.get(&blob_id).copied()) {
Some(committed) => Ok(BlobFrontier::Punched(committed)),
None => derive_blob_frontier(&config.fs, blob_path, blob_id),
}
};
let opens_cleanly = |blob_id, blob_path: &std::path::Path| -> crate::Result<()> {
let frontier = match resolve_frontier(blob_id, blob_path)? {
BlobFrontier::Whole => 0,
BlobFrontier::Punched(f) => f,
BlobFrontier::FullyConsumed => return Ok(()),
};
let checksum = crate::Checksum::from_raw(compute_table_checksum_from(
&*config.fs,
blob_path,
frontier,
)?);
let handle = crate::vlog::recover_blob_file_from(
blob_path, blob_id, checksum, 0, &config.fs, frontier,
)?;
match validate_blob_frames(config, blob_path, blob_id, frontier, &handle)? {
Some(_) => Ok(()),
None => Err(crate::Error::InvalidHeader(
"blob frame validation failed on this copy",
)),
}
};
let candidates = {
let mut ordered: Vec<(crate::vlog::BlobFileId, PathBuf, String)> =
Vec::with_capacity(candidates.len());
let mut rest = candidates.into_iter().peekable();
while let Some(first) = rest.next() {
let blob_id = first.0;
let mut group = vec![first];
while rest.peek().is_some_and(|c| c.0 == blob_id) {
if let Some(c) = rest.next() {
group.push(c);
}
}
if group.len() > 1 {
let committed = committed_checksums.and_then(|m| m.get(&blob_id).copied());
let mut authoritative = None;
let mut whole = None;
for (k, candidate) in group.iter().enumerate() {
if let Some(expected) = committed
&& authoritative.is_none()
{
let frontier = match resolve_frontier(blob_id, &candidate.1)? {
BlobFrontier::Whole => Some(0),
BlobFrontier::Punched(f) => Some(f),
BlobFrontier::FullyConsumed => None,
};
if let Some(frontier) = frontier {
match compute_table_checksum_from(&*config.fs, &candidate.1, frontier) {
Ok(d) if crate::Checksum::from_raw(d) == expected => {
authoritative = Some(k);
continue;
}
Err(e) if is_environmental(&e) => return Err(e),
Ok(_) | Err(_) => {}
}
}
}
if whole.is_none() {
match opens_cleanly(blob_id, &candidate.1) {
Ok(()) => whole = Some(k),
Err(e) if is_environmental(&e) => return Err(e),
Err(_) => {}
}
}
}
if let Some(k) = authoritative.or(whole) {
group.swap(0, k);
}
}
ordered.extend(group);
}
ordered
};
for (blob_id, blob_path, _file_name) in candidates {
if let Some(kept) = kept_paths.get(&blob_id) {
if same_physical_file(&*config.fs, kept, &*config.fs, &blob_path)? {
continue;
}
match opens_cleanly(blob_id, &blob_path) {
Ok(()) => {
let reason = format!("duplicate of blob file id {blob_id}");
discard.push((blob_path.clone(), reason.clone()));
excluded.push((blob_path, reason));
}
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
let reason = format!(
"damaged duplicate of blob file id {blob_id} (kept copy is intact): {e}"
);
discard.push((blob_path.clone(), reason.clone()));
unreadable.push((blob_path, reason));
}
}
continue;
}
check_cancel(config)?;
if let Some(p) = &config.recovery_progress {
p.blob_file_discovered();
}
let discard_unreadable =
|blob_path: PathBuf,
e: &crate::Error,
unreadable: &mut UnreadableFiles,
discard: &mut Vec<(PathBuf, String)>| {
discard.push((blob_path.clone(), e.to_string()));
unreadable.push((blob_path, e.to_string()));
};
let frontier = match resolve_frontier(blob_id, &blob_path) {
Ok(BlobFrontier::Whole) => 0,
Ok(BlobFrontier::Punched(f)) => f,
Ok(BlobFrontier::FullyConsumed) => {
log::info!(
"blob file {blob_id} at {}: its punch consumed every frame — \
queueing the relocation's lagged file drop for after the commit",
blob_path.display(),
);
discard.push((
blob_path,
"fully punched blob file: a completed relocation's lagged drop".to_string(),
));
continue;
}
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
};
let handle = match crate::vlog::recover_blob_file(
&blob_path,
blob_id,
crate::Checksum::from_raw(0),
0,
&config.fs,
) {
Ok(handle) => handle,
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
};
let stored = handle.meta().id;
if stored != blob_id {
let e = crate::Error::InvalidHeader(
"blob file's stored metadata id disagrees with its file name",
);
log::warn!(
"blob file at {}: metadata records id {stored}, file name says \
{blob_id} — a renamed or swapped file; setting it aside",
blob_path.display(),
);
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
let Some(live) = validate_blob_frames(config, &blob_path, blob_id, frontier, &handle)?
else {
if !referenced.contains(&blob_id) {
log::debug!(
"blob file {blob_id} is invalid and referenced by no recovered \
table; skipping its salvage and leaving it for the post-commit \
sweep"
);
discard.push((
blob_path,
"invalid and referenced by no recovered table; salvage skipped".to_string(),
));
continue;
}
let Some(new_id) = next_blob_id else {
log::error!(
"blob file {blob_id} needs salvaging but the blob id space is \
exhausted: there is no fresh name to publish a replacement \
under, and reusing one would destroy an existing file"
);
return Err(crate::Error::Unrecoverable);
};
next_blob_id = new_id.checked_add(1);
let temp = blobs_folder.join(format!("{new_id}.salvage-tmp"));
remove_temp(config, &temp)?;
let salvage = (|| -> crate::Result<Option<(crate::vlog::BlobFile, crate::salvage::BlobSalvageReport)>> {
let report = crate::salvage::salvage_blob_file(
&blob_path,
temp.clone(),
&config.fs,
new_id,
&config.comparator,
frontier,
#[cfg(zstd_any)]
config.zstd_dictionary.as_ref(),
)?;
let Some(salvaged_path) = report.salvaged_path.clone() else {
return Ok(None);
};
let checksum = crate::Checksum::from_raw(compute_table_checksum(
&*config.fs,
&salvaged_path,
)?);
let bf = crate::vlog::recover_blob_file_from(
&salvaged_path,
new_id,
checksum,
0,
&config.fs,
0,
)?;
Ok(Some((bf, report)))
})();
let (bf, report) = match salvage {
Ok(Some(pair)) => pair,
Ok(None) => {
remove_temp(config, &temp)?;
let e = crate::Error::InvalidHeader(
"blob value frames failed validation and no record was recoverable",
);
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
Err(e) if is_environmental(&e) => {
let _ = config.fs.remove_file(&temp);
return Err(e);
}
Err(e) => {
remove_temp(config, &temp)?;
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
};
let new_path = blobs_folder.join(new_id.to_string());
if let Err(e) = config.fs.rename(&temp, &new_path) {
let _ = config.fs.remove_file(&temp);
return Err(e.into());
}
published.publish(new_id);
config
.fs
.sync_directory_with(&blobs_folder, config.sync_mode)?;
blob_files.push(bf);
rewrites.insert(
blob_id,
crate::salvage::BlobFileRewrite::Remap {
new_id,
offsets: report.offset_remap.iter().copied().collect(),
},
);
let surrendered_tail = report
.dropped
.iter()
.any(|d| matches!(d.reason, crate::salvage::BlobDropReason::Corrupt(_)));
let note = if surrendered_tail {
format!(
"{} records salvaged into blob file {new_id}; the record \
stream then desynchronized and the remainder of the file \
was surrendered, so the number of records lost with it is \
not knowable",
report.records_salvaged,
)
} else {
format!(
"{} of {} records salvaged into blob file {new_id} \
(the rest failed their checksums)",
report.records_salvaged, report.records_total,
)
};
stale_originals.push((blob_path.clone(), note));
kept_paths.insert(blob_id, blob_path);
continue;
};
let checksum = match compute_table_checksum_from(&*config.fs, &blob_path, frontier) {
Ok(c) => crate::Checksum::from_raw(c),
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
continue;
}
};
match crate::vlog::recover_blob_file_from(
&blob_path, blob_id, checksum, 0, &config.fs, frontier,
) {
Ok(bf) => {
if frontier > 0 {
rewrites.insert(
blob_id,
crate::salvage::BlobFileRewrite::DropBelow(frontier),
);
let meta = bf.meta();
let prefix_items = meta.item_count - live.items;
let prefix_len =
usize::try_from(prefix_items).map_err(|_| crate::Error::Unrecoverable)?;
frag.insert(
blob_id,
crate::blob_tree::FragmentationEntry::new(
prefix_len,
meta.total_uncompressed_bytes - live.uncompressed_bytes,
meta.total_compressed_bytes - live.compressed_bytes,
),
);
}
kept_paths.insert(blob_id, blob_path);
blob_files.push(bf);
}
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
discard_unreadable(blob_path, &e, &mut unreadable, &mut discard);
}
}
}
Ok(BlobRecovery {
files: blob_files,
unreadable,
rewrites,
frag,
stale: stale_originals,
discard,
excluded,
})
}
impl Config {
pub fn repair(&self) -> crate::Result<RepairReport> {
repair_tree(self, false, false)
}
pub fn repair_with_salvage(&self, salvage: bool) -> crate::Result<RepairReport> {
repair_tree(self, salvage, false)
}
pub fn repair_with_resurrection(
&self,
salvage: bool,
allow_resurrection: bool,
) -> crate::Result<RepairReport> {
repair_tree(self, salvage, allow_resurrection)
}
pub fn open_or_repair(
self,
policy: RepairPolicy,
) -> crate::Result<(crate::AnyTree, Option<RepairReport>)> {
let retry = self.clone();
match self.open() {
Ok(tree) => Ok((tree, None)),
Err(e) if is_repairable_structural(&e) => {
let report =
retry.repair_with_resurrection(policy.salvage, policy.allow_resurrection)?;
let tree = match retry.open() {
Ok(tree) => tree,
Err(cause) => {
return Err(crate::Error::RepairedButUnopened {
report: alloc::boxed::Box::new(report),
cause: alloc::boxed::Box::new(cause),
});
}
};
Ok((tree, Some(report)))
}
Err(e) => Err(e),
}
}
}
fn is_repairable_structural(e: &crate::Error) -> bool {
use crate::Error;
match e {
Error::Io(io) => matches!(
io.kind(),
crate::io::ErrorKind::NotFound
| crate::io::ErrorKind::InvalidData
| crate::io::ErrorKind::UnexpectedEof
),
Error::Decompress(_)
| Error::Excised { .. }
| Error::ChecksumMismatch { .. }
| Error::HeaderCrcMismatch { .. }
| Error::InvalidTag(_)
| Error::InvalidTrailer
| Error::InvalidHeader(_)
| Error::DecompressedSizeTooLarge { .. }
| Error::ManifestFrameChecksumMismatch { .. }
| Error::ManifestFooterInvalid(_)
| Error::ManifestSectionInvalid(_)
| Error::TornManifestEditLog { .. }
| Error::RangeTombstoneDecode { .. }
| Error::PageEccUnrecoverable { .. } => true,
_ => false,
}
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct RepairPolicy {
pub salvage: bool,
pub allow_resurrection: bool,
}
impl RepairPolicy {
#[must_use]
pub const fn salvage(mut self, enable: bool) -> Self {
self.salvage = enable;
self
}
#[must_use]
pub const fn allow_resurrection(mut self, enable: bool) -> Self {
self.allow_resurrection = enable;
self
}
}
struct CommittedManifest {
tables: crate::HashMap<TableId, crate::version::recovery::RecoveredTable>,
restrictions: crate::HashMap<TableId, crate::UserKey>,
blob_frontiers: crate::HashMap<crate::vlog::BlobFileId, u64>,
blob_checksums: crate::HashMap<crate::vlog::BlobFileId, crate::Checksum>,
}
#[cfg(feature = "std")]
impl CommittedManifest {
fn restriction_of(&self, table_id: TableId) -> ManifestRestriction {
if !self.tables.contains_key(&table_id) {
return ManifestRestriction::Unknown;
}
self.restrictions
.get(&table_id)
.map_or(ManifestRestriction::Unrestricted, |bound| {
ManifestRestriction::Restricted(bound.clone())
})
}
}
#[cfg(feature = "std")]
enum ManifestRestriction {
Restricted(crate::UserKey),
Unrestricted,
Unknown,
}
#[cfg(feature = "std")]
fn sweep_superseded_by_committed_manifest(
config: &Config,
) -> crate::Result<Option<CommittedManifest>> {
let recovery = match crate::version::recovery::recover(
&config.path,
&*config.fs,
crate::config::ManifestRecoveryMode::AbsoluteConsistency,
config.encryption.clone(),
) {
Ok(recovery) => recovery,
Err(e) if is_environmental(&e) => return Err(e),
Err(_) => return Ok(None),
};
let requested = if config.kv_separation_opts.is_some() {
TreeType::Blob
} else {
TreeType::Standard
};
if recovery.tree_type != requested {
log::error!(
"repair: the committed manifest describes a {:?} tree but this repair is \
configured for {requested:?}; rebuilding would leave the store openable \
only under the wrong configuration",
recovery.tree_type,
);
return Err(crate::Error::TreeTypeMismatch {
requested,
actual: recovery.tree_type,
});
}
let referenced_tables: crate::HashMap<TableId, crate::version::recovery::RecoveredTable> =
recovery
.table_ids
.iter()
.flatten()
.flatten()
.map(|t| (t.id, *t))
.collect();
for (table_base_folder, folder_fs) in config.all_tables_folders() {
if !folder_fs.exists(&table_base_folder)? {
continue;
}
for dirent in folder_fs.read_dir(&table_base_folder)? {
if dirent.is_dir {
continue;
}
if let Some(id) = table_id_from_repair_tmp_name(&dirent.file_name) {
let table_path = table_base_folder.join(id.to_string());
let published = match referenced_tables.get(&id).map(|t| t.checksum) {
Some(manifest_checksum) => repair_tmp_is_published(
config,
&folder_fs,
&dirent.path,
id,
manifest_checksum,
recovery.restrictions.get(&id),
)?,
None => false,
};
if published {
commit_repair_tmp(
&*folder_fs,
&dirent.path,
&table_path,
config.sync_mode,
recovery.restrictions.contains_key(&id),
)?;
log::info!(
"repair: finished the pending swap of table {id} from a previous run",
);
} else {
discard_unreferenced(&*folder_fs, &dirent.path, config.sync_mode)?;
log::info!("repair: dropped an abandoned replacement for table {id}");
}
continue;
}
let Ok(id) = dirent.file_name.parse::<TableId>() else {
continue;
};
if referenced_tables.contains_key(&id) {
continue;
}
discard_unreferenced(&*folder_fs, &dirent.path, config.sync_mode)?;
log::info!("repair: table {id} is superseded by the committed manifest; removed");
}
}
if config.kv_separation_opts.is_some() {
let referenced_blobs: crate::HashSet<crate::vlog::BlobFileId> =
recovery.blob_file_ids.iter().map(|(id, _)| *id).collect();
let blobs_folder = config.path.join(crate::file::BLOBS_FOLDER);
if config.fs.exists(&blobs_folder)? {
for dirent in config.fs.read_dir(&blobs_folder)? {
let Ok(id) = dirent.file_name.parse::<crate::vlog::BlobFileId>() else {
continue;
};
if dirent.is_dir || referenced_blobs.contains(&id) {
continue;
}
match config.fs.remove_file(&dirent.path) {
Ok(()) => log::info!(
"repair: blob file {id} is superseded by the committed manifest; removed",
),
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => return Err(e.into()),
}
}
}
}
Ok(Some(CommittedManifest {
tables: referenced_tables,
restrictions: recovery.restrictions,
blob_frontiers: recovery.blob_restrictions,
blob_checksums: recovery.blob_file_ids.iter().copied().collect(),
}))
}
fn remove_published_blob_replacements(
config: &Config,
blobs_folder: &std::path::Path,
replacement_ids: impl IntoIterator<Item = crate::vlog::BlobFileId>,
) -> crate::Result<()> {
let mut removed = false;
for id in replacement_ids {
match config.fs.remove_file(&blobs_folder.join(id.to_string())) {
Ok(()) => removed = true,
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => return Err(e.into()),
}
}
if removed {
config
.fs
.sync_directory_with(blobs_folder, config.sync_mode)?;
}
Ok(())
}
struct PublishedBlobReplacements<'a> {
config: &'a Config,
blobs_folder: PathBuf,
ids: Vec<crate::vlog::BlobFileId>,
armed: bool,
}
impl<'a> PublishedBlobReplacements<'a> {
fn new(config: &'a Config) -> Self {
Self {
config,
blobs_folder: config.path.join(crate::file::BLOBS_FOLDER),
ids: Vec::new(),
armed: true,
}
}
fn publish(&mut self, id: crate::vlog::BlobFileId) {
self.ids.push(id);
}
fn disarm(&mut self) {
self.armed = false;
}
}
enum CurrentProbe {
Switched,
NotSwitched,
Inconclusive,
}
fn probe_current(
fs: &dyn crate::fs::Fs,
folder: &std::path::Path,
version_id: u64,
) -> CurrentProbe {
let file = match fs.open(
&folder.join(crate::file::CURRENT_VERSION_FILE),
&crate::fs::FsOpenOptions::new().read(true),
) {
Ok(file) => file,
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => return CurrentProbe::NotSwitched,
Err(_) => return CurrentProbe::Inconclusive,
};
let Ok(bytes) = crate::file::read_exact(&*file, 0, 8) else {
return CurrentProbe::Inconclusive;
};
let Ok(raw) = <[u8; 8]>::try_from(bytes.as_ref()) else {
return CurrentProbe::Inconclusive;
};
if u64::from_le_bytes(raw) == version_id {
CurrentProbe::Switched
} else {
CurrentProbe::NotSwitched
}
}
impl Drop for PublishedBlobReplacements<'_> {
fn drop(&mut self) {
if !self.armed || self.ids.is_empty() {
return;
}
if let Err(e) =
remove_published_blob_replacements(self.config, &self.blobs_folder, self.ids.drain(..))
{
log::warn!(
"repair: could not remove a published blob replacement while \
aborting ({e}); the next open's orphan sweep removes it",
);
}
}
}
fn check_cancel(config: &Config) -> crate::Result<()> {
if let Some(p) = &config.recovery_progress
&& p.is_cancel_requested()
{
return Err(crate::Error::Cancelled);
}
Ok(())
}
fn publish_recovery_bytes_total(config: &Config) {
let Some(progress) = &config.recovery_progress else {
return;
};
let mut total: u64 = 0;
let mut add_folder = |folder: &std::path::Path, fs: &dyn crate::fs::Fs| {
let Ok(true) = fs.exists(folder) else {
return;
};
let Ok(dirents) = fs.read_dir(folder) else {
return;
};
for dirent in dirents {
if dirent.is_dir {
continue;
}
if let Ok(meta) = fs.metadata(&dirent.path) {
total = total.saturating_add(meta.len);
}
}
};
for (folder, fs) in config.all_tables_folders() {
add_folder(&folder, &*fs);
}
if config.kv_separation_opts.is_some() {
add_folder(&config.path.join(crate::file::BLOBS_FOLDER), &*config.fs);
}
progress.set_bytes_total(total);
}
fn repair_tree(
config: &Config,
salvage: bool,
allow_resurrection: bool,
) -> crate::Result<RepairReport> {
#[cfg(feature = "std")]
let _directory_lock =
crate::config::acquire_directory_lock(&*config.fs, &config.path, config.directory_lock)?;
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::PendingSwaps);
}
check_cancel(config)?;
#[cfg(feature = "std")]
let manifest_referenced: Option<CommittedManifest> =
sweep_superseded_by_committed_manifest(config)?;
#[cfg(not(feature = "std"))]
let manifest_referenced: Option<CommittedManifest> = None;
publish_recovery_bytes_total(config);
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::ScanningTables);
}
let scan = scan_table_folders(
config,
salvage,
allow_resurrection,
manifest_referenced.as_ref(),
)?;
rebuild_from_scan(config, allow_resurrection, manifest_referenced, scan)
}
#[cfg(feature = "std")]
struct TableScan {
recovered_by_id: crate::HashMap<TableId, TableCandidate>,
unreadable_files: Vec<(PathBuf, String)>,
redundant_unreadable: crate::HashSet<PathBuf>,
excluded_files: Vec<(PathBuf, String)>,
coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)>,
discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
live_sidecars: Vec<(TableId, PathBuf, Arc<dyn crate::fs::Fs>)>,
scanned_table_ids: crate::HashSet<TableId>,
}
#[cfg(feature = "std")]
fn scan_table_folders(
config: &Config,
salvage: bool,
allow_resurrection: bool,
manifest_referenced: Option<&CommittedManifest>,
) -> crate::Result<TableScan> {
let mut scanned_table_ids: crate::HashSet<TableId> = crate::HashSet::default();
let mut recovered_by_id: crate::HashMap<TableId, TableCandidate> = crate::HashMap::default();
let mut unreadable_files: Vec<(PathBuf, String)> = Vec::new();
let mut redundant_unreadable: crate::HashSet<PathBuf> = crate::HashSet::default();
let mut excluded_files: Vec<(PathBuf, String)> = Vec::new();
let mut coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)> =
crate::HashMap::default();
let mut discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)> = Vec::new();
let mut swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)> = Vec::new();
let mut live_sidecars: Vec<(TableId, PathBuf, Arc<dyn crate::fs::Fs>)> = Vec::new();
for (table_base_folder, folder_fs) in config.all_tables_folders() {
if !folder_fs.exists(&table_base_folder)? {
continue;
}
let mut dirents = folder_fs.read_dir(&table_base_folder)?;
dirents.sort_by(|a, b| {
let key = |e: &crate::fs::FsDirEntry| {
let numeric = e.file_name.parse::<TableId>().ok();
let temp = table_id_from_repair_tmp_name(&e.file_name);
let id = numeric.or(temp);
let rank = if temp.is_some() {
0
} else if numeric.is_some_and(|id| e.file_name == id.to_string()) {
1
} else {
2
};
(id.is_none(), id.unwrap_or(0), rank)
};
key(a)
.cmp(&key(b))
.then_with(|| a.file_name.cmp(&b.file_name))
});
'dirent: for dirent in dirents {
let crate::fs::FsDirEntry {
path: table_path,
file_name,
is_dir,
} = dirent;
if is_dir {
continue;
}
check_cancel(config)?;
if let Some(p) = &config.recovery_progress
&& let Ok(meta) = folder_fs.metadata(&table_path)
{
p.add_bytes_processed(meta.len);
}
use crate::file::TableDirEntry;
let table_id = match TableDirEntry::classify(&file_name) {
TableDirEntry::HealAttest(id) | TableDirEntry::RestrictBound(id) => {
live_sidecars.push((id, table_path, Arc::clone(&folder_fs)));
continue;
}
TableDirEntry::HealAttestTmp(_)
| TableDirEntry::HealTmp(_)
| TableDirEntry::RestrictBoundTmp(_) => {
discard_after_commit.push((
Arc::clone(&folder_fs),
table_path,
"disposable crashed-heal artifact; the next open would \
sweep it and fail on a refused removal"
.to_string(),
));
continue;
}
TableDirEntry::RepairTmpCompanion(_) => {
discard_after_commit.push((
Arc::clone(&folder_fs),
table_path,
"restriction sidecar of an abandoned repair replacement".to_string(),
));
continue;
}
TableDirEntry::RepairTmp(_) => {
if let Some(p) = &config.recovery_progress
&& let Ok(meta) =
folder_fs.metadata(&crate::restrict_bound::sidecar_path(&table_path))
{
p.add_bytes_processed(meta.len);
}
discard_unreferenced(&*folder_fs, &table_path, config.sync_mode)?;
log::warn!(
"repair: dropped an abandoned replacement {}",
table_path.display(),
);
continue;
}
TableDirEntry::Foreign => {
log::debug!(
"repair: ignoring {} in the tables folder: not an engine file",
table_path.display(),
);
continue;
}
TableDirEntry::Table(id) => id,
};
scanned_table_ids.insert(table_id);
if let Some(p) = &config.recovery_progress {
p.table_discovered();
}
let manifest_global_seqno: Option<SeqNo> = manifest_referenced
.as_ref()
.and_then(|m| m.tables.get(&table_id))
.map(|t| t.global_seqno);
let manifest_restriction = manifest_referenced
.as_ref()
.map_or(ManifestRestriction::Unknown, |m| m.restriction_of(table_id));
let own_digest = match compute_table_checksum(&*folder_fs, &table_path) {
Ok(c) => Ok(crate::Checksum::from_raw(c)),
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => Err(e),
};
let committed_digest = manifest_referenced
.as_ref()
.and_then(|m| m.tables.get(&table_id))
.map(|t| t.checksum);
let matches_manifest = match (&own_digest, committed_digest) {
(Ok(d), Some(committed)) => {
match trustworthy_restriction_bound(
config,
&*folder_fs,
&table_path,
table_id,
&manifest_restriction,
)? {
Some(bound) => restricted_suffix_digest(
config,
&folder_fs,
&table_path,
table_id,
&bound,
)?
.is_some_and(|suffix| suffix == committed),
None => *d == committed,
}
}
_ => false,
};
if let Some(existing) = recovered_by_id
.get(&table_id)
.filter(|c| c.fidelity.is_complete())
.filter(|c| c.matches_manifest || !matches_manifest)
{
if same_physical_file(&*folder_fs, &table_path, &*existing.fs, &existing.path)? {
continue;
}
let verdict = match own_digest {
Ok(digest) => Table::recover(repair_recover_params(
config,
table_path.clone(),
digest,
table_id,
folder_fs.clone(),
manifest_global_seqno,
))
.and_then(|table| {
let table = match trustworthy_restriction_bound(
config,
&*folder_fs,
&table_path,
table_id,
&manifest_restriction,
)? {
Some(bound) => table.reopen_restricted(bound)?,
None => table,
};
match block_verify_verdict(config, &folder_fs, &table_path, &table)? {
BlockVerifyVerdict::Clean | BlockVerifyVerdict::DegradedButReadable => {
Ok(())
}
BlockVerifyVerdict::Corrupt => Err(crate::Error::InvalidHeader(
"block verification failed on this copy",
)),
BlockVerifyVerdict::DegradedUnscanned => Err(
crate::Error::InvalidHeader("this copy could not be verified"),
),
}
}),
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => Err(e),
};
let verdict = match verdict {
Err(e) if is_environmental(&e) => return Err(e),
other => other,
};
match verdict {
Ok(()) => {
let reason = "duplicate table id; a complete copy is already held";
excluded_files.push((table_path.clone(), reason.to_string()));
discard_after_commit.push((
Arc::clone(&folder_fs),
table_path,
reason.to_string(),
));
}
Err(e) => {
let reason = format!(
"damaged duplicate of table {table_id} (kept copy is intact): {e}"
);
redundant_unreadable.insert(table_path.clone());
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
continue;
}
let recovered = match own_digest {
Ok(digest) => Table::recover(repair_recover_params(
config,
table_path.clone(),
digest,
table_id,
folder_fs.clone(),
manifest_global_seqno,
)),
Err(e) => Err(e),
};
if let Ok(t) = &recovered {
let range = t.metadata.key_range.clone();
let seqno = (manifest_global_seqno.is_some()
|| !has_unrecoverable_ingest_offset(
t.metadata.bulk_ingested,
t.metadata.item_count,
t.max_local_seqno(),
))
.then(|| t.get_highest_seqno());
coverage_by_path.insert(
table_path.clone(),
(range.min().clone(), range.max().clone(), seqno),
);
}
if manifest_global_seqno.is_none()
&& matches!(&recovered, Ok(t) if has_unrecoverable_ingest_offset(
t.metadata.bulk_ingested,
t.metadata.item_count,
t.max_local_seqno(),
))
{
drop(recovered); set_aside_path(
&folder_fs,
&table_path,
"bulk-ingest sequence offset cannot be reconstructed from the SST",
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
let mut geometry_lossy = false;
let recovered = 'restrict: {
let Ok(table) = recovered else {
break 'restrict recovered;
};
let exact_bound = match trustworthy_restriction_bound(
config,
&*folder_fs,
&table_path,
table_id,
&manifest_restriction,
) {
Ok(bound) => bound,
Err(e) => break 'restrict Err(e),
};
if let Some(bound) = &exact_bound {
break 'restrict table.reopen_restricted(bound.clone());
}
let geometry = match table.punch_geometry() {
Ok(geometry) => geometry,
Err(e) => break 'restrict Err(e),
};
match geometry.verdict {
crate::table::PunchProbe::Unpunched => break 'restrict Ok(table),
crate::table::PunchProbe::Unproven if !allow_resurrection => {
drop(table);
set_aside_path(
&folder_fs,
&table_path,
"zeroed data blocks whose allocation state the backend \
cannot attribute (a lost-sidecar punch and damage are \
indistinguishable); a resurrection repair restricts past \
the zeroed region instead",
&mut unreadable_files,
&mut discard_after_commit,
);
continue 'dirent;
}
crate::table::PunchProbe::Punched | crate::table::PunchProbe::Unproven => {
use crate::table::DerivedRestriction;
let derived = if allow_resurrection {
match table.greedy_restriction_bound(&geometry) {
Ok(Some(bound)) => Ok(DerivedRestriction::Bound(bound)),
Ok(None) => Ok(DerivedRestriction::NoLiveData),
Err(e) => Err(e),
}
} else {
Ok(Table::conservative_restriction(&geometry))
};
match derived {
Ok(DerivedRestriction::Bound(bound)) => {
geometry_lossy = !allow_resurrection;
break 'restrict table.reopen_restricted(bound);
}
Err(e) => break 'restrict Err(e),
Ok(
reason @ (DerivedRestriction::NoLiveData
| DerivedRestriction::IrregularPunch),
) => {
let reason = match reason {
DerivedRestriction::NoLiveData => {
"fully hole-punched SST with no live data"
}
_ => {
"partially punched SST with punch failures and no \
trustworthy bound; the consumed/live boundary is \
unknowable (a resurrection repair keeps the readable \
region instead)"
}
};
drop(table);
set_aside_path(
&folder_fs,
&table_path,
reason,
&mut unreadable_files,
&mut discard_after_commit,
);
continue 'dirent;
}
}
}
}
};
match recovered {
Ok(table) if salvage => {
match verify_keep_decision(
config,
&folder_fs,
&table_path,
&table,
allow_resurrection,
true,
)? {
RepairKeepDecision::Keep => {
record_best(
&mut recovered_by_id,
&mut unreadable_files,
&mut redundant_unreadable,
&mut discard_after_commit,
&mut swap_after_commit,
table_id,
table,
if geometry_lossy {
Fidelity::GeometryRestricted
} else {
Fidelity::Complete
},
&folder_fs,
&table_path,
matches_manifest,
)?;
}
RepairKeepDecision::Drop(reason) => {
drop(table);
set_aside_path(
&folder_fs,
&table_path,
reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
RepairKeepDecision::Salvage => {
let restrict_bound = table.restrict_lower_bound().cloned();
drop(table);
let output_path = repair_tmp_path(&table_path);
match try_salvage_table(
config,
&folder_fs,
allow_resurrection,
TableSalvage {
source: &table_path,
table_path: &output_path,
table_id,
reject_punched_without_bound: false,
blob_rewrite: None,
recovered_global_seqno: manifest_global_seqno,
},
) {
Ok(SalvageOutcome::Salvaged(salvaged)) => {
let table = restrict_salvaged_output(
&*folder_fs,
config,
&output_path,
salvaged,
restrict_bound.clone(),
allow_resurrection,
)?;
keep_salvaged_replacement(
&mut recovered_by_id,
&mut unreadable_files,
&mut redundant_unreadable,
&mut discard_after_commit,
&mut swap_after_commit,
table_id,
table,
&folder_fs,
&table_path,
output_path,
)?;
}
Ok(SalvageOutcome::Unusable | SalvageOutcome::PunchedBoundLost) => {
let reason = "verify found corrupt blocks; nothing salvageable";
set_aside_path(
&folder_fs,
&table_path,
reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
Err(salvage_err) if is_environmental(&salvage_err) => {
return Err(salvage_err);
}
Err(salvage_err) => {
let reason = format!(
"verify found corrupt blocks; salvage failed \
({salvage_err})"
);
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
}
}
}
Ok(table) => {
match verify_keep_decision(
config,
&folder_fs,
&table_path,
&table,
allow_resurrection,
false,
)? {
RepairKeepDecision::Keep => {
record_best(
&mut recovered_by_id,
&mut unreadable_files,
&mut redundant_unreadable,
&mut discard_after_commit,
&mut swap_after_commit,
table_id,
table,
if geometry_lossy {
Fidelity::GeometryRestricted
} else {
Fidelity::Complete
},
&folder_fs,
&table_path,
matches_manifest,
)?;
}
decision @ (RepairKeepDecision::Drop(_) | RepairKeepDecision::Salvage) => {
let reason = match decision {
RepairKeepDecision::Drop(reason) => reason,
_ => {
"verification found corrupt data blocks; run a \
salvage-enabled repair to rewrite the readable blocks"
}
};
drop(table);
set_aside_path(
&folder_fs,
&table_path,
reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
}
Err(e) if salvage => {
if is_environmental(&e) {
return Err(e);
}
let restrict_bound = trustworthy_restriction_bound(
config,
&*folder_fs,
&table_path,
table_id,
&manifest_restriction,
)?;
if restrict_bound.is_none()
&& !allow_resurrection
&& source_prefix_is_punched(&*folder_fs, &table_path)?
{
let reason = "punched SST with no recoverable restriction bound \
(missing / corrupt sidecar and failed recovery); a \
resurrection repair keeps its readable region instead";
set_aside_path(
&folder_fs,
&table_path,
reason,
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
let output_path = repair_tmp_path(&table_path);
let reject_punched = restrict_bound.is_none() && !allow_resurrection;
match try_salvage_table(
config,
&folder_fs,
allow_resurrection,
TableSalvage {
source: &table_path,
table_path: &output_path,
table_id,
reject_punched_without_bound: reject_punched,
blob_rewrite: None,
recovered_global_seqno: manifest_global_seqno,
},
) {
Ok(SalvageOutcome::Salvaged(salvaged)) => {
let table = restrict_salvaged_output(
&*folder_fs,
config,
&output_path,
salvaged,
restrict_bound,
allow_resurrection,
)?;
keep_salvaged_replacement(
&mut recovered_by_id,
&mut unreadable_files,
&mut redundant_unreadable,
&mut discard_after_commit,
&mut swap_after_commit,
table_id,
table,
&folder_fs,
&table_path,
output_path,
)?;
}
Ok(SalvageOutcome::Unusable) => {
let reason = format!("unrecoverable ({e}); nothing salvageable");
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
Ok(SalvageOutcome::PunchedBoundLost) => {
let reason = format!(
"punched SST with no recoverable restriction bound \
(missing / corrupt sidecar and failed recovery, punched \
extents found during salvage): {e}; a resurrection repair \
keeps its readable region instead"
);
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
Err(salvage_err) if is_environmental(&salvage_err) => {
return Err(salvage_err);
}
Err(salvage_err) => {
let reason =
format!("recovery failed ({e}); salvage failed ({salvage_err})");
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
}
Err(e) => {
if is_environmental(&e) {
return Err(e);
}
let reason = e.to_string();
set_aside_path(
&folder_fs,
&table_path,
&reason,
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
}
}
Ok(TableScan {
recovered_by_id,
unreadable_files,
redundant_unreadable,
excluded_files,
coverage_by_path,
discard_after_commit,
swap_after_commit,
live_sidecars,
scanned_table_ids,
})
}
#[cfg(feature = "std")]
fn rebuild_from_scan(
config: &Config,
allow_resurrection: bool,
manifest_referenced: Option<CommittedManifest>,
scan: TableScan,
) -> crate::Result<RepairReport> {
let TableScan {
recovered_by_id,
mut unreadable_files,
redundant_unreadable,
mut excluded_files,
coverage_by_path,
mut discard_after_commit,
mut swap_after_commit,
live_sidecars,
scanned_table_ids,
} = scan;
let mut recovered_tables: Vec<(Table, Fidelity, PathBuf, Arc<dyn crate::fs::Fs>)> =
recovered_by_id
.into_values()
.map(|c| (c.table, c.fidelity, c.path, c.fs))
.collect();
recovered_tables
.sort_by_key(|(t, ..)| (std::cmp::Reverse(t.l0_recency()), std::cmp::Reverse(t.id())));
let mut blob_rewrites: crate::HashMap<
crate::vlog::BlobFileId,
crate::salvage::BlobFileRewrite,
> = crate::HashMap::default();
let mut published_blob_replacements = PublishedBlobReplacements::new(config);
let blob_files_salvaged: Vec<(PathBuf, String)> = Vec::new();
let mut blob_frag = crate::blob_tree::FragmentationMap::default();
let mut stale_blob_originals: Vec<(PathBuf, String)> = Vec::new();
let mut unreferenced_blob_files: Vec<PathBuf> = Vec::new();
let (tree_type, mut blob_file_list) = if config.kv_separation_opts.is_some() {
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::RecoveringBlobFiles);
}
let mut referenced_blob_ids: crate::HashSet<crate::vlog::BlobFileId> =
crate::HashSet::default();
for (table, ..) in &recovered_tables {
match table.list_blob_file_references() {
Ok(links) => {
for link in links.into_iter().flatten() {
referenced_blob_ids.insert(link.blob_file_id);
}
}
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
log::warn!(
"repair: table {} has an unreadable blob-reference section \
({e}); the dependency filter will set it aside",
table.id(),
);
}
}
}
let recovery = recover_blob_files(
config,
&mut published_blob_replacements,
&referenced_blob_ids,
manifest_referenced.as_ref().map(|m| &m.blob_frontiers),
manifest_referenced.as_ref().map(|m| &m.blob_checksums),
)?;
unreadable_files.extend(recovery.unreadable);
excluded_files.extend(recovery.excluded);
blob_rewrites = recovery.rewrites;
blob_frag = recovery.frag;
stale_blob_originals = recovery.stale;
discard_after_commit.extend(
recovery
.discard
.into_iter()
.map(|(path, note)| (Arc::clone(&config.fs), path, note)),
);
let map: crate::HashMap<crate::vlog::BlobFileId, crate::vlog::BlobFile> =
recovery.files.into_iter().map(|bf| (bf.id(), bf)).collect();
(TreeType::Blob, BlobFileList::new(map))
} else {
let candidates = core::mem::take(&mut recovered_tables);
recovered_tables.reserve(candidates.len());
for candidate in candidates {
let is_blob_backed = match candidate.0.list_blob_file_references() {
Ok(refs) => refs.is_some_and(|r| !r.is_empty()),
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
let (table, ..) = candidate;
set_aside_table(
table,
&format!(
"blob-file reference list unreadable ({e}) on a standard \
rebuild; the table cannot prove its tree type"
),
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
};
if is_blob_backed {
log::error!(
"repair: table {} references blob files, so this store is a \
KV-separated (blob) tree; rebuilding a Standard manifest \
over it would return indirection handles as values — \
configure kv separation and retry",
candidate.0.id(),
);
return Err(crate::Error::TreeTypeMismatch {
requested: TreeType::Standard,
actual: TreeType::Blob,
});
}
recovered_tables.push(candidate);
}
(
TreeType::Standard,
BlobFileList::new(crate::HashMap::default()),
)
};
if config.kv_separation_opts.is_some() {
let punched_frontiers: crate::HashMap<crate::vlog::BlobFileId, u64> = blob_rewrites
.iter()
.filter_map(|(id, rw)| match rw {
crate::salvage::BlobFileRewrite::DropBelow(f) => Some((*id, *f)),
crate::salvage::BlobFileRewrite::Remap { .. } => None,
})
.collect();
let blob_rewrites = Arc::new(blob_rewrites);
let mut kept: Vec<(Table, Fidelity, PathBuf, Arc<dyn crate::fs::Fs>)> =
Vec::with_capacity(recovered_tables.len());
for (table, fidelity, source_path, source_fs) in recovered_tables {
let links = match table.list_blob_file_references() {
Ok(links) => links,
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
set_aside_table(
table,
&format!("blob-file reference list unreadable ({e})"),
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
};
if let Some(l) = links.as_ref().and_then(|links| {
links.iter().find(|l| {
!blob_file_list.contains_key(l.blob_file_id)
&& !blob_rewrites.contains_key(&l.blob_file_id)
})
}) {
set_aside_table(
table,
&format!("blob file {} is not recoverable", l.blob_file_id),
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
let mut needs_rewrite = false;
if let Some(links) = &links {
if links.iter().any(|l| {
matches!(
blob_rewrites.get(&l.blob_file_id),
Some(crate::salvage::BlobFileRewrite::Remap { .. })
)
}) {
needs_rewrite = true;
} else if !punched_frontiers.is_empty()
&& links
.iter()
.any(|l| punched_frontiers.contains_key(&l.blob_file_id))
{
match handle_below_blob_frontier(&table, &punched_frontiers) {
Ok(hit) => needs_rewrite = hit.is_some(),
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
set_aside_table(
table,
&format!("blob handles unreadable ({e})"),
&mut unreadable_files,
&mut discard_after_commit,
);
continue;
}
}
}
}
if !needs_rewrite {
kept.push((table, fidelity, source_path, source_fs));
continue;
}
let source_id = table.id();
let path = (*table.path).clone();
let output_path = repair_tmp_path(&path);
let restrict_bound = table.restrict_lower_bound().cloned();
let source_global_seqno = table.global_seqno();
let fs = table.fs.clone();
drop(table); match try_salvage_table(
config,
&fs,
allow_resurrection,
TableSalvage {
source: &path,
table_path: &output_path,
table_id: source_id,
reject_punched_without_bound: false,
blob_rewrite: Some(Arc::clone(&blob_rewrites)),
recovered_global_seqno: Some(source_global_seqno),
},
) {
Ok(SalvageOutcome::Salvaged(rewritten)) => {
let rewritten = restrict_salvaged_output(
&*fs,
config,
&output_path,
rewritten,
restrict_bound,
allow_resurrection,
)?;
let restricted = rewritten.restrict_lower_bound().is_some();
kept.push((rewritten, Fidelity::Salvaged, source_path, source_fs));
swap_after_commit.push((Arc::clone(&fs), output_path, path, restricted));
}
Ok(SalvageOutcome::Unusable | SalvageOutcome::PunchedBoundLost) => {
set_aside_path(
&fs,
&path,
"blob-handle rewrite produced nothing",
&mut unreadable_files,
&mut discard_after_commit,
);
}
Err(e) if is_environmental(&e) => return Err(e),
Err(e) => {
set_aside_path(
&fs,
&path,
&format!("blob-handle rewrite failed ({e})"),
&mut unreadable_files,
&mut discard_after_commit,
);
}
}
}
recovered_tables = kept;
}
let present_ids: crate::HashSet<TableId> = recovered_tables
.iter()
.filter(|(_, fidelity, ..)| fidelity.is_complete())
.map(|(t, ..)| t.id())
.collect();
let all_ranges: crate::HashMap<TableId, (UserKey, UserKey, SeqNo)> = recovered_tables
.iter()
.map(|(t, ..)| {
(
t.id(),
(
t.metadata.key_range.min().clone(),
t.metadata.key_range.max().clone(),
t.get_highest_seqno(),
),
)
})
.collect();
let lineage_by_id: crate::HashMap<TableId, Vec<TableId>> = recovered_tables
.iter()
.filter_map(|(t, ..)| t.metadata.lineage.clone().map(|l| (t.id(), l)))
.collect();
let mut lineage_partial: Vec<(TableId, PathBuf, UserKey, UserKey, Option<SeqNo>, bool)> =
Vec::new();
let mut inputs_superseded: crate::HashSet<TableId> = crate::HashSet::default();
let mut redundant_excluded_ids: crate::HashSet<TableId> = crate::HashSet::default();
{
let candidates = core::mem::take(&mut recovered_tables);
for candidate in candidates {
let lineage = candidate.0.metadata.lineage.clone();
match lineage.as_deref() {
Some(inputs)
if !candidate.0.metadata.lineage_transformed
&& !inputs.is_empty()
&& inputs
.iter()
.all(|id| *id != candidate.0.id() && present_ids.contains(id)) =>
{
let (table, _, path, fs) = candidate;
let table_id = table.id();
log::info!(
"repair: table {table_id} is a compaction output whose inputs \
{inputs:?} all survived; excluding the derived copy so its \
merge operands are not applied twice",
);
drop(table);
let reason = "derived output of an uncommitted compaction whose inputs \
all survived; excluded so its merge operands are not \
applied twice";
redundant_excluded_ids.insert(table_id);
excluded_files.push((path.clone(), reason.to_string()));
discard_after_commit.push((fs, path, reason.to_string()));
}
Some(inputs) => {
let cmp = config.comparator.as_ref();
for input in inputs {
if *input == candidate.0.id() {
continue;
}
let Some((in_min, in_max, in_hi)) = all_ranges.get(input) else {
continue;
};
let out_range = &candidate.0.metadata.key_range;
if cmp.compare(out_range.min(), in_max) == core::cmp::Ordering::Greater
|| cmp.compare(in_min, out_range.max()) == core::cmp::Ordering::Greater
{
continue;
}
let covers = candidate.1.is_complete()
&& cmp.compare(out_range.min(), in_min) != core::cmp::Ordering::Greater
&& cmp.compare(in_max, out_range.max()) != core::cmp::Ordering::Greater;
if covers {
inputs_superseded.insert(*input);
continue;
}
let lo =
if cmp.compare(out_range.min(), in_min) == core::cmp::Ordering::Less {
in_min
} else {
out_range.min()
};
let hi =
if cmp.compare(out_range.max(), in_max) == core::cmp::Ordering::Less {
out_range.max()
} else {
in_max
};
lineage_partial.push((
*input,
candidate.2.clone(),
lo.clone(),
hi.clone(),
Some(candidate.0.get_highest_seqno().min(*in_hi)),
candidate.0.metadata.lineage_transformed,
));
}
recovered_tables.push(candidate);
}
_ => recovered_tables.push(candidate),
}
}
}
{
let mut by_id: crate::HashMap<TableId, usize> = crate::HashMap::default();
for (idx, (t, fidelity, ..)) in recovered_tables.iter().enumerate() {
if fidelity.is_complete() && t.metadata.lineage.is_some() {
by_id.insert(t.id(), idx);
}
}
let same_lineage = |a: usize, b: usize| {
recovered_tables.get(a).map(|(t, ..)| &t.metadata.lineage)
== recovered_tables.get(b).map(|(t, ..)| &t.metadata.lineage)
};
let mut next_of: crate::HashMap<TableId, TableId> = crate::HashMap::default();
for &idx in by_id.values() {
let Some((t, ..)) = recovered_tables.get(idx) else {
continue;
};
if let Some(prev) = t.metadata.lineage_prev
&& by_id.get(&prev).is_some_and(|&p| same_lineage(p, idx))
{
next_of.insert(prev, t.id());
}
}
let cmp = config.comparator.as_ref();
for (&head_id, &head_idx) in &by_id {
let Some((head, ..)) = recovered_tables.get(head_idx) else {
continue;
};
if head
.metadata
.lineage_prev
.is_some_and(|prev| by_id.contains_key(&prev))
{
continue;
}
let open_lo = head.metadata.lineage_prev.is_none();
let union_min = head.metadata.key_range.min().clone();
let mut union_max = head.metadata.key_range.max().clone();
let mut open_hi = head.metadata.lineage_last;
let mut cursor = head_id;
while let Some(&next) = next_of.get(&cursor) {
if let Some(&next_idx) = by_id.get(&next)
&& let Some((t, ..)) = recovered_tables.get(next_idx)
{
union_max = t.metadata.key_range.max().clone();
open_hi = t.metadata.lineage_last;
}
cursor = next;
}
let complete_run = open_lo && open_hi;
let Some(inputs) = &head.metadata.lineage else {
continue;
};
for input in inputs {
if by_id.contains_key(input) {
continue;
}
let Some((in_min, in_max, _)) = all_ranges.get(input) else {
continue;
};
if complete_run
|| (cmp.compare(&union_min, in_min) != core::cmp::Ordering::Greater
&& cmp.compare(in_max, &union_max) != core::cmp::Ordering::Greater)
{
inputs_superseded.insert(*input);
}
}
}
}
{
let mut worklist: Vec<TableId> = inputs_superseded.iter().copied().collect();
while let Some(id) = worklist.pop() {
let Some(inputs) = lineage_by_id.get(&id) else {
continue;
};
for input in inputs {
if all_ranges.contains_key(input) && inputs_superseded.insert(*input) {
worklist.push(*input);
}
}
}
}
if !inputs_superseded.is_empty() {
let candidates = core::mem::take(&mut recovered_tables);
for candidate in candidates {
if inputs_superseded.contains(&candidate.0.id()) {
let (table, _, path, fs) = candidate;
log::info!(
"repair: table {} is fully covered by a surviving compaction \
output; excluding it so its merge operands are not applied twice",
table.id(),
);
drop(table);
let reason = "input fully covered by a surviving compaction output; \
excluded so its merge operands are not applied twice";
excluded_files.push((path.clone(), reason.to_string()));
discard_after_commit.push((fs, path, reason.to_string()));
} else {
recovered_tables.push(candidate);
}
}
lineage_partial.retain(|(input, ..)| !inputs_superseded.contains(input));
}
if let Some((input, output_path, ..)) = lineage_partial
.iter()
.find(|(.., transformed)| config.merge_operator.is_some() || *transformed)
{
log::error!(
"repair: input table {} overlaps the surviving compaction output {} \
that folded part of it, and no surviving output set covers the \
input whole; publishing both would resurrect filter-removed keys \
or double-apply merge operands, and no replay can undo either",
input,
output_path.display(),
);
return Err(crate::Error::Unrecoverable);
}
for (table, ..) in &recovered_tables {
if !table.metadata.lineage_transformed {
continue;
}
let Some(inputs) = &table.metadata.lineage else {
continue;
};
let mut visited: crate::HashSet<TableId> = crate::HashSet::default();
let mut worklist: Vec<TableId> = inputs.clone();
while let Some(id) = worklist.pop() {
if id == table.id() || !visited.insert(id) || inputs_superseded.contains(&id) {
continue;
}
if redundant_excluded_ids.contains(&id) {
if let Some(carried) = lineage_by_id.get(&id) {
worklist.extend(carried.iter().copied());
}
continue;
}
if all_ranges.contains_key(&id) {
log::error!(
"repair: table {} carries history the transformed compaction \
output {} filtered, and no surviving output set proves that \
history incorporated; publishing both would resurrect the \
filter-removed rows, and no replay can re-delete them",
id,
table.id(),
);
return Err(crate::Error::Unrecoverable);
}
}
}
if config.kv_separation_opts.is_some() {
let mut referenced: crate::HashSet<crate::vlog::BlobFileId> = crate::HashSet::default();
for (table, ..) in &recovered_tables {
for link in table.list_blob_file_references()?.into_iter().flatten() {
referenced.insert(link.blob_file_id);
}
}
let dropped: Vec<(crate::vlog::BlobFileId, PathBuf)> = blob_file_list
.iter()
.filter(|bf| !referenced.contains(&bf.id()))
.map(|bf| (bf.id(), bf.path().to_path_buf()))
.collect();
for (id, path) in dropped {
log::debug!(
"blob file {id} is referenced by no recovered table; leaving it out \
of the rebuilt manifest and removing it"
);
blob_file_list.remove(id);
blob_frag.remove(&id);
unreferenced_blob_files.push(path);
}
}
{
let final_ids: crate::HashSet<TableId> =
recovered_tables.iter().map(|(t, ..)| t.id()).collect();
for (id, path, fs) in live_sidecars {
if final_ids.contains(&id) {
continue;
}
discard_after_commit.push((
fs,
path,
"orphaned sidecar; its table did not survive the rebuild".to_string(),
));
}
}
let salvaged = recovered_tables
.iter()
.filter(|(_, fidelity, ..)| *fidelity == Fidelity::Salvaged)
.count();
let mut salvaged_unknowable: Vec<PathBuf> = Vec::new();
let salvaged_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)> = recovered_tables
.iter()
.filter(|(_, fidelity, ..)| !fidelity.is_complete())
.filter_map(|(_, _, source_path, _)| {
if let Some((lo, hi, bound)) = coverage_by_path.get(source_path) {
Some((source_path.clone(), lo.clone(), hi.clone(), *bound))
} else {
salvaged_unknowable.push(source_path.clone());
None
}
})
.collect();
struct RecencyProbe {
min: UserKey,
max: UserKey,
lo_seqno: SeqNo,
hi_seqno: SeqNo,
modern: bool,
path: PathBuf,
}
let mut ambiguous_order_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)> = Vec::new();
if recovered_tables
.iter()
.any(|(t, ..)| t.metadata.recency.is_none())
{
let cmp = config.comparator.as_ref();
let mut probes: Vec<RecencyProbe> = recovered_tables
.iter()
.map(|(t, _, path, _)| RecencyProbe {
min: t.metadata.key_range.min().clone(),
max: t.metadata.key_range.max().clone(),
lo_seqno: t.get_lowest_seqno(),
hi_seqno: t.get_highest_seqno(),
modern: t.metadata.recency.is_some(),
path: path.clone(),
})
.collect();
probes.sort_by(|a, b| cmp.compare(&a.min, &b.min));
let mut active: Vec<&RecencyProbe> = Vec::new();
for probe in &probes {
active.retain(|other| cmp.compare(&other.max, &probe.min) != core::cmp::Ordering::Less);
for other in &active {
if probe.modern && other.modern {
continue;
}
if probe.lo_seqno > other.hi_seqno || other.lo_seqno > probe.hi_seqno {
continue;
}
if config.merge_operator.is_some() {
log::error!(
"repair: tables {} and {} overlap with intersecting seqno \
ranges and no trustworthy order or lineage; under a merge \
operator publishing both would double-apply operands, and \
no replay can undo that",
probe.path.display(),
other.path.display(),
);
return Err(crate::Error::Unrecoverable);
}
let hi = if cmp.compare(&probe.max, &other.max) == core::cmp::Ordering::Less {
&probe.max
} else {
&other.max
};
ambiguous_order_coverage.push((
probe.path.clone(),
probe.min.clone(),
hi.clone(),
Some(probe.hi_seqno.min(other.hi_seqno)),
));
}
active.push(probe);
}
}
let recovered_tables: Vec<Table> = recovered_tables.into_iter().map(|(t, ..)| t).collect();
publish_repaired_manifest(
config,
RepairPublication {
recovered_tables,
tree_type,
blob_file_list,
blob_frag,
published_blob_replacements,
unreadable_files,
redundant_unreadable,
excluded_files,
lost_coverage_scoped: (
salvaged_coverage,
ambiguous_order_coverage,
lineage_partial,
salvaged_unknowable,
),
coverage_by_path,
manifest_referenced,
scanned_table_ids,
discard_after_commit,
swap_after_commit,
stale_blob_originals,
unreferenced_blob_files,
blob_files_salvaged,
salvaged,
},
)
}
#[cfg(feature = "std")]
struct RepairPublication<'a> {
recovered_tables: Vec<Table>,
tree_type: TreeType,
blob_file_list: BlobFileList,
blob_frag: crate::blob_tree::FragmentationMap,
published_blob_replacements: PublishedBlobReplacements<'a>,
unreadable_files: Vec<(PathBuf, String)>,
redundant_unreadable: crate::HashSet<PathBuf>,
excluded_files: Vec<(PathBuf, String)>,
#[expect(clippy::type_complexity, reason = "the four coverage channels")]
lost_coverage_scoped: (
Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)>,
Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)>,
Vec<(TableId, PathBuf, UserKey, UserKey, Option<SeqNo>, bool)>,
Vec<PathBuf>,
),
coverage_by_path: crate::HashMap<PathBuf, (UserKey, UserKey, Option<SeqNo>)>,
manifest_referenced: Option<CommittedManifest>,
scanned_table_ids: crate::HashSet<TableId>,
discard_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, String)>,
swap_after_commit: Vec<(Arc<dyn crate::fs::Fs>, PathBuf, PathBuf, bool)>,
stale_blob_originals: Vec<(PathBuf, String)>,
unreferenced_blob_files: Vec<PathBuf>,
blob_files_salvaged: Vec<(PathBuf, String)>,
salvaged: usize,
}
#[cfg(feature = "std")]
fn publish_repaired_manifest(
config: &Config,
publication: RepairPublication<'_>,
) -> crate::Result<RepairReport> {
let RepairPublication {
recovered_tables,
tree_type,
blob_file_list,
blob_frag,
mut published_blob_replacements,
unreadable_files,
redundant_unreadable,
excluded_files,
lost_coverage_scoped:
(salvaged_coverage, ambiguous_order_coverage, lineage_partial, salvaged_unknowable),
coverage_by_path,
manifest_referenced,
scanned_table_ids,
discard_after_commit,
swap_after_commit,
stale_blob_originals,
unreferenced_blob_files,
mut blob_files_salvaged,
salvaged,
} = publication;
if recovered_tables.iter().map(Table::id).max() == Some(crate::TableId::MAX) {
log::error!(
"repair: the table id space is exhausted (id {} is in use); a rebuilt \
manifest would make the next open's id allocator overflow",
crate::TableId::MAX,
);
return Err(crate::Error::Unrecoverable);
}
if blob_file_list.iter().map(crate::vlog::BlobFile::id).max()
== Some(crate::vlog::BlobFileId::MAX)
{
log::error!(
"repair: the blob file id space is exhausted (id {} is referenced); a \
rebuilt manifest would make the next open's blob id allocator overflow",
crate::vlog::BlobFileId::MAX,
);
return Err(crate::Error::Unrecoverable);
}
if let Some(p) = &config.recovery_progress {
p.tables_recovered_add(recovered_tables.len() as u64);
p.blob_files_recovered_add(blob_file_list.len() as u64);
}
let l0_runs = recovered_tables
.iter()
.cloned()
.filter_map(|t| Run::new(vec![t]).map(Arc::new))
.collect::<Vec<_>>();
let recovered = l0_runs.len();
let mut levels = Vec::with_capacity(config.level_count.into());
levels.push(Level::from_runs(l0_runs));
for _ in 1..config.level_count {
levels.push(Level::empty());
}
let version_id = match highest_existing_version_id(&*config.fs, &config.path)? {
Some(max) => {
let next = max.checked_add(1).ok_or(crate::Error::Unrecoverable)?;
if next == u64::MAX {
log::error!(
"repair: the version id space is exhausted (v{max} exists); a \
rebuilt manifest at v{next} would make the next version edit \
overflow",
);
return Err(crate::Error::Unrecoverable);
}
next
}
None => 0,
};
let version = Version::from_levels(version_id, tree_type, levels, blob_file_list, blob_frag);
check_cancel(config)?;
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::Committing);
}
let persisted = crate::version::persist_version(
&config.path,
&version,
config.comparator.name(),
&*config.fs,
Arc::new(config.initial_runtime_config.clone()),
config.encryption.clone(),
config.sync_mode,
);
let mut post_commit_error: Option<crate::Error> = None;
match persisted {
Ok(()) => published_blob_replacements.disarm(),
Err(e) => match probe_current(&*config.fs, &config.path, version_id) {
CurrentProbe::NotSwitched => return Err(e),
CurrentProbe::Switched => {
published_blob_replacements.disarm();
post_commit_error = Some(e);
}
CurrentProbe::Inconclusive => {
published_blob_replacements.disarm();
return Err(e);
}
},
}
if post_commit_error.is_none() {
match config.fs.read_dir(&config.path) {
Ok(dirents) => {
for dirent in dirents {
if dirent.is_dir || !dirent.file_name.starts_with("edits-") {
continue;
}
match config.fs.remove_file(&dirent.path) {
Ok(()) => {}
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => {
post_commit_error = Some(e.into());
break;
}
}
}
}
Err(e) => post_commit_error = Some(e.into()),
}
}
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::Cleanup);
}
if post_commit_error.is_none() {
for (fs, tmp_path, table_path, restricted) in swap_after_commit {
if let Err(e) =
commit_repair_tmp(&*fs, &tmp_path, &table_path, config.sync_mode, restricted)
{
log::error!(
"repair: cannot swap the replacement {} onto {} ({e}); failing the \
repair — the committed manifest names the replacement's content",
tmp_path.display(),
table_path.display(),
);
post_commit_error = Some(e);
break;
}
}
}
let salvage_report_base = blob_files_salvaged.len();
blob_files_salvaged.extend(
stale_blob_originals
.iter()
.map(|(path, note)| (path.clone(), format!("{note}; original NOT removed"))),
);
if post_commit_error.is_none() {
for (i, (path, note)) in stale_blob_originals.iter().enumerate() {
match discard_unreferenced(&*config.fs, path, config.sync_mode) {
Ok(()) => {
if let Some(entry) = blob_files_salvaged.get_mut(salvage_report_base + i) {
*entry = (path.clone(), format!("{note}; original removed"));
}
}
Err(e) => {
log::error!(
"repair: cannot remove the superseded blob original {} ({e}); \
failing the repair — left in blobs/ it is an orphan the next \
open must remove, and that removal would hit the same error",
path.display(),
);
post_commit_error = Some(e);
break;
}
}
}
}
if post_commit_error.is_none() {
for (fs, path, note) in discard_after_commit {
match discard_unreferenced(&*fs, &path, config.sync_mode) {
Ok(()) => log::info!("repair: {} removed ({note})", path.display()),
Err(e) => {
log::error!(
"repair: cannot remove {} ({e}); failing the repair — left in \
place it is a file the next open must reject or sweep, and it \
would hit the same error",
path.display(),
);
post_commit_error = Some(e);
break;
}
}
}
}
if post_commit_error.is_none() {
let mut removed_dir: Option<std::path::PathBuf> = None;
for path in unreferenced_blob_files {
match config.fs.remove_file(&path) {
Ok(()) => removed_dir = path.parent().map(std::path::Path::to_path_buf),
Err(e) if e.kind() == crate::io::ErrorKind::NotFound => {}
Err(e) => {
log::error!(
"repair: cannot remove the unreferenced blob file {} ({e}); \
failing the repair — left in blobs/ it is an orphan the next \
open must remove, and that removal would hit the same error",
path.display(),
);
post_commit_error = Some(e.into());
break;
}
}
}
if let Some(dir) = removed_dir
&& post_commit_error.is_none()
&& let Err(e) = config.fs.sync_directory_with(&dir, config.sync_mode)
{
log::error!(
"repair: cannot make the removal of unreferenced blob files durable \
in {} ({e}); failing the repair: a power loss would restore them \
as orphans the next open must sweep",
dir.display(),
);
post_commit_error = Some(e.into());
}
}
let mut warnings = vec![
"All recovered tables placed at L0; background compaction will redistribute them",
"Recent unlogged version edits (in-flight compactions, recent deletions) are lost",
];
if config.kv_separation_opts.is_some() {
warnings.push(
"Blob fragmentation stats reset (punched prefixes reseeded); blob GC re-learns the rest over time",
);
}
let table_folders: Vec<PathBuf> = config
.all_tables_folders()
.into_iter()
.map(|(folder, _)| folder)
.collect();
let mut lost_coverage: Vec<(PathBuf, UserKey, UserKey, Option<SeqNo>)> = Vec::new();
let mut unknowable_losses: Vec<PathBuf> = Vec::new();
let is_table_candidate = |path: &std::path::Path| {
path.parent()
.is_some_and(|parent| table_folders.iter().any(|f| f.as_path() == parent))
&& path.file_name().and_then(|n| n.to_str()).is_some_and(|n| {
matches!(
crate::file::TableDirEntry::classify(n),
crate::file::TableDirEntry::Table(_)
)
})
};
for (path, _) in &unreadable_files {
if redundant_unreadable.contains(path) {
continue;
}
match coverage_by_path.get(path) {
Some((lo, hi, seqno)) => {
lost_coverage.push((path.clone(), lo.clone(), hi.clone(), *seqno));
}
None if is_table_candidate(path) => unknowable_losses.push(path.clone()),
None => {}
}
}
lost_coverage.extend(salvaged_coverage);
lost_coverage.extend(ambiguous_order_coverage);
lost_coverage.extend(
lineage_partial
.into_iter()
.map(|(_, path, lo, hi, bound, _)| (path, lo, hi, bound)),
);
unknowable_losses.extend(salvaged_unknowable);
if let Some(referenced) = manifest_referenced {
let primary_tables = config.path.join("tables");
let mut missing: Vec<TableId> = referenced
.tables
.into_keys()
.filter(|id| !scanned_table_ids.contains(id))
.collect();
missing.sort_unstable();
for id in missing {
log::warn!(
"repair: table {id} is referenced by the recovered manifest but has \
no file on disk; its loss is unscopable",
);
unknowable_losses.push(primary_tables.join(id.to_string()));
}
}
let report = RepairReport {
recovered,
salvaged,
unreadable: unreadable_files.len(),
unreadable_files,
excluded_files,
lost_coverage,
unknowable_losses,
blob_files_salvaged,
method: "all-to-L0 with sequence-number ordering",
warnings,
};
if let Some(cause) = post_commit_error {
return Err(crate::Error::RepairedButUnopened {
report: Box::new(report),
cause: Box::new(cause),
});
}
if let Some(p) = &config.recovery_progress {
p.set_phase(crate::RecoveryPhase::Done);
}
Ok(report)
}
#[cfg(test)]
mod tests;