reddb_engine/storage/
overflow.rs

1//! Overflow chain storage
2//!
3//! Owns chains of dedicated overflow pages (PageType::Overflow), independent of
4//! B-tree semantics and MVCC. Backs the large-value spill path from ADR 0023
5//! (slice B of PRD #662).
6//!
7//! Each overflow page lays a small chain-local header at the start of its
8//! content area:
9//!
10//! ```text
11//! Offset  Size  Field
12//! ------  ----  -----
13//!   0      4    next_page_id (u32 LE, 0 = tail sentinel)
14//!   4      4    payload_len  (u32 LE, payload bytes on this page)
15//!   8     ..    payload bytes
16//! ```
17//!
18//! `version` (per ADR 0025) is added in slice C (#700); this slice only owns
19//! the chain mechanics — allocate, link, walk, free.
20//!
21//! Whole-value materialisation only — no streaming reads.
22
23use super::page::{Page, PageType, CONTENT_SIZE, HEADER_SIZE, PAGE_SIZE};
24use super::page_store::PageStore;
25use super::pager::PagerError;
26
27/// Bytes consumed by the per-page chain header (next + payload_len).
28pub const OVERFLOW_PAGE_HEADER: usize = 8;
29
30/// Payload bytes that fit on a single overflow page.
31pub const OVERFLOW_PAYLOAD_PER_PAGE: usize = CONTENT_SIZE - OVERFLOW_PAGE_HEADER;
32
33const _: () = assert!(OVERFLOW_PAYLOAD_PER_PAGE == PAGE_SIZE - HEADER_SIZE - OVERFLOW_PAGE_HEADER);
34
35/// Errors returned by overflow-chain operations.
36#[derive(Debug)]
37pub enum OverflowError {
38    /// Underlying pager call failed.
39    Pager(PagerError),
40    /// A page reached while walking the chain is not an Overflow page.
41    NotOverflowPage { page_id: u32 },
42    /// A page declared a payload longer than the per-page capacity.
43    PayloadTooLarge { page_id: u32, len: u32 },
44    /// The advertised `total_len` disagrees with what the chain actually holds.
45    LengthMismatch { expected: u64, actual: u64 },
46    /// Caller asked to free a non-overflow page as a chain head.
47    InvalidHead { page_id: u32 },
48}
49
50impl std::fmt::Display for OverflowError {
51    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
52        match self {
53            Self::Pager(e) => write!(f, "pager error: {}", e),
54            Self::NotOverflowPage { page_id } => {
55                write!(f, "page {} is not an overflow page", page_id)
56            }
57            Self::PayloadTooLarge { page_id, len } => {
58                write!(
59                    f,
60                    "overflow page {} declares payload_len {} (max {})",
61                    page_id, len, OVERFLOW_PAYLOAD_PER_PAGE
62                )
63            }
64            Self::LengthMismatch { expected, actual } => write!(
65                f,
66                "overflow chain length mismatch: expected {} bytes, chain holds {}",
67                expected, actual
68            ),
69            Self::InvalidHead { page_id } => {
70                write!(f, "free called with non-overflow head page {}", page_id)
71            }
72        }
73    }
74}
75
76impl std::error::Error for OverflowError {}
77
78impl From<PagerError> for OverflowError {
79    fn from(e: PagerError) -> Self {
80        Self::Pager(e)
81    }
82}
83
84/// Number of overflow pages required to hold `len` bytes.
85pub fn pages_needed(len: usize) -> usize {
86    if len == 0 {
87        1
88    } else {
89        len.div_ceil(OVERFLOW_PAYLOAD_PER_PAGE)
90    }
91}
92
93fn read_chain_header(page: &Page) -> Result<(u32, u32), OverflowError> {
94    if page.page_type().map_err(PagerError::from)? != PageType::Overflow {
95        return Err(OverflowError::NotOverflowPage {
96            page_id: page.page_id(),
97        });
98    }
99    let content = page.content();
100    let next = u32::from_le_bytes([content[0], content[1], content[2], content[3]]);
101    let len = u32::from_le_bytes([content[4], content[5], content[6], content[7]]);
102    if len as usize > OVERFLOW_PAYLOAD_PER_PAGE {
103        return Err(OverflowError::PayloadTooLarge {
104            page_id: page.page_id(),
105            len,
106        });
107    }
108    Ok((next, len))
109}
110
111fn write_chain_header(page: &mut Page, next: u32, payload_len: u32) {
112    let content = page.content_mut();
113    content[0..4].copy_from_slice(&next.to_le_bytes());
114    content[4..8].copy_from_slice(&payload_len.to_le_bytes());
115}
116
117/// Chain operations over a pager.
118///
119/// Held by value rather than as `impl Pager` methods to keep overflow concerns
120/// out of `engine::pager`. Slice E will hold one of these from inside the
121/// B-tree write path.
122pub struct OverflowChain<'p, P: PageStore> {
123    pager: &'p P,
124}
125
126impl<'p, P: PageStore> OverflowChain<'p, P> {
127    pub fn new(pager: &'p P) -> Self {
128        Self { pager }
129    }
130
131    /// Allocate enough overflow pages to hold `bytes`, link them, and return
132    /// the head page id together with the total length.
133    ///
134    /// Empty input still produces a single zero-length head page so that the
135    /// leaf-side pointer is always valid.
136    pub fn store(&self, bytes: &[u8]) -> Result<(u32, u64), OverflowError> {
137        let total_len = bytes.len() as u64;
138        let n_pages = pages_needed(bytes.len());
139
140        let mut page_ids = Vec::with_capacity(n_pages);
141        for _ in 0..n_pages {
142            let page = self.pager.allocate_page(PageType::Overflow)?;
143            page_ids.push(page.page_id());
144        }
145
146        let mut offset = 0usize;
147        for (i, &pid) in page_ids.iter().enumerate() {
148            let next = if i + 1 < page_ids.len() {
149                page_ids[i + 1]
150            } else {
151                0
152            };
153            let chunk_end = (offset + OVERFLOW_PAYLOAD_PER_PAGE).min(bytes.len());
154            let chunk = &bytes[offset..chunk_end];
155            offset = chunk_end;
156
157            let mut page = Page::new(PageType::Overflow, pid);
158            write_chain_header(&mut page, next, chunk.len() as u32);
159            page.content_mut()[OVERFLOW_PAGE_HEADER..OVERFLOW_PAGE_HEADER + chunk.len()]
160                .copy_from_slice(chunk);
161
162            self.pager.write_page(pid, page)?;
163        }
164
165        Ok((page_ids[0], total_len))
166    }
167
168    /// Walk the chain and return the concatenated payload.
169    ///
170    /// `total_len` must match the actual bytes carried by the chain; if it
171    /// does not, a `LengthMismatch` error is returned (no truncation, no
172    /// silent extension).
173    pub fn read(&self, head_page_id: u32, total_len: u64) -> Result<Vec<u8>, OverflowError> {
174        let expected = total_len as usize;
175        let mut out = Vec::with_capacity(expected);
176        let mut current = head_page_id;
177        let mut collected: u64 = 0;
178
179        while current != 0 {
180            let page = self.pager.read_page(current).map_err(OverflowError::from)?;
181            let (next, len) = read_chain_header(&page)?;
182            let len_usize = len as usize;
183            collected += len as u64;
184
185            if collected > total_len {
186                return Err(OverflowError::LengthMismatch {
187                    expected: total_len,
188                    actual: collected,
189                });
190            }
191
192            let payload = &page.content()[OVERFLOW_PAGE_HEADER..OVERFLOW_PAGE_HEADER + len_usize];
193            out.extend_from_slice(payload);
194            current = next;
195        }
196
197        if collected != total_len {
198            return Err(OverflowError::LengthMismatch {
199                expected: total_len,
200                actual: collected,
201            });
202        }
203
204        Ok(out)
205    }
206
207    /// Walk the chain starting at `head_page_id` and return every page to the
208    /// free list.
209    pub fn free(&self, head_page_id: u32) -> Result<(), OverflowError> {
210        let mut current = head_page_id;
211        let mut first = true;
212        while current != 0 {
213            let page = self.pager.read_page(current).map_err(OverflowError::from)?;
214            if page.page_type().map_err(PagerError::from)? != PageType::Overflow {
215                return Err(if first {
216                    OverflowError::InvalidHead { page_id: current }
217                } else {
218                    OverflowError::NotOverflowPage { page_id: current }
219                });
220            }
221            let (next, _) = read_chain_header(&page)?;
222            self.pager.free_page(current)?;
223            current = next;
224            first = false;
225        }
226        Ok(())
227    }
228}
reddb_engine/storage/overflow.rs

reddb_engine/storage/
overflow.rs
