147 lines
3.7 KiB
Go
147 lines
3.7 KiB
Go
// Package wal 实现 Agent 本地 WAL（Write-Ahead Log）。
|
||
//
|
||
// WAL 以追加方式持久化待上报的批次，配合 checkpoint.Store 记录的已确认
|
||
// 序号，实现 Agent 进程重启后从未确认点继续上报（断点续传）。每个批次
|
||
// 使用单调递增的 seq 标识，写入顺序与发送顺序一致。
|
||
package wal
|
||
|
||
import (
|
||
"encoding/binary"
|
||
"fmt"
|
||
"io"
|
||
"os"
|
||
"sync"
|
||
)
|
||
|
||
// Record 是 WAL 中的一条记录。Payload 为已序列化的批次（protobuf 字节）。
|
||
type Record struct {
|
||
Seq int64
|
||
Payload []byte
|
||
}
|
||
|
||
// WAL 是一个简单的追加式文件日志。记录格式：
|
||
//
|
||
// [seq:8字节 big-endian][len:4字节 big-endian][payload:len字节]
|
||
type WAL struct {
|
||
path string
|
||
|
||
mu sync.Mutex
|
||
f *os.File
|
||
}
|
||
|
||
// Open 打开（不存在则创建）指定路径的 WAL 文件。
|
||
func Open(path string) (*WAL, error) {
|
||
f, err := os.OpenFile(path, os.O_CREATE|os.O_RDWR, 0o644)
|
||
if err != nil {
|
||
return nil, fmt.Errorf("wal: open %s: %w", path, err)
|
||
}
|
||
return &WAL{path: path, f: f}, nil
|
||
}
|
||
|
||
// Append 追加一条记录并立即刷盘，保证进程崩溃后可恢复。
|
||
func (w *WAL) Append(seq int64, payload []byte) error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
|
||
var hdr [12]byte
|
||
binary.BigEndian.PutUint64(hdr[0:8], uint64(seq))
|
||
binary.BigEndian.PutUint32(hdr[8:12], uint32(len(payload)))
|
||
|
||
if _, err := w.f.Write(hdr[:]); err != nil {
|
||
return fmt.Errorf("wal: write header: %w", err)
|
||
}
|
||
if _, err := w.f.Write(payload); err != nil {
|
||
return fmt.Errorf("wal: write payload: %w", err)
|
||
}
|
||
if err := w.f.Sync(); err != nil {
|
||
return fmt.Errorf("wal: sync: %w", err)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Load 读取 WAL 中的全部记录，按写入顺序返回。
|
||
func (w *WAL) Load() ([]Record, error) {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
return w.loadLocked()
|
||
}
|
||
|
||
func (w *WAL) loadLocked() ([]Record, error) {
|
||
if _, err := w.f.Seek(0, io.SeekStart); err != nil {
|
||
return nil, fmt.Errorf("wal: seek: %w", err)
|
||
}
|
||
|
||
var out []Record
|
||
for {
|
||
var hdr [12]byte
|
||
_, err := io.ReadFull(w.f, hdr[:])
|
||
switch {
|
||
case err == io.EOF:
|
||
return out, nil
|
||
case err == io.ErrUnexpectedEOF:
|
||
// 尾部存在半条记录（进程被强杀），视为损坏并忽略。
|
||
return out, nil
|
||
case err != nil:
|
||
return nil, fmt.Errorf("wal: read header: %w", err)
|
||
}
|
||
|
||
seq := int64(binary.BigEndian.Uint64(hdr[0:8]))
|
||
n := binary.BigEndian.Uint32(hdr[8:12])
|
||
payload := make([]byte, n)
|
||
if _, err := io.ReadFull(w.f, payload); err != nil {
|
||
return nil, fmt.Errorf("wal: read payload: %w", err)
|
||
}
|
||
out = append(out, Record{Seq: seq, Payload: payload})
|
||
}
|
||
}
|
||
|
||
// TruncateBefore 删除 seq 小于 keepSeq 的记录（即保留 seq >= keepSeq）。
|
||
// 通常在服务端确认后调用，以避免 WAL 无限增长。
|
||
func (w *WAL) TruncateBefore(keepSeq int64) error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
|
||
recs, err := w.loadLocked()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
kept := make([]Record, 0, len(recs))
|
||
for _, r := range recs {
|
||
if r.Seq >= keepSeq {
|
||
kept = append(kept, r)
|
||
}
|
||
}
|
||
|
||
if err := w.f.Truncate(0); err != nil {
|
||
return fmt.Errorf("wal: truncate: %w", err)
|
||
}
|
||
if _, err := w.f.Seek(0, io.SeekStart); err != nil {
|
||
return fmt.Errorf("wal: seek: %w", err)
|
||
}
|
||
for _, r := range kept {
|
||
var hdr [12]byte
|
||
binary.BigEndian.PutUint64(hdr[0:8], uint64(r.Seq))
|
||
binary.BigEndian.PutUint32(hdr[8:12], uint32(len(r.Payload)))
|
||
if _, err := w.f.Write(hdr[:]); err != nil {
|
||
return fmt.Errorf("wal: rewrite header: %w", err)
|
||
}
|
||
if _, err := w.f.Write(r.Payload); err != nil {
|
||
return fmt.Errorf("wal: rewrite payload: %w", err)
|
||
}
|
||
}
|
||
return w.f.Sync()
|
||
}
|
||
|
||
// Close 关闭底层文件。
|
||
func (w *WAL) Close() error {
|
||
w.mu.Lock()
|
||
defer w.mu.Unlock()
|
||
if w.f == nil {
|
||
return nil
|
||
}
|
||
err := w.f.Close()
|
||
w.f = nil
|
||
return err
|
||
}
|