npc0-hue
e73fe765c3
A segment that the platform already acknowledged could still be extended by the next tailed line, so its checksum covered entries stored under a different batch. The platform then rejected the same body every second while RejectStreamAfter skipped it, because it only quarantined segments that start after the platform latest. Every newly allocated line was quarantined by the following recovery pass, so live log ingest never resumed. Quarantine pending segments that reach beyond the acknowledged range, never extend a segment the platform already stored, and log the gap recovery so a stalled stream is diagnosable from the spool alone.
911 lines
31 KiB
Go
911 lines
31 KiB
Go
package spool
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"path/filepath"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"browser.local/run/protocol"
|
|
)
|
|
|
|
type LogSpool struct {
|
|
dir string
|
|
statePath string
|
|
mu *sync.Mutex
|
|
watermarks map[string]logStreamWatermark
|
|
inflight map[string]bool
|
|
}
|
|
|
|
const (
|
|
maxAggregatedLogEntries = 128
|
|
maxDurableLogSegmentBytes = 8 * 1024 * 1024
|
|
)
|
|
|
|
type logStreamWatermark struct {
|
|
Allocated uint64 `json:"allocated"`
|
|
Acknowledged uint64 `json:"acknowledged"`
|
|
SourceOffset int64 `json:"sourceOffset,omitempty"`
|
|
}
|
|
type logSpoolState struct {
|
|
Streams map[string]logStreamWatermark `json:"streams"`
|
|
}
|
|
|
|
type LogSourceCursor struct {
|
|
StartOffset int64 `json:"startOffset"`
|
|
EndOffset int64 `json:"endOffset"`
|
|
}
|
|
|
|
type durableLogBatch struct {
|
|
protocol.LogBatchIngestRequest
|
|
SourceCursor *LogSourceCursor `json:"_sourceCursor,omitempty"`
|
|
}
|
|
|
|
type pendingLogSegment struct {
|
|
path string
|
|
modTime time.Time
|
|
batch durableLogBatch
|
|
}
|
|
|
|
type pendingLogSegmentWatermark struct {
|
|
logStreamID string
|
|
lastSeq uint64
|
|
sourceCursor *LogSourceCursor
|
|
}
|
|
|
|
type pendingSegmentScanOptions struct {
|
|
diagnosticPhase string
|
|
}
|
|
|
|
type pendingSegmentScanStats struct {
|
|
candidates int
|
|
loaded int
|
|
skipped int
|
|
quarantined int
|
|
bytes int64
|
|
}
|
|
|
|
func NewLogSpool(dir string) (LogSpool, error) {
|
|
startedAt := time.Now()
|
|
if strings.TrimSpace(dir) == "" {
|
|
return LogSpool{}, fmt.Errorf("spool directory is required")
|
|
}
|
|
logDir := filepath.Join(dir, "logs")
|
|
if err := os.MkdirAll(logDir, 0o755); err != nil {
|
|
return LogSpool{}, fmt.Errorf("create log spool: %w", err)
|
|
}
|
|
spool := LogSpool{dir: logDir, statePath: filepath.Join(dir, "log-watermarks.json"), mu: &sync.Mutex{}, watermarks: map[string]logStreamWatermark{}, inflight: map[string]bool{}}
|
|
if err := spool.loadWatermarks(); err != nil {
|
|
return LogSpool{}, err
|
|
}
|
|
log.Printf("RUN phase=log_spool.init status=watermarks_loaded streams=%d durationMs=%d", len(spool.watermarks), time.Since(startedAt).Milliseconds())
|
|
if err := spool.inspectPendingSegmentFiles(); err != nil {
|
|
return LogSpool{}, err
|
|
}
|
|
log.Printf("RUN phase=log_spool.init status=complete streams=%d durationMs=%d", len(spool.watermarks), time.Since(startedAt).Milliseconds())
|
|
return spool, nil
|
|
}
|
|
|
|
func (spool LogSpool) inspectPendingSegmentFiles() error {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
entries, err := os.ReadDir(spool.dir)
|
|
if err != nil {
|
|
return fmt.Errorf("read log spool: %w", err)
|
|
}
|
|
paths, stats := spool.pendingSegmentPathsLocked(entries, pendingSegmentScanOptions{diagnosticPhase: "startup_inspect"})
|
|
log.Printf("RUN phase=log_spool.inspect status=complete candidates=%d accepted=%d skipped=%d quarantined=%d bytes=%d", stats.candidates, len(paths), stats.skipped, stats.quarantined, stats.bytes)
|
|
return nil
|
|
}
|
|
|
|
func (spool LogSpool) NextSequence(ctx context.Context, streamID string, recover func(context.Context, string) (uint64, error)) (uint64, error) {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
watermark, known := spool.watermarks[streamID]
|
|
if !known {
|
|
restored, restoredKnown, err := spool.restorePendingWatermarkLocked(streamID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if restoredKnown {
|
|
watermark = restored
|
|
known = true
|
|
}
|
|
}
|
|
if !known && recover != nil {
|
|
latest, err := recover(ctx, streamID)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
watermark = logStreamWatermark{Allocated: latest, Acknowledged: latest}
|
|
}
|
|
watermark.Allocated++
|
|
spool.watermarks[streamID] = watermark
|
|
if err := spool.persistWatermarksLocked(); err != nil {
|
|
return 0, err
|
|
}
|
|
return watermark.Allocated, nil
|
|
}
|
|
|
|
func (spool LogSpool) Enqueue(batch protocol.LogBatchIngestRequest) error {
|
|
return spool.enqueue(batch, nil)
|
|
}
|
|
|
|
// EnqueueAggregated extends the newest compatible durable segment so callers
|
|
// do not create one upload request for every process-output line.
|
|
func (spool LogSpool) EnqueueAggregated(batch protocol.LogBatchIngestRequest, checksum func([]protocol.LogEntry) (string, error)) error {
|
|
return spool.enqueue(batch, checksum)
|
|
}
|
|
|
|
// EnqueueNextAggregated allocates the next sequence and commits its batch
|
|
// under one spool lock. The durable segment is the source of truth; startup
|
|
// restores its watermark if the separate watermark snapshot was interrupted.
|
|
func (spool LogSpool) EnqueueNextAggregated(ctx context.Context, batch protocol.LogBatchIngestRequest, cursor *LogSourceCursor, recover func(context.Context, string) (uint64, error), checksum func([]protocol.LogEntry) (string, error)) (uint64, bool, error) {
|
|
if len(batch.Entries) != 1 {
|
|
return 0, false, fmt.Errorf("next aggregated log batch requires exactly one entry")
|
|
}
|
|
if cursor != nil && (cursor.StartOffset < 0 || cursor.EndOffset <= cursor.StartOffset) {
|
|
return 0, false, fmt.Errorf("source cursor range is invalid")
|
|
}
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
watermark, known := spool.watermarks[batch.LogStreamID]
|
|
if !known {
|
|
restored, restoredKnown, err := spool.restorePendingWatermarkLocked(batch.LogStreamID)
|
|
if err != nil {
|
|
return 0, false, err
|
|
}
|
|
if restoredKnown {
|
|
watermark = restored
|
|
known = true
|
|
}
|
|
}
|
|
if !known && recover != nil {
|
|
latest, err := recover(ctx, batch.LogStreamID)
|
|
if err != nil {
|
|
return 0, false, err
|
|
}
|
|
watermark = logStreamWatermark{Allocated: latest, Acknowledged: latest}
|
|
}
|
|
if cursor != nil && cursor.EndOffset <= watermark.SourceOffset {
|
|
return watermark.Allocated, false, nil
|
|
}
|
|
sequence := watermark.Allocated + 1
|
|
batch.FirstSeq = sequence
|
|
batch.LastSeq = sequence
|
|
batch.Entries[0].Seq = sequence
|
|
var err error
|
|
batch.Checksum, err = checksum(batch.Entries)
|
|
if err != nil {
|
|
return 0, false, err
|
|
}
|
|
if err := spool.writeAggregatedLocked(durableLogBatch{LogBatchIngestRequest: batch, SourceCursor: cursor}, checksum); err != nil {
|
|
return 0, false, err
|
|
}
|
|
watermark.Allocated = sequence
|
|
if cursor != nil {
|
|
watermark.SourceOffset = cursor.EndOffset
|
|
}
|
|
spool.watermarks[batch.LogStreamID] = watermark
|
|
// The batch is already synced and durable. A later startup reconstructs
|
|
// this watermark from pending segments if this snapshot cannot be written.
|
|
_ = spool.persistWatermarksLocked()
|
|
return sequence, true, nil
|
|
}
|
|
|
|
func (spool LogSpool) enqueue(batch protocol.LogBatchIngestRequest, checksum func([]protocol.LogEntry) (string, error)) error {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
watermark := spool.watermarks[batch.LogStreamID]
|
|
if batch.LastSeq > watermark.Allocated {
|
|
watermark.Allocated = batch.LastSeq
|
|
spool.watermarks[batch.LogStreamID] = watermark
|
|
if err := spool.persistWatermarksLocked(); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return spool.writeAggregatedLocked(durableLogBatch{LogBatchIngestRequest: batch}, checksum)
|
|
}
|
|
|
|
func (spool LogSpool) writeAggregatedLocked(batch durableLogBatch, checksum func([]protocol.LogEntry) (string, error)) error {
|
|
if checksum != nil && len(batch.Entries) == 1 {
|
|
if merged, previous, ok, err := spool.mergeLatest(batch, checksum); err != nil {
|
|
return err
|
|
} else if ok {
|
|
return spool.replaceBatch(previous.path, merged)
|
|
}
|
|
}
|
|
return spool.writeBatch(batch)
|
|
}
|
|
|
|
func (spool LogSpool) mergeLatest(batch durableLogBatch, checksum func([]protocol.LogEntry) (string, error)) (durableLogBatch, pendingLogSegment, bool, error) {
|
|
pending, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
return durableLogBatch{}, pendingLogSegment{}, false, err
|
|
}
|
|
acknowledged := spool.watermarks[batch.LogStreamID].Acknowledged
|
|
for index := len(pending) - 1; index >= 0; index-- {
|
|
previous := pending[index]
|
|
// Extending a segment the platform already stored would rewrite its
|
|
// checksum for a range that can no longer be uploaded, and the segment
|
|
// then becomes permanently unresendable. Leave it for the deduplicating
|
|
// retry and start a new segment instead.
|
|
if spool.inflight[previous.path] || previous.batch.LastSeq <= acknowledged || !compatibleLogBatch(previous.batch, batch) || previous.batch.LastSeq+1 != batch.FirstSeq || len(previous.batch.Entries)+len(batch.Entries) > maxAggregatedLogEntries || !compatibleSourceCursor(previous.batch.SourceCursor, batch.SourceCursor) {
|
|
continue
|
|
}
|
|
merged := previous.batch
|
|
merged.LastSeq = batch.LastSeq
|
|
merged.Entries = append(append([]protocol.LogEntry(nil), previous.batch.Entries...), batch.Entries...)
|
|
if merged.SourceCursor != nil {
|
|
cursor := *merged.SourceCursor
|
|
cursor.EndOffset = batch.SourceCursor.EndOffset
|
|
merged.SourceCursor = &cursor
|
|
}
|
|
merged.Checksum, err = checksum(merged.Entries)
|
|
if err != nil {
|
|
return durableLogBatch{}, pendingLogSegment{}, false, err
|
|
}
|
|
return merged, previous, true, nil
|
|
}
|
|
return durableLogBatch{}, pendingLogSegment{}, false, nil
|
|
}
|
|
|
|
func compatibleLogBatch(previous durableLogBatch, next durableLogBatch) bool {
|
|
return previous.LogStreamID == next.LogStreamID && previous.ServerInstanceID == next.ServerInstanceID && previous.StreamKey == next.StreamKey && previous.Source == next.Source && previous.Compression == next.Compression && previous.LogSessionID == next.LogSessionID && previous.SessionStartedAt.Equal(next.SessionStartedAt)
|
|
}
|
|
|
|
func compatibleSourceCursor(previous *LogSourceCursor, next *LogSourceCursor) bool {
|
|
if previous == nil || next == nil {
|
|
return previous == nil && next == nil
|
|
}
|
|
return next.StartOffset >= previous.EndOffset
|
|
}
|
|
|
|
func (spool LogSpool) writeBatch(batch durableLogBatch) error {
|
|
path := spool.batchPath(batch)
|
|
if existing, err := readDurableLogBatch(path); err == nil {
|
|
if existing.LogStreamID == batch.LogStreamID && existing.FirstSeq == batch.FirstSeq && existing.LastSeq == batch.LastSeq && existing.Checksum == batch.Checksum {
|
|
return nil
|
|
}
|
|
return fmt.Errorf("log spool segment conflicts with committed batch")
|
|
} else if !os.IsNotExist(err) {
|
|
return err
|
|
}
|
|
return spool.writeBatchAt(path, batch)
|
|
}
|
|
|
|
func (spool LogSpool) replaceBatch(path string, batch durableLogBatch) error {
|
|
if strings.TrimSpace(path) == "" {
|
|
return fmt.Errorf("log spool segment path is required")
|
|
}
|
|
return spool.writeBatchAt(path, batch)
|
|
}
|
|
|
|
func (spool LogSpool) writeBatchAt(path string, batch durableLogBatch) error {
|
|
tmp := path + ".tmp"
|
|
file, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
|
|
if err != nil {
|
|
return fmt.Errorf("open log spool segment: %w", err)
|
|
}
|
|
encodeErr := json.NewEncoder(file).Encode(batch)
|
|
closeErr := file.Close()
|
|
if encodeErr != nil {
|
|
_ = os.Remove(tmp)
|
|
return fmt.Errorf("encode log spool segment: %w", encodeErr)
|
|
}
|
|
if closeErr != nil {
|
|
_ = os.Remove(tmp)
|
|
return fmt.Errorf("close log spool segment: %w", closeErr)
|
|
}
|
|
if err := syncFile(tmp); err != nil {
|
|
_ = os.Remove(tmp)
|
|
return err
|
|
}
|
|
if err := os.Rename(tmp, path); err != nil {
|
|
_ = os.Remove(tmp)
|
|
return fmt.Errorf("commit log spool segment: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// restorePendingWatermarkLocked lazily supports upgrades from spools created
|
|
// before per-stream watermarks existed without making worker startup decode the
|
|
// entire durable log backlog.
|
|
func (spool LogSpool) restorePendingWatermarkLocked(streamID string) (logStreamWatermark, bool, error) {
|
|
if strings.TrimSpace(streamID) == "" {
|
|
return logStreamWatermark{}, false, nil
|
|
}
|
|
startedAt := time.Now()
|
|
pending, stats, err := spool.pendingSegmentWatermarksLocked(pendingSegmentScanOptions{diagnosticPhase: "lazy_restore"})
|
|
if err != nil {
|
|
return logStreamWatermark{}, false, err
|
|
}
|
|
watermark := spool.watermarks[streamID]
|
|
changed := false
|
|
for _, segment := range pending {
|
|
if segment.logStreamID != streamID {
|
|
continue
|
|
}
|
|
if segment.lastSeq > watermark.Allocated {
|
|
watermark.Allocated = segment.lastSeq
|
|
changed = true
|
|
}
|
|
if segment.sourceCursor != nil && segment.sourceCursor.EndOffset > watermark.SourceOffset {
|
|
watermark.SourceOffset = segment.sourceCursor.EndOffset
|
|
changed = true
|
|
}
|
|
}
|
|
if changed {
|
|
spool.watermarks[streamID] = watermark
|
|
if err := spool.persistWatermarksLocked(); err != nil {
|
|
return logStreamWatermark{}, false, err
|
|
}
|
|
}
|
|
log.Printf("RUN phase=log_spool.restore status=lazy_complete stream=%s found=%t candidates=%d loaded=%d skipped=%d quarantined=%d durationMs=%d", safeSpoolStreamID(streamID), changed, stats.candidates, stats.loaded, stats.skipped, stats.quarantined, time.Since(startedAt).Milliseconds())
|
|
return watermark, changed, nil
|
|
}
|
|
|
|
func (spool LogSpool) Pending() ([]protocol.LogBatchIngestRequest, error) {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
segments, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
batches := make([]protocol.LogBatchIngestRequest, 0, len(segments))
|
|
for _, segment := range segments {
|
|
batches = append(batches, segment.batch.LogBatchIngestRequest)
|
|
}
|
|
return batches, nil
|
|
}
|
|
|
|
func (spool LogSpool) pendingSegmentsLocked() ([]pendingLogSegment, error) {
|
|
segments, _, err := spool.pendingSegmentsLockedWithOptions(pendingSegmentScanOptions{})
|
|
return segments, err
|
|
}
|
|
|
|
func (spool LogSpool) pendingSegmentsLockedWithOptions(options pendingSegmentScanOptions) ([]pendingLogSegment, pendingSegmentScanStats, error) {
|
|
entries, err := os.ReadDir(spool.dir)
|
|
if err != nil {
|
|
return nil, pendingSegmentScanStats{}, fmt.Errorf("read log spool: %w", err)
|
|
}
|
|
paths, stats := spool.pendingSegmentPathsLocked(entries, options)
|
|
segments := make([]pendingLogSegment, 0, len(paths))
|
|
for _, path := range paths {
|
|
batch, err := readDurableLogBatch(path)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
stats.skipped++
|
|
continue
|
|
}
|
|
if spool.quarantineLogSegmentLocked(path, "unreadable") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined reason=unreadable file=%s error=%s", safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_skipped reason=unreadable file=%s error=%s", safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
var modTime time.Time
|
|
if info, statErr := os.Stat(path); statErr == nil {
|
|
modTime = info.ModTime()
|
|
}
|
|
segments = append(segments, pendingLogSegment{path: path, modTime: modTime, batch: batch})
|
|
stats.loaded++
|
|
}
|
|
sort.SliceStable(segments, func(i, j int) bool { return pendingLogSegmentBefore(segments[i], segments[j]) })
|
|
return segments, stats, nil
|
|
}
|
|
|
|
func (spool LogSpool) pendingSegmentWatermarksLocked(options pendingSegmentScanOptions) ([]pendingLogSegmentWatermark, pendingSegmentScanStats, error) {
|
|
entries, err := os.ReadDir(spool.dir)
|
|
if err != nil {
|
|
return nil, pendingSegmentScanStats{}, fmt.Errorf("read log spool: %w", err)
|
|
}
|
|
paths, stats := spool.pendingSegmentPathsLocked(entries, options)
|
|
segments := make([]pendingLogSegmentWatermark, 0, len(paths))
|
|
for _, path := range paths {
|
|
watermark, err := readDurableLogBatchWatermark(path)
|
|
if err != nil {
|
|
if os.IsNotExist(err) {
|
|
stats.skipped++
|
|
continue
|
|
}
|
|
if spool.quarantineLogSegmentLocked(path, "unreadable") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined phase=%s reason=unreadable file=%s error=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_skipped phase=%s reason=unreadable file=%s error=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
if strings.TrimSpace(watermark.LogStreamID) == "" {
|
|
if spool.quarantineLogSegmentLocked(path, "missing_stream") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined phase=%s reason=missing_stream file=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path))
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
continue
|
|
}
|
|
segments = append(segments, pendingLogSegmentWatermark{logStreamID: watermark.LogStreamID, lastSeq: watermark.LastSeq, sourceCursor: watermark.SourceCursor})
|
|
stats.loaded++
|
|
}
|
|
return segments, stats, nil
|
|
}
|
|
|
|
func (spool LogSpool) pendingSegmentPathsLocked(entries []os.DirEntry, options pendingSegmentScanOptions) ([]string, pendingSegmentScanStats) {
|
|
startedAt := time.Now()
|
|
paths := make([]string, 0, len(entries))
|
|
stats := pendingSegmentScanStats{}
|
|
if options.diagnosticPhase != "" {
|
|
log.Printf("RUN phase=log_spool.scan status=starting phase=%s entries=%d", safeDiagnosticPhase(options.diagnosticPhase), len(entries))
|
|
}
|
|
for _, entry := range entries {
|
|
if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
|
|
continue
|
|
}
|
|
stats.candidates++
|
|
path := filepath.Join(spool.dir, entry.Name())
|
|
info, err := entry.Info()
|
|
if err != nil {
|
|
if spool.quarantineLogSegmentLocked(path, "stat_failed") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined phase=%s reason=stat_failed file=%s error=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_skipped phase=%s reason=stat_failed file=%s error=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path), logSpoolErrorKind(err))
|
|
continue
|
|
}
|
|
if !info.Mode().IsRegular() {
|
|
if spool.quarantineLogSegmentLocked(path, "non_regular") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined phase=%s reason=non_regular file=%s", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path))
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
continue
|
|
}
|
|
if info.Size() > maxDurableLogSegmentBytes {
|
|
if spool.quarantineLogSegmentLocked(path, "oversized") == nil {
|
|
stats.quarantined++
|
|
log.Printf("RUN phase=log_spool.scan status=segment_quarantined phase=%s reason=oversized file=%s bytes=%d limit=%d", safeDiagnosticPhase(options.diagnosticPhase), safeSpoolFileName(path), info.Size(), maxDurableLogSegmentBytes)
|
|
continue
|
|
}
|
|
stats.skipped++
|
|
continue
|
|
}
|
|
stats.bytes += info.Size()
|
|
paths = append(paths, path)
|
|
if options.diagnosticPhase != "" && stats.candidates%100 == 0 {
|
|
log.Printf("RUN phase=log_spool.scan status=progress phase=%s candidates=%d accepted=%d quarantined=%d skipped=%d durationMs=%d", safeDiagnosticPhase(options.diagnosticPhase), stats.candidates, len(paths), stats.quarantined, stats.skipped, time.Since(startedAt).Milliseconds())
|
|
}
|
|
}
|
|
sort.Strings(paths)
|
|
if options.diagnosticPhase != "" {
|
|
log.Printf("RUN phase=log_spool.scan status=ready phase=%s candidates=%d accepted=%d quarantined=%d skipped=%d bytes=%d durationMs=%d", safeDiagnosticPhase(options.diagnosticPhase), stats.candidates, len(paths), stats.quarantined, stats.skipped, stats.bytes, time.Since(startedAt).Milliseconds())
|
|
}
|
|
return paths, stats
|
|
}
|
|
|
|
func pendingLogSegmentBefore(left pendingLogSegment, right pendingLogSegment) bool {
|
|
leftBatch := left.batch.LogBatchIngestRequest
|
|
rightBatch := right.batch.LogBatchIngestRequest
|
|
if leftBatch.LogStreamID == rightBatch.LogStreamID {
|
|
if leftBatch.FirstSeq != rightBatch.FirstSeq {
|
|
return leftBatch.FirstSeq < rightBatch.FirstSeq
|
|
}
|
|
return leftBatch.LastSeq < rightBatch.LastSeq
|
|
}
|
|
leftTime := pendingLogSegmentPriorityTime(left)
|
|
rightTime := pendingLogSegmentPriorityTime(right)
|
|
if !leftTime.Equal(rightTime) {
|
|
return leftTime.After(rightTime)
|
|
}
|
|
return left.path < right.path
|
|
}
|
|
|
|
func pendingLogSegmentPriorityTime(segment pendingLogSegment) time.Time {
|
|
batch := segment.batch.LogBatchIngestRequest
|
|
if batch.Source == "process" && strings.TrimSpace(batch.LogSessionID) != "" && !batch.SessionStartedAt.IsZero() {
|
|
return batch.SessionStartedAt
|
|
}
|
|
return segment.modTime
|
|
}
|
|
|
|
func (spool LogSpool) Ack(response protocol.LogBatchIngestResponse) error {
|
|
if response.AcceptedFrom == 0 || response.AcceptedTo < response.AcceptedFrom {
|
|
return nil
|
|
}
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
watermark := spool.watermarks[response.LogStreamID]
|
|
if response.AcceptedTo > watermark.Acknowledged {
|
|
watermark.Acknowledged = response.AcceptedTo
|
|
}
|
|
if watermark.Allocated < watermark.Acknowledged {
|
|
watermark.Allocated = watermark.Acknowledged
|
|
}
|
|
spool.watermarks[response.LogStreamID] = watermark
|
|
if err := spool.persistWatermarksLocked(); err != nil {
|
|
return err
|
|
}
|
|
segments, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, segment := range segments {
|
|
batch := segment.batch
|
|
if batch.LogStreamID == response.LogStreamID && batch.FirstSeq >= response.AcceptedFrom && batch.LastSeq <= response.AcceptedTo {
|
|
if err := os.Remove(segment.path); err != nil && !os.IsNotExist(err) {
|
|
return fmt.Errorf("remove acknowledged log spool segment: %w", err)
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (spool *LogSpool) loadWatermarks() error {
|
|
body, err := os.ReadFile(spool.statePath)
|
|
if os.IsNotExist(err) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("read log watermarks: %w", err)
|
|
}
|
|
var state logSpoolState
|
|
if err := json.Unmarshal(body, &state); err != nil {
|
|
return fmt.Errorf("decode log watermarks: %w", err)
|
|
}
|
|
if state.Streams != nil {
|
|
spool.watermarks = state.Streams
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (spool LogSpool) persistWatermarksLocked() error {
|
|
body, err := json.Marshal(logSpoolState{Streams: spool.watermarks})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
temporary := spool.statePath + ".tmp"
|
|
if err := os.WriteFile(temporary, body, 0o600); err != nil {
|
|
return fmt.Errorf("write log watermarks: %w", err)
|
|
}
|
|
if err := syncFile(temporary); err != nil {
|
|
_ = os.Remove(temporary)
|
|
return err
|
|
}
|
|
if err := os.Rename(temporary, spool.statePath); err != nil {
|
|
_ = os.Remove(temporary)
|
|
return fmt.Errorf("commit log watermarks: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type LogBatchClient interface {
|
|
IngestLogBatch(context.Context, protocol.LogBatchIngestRequest) (protocol.LogBatchIngestResponse, error)
|
|
}
|
|
|
|
type LogStreamProgressClient interface {
|
|
LogStreamLatestSeq(context.Context, protocol.LogBatchIngestRequest) (uint64, error)
|
|
}
|
|
|
|
type PermanentLogBatchError struct {
|
|
Reason string
|
|
Err error
|
|
}
|
|
|
|
func (err PermanentLogBatchError) Error() string {
|
|
if err.Err == nil {
|
|
return "permanent log batch rejection: " + err.Reason
|
|
}
|
|
return "permanent log batch rejection: " + err.Reason + ": " + err.Err.Error()
|
|
}
|
|
|
|
func (err PermanentLogBatchError) Unwrap() error {
|
|
return err.Err
|
|
}
|
|
|
|
func PermanentLogBatchRejection(reason string, err error) error {
|
|
return PermanentLogBatchError{Reason: sanitizeSegmentName(reason), Err: err}
|
|
}
|
|
|
|
func (spool LogSpool) Flush(ctx context.Context, client LogBatchClient) (int, error) {
|
|
if client == nil {
|
|
return 0, fmt.Errorf("log batch client is required")
|
|
}
|
|
spool.mu.Lock()
|
|
pending, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
spool.mu.Unlock()
|
|
return 0, err
|
|
}
|
|
for _, segment := range pending {
|
|
spool.inflight[segment.path] = true
|
|
}
|
|
spool.mu.Unlock()
|
|
defer func() {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
for _, segment := range pending {
|
|
delete(spool.inflight, segment.path)
|
|
}
|
|
}()
|
|
acknowledged := 0
|
|
for _, segment := range pending {
|
|
batch := segment.batch.LogBatchIngestRequest
|
|
if err := ctx.Err(); err != nil {
|
|
return acknowledged, err
|
|
}
|
|
response, err := client.IngestLogBatch(ctx, batch)
|
|
if err != nil {
|
|
var permanent PermanentLogBatchError
|
|
if errors.As(err, &permanent) {
|
|
if permanent.Reason == "platform_sequence_gap" {
|
|
latestSeq, progressErr := recoverLogStreamLatestSeq(ctx, client, batch)
|
|
if progressErr != nil {
|
|
return acknowledged, err
|
|
}
|
|
log.Printf("RUN phase=log_spool.flush status=sequence_gap stream=%s platformLatestSeq=%d batchFirstSeq=%d batchLastSeq=%d", safeSpoolStreamID(batch.LogStreamID), latestSeq, batch.FirstSeq, batch.LastSeq)
|
|
rejected, rejectErr := spool.RejectStreamAfter(batch.LogStreamID, latestSeq, permanent.Reason)
|
|
if rejectErr != nil {
|
|
return acknowledged, rejectErr
|
|
}
|
|
log.Printf("RUN phase=log_spool.flush status=sequence_gap_recovered stream=%s platformLatestSeq=%d rejectedSegments=%d", safeSpoolStreamID(batch.LogStreamID), latestSeq, rejected)
|
|
if resetErr := spool.ResetStreamWatermark(batch.LogStreamID, latestSeq); resetErr != nil {
|
|
return acknowledged, resetErr
|
|
}
|
|
acknowledged += rejected
|
|
return acknowledged, nil
|
|
}
|
|
if rejectErr := spool.Reject(batch, permanent.Reason); rejectErr != nil {
|
|
return acknowledged, rejectErr
|
|
}
|
|
acknowledged++
|
|
continue
|
|
}
|
|
return acknowledged, err
|
|
}
|
|
if !response.Accepted || response.LogStreamID != batch.LogStreamID || response.AcceptedFrom > batch.FirstSeq || response.AcceptedTo < batch.LastSeq {
|
|
return acknowledged, fmt.Errorf("platform log acknowledgement does not cover pending batch")
|
|
}
|
|
if err := spool.Ack(response); err != nil {
|
|
return acknowledged, err
|
|
}
|
|
acknowledged++
|
|
}
|
|
return acknowledged, nil
|
|
}
|
|
|
|
func recoverLogStreamLatestSeq(ctx context.Context, client LogBatchClient, batch protocol.LogBatchIngestRequest) (uint64, error) {
|
|
progressClient, ok := client.(LogStreamProgressClient)
|
|
if !ok {
|
|
return 0, fmt.Errorf("log stream progress client is required after sequence gap")
|
|
}
|
|
return progressClient.LogStreamLatestSeq(ctx, batch)
|
|
}
|
|
|
|
func (spool LogSpool) ResetStreamWatermark(logStreamID string, latestSeq uint64) error {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
watermark := spool.watermarks[logStreamID]
|
|
watermark.Allocated = latestSeq
|
|
watermark.Acknowledged = latestSeq
|
|
spool.watermarks[logStreamID] = watermark
|
|
return spool.persistWatermarksLocked()
|
|
}
|
|
|
|
func (spool LogSpool) RejectStreamAfter(logStreamID string, latestSeq uint64, reason string) (int, error) {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
segments, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
rejected := 0
|
|
for _, segment := range segments {
|
|
batch := segment.batch.LogBatchIngestRequest
|
|
// A pending segment is only resendable when it lies entirely inside the
|
|
// platform's acknowledged range (it is then deduplicated by sequence
|
|
// checksum). Every segment that reaches beyond the platform latest, or
|
|
// that straddles it, can never be accepted again: its checksum covers
|
|
// entries the platform already stored under a different batch. Keeping
|
|
// such a straddling segment pending made the spool resend the same
|
|
// rejected body forever, and every newly allocated line was quarantined
|
|
// by the next recovery pass, so the stream never resumed.
|
|
if batch.LogStreamID != logStreamID || batch.LastSeq <= latestSeq {
|
|
continue
|
|
}
|
|
if err := spool.moveLogSegmentToRejectedLocked(segment.path, reason); err != nil {
|
|
return rejected, err
|
|
}
|
|
rejected++
|
|
}
|
|
return rejected, nil
|
|
}
|
|
|
|
func (spool LogSpool) Reject(batch protocol.LogBatchIngestRequest, reason string) error {
|
|
spool.mu.Lock()
|
|
defer spool.mu.Unlock()
|
|
segments, err := spool.pendingSegmentsLocked()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
path := ""
|
|
for _, segment := range segments {
|
|
current := segment.batch
|
|
if current.LogStreamID == batch.LogStreamID && current.FirstSeq == batch.FirstSeq && current.LastSeq == batch.LastSeq && current.Checksum == batch.Checksum {
|
|
path = segment.path
|
|
break
|
|
}
|
|
}
|
|
if path == "" {
|
|
return nil
|
|
}
|
|
return spool.moveLogSegmentToRejectedLocked(path, reason)
|
|
}
|
|
|
|
func readDurableLogBatch(path string) (durableLogBatch, error) {
|
|
if err := validateDurableLogSegment(path); err != nil {
|
|
return durableLogBatch{}, err
|
|
}
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return durableLogBatch{}, err
|
|
}
|
|
defer file.Close()
|
|
var batch durableLogBatch
|
|
if err := json.NewDecoder(file).Decode(&batch); err != nil {
|
|
return durableLogBatch{}, fmt.Errorf("decode log spool segment: %w", err)
|
|
}
|
|
return batch, nil
|
|
}
|
|
|
|
func readDurableLogBatchWatermark(path string) (struct {
|
|
LogStreamID string `json:"logStreamId"`
|
|
LastSeq uint64 `json:"lastSeq"`
|
|
SourceCursor *LogSourceCursor `json:"_sourceCursor,omitempty"`
|
|
}, error) {
|
|
var watermark struct {
|
|
LogStreamID string `json:"logStreamId"`
|
|
LastSeq uint64 `json:"lastSeq"`
|
|
SourceCursor *LogSourceCursor `json:"_sourceCursor,omitempty"`
|
|
}
|
|
if err := validateDurableLogSegment(path); err != nil {
|
|
return watermark, err
|
|
}
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return watermark, err
|
|
}
|
|
defer file.Close()
|
|
if err := json.NewDecoder(file).Decode(&watermark); err != nil {
|
|
return watermark, fmt.Errorf("decode log spool segment watermark: %w", err)
|
|
}
|
|
return watermark, nil
|
|
}
|
|
|
|
func validateDurableLogSegment(path string) error {
|
|
info, err := os.Stat(path)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if !info.Mode().IsRegular() {
|
|
return fmt.Errorf("log spool segment is not a regular file")
|
|
}
|
|
if info.Size() > maxDurableLogSegmentBytes {
|
|
return fmt.Errorf("log spool segment exceeds %d bytes", maxDurableLogSegmentBytes)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (spool LogSpool) quarantineLogSegmentLocked(path string, reason string) error {
|
|
return spool.moveLogSegmentToRejectedLocked(path, reason)
|
|
}
|
|
|
|
func (spool LogSpool) moveLogSegmentToRejectedLocked(path string, reason string) error {
|
|
rejectedDir := filepath.Join(filepath.Dir(spool.dir), "logs-rejected")
|
|
if err := os.MkdirAll(rejectedDir, 0o755); err != nil {
|
|
return fmt.Errorf("create rejected log spool directory: %w", err)
|
|
}
|
|
rejectedPath := filepath.Join(rejectedDir, fmt.Sprintf("%s.%s.%d", filepath.Base(path), sanitizeSegmentName(reason), time.Now().UTC().UnixNano()))
|
|
if err := os.Rename(path, rejectedPath); err != nil {
|
|
return fmt.Errorf("move rejected log spool segment: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func safeSpoolFileName(path string) string {
|
|
return sanitizeSegmentName(filepath.Base(path))
|
|
}
|
|
|
|
func safeSpoolStreamID(streamID string) string {
|
|
return sanitizeSegmentName(streamID)
|
|
}
|
|
|
|
func safeDiagnosticPhase(value string) string {
|
|
if strings.TrimSpace(value) == "" {
|
|
return "-"
|
|
}
|
|
return sanitizeSegmentName(value)
|
|
}
|
|
|
|
func logSpoolErrorKind(err error) string {
|
|
if err == nil {
|
|
return "-"
|
|
}
|
|
if os.IsNotExist(err) {
|
|
return "not_found"
|
|
}
|
|
if os.IsPermission(err) {
|
|
return "permission"
|
|
}
|
|
var pathErr *os.PathError
|
|
if errors.As(err, &pathErr) {
|
|
return sanitizeSegmentName(pathErr.Op + "_" + pathErr.Err.Error())
|
|
}
|
|
return sanitizeSegmentName(err.Error())
|
|
}
|
|
|
|
func syncFile(path string) error {
|
|
file, err := os.OpenFile(path, os.O_RDWR, 0)
|
|
if err != nil {
|
|
return fmt.Errorf("open spool file for sync: %w", err)
|
|
}
|
|
defer file.Close()
|
|
if err := file.Sync(); err != nil {
|
|
return fmt.Errorf("sync spool file: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (spool LogSpool) batchPath(batch durableLogBatch) string {
|
|
streamID := sanitizeSegmentName(batch.LogStreamID)
|
|
return filepath.Join(spool.dir, fmt.Sprintf("%s-%020d-%020d.json", streamID, batch.FirstSeq, batch.LastSeq))
|
|
}
|
|
|
|
func (spool LogSpool) watermarkPath(streamID string) string {
|
|
return filepath.Join(filepath.Dir(spool.dir), "log-watermarks", sanitizeSegmentName(streamID)+".json")
|
|
}
|
|
|
|
func (spool LogSpool) readWatermark(streamID string) (uint64, error) {
|
|
file, err := os.Open(spool.watermarkPath(streamID))
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
defer file.Close()
|
|
var watermark struct {
|
|
LatestSeq uint64 `json:"latestSeq"`
|
|
}
|
|
if err := json.NewDecoder(file).Decode(&watermark); err != nil {
|
|
return 0, fmt.Errorf("decode log watermark: %w", err)
|
|
}
|
|
return watermark.LatestSeq, nil
|
|
}
|
|
|
|
func sanitizeSegmentName(value string) string {
|
|
var builder strings.Builder
|
|
for _, r := range value {
|
|
if r >= 'a' && r <= 'z' || r >= 'A' && r <= 'Z' || r >= '0' && r <= '9' || r == '-' || r == '_' || r == '.' {
|
|
builder.WriteRune(r)
|
|
continue
|
|
}
|
|
builder.WriteByte('_')
|
|
}
|
|
if builder.Len() == 0 {
|
|
return "stream"
|
|
}
|
|
return builder.String()
|
|
}
|