use super::*;
use crate::{
AbstractTree, Config, SequenceNumberCounter, compression::CompressionType,
config::CompressionPolicy,
};
use std::io::{Read, Seek, SeekFrom, Write};
use test_log::test;
fn populate_tree(dir: &std::path::Path, items: usize) {
let cfg = Config::new(
dir,
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let tree = cfg.open().unwrap();
for i in 0u64..items as u64 {
let key = format!("k{i:08}");
let val = format!("v{i:08}");
tree.insert(key.as_bytes(), val.as_bytes(), 1 + i);
}
tree.flush_active_memtable(1 + items as u64).unwrap();
drop(tree);
}
fn reopen_tree(dir: &std::path::Path) -> crate::AnyTree {
Config::new(
dir,
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.open()
.unwrap()
}
fn populate_tree_kv_checked(dir: &std::path::Path, items: usize) {
use crate::AbstractTree;
use crate::runtime_config::KvChecksumPolicy;
let cfg = Config::new(
dir,
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let any = cfg.open().unwrap();
let crate::AnyTree::Standard(tree) = any else {
panic!("expected Standard tree");
};
tree.update_runtime_config(|c| {
c.kv_checksums = KvChecksumPolicy::AllLevels;
})
.unwrap();
for i in 0u64..items as u64 {
let key = format!("k{i:08}");
let val = format!("v{i:08}");
tree.insert(key.as_bytes(), val.as_bytes(), 1 + i);
}
tree.flush_active_memtable(1 + items as u64).unwrap();
drop(tree);
}
#[test]
fn verify_block_checksums_clean_tree_has_no_errors() {
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 1_000);
let tree = reopen_tree(dir.path());
let report = verify_block_checksums(&tree);
assert!(
report.is_ok(),
"expected clean tree to verify with zero errors, got {:?}",
report.errors
);
assert!(
report.blocks_scanned > 0,
"expected at least one block scanned",
);
assert!(
report.sst_files_scanned >= 1,
"expected at least one SST scanned",
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn verify_block_checksums_clean_page_ecc_tree_has_no_errors() {
use crate::AbstractTree;
let dir = tempfile::tempdir().unwrap();
{
let any = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.page_ecc(true)
.ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 4,
parity_shards: 2,
})
.open()
.unwrap();
for i in 0u64..2_000 {
let key = format!("k{i:08}");
let val = format!("v{i:08}");
any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
}
any.flush_active_memtable(2_001).unwrap();
drop(any);
}
let tree = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.page_ecc(true)
.ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 4,
parity_shards: 2,
})
.open()
.unwrap();
let report = verify_block_checksums(&tree);
assert!(
report.is_ok(),
"page_ecc tree must verify with zero errors (parity trailers skipped \
per block), got {:?}",
report.errors,
);
assert!(
report.blocks_scanned > 1,
"expected multiple blocks scanned to exercise cross-block alignment",
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn verify_block_checksums_clean_nondefault_ecc_tree_has_no_errors() {
use crate::AbstractTree;
let dir = tempfile::tempdir().unwrap();
{
let any = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.page_ecc(true)
.ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 8,
parity_shards: 2,
})
.open()
.unwrap();
for i in 0u64..2_000 {
let key = format!("k{i:08}");
let val = format!("v{i:08}");
any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
}
any.flush_active_memtable(2_001).unwrap();
drop(any);
}
let tree = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.page_ecc(true)
.ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 8,
parity_shards: 2,
})
.open()
.unwrap();
let report = verify_block_checksums(&tree);
assert!(
report.is_ok(),
"non-default-scheme ECC tree must verify with zero errors \
(parity sized from the descriptor, not RS(4,2)), got {:?}",
report.errors,
);
assert!(
report.blocks_scanned > 1,
"expected multiple blocks scanned to exercise cross-block alignment",
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_detects_a_rotted_parity_trailer() {
use crate::coding::Decode;
use crate::table::block::Header;
let dir = tempfile::tempdir().unwrap();
{
let any = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None))
.page_ecc(true)
.ecc_scheme(crate::runtime_config::EccScheme::ReedSolomon {
data_shards: 4,
parity_shards: 2,
})
.open()
.unwrap();
for i in 0u64..2_000 {
let key = format!("k{i:08}");
let val = format!("v{i:08}");
any.insert(key.as_bytes(), val.as_bytes(), 1 + i);
}
any.flush_active_memtable(2_001).unwrap();
drop(any);
}
let sst_path = pick_first_sst_path(dir.path());
assert!(
verify_sst_file(&sst_path).is_ok(),
"the freshly written SST must verify clean before corruption",
);
let mut bytes = std::fs::read(&sst_path).unwrap();
let mut cursor = bytes.as_slice();
let header = Header::decode_from(&mut cursor).unwrap();
let trailer_pos = Header::header_len(header.block_type) + header.data_length as usize;
let slot = bytes
.get_mut(trailer_pos)
.expect("parity trailer within the file");
*slot ^= 0xFF;
std::fs::write(&sst_path, &bytes).unwrap();
let report = verify_sst_file(&sst_path);
assert!(
!report.is_ok(),
"a rotted parity trailer under a clean payload checksum must be \
flagged (dead ECC), got {report:?}",
);
let mismatch = report
.errors
.iter()
.find(|e| matches!(e, BlockVerifyError::EccParityMismatch { .. }))
.unwrap_or_else(|| panic!("expected an EccParityMismatch error, got {report:?}"));
let rendered = mismatch.to_string();
assert!(
rendered.contains("parity trailer") && rendered.contains("offset 0"),
"display names the block and the condition: {rendered}",
);
}
fn pick_first_sst_path(dir: &std::path::Path) -> std::path::PathBuf {
let tree = reopen_tree(dir);
let path = tree
.current_version()
.iter_tables()
.next()
.map(|table| (*table.path).clone())
.expect("at least one populated SST file");
drop(tree);
path
}
#[test]
fn verify_block_checksums_detects_flipped_byte_in_data_block() {
use crate::table::block::Header;
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 1_000);
let sst_path = pick_first_sst_path(dir.path());
let flip_offset = Header::MIN_LEN as u64;
{
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(&sst_path)
.unwrap();
f.seek(SeekFrom::Start(flip_offset)).unwrap();
let mut byte = [0u8; 1];
f.read_exact(&mut byte).unwrap();
byte[0] ^= 0xFF;
f.seek(SeekFrom::Start(flip_offset)).unwrap();
f.write_all(&byte).unwrap();
f.sync_all().unwrap();
}
let tree = reopen_tree(dir.path());
let report = verify_block_checksums(&tree);
assert!(
!report.is_ok(),
"expected corruption to surface as report errors, got {report:?}",
);
let has_data_corruption = report.errors.iter().any(|e| {
matches!(
e,
BlockVerifyError::DataCorrupted { path, .. } if path == &sst_path,
)
});
assert!(
has_data_corruption,
"expected a DataCorrupted error for {}, got {:?}",
sst_path.display(),
report.errors,
);
}
#[test]
fn verify_kv_checksums_clean_kv_checked_tree_passes() {
let dir = tempfile::tempdir().unwrap();
populate_tree_kv_checked(dir.path(), 500);
let tree = reopen_tree(dir.path());
let crate::AnyTree::Standard(tree) = tree else {
panic!("expected Standard tree");
};
verify_kv_checksums(&tree).expect("clean kv-checked tree must pass per-KV scrub");
}
#[test]
fn verify_kv_checked_detects_corrupted_digest_under_valid_block_checksum() {
use crate::InternalValue;
use crate::ValueType::Value;
use crate::comparator::default_comparator;
use crate::runtime_config::ChecksumAlgorithm;
use crate::table::block::header::block_flags;
use crate::table::block::{Block, BlockIdentity, BlockTransform, BlockType, kv_checksum};
use crate::table::data_block::DataBlock;
let algo = ChecksumAlgorithm::Xxh3_64;
let items = [
InternalValue::from_components(b"alpha".to_vec(), b"one".to_vec(), 3, Value),
InternalValue::from_components(b"bravo".to_vec(), b"two".to_vec(), 2, Value),
];
let digests: Vec<u64> = items
.iter()
.map(|it| kv_checksum::kv_digest(it, algo).expect("xxh3 always available"))
.collect();
let mut payload = Vec::new();
DataBlock::encode_kv_checked_into(&mut payload, &items, &digests, algo, 2, 0.0).unwrap();
let inner_len = kv_checksum::split_inner(&payload).unwrap().len();
*payload.get_mut(inner_len).expect("digest array byte") ^= 0xFF;
let id = BlockIdentity::for_test(0, BlockType::Data);
let at = crate::table::block::ChecksumAt::table(0, 0);
let mut buf = Vec::new();
Block::write_into_with_flags(
&mut buf,
&payload,
id,
&BlockTransform::PLAIN,
block_flags::KV_CHECKSUM_FOOTER,
at,
)
.unwrap();
let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN, at).unwrap();
let err = DataBlock::verify_kv_checked(&block.data, block.header, default_comparator(), None)
.expect_err("corrupted stored digest must fail the per-KV verifier");
assert!(
matches!(err, crate::Error::ChecksumMismatch { .. }),
"expected ChecksumMismatch, got {err:?}"
);
}
#[test]
fn verify_kv_checked_rejects_non_data_block_type() {
use crate::InternalValue;
use crate::ValueType::Value;
use crate::comparator::default_comparator;
use crate::runtime_config::ChecksumAlgorithm;
use crate::table::block::header::block_flags;
use crate::table::block::{Block, BlockIdentity, BlockTransform, BlockType, kv_checksum};
use crate::table::data_block::DataBlock;
let algo = ChecksumAlgorithm::Xxh3_64;
let items = [
InternalValue::from_components(b"alpha".to_vec(), b"one".to_vec(), 3, Value),
InternalValue::from_components(b"bravo".to_vec(), b"two".to_vec(), 2, Value),
];
let digests: Vec<u64> = items
.iter()
.map(|it| kv_checksum::kv_digest(it, algo).expect("xxh3 always available"))
.collect();
let mut payload = Vec::new();
DataBlock::encode_kv_checked_into(&mut payload, &items, &digests, algo, 2, 0.0).unwrap();
let id = BlockIdentity::for_test(0, BlockType::Data);
let at = crate::table::block::ChecksumAt::table(0, 0);
let mut buf = Vec::new();
Block::write_into_with_flags(
&mut buf,
&payload,
id,
&BlockTransform::PLAIN,
block_flags::KV_CHECKSUM_FOOTER,
at,
)
.unwrap();
let block = Block::from_reader(&mut &buf[..], id, &BlockTransform::PLAIN, at).unwrap();
let mut bad_header = block.header;
bad_header.block_type = BlockType::Index;
let err = DataBlock::verify_kv_checked(&block.data, bad_header, default_comparator(), None)
.expect_err("non-Data block_type must be rejected, not coerced");
assert!(
matches!(err, crate::Error::InvalidTag(("BlockType", _))),
"expected InvalidTag(BlockType), got {err:?}"
);
}
#[test]
fn verify_sst_file_clean_file_has_no_errors() {
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 1_000);
let sst_path = pick_first_sst_path(dir.path());
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"expected clean SST to verify with zero errors, got {:?}",
report.errors,
);
assert_eq!(
report.sst_files_scanned, 1,
"wrapper must always stamp sst_files_scanned = 1",
);
assert!(
report.blocks_scanned > 0,
"expected at least one block scanned in a populated SST",
);
}
#[test]
fn verify_sst_file_flags_an_omitted_toc_section() {
let dir = tempfile::tempdir().unwrap();
{
let cfg = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let tree = cfg.open().unwrap();
for i in 0u64..100 {
let key = format!("k{i:08}");
tree.insert(key.as_bytes(), b"v", 1 + i);
}
tree.remove_range("k00000010", "k00000020", 200);
tree.flush_active_memtable(300).unwrap();
drop(tree);
}
let sst_path = pick_first_sst_path(dir.path());
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"intact SST must be clean: {:?}",
report.errors
);
crate::test_forge::forge_section_omitted(&sst_path, b"range_tombstones").unwrap();
let report = verify_sst_file(&sst_path);
assert!(
report.errors.iter().any(|e| matches!(
e,
BlockVerifyError::TocCorrupted { reason, .. }
if reason.contains("the gap hides an omitted TOC entry")
)),
"an omitted TOC entry leaves a tiling gap the walk must flag with the \
gap-specific TocCorrupted reason, got {:?}",
report.errors,
);
}
#[test]
fn verify_sst_file_flags_diverging_mirrors_behind_an_unrecognized_ecc() {
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 200);
let sst_path = pick_first_sst_path(dir.path());
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"intact SST must be clean: {:?}",
report.errors
);
crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
.unwrap();
crate::test_forge::forge_tail_meta_value(&sst_path, b"created_at", &[0xFF; 16]).unwrap();
let report = verify_sst_file(&sst_path);
assert!(
report.errors.iter().any(|e| matches!(
e,
BlockVerifyError::TocCorrupted { reason, .. }
if reason.contains("mirrors decode to different metadata")
)),
"a tail mirror that changed created_at behind an unrecognized ECC \
descriptor must still diverge from meta_mid, got {:?}",
report.errors,
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_reports_incomplete_when_ecc_is_unrecognized_in_both_mirrors() -> crate::Result<()>
{
use crate::table::Writer;
use crate::table::block::EccParams;
let dir = tempfile::tempdir()?;
let sst_path = dir.path().join("t");
let mut writer = Writer::new(
sst_path.clone(),
0,
0,
std::sync::Arc::new(crate::fs::StdFs),
)?
.use_ecc(Some(EccParams::RS_4_2));
for i in 0u64..200 {
writer.write(crate::InternalValue::from_components(
format!("key-{i:05}").into_bytes(),
format!("value-{i:05}").into_bytes(),
i + 1,
crate::ValueType::Value,
))?;
}
assert!(writer.finish()?.is_some(), "the fixture is non-empty");
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"intact SST must be clean: {:?}",
report.errors
);
crate::test_forge::forge_meta_value_both_mirrors(
&sst_path,
b"descriptor#page_ecc",
&[9, 0, 0, 0],
)?;
let report = verify_sst_file(&sst_path);
assert!(
report.errors.is_empty(),
"no corruption, only an unwalkable ECC scheme: {:?}",
report.errors,
);
assert!(
report.incomplete,
"the walk skipped the data blocks, so the scan is incomplete",
);
assert!(
!report.is_ok(),
"an incomplete scan (data blocks never verified) must not report OK",
);
assert!(
report
.warnings
.iter()
.any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
"the unrecognized-ECC warning must still be recorded: {:?}",
report.warnings,
);
Ok(())
}
#[cfg(feature = "std")]
#[test]
fn verify_block_checksums_stays_incomplete_when_one_sst_has_unrecognized_ecc() {
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 3, 200);
let tree = reopen_tree(dir.path());
let report = verify_block_checksums(&tree);
assert!(
report.is_ok(),
"intact tree must be clean: {:?}",
report.errors
);
drop(tree);
let sst_path = pick_first_sst_path(dir.path());
crate::test_forge::forge_meta_value_both_mirrors(
&sst_path,
b"descriptor#page_ecc",
&[9, 0, 0, 0],
)
.unwrap();
let tree = reopen_tree(dir.path());
let report = verify_block_checksums(&tree);
assert!(
report.errors.is_empty(),
"no corruption, only an unwalkable ECC scheme: {:?}",
report.errors,
);
assert!(
report.incomplete,
"the merged report must inherit the incomplete flag from the skipped SST",
);
assert!(
!report.is_ok(),
"a merged report that skipped a whole SST's data blocks must not report OK",
);
}
#[test]
fn verify_sst_file_flags_a_duplicate_toc_section_name() {
let dir = tempfile::tempdir().unwrap();
{
let cfg = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let tree = cfg.open().unwrap();
for i in 0u64..100 {
let key = format!("k{i:08}");
tree.insert(key.as_bytes(), b"v", 1 + i);
}
tree.remove_range("k00000010", "k00000020", 200);
tree.flush_active_memtable(300).unwrap();
drop(tree);
}
let sst_path = pick_first_sst_path(dir.path());
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"intact SST must be clean: {:?}",
report.errors
);
crate::test_forge::forge_duplicate_section_name(
&sst_path,
b"range_tombstones",
b"data",
crate::table::block::BlockType::Data,
)
.unwrap();
let report = verify_sst_file(&sst_path);
assert!(
report.errors.iter().any(|e| matches!(
e,
BlockVerifyError::TocCorrupted { section_name, reason, .. }
if section_name == b"data" && reason.contains("duplicate TOC section name")
)),
"a duplicate recognized section name must be flagged as TocCorrupted, \
got {:?}",
report.errors,
);
}
#[test]
fn verify_sst_file_accepts_a_partitioned_filter() {
use crate::InternalValue;
use crate::ValueType::Value;
use crate::table::Writer;
use std::sync::Arc;
let dir = tempfile::tempdir().unwrap();
let sst_path = dir.path().join("partitioned");
let mut writer = Writer::new(sst_path.clone(), 0, 0, Arc::new(crate::fs::StdFs))
.unwrap()
.use_partitioned_filter()
.use_meta_partition_size(3);
for i in 0u64..64 {
writer
.write(InternalValue::from_components(
format!("key-{i:03}").into_bytes(),
format!("val-{i:03}").into_bytes(),
i + 1,
Value,
))
.unwrap();
}
assert!(writer.finish().unwrap().is_some(), "SST is non-empty");
{
let mut f = std::fs::File::open(&sst_path).unwrap();
let reader = crate::sfa::Reader::from_reader(&mut f).unwrap();
assert!(
reader.toc().iter().any(|e| e.name() == b"filter_tli"),
"the fixture must produce a filter_tli section",
);
}
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"a healthy partitioned-filter SST must verify clean: {:?}",
report.errors,
);
}
#[test]
fn verify_sst_file_missing_file_reports_unreadable() {
let dir = tempfile::tempdir().unwrap();
let missing_path = dir.path().join("does-not-exist-sst-12345.sst");
assert!(
!missing_path.exists(),
"tempdir entry must be absent for this test to exercise the missing-file branch",
);
let report = verify_sst_file(&missing_path);
assert_eq!(
report.sst_files_scanned, 1,
"wrapper stamps sst_files_scanned = 1 even on file-open failure \
so callers see the attempt was made",
);
assert_eq!(
report.blocks_scanned, 0,
"no blocks could be walked because the file couldn't be opened",
);
assert_eq!(
report.errors.len(),
1,
"expected exactly one error, got {:?}",
report.errors,
);
let err = report.errors.first().unwrap();
assert!(
matches!(
err,
BlockVerifyError::SstFileUnreadable { table_id: 0, path, .. }
if path == &missing_path,
),
"expected SstFileUnreadable for {}, got {err:?}",
missing_path.display(),
);
}
#[test]
#[expect(
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "synthetic SFA forgery — offsets are all in-bounds by \
construction (we just wrote the bytes ourselves), and \
the u64 -> usize cast cannot overflow on any target \
the test runs on (the forged archive is < 1 KiB)"
)]
fn walk_block_region_reports_data_read_error_on_truncated_data_segment() -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, Header};
const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
const DATA_LENGTH: u32 = 4096;
const HEADER_LEN: u64 = Header::MIN_LEN as u64;
let header = Header {
stored_checksum: Checksum::from_raw(0xDEAD_BEEF_DEAD_BEEF),
data_length: DATA_LENGTH,
uncompressed_length: DATA_LENGTH,
..Header::test_dummy(BlockType::Data)
};
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("data").unwrap();
writer.write_all(&header.encode_into_vec()).unwrap();
writer.finish().unwrap();
}
let trailer_start = archive_bytes.len() - TRAILER_LEN;
let toc_pos_bytes: [u8; 8] = archive_bytes[trailer_start + 22..trailer_start + 30]
.try_into()
.unwrap();
let toc_len_bytes: [u8; 8] = archive_bytes[trailer_start + 30..trailer_start + 38]
.try_into()
.unwrap();
let toc_pos = u64::from_le_bytes(toc_pos_bytes) as usize;
let toc_len = u64::from_le_bytes(toc_len_bytes) as usize;
let first_entry_offset = toc_pos + 4 + 4;
let len_field_offset = first_entry_offset + 8;
let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH);
archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
let csum_field_offset = trailer_start + 4 + 1 + 1;
archive_bytes[csum_field_offset..csum_field_offset + 16]
.copy_from_slice(&new_toc_checksum.to_le_bytes());
let dir = tempfile::tempdir()?;
let fs = StdFs;
let forged = dir.path().join("forged.sst");
let path = forged.as_path();
{
let mut f = fs.open(
path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let table_id: TableId = 42;
let scan = scan_sst_blocks(&fs, path, table_id, SstLayout::PLAIN, None)?;
assert_eq!(
scan.errors.len(),
2,
"expected the tiling finding plus the read error, got {:?}",
scan.errors,
);
assert!(
scan.errors
.iter()
.any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
"the inflated section length must break the TOC tiling: {:?}",
scan.errors,
);
assert!(
scan.errors.iter().any(|err| matches!(
err,
BlockVerifyError::DataReadError {
table_id: t,
offset: 0,
data_length: d,
..
} if *t == table_id && *d == DATA_LENGTH,
)),
"expected DataReadError {{ table_id: {table_id}, offset: 0, \
data_length: {DATA_LENGTH}, .. }}; got {:?}",
scan.errors,
);
assert_eq!(
scan.blocks_scanned, 1,
"header decoded successfully, so blocks_scanned must count this block \
even though the data segment read failed",
);
Ok(())
}
#[test]
#[expect(
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "synthetic SFA forgery: the offsets are in bounds by construction and the \
archive is under 1 KiB"
)]
fn a_rated_scan_charges_the_bytes_read_not_a_forged_section_length() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("meta_separator").unwrap();
writer.write_all(&[0u8; 16]).unwrap();
writer.finish().unwrap();
}
let trailer_start = archive_bytes.len() - TRAILER_LEN;
let field = |at: usize| -> usize {
u64::from_le_bytes(archive_bytes[at..at + 8].try_into().unwrap()) as usize
};
let (toc_pos, toc_len) = (field(trailer_start + 22), field(trailer_start + 30));
let len_at = toc_pos + 4 + 4 + 8;
archive_bytes[len_at..len_at + 8].copy_from_slice(&(1u64 << 40).to_le_bytes());
let checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
let checksum_at = trailer_start + 4 + 1 + 1;
archive_bytes[checksum_at..checksum_at + 16].copy_from_slice(&checksum.to_le_bytes());
let dir = tempfile::tempdir()?;
let path = dir.path().join("forged.sst");
{
let mut f = StdFs.open(
&path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let limiter = crate::rate_limiter::RateLimiter::new(1 << 20);
let start = std::time::Instant::now();
let scan = scan_sst_blocks(&StdFs, &path, 9, SstLayout::PLAIN, Some(&limiter))?;
let elapsed = start.elapsed();
assert!(
elapsed < std::time::Duration::from_secs(2),
"the scan waited {elapsed:?} on a forged section length"
);
assert!(
scan.errors
.iter()
.any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
"the forged length is still reported: {:?}",
scan.errors,
);
Ok(())
}
struct ShortReads(Box<dyn crate::fs::FsFile>);
impl std::io::Read for ShortReads {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
let cap = buf.len().min(512);
match buf.get_mut(..cap) {
Some(head) => self.0.read(head),
None => Ok(0),
}
}
}
impl std::io::Write for ShortReads {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.0.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.0.flush()
}
}
impl std::io::Seek for ShortReads {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.0.seek(to)
}
}
impl crate::fs::FsFile for ShortReads {
fn sync_all(&self) -> crate::io::Result<()> {
self.0.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.0.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.0.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.0.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.0.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.0.lock_exclusive()
}
}
#[test]
fn a_paced_file_charges_short_reads_once() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
let fs = MemFs::new();
let path = std::path::Path::new("/f");
let bytes: Vec<u8> = (0..64 * 1024u32).map(|i| (i % 251) as u8).collect();
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&bytes)?;
}
let limiter = crate::rate_limiter::RateLimiter::new(bytes.len() as u64);
let file = ShortReads(fs.open(path, &FsOpenOptions::new().read(true))?);
let mut reader = std::io::BufReader::with_capacity(
64 * 1024,
PacedFile::new(Box::new(file), Some(&limiter))?,
);
let mut read = vec![0u8; bytes.len()];
let start = std::time::Instant::now();
reader.read_exact(&mut read[..])?;
let elapsed = start.elapsed();
assert_eq!(read, bytes, "the file reads back whole");
assert!(
elapsed < std::time::Duration::from_millis(500),
"short reads were charged more than once: the read waited {elapsed:?}"
);
Ok(())
}
struct InterruptedReads {
inner: Box<dyn crate::fs::FsFile>,
calls: usize,
}
impl std::io::Read for InterruptedReads {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.calls += 1;
if self.calls.is_multiple_of(2) {
return Err(std::io::ErrorKind::Interrupted.into());
}
let cap = buf.len().min(512);
match buf.get_mut(..cap) {
Some(head) => self.inner.read(head),
None => Ok(0),
}
}
}
impl std::io::Write for InterruptedReads {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
impl std::io::Seek for InterruptedReads {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.inner.seek(to)
}
}
impl crate::fs::FsFile for InterruptedReads {
fn sync_all(&self) -> crate::io::Result<()> {
self.inner.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.inner.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.inner.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.inner.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.inner.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.inner.lock_exclusive()
}
}
struct ScriptedReads {
inner: Box<dyn crate::fs::FsFile>,
cap: usize,
fail_at: Option<usize>,
calls: usize,
largest: std::sync::Arc<std::sync::atomic::AtomicUsize>,
}
impl std::io::Read for ScriptedReads {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.calls += 1;
self.largest
.fetch_max(buf.len(), std::sync::atomic::Ordering::Relaxed);
if self.fail_at == Some(self.calls) {
return Err(std::io::ErrorKind::Other.into());
}
let cap = buf.len().min(self.cap);
match buf.get_mut(..cap) {
Some(head) => self.inner.read(head),
None => Ok(0),
}
}
}
impl std::io::Write for ScriptedReads {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
impl std::io::Seek for ScriptedReads {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.inner.seek(to)
}
}
impl crate::fs::FsFile for ScriptedReads {
fn sync_all(&self) -> crate::io::Result<()> {
self.inner.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.inner.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.inner.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.inner.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.inner.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.inner.lock_exclusive()
}
}
struct LoggedReadAt {
inner: Box<dyn crate::fs::FsFile>,
log: std::sync::Arc<std::sync::Mutex<Vec<(char, u64, u64)>>>,
}
impl std::io::Read for LoggedReadAt {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.inner.read(buf)
}
}
impl std::io::Write for LoggedReadAt {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
impl std::io::Seek for LoggedReadAt {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.inner.seek(to)
}
}
impl crate::fs::FsFile for LoggedReadAt {
fn sync_all(&self) -> crate::io::Result<()> {
self.inner.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.inner.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.inner.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.inner.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.log
.lock()
.expect("lock is not poisoned")
.push(('r', offset, buf.len() as u64));
self.inner.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.inner.lock_exclusive()
}
}
#[test]
fn a_paced_read_alternates_each_portion_with_its_read() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
let portion = PACE_PORTION as u64;
let size = 2 * PACE_PORTION + 100;
let fs = MemFs::new();
let path = std::path::Path::new("/f");
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&vec![7u8; size])?;
}
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
let file = LoggedReadAt {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
log: std::sync::Arc::clone(&log),
};
let pace = LogPacer {
active: true,
log: std::sync::Arc::clone(&log),
};
let read = crate::file::read_exact_paced(&file, 0, size, Some(&pace))?;
assert_eq!(read.len(), size);
let tail = size as u64 - 2 * portion;
assert_eq!(
*log.lock().expect("lock is not poisoned"),
vec![
('p', 0, portion),
('r', 0, portion),
('p', portion, portion),
('r', portion, portion),
('p', 2 * portion, tail),
('r', 2 * portion, tail),
]
);
log.lock().expect("lock is not poisoned").clear();
crate::file::read_exact(&file, 0, size)?;
assert_eq!(
*log.lock().expect("lock is not poisoned"),
vec![('r', 0, size as u64)],
"an unpaced read is one read"
);
log.lock().expect("lock is not poisoned").clear();
let idle = LogPacer {
active: false,
log: std::sync::Arc::clone(&log),
};
crate::file::read_exact_paced(&file, 0, size, Some(&idle))?;
assert_eq!(
*log.lock().expect("lock is not poisoned"),
vec![('r', 0, size as u64)],
"a read under an idle pacer is one read, not charged"
);
Ok(())
}
struct LogPacer {
active: bool,
log: std::sync::Arc<std::sync::Mutex<Vec<(char, u64, u64)>>>,
}
impl crate::table::util::ReadPacer for LogPacer {
fn active(&self) -> bool {
self.active
}
fn pace(&self, offset: u64, len: u64) {
self.log
.lock()
.expect("lock is not poisoned")
.push(('p', offset, len));
}
}
fn scripted_file(
bytes: &[u8],
cap: usize,
fail_at: Option<usize>,
) -> crate::Result<(
ScriptedReads,
std::sync::Arc<std::sync::atomic::AtomicUsize>,
)> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
let fs = MemFs::new();
let path = std::path::Path::new("/f");
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(bytes)?;
}
let largest = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
let file = ScriptedReads {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
cap,
fail_at,
calls: 0,
largest: std::sync::Arc::clone(&largest),
};
Ok((file, largest))
}
#[test]
fn a_paced_read_reaches_the_file_a_portion_at_a_time() -> crate::Result<()> {
let bytes: Vec<u8> = (0..1024 * 1024u32).map(|i| (i % 251) as u8).collect();
let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
let mut read = vec![0u8; bytes.len()];
std::io::Read::read_exact(&mut paced, &mut read)?;
assert_eq!(read, bytes, "the file reads back whole");
let largest = largest.load(std::sync::atomic::Ordering::Relaxed);
assert!(
largest <= PACE_PORTION,
"a read of {largest} bytes reached the file at once"
);
let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
let mut unpaced = PacedFile::new(Box::new(file), None)?;
std::io::Read::read_exact(&mut unpaced, &mut read)?;
assert_eq!(read, bytes, "the file reads back whole");
assert_eq!(
largest.load(std::sync::atomic::Ordering::Relaxed),
bytes.len(),
"an unrated read is not split"
);
Ok(())
}
#[test]
fn a_paced_file_reports_an_error_that_followed_a_partial_read() -> crate::Result<()> {
let bytes = vec![7u8; 16 * 1024];
let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
let (file, _) = scripted_file(&bytes, 512, Some(2))?;
let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
let mut buf = vec![0u8; 4096];
assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
let next = std::io::Read::read(&mut paced, &mut buf);
assert!(
matches!(&next, Err(e) if e.kind() == std::io::ErrorKind::Other),
"the failure reached the caller: {next:?}"
);
Ok(())
}
#[test]
fn a_seek_drops_an_error_held_back_from_a_partial_read() -> crate::Result<()> {
let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
let (file, _) = scripted_file(&bytes, 512, Some(2))?;
let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
let mut buf = vec![0u8; 4096];
assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
std::io::Seek::seek(&mut paced, SeekFrom::Start(8192))?;
let read = std::io::Read::read(&mut paced, &mut buf)?;
assert!(read > 0, "the read after the seek returns bytes");
assert_eq!(
buf.get(..read),
bytes.get(8192..8192 + read),
"the bytes at the seek"
);
Ok(())
}
#[test]
fn a_zero_rate_limiter_does_not_split_reads() -> crate::Result<()> {
let bytes: Vec<u8> = (0..1024 * 1024u32).map(|i| (i % 251) as u8).collect();
let limiter = crate::rate_limiter::RateLimiter::new(0);
let (file, largest) = scripted_file(&bytes, usize::MAX, None)?;
let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
let mut read = vec![0u8; bytes.len()];
std::io::Read::read_exact(&mut paced, &mut read)?;
assert_eq!(read, bytes, "the file reads back whole");
assert_eq!(
largest.load(std::sync::atomic::Ordering::Relaxed),
bytes.len(),
"an unrated read is not split"
);
Ok(())
}
#[test]
fn a_large_charge_is_made_a_portion_at_a_time() {
let portion = PACE_PORTION as u64;
let parts: Vec<u64> = pace_portions(3 * portion + 5).collect();
assert_eq!(parts, vec![portion, portion, portion, 5]);
assert_eq!(pace_portions(0).count(), 0, "nothing to charge, no request");
assert_eq!(pace_portions(portion).collect::<Vec<_>>(), vec![portion]);
}
#[test]
fn bytes_charged_but_not_read_are_credited_to_the_next_read() -> crate::Result<()> {
let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
let limiter = crate::rate_limiter::RateLimiter::new(8 * 1024);
let (file, _) = scripted_file(&bytes, 512, Some(2))?;
let mut paced = PacedFile::new(Box::new(file), Some(&limiter))?;
let mut buf = vec![0u8; 4096];
assert_eq!(std::io::Read::read(&mut paced, &mut buf)?, 512);
std::io::Seek::seek(&mut paced, SeekFrom::Start(8192))?;
assert!(std::io::Read::read(&mut paced, &mut buf)? > 0);
let started = std::time::Instant::now();
assert!(!limiter.request_interruptible(3 * 1024, || false));
assert!(
started.elapsed() < std::time::Duration::from_millis(150),
"the next request waited {:?} for bytes never read",
started.elapsed()
);
Ok(())
}
#[test]
fn a_paced_portion_is_charged_only_the_bytes_inside_the_file() {
assert_eq!(bytes_in_file(0, 4096, 8192), 4096);
assert_eq!(bytes_in_file(8092, 4096, 8192), 100);
assert_eq!(bytes_in_file(8192, 4096, 8192), 0);
assert_eq!(bytes_in_file(u64::MAX, 4096, 8192), 0);
}
#[test]
fn a_restricted_lookup_at_rate_zero_reads_no_metadata() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
let dir = tempfile::tempdir()?;
let fs = FaultFs::new(StdFs);
let injector = fs.injector();
let tree = crate::Config::new(
dir.path(),
crate::SequenceNumberCounter::default(),
crate::SequenceNumberCounter::default(),
)
.with_shared_fs(std::sync::Arc::new(fs))
.open()?;
for i in 0..100u32 {
tree.insert(format!("k{i:05}"), "v", u64::from(i));
}
tree.flush_active_memtable(0)?;
let version = tree.current_version();
let table = version.iter_tables().next().expect("one table");
let expected = table.punch_offset_for(b"k00050")?;
injector.arm(FaultRule::new(
FaultOp::Metadata,
Fault::Error(crate::io::ErrorKind::Other),
));
let idle = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(0));
let found = super::paced_punch_offset(table, b"k00050", Some(&idle));
injector.clear();
assert_eq!(found?, expected, "an idle limiter walks as no limiter does");
Ok(())
}
#[test]
fn a_walk_started_idle_is_paced_once_a_rate_is_set() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, StdFs};
use crate::table::util::ReadPacer;
let dir = tempfile::tempdir()?;
let fs = FaultFs::new(StdFs);
let injector = fs.injector();
let tree = crate::Config::new(
dir.path(),
crate::SequenceNumberCounter::default(),
crate::SequenceNumberCounter::default(),
)
.with_shared_fs(std::sync::Arc::new(fs))
.open()?;
tree.insert("k", "v", 0);
tree.flush_active_memtable(0)?;
let version = tree.current_version();
let table = version.iter_tables().next().expect("one table");
let limiter = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(0));
let pacer = super::walk_pacer(table, Some(&limiter)).expect("a limiter attaches a pacer");
assert!(!pacer.active(), "an idle limiter reads at full speed");
limiter.set_rate(1 << 30);
assert!(
pacer.active(),
"a rate set mid-walk paces the reads after it"
);
injector.arm(FaultRule::new(
FaultOp::Metadata,
Fault::Error(crate::io::ErrorKind::Other),
));
pacer.pace(0, 4096);
injector.clear();
assert!(
super::walk_pacer(table, None).is_none(),
"no limiter, no pacer"
);
Ok(())
}
#[test]
fn a_restricted_index_lookup_is_charged_what_it_reads() -> crate::Result<()> {
use crate::config::{BlockSizePolicy, PinningPolicy};
use crate::table::block::ParsedItem;
use crate::table::block_index::BlockIndexImpl;
let keys: Vec<String> = (0..2_000).map(|i| format!("k{i:05}")).collect();
for shape in ["pinned", "volatile", "partitioned"] {
let dir = tempfile::tempdir()?;
let config = crate::Config::new(
dir.path(),
crate::SequenceNumberCounter::default(),
crate::SequenceNumberCounter::default(),
)
.use_cache(std::sync::Arc::new(crate::Cache::with_capacity_bytes(1)));
let config = match shape {
"pinned" => config
.index_block_partitioning_policy(PinningPolicy::all(false))
.index_block_pinning_policy(PinningPolicy::all(true)),
"volatile" => config
.index_block_partitioning_policy(PinningPolicy::all(false))
.index_block_pinning_policy(PinningPolicy::all(false)),
_ => {
let runtime = crate::runtime_config::RuntimeConfig {
index_partition_spill_threshold: 0,
..Default::default()
};
config
.data_block_size_policy(BlockSizePolicy::all(256))
.index_block_partitioning_policy(PinningPolicy::all(true))
.index_block_partition_size_policy(BlockSizePolicy::all(256))
.with_runtime_config(runtime)
}
};
let tree = config.open()?;
for (seqno, key) in keys.iter().enumerate() {
tree.insert(key, "v", seqno as u64);
}
tree.flush_active_memtable(0)?;
let version = tree.current_version();
let table = version.iter_tables().next().expect("one table");
let file_size = table.fs.metadata(&table.path)?.len;
let charged = |bound: &[u8]| -> crate::Result<u64> {
let log = std::sync::Arc::new(std::sync::Mutex::new(Vec::new()));
table.punch_offset_paced(
bound,
crate::table::util::ReadCharge::Untraced,
std::sync::Arc::new(LogPacer {
active: true,
log: std::sync::Arc::clone(&log),
}),
)?;
let paced = log.lock().expect("lock is not poisoned");
Ok(paced.iter().map(|&(_, _, len)| len).sum())
};
let first = charged(b"k00000")?;
let middle = charged(b"k01000")?;
let past = charged(b"z")?;
assert!(past <= file_size, "never more than the file holds");
match &*table.block_index {
BlockIndexImpl::Full(_) => {
assert_eq!(shape, "pinned");
assert_eq!((0, 0, 0), (first, middle, past));
}
BlockIndexImpl::VolatileFull(index) => {
assert_eq!(shape, "volatile");
let size = u64::from(index.handle.size());
assert!(size > 0);
assert_eq!((size, size, size), (first, middle, past));
}
BlockIndexImpl::TwoLevel(index) => {
assert_eq!(shape, "partitioned");
let tli = &index.top_level_index;
let all: u64 = tli
.iter(table.comparator.clone())
.map(|item| u64::from(item.materialize(tli.as_slice()).size()))
.sum();
assert!(tli.len() > 2, "several partitions");
assert!(0 < first && first < middle && middle < past);
assert_eq!(all, past, "a bound past the end walks every partition");
}
BlockIndexImpl::Closed => panic!("a live table"),
}
}
Ok(())
}
#[test]
fn a_paced_file_keeps_the_bytes_read_before_an_interruption() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
let fs = MemFs::new();
let path = std::path::Path::new("/f");
let bytes: Vec<u8> = (0..16 * 1024u32).map(|i| (i % 251) as u8).collect();
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&bytes)?;
}
let limiter = crate::rate_limiter::RateLimiter::new(1 << 30);
let file = InterruptedReads {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
calls: 0,
};
let mut reader = std::io::BufReader::with_capacity(
4 * 1024,
PacedFile::new(Box::new(file), Some(&limiter))?,
);
let mut read = vec![0u8; bytes.len()];
reader.read_exact(&mut read[..])?;
assert_eq!(read, bytes, "the file reads back whole");
Ok(())
}
struct SlowReads {
inner: Box<dyn crate::fs::FsFile>,
delay: std::time::Duration,
fails: bool,
}
impl std::io::Read for SlowReads {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
std::thread::sleep(self.delay);
if self.fails {
return Err(std::io::ErrorKind::Other.into());
}
self.inner.read(buf)
}
}
impl std::io::Write for SlowReads {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
impl std::io::Seek for SlowReads {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.inner.seek(to)
}
}
impl crate::fs::FsFile for SlowReads {
fn sync_all(&self) -> crate::io::Result<()> {
self.inner.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.inner.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.inner.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.inner.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.inner.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.inner.lock_exclusive()
}
}
#[test]
fn a_paced_file_reports_its_read_latency_to_the_limiter() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use crate::rate_limiter::{LatencyBackoff, RateLimiter};
use std::time::Duration;
let fs = MemFs::new();
let path = std::path::Path::new("/f");
let bytes = vec![7u8; 16 * 1024];
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&bytes)?;
}
let slow = || -> crate::Result<SlowReads> {
Ok(SlowReads {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
delay: Duration::from_millis(5),
fails: false,
})
};
let configured = 1 << 30;
let limiter = RateLimiter::new(configured);
limiter.set_latency_backoff(Some(
LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
));
let mut reader =
std::io::BufReader::with_capacity(1024, PacedFile::new(Box::new(slow()?), Some(&limiter))?);
let mut read = vec![0u8; bytes.len()];
reader.read_exact(&mut read[..])?;
assert_eq!(read, bytes);
let granted = limiter.effective_rate();
assert!(granted < configured, "a slow device did not lower the rate");
assert!(
granted >= configured / 20,
"the rate fell below the 5% floor: {granted}"
);
let unmanaged = RateLimiter::new(configured);
let mut reader = std::io::BufReader::with_capacity(
1024,
PacedFile::new(Box::new(slow()?), Some(&unmanaged))?,
);
reader.read_exact(&mut read[..])?;
assert_eq!(configured, unmanaged.effective_rate());
Ok(())
}
#[test]
fn a_slow_failed_read_lowers_the_granted_rate() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use crate::rate_limiter::{LatencyBackoff, RateLimiter};
use std::time::Duration;
let fs = MemFs::new();
let path = std::path::Path::new("/f");
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&[7u8; 4096])?;
}
let configured = 1 << 30;
let limiter = RateLimiter::new(configured);
limiter.set_latency_backoff(Some(
LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
));
let failing = SlowReads {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
delay: Duration::from_millis(5),
fails: true,
};
let mut paced = PacedFile::new(Box::new(failing), Some(&limiter))?;
let mut buf = [0u8; 1024];
for _ in 0..4 {
assert!(std::io::Read::read(&mut paced, &mut buf).is_err());
}
assert!(
limiter.effective_rate() < configured,
"slow failures did not lower the rate"
);
Ok(())
}
#[test]
fn a_failed_read_does_not_hand_back_the_credit_it_spent() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use crate::rate_limiter::RateLimiter;
use std::time::Duration;
let fs = MemFs::new();
let path = std::path::Path::new("/f");
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&[7u8; 4096])?;
}
let limiter = RateLimiter::new(1 << 30);
let failing = SlowReads {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
delay: Duration::ZERO,
fails: true,
};
let mut paced = PacedFile::new(Box::new(failing), Some(&limiter))?;
let mut buf = [0u8; 1024];
let mut credits = Vec::new();
for _ in 0..4 {
assert!(std::io::Read::read(&mut paced, &mut buf).is_err());
credits.push(paced.credit);
}
assert_eq!(credits, vec![1024, 0, 1024, 0]);
Ok(())
}
struct RetunedDuringRead {
inner: Box<dyn crate::fs::FsFile>,
limiter: std::sync::Arc<crate::rate_limiter::RateLimiter>,
delay: std::time::Duration,
}
impl std::io::Read for RetunedDuringRead {
fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
self.limiter.set_rate(1 << 30);
std::thread::sleep(self.delay);
self.inner.read(buf)
}
}
impl std::io::Write for RetunedDuringRead {
fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
self.inner.write(buf)
}
fn flush(&mut self) -> std::io::Result<()> {
self.inner.flush()
}
}
impl std::io::Seek for RetunedDuringRead {
fn seek(&mut self, to: SeekFrom) -> std::io::Result<u64> {
self.inner.seek(to)
}
}
impl crate::fs::FsFile for RetunedDuringRead {
fn sync_all(&self) -> crate::io::Result<()> {
self.inner.sync_all()
}
fn sync_data(&self) -> crate::io::Result<()> {
self.inner.sync_data()
}
fn metadata(&self) -> crate::io::Result<crate::fs::FsMetadata> {
self.inner.metadata()
}
fn set_len(&self, size: u64) -> crate::io::Result<()> {
self.inner.set_len(size)
}
fn read_at(&self, buf: &mut [u8], offset: u64) -> crate::io::Result<usize> {
self.inner.read_at(buf, offset)
}
fn lock_exclusive(&self) -> crate::io::Result<()> {
self.inner.lock_exclusive()
}
}
#[test]
fn a_read_decided_without_a_rate_reports_no_latency() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use crate::rate_limiter::{LatencyBackoff, RateLimiter};
use std::time::Duration;
let fs = MemFs::new();
let path = std::path::Path::new("/f");
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&[7u8; 4096])?;
}
let limiter = std::sync::Arc::new(RateLimiter::new(0));
limiter.set_latency_backoff(Some(
LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
));
let retuned = RetunedDuringRead {
inner: fs.open(path, &FsOpenOptions::new().read(true))?,
limiter: std::sync::Arc::clone(&limiter),
delay: Duration::from_millis(5),
};
let mut paced = PacedFile::new(Box::new(retuned), Some(&limiter))?;
let mut buf = [0u8; 4096];
std::io::Read::read(&mut paced, &mut buf)?;
assert_eq!(
limiter.effective_rate(),
limiter.rate(),
"an uncharged read lowered the rate"
);
Ok(())
}
#[test]
fn a_paced_index_read_reports_its_latency() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use crate::rate_limiter::{LatencyBackoff, RateLimiter};
use crate::table::util::ReadPacer;
use std::time::Duration;
struct Timed(std::sync::atomic::AtomicUsize);
impl ReadPacer for Timed {
fn active(&self) -> bool {
true
}
fn pace(&self, _offset: u64, _len: u64) {}
fn read_took(&self, _elapsed: Duration) {
self.0.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
}
}
let fs = MemFs::new();
let path = std::path::Path::new("/f");
let size = 2 * PACE_PORTION + 100;
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&vec![7u8; size])?;
}
let file = fs.open(path, &FsOpenOptions::new().read(true))?;
let timed = Timed(std::sync::atomic::AtomicUsize::new(0));
crate::file::read_exact_paced(file.as_ref(), 0, size, Some(&timed))?;
assert_eq!(
3,
timed.0.load(std::sync::atomic::Ordering::Relaxed),
"every portion read reports its time"
);
let dir = tempfile::tempdir()?;
let tree = crate::Config::new(
dir.path(),
crate::SequenceNumberCounter::default(),
crate::SequenceNumberCounter::default(),
)
.open()?;
tree.insert("k", "v", 0);
tree.flush_active_memtable(0)?;
let version = tree.current_version();
let table = version.iter_tables().next().expect("one table");
let configured = 1 << 30;
let limiter = std::sync::Arc::new(RateLimiter::new(configured));
limiter.set_latency_backoff(Some(
LatencyBackoff::new(Duration::from_millis(1)).with_period(Duration::ZERO),
));
let pacer = super::walk_pacer(table, Some(&limiter)).expect("a limiter attaches a pacer");
pacer.read_took(Duration::from_millis(10));
assert!(
limiter.effective_rate() < configured,
"a slow index read did not lower the rate"
);
Ok(())
}
#[test]
fn a_large_paced_read_lets_a_shared_limiter_serve_others_between_portions() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, MemFs};
use std::time::{Duration, Instant};
let fs = MemFs::new();
let path = std::path::Path::new("/f");
let rate: u64 = 256 * 1024;
let bytes = vec![3u8; 8 * 256 * 1024];
{
let mut f = fs.open(path, &FsOpenOptions::new().write(true).create(true))?;
f.write_all(&bytes)?;
}
let limiter = std::sync::Arc::new(crate::rate_limiter::RateLimiter::new(rate));
assert!(!limiter.request_interruptible(rate, || false));
let scan = {
let limiter = std::sync::Arc::clone(&limiter);
let file = fs.open(path, &FsOpenOptions::new().read(true))?;
std::thread::spawn(move || -> crate::Result<()> {
let mut paced = PacedFile::new(file, Some(&limiter))?;
let mut buf = vec![0u8; 8 * 256 * 1024];
std::io::Read::read_exact(&mut paced, &mut buf)?;
Ok(())
})
};
std::thread::sleep(Duration::from_millis(300));
let started = Instant::now();
assert!(!limiter.request_interruptible(1024, || false));
let waited = started.elapsed();
scan.join().expect("scan thread")?;
assert!(
waited < Duration::from_secs(2),
"a small request waited {waited:?} behind one large read"
);
Ok(())
}
#[test]
#[expect(
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "synthetic SFA forgery — offsets are in-bounds by construction (we wrote the \
bytes ourselves) and the archive is < 8 KiB, so the casts cannot overflow"
)]
fn walk_block_region_reports_data_read_error_on_truncated_parity_trailer() -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};
const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
const DATA_LENGTH: u32 = 4096;
const HEADER_LEN: u64 = Header::MIN_LEN as u64;
let data = vec![0xABu8; DATA_LENGTH as usize];
let mut header = Header {
data_length: DATA_LENGTH,
uncompressed_length: DATA_LENGTH,
..Header::test_dummy(BlockType::Data)
};
header.bind_checksum(
Checksum::from_raw(crate::hash::hash128(&data)),
crate::table::block::ChecksumAt::table(7, 0),
);
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("data").unwrap();
writer.write_all(&header.encode_into_vec()).unwrap();
writer.write_all(&data).unwrap();
writer.finish().unwrap();
}
let parity_len = u64::from(expected_parity_len(DATA_LENGTH, EccParams::RS_4_2));
let trailer_start = archive_bytes.len() - TRAILER_LEN;
let toc_pos = u64::from_le_bytes(
archive_bytes[trailer_start + 22..trailer_start + 30]
.try_into()
.unwrap(),
) as usize;
let toc_len = u64::from_le_bytes(
archive_bytes[trailer_start + 30..trailer_start + 38]
.try_into()
.unwrap(),
) as usize;
let len_field_offset = toc_pos + 4 + 4 + 8;
let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH) + parity_len;
archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
let csum_field_offset = trailer_start + 4 + 1 + 1;
archive_bytes[csum_field_offset..csum_field_offset + 16]
.copy_from_slice(&new_toc_checksum.to_le_bytes());
let dir = tempfile::tempdir()?;
let fs = StdFs;
let forged = dir.path().join("forged-parity.sst");
let path = forged.as_path();
{
let mut f = fs.open(
path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let table_id: TableId = 7;
let layout = SstLayout {
ecc: Some(EccParams::RS_4_2),
..SstLayout::PLAIN
};
let scan = scan_sst_blocks(&fs, path, table_id, layout, None)?;
assert!(
scan.errors.iter().any(|e| matches!(
e,
BlockVerifyError::DataReadError { table_id: t, offset: 0, error, .. }
if *t == table_id && error.kind() == crate::io::ErrorKind::UnexpectedEof
)),
"expected a truncated-parity DataReadError, got {:?}",
scan.errors,
);
Ok(())
}
#[test]
#[expect(
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "synthetic SFA forgery — offsets are in-bounds by construction (we wrote the \
bytes ourselves) and the archive is small, so the casts cannot overflow"
)]
fn walk_block_region_caps_an_absurd_parity_trailer_length() -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};
const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
const DATA_LENGTH: u32 = 2 * 1024 * 1024;
const HEADER_LEN: u64 = Header::MIN_LEN as u64;
let params = EccParams::try_new(1, 255).expect("a 1/255 shard layout parses");
let parity_len = u64::from(expected_parity_len(DATA_LENGTH, params));
assert!(
parity_len > u64::from(DATA_LENGTH) * 200,
"the forged scheme must amplify parity far past the payload",
);
let data = vec![0xABu8; DATA_LENGTH as usize];
let mut header = Header {
data_length: DATA_LENGTH,
uncompressed_length: DATA_LENGTH,
..Header::test_dummy(BlockType::Data)
};
header.bind_checksum(
Checksum::from_raw(crate::hash::hash128(&data)),
crate::table::block::ChecksumAt::table(7, 0),
);
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("data").unwrap();
writer.write_all(&header.encode_into_vec()).unwrap();
writer.write_all(&data).unwrap();
writer.finish().unwrap();
}
let trailer_start = archive_bytes.len() - TRAILER_LEN;
let toc_pos = u64::from_le_bytes(
archive_bytes[trailer_start + 22..trailer_start + 30]
.try_into()
.unwrap(),
) as usize;
let toc_len = u64::from_le_bytes(
archive_bytes[trailer_start + 30..trailer_start + 38]
.try_into()
.unwrap(),
) as usize;
let len_field_offset = toc_pos + 4 + 4 + 8;
let lied_len: u64 = HEADER_LEN + u64::from(DATA_LENGTH) + parity_len;
archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
let csum_field_offset = trailer_start + 4 + 1 + 1;
archive_bytes[csum_field_offset..csum_field_offset + 16]
.copy_from_slice(&new_toc_checksum.to_le_bytes());
let dir = tempfile::tempdir()?;
let fs = StdFs;
let forged = dir.path().join("forged-parity-cap.sst");
let path = forged.as_path();
{
let mut f = fs.open(
path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let table_id: TableId = 7;
let layout = SstLayout {
ecc: Some(params),
..SstLayout::PLAIN
};
let scan = scan_sst_blocks(&fs, path, table_id, layout, None)?;
assert!(
scan.errors.iter().any(|e| matches!(
e,
BlockVerifyError::HeaderCorrupted { table_id: t, offset: 0, reason, .. }
if *t == table_id && reason.contains("parity trailer length")
)),
"an over-cap parity trailer must be HeaderCorrupted without reserving \
the buffer, got {:?}",
scan.errors,
);
Ok(())
}
#[test]
#[expect(
clippy::indexing_slicing,
clippy::cast_possible_truncation,
reason = "synthetic SFA forgery — offsets are in-bounds by construction \
and the forged archive is < 1 KiB"
)]
fn walk_block_region_reports_header_crossing_section_boundary() -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, Header};
const TRAILER_LEN: usize = 4 + 1 + 1 + 16 + 8 + 8;
let header = Header {
stored_checksum: Checksum::from_raw(0xDEAD_BEEF_DEAD_BEEF),
data_length: 0,
uncompressed_length: 0,
..Header::test_dummy(BlockType::Meta)
};
assert_eq!(
Header::header_len(BlockType::Meta) as u64,
Header::MIN_LEN as u64 + 1,
);
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("meta").unwrap();
writer.write_all(&header.encode_into_vec()).unwrap();
writer.finish().unwrap();
}
let trailer_start = archive_bytes.len() - TRAILER_LEN;
let toc_pos_bytes: [u8; 8] = archive_bytes[trailer_start + 22..trailer_start + 30]
.try_into()
.unwrap();
let toc_len_bytes: [u8; 8] = archive_bytes[trailer_start + 30..trailer_start + 38]
.try_into()
.unwrap();
let toc_pos = u64::from_le_bytes(toc_pos_bytes) as usize;
let toc_len = u64::from_le_bytes(toc_len_bytes) as usize;
let first_entry_offset = toc_pos + 4 + 4;
let len_field_offset = first_entry_offset + 8;
let lied_len: u64 = Header::MIN_LEN as u64;
archive_bytes[len_field_offset..len_field_offset + 8].copy_from_slice(&lied_len.to_le_bytes());
let new_toc_checksum = crate::hash::hash128(&archive_bytes[toc_pos..toc_pos + toc_len]);
let csum_field_offset = trailer_start + 4 + 1 + 1;
archive_bytes[csum_field_offset..csum_field_offset + 16]
.copy_from_slice(&new_toc_checksum.to_le_bytes());
let dir = tempfile::tempdir()?;
let fs = StdFs;
let forged = dir.path().join("forged-boundary.sst");
let path = forged.as_path();
{
let mut f = fs.open(
path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let table_id: TableId = 7;
let scan = scan_sst_blocks(&fs, path, table_id, SstLayout::PLAIN, None)?;
assert_eq!(
scan.errors.len(),
2,
"expected the tiling finding plus the boundary violation, got {:?}",
scan.errors,
);
assert!(
scan.errors
.iter()
.any(|e| matches!(e, BlockVerifyError::TocCorrupted { .. })),
"the shrunken section length must break the TOC tiling: {:?}",
scan.errors,
);
assert!(
scan.errors.iter().any(|err| matches!(
err,
BlockVerifyError::HeaderCorrupted { table_id: t, offset: 0, reason, .. }
if *t == table_id && reason.contains("extends past the section end"),
)),
"expected a section-boundary HeaderCorrupted; got {:?}",
scan.errors,
);
Ok(())
}
fn populate_multi_sst(dir: &std::path::Path, batches: usize, per_batch: usize) {
let cfg = Config::new(
dir,
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let tree = cfg.open().unwrap();
let mut seqno = 1u64;
for b in 0..batches {
for i in 0..per_batch {
let key = format!("b{b:03}k{i:08}");
tree.insert(key.as_bytes(), b"v".as_slice(), seqno);
seqno += 1;
}
tree.flush_active_memtable(seqno).unwrap();
seqno += 1;
}
drop(tree);
}
#[test]
fn verify_checksum_method_on_clean_tree_is_ok() {
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 500);
let tree = reopen_tree(dir.path());
let report = tree.verify_checksum();
assert!(report.is_ok(), "clean tree must verify clean: {report:?}");
assert!(report.sst_files_scanned >= 1);
assert!(report.blocks_scanned >= 1);
}
#[test]
fn verify_checksum_with_parallel_matches_sequential() {
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 5, 300);
let tree = reopen_tree(dir.path());
let seq = tree.verify_checksum_with(&VerifyOptions::default());
let par = tree.verify_checksum_with(&VerifyOptions::default().parallelism(4));
assert!(
seq.sst_files_scanned >= 2,
"need >1 SST to exercise parallelism, got {}",
seq.sst_files_scanned,
);
assert_eq!(seq.sst_files_scanned, par.sst_files_scanned);
assert_eq!(seq.blocks_scanned, par.blocks_scanned);
assert_eq!(seq.errors.len(), par.errors.len());
assert!(
seq.is_ok() && par.is_ok(),
"clean tree: seq={seq:?} par={par:?}"
);
}
#[test]
fn verify_checksum_with_throttle_runs_inter_sst_pause() {
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 3, 300);
let tree = reopen_tree(dir.path());
let report = tree.verify_checksum_with(
&VerifyOptions::default().throttle(std::time::Duration::from_nanos(1)),
);
assert!(
report.sst_files_scanned >= 2,
"need >1 SST to exercise the inter-SST throttle, got {}",
report.sst_files_scanned,
);
assert!(report.is_ok(), "clean tree must verify clean: {report:?}");
}
#[test]
fn verify_checksum_with_parallel_detects_corruption() {
use crate::table::block::Header;
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 4, 300);
let sst_path = pick_first_sst_path(dir.path());
let flip_offset = Header::MIN_LEN as u64;
{
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(&sst_path)
.unwrap();
f.seek(SeekFrom::Start(flip_offset)).unwrap();
let mut byte = [0u8; 1];
f.read_exact(&mut byte).unwrap();
byte[0] ^= 0xFF;
f.seek(SeekFrom::Start(flip_offset)).unwrap();
f.write_all(&byte).unwrap();
f.sync_all().unwrap();
}
let tree = reopen_tree(dir.path());
let report = tree.verify_checksum_with(&VerifyOptions::default().parallelism(4));
assert!(
!report.is_ok(),
"parallel scrub must surface the flipped byte: {report:?}",
);
}
#[test]
fn verify_checksum_with_throttle_completes_clean() {
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 3, 200);
let tree = reopen_tree(dir.path());
let opts = VerifyOptions::default()
.parallelism(2)
.throttle(std::time::Duration::from_millis(1));
let report = tree.verify_checksum_with(&opts);
assert!(
report.is_ok(),
"throttled scrub must still verify clean: {report:?}"
);
assert!(report.sst_files_scanned >= 2);
}
#[test]
fn verify_sst_file_flags_a_corrupt_blob_link_count() {
let dir = tempfile::tempdir().unwrap();
let crate::AnyTree::Blob(tree) = Config::new(
dir.path(),
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.with_kv_separation(Some(crate::KvSeparationOptions::default()))
.open()
.unwrap() else {
unreachable!("kv separation configured");
};
let big = |i: u32| format!("{i:08}").repeat(512);
for i in 0u32..10 {
tree.insert(format!("key{i:05}"), big(i), u64::from(i) + 1);
}
tree.flush_active_memtable(10).unwrap();
let sst_path = {
let binding = tree.index.version_history.read().latest_version();
let table = binding
.version
.iter_tables()
.next()
.expect("flush produced one table");
(*table.path).clone()
};
drop(tree);
let pos = {
let mut f = std::fs::File::open(&sst_path).unwrap();
let reader = crate::sfa::Reader::from_reader(&mut f).expect("SFA trailer reads");
let entry = reader
.toc()
.iter()
.find(|e| e.name() == b"linked_blob_files")
.expect("the SST carries a linked_blob_files section");
usize::try_from(entry.pos()).expect("section offset fits usize")
};
let mut bytes = std::fs::read(&sst_path).unwrap();
*bytes.get_mut(pos).expect("count prefix within the file") ^= 0xFF;
std::fs::write(&sst_path, &bytes).unwrap();
let fs: alloc::sync::Arc<dyn crate::fs::Fs> = alloc::sync::Arc::new(crate::fs::StdFs);
let report = verify_sst_file_with_fs(&fs, &sst_path);
assert!(
report.errors.iter().any(|e| matches!(
e,
BlockVerifyError::TocCorrupted { section_name, reason, .. }
if section_name == b"linked_blob_files" && reason.contains("blob-link count")
)),
"a corrupt blob-link count must fail the out-of-band walk, not be \
skipped as an unchecked raw section: {report:?}",
);
}
fn blob_link_shape(bytes: &[u8]) -> Option<String> {
use crate::fs::{Fs as _, FsOpenOptions};
let fs = crate::fs::MemFs::new();
let path = std::path::Path::new("/linked_blob_files");
{
let mut file = fs
.open(path, &FsOpenOptions::new().write(true).create(true))
.expect("create");
file.write_all(bytes).expect("write");
}
let file = fs
.open(path, &FsOpenOptions::new().read(true))
.expect("open");
let mut reader = std::io::BufReader::new(file);
let len = u64::try_from(bytes.len()).expect("test section fits u64");
raw_section_shape_error(&mut reader, b"linked_blob_files", 0, len).expect("in-memory read")
}
fn blob_link_record(first: &[u8], last: &[u8]) -> Vec<u8> {
let mut record = vec![0u8; crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN];
for key in [first, last] {
let len = u16::try_from(key.len()).expect("test key fits u16");
record.extend_from_slice(&len.to_le_bytes());
record.extend_from_slice(key);
}
record
}
fn blob_link_section(count: u32, records: &[Vec<u8>]) -> Vec<u8> {
let mut section = count.to_le_bytes().to_vec();
for record in records {
section.extend_from_slice(record);
}
section
}
#[test]
fn blob_link_shape_accepts_records_that_fill_the_section() {
let records = [blob_link_record(b"a", b"m"), blob_link_record(b"", b"zz")];
assert_eq!(blob_link_shape(&blob_link_section(2, &records)), None);
}
#[test]
fn blob_link_shape_names_the_record_that_runs_past_the_section() {
let record = blob_link_record(b"a", b"b");
let fixed = blob_link_shape(&blob_link_section(2, core::slice::from_ref(&record)));
assert!(
fixed
.as_deref()
.is_some_and(|r| r.contains("record 1 runs past")),
"{fixed:?}"
);
let mut prefix = blob_link_section(2, core::slice::from_ref(&record));
prefix.extend_from_slice(&[0u8; crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN + 1]);
let short = blob_link_shape(&prefix);
assert!(
short
.as_deref()
.is_some_and(|r| r.contains("record 1 runs past")),
"{short:?}"
);
let mut long_key = blob_link_record(b"a", b"b");
let last_len_at = crate::table::writer::LINKED_BLOB_FILE_FIXED_LEN + 2 + 1;
long_key
.get_mut(last_len_at..last_len_at + 2)
.expect("last key length")
.copy_from_slice(&u16::MAX.to_le_bytes());
let past = blob_link_shape(&blob_link_section(1, &[long_key]));
assert!(
past.as_deref()
.is_some_and(|r| r.contains("record 0 runs past")),
"{past:?}"
);
}
#[test]
fn blob_link_shape_rejects_bytes_past_the_declared_records() {
let record = blob_link_record(b"a", b"b");
let mut section = blob_link_section(1, core::slice::from_ref(&record));
section.push(0);
let reason = blob_link_shape(§ion);
let taken = 4 + record.len();
assert!(
reason
.as_deref()
.is_some_and(|r| r.contains(&format!("({taken} bytes of records)"))),
"{reason:?}"
);
}
#[test]
fn verify_checksum_with_throttle_does_not_sleep_after_last_sst() {
let dir = tempfile::tempdir().unwrap();
populate_multi_sst(dir.path(), 1, 50);
let tree = reopen_tree(dir.path());
let throttle = std::time::Duration::from_millis(400);
let opts = VerifyOptions::default().parallelism(1).throttle(throttle);
let start = std::time::Instant::now();
let report = tree.verify_checksum_with(&opts);
let elapsed = start.elapsed();
assert!(report.is_ok(), "clean single-SST scrub: {report:?}");
assert_eq!(report.sst_files_scanned, 1, "test needs exactly one SST");
assert!(
elapsed < throttle / 2,
"a single-SST scrub must not sleep the inter-SST throttle after the \
last table: took {elapsed:?} with a {throttle:?} throttle",
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn verify_sst_file_lone_recognized_mirror_that_misframes_falls_back_to_unrecognized() {
use crate::table::Writer;
use crate::table::block::EccParams;
let dir = tempfile::tempdir().unwrap();
let path = dir.path().join("t");
let mut writer = Writer::new(path.clone(), 0, 0, std::sync::Arc::new(crate::fs::StdFs))
.unwrap()
.use_ecc(Some(EccParams::RS_4_2));
for i in 0u64..200 {
writer
.write(crate::InternalValue::from_components(
format!("key-{i:05}").into_bytes(),
format!("value-{i:05}").into_bytes(),
i + 1,
crate::ValueType::Value,
))
.unwrap();
}
assert!(
writer.finish().unwrap().is_some(),
"the fixture is non-empty"
);
let report = verify_sst_file(&path);
assert!(
report.is_ok(),
"an intact RS(4,2) table must verify clean: {:?}",
report.errors,
);
crate::test_forge::forge_mid_meta_value(&path, b"descriptor#page_ecc", &[0, 0, 0, 0]).unwrap();
crate::test_forge::forge_tail_meta_value(&path, b"descriptor#page_ecc", &[9, 0, 0, 0]).unwrap();
let report = verify_sst_file(&path);
assert!(
report.errors.is_empty(),
"the blocks are healthy — a mis-framing descriptor must not be trusted \
into reporting them corrupt: {:?}",
report.errors,
);
assert!(
report
.warnings
.iter()
.any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
"the descriptor that does not frame the data must fail safe to \
unrecognized: {:?}",
report.warnings,
);
assert!(
report.incomplete,
"the ECC-dependent sections were skipped, so the scan is incomplete",
);
}
#[test]
fn verify_sst_file_lone_recognized_mirror_that_frames_stays_authoritative() {
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 200);
let sst_path = pick_first_sst_path(dir.path());
let report = verify_sst_file(&sst_path);
assert!(
report.is_ok(),
"intact SST must be clean: {:?}",
report.errors,
);
crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
.unwrap();
let report = verify_sst_file(&sst_path);
assert!(
report.errors.is_empty(),
"the surviving descriptor frames the data, so the walk must proceed \
under it: {:?}",
report.errors,
);
assert!(
!report.incomplete,
"the data blocks were walked, so the scan is complete",
);
assert!(
report.is_ok(),
"a descriptor-only forge on one mirror must not condemn a healthy table",
);
}
#[test]
fn verify_sst_file_both_mirrors_unrecognized_falls_back_to_off() -> crate::Result<()> {
use crate::table::Writer;
let dir = tempfile::tempdir()?;
let path = dir.path().join("t");
let mut writer =
Writer::new(path.clone(), 0, 0, std::sync::Arc::new(crate::fs::StdFs))?.use_ecc(None);
for i in 0u64..200 {
writer.write(crate::InternalValue::from_components(
format!("key-{i:05}").into_bytes(),
format!("value-{i:05}").into_bytes(),
i + 1,
crate::ValueType::Value,
))?;
}
assert!(writer.finish()?.is_some(), "the fixture is non-empty");
let report = verify_sst_file(&path);
assert!(
report.is_ok(),
"an intact parity-less table must verify clean: {:?}",
report.errors,
);
crate::test_forge::forge_mid_meta_value(&path, b"descriptor#page_ecc", &[9, 0, 0, 0])?;
crate::test_forge::forge_tail_meta_value(&path, b"descriptor#page_ecc", &[8, 0, 0, 0])?;
let report = verify_sst_file(&path);
assert!(
report.errors.is_empty(),
"the blocks are untouched: {:?}",
report.errors,
);
assert!(
!report
.warnings
.iter()
.any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
"the file frames without a trailer, which answers the question the \
descriptors no longer can: {:?}",
report.warnings,
);
assert!(
!report.incomplete,
"every section was walked, so nothing is left unverified — the whole \
point of trying `Off` rather than giving up",
);
assert!(
report.warnings.iter().any(|w| matches!(
w,
crate::verify::BlockVerifyWarning::EccDescriptorsUnreadable { .. }
)),
"the inferred layout must not hide the malformed descriptors: {:?}",
report.warnings,
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_same_length_scheme_is_flagged_but_not_refused() -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, EccParams, Header, expected_parity_len};
let real = EccParams::RS_4_2;
let impostor = EccParams::try_new(2, 1)?;
const DATA_LENGTH: u32 = 4096;
assert_eq!(
expected_parity_len(DATA_LENGTH, real),
expected_parity_len(DATA_LENGTH, impostor),
"the fixture must exercise a length collision, or it proves nothing",
);
let payload = discriminating_payload(DATA_LENGTH);
let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes the fixture");
let mut header = Header {
data_length: DATA_LENGTH,
uncompressed_length: DATA_LENGTH,
..Header::test_dummy(BlockType::Data)
};
header.bind_checksum(
Checksum::from_raw(crate::hash::hash128(&payload)),
crate::table::block::ChecksumAt::table(0, 0),
);
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
writer.start("data").unwrap();
writer.write_all(&header.encode_into_vec()).unwrap();
writer.write_all(&payload).unwrap();
writer.write_all(&parity).unwrap();
writer.finish().unwrap();
}
let dir = tempfile::tempdir()?;
let fs = StdFs;
let path = dir.path().join("rs42.sst");
{
let mut f = fs.open(
&path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
}
let mut probe = fs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let entry = toc
.section(b"data")
.expect("the fixture has a data section");
let (start, end) = (entry.pos(), entry.pos() + entry.len());
assert_eq!(
scheme_frames_region(probe.as_ref(), ScrubEcc::Scheme(real), start, end)?,
Some(true),
"the real scheme must frame its own block",
);
assert_eq!(
scheme_frames_region(probe.as_ref(), ScrubEcc::Scheme(impostor), start, end)?,
Some(true),
"the same-length impostor frames identically — framing cannot separate them",
);
let cap = block_data_length_cap(0);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(true),
"the real scheme sizes its own blocks",
);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(impostor), 0)?,
Some(true),
"the impostor sizes them identically, so the walk can still proceed — \
refusing it would cost the whole table for a parity question",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
"the real scheme reproduces the trailer, so nothing is suspect",
);
assert!(
codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
"the impostor reproduces no trailer — the signature of a mis-identified \
scheme, which is what the operator is told",
);
Ok(())
}
#[test]
fn verify_sst_file_unframeable_data_region_refuses_the_descriptor() {
use crate::fs::{Fs, FsOpenOptions, StdFs};
let dir = tempfile::tempdir().unwrap();
populate_tree(dir.path(), 200);
let sst_path = pick_first_sst_path(dir.path());
crate::test_forge::forge_tail_meta_value(&sst_path, b"descriptor#page_ecc", &[9, 0, 0, 0])
.unwrap();
let fs = StdFs;
let data_pos = {
let mut probe = fs
.open(&sst_path, &FsOpenOptions::new().read(true))
.unwrap();
let reader = crate::sfa::Reader::from_reader(&mut probe).unwrap();
reader
.toc()
.section(b"data")
.expect("the SST has a data section")
.pos()
};
{
let mut f = fs
.open(&sst_path, &FsOpenOptions::new().write(true))
.unwrap();
f.seek(SeekFrom::Start(data_pos)).unwrap();
f.write_all(&[0xFFu8; 16]).unwrap();
}
let report = verify_sst_file(&sst_path);
assert!(
report
.warnings
.iter()
.any(|w| matches!(w, crate::verify::BlockVerifyWarning::UnrecognizedEcc { .. })),
"a region that will not frame refuses the descriptor, whatever the \
other regions did: {:?}",
report.warnings,
);
assert!(
report.incomplete,
"the ECC-dependent sections were skipped, so the scan is incomplete",
);
assert!(
!report.is_ok(),
"the table must not verify clean: its data region holds a corrupt header",
);
}
#[cfg(feature = "page_ecc")]
#[test]
fn arbitrate_by_framing_propagates_an_unreadable_region() -> crate::Result<()> {
use crate::fs::{Fault, FaultFs, FaultOp, FaultRule, Fs, FsOpenOptions, StdFs};
use crate::io::ErrorKind;
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let bare = discriminating_payload(LEN);
let framed = discriminating_payload(LEN);
let framed_parity = crate::ecc::encode_parity(&framed, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("unreadable-region.sst");
write_block_archive(
&path,
&[
("data", vec![(bare, Vec::new())]),
("tli", vec![(framed, framed_parity)]),
],
)?;
let mut plain = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut plain)?;
let toc = sfa_reader.toc();
assert_eq!(
arbitrate_by_framing(plain.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(false),
"the parity-less data region cannot frame under RS(4,2) — the premise \
of this test",
);
let fault = FaultFs::new(StdFs);
let injector = fault.injector();
injector.arm(FaultRule::new(FaultOp::ReadAt, Fault::Error(ErrorKind::Other)).once());
let faulted = fault.open(&path, &FsOpenOptions::new().read(true))?;
let verdict = arbitrate_by_framing(faulted.as_ref(), toc, ScrubEcc::Scheme(real), 0);
injector.clear();
assert!(
verdict.is_err(),
"an unread region must surface as a read failure, not as a region that \
had nothing to say: got {verdict:?}",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
type SyntheticBlock = (Vec<u8>, Vec<u8>);
#[cfg(feature = "page_ecc")]
fn write_block_archive(
path: &std::path::Path,
sections: &[(&str, Vec<SyntheticBlock>)],
) -> crate::Result<()> {
use crate::coding::Encode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{BlockType, Header};
let mut archive_bytes: Vec<u8> = Vec::new();
{
let mut writer = crate::sfa::Writer::from_writer(std::io::Cursor::new(&mut archive_bytes));
for (name, blocks) in sections {
writer.start(*name).unwrap();
for (payload, parity) in blocks {
#[expect(
clippy::cast_possible_truncation,
reason = "test payloads are kilobytes"
)]
let mut header = Header {
data_length: payload.len() as u32,
uncompressed_length: payload.len() as u32,
..Header::test_dummy(BlockType::Data)
};
header.bind_checksum(
Checksum::from_raw(crate::hash::hash128(payload)),
crate::table::block::ChecksumAt::table(0, writer.get_ref().position()),
);
writer.write_all(&header.encode_into_vec()).unwrap();
writer.write_all(payload).unwrap();
writer.write_all(parity).unwrap();
}
}
writer.finish().unwrap();
}
let mut f = StdFs.open(
path,
&FsOpenOptions::new().write(true).create(true).truncate(true),
)?;
f.write_all(&archive_bytes)?;
Ok(())
}
#[cfg(feature = "page_ecc")]
fn discriminating_payload(len: u32) -> Vec<u8> {
(0..len)
.map(|i| u8::try_from(i % 251).expect("the modulus keeps every value below 256"))
.collect()
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_keeps_scanning_past_a_mismatch() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let impostor = EccParams::try_new(2, 1)?;
let degenerate = vec![0xABu8; LEN as usize];
let degenerate_parity = crate::ecc::encode_parity(°enerate, 4, 2).expect("RS(4,2) encodes");
assert_eq!(
degenerate_parity,
crate::ecc::encode_parity(°enerate, 2, 1).expect("XOR(2,1) encodes"),
"the fixture's second block must be one the two codecs agree on, or the \
scan has no later match to find",
);
let discriminating = discriminating_payload(LEN);
let discriminating_parity =
crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("two-blocks.sst");
write_block_archive(
&path,
&[(
"data",
vec![
(discriminating, discriminating_parity),
(degenerate, degenerate_parity),
],
)],
)?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let data = toc
.section(b"data")
.expect("the fixture has a data section");
let (start, end) = (data.pos(), data.pos() + data.len());
assert_eq!(
codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, cap),
CodecVerdict::Confirmed,
"the real codec reproduces both trailers",
);
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
start,
end,
cap
),
CodecVerdict::Confirmed,
"the second block's trailer IS reproduced, and a scan that stopped at \
the first mismatch would never see it",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
"one reproduced trailer refutes the report's claim, so a table whose \
mismatches are ordinary rot is not blamed on its scheme",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_scan_stops_early_reports_incomplete() -> crate::Result<()> {
use crate::coding::Decode;
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{EccParams, Header};
const LEN: u32 = 4096;
let impostor = EccParams::try_new(2, 1)?;
let discriminating = discriminating_payload(LEN);
let discriminating_parity =
crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
let degenerate = vec![0xABu8; LEN as usize];
let degenerate_parity = crate::ecc::encode_parity(°enerate, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("truncated-scan.sst");
write_block_archive(
&path,
&[(
"data",
vec![
(discriminating, discriminating_parity.clone()),
(degenerate, degenerate_parity),
],
)],
)?;
let (start, end, second_at) = {
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let reader = crate::sfa::Reader::from_reader(&mut probe)?;
let data = reader
.toc()
.section(b"data")
.expect("the fixture has a data section");
let head = crate::file::read_exact(probe.as_ref(), data.pos(), Header::MAX_LEN)?;
let header = Header::decode_from(&mut &head[..])?;
let frame = Header::header_len(header.block_type) as u64
+ u64::from(header.data_length)
+ discriminating_parity.len() as u64;
(data.pos(), data.pos() + data.len(), data.pos() + frame)
};
{
let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
f.seek(SeekFrom::Start(second_at))?;
f.write_all(&[0xFFu8; 16])?;
}
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
start,
end,
cap
),
CodecVerdict::Incomplete,
"the region was not inspected to its end, so it cannot say the scheme \
reproduces nothing",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
"and no diagnosis is reported off a truncated probe",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_empty_region_reports_no_evidence() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let payload = discriminating_payload(LEN);
let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
*rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;
let dir = tempfile::tempdir()?;
let path = dir.path().join("empty-region.sst");
write_block_archive(
&path,
&[("data", vec![(payload, rotted)]), ("filter", Vec::new())],
)?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let filter = toc
.section(b"filter")
.expect("the fixture has a filter section");
assert_eq!(filter.len(), 0, "the fixture's filter section is empty");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
filter.pos(),
filter.pos() + filter.len(),
cap,
),
CodecVerdict::NoEvidence,
"an empty region holds nothing to judge, which is not the same as a \
traversal that stopped",
);
assert!(
codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
"so it leaves the answer to the regions that do hold blocks, instead of \
silencing the whole table",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_is_silenced_by_a_region_it_could_not_finish() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let payload = discriminating_payload(LEN);
let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
*rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;
let other = discriminating_payload(LEN / 2);
let other_parity = crate::ecc::encode_parity(&other, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("mixed-regions.sst");
write_block_archive(
&path,
&[
("data", vec![(payload, rotted)]),
("tli", vec![(other, other_parity)]),
],
)?;
let tli_pos = {
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let reader = crate::sfa::Reader::from_reader(&mut probe)?;
reader
.toc()
.section(b"tli")
.expect("the fixture has a tli section")
.pos()
};
{
let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
f.seek(SeekFrom::Start(tli_pos))?;
f.write_all(&[0xFFu8; 16])?;
}
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let data = toc
.section(b"data")
.expect("the fixture has a data section");
let tli = toc.section(b"tli").expect("the fixture has a tli section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
data.pos(),
data.pos() + data.len(),
cap,
),
CodecVerdict::Rejected,
"the rotted trailer makes the fully-scanned region reject — one half of \
the premise",
);
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
tli.pos(),
tli.pos() + tli.len(),
cap,
),
CodecVerdict::Incomplete,
"and the other region was never finished — the other half",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
"an unfinished region leaves the table-wide claim unavailable, whatever \
a finished one found",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn rotted_trailer_keeps_the_descriptor_and_is_not_reported_suspect() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let payload = discriminating_payload(LEN);
let mut rotted = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
*rotted.first_mut().expect("the trailer is non-empty") ^= 0xFF;
let healthy = discriminating_payload(LEN / 2);
let healthy_parity = crate::ecc::encode_parity(&healthy, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("rotted-trailer.sst");
write_block_archive(
&path,
&[
("data", vec![(payload, rotted)]),
("tli", vec![(healthy, healthy_parity)]),
],
)?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let data = toc
.section(b"data")
.expect("the fixture has a data section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
data.pos(),
data.pos() + data.len(),
cap,
),
CodecVerdict::Rejected,
"the rotted trailer makes the data region reject on its own",
);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(true),
"damage does not change the trailer LENGTH, so the walk can still read \
the section and name the damaged block",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
"a region that reproduces the trailer rules the scheme out as the \
explanation, leaving the mismatch reported as what it is: damage",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_skips_a_block_whose_declared_length_exceeds_the_cap() -> crate::Result<()>
{
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let payload = discriminating_payload(LEN);
let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("capped.sst");
write_block_archive(&path, &[("data", vec![(payload, parity)])])?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let data = sfa_reader
.toc()
.section(b"data")
.expect("the fixture has a data section");
let (start, end) = (data.pos(), data.pos() + data.len());
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
start,
end,
cap_for_test()
),
CodecVerdict::Confirmed,
"under the real cap the block is read and confirms the codec",
);
assert_eq!(
codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, 16),
CodecVerdict::Incomplete,
"a declared length past the cap stops the traversal instead of sizing a \
read, and a stopped traversal answers nothing about the region",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
fn cap_for_test() -> u64 {
block_data_length_cap(0)
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_confirms_region_scans_past_a_run_of_mismatches() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
const MISMATCH_RUN: usize = 8;
let real = EccParams::RS_4_2;
let impostor = EccParams::try_new(2, 1)?;
let degenerate = vec![0xABu8; LEN as usize];
let degenerate_parity = crate::ecc::encode_parity(°enerate, 4, 2).expect("RS(4,2) encodes");
assert_eq!(
degenerate_parity,
crate::ecc::encode_parity(°enerate, 2, 1).expect("XOR(2,1) encodes"),
"the LAST block must be one the two codecs agree on, or there is no \
match at the far end to find",
);
let discriminating = discriminating_payload(LEN);
let discriminating_parity =
crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
let mut blocks: Vec<SyntheticBlock> = (0..MISMATCH_RUN)
.map(|_| (discriminating.clone(), discriminating_parity.clone()))
.collect();
blocks.push((degenerate, degenerate_parity));
let dir = tempfile::tempdir()?;
let path = dir.path().join("nine-blocks.sst");
write_block_archive(&path, &[("data", blocks)])?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let data = toc
.section(b"data")
.expect("the fixture has a data section");
let (start, end) = (data.pos(), data.pos() + data.len());
assert_eq!(
codec_confirms_region((probe.as_ref(), 0), ScrubEcc::Scheme(real), start, end, cap),
CodecVerdict::Confirmed,
"the real codec reproduces every trailer",
);
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
start,
end,
cap
),
CodecVerdict::Confirmed,
"the ninth block's trailer IS reproduced, and a scan that gave up over \
the eight before it would never reach the evidence",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn arbitrate_by_framing_rejects_a_split_verdict() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{EccParams, Header};
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let payload = discriminating_payload(LEN);
let parity = crate::ecc::encode_parity(&payload, 4, 2).expect("RS(4,2) encodes");
let other = discriminating_payload(LEN / 2);
let short_parity = vec![0u8; 8];
let dir = tempfile::tempdir()?;
let path = dir.path().join("split.sst");
write_block_archive(
&path,
&[
("data", vec![(payload, parity)]),
("tli", vec![(other, short_parity)]),
],
)?;
let data_pos = {
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let reader = crate::sfa::Reader::from_reader(&mut probe)?;
reader
.toc()
.section(b"data")
.expect("the fixture has a data section")
.pos()
};
{
let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
f.write_all(&[0xFF])?;
}
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let data = toc
.section(b"data")
.expect("the fixture has a data section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
data.pos(),
data.pos() + data.len(),
cap,
),
CodecVerdict::NoEvidence,
"no clean block, so the codec has nothing to say here either",
);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(false),
"one region framed and the other did not — accepting here walks a \
region with the wrong trailer length and reports corruption that is \
not there",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_is_silenced_by_any_agreement() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::EccParams;
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let impostor = EccParams::try_new(2, 1)?;
let discriminating = discriminating_payload(LEN);
let discriminating_parity =
crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
let degenerate = vec![0xABu8; LEN as usize];
let degenerate_parity = crate::ecc::encode_parity(°enerate, 4, 2).expect("RS(4,2) encodes");
assert!(
degenerate_parity.iter().all(|b| *b == 0),
"the degenerate region's trailer must be all-zero, which is what makes \
its agreement uninformative",
);
let dir = tempfile::tempdir()?;
let path = dir.path().join("degenerate-vs-healthy.sst");
write_block_archive(
&path,
&[
("data", vec![(discriminating, discriminating_parity)]),
("tli", vec![(degenerate, degenerate_parity)]),
],
)?;
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let tli = toc.section(b"tli").expect("the fixture has a tli section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
tli.pos(),
tli.pos() + tli.len(),
cap,
),
CodecVerdict::Confirmed,
"the impostor agrees with the all-zero region — the premise of this test",
);
assert!(
!codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
"one region disagreed and another agreed, so the report's claim does \
not hold and it stays silent",
);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(true),
"the real scheme sizes both regions",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_consults_the_tli_tail_mirror() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{EccParams, Header};
const LEN: u32 = 4096;
let real = EccParams::RS_4_2;
let data_payload = discriminating_payload(LEN);
let data_parity = crate::ecc::encode_parity(&data_payload, 4, 2).expect("RS(4,2) encodes");
let head = discriminating_payload(LEN / 2);
let head_parity = crate::ecc::encode_parity(&head, 4, 2).expect("RS(4,2) encodes");
let tail = discriminating_payload(LEN / 2);
let mut tail_parity = crate::ecc::encode_parity(&tail, 4, 2).expect("RS(4,2) encodes");
*tail_parity.first_mut().expect("the trailer is non-empty") ^= 0xFF;
let dir = tempfile::tempdir()?;
let path = dir.path().join("tail-mirror.sst");
write_block_archive(
&path,
&[
("data", vec![(data_payload, data_parity)]),
("tli", vec![(head, head_parity)]),
("tli_tail", vec![(tail, tail_parity)]),
],
)?;
let (data_pos, tli_pos) = {
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = reader.toc();
(
toc.section(b"data")
.expect("the fixture has a data section")
.pos(),
toc.section(b"tli")
.expect("the fixture has a tli section")
.pos(),
)
};
{
let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
f.write_all(&[0xFF])?;
f.seek(SeekFrom::Start(tli_pos + Header::MIN_LEN as u64))?;
f.write_all(&[0xFF])?;
}
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let tli = toc.section(b"tli").expect("the fixture has a tli section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(real),
tli.pos(),
tli.pos() + tli.len(),
cap,
),
CodecVerdict::NoEvidence,
"the head offers nothing to judge on — the premise of this test",
);
assert_eq!(
arbitrate_by_framing(probe.as_ref(), toc, ScrubEcc::Scheme(real), 0)?,
Some(true),
"every region frames, so the descriptor stands whatever the trailers say",
);
assert!(
codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(real), 0, cap),
"only the tail mirror had anything to say, so a region set omitting it \
would report nothing at all",
);
Ok(())
}
#[cfg(feature = "page_ecc")]
#[test]
fn codec_disagrees_everywhere_consults_sections_outside_the_mirrors() -> crate::Result<()> {
use crate::fs::{Fs, FsOpenOptions, StdFs};
use crate::table::block::{EccParams, Header};
const LEN: u32 = 4096;
let impostor = EccParams::try_new(2, 1)?;
let silent = discriminating_payload(LEN);
let silent_parity = crate::ecc::encode_parity(&silent, 4, 2).expect("RS(4,2) encodes");
let discriminating = discriminating_payload(LEN);
let discriminating_parity =
crate::ecc::encode_parity(&discriminating, 4, 2).expect("RS(4,2) encodes");
let dir = tempfile::tempdir()?;
let path = dir.path().join("outside-mirrors.sst");
write_block_archive(
&path,
&[
("data", vec![(silent, silent_parity)]),
(
"range_tombstones",
vec![(discriminating, discriminating_parity)],
),
],
)?;
let data_pos = {
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let reader = crate::sfa::Reader::from_reader(&mut probe)?;
reader
.toc()
.section(b"data")
.expect("the fixture has a data section")
.pos()
};
{
let mut f = StdFs.open(&path, &FsOpenOptions::new().write(true))?;
f.seek(SeekFrom::Start(data_pos + Header::MIN_LEN as u64))?;
f.write_all(&[0xFF])?;
}
let mut probe = StdFs.open(&path, &FsOpenOptions::new().read(true))?;
let sfa_reader = crate::sfa::Reader::from_reader(&mut probe)?;
let toc = sfa_reader.toc();
let cap = block_data_length_cap(0);
let rt = toc
.section(b"range_tombstones")
.expect("the fixture has a range_tombstones section");
let data = toc
.section(b"data")
.expect("the fixture has a data section");
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
rt.pos(),
rt.pos() + rt.len(),
cap,
),
CodecVerdict::Rejected,
"the section outside the mirrors is the only one that disagrees — the \
premise of this test",
);
assert_eq!(
codec_confirms_region(
(probe.as_ref(), 0),
ScrubEcc::Scheme(impostor),
data.pos(),
data.pos() + data.len(),
cap,
),
CodecVerdict::NoEvidence,
"and the data region says nothing, so it cannot silence the report",
);
assert!(
codec_disagrees_everywhere((probe.as_ref(), 0), toc, ScrubEcc::Scheme(impostor), 0, cap),
"a section the descriptor sizes is consulted wherever it sits",
);
Ok(())
}
#[cfg(feature = "columnar")]
#[test]
fn verify_sst_file_walks_a_healthy_columnar_sst_clean() -> crate::Result<()> {
use crate::table::Writer;
let dir = tempfile::tempdir()?;
let sst = dir.path().join("table");
let mut writer = Writer::new(sst.clone(), 0, 0, alloc::sync::Arc::new(crate::fs::StdFs))?
.use_columnar(true)
.use_zone_map(true)
.use_row_group_size(256);
for i in 0u32..200 {
writer.write(crate::InternalValue::from_components(
format!("key{i:05}").into_bytes(),
format!("val{i:05}").into_bytes(),
1,
crate::ValueType::Value,
))?;
}
assert!(writer.finish()?.is_some(), "the SST is non-empty");
let report = verify_sst_file(&sst);
assert!(
report.is_ok(),
"a healthy columnar SST must verify clean: {:?}",
report.errors,
);
let groups = {
let checksum = crate::Checksum::from_raw(crate::repair::compute_table_checksum(
&crate::fs::StdFs,
&sst,
)?);
let mut params = crate::table::RecoverParams::new(
sst,
checksum,
0,
alloc::sync::Arc::new(crate::fs::StdFs),
crate::comparator::default_comparator(),
alloc::sync::Arc::new(crate::cache::Cache::with_capacity_bytes(1 << 20)),
);
params.descriptor_table = Some(alloc::sync::Arc::new(
crate::descriptor_table::DescriptorTable::new(4),
));
crate::table::Table::recover(params)?
.data_block_handles()
.filter_map(Result::ok)
.count()
};
assert!(groups > 1, "the fixture must span several row groups");
assert!(
report.blocks_scanned >= groups * 5,
"every directory and page must be walked: {} blocks for {groups} groups",
report.blocks_scanned,
);
Ok(())
}
fn populate_rated_fixture(dir: &std::path::Path) {
let cfg = Config::new(
dir,
SequenceNumberCounter::default(),
SequenceNumberCounter::default(),
)
.data_block_compression_policy(CompressionPolicy::all(CompressionType::None));
let tree = cfg.open().unwrap();
let mut seqno = 1u64;
for b in 0..4u32 {
for i in 0..300u32 {
let key = format!("b{b:03}k{i:08}");
let value: Vec<u8> = (0..1024u32)
.map(|j| {
let [low, ..] = (j.wrapping_mul(2_654_435_761) ^ i ^ (b << 16)).to_le_bytes();
low
})
.collect();
tree.insert(key.as_bytes(), value, seqno);
seqno += 1;
}
tree.flush_active_memtable(seqno).unwrap();
seqno += 1;
}
}
fn table_bytes(tree: &impl AbstractTree) -> u64 {
tree.current_version()
.iter_tables()
.map(|t| std::fs::metadata(&*t.path).unwrap().len())
.sum()
}
fn assert_rated_scan(parallelism: usize) {
let dir = tempfile::tempdir().unwrap();
populate_rated_fixture(dir.path());
let tree = reopen_tree(dir.path());
let bytes = table_bytes(&tree);
let rate = 400 * 1024u64;
assert!(
bytes > 2 * rate,
"the fixture must outlast the burst: {bytes} bytes"
);
let opts = VerifyOptions::default()
.parallelism(parallelism)
.max_bytes_per_sec(rate);
let start = std::time::Instant::now();
let report = tree.verify_checksum_with(&opts);
let elapsed = start.elapsed();
assert!(report.is_ok(), "a rated scan verifies the same: {report:?}");
assert_eq!(report.sst_files_scanned, 4);
let owed = |bytes: u64| std::time::Duration::from_millis((bytes - rate) * 1000 / rate);
let floor = owed(bytes - 4 * 1024) * 19 / 20;
let ceiling = owed(bytes) * 3 / 2 + std::time::Duration::from_secs(1);
assert!(
elapsed >= floor,
"{parallelism} worker(s): {bytes} bytes at {rate} B/s took {elapsed:?}, \
under the {floor:?} the limiter owes",
);
assert!(
elapsed <= ceiling,
"{parallelism} worker(s): {bytes} bytes at {rate} B/s took {elapsed:?}, \
past {ceiling:?}",
);
}
#[test]
fn verify_at_a_byte_rate_takes_its_bytes_over_the_rate_with_one_worker() {
assert_rated_scan(1);
}
#[test]
fn verify_at_a_byte_rate_takes_its_bytes_over_the_rate_with_four_workers() {
assert_rated_scan(4);
}
#[test]
fn verify_without_a_byte_rate_reads_at_full_speed() {
let dir = tempfile::tempdir().unwrap();
populate_rated_fixture(dir.path());
let tree = reopen_tree(dir.path());
for opts in [
VerifyOptions::default(),
VerifyOptions::default().max_bytes_per_sec(0),
] {
let start = std::time::Instant::now();
let report = tree.verify_checksum_with(&opts);
let elapsed = start.elapsed();
assert!(report.is_ok(), "{report:?}");
assert!(
elapsed < std::time::Duration::from_millis(900),
"an unrated scan of {} bytes took {elapsed:?}",
table_bytes(&tree),
);
}
}
#[test]
fn verify_at_a_byte_rate_still_reports_a_corrupt_block() {
use crate::table::block::Header;
let dir = tempfile::tempdir().unwrap();
populate_rated_fixture(dir.path());
let sst_path = pick_first_sst_path(dir.path());
let flip_offset = Header::MIN_LEN as u64 + 8;
{
let mut f = std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(&sst_path)
.unwrap();
f.seek(SeekFrom::Start(flip_offset)).unwrap();
let mut byte = [0u8; 1];
f.read_exact(&mut byte).unwrap();
byte[0] ^= 0xFF;
f.seek(SeekFrom::Start(flip_offset)).unwrap();
f.write_all(&byte).unwrap();
f.sync_all().unwrap();
}
let tree = reopen_tree(dir.path());
let rate = table_bytes(&tree);
let report = tree.verify_checksum_with(&VerifyOptions::default().max_bytes_per_sec(rate));
assert!(
report
.errors
.iter()
.any(|e| matches!(e, BlockVerifyError::DataCorrupted { .. })),
"the flipped payload byte is reported under a limit: {report:?}",
);
}