From 5ab73387b6e8c347289593e3a1cafe9d72bc81f4 Mon Sep 17 00:00:00 2001 From: 0x4D31 Date: Wed, 3 Dec 2025 21:14:06 +0000 Subject: [PATCH] spool: handle startup backlog and resync after overflow - seed existing spool files into stability tracking so recent pre-start files are processed - avoid blocking on large startup backlogs by enqueueing stably or tracking for later delivery - rescan the spool when fsnotify reports ErrEventOverflow to recover missed files - add tests for startup-recent files, backlog handling, and overflow resync Tests: GOCACHE=/tmp/go-build-cache go test ./internal/spool --- internal/spool/watcher.go | 98 ++++++++++++++++++------- internal/spool/watcher_test.go | 130 +++++++++++++++++++++++++++++++++ 2 files changed, 200 insertions(+), 28 deletions(-) diff --git a/internal/spool/watcher.go b/internal/spool/watcher.go index f16b677..645540e 100644 --- a/internal/spool/watcher.go +++ b/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,24 +129,7 @@ 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: @@ -151,6 +137,9 @@ func (w *Watcher) Start(ctx context.Context) error { 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 +} diff --git a/internal/spool/watcher_test.go b/internal/spool/watcher_test.go index 4f8f3dc..fc1e726 100644 --- a/internal/spool/watcher_test.go +++ b/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") + } +}