admin343/browser
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Remove file log store memory mirror

Browse Source
This commit is contained in:
npc0-hue
2026-09-03 12:21:07 +08:00
parent 6311eb72ba
commit 5b82a42cc4
3 changed files with 162 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
platform/service/log_body_store.go
+126 -52
View File
@@ -10,6 +10,7 @@ import (
"os"
"path/filepath"
"sort"
"strconv"
"strings"
"sync"
@@ -103,7 +104,11 @@ func (store *MemoryLogBodyStore) LatestSeq(streamID string) (uint64, error) {
type FileLogBodyStore struct {
mu sync.Mutex
rootDir string
memory *MemoryLogBodyStore
}
type logSegmentRef struct {
firstSeq uint64
path string
}
func NewFileLogBodyStore(rootDir string) (*FileLogBodyStore, error) {
@@ -113,14 +118,10 @@ func NewFileLogBodyStore(rootDir string) (*FileLogBodyStore, error) {
}
store := &FileLogBodyStore{
rootDir: rootDir,
memory: NewMemoryLogBodyStore(),
}
if err := os.MkdirAll(rootDir, 0o700); err != nil {
return nil, fmt.Errorf("create log directory: %w", err)
}
if err := store.load(); err != nil {
return nil, err
}
return store, nil
}
@@ -132,17 +133,19 @@ func (store *FileLogBodyStore) AppendBatch(streamID string, record domain.LogBat
store.mu.Lock()
defer store.mu.Unlock()
if _, exists, err := store.memory.GetBatch(streamID, record.FirstSeq); err != nil {
return err
} else if exists {
return store.memory.AppendBatch(streamID, record)
}
streamDir := store.streamDir(streamID)
if err := os.MkdirAll(streamDir, 0o700); err != nil {
return fmt.Errorf("create log stream directory: %w", err)
}
segmentPath := store.segmentPath(streamID, record.FirstSeq)
if _, err := os.Stat(segmentPath); err == nil {
existing, err := readLogSegment(segmentPath)
if err != nil {
return err
}
if sameLogBatchRecord(existing, record) {
return nil
}
return validationError("log batch conflicts with acknowledged range")
} else if !os.IsNotExist(err) {
return fmt.Errorf("stat log segment: %w", err)
@@ -157,64 +160,76 @@ func (store *FileLogBodyStore) AppendBatch(streamID string, record domain.LogBat
if err := writeAtomicFile(segmentPath, []byte(body.String()), 0o600); err != nil {
return fmt.Errorf("persist log segment: %w", err)
}
return store.memory.AppendBatch(streamID, record)
return nil
}
func (store *FileLogBodyStore) GetBatch(streamID string, firstSeq uint64) (domain.LogBatchRecord, bool, error) {
return store.memory.GetBatch(streamID, firstSeq)
store.mu.Lock()
defer store.mu.Unlock()
record, err := readLogSegment(store.segmentPath(streamID, firstSeq))
if errors.Is(err, os.ErrNotExist) {
return domain.LogBatchRecord{}, false, nil
}
if err != nil {
return domain.LogBatchRecord{}, false, err
}
if len(record.Entries) == 0 {
return domain.LogBatchRecord{}, false, nil
}
return record, true, nil
}
func (store *FileLogBodyStore) Query(streamID string, afterSeq uint64, limit int) ([]domain.LogEntry, uint64, error) {
return store.memory.Query(streamID, afterSeq, limit)
store.mu.Lock()
defer store.mu.Unlock()
if limit <= 0 {
return nil, afterSeq, nil
}
segments, err := store.listSegments(streamID)
if err != nil {
return nil, 0, err
}
selected := make([]domain.LogEntry, 0, limit)
nextSeq := afterSeq
for index := firstLogSegmentIndex(segments, afterSeq); index < len(segments) && len(selected) < limit; index++ {
record, err := readLogSegment(segments[index].path)
if err != nil {
return nil, 0, err
}
for _, entry := range record.Entries {
if entry.Seq <= afterSeq {
continue
}
if len(selected) >= limit {
break
}
selected = append(selected, domain.CopyLogEntry(entry))
nextSeq = entry.Seq
}
}
return selected, nextSeq, nil
}
func (store *FileLogBodyStore) LatestSeq(streamID string) (uint64, error) {
return store.memory.LatestSeq(streamID)
}
store.mu.Lock()
defer store.mu.Unlock()
func (store *FileLogBodyStore) load() error {
entries, err := os.ReadDir(store.rootDir)
segments, err := store.listSegments(streamID)
if err != nil {
return fmt.Errorf("read log root: %w", err)
return 0, err
}
for _, entry := range entries {
if !entry.IsDir() {
continue
}
streamID, err := url.PathUnescape(entry.Name())
for index := len(segments) - 1; index >= 0; index-- {
record, err := readLogSegment(segments[index].path)
if err != nil {
return fmt.Errorf("decode log stream directory: %w", err)
return 0, err
}
if err := store.loadStream(streamID, filepath.Join(store.rootDir, entry.Name())); err != nil {
return err
if len(record.Entries) > 0 {
return record.LastSeq, nil
}
}
return nil
}
func (store *FileLogBodyStore) loadStream(streamID string, streamDir string) error {
segments, err := os.ReadDir(streamDir)
if err != nil {
return fmt.Errorf("read log stream directory: %w", err)
}
sort.SliceStable(segments, func(i, j int) bool { return segments[i].Name() < segments[j].Name() })
for _, segment := range segments {
if segment.IsDir() || !strings.HasPrefix(segment.Name(), "segment-") || !strings.HasSuffix(segment.Name(), ".jsonl") {
continue
}
segmentPath := filepath.Join(streamDir, segment.Name())
record, err := readLogSegment(segmentPath)
if err != nil {
return err
}
if len(record.Entries) == 0 {
continue
}
if err := store.memory.AppendBatch(streamID, record); err != nil {
return err
}
}
return nil
return 0, nil
}
func (store *FileLogBodyStore) streamDir(streamID string) string {
@@ -225,6 +240,65 @@ func (store *FileLogBodyStore) segmentPath(streamID string, firstSeq uint64) str
return filepath.Join(store.streamDir(streamID), fmt.Sprintf("segment-%020d.jsonl", firstSeq))
}
func (store *FileLogBodyStore) listSegments(streamID string) ([]logSegmentRef, error) {
streamDir := store.streamDir(streamID)
entries, err := os.ReadDir(streamDir)
if errors.Is(err, os.ErrNotExist) {
return nil, nil
}
if err != nil {
return nil, fmt.Errorf("read log stream directory: %w", err)
}
segments := make([]logSegmentRef, 0, len(entries))
for _, entry := range entries {
if entry.IsDir() {
continue
}
firstSeq, ok, err := logSegmentFirstSeq(entry.Name())
if err != nil {
return nil, err
}
if !ok {
continue
}
segments = append(segments, logSegmentRef{firstSeq: firstSeq, path: filepath.Join(streamDir, entry.Name())})
}
sort.SliceStable(segments, func(i, j int) bool { return segments[i].firstSeq < segments[j].firstSeq })
return segments, nil
}
func logSegmentFirstSeq(name string) (uint64, bool, error) {
if !strings.HasPrefix(name, "segment-") || !strings.HasSuffix(name, ".jsonl") {
return 0, false, nil
}
firstSeq, err := strconv.ParseUint(strings.TrimSuffix(strings.TrimPrefix(name, "segment-"), ".jsonl"), 10, 64)
if err != nil {
return 0, true, fmt.Errorf("parse log segment %s: %w", name, err)
}
return firstSeq, true, nil
}
func firstLogSegmentIndex(segments []logSegmentRef, afterSeq uint64) int {
index := sort.Search(len(segments), func(i int) bool { return segments[i].firstSeq > afterSeq })
if index < len(segments) && afterSeq != ^uint64(0) && segments[index].firstSeq == afterSeq+1 {
return index
}
if index > 0 {
return index - 1
}
return index
}
func sameLogBatchRecord(left domain.LogBatchRecord, right domain.LogBatchRecord) bool {
if left.FirstSeq != right.FirstSeq || left.LastSeq != right.LastSeq {
return false
}
if left.Checksum == right.Checksum {
return true
}
return len(left.Entries) == 1 && right.Checksum == validator.LogLineChecksum(left.Entries[0].Line)
}
func readLogSegment(path string) (domain.LogBatchRecord, error) {
file, err := os.Open(path)
if err != nil {
platform/service/log_ingest.go
+4 -5
View File
@@ -67,13 +67,12 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB
return domain.LogBatchIngestResult{}, validationError("log batch firstSeq must follow latest acknowledged sequence")
}
storedBatch := domain.CopyLogBatchIngest(batch)
record := domain.CopyLogBatchRecord(domain.LogBatchRecord{
record := domain.LogBatchRecord{
Checksum: batch.Checksum,
FirstSeq: batch.FirstSeq,
LastSeq: batch.LastSeq,
Entries: storedBatch.Entries,
})
Entries: batch.Entries,
}
if err := svc.logStore.AppendBatch(batch.LogStreamID, record); err != nil {
return domain.LogBatchIngestResult{}, err
}
@@ -84,7 +83,7 @@ func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogB
}
locked = false
lock.Unlock()
svc.publishLogEvents(stream, storedBatch.Entries)
svc.publishLogEvents(stream, batch.Entries)
return domain.LogBatchIngestResult{
Accepted: true,
LogStreamID: batch.LogStreamID,
platform/service/log_ingest_test.go
+32
View File
@@ -1,6 +1,7 @@
package service
import (
"os"
"path/filepath"
"strings"
"sync"
@@ -554,6 +555,37 @@ func TestFileLogBodyStoreReloadsVerbatimLongLogLine(t *testing.T) {
}
}
func TestFileLogBodyStoreStartupSkipsBodyLoadAndQueriesFromCursorSegment(t *testing.T) {
rootDir := filepath.Join(t.TempDir(), "logs")
store, err := NewFileLogBodyStore(rootDir)
if err != nil {
t.Fatalf("create file log store: %v", err)
}
first := []domain.LogEntry{{Seq: 1, Timestamp: time.Date(2026, 7, 3, 12, 0, 1, 0, time.UTC), Line: "old"}}
second := []domain.LogEntry{{Seq: 2, Timestamp: time.Date(2026, 7, 3, 12, 0, 2, 0, time.UTC), Line: "new"}}
if err := store.AppendBatch("log-1", domain.LogBatchRecord{Checksum: checksumForEntries(t, first), FirstSeq: 1, LastSeq: 1, Entries: first}); err != nil {
t.Fatalf("append first segment: %v", err)
}
if err := store.AppendBatch("log-1", domain.LogBatchRecord{Checksum: checksumForEntries(t, second), FirstSeq: 2, LastSeq: 2, Entries: second}); err != nil {
t.Fatalf("append second segment: %v", err)
}
if err := os.WriteFile(store.segmentPath("log-1", 1), []byte("{not-json\n"), 0o600); err != nil {
t.Fatalf("corrupt old segment: %v", err)
}
reloaded, err := NewFileLogBodyStore(rootDir)
if err != nil {
t.Fatalf("reload should not read all log bodies: %v", err)
}
selected, nextSeq, err := reloaded.Query("log-1", 1, 1)
if err != nil {
t.Fatalf("query from cursor segment: %v", err)
}
if len(selected) != 1 || selected[0].Seq != 2 || selected[0].Line != "new" || nextSeq != 2 {
t.Fatalf("unexpected cursor query: entries=%+v next=%d", selected, nextSeq)
}
}
func newRegisteredLogIngestService(t *testing.T) (*CoreService, string) {
t.Helper()
svc := newTestCoreService()