mq_db/storage/
page.rs

1use std::{
2    fs::{File, OpenOptions},
3    io::{Read, Seek, SeekFrom, Write},
4    path::Path,
5};
6
7use crate::error::MqdbError;
8
9pub const PAGE_SIZE: usize = 8192;
10pub const PAGE_HEADER_SIZE: usize = 16;
11pub const PAGE_BODY_SIZE: usize = PAGE_SIZE - PAGE_HEADER_SIZE;
12
13pub const PAGE_TYPE_FREE: u32 = 0;
14pub(crate) const PAGE_TYPE_FILE_HEADER: u32 = 1;
15pub(crate) const PAGE_TYPE_CATALOG: u32 = 2;
16pub(crate) const PAGE_TYPE_BLOCK_DATA: u32 = 3;
17pub(crate) const PAGE_TYPE_OVERFLOW: u32 = 4;
18pub(crate) const PAGE_TYPE_INDEX: u32 = 5;
19pub(crate) const PAGE_TYPE_TABLE_DATA: u32 = 6;
20
21const FILE_MAGIC: u32 = 0x4D51_4442;
22pub const FILE_VERSION: u32 = 5;
23const LEGACY_VERSIONS: &[u32] = &[4];
24const CATALOG_START_PAGE: u32 = 1;
25
26fn invalid_data(message: impl Into<String>) -> MqdbError {
27    MqdbError::Storage(message.into())
28}
29
30fn file_header_body(num_pages: u32) -> [u8; PAGE_BODY_SIZE] {
31    let mut body = [0u8; PAGE_BODY_SIZE];
32    body[0..4].copy_from_slice(&FILE_MAGIC.to_le_bytes());
33    body[4..8].copy_from_slice(&FILE_VERSION.to_le_bytes());
34    body[8..12].copy_from_slice(&(PAGE_SIZE as u32).to_le_bytes());
35    body[12..16].copy_from_slice(&num_pages.to_le_bytes());
36    body[16..20].copy_from_slice(&CATALOG_START_PAGE.to_le_bytes());
37    body
38}
39
40pub fn compute_checksum(page: &[u8; PAGE_SIZE]) -> u32 {
41    let mut checksum = 0u32;
42    for byte in &page[0..4] {
43        checksum = checksum.wrapping_add(u32::from(*byte));
44    }
45    for byte in &page[8..PAGE_SIZE] {
46        checksum = checksum.wrapping_add(u32::from(*byte));
47    }
48    checksum
49}
50
51pub fn verify_checksum(page: &[u8; PAGE_SIZE]) -> bool {
52    let (_, checksum, _, _) = parse_page_header(page);
53    checksum == compute_checksum(page)
54}
55
56pub fn parse_page_header(page: &[u8; PAGE_SIZE]) -> (u32, u32, u32, u32) {
57    let page_type = u32::from_le_bytes(page[0..4].try_into().expect("page type slice"));
58    let checksum = u32::from_le_bytes(page[4..8].try_into().expect("checksum slice"));
59    let page_id = u32::from_le_bytes(page[8..12].try_into().expect("page id slice"));
60    let next_page = u32::from_le_bytes(page[12..16].try_into().expect("next page slice"));
61    (page_type, checksum, page_id, next_page)
62}
63
64pub fn make_page(page_type: u32, page_id: u32, next_page: u32, body: &[u8]) -> [u8; PAGE_SIZE] {
65    let mut page = [0u8; PAGE_SIZE];
66    page[0..4].copy_from_slice(&page_type.to_le_bytes());
67    page[8..12].copy_from_slice(&page_id.to_le_bytes());
68    page[12..16].copy_from_slice(&next_page.to_le_bytes());
69
70    let copy_len = body.len().min(PAGE_BODY_SIZE);
71    page[PAGE_HEADER_SIZE..PAGE_HEADER_SIZE + copy_len].copy_from_slice(&body[..copy_len]);
72
73    let checksum = compute_checksum(&page);
74    page[4..8].copy_from_slice(&checksum.to_le_bytes());
75    page
76}
77
78pub struct PageFile {
79    file: File,
80    pub num_pages: u32,
81    /// `true` if `num_pages` has advanced since the header page was last
82    /// written to disk; `append_page` no longer writes it eagerly.
83    header_dirty: bool,
84    /// File-format version read from the header. `FILE_VERSION` for files
85    /// created by this build; an older value if `open` accepted a legacy
86    /// version (see `LEGACY_VERSIONS`).
87    pub version: u32,
88}
89
90impl PageFile {
91    pub fn create(path: &Path) -> Result<Self, MqdbError> {
92        let file = OpenOptions::new()
93            .read(true)
94            .write(true)
95            .create(true)
96            .truncate(true)
97            .open(path)?;
98        let mut page_file = Self {
99            file,
100            num_pages: 1,
101            header_dirty: false,
102            version: FILE_VERSION,
103        };
104        page_file.write_file_header()?;
105        Ok(page_file)
106    }
107
108    pub fn open(path: &Path) -> Result<Self, MqdbError> {
109        let mut file = OpenOptions::new().read(true).write(true).open(path)?;
110        let mut page = [0u8; PAGE_SIZE];
111        file.seek(SeekFrom::Start(0))?;
112        file.read_exact(&mut page)?;
113
114        if !verify_checksum(&page) {
115            return Err(invalid_data("invalid file header checksum"));
116        }
117
118        let (page_type, _, page_id, _) = parse_page_header(&page);
119        if page_type != PAGE_TYPE_FILE_HEADER || page_id != 0 {
120            return Err(invalid_data("page 0 is not a valid file header page"));
121        }
122
123        let body = &page[PAGE_HEADER_SIZE..];
124        let magic = u32::from_le_bytes(body[0..4].try_into().expect("magic slice"));
125        let version = u32::from_le_bytes(body[4..8].try_into().expect("version slice"));
126        let page_size = u32::from_le_bytes(body[8..12].try_into().expect("page size slice"));
127        let num_pages = u32::from_le_bytes(body[12..16].try_into().expect("num pages slice"));
128        let catalog_start = u32::from_le_bytes(body[16..20].try_into().expect("catalog slice"));
129
130        if magic != FILE_MAGIC {
131            return Err(invalid_data("invalid MQDB magic number"));
132        }
133        if version != FILE_VERSION && !LEGACY_VERSIONS.contains(&version) {
134            return Err(invalid_data(format!(
135                "unsupported file version {version} (expected {FILE_VERSION}); run `mq-db index` to recreate the store"
136            )));
137        }
138        if page_size != PAGE_SIZE as u32 {
139            return Err(invalid_data(format!("unsupported page size: {page_size}")));
140        }
141        if catalog_start != CATALOG_START_PAGE {
142            return Err(invalid_data(format!(
143                "unexpected catalog start page: {catalog_start}"
144            )));
145        }
146        if num_pages == 0 {
147            return Err(invalid_data("invalid page count 0 in file header"));
148        }
149
150        let file_len = file.metadata()?.len();
151        if file_len % PAGE_SIZE as u64 != 0 {
152            return Err(invalid_data(
153                "database file size is not aligned to page size",
154            ));
155        }
156        if file_len < u64::from(num_pages) * PAGE_SIZE as u64 {
157            return Err(invalid_data(
158                "database file is shorter than file header page count",
159            ));
160        }
161
162        Ok(Self {
163            file,
164            num_pages,
165            header_dirty: false,
166            version,
167        })
168    }
169
170    pub fn read_page(&mut self, page_id: u32) -> Result<[u8; PAGE_SIZE], MqdbError> {
171        if page_id >= self.num_pages {
172            return Err(invalid_data(format!("page out of bounds: {page_id}")));
173        }
174
175        let mut page = [0u8; PAGE_SIZE];
176        self.file
177            .seek(SeekFrom::Start(u64::from(page_id) * PAGE_SIZE as u64))?;
178        self.file.read_exact(&mut page)?;
179
180        if !verify_checksum(&page) {
181            return Err(invalid_data(format!(
182                "checksum mismatch for page {page_id}"
183            )));
184        }
185
186        let (_, _, stored_page_id, _) = parse_page_header(&page);
187        if stored_page_id != page_id {
188            return Err(invalid_data(format!(
189                "page header id mismatch: expected {page_id}, found {stored_page_id}"
190            )));
191        }
192
193        Ok(page)
194    }
195
196    pub fn write_page(&mut self, page_id: u32, data: &[u8; PAGE_SIZE]) -> Result<(), MqdbError> {
197        if page_id >= self.num_pages {
198            return Err(invalid_data(format!("page out of bounds: {page_id}")));
199        }
200
201        self.file
202            .seek(SeekFrom::Start(u64::from(page_id) * PAGE_SIZE as u64))?;
203        self.file.write_all(data)?;
204        Ok(())
205    }
206
207    pub fn append_page(&mut self, data: &[u8; PAGE_SIZE]) -> Result<u32, MqdbError> {
208        let page_id = self.num_pages;
209        self.file.seek(SeekFrom::End(0))?;
210        self.file.write_all(data)?;
211        self.num_pages = self
212            .num_pages
213            .checked_add(1)
214            .ok_or_else(|| invalid_data("page count overflow"))?;
215        self.header_dirty = true;
216        Ok(page_id)
217    }
218
219    /// Persists `num_pages` if it changed since the last write. Must run
220    /// before the file is reopened from disk (see `Storage::flush_catalog`).
221    pub fn sync_header(&mut self) -> Result<(), MqdbError> {
222        if self.header_dirty {
223            self.write_file_header()?;
224            self.header_dirty = false;
225        }
226        Ok(())
227    }
228
229    fn write_file_header(&mut self) -> Result<(), MqdbError> {
230        let body = file_header_body(self.num_pages);
231        let page = make_page(PAGE_TYPE_FILE_HEADER, 0, 0, &body);
232        self.file.seek(SeekFrom::Start(0))?;
233        self.file.write_all(&page)?;
234        Ok(())
235    }
236}
mq_db/storage/page.rs

mq_db/storage/
page.rs
