use super::{Page, PageSource};
use crate::error::{Error, Result};
use crate::format::DatabaseHeader;
use crate::format::header::HEADER_LEN;
use crate::vfs::File;
use alloc::boxed::Box;
use alloc::collections::BTreeMap;
#[cfg(not(feature = "std"))]
use alloc::rc::Rc;
use alloc::vec;
use alloc::vec::Vec;
#[cfg(not(feature = "std"))]
use core::cell::RefCell;
const WAL_HEADER_LEN: usize = 32;
const FRAME_HEADER_LEN: usize = 24;
const WAL_MAGIC_BE: u32 = 0x377f_0682;
pub(crate) const NATIVE_BIG_ENDIAN: bool = cfg!(target_endian = "big");
pub struct WalReader {
main: Box<dyn File>,
header: DatabaseHeader,
page_size: usize,
db_size: u32,
frames: BTreeMap<u32, Vec<u8>>,
}
impl WalReader {
pub fn open(main: Box<dyn File>, wal: &mut dyn File) -> Result<WalReader> {
let wal_size = wal.size()?;
let mut frames = BTreeMap::new();
let mut commit_db_size = 0u32;
let mut wal_page_size = 0usize;
if wal_size >= WAL_HEADER_LEN as u64 {
let mut hdr = [0u8; WAL_HEADER_LEN];
wal.read_exact_at(&mut hdr, 0)?;
let magic = be32(&hdr, 0);
if magic & 0xFFFF_FFFE != WAL_MAGIC_BE {
return Err(Error::Corrupt("bad WAL magic".into()));
}
let big_endian = (magic & 1) == 1;
let page_size = be32(&hdr, 8) as usize;
if page_size < 512 || !page_size.is_power_of_two() {
return Err(Error::Corrupt("bad WAL page size".into()));
}
wal_page_size = page_size;
let salt = &hdr[16..24];
let (h0, h1) = checksum(big_endian, 0, 0, &hdr[0..24]);
if h0 != be32(&hdr, 24) || h1 != be32(&hdr, 28) {
return Err(Error::Corrupt("WAL header checksum mismatch".into()));
}
let frame_len = FRAME_HEADER_LEN + page_size;
let (mut s0, mut s1) = (h0, h1);
let mut off = WAL_HEADER_LEN as u64;
let mut pending: Vec<(u32, Vec<u8>)> = Vec::new();
while off + frame_len as u64 <= wal_size {
let mut fhdr = [0u8; FRAME_HEADER_LEN];
wal.read_exact_at(&mut fhdr, off)?;
let mut page = vec![0u8; page_size];
wal.read_exact_at(&mut page, off + FRAME_HEADER_LEN as u64)?;
if fhdr[8..16] != *salt {
break;
}
let (c0, c1) = checksum(big_endian, s0, s1, &fhdr[0..8]);
let (c0, c1) = checksum(big_endian, c0, c1, &page);
if c0 != be32(&fhdr, 16) || c1 != be32(&fhdr, 20) {
break; }
s0 = c0;
s1 = c1;
let page_no = be32(&fhdr, 0);
let db_size = be32(&fhdr, 4);
pending.push((page_no, page));
if db_size != 0 {
for (p, data) in pending.drain(..) {
frames.insert(p, data);
}
commit_db_size = db_size;
}
off += frame_len as u64;
}
}
let page1 = match frames.get(&1) {
Some(p) => p.clone(),
None => {
let mut buf = vec![0u8; HEADER_LEN.max(512)];
let want = main.size()?.min(buf.len() as u64) as usize;
buf.truncate(want);
main.read_exact_at(&mut buf, 0)?;
buf
}
};
let header = DatabaseHeader::parse(&page1)?;
let page_size = if wal_page_size != 0 {
wal_page_size
} else {
header.page_size as usize
};
let db_size = if commit_db_size != 0 {
commit_db_size
} else {
(main.size()? / page_size as u64) as u32
};
Ok(WalReader {
main,
header,
page_size,
db_size,
frames,
})
}
}
impl PageSource for WalReader {
fn page(&self, number: u32) -> Result<Page> {
if number == 0 || number > self.db_size {
return Err(Error::Corrupt(alloc::format!(
"page {number} out of range 1..={}",
self.db_size
)));
}
if let Some(data) = self.frames.get(&number) {
return Ok(Page::from_bytes(number, data.clone()));
}
let mut buf = vec![0u8; self.page_size];
self.main
.read_exact_at(&mut buf, (number as u64 - 1) * self.page_size as u64)?;
Ok(Page::from_bytes(number, buf))
}
fn header(&self) -> &DatabaseHeader {
&self.header
}
fn usable_size(&self) -> usize {
self.header.usable_size() as usize
}
fn page_count(&self) -> u32 {
self.db_size
}
}
#[inline]
fn be32(b: &[u8], at: usize) -> u32 {
u32::from_be_bytes([b[at], b[at + 1], b[at + 2], b[at + 3]])
}
pub type WriterState = (u64, (u32, u32), [u8; 8], u32);
#[derive(Clone)]
struct IndexFrame {
page_no: u32,
commit_db_size: u32,
data: Vec<u8>,
}
pub struct WalIndex {
log: Vec<IndexFrame>,
mx_frame: u32,
salt: [u8; 8],
next_offset: u64,
cksum: (u32, u32),
page_size: u32,
n_backfill: u32,
ckpt_seq: u32,
big_end_cksum: bool,
generation: u64,
readers: Vec<u32>,
}
impl WalIndex {
fn new() -> WalIndex {
WalIndex {
log: Vec::new(),
mx_frame: 0,
salt: [0; 8],
next_offset: 0,
cksum: (0, 0),
page_size: 0,
n_backfill: 0,
ckpt_seq: 0,
big_end_cksum: NATIVE_BIG_ENDIAN,
generation: 0,
readers: Vec::new(),
}
}
pub fn mx_frame(&self) -> u32 {
self.mx_frame
}
pub fn generation(&self) -> u64 {
self.generation
}
pub fn n_backfill(&self) -> u32 {
self.n_backfill
}
pub fn set_n_backfill(&mut self, n: u32) {
if n > self.n_backfill {
self.n_backfill = n;
}
}
pub fn ckpt_seq(&self) -> u32 {
self.ckpt_seq
}
pub fn salt(&self) -> [u8; 8] {
self.salt
}
pub fn big_end_cksum(&self) -> bool {
self.big_end_cksum
}
pub fn restart_hdr(&mut self) {
self.ckpt_seq = self.ckpt_seq.wrapping_add(1);
let s1 = u32::from_be_bytes([self.salt[0], self.salt[1], self.salt[2], self.salt[3]])
.wrapping_add(1);
self.salt[0..4].copy_from_slice(&s1.to_be_bytes());
let mut s2 = u32::from_be_bytes([self.salt[4], self.salt[5], self.salt[6], self.salt[7]]);
s2 ^= s2 << 13;
s2 ^= s2 >> 17;
s2 ^= s2 << 5;
s2 = s2.wrapping_add(0x9E37_79B9);
self.salt[4..8].copy_from_slice(&s2.to_be_bytes());
self.log.clear();
self.mx_frame = 0;
self.n_backfill = 0;
self.next_offset = 0;
self.cksum = (0, 0);
self.big_end_cksum = NATIVE_BIG_ENDIAN;
self.generation = self.generation.wrapping_add(1);
}
pub fn backfill_pages(&self, mx_safe: u32) -> Vec<(u32, Vec<u8>)> {
let lo = self.n_backfill as usize;
let hi = (self.mx_frame as usize).min(self.log.len());
if lo >= hi {
return Vec::new();
}
let mut newest: BTreeMap<u32, usize> = BTreeMap::new();
for (i, frame) in self.log[lo..hi].iter().enumerate() {
newest.insert(frame.page_no, lo + i);
}
newest
.into_iter()
.filter(|&(_, i)| (i as u32 + 1) <= mx_safe)
.map(|(page_no, i)| (page_no, self.log[i].data.clone()))
.collect()
}
pub fn writer_state(&self) -> Option<WriterState> {
if self.mx_frame == 0 {
None
} else {
Some((self.next_offset, self.cksum, self.salt, self.page_size))
}
}
pub fn is_empty(&self) -> bool {
self.mx_frame == 0
}
pub fn find_frame(&self, page_no: u32, snapshot_mx: u32) -> Option<Vec<u8>> {
let hi = (snapshot_mx as usize).min(self.log.len());
for frame in self.log[..hi].iter().rev() {
if frame.page_no == page_no {
return Some(frame.data.clone());
}
}
None
}
pub fn snapshot_db_size(&self, snapshot_mx: u32) -> Option<u32> {
let hi = (snapshot_mx as usize).min(self.log.len());
for frame in self.log[..hi].iter().rev() {
if frame.commit_db_size != 0 {
return Some(frame.commit_db_size);
}
}
None
}
pub fn reset(&mut self, salt: [u8; 8]) {
self.log.clear();
self.mx_frame = 0;
self.salt = salt;
self.next_offset = 0;
self.cksum = (0, 0);
self.page_size = 0;
self.n_backfill = 0;
self.ckpt_seq = 0;
self.big_end_cksum = NATIVE_BIG_ENDIAN;
self.generation = self.generation.wrapping_add(1);
}
pub fn register_reader(&mut self, mx_frame: u32) {
self.readers.push(mx_frame);
}
pub fn unregister_reader(&mut self, mx_frame: u32) {
if let Some(pos) = self.readers.iter().position(|&m| m == mx_frame) {
self.readers.swap_remove(pos);
}
}
pub fn min_reader_mark(&self) -> Option<u32> {
self.readers.iter().copied().min()
}
pub fn full_db_size(&self) -> u32 {
self.log
.iter()
.rev()
.find(|f| f.commit_db_size != 0)
.map(|f| f.commit_db_size)
.unwrap_or(0)
}
}
#[derive(Clone)]
pub enum SharedWalIndex {
#[cfg(not(feature = "std"))]
Local(Rc<RefCell<WalIndex>>),
#[cfg(feature = "std")]
Shared(std::sync::Arc<std::sync::Mutex<WalIndex>>),
}
impl SharedWalIndex {
pub fn new() -> SharedWalIndex {
#[cfg(feature = "std")]
{
SharedWalIndex::Shared(std::sync::Arc::new(std::sync::Mutex::new(WalIndex::new())))
}
#[cfg(not(feature = "std"))]
{
SharedWalIndex::Local(Rc::new(RefCell::new(WalIndex::new())))
}
}
pub fn with<R>(&self, f: impl FnOnce(&mut WalIndex) -> R) -> R {
match self {
#[cfg(not(feature = "std"))]
SharedWalIndex::Local(rc) => f(&mut rc.borrow_mut()),
#[cfg(feature = "std")]
SharedWalIndex::Shared(arc) => f(&mut arc.lock().expect("wal-index mutex poisoned")),
}
}
}
impl Default for SharedWalIndex {
fn default() -> Self {
Self::new()
}
}
#[derive(Clone, Copy)]
pub struct WalSnapshot {
pub mx_frame: u32,
pub generation: u64,
}
impl WalIndex {
pub fn snapshot(&self) -> WalSnapshot {
WalSnapshot {
mx_frame: self.mx_frame,
generation: self.generation,
}
}
#[allow(clippy::too_many_arguments)]
pub fn append(
&mut self,
page_no: u32,
commit_db_size: u32,
salt: [u8; 8],
data: Vec<u8>,
next_offset: u64,
cksum: (u32, u32),
page_size: u32,
) {
self.log.push(IndexFrame {
page_no,
commit_db_size,
data,
});
self.mx_frame = self.log.len() as u32;
self.salt = salt;
self.next_offset = next_offset;
self.cksum = cksum;
self.page_size = page_size;
}
#[allow(clippy::too_many_arguments)]
pub fn seed(
&mut self,
frames: Vec<(u32, u32, Vec<u8>)>,
salt: [u8; 8],
next_offset: u64,
cksum: (u32, u32),
page_size: u32,
ckpt_seq: u32,
big_end_cksum: bool,
) {
self.log = frames
.into_iter()
.map(|(page_no, commit_db_size, data)| IndexFrame {
page_no,
commit_db_size,
data,
})
.collect();
self.mx_frame = self.log.len() as u32;
self.salt = salt;
self.next_offset = next_offset;
self.cksum = cksum;
self.page_size = page_size;
self.n_backfill = 0;
self.ckpt_seq = ckpt_seq;
self.big_end_cksum = big_end_cksum;
}
}
pub(crate) fn checksum(big_endian: bool, mut s0: u32, mut s1: u32, data: &[u8]) -> (u32, u32) {
let read = |at: usize| -> u32 {
if big_endian {
u32::from_be_bytes([data[at], data[at + 1], data[at + 2], data[at + 3]])
} else {
u32::from_le_bytes([data[at], data[at + 1], data[at + 2], data[at + 3]])
}
};
let mut i = 0;
while i + 8 <= data.len() {
s0 = s0.wrapping_add(read(i)).wrapping_add(s1);
s1 = s1.wrapping_add(read(i + 4)).wrapping_add(s0);
i += 8;
}
(s0, s1)
}
#[cfg(all(test, feature = "std"))]
const _: fn() = || {
fn assert<T: Send + Sync + std::panic::RefUnwindSafe>() {}
assert::<SharedWalIndex>();
};