diff --git a/AGENTS.md b/AGENTS.md index ee090e6..e6d2c1a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -28,5 +28,5 @@ - 改动说明只在交流中记录。 ## 当前重点 -- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复。 +- 主线已完成:高级优化器 → 并行查询 → M1 会话隔离 → M2 快照事务 → M3 WAL 与崩溃恢复 → M4 表结构进 WAL → M5 系统表统一命名 → M6 检查点按 LSN 部分截断。 - 进度与计数见 `agent_credientials/project-plan.txt`,限制与风险见 `agent_credientials/known-issues.txt`。 diff --git a/chusql-core/storage/src/runtime.rs b/chusql-core/storage/src/runtime.rs index a28591e..96db318 100644 --- a/chusql-core/storage/src/runtime.rs +++ b/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(()) } diff --git a/chusql-core/storage/src/wal.rs b/chusql-core/storage/src/wal.rs index 0f788e0..81b2eea 100644 --- a/chusql-core/storage/src/wal.rs +++ b/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 { + 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, 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 { 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)?; diff --git a/chusql-core/storage/tests/runtime_test.rs b/chusql-core/storage/tests/runtime_test.rs index 0c1eb23..86020a9 100644 --- a/chusql-core/storage/tests/runtime_test.rs +++ b/chusql-core/storage/tests/runtime_test.rs @@ -1394,6 +1394,43 @@ fn already_applied_create_table_is_skipped_on_restart() -> Result<(), Box Result<(), Box> { + let checkpoint = |data: &tempfile::TempDir| -> Result> { + 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() { diff --git a/chusql-core/storage/tests/wal_test.rs b/chusql-core/storage/tests/wal_test.rs index 930f562..7185932 100644 --- a/chusql-core/storage/tests/wal_test.rs +++ b/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 = 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); +} + diff --git a/docs/architecture.md b/docs/architecture.md index 872c95f..e96ef7e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -51,7 +51,7 @@ chusql-server 显式事务在会话层实现:`BEGIN` 把当前数据库整库读成内存快照,事务内的语句都在快照上执行;`COMMIT` 计算快照与库的差分,用一条 `apply_transaction` 请求交给存储层应用;`ROLLBACK` 直接丢弃快照。因此隔离级别是快照读加原子提交:事务里看不到别的事务未提交或后来提交的改动,提交对外一次性可见。 -WAL 负责崩溃恢复:每条记录带自增 LSN,一组写操作与一条提交标记一次落盘;进程重启时按标记分组重放已提交的记录、丢弃没有标记的整组,重放前逐条做幂等检查,恢复结束把最大 LSN 写进 `wal.checkpoint` 并清空 WAL。`SystemStatus` 应答里的 `last_lsn` 报告该检查点 LSN。建表、删表、建索引、删索引与数据改动走同一条 WAL,恢复时一并重放。 +WAL 负责崩溃恢复:每条记录带自增 LSN,一组写操作与一条提交标记一次落盘;每组提交在数据和字典都刷盘之后把 `wal.checkpoint` 推进到这条提交的 LSN,含义是「到此为止的改动都已落盘」,再按这个边界丢掉日志前缀——所以日志里通常只有当组在飞的记录。进程重启时只重放检查点之后的记录,按标记分组取已提交的组、丢弃没有标记的整组,重放前逐条做幂等检查作为第二道保险,结束后再把检查点推到最后一条提交。LSN 从检查点与文件尾里较大的那个续号,跨重启单调,`SystemStatus` 应答里的 `last_lsn` 报告该检查点 LSN。建表、删表、建索引、删索引与数据改动走同一条 WAL,恢复时一并重放。 系统数据库及部分管理操作仅允许管理员访问。系统表统一用 `__system_` 前缀:账号表 `__system_users` 对语句通道隐藏,角色、授权、成员三张表(`__system_roles` / `__system_grants` / `__system_members`)只在 `system` 库里可见,权限服务自身也通过 SQL 读写它们;存储启动时把旧名 `sys_roles` / `sys_grants` / `sys_members` 连同数据与索引文件一起迁移过来。