reddb_server/storage/wal/
checkpoint.rs

1//! Checkpoint Manager
2//!
3//! Responsible for transferring committed transactions from the WAL to the main
4//! database file. Checkpointing ensures durability and allows WAL truncation.
5//!
6//! # Algorithm
7//!
8//! 1. Read all WAL records sequentially
9//! 2. Track transaction states (Begin, Commit, Rollback)
10//! 3. For committed transactions, collect PageWrite records
11//! 4. Apply committed pages to the Pager in LSN order
12//! 5. Sync Pager to disk
13//! 6. Update checkpoint LSN in database header
14//! 7. Truncate WAL
15//!
16//! # References
17//!
18//! - Turso `core/storage/wal.rs:checkpoint()` - Checkpoint logic
19//! - SQLite WAL documentation
20
21use std::collections::{HashMap, HashSet};
22use std::io;
23use std::path::Path;
24
25use super::reader::WalReader;
26use super::record::WalRecord;
27use super::writer::WalWriter;
28use crate::storage::engine::{Page, Pager, PAGE_SIZE};
29
30/// Checkpoint mode
31#[derive(Debug, Clone, Copy, PartialEq, Eq)]
32pub enum CheckpointMode {
33    /// Passive: Only checkpoint if no active writers
34    Passive,
35    /// Full: Wait for active writers to finish, then checkpoint all
36    Full,
37    /// Restart: Like Full, but also truncates the WAL
38    Restart,
39    /// Truncate: Checkpoint all and truncate WAL
40    Truncate,
41}
42
43/// Checkpoint result statistics
44#[derive(Debug, Clone, Default)]
45pub struct CheckpointResult {
46    /// Number of transactions processed
47    pub transactions_processed: u64,
48    /// Highest transaction id observed while scanning the WAL.
49    pub max_transaction_id: u64,
50    /// Number of pages checkpointed
51    pub pages_checkpointed: u64,
52    /// Number of records processed
53    pub records_processed: u64,
54    /// Final LSN after checkpoint
55    pub checkpoint_lsn: u64,
56    /// Whether WAL was truncated
57    pub wal_truncated: bool,
58}
59
60/// Checkpoint error types
61#[derive(Debug)]
62pub enum CheckpointError {
63    /// I/O error
64    Io(io::Error),
65    /// Pager error
66    Pager(String),
67    /// WAL is corrupted
68    CorruptedWal(String),
69    /// No WAL file found
70    NoWal,
71}
72
73impl std::fmt::Display for CheckpointError {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        match self {
76            Self::Io(e) => write!(f, "I/O error: {}", e),
77            Self::Pager(msg) => write!(f, "Pager error: {}", msg),
78            Self::CorruptedWal(msg) => write!(f, "Corrupted WAL: {}", msg),
79            Self::NoWal => write!(f, "No WAL file found"),
80        }
81    }
82}
83
84impl std::error::Error for CheckpointError {}
85
86impl From<io::Error> for CheckpointError {
87    fn from(e: io::Error) -> Self {
88        Self::Io(e)
89    }
90}
91
92/// Transaction state during checkpoint
93#[derive(Debug, Clone, Copy, PartialEq, Eq)]
94enum TxState {
95    Active,
96    Committed,
97    Aborted,
98}
99
100/// Pending page write from a transaction
101#[derive(Debug)]
102struct PendingWrite {
103    tx_id: u64,
104    page_id: u32,
105    data: Vec<u8>,
106    lsn: u64,
107}
108
109fn result_observe_tx_id(max_transaction_id: &mut u64, tx_id: u64) {
110    *max_transaction_id = (*max_transaction_id).max(tx_id);
111}
112
113/// Checkpoint manager
114///
115/// Responsible for transferring committed transactions from the WAL to the main database file.
116pub struct Checkpointer {
117    /// Checkpoint mode
118    mode: CheckpointMode,
119}
120
121impl Checkpointer {
122    /// Create a new checkpointer with the given mode
123    pub fn new(mode: CheckpointMode) -> Self {
124        Self { mode }
125    }
126
127    /// Create a checkpointer with default mode (Full)
128    pub fn default_mode() -> Self {
129        Self::new(CheckpointMode::Full)
130    }
131
132    /// Perform a checkpoint
133    ///
134    /// Reads all records from the WAL and applies committed changes to the database.
135    ///
136    /// # Arguments
137    ///
138    /// * `pager` - The Pager to write committed pages to
139    /// * `wal_path` - Path to the WAL file
140    ///
141    /// # Returns
142    ///
143    /// Checkpoint statistics or error
144    pub fn checkpoint(
145        &self,
146        pager: &Pager,
147        wal_path: &Path,
148    ) -> Result<CheckpointResult, CheckpointError> {
149        // Open WAL for reading
150        let wal_reader = match WalReader::open(wal_path) {
151            Ok(r) => r,
152            Err(e) if e.kind() == io::ErrorKind::NotFound => {
153                // No WAL file - nothing to checkpoint
154                return Ok(CheckpointResult::default());
155            }
156            Err(e) => return Err(CheckpointError::Io(e)),
157        };
158
159        // Phase 1: Read and categorize all records
160        let mut tx_states: HashMap<u64, TxState> = HashMap::new();
161        let mut pending_writes: Vec<PendingWrite> = Vec::new();
162        let mut records_processed: u64 = 0;
163        let mut last_lsn: u64 = 0;
164        let mut max_transaction_id: u64 = 0;
165
166        for record_result in wal_reader.iter() {
167            let (lsn, record) = record_result.map_err(CheckpointError::Io)?;
168            records_processed += 1;
169            last_lsn = lsn;
170
171            match record {
172                WalRecord::Begin { tx_id } => {
173                    result_observe_tx_id(&mut max_transaction_id, tx_id);
174                    tx_states.insert(tx_id, TxState::Active);
175                }
176                WalRecord::Commit { tx_id } => {
177                    result_observe_tx_id(&mut max_transaction_id, tx_id);
178                    tx_states.insert(tx_id, TxState::Committed);
179                }
180                WalRecord::Rollback { tx_id } => {
181                    result_observe_tx_id(&mut max_transaction_id, tx_id);
182                    tx_states.insert(tx_id, TxState::Aborted);
183                }
184                WalRecord::PageWrite {
185                    tx_id,
186                    page_id,
187                    data,
188                } => {
189                    result_observe_tx_id(&mut max_transaction_id, tx_id);
190                    pending_writes.push(PendingWrite {
191                        tx_id,
192                        page_id,
193                        data,
194                        lsn,
195                    });
196                }
197                WalRecord::Checkpoint {
198                    lsn: _checkpoint_lsn,
199                } => {
200                    // Checkpoint marker - we can skip records before this LSN
201                    // For now, we process everything
202                }
203                WalRecord::TxCommitBatch { tx_id, .. } => {
204                    result_observe_tx_id(&mut max_transaction_id, tx_id);
205                    // Store-level logical commit batches are replayed by
206                    // UnifiedStore, not by the pager page checkpoint path.
207                }
208                WalRecord::VectorInsert { .. } => {
209                    // Vector-turbo logical inserts are replayed by the
210                    // vector index layer, not by page checkpointing.
211                    // Issue #694 — named crash boundary for #673.
212                    crate::runtime::turbo_crash_inject::fire(
213                        crate::runtime::turbo_crash_inject::InjectionPoint::MidCheckpoint,
214                    );
215                }
216                WalRecord::FullPageImage { tx_id, .. } => {
217                    result_observe_tx_id(&mut max_transaction_id, tx_id);
218                    // FPI records (gh-478) are consumed by the pager
219                    // recovery path before redo, not by checkpoint
220                    // accounting.
221                }
222            }
223        }
224
225        // Phase 2: Filter for committed transactions only
226        let committed_txs: HashSet<u64> = tx_states
227            .iter()
228            .filter(|(_, state)| **state == TxState::Committed)
229            .map(|(tx_id, _)| *tx_id)
230            .collect();
231
232        // Phase 3: Collect pages from committed transactions
233        // Keep only the latest write for each page (from committed txs)
234        let mut latest_writes: HashMap<u32, Vec<u8>> = HashMap::new();
235
236        for write in pending_writes {
237            if committed_txs.contains(&write.tx_id) {
238                // Always overwrite with later writes (they have higher LSN)
239                latest_writes.insert(write.page_id, write.data);
240            }
241        }
242
243        // Phase 4 (PREPARE): Mark checkpoint in progress in header
244        if !latest_writes.is_empty() {
245            pager
246                .set_checkpoint_in_progress(true, last_lsn)
247                .map_err(|e| CheckpointError::Pager(e.to_string()))?;
248        }
249
250        // Phase 5 (APPLY): Write committed pages to Pager
251        let mut pages_checkpointed: u64 = 0;
252
253        for (page_id, data) in &latest_writes {
254            // Reconstruct page from WAL data
255            if data.len() != PAGE_SIZE {
256                return Err(CheckpointError::CorruptedWal(format!(
257                    "Page {} has wrong size: {} (expected {})",
258                    page_id,
259                    data.len(),
260                    PAGE_SIZE
261                )));
262            }
263
264            let mut page_data = [0u8; PAGE_SIZE];
265            page_data.copy_from_slice(data);
266            let page = Page::from_bytes(page_data);
267
268            // Write to pager
269            pager
270                .write_page(*page_id, page)
271                .map_err(|e| CheckpointError::Pager(e.to_string()))?;
272
273            pages_checkpointed += 1;
274        }
275
276        // Phase 6: Sync Pager to disk
277        pager
278            .sync()
279            .map_err(|e| CheckpointError::Pager(e.to_string()))?;
280
281        // Phase 7 (COMPLETE): Clear in-progress flag and update checkpoint LSN
282        if !latest_writes.is_empty() {
283            pager
284                .complete_checkpoint(last_lsn)
285                .map_err(|e| CheckpointError::Pager(e.to_string()))?;
286        }
287
288        // Phase 8: Truncate WAL if requested
289        let wal_truncated = matches!(
290            self.mode,
291            CheckpointMode::Restart | CheckpointMode::Truncate
292        );
293
294        if wal_truncated {
295            let mut wal_writer = WalWriter::open(wal_path)?;
296            wal_writer.truncate()?;
297
298            // Write checkpoint marker with current LSN
299            let checkpoint_record = WalRecord::Checkpoint { lsn: last_lsn };
300            wal_writer.append(&checkpoint_record)?;
301            wal_writer.sync()?;
302        }
303
304        Ok(CheckpointResult {
305            transactions_processed: committed_txs.len() as u64,
306            max_transaction_id,
307            pages_checkpointed,
308            records_processed,
309            checkpoint_lsn: last_lsn,
310            wal_truncated,
311        })
312    }
313
314    /// Perform crash recovery
315    ///
316    /// Called on database open to apply any committed transactions from the WAL
317    /// that weren't checkpointed before the crash. If a checkpoint was interrupted
318    /// (checkpoint_in_progress flag set), re-applies all WAL records from scratch.
319    ///
320    /// # Arguments
321    ///
322    /// * `pager` - The Pager to recover into
323    /// * `wal_path` - Path to the WAL file
324    ///
325    /// # Returns
326    ///
327    /// Recovery statistics or error
328    pub fn recover(pager: &Pager, wal_path: &Path) -> Result<CheckpointResult, CheckpointError> {
329        // Check if a checkpoint was interrupted
330        if let Ok(header) = pager.header() {
331            if header.checkpoint_in_progress {
332                // Previous checkpoint was interrupted — re-apply everything from WAL
333                // This is safe because page writes are idempotent (last-write-wins)
334                let _ = pager.set_checkpoint_in_progress(false, 0);
335            }
336        }
337        let checkpointer = Self::new(CheckpointMode::Truncate);
338        checkpointer.checkpoint(pager, wal_path)
339    }
340}
341
342#[cfg(test)]
343mod tests {
344    use super::*;
345    use crate::storage::engine::PageType;
346    use std::fs;
347    use std::time::{SystemTime, UNIX_EPOCH};
348
349    fn temp_dir() -> std::path::PathBuf {
350        let timestamp = SystemTime::now()
351            .duration_since(UNIX_EPOCH)
352            .unwrap()
353            .as_nanos();
354        std::env::temp_dir().join(format!("reddb_checkpoint_test_{}", timestamp))
355    }
356
357    fn cleanup(dir: &Path) {
358        let _ = fs::remove_dir_all(dir);
359    }
360
361    fn temp_wal_path(dir: &Path, name: &str) -> std::path::PathBuf {
362        reddb_file::layout::wal_component_temp_path(dir, "checkpoint", name, std::process::id())
363    }
364
365    #[test]
366    fn test_checkpoint_empty_wal() {
367        let dir = temp_dir();
368        let _ = fs::create_dir_all(&dir);
369        let db_path = dir.join("test.db");
370        let wal_path = temp_wal_path(&dir, "empty");
371
372        // Create pager
373        let pager = Pager::open_default(&db_path).unwrap();
374
375        // No WAL file - should succeed with empty result
376        let checkpointer = Checkpointer::default_mode();
377        let result = checkpointer.checkpoint(&pager, &wal_path).unwrap();
378
379        assert_eq!(result.transactions_processed, 0);
380        assert_eq!(result.pages_checkpointed, 0);
381
382        cleanup(&dir);
383    }
384
385    #[test]
386    fn test_checkpoint_committed_transaction() {
387        let dir = temp_dir();
388        let _ = fs::create_dir_all(&dir);
389        let db_path = dir.join("test.db");
390        let wal_path = temp_wal_path(&dir, "committed");
391
392        // Create pager
393        let pager = Pager::open_default(&db_path).unwrap();
394
395        // Allocate a page to get its ID
396        let page = pager.allocate_page(PageType::BTreeLeaf).unwrap();
397        let page_id = page.page_id();
398
399        // Create WAL with a committed transaction
400        {
401            let mut wal_writer = WalWriter::open(&wal_path).unwrap();
402
403            // Begin transaction
404            wal_writer.append(&WalRecord::Begin { tx_id: 1 }).unwrap();
405
406            // Write a page
407            let mut page_data = [0u8; PAGE_SIZE];
408            page_data[0] = 0x42; // Mark with test byte
409            wal_writer
410                .append(&WalRecord::PageWrite {
411                    tx_id: 1,
412                    page_id,
413                    data: page_data.to_vec(),
414                })
415                .unwrap();
416
417            // Commit transaction
418            wal_writer.append(&WalRecord::Commit { tx_id: 1 }).unwrap();
419
420            wal_writer.sync().unwrap();
421        }
422
423        // Checkpoint
424        let checkpointer = Checkpointer::new(CheckpointMode::Full);
425        let result = checkpointer.checkpoint(&pager, &wal_path).unwrap();
426
427        assert_eq!(result.transactions_processed, 1);
428        assert_eq!(result.pages_checkpointed, 1);
429        assert_eq!(result.records_processed, 3);
430
431        // Verify page was written
432        let read_page = pager.read_page(page_id).unwrap();
433        assert_eq!(read_page.as_bytes()[0], 0x42);
434
435        cleanup(&dir);
436    }
437
438    #[test]
439    fn test_checkpoint_aborted_transaction() {
440        let dir = temp_dir();
441        let _ = fs::create_dir_all(&dir);
442        let db_path = dir.join("test.db");
443        let wal_path = temp_wal_path(&dir, "aborted");
444
445        // Create pager
446        let pager = Pager::open_default(&db_path).unwrap();
447
448        // Allocate a page to get its ID
449        let page = pager.allocate_page(PageType::BTreeLeaf).unwrap();
450        let page_id = page.page_id();
451
452        // Create WAL with an aborted transaction
453        {
454            let mut wal_writer = WalWriter::open(&wal_path).unwrap();
455
456            // Begin transaction
457            wal_writer.append(&WalRecord::Begin { tx_id: 1 }).unwrap();
458
459            // Write a page
460            let mut page_data = [0u8; PAGE_SIZE];
461            page_data[0] = 0x42;
462            wal_writer
463                .append(&WalRecord::PageWrite {
464                    tx_id: 1,
465                    page_id,
466                    data: page_data.to_vec(),
467                })
468                .unwrap();
469
470            // Rollback transaction
471            wal_writer
472                .append(&WalRecord::Rollback { tx_id: 1 })
473                .unwrap();
474
475            wal_writer.sync().unwrap();
476        }
477
478        // Checkpoint
479        let checkpointer = Checkpointer::new(CheckpointMode::Full);
480        let result = checkpointer.checkpoint(&pager, &wal_path).unwrap();
481
482        // Aborted transaction should not be checkpointed
483        assert_eq!(result.transactions_processed, 0);
484        assert_eq!(result.pages_checkpointed, 0);
485
486        // Verify page was NOT written (should still be zeros)
487        let read_page = pager.read_page(page_id).unwrap();
488        assert_ne!(read_page.as_bytes()[0], 0x42);
489
490        cleanup(&dir);
491    }
492
493    #[test]
494    fn test_checkpoint_mixed_transactions() {
495        let dir = temp_dir();
496        let _ = fs::create_dir_all(&dir);
497        let db_path = dir.join("test.db");
498        let wal_path = temp_wal_path(&dir, "truncate");
499
500        // Create pager
501        let pager = Pager::open_default(&db_path).unwrap();
502
503        // Allocate pages
504        let page1 = pager.allocate_page(PageType::BTreeLeaf).unwrap();
505        let page2 = pager.allocate_page(PageType::BTreeLeaf).unwrap();
506        let page1_id = page1.page_id();
507        let page2_id = page2.page_id();
508
509        // Create WAL with mixed transactions
510        {
511            let mut wal_writer = WalWriter::open(&wal_path).unwrap();
512
513            // Transaction 1: Committed
514            wal_writer.append(&WalRecord::Begin { tx_id: 1 }).unwrap();
515            let mut page_data1 = [0u8; PAGE_SIZE];
516            page_data1[0] = 0x11;
517            wal_writer
518                .append(&WalRecord::PageWrite {
519                    tx_id: 1,
520                    page_id: page1_id,
521                    data: page_data1.to_vec(),
522                })
523                .unwrap();
524            wal_writer.append(&WalRecord::Commit { tx_id: 1 }).unwrap();
525
526            // Transaction 2: Aborted
527            wal_writer.append(&WalRecord::Begin { tx_id: 2 }).unwrap();
528            let mut page_data2 = [0u8; PAGE_SIZE];
529            page_data2[0] = 0x22;
530            wal_writer
531                .append(&WalRecord::PageWrite {
532                    tx_id: 2,
533                    page_id: page2_id,
534                    data: page_data2.to_vec(),
535                })
536                .unwrap();
537            wal_writer
538                .append(&WalRecord::Rollback { tx_id: 2 })
539                .unwrap();
540
541            // Transaction 3: Committed
542            wal_writer.append(&WalRecord::Begin { tx_id: 3 }).unwrap();
543            let mut page_data3 = [0u8; PAGE_SIZE];
544            page_data3[0] = 0x33;
545            wal_writer
546                .append(&WalRecord::PageWrite {
547                    tx_id: 3,
548                    page_id: page2_id,
549                    data: page_data3.to_vec(),
550                })
551                .unwrap();
552            wal_writer.append(&WalRecord::Commit { tx_id: 3 }).unwrap();
553
554            wal_writer.sync().unwrap();
555        }
556
557        // Checkpoint
558        let checkpointer = Checkpointer::new(CheckpointMode::Full);
559        let result = checkpointer.checkpoint(&pager, &wal_path).unwrap();
560
561        // Only committed transactions (1 and 3) should be processed
562        assert_eq!(result.transactions_processed, 2);
563        assert_eq!(result.max_transaction_id, 3);
564        assert_eq!(result.pages_checkpointed, 2);
565
566        // Verify pages
567        let read_page1 = pager.read_page(page1_id).unwrap();
568        assert_eq!(read_page1.as_bytes()[0], 0x11);
569
570        let read_page2 = pager.read_page(page2_id).unwrap();
571        assert_eq!(read_page2.as_bytes()[0], 0x33); // From tx 3, not tx 2
572
573        cleanup(&dir);
574    }
575
576    #[test]
577    fn test_checkpoint_truncate() {
578        let dir = temp_dir();
579        let _ = fs::create_dir_all(&dir);
580        let db_path = dir.join("test.db");
581        let wal_path = temp_wal_path(&dir, "full-page-images");
582
583        // Create pager
584        let pager = Pager::open_default(&db_path).unwrap();
585
586        // Create WAL with a committed transaction
587        {
588            let mut wal_writer = WalWriter::open(&wal_path).unwrap();
589            wal_writer.append(&WalRecord::Begin { tx_id: 1 }).unwrap();
590            wal_writer.append(&WalRecord::Commit { tx_id: 1 }).unwrap();
591            wal_writer.sync().unwrap();
592        }
593
594        // Checkpoint with truncate
595        let checkpointer = Checkpointer::new(CheckpointMode::Truncate);
596        let result = checkpointer.checkpoint(&pager, &wal_path).unwrap();
597
598        assert!(result.wal_truncated);
599
600        let records: Vec<_> = WalReader::open(&wal_path)
601            .unwrap()
602            .iter()
603            .collect::<Result<Vec<_>, _>>()
604            .unwrap();
605        assert_eq!(records.len(), 1);
606        assert_eq!(
607            records[0].1,
608            WalRecord::Checkpoint {
609                lsn: result.checkpoint_lsn
610            }
611        );
612
613        cleanup(&dir);
614    }
615}
reddb_server/storage/wal/checkpoint.rs

reddb_server/storage/wal/
checkpoint.rs
