From 0f013804fff33adba5095fce74d28f25c8db9cb6 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 9 Nov 2026 17:06:30 -0500 Subject: [PATCH 1/4] engine: on macOS a held chat's folder costs one watch handle, not one per file (#2471) The presence watch handed each held chat's session folder to fsnotify, whose kqueue backend on macOS opens a descriptor for every file in a watched folder (transcript, meta, lock, presence, card, told files). A long-lived app kept about that many descriptors per chat it had shown. Linux and Windows already use one handle per folder. The watch only needs request files to appear (takeover.json, .conversation-delete-request, .task-delete-request-*), and every writer publishes them by create or rename in that folder. On darwin a kqueue watch on the folder itself now sees those entry changes and lists the folder for requests; a removed or renamed folder is dropped and re-added by the safety pass as before. Other platforms keep fsnotify behind the same small seam. Holds, viewOpen, the idle sweep and the heartbeat clocks are unchanged. Co-Authored-By: Claude Opus 5.5 --- internal/session/presence_watch.go | 32 ++- internal/session/presence_watch_darwin.go | 224 +++++++++++++++ .../session/presence_watch_darwin_test.go | 269 ++++++++++++++++++ internal/session/presence_watch_fsnotify.go | 20 ++ .../session/presence_watch_fsnotify_test.go | 32 +++ internal/session/presence_watch_test.go | 245 ++++++++++++++++ 6 files changed, 815 insertions(+), 7 deletions(-) create mode 100644 internal/session/presence_watch_darwin.go create mode 100644 internal/session/presence_watch_darwin_test.go create mode 100644 internal/session/presence_watch_fsnotify.go create mode 100644 internal/session/presence_watch_fsnotify_test.go create mode 100644 internal/session/presence_watch_test.go diff --git a/internal/session/presence_watch.go b/internal/session/presence_watch.go index 1411995086..c45590337a 100644 --- a/internal/session/presence_watch.go +++ b/internal/session/presence_watch.go @@ -16,13 +16,26 @@ import ( var presenceDoors presenceDoorWatch type presenceDoorWatch struct { - mu sync.Mutex - run *presenceDoorRun + mu sync.Mutex + run *presenceDoorRun + newWatcher func() (doorWatcher, error) +} + +// Keep fsnotify's event shape so other platforms need no event forwarding loop. +type doorEvent = fsnotify.Event + +type doorWatcher interface { + Add(string) error + Remove(string) error + Close() error + Events() <-chan doorEvent + Errors() <-chan error } type presenceDoorRun struct { - watcher *fsnotify.Watcher + watcher doorWatcher stop chan struct{} + done chan struct{} clients map[*presenceDoorClient]bool watched map[string]bool } @@ -38,8 +51,12 @@ func (w *presenceDoorWatch) subscribe(dir string) (<-chan struct{}, <-chan time. w.mu.Lock() defer w.mu.Unlock() if w.run == nil { - watcher, _ := fsnotify.NewWatcher() - w.run = &presenceDoorRun{watcher: watcher, stop: make(chan struct{}), clients: map[*presenceDoorClient]bool{}, watched: map[string]bool{}} + newWatcher := w.newWatcher + if newWatcher == nil { + newWatcher = newDoorWatcher + } + watcher, _ := newWatcher() + w.run = &presenceDoorRun{watcher: watcher, stop: make(chan struct{}), done: make(chan struct{}), clients: map[*presenceDoorClient]bool{}, watched: map[string]bool{}} go w.loop(w.run) } run := w.run @@ -99,14 +116,15 @@ func presenceDoorstepEvent(event fsnotify.Event) bool { } func (w *presenceDoorWatch) loop(run *presenceDoorRun) { + defer close(run.done) ticker := time.NewTicker(presenceHeartbeat) defer ticker.Stop() refresh := time.NewTicker(presenceRefresh) defer refresh.Stop() - var events <-chan fsnotify.Event + var events <-chan doorEvent var errors <-chan error if run.watcher != nil { - events, errors = run.watcher.Events, run.watcher.Errors + events, errors = run.watcher.Events(), run.watcher.Errors() } for { select { diff --git a/internal/session/presence_watch_darwin.go b/internal/session/presence_watch_darwin.go new file mode 100644 index 0000000000..24437d9653 --- /dev/null +++ b/internal/session/presence_watch_darwin.go @@ -0,0 +1,224 @@ +//go:build darwin + +package session + +import ( + "errors" + "os" + "path/filepath" + "sync" + + "github.com/fsnotify/fsnotify" + "golang.org/x/sys/unix" +) + +type darwinDoor struct { + dir string + fd int + token byte +} + +// Only directory entries matter here. Watching their files would keep a +// descriptor for every transcript, presence refresh and lock in a held chat. +type darwinDoorWatcher struct { + mu sync.Mutex + kq int + dirs map[string]*darwinDoor + fds map[int]*darwinDoor + retired []*darwinDoor + events chan doorEvent + errors chan error + stop chan struct{} + done chan struct{} + err error +} + +func newDoorWatcher() (doorWatcher, error) { + kq, err := unix.Kqueue() + if err != nil { + return nil, err + } + unix.CloseOnExec(kq) + var wake unix.Kevent_t + unix.SetKevent(&wake, 0, unix.EVFILT_USER, unix.EV_ADD|unix.EV_CLEAR) + if err := changeDoorKevent(kq, wake); err != nil { + unix.Close(kq) + return nil, err + } + w := &darwinDoorWatcher{ + kq: kq, dirs: map[string]*darwinDoor{}, fds: map[int]*darwinDoor{}, + events: make(chan doorEvent), errors: make(chan error, 1), + stop: make(chan struct{}), done: make(chan struct{}), + } + go w.read() + return w, nil +} + +func changeDoorKevent(kq int, change unix.Kevent_t) error { + for { + _, err := unix.Kevent(kq, []unix.Kevent_t{change}, nil, nil) + if err != unix.EINTR { + return err + } + } +} + +func (w *darwinDoorWatcher) Add(dir string) error { + w.mu.Lock() + defer w.mu.Unlock() + select { + case <-w.stop: + return os.ErrClosed + default: + } + dir = filepath.Clean(dir) + if w.dirs[dir] != nil { + return nil + } + fd, err := unix.Open(dir, unix.O_EVTONLY|unix.O_CLOEXEC|unix.O_DIRECTORY, 0) + if err != nil { + return err + } + door := &darwinDoor{dir: dir, fd: fd} + var change unix.Kevent_t + unix.SetKevent(&change, fd, unix.EVFILT_VNODE, unix.EV_ADD|unix.EV_CLEAR) + change.Fflags = unix.NOTE_WRITE | unix.NOTE_DELETE | unix.NOTE_RENAME | unix.NOTE_REVOKE + change.Udata = &door.token + if err := changeDoorKevent(w.kq, change); err != nil { + unix.Close(fd) + return err + } + w.dirs[dir], w.fds[fd] = door, door + return nil +} + +func (w *darwinDoorWatcher) Remove(dir string) error { + w.mu.Lock() + defer w.mu.Unlock() + if door := w.dirs[filepath.Clean(dir)]; door != nil { + err := w.remove(door) + w.report(w.wake()) + return err + } + return nil +} + +// Keep tokens until the next batch is read, so an old event cannot name a +// different folder that reused its descriptor or its token's address. +func (w *darwinDoorWatcher) remove(door *darwinDoor) error { + delete(w.dirs, door.dir) + delete(w.fds, door.fd) + w.retired = append(w.retired, door) + return unix.Close(door.fd) +} + +func (w *darwinDoorWatcher) wake() error { + var wake unix.Kevent_t + unix.SetKevent(&wake, 0, unix.EVFILT_USER, 0) + wake.Fflags = unix.NOTE_TRIGGER + return changeDoorKevent(w.kq, wake) +} + +func (w *darwinDoorWatcher) Close() error { + w.mu.Lock() + select { + case <-w.stop: + default: + close(w.stop) + w.err = errors.Join(w.err, w.wake()) + } + w.mu.Unlock() + <-w.done + return w.err +} + +func (w *darwinDoorWatcher) Events() <-chan doorEvent { return w.events } +func (w *darwinDoorWatcher) Errors() <-chan error { return w.errors } + +func (w *darwinDoorWatcher) report(err error) { + if err != nil { + select { + case w.errors <- err: + default: + } + } +} + +func (w *darwinDoorWatcher) read() { + defer func() { + w.mu.Lock() + defer w.mu.Unlock() + select { + case <-w.stop: + default: + close(w.stop) + } + for _, door := range w.dirs { + w.err = errors.Join(w.err, w.remove(door)) + } + w.retired = nil + w.err = errors.Join(w.err, unix.Close(w.kq)) + close(w.events) + close(w.errors) + close(w.done) + }() + events := make([]unix.Kevent_t, 32) + for { + select { + case <-w.stop: + return + default: + } + // No timeout: an idle engine must not wake once a second. Close and + // Remove trigger the user event, which ends this wait. + n, err := unix.Kevent(w.kq, nil, events, nil) + if err == unix.EINTR { + continue + } + if err != nil { + w.report(err) + return + } + w.mu.Lock() + var pending []doorEvent + for _, event := range events[:n] { + if event.Filter != unix.EVFILT_VNODE { + continue + } + door := w.fds[int(event.Ident)] + if door == nil || event.Udata != &door.token { + continue + } + if event.Flags&unix.EV_ERROR != 0 { + w.report(unix.Errno(event.Data)) + continue + } + if event.Fflags&(unix.NOTE_DELETE|unix.NOTE_RENAME|unix.NOTE_REVOKE) != 0 { + w.report(w.remove(door)) + pending = append(pending, doorEvent{Name: door.dir, Op: fsnotify.Remove}) + continue + } + if event.Fflags&unix.NOTE_WRITE != 0 { + names, err := os.ReadDir(door.dir) + if err != nil { + w.report(err) + continue + } + for _, name := range names { + if presenceRequest(name.Name()) { + pending = append(pending, doorEvent{Name: filepath.Join(door.dir, name.Name()), Op: fsnotify.Create}) + } + } + } + } + w.retired = nil + w.mu.Unlock() + for _, event := range pending { + select { + case w.events <- event: + case <-w.stop: + return + } + } + } +} diff --git a/internal/session/presence_watch_darwin_test.go b/internal/session/presence_watch_darwin_test.go new file mode 100644 index 0000000000..ad19700aa7 --- /dev/null +++ b/internal/session/presence_watch_darwin_test.go @@ -0,0 +1,269 @@ +//go:build darwin + +package session + +import ( + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "golang.org/x/sys/unix" +) + +func doorFDCount(t *testing.T) int { + t.Helper() + fds, err := os.ReadDir("/dev/fd") + if err != nil { + t.Fatal(err) + } + return len(fds) +} + +func watchedDoorFDCount(t *testing.T, backend *darwinDoorWatcher) int { + t.Helper() + // Exclude the short-lived ReadDir descriptor from the steady watch count. + backend.mu.Lock() + defer backend.mu.Unlock() + return doorFDCount(t) +} + +func darwinDoorRun(t *testing.T, watch *presenceDoorWatch) (*presenceDoorRun, *darwinDoorWatcher) { + t.Helper() + watch.mu.Lock() + defer watch.mu.Unlock() + if watch.run == nil || watch.run.watcher == nil { + t.Fatal("directory watcher did not start") + } + return watch.run, watch.run.watcher.(*darwinDoorWatcher) +} + +func writeDoorOrdinaryFiles(t *testing.T, dir string, start, end int) { + t.Helper() + for i := start; i < end; i++ { + if err := os.WriteFile(filepath.Join(dir, fmt.Sprintf("ordinary-%d", i)), []byte("data"), 0o600); err != nil { + t.Fatal(err) + } + } +} + +func TestDarwinPresenceDoorC1OneHandlePerFolder(t *testing.T) { + dir := t.TempDir() + writeDoorOrdinaryFiles(t, dir, 0, 20) + before := doorFDCount(t) + var watch presenceDoorWatch + _, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + _, backend := darwinDoorRun(t, &watch) + after := watchedDoorFDCount(t, backend) + if delta := after - before; delta < 1 || delta > 2 { + t.Fatalf("watching a folder with 20 files added %d descriptors, want at most 2", delta) + } + writeDoorOrdinaryFiles(t, dir, 20, 40) + withMoreFiles := watchedDoorFDCount(t, backend) + if withMoreFiles != after { + t.Fatalf("20 more files changed descriptors from %d to %d", after, withMoreFiles) + } + otherDir := t.TempDir() + _, _, _, otherCancel := watch.subscribe(otherDir) + t.Cleanup(otherCancel) + withOtherFolder := watchedDoorFDCount(t, backend) + if withOtherFolder != after+1 { + t.Fatalf("second folder added %d descriptors, want 1", withOtherFolder-after) + } +} + +func TestDarwinPresenceDoorC2PublishedRequestsWakeByEvent(t *testing.T) { + for _, name := range []string{takeoverName, conversationDeleteRequest, ".task-delete-request-123"} { + for _, replace := range []bool{false, true} { + t.Run(fmt.Sprintf("%s/replace=%t", name, replace), func(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, name) + if replace { + if err := writePresenceRequest(path, []byte("old request")); err != nil { + t.Fatal(err) + } + } + var watch presenceDoorWatch + requests, beats, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + started := time.Now() + if err := writePresenceRequest(path, []byte("new request")); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) + if elapsed := time.Since(started); elapsed >= 500*time.Millisecond { + t.Fatalf("request took %s, want less than 500ms", elapsed) + } + select { + case <-beats: + t.Fatal("request needed a safety pass") + default: + } + }) + } + } +} + +func waitDoorDropped(t *testing.T, watch *presenceDoorWatch, run *presenceDoorRun, dir string) { + t.Helper() + deadline := time.Now().Add(500 * time.Millisecond) + for { + watch.mu.Lock() + watched := run.watched[dir] + watch.mu.Unlock() + if !watched { + return + } + if time.Now().After(deadline) { + t.Fatal("removed or renamed folder stayed watched") + } + time.Sleep(time.Millisecond) + } +} + +func TestDarwinPresenceDoorC4ReplacedFolderIsWatchedAgain(t *testing.T) { + for _, rename := range []bool{false, true} { + t.Run(fmt.Sprintf("rename=%t", rename), func(t *testing.T) { + root := t.TempDir() + dir := filepath.Join(root, "session") + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatal(err) + } + var watch presenceDoorWatch + requests, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + run, backend := darwinDoorRun(t, &watch) + backend.mu.Lock() + fd := backend.dirs[dir].fd + backend.mu.Unlock() + var err error + if rename { + err = os.Rename(dir, filepath.Join(root, "old-session")) + } else { + err = os.Remove(dir) + } + if err != nil { + t.Fatal(err) + } + waitDoorDropped(t, &watch, run, dir) + backend.mu.Lock() + oldWatch := backend.dirs[dir] + backend.mu.Unlock() + if oldWatch != nil { + t.Fatal("old folder retained its kernel watch") + } + if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != unix.EBADF { + t.Fatalf("old folder descriptor is still open: %v", err) + } + watch.rescan(run, time.Now()) + watch.mu.Lock() + watched := run.watched[dir] + watch.mu.Unlock() + if watched { + t.Fatal("safety pass marked a missing folder watched") + } + if err := os.Mkdir(dir, 0o700); err != nil { + t.Fatal(err) + } + watch.rescan(run, time.Now()) + watch.mu.Lock() + watched = run.watched[dir] + watch.mu.Unlock() + if !watched { + t.Fatal("safety pass did not watch the replacement folder") + } + if err := writePresenceRequest(filepath.Join(dir, takeoverName), []byte("request")); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) + }) + } +} + +func TestDarwinPresenceDoorC5ClosesHandlesAndRestarts(t *testing.T) { + var watch presenceDoorWatch + dir := t.TempDir() + _, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + _, _, _, siblingCancel := watch.subscribe(dir) + t.Cleanup(siblingCancel) + run, backend := darwinDoorRun(t, &watch) + backend.mu.Lock() + fd, kq := backend.dirs[dir].fd, backend.kq + backend.mu.Unlock() + cancel() + cancel() + if _, err := unix.FcntlInt(uintptr(fd), unix.F_GETFD, 0); err != nil { + t.Fatalf("folder closed while a client still needed it: %v", err) + } + siblingCancel() + siblingCancel() + for _, descriptor := range []int{fd, kq} { + if _, err := unix.FcntlInt(uintptr(descriptor), unix.F_GETFD, 0); err != unix.EBADF { + t.Fatalf("descriptor %d remained open after last cancel: %v", descriptor, err) + } + } + select { + case <-run.done: + case <-time.After(500 * time.Millisecond): + t.Fatal("shared loop remained running after last cancel") + } + if err := backend.Close(); err != nil { + t.Fatalf("second backend close: %v", err) + } + requests, _, _, restartCancel := watch.subscribe(dir) + t.Cleanup(restartCancel) + _, restarted := darwinDoorRun(t, &watch) + if restarted == backend { + t.Fatal("subscribe after a full stop reused a closed backend") + } + if err := writePresenceRequest(filepath.Join(dir, takeoverName), []byte("request")); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) +} + +func TestDarwinPresenceDoorC5CloseWithoutDrainingEvents(t *testing.T) { + for _, pending := range []bool{false, true} { + t.Run(fmt.Sprintf("pending=%t", pending), func(t *testing.T) { + watcher, err := newDoorWatcher() + if err != nil { + t.Fatal(err) + } + closeStarted := false + t.Cleanup(func() { + if !closeStarted { + _ = watcher.Close() + } + }) + dir := t.TempDir() + if err := watcher.Add(dir); err != nil { + t.Fatal(err) + } + if pending { + if err := writePresenceRequest(filepath.Join(dir, takeoverName), []byte("request")); err != nil { + t.Fatal(err) + } + } + closed := make(chan error, 1) + closeStarted = true + go func() { closed <- watcher.Close() }() + select { + case err := <-closed: + if err != nil { + t.Fatal(err) + } + case <-time.After(500 * time.Millisecond): + t.Fatal("Close left the kernel reader or event sender blocked") + } + if _, ok := <-watcher.Events(); ok { + t.Fatal("event channel remained open after Close") + } + if _, ok := <-watcher.Errors(); ok { + t.Fatal("error channel remained open after Close") + } + }) + } +} diff --git a/internal/session/presence_watch_fsnotify.go b/internal/session/presence_watch_fsnotify.go new file mode 100644 index 0000000000..c99dba27a7 --- /dev/null +++ b/internal/session/presence_watch_fsnotify.go @@ -0,0 +1,20 @@ +//go:build !darwin + +package session + +import "github.com/fsnotify/fsnotify" + +type fsnotifyDoorWatcher struct { + *fsnotify.Watcher +} + +func newDoorWatcher() (doorWatcher, error) { + watcher, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + return &fsnotifyDoorWatcher{watcher}, nil +} + +func (w *fsnotifyDoorWatcher) Events() <-chan doorEvent { return w.Watcher.Events } +func (w *fsnotifyDoorWatcher) Errors() <-chan error { return w.Watcher.Errors } diff --git a/internal/session/presence_watch_fsnotify_test.go b/internal/session/presence_watch_fsnotify_test.go new file mode 100644 index 0000000000..b7f2b80954 --- /dev/null +++ b/internal/session/presence_watch_fsnotify_test.go @@ -0,0 +1,32 @@ +//go:build !darwin + +package session + +import ( + "path/filepath" + "testing" +) + +func TestPresenceDoorC7KeepsFsnotifyDelivery(t *testing.T) { + for _, name := range []string{takeoverName, conversationDeleteRequest, ".task-delete-request-123"} { + t.Run(name, func(t *testing.T) { + var watch presenceDoorWatch + dir := t.TempDir() + requests, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + watch.mu.Lock() + watcher, ok := watch.run.watcher.(*fsnotifyDoorWatcher) + watch.mu.Unlock() + if !ok { + t.Fatal("non-darwin subscription did not use fsnotify") + } + if watcher.Events() != watcher.Watcher.Events || watcher.Errors() != watcher.Watcher.Errors { + t.Fatal("fsnotify channels were replaced by another delivery path") + } + if err := writePresenceRequest(filepath.Join(dir, name), []byte("request")); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) + }) + } +} diff --git a/internal/session/presence_watch_test.go b/internal/session/presence_watch_test.go new file mode 100644 index 0000000000..1921edcf18 --- /dev/null +++ b/internal/session/presence_watch_test.go @@ -0,0 +1,245 @@ +package session + +import ( + "errors" + "path/filepath" + "sync" + "testing" + "time" + + "github.com/fsnotify/fsnotify" +) + +type fakeDoorWatcher struct { + mu sync.Mutex + events chan doorEvent + errors chan error + adds map[string]int + removes map[string]int + addError map[string]error + closes int +} + +func newFakeDoorWatcher() *fakeDoorWatcher { + return &fakeDoorWatcher{ + events: make(chan doorEvent), errors: make(chan error), + adds: map[string]int{}, removes: map[string]int{}, addError: map[string]error{}, + } +} + +func (w *fakeDoorWatcher) Add(dir string) error { + w.mu.Lock() + defer w.mu.Unlock() + w.adds[dir]++ + return w.addError[dir] +} + +func (w *fakeDoorWatcher) Remove(dir string) error { + w.mu.Lock() + defer w.mu.Unlock() + w.removes[dir]++ + return nil +} + +func (w *fakeDoorWatcher) Close() error { + w.mu.Lock() + defer w.mu.Unlock() + w.closes++ + if w.closes == 1 { + close(w.events) + close(w.errors) + } + return nil +} + +func (w *fakeDoorWatcher) Events() <-chan doorEvent { return w.events } +func (w *fakeDoorWatcher) Errors() <-chan error { return w.errors } + +func (w *fakeDoorWatcher) event(t *testing.T, event doorEvent) { + t.Helper() + select { + case w.events <- event: + case <-time.After(500 * time.Millisecond): + t.Fatal("shared loop did not receive an event") + } +} + +// The next receive proves the shared loop finished handling the previous input. +func (w *fakeDoorWatcher) flush(t *testing.T) { + t.Helper() + w.event(t, doorEvent{}) +} + +func waitDoorRequest(t *testing.T, requests <-chan struct{}) { + t.Helper() + select { + case <-requests: + case <-time.After(500 * time.Millisecond): + t.Fatal("request waited for the five-second safety pass") + } +} + +func TestPresenceDoorC2C3RoutesRequestEvents(t *testing.T) { + for _, name := range []string{takeoverName, conversationDeleteRequest, ".task-delete-request-123"} { + for _, op := range []fsnotify.Op{fsnotify.Create, fsnotify.Write} { + t.Run(name+"/"+op.String(), func(t *testing.T) { + fake := newFakeDoorWatcher() + watch := presenceDoorWatch{newWatcher: func() (doorWatcher, error) { return fake, nil }} + dir := t.TempDir() + requests, beats, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + sibling, _, _, siblingCancel := watch.subscribe(dir) + t.Cleanup(siblingCancel) + other, _, _, otherCancel := watch.subscribe(t.TempDir()) + t.Cleanup(otherCancel) + path := filepath.Join(dir, name) + if err := writePresenceRequest(path, []byte("request")); err != nil { + t.Fatal(err) + } + fake.event(t, doorEvent{Name: path, Op: op}) + waitDoorRequest(t, requests) + waitDoorRequest(t, sibling) + fake.flush(t) + select { + case <-other: + t.Fatal("C3: request woke a different folder") + default: + } + select { + case <-beats: + t.Fatal("C2: request delivery needed a safety pass") + default: + } + }) + } + } +} + +func TestPresenceDoorC4SafetyPassReaddsDroppedFolders(t *testing.T) { + for _, op := range []fsnotify.Op{fsnotify.Remove, fsnotify.Rename} { + t.Run(op.String(), func(t *testing.T) { + fake := newFakeDoorWatcher() + watch := presenceDoorWatch{newWatcher: func() (doorWatcher, error) { return fake, nil }} + dir := t.TempDir() + _, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + run := watch.run + fake.event(t, doorEvent{Name: dir, Op: op}) + fake.flush(t) + watch.mu.Lock() + watched := run.watched[dir] + watch.mu.Unlock() + if watched { + t.Fatal("removed folder remained watched") + } + watch.rescan(run, time.Now()) + watch.mu.Lock() + watched = run.watched[dir] + watch.mu.Unlock() + if !watched || fake.adds[dir] != 2 { + t.Fatal("safety pass did not re-add the folder") + } + }) + } +} + +func TestPresenceDoorC5CancelClosesOnceAndCanRestart(t *testing.T) { + var watchers []*fakeDoorWatcher + watch := presenceDoorWatch{newWatcher: func() (doorWatcher, error) { + fake := newFakeDoorWatcher() + watchers = append(watchers, fake) + return fake, nil + }} + dir, otherDir := t.TempDir(), t.TempDir() + _, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + _, _, _, siblingCancel := watch.subscribe(dir) + t.Cleanup(siblingCancel) + _, _, _, otherCancel := watch.subscribe(otherDir) + t.Cleanup(otherCancel) + run, fake := watch.run, watchers[0] + if len(watchers) != 1 || fake.adds[dir] != 1 || fake.adds[otherDir] != 1 { + t.Fatal("subscriptions did not share one watcher and one watch per folder") + } + cancel() + cancel() + if fake.removes[dir] != 0 || fake.closes != 0 { + t.Fatal("cancel closed a folder still used by another client") + } + siblingCancel() + siblingCancel() + if fake.removes[dir] != 1 || fake.closes != 0 { + t.Fatal("last folder client did not remove exactly one watch") + } + otherCancel() + otherCancel() + if fake.removes[otherDir] != 1 || fake.closes != 1 || watch.run != nil { + t.Fatal("last client did not close and release the shared watcher exactly once") + } + select { + case <-run.done: + case <-time.After(500 * time.Millisecond): + t.Fatal("shared loop did not stop") + } + requests, _, _, restartCancel := watch.subscribe(dir) + t.Cleanup(restartCancel) + if len(watchers) != 2 || watch.run == run { + t.Fatal("subscribe after full stop reused the old run") + } + watchers[1].event(t, doorEvent{Name: filepath.Join(dir, takeoverName), Op: fsnotify.Create}) + waitDoorRequest(t, requests) +} + +func TestPresenceDoorC6ErrorsRunBoundedSafetyPass(t *testing.T) { + fake := newFakeDoorWatcher() + dir, missing := t.TempDir(), t.TempDir() + fake.addError[missing] = errors.New("folder not yet present") + watch := presenceDoorWatch{newWatcher: func() (doorWatcher, error) { return fake, nil }} + requests, beats, refreshes, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + _, missingBeats, _, missingCancel := watch.subscribe(missing) + t.Cleanup(missingCancel) + fake.mu.Lock() + delete(fake.addError, missing) + fake.mu.Unlock() + for i := 0; i < 3; i++ { + select { + case fake.errors <- errors.New("watcher overflow"): + case <-time.After(500 * time.Millisecond): + t.Fatal("watcher error blocked the shared loop") + } + fake.flush(t) + } + for _, ticks := range []<-chan time.Time{beats, missingBeats} { + if len(ticks) != 1 { + t.Fatal("safety pass did not coalesce into one pending heartbeat per client") + } + if (<-ticks).IsZero() { + t.Fatal("safety pass delivered an empty heartbeat") + } + } + watch.mu.Lock() + watchingMissing := watch.run.watched[missing] + watch.mu.Unlock() + if !watchingMissing || fake.adds[missing] != 2 { + t.Fatal("error safety pass did not retry a failed folder watch") + } + if len(requests) != 0 || len(refreshes) != 0 { + t.Fatal("watcher error changed request or refresh cadence") + } +} + +func TestPresenceDoorC8IgnoresOrdinaryFileEvents(t *testing.T) { + fake := newFakeDoorWatcher() + watch := presenceDoorWatch{newWatcher: func() (doorWatcher, error) { return fake, nil }} + dir := t.TempDir() + requests, beats, refreshes, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + for _, name := range []string{"transcript.jsonl", presenceName, ".request-temporary"} { + fake.event(t, doorEvent{Name: filepath.Join(dir, name), Op: fsnotify.Create | fsnotify.Write}) + } + fake.flush(t) + if len(requests) != 0 || len(beats) != 0 || len(refreshes) != 0 { + t.Fatal("ordinary file event woke a client") + } +} From 7cb68a490b5178f602e2e15d111a5e337c7d4200 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 9 Nov 2026 17:48:08 -0500 Subject: [PATCH 2/4] engine: the macOS folder watch also sees answers appended to answers.jsonl #2478 lets a quiet session skip its doorstep pass for a minute because the watch wakes it when an answer is WRITTEN. deliverAnswer appends to answers.jsonl, and once that file exists an append does not change the folder's entries, so a folder-only kqueue watch missed it. The darwin watcher now also watches answers.jsonl while it is there: a write reports the answer, an empty just-created file reports nothing (the line comes after the create), and a drain's rename gives the handle back. A held chat costs its folder handle plus one for an answers file only while an answer waits. TestPresenceDoorC6 follows #2478's contract: a watcher error now knocks every session once as well as running the safety pass. Co-Authored-By: Claude Opus 5.5 --- internal/session/presence_watch_darwin.go | 116 +++++++++++++++--- .../session/presence_watch_darwin_test.go | 71 +++++++++++ internal/session/presence_watch_test.go | 11 +- 3 files changed, 178 insertions(+), 20 deletions(-) diff --git a/internal/session/presence_watch_darwin.go b/internal/session/presence_watch_darwin.go index 24437d9653..08c92f289c 100644 --- a/internal/session/presence_watch_darwin.go +++ b/internal/session/presence_watch_darwin.go @@ -12,14 +12,24 @@ import ( "golang.org/x/sys/unix" ) +// darwinDoor is one kernel handle: a session folder, or the folder's answers +// file while one is waiting to be drained. type darwinDoor struct { dir string fd int token byte + // answers is a folder's answers-file handle; folder is set on that handle. + answers *darwinDoor + folder *darwinDoor } -// Only directory entries matter here. Watching their files would keep a -// descriptor for every transcript, presence refresh and lock in a held chat. +// Only directory entries matter here, plus one file. Watching every file would +// keep a descriptor for each transcript, presence refresh and lock in a held +// chat. Requests are published by create or rename, which change the folder's +// entries. An answer is APPENDED to answers.jsonl, which a folder watch cannot +// see once the file exists, and the doorstep counts an answer only once it is +// written ([presenceDoorstepEvent]), so that one file is watched while it is +// there. type darwinDoorWatcher struct { mu sync.Mutex kq int @@ -63,6 +73,25 @@ func changeDoorKevent(kq int, change unix.Kevent_t) error { } } +// open takes an event-only handle on path and registers it with the kqueue. +func (w *darwinDoorWatcher) open(path string, flags int, fflags uint32) (*darwinDoor, error) { + fd, err := unix.Open(path, unix.O_EVTONLY|unix.O_CLOEXEC|flags, 0) + if err != nil { + return nil, err + } + door := &darwinDoor{fd: fd} + var change unix.Kevent_t + unix.SetKevent(&change, fd, unix.EVFILT_VNODE, unix.EV_ADD|unix.EV_CLEAR) + change.Fflags = fflags + change.Udata = &door.token + if err := changeDoorKevent(w.kq, change); err != nil { + unix.Close(fd) + return nil, err + } + w.fds[fd] = door + return door, nil +} + func (w *darwinDoorWatcher) Add(dir string) error { w.mu.Lock() defer w.mu.Unlock() @@ -75,23 +104,43 @@ func (w *darwinDoorWatcher) Add(dir string) error { if w.dirs[dir] != nil { return nil } - fd, err := unix.Open(dir, unix.O_EVTONLY|unix.O_CLOEXEC|unix.O_DIRECTORY, 0) + door, err := w.open(dir, unix.O_DIRECTORY, unix.NOTE_WRITE|unix.NOTE_DELETE|unix.NOTE_RENAME|unix.NOTE_REVOKE) if err != nil { return err } - door := &darwinDoor{dir: dir, fd: fd} - var change unix.Kevent_t - unix.SetKevent(&change, fd, unix.EVFILT_VNODE, unix.EV_ADD|unix.EV_CLEAR) - change.Fflags = unix.NOTE_WRITE | unix.NOTE_DELETE | unix.NOTE_RENAME | unix.NOTE_REVOKE - change.Udata = &door.token - if err := changeDoorKevent(w.kq, change); err != nil { - unix.Close(fd) - return err + door.dir = dir + w.dirs[dir] = door + // An answers file already there is watched from now on. What it holds is + // not reported: the session looks at its doorstep right after it subscribes. + if _, err := w.watchAnswers(door); err != nil { + w.report(err) } - w.dirs[dir], w.fds[fd] = door, door return nil } +// watchAnswers starts watching folder's answers file if it exists and is not +// watched yet. It says whether the file already holds something, which a write +// made before the watch began would otherwise leave unreported. An empty file +// is not reported: deliverAnswer creates it and only then writes the line. +func (w *darwinDoorWatcher) watchAnswers(folder *darwinDoor) (bool, error) { + if folder.answers != nil { + return false, nil + } + door, err := w.open(filepath.Join(folder.dir, answersName), 0, unix.NOTE_WRITE|unix.NOTE_EXTEND|unix.NOTE_DELETE|unix.NOTE_RENAME|unix.NOTE_REVOKE) + if errors.Is(err, unix.ENOENT) { + return false, nil + } + if err != nil { + return false, err + } + door.dir, door.folder, folder.answers = folder.dir, folder, door + var st unix.Stat_t + if err := unix.Fstat(door.fd, &st); err != nil { + return false, err + } + return st.Size > 0, nil +} + func (w *darwinDoorWatcher) Remove(dir string) error { w.mu.Lock() defer w.mu.Unlock() @@ -103,13 +152,24 @@ func (w *darwinDoorWatcher) Remove(dir string) error { return nil } -// Keep tokens until the next batch is read, so an old event cannot name a -// different folder that reused its descriptor or its token's address. +// remove closes one handle; a folder's answers handle goes with it. Tokens are +// kept until the next batch is read, so an old event cannot name a different +// handle that reused its descriptor or its token's address. func (w *darwinDoorWatcher) remove(door *darwinDoor) error { - delete(w.dirs, door.dir) + var err error + if door.answers != nil { + err = w.remove(door.answers) + } + if door.folder != nil { + if door.folder.answers == door { + door.folder.answers = nil + } + } else { + delete(w.dirs, door.dir) + } delete(w.fds, door.fd) w.retired = append(w.retired, door) - return unix.Close(door.fd) + return errors.Join(err, unix.Close(door.fd)) } func (w *darwinDoorWatcher) wake() error { @@ -163,6 +223,7 @@ func (w *darwinDoorWatcher) read() { close(w.done) }() events := make([]unix.Kevent_t, 32) + gone := uint32(unix.NOTE_DELETE | unix.NOTE_RENAME | unix.NOTE_REVOKE) for { select { case <-w.stop: @@ -193,7 +254,19 @@ func (w *darwinDoorWatcher) read() { w.report(unix.Errno(event.Data)) continue } - if event.Fflags&(unix.NOTE_DELETE|unix.NOTE_RENAME|unix.NOTE_REVOKE) != 0 { + if door.folder != nil { + // The answers file: a write is an answer landing. Drained + // (renamed away) or deleted, its handle goes; a new file shows + // up as a folder change. + if event.Fflags&(unix.NOTE_WRITE|unix.NOTE_EXTEND) != 0 { + pending = append(pending, doorEvent{Name: filepath.Join(door.dir, answersName), Op: fsnotify.Write}) + } + if event.Fflags&gone != 0 { + w.report(w.remove(door)) + } + continue + } + if event.Fflags&gone != 0 { w.report(w.remove(door)) pending = append(pending, doorEvent{Name: door.dir, Op: fsnotify.Remove}) continue @@ -204,10 +277,19 @@ func (w *darwinDoorWatcher) read() { w.report(err) continue } + answers := false for _, name := range names { if presenceRequest(name.Name()) { pending = append(pending, doorEvent{Name: filepath.Join(door.dir, name.Name()), Op: fsnotify.Create}) } + answers = answers || name.Name() == answersName + } + if answers { + wrote, err := w.watchAnswers(door) + w.report(err) + if wrote { + pending = append(pending, doorEvent{Name: filepath.Join(door.dir, answersName), Op: fsnotify.Write}) + } } } } diff --git a/internal/session/presence_watch_darwin_test.go b/internal/session/presence_watch_darwin_test.go index ad19700aa7..3f99dc1012 100644 --- a/internal/session/presence_watch_darwin_test.go +++ b/internal/session/presence_watch_darwin_test.go @@ -267,3 +267,74 @@ func TestDarwinPresenceDoorC5CloseWithoutDrainingEvents(t *testing.T) { }) } } + +func waitDoorFDs(t *testing.T, backend *darwinDoorWatcher, want int, what string) { + t.Helper() + deadline := time.Now().Add(500 * time.Millisecond) + for { + got := watchedDoorFDCount(t, backend) + if got == want { + return + } + if time.Now().After(deadline) { + t.Fatalf("%s: %d descriptors, want %d", what, got, want) + } + time.Sleep(time.Millisecond) + } +} + +func noDoorRequest(t *testing.T, requests <-chan struct{}, what string) { + t.Helper() + select { + case <-requests: + t.Fatal(what) + case <-time.After(200 * time.Millisecond): + } +} + +// C9: an answer appended to answers.jsonl wakes its folder's session by event, +// whether the file is new or already waiting, and a created but still empty +// file does not (deliverAnswer creates it and then writes the line). +// C10: the answers file costs one handle while it exists, and draining it +// gives that handle back. +func TestDarwinPresenceDoorC9C10AnswersWakeAndReleaseTheirHandle(t *testing.T) { + dir := t.TempDir() + var watch presenceDoorWatch + requests, _, _, cancel := watch.subscribe(dir) + t.Cleanup(cancel) + _, backend := darwinDoorRun(t, &watch) + folderOnly := watchedDoorFDCount(t, backend) + + empty, err := os.OpenFile(AnswersPath(dir), os.O_CREATE|os.O_WRONLY, 0o600) + if err != nil { + t.Fatal(err) + } + empty.Close() + noDoorRequest(t, requests, "C9: an empty answers file woke the session before its line was written") + waitDoorFDs(t, backend, folderOnly+1, "C10: answers file present") + + if err := deliverAnswer(dir, Answer{At: time.Now()}); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) + if err := deliverAnswer(dir, Answer{At: time.Now()}); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) // C9: an append to a file already there + + if _, err := DrainAnswers(dir); err != nil { + t.Fatal(err) + } + waitDoorFDs(t, backend, folderOnly, "C10: answers drained") + // Draining leaves a stray wake at most; take it before the next check. + select { + case <-requests: + case <-time.After(50 * time.Millisecond): + } + + if err := deliverAnswer(dir, Answer{At: time.Now()}); err != nil { + t.Fatal(err) + } + waitDoorRequest(t, requests) // C9: a fresh file after a drain + waitDoorFDs(t, backend, folderOnly+1, "C10: a fresh answers file") +} diff --git a/internal/session/presence_watch_test.go b/internal/session/presence_watch_test.go index 1921edcf18..835f1b5467 100644 --- a/internal/session/presence_watch_test.go +++ b/internal/session/presence_watch_test.go @@ -190,7 +190,7 @@ func TestPresenceDoorC5CancelClosesOnceAndCanRestart(t *testing.T) { waitDoorRequest(t, requests) } -func TestPresenceDoorC6ErrorsRunBoundedSafetyPass(t *testing.T) { +func TestPresenceDoorC6ErrorsRunBoundedSafetyPassAndKnock(t *testing.T) { fake := newFakeDoorWatcher() dir, missing := t.TempDir(), t.TempDir() fake.addError[missing] = errors.New("folder not yet present") @@ -224,8 +224,13 @@ func TestPresenceDoorC6ErrorsRunBoundedSafetyPass(t *testing.T) { if !watchingMissing || fake.adds[missing] != 2 { t.Fatal("error safety pass did not retry a failed folder watch") } - if len(requests) != 0 || len(refreshes) != 0 { - t.Fatal("watcher error changed request or refresh cadence") + // An error can have dropped any folder's event, so every session is knocked + // once to look at its doorstep now; the knocks coalesce like the beats. + if len(requests) != 1 { + t.Fatalf("watcher error must knock every session once: %d request(s) pending", len(requests)) + } + if len(refreshes) != 0 { + t.Fatal("watcher error changed the refresh cadence") } } From 47c07cb490ebb24b6e2a6389ddd67fbfbb84f5b3 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 9 Nov 2026 17:48:08 -0500 Subject: [PATCH 3/4] docs: change entry for #2551 Co-Authored-By: Claude Opus 5.5 --- docs/changes/unreleased/2551-darwin-folder-watch.md | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 docs/changes/unreleased/2551-darwin-folder-watch.md diff --git a/docs/changes/unreleased/2551-darwin-folder-watch.md b/docs/changes/unreleased/2551-darwin-folder-watch.md new file mode 100644 index 0000000000..09f6dbd3ac --- /dev/null +++ b/docs/changes/unreleased/2551-darwin-folder-watch.md @@ -0,0 +1,8 @@ +--- +kind: fixed +title: On macOS a held chat's folder costs one watch handle, not one per file in it +pr: 2551 +surface: [engine, app] +invalidates: + - "On macOS every chat the app had held kept a file descriptor open in codeaf app for each file in its session folder, about ten per chat for the life of the app, because fsnotify's kqueue backend watches every file of a watched folder. The presence watch now holds one handle for the folder, plus one for answers.jsonl only while an answer waits to be drained. Linux and Windows already used one handle per folder." +--- From 1cfcceb7c45a1652ed8682a91a5a247781940ee6 Mon Sep 17 00:00:00 2001 From: Abir Abbas Date: Mon, 9 Nov 2026 17:50:45 -0500 Subject: [PATCH 4/4] engine: the darwin watch tests count descriptors by name os.ReadDir("/dev/fd") stats every entry and fails with EBADF inside a test binary on macOS 15.6, so TestDarwinPresenceDoorC1 failed before any watch was made. Readdirnames only counts names (found by the Mac proof run). Co-Authored-By: Claude Opus 5.5 --- internal/session/presence_watch_darwin_test.go | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/internal/session/presence_watch_darwin_test.go b/internal/session/presence_watch_darwin_test.go index 3f99dc1012..73bc0c9da0 100644 --- a/internal/session/presence_watch_darwin_test.go +++ b/internal/session/presence_watch_darwin_test.go @@ -12,13 +12,20 @@ import ( "golang.org/x/sys/unix" ) +// doorFDCount counts this process's descriptors by name only. os.ReadDir would +// stat every entry, and on macOS some of /dev/fd cannot be stat'ed (EBADF). func doorFDCount(t *testing.T) int { t.Helper() - fds, err := os.ReadDir("/dev/fd") + dir, err := os.Open("/dev/fd") if err != nil { t.Fatal(err) } - return len(fds) + defer dir.Close() + names, err := dir.Readdirnames(-1) + if err != nil { + t.Fatal(err) + } + return len(names) } func watchedDoorFDCount(t *testing.T, backend *darwinDoorWatcher) int {