reddb_engine/storage/
pager_io.rs

1//! Byte I/O shared by the native pager and exclusively owned browser handles.
2//!
3//! The owner supplies locking, read-only policy, WAL-first ordering, allocation
4//! fences and encryption. It must stop using a handle after failed I/O and
5//! reopen/recover before resuming. These functions do not form a database API.
6
7#![deny(
8    clippy::unwrap_used,
9    clippy::too_many_lines,
10    clippy::cast_possible_truncation
11)]
12
13mod staging;
14pub(crate) use staging::StagedPageBatch;
15
16use super::page::{Page, PageError, PageType, PAGE_SIZE};
17use reddb_file::{VfsFile, PAGED_SUPERBLOCK_SLOT_SIZE, PAGED_SUPERBLOCK_ZONE_SIZE};
18use std::io::{self, SeekFrom};
19
20mod dwb_publication;
21
22pub(crate) fn requires_dwb_publication<F: VfsFile>(file: &mut F) -> Result<bool, PagerIoError> {
23    dwb_publication::enabled(file)
24}
25
26pub const DWB_ZONE_START_PAGE: u32 = 3;
27pub const DWB_ZONE_PAGES: u32 = 64;
28pub const DWB_ZONE_BYTES: usize = DWB_ZONE_PAGES as usize * PAGE_SIZE;
29pub const FIRST_ALLOCATABLE_PAGE: u32 = DWB_ZONE_START_PAGE + DWB_ZONE_PAGES;
30pub const DWB_MAX_PAGES_PER_FRAME: usize =
31    (DWB_ZONE_BYTES - reddb_file::PAGED_DWB_HEADER_SIZE) / reddb_file::PAGED_DWB_ENTRY_SIZE;
32
33#[derive(Debug)]
34pub enum PagerIoError {
35    Io(io::Error),
36    Page(PageError),
37    InvalidDatabase(String),
38}
39
40impl std::fmt::Display for PagerIoError {
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        match self {
43            Self::Io(error) => write!(f, "I/O error: {error}"),
44            Self::Page(error) => write!(f, "Page error: {error}"),
45            Self::InvalidDatabase(message) => write!(f, "Invalid database: {message}"),
46        }
47    }
48}
49
50impl std::error::Error for PagerIoError {
51    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
52        match self {
53            Self::Io(error) => Some(error),
54            Self::Page(error) => Some(error),
55            Self::InvalidDatabase(_) => None,
56        }
57    }
58}
59
60impl From<io::Error> for PagerIoError {
61    fn from(error: io::Error) -> Self {
62        Self::Io(error)
63    }
64}
65
66impl From<PageError> for PagerIoError {
67    fn from(error: PageError) -> Self {
68        Self::Page(error)
69    }
70}
71
72pub fn read_page<F: VfsFile>(
73    file: &mut F,
74    page_id: u32,
75    verify_checksum: bool,
76) -> Result<Page, PagerIoError> {
77    file.seek(SeekFrom::Start(u64::from(page_id) * PAGE_SIZE as u64))?;
78    let mut buffer = [0u8; PAGE_SIZE];
79    file.read_exact(&mut buffer)?;
80    let page = Page::from_bytes(buffer);
81    // Page 0 contains independently sealed superblocks, not a page checksum.
82    if verify_checksum && page_id != 0 {
83        page.verify_checksum()?;
84        check_page_identity(&page, page_id)?;
85    }
86    Ok(page)
87}
88
89/// Check plaintext identity only; encrypted envelopes are authenticated by the host.
90pub fn check_page_identity(page: &Page, expected: u32) -> Result<(), PagerIoError> {
91    if expected != 0 && page.page_id() != expected {
92        return Err(PagerIoError::InvalidDatabase(format!(
93            "page identity mismatch: requested {expected}, found {}",
94            page.page_id()
95        )));
96    }
97    Ok(())
98}
99
100/// Raw write, without a durability barrier. The owner controls publication.
101pub fn write_page<F: VfsFile>(file: &mut F, page_id: u32, page: &Page) -> Result<(), PagerIoError> {
102    file.seek(SeekFrom::Start(u64::from(page_id) * PAGE_SIZE as u64))?;
103    file.write_all(page.as_bytes())?;
104    Ok(())
105}
106
107pub fn read_superblock_zone<F: VfsFile>(
108    file: &mut F,
109) -> Result<[u8; PAGED_SUPERBLOCK_ZONE_SIZE], PagerIoError> {
110    let len = file.file_len()?;
111    let mut zone = [0u8; PAGED_SUPERBLOCK_ZONE_SIZE];
112    let readable = usize::try_from(len.min(PAGED_SUPERBLOCK_ZONE_SIZE as u64))
113        .expect("readable prefix is bounded by the superblock zone");
114    if readable > 0 {
115        file.seek(SeekFrom::Start(0))?;
116        file.read_exact(&mut zone[..readable])?;
117    }
118    Ok(zone)
119}
120
121/// Seal and sync only the stale slot; the owner selects it and its generation.
122pub fn write_superblock_slot<F: VfsFile>(
123    file: &mut F,
124    slot: &mut [u8; PAGED_SUPERBLOCK_SLOT_SIZE],
125    copy_index: usize,
126    generation: u64,
127) -> Result<(), PagerIoError> {
128    reddb_file::seal_paged_superblock_slot(slot, copy_index, generation)
129        .map_err(|error| PagerIoError::InvalidDatabase(error.to_string()))?;
130    file.seek(SeekFrom::Start(reddb_file::paged_superblock_slot_offset(
131        copy_index,
132    )))?;
133    file.write_all(slot)?;
134    file.sync_all()?;
135    Ok(())
136}
137
138pub fn dwb_zone_offset() -> u64 {
139    u64::from(DWB_ZONE_START_PAGE) * PAGE_SIZE as u64
140}
141
142/// Clear the in-file DWB zone with valid reserved pages and sync it.
143pub fn clear_dwb_zone<F: VfsFile>(file: &mut F) -> Result<(), PagerIoError> {
144    for page_id in DWB_ZONE_START_PAGE..FIRST_ALLOCATABLE_PAGE {
145        let mut page = Page::new(PageType::Header, page_id);
146        page.update_checksum();
147        write_page(file, page_id, &page)?;
148    }
149    file.sync_all()?;
150    Ok(())
151}
152
153pub fn dwb_zone_is_clear(buffer: &[u8]) -> bool {
154    if buffer.len() != DWB_ZONE_BYTES {
155        return false;
156    }
157    for (index, chunk) in buffer.chunks_exact(PAGE_SIZE).enumerate() {
158        let Ok(page) = Page::from_slice(chunk) else {
159            return false;
160        };
161        if page.verify_checksum().is_err()
162            || page.page_id()
163                != DWB_ZONE_START_PAGE
164                    + u32::try_from(index).expect("DWB page index is bounded by its zone")
165            || page.page_type().ok() != Some(PageType::Header)
166        {
167            return false;
168        }
169    }
170    true
171}
172
173/// The caller must make the batch's WAL durable before entering this function.
174/// With DWB enabled: stage+sync and home pages+sync for each bounded frame,
175/// then clear+sync once. With DWB disabled, the caller's proven atomic-write
176/// policy applies.
177pub fn write_pages<F: VfsFile>(
178    file: &mut F,
179    pages: &[(u32, Page)],
180    double_write: bool,
181) -> Result<(), PagerIoError> {
182    if !double_write {
183        for (page_id, page) in pages {
184            write_page(file, *page_id, page)?;
185        }
186        return Ok(());
187    }
188
189    let protected = dwb_publication::enabled(file)?;
190    for chunk in pages.chunks(DWB_MAX_PAGES_PER_FRAME.max(1)) {
191        let buffer = reddb_file::encode_paged_dwb_frame(
192            chunk
193                .iter()
194                .map(|(page_id, page)| (*page_id, page.as_bytes())),
195        );
196        assert!(buffer.len() <= DWB_ZONE_BYTES, "DWB frame fits its zone");
197        let mut publication = protected.then(|| dwb_publication::publication(&buffer));
198        if protected {
199            let mut seen = std::collections::HashSet::new();
200            if chunk.iter().any(|(id, _)| {
201                *id == 0
202                    || (DWB_ZONE_START_PAGE..FIRST_ALLOCATABLE_PAGE).contains(id)
203                    || !seen.insert(*id)
204            }) {
205                return Err(PagerIoError::InvalidDatabase(
206                    "invalid DWB publication ownership".into(),
207                ));
208            }
209        }
210        if let Some(publication) = publication {
211            dwb_publication::publish(file, publication)?;
212        }
213        file.seek(SeekFrom::Start(dwb_zone_offset()))?;
214        file.write_all(&buffer)?;
215        file.sync_all()?;
216        if let Some(publication) = publication.as_mut() {
217            publication.phase = reddb_file::file_format::DwbPhase::Ready;
218            dwb_publication::publish(file, *publication)?;
219        }
220        for (page_id, page) in chunk {
221            write_page(file, *page_id, page)?;
222        }
223        file.sync_all()?;
224        if let Some(publication) = publication {
225            dwb_publication::consume(file, publication)?;
226        }
227    }
228    // Legacy frames' home pages are durable before the next frame replaces their
229    // staging image. An interrupted replacement can replay the old frame or
230    // discard an incomplete new one: neither can lose those durable homes.
231    // Only the final legacy frame needs clearing before returning to the owner.
232    // Guarded frames fence Consumed and clear before the next Preparing phase.
233    if !protected && !pages.is_empty() {
234        clear_dwb_zone(file)?;
235    }
236    Ok(())
237}
238
239fn has_complete_dwb_zone<F: VfsFile>(file: &mut F) -> Result<bool, PagerIoError> {
240    if file.file_len()? >= dwb_zone_offset() + DWB_ZONE_BYTES as u64 {
241        return Ok(true);
242    }
243    // The complete reserved zone precedes the guarded capability fence. A
244    // shorter file lost durable bytes; it is not an interrupted stage/clear.
245    if dwb_publication::enabled(file)? {
246        return Err(PagerIoError::InvalidDatabase(
247            "DWB generation authority: reserved zone is truncated".into(),
248        ));
249    }
250    Ok(false)
251}
252
253/// Replay a complete staged DWB frame before allowing page access. Incomplete
254/// staging is discarded as on the native path: home writes start only after
255/// the complete frame crossed its barrier. No database root is recreated.
256pub fn recover_from_dwb<F: VfsFile>(file: &mut F) -> Result<(), PagerIoError> {
257    if !has_complete_dwb_zone(file)? {
258        return Ok(());
259    }
260    if !dwb_publication::enabled(file)? {
261        return staging::recover(file);
262    }
263    let buffer = read_guarded_dwb_zone(file)?;
264    let plan = dwb_publication::plan(file, &buffer)?.ok_or_else(|| {
265        PagerIoError::InvalidDatabase("DWB generation authority disappeared during recovery".into())
266    })?;
267    let authority = match plan {
268        dwb_publication::Recovery::Unchanged => return Ok(()),
269        dwb_publication::Recovery::Discard(authority) => authority,
270        dwb_publication::Recovery::Apply(authority) => {
271            // Validate the independent digest and every address before the
272            // first home write. Repeat both Ready fences even when recovery
273            // selected a Preparing fallback after an interrupted publication.
274            let mut ready = authority;
275            ready.phase = reddb_file::file_format::DwbPhase::Ready;
276            dwb_publication::publish(file, ready)?;
277            let entries = reddb_file::decode_paged_dwb_frame(&buffer)
278                .map_err(|error| PagerIoError::InvalidDatabase(error.to_string()))?;
279            for entry in entries {
280                write_page(file, entry.page_id, &Page::from_bytes(entry.page))?;
281            }
282            file.sync_all()?;
283            authority
284        }
285    };
286    dwb_publication::consume(file, authority)
287}
288
289/// Read a complete staged frame without consuming or repairing the source.
290/// Incomplete staging cannot have reached home pages and is ignored, just as
291/// in writer recovery. The frame decoder bounds the number of pages.
292pub fn read_dwb_pages<F: VfsFile>(file: &mut F) -> Result<Vec<(u32, Page)>, PagerIoError> {
293    if !has_complete_dwb_zone(file)? {
294        return Ok(Vec::new());
295    }
296    if !dwb_publication::enabled(file)? {
297        return staging::read_pages(file);
298    }
299    let buffer = read_guarded_dwb_zone(file)?;
300    let plan = dwb_publication::plan(file, &buffer)?.ok_or_else(|| {
301        PagerIoError::InvalidDatabase("DWB generation authority disappeared during recovery".into())
302    })?;
303    if !matches!(plan, dwb_publication::Recovery::Apply(_)) {
304        return Ok(Vec::new());
305    }
306    Ok(reddb_file::decode_paged_dwb_frame(&buffer)
307        .map_err(|error| PagerIoError::InvalidDatabase(error.to_string()))?
308        .into_iter()
309        .map(|entry| (entry.page_id, Page::from_bytes(entry.page)))
310        .collect())
311}
312
313fn read_guarded_dwb_zone<F: VfsFile>(file: &mut F) -> Result<Vec<u8>, PagerIoError> {
314    if DWB_ZONE_BYTES > file.max_read_buffer_bytes() {
315        return Err(PagerIoError::InvalidDatabase(
316            "guarded double-write recovery exceeds the owned read budget".into(),
317        ));
318    }
319    let mut buffer = vec![0u8; DWB_ZONE_BYTES];
320    file.seek(SeekFrom::Start(dwb_zone_offset()))?;
321    file.read_exact(&mut buffer)?;
322    Ok(buffer)
323}
reddb_engine/storage/pager_io.rs

reddb_engine/storage/
pager_io.rs
