fix: replace fsnotify with git polling to prevent fd exhaustion (#356)

This commit is contained in:
João Salavisa
2026-03-03 20:07:01 +00:00
committed by GitHub
parent 43eccfaee2
commit 8cbaf950d2
6 changed files with 128 additions and 231 deletions
+1 -1
View File
@@ -7,7 +7,6 @@ require (
github.com/coder/acp-go-sdk v0.6.3
github.com/creack/pty v1.1.21
github.com/docker/docker v28.5.2+incompatible
github.com/fsnotify/fsnotify v1.6.0
github.com/gin-gonic/gin v1.9.1
github.com/github/copilot-sdk/go v0.1.18
github.com/google/uuid v1.6.0
@@ -45,6 +44,7 @@ require (
github.com/docker/go-connections v0.6.0 // indirect
github.com/docker/go-units v0.5.0 // indirect
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/fsnotify/fsnotify v1.6.0 // indirect
github.com/gabriel-vasile/mimetype v1.4.2 // indirect
github.com/gin-contrib/sse v0.1.0 // indirect
github.com/go-logr/logr v1.4.3 // indirect
@@ -168,19 +168,12 @@ func filterGitEnv(env []string) []string {
return result
}
// triggerFsNotify creates and removes a temporary file to trigger fsnotify and refresh git status.
// This is OS-agnostic and reliably triggers filesystem events that the workspace tracker watches.
// Called after git operations like commit, push, pull, etc. to refresh the UI.
func (g *GitOperator) triggerFsNotify() {
sentinelPath := filepath.Join(g.workDir, ".git-op-complete")
f, err := os.Create(sentinelPath)
if err != nil {
g.logger.Debug("failed to create sentinel file", zap.Error(err))
return
}
_ = f.Close()
if err := os.Remove(sentinelPath); err != nil {
g.logger.Debug("failed to remove sentinel file", zap.Error(err))
// triggerRefresh refreshes git status in the workspace tracker immediately.
// Called after git operations like commit, push, pull, etc. to refresh the UI
// without waiting for the next poll cycle.
func (g *GitOperator) triggerRefresh() {
if g.workspaceTracker != nil {
g.workspaceTracker.RefreshGitStatus(context.Background())
}
}
@@ -669,7 +662,7 @@ func (g *GitOperator) Discard(ctx context.Context, paths []string) (*GitOperatio
result.Success = true
}
g.triggerFsNotify()
g.triggerRefresh()
g.logger.Info("discard completed",
zap.Int("total_files", len(paths)),
zap.Int("untracked_files", len(untrackedFiles)),
@@ -1098,9 +1091,9 @@ func (g *GitOperator) unlock() {
g.inProgress = false
g.currentOp = ""
// Trigger fsnotify to refresh git status in the workspace tracker.
// Refresh git status in the workspace tracker immediately.
// This is called after every git operation completes.
g.triggerFsNotify()
g.triggerRefresh()
}
// PRCreateResult represents the result of a PR creation operation.
@@ -325,7 +325,7 @@ func (wt *WorkspaceTracker) ApplyFileDiff(reqPath string, unifiedDiff string, or
// Notify with the original relative path (not the resolved symlink target)
relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator))
wt.addPendingChange(relPath, types.FileOpWrite)
wt.notifyFileChange(relPath, types.FileOpWrite)
wt.logger.Debug("applied file diff using git apply",
zap.String("path", relPath),
@@ -389,7 +389,7 @@ func (wt *WorkspaceTracker) CreateFile(reqPath string) error {
// Notify with the relative path
cleanWorkDir := filepath.Clean(wt.workDir)
relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator))
wt.addPendingChange(relPath, types.FileOpCreate)
wt.notifyFileChange(relPath, types.FileOpCreate)
return nil
}
@@ -429,7 +429,7 @@ func (wt *WorkspaceTracker) DeleteFile(reqPath string) error {
}
relPath := strings.TrimPrefix(safePath, cleanWorkDir+string(os.PathSeparator))
wt.addPendingChange(relPath, types.FileOpRemove)
wt.notifyFileChange(relPath, types.FileOpRemove)
return nil
}
@@ -475,9 +475,9 @@ func (wt *WorkspaceTracker) RenameFile(oldPath, newPath string) error {
cleanWorkDir := wt.resolvedWorkDir()
oldRelPath := strings.TrimPrefix(oldSafePath, cleanWorkDir+string(os.PathSeparator))
newRelPath := strings.TrimPrefix(newSafePath, cleanWorkDir+string(os.PathSeparator))
wt.addPendingChange(oldRelPath, types.FileOpRename)
wt.notifyFileChange(oldRelPath, types.FileOpRename)
if newRelPath != oldRelPath {
wt.addPendingChange(newRelPath, types.FileOpRename)
wt.notifyFileChange(newRelPath, types.FileOpRename)
}
return nil
@@ -2,185 +2,96 @@ package process
import (
"context"
"io/fs"
"fmt"
"os"
"path/filepath"
"os/exec"
"time"
"github.com/fsnotify/fsnotify"
"github.com/kandev/kandev/internal/agentctl/types"
"go.uber.org/zap"
)
// monitorLoop handles debounced filesystem change events
// When files change, we wait for activity to settle, then update everything at once
// DefaultFilePollInterval is the interval for polling file modification times.
// This replaces fsnotify which on macOS (kqueue) opens a file descriptor per file,
// causing "too many open files" errors in large workspaces.
const DefaultFilePollInterval = 2 * time.Second
// monitorLoop polls for file changes using file mtimes and quick git checks.
// This is more efficient than fsnotify on macOS and avoids file descriptor exhaustion.
func (wt *WorkspaceTracker) monitorLoop(ctx context.Context) {
defer wt.wg.Done()
// Debounce duration - wait this long after last file change before updating
const debounceDuration = 300 * time.Millisecond
var debounceTimer *time.Timer
var pendingUpdate bool
ticker := time.NewTicker(DefaultFilePollInterval)
defer ticker.Stop()
// Initial update
wt.updateGitStatus(ctx)
wt.updateFiles(ctx)
// Cache the last known state
lastState := wt.getWorkspaceState(ctx)
wt.logger.Info("file polling started", zap.Duration("interval", DefaultFilePollInterval))
for {
select {
case <-ctx.Done():
if debounceTimer != nil {
debounceTimer.Stop()
}
return
case <-wt.stopCh:
if debounceTimer != nil {
debounceTimer.Stop()
}
return
case <-wt.fsChangeTrigger:
// File change detected - start or reset debounce timer
if debounceTimer == nil {
debounceTimer = time.NewTimer(debounceDuration)
} else {
// Reset the timer if already running
if !debounceTimer.Stop() {
select {
case <-debounceTimer.C:
default:
}
}
debounceTimer.Reset(debounceDuration)
}
pendingUpdate = true
case <-func() <-chan time.Time {
if debounceTimer != nil {
return debounceTimer.C
}
return nil
}():
// Debounce timer fired - update everything
if pendingUpdate {
wt.updateGitStatus(ctx)
case <-ticker.C:
// Quick state check using mtime + diff-files
currentState := wt.getWorkspaceState(ctx)
if currentState.changed(lastState) {
lastState = currentState
wt.logger.Debug("workspace state changed, updating")
// Update file list and notify subscribers
wt.updateFiles(ctx)
// Drain pending changes and emit specific notifications
wt.pendingChangesMu.Lock()
changes := wt.pendingChanges
wt.pendingChanges = nil
wt.pendingChangesMu.Unlock()
wt.emitFileChanges(changes)
pendingUpdate = false
wt.notifyWorkspaceStreamFileChange(types.FileChangeNotification{
Timestamp: time.Now(),
Operation: types.FileOpRefresh,
})
}
debounceTimer = nil
}
}
}
// watchFilesystem watches for filesystem changes and triggers debounced updates
func (wt *WorkspaceTracker) watchFilesystem(ctx context.Context) {
defer wt.wg.Done()
for {
select {
case <-ctx.Done():
return
case <-wt.stopCh:
return
case event, ok := <-wt.watcher.Events:
if !ok {
return
}
// Ignore CHMOD events - permission changes don't affect file content
// and shouldn't trigger UI refreshes. This prevents loops from filesystem
// scanners, git operations, or other tools that touch file permissions.
if event.Op == fsnotify.Chmod {
continue
}
// If a directory was created, watch it
if event.Op&fsnotify.Create == fsnotify.Create {
if info, err := os.Stat(event.Name); err == nil && info.IsDir() {
if err := wt.addDirectoryRecursive(event.Name); err != nil {
wt.logger.Debug("failed to watch new directory", zap.Error(err))
}
}
}
// Record specific change with relative path
relPath, relErr := filepath.Rel(wt.workDir, event.Name)
if relErr != nil {
relPath = event.Name
}
wt.addPendingChange(relPath, fsOpToString(event.Op))
case err, ok := <-wt.watcher.Errors:
if !ok {
return
}
wt.logger.Debug("filesystem watcher error", zap.Error(err))
}
}
// workspaceState holds quick-check state for detecting workspace changes.
type workspaceState struct {
indexMtime time.Time // index mtime - changes on stage/unstage/commit
diffFilesID string // hash of git diff-files output - changes on any file modification
}
// fsOpToString converts an fsnotify operation to a FileOp string constant.
func fsOpToString(op fsnotify.Op) string {
switch {
case op&fsnotify.Create != 0:
return types.FileOpCreate
case op&fsnotify.Write != 0:
return types.FileOpWrite
case op&fsnotify.Remove != 0:
return types.FileOpRemove
case op&fsnotify.Rename != 0:
return types.FileOpRename
default:
return types.FileOpRefresh
// getWorkspaceState returns the current workspace state using fast checks.
// Uses index mtime + git diff-files output which includes file content hashes.
func (wt *WorkspaceTracker) getWorkspaceState(ctx context.Context) workspaceState {
var state workspaceState
// Check index mtime (gitIndexPath is pre-validated at startup)
if wt.gitIndexPath != "" {
if info, err := os.Stat(wt.gitIndexPath); err == nil {
state.indexMtime = info.ModTime()
}
}
// git diff-files shows changed files with their blob hashes
// Output changes whenever file content changes (not just dirty/clean boolean)
cmd := exec.CommandContext(ctx, "git", "diff-files")
cmd.Dir = wt.workDir
out, _ := cmd.Output()
// Use length + first/last bytes as cheap identity check
// (full hash would work too but this is faster for our use case)
if len(out) > 0 {
state.diffFilesID = fmt.Sprintf("%d:%x:%x", len(out), out[0], out[len(out)-1])
}
return state
}
// addDirectoryRecursive adds a directory and all its subdirectories to the watcher
func (wt *WorkspaceTracker) addDirectoryRecursive(dir string) error {
return filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
if err != nil {
return err
}
// Skip hidden directories and common ignore patterns
if d.IsDir() {
name := d.Name()
if name == ".git" || name == "node_modules" || name == ".next" || name == "dist" || name == "build" {
return filepath.SkipDir
}
// Add directory to watcher
if err := wt.watcher.Add(path); err != nil {
wt.logger.Debug("failed to watch directory", zap.String("path", path), zap.Error(err))
}
}
return nil
})
}
// addPendingChange records a specific file change and triggers a debounced update.
func (wt *WorkspaceTracker) addPendingChange(relPath, operation string) {
wt.pendingChangesMu.Lock()
wt.pendingChanges = append(wt.pendingChanges, types.FileChangeNotification{
Timestamp: time.Now(),
Path: relPath,
Operation: operation,
})
wt.pendingChangesMu.Unlock()
// Trigger debounced update (non-blocking)
select {
case wt.fsChangeTrigger <- struct{}{}:
default:
}
// changed returns true if the workspace state has changed.
func (s workspaceState) changed(other workspaceState) bool {
return s.indexMtime != other.indexMtime || s.diffFilesID != other.diffFilesID
}
// emitFileChanges sends accumulated file changes to subscribers.
@@ -198,3 +109,14 @@ func (wt *WorkspaceTracker) emitFileChanges(changes []types.FileChangeNotificati
wt.notifyWorkspaceStreamFileChange(changes[i])
}
}
// notifyFileChange sends a file change notification immediately.
// Used by direct file operations (create, delete, rename, apply diff) to notify
// subscribers without waiting for the next poll cycle.
func (wt *WorkspaceTracker) notifyFileChange(relPath, operation string) {
wt.notifyWorkspaceStreamFileChange(types.FileChangeNotification{
Timestamp: time.Now(),
Path: relPath,
Operation: operation,
})
}
@@ -2,10 +2,13 @@ package process
import (
"context"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"time"
"github.com/fsnotify/fsnotify"
"github.com/kandev/kandev/internal/agentctl/types"
"github.com/kandev/kandev/internal/common/logger"
"go.uber.org/zap"
@@ -20,10 +23,13 @@ const (
fileStatusModified = "modified"
)
// WorkspaceTracker monitors workspace changes and provides real-time updates
// WorkspaceTracker monitors workspace changes and provides real-time updates.
// It uses git status polling instead of fsnotify to avoid file descriptor exhaustion
// on macOS (where kqueue opens a file descriptor for every watched file).
type WorkspaceTracker struct {
workDir string
logger *logger.Logger
workDir string
gitIndexPath string // Cached, validated path to git index file (works with worktrees)
logger *logger.Logger
// Current state
currentStatus types.GitStatusUpdate
@@ -39,16 +45,6 @@ type WorkspaceTracker struct {
workspaceStreamSubscribers map[types.WorkspaceStreamSubscriber]struct{}
workspaceSubMu sync.RWMutex
// Filesystem watcher
watcher *fsnotify.Watcher
// Debounce channel for filesystem change events
fsChangeTrigger chan struct{}
// Pending file changes accumulated between debounce flushes
pendingChanges []types.FileChangeNotification
pendingChangesMu sync.Mutex
// Git polling interval
gitPollInterval time.Duration
@@ -62,24 +58,48 @@ type WorkspaceTracker struct {
// NewWorkspaceTracker creates a new workspace tracker
func NewWorkspaceTracker(workDir string, log *logger.Logger) *WorkspaceTracker {
resolvedWorkDir := resolveExistingWorkDir(workDir, log.WithFields(zap.String("component", "workspace-tracker")))
watcher, err := fsnotify.NewWatcher()
if err != nil {
log.Error("failed to create filesystem watcher", zap.Error(err))
watcher = nil
}
// Cache validated git index path (works with worktrees where .git is a file)
gitIndexPath := resolveGitIndexPath(resolvedWorkDir)
return &WorkspaceTracker{
workDir: resolvedWorkDir,
gitIndexPath: gitIndexPath,
logger: log.WithFields(zap.String("component", "workspace-tracker")),
workspaceStreamSubscribers: make(map[types.WorkspaceStreamSubscriber]struct{}),
watcher: watcher,
fsChangeTrigger: make(chan struct{}, 1), // Buffered to avoid blocking
gitPollInterval: DefaultGitPollInterval,
stopCh: make(chan struct{}),
}
}
// Start begins monitoring the workspace
// resolveGitIndexPath returns the validated path to the git index file.
// Returns empty string if the path cannot be resolved or validated.
// This handles worktrees where .git is a file pointing elsewhere.
func resolveGitIndexPath(workDir string) string {
cmd := exec.Command("git", "rev-parse", "--git-dir")
cmd.Dir = workDir
out, err := cmd.Output()
if err != nil {
return ""
}
gitDir := strings.TrimSpace(string(out))
if !filepath.IsAbs(gitDir) {
gitDir = filepath.Join(workDir, gitDir)
}
// Clean and construct the index path
gitDir = filepath.Clean(gitDir)
indexPath := filepath.Join(gitDir, "index")
// Validate the index file exists (this proves gitDir is a valid git directory)
info, err := os.Stat(indexPath)
if err != nil || info.IsDir() {
return ""
}
return indexPath
}
// Start begins monitoring the workspace using polling (no fsnotify).
// File changes are detected via git status polling, which is efficient and
// doesn't consume file descriptors like fsnotify/kqueue does on macOS.
func (wt *WorkspaceTracker) Start(ctx context.Context) {
wt.mu.Lock()
if wt.started {
@@ -90,34 +110,19 @@ func (wt *WorkspaceTracker) Start(ctx context.Context) {
wt.started = true
wt.mu.Unlock()
// Start file change monitoring (uses git status polling)
wt.wg.Add(1)
go wt.monitorLoop(ctx)
// Start git polling for detecting manual git operations
// Start git polling for detecting manual git operations (commits, resets, etc.)
wt.wg.Add(1)
go wt.pollGitChanges(ctx)
// Start filesystem watcher if available
if wt.watcher != nil {
wt.wg.Add(1)
go wt.watchFilesystem(ctx)
// Add workspace root to watcher
if err := wt.addDirectoryRecursive(wt.workDir); err != nil {
wt.logger.Error("failed to watch workspace directory", zap.Error(err))
}
}
}
// Stop stops the workspace tracker
func (wt *WorkspaceTracker) Stop() {
wt.stopOnce.Do(func() {
close(wt.stopCh)
if wt.watcher != nil {
if err := wt.watcher.Close(); err != nil {
wt.logger.Debug("failed to close watcher", zap.Error(err))
}
}
wt.wg.Wait()
wt.logger.Info("workspace tracker stopped")
})
@@ -10,7 +10,6 @@ import (
"testing"
"time"
"github.com/fsnotify/fsnotify"
"github.com/kandev/kandev/internal/agentctl/types"
"github.com/kandev/kandev/internal/agentctl/types/streams"
)
@@ -577,28 +576,6 @@ func TestRenameFile(t *testing.T) {
})
}
func TestFsOpToString(t *testing.T) {
tests := []struct {
op fsnotify.Op
want string
}{
{fsnotify.Create, types.FileOpCreate},
{fsnotify.Write, types.FileOpWrite},
{fsnotify.Remove, types.FileOpRemove},
{fsnotify.Rename, types.FileOpRename},
{fsnotify.Chmod, types.FileOpRefresh},
}
for _, tt := range tests {
t.Run(tt.op.String(), func(t *testing.T) {
got := fsOpToString(tt.op)
if got != tt.want {
t.Errorf("fsOpToString(%v) = %q, want %q", tt.op, got, tt.want)
}
})
}
}
func TestRewriteDiffPaths(t *testing.T) {
tests := []struct {
name string