use crate::{Blob, Buf, BufMut, Error, IoBuf, ReadOptions};
#[cfg(any(test, feature = "test-utils"))]
use crate::{Storage, WriteOptions};
use commonware_codec::{EncodeFixed, FixedSize, Read as CodecRead, ReadExt, Write};
use commonware_cryptography::{Crc32, crc32};
use std::num::NonZeroU16;
mod cache;
mod read;
mod sealed;
mod view;
mod writer;
pub use cache::CacheRef;
pub use read::Replay;
pub use sealed::Sealed;
use tracing::{debug, error};
pub use writer::Writer;
pub const CHECKSUM_SIZE: u64 = Checksum::SIZE as u64;
pub(crate) const STORAGE_PAGE_SIZE: u64 = 4096;
const _: () = assert!(
crate::DEFAULT_BLOB_LAYOUT
.data_offset()
.is_multiple_of(STORAGE_PAGE_SIZE)
);
const CHECKSUM_SLOT_LEN_SIZE: usize = u16::SIZE;
const CHECKSUM_SLOT_SIZE: usize = CHECKSUM_SLOT_LEN_SIZE + crc32::Digest::SIZE;
pub const fn page_size(physical_page_size: u32) -> NonZeroU16 {
assert!(
physical_page_size.is_power_of_two(),
"physical page size must be a power of two"
);
assert!(
physical_page_size as u64 > CHECKSUM_SIZE,
"physical page size must exceed the CRC record size"
);
let logical = physical_page_size as u64 - CHECKSUM_SIZE;
assert!(
logical <= u16::MAX as u64,
"logical page size must fit in a u16"
);
match NonZeroU16::new(logical as u16) {
Some(size) => size,
None => unreachable!(),
}
}
#[cfg(test)]
pub(crate) fn validate_page_for_tests(page: &[u8]) -> bool {
Checksum::validate_page(page).is_some()
}
#[cfg(any(test, feature = "test-utils"))]
pub fn page_len(page: &[u8], logical_page_size: usize) -> Option<usize> {
let footer = page.get(logical_page_size..)?;
if footer.len() != CHECKSUM_SIZE as usize {
return None;
}
let slots = [
(
u16::from_be_bytes(footer[0..2].try_into().unwrap()) as usize,
u32::from_be_bytes(footer[2..6].try_into().unwrap()),
),
(
u16::from_be_bytes(footer[6..8].try_into().unwrap()) as usize,
u32::from_be_bytes(footer[8..12].try_into().unwrap()),
),
];
let authoritative = usize::from(slots[1].0 > slots[0].0);
for slot in [authoritative, authoritative ^ 1] {
let (len, checksum) = slots[slot];
if len > 0 && len <= logical_page_size && Crc32::checksum(&page[..len]) == checksum {
return Some(len);
}
}
None
}
#[cfg(any(test, feature = "test-utils"))]
pub async fn corrupt_page(
storage: &impl Storage,
partition: &str,
name: &[u8],
page: u64,
logical_page_size: u64,
) {
let physical_page_size = logical_page_size + CHECKSUM_SIZE;
let offset = page * physical_page_size;
let (blob, size) = storage.open(partition, name).await.unwrap();
assert!(
offset
.checked_add(physical_page_size * 2)
.is_some_and(|end| end <= size),
"corruption target must be an interior page"
);
let byte = blob
.read_at(offset, 1, ReadOptions::default())
.await
.unwrap()
.coalesce();
blob.write_at(
offset,
vec![byte.as_ref()[0] ^ 0xFF],
WriteOptions::default(),
)
.await
.unwrap();
blob.sync().await.unwrap();
}
fn validate_read_ranges(
buf_len: usize,
ranges: impl Iterator<Item = (u64, usize)>,
size: u64,
) -> Result<(), Error> {
let mut expected_len = 0usize;
let mut previous_end = None;
for (offset, len) in ranges {
expected_len = expected_len
.checked_add(len)
.expect("buf must hold one slot per range totaling its length");
let end = offset
.checked_add(len as u64)
.ok_or(Error::OffsetOverflow)?;
if let Some(previous_end) = previous_end {
assert!(
offset >= previous_end,
"ranges must be sorted and non-overlapping"
);
}
if end > size {
return Err(Error::BlobInsufficientLength);
}
previous_end = Some(end);
}
assert_eq!(
buf_len, expected_len,
"buf must hold one slot per range totaling its length"
);
Ok(())
}
fn split_read_ranges<'a>(
mut buf: &'a mut [u8],
ranges: impl ExactSizeIterator<Item = (u64, usize)>,
tail_offset: u64,
tail: &[u8],
) -> Vec<(&'a mut [u8], u64)> {
let mut cache_ranges = Vec::with_capacity(ranges.len());
for (offset, len) in ranges {
let (slot, rest) = buf.split_at_mut(len);
buf = rest;
if len == 0 {
continue;
}
let end = offset + len as u64;
if end <= tail_offset {
cache_ranges.push((slot, offset));
} else if offset >= tail_offset {
let src = (offset - tail_offset) as usize;
slot.copy_from_slice(&tail[src..src + len]);
} else {
let prefix_len = (tail_offset - offset) as usize;
let (prefix, suffix) = slot.split_at_mut(prefix_len);
suffix.copy_from_slice(&tail[..len - prefix_len]);
cache_ranges.push((prefix, offset));
}
}
cache_ranges
}
async fn get_page_from_blob(
blob: &impl Blob,
page_num: u64,
page_size: u64,
read_options: ReadOptions,
) -> Result<IoBuf, Error> {
let (page, _) =
get_page_with_checksum_from_blob(blob, page_num, page_size, read_options).await?;
Ok(page)
}
async fn get_page_with_checksum_from_blob(
blob: &impl Blob,
page_num: u64,
page_size: u64,
read_options: ReadOptions,
) -> Result<(IoBuf, ActiveChecksum), Error> {
let physical_page_size = page_size
.checked_add(CHECKSUM_SIZE)
.ok_or(Error::OffsetOverflow)?;
let physical_page_start = page_num
.checked_mul(physical_page_size)
.ok_or(Error::OffsetOverflow)?;
let page = blob
.read_at(
physical_page_start,
physical_page_size as usize,
read_options,
)
.await?
.coalesce();
let Some(checksum) = Checksum::validate_page(page.as_ref()) else {
return Err(Error::InvalidChecksum);
};
Ok((page.freeze().slice(..checksum.len as usize), checksum))
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum Slot {
First,
Second,
}
impl Slot {
const fn offset(self) -> usize {
match self {
Self::First => 0,
Self::Second => CHECKSUM_SLOT_SIZE,
}
}
const fn other(self) -> Self {
match self {
Self::First => Self::Second,
Self::Second => Self::First,
}
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
struct ActiveChecksum {
slot: Slot,
len: u16,
crc: u32,
}
impl ActiveChecksum {
const fn new(slot: Slot, len: u16, crc: u32) -> Self {
Self { slot, len, crc }
}
}
struct Checksum {
len1: u16,
crc1: u32,
len2: u16,
crc2: u32,
}
impl Checksum {
const fn new(len: u16, crc: u32) -> Self {
Self {
len1: len,
crc1: crc,
len2: 0,
crc2: 0,
}
}
const fn authoritative(&self) -> Slot {
if self.len1 >= self.len2 {
Slot::First
} else {
Slot::Second
}
}
fn validate_page(buf: &[u8]) -> Option<ActiveChecksum> {
let physical_page_size = buf.len() as u64;
if physical_page_size < CHECKSUM_SIZE {
error!(
physical_page_size,
required = CHECKSUM_SIZE,
"read page smaller than CRC record"
);
return None;
}
let crc_start_idx = (physical_page_size - CHECKSUM_SIZE) as usize;
let mut crc_bytes = &buf[crc_start_idx..];
let crc_record = Self::read(&mut crc_bytes).expect("CRC record read should not fail");
let authoritative = crc_record.authoritative();
if let Some(checksum) = crc_record.validate_slot(authoritative, buf, crc_start_idx) {
return Some(checksum);
}
debug!("Invalid authoritative CRC, using fallback CRC");
let checksum = crc_record.validate_slot(authoritative.other(), buf, crc_start_idx);
if checksum.is_none() {
debug!("Invalid fallback CRC");
}
checksum
}
fn validate_slot(
&self,
slot: Slot,
buf: &[u8],
crc_start_idx: usize,
) -> Option<ActiveChecksum> {
let (len, crc) = self.get_slot(slot);
let len_usize = len as usize;
if len_usize == 0 {
return None;
}
if len_usize > crc_start_idx {
return None;
}
if Crc32::checksum(&buf[..len_usize]) != crc {
return None;
}
Some(ActiveChecksum::new(slot, len, crc))
}
const fn get_slot(&self, slot: Slot) -> (u16, u32) {
match slot {
Slot::First => (self.len1, self.crc1),
Slot::Second => (self.len2, self.crc2),
}
}
fn to_bytes(&self) -> [u8; CHECKSUM_SIZE as usize] {
self.encode_fixed()
}
fn slot_bytes(len: u16, crc: u32) -> [u8; CHECKSUM_SLOT_SIZE] {
let mut bytes = [0; CHECKSUM_SLOT_SIZE];
let mut buf = bytes.as_mut_slice();
len.write(&mut buf);
crc.write(&mut buf);
bytes
}
fn slot_len_bytes(len: u16) -> [u8; CHECKSUM_SLOT_LEN_SIZE] {
let mut bytes = [0; CHECKSUM_SLOT_LEN_SIZE];
let mut buf = bytes.as_mut_slice();
len.write(&mut buf);
bytes
}
}
impl Write for Checksum {
fn write(&self, buf: &mut impl BufMut) {
self.len1.write(buf);
self.crc1.write(buf);
self.len2.write(buf);
self.crc2.write(buf);
}
}
impl CodecRead for Checksum {
type Cfg = ();
fn read_cfg(buf: &mut impl Buf, _: &Self::Cfg) -> Result<Self, commonware_codec::Error> {
Ok(Self {
len1: u16::read(buf)?,
crc1: u32::read(buf)?,
len2: u16::read(buf)?,
crc2: u32::read(buf)?,
})
}
}
impl FixedSize for Checksum {
const SIZE: usize = 2 * u16::SIZE + 2 * crc32::Digest::SIZE;
}
#[cfg(feature = "arbitrary")]
impl arbitrary::Arbitrary<'_> for Checksum {
fn arbitrary(u: &mut arbitrary::Unstructured<'_>) -> arbitrary::Result<Self> {
Ok(Self {
len1: u.arbitrary()?,
crc1: u.arbitrary()?,
len2: u.arbitrary()?,
crc2: u.arbitrary()?,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use rstest::rstest;
#[test]
#[should_panic(expected = "corruption target must be an interior page")]
fn test_corrupt_page_rejects_short_blob() {
use crate::Runner as _;
crate::deterministic::Runner::default().start(|context| async move {
corrupt_page(&context, "short-blob", b"blob", 0, 64).await;
});
}
enum ValidationExpectation {
Ok,
OffsetOverflow,
BlobInsufficientLength,
}
#[rstest]
#[case::ok(12, vec![(0, 4), (4, 8)], 16, ValidationExpectation::Ok)]
#[case::empty_ranges_are_a_noop(0, vec![], 0, ValidationExpectation::Ok)]
#[case::zero_length_range(4, vec![(0, 0), (0, 4)], 16, ValidationExpectation::Ok)]
#[case::offset_overflow(4, vec![(u64::MAX, 4)], 16, ValidationExpectation::OffsetOverflow)]
#[case::insufficient_length(4, vec![(14, 4)], 16, ValidationExpectation::BlobInsufficientLength)]
#[case::range_may_end_exactly_at_logical_size(4, vec![(12, 4)], 16, ValidationExpectation::Ok)]
fn test_validate_read_ranges(
#[case] buf_len: usize,
#[case] ranges: Vec<(u64, usize)>,
#[case] size: u64,
#[case] expected: ValidationExpectation,
) {
let result = validate_read_ranges(buf_len, ranges.iter().copied(), size);
match expected {
ValidationExpectation::Ok => assert!(result.is_ok()),
ValidationExpectation::OffsetOverflow => {
assert!(matches!(result, Err(Error::OffsetOverflow)))
}
ValidationExpectation::BlobInsufficientLength => {
assert!(matches!(result, Err(Error::BlobInsufficientLength)))
}
}
}
#[test]
#[should_panic(expected = "buf must hold one slot per range totaling its length")]
fn test_validate_read_ranges_rejects_buffer_len_mismatch() {
let _ = validate_read_ranges(7, [(0, 4), (4, 4)].into_iter(), 16);
}
#[test]
#[should_panic(expected = "ranges must be sorted and non-overlapping")]
fn test_validate_read_ranges_rejects_overlapping_ranges() {
let _ = validate_read_ranges(8, [(0, 4), (2, 4)].into_iter(), 16);
}
#[test]
#[should_panic(expected = "ranges must be sorted and non-overlapping")]
fn test_validate_read_ranges_rejects_unsorted_ranges() {
let _ = validate_read_ranges(8, [(8, 4), (4, 4)].into_iter(), 16);
}
#[test]
#[should_panic(expected = "buf must hold one slot per range totaling its length")]
fn test_validate_read_ranges_rejects_length_overflow() {
let _ = validate_read_ranges(
usize::MAX,
[(0, usize::MAX), (u64::MAX, 1)].into_iter(),
u64::MAX,
);
}
#[test]
fn test_crc_record_encode_read_roundtrip() {
let record = Checksum {
len1: 0x1234,
crc1: 0xAABBCCDD,
len2: 0x5678,
crc2: 0x11223344,
};
let bytes = record.to_bytes();
let restored = Checksum::read(&mut &bytes[..]).unwrap();
assert_eq!(restored.len1, 0x1234);
assert_eq!(restored.crc1, 0xAABBCCDD);
assert_eq!(restored.len2, 0x5678);
assert_eq!(restored.crc2, 0x11223344);
}
#[test]
fn test_crc_record_encoding() {
let record = Checksum {
len1: 0x0102,
crc1: 0x03040506,
len2: 0x0708,
crc2: 0x090A0B0C,
};
let bytes = record.to_bytes();
assert_eq!(
bytes,
[
0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08, 0x09, 0x0A, 0x0B, 0x0C
]
);
}
#[test]
fn test_crc_record_authoritative_len1_larger() {
let record = Checksum {
len1: 200,
crc1: 0xAAAAAAAA,
len2: 100,
crc2: 0xBBBBBBBB,
};
assert_eq!(record.authoritative(), Slot::First);
}
#[test]
fn test_crc_record_authoritative_len2_larger() {
let record = Checksum {
len1: 100,
crc1: 0xAAAAAAAA,
len2: 200,
crc2: 0xBBBBBBBB,
};
assert_eq!(record.authoritative(), Slot::Second);
}
#[test]
fn test_crc_record_authoritative_equal_lengths() {
let record = Checksum {
len1: 100,
crc1: 0xAAAAAAAA,
len2: 100,
crc2: 0xBBBBBBBB,
};
assert_eq!(record.authoritative(), Slot::First);
}
#[test]
fn test_validate_page_valid() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"hello world";
page[..data.len()].copy_from_slice(data);
let crc = Crc32::checksum(&page[..data.len()]);
let record = Checksum::new(data.len() as u16, crc);
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
let validated = Checksum::validate_page(&page);
assert!(validated.is_some());
assert_eq!(validated.unwrap().len as usize, data.len());
}
#[test]
fn test_validate_page_invalid_crc() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"hello world";
page[..data.len()].copy_from_slice(data);
let wrong_crc = 0xBADBADBA;
let record = Checksum::new(data.len() as u16, wrong_crc);
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
let validated = Checksum::validate_page(&page);
assert!(validated.is_none());
}
#[test]
fn test_validate_page_corrupted_data() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"hello world";
page[..data.len()].copy_from_slice(data);
let crc = Crc32::checksum(&page[..data.len()]);
let record = Checksum::new(data.len() as u16, crc);
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
page[0] = 0xFF;
let validated = Checksum::validate_page(&page);
assert!(validated.is_none());
}
#[test]
fn test_validate_page_uses_larger_len() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"hello world, this is longer";
page[..data.len()].copy_from_slice(data);
let crc = Crc32::checksum(&page[..data.len()]);
let record = Checksum {
len1: 5,
crc1: 0xDEADBEEF, len2: data.len() as u16,
crc2: crc,
};
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
let validated = Checksum::validate_page(&page);
assert!(validated.is_some());
assert_eq!(validated.unwrap().len as usize, data.len());
}
#[test]
fn test_validate_page_uses_fallback() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"fallback data";
page[..data.len()].copy_from_slice(data);
let valid_crc = Crc32::checksum(&page[..data.len()]);
let valid_len = data.len() as u16;
let record = Checksum {
len1: valid_len + 10, crc1: 0xBAD1DEA, len2: valid_len, crc2: valid_crc, };
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
let validated = Checksum::validate_page(&page);
assert!(validated.is_some(), "Should have validated using fallback");
let validated = validated.unwrap();
assert_eq!(
validated,
ActiveChecksum::new(Slot::Second, valid_len, valid_crc)
);
}
#[test]
fn test_validate_page_no_fallback_available() {
let page_size = 64usize;
let physical_page_size = page_size + Checksum::SIZE;
let mut page = vec![0u8; physical_page_size];
let data = b"some data";
page[..data.len()].copy_from_slice(data);
let record = Checksum {
len1: data.len() as u16,
crc1: 0xBAD1DEA, len2: 0, crc2: 0,
};
let crc_start = physical_page_size - Checksum::SIZE;
page[crc_start..].copy_from_slice(&record.to_bytes());
let validated = Checksum::validate_page(&page);
assert!(
validated.is_none(),
"Should fail when primary is invalid and fallback has len=0"
);
}
#[cfg(feature = "arbitrary")]
mod conformance {
use super::*;
use commonware_codec::conformance::CodecConformance;
commonware_conformance::conformance_tests! {
CodecConformance<Checksum>,
}
}
}