检查点按 LSN 部分截断

chusql-core/storage/src/runtime.rs

@@ -11,7 +11,7 @@ use crate::protocol::{
     blocked_table, reserved_table, Account, ColumnStatWire, IndexWire, Request, Response, Row,
     SchemaColumn, StorageOp, TableSchemaWire, USERS_TABLE,
 };
-use crate::wal::{committed_ops, Wal, WalOp};
+use crate::wal::{committed_ops, records_after, Wal, WalOp};
 use crate::{log_debug, log_error, log_info, log_warn};
 
 // 进程内存储运行时：Haskell 侧经 ffi.rs 调进来。
@@ -412,11 +412,13 @@ impl Server {
         self.group_via_wal(&wal_ops)
     }
 
-    /// 一组操作加提交标记落盘，成功后清空日志
+    /// 一组操作加提交标记落盘，成功后把检查点推到这条提交
+    ///
+    /// 检查点写在这组数据和字典都刷盘之后，含义是「LSN 到此为止的改动都已落盘」；
+    /// 写完再按这个边界丢掉日志前缀，中途崩掉只会留下更长的日志，不会少数据。
     fn group_via_wal(&self, ops: &[WalOp]) -> std::io::Result<()> {
         std::fs::create_dir_all(&self.cfg.data_dir)?;
         let wal = &self.wal;
-        wal.truncate()?;
         let commit_lsn = wal.append_group(ops)?;
         for op in ops {
             let result = self.apply_op(op);
@@ -430,7 +432,8 @@ impl Server {
             self.flush_op_tables(op)?;
         }
         self.flush_all_tables()?;
-        wal.clear()?;
+        wal.write_checkpoint(commit_lsn)?;
+        wal.truncate_before(commit_lsn)?;
         log_debug!(core, "WAL committed group at LSN {}", commit_lsn);
         Ok(())
     }
@@ -472,7 +475,7 @@ impl Server {
         self.with_table(table, |t| t.flush())
     }
 
-    /// 重放残留 WAL：已提交组重放，未提交组丢弃
+    /// 重放残留 WAL：检查点之后的已提交组重放，未提交组丢弃
     fn recover_wal(&self) -> std::io::Result<()> {
         std::fs::create_dir_all(&self.cfg.data_dir)?;
         let wal = &self.wal;
@@ -483,14 +486,17 @@ impl Server {
             }
             return Ok(());
         }
-        let records = committed_ops(&scan.records);
+        // 检查点以内的记录已经落盘，直接按边界截掉
+        let checkpoint = wal.read_checkpoint()?.unwrap_or(0);
+        let pending = records_after(&scan.records, checkpoint);
+        let records = committed_ops(&pending);
         if records.is_empty() {
             log_warn!(core, "discarding uncommitted WAL records");
-            wal.clear()?;
+            // 边界之后的都是没提交的尾巴，直接丢
+            wal.truncate_after(checkpoint)?;
             return Ok(());
         }
-        let max_lsn = scan
-            .records
+        let max_lsn = pending
             .iter()
             .rev()
             .find(|(_, op)| *op == WalOp::Commit)
@@ -504,7 +510,7 @@ impl Server {
         }
         self.flush_all_tables()?;
         wal.write_checkpoint(max_lsn)?;
-        wal.clear()?;
+        wal.truncate_before(max_lsn)?;
         log_info!(core, "WAL replay complete");
         Ok(())
     }

chusql-core/storage/src/wal.rs

@@ -7,6 +7,9 @@ use serde::{Deserialize, Serialize};
 
 use crate::protocol::{Account, Row, SchemaColumn, USERS_TABLE};
 
+/// 一帧的位置：LSN、操作、起止字节偏移
+type Frame = (u64, WalOp, usize, usize);
+
 // 预写日志：先记操作并 fsync，再改数据；重启按 LSN 重放已提交的组。
 
 const OP_INSERT: u8 = 1;
@@ -170,6 +173,51 @@ impl Wal {
         })
     }
 
+    /// 丢掉 LSN 不大于边界的帧，其余记录原样留下
+    ///
+    /// 边界以内的记录代表「已经落盘」，所以整段前缀可以丢弃；做法是把尾巴
+    /// 写进同目录的临时文件再改名，崩在中间也不会留下半新半旧的文件。
+    pub fn truncate_before(&self, lsn: u64) -> io::Result<()> {
+        let (frames, good_len) = self.parse()?;
+        let keep = frames
+            .iter()
+            .find(|(frame_lsn, _, _, _)| *frame_lsn > lsn)
+            .map(|(_, _, start, _)| *start);
+        match keep {
+            None => self.clear(),
+            Some(0) => Ok(()),
+            Some(start) => self.rewrite_tail(start, good_len),
+        }
+    }
+
+    /// 丢掉 LSN 大于边界的帧（未提交的尾巴），前缀原样留下
+    pub fn truncate_after(&self, lsn: u64) -> io::Result<()> {
+        let (frames, good_len) = self.parse()?;
+        let keep = frames
+            .iter()
+            .find(|(frame_lsn, _, _, _)| *frame_lsn > lsn)
+            .map(|(_, _, start, _)| *start)
+            .unwrap_or(good_len);
+        self.truncate_to(keep.min(good_len))
+    }
+
+    /// 把 [start, end) 一段写到文件头
+    fn rewrite_tail(&self, start: usize, end: usize) -> io::Result<()> {
+        // 常驻句柄认的是旧文件，改名后必须重新打开
+        *self.handle.lock().unwrap() = None;
+        let bytes = std::fs::read(&self.path)?;
+        let tail = bytes.get(start..end).unwrap_or(&[]);
+        let mut temp = self.path.clone().into_os_string();
+        temp.push(".tmp");
+        let temp = PathBuf::from(temp);
+        {
+            let mut f = File::create(&temp)?;
+            f.write_all(tail)?;
+            f.sync_all()?;
+        }
+        std::fs::rename(&temp, &self.path)
+    }
+
     /// 追加一条并 fsync
     pub fn append(&self, op: &WalOp) -> io::Result<()> {
         let bytes = self.encode_next(op)?;
@@ -202,16 +250,25 @@ impl Wal {
 
     /// 读全部记录，撕裂或损坏的尾巴按长度截断
     pub fn read_all(&self) -> io::Result<WalScan> {
+        let (frames, good_len) = self.parse()?;
+        Ok(WalScan {
+            records: frames
+                .into_iter()
+                .map(|(lsn, op, _, _)| (lsn, op))
+                .collect(),
+            good_len,
+        })
+    }
+
+    /// 解析整个文件：每帧的 LSN、操作与字节范围，外加完好长度
+    fn parse(&self) -> io::Result<(Vec<Frame>, usize)> {
         if !self.path.exists() {
-            return Ok(WalScan {
-                records: Vec::new(),
-                good_len: 0,
-            });
+            return Ok((Vec::new(), 0));
         }
         let mut f = File::open(&self.path)?;
         let mut buf = Vec::new();
         f.read_to_end(&mut buf)?;
-        let mut records = Vec::new();
+        let mut frames = Vec::new();
         let mut offset = 0usize;
         while offset + 4 <= buf.len() {
             let body_len =
@@ -222,17 +279,14 @@ impl Wal {
                 break;
             }
             match decode_body(&buf[offset + 4..end]) {
-                Ok(record) => {
-                    records.push(record);
+                Ok((lsn, op)) => {
+                    frames.push((lsn, op, offset, end));
                     offset = end;
                 }
                 Err(_) => break,
             }
         }
-        Ok(WalScan {
-            records,
-            good_len: offset,
-        })
+        Ok((frames, offset))
     }
 
     /// 记录检查点：到此为止的 LSN 都已落盘
@@ -258,11 +312,17 @@ impl Wal {
     }
 
     /// 取下一条 LSN
+    ///
+    /// 号从检查点与文件尾里较大的那个续，因此日志被按边界截断之后 LSN 也不回头。
     fn take_lsn(&self) -> io::Result<u64> {
         let mut slot = self.next_lsn.lock().unwrap();
         if *slot == 0 {
-            let last = self.read_all()?.records.last().map(|(lsn, _)| *lsn).unwrap_or(0);
-            *slot = last + 1;
+            let mut base = self.read_checkpoint()?.unwrap_or(0);
+            let (frames, _) = self.parse()?;
+            if let Some((lsn, _, _, _)) = frames.last() {
+                base = base.max(*lsn);
+            }
+            *slot = base + 1;
         }
         let lsn = *slot;
         *slot += 1;
@@ -293,6 +353,15 @@ pub fn committed_ops(records: &[(u64, WalOp)]) -> Vec<(u64, WalOp)> {
     out
 }
 
+/// 只留下检查点之后的记录：边界以内的已经落盘，不必重放
+pub fn records_after(records: &[(u64, WalOp)], checkpoint: u64) -> Vec<(u64, WalOp)> {
+    records
+        .iter()
+        .filter(|(lsn, _)| *lsn > checkpoint)
+        .cloned()
+        .collect()
+}
+
 /// 清空文件并把读写位置归零
 fn truncate(f: &mut File) -> io::Result<()> {
     f.set_len(0)?;

chusql-core/storage/tests/runtime_test.rs

@@ -1394,6 +1394,43 @@ fn already_applied_create_table_is_skipped_on_restart() -> Result<(), Box<dyn st
     Ok(())
 }
 
+/// 检查点跟着每组提交走，LSN 跨重启不回头
+#[test]
+fn checkpoint_advances_and_lsn_survives_restart() -> Result<(), Box<dyn std::error::Error>> {
+    let checkpoint = |data: &tempfile::TempDir| -> Result<u64, Box<dyn std::error::Error>> {
+        let bytes = std::fs::read(data.path().join("databases/main/wal.checkpoint"))?;
+        Ok(u64::from_le_bytes(bytes[..8].try_into()?))
+    };
+
+    let (server, data) = start_server();
+    let mut c = connect()?;
+    request_ok(&mut c, serde_json::json!({"method":"create_table","table":"m6_seq","columns":[{"name":"id","ty":"int"}]}))?;
+    request_ok(&mut c, serde_json::json!({"method":"insert","table":"m6_seq","row":{"id":1}}))?;
+    drop(c);
+    let first = checkpoint(&data)?;
+    assert!(first > 0, "提交后检查点要前进: {first}");
+    drop(server);
+
+    // 重启后接着写，新的 LSN 必须落在检查点之后
+    let _server = start_server_in(data.path());
+    let mut c = connect()?;
+    request_ok(&mut c, serde_json::json!({"method":"insert","table":"m6_seq","row":{"id":2}}))?;
+    drop(c);
+    let second = checkpoint(&data)?;
+    assert!(second > first, "重启后 LSN 要接着检查点走: {second} vs {first}");
+
+    // 数据都在，日志本身是空的：每组提交后就按边界截掉了
+    let mut c = connect()?;
+    let rows: serde_json::Value =
+        serde_json::from_str(&send(&mut c, r#"{"method":"scan","table":"m6_seq"}"#))?;
+    assert_eq!(rows["rows"].as_array().map(Vec::len), Some(2), "{rows}");
+    assert_eq!(
+        std::fs::metadata(data.path().join("databases/main/wal.log"))?.len(),
+        0
+    );
+    Ok(())
+}
+
 /// 给第二列建索引后能按那一列查
 #[test]
 fn create_index_on_secondary_column() {

chusql-core/storage/tests/wal_test.rs

@@ -1,5 +1,5 @@
 use chusql_core_storage::protocol::Row;
-use chusql_core_storage::wal::{committed_ops, Wal, WalOp};
+use chusql_core_storage::wal::{committed_ops, records_after, Wal, WalOp};
 use serde_json::json;
 
 // WAL 测试：追加、读取、清空、截断尾巴、LSN 与提交标记。
@@ -218,3 +218,81 @@ fn checkpoint_round_trips() {
     assert_eq!(wal.read_checkpoint().unwrap(), Some(7));
 }
 
+/// 按 LSN 截断只丢边界以内的帧
+#[test]
+fn truncate_before_keeps_newer_records() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("wal.log");
+    let wal = Wal::new(&path);
+
+    let second_op = WalOp::Insert {
+        table: "t".into(),
+        row: row(&[("id", json!(2))]),
+    };
+    let first = wal
+        .append_group(&[WalOp::Insert {
+            table: "t".into(),
+            row: row(&[("id", json!(1))]),
+        }])
+        .unwrap();
+    let second = wal.append_group(std::slice::from_ref(&second_op)).unwrap();
+    assert!(second > first);
+
+    // 提交号是第一组的最后一条；边界以内的操作与提交标记都该丢掉
+    wal.truncate_before(first).unwrap();
+    let kept = wal.read_all().unwrap();
+    assert!(kept.records.iter().all(|(lsn, _)| *lsn > first));
+    let ops: Vec<&WalOp> = kept
+        .records
+        .iter()
+        .map(|(_, op)| op)
+        .filter(|op| **op != WalOp::Commit)
+        .collect();
+    assert_eq!(ops, vec![&second_op], "只剩第二组的操作");
+
+    // 边界划到最后一条之后，日志就空了
+    wal.truncate_before(second).unwrap();
+    assert_eq!(std::fs::metadata(&path).unwrap().len(), 0);
+    assert!(wal.read_all().unwrap().records.is_empty());
+}
+
+/// 日志被截断后 LSN 也不回头
+#[test]
+fn lsn_continues_from_the_checkpoint() {
+    let dir = tempfile::tempdir().unwrap();
+    let path = dir.path().join("wal.log");
+    let op = WalOp::Insert {
+        table: "t".into(),
+        row: row(&[("id", json!(1))]),
+    };
+
+    let first = {
+        let wal = Wal::new(&path);
+        let lsn = wal.append_group(std::slice::from_ref(&op)).unwrap();
+        wal.write_checkpoint(lsn).unwrap();
+        wal.truncate_before(lsn).unwrap();
+        lsn
+    };
+
+    // 重新打开：号从检查点续
+    let wal = Wal::new(&path);
+    let next = wal.append_group(std::slice::from_ref(&op)).unwrap();
+    assert!(next > first, "重启后 LSN 要接着检查点走: {next} vs {first}");
+}
+
+/// 过滤只留下检查点之后的记录
+#[test]
+fn records_after_keeps_only_the_tail() {
+    let records = vec![
+        (1u64, WalOp::Insert { table: "t".into(), row: row(&[("id", json!(1))]) }),
+        (2, WalOp::Commit),
+        (3, WalOp::Insert { table: "t".into(), row: row(&[("id", json!(2))]) }),
+        (4, WalOp::Commit),
+    ];
+    let kept = records_after(&records, 2);
+    let lsn: Vec<u64> = kept.iter().map(|(lsn, _)| *lsn).collect();
+    assert_eq!(lsn, vec![3, 4]);
+    assert_eq!(records_after(&records, 4).len(), 0);
+    assert_eq!(records_after(&records, 0).len(), 4);
+}
+