mentedb_storage/
legacy.rs

1//! Migration from the page file layout (format 1) to the record file.
2//!
3//! Format 1 stored every memory in its own 64 KB page (`pages.db`), with a
4//! WAL of full page images (`wal.log`) and a page keyed memory map snapshot
5//! (`memory_map.bin`). A typical memory used under a tenth of its page.
6//!
7//! Migration reads that state without modifying a single byte of it, writes
8//! the complete record database under new file names, and commits with one
9//! atomic rename:
10//!
11//! 1. Read the old state: page file, plus the old WAL replayed in memory
12//!    (last operation per page wins, exactly as the old recovery did), plus
13//!    the old snapshot's write stamps when it is still anchored.
14//! 2. Write every memory into `records.db.tmp`, keeping its page number as its
15//!    locator; fsync.
16//! 3. Write the new WAL (`records.wal`, a single checkpoint entry) and the new
17//!    record map (`records_map.bin`) anchored on it; fsync both.
18//! 4. Rename `records.db.tmp` to `records.db` and fsync the directory. This
19//!    is the commit point: the new database exists exactly when `records.db`
20//!    does.
21//! 5. Move the old files aside (`pages.db.v1`, `wal.log.v1`,
22//!    `memory_map.bin.v1`), kept as a copy of the database as it was before
23//!    migration until an operator deletes them.
24//!
25//! A crash before step 4 leaves the old database untouched (open migrates it
26//! again from scratch, discarding the partial new files); a crash after it
27//! leaves a complete new database whose leftover old files are removed on the
28//! next open. Engines older than this layout refuse a directory holding
29//! `records.db`, so they never open the moved files as a fresh database.
30//! Locators and stamps are preserved, so every snapshot keyed by
31//! page and stamp (salience bounds, mode probes) stays valid.
32
33use std::fs::File;
34use std::io::{BufReader, BufWriter, Read, Write};
35use std::path::Path;
36use std::time::Instant;
37
38use mentedb_core::MemoryNode;
39use mentedb_core::error::{MenteError, MenteResult};
40use mentedb_core::types::MemoryId;
41use tracing::{info, warn};
42
43use crate::memory_map::{self, RecordEntry, RecordTable, Token, entry_stamp};
44use crate::record::{self, DATA_FILE, Extent, WAL_FILE, le_u32, le_u64};
45use crate::wal::{Wal, WalEntry, WalEntryType};
46
47/// Page size of format 1.
48pub const LEGACY_PAGE_SIZE: usize = 64 * 1024;
49/// Page header size of format 1.
50const LEGACY_HEADER_SIZE: usize = 32;
51/// Usable bytes per format 1 page.
52const LEGACY_DATA_SIZE: usize = LEGACY_PAGE_SIZE - LEGACY_HEADER_SIZE;
53const LEGACY_MAGIC: u64 = 0x4D454E_5445444231;
54const LEGACY_VERSION: u32 = 1;
55const LEGACY_DATA_PAGE: u8 = 1;
56
57/// Format 1 page file.
58pub const LEGACY_PAGE_FILE: &str = "pages.db";
59/// Format 1 WAL.
60pub const LEGACY_WAL_FILE: &str = "wal.log";
61/// Format 1 memory map snapshot.
62pub const LEGACY_MAP_FILE: &str = "memory_map.bin";
63
64const LEGACY_MAP_MAGIC: &[u8; 8] = b"MDBMMAP1";
65
66/// Where an injected fault stops a migration, to test crash safety.
67#[doc(hidden)]
68#[derive(Debug, Clone, Copy, PartialEq, Eq)]
69pub enum MigrationFault {
70    /// After part of the new record file is written, before its fsync.
71    MidDataWrite,
72    /// After the new record file is written and synced.
73    AfterDataSync,
74    /// After the new WAL is written.
75    AfterWal,
76    /// After the new record map snapshot is written, right before commit.
77    BeforeCommit,
78    /// After the commit rename, before the old files are moved aside.
79    AfterCommit,
80}
81
82/// What a migration did.
83#[derive(Debug, Clone, Default)]
84pub struct MigrationReport {
85    /// Memories carried over.
86    pub memories: u64,
87    /// Pages in the old file (including the header page).
88    pub legacy_pages: u64,
89    /// Old pages that held no decodable memory (free, empty, or corrupt).
90    pub skipped_pages: u64,
91    /// Old WAL operations applied during the migration.
92    pub wal_ops: u64,
93    /// Write stamps carried over from the old memory map snapshot.
94    pub stamps_kept: u64,
95    /// Bytes of the old page file, WAL and snapshot.
96    pub bytes_before: u64,
97    /// Bytes of the new record file, WAL and snapshot.
98    pub bytes_after: u64,
99    /// Wall time of the migration.
100    pub elapsed_ms: u64,
101}
102
103fn file_len(path: &Path) -> u64 {
104    std::fs::metadata(path).map(|m| m.len()).unwrap_or(0)
105}
106
107/// Whether `dir` holds a format 1 database and no record database.
108pub(crate) fn is_legacy(dir: &Path) -> bool {
109    !dir.join(DATA_FILE).exists() && file_len(&dir.join(LEGACY_PAGE_FILE)) > 0
110}
111
112/// Suffix of the format 1 files kept after a migration.
113pub const LEGACY_KEPT_SUFFIX: &str = ".v1";
114
115/// Settle format 1 files after a committed migration (also run on every open
116/// while `records.db` exists, which finishes a migration that crashed after
117/// its commit). The old page file, WAL and snapshot are moved aside with the
118/// `.v1` suffix rather than deleted, so the pre-migration database survives
119/// a migration bug until an operator removes it; temporary files are removed.
120pub(crate) fn remove_leftovers(dir: &Path) -> MenteResult<()> {
121    let mut kept = Vec::new();
122    for name in [LEGACY_PAGE_FILE, LEGACY_WAL_FILE, LEGACY_MAP_FILE] {
123        let p = dir.join(name);
124        if p.exists() {
125            std::fs::rename(&p, dir.join(format!("{name}{LEGACY_KEPT_SUFFIX}")))?;
126            kept.push(name);
127        }
128    }
129    let mut removed = Vec::new();
130    for name in ["wal.log.tmp", "memory_map.bin.tmp", "records.db.tmp"] {
131        let p = dir.join(name);
132        if p.exists() {
133            std::fs::remove_file(&p)?;
134            removed.push(name);
135        }
136    }
137    if !kept.is_empty() || !removed.is_empty() {
138        record::sync_dir(dir)?;
139        info!(
140            ?kept,
141            ?removed,
142            suffix = LEGACY_KEPT_SUFFIX,
143            "settled format 1 files after migration"
144        );
145    }
146    Ok(())
147}
148
149/// Decode a memory from a format 1 page data region: `[len u32][json]`.
150fn decode_legacy_memory(data: &[u8]) -> Option<MemoryNode> {
151    let data = &data[..data.len().min(LEGACY_DATA_SIZE)];
152    if data.len() < 4 {
153        return None;
154    }
155    let len = le_u32(&data[..4]) as usize;
156    if len == 0 || len + 4 > LEGACY_DATA_SIZE || 4 + len > data.len() {
157        return None;
158    }
159    serde_json::from_slice(&data[4..4 + len]).ok()
160}
161
162/// A format 1 page decoded from its on-disk bytes: the memory it holds, if
163/// any. A checksum mismatch is reported as an error.
164fn decode_legacy_page(page_id: u64, buf: &[u8]) -> MenteResult<Option<MemoryNode>> {
165    let header = &buf[..LEGACY_HEADER_SIZE];
166    let data = &buf[LEGACY_HEADER_SIZE..LEGACY_PAGE_SIZE];
167    let checksum = le_u32(&header[16..20]);
168    if checksum != 0 {
169        let mut h = crc32fast::Hasher::new();
170        h.update(&header[0..8]); // page_id
171        h.update(&header[8..16]); // lsn
172        h.update(&header[20..22]); // free_space
173        h.update(&header[22..24]); // num_slots
174        h.update(&header[24..25]); // page_type
175        h.update(data);
176        if h.finalize() != checksum {
177            return Err(MenteError::Storage(format!(
178                "legacy page {page_id} checksum mismatch"
179            )));
180        }
181    }
182    if header[24] != LEGACY_DATA_PAGE {
183        return Ok(None);
184    }
185    Ok(decode_legacy_memory(data))
186}
187
188/// The write stamps the old memory map recorded for each page, when its
189/// snapshot is intact and still anchored in the old WAL.
190fn legacy_stamps(
191    dir: &Path,
192    entries: &[WalEntry],
193) -> Option<std::collections::HashMap<u64, (MemoryId, u64)>> {
194    let bytes = std::fs::read(dir.join(LEGACY_MAP_FILE)).ok()?;
195    if bytes.len() < 8 + 4 + 16 + 8 + 8 + 8 + 4 {
196        return None;
197    }
198    let (body, crc) = bytes.split_at(bytes.len() - 4);
199    if crc32fast::hash(body) != le_u32(crc) || &body[..8] != LEGACY_MAP_MAGIC {
200        return None;
201    }
202    if le_u32(&body[8..12]) != 1 {
203        return None;
204    }
205    let mut token: Token = [0; 16];
206    token.copy_from_slice(&body[12..28]);
207    let mut pos = 28 + 8; // token, page_count
208    let take = |pos: &mut usize, n: usize| -> Option<&[u8]> {
209        let s = body.get(*pos..*pos + n)?;
210        *pos += n;
211        Some(s)
212    };
213    let n = le_u64(take(&mut pos, 8)?);
214    let mut stamps = std::collections::HashMap::new();
215    for _ in 0..n {
216        let page = le_u64(take(&mut pos, 8)?);
217        let id = uuid::Uuid::from_slice(take(&mut pos, 16)?).ok()?;
218        let stamp = le_u64(take(&mut pos, 8)?);
219        stamps.insert(page, (MemoryId(id), stamp));
220    }
221    // Roll forward over the WAL after the anchoring checkpoint, with the
222    // rules the old memory map applied.
223    let at = entries
224        .iter()
225        .rposition(|e| e.entry_type == WalEntryType::Checkpoint && e.data == token)?;
226    for e in &entries[at + 1..] {
227        match e.entry_type {
228            WalEntryType::PageWrite => match decode_legacy_memory(&e.data) {
229                Some(node) => {
230                    stamps.insert(e.page_id, (node.id, entry_stamp(&token, e.lsn)));
231                }
232                None => {
233                    stamps.remove(&e.page_id);
234                }
235            },
236            WalEntryType::PageFree => {
237                stamps.remove(&e.page_id);
238            }
239            _ => {}
240        }
241    }
242    Some(stamps)
243}
244
245/// Read the format 1 file header: page count.
246fn legacy_page_count(file: &mut File) -> MenteResult<u64> {
247    let mut hdr = [0u8; 32];
248    file.read_exact(&mut hdr)?;
249    if le_u64(&hdr[0..8]) != LEGACY_MAGIC {
250        return Err(MenteError::Storage("invalid page file magic number".into()));
251    }
252    let version = le_u32(&hdr[8..12]);
253    if version != LEGACY_VERSION {
254        return Err(MenteError::Storage(format!(
255            "unsupported page file version: {version}"
256        )));
257    }
258    Ok(le_u64(&hdr[16..24]))
259}
260
261fn fault(at: Option<MigrationFault>, here: MigrationFault) -> MenteResult<()> {
262    if at == Some(here) {
263        return Err(MenteError::Storage(format!(
264            "injected migration fault: {here:?}"
265        )));
266    }
267    Ok(())
268}
269
270/// Migrate the format 1 database in `dir` to the record layout. The caller
271/// holds the directory lock.
272pub(crate) fn migrate(dir: &Path, at: Option<MigrationFault>) -> MenteResult<MigrationReport> {
273    let started = Instant::now();
274    let pages_path = dir.join(LEGACY_PAGE_FILE);
275    let tmp_path = dir.join(format!("{DATA_FILE}.tmp"));
276    let wal_path = dir.join(WAL_FILE);
277
278    // Partial output of an earlier attempt that crashed before its commit.
279    for p in [
280        tmp_path.clone(),
281        wal_path.clone(),
282        dir.join(format!("{WAL_FILE}.tmp")),
283        dir.join(memory_map::MEMORY_MAP_FILE),
284        dir.join(format!("{}.tmp", memory_map::MEMORY_MAP_FILE)),
285    ] {
286        if p.exists() {
287            std::fs::remove_file(&p)?;
288        }
289    }
290
291    let mut report = MigrationReport {
292        bytes_before: [LEGACY_PAGE_FILE, LEGACY_WAL_FILE, LEGACY_MAP_FILE]
293            .iter()
294            .map(|n| file_len(&dir.join(n)))
295            .sum(),
296        ..Default::default()
297    };
298    info!(
299        dir = %dir.display(),
300        bytes = report.bytes_before,
301        "migrating page file database to the record layout"
302    );
303
304    // Old state: page file plus the old WAL, last operation per page wins.
305    let entries = Wal::read_entries(&dir.join(LEGACY_WAL_FILE))?;
306    let mut last_op: std::collections::HashMap<u64, &WalEntry> = Default::default();
307    for e in &entries {
308        if matches!(
309            e.entry_type,
310            WalEntryType::PageWrite | WalEntryType::PageFree
311        ) {
312            last_op.insert(e.page_id, e);
313        }
314    }
315    report.wal_ops = last_op.len() as u64;
316    let stamps = legacy_stamps(dir, &entries);
317
318    let mut pages_file = File::open(&pages_path)?;
319    let header_count = legacy_page_count(&mut pages_file)?;
320    let file_pages = file_len(&pages_path) / LEGACY_PAGE_SIZE as u64;
321    let wal_max = last_op.keys().max().map_or(0, |m| m + 1);
322    let page_count = header_count.max(wal_max);
323    report.legacy_pages = page_count;
324
325    // Sequential, buffered reads of the whole old file: one pass, large
326    // requests, which is what network file systems serve best.
327    let mut reader = BufReader::with_capacity(64 * LEGACY_PAGE_SIZE, pages_file);
328    let mut skip = vec![0u8; LEGACY_PAGE_SIZE - 32];
329    reader.read_exact(&mut skip)?; // rest of page 0
330
331    let mut out = BufWriter::with_capacity(4 << 20, File::create(&tmp_path)?);
332    out.write_all(&record::file_header())?;
333    let mut block = 1u64;
334    let mut table = RecordTable::default();
335    let mut page = vec![0u8; LEGACY_PAGE_SIZE];
336    for page_id in 1..page_count {
337        let on_disk = page_id < file_pages;
338        if on_disk {
339            reader.read_exact(&mut page)?;
340        }
341        let node = match last_op.get(&page_id) {
342            Some(e) if e.entry_type == WalEntryType::PageWrite => decode_legacy_memory(&e.data),
343            Some(_) => None,
344            None if on_disk => match decode_legacy_page(page_id, &page) {
345                Ok(n) => n,
346                Err(e) => {
347                    warn!(page_id, error = %e, "skipping unreadable legacy page");
348                    None
349                }
350            },
351            None => None,
352        };
353        let Some(node) = node else {
354            report.skipped_pages += 1;
355            continue;
356        };
357        let payload = record::encode_node(&node)?;
358        let blocks = record::blocks_for(payload.len());
359        out.write_all(&record::live_image(page_id, node.id, &payload, blocks))?;
360        let stamp = match stamps.as_ref().and_then(|s| s.get(&page_id)) {
361            Some(&(id, stamp)) if id == node.id => {
362                report.stamps_kept += 1;
363                stamp
364            }
365            _ => memory_map::random_u64(),
366        };
367        table.records.insert(
368            page_id,
369            RecordEntry {
370                id: node.id,
371                stamp,
372                extent: Extent {
373                    start: block,
374                    blocks,
375                },
376            },
377        );
378        block += blocks;
379        report.memories += 1;
380        if report.memories == 1 {
381            fault(at, MigrationFault::MidDataWrite)?;
382        }
383    }
384    table.next_locator = page_count.max(1);
385    let out = out
386        .into_inner()
387        .map_err(|e| MenteError::Storage(format!("record file write failed: {e}")))?;
388    out.sync_all()?;
389    drop(out);
390    fault(at, MigrationFault::AfterDataSync)?;
391
392    // New WAL holding only the checkpoint the snapshot is anchored on.
393    let token = memory_map::new_token();
394    {
395        let mut wal = Wal::open_file(&wal_path)?;
396        wal.append(WalEntryType::Checkpoint, 0, &token)?;
397        wal.sync()?;
398    }
399    fault(at, MigrationFault::AfterWal)?;
400    memory_map::write_snapshot(dir, &token, &table)?;
401    record::sync_dir(dir)?;
402    fault(at, MigrationFault::BeforeCommit)?;
403
404    // Commit.
405    std::fs::rename(&tmp_path, dir.join(DATA_FILE))?;
406    record::sync_dir(dir)?;
407    fault(at, MigrationFault::AfterCommit)?;
408
409    remove_leftovers(dir)?;
410    report.bytes_after = [DATA_FILE, WAL_FILE, memory_map::MEMORY_MAP_FILE]
411        .iter()
412        .map(|n| file_len(&dir.join(n)))
413        .sum();
414    report.elapsed_ms = started.elapsed().as_millis() as u64;
415    info!(
416        dir = %dir.display(),
417        memories = report.memories,
418        legacy_pages = report.legacy_pages,
419        skipped_pages = report.skipped_pages,
420        wal_ops = report.wal_ops,
421        stamps_kept = report.stamps_kept,
422        bytes_before = report.bytes_before,
423        bytes_after = report.bytes_after,
424        elapsed_ms = report.elapsed_ms,
425        "migrated page file database to the record layout"
426    );
427    Ok(report)
428}
mentedb_storage/legacy.rs

mentedb_storage/
legacy.rs
