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

reddb_engine/storage/
overflow.rs
