use crate::config::{SQLITE_PAGE_SIZE, STABLE_PAGE_SIZE, SUPERBLOCK_SIZE};
use crate::sqlite_vfs::overlay::{self, Overlay};
#[cfg(test)]
use crate::stable::memory::ContextId;
use crate::stable::memory::{self, StableMemoryError};
use crate::stable::meta::{
fnv1a64, Superblock, ZeroExtent, CURRENT_LAYOUT_VERSION, FLAG_CHECKSUM_REFRESHING,
FLAG_CHECKSUM_STALE, FLAG_IMPORTING, MAX_ZERO_EXTENTS,
};
#[cfg(test)]
use std::cell::RefCell;
#[cfg(test)]
use std::collections::BTreeMap;
const CHECKSUM_CHUNK_LEN: u64 = 16 * 1024;
#[cfg(test)]
const FAR_PAGE_NO: u64 = 257;
const FILE_PAGE_OFFSET_CACHE_CAPACITY: usize = 64;
const FILE_PAGE_DATA_CACHE_CAPACITY: usize = 8;
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ChecksumRefresh {
pub complete: bool,
pub checksum: u64,
pub scanned_bytes: u64,
pub db_size: u64,
}
#[allow(dead_code)]
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StorageStats {
pub layout_version: u64,
pub page_count: u64,
pub page_table_bytes: u64,
pub zero_extent_count: u64,
pub active_bytes: u64,
pub allocated_bytes: u64,
pub orphan_bytes_estimate: u64,
pub orphan_ratio_basis_points: u64,
pub compact_recommended: bool,
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(crate) enum StableBlobFailpoint {
OverlayWrite,
OverlayTruncate,
CommitCapacity,
CommitChunkWrite,
CommitSuperblockStore,
}
thread_local! {
#[cfg(test)]
static FAILPOINTS: RefCell<BTreeMap<ContextId, StableBlobFailpoint>> = const { RefCell::new(BTreeMap::new()) };
}
#[derive(Debug)]
pub(crate) struct PageOffsetCache {
entries: Vec<(u64, u64)>,
pages: Vec<(u64, Vec<u8>)>,
}
impl PageOffsetCache {
pub(crate) fn new() -> Self {
Self {
entries: Vec::with_capacity(FILE_PAGE_OFFSET_CACHE_CAPACITY),
pages: Vec::new(),
}
}
fn get(&self, page_no: u64) -> Option<u64> {
match self.entries.as_slice() {
[] => None,
[(cached_page, physical)] => (*cached_page == page_no).then_some(*physical),
entries => {
for (cached_page, physical) in entries {
if *cached_page == page_no {
return Some(*physical);
}
}
None
}
}
}
fn insert(&mut self, page_no: u64, physical: u64) {
if self.entries.len() == FILE_PAGE_OFFSET_CACHE_CAPACITY {
self.entries.remove(0);
}
self.entries.push((page_no, physical));
}
#[inline(always)]
fn copy_page_slice(&self, page_no: u64, in_page: usize, dst: &mut [u8]) -> bool {
if self.pages.is_empty() {
return false;
}
if self.pages.len() == 1 {
let (cached_page, page) = &self.pages[0];
if *cached_page == page_no {
let end = in_page + dst.len();
dst.copy_from_slice(&page[in_page..end]);
return true;
}
return false;
}
for (cached_page, page) in &self.pages {
if *cached_page == page_no {
let end = in_page + dst.len();
dst.copy_from_slice(&page[in_page..end]);
return true;
}
}
false
}
fn insert_page(&mut self, page_no: u64, page: Vec<u8>) {
if self.pages.len() == FILE_PAGE_DATA_CACHE_CAPACITY {
self.pages.remove(0);
}
self.pages.push((page_no, page));
}
}
#[cfg(test)]
pub(crate) fn set_failpoint(failpoint: StableBlobFailpoint) {
if let Ok(context) = memory::active_context_id() {
FAILPOINTS.with(|slot| {
slot.borrow_mut().insert(context, failpoint);
});
}
}
#[cfg(test)]
pub(crate) fn clear_failpoint() {
FAILPOINTS.with(|slot| slot.borrow_mut().clear());
}
pub(crate) fn ensure_current_layout() -> Result<(), StableMemoryError> {
let block = Superblock::load()?;
if block.layout_version == CURRENT_LAYOUT_VERSION {
return Ok(());
}
Err(StableMemoryError::UnsupportedLayoutVersion(
block.layout_version,
))
}
pub(crate) fn begin_update() -> Result<u64, StableMemoryError> {
let block = Superblock::load()?;
if block.layout_version != CURRENT_LAYOUT_VERSION {
return Err(StableMemoryError::UnsupportedLayoutVersion(
block.layout_version,
));
}
if block.is_importing() {
return Err(StableMemoryError::ImportAlreadyStarted);
}
overlay::begin(block.db_size)?;
Ok(block.db_size)
}
pub(crate) fn rollback_update() {
overlay::rollback();
}
pub(crate) fn commit_update() -> Result<(), StableMemoryError> {
let Some(overlay) = overlay::take() else {
return Ok(());
};
if overlay.is_empty() {
return Ok(());
}
commit_overlay(overlay, true)
}
pub(crate) fn read_at(offset: u64, dst: &mut [u8]) -> Result<bool, StableMemoryError> {
if let Some(result) = overlay::read_at(offset, dst) {
return result;
}
read_base_at(offset, dst)
}
pub(crate) fn read_base_at(offset: u64, dst: &mut [u8]) -> Result<bool, StableMemoryError> {
if dst.is_empty() {
return Ok(true);
}
let block = Superblock::load()?;
read_base_at_with_block(&block, offset, dst)
}
pub(crate) fn read_base_at_with_block(
block: &Superblock,
offset: u64,
dst: &mut [u8],
) -> Result<bool, StableMemoryError> {
if dst.is_empty() {
return Ok(true);
}
if offset >= block.db_size {
dst.fill(0);
return Ok(false);
}
let requested = u64::try_from(dst.len()).map_err(|_| StableMemoryError::OffsetOverflow)?;
if requested <= block.db_size - offset {
read_logical_range(block, offset, dst)?;
return Ok(true);
}
let copied = requested.min(block.db_size - offset);
let copied_len = usize::try_from(copied).map_err(|_| StableMemoryError::OffsetOverflow)?;
read_logical_range(block, offset, &mut dst[..copied_len])?;
dst[copied_len..].fill(0);
Ok(copied == requested)
}
#[inline(always)]
pub(crate) fn read_base_at_with_page_cache(
block: &Superblock,
offset: u64,
dst: &mut [u8],
page_offsets: &mut PageOffsetCache,
) -> Result<bool, StableMemoryError> {
if dst.is_empty() {
return Ok(true);
}
if offset >= block.db_size {
dst.fill(0);
return Ok(false);
}
let requested = u64::try_from(dst.len()).map_err(|_| StableMemoryError::OffsetOverflow)?;
if requested <= block.db_size - offset {
read_logical_range_with_page_cache(block, offset, dst, page_offsets)?;
return Ok(true);
}
let copied = requested.min(block.db_size - offset);
let copied_len = usize::try_from(copied).map_err(|_| StableMemoryError::OffsetOverflow)?;
read_logical_range_with_page_cache(block, offset, &mut dst[..copied_len], page_offsets)?;
dst[copied_len..].fill(0);
Ok(copied == requested)
}
pub(crate) fn read_base_page(page_no: u64) -> Result<Vec<u8>, StableMemoryError> {
let block = Superblock::load()?;
let mut page = zero_page();
read_base_at_with_block(
&block,
page_no
.checked_mul(page_size())
.ok_or(StableMemoryError::OffsetOverflow)?,
&mut page,
)?;
Ok(page)
}
pub(crate) fn write_at(offset: u64, bytes: &[u8]) -> Result<(), StableMemoryError> {
if let Some(result) = overlay::write_at(offset, bytes) {
hit_failpoint(StableBlobFailpoint::OverlayWrite)?;
return result;
}
if bytes.is_empty() {
return Ok(());
}
ensure_current_layout()?;
let mut direct = Overlay::new(Superblock::load()?.db_size);
direct.write_at(offset, bytes)?;
commit_overlay(direct, false)
}
pub(crate) fn truncate(size: u64) -> Result<(), StableMemoryError> {
if let Some(result) = overlay::truncate(size) {
hit_failpoint(StableBlobFailpoint::OverlayTruncate)?;
return result;
}
ensure_current_layout()?;
let mut direct = Overlay::new(Superblock::load()?.db_size);
direct.truncate(size)?;
if direct.is_empty() {
return Ok(());
}
commit_overlay(direct, false)
}
pub(crate) fn file_size() -> Result<u64, StableMemoryError> {
if let Some(size) = overlay::file_size() {
return Ok(size);
}
Ok(Superblock::load()?.db_size)
}
#[allow(dead_code)]
pub fn export_chunk(offset: u64, len: u64) -> Result<Vec<u8>, StableMemoryError> {
reject_during_update()?;
let block = Superblock::load()?;
if offset >= block.db_size {
return Ok(Vec::new());
}
let copied = len.min(block.db_size - offset);
let copied_len = usize::try_from(copied).map_err(|_| StableMemoryError::OffsetOverflow)?;
let mut out = vec![0_u8; copied_len];
read_logical_range(&block, offset, &mut out)?;
Ok(out)
}
#[allow(dead_code)]
pub fn import_chunk(offset: u64, bytes: &[u8]) -> Result<(), StableMemoryError> {
reject_during_update()?;
let mut block = Superblock::load()?;
if !block.is_importing() {
return Err(StableMemoryError::ImportNotStarted);
}
let len = u64::try_from(bytes.len()).map_err(|_| StableMemoryError::OffsetOverflow)?;
if offset != block.import_written_until {
return Err(StableMemoryError::ImportOutOfOrder {
offset,
expected: block.import_written_until,
});
}
let end = checked_add(offset, len)?;
if end > block.import_total_size {
return Err(StableMemoryError::ImportOutOfBounds {
offset,
len,
db_size: block.import_total_size,
});
}
memory::write(import_offset(&block, offset)?, bytes)?;
block.import_written_until = end;
block.store()?;
Ok(())
}
#[allow(dead_code)]
pub fn begin_import(total_size: u64, expected_checksum: u64) -> Result<(), StableMemoryError> {
reject_during_update()?;
let mut block = Superblock::load()?;
if block.is_importing() {
return Err(StableMemoryError::ImportAlreadyStarted);
}
let import_base_offset = append_base()?;
checked_add(import_base_offset, total_size)?;
block.flags |= FLAG_IMPORTING;
block.clear_checksum_refresh();
block.import_expected_checksum = expected_checksum;
block.import_written_until = 0;
block.import_total_size = total_size;
block.import_base_offset = import_base_offset;
block.store()?;
Ok(())
}
#[allow(dead_code)]
pub fn finish_import() -> Result<(), StableMemoryError> {
reject_during_update()?;
let mut block = Superblock::load()?;
if !block.is_importing() {
return Err(StableMemoryError::ImportNotStarted);
}
if block.import_written_until != block.import_total_size {
return Err(StableMemoryError::ImportIncomplete {
written_until: block.import_written_until,
db_size: block.import_total_size,
});
}
let checksum = checksum_physical_range(block.import_base_offset, block.import_total_size)?;
if checksum != block.import_expected_checksum {
let expected = block.import_expected_checksum;
clear_import(&mut block)?;
return Err(StableMemoryError::ChecksumMismatch {
expected,
actual: checksum,
});
}
block.db_size = block.import_total_size;
block.db_base_offset = block.import_base_offset;
block.page_table_offset = 0;
block.page_count = page_count_for_size(block.db_size)?;
block.layout_version = CURRENT_LAYOUT_VERSION;
block.schema_version = 0;
block.flags &= !FLAG_IMPORTING;
block.flags &= !FLAG_CHECKSUM_STALE;
block.clear_zero_extents();
block.clear_checksum_refresh();
block.checksum = checksum;
block.import_expected_checksum = 0;
block.import_written_until = 0;
block.import_total_size = 0;
block.import_base_offset = 0;
block.store()?;
Ok(())
}
#[allow(dead_code)]
pub fn cancel_import() -> Result<(), StableMemoryError> {
reject_during_update()?;
let mut block = Superblock::load()?;
if !block.is_importing() {
return Err(StableMemoryError::ImportNotStarted);
}
clear_import(&mut block)
}
pub fn refresh_checksum() -> Result<u64, StableMemoryError> {
reject_during_update()?;
let checksum = checksum()?;
let mut block = Superblock::load()?;
block.checksum = checksum;
block.flags &= !FLAG_CHECKSUM_STALE;
block.clear_checksum_refresh();
block.store()?;
Ok(checksum)
}
pub fn refresh_checksum_chunk(max_bytes: u64) -> Result<ChecksumRefresh, StableMemoryError> {
reject_during_update()?;
if max_bytes == 0 {
return Err(StableMemoryError::ChecksumRefreshChunkEmpty);
}
let mut block = Superblock::load()?;
if block.is_importing() {
return Err(StableMemoryError::ImportAlreadyStarted);
}
if !block.is_checksum_refreshing() {
block.flags |= FLAG_CHECKSUM_REFRESHING;
block.checksum_refresh_offset = 0;
block.checksum_refresh_hash = fnv1a64(&[]);
block.checksum_refresh_tx_id = block.last_tx_id;
}
if block.checksum_refresh_tx_id != block.last_tx_id {
block.clear_checksum_refresh();
block.store()?;
return refresh_checksum_chunk(max_bytes);
}
let start = block.checksum_refresh_offset;
let end = block.db_size.min(start.saturating_add(max_bytes));
let mut offset = start;
let mut hash = block.checksum_refresh_hash;
while offset < end {
let len = (end - offset).min(CHECKSUM_CHUNK_LEN);
let copied_len = usize::try_from(len).map_err(|_| StableMemoryError::OffsetOverflow)?;
let mut bytes = vec![0_u8; copied_len];
read_logical_range(&block, offset, &mut bytes)?;
hash = fold_fnv1a64(hash, &bytes);
offset += len;
}
block.checksum_refresh_offset = offset;
block.checksum_refresh_hash = hash;
if offset == block.db_size {
block.checksum = hash;
block.flags &= !FLAG_CHECKSUM_STALE;
block.clear_checksum_refresh();
}
let out = ChecksumRefresh {
complete: offset == block.db_size,
checksum: hash,
scanned_bytes: offset,
db_size: block.db_size,
};
block.store()?;
Ok(out)
}
pub fn checksum() -> Result<u64, StableMemoryError> {
reject_during_update()?;
let block = Superblock::load()?;
checksum_logical_range(&block, block.db_size)
}
#[allow(dead_code)]
pub fn compact() -> Result<(), StableMemoryError> {
reject_during_update()?;
ensure_current_layout()?;
Ok(())
}
#[allow(dead_code)]
pub fn storage_stats() -> Result<StorageStats, StableMemoryError> {
let block = Superblock::load()?;
ensure_current_layout()?;
let page_table_bytes = 0;
let zero_extent_count =
u64::try_from(block.zero_extent_count()).map_err(|_| StableMemoryError::OffsetOverflow)?;
let active_bytes = SUPERBLOCK_SIZE
.checked_add(block.db_size)
.ok_or(StableMemoryError::OffsetOverflow)?;
let allocated_bytes = memory::size_pages()
.checked_mul(STABLE_PAGE_SIZE)
.ok_or(StableMemoryError::OffsetOverflow)?;
let orphan_bytes_estimate = allocated_bytes.saturating_sub(active_bytes);
let orphan_ratio_basis_points = orphan_bytes_estimate
.saturating_mul(10_000)
.checked_div(active_bytes)
.unwrap_or(0);
Ok(StorageStats {
layout_version: block.layout_version,
page_count: active_page_count(&block)?,
page_table_bytes,
zero_extent_count,
active_bytes,
allocated_bytes,
orphan_bytes_estimate,
orphan_ratio_basis_points,
compact_recommended: false,
})
}
pub(crate) fn page_count_for_size(size: u64) -> Result<u64, StableMemoryError> {
Ok(size.div_ceil(page_size()))
}
fn commit_overlay(overlay: Overlay, advance_tx: bool) -> Result<(), StableMemoryError> {
hit_failpoint(StableBlobFailpoint::CommitCapacity)?;
let profile_enabled = commit_profile_enabled();
let profile_start = commit_profile_start(profile_enabled);
let block = Superblock::load()?;
commit_profile_record_load(profile_start);
commit_overlay_in_place(&block, overlay, advance_tx, profile_enabled)
}
fn commit_overlay_in_place(
block: &Superblock,
overlay: Overlay,
advance_tx: bool,
profile_enabled: bool,
) -> Result<(), StableMemoryError> {
let overlay_size = overlay.size();
let final_page_count = page_count_for_size(overlay_size)?;
debug_assert!(overlay
.dirty_pages()
.iter()
.all(|(page_no, _)| *page_no < final_page_count));
let dirty_pages = overlay.dirty_pages();
let zero_extents = zero_extents_after_commit(block, &overlay, final_page_count)?;
let mut required_end = checked_add(block.db_base_offset, overlay_size)?;
for (page_no, _) in dirty_pages {
if *page_no >= final_page_count {
continue;
}
required_end = required_end.max(checked_add(
page_physical_offset(block, *page_no)?,
page_size(),
)?);
}
let profile_start = commit_profile_start(profile_enabled);
memory::ensure_capacity(required_end)?;
commit_profile_record_capacity(profile_start);
let profile_start = commit_profile_start(profile_enabled);
for (page_no, page) in dirty_pages {
if *page_no >= final_page_count {
continue;
}
hit_failpoint(StableBlobFailpoint::CommitChunkWrite)?;
write_commit_page(
page_physical_offset(block, *page_no)?,
page,
profile_enabled,
)?;
}
commit_profile_record_page_write(profile_start);
if let Err(error) = hit_failpoint(StableBlobFailpoint::CommitSuperblockStore) {
panic!("failed to publish in-place commit after page writes: {error}");
}
let profile_start = commit_profile_start(profile_enabled);
let result =
store_commit_db_image(advance_tx, block.db_base_offset, overlay_size, zero_extents);
commit_profile_record_superblock_store(profile_start);
if let Err(error) = result {
panic!("failed to publish in-place commit after page writes: {error}");
}
Ok(())
}
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
#[inline(always)]
fn commit_profile_enabled() -> bool {
crate::read_metrics::metrics_enabled()
}
#[cfg(not(any(test, debug_assertions, feature = "bench-profile")))]
#[inline(always)]
fn commit_profile_enabled() -> bool {
false
}
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
#[inline(always)]
fn commit_profile_start(enabled: bool) -> Option<u64> {
if enabled {
Some(crate::read_metrics::instruction_counter())
} else {
None
}
}
#[cfg(not(any(test, debug_assertions, feature = "bench-profile")))]
#[inline(always)]
fn commit_profile_start(_enabled: bool) -> Option<u64> {
None
}
macro_rules! commit_profile_recorder {
($name:ident, $record:ident) => {
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
#[inline(always)]
fn $name(start: Option<u64>) {
if let Some(start) = start {
crate::read_metrics::$record(
crate::read_metrics::instruction_counter().saturating_sub(start),
);
}
}
#[cfg(not(any(test, debug_assertions, feature = "bench-profile")))]
#[inline(always)]
fn $name(_start: Option<u64>) {}
};
}
commit_profile_recorder!(commit_profile_record_capacity, record_commit_capacity);
commit_profile_recorder!(commit_profile_record_load, record_commit_load);
commit_profile_recorder!(commit_profile_record_page_write, record_commit_page_write);
commit_profile_recorder!(
commit_profile_record_superblock_store,
record_commit_superblock_store
);
#[inline(always)]
fn write_commit_page(
offset: u64,
page: &[u8],
profile_enabled: bool,
) -> Result<(), StableMemoryError> {
if profile_enabled {
memory::write_prechecked(offset, page)
} else {
memory::write_prechecked_unmetered(offset, page)
}
}
fn store_commit_db_image(
advance_tx: bool,
db_base_offset: u64,
overlay_size: u64,
zero_extents: Vec<ZeroExtent>,
) -> Result<(), StableMemoryError> {
match advance_tx {
true => Superblock::commit_db_image(db_base_offset, overlay_size, zero_extents),
false => Superblock::store_db_image_without_tx(db_base_offset, overlay_size, zero_extents),
}
}
fn zero_extents_after_commit(
block: &Superblock,
overlay: &Overlay,
final_page_count: u64,
) -> Result<Vec<ZeroExtent>, StableMemoryError> {
let mut extents = block.zero_extents().to_vec();
let old_page_count = page_count_for_size(block.db_size)?;
let dirty_page_slack = overlay
.dirty_pages()
.iter()
.filter(|(page_no, _)| *page_no < final_page_count)
.count();
let temporary_limit = MAX_ZERO_EXTENTS.checked_add(dirty_page_slack).ok_or(
StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS,
},
)?;
if overlay.size() > block.db_size {
let first_new_page = page_count_for_size(block.db_size)?;
add_zero_extent_without_limit(&mut extents, first_new_page, final_page_count);
enforce_temporary_zero_extent_limit(&extents, temporary_limit)?;
}
if overlay.size() < block.db_size {
let first_zero_page = page_count_for_size(overlay.size())?;
add_zero_extent_without_limit(&mut extents, first_zero_page, old_page_count);
enforce_temporary_zero_extent_limit(&extents, temporary_limit)?;
}
for extent in overlay.zero_extents() {
add_zero_extent_without_limit(&mut extents, extent.start_page, extent.end_page);
enforce_temporary_zero_extent_limit(&extents, temporary_limit)?;
}
for (page_no, _) in overlay.dirty_pages() {
if *page_no < final_page_count {
subtract_zero_extent_without_limit(&mut extents, *page_no, page_no.saturating_add(1));
}
}
enforce_zero_extent_limit(&extents)?;
Ok(extents)
}
fn enforce_temporary_zero_extent_limit(
extents: &[ZeroExtent],
temporary_limit: usize,
) -> Result<(), StableMemoryError> {
if extents.len() > temporary_limit {
return Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS,
});
}
Ok(())
}
#[cfg(test)]
fn add_zero_extent(
extents: &mut Vec<ZeroExtent>,
start_page: u64,
end_page: u64,
) -> Result<(), StableMemoryError> {
if start_page >= end_page {
return Ok(());
}
extents.push(ZeroExtent {
start_page,
end_page,
});
normalize_zero_extents(extents)
}
fn add_zero_extent_without_limit(extents: &mut Vec<ZeroExtent>, start_page: u64, end_page: u64) {
if start_page >= end_page {
return;
}
extents.push(ZeroExtent {
start_page,
end_page,
});
normalize_zero_extents_without_limit(extents);
}
#[cfg(test)]
fn subtract_zero_extent(
extents: &mut Vec<ZeroExtent>,
start_page: u64,
end_page: u64,
) -> Result<(), StableMemoryError> {
if start_page >= end_page || extents.is_empty() {
return Ok(());
}
let mut next = Vec::with_capacity(extents.len() + 1);
for extent in extents.iter() {
if end_page <= extent.start_page || start_page >= extent.end_page {
next.push(extent.clone());
continue;
}
if extent.start_page < start_page {
next.push(ZeroExtent {
start_page: extent.start_page,
end_page: start_page,
});
}
if end_page < extent.end_page {
next.push(ZeroExtent {
start_page: end_page,
end_page: extent.end_page,
});
}
}
*extents = next;
normalize_zero_extents(extents)
}
fn subtract_zero_extent_without_limit(
extents: &mut Vec<ZeroExtent>,
start_page: u64,
end_page: u64,
) {
if start_page >= end_page || extents.is_empty() {
return;
}
let mut next = Vec::with_capacity(extents.len() + 1);
for extent in extents.iter() {
if end_page <= extent.start_page || start_page >= extent.end_page {
next.push(extent.clone());
continue;
}
if extent.start_page < start_page {
next.push(ZeroExtent {
start_page: extent.start_page,
end_page: start_page,
});
}
if end_page < extent.end_page {
next.push(ZeroExtent {
start_page: end_page,
end_page: extent.end_page,
});
}
}
*extents = next;
normalize_zero_extents_without_limit(extents);
}
#[cfg(test)]
fn normalize_zero_extents(extents: &mut Vec<ZeroExtent>) -> Result<(), StableMemoryError> {
normalize_zero_extents_without_limit(extents);
enforce_zero_extent_limit(extents)
}
fn normalize_zero_extents_without_limit(extents: &mut Vec<ZeroExtent>) {
extents.retain(|extent| extent.start_page < extent.end_page);
extents.sort_by_key(|extent| extent.start_page);
let mut normalized: Vec<ZeroExtent> = Vec::with_capacity(extents.len());
for extent in extents.drain(..) {
if let Some(last) = normalized.last_mut() {
if extent.start_page <= last.end_page {
last.end_page = last.end_page.max(extent.end_page);
continue;
}
}
normalized.push(extent);
}
*extents = normalized;
}
fn enforce_zero_extent_limit(extents: &[ZeroExtent]) -> Result<(), StableMemoryError> {
if extents.len() > MAX_ZERO_EXTENTS {
return Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS,
});
}
Ok(())
}
fn reject_during_update() -> Result<(), StableMemoryError> {
if overlay::is_active() {
Err(StableMemoryError::UpdateInProgress)
} else {
Ok(())
}
}
fn read_logical_range(
block: &Superblock,
offset: u64,
dst: &mut [u8],
) -> Result<(), StableMemoryError> {
if dst.is_empty() {
return Ok(());
}
let in_page =
usize::try_from(offset % page_size()).map_err(|_| StableMemoryError::OffsetOverflow)?;
if dst.len() <= page_len() - in_page {
return read_logical_page_slice(block, offset / page_size(), in_page, dst);
}
let mut copied_total = 0_usize;
while copied_total < dst.len() {
let absolute = checked_add(
offset,
u64::try_from(copied_total).map_err(|_| StableMemoryError::OffsetOverflow)?,
)?;
let page_no = absolute / page_size();
let in_page = usize::try_from(absolute % page_size())
.map_err(|_| StableMemoryError::OffsetOverflow)?;
let copied = (page_len() - in_page).min(dst.len() - copied_total);
read_logical_page_slice(
block,
page_no,
in_page,
&mut dst[copied_total..copied_total + copied],
)?;
copied_total += copied;
}
Ok(())
}
fn read_logical_range_with_page_cache(
block: &Superblock,
offset: u64,
dst: &mut [u8],
page_offsets: &mut PageOffsetCache,
) -> Result<(), StableMemoryError> {
let in_page =
usize::try_from(offset % page_size()).map_err(|_| StableMemoryError::OffsetOverflow)?;
if dst.len() <= page_len() - in_page {
return read_logical_page_slice_with_page_cache(
block,
offset / page_size(),
in_page,
dst,
page_offsets,
);
}
let mut copied_total = 0_usize;
while copied_total < dst.len() {
let absolute = checked_add(
offset,
u64::try_from(copied_total).map_err(|_| StableMemoryError::OffsetOverflow)?,
)?;
let page_no = absolute / page_size();
let in_page = usize::try_from(absolute % page_size())
.map_err(|_| StableMemoryError::OffsetOverflow)?;
let copied = (page_len() - in_page).min(dst.len() - copied_total);
read_logical_page_slice_with_page_cache(
block,
page_no,
in_page,
&mut dst[copied_total..copied_total + copied],
page_offsets,
)?;
copied_total += copied;
}
Ok(())
}
fn read_logical_page_slice(
block: &Superblock,
page_no: u64,
in_page: usize,
dst: &mut [u8],
) -> Result<(), StableMemoryError> {
if page_is_zero_masked(block, page_no) {
dst.fill(0);
return Ok(());
}
let physical = page_offset_for(block, page_no)?;
if physical == 0 {
dst.fill(0);
return Ok(());
}
let stable_offset = checked_add(
physical,
u64::try_from(in_page).map_err(|_| StableMemoryError::OffsetOverflow)?,
)?;
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
crate::read_metrics::record_stable_data_read(dst.len());
memory::read_preallocated(stable_offset, dst)
}
#[inline(always)]
fn read_logical_page_slice_with_page_cache(
block: &Superblock,
page_no: u64,
in_page: usize,
dst: &mut [u8],
page_offsets: &mut PageOffsetCache,
) -> Result<(), StableMemoryError> {
if page_is_zero_masked(block, page_no) {
dst.fill(0);
return Ok(());
}
if dst.len() < page_len() && page_offsets.copy_page_slice(page_no, in_page, dst) {
return Ok(());
}
let physical = match page_offsets.get(page_no) {
Some(physical) => physical,
None => {
let physical = page_offset_for(block, page_no)?;
page_offsets.insert(page_no, physical);
physical
}
};
if physical == 0 {
dst.fill(0);
return Ok(());
}
if in_page == 0 && dst.len() == page_len() {
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
crate::read_metrics::record_stable_data_read(dst.len());
return memory::read_preallocated(physical, dst);
}
if dst.len() < page_len() {
let mut page = zero_page();
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
crate::read_metrics::record_stable_data_read(page.len());
memory::read_preallocated(physical, &mut page)?;
let end = in_page + dst.len();
dst.copy_from_slice(&page[in_page..end]);
page_offsets.insert_page(page_no, page);
return Ok(());
}
let stable_offset = checked_add(
physical,
u64::try_from(in_page).map_err(|_| StableMemoryError::OffsetOverflow)?,
)?;
#[cfg(any(test, debug_assertions, feature = "bench-profile"))]
crate::read_metrics::record_stable_data_read(dst.len());
memory::read_preallocated(stable_offset, dst)
}
fn page_offset_for(block: &Superblock, page_no: u64) -> Result<u64, StableMemoryError> {
if page_no >= active_page_count(block)? {
return Ok(0);
}
page_physical_offset(block, page_no)
}
fn page_is_zero_masked(block: &Superblock, page_no: u64) -> bool {
block
.zero_extents()
.iter()
.any(|extent| page_no >= extent.start_page && page_no < extent.end_page)
}
fn page_physical_offset(block: &Superblock, page_no: u64) -> Result<u64, StableMemoryError> {
checked_add(
block.db_base_offset,
page_no
.checked_mul(page_size())
.ok_or(StableMemoryError::OffsetOverflow)?,
)
}
fn checksum_logical_range(block: &Superblock, len: u64) -> Result<u64, StableMemoryError> {
let mut offset = 0_u64;
let mut hash = fnv1a64(&[]);
while offset < len {
let chunk_len = (len - offset).min(CHECKSUM_CHUNK_LEN);
let copied_len =
usize::try_from(chunk_len).map_err(|_| StableMemoryError::OffsetOverflow)?;
let mut bytes = vec![0_u8; copied_len];
read_logical_range(block, offset, &mut bytes)?;
hash = fold_fnv1a64(hash, &bytes);
offset += chunk_len;
}
Ok(hash)
}
#[allow(dead_code)]
fn checksum_physical_range(base_offset: u64, len: u64) -> Result<u64, StableMemoryError> {
let mut offset = 0_u64;
let mut hash = fnv1a64(&[]);
while offset < len {
let chunk_len = (len - offset).min(CHECKSUM_CHUNK_LEN);
let copied_len =
usize::try_from(chunk_len).map_err(|_| StableMemoryError::OffsetOverflow)?;
let mut bytes = vec![0_u8; copied_len];
memory::read_preallocated(checked_add(base_offset, offset)?, &mut bytes)?;
hash = fold_fnv1a64(hash, &bytes);
offset += chunk_len;
}
Ok(hash)
}
#[allow(dead_code)]
fn clear_import(block: &mut Superblock) -> Result<(), StableMemoryError> {
block.flags &= !FLAG_IMPORTING;
block.import_expected_checksum = 0;
block.import_written_until = 0;
block.import_total_size = 0;
block.import_base_offset = 0;
block.store()?;
Ok(())
}
#[allow(dead_code)]
fn import_offset(block: &Superblock, offset: u64) -> Result<u64, StableMemoryError> {
checked_add(block.import_base_offset, offset)
}
fn active_page_count(block: &Superblock) -> Result<u64, StableMemoryError> {
page_count_for_size(block.db_size)
}
#[allow(dead_code)]
fn append_base() -> Result<u64, StableMemoryError> {
memory::size_pages()
.checked_mul(STABLE_PAGE_SIZE)
.ok_or(StableMemoryError::OffsetOverflow)
}
fn page_size() -> u64 {
u64::from(SQLITE_PAGE_SIZE)
}
fn page_len() -> usize {
usize::try_from(SQLITE_PAGE_SIZE).expect("SQLite page size fits usize")
}
fn zero_page() -> Vec<u8> {
vec![0_u8; page_len()]
}
fn checked_add(left: u64, right: u64) -> Result<u64, StableMemoryError> {
left.checked_add(right)
.ok_or(StableMemoryError::OffsetOverflow)
}
fn fold_fnv1a64(mut hash: u64, bytes: &[u8]) -> u64 {
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
hash
}
#[cfg(test)]
fn hit_failpoint(failpoint: StableBlobFailpoint) -> Result<(), StableMemoryError> {
let Ok(context) = memory::active_context_id() else {
return Ok(());
};
FAILPOINTS.with(|slot| {
let mut slot = slot.borrow_mut();
if slot.get(&context).copied() == Some(failpoint) {
slot.remove(&context);
Err(StableMemoryError::Failpoint(failpoint.name()))
} else {
Ok(())
}
})
}
#[cfg(not(test))]
fn hit_failpoint(_failpoint: StableBlobFailpoint) -> Result<(), StableMemoryError> {
Ok(())
}
#[cfg(test)]
impl StableBlobFailpoint {
fn name(self) -> &'static str {
match self {
Self::OverlayWrite => "before overlay write",
Self::OverlayTruncate => "before overlay truncate",
Self::CommitCapacity => "before commit capacity",
Self::CommitChunkWrite => "before commit page write",
Self::CommitSuperblockStore => "before commit superblock store",
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use proptest::prelude::*;
use proptest::test_runner::{Config, TestRunner};
use std::collections::BTreeSet;
#[test]
fn layout_math_matches_expected_boundaries() {
assert_eq!(page_count_for_size(0).unwrap(), 0);
assert_eq!(page_count_for_size(1).unwrap(), 1);
assert_eq!(page_count_for_size(page_size()).unwrap(), 1);
assert_eq!(page_count_for_size(page_size() + 1).unwrap(), 2);
}
#[test]
fn layout_math_rejects_u64_max_overflow_boundaries() {
assert!(matches!(
checked_add(u64::MAX, 1),
Err(StableMemoryError::OffsetOverflow)
));
let mut block = Superblock::fresh();
block.import_base_offset = u64::MAX;
assert!(matches!(
import_offset(&block, 1),
Err(StableMemoryError::OffsetOverflow)
));
block.db_base_offset = u64::MAX - page_size() + 1;
assert!(matches!(
page_physical_offset(&block, 1),
Err(StableMemoryError::OffsetOverflow)
));
}
#[test]
fn pbt_layout_math_matches_verus_model() {
let mut runner = TestRunner::new(Config {
cases: 512,
..Config::default()
});
runner
.run(
&(boundary_size_strategy(), boundary_page_strategy()),
|(size, page_no)| {
let page_count = page_count_for_size(size).unwrap();
let page_size = u128::from(page_size());
if size == 0 {
prop_assert_eq!(page_count, 0);
} else {
prop_assert!(u128::from(page_count - 1) * page_size < u128::from(size));
prop_assert!(u128::from(size) <= u128::from(page_count) * page_size);
}
let mut block = Superblock::fresh();
block.db_base_offset = SUPERBLOCK_SIZE;
let expected = u128::from(SUPERBLOCK_SIZE) + u128::from(page_no) * page_size;
match page_physical_offset(&block, page_no) {
Ok(physical) => prop_assert_eq!(u128::from(physical), expected),
Err(StableMemoryError::OffsetOverflow) => {
prop_assert!(expected > u128::from(u64::MAX));
}
Err(error) => return Err(TestCaseError::fail(error.to_string())),
}
Ok(())
},
)
.unwrap();
}
fn boundary_size_strategy() -> impl Strategy<Value = u64> {
let page = page_size();
let far_page_bytes = FAR_PAGE_NO * page;
prop_oneof![
any::<u64>(),
prop::sample::select(boundary_values(&[
0,
1,
page - 1,
page,
page + 1,
far_page_bytes - 1,
far_page_bytes,
far_page_bytes + 1,
u64::MAX,
])),
]
}
fn boundary_page_strategy() -> impl Strategy<Value = u64> {
prop_oneof![
any::<u64>(),
prop::sample::select(boundary_values(&[
0,
1,
FAR_PAGE_NO - 1,
FAR_PAGE_NO,
FAR_PAGE_NO + 1,
u64::MAX,
])),
]
}
fn boundary_values(values: &[u64]) -> Vec<u64> {
values
.iter()
.flat_map(|value| [value.saturating_sub(1), *value, value.saturating_add(1)])
.collect()
}
#[test]
fn fnv_fold_matches_one_pass_for_multiple_partitions() {
let bytes: Vec<u8> = (0..97)
.map(|index| (index as u8).wrapping_mul(37).wrapping_add(11))
.collect();
let expected = fnv1a64(&bytes);
for split in [0_usize, 1, 2, 7, 31, 64, bytes.len()] {
let split = split.min(bytes.len());
let mut hash = fnv1a64(&[]);
hash = fold_fnv1a64(hash, &bytes[..split]);
hash = fold_fnv1a64(hash, &bytes[split..]);
assert_eq!(hash, expected);
}
let mut hash = fnv1a64(&[]);
for chunk in bytes.chunks(13) {
hash = fold_fnv1a64(hash, chunk);
}
assert_eq!(hash, expected);
}
#[test]
#[serial_test::serial]
fn in_place_commit_keeps_dirty_page_offsets_stable() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let page_zero = vec![1_u8; page_len()];
let page_later = vec![2_u8; page_len()];
let later_page_no = FAR_PAGE_NO;
write_at(0, &page_zero).unwrap();
write_at(later_page_no * page_size(), &page_later).unwrap();
let block = Superblock::load().unwrap();
let old_page_zero_offset = page_physical_offset(&block, 0).unwrap();
let old_later_offset = page_physical_offset(&block, later_page_no).unwrap();
let updated_page_zero = vec![3_u8; page_len()];
write_at(0, &updated_page_zero).unwrap();
let updated_block = Superblock::load().unwrap();
let mut out = vec![0_u8; page_len()];
read_base_at(0, &mut out).unwrap();
assert_eq!(
page_physical_offset(&updated_block, 0).unwrap(),
old_page_zero_offset
);
assert_eq!(
page_physical_offset(&updated_block, later_page_no).unwrap(),
old_later_offset
);
assert_eq!(out, updated_page_zero);
}
#[test]
#[serial_test::serial]
fn in_place_commit_tracks_multi_segment_dirty_and_clean_pages() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let clean_page_no = 1;
let later_page_no = FAR_PAGE_NO;
write_at(0, &vec![1_u8; page_len()]).unwrap();
write_at(clean_page_no * page_size(), &vec![2_u8; page_len()]).unwrap();
write_at(later_page_no * page_size(), &vec![3_u8; page_len()]).unwrap();
let before = Superblock::load().unwrap();
let before_first = page_physical_offset(&before, 0).unwrap();
let before_clean = page_physical_offset(&before, clean_page_no).unwrap();
let before_later = page_physical_offset(&before, later_page_no).unwrap();
begin_update().unwrap();
write_at(0, &vec![4_u8; page_len()]).unwrap();
write_at(later_page_no * page_size(), &vec![5_u8; page_len()]).unwrap();
commit_update().unwrap();
let after = Superblock::load().unwrap();
assert_eq!(after.page_count, active_page_count(&after).unwrap());
assert_eq!(page_physical_offset(&after, 0).unwrap(), before_first);
assert_eq!(
page_physical_offset(&after, clean_page_no).unwrap(),
before_clean
);
assert_eq!(
page_physical_offset(&after, later_page_no).unwrap(),
before_later
);
}
#[test]
#[serial_test::serial]
fn in_place_truncate_hides_old_tail_pages() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
write_at(0, &vec![1_u8; page_len()]).unwrap();
write_at(page_size(), &vec![2_u8; page_len()]).unwrap();
write_at(2 * page_size(), &vec![3_u8; page_len()]).unwrap();
truncate(page_size()).unwrap();
let block = Superblock::load().unwrap();
let mut hidden = vec![1_u8; page_len()];
read_base_at(page_size(), &mut hidden).unwrap();
assert_eq!(block.db_size, page_size());
assert_eq!(block.page_count, 1);
assert_eq!(block.zero_extent_count(), 1);
assert_eq!(hidden, vec![0_u8; page_len()]);
}
#[test]
#[serial_test::serial]
fn same_overlay_truncate_reextend_zeroes_active_gap() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
write_at(0, &vec![1_u8; page_len()]).unwrap();
write_at(page_size(), &vec![0xAA; page_len()]).unwrap();
write_at(2 * page_size(), &vec![3_u8; page_len()]).unwrap();
begin_update().unwrap();
truncate(page_size()).unwrap();
truncate(2 * page_size()).unwrap();
let mut overlay_gap = vec![1_u8; page_len()];
read_at(page_size(), &mut overlay_gap).unwrap();
commit_update().unwrap();
let committed_gap = export_chunk(page_size(), page_size()).unwrap();
assert_eq!(overlay_gap, vec![0_u8; page_len()]);
assert_eq!(committed_gap, vec![0_u8; page_len()]);
}
#[test]
#[serial_test::serial]
fn compact_is_noop_for_in_place_layout() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let later_page_no = FAR_PAGE_NO;
let first_page = vec![7_u8; page_len()];
let later_page = vec![9_u8; page_len()];
write_at(0, &first_page).unwrap();
write_at(later_page_no * page_size(), &later_page).unwrap();
compact().unwrap();
let block = Superblock::load().unwrap();
let mut first_out = vec![0_u8; page_len()];
let mut later_out = vec![0_u8; page_len()];
read_base_at(0, &mut first_out).unwrap();
read_base_at(later_page_no * page_size(), &mut later_out).unwrap();
assert_eq!(
page_physical_offset(&block, later_page_no).unwrap(),
block.db_base_offset + later_page_no * page_size()
);
assert_eq!(first_out, first_page);
assert_eq!(later_out, later_page);
}
#[test]
#[serial_test::serial]
fn in_place_expand_only_commit_zeroes_new_logical_tail() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
write_at(0, &[0]).unwrap();
truncate(page_size() * 4).unwrap();
truncate(page_size() * 4 + 1).unwrap();
let block = Superblock::load().unwrap();
let mut first = [1_u8; 1];
let mut expanded_tail = [1_u8; 1];
read_base_at(0, &mut first).unwrap();
read_base_at(page_size() * 4, &mut expanded_tail).unwrap();
assert_eq!(block.db_size, page_size() * 4 + 1);
assert_eq!(
page_physical_offset(&block, 0).unwrap(),
block.db_base_offset
);
assert_eq!(
page_physical_offset(&block, 4).unwrap(),
block.db_base_offset + 4 * page_size()
);
assert_eq!(first, [0]);
assert_eq!(expanded_tail, [0]);
}
#[test]
#[serial_test::serial]
fn truncate_grow_zeroes_stale_physical_gap() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let old_size = page_size() + 17;
let grow_len = page_size() * 2 + 111;
let grow_len_usize = usize::try_from(grow_len).unwrap();
write_at(old_size - 1, b"a").unwrap();
let block = Superblock::load().unwrap();
crate::stable::memory::write(block.db_base_offset + old_size, &vec![0xAA; grow_len_usize])
.unwrap();
truncate(old_size + grow_len).unwrap();
let gap = export_chunk(old_size, grow_len).unwrap();
assert!(gap.iter().all(|byte| *byte == 0));
}
#[test]
#[serial_test::serial]
fn sparse_write_zeroes_stale_physical_gap() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let old_size = page_size() + 17;
let write_offset = old_size + page_size() * 2 + 33;
let gap_len = write_offset - old_size;
let gap_len_usize = usize::try_from(gap_len).unwrap();
write_at(old_size - 1, b"a").unwrap();
let block = Superblock::load().unwrap();
crate::stable::memory::write(block.db_base_offset + old_size, &vec![0xAA; gap_len_usize])
.unwrap();
write_at(write_offset, b"z").unwrap();
let gap = export_chunk(old_size, gap_len).unwrap();
let written = export_chunk(write_offset, 1).unwrap();
assert!(gap.iter().all(|byte| *byte == 0));
assert_eq!(written, b"z");
}
#[test]
#[serial_test::serial]
fn failed_import_slack_stays_hidden_after_normal_grow() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
write_at(0, b"a").unwrap();
begin_import(page_size(), 0).unwrap();
let importing = Superblock::load().unwrap();
import_chunk(0, &vec![0xAA; page_len()]).unwrap();
assert!(finish_import().is_err());
let block = Superblock::load().unwrap();
let logical_import_offset = importing.import_base_offset - block.db_base_offset;
truncate(logical_import_offset + page_size()).unwrap();
let grown_over_import_slack = export_chunk(logical_import_offset, page_size()).unwrap();
assert!(grown_over_import_slack.iter().all(|byte| *byte == 0));
}
#[test]
#[serial_test::serial]
fn zero_mask_hides_truncated_tail_without_moving_base() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
write_at(0, &vec![b'a'; page_len()]).unwrap();
write_at(page_size(), &vec![b'b'; page_len()]).unwrap();
let old_base = Superblock::load().unwrap().db_base_offset;
truncate(1).unwrap();
let mut stale_tail = [0_u8; 1];
crate::stable::memory::read_preallocated(old_base + page_size(), &mut stale_tail).unwrap();
assert_eq!(stale_tail, [b'b']);
let grow_offset = 2 * page_size() + 2;
write_at(grow_offset, b"z").unwrap();
let block = Superblock::load().unwrap();
let image = export_chunk(0, block.db_size).unwrap();
let z_index = usize::try_from(grow_offset).unwrap();
assert_eq!(block.db_base_offset, old_base);
assert_eq!(block.db_size, grow_offset + 1);
assert_eq!(block.zero_extent_count(), 1);
assert_eq!(image[0], b'a');
assert!(image[1..z_index].iter().all(|byte| *byte == 0));
assert_eq!(image[z_index], b'z');
}
#[test]
fn zero_extent_add_subtract_merges_and_splits_ranges() {
let mut extents = Vec::new();
add_zero_extent(&mut extents, 10, 20).unwrap();
add_zero_extent(&mut extents, 30, 40).unwrap();
add_zero_extent(&mut extents, 20, 30).unwrap();
assert_eq!(
extents,
vec![ZeroExtent {
start_page: 10,
end_page: 40
}]
);
subtract_zero_extent(&mut extents, 15, 17).unwrap();
assert_eq!(
extents,
vec![
ZeroExtent {
start_page: 10,
end_page: 15
},
ZeroExtent {
start_page: 17,
end_page: 40
}
]
);
subtract_zero_extent(&mut extents, 8, 18).unwrap();
assert_eq!(
extents,
vec![ZeroExtent {
start_page: 18,
end_page: 40
}]
);
}
#[test]
fn zero_extent_truncate_tail_merges_with_existing_range() {
let mut extents = vec![ZeroExtent {
start_page: 10,
end_page: 20,
}];
add_zero_extent(&mut extents, 10, 40).unwrap();
assert_eq!(
extents,
vec![ZeroExtent {
start_page: 10,
end_page: 40
}]
);
}
#[test]
fn zero_extent_commit_allows_temporary_limit_excess_removed_by_dirty_page() {
let mut block = Superblock::fresh();
block.db_size = (MAX_ZERO_EXTENTS as u64) * 2 * page_size();
block.zero_extents = (0..MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: (index as u64) * 2,
end_page: (index as u64) * 2 + 1,
})
.collect();
let mut overlay = Overlay::new(block.db_size);
overlay
.write_at(block.db_size, &vec![0xAA; page_len()])
.unwrap();
let final_page_count = page_count_for_size(overlay.size()).unwrap();
let extents = zero_extents_after_commit(&block, &overlay, final_page_count).unwrap();
assert_eq!(extents.len(), MAX_ZERO_EXTENTS);
assert_eq!(extents, block.zero_extents);
}
#[test]
fn zero_extent_commit_rejects_final_limit_excess() {
let mut block = Superblock::fresh();
block.db_size = (MAX_ZERO_EXTENTS as u64) * 2 * page_size();
block.zero_extents = (0..MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: (index as u64) * 2,
end_page: (index as u64) * 2 + 1,
})
.collect();
let mut overlay = Overlay::new(block.db_size);
overlay.truncate(block.db_size + page_size()).unwrap();
let final_page_count = page_count_for_size(overlay.size()).unwrap();
assert!(matches!(
zero_extents_after_commit(&block, &overlay, final_page_count),
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
));
}
#[test]
#[serial_test::serial]
fn zero_extent_limit_error_keeps_active_image_and_metadata() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let mut block = Superblock::fresh();
let max_zero_extents = u64::try_from(MAX_ZERO_EXTENTS).unwrap();
block.db_size = max_zero_extents * 2 * page_size();
block.page_count = page_count_for_size(block.db_size).unwrap();
block.last_tx_id = 41;
block.flags = FLAG_CHECKSUM_STALE | FLAG_CHECKSUM_REFRESHING;
block.checksum = 0xA5A5;
block.checksum_refresh_offset = 17;
block.checksum_refresh_hash = 0x5A5A;
block.checksum_refresh_tx_id = 40;
block.zero_extents = (0..MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: u64::try_from(index).unwrap() * 2,
end_page: u64::try_from(index).unwrap() * 2 + 1,
})
.collect();
let sample_page_no = 1_u64;
let sample_offset = page_physical_offset(&block, sample_page_no).unwrap();
let sample = b"active-before";
let sample_len = u64::try_from(sample.len()).unwrap();
crate::stable::memory::ensure_capacity(checked_add(sample_offset, sample_len).unwrap())
.unwrap();
crate::stable::memory::write(sample_offset, sample).unwrap();
block.store().unwrap();
let before_block = Superblock::load().unwrap();
let mut before_sample = vec![0_u8; sample.len()];
read_base_at(sample_page_no * page_size(), &mut before_sample).unwrap();
assert_eq!(before_sample, sample);
begin_update().unwrap();
truncate(block.db_size + page_size()).unwrap();
let result = commit_update();
assert!(matches!(
result,
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
));
assert_eq!(Superblock::load().unwrap(), before_block);
let mut after_sample = vec![0_u8; sample.len()];
read_base_at(sample_page_no * page_size(), &mut after_sample).unwrap();
assert_eq!(after_sample, before_sample);
}
#[test]
fn zero_extent_temporary_limit_allows_only_dirty_page_slack() {
let extents = (0..=MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: index as u64,
end_page: index as u64 + 1,
})
.collect::<Vec<_>>();
enforce_temporary_zero_extent_limit(&extents, MAX_ZERO_EXTENTS + 1).unwrap();
assert!(matches!(
enforce_temporary_zero_extent_limit(&extents, MAX_ZERO_EXTENTS),
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
));
}
#[test]
fn zero_extent_limit_counts_normalized_ranges() {
let mut extents = (0..MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: (index as u64) * 2,
end_page: (index as u64) * 2 + 1,
})
.collect::<Vec<_>>();
add_zero_extent(
&mut extents,
(MAX_ZERO_EXTENTS as u64) * 2 - 2,
(MAX_ZERO_EXTENTS as u64) * 2,
)
.unwrap();
assert_eq!(extents.len(), MAX_ZERO_EXTENTS);
assert_eq!(
extents[MAX_ZERO_EXTENTS - 1],
ZeroExtent {
start_page: (MAX_ZERO_EXTENTS as u64) * 2 - 2,
end_page: (MAX_ZERO_EXTENTS as u64) * 2
}
);
}
#[test]
fn zero_extent_limit_is_rejected_without_fallback() {
let mut extents = (0..MAX_ZERO_EXTENTS)
.map(|index| ZeroExtent {
start_page: (index as u64) * 2,
end_page: (index as u64) * 2 + 1,
})
.collect::<Vec<_>>();
let result = add_zero_extent(
&mut extents,
(MAX_ZERO_EXTENTS as u64) * 2,
(MAX_ZERO_EXTENTS as u64) * 2 + 1,
);
assert!(matches!(
result,
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
));
}
#[test]
fn pbt_zero_extent_normalizer_matches_independent_model() {
let mut runner = TestRunner::new(Config {
cases: 512,
..Config::default()
});
runner
.run(&zero_extent_vec_strategy(), |mut extents| {
let expected = model_normalize_zero_extents(&extents);
let result = normalize_zero_extents(&mut extents);
if expected.len() > MAX_ZERO_EXTENTS {
let rejected = matches!(
result,
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
);
prop_assert!(rejected);
} else {
result.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(extents, expected);
}
Ok(())
})
.unwrap();
}
#[test]
fn pbt_zero_extent_add_subtract_matches_independent_model() {
let mut runner = TestRunner::new(Config {
cases: 256,
..Config::default()
});
runner
.run(&zero_extent_operation_sequence(), |operations| {
let mut production = Vec::new();
let mut model = Vec::new();
for operation in operations {
let result = match operation {
ZeroExtentOp::Add {
start_page,
end_page,
} => {
model_add_zero_extent(&mut model, start_page, end_page);
add_zero_extent(&mut production, start_page, end_page)
}
ZeroExtentOp::Subtract {
start_page,
end_page,
} => {
model_subtract_zero_extent(&mut model, start_page, end_page);
subtract_zero_extent(&mut production, start_page, end_page)
}
};
let expected = model_normalize_zero_extents(&model);
if expected.len() > MAX_ZERO_EXTENTS {
let rejected = matches!(
result,
Err(StableMemoryError::ZeroExtentLimitExceeded {
limit: MAX_ZERO_EXTENTS
})
);
prop_assert!(rejected);
return Ok(());
}
result.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(&production, &expected);
model = expected;
}
Ok(())
})
.unwrap();
}
#[test]
#[serial_test::serial]
fn pbt_noop_compact_preserves_sparse_page_model() {
let mut runner = TestRunner::new(Config {
cases: 32,
..Config::default()
});
runner
.run(
&proptest::collection::vec(prop::option::of(any::<u8>()), 0..=300),
|pages| {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let active_len = pages
.iter()
.rposition(Option::is_some)
.map(|index| index + 1)
.unwrap_or(0);
for (page_no, byte) in pages.iter().take(active_len).enumerate() {
if let Some(byte) = byte {
write_at(
u64::try_from(page_no).unwrap() * page_size(),
&vec![*byte; page_len()],
)
.unwrap();
}
}
compact().unwrap();
let block = Superblock::load().unwrap();
prop_assert_eq!(
block.db_size,
u64::try_from(active_len).unwrap() * page_size()
);
prop_assert_eq!(block.page_count, active_page_count(&block).unwrap());
for (page_no, byte) in pages.iter().take(active_len).enumerate() {
let mut page = vec![0_u8; page_len()];
read_base_at(u64::try_from(page_no).unwrap() * page_size(), &mut page)
.unwrap();
if let Some(byte) = byte {
prop_assert_eq!(page, vec![*byte; page_len()]);
} else {
prop_assert_eq!(page, vec![0_u8; page_len()]);
}
}
Ok(())
},
)
.unwrap();
}
#[derive(Clone, Debug)]
enum BlobOp {
Write { offset: u64, len: usize, byte: u8 },
Truncate { size: u64 },
Compact,
}
#[derive(Clone, Debug)]
enum ZeroExtentOp {
Add { start_page: u64, end_page: u64 },
Subtract { start_page: u64, end_page: u64 },
}
fn zero_extent_vec_strategy() -> impl Strategy<Value = Vec<ZeroExtent>> {
let page_limit = zero_extent_model_page_limit();
proptest::collection::vec(
(0_u64..=page_limit, 0_u64..=page_limit).prop_map(|(left, right)| ZeroExtent {
start_page: left.min(right),
end_page: left.max(right),
}),
0..=MAX_ZERO_EXTENTS + 8,
)
}
fn zero_extent_operation_sequence() -> impl Strategy<Value = Vec<ZeroExtentOp>> {
let page_limit = zero_extent_model_page_limit();
let range = (0_u64..=page_limit, 0_u64..=page_limit)
.prop_map(|(left, right)| (left.min(right), left.max(right)));
proptest::collection::vec(
prop_oneof![
range
.clone()
.prop_map(|(start_page, end_page)| ZeroExtentOp::Add {
start_page,
end_page,
}),
range.prop_map(|(start_page, end_page)| ZeroExtentOp::Subtract {
start_page,
end_page,
}),
],
0..=96,
)
}
fn zero_extent_model_page_limit() -> u64 {
(u64::try_from(MAX_ZERO_EXTENTS).unwrap() + 8) * 2
}
fn model_add_zero_extent(extents: &mut Vec<ZeroExtent>, start_page: u64, end_page: u64) {
if start_page < end_page {
extents.push(ZeroExtent {
start_page,
end_page,
});
}
}
fn model_subtract_zero_extent(extents: &mut Vec<ZeroExtent>, start_page: u64, end_page: u64) {
if start_page >= end_page {
return;
}
let mut next = Vec::new();
for extent in extents.iter() {
if end_page <= extent.start_page || start_page >= extent.end_page {
next.push(extent.clone());
continue;
}
if extent.start_page < start_page {
next.push(ZeroExtent {
start_page: extent.start_page,
end_page: start_page,
});
}
if end_page < extent.end_page {
next.push(ZeroExtent {
start_page: end_page,
end_page: extent.end_page,
});
}
}
*extents = next;
}
fn model_normalize_zero_extents(extents: &[ZeroExtent]) -> Vec<ZeroExtent> {
let page_limit = usize::try_from(zero_extent_model_page_limit()).unwrap();
let mut mask = vec![false; page_limit + 1];
for extent in extents {
let start = usize::try_from(extent.start_page).unwrap().min(mask.len());
let end = usize::try_from(extent.end_page).unwrap().min(mask.len());
for page in &mut mask[start..end] {
*page = true;
}
}
let mut normalized = Vec::new();
let mut page = 0_usize;
while page < mask.len() {
if !mask[page] {
page += 1;
continue;
}
let start = page;
while page < mask.len() && mask[page] {
page += 1;
}
normalized.push(ZeroExtent {
start_page: u64::try_from(start).unwrap(),
end_page: u64::try_from(page).unwrap(),
});
}
normalized
}
#[test]
#[serial_test::serial]
fn pbt_blob_operations_match_logical_model_across_noop_compact() {
let mut runner = TestRunner::new(Config {
cases: 48,
..Config::default()
});
runner
.run(&blob_operation_sequence(), |operations| {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let mut model = Vec::new();
let mut materialized = BTreeSet::new();
assert_blob_model(&model, &materialized, false)?;
for operation in operations {
let compacted = apply_blob_op(operation, &mut model, &mut materialized)?;
assert_blob_model(&model, &materialized, compacted)?;
}
Ok(())
})
.unwrap();
}
fn blob_operation_sequence() -> impl Strategy<Value = Vec<BlobOp>> {
let write = (blob_offset_strategy(), blob_len_strategy(), any::<u8>())
.prop_map(|(offset, len, byte)| BlobOp::Write { offset, len, byte });
let truncate = blob_offset_strategy().prop_map(|size| BlobOp::Truncate { size });
proptest::collection::vec(prop_oneof![write, truncate, Just(BlobOp::Compact)], 0..=48)
}
fn blob_offset_strategy() -> impl Strategy<Value = u64> {
let limit = blob_model_limit();
let page = page_size();
let far_page = FAR_PAGE_NO * page;
prop_oneof![
0_u64..=limit,
prop::sample::select(boundary_values(&[
0,
1,
page - 1,
page,
page + 1,
far_page - 1,
far_page,
far_page + 1,
limit - 1,
limit,
]))
.prop_map(move |value| value.min(limit)),
]
}
fn blob_len_strategy() -> impl Strategy<Value = usize> {
prop_oneof![
0_usize..=(page_len() * 2 + 17),
prop::sample::select(vec![
0,
1,
page_len() - 1,
page_len(),
page_len() + 1,
page_len() * 2 + 1,
]),
]
}
fn blob_model_limit() -> u64 {
(FAR_PAGE_NO + 3) * page_size()
}
fn apply_blob_op(
operation: BlobOp,
model: &mut Vec<u8>,
materialized: &mut BTreeSet<u64>,
) -> Result<bool, TestCaseError> {
match operation {
BlobOp::Write { offset, len, byte } => {
let len = len.min(usize::try_from(blob_model_limit() - offset).unwrap());
let bytes = vec![byte; len];
write_at(offset, &bytes).map_err(|error| TestCaseError::fail(error.to_string()))?;
if len == 0 {
return Ok(false);
}
let start = usize::try_from(offset).unwrap();
let end = start + len;
if model.len() < start {
model.resize(start, 0);
}
if model.len() < end {
model.resize(end, 0);
}
model[start..end].copy_from_slice(&bytes);
mark_materialized_range(offset, len, materialized);
Ok(false)
}
BlobOp::Truncate { size } => {
truncate(size).map_err(|error| TestCaseError::fail(error.to_string()))?;
let new_len = usize::try_from(size).unwrap();
model.resize(new_len, 0);
let active_pages = page_count_for_size(size)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
materialized.retain(|page_no| *page_no < active_pages);
if size > 0 && !size.is_multiple_of(page_size()) {
materialized.insert(size / page_size());
}
Ok(false)
}
BlobOp::Compact => {
compact().map_err(|error| TestCaseError::fail(error.to_string()))?;
Ok(true)
}
}
}
fn mark_materialized_range(offset: u64, len: usize, materialized: &mut BTreeSet<u64>) {
let end = offset + u64::try_from(len).unwrap();
let first_page = offset / page_size();
let last_page = (end - 1) / page_size();
for page_no in first_page..=last_page {
materialized.insert(page_no);
}
}
fn assert_blob_model(
model: &[u8],
_materialized: &BTreeSet<u64>,
_expect_compacted: bool,
) -> Result<(), TestCaseError> {
let block = Superblock::load().map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(block.db_size, u64::try_from(model.len()).unwrap());
if !model.is_empty() {
let mut out = vec![0_u8; model.len()];
read_base_at(0, &mut out).map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(out, model);
}
let mut tail = vec![1_u8; 32];
read_base_at(u64::try_from(model.len()).unwrap(), &mut tail)
.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(tail, vec![0_u8; 32]);
let active_pages = page_count_for_size(u64::try_from(model.len()).unwrap())
.map_err(|error| TestCaseError::fail(error.to_string()))?;
prop_assert_eq!(block.page_count, active_pages);
Ok(())
}
#[test]
#[serial_test::serial]
fn read_metrics_track_in_place_data_reads() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let page = vec![7_u8; page_len()];
write_at(0, &page).unwrap();
crate::read_metrics::reset_read_metrics();
let first = read_base_page(0).unwrap();
let second = read_base_page(0).unwrap();
let metrics = crate::read_metrics::read_metrics_snapshot();
assert_eq!(first, page);
assert_eq!(second, page);
assert!(metrics.stable_data_read_calls >= 2);
assert!(metrics.stable_data_read_bytes >= page_size() * 2);
#[cfg(feature = "bench-profile")]
assert!(metrics.superblock_loads <= 1);
#[cfg(not(feature = "bench-profile"))]
assert_eq!(metrics.superblock_loads, 0);
}
#[test]
#[serial_test::serial]
fn page_offset_cache_reuses_page_data_for_small_reads() {
crate::stable::memory::reset_for_tests();
crate::stable::memory::init(crate::stable::memory::memory_for_tests()).unwrap();
let page = vec![9_u8; page_len()];
write_at(0, &page).unwrap();
let block = Superblock::load().unwrap();
let mut cache = PageOffsetCache::new();
let mut first = [0_u8; 16];
let mut second = [0_u8; 16];
crate::read_metrics::reset_read_metrics();
read_base_at_with_page_cache(&block, 0, &mut first, &mut cache).unwrap();
read_base_at_with_page_cache(&block, 8, &mut second, &mut cache).unwrap();
let metrics = crate::read_metrics::read_metrics_snapshot();
assert_eq!(first, [9_u8; 16]);
assert_eq!(second, [9_u8; 16]);
assert_eq!(metrics.stable_data_read_calls, 1);
assert_eq!(metrics.stable_data_read_bytes, page_size());
}
}