mentedb_storage/
memory_map.rs

1//! Record map: which extent holds which memory, persisted at checkpoints.
2//!
3//! The engine keeps the whole locator to extent table in memory; every read
4//! resolves its locator through it. The table is persisted at checkpoint time
5//! and replayed forward from the WAL on open, so a cold open reads no records.
6//!
7//! # Anchoring
8//!
9//! Each checkpoint that changed the table appends a `Checkpoint` WAL entry
10//! whose payload is a random 16 byte token, then writes the snapshot stamped
11//! with the same token, then truncates the WAL to that entry. The table now is
12//! the snapshot plus every WAL entry after the entry carrying its token. A
13//! snapshot whose token is not in the WAL can never be matched again, because
14//! tokens are never reused for a changed table. On any doubt (missing,
15//! corrupt, unmatched, inconsistent with the file) open falls back to a scan of
16//! the record chain.
17//!
18//! Each entry also carries a stamp, a value refreshed on every write to the
19//! record, which lets derived caches (mode probes, salience bounds) persist
20//! alongside and detect, without loading the record, that it changed since
21//! they were captured.
22
23use std::collections::BTreeMap;
24use std::io::Write;
25use std::path::Path;
26
27use mentedb_core::error::{MenteError, MenteResult};
28use mentedb_core::types::MemoryId;
29
30use crate::record::{Extent, FreeSpace, PageId, RecordOp, le_u32, le_u64};
31
32/// File name of the persisted record map inside the database directory.
33pub const MEMORY_MAP_FILE: &str = "records_map.bin";
34
35const MAGIC: &[u8; 8] = b"MDBRMAP1";
36const FORMAT_VERSION: u32 = 1;
37
38/// Checkpoint token carried in the `Checkpoint` WAL entry payload.
39pub(crate) type Token = [u8; 16];
40
41/// A memory the map knows about: its id, its locator, and its write stamp.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub struct MappedMemory {
44    pub id: MemoryId,
45    pub page: PageId,
46    /// Changes whenever the record is written. Equal stamps for the same
47    /// locator mean no write was applied to it in between.
48    pub stamp: u64,
49}
50
51/// A fresh random stamp.
52pub(crate) fn random_u64() -> u64 {
53    let (a, b) = uuid::Uuid::new_v4().as_u64_pair();
54    a ^ b.rotate_left(17)
55}
56
57pub(crate) fn new_token() -> Token {
58    *uuid::Uuid::new_v4().as_bytes()
59}
60
61/// The stamp of the WAL entry at `lsn` written while `token` anchored the
62/// map. Deterministic, so the stamp a write gets live and the stamp a reopen
63/// gives it replaying the WAL are the same value. LSNs only grow while a
64/// token anchors the map, so (token, lsn) never repeats.
65pub(crate) fn entry_stamp(token: &Token, lsn: u64) -> u64 {
66    fn mix(mut z: u64) -> u64 {
67        z = z.wrapping_add(0x9E37_79B9_7F4A_7C15);
68        z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9);
69        z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB);
70        z ^ (z >> 31)
71    }
72    let a = le_u64(&token[..8]);
73    let b = le_u64(&token[8..]);
74    mix(a ^ mix(b ^ mix(lsn)))
75}
76
77/// Where one record lives and what it holds.
78#[derive(Debug, Clone, Copy, PartialEq, Eq)]
79pub(crate) struct RecordEntry {
80    pub(crate) id: MemoryId,
81    pub(crate) stamp: u64,
82    pub(crate) extent: Extent,
83}
84
85/// The locator to extent table.
86#[derive(Debug, Clone, Default, PartialEq, Eq)]
87pub(crate) struct RecordTable {
88    pub(crate) records: BTreeMap<u64, RecordEntry>,
89    /// Next locator to hand out; above every locator ever used.
90    pub(crate) next_locator: u64,
91}
92
93impl RecordTable {
94    /// Apply one logged operation with the given stamp.
95    pub(crate) fn apply(&mut self, op: &RecordOp, stamp: u64) {
96        match op {
97            RecordOp::Write {
98                locator,
99                id,
100                extent,
101                ..
102            } => {
103                self.records.insert(
104                    *locator,
105                    RecordEntry {
106                        id: *id,
107                        stamp,
108                        extent: *extent,
109                    },
110                );
111            }
112            RecordOp::Free { locator, .. } => {
113                self.records.remove(locator);
114            }
115        }
116        self.next_locator = self.next_locator.max(op.locator() + 1);
117    }
118
119    /// Every mapped memory in locator order.
120    pub(crate) fn memories(&self) -> Vec<MappedMemory> {
121        self.records
122            .iter()
123            .map(|(&page, e)| MappedMemory {
124                id: e.id,
125                page: PageId(page),
126                stamp: e.stamp,
127            })
128            .collect()
129    }
130
131    /// Highest block any record occupies, exclusive (1 when empty).
132    pub(crate) fn max_end(&self) -> u64 {
133        self.records
134            .values()
135            .map(|e| e.extent.end())
136            .max()
137            .unwrap_or(1)
138    }
139
140    /// The free space of a file of `end` blocks holding these records.
141    pub(crate) fn free_space(&self, end: u64) -> MenteResult<FreeSpace> {
142        FreeSpace::from_live(self.records.values().map(|e| e.extent).collect(), end)
143    }
144}
145
146/// A persisted table and the checkpoint token it is anchored on.
147#[derive(Debug, Clone)]
148pub(crate) struct Snapshot {
149    pub(crate) token: Token,
150    pub(crate) table: RecordTable,
151}
152
153fn put_u64(buf: &mut Vec<u8>, v: u64) {
154    buf.extend_from_slice(&v.to_le_bytes());
155}
156
157pub(crate) fn encode(token: &Token, t: &RecordTable) -> Vec<u8> {
158    let mut buf = Vec::with_capacity(64 + t.records.len() * 56);
159    buf.extend_from_slice(MAGIC);
160    buf.extend_from_slice(&FORMAT_VERSION.to_le_bytes());
161    buf.extend_from_slice(token);
162    put_u64(&mut buf, t.next_locator);
163    put_u64(&mut buf, t.records.len() as u64);
164    for (&locator, e) in &t.records {
165        put_u64(&mut buf, locator);
166        buf.extend_from_slice(e.id.0.as_bytes());
167        put_u64(&mut buf, e.stamp);
168        put_u64(&mut buf, e.extent.start);
169        put_u64(&mut buf, e.extent.blocks);
170    }
171    let crc = crc32fast::hash(&buf);
172    buf.extend_from_slice(&crc.to_le_bytes());
173    buf
174}
175
176struct Reader<'a> {
177    buf: &'a [u8],
178    pos: usize,
179}
180
181impl Reader<'_> {
182    fn take(&mut self, n: usize) -> MenteResult<&[u8]> {
183        let end = self
184            .pos
185            .checked_add(n)
186            .filter(|&e| e <= self.buf.len())
187            .ok_or_else(|| MenteError::Storage("record map snapshot truncated".into()))?;
188        let s = &self.buf[self.pos..end];
189        self.pos = end;
190        Ok(s)
191    }
192
193    fn u64(&mut self) -> MenteResult<u64> {
194        Ok(le_u64(self.take(8)?))
195    }
196}
197
198pub(crate) fn decode(bytes: &[u8]) -> MenteResult<Snapshot> {
199    let bad = |what: &str| MenteError::Storage(format!("record map snapshot {what}"));
200    if bytes.len() < MAGIC.len() + 4 + 16 + 8 + 8 + 4 {
201        return Err(bad("too short"));
202    }
203    let (body, crc_bytes) = bytes.split_at(bytes.len() - 4);
204    if crc32fast::hash(body) != le_u32(crc_bytes) {
205        return Err(bad("checksum mismatch"));
206    }
207    let mut r = Reader { buf: body, pos: 0 };
208    if r.take(8)? != MAGIC {
209        return Err(bad("magic mismatch"));
210    }
211    if le_u32(r.take(4)?) != FORMAT_VERSION {
212        return Err(bad("version mismatch"));
213    }
214    let mut token = [0u8; 16];
215    token.copy_from_slice(r.take(16)?);
216    let next_locator = r.u64()?;
217    let n = r.u64()?;
218    let mut table = RecordTable {
219        records: BTreeMap::new(),
220        next_locator,
221    };
222    for _ in 0..n {
223        let locator = r.u64()?;
224        let id = uuid::Uuid::from_slice(r.take(16)?).map_err(|e| bad(&e.to_string()))?;
225        let stamp = r.u64()?;
226        let start = r.u64()?;
227        let blocks = r.u64()?;
228        if locator >= next_locator {
229            return Err(bad("has a locator past its counter"));
230        }
231        table.records.insert(
232            locator,
233            RecordEntry {
234                id: MemoryId(id),
235                stamp,
236                extent: Extent { start, blocks },
237            },
238        );
239    }
240    if r.pos != body.len() {
241        return Err(bad("has trailing bytes"));
242    }
243    Ok(Snapshot { token, table })
244}
245
246/// Read the snapshot in `dir`. `Ok(None)` when there is none.
247pub(crate) fn read_snapshot(dir: &Path) -> MenteResult<Option<Snapshot>> {
248    match std::fs::read(dir.join(MEMORY_MAP_FILE)) {
249        Ok(bytes) => decode(&bytes).map(Some),
250        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(None),
251        Err(e) => Err(e.into()),
252    }
253}
254
255/// Write the snapshot atomically: a reader sees the old file or the complete
256/// new one, never a mix (temp file, fsync, rename).
257pub(crate) fn write_snapshot(dir: &Path, token: &Token, table: &RecordTable) -> MenteResult<()> {
258    let path = dir.join(MEMORY_MAP_FILE);
259    let tmp = dir.join(format!("{MEMORY_MAP_FILE}.tmp"));
260    {
261        let mut file = std::fs::File::create(&tmp)?;
262        file.write_all(&encode(token, table))?;
263        file.sync_all()?;
264    }
265    std::fs::rename(&tmp, &path)?;
266    Ok(())
267}
268
269/// Remove the snapshot, ignoring a missing file.
270pub(crate) fn remove_snapshot(dir: &Path) {
271    let _ = std::fs::remove_file(dir.join(MEMORY_MAP_FILE));
272}
273
274#[cfg(test)]
275mod tests {
276    use super::*;
277
278    fn snap() -> Snapshot {
279        let mut table = RecordTable::default();
280        for (loc, start) in [(1u64, 1u64), (3, 4)] {
281            table.apply(
282                &RecordOp::Write {
283                    locator: loc,
284                    id: MemoryId::new(),
285                    extent: Extent { start, blocks: 3 },
286                    payload: vec![],
287                    remainder: None,
288                    freed: None,
289                },
290                random_u64(),
291            );
292        }
293        Snapshot {
294            token: new_token(),
295            table,
296        }
297    }
298
299    #[test]
300    fn snapshot_roundtrips() {
301        let dir = tempfile::tempdir().unwrap();
302        let a = snap();
303        assert_eq!(a.table.next_locator, 4);
304        write_snapshot(dir.path(), &a.token, &a.table).unwrap();
305        let b = read_snapshot(dir.path()).unwrap().unwrap();
306        assert_eq!(a.token, b.token);
307        assert_eq!(a.table, b.table);
308        assert!(!dir.path().join(format!("{MEMORY_MAP_FILE}.tmp")).exists());
309    }
310
311    #[test]
312    fn corrupt_snapshot_is_rejected() {
313        let dir = tempfile::tempdir().unwrap();
314        let s = snap();
315        write_snapshot(dir.path(), &s.token, &s.table).unwrap();
316        let path = dir.path().join(MEMORY_MAP_FILE);
317        let mut bytes = std::fs::read(&path).unwrap();
318        let mid = bytes.len() / 2;
319        bytes[mid] ^= 0xff;
320        std::fs::write(&path, &bytes).unwrap();
321        assert!(read_snapshot(dir.path()).is_err());
322        std::fs::write(&path, &bytes[..10]).unwrap();
323        assert!(read_snapshot(dir.path()).is_err());
324        std::fs::remove_file(&path).unwrap();
325        assert!(read_snapshot(dir.path()).unwrap().is_none());
326    }
327
328    #[test]
329    fn free_space_follows_the_table() {
330        let s = snap();
331        assert_eq!(s.table.max_end(), 7);
332        let f = s.table.free_space(10).unwrap();
333        assert_eq!(f.free_blocks(), 3);
334        assert!(s.table.free_space(6).is_err(), "extent past the end");
335    }
336}
mentedb_storage/memory_map.rs

mentedb_storage/
memory_map.rs
