fix(wal): checkpoint 重写改为临时文件+原子 rename，堵住崩溃丢数据窗口

storage/wal.go

@@ -161,33 +161,71 @@ func (w *WAL) commit(batch []*walReq) {
 	}
 }
 
-// doRewrite 在 flushLoop 内把文件截断为空后重写全部 records 并 fsync。
+// doRewrite 通过"临时文件 + 原子 rename"把 WAL 内容整体替换为 records。
+// 不能就地 Truncate(0)+重写：那三步非原子，若在 Truncate 之后、重写 fsync 之前崩溃，
+// WAL 会残缺/清空——而快照里的 active+dirty 尚无 SSTable 副本，会丢已 ack 的写。
+// 先把 records 写满并 fsync 到 tmp，再 rename 覆盖（POSIX 原子），最后 fsync 父目录
+// 让 rename 本身持久。恢复时只会看到完整的旧 WAL 或完整的新 WAL，绝不残缺。
 // 只由 flushLoop 调用，保持文件唯一写者不变式。
 func (w *WAL) doRewrite(records []WALRecord) error {
-	if err := w.file.Truncate(0); err != nil {
+	tmpPath := w.path + ".tmp"
+	tmp, err := os.OpenFile(tmpPath, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0644)
+	if err != nil {
 		return err
 	}
-	// O_APPEND 下写入总落到文件末尾；Truncate(0) 后末尾即 0，故重写从头开始。
 	for _, r := range records {
-		if err := w.writeRecord(r.Op, r.Key, r.Value); err != nil {
+		if _, err := tmp.Write(encodeRecord(r.Op, r.Key, r.Value)); err != nil {
+			tmp.Close()
 			return err
 		}
 	}
-	return w.file.Sync()
+	if err := tmp.Sync(); err != nil {
+		tmp.Close()
+		return err
+	}
+	if err := tmp.Close(); err != nil {
+		return err
+	}
+	if err := os.Rename(tmpPath, w.path); err != nil {
+		return err
+	}
+	// 重新打开为新文件（旧 fd 指向已被覆盖的旧 inode）。flushLoop 是唯一写者，无并发。
+	_ = w.file.Close()
+	f, err := os.OpenFile(w.path, os.O_APPEND|os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return err
+	}
+	w.file = f
+	return fsyncDir(filepath.Dir(w.path))
 }
 
-// writeRecord 把单条记录字节写入文件（不 fsync）。O_APPEND 保证单次 Write 原子追加。
-func (w *WAL) writeRecord(op uint8, key, value []byte) error {
+// encodeRecord 编码一条 WAL 记录：[op u8][klen u32][vlen u32][key][value]（BigEndian）。
+func encodeRecord(op uint8, key, value []byte) []byte {
 	buf := make([]byte, 9, 9+len(key)+len(value))
 	buf[0] = op
 	binary.BigEndian.PutUint32(buf[1:5], uint32(len(key)))
 	binary.BigEndian.PutUint32(buf[5:9], uint32(len(value)))
 	buf = append(buf, key...)
 	buf = append(buf, value...)
-	_, err := w.file.Write(buf)
+	return buf
+}
+
+// writeRecord 把单条记录字节写入文件（不 fsync）。O_APPEND 保证单次 Write 原子追加。
+func (w *WAL) writeRecord(op uint8, key, value []byte) error {
+	_, err := w.file.Write(encodeRecord(op, key, value))
 	return err
 }
 
+// fsyncDir fsync 目录项，使其中文件的 rename 本身持久（崩溃后 rename 不丢失）。
+func fsyncDir(dir string) error {
+	d, err := os.Open(dir)
+	if err != nil {
+		return err
+	}
+	defer d.Close()
+	return d.Sync()
+}
+
 // Replay 从头读取全部记录，对每条调用 fn。读到残缺记录（撕裂尾写）即停止重放，
 // 返回 nil；底层 IO 错误或 fn 返回错误则向上抛出。
 func (w *WAL) Replay(fn func(op uint8, key, value []byte) error) error {

storage/wal_test.go

@@ -158,6 +158,71 @@ func TestWALConcurrentAppendGroupCommit(t *testing.T) {
 	_ = w.Close()
 }
 
+func TestWALRewriteAtomicReplace(t *testing.T) {
+	dir := t.TempDir()
+	path := filepath.Join(dir, "wal.log")
+	w, err := NewWAL(path)
+	if err != nil {
+		t.Fatalf("new wal: %v", err)
+	}
+
+	// 先写入若干旧记录。
+	for i := 0; i < 5; i++ {
+		if err := w.Append(WALOpPut, []byte(fmt.Sprintf("old%d", i)), []byte("x")); err != nil {
+			t.Fatalf("append: %v", err)
+		}
+	}
+
+	// 预置一个陈旧的 .tmp（模拟上次 rewrite 崩在中途的残留），不得干扰恢复。
+	if err := os.WriteFile(path+".tmp", []byte("garbage"), 0644); err != nil {
+		t.Fatalf("stale tmp: %v", err)
+	}
+
+	// Rewrite 为一批新记录（含墓碑、含空值），原子替换。
+	newRecs := []WALRecord{
+		{Op: WALOpPut, Key: []byte("a"), Value: []byte("1")},
+		{Op: WALOpDelete, Key: []byte("b"), Value: nil},
+		{Op: WALOpPut, Key: []byte("c"), Value: []byte{}},
+	}
+	if err := w.Rewrite(newRecs); err != nil {
+		t.Fatalf("rewrite: %v", err)
+	}
+
+	// 重写后应恰好只剩新记录，旧记录被整体替换，无残缺。
+	recs := replayAll(t, w)
+	if len(recs) != len(newRecs) {
+		t.Fatalf("want %d records after rewrite, got %d: %+v", len(newRecs), len(recs), recs)
+	}
+	if recs[0].op != WALOpPut || !bytes.Equal(recs[0].key, []byte("a")) || !bytes.Equal(recs[0].value, []byte("1")) {
+		t.Errorf("rec0 mismatch: %+v", recs[0])
+	}
+	if recs[1].op != WALOpDelete || !bytes.Equal(recs[1].key, []byte("b")) || len(recs[1].value) != 0 {
+		t.Errorf("rec1 (tombstone) mismatch: %+v", recs[1])
+	}
+	if recs[2].op != WALOpPut || !bytes.Equal(recs[2].key, []byte("c")) || len(recs[2].value) != 0 {
+		t.Errorf("rec2 (empty-value put) mismatch: %+v", recs[2])
+	}
+
+	// 重写后仍可继续正常 Append。
+	if err := w.Append(WALOpPut, []byte("d"), []byte("4")); err != nil {
+		t.Fatalf("append after rewrite: %v", err)
+	}
+	if got := replayAll(t, w); len(got) != len(newRecs)+1 {
+		t.Fatalf("append after rewrite not persisted: %d", len(got))
+	}
+	_ = w.Close()
+
+	// 重开（模拟重启）应仍读到完整数据——证明 rename 后的文件是完整的。
+	w2, err := NewWAL(path)
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer w2.Close()
+	if got := replayAll(t, w2); len(got) != len(newRecs)+1 {
+		t.Fatalf("reopen replay mismatch: %d", len(got))
+	}
+}
+
 func TestWALReplayMissingFile(t *testing.T) {
 	path := filepath.Join(t.TempDir(), "absent", "wal.log")
 	// 不创建文件，直接构造一个指向不存在路径的 WAL 实例做重放