use crate::path::{Path, PathBuf};
use crate::{checksum::Checksum, coding::Decode, io, table::TableId, table::block::Header};
#[cfg(not(feature = "std"))]
use alloc::{boxed::Box, string::String, vec::Vec};
#[cfg(feature = "std")]
#[derive(Debug)]
#[non_exhaustive]
pub enum IntegrityError {
SstFileCorrupted {
table_id: TableId,
path: PathBuf,
expected: Checksum,
got: Checksum,
},
BlobFileCorrupted {
blob_file_id: u64,
path: PathBuf,
expected: Checksum,
got: Checksum,
},
IoError {
path: PathBuf,
error: io::Error,
},
}
#[cfg(feature = "std")]
impl core::fmt::Display for IntegrityError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::SstFileCorrupted {
table_id,
path,
expected,
got,
} => write!(
f,
"SST table {table_id} corrupted at {}: expected {expected}, got {got}",
path.display()
),
Self::BlobFileCorrupted {
blob_file_id,
path,
expected,
got,
} => write!(
f,
"blob file {blob_file_id} corrupted at {}: expected {expected}, got {got}",
path.display()
),
Self::IoError { path, error } => {
write!(f, "I/O error reading {}: {}", path.display(), error)
}
}
}
}
#[cfg(feature = "std")]
impl core::error::Error for IntegrityError {
fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
match self {
Self::IoError { error, .. } => Some(error),
_ => None,
}
}
}
#[cfg(feature = "std")]
#[derive(Debug)]
#[non_exhaustive]
pub struct IntegrityReport {
pub sst_files_checked: usize,
pub blob_files_checked: usize,
pub errors: Vec<IntegrityError>,
}
#[cfg(feature = "std")]
impl IntegrityReport {
#[must_use]
pub fn is_ok(&self) -> bool {
self.errors.is_empty()
}
#[must_use]
pub fn files_checked(&self) -> usize {
self.sst_files_checked + self.blob_files_checked
}
}
#[cfg(feature = "std")]
pub(crate) fn stream_checksum_from(
path: &std::path::Path,
start: u64,
) -> std::io::Result<Checksum> {
use std::io::{Read, Seek, SeekFrom};
let mut reader = std::fs::File::open(path)?;
if start != 0 {
reader.seek(SeekFrom::Start(start))?;
}
let mut hasher = xxhash_rust::xxh3::Xxh3Default::new();
let mut buf = vec![0u8; 64 * 1024];
loop {
let n = match reader.read(&mut buf) {
Ok(n) => n,
Err(e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
Err(e) => return Err(e),
};
if n == 0 {
break;
}
if let Some(chunk) = buf.get(..n) {
hasher.update(chunk);
}
}
Ok(Checksum::from_raw(hasher.digest128()))
}
#[cfg(feature = "std")]
#[must_use]
pub fn verify_integrity(tree: &impl crate::AbstractTree) -> IntegrityReport {
let version = tree.current_version();
let mut report = IntegrityReport {
sst_files_checked: 0,
blob_files_checked: 0,
errors: Vec::new(),
};
for table in version.iter_tables() {
let path = &*table.path;
let expected = table.checksum();
let start = match table.restrict_lower_bound() {
Some(bound) => match table.punch_offset_for(bound) {
Ok(offset) => offset,
Err(e) if e.is_environmental() => {
report.errors.push(IntegrityError::IoError {
path: (*table.path).clone(),
error: environmental_as_io(e),
});
report.sst_files_checked += 1;
continue;
}
Err(_) => 0,
},
None => 0,
};
match stream_checksum_from(path, start) {
Ok(got) if got != expected => {
report.errors.push(IntegrityError::SstFileCorrupted {
table_id: table.id(),
path: (*table.path).clone(),
expected,
got,
});
}
Ok(_) => {}
Err(e) => {
report.errors.push(IntegrityError::IoError {
path: (*table.path).clone(),
error: e.into(),
});
}
}
report.sst_files_checked += 1;
}
for blob_file in version.blob_files.iter() {
let path = blob_file.path();
let expected = blob_file.checksum();
match stream_checksum_from(path, blob_file.live_data_start()) {
Ok(got) if got != expected => {
report.errors.push(IntegrityError::BlobFileCorrupted {
blob_file_id: blob_file.id(),
path: path.to_path_buf(),
expected,
got,
});
}
Ok(_) => {}
Err(e) => {
report.errors.push(IntegrityError::IoError {
path: path.to_path_buf(),
error: e.into(),
});
}
}
report.blob_files_checked += 1;
}
report
}
#[derive(Debug)]
#[non_exhaustive]
pub enum BlockVerifyError {
SstFileUnreadable {
table_id: TableId,
path: PathBuf,
error: io::Error,
},
HeaderCorrupted {
table_id: TableId,
path: PathBuf,
offset: u64,
reason: String,
},
DataCorrupted {
table_id: TableId,
path: PathBuf,
offset: u64,
data_length: u32,
expected: Checksum,
got: Checksum,
},
DataReadError {
table_id: TableId,
path: PathBuf,
offset: u64,
data_length: u32,
error: io::Error,
},
EccParityMismatch {
table_id: TableId,
path: PathBuf,
offset: u64,
data_length: u32,
},
TocCorrupted {
table_id: TableId,
path: PathBuf,
section_name: Vec<u8>,
section_offset: u64,
reason: String,
},
}
impl core::fmt::Display for BlockVerifyError {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
match self {
Self::SstFileUnreadable {
table_id,
path,
error,
} => write!(
f,
"SST table {table_id} at {} could not be opened/parsed: {error}",
path.display(),
),
Self::HeaderCorrupted {
table_id,
path,
offset,
reason,
} => write!(
f,
"SST table {table_id} at {}: block header at offset {offset} is corrupt ({reason})",
path.display(),
),
Self::DataCorrupted {
table_id,
path,
offset,
data_length,
expected,
got,
} => write!(
f,
"SST table {table_id} at {}: block at offset {offset} ({data_length} bytes) data \
checksum mismatch, expected {expected}, got {got}",
path.display(),
),
Self::DataReadError {
table_id,
path,
offset,
data_length,
error,
} => write!(
f,
"SST table {table_id} at {}: failed to read {data_length}-byte data segment for \
block at offset {offset}: {error}",
path.display(),
),
Self::EccParityMismatch {
table_id,
path,
offset,
data_length,
} => write!(
f,
"SST table {table_id} at {}: block at offset {offset} ({data_length} bytes) has a \
clean payload but its ECC parity trailer does not match freshly computed parity \
(dead ECC — recompact or heal in place)",
path.display(),
),
Self::TocCorrupted {
table_id,
path,
section_name,
section_offset,
reason,
} => write!(
f,
"SST table {table_id} at {}: TOC section {:?} at offset {section_offset} is \
unreachable ({reason})",
path.display(),
String::from_utf8_lossy(section_name),
),
}
}
}
impl core::error::Error for BlockVerifyError {
fn source(&self) -> Option<&(dyn core::error::Error + 'static)> {
match self {
Self::SstFileUnreadable { error, .. } | Self::DataReadError { error, .. } => {
Some(error)
}
_ => None,
}
}
}
#[derive(Debug)]
#[non_exhaustive]
pub enum BlockVerifyWarning {
UnrecognizedEcc {
table_id: TableId,
path: PathBuf,
},
ParityUnverifiable {
table_id: TableId,
path: PathBuf,
},
EccCodecSuspect {
table_id: TableId,
path: PathBuf,
},
EccDescriptorsUnreadable {
table_id: TableId,
path: PathBuf,
},
}
#[derive(Debug, Default)]
#[non_exhaustive]
pub struct BlockVerifyReport {
pub sst_files_scanned: usize,
pub blocks_scanned: usize,
pub errors: Vec<BlockVerifyError>,
pub warnings: Vec<BlockVerifyWarning>,
pub incomplete: bool,
}
impl BlockVerifyReport {
#[must_use]
pub fn is_ok(&self) -> bool {
self.errors.is_empty() && !self.incomplete
}
#[must_use]
pub fn has_warnings(&self) -> bool {
!self.warnings.is_empty()
}
}
#[derive(Clone, Debug)]
pub struct VerifyOptions {
pub parallelism: usize,
pub throttle: Option<core::time::Duration>,
}
impl Default for VerifyOptions {
fn default() -> Self {
Self {
parallelism: 1,
throttle: None,
}
}
}
impl VerifyOptions {
#[must_use]
pub const fn parallelism(mut self, workers: usize) -> Self {
self.parallelism = workers;
self
}
#[must_use]
pub const fn throttle(mut self, delay: core::time::Duration) -> Self {
self.throttle = Some(delay);
self
}
}
fn environmental_as_io(e: crate::Error) -> io::Error {
match e {
crate::Error::Io(io) => io,
other => io::Error::new(
io::ErrorKind::Other,
alloc::string::ToString::to_string(&other),
),
}
}
fn merge_report(dst: &mut BlockVerifyReport, src: BlockVerifyReport) {
dst.sst_files_scanned += src.sst_files_scanned;
dst.blocks_scanned += src.blocks_scanned;
dst.errors.extend(src.errors);
dst.warnings.extend(src.warnings);
dst.incomplete |= src.incomplete;
}
fn scan_one_table(table: &crate::table::Table) -> BlockVerifyReport {
let mut report = BlockVerifyReport {
sst_files_scanned: 1,
..BlockVerifyReport::default()
};
let path: &Path = &table.path;
let table_id = table.id();
let ecc_unrecognized = table.metadata.ecc_unrecognized;
if ecc_unrecognized {
log::warn!(
"table {table_id} at {}: unrecognized ECC scheme — skipping the \
ECC-dependent block sections; recompact to re-stamp with a \
supported scheme",
path.display(),
);
report.warnings.push(BlockVerifyWarning::UnrecognizedEcc {
table_id,
path: path.to_path_buf(),
});
report.incomplete = true;
}
#[cfg(not(feature = "page_ecc"))]
if table.metadata.ecc_params.is_some() {
report
.warnings
.push(BlockVerifyWarning::ParityUnverifiable {
table_id,
path: path.to_path_buf(),
});
}
let max_enc_overhead = table.encryption.as_ref().map_or(0u32, |e| e.max_overhead());
let data_start = match table.restrict_lower_bound() {
Some(bound) => match table.punch_offset_for(bound) {
Ok(offset) => offset,
Err(e) if e.is_environmental() => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error: environmental_as_io(e),
});
return report;
}
Err(_) => 0,
},
None => 0,
};
match scan_sst_blocks(
&*table.fs,
path,
table_id,
max_enc_overhead,
table.metadata.ecc_params,
ecc_unrecognized,
data_start,
) {
Ok(per_file) => {
report.blocks_scanned += per_file.blocks_scanned;
report.errors.extend(per_file.errors);
}
Err(error) => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error,
});
}
}
report
}
#[must_use]
pub fn verify_block_checksums(tree: &impl crate::AbstractTree) -> BlockVerifyReport {
verify_block_checksums_with(tree, &VerifyOptions::default())
}
#[must_use]
pub fn verify_block_checksums_with(
tree: &impl crate::AbstractTree,
options: &VerifyOptions,
) -> BlockVerifyReport {
let version = tree.current_version();
let tables: Vec<crate::table::Table> = version.iter_tables().cloned().collect();
#[cfg(not(feature = "std"))]
let _ = options;
#[cfg(feature = "std")]
{
let workers = options.parallelism.max(1).min(tables.len().max(1));
if workers > 1 {
let cursor = core::sync::atomic::AtomicUsize::new(0);
let partials = std::thread::scope(|scope| {
let handles: Vec<_> = (0..workers)
.map(|_| {
scope.spawn(|| {
let mut local = BlockVerifyReport::default();
let mut idx =
cursor.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
while let Some(table) = tables.get(idx) {
merge_report(&mut local, scan_one_table(table));
idx = cursor.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
if tables.get(idx).is_some()
&& let Some(delay) = options.throttle
{
std::thread::sleep(delay);
}
}
local
})
})
.collect();
handles
.into_iter()
.map(|handle| match handle.join() {
Ok(local) => local,
Err(payload) => std::panic::resume_unwind(payload),
})
.collect::<Vec<_>>()
});
let mut report = BlockVerifyReport::default();
for partial in partials {
merge_report(&mut report, partial);
}
return report;
}
}
let mut report = BlockVerifyReport::default();
for (idx, table) in tables.iter().enumerate() {
merge_report(&mut report, scan_one_table(table));
#[cfg(feature = "std")]
if idx + 1 < tables.len()
&& let Some(delay) = options.throttle
{
std::thread::sleep(delay);
}
#[cfg(not(feature = "std"))]
let _ = idx;
}
report
}
pub fn verify_kv_checksums(tree: &impl crate::AbstractTree) -> crate::Result<()> {
let version = tree.current_version();
for table in version.iter_tables() {
table.verify_kv_checksums()?;
}
Ok(())
}
#[cfg(feature = "std")]
#[must_use]
pub fn verify_sst_file(path: &std::path::Path) -> BlockVerifyReport {
let fs: alloc::sync::Arc<dyn crate::fs::Fs> = alloc::sync::Arc::new(crate::fs::StdFs);
verify_sst_file_with_fs(&fs, path)
}
#[cfg(feature = "std")]
pub(crate) fn verify_sst_file_with_fs(
fs: &alloc::sync::Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
) -> BlockVerifyReport {
verify_sst_file_with_context(fs, path, None, None, 0)
}
#[cfg(feature = "std")]
pub(crate) fn verify_sst_file_with_context(
fs: &alloc::sync::Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
encryption: Option<&alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
known_table_id: Option<crate::TableId>,
data_start: u64,
) -> BlockVerifyReport {
let table_id = known_table_id.unwrap_or(0);
let derived = if data_start == 0 {
restricted_data_start(fs, path, encryption, known_table_id)
} else {
Ok(data_start)
};
let mut report = BlockVerifyReport {
sst_files_scanned: 1,
..BlockVerifyReport::default()
};
let data_start = match derived {
Ok(offset) => offset,
Err(e) => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error: environmental_as_io(e),
});
return report;
}
};
let mut ecc_unrecognized = false;
let provider = encryption.map(|e| &**e);
let probe = match read_ecc_params_out_of_band(&**fs, path, provider, known_table_id, data_start)
{
Ok(p) => p,
Err(error) => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error: error.into(),
});
return report;
}
};
if probe.mirrors_diverge {
report.errors.push(BlockVerifyError::TocCorrupted {
table_id,
path: path.to_path_buf(),
section_name: b"meta".to_vec(),
section_offset: 0,
reason: alloc::string::String::from(
"the tail meta and meta_mid mirrors decode to different metadata; \
one copy is forged or rotted behind a re-stamped checksum",
),
});
}
let ecc = match probe.ecc {
Some(ScrubEcc::Off) => None,
Some(ScrubEcc::Scheme(params)) => Some(params),
Some(ScrubEcc::Unrecognized) => {
log::warn!(
"{}: unrecognized ECC scheme — skipping the ECC-dependent block \
sections; recompact to re-stamp with a supported scheme",
path.display(),
);
report.warnings.push(BlockVerifyWarning::UnrecognizedEcc {
table_id,
path: path.to_path_buf(),
});
report.incomplete = true;
ecc_unrecognized = true;
None
}
None => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error: io::Error::new(
io::ErrorKind::InvalidData,
"could not decode the SST meta block to determine the ECC scheme \
(corrupt meta, or an encrypted SST with no key out-of-band); \
skipping the block walk — use verify_block_checksums on a live \
tree for ECC-aware verification",
),
});
return report;
}
};
#[cfg(not(feature = "page_ecc"))]
if ecc.is_some() {
report
.warnings
.push(BlockVerifyWarning::ParityUnverifiable {
table_id,
path: path.to_path_buf(),
});
}
let max_enc_overhead =
provider.map_or(0u32, crate::encryption::EncryptionProvider::max_overhead);
match scan_sst_blocks(
&**fs,
path,
table_id,
max_enc_overhead,
ecc,
ecc_unrecognized,
data_start,
) {
Ok(per_file) => {
report.blocks_scanned = per_file.blocks_scanned;
report.errors.extend(per_file.errors);
}
Err(error) => {
report.errors.push(BlockVerifyError::SstFileUnreadable {
table_id,
path: path.to_path_buf(),
error,
});
}
}
if let Some(scheme @ ScrubEcc::Scheme(_)) = probe.ecc
&& report
.errors
.iter()
.any(|e| matches!(e, BlockVerifyError::EccParityMismatch { .. }))
&& codec_suspect_for(
&**fs,
path,
scheme,
data_start,
block_data_length_cap(max_enc_overhead),
)
{
report.warnings.push(BlockVerifyWarning::EccCodecSuspect {
table_id,
path: path.to_path_buf(),
});
}
if probe.descriptors_unreadable {
report
.warnings
.push(BlockVerifyWarning::EccDescriptorsUnreadable {
table_id,
path: path.to_path_buf(),
});
}
report
}
#[derive(Clone, Copy, PartialEq, Eq)]
#[cfg(feature = "std")]
enum ScrubEcc {
Off,
Scheme(crate::table::block::EccParams),
Unrecognized,
}
#[cfg(feature = "std")]
fn descriptor_sized_regions(toc: &crate::sfa::Toc, data_start: u64) -> Vec<(u64, u64)> {
let mut regions: Vec<(u64, u64)> = Vec::new();
for entry in toc.iter() {
let Some(roles) = expected_section_roles(entry.name()) else {
continue;
};
if roles
.iter()
.any(|role| crate::table::block::Header::has_block_flags(*role))
{
continue;
}
let Some(end) = entry.pos().checked_add(entry.len()) else {
continue;
};
let floor = if entry.name() == b"data" {
data_start
} else {
0
};
regions.push((core::cmp::max(entry.pos(), floor), end));
}
regions
}
#[cfg(feature = "std")]
fn arbitrate_by_framing(
file: &dyn crate::fs::FsFile,
toc: &crate::sfa::Toc,
scheme: ScrubEcc,
data_start: u64,
) -> crate::io::Result<Option<bool>> {
let regions = descriptor_sized_regions(toc, data_start);
let (mut judged, mut framed_any, mut framed_all) = (false, false, true);
for &(start, end) in ®ions {
if let Some(verdict) = scheme_frames_region(file, scheme, start, end)? {
judged = true;
framed_any |= verdict;
framed_all &= verdict;
}
}
if !judged {
return Ok(None);
}
if !framed_any || !framed_all {
return Ok(Some(false));
}
Ok(Some(true))
}
#[cfg(feature = "std")]
fn codec_disagrees_everywhere(
file: &dyn crate::fs::FsFile,
toc: &crate::sfa::Toc,
scheme: ScrubEcc,
data_start: u64,
payload_cap: u64,
) -> bool {
let regions = descriptor_sized_regions(toc, data_start);
let mut judged = false;
for &(start, end) in ®ions {
match codec_confirms_region(file, scheme, start, end, payload_cap) {
CodecVerdict::Confirmed | CodecVerdict::Incomplete => return false,
CodecVerdict::Rejected => judged = true,
CodecVerdict::NoEvidence => {}
}
}
judged
}
#[cfg(feature = "std")]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum CodecVerdict {
#[cfg_attr(
not(feature = "page_ecc"),
expect(
dead_code,
reason = "recomputing parity is what produces a match, and it needs the ECC codecs"
)
)]
Confirmed,
Rejected,
#[cfg_attr(
not(feature = "page_ecc"),
expect(
dead_code,
reason = "the traversal that can stop short is the parity recomputation, which needs the ECC codecs"
)
)]
Incomplete,
NoEvidence,
}
#[cfg(feature = "std")]
fn codec_confirms_region(
file: &dyn crate::fs::FsFile,
scheme: ScrubEcc,
start: u64,
end: u64,
payload_cap: u64,
) -> CodecVerdict {
let params = match scheme {
ScrubEcc::Off => return CodecVerdict::NoEvidence,
ScrubEcc::Scheme(params) => params,
ScrubEcc::Unrecognized => return CodecVerdict::Rejected,
};
#[cfg(not(feature = "page_ecc"))]
{
let _ = (file, params, start, end, payload_cap);
CodecVerdict::NoEvidence
}
#[cfg(feature = "page_ecc")]
{
use crate::table::block::Header;
let mut offset = start;
let mut matched = false;
let mut mismatched = false;
let mut complete = offset >= end;
while offset < end {
let remaining = end - offset;
let want =
usize::try_from(remaining).map_or(Header::MAX_LEN, |r| r.min(Header::MAX_LEN));
let Ok(buf) = crate::file::read_exact(file, offset, want) else {
break;
};
let Ok(header) = Header::decode_from(&mut &buf[..]) else {
break;
};
if u64::from(header.data_length) > payload_cap {
break;
}
let header_len = Header::header_len(header.block_type) as u64;
let parity_bytes = crate::table::block::expected_parity_len(header.data_length, params);
let parity_len = u64::from(parity_bytes);
if parity_len > MAX_BLOCK_DATA_LENGTH {
break;
}
let Some(payload_at) = offset.checked_add(header_len) else {
break;
};
let Some(trailer_at) = payload_at.checked_add(u64::from(header.data_length)) else {
break;
};
let Some(next) = trailer_at.checked_add(parity_len) else {
break;
};
if next > end {
break;
}
if let (Ok(payload_size), Ok(trailer_size)) = (
usize::try_from(header.data_length),
usize::try_from(parity_bytes),
) && payload_size > 0
&& trailer_size > 0
{
let payload = crate::file::read_exact(file, payload_at, payload_size);
let trailer = crate::file::read_exact(file, trailer_at, trailer_size);
if let (Ok(payload), Ok(trailer)) = (payload, trailer)
&& Checksum::from_raw(crate::hash::hash128(&payload)) == header.checksum
{
let fresh = match params {
crate::table::block::EccParams::Secded => {
Some(crate::secded::encode_block_parity(&payload))
}
crate::table::block::EccParams::Shard { .. } => {
let (ds, ps) = params.as_shards();
crate::ecc::encode_parity(&payload, ds, ps).ok()
}
};
if fresh.as_deref() == Some(&trailer[..]) {
matched = true;
} else {
mismatched = true;
}
}
}
offset = next;
complete = offset >= end;
}
if matched {
CodecVerdict::Confirmed
} else if !complete {
CodecVerdict::Incomplete
} else if mismatched {
CodecVerdict::Rejected
} else {
CodecVerdict::NoEvidence
}
}
}
#[cfg(feature = "std")]
fn scheme_frames_region(
file: &dyn crate::fs::FsFile,
scheme: ScrubEcc,
start: u64,
end: u64,
) -> crate::io::Result<Option<bool>> {
use crate::table::block::Header;
let params = match scheme {
ScrubEcc::Off => None,
ScrubEcc::Scheme(params) => Some(params),
ScrubEcc::Unrecognized => return Ok(None),
};
if end <= start {
return Ok(None);
}
let mut offset = start;
let mut framed = 0usize;
while offset < end {
let remaining = end - offset;
if remaining < Header::MIN_LEN as u64 {
return Ok(Some(false));
}
let want = usize::try_from(remaining).map_or(Header::MAX_LEN, |r| r.min(Header::MAX_LEN));
let buf = crate::file::read_exact(file, offset, want)?;
let Ok(header) = Header::decode_from(&mut &buf[..]) else {
return Ok(Some(false));
};
let parity_len = params.map_or(0, |p| {
u64::from(crate::table::block::expected_parity_len(
header.data_length,
p,
))
});
let Some(frame) = (Header::header_len(header.block_type) as u64)
.checked_add(u64::from(header.data_length))
.and_then(|n| n.checked_add(parity_len))
else {
return Ok(Some(false));
};
let Some(next) = offset.checked_add(frame) else {
return Ok(Some(false));
};
if next > end {
return Ok(Some(false));
}
offset = next;
framed += 1;
}
Ok(if framed == 0 { None } else { Some(true) })
}
#[cfg(feature = "std")]
fn read_ecc_params_out_of_band(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
encryption: Option<&dyn crate::encryption::EncryptionProvider>,
known_table_id: Option<crate::TableId>,
data_start: u64,
) -> std::io::Result<EccProbe> {
let mut probe = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)
.map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?;
let toc = sfa_reader.toc();
let mut unrecognized_seen = false;
let mut recognized: Vec<ScrubEcc> = Vec::new();
let mut decoded: Vec<crate::table::meta::ParsedMeta> = Vec::new();
for name in [b"meta".as_slice(), b"meta_mid".as_slice()] {
let Some((pos, len)) = toc.section(name).map(|e| (e.pos(), e.len())) else {
continue;
};
let Ok(size) = u32::try_from(len) else {
continue;
};
let handle = crate::table::BlockHandle::new(crate::table::BlockOffset(pos), size);
let expected_id = if encryption.is_some() {
Some(known_table_id.unwrap_or(0))
} else {
known_table_id
};
match crate::table::meta::ParsedMeta::load_with_handle(
probe.as_ref(),
&handle,
expected_id,
encryption,
) {
Ok(meta) => {
if meta.ecc_unrecognized {
unrecognized_seen = true;
} else {
recognized.push(if let Some(params) = meta.ecc_params {
ScrubEcc::Scheme(params)
} else {
ScrubEcc::Off
});
}
decoded.push(meta);
}
Err(e) if e.is_environmental() => {
return Err(match e {
crate::Error::Io(io) => io.into(),
other => std::io::Error::other(other),
});
}
Err(_) => {}
}
}
let mirrors_diverge = match decoded.as_slice() {
[a, b] if unrecognized_seen => a.clone().without_ecc() != b.clone().without_ecc(),
[a, b] => a != b,
_ => false,
};
let mut descriptors_unreadable = false;
let ecc = match recognized.as_slice() {
[a, b] if a == b => Some(*a),
[_, _] => Some(ScrubEcc::Unrecognized),
[one] if unrecognized_seen => Some(
match arbitrate_by_framing(probe.as_ref(), toc, *one, data_start)? {
Some(false) => ScrubEcc::Unrecognized,
Some(true) | None => *one,
},
),
[one] => Some(*one),
[] if unrecognized_seen => Some(
match arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Off, data_start)? {
Some(true) => {
descriptors_unreadable = true;
ScrubEcc::Off
}
Some(false) | None => ScrubEcc::Unrecognized,
},
),
[..] => None,
};
Ok(EccProbe {
ecc,
mirrors_diverge,
descriptors_unreadable,
})
}
#[cfg(feature = "std")]
fn codec_suspect_for(
fs: &dyn crate::fs::Fs,
path: &std::path::Path,
scheme: ScrubEcc,
data_start: u64,
payload_cap: u64,
) -> bool {
let Ok(mut probe) = fs.open(path, &crate::fs::FsOpenOptions::new().read(true)) else {
return false;
};
let Ok(sfa_reader) = crate::sfa::Reader::from_reader(&mut probe) else {
return false;
};
codec_disagrees_everywhere(
probe.as_ref(),
sfa_reader.toc(),
scheme,
data_start,
payload_cap,
)
}
#[cfg(feature = "std")]
fn restricted_data_start(
fs: &alloc::sync::Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
encryption: Option<&alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
known_table_id: Option<crate::TableId>,
) -> crate::Result<u64> {
let expected_id = known_table_id.or_else(|| {
path.file_name()
.and_then(|n| n.to_str())
.and_then(|n| n.parse::<crate::TableId>().ok())
});
let bound = match crate::restrict_bound::read(&**fs, path, encryption.map(|e| &**e)) {
Ok(crate::restrict_bound::SidecarRead::Present(sidecar_id, bound))
if expected_id == Some(sidecar_id) =>
{
bound
}
Err(e) if e.is_environmental() => return Err(e),
_ => return Ok(0),
};
let index_frontier = index_derived_frontier(fs, path, encryption, expected_id, &bound);
let Ok(mut file) = fs.open(path, &crate::fs::FsOpenOptions::new().read(true)) else {
return Ok(0);
};
let Ok(meta) = crate::fs::FsFile::metadata(&*file) else {
return Ok(0);
};
let file_len = meta.len;
let Ok(reader) = crate::sfa::Reader::from_reader(&mut file) else {
return Ok(0);
};
let Some((data_pos, data_len)) = reader
.toc()
.iter()
.find(|e| e.name() == b"data")
.map(|e| (e.pos(), e.len()))
else {
return Ok(0);
};
let data_end = data_pos.saturating_add(data_len).min(file_len);
let mut offset = data_pos;
let mut frontier = data_pos;
while offset < data_end {
if let Some(header) = block_header_at(&*file, offset) {
let step = u64::from(header.on_disk_size());
if step == 0 {
return Ok(0); }
offset = offset.saturating_add(step);
continue;
}
let Some(next) = next_block_header(&*file, offset, data_end) else {
if extent_is_zeroed(&*file, offset, data_end) {
frontier = data_end;
}
break;
};
if extent_is_zeroed(&*file, offset, next) {
frontier = next;
break;
}
offset = next;
}
if frontier == data_pos {
return Ok(0);
}
Ok(index_frontier.map_or(0, |from_index| from_index.min(frontier)))
}
#[cfg(feature = "std")]
fn index_derived_frontier(
fs: &alloc::sync::Arc<dyn crate::fs::Fs>,
path: &std::path::Path,
encryption: Option<&alloc::sync::Arc<dyn crate::encryption::EncryptionProvider>>,
table_id: Option<crate::TableId>,
bound: &[u8],
) -> Option<u64> {
let checksum =
crate::Checksum::from_raw(crate::repair::compute_table_checksum_from(&**fs, path, 0).ok()?);
let mut params = crate::table::RecoverParams::new(
path.to_path_buf(),
checksum,
table_id.unwrap_or(0),
alloc::sync::Arc::clone(fs),
crate::comparator::default_comparator(),
alloc::sync::Arc::new(crate::cache::Cache::with_capacity_bytes(1_000_000)),
);
params.encryption = encryption.map(alloc::sync::Arc::clone);
let table = crate::table::Table::recover(params).ok()?;
table.punch_offset_for(bound).ok()
}
#[cfg(feature = "std")]
fn block_header_at(
file: &dyn crate::fs::FsFile,
offset: u64,
) -> Option<crate::table::block::Header> {
use crate::coding::Decode;
let bytes = crate::file::read_exact(file, offset, crate::table::block::Header::MAX_LEN).ok()?;
crate::table::block::Header::decode_from(&mut &bytes[..]).ok()
}
#[cfg(feature = "std")]
fn next_block_header(file: &dyn crate::fs::FsFile, from: u64, end: u64) -> Option<u64> {
const CHUNK: usize = 64 * 1024;
let lead = *crate::file::MAGIC_BYTES.first()?;
let mut at = from;
while at < end {
let want = usize::try_from(end - at).unwrap_or(CHUNK).min(CHUNK);
let chunk = crate::file::read_exact(file, at, want).ok()?;
for (i, _) in chunk.iter().enumerate().filter(|&(_, &b)| b == lead) {
let offset = at.saturating_add(i as u64);
if block_header_at(file, offset).is_some() {
return Some(offset);
}
}
at = at.saturating_add(want as u64);
}
None
}
#[cfg(feature = "std")]
fn extent_is_zeroed(file: &dyn crate::fs::FsFile, start: u64, end: u64) -> bool {
const CHUNK: usize = 64 * 1024;
let mut at = start;
while at < end {
let want = usize::try_from(end - at).unwrap_or(CHUNK).min(CHUNK);
let Ok(bytes) = crate::file::read_exact(file, at, want) else {
return false;
};
if bytes.iter().any(|&b| b != 0) {
return false;
}
at += want as u64;
}
end > start
}
#[cfg(feature = "std")]
struct EccProbe {
ecc: Option<ScrubEcc>,
mirrors_diverge: bool,
descriptors_unreadable: bool,
}
struct PerFileScan {
blocks_scanned: usize,
errors: Vec<BlockVerifyError>,
}
fn scan_sst_blocks(
fs: &dyn crate::fs::Fs,
path: &Path,
table_id: TableId,
max_enc_overhead: u32,
ecc: Option<crate::table::block::EccParams>,
ecc_unrecognized: bool,
data_start: u64,
) -> io::Result<PerFileScan> {
use io::BufReader;
#[cfg(not(feature = "std"))]
use io::{Seek, SeekFrom};
#[cfg(feature = "std")]
use std::io::{Seek, SeekFrom};
let mut file = fs.open(path, &crate::fs::FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut file)
.map_err(|e| io::Error::new(io::ErrorKind::InvalidData, alloc::format!("{e:?}")))?;
let toc = sfa_reader.toc();
let mut reader = BufReader::with_capacity(64 * 1024, file);
let mut blocks_scanned: usize = 0;
let mut errors: Vec<BlockVerifyError> = Vec::new();
for defect in toc_catalogue_defects(toc, sfa_reader.toc_pos()) {
errors.push(BlockVerifyError::TocCorrupted {
table_id,
path: path.to_path_buf(),
section_name: defect.name,
section_offset: defect.offset,
reason: defect.reason,
});
}
let mut data_buf: Vec<u8> = Vec::new();
let mut parity_buf: Vec<u8> = Vec::new();
for entry in toc.iter() {
if RAW_FORMAT_SECTIONS.contains(&entry.name()) {
match raw_section_shape_error(&mut reader, entry.name(), entry.pos(), entry.len()) {
Ok(Some(reason)) => {
errors.push(BlockVerifyError::TocCorrupted {
table_id,
path: path.to_path_buf(),
section_name: entry.name().to_vec(),
section_offset: entry.pos(),
reason,
});
}
Ok(None) => {}
Err(e) => {
errors.push(BlockVerifyError::DataReadError {
table_id,
path: path.to_path_buf(),
offset: entry.pos(),
data_length: 0,
error: e,
});
}
}
continue;
}
let start = if entry.name() == b"data" {
entry.pos().max(data_start)
} else {
entry.pos()
};
let Some(end) = entry.pos().checked_add(entry.len()) else {
let declared = entry.pos();
errors.push(BlockVerifyError::TocCorrupted {
table_id,
path: path.to_path_buf(),
section_name: entry.name().to_vec(),
section_offset: declared,
reason: format!(
"section length {} overflows u64 when added to start offset {declared}",
entry.len(),
),
});
continue;
};
if let Err(e) = reader.seek(SeekFrom::Start(start)) {
errors.push(BlockVerifyError::DataReadError {
table_id,
path: path.to_path_buf(),
offset: start,
data_length: 0,
error: e.into(),
});
continue;
}
let Some(expected_roles) = expected_section_roles(entry.name()) else {
continue;
};
let mut ctx = WalkCtx {
reader: &mut reader,
table_id,
path,
data_buf: &mut data_buf,
parity_buf: &mut parity_buf,
blocks_scanned: &mut blocks_scanned,
errors: &mut errors,
max_data_length: block_data_length_cap(max_enc_overhead),
ecc,
ecc_unrecognized,
expected_roles,
};
walk_block_region(&mut ctx, start, end);
}
Ok(PerFileScan {
blocks_scanned,
errors,
})
}
const RAW_FORMAT_SECTIONS: &[&[u8]] = &[b"linked_blob_files", b"table_version", b"meta_separator"];
fn expected_section_roles(name: &[u8]) -> Option<&'static [crate::table::block::BlockType]> {
use crate::table::block::BlockType;
Some(match name {
b"data" => &[BlockType::Data, BlockType::Columnar],
b"index" | b"tli" | b"tli_tail" | b"filter_tli" => &[BlockType::Index],
b"filter" => &[BlockType::Filter],
b"range_tombstones" => &[BlockType::RangeTombstone],
b"meta" | b"meta_mid" => &[BlockType::Meta],
b"block_layout" => &[BlockType::BlockLayout],
b"seqno_bounds" => &[BlockType::SeqnoBounds],
b"zone_map" => &[BlockType::ZoneMap],
b"delete_bitmap" => &[BlockType::DeleteBitmap],
b"locator" => &[BlockType::Locator],
_ => return None,
})
}
struct TocCatalogueDefect {
name: Vec<u8>,
offset: u64,
reason: String,
}
fn toc_catalogue_defects(toc: &crate::sfa::Toc, toc_pos: u64) -> Vec<TocCatalogueDefect> {
let mut defects = Vec::new();
let mut expected_pos: u64 = 0;
let mut seen: Vec<&[u8]> = Vec::new();
for entry in toc.iter() {
let name = entry.name();
if seen.contains(&name) {
defects.push(TocCatalogueDefect {
name: name.to_vec(),
offset: entry.pos(),
reason: format!(
"duplicate TOC section name {:?}; a renamed section can shadow \
another and hide it from the readers that look it up by name",
alloc::string::String::from_utf8_lossy(name),
),
});
} else {
seen.push(name);
}
if entry.pos() != expected_pos {
defects.push(TocCatalogueDefect {
name: name.to_vec(),
offset: entry.pos(),
reason: format!(
"section starts at {} but the previous section ended at \
{expected_pos}; the gap hides an omitted TOC entry",
entry.pos(),
),
});
}
if expected_section_roles(name).is_none() && !RAW_FORMAT_SECTIONS.contains(&name) {
defects.push(TocCatalogueDefect {
name: name.to_vec(),
offset: entry.pos(),
reason: String::from(
"unrecognized block-format section name; a renamed TOC entry \
hides a known section from every reader",
),
});
}
let Some(end) = entry.pos().checked_add(entry.len()) else {
expected_pos = u64::MAX;
break;
};
expected_pos = end;
}
if expected_pos != toc_pos {
defects.push(TocCatalogueDefect {
name: b"<tiling>".to_vec(),
offset: expected_pos,
reason: format!(
"sections end at {expected_pos} but the TOC begins at {toc_pos}; a \
trailing TOC entry was omitted or truncated",
),
});
}
defects
}
pub(crate) fn toc_may_hide_deletion_section(toc: &crate::sfa::Toc, toc_pos: u64) -> bool {
!toc_catalogue_defects(toc, toc_pos).is_empty()
}
fn raw_section_shape_error(
reader: &mut io::BufReader<Box<dyn crate::fs::FsFile>>,
name: &[u8],
pos: u64,
len: u64,
) -> Result<Option<String>, io::Error> {
use alloc::string::ToString as _;
#[cfg(not(feature = "std"))]
use io::{Read as _, Seek as _, SeekFrom};
#[cfg(feature = "std")]
use std::io::{Read as _, Seek as _, SeekFrom};
match name {
b"linked_blob_files" => {
if len < 4 {
return Ok(Some(format!(
"linked_blob_files section is {len} bytes, too short for its count prefix"
)));
}
reader.seek(SeekFrom::Start(pos))?;
let mut count_le = [0u8; 4];
reader.read_exact(&mut count_le)?;
let count = u64::from(u32::from_le_bytes(count_le));
let expected = count
.checked_mul(32)
.and_then(|records| records.checked_add(4));
if expected != Some(len) {
return Ok(Some(format!(
"blob-link count {count} disagrees with the section length {len} \
(expected {} bytes)",
expected.map_or_else(|| "overflowing".to_string(), |e| e.to_string()),
)));
}
Ok(None)
}
b"table_version" => {
Ok((len != 1).then(|| format!("table_version section is {len} bytes, expected 1")))
}
_ => Ok(None),
}
}
const MAX_BLOCK_DATA_LENGTH: u64 = 256 * 1024 * 1024;
fn block_data_length_cap(max_enc_overhead: u32) -> u64 {
MAX_BLOCK_DATA_LENGTH + u64::from(max_enc_overhead)
}
struct WalkCtx<'a> {
reader: &'a mut io::BufReader<Box<dyn crate::fs::FsFile>>,
table_id: TableId,
path: &'a Path,
data_buf: &'a mut Vec<u8>,
parity_buf: &'a mut Vec<u8>,
blocks_scanned: &'a mut usize,
errors: &'a mut Vec<BlockVerifyError>,
max_data_length: u64,
ecc: Option<crate::table::block::EccParams>,
ecc_unrecognized: bool,
expected_roles: &'static [crate::table::block::BlockType],
}
fn walk_block_region(ctx: &mut WalkCtx<'_>, start_offset: u64, end_offset: u64) {
#[cfg(not(feature = "std"))]
use io::Read;
#[cfg(feature = "std")]
use std::io::Read;
let mut offset = start_offset;
while offset < end_offset {
let remaining_in_section = end_offset - offset;
if remaining_in_section < Header::MIN_LEN as u64 {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"section has only {remaining_in_section} bytes left at this offset, \
less than Header::MIN_LEN = {}",
Header::MIN_LEN,
),
});
return;
}
let header = match Header::decode_from(ctx.reader) {
Ok(h) => h,
Err(crate::Error::Io(e)) => {
ctx.errors.push(BlockVerifyError::DataReadError {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
data_length: 0,
error: e,
});
return;
}
Err(e) => {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!("{e:?}"),
});
return;
}
};
if ctx.ecc_unrecognized && !Header::has_block_flags(header.block_type) {
return;
}
if !ctx.expected_roles.contains(&header.block_type) {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"block role {:?} does not belong to this section (expected one of {:?})",
header.block_type, ctx.expected_roles,
),
});
}
*ctx.blocks_scanned += 1;
let header_len = Header::header_len(header.block_type) as u64;
let block_ecc = if Header::has_block_flags(header.block_type) {
(header.block_flags & crate::table::block::header::block_flags::ECC_PARITY != 0)
.then_some(crate::table::block::EccParams::RS_4_2)
} else {
ctx.ecc
};
let parity_len = block_ecc.map_or(0, |scheme| {
u64::from(crate::table::block::expected_parity_len(
header.data_length,
scheme,
))
});
if parity_len > MAX_BLOCK_DATA_LENGTH {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"parity trailer length {parity_len} exceeds hard cap {MAX_BLOCK_DATA_LENGTH}",
),
});
return;
}
let data_length_u64 = u64::from(header.data_length);
if data_length_u64 > ctx.max_data_length {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"header data_length {data_length_u64} exceeds hard cap {}",
ctx.max_data_length,
),
});
return;
}
if header_len > remaining_in_section {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"block header ({header_len} bytes) extends past the section end \
({remaining_in_section} bytes remain)",
),
});
return;
}
let remaining = remaining_in_section - header_len;
let on_disk_payload = data_length_u64 + parity_len;
if on_disk_payload > remaining {
ctx.errors.push(BlockVerifyError::HeaderCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
reason: format!(
"header data_length {data_length_u64} + parity {parity_len} exceeds \
remaining section bytes {remaining}",
),
});
return;
}
let data_length = header.data_length as usize;
ctx.data_buf.resize(data_length, 0);
if let Err(e) = ctx.reader.read_exact(ctx.data_buf.as_mut_slice()) {
ctx.errors.push(BlockVerifyError::DataReadError {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
data_length: header.data_length,
error: e.into(),
});
return;
}
let computed = Checksum::from_raw(crate::hash::hash128(ctx.data_buf));
let payload_clean = computed == header.checksum;
if !payload_clean {
ctx.errors.push(BlockVerifyError::DataCorrupted {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
data_length: header.data_length,
expected: header.checksum,
got: computed,
});
}
if parity_len > 0 {
let parity_usize = usize::try_from(parity_len).unwrap_or(usize::MAX);
ctx.parity_buf.resize(parity_usize, 0);
if let Err(e) = ctx.reader.read_exact(ctx.parity_buf.as_mut_slice()) {
ctx.errors.push(BlockVerifyError::DataReadError {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
data_length: header.data_length,
error: e.into(),
});
return;
}
#[cfg(feature = "page_ecc")]
if payload_clean && let Some(scheme) = block_ecc {
let fresh = match scheme {
crate::table::block::EccParams::Secded => {
Some(crate::secded::encode_block_parity(ctx.data_buf))
}
crate::table::block::EccParams::Shard { .. } => {
let (ds, ps) = scheme.as_shards();
crate::ecc::encode_parity(ctx.data_buf, ds, ps).ok()
}
};
if fresh.as_deref() != Some(ctx.parity_buf.as_slice()) {
ctx.errors.push(BlockVerifyError::EccParityMismatch {
table_id: ctx.table_id,
path: ctx.path.to_path_buf(),
offset,
data_length: header.data_length,
});
}
}
}
offset += header_len + data_length_u64 + parity_len;
}
}
#[cfg(test)]
#[expect(clippy::unwrap_used, clippy::expect_used, reason = "test assertions")]
mod block_verify_tests;