spool: handle startup backlog and resync after overflow

internal/spool/watcher.go

@@ -2,6 +2,7 @@ package spool
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -95,14 +96,16 @@ func (w *Watcher) Events() <-chan string {
 
 // Start begins watching for new files
 func (w *Watcher) Start(ctx context.Context) error {
+	// Track file modification times for stability check
+	fileStability := make(map[string]time.Time)
+
 	// First, process any existing files in the spool
-	if err := w.processExistingFiles(); err != nil {
+	if existing, err := w.processExistingFiles(); err != nil {
 		logutil.Warn("Failed to process existing files: %v", err)
+	} else {
+		w.seedExistingFiles(existing, fileStability)
 	}
 
-	// Track file modification times for stability check
-	fileStability := make(map[string]time.Time)
-
 	// Start stability checker goroutine
 	stabilityTicker := time.NewTicker(w.checkInterval)
 	defer stabilityTicker.Stop()
@@ -126,31 +129,17 @@ func (w *Watcher) Start(ctx context.Context) error {
 			// Only care about Create and Write events
 			if event.Op&fsnotify.Create == fsnotify.Create ||
 				event.Op&fsnotify.Write == fsnotify.Write {
-				w.stabMu.Lock()
-				// Check if we're at max capacity
-				if len(fileStability) >= w.maxPendingFiles {
-					log.Printf("Warning: max pending files reached (%d), dropping oldest", w.maxPendingFiles)
-					// Remove oldest entry
-					var oldest string
-					var oldestTime time.Time
-					for p, t := range fileStability {
-						if oldest == "" || t.Before(oldestTime) {
-							oldest = p
-							oldestTime = t
-						}
-					}
-					delete(fileStability, oldest)
-				}
-				// Mark file as recently modified
-				fileStability[event.Name] = time.Now()
-				w.stabMu.Unlock()
+				w.trackFile(fileStability, event.Name, time.Now())
 			}
 
 		case err, ok := <-w.watcher.Errors:
 			if !ok {
 				return fmt.Errorf("watcher errors channel closed")
 			}
 			log.Printf("Watcher error: %v", err)
+			if errors.Is(err, fsnotify.ErrEventOverflow) {
+				w.resyncFiles(fileStability)
+			}
 
 		case <-stabilityTicker.C:
 			// Check for stable files
@@ -242,14 +231,20 @@ func (w *Watcher) copyFile(src, dst string) error {
 	return dstFile.Sync()
 }
 
+type existingFile struct {
+	path    string
+	modTime time.Time
+}
+
 // processExistingFiles scans the spool directory for existing files
-func (w *Watcher) processExistingFiles() error {
+func (w *Watcher) processExistingFiles() ([]existingFile, error) {
 	newDir := filepath.Join(w.spoolDir, "new")
 	entries, err := os.ReadDir(newDir)
 	if err != nil {
-		return err
+		return nil, err
 	}
 
+	var existing []existingFile
 	for _, entry := range entries {
 		if entry.IsDir() {
 			continue
@@ -264,15 +259,62 @@ func (w *Watcher) processExistingFiles() error {
 			continue
 		}
 
-		if time.Since(info.ModTime()) >= w.stabilityWait {
-			w.eventChan <- path
-		}
+		existing = append(existing, existingFile{path: path, modTime: info.ModTime()})
 	}
 
-	return nil
+	return existing, nil
 }
 
 // Close stops the watcher and releases resources
 func (w *Watcher) Close() error {
 	return w.watcher.Close()
 }
+
+// seedExistingFiles enqueues existing files without blocking the watcher startup.
+func (w *Watcher) seedExistingFiles(existing []existingFile, fileStability map[string]time.Time) {
+	now := time.Now()
+	for _, f := range existing {
+		age := now.Sub(f.modTime)
+		if age >= w.stabilityWait {
+			select {
+			case w.eventChan <- f.path:
+				continue
+			default:
+			}
+		}
+		w.trackFile(fileStability, f.path, f.modTime)
+	}
+}
+
+// resyncFiles rescans the spool directory and seeds any files that may have been missed (e.g., after fsnotify overflow).
+func (w *Watcher) resyncFiles(fileStability map[string]time.Time) {
+	existing, err := w.processExistingFiles()
+	if err != nil {
+		logutil.Warn("Failed to resync spool directory: %v", err)
+		return
+	}
+	w.seedExistingFiles(existing, fileStability)
+}
+
+// trackFile records a path in the stability map, respecting the maxPendingFiles limit.
+func (w *Watcher) trackFile(fileStability map[string]time.Time, path string, modTime time.Time) {
+	w.stabMu.Lock()
+	defer w.stabMu.Unlock()
+
+	// Check if we're at max capacity
+	if len(fileStability) >= w.maxPendingFiles {
+		log.Printf("Warning: max pending files reached (%d), dropping oldest", w.maxPendingFiles)
+		// Remove oldest entry
+		var oldest string
+		var oldestTime time.Time
+		for p, t := range fileStability {
+			if oldest == "" || t.Before(oldestTime) {
+				oldest = p
+				oldestTime = t
+			}
+		}
+		delete(fileStability, oldest)
+	}
+	// Mark file as recently modified
+	fileStability[path] = modTime
+}

internal/spool/watcher_test.go

@@ -2,10 +2,13 @@ package spool
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
+
+	"github.com/fsnotify/fsnotify"
 )
 
 func TestNewWatcher(t *testing.T) {
@@ -353,3 +356,130 @@ func TestWatcherContextCancellation(t *testing.T) {
 		t.Error("Events channel not closed")
 	}
 }
+
+func TestWatcherStartupRecentFile(t *testing.T) {
+	spoolDir := t.TempDir()
+	newDir := filepath.Join(spoolDir, "new")
+	if err := os.MkdirAll(newDir, 0755); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create a file shortly before watcher start (younger than stabilityWait)
+	testFile := filepath.Join(newDir, "recent.pb")
+	if err := os.WriteFile(testFile, []byte("data"), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	opts := WatcherOptions{
+		CheckInterval: 10 * time.Millisecond,
+	}
+	w, err := NewWatcherWithOptions(spoolDir, 200*time.Millisecond, opts)
+	if err != nil {
+		t.Fatalf("NewWatcherWithOptions failed: %v", err)
+	}
+	defer func() { _ = w.Close() }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+
+	start := time.Now()
+	go func() { _ = w.Start(ctx) }()
+
+	select {
+	case path := <-w.Events():
+		if path != testFile {
+			t.Fatalf("Expected path %s, got %s", testFile, path)
+		}
+		if elapsed := time.Since(start); elapsed < 180*time.Millisecond {
+			t.Fatalf("File delivered too soon after startup: %v", elapsed)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for recently-created startup file")
+	}
+}
+
+func TestWatcherStartupBacklogDoesNotBlock(t *testing.T) {
+	spoolDir := t.TempDir()
+	newDir := filepath.Join(spoolDir, "new")
+	if err := os.MkdirAll(newDir, 0755); err != nil {
+		t.Fatal(err)
+	}
+
+	// Create more files than the channel buffer
+	fileCount := 5
+	for i := 0; i < fileCount; i++ {
+		f := filepath.Join(newDir, fmt.Sprintf("file%d.pb", i))
+		if err := os.WriteFile(f, []byte("data"), 0644); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// Ensure they are stable before start
+	time.Sleep(50 * time.Millisecond)
+
+	opts := WatcherOptions{
+		ChannelBuffer: 1, // Force backlog
+		CheckInterval: 10 * time.Millisecond,
+	}
+	w, err := NewWatcherWithOptions(spoolDir, 20*time.Millisecond, opts)
+	if err != nil {
+		t.Fatalf("NewWatcherWithOptions failed: %v", err)
+	}
+	defer func() { _ = w.Close() }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	go func() { _ = w.Start(ctx) }()
+
+	seen := make(map[string]bool)
+	timeout := time.After(2 * time.Second)
+	for len(seen) < fileCount {
+		select {
+		case path := <-w.Events():
+			seen[path] = true
+		case <-timeout:
+			t.Fatalf("Timed out waiting for backlog files, saw %d/%d", len(seen), fileCount)
+		}
+	}
+}
+
+func TestWatcherOverflowResyncs(t *testing.T) {
+	spoolDir := t.TempDir()
+	opts := WatcherOptions{
+		CheckInterval: 10 * time.Millisecond,
+	}
+	w, err := NewWatcherWithOptions(spoolDir, 50*time.Millisecond, opts)
+	if err != nil {
+		t.Fatalf("NewWatcherWithOptions failed: %v", err)
+	}
+	defer func() { _ = w.Close() }()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	go func() { _ = w.Start(ctx) }()
+
+	time.Sleep(50 * time.Millisecond) // allow Start to begin
+
+	newDir := filepath.Join(spoolDir, "new")
+	if err := w.watcher.Remove(newDir); err != nil {
+		t.Fatalf("Failed to remove watch: %v", err)
+	}
+
+	testFile := filepath.Join(newDir, "overflow.pb")
+	if err := os.WriteFile(testFile, []byte("data"), 0644); err != nil {
+		t.Fatal(err)
+	}
+
+	// Trigger overflow handling to force rescan
+	go func() {
+		w.watcher.Errors <- fsnotify.ErrEventOverflow
+	}()
+
+	select {
+	case path := <-w.Events():
+		if path != testFile {
+			t.Fatalf("Expected path %s from resync, got %s", testFile, path)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("Timeout waiting for resynced file after overflow")
+	}
+}