diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c359e9a..a263a7e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,7 +14,7 @@ jobs: fail-fast: false matrix: os: [ubuntu-latest, macos-latest] - go-version: ['1.21', '1.22', '1.23'] + go-version: ['1.24', '1.25', '1.26'] steps: - uses: actions/checkout@v4 diff --git a/watch/daemon.go b/watch/daemon.go index c1029ff..54936bf 100644 --- a/watch/daemon.go +++ b/watch/daemon.go @@ -6,6 +6,7 @@ import ( "os" "path/filepath" "strings" + "sync" "time" "codemap/limits" @@ -23,6 +24,8 @@ type Daemon struct { eventLog string // path to event log file verbose bool done chan struct{} + + eventLoopWG sync.WaitGroup } // NewDaemon creates a new watch daemon for the given root @@ -97,7 +100,11 @@ func (d *Daemon) Start() error { d.writeState() // Start event loop - go d.eventLoop() + d.eventLoopWG.Add(1) + go func() { + defer d.eventLoopWG.Done() + d.eventLoop() + }() return nil } @@ -106,6 +113,7 @@ func (d *Daemon) Start() error { func (d *Daemon) Stop() { close(d.done) d.watcher.Close() + d.eventLoopWG.Wait() } // GetGraph returns the current graph (thread-safe) diff --git a/watch/events.go b/watch/events.go index 5ba6cf9..f509d94 100644 --- a/watch/events.go +++ b/watch/events.go @@ -25,8 +25,22 @@ type eventDebouncer struct { pruneAfter time.Duration lastSeen map[string]time.Time lastPruned time.Time + pending map[string]pendingWrite } +type pendingWrite struct { + event fsnotify.Event + due time.Time +} + +type debounceAction uint8 + +const ( + debounceProcess debounceAction = iota + debounceSkip + debounceDefer +) + func newEventDebouncer(window time.Duration) *eventDebouncer { pruneAfter := 10 * window if pruneAfter < time.Second { @@ -36,6 +50,7 @@ func newEventDebouncer(window time.Duration) *eventDebouncer { window: window, pruneAfter: pruneAfter, lastSeen: make(map[string]time.Time), + pending: make(map[string]pendingWrite), } } @@ -55,9 +70,7 @@ func (d *eventDebouncer) shouldSkip(event fsnotify.Event, now time.Time) bool { return false } - if last, exists := d.lastSeen[event.Name]; exists && now.Sub(last) < d.window { - return true - } + last, exists := d.lastSeen[event.Name] d.lastSeen[event.Name] = now if d.lastPruned.IsZero() || now.Sub(d.lastPruned) >= d.pruneAfter { @@ -65,7 +78,7 @@ func (d *eventDebouncer) shouldSkip(event fsnotify.Event, now time.Time) bool { d.lastPruned = now } - return false + return exists && now.Sub(last) < d.window } func (d *eventDebouncer) prune(now time.Time) { @@ -77,19 +90,111 @@ func (d *eventDebouncer) prune(now time.Time) { } } +func (d *eventDebouncer) deferEvent(event fsnotify.Event, now time.Time) { + d.pending[event.Name] = pendingWrite{event: event, due: now.Add(d.window)} +} + +func (d *eventDebouncer) cancelPending(path string) { + delete(d.pending, path) +} + +func (d *eventDebouncer) takeDue(now time.Time) []fsnotify.Event { + var events []fsnotify.Event + for path, pending := range d.pending { + if pending.due.After(now) { + continue + } + events = append(events, pending.event) + delete(d.pending, path) + } + return events +} + +func (d *eventDebouncer) takeDueBeforeEvent(event fsnotify.Event, now time.Time) []fsnotify.Event { + if event.Op&(fsnotify.Create|fsnotify.Remove|fsnotify.Rename) != 0 { + d.cancelPending(event.Name) + } + return d.takeDue(now) +} + +func (d *eventDebouncer) takeAll() []fsnotify.Event { + events := make([]fsnotify.Event, 0, len(d.pending)) + for path, pending := range d.pending { + events = append(events, pending.event) + delete(d.pending, path) + } + return events +} + +func (d *eventDebouncer) nextDelay(now time.Time) (time.Duration, bool) { + var earliest time.Time + for _, pending := range d.pending { + if earliest.IsZero() || pending.due.Before(earliest) { + earliest = pending.due + } + } + if earliest.IsZero() { + return 0, false + } + delay := earliest.Sub(now) + if delay < 0 { + delay = 0 + } + return delay, true +} + // eventLoop processes file system events func (d *Daemon) eventLoop() { debouncer := newEventDebouncer(100 * time.Millisecond) + timer := time.NewTimer(time.Hour) + timer.Stop() + defer timer.Stop() + + var timerC <-chan time.Time + armTimer := func(now time.Time) { + if !timer.Stop() { + select { + case <-timer.C: + default: + } + } + delay, ok := debouncer.nextDelay(now) + if !ok { + timerC = nil + return + } + timer.Reset(delay) + timerC = timer.C + } + flushDue := func(now time.Time) { + for _, event := range debouncer.takeDue(now) { + d.handleEvent(event) + } + } + defer func() { + for _, event := range debouncer.takeAll() { + d.handleEvent(event) + } + }() for { select { case <-d.done: return + case <-timerC: + now := time.Now() + flushDue(now) + armTimer(time.Now()) + case event, ok := <-d.watcher.Events: if !ok { return } + now := time.Now() + for _, pending := range debouncer.takeDueBeforeEvent(event, now) { + d.handleEvent(pending) + } // Allow directory creates through (to add new dirs to watcher) // but skip non-source files otherwise @@ -114,12 +219,15 @@ func (d *Daemon) eventLoop() { continue } - if d.shouldSkipEvent(debouncer, event, time.Now()) { - continue + switch d.debounceAction(debouncer, event, now) { + case debounceSkip: + debouncer.cancelPending(event.Name) + case debounceDefer: + debouncer.deferEvent(event, now) + case debounceProcess: + d.handleEvent(event) } - - // Process the event - d.handleEvent(event) + armTimer(time.Now()) case err, ok := <-d.watcher.Errors: if !ok { @@ -132,24 +240,35 @@ func (d *Daemon) eventLoop() { } } -func (d *Daemon) shouldSkipEvent(debouncer *eventDebouncer, event fsnotify.Event, now time.Time) bool { +func (d *Daemon) debounceAction(debouncer *eventDebouncer, event fsnotify.Event, now time.Time) debounceAction { if !debouncer.shouldSkip(event, now) { - return false - } - info, err := os.Stat(event.Name) - if err != nil { - return false + return debounceProcess } relPath, err := filepath.Rel(d.root, event.Name) if err != nil { - return false + return debounceProcess } d.graph.mu.RLock() cached := d.graph.State[relPath] - matches := cached != nil && cached.Size == info.Size() + var cachedSize, cachedModTime int64 + if cached != nil { + cachedSize = cached.Size + cachedModTime = cached.ModTime + } d.graph.mu.RUnlock() - return matches + if cached == nil { + return debounceProcess + } + + info, err := os.Stat(event.Name) + if err != nil { + return debounceProcess + } + if cachedModTime != 0 && cachedSize == info.Size() && cachedModTime == info.ModTime().UnixNano() { + return debounceSkip + } + return debounceDefer } // isSourceFile checks if a file should be tracked. @@ -262,7 +381,11 @@ func (d *Daemon) handleEvent(fsEvent fsnotify.Event) { } // Update cached state - d.graph.State[relPath] = &FileState{Lines: newLines, Size: info.Size()} + d.graph.State[relPath] = &FileState{ + Lines: newLines, + Size: info.Size(), + ModTime: info.ModTime().UnixNano(), + } // Update file info d.graph.Files[relPath] = &scanner.FileInfo{ diff --git a/watch/events_debounce_test.go b/watch/events_debounce_test.go index a1af383..30bc8cd 100644 --- a/watch/events_debounce_test.go +++ b/watch/events_debounce_test.go @@ -9,37 +9,86 @@ import ( "github.com/fsnotify/fsnotify" ) -func TestDaemonDoesNotDebounceWriteWhenCachedSizeIsStale(t *testing.T) { +func TestDaemonDebounceActionUsesSizeAndModTime(t *testing.T) { root := t.TempDir() path := filepath.Join(root, "main.go") - if err := os.WriteFile(path, []byte("package main\n\nfunc changed() {}\n"), 0o644); err != nil { + if err := os.WriteFile(path, []byte("package foo\n"), 0o644); err != nil { + t.Fatal(err) + } + initialModTime := time.Unix(1_700_000_000, 0) + if err := os.Chtimes(path, initialModTime, initialModTime); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { t.Fatal(err) } d := &Daemon{ root: root, graph: &Graph{State: map[string]*FileState{ - "main.go": {Size: 0}, + "main.go": {Size: info.Size(), ModTime: info.ModTime().UnixNano()}, }}, } debouncer := newEventDebouncer(100 * time.Millisecond) event := fsnotify.Event{Name: path, Op: fsnotify.Write} base := time.Unix(0, 0) - if d.shouldSkipEvent(debouncer, event, base) { - t.Fatal("first write should not be skipped") + if got := d.debounceAction(debouncer, event, base); got != debounceProcess { + t.Fatalf("first write action = %v, want process", got) } - if d.shouldSkipEvent(debouncer, event, base.Add(10*time.Millisecond)) { - t.Fatal("final write should refresh stale cached state") + + if err := os.WriteFile(path, []byte("package bar\n"), 0o644); err != nil { + t.Fatal(err) + } + changedModTime := initialModTime.Add(time.Second) + if err := os.Chtimes(path, changedModTime, changedModTime); err != nil { + t.Fatal(err) + } + if got := d.debounceAction(debouncer, event, base.Add(10*time.Millisecond)); got != debounceDefer { + t.Fatalf("same-size changed write action = %v, want defer", got) } - info, err := os.Stat(path) + info, err = os.Stat(path) if err != nil { t.Fatal(err) } d.graph.State["main.go"].Size = info.Size() - if !d.shouldSkipEvent(debouncer, event, base.Add(20*time.Millisecond)) { - t.Fatal("duplicate write should be skipped once cached state is current") + d.graph.State["main.go"].ModTime = info.ModTime().UnixNano() + if got := d.debounceAction(debouncer, event, base.Add(20*time.Millisecond)); got != debounceSkip { + t.Fatalf("duplicate write action = %v, want skip", got) + } +} + +func TestDaemonDebounceActionProcessesMissingCache(t *testing.T) { + root := t.TempDir() + d := &Daemon{root: root, graph: &Graph{State: map[string]*FileState{}}} + debouncer := newEventDebouncer(100 * time.Millisecond) + event := fsnotify.Event{Name: filepath.Join(root, "missing.go"), Op: fsnotify.Write} + base := time.Unix(0, 0) + + if got := d.debounceAction(debouncer, event, base); got != debounceProcess { + t.Fatalf("first missing-cache action = %v, want process", got) + } + if got := d.debounceAction(debouncer, event, base.Add(10*time.Millisecond)); got != debounceProcess { + t.Fatalf("rapid missing-cache action = %v, want process", got) + } +} + +func TestDaemonDebounceActionProcessesMissingTrackedFile(t *testing.T) { + root := t.TempDir() + d := &Daemon{root: root, graph: &Graph{State: map[string]*FileState{ + "missing.go": {}, + }}} + debouncer := newEventDebouncer(100 * time.Millisecond) + event := fsnotify.Event{Name: filepath.Join(root, "missing.go"), Op: fsnotify.Write} + base := time.Unix(0, 0) + + if got := d.debounceAction(debouncer, event, base); got != debounceProcess { + t.Fatalf("first missing-file action = %v, want process", got) + } + if got := d.debounceAction(debouncer, event, base.Add(10*time.Millisecond)); got != debounceProcess { + t.Fatalf("rapid missing-file action = %v, want process", got) } } @@ -54,7 +103,10 @@ func TestEventDebouncerSkipsRapidWrites(t *testing.T) { if !debouncer.shouldSkip(event, base.Add(50*time.Millisecond)) { t.Fatal("rapid write should be skipped") } - if debouncer.shouldSkip(event, base.Add(150*time.Millisecond)) { + if !debouncer.shouldSkip(event, base.Add(120*time.Millisecond)) { + t.Fatal("rapid write should extend the quiet window") + } + if debouncer.shouldSkip(event, base.Add(220*time.Millisecond)) { t.Fatal("write outside debounce window should not be skipped") } } @@ -112,3 +164,130 @@ func TestEventDebouncerPrunesStaleEntries(t *testing.T) { t.Fatal("expected recent path entry to be retained") } } + +func TestEventDebouncerPendingWritesKeepLatestUntilDue(t *testing.T) { + debouncer := newEventDebouncer(100 * time.Millisecond) + base := time.Unix(0, 0) + first := fsnotify.Event{Name: "src/file.go", Op: fsnotify.Write} + latest := fsnotify.Event{Name: "src/file.go", Op: fsnotify.Write | fsnotify.Chmod} + + debouncer.deferEvent(first, base) + debouncer.deferEvent(latest, base.Add(50*time.Millisecond)) + + if got := debouncer.takeDue(base.Add(149 * time.Millisecond)); len(got) != 0 { + t.Fatalf("takeDue before quiet window returned %v", got) + } + if delay, ok := debouncer.nextDelay(base.Add(100 * time.Millisecond)); !ok || delay != 50*time.Millisecond { + t.Fatalf("nextDelay = (%v, %v), want (50ms, true)", delay, ok) + } + if delay, ok := debouncer.nextDelay(base.Add(151 * time.Millisecond)); !ok || delay != 0 { + t.Fatalf("overdue nextDelay = (%v, %v), want (0, true)", delay, ok) + } + + got := debouncer.takeDue(base.Add(150 * time.Millisecond)) + if len(got) != 1 || got[0] != latest { + t.Fatalf("takeDue after quiet window = %v, want latest event %v", got, latest) + } + if _, ok := debouncer.nextDelay(base.Add(150 * time.Millisecond)); ok { + t.Fatal("expected no timer deadline after pending event drained") + } +} + +func TestEventLoopDrainsLatestPendingWriteOnShutdown(t *testing.T) { + root := t.TempDir() + if err := os.Mkdir(filepath.Join(root, ".codemap"), 0o755); err != nil { + t.Fatal(err) + } + path := filepath.Join(root, "main.go") + if err := os.WriteFile(path, []byte("package foo\n"), 0o644); err != nil { + t.Fatal(err) + } + + d, err := NewDaemon(root, false) + if err != nil { + t.Fatal(err) + } + if err := d.watcher.Close(); err != nil { + t.Fatal(err) + } + d.watcher.Events = make(chan fsnotify.Event) + d.watcher.Errors = make(chan error) + d.eventLoopWG.Add(1) + go func() { + defer d.eventLoopWG.Done() + d.eventLoop() + }() + stopped := false + t.Cleanup(func() { + if !stopped { + d.Stop() + } + }) + + event := fsnotify.Event{Name: path, Op: fsnotify.Write} + d.watcher.Events <- event + waitForWatchCondition(t, time.Second, func() bool { + return len(d.GetEvents(10)) == 1 + }) + d.watcher.Events <- event + time.Sleep(20 * time.Millisecond) + if got := len(d.GetEvents(10)); got != 1 { + t.Fatalf("duplicate write produced %d events, want 1", got) + } + + if err := os.WriteFile(path, []byte("package bar\n"), 0o644); err != nil { + t.Fatal(err) + } + info, err := os.Stat(path) + if err != nil { + t.Fatal(err) + } + changedModTime := info.ModTime().Add(time.Second) + if err := os.Chtimes(path, changedModTime, changedModTime); err != nil { + t.Fatal(err) + } + d.watcher.Events <- event + d.Stop() + stopped = true + got := d.GetEvents(10) + if len(got) != 2 || got[1].Lines != 1 { + t.Fatalf("events after shutdown drain = %#v, want latest pending write", got) + } +} + +func TestEventDebouncerCancelsAndDrainsPendingWrites(t *testing.T) { + debouncer := newEventDebouncer(100 * time.Millisecond) + base := time.Unix(0, 0) + canceled := fsnotify.Event{Name: "src/canceled.go", Op: fsnotify.Write} + remaining := fsnotify.Event{Name: "src/remaining.go", Op: fsnotify.Write} + + debouncer.deferEvent(canceled, base) + debouncer.deferEvent(remaining, base.Add(10*time.Millisecond)) + debouncer.cancelPending(canceled.Name) + + got := debouncer.takeAll() + if len(got) != 1 || got[0] != remaining { + t.Fatalf("takeAll = %v, want remaining event %v", got, remaining) + } + if got := debouncer.takeAll(); len(got) != 0 { + t.Fatalf("second takeAll returned already-drained events: %v", got) + } +} + +func TestEventDebouncerLifecycleCancelsSamePathBeforeDueFlush(t *testing.T) { + debouncer := newEventDebouncer(100 * time.Millisecond) + base := time.Unix(0, 0) + canceled := fsnotify.Event{Name: "src/canceled.go", Op: fsnotify.Write} + remaining := fsnotify.Event{Name: "src/remaining.go", Op: fsnotify.Write} + + debouncer.deferEvent(canceled, base) + debouncer.deferEvent(remaining, base) + got := debouncer.takeDueBeforeEvent( + fsnotify.Event{Name: canceled.Name, Op: fsnotify.Remove}, + base.Add(100*time.Millisecond), + ) + + if len(got) != 1 || got[0] != remaining { + t.Fatalf("due writes before lifecycle event = %v, want only %v", got, remaining) + } +} diff --git a/watch/types.go b/watch/types.go index e5362e3..174b779 100644 --- a/watch/types.go +++ b/watch/types.go @@ -26,8 +26,9 @@ type Event struct { // FileState tracks lightweight per-file state for delta calculations type FileState struct { - Lines int - Size int64 + Lines int + Size int64 + ModTime int64 } // DepContext holds pre-computed dependency context for a file @@ -43,7 +44,7 @@ type Graph struct { Files map[string]*scanner.FileInfo // path -> file info FileGraph *scanner.FileGraph // internal file-to-file dependencies DepCtx map[string]*DepContext // path -> dependency context (precomputed) - State map[string]*FileState // path -> line/size cache for deltas + State map[string]*FileState // path -> line/size/mtime cache for deltas Events []Event WorkingSet *WorkingSet // session working set LastScan time.Time diff --git a/watch/watch_test.go b/watch/watch_test.go index cf430a3..7a3404a 100644 --- a/watch/watch_test.go +++ b/watch/watch_test.go @@ -281,7 +281,7 @@ func TestDebounce(t *testing.T) { // Rapid fire writes (within debounce window) for i := 0; i < 5; i++ { - content := []byte("package rapid\n// line " + string(rune('a'+i)) + "\n") + content := []byte("package rapid\n" + strings.Repeat("// line\n", i+1)) if err := os.WriteFile(testFile, content, 0644); err != nil { t.Fatalf("Failed to modify file: %v", err) } @@ -292,15 +292,21 @@ func TestDebounce(t *testing.T) { events := daemon.GetEvents(100) writeCount := 0 + lastWriteLines := 0 for _, e := range events { if e.Op == "WRITE" && e.Path == "rapid.go" { writeCount++ + lastWriteLines = e.Lines } } - // Should have fewer events than writes due to debouncing - if writeCount >= 5 { - t.Errorf("Expected debounced events (< 5), got %d", writeCount) + // The first write may be processed immediately; the changed-size burst must + // collapse to at most one trailing event with the latest contents. + if writeCount == 0 || writeCount > 2 { + t.Errorf("Expected 1-2 debounced events, got %d", writeCount) + } + if lastWriteLines != 6 { + t.Errorf("Expected trailing event with 6 lines, got %d", lastWriteLines) } }