diff --git a/platform/service/log_body_store.go b/platform/service/log_body_store.go index 3a7d66b..4f15394 100644 --- a/platform/service/log_body_store.go +++ b/platform/service/log_body_store.go @@ -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 { diff --git a/platform/service/log_ingest.go b/platform/service/log_ingest.go index 79f4c62..332bb39 100644 --- a/platform/service/log_ingest.go +++ b/platform/service/log_ingest.go @@ -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, diff --git a/platform/service/log_ingest_test.go b/platform/service/log_ingest_test.go index a7f405f..ddcfef9 100644 --- a/platform/service/log_ingest_test.go +++ b/platform/service/log_ingest_test.go @@ -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()