Keep Run startup independent from log backlog

spool/log_spool.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"os"
 	"path/filepath"
 	"sort"
@@ -23,7 +24,10 @@ type LogSpool struct {
 	inflight   map[string]bool
 }
 
-const maxAggregatedLogEntries = 128
+const (
+	maxAggregatedLogEntries   = 128
+	maxDurableLogSegmentBytes = 8 * 1024 * 1024
+)
 
 type logStreamWatermark struct {
 	Allocated    uint64 `json:"allocated"`
@@ -50,7 +54,26 @@ type pendingLogSegment struct {
 	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")
 	}
@@ -62,39 +85,23 @@ func NewLogSpool(dir string) (LogSpool, error) {
 	if err := spool.loadWatermarks(); err != nil {
 		return LogSpool{}, err
 	}
-	if err := spool.restorePendingWatermarks(); err != nil {
+	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
 }
 
-// restorePendingWatermarks supports upgrades from spools created before
+func (spool LogSpool) inspectPendingSegmentFiles() error {
-// per-stream watermarks existed. Pending durable segments are still an
-// allocation fact and must win over a remote progress lookup.
-func (spool *LogSpool) restorePendingWatermarks() error {
 	spool.mu.Lock()
 	defer spool.mu.Unlock()
-	pending, err := spool.pendingSegmentsLocked()
+	entries, err := os.ReadDir(spool.dir)
 	if err != nil {
-		return err
+		return fmt.Errorf("read log spool: %w", err)
-	}
-	changed := false
-	for _, segment := range pending {
-		batch := segment.batch
-		watermark := spool.watermarks[batch.LogStreamID]
-		if batch.LastSeq > watermark.Allocated {
-			watermark.Allocated = batch.LastSeq
-			changed = true
-		}
-		if batch.SourceCursor != nil && batch.SourceCursor.EndOffset > watermark.SourceOffset {
-			watermark.SourceOffset = batch.SourceCursor.EndOffset
-			changed = true
-		}
-		spool.watermarks[batch.LogStreamID] = watermark
-	}
-	if changed {
-		return spool.persistWatermarksLocked()
 	}
+	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
 }
 
@@ -102,6 +109,16 @@ func (spool LogSpool) NextSequence(ctx context.Context, streamID string, recover
 	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 {
@@ -140,6 +157,16 @@ func (spool LogSpool) EnqueueNextAggregated(ctx context.Context, batch protocol.
 	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 {
@@ -283,6 +310,43 @@ func (spool LogSpool) writeBatchAt(path string, batch durableLogBatch) error {
 	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()
@@ -298,35 +362,135 @@ func (spool LogSpool) Pending() ([]protocol.LogBatchIngestRequest, error) {
 }
 
 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, fmt.Errorf("read log spool: %w", err)
+		return nil, pendingSegmentScanStats{}, fmt.Errorf("read log spool: %w", err)
 	}
-	paths := make([]string, 0, len(entries))
+	paths, stats := spool.pendingSegmentPathsLocked(entries, options)
-	for _, entry := range entries {
-		if entry.IsDir() || !strings.HasSuffix(entry.Name(), ".json") {
-			continue
-		}
-		paths = append(paths, filepath.Join(spool.dir, entry.Name()))
-	}
-	sort.Strings(paths)
 	segments := make([]pendingLogSegment, 0, len(paths))
 	for _, path := range paths {
 		batch, err := readDurableLogBatch(path)
 		if err != nil {
 			if os.IsNotExist(err) {
+				stats.skipped++
 				continue
 			}
-			return nil, err
+			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, nil
+	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 {
@@ -542,19 +706,14 @@ func (spool LogSpool) RejectStreamAfter(logStreamID string, latestSeq uint64, re
 	if err != nil {
 		return 0, err
 	}
-	rejectedDir := filepath.Join(filepath.Dir(spool.dir), "logs-rejected")
 	rejected := 0
 	for _, segment := range segments {
 		batch := segment.batch.LogBatchIngestRequest
 		if batch.LogStreamID != logStreamID || batch.FirstSeq <= latestSeq {
 			continue
 		}
-		if err := os.MkdirAll(rejectedDir, 0o755); err != nil {
+		if err := spool.moveLogSegmentToRejectedLocked(segment.path, reason); err != nil {
-			return rejected, fmt.Errorf("create rejected log spool directory: %w", err)
+			return rejected, err
-		}
-		rejectedPath := filepath.Join(rejectedDir, fmt.Sprintf("%s.%s", filepath.Base(segment.path), sanitizeSegmentName(reason)))
-		if err := os.Rename(segment.path, rejectedPath); err != nil {
-			return rejected, fmt.Errorf("move rejected log spool segment: %w", err)
 		}
 		rejected++
 	}
@@ -579,18 +738,13 @@ func (spool LogSpool) Reject(batch protocol.LogBatchIngestRequest, reason string
 	if path == "" {
 		return nil
 	}
-	rejectedDir := filepath.Join(filepath.Dir(spool.dir), "logs-rejected")
+	return spool.moveLogSegmentToRejectedLocked(path, reason)
-	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", filepath.Base(path), sanitizeSegmentName(reason)))
-	if err := os.Rename(path, rejectedPath); err != nil {
-		return fmt.Errorf("move rejected log spool segment: %w", err)
-	}
-	return nil
 }
 
 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
@@ -603,6 +757,92 @@ func readDurableLogBatch(path string) (durableLogBatch, error) {
 	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 {

spool/log_spool_test.go

@@ -219,6 +219,53 @@ func TestLogSpoolRestoresPendingAllocationWithoutWatermark(t *testing.T) {
 	}
 }
 
+func TestLogSpoolStartupDoesNotDecodeCorruptPendingSegment(t *testing.T) {
+	root := t.TempDir()
+	logDir := filepath.Join(root, "logs")
+	if err := os.MkdirAll(logDir, 0o755); err != nil {
+		t.Fatalf("create log dir: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(logDir, "bad-00000000000000000001-00000000000000000001.json"), []byte("{"), 0o600); err != nil {
+		t.Fatalf("write corrupt segment: %v", err)
+	}
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("startup should not decode corrupt segment: %v", err)
+	}
+	pending, err := logSpool.Pending()
+	if err != nil || len(pending) != 0 {
+		t.Fatalf("expected corrupt segment to be quarantined when read, pending=%+v err=%v", pending, err)
+	}
+	rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
+	if err != nil || len(rejected) != 1 {
+		t.Fatalf("expected one rejected corrupt segment, rejected=%+v err=%v", rejected, err)
+	}
+}
+
+func TestLogSpoolStartupQuarantinesOversizedPendingSegment(t *testing.T) {
+	root := t.TempDir()
+	logDir := filepath.Join(root, "logs")
+	if err := os.MkdirAll(logDir, 0o755); err != nil {
+		t.Fatalf("create log dir: %v", err)
+	}
+	oversized := make([]byte, maxDurableLogSegmentBytes+1)
+	if err := os.WriteFile(filepath.Join(logDir, "huge-00000000000000000001-00000000000000000001.json"), oversized, 0o600); err != nil {
+		t.Fatalf("write oversized segment: %v", err)
+	}
+	logSpool, err := NewLogSpool(root)
+	if err != nil {
+		t.Fatalf("startup should quarantine oversized segment: %v", err)
+	}
+	pending, err := logSpool.Pending()
+	if err != nil || len(pending) != 0 {
+		t.Fatalf("expected oversized segment to be quarantined, pending=%+v err=%v", pending, err)
+	}
+	rejected, err := os.ReadDir(filepath.Join(root, "logs-rejected"))
+	if err != nil || len(rejected) != 1 {
+		t.Fatalf("expected one rejected oversized segment, rejected=%+v err=%v", rejected, err)
+	}
+}
+
 func TestLogSpoolSourceCursorDeduplicatesAcknowledgedReplayAfterRestart(t *testing.T) {
 	root := t.TempDir()
 	checksum := func(entries []protocol.LogEntry) (string, error) {