94 lines
1.9 KiB
Go
94 lines
1.9 KiB
Go
package checkpoint
|
|
|
|
import (
|
|
"encoding/json"
|
|
"os"
|
|
"path/filepath"
|
|
"sync"
|
|
)
|
|
|
|
// FileStore 将检查点持久化到本地 JSON 文件，适合 Agent 本地 WAL 场景。
|
|
type FileStore struct {
|
|
path string
|
|
|
|
mu sync.RWMutex
|
|
data fileData
|
|
}
|
|
|
|
type fileData struct {
|
|
LogOffsets map[string]int64 `json:"log_offsets"`
|
|
AckedSeq map[string]int64 `json:"acked_seq"`
|
|
}
|
|
|
|
// NewFileStore 加载（或初始化）指定路径的检查点文件。
|
|
func NewFileStore(path string) (*FileStore, error) {
|
|
s := &FileStore{
|
|
path: path,
|
|
data: fileData{
|
|
LogOffsets: map[string]int64{},
|
|
AckedSeq: map[string]int64{},
|
|
},
|
|
}
|
|
|
|
raw, err := os.ReadFile(path)
|
|
switch {
|
|
case err == nil:
|
|
if err := json.Unmarshal(raw, &s.data); err != nil {
|
|
return nil, err
|
|
}
|
|
case os.IsNotExist(err):
|
|
// 首次运行，使用空检查点。
|
|
default:
|
|
return nil, err
|
|
}
|
|
|
|
if s.data.LogOffsets == nil {
|
|
s.data.LogOffsets = map[string]int64{}
|
|
}
|
|
if s.data.AckedSeq == nil {
|
|
s.data.AckedSeq = map[string]int64{}
|
|
}
|
|
return s, nil
|
|
}
|
|
|
|
func (s *FileStore) persistLocked() error {
|
|
raw, err := json.MarshalIndent(s.data, "", " ")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if dir := filepath.Dir(s.path); dir != "." {
|
|
if err := os.MkdirAll(dir, 0o755); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return os.WriteFile(s.path, raw, 0o644)
|
|
}
|
|
|
|
func (s *FileStore) GetLogOffset(file string) (int64, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.data.LogOffsets[file], nil
|
|
}
|
|
|
|
func (s *FileStore) SetLogOffset(file string, offset int64) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.data.LogOffsets[file] = offset
|
|
return s.persistLocked()
|
|
}
|
|
|
|
func (s *FileStore) GetAckedSeq(hostID string) (int64, error) {
|
|
s.mu.RLock()
|
|
defer s.mu.RUnlock()
|
|
return s.data.AckedSeq[hostID], nil
|
|
}
|
|
|
|
func (s *FileStore) SetAckedSeq(hostID string, seq int64) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
s.data.AckedSeq[hostID] = seq
|
|
return s.persistLocked()
|
|
}
|
|
|
|
func (s *FileStore) Close() error { return nil }
|