diff --git a/storage/wal.go b/storage/wal.go index 39b6f63..d887db9 100644 --- a/storage/wal.go +++ b/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 { diff --git a/storage/wal_test.go b/storage/wal_test.go index de458a1..6877967 100644 --- a/storage/wal_test.go +++ b/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 实例做重放