// Watcher tests: CRUD events for files, directories, sub-entries, and // symlinks; event coalescing; multiple subscriptions; error handling; // watch lifecycle; public API validation; and watcherBase/ // dirWatchError internals. Each test runs against every watcher available // on the host OS unless it exercises internal types directly. package fswatch import ( "cmp" "errors" "fmt" "math/rand" "os" "os/exec" "path/filepath" "runtime" "slices" "strings" "sync" "sync/atomic" "testing" "time" ) // ----- helpers ----------------------------------------------------------- // defaultEventTimeout is the per-`next` wait used by subscribe tests for // the fast/responsive backends (inotify, fanotify, Windows). Scales up // on retry via [watcherEventTimeout] so the fast path is cheap. func defaultEventTimeout() time.Duration { return 1 * time.Second } // kqueueFSEventsTimeout is the per-event deadline for the kqueue and // fsevents backends. Those have materially higher kernel-to-userspace // latency than inotify/fanotify/Windows: kqueue uses directory // NOTE_WRITE + compareDir which takes a scheduling round-trip per // change, and fsevents introduces its own batching on top of the GCD // dispatch queue. Scales up on retry via [watcherEventTimeout]. func kqueueFSEventsTimeout() time.Duration { return 2 * time.Second } // watcherEventTimeout returns the appropriate per-event deadline for // the backend under test, scaled by the current [testingT]'s retry attempt // number. The fast-path uses the base timeout (1-2 seconds); retries // scale up so a single environmental hiccup gets a longer wait without // inflating every passing run's wall-clock. func watcherEventTimeout(t testingT, w Watcher) time.Duration { base := defaultEventTimeout() if w == FSEvents() || w == Kqueue() { base = kqueueFSEventsTimeout() } scale := 1 if rt, ok := t.(*retryT); ok { scale = retryTimeoutScale(rt.attempt) } return base * time.Duration(scale) } // availableWatchers is populated at init time from whichever backends // are available on the current platform, plus any test-only watcher // variants registered in additionalTestWatchers (see e.g. // fanotify_linux_test.go). var availableWatchers []Watcher // additionalTestWatchers is appended to by platform-specific *_test.go // init() functions to register test-only watcher variants (e.g. the // fanotify-no-rename backend that exercises the FAN_MOVED_FROM/_TO // fallback path). Producers' init() must run before this file's init(); // since Go runs file inits in lexicographic file-name order and this // file is watcher_test.go, that ordering is satisfied for every other // *_test.go file in the package. var additionalTestWatchers []Watcher func init() { for _, b := range AllWatchers() { if b.Available() { availableWatchers = append(availableWatchers, b) } } for _, b := range additionalTestWatchers { if b.Available() { availableWatchers = append(availableWatchers, b) } } } // runForEachWatcher runs fn as a subtest for every available watcher. // // The per-backend test body receives a [testingT] (a subset of *testing.T) // rather than the real *testing.T. This lets [runWithRetry] re-run a // body that fails due to environmental flakes (macOS event-delivery // stalls under load) before propagating the failure to the real test // runner. func runForEachWatcher(t *testing.T, fn func(t testingT, watcherImpl Watcher)) { t.Helper() for _, b := range availableWatchers { t.Run(b.Name(), func(t *testing.T) { t.Parallel() runWithRetry(t, func(rt testingT) { fn(rt, b) }) }) } } // newTmpDir creates a fresh temp dir, resolves any symlinks in the path so // it matches what backends report, and registers cleanup. func newTmpDir(t testingT) string { t.Helper() d := t.TempDir() resolved, err := filepath.EvalSymlinks(d) if err != nil { t.Fatal(err) } return resolved } func makeDirSymlink(t testingT, target string, link string) { t.Helper() if runtime.GOOS == "windows" { if err := exec.Command("cmd", "/c", "mklink", "/J", link, target).Run(); err == nil { return } } if err := os.Symlink(target, link); err != nil { t.Skipf("directory symlink support is not available: %v", err) } } // nameCounter generates unique file names per test to avoid collisions. var nameCounter atomic.Uint64 func uniqueName(parts ...string) string { n := nameCounter.Add(1) suffix := fmt.Sprintf("test%d%d", n, rand.Int63()) return filepath.Join(append(parts, suffix)...) } // subPath produces a unique name in dir. func subPath(dir string) string { return uniqueName(dir) } // newDirectWatcher creates a bare dirWatch for unit-testing tree/debounce // helpers without going through the full backend subscribe path. Each // test gets its own debouncer so tests don't share goroutine state. func newDirectWatcher(t testingT, dir string) *dirWatch { t.Helper() w := newDirWatch(dir, dir, newDebounce()) w.recursive = true t.Cleanup(func() { w.destroyDebounce() }) return w } // subscribeFor sets up a recorder + WatchDirectory and registers cleanup. func subscribeFor(t testingT, dir string, watcherImpl Watcher) (*recordingWatcher, Watch) { return subscribeForOpts(t, dir, watcherImpl, WithRecursive()) } // settleSleep is the post-subscribe settle wait. Empirically tuned per // backend: fsevents and kqueue need a couple of hundred ms to actually // arm their watches on the freshly-created tmp dir, while inotify/ // fanotify/Windows are essentially synchronous. func settleSleep(w Watcher) time.Duration { if w == FSEvents() || w == Kqueue() { return 300 * time.Millisecond } return 60 * time.Millisecond } // preSubscribeSleep gives the macOS fsevents stream timestamp enough // distance from any tmp-dir creation just before subscribe; without // it the initial event batch may include the watched dir's own create. func preSubscribeSleep(w Watcher) time.Duration { if w == FSEvents() || w == Kqueue() { return 50 * time.Millisecond } return 0 } // subscribeFileFor sets up a recorder + WatchFile and registers cleanup. func subscribeFileFor(t testingT, path string, watcherImpl Watcher) (*recordingWatcher, Watch) { t.Helper() if d := preSubscribeSleep(watcherImpl); d > 0 { time.Sleep(d) } r := newRecorder(t) r.watcher = watcherImpl sub, err := watcherImpl.WatchFile(path, r.callback) if err != nil { t.Fatalf("subscribeFile: %v", err) } t.Cleanup(func() { _ = sub.Close() }) time.Sleep(settleSleep(watcherImpl)) return r, sub } // subscribeForOpts sets up a recorder + WatchDirectory with options and registers cleanup. func subscribeForOpts(t testingT, dir string, watcherImpl Watcher, opts ...WatchOption) (*recordingWatcher, Watch) { t.Helper() if d := preSubscribeSleep(watcherImpl); d > 0 { time.Sleep(d) } r := newRecorder(t) r.watcher = watcherImpl sub, err := watcherImpl.WatchDirectory(dir, r.callback, opts...) if err != nil { t.Fatalf("subscribe: %v", err) } t.Cleanup(func() { _ = sub.Close() }) time.Sleep(settleSleep(watcherImpl)) return r, sub } // ----- recordingWatcher -------------------------------------------------- type recordingWatcher struct { t testingT watcher Watcher // bound at subscribe time so expect* helpers can choose timeouts mu sync.Mutex cond *sync.Cond buf []Event errs []error } func newRecorder(t testingT) *recordingWatcher { r := &recordingWatcher{t: t} r.cond = sync.NewCond(&r.mu) return r } // deadline returns the per-event timeout appropriate for the recorder's // bound watcher backend, or the default if no watcher was attached. // The returned duration scales with the current retry attempt when the // recorder is bound to a [retryT]. func (r *recordingWatcher) deadline() time.Duration { if r.watcher == nil { return scaledDeadline(r.t, defaultEventTimeout()) } return watcherEventTimeout(r.t, r.watcher) } // scaledDeadline multiplies base by the retry scale for t (if t is a // retryT), so per-event timeouts grow on retries without inflating the // fast path. func scaledDeadline(t testingT, base time.Duration) time.Duration { if rt, ok := t.(*retryT); ok { return base * time.Duration(retryTimeoutScale(rt.attempt)) } return base } func (r *recordingWatcher) callback(events []Event, err error) { r.mu.Lock() defer r.mu.Unlock() if err != nil { r.errs = append(r.errs, err) } r.buf = append(r.buf, events...) r.cond.Broadcast() } // next blocks for up to d for at least one event, then drains and returns // everything that has accumulated. func (r *recordingWatcher) next(d time.Duration) []Event { r.t.Helper() deadline := time.Now().Add(d) r.mu.Lock() defer r.mu.Unlock() for len(r.buf) == 0 { remaining := time.Until(deadline) if remaining <= 0 { return nil } stopper := time.AfterFunc(remaining, func() { r.mu.Lock() r.cond.Broadcast() r.mu.Unlock() }) r.cond.Wait() stopper.Stop() } out := slices.Clone(r.buf) r.buf = nil return out } // drainQuiet drains any buffered events, then waits at most d to make sure // no further events arrive. Returns whatever shows up. func (r *recordingWatcher) drainQuiet(d time.Duration) []Event { r.t.Helper() r.mu.Lock() r.buf = nil r.mu.Unlock() time.Sleep(d) r.mu.Lock() out := slices.Clone(r.buf) r.buf = nil r.mu.Unlock() return out } // gather waits up to `wait` for at least one event, then settles for // `settle` to give the rest of the debounced batch a chance to arrive. func (r *recordingWatcher) gather(wait, settle time.Duration) []Event { first := r.next(wait) if len(first) == 0 { return nil } time.Sleep(settle) r.mu.Lock() defer r.mu.Unlock() more := slices.Clone(r.buf) r.buf = nil return append(first, more...) } // gatherUntilQuiet collects events until either the initial wait expires // without seeing one, or the recorder has gone quiet for `quiet`. Useful // for assertions that need to observe events possibly spread across // multiple debounce batches (e.g. rapid-coalescing tests). func (r *recordingWatcher) gatherUntilQuiet(initialWait, quiet time.Duration) []Event { first := r.next(initialWait) if len(first) == 0 { return nil } all := first for { more := r.next(quiet) if len(more) == 0 { return all } all = append(all, more...) } } // waitForEvent blocks until an event matching pred is observed in the // recorder's accumulating buffer, or until deadline elapses. Returns all // events collected up to the success / timeout (and drains them from the // buffer). Useful for tests where the kernel backend takes a variable // amount of time to install/propagate a fresh watch, instead of betting // on a fixed sleep that breaks under host CPU/IO contention. func (r *recordingWatcher) waitForEvent(d time.Duration, pred func(Event) bool) []Event { r.t.Helper() deadline := time.Now().Add(d) for { r.mu.Lock() if slices.ContainsFunc(r.buf, pred) { out := slices.Clone(r.buf) r.buf = nil r.mu.Unlock() return out } remaining := time.Until(deadline) if remaining <= 0 { out := slices.Clone(r.buf) r.buf = nil r.mu.Unlock() return out } stopper := time.AfterFunc(remaining, func() { r.mu.Lock() r.cond.Broadcast() r.mu.Unlock() }) r.cond.Wait() stopper.Stop() r.mu.Unlock() } } // waitForAll polls the recorder's buffer until every event in want is // observed (paths matched, kind matched) or d elapses. Returns the full // accumulated set drained from the buffer either way. Extra events // outside of want are kept in the returned slice but do not count // against the deadline. // // Use this instead of one-shot r.next() in any test where the kernel // might split events across multiple debounce batches or take a moment // to install a watch on a freshly created dir. The retry behavior makes // the test robust to host CPU/IO contention. func (r *recordingWatcher) waitForAll(d time.Duration, want []wantEvent) []Event { r.t.Helper() if len(want) == 0 { return nil } deadline := time.Now().Add(d) collected := make([]Event, 0) for { r.mu.Lock() collected = append(collected, r.buf...) r.buf = nil r.mu.Unlock() if haveAll(collected, want) { return collected } remaining := time.Until(deadline) if remaining <= 0 { return collected } r.mu.Lock() if len(r.buf) > 0 { r.mu.Unlock() continue } stopper := time.AfterFunc(remaining, func() { r.mu.Lock() r.cond.Broadcast() r.mu.Unlock() }) r.cond.Wait() stopper.Stop() r.mu.Unlock() } } // haveAll reports whether every event in want is matched at least once // in got. Extra got events are ignored. func haveAll(got []Event, want []wantEvent) bool { for _, w := range want { found := false for _, e := range got { if e.Kind == w.Kind && e.Path == w.Path { found = true break } } if !found { return false } } return true } // expectEventSet polls until every wanted event has arrived (or the // scaled deadline elapses) and then asserts the matching set // (ignoring order). Use everywhere the test had next/gather followed // by assertEventSet; it removes the timing assumption that the events // land in one debounce batch. func expectEventSet(t testingT, r *recordingWatcher, want []wantEvent) []Event { t.Helper() got := r.waitForAll(r.deadline(), want) assertEventSet(t, got, want) return got } // expectEventSequence polls until every wanted event has arrived, then // asserts they appear in the exact specified order (filtered to // wanted paths). Order-sensitive callers that previously used // assertEventSequence on a one-shot next/gather. func expectEventSequence(t testingT, r *recordingWatcher, want []wantEvent) []Event { t.Helper() got := r.waitForAll(r.deadline(), want) assertEventSequence(t, got, want) return got } // expectContains polls until any event matching kind+path arrives, then // returns the accumulated event slice. Use for tests that don't care // about a specific set of events but want to verify at least one // specific event surfaced. func expectContains(t testingT, r *recordingWatcher, kind EventKind, path string) []Event { t.Helper() d := r.deadline() got := r.waitForEvent(d, func(e Event) bool { return e.Kind == kind && e.Path == path }) if !containsEvent(got, kind, path) { t.Fatalf("expected event %s %s within %s, got %v", kind, path, d, toWantEvents(got)) } return got } func expectNoBufferedEvents(t testingT, r *recordingWatcher, msg string) { t.Helper() r.mu.Lock() got := slices.Clone(r.buf) r.buf = nil r.mu.Unlock() if len(got) > 0 { t.Fatalf("%s, got %v", msg, toWantEvents(got)) } } func assertNoEventsForPath(t testingT, got []Event, path, msg string) { t.Helper() got = filterEventsForPaths(got, path) if len(got) > 0 { t.Fatalf("%s %s, got %v", msg, path, toWantEvents(got)) } } // ----- assertion helpers ------------------------------------------------- type wantEvent struct { Kind EventKind Path string } func toWantEvents(events []Event) []wantEvent { out := make([]wantEvent, len(events)) for i, e := range events { out[i] = wantEvent{Kind: e.Kind, Path: e.Path} } return out } // assertEventSet compares two event sets ignoring order. // Events for paths not in want are ignored (e.g. parent-dir update noise). func assertEventSet(t testingT, got []Event, want []wantEvent) { t.Helper() got = filterToWantedPaths(got, want) gotW := toWantEvents(got) cmpEvents := func(a, b wantEvent) int { if a.Kind != b.Kind { return cmp.Compare(a.Kind, b.Kind) } return cmp.Compare(a.Path, b.Path) } slices.SortFunc(gotW, cmpEvents) slices.SortFunc(want, cmpEvents) if !equalWantEvents(gotW, want) { t.Fatalf("event mismatch\nwant: %v\n got: %v", want, gotW) } } // assertEventSequence is like assertEventSet but order-sensitive. // Events for paths not in want are ignored (e.g. parent-dir update noise). func assertEventSequence(t testingT, got []Event, want []wantEvent) { t.Helper() got = filterToWantedPaths(got, want) gotW := toWantEvents(got) if !equalWantEvents(gotW, want) { t.Fatalf("event sequence mismatch\nwant: %v\n got: %v", want, gotW) } } // filterToWantedPaths returns only events whose path appears in want. func filterToWantedPaths(got []Event, want []wantEvent) []Event { paths := make(map[string]struct{}, len(want)) for _, w := range want { paths[w.Path] = struct{}{} } filtered := make([]Event, 0, len(got)) for _, e := range got { if _, ok := paths[e.Path]; ok { filtered = append(filtered, e) } } return filtered } func equalWantEvents(a, b []wantEvent) bool { if len(a) != len(b) { return false } for i := range a { if a[i] != b[i] { return false } } return true } // containsEvent reports whether got contains an event with the given type+path. func containsEvent(got []Event, typ EventKind, path string) bool { for _, e := range got { if e.Kind == typ && e.Path == path { return true } } return false } func TestRebasePath(t *testing.T) { t.Parallel() volume := filepath.VolumeName(os.TempDir()) root := volume + string(filepath.Separator) from := filepath.Join(root, "from") to := filepath.Join(root, "to") tests := []struct { name string path string from string to string want string }{ { name: "exact root", path: from, from: from, to: to, want: to, }, { name: "child", path: filepath.Join(from, "child"), from: from, to: to, want: filepath.Join(to, "child"), }, { name: "sibling", path: filepath.Join(root, "from-sibling", "child"), from: from, to: to, want: filepath.Join(root, "from-sibling", "child"), }, { name: "from root", path: filepath.Join(root, "child"), from: root, to: to, want: filepath.Join(to, "child"), }, { name: "to root", path: filepath.Join(from, "child"), from: from, to: root, want: filepath.Join(root, "child"), }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() if got := rebasePath(tt.path, tt.from, tt.to); got != tt.want { t.Fatalf("rebasePath(%q, %q, %q) = %q, want %q", tt.path, tt.from, tt.to, got, tt.want) } }) } } func TestPhysicalDirForResolvesSymlinkAncestor(t *testing.T) { t.Parallel() root := t.TempDir() target := filepath.Join(root, "target") if err := os.MkdirAll(filepath.Join(target, "nested"), 0o755); err != nil { t.Fatal(err) } link := filepath.Join(root, "link") makeDirSymlink(t, target, link) dir := filepath.Join(link, "nested") want := physicalDirFor(filepath.Join(target, "nested")) if got := physicalDirFor(dir); got != want { t.Fatalf("physicalDirFor(%q) = %q, want %q", dir, got, want) } } func TestIsInDirectoryOrSelf(t *testing.T) { t.Parallel() volume := filepath.VolumeName(os.TempDir()) root := volume + string(filepath.Separator) parent := filepath.Join(root, "parent") child := filepath.Join(parent, "child") nested := filepath.Join(child, "nested") siblingPrefix := filepath.Join(root, "parent-sibling") tests := []struct { name string dir string path string want bool }{ {name: "exact", dir: parent, path: parent, want: true}, {name: "child", dir: parent, path: child, want: true}, {name: "nested", dir: parent, path: nested, want: true}, {name: "sibling prefix", dir: parent, path: siblingPrefix, want: false}, {name: "root self", dir: root, path: root, want: true}, {name: "root child", dir: root, path: filepath.Join(root, "child"), want: true}, {name: "empty dir", dir: "", path: child, want: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { t.Parallel() if got := isInDirectoryOrSelf(tt.dir, tt.path); got != tt.want { t.Fatalf("isInDirectoryOrSelf(%q, %q) = %v, want %v", tt.dir, tt.path, got, tt.want) } }) } } // filterEventsForPaths returns only the events whose Path is in the // allowed set. Used to discard incidental dir-update events that some // backends emit for the parent dir of a touched file. func filterEventsForPaths(events []Event, paths ...string) []Event { allow := make(map[string]struct{}, len(paths)) for _, p := range paths { allow[p] = struct{}{} } out := make([]Event, 0, len(events)) for _, e := range events { if _, ok := allow[e.Path]; ok { out = append(out, e) } } return out } // replayEventList re-applies a sequence of events through a fresh // eventList and returns the coalesced result. This is what the directory watch // would have produced if every event had landed in the same debounce // batch; useful for assertions that must be tolerant to batch splitting. func replayEventList(events []Event) []Event { var el eventList for _, e := range events { switch e.Kind { case EventUpdate: el.update(e.Path) case EventDelete: el.remove(e.Path) } } return el.getEvents() } // ----- files ------------------------------------------------------------- func TestWatchFileCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestWatchFileUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) // Mirror upstream JS: create file AFTER subscribe so the create // event populates the watcherImpl's internal tree, then update it so // the subsequent modify event is correctly classified as update. if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) // consume the create event if err := os.WriteFile(f, []byte("v2-longer"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestWatchFileRename(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f1 := subPath(dir) f2 := subPath(dir) if err := os.WriteFile(f1, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Rename(f1, f2); err != nil { t.Fatal(err) } expectEventSet(t, r, []wantEvent{ {EventDelete, f1}, {EventUpdate, f2}, }) }) } func TestWatchFileRenameExisting(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) // Existing file present at subscribe time. f1 := subPath(dir) if err := os.WriteFile(f1, []byte("hi"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) f2 := subPath(dir) if err := os.Rename(f1, f2); err != nil { t.Fatal(err) } expectEventSet(t, r, []wantEvent{ {EventDelete, f1}, {EventUpdate, f2}, }) }) } func TestWatchFileDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Remove(f); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f}}) }) } // ----- directories ------------------------------------------------------- func TestSubscribeDirCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) if err := os.Mkdir(f, 0o755); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } // TestSubscribeNonASCIIPath checks that every backend round-trips a // non-ASCII path byte-for-byte: subscribe to a directory whose name // contains precomposed (NFC) Unicode, create a child with non-ASCII // bytes in its name, and assert the event's Path equals what we would // have produced with filepath.Join. Guards against any backend (or the // shared event path) silently mutating the bytes. func TestSubscribeNonASCIIPath(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { parent := newTmpDir(t) // "café" + "résumé"; both precomposed NFC. dir := filepath.Join(parent, "caf\u00e9-dir") if err := os.Mkdir(dir, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) child := filepath.Join(dir, "r\u00e9sum\u00e9.txt") if err := os.WriteFile(child, []byte("hi"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, child}}) }) } func TestSubscribeDirRename(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f1 := subPath(dir) if err := os.Mkdir(f1, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) f2 := subPath(dir) if err := os.Rename(f1, f2); err != nil { t.Fatal(err) } expectEventSet(t, r, []wantEvent{ {EventDelete, f1}, {EventUpdate, f2}, }) }) } func TestSubscribeDirDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.Mkdir(f, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.RemoveAll(f); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f}}) }) } func TestSubscribeWatchedDirDeleted(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) if err := os.RemoveAll(dir); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, dir}}) // Give the backend a moment to surface ErrWatchTerminated alongside // the delete; some backends batch the error into a later debounce // tick than the event itself. deadline := time.Now().Add(r.deadline()) for time.Now().Before(deadline) { r.mu.Lock() n := len(r.errs) r.mu.Unlock() if n > 0 { break } time.Sleep(20 * time.Millisecond) } r.mu.Lock() errs := slices.Clone(r.errs) r.errs = nil r.mu.Unlock() sawTerminated := false for _, e := range errs { if errors.Is(e, ErrWatchTerminated) { sawTerminated = true break } } if !sawTerminated { t.Fatalf("expected ErrWatchTerminated after watched dir delete, got errs=%v", errs) } // Re-create; should not emit events for a now-stale watch. if err := os.MkdirAll(dir, 0o755); err != nil { t.Fatal(err) } extra := r.drainQuiet(200 * time.Millisecond) if len(extra) != 0 { t.Fatalf("expected no follow-up events, got %v", extra) } }) } // ----- sub-files --------------------------------------------------------- func TestSubscribeSubfileCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, sub) // Wait for the inotify watcherImpl to finish setting up the watch on // the new dir before mutating it. time.Sleep(100 * time.Millisecond) f := subPath(sub) if err := os.WriteFile(f, []byte("hi"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeSubfileUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(sub) // WatchDirectory-then-create so the create event populates the // watcherImpl's tree before the modify arrives. if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) if err := os.WriteFile(f, []byte("v2-longer"), 0o644); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, f) }) } func TestSubscribeSubfileRename(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } f1 := subPath(sub) if err := os.WriteFile(f1, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) f2 := subPath(sub) if err := os.Rename(f1, f2); err != nil { t.Fatal(err) } // Wait for both events to arrive before checking the set. want := []wantEvent{{EventDelete, f1}, {EventUpdate, f2}} got := r.waitForAll(r.deadline(), want) filtered := filterEventsForPaths(got, f1, f2) assertEventSet(t, filtered, want) }) } func TestSubscribeSubfileDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } f := subPath(sub) if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Remove(f); err != nil { t.Fatal(err) } want := []wantEvent{{EventDelete, f}} got := r.waitForAll(r.deadline(), want) filtered := filterEventsForPaths(got, f) assertEventSequence(t, filtered, want) }) } // ----- sub-directories --------------------------------------------------- func TestSubscribeSubdirCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) nested := subPath(sub) if err := os.Mkdir(nested, 0o755); err != nil { t.Fatal(err) } want := []wantEvent{{EventUpdate, nested}} got := r.waitForAll(r.deadline(), want) filtered := filterEventsForPaths(got, nested) assertEventSequence(t, filtered, want) }) } func TestSubscribeSubdirDeleteWithFiles(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) subDir := subPath(dir) if err := os.Mkdir(subDir, 0o755); err != nil { t.Fatal(err) } child := subPath(subDir) if err := os.WriteFile(child, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.RemoveAll(subDir); err != nil { t.Fatal(err) } expectEventSet(t, r, []wantEvent{ {EventDelete, subDir}, {EventDelete, child}, }) }) } // ----- symlinks ---------------------------------------------------------- func TestSubscribeSymlinkCreate(t *testing.T) { t.Parallel() if runtime.GOOS == "dragonfly" { t.Skip("DragonFlyBSD kqueue doesn't fire NOTE_WRITE on symlink creation") } runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f1 := subPath(dir) if err := os.WriteFile(f1, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) f2 := subPath(dir) if err := os.Symlink(f1, f2); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f2}}) }) } func TestSubscribeSymlinkDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f1 := subPath(dir) f2 := subPath(dir) if err := os.WriteFile(f1, []byte("x"), 0o644); err != nil { t.Fatal(err) } if err := os.Symlink(f1, f2); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Remove(f2); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f2}}) }) } func TestSubscribeSymlinkedDirectoryRebasesTargetEvents(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) target := filepath.Join(dir, "target") if err := os.Mkdir(target, 0o755); err != nil { t.Fatal(err) } link := filepath.Join(dir, "link") makeDirSymlink(t, target, link) r, _ := subscribeFor(t, link, watcherImpl) targetChild := filepath.Join(target, "child") if err := os.WriteFile(targetChild, []byte("x"), 0o644); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, filepath.Join(link, "child")) }) } func TestRecursiveSubscribeSymlinkedDirectoryDoesNotFollowDescendantSymlink(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { if watcherImpl.HasFastRecursiveBackend() { t.Skip("fast recursive backends do not use the userspace recursive walk") } dir := newTmpDir(t) target := filepath.Join(dir, "target") if err := os.Mkdir(target, 0o755); err != nil { t.Fatal(err) } link := filepath.Join(dir, "link") makeDirSymlink(t, target, link) descendantTarget := filepath.Join(dir, "descendant-target") if err := os.Mkdir(descendantTarget, 0o755); err != nil { t.Fatal(err) } descendantLink := filepath.Join(target, "descendant-link") makeDirSymlink(t, descendantTarget, descendantLink) r, _ := subscribeFor(t, link, watcherImpl) logicalGrandchild := filepath.Join(link, "descendant-link", "grandchild") physicalGrandchild := filepath.Join(descendantTarget, "grandchild") if err := os.WriteFile(logicalGrandchild, []byte("x"), 0o644); err != nil { t.Fatal(err) } marker := filepath.Join(target, "marker") if err := os.WriteFile(marker, []byte("flush"), 0o644); err != nil { t.Fatal(err) } got := expectContains(t, r, EventUpdate, filepath.Join(link, "marker")) got = append(got, r.drainQuiet(2*maxWaitTime)...) assertNoEventsForPath(t, got, logicalGrandchild, "expected no events through descendant symlink") assertNoEventsForPath(t, got, physicalGrandchild, "expected no events for descendant symlink target") }) } // ----- event coalescing -------------------------------------------------- func TestSubscribeCoalesceCreateUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } if err := os.WriteFile(f, []byte("v2"), 0o644); err != nil { t.Fatal(err) } // The two writes should net to one update. Under host load the // debounce may split them across batches, so check the coalesced // effect via replayEventList rather than insisting on a single // delivered event. got := r.gatherUntilQuiet(r.deadline(), 3*maxWaitTime) net := replayEventList(filterEventsForPaths(got, f)) assertEventSet(t, net, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeCoalesceDeleteCreateAsUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) if err := os.Remove(f); err != nil { t.Fatal(err) } if err := os.WriteFile(f, []byte("v2"), 0o644); err != nil { t.Fatal(err) } // Net: delete+create coalesces to update. got := r.gatherUntilQuiet(r.deadline(), 3*maxWaitTime) net := replayEventList(filterEventsForPaths(got, f)) assertEventSet(t, net, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeCoalesceCreateThenDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f1 := subPath(dir) f2 := subPath(dir) if err := os.WriteFile(f1, []byte("x"), 0o644); err != nil { t.Fatal(err) } if err := os.WriteFile(f2, []byte("x"), 0o644); err != nil { t.Fatal(err) } if err := os.Remove(f2); err != nil { t.Fatal(err) } // Whether all three operations land in one debounce batch (perfect // coalescing → just [update f1]) or split across batches (we may // see [update f1] + [update f2] + [delete f2]) depends on kernel // timing. Either is correct as long as the *net effect*, // replaying the events through eventList, leaves only [update f1]. // Quiet window must exceed the debouncer's maxWaitTime so a delayed // follow-up batch doesn't get cut off by the gatherUntilQuiet timer. // Use 3× maxWaitTime to leave headroom for -race overhead. got := r.gatherUntilQuiet(r.deadline(), 3*maxWaitTime) net := replayEventList(got) assertEventSet(t, net, []wantEvent{{EventUpdate, f1}}) }) } func TestSubscribeCoalesceMultipleUpdates(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) // consume initial update for _, v := range []string{"v2", "v3", "v4"} { if err := os.WriteFile(f, []byte(v), 0o644); err != nil { t.Fatal(err) } } got := r.gatherUntilQuiet(r.deadline(), 3*maxWaitTime) net := replayEventList(filterEventsForPaths(got, f)) assertEventSet(t, net, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeCoalesceUpdateDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) f := subPath(dir) // Upstream's debouncer (by design) fires the first event in a quiet // window immediately. To exercise the coalescing path, we create // the file post-subscribe and consume that initial event so the // debouncer's lastTime is recent before the update+delete pair. if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) if err := os.WriteFile(f, []byte("v2"), 0o644); err != nil { t.Fatal(err) } if err := os.Remove(f); err != nil { t.Fatal(err) } got := r.gatherUntilQuiet(r.deadline(), 3*maxWaitTime) net := replayEventList(filterEventsForPaths(got, f)) assertEventSet(t, net, []wantEvent{{EventDelete, f}}) }) } // ----- multiple subscriptions -------------------------------------------- func TestSubscribeMultipleSameDir(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) // Let fseventsd register the freshly-created tmpDir before we // subscribe; otherwise the dir's own creation can appear in the // initial event batch on macOS. time.Sleep(50 * time.Millisecond) r1 := newRecorder(t) s1, err := watcherImpl.WatchDirectory(dir, r1.callback) if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = s1.Close() }) r2 := newRecorder(t) s2, err := watcherImpl.WatchDirectory(dir, r2.callback) if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = s2.Close() }) time.Sleep(100 * time.Millisecond) f := subPath(dir) if err := os.WriteFile(f, []byte("hi"), 0o644); err != nil { t.Fatal(err) } assertEventSequence(t, r1.next(r1.deadline()), []wantEvent{{EventUpdate, f}}) assertEventSequence(t, r2.next(r2.deadline()), []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeMultipleDifferentDirs(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir1 := newTmpDir(t) dir2 := newTmpDir(t) r1, _ := subscribeFor(t, dir1, watcherImpl) r2, _ := subscribeFor(t, dir2, watcherImpl) f1 := subPath(dir1) f2 := subPath(dir2) if err := os.WriteFile(f1, []byte("a"), 0o644); err != nil { t.Fatal(err) } if err := os.WriteFile(f2, []byte("b"), 0o644); err != nil { t.Fatal(err) } assertEventSequence(t, r1.next(r1.deadline()), []wantEvent{{EventUpdate, f1}}) assertEventSequence(t, r2.next(r2.deadline()), []wantEvent{{EventUpdate, f2}}) }) } func TestWatchDirectoriesBatch(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir1 := newTmpDir(t) dir2 := newTmpDir(t) r1 := newRecorder(t) r1.watcher = watcherImpl r2 := newRecorder(t) r2.watcher = watcherImpl watches, err := watcherImpl.WatchDirectories([]WatchDirectoryRequest{ {Dir: dir1, Callback: r1.callback, Options: []WatchOption{WithRecursive()}}, {Dir: dir2, Callback: r2.callback, Options: []WatchOption{WithRecursive()}}, }) if err != nil { t.Fatal(err) } t.Cleanup(func() { for _, watch := range watches { _ = watch.Close() } }) time.Sleep(settleSleep(watcherImpl)) f1 := subPath(dir1) f2 := subPath(dir2) if err := os.WriteFile(f1, []byte("a"), 0o644); err != nil { t.Fatal(err) } if err := os.WriteFile(f2, []byte("b"), 0o644); err != nil { t.Fatal(err) } assertEventSequence(t, r1.next(r1.deadline()), []wantEvent{{EventUpdate, f1}}) assertEventSequence(t, r2.next(r2.deadline()), []wantEvent{{EventUpdate, f2}}) }) } type countingWatcherImpl struct { watcherBase subscribed []*dirWatch closed []*dirWatch } func newCountingWatcherImpl() *countingWatcherImpl { impl := &countingWatcherImpl{} impl.watcherBase.init(impl) return impl } func (b *countingWatcherImpl) start() error { b.notifyStarted() return nil } func (b *countingWatcherImpl) subscribe(w *dirWatch) error { b.subscribed = append(b.subscribed, w) return nil } func (b *countingWatcherImpl) closeWatch(w *dirWatch) error { b.closed = append(b.closed, w) return nil } func TestFastRecursiveWatcherConsolidatesSiblingDirectories(t *testing.T) { t.Parallel() root := t.TempDir() parent := filepath.Join(root, "node_modules", ".bun") if err := os.MkdirAll(parent, 0o755); err != nil { t.Fatal(err) } var impl *countingWatcherImpl watcherImpl := &watcher{ name: "fsevents", factory: func() watcherImpl { impl = newCountingWatcherImpl() return impl }, } var subs []Watch for i := range recursiveConsolidateThreshold + 2 { dir := filepath.Join(parent, fmt.Sprintf("pkg%d", i)) if err := os.MkdirAll(dir, 0o755); err != nil { t.Fatal(err) } sub, err := watcherImpl.WatchDirectory(dir, func([]Event, error) {}) if err != nil { t.Fatal(err) } subs = append(subs, sub) } t.Cleanup(func() { for _, sub := range subs { _ = sub.Close() } }) if got := len(impl.subscribed); got != recursiveConsolidateThreshold { t.Fatalf("expected %d subscriptions after consolidation, got %d", recursiveConsolidateThreshold, got) } consolidated := impl.subscribed[len(impl.subscribed)-1] if consolidated.dir != parent || !consolidated.recursive { t.Fatalf("expected consolidated recursive watch on %s, got dir=%s recursive=%v", parent, consolidated.dir, consolidated.recursive) } watcherImpl.mu.Lock() _, hasPkgWatch := watcherImpl.dirWatches[watcherImpl.keyForDirWatch(filepath.Join(parent, "pkg11"), false)] watcherImpl.mu.Unlock() if hasPkgWatch { t.Fatal("expected later package watch to reuse consolidated parent instead of creating its own stream") } } func TestFastRecursiveWatcherDoesNotConsolidateSymlinkOutsideRoot(t *testing.T) { t.Parallel() root := t.TempDir() parent := filepath.Join(root, "node_modules", ".bun") if err := os.MkdirAll(parent, 0o755); err != nil { t.Fatal(err) } var impl *countingWatcherImpl watcherImpl := &watcher{ name: "fsevents", factory: func() watcherImpl { impl = newCountingWatcherImpl() return impl }, } var subs []Watch for i := range recursiveConsolidateThreshold { dir := filepath.Join(parent, fmt.Sprintf("pkg%d", i)) if err := os.MkdirAll(dir, 0o755); err != nil { t.Fatal(err) } sub, err := watcherImpl.WatchDirectory(dir, func([]Event, error) {}) if err != nil { t.Fatal(err) } subs = append(subs, sub) } t.Cleanup(func() { for _, sub := range subs { _ = sub.Close() } }) consolidated := impl.subscribed[len(impl.subscribed)-1] if consolidated.dir != parent || !consolidated.recursive { t.Fatalf("expected consolidated recursive watch on %s, got dir=%s recursive=%v", parent, consolidated.dir, consolidated.recursive) } target := filepath.Join(root, "outside") if err := os.MkdirAll(target, 0o755); err != nil { t.Fatal(err) } link := filepath.Join(parent, "linked") makeDirSymlink(t, target, link) sub, err := watcherImpl.WatchDirectory(link, func([]Event, error) {}) if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = sub.Close() }) watch := sub.(*watch) if watch.dw == consolidated { t.Fatal("symlink outside consolidated physical root should keep its own watch") } if watch.dw.dir != link || watch.dw.physicalDir != physicalDirFor(link) { t.Fatalf("expected watch on symlink root %s (%s), got %s (%s)", link, physicalDirFor(link), watch.dw.dir, watch.dw.physicalDir) } } func TestConsolidatedSymlinkChildMapsSharedLogicalPath(t *testing.T) { t.Parallel() root := t.TempDir() physicalParent := filepath.Join(root, "physical-parent") if err := os.MkdirAll(filepath.Join(physicalParent, "target"), 0o755); err != nil { t.Fatal(err) } logicalParent := filepath.Join(root, "logical-parent") makeDirSymlink(t, physicalParent, logicalParent) link := filepath.Join(logicalParent, "link") makeDirSymlink(t, filepath.Join(physicalParent, "target"), link) cb := callback{ dir: link, physicalDir: physicalDirFor(link), watchDir: logicalParent, watchPhysicalDir: physicalDirFor(logicalParent), } event := Event{Kind: EventUpdate, Path: filepath.Join(logicalParent, "target", "file.ts")} got := cb.mapEvent(event) want := filepath.Join(link, "file.ts") if got.Path != want { t.Fatalf("mapEvent path = %q, want %q", got.Path, want) } } func TestConsolidatedSymlinkChildTerminatesFromSharedLogicalPath(t *testing.T) { t.Parallel() root := t.TempDir() physicalParent := filepath.Join(root, "physical-parent") if err := os.MkdirAll(filepath.Join(physicalParent, "target"), 0o755); err != nil { t.Fatal(err) } logicalParent := filepath.Join(root, "logical-parent") makeDirSymlink(t, physicalParent, logicalParent) link := filepath.Join(logicalParent, "link") makeDirSymlink(t, filepath.Join(physicalParent, "target"), link) dw := newDirectWatcher(t, logicalParent) dw.physicalDir = physicalDirFor(logicalParent) id, _ := dw.watch(link, physicalDirFor(link), true, func([]Event, error) {}, nil) err := errors.New("terminated") if !dw.terminateCallbacksForDeletedRoot(filepath.Join(logicalParent, "target"), 1, err) { t.Fatal("expected symlink child callback to terminate") } dw.mu.Lock() defer dw.mu.Unlock() for _, cb := range dw.callbacks { if cb.id == id && !errors.Is(cb.terminal, err) { t.Fatalf("terminal error = %v, want %v", cb.terminal, err) } } } func TestConsolidatedChildWatchFiltersAgainstRequestedDir(t *testing.T) { t.Parallel() parent := filepath.Join(t.TempDir(), "parent") child := filepath.Join(parent, "child") sibling := filepath.Join(parent, "sibling") dw := newDirectWatcher(t, parent) var got []Event dw.watch(child, child, false, func(events []Event, err error) { if err != nil { t.Fatal(err) } got = append(got, events...) }, nil) dw.events.updateWatchRootAt(child, 1) dw.events.update(filepath.Join(child, "file.ts")) dw.events.update(filepath.Join(child, "nested", "file.ts")) dw.events.update(filepath.Join(sibling, "file.ts")) dw.triggerCallbacks() gotW := toWantEvents(got) want := []wantEvent{ {EventUpdate, child}, {EventUpdate, filepath.Join(child, "file.ts")}, } cmpEvents := func(a, b wantEvent) int { if a.Kind != b.Kind { return cmp.Compare(a.Kind, b.Kind) } return cmp.Compare(a.Path, b.Path) } slices.SortFunc(gotW, cmpEvents) slices.SortFunc(want, cmpEvents) if !equalWantEvents(gotW, want) { t.Fatalf("event mismatch\nwant: %v\n got: %v", want, gotW) } } func TestConsolidatedChildWatchIgnoresEventsBeforeSubscribe(t *testing.T) { t.Parallel() parent := filepath.Join(t.TempDir(), "parent") child := filepath.Join(parent, "child") dw := newDirectWatcher(t, parent) dw.events.update(child) var got []Event dw.watch(child, child, true, func(events []Event, err error) { if err != nil { t.Fatal(err) } got = append(got, events...) }, nil) dw.events.remove(child) dw.triggerCallbacks() assertEventSequence(t, got, []wantEvent{{EventDelete, child}}) } func TestRecursiveWatchWithIgnoreDoesNotFilterByLogicalRoot(t *testing.T) { t.Parallel() dir := filepath.Join(t.TempDir(), "root") dw := newDirectWatcher(t, dir) var got []Event outsidePath := filepath.Join(t.TempDir(), "outside", "pkg", "index.ts") dw.watch(dir, dir, true, func(events []Event, err error) { if err != nil { t.Fatal(err) } got = append(got, events...) }, func(string) bool { return false }) dw.events.update(outsidePath) dw.triggerCallbacks() assertEventSequence(t, got, []wantEvent{{EventUpdate, outsidePath}}) } // ----- errors ------------------------------------------------------------ func TestSubscribeMissingDirError(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { bogus := filepath.Join(newTmpDir(t), "definitely-not-here") _, err := watcherImpl.WatchDirectory(bogus, func([]Event, error) {}) if err == nil { t.Fatal("expected error subscribing to non-existent dir") } }) } func TestSubscribeNotADirError(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } _, err := watcherImpl.WatchDirectory(f, func([]Event, error) {}) if err == nil { t.Fatal("expected error subscribing to a file") } }) } func TestSubscribeRejectsNilCallback(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { if _, err := watcherImpl.WatchDirectory(t.TempDir(), nil); err == nil { t.Fatal("WatchDirectory(nil callback) should return an error") } }) } func TestSubscribeRejectsRelativePath(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { _, err := watcherImpl.WatchDirectory("relative/path", func([]Event, error) {}) if err == nil { t.Fatal("WatchDirectory with relative path should return an error") } _, err = watcherImpl.WatchFile("relative/path/file.txt", func([]Event, error) {}) if err == nil { t.Fatal("WatchFile with relative path should return an error") } }) } // ----- watch lifecycle -------------------------------------------- func TestSubscribeUnsubscribeIdempotent(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r := newRecorder(t) sub, err := watcherImpl.WatchDirectory(dir, r.callback) if err != nil { t.Fatal(err) } if err := sub.Close(); err != nil { t.Fatal(err) } if err := sub.Close(); err != nil { t.Fatalf("second Close should be a no-op, got %v", err) } }) } // TestSubscribeCloseThenReSubscribe verifies that Close fully tears down // any kernel-side resources before returning, so a follow-on subscribe // on the same path immediately afterwards observes events from a fresh // watch instead of getting stuck on a stale handle / fd / mark. // // Before Q7's fix on Windows, Close returned while the per-watch // goroutine was still completing GetOverlappedResult and the directory // handle remained open. A test (or a real program) racing to delete the // watched dir or to install a different watcher could see flakes. func TestSubscribeCloseThenReSubscribe(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r1 := newRecorder(t) s1, err := watcherImpl.WatchDirectory(dir, r1.callback) if err != nil { t.Fatal(err) } if err = s1.Close(); err != nil { t.Fatal(err) } // Immediately re-watch the same directory and verify a fresh // event still flows. Use a separate recorder so we know the // event isn't a leftover from the first watch. r2 := newRecorder(t) s2, err := watcherImpl.WatchDirectory(dir, r2.callback) if err != nil { t.Fatalf("re-WatchDirectory after Close: %v", err) } t.Cleanup(func() { _ = s2.Close() }) // Give the second watcher a moment to settle (fsevents/kqueue // need it; inotify/fanotify/Windows don't but the wait is cheap). if watcherImpl == FSEvents() || watcherImpl == Kqueue() { time.Sleep(300 * time.Millisecond) } else { time.Sleep(60 * time.Millisecond) } f := subPath(dir) if err := os.WriteFile(f, []byte("hi"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r2, []wantEvent{{EventUpdate, f}}) // The first recorder must not have seen the event meant for r2. stale := r1.drainQuiet(50 * time.Millisecond) if len(stale) != 0 { t.Fatalf("closed watch saw events: %v", toWantEvents(stale)) } }) } func TestSubscribeNoGoroutineLeak(t *testing.T) { //nolint:paralleltest // goroutine counting requires sequential execution // No t.Parallel(): goroutine counting requires sequential execution. for _, b := range availableWatchers { //nolint:paralleltest // goroutine counting requires sequential execution t.Run(b.Name(), func(t *testing.T) { dir := newTmpDir(t) // Warm up: trigger any lazy singleton init (backend, // debouncer) so it doesn't inflate the post-loop count. warmup, err := b.WatchDirectory(dir, func([]Event, error) {}) if err != nil { t.Fatal(err) } if err := warmup.Close(); err != nil { t.Fatal(err) } runtime.GC() time.Sleep(100 * time.Millisecond) baseline := runtime.NumGoroutine() for range 8 { r := newRecorder(t) sub, err := b.WatchDirectory(dir, r.callback) if err != nil { t.Fatal(err) } if err := sub.Close(); err != nil { t.Fatal(err) } } // Allow lazy backend/debounce shutdown to settle. deadline := time.Now().Add(2 * time.Second) for time.Now().Before(deadline) { runtime.GC() if runtime.NumGoroutine() <= baseline+2 { return } time.Sleep(50 * time.Millisecond) } t.Fatalf("goroutine leak: baseline=%d now=%d", baseline, runtime.NumGoroutine()) }) } } // ----- additional coverage ----------------------------------------------- func TestSubscribeDeepNestedCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) // Create a/b/c one level at a time so the watcher can keep up. a := filepath.Join(dir, "a") b := filepath.Join(a, "b") c := filepath.Join(b, "c") for _, d := range []string{a, b, c} { if err := os.Mkdir(d, 0o755); err != nil { t.Fatal(err) } time.Sleep(150 * time.Millisecond) } f := filepath.Join(c, "deep.txt") if err := os.WriteFile(f, []byte("deep"), 0o644); err != nil { t.Fatal(err) } want := []wantEvent{{EventUpdate, a}, {EventUpdate, f}} got := r.waitForAll(r.deadline(), want) for _, w := range want { if !containsEvent(got, w.Kind, w.Path) { t.Fatalf("expected %s for %s, got %v", w.Kind, w.Path, toWantEvents(got)) } } }) } func TestSubscribeManyFilesAtOnce(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeFor(t, dir, watcherImpl) const count = 50 paths := make([]string, count) for i := range count { paths[i] = subPath(dir) if err := os.WriteFile(paths[i], []byte("x"), 0o644); err != nil { t.Fatal(err) } } want := make([]wantEvent, count) for i, p := range paths { want[i] = wantEvent{EventUpdate, p} } // Some kqueue kernels coalesce dir-NOTE_WRITE events under // load and miss a few. Retry the missing files (a fresh write // provokes a new NOTE_WRITE on the parent) up to a couple of // times before declaring failure. got := r.waitForAll(r.deadline(), want) for attempt := 0; attempt < 3 && !haveAll(got, want); attempt++ { for _, p := range paths { if !containsEvent(got, EventUpdate, p) { _ = os.WriteFile(p, []byte("x"), 0o644) } } more := r.waitForAll(r.deadline(), want) got = append(got, more...) } for _, p := range paths { if !containsEvent(got, EventUpdate, p) { t.Fatalf("missing create for %s (got %d events total)", p, len(got)) } } }) } func TestSubscribeTruncateFile(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.WriteFile(f, []byte("hello world"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Truncate(f, 0); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeConcurrentSubscribeUnsubscribe(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) done := make(chan struct{}) for range 8 { go func() { defer func() { done <- struct{}{} }() rec := newRecorder(t) sub, err := watcherImpl.WatchDirectory(dir, rec.callback) if err != nil { return } _ = sub.Close() }() } for range 8 { <-done } }) } func TestSubscribeRenameDir(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "before") if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } child := filepath.Join(sub, "file.txt") if err := os.WriteFile(child, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) after := filepath.Join(dir, "after") if err := os.Rename(sub, after); err != nil { t.Fatal(err) } want := []wantEvent{{EventUpdate, after}, {EventDelete, sub}} got := r.waitForAll(r.deadline(), want) for _, w := range want { if !containsEvent(got, w.Kind, w.Path) { t.Fatalf("expected %s for %s, got %v", w.Kind, w.Path, toWantEvents(got)) } } }) } func TestSubscribeReplaceFileWithDir(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) target := subPath(dir) if err := os.WriteFile(target, []byte("file"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Remove(target); err != nil { t.Fatal(err) } if err := os.Mkdir(target, 0o755); err != nil { t.Fatal(err) } // Should see at least one event for target (delete and/or update). got := r.waitForEvent(r.deadline(), func(e Event) bool { return e.Path == target }) if !containsEvent(got, EventDelete, target) && !containsEvent(got, EventUpdate, target) { t.Fatalf("expected events for file-to-dir replacement, got %v", toWantEvents(got)) } }) } func TestSubscribeAppendToFile(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.WriteFile(f, []byte("initial"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) fh, err := os.OpenFile(f, os.O_APPEND|os.O_WRONLY, 0) if err != nil { t.Fatal(err) } _, _ = fh.WriteString(" appended") fh.Close() expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestSubscribeNoEventsAfterUnsubscribe(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, sub := subscribeFor(t, dir, watcherImpl) if err := sub.Close(); err != nil { t.Fatal(err) } // Create a file after closeWatch; should produce nothing. f := subPath(dir) if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } got := r.drainQuiet(500 * time.Millisecond) if len(got) != 0 { t.Fatalf("expected no events after closeWatch, got %v", toWantEvents(got)) } }) } // ----- watcherBase / dirWatchError internals ----------------------------- type failingBackend struct { watcherBase err error } func newFailingBackend(err error) *failingBackend { b := &failingBackend{err: err} b.watcherBase.init(b) return b } func (b *failingBackend) start() error { return b.err } func (b *failingBackend) subscribe(*dirWatch) error { return nil } func (b *failingBackend) closeWatch(*dirWatch) error { return nil } func TestBackendRunReturnsStartError(t *testing.T) { t.Parallel() want := errors.New("startup failed") b := newFailingBackend(want) if err := b.run(); !errors.Is(err, want) { t.Fatalf("run() error = %v, want %v", err, want) } } func TestDirWatchErrorImplementsError(t *testing.T) { t.Parallel() var err error = &dirWatchError{err: errors.New("boom")} if err.Error() != "boom" { t.Fatalf("dirWatchError.Error want boom, got %q", err.Error()) } } func TestFileCallbackForwardsErrAlongsideEvents(t *testing.T) { t.Parallel() target := "/abs/dir/target.txt" other := "/abs/dir/sibling.txt" overflow := errors.New("overflow") type call struct { events []Event err error } var got []call cb := fileCallback(target, func(events []Event, err error) { got = append(got, call{events: events, err: err}) }) // Plain events: only target events pass through, sibling dropped. cb([]Event{{Kind: EventUpdate, Path: target}, {Kind: EventUpdate, Path: other}}, nil) if len(got) != 1 || len(got[0].events) != 1 || got[0].events[0].Path != target || got[0].err != nil { t.Fatalf("plain delivery: got %+v", got) } // Err only, no matching events: still forwarded with empty slice. got = nil cb([]Event{{Kind: EventUpdate, Path: other}}, overflow) if len(got) != 1 || len(got[0].events) != 0 || !errors.Is(got[0].err, overflow) { t.Fatalf("err-only delivery: got %+v", got) } // Err with matching events: deliver both the filtered events and err. got = nil cb([]Event{{Kind: EventDelete, Path: target}, {Kind: EventUpdate, Path: other}}, overflow) if len(got) != 1 || len(got[0].events) != 1 || got[0].events[0].Path != target || got[0].events[0].Kind != EventDelete || !errors.Is(got[0].err, overflow) { t.Fatalf("combined delivery: got %+v", got) } // No events, no err: callback not invoked at all. got = nil cb(nil, nil) if len(got) != 0 { t.Fatalf("no-op delivery: got %+v", got) } } // TestRenameDirOutOfTreeNoStaleEvents pins the cross-backend contract: // once a subdirectory is renamed out of the watched root, modifications // to files at its new location must not surface against the old paths. // // This passes today on every backend even without B6's fanotify fix: // the FAN_RENAME path (kernel >= 5.17) already handles descendant // cleanup correctly, and inotify/kqueue/fsevents/Windows track watches // at a level where the moved subtree drops out naturally. The harder // case (forced FAN_MOVED_FROM fallback) is in // TestFanotifyNoRenameFallback/RenameDirOutDropsDescendants. func TestRenameDirOutOfTreeNoStaleEvents(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { watched := newTmpDir(t) outside := newTmpDir(t) // separate watch root, NOT watched. // Build a nested subtree under sub/. The descendant subdirs are // the ones that exercise the bug: with the broken fanotify // handleSubscription, sub itself got cleaned (exact-match) but // sub/inner stayed in b.subscriptions with a stale path. Later // modifications to files at the new location of sub/inner would // then surface against the old (now-invalid) path. sub := filepath.Join(watched, "sub") inner := filepath.Join(sub, "inner") if err := os.MkdirAll(inner, 0o755); err != nil { t.Fatal(err) } nested := filepath.Join(inner, "leaf.txt") if err := os.WriteFile(nested, []byte("v1"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeForOpts(t, watched, watcherImpl, WithRecursive()) // Rename the whole subtree out of the watched dir. dest := filepath.Join(outside, "moved") if err := os.Rename(sub, dest); err != nil { t.Fatal(err) } // Consume the rename-away events. _ = r.drainQuiet(500 * time.Millisecond) // Modify the file at its new location. movedNested := filepath.Join(dest, "inner", "leaf.txt") if err := os.WriteFile(movedNested, []byte("v2-longer"), 0o644); err != nil { t.Fatal(err) } extra := r.drainQuiet(800 * time.Millisecond) // Allow events for unrelated parts of the watched dir, but no // event whose path is at or under the moved subtree may appear. oldPrefix := sub + string(filepath.Separator) for _, e := range extra { if e.Path == sub || strings.HasPrefix(e.Path, oldPrefix) { t.Fatalf("stale event for moved-out path %s: %+v\nall extras: %v", e.Path, e, toWantEvents(extra)) } } }) } // ----- platform-specific ------------------------------------------------- func TestDefaultBackendMatchesPlatform(t *testing.T) { t.Parallel() d := Default() var wantName string switch runtime.GOOS { case "linux": if Fanotify().Available() { wantName = "fanotify" } else { wantName = "inotify" } case "darwin": wantName = "fsevents" case "windows": wantName = "windows" case "freebsd", "openbsd", "netbsd", "dragonfly": wantName = "kqueue" default: t.Skipf("no expected default watcher for %s", runtime.GOOS) } if !d.Available() { t.Fatalf("Default() should be available on %s", runtime.GOOS) } if d.Name() != wantName { t.Fatalf("Default().Name() = %q, want %q", d.Name(), wantName) } } func TestUnavailableBackendReturnsError(t *testing.T) { t.Parallel() // Pick a watcher that is definitely unavailable on the current OS. var unavailable Watcher for _, w := range AllWatchers() { if !w.Available() { unavailable = w break } } if unavailable == nil { t.Skip("all watchers are available on this platform") } dir := newTmpDir(t) _, err := unavailable.WatchDirectory(dir, func([]Event, error) {}) if !errors.Is(err, ErrUnavailable) { t.Fatalf("expected ErrUnavailable from %s, got %v", unavailable.Name(), err) } } func TestSubscribeNestedDirDeletionCleansDescendants(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "parent") nested := filepath.Join(sub, "child") if err := os.MkdirAll(nested, 0o755); err != nil { t.Fatal(err) } childFile := filepath.Join(nested, "file.txt") if err := os.WriteFile(childFile, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.RemoveAll(sub); err != nil { t.Fatal(err) } expectContains(t, r, EventDelete, sub) }) } // ----- non-recursive tests ----------------------------------------------- func TestNonRecursiveFileCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeForOpts(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestNonRecursiveFileUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeForOpts(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } _ = r.waitForEvent(r.deadline(), func(Event) bool { return true }) // consume create if err := os.WriteFile(f, []byte("v2-longer"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestNonRecursiveFileDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := subPath(dir) if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeForOpts(t, dir, watcherImpl) if err := os.Remove(f); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f}}) }) } func TestNonRecursiveDirCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeForOpts(t, dir, watcherImpl) sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, sub) }) } func TestNonRecursiveGrandchildIgnored(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "child") if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeForOpts(t, dir, watcherImpl) // Create a file inside the child directory (grandchild). grandchild := subPath(sub) if err := os.WriteFile(grandchild, []byte("deep"), 0o644); err != nil { t.Fatal(err) } marker := subPath(dir) if err := os.WriteFile(marker, []byte("flush"), 0o644); err != nil { t.Fatal(err) } // The marker proves the non-recursive watcher processed a later // direct-child batch. It still must not report the grandchild. got := expectContains(t, r, EventUpdate, marker) got = append(got, r.drainQuiet(2*maxWaitTime)...) assertNoEventsForPath(t, got, grandchild, "expected no events for grandchild") }) } func TestNonRecursiveNewSubdirContentIgnored(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) r, _ := subscribeForOpts(t, dir, watcherImpl) // Create a new subdirectory. sub := subPath(dir) if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } // Wait for the dir create event. expectContains(t, r, EventUpdate, sub) // Write a file inside the new subdirectory. grandchild := subPath(sub) if err := os.WriteFile(grandchild, []byte("nested"), 0o644); err != nil { t.Fatal(err) } marker := subPath(dir) if err := os.WriteFile(marker, []byte("flush"), 0o644); err != nil { t.Fatal(err) } // Should NOT see the grandchild event. got := expectContains(t, r, EventUpdate, marker) got = append(got, r.drainQuiet(2*maxWaitTime)...) assertNoEventsForPath(t, got, grandchild, "expected no events for nested file") }) } func TestNonRecursiveAndRecursiveSameDir(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "child") if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } rNonRec, _ := subscribeForOpts(t, dir, watcherImpl) rRec, _ := subscribeForOpts(t, dir, watcherImpl, WithRecursive()) // Create a grandchild file. grandchild := subPath(sub) if err := os.WriteFile(grandchild, []byte("deep"), 0o644); err != nil { t.Fatal(err) } marker := subPath(dir) if err := os.WriteFile(marker, []byte("flush"), 0o644); err != nil { t.Fatal(err) } // Recursive should see the grandchild. expectContains(t, rRec, EventUpdate, grandchild) // Non-recursive should NOT see the grandchild. gotNonRec := expectContains(t, rNonRec, EventUpdate, marker) gotNonRec = append(gotNonRec, rNonRec.drainQuiet(2*maxWaitTime)...) assertNoEventsForPath(t, gotNonRec, grandchild, "non-recursive: expected no events for") }) } func TestNonRecursiveWithDeniedSubdir(t *testing.T) { t.Parallel() if runtime.GOOS == "windows" { t.Skip("chmod is not meaningful on Windows") } runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) // Create a permission-denied subdirectory. denied := filepath.Join(dir, "denied") if err := os.Mkdir(denied, 0o755); err != nil { t.Fatal(err) } if err := os.Chmod(denied, 0); err != nil { t.Fatal(err) } t.Cleanup(func() { _ = os.Chmod(denied, 0o700) }) // Non-recursive watch should succeed despite the inaccessible child. r, _ := subscribeForOpts(t, dir, watcherImpl) f := subPath(dir) if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } // ----- file watch tests -------------------------------------------------- func TestFileWatchCreate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := filepath.Join(dir, "target.txt") r, _ := subscribeFileFor(t, f, watcherImpl) if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestFileWatchUpdate(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := filepath.Join(dir, "target.txt") if err := os.WriteFile(f, []byte("v1"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFileFor(t, f, watcherImpl) if err := os.WriteFile(f, []byte("v2-longer"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } func TestFileWatchDelete(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := filepath.Join(dir, "target.txt") if err := os.WriteFile(f, []byte("x"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFileFor(t, f, watcherImpl) if err := os.Remove(f); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f}}) }) } func TestFileWatchIgnoresSiblings(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) target := filepath.Join(dir, "target.txt") sibling := filepath.Join(dir, "sibling.txt") r, _ := subscribeFileFor(t, target, watcherImpl) witness, _ := subscribeForOpts(t, dir, watcherImpl) // Write to sibling; should NOT see this. if err := os.WriteFile(sibling, []byte("noise"), 0o644); err != nil { t.Fatal(err) } expectContains(t, witness, EventUpdate, sibling) expectNoBufferedEvents(t, r, "expected no events for sibling") }) } // Not parallel: under load on macOS, this test (which subscribes // twice to files in the same directory via WatchFile) intermittently // stalls for the full FSEvents-timeout window. Running serially keeps // the multi-WatchFile-share path predictable. func TestFileWatchMultipleSameDir(t *testing.T) { //nolint:tparallel,paralleltest // see comment runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f1 := filepath.Join(dir, "a.txt") f2 := filepath.Join(dir, "b.txt") r1, _ := subscribeFileFor(t, f1, watcherImpl) r2, _ := subscribeFileFor(t, f2, watcherImpl) // Write to f1; only r1 should see it. if err := os.WriteFile(f1, []byte("hello"), 0o644); err != nil { t.Fatal(err) } got1 := r1.next(r1.deadline()) assertEventSequence(t, got1, []wantEvent{{EventUpdate, f1}}) expectNoBufferedEvents(t, r2, "r2 should not see f1 events") // Write to f2; only r2 should see it. if err := os.WriteFile(f2, []byte("world"), 0o644); err != nil { t.Fatal(err) } got2 := r2.next(r2.deadline()) assertEventSequence(t, got2, []wantEvent{{EventUpdate, f2}}) expectNoBufferedEvents(t, r1, "r1 should not see f2 events") }) } // Not parallel: under load on macOS, this test (delete then recreate // a file inside a WatchFile target) intermittently stalls for the full // FSEvents-timeout window. Running serially eliminates the flake. func TestFileWatchDeleteAndRecreate(t *testing.T) { //nolint:tparallel,paralleltest // see comment runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := filepath.Join(dir, "config.json") if err := os.WriteFile(f, []byte(`{"v":1}`), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFileFor(t, f, watcherImpl) // Delete the file. if err := os.Remove(f); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventDelete, f}}) // Recreate it. if err := os.WriteFile(f, []byte(`{"v":2}`), 0o644); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, f) }) } func TestFileWatchNonExistentTarget(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) f := filepath.Join(dir, "doesnotexist.txt") // File doesn't exist; subscribe should still succeed // (watches the parent dir). r, _ := subscribeFileFor(t, f, watcherImpl) // Now create it. if err := os.WriteFile(f, []byte("appeared"), 0o644); err != nil { t.Fatal(err) } expectEventSequence(t, r, []wantEvent{{EventUpdate, f}}) }) } // TestRecursiveMoveInPrePopulated verifies that moving a pre-populated // directory tree into a recursive watch detects changes in nested subdirs. func TestRecursiveMoveInPrePopulated(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) outside := newTmpDir(t) // Build a tree outside the watched directory. nested := filepath.Join(outside, "a", "b", "c") if err := os.MkdirAll(nested, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeForOpts(t, dir, watcherImpl, WithRecursive()) // Move the pre-populated tree into the watched directory. dest := filepath.Join(dir, "tree") if err := os.Rename(outside, dest); err != nil { t.Fatal(err) } // Consume the move-in events. _ = r.drainQuiet(500 * time.Millisecond) // Now modify a file deep inside the moved tree. There's a // race against the backend's recursive re-arm of the moved // subtree, so retry with fresh filenames until one surfaces // rather than betting on the first write being seen. nestedDir := filepath.Join(dest, "a", "b", "c") deadline := time.Now().Add(r.deadline()) var allSeen []Event for attempt := 0; time.Now().Before(deadline); attempt++ { f := filepath.Join(nestedDir, fmt.Sprintf("deep-%d.txt", attempt)) if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } more := r.waitForEvent(750*time.Millisecond, func(e Event) bool { return e.Kind == EventUpdate && strings.HasPrefix(e.Path, nestedDir+string(filepath.Separator)) }) allSeen = append(allSeen, more...) for _, e := range more { if e.Kind == EventUpdate && strings.HasPrefix(e.Path, nestedDir+string(filepath.Separator)) { return } } } t.Fatalf("expected update for a file inside moved-in tree (gave up after %s), got %v", r.deadline(), toWantEvents(allSeen)) }) } // TestAtomicSave verifies that the "safe save" pattern (write tmp, rename // over target) is detected as an update, not a delete+create or nothing. func TestAtomicSave(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) target := filepath.Join(dir, "config.json") if err := os.WriteFile(target, []byte(`{"v":1}`), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) // Atomic save: write to temp, rename over target. tmp := target + ".tmp" if err := os.WriteFile(tmp, []byte(`{"v":2}`), 0o644); err != nil { t.Fatal(err) } if err := os.Rename(tmp, target); err != nil { t.Fatal(err) } // Any event for target proves the atomic save was observed. got := r.waitForEvent(r.deadline(), func(e Event) bool { return e.Path == target }) got = filterEventsForPaths(got, target) if len(got) == 0 { t.Fatalf("expected events for %s after atomic save, got none", target) } }) } // TestAtomicSaveFileWatch verifies atomic save detection through WatchFile. func TestAtomicSaveFileWatch(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) target := filepath.Join(dir, "target.txt") if err := os.WriteFile(target, []byte("v1"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFileFor(t, target, watcherImpl) tmp := target + ".tmp" if err := os.WriteFile(tmp, []byte("v2"), 0o644); err != nil { t.Fatal(err) } if err := os.Rename(tmp, target); err != nil { t.Fatal(err) } got := r.waitForEvent(r.deadline(), func(e Event) bool { return e.Path == target }) got = filterEventsForPaths(got, target) if len(got) == 0 { t.Fatalf("expected events for %s after atomic save, got none", target) } }) } // TestReplaceDirWithFile verifies that replacing a directory with a file // of the same name emits appropriate events. func TestReplaceDirWithFile(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) child := filepath.Join(dir, "child") if err := os.Mkdir(child, 0o755); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) if err := os.Remove(child); err != nil { t.Fatal(err) } if err := os.WriteFile(child, []byte("now a file"), 0o644); err != nil { t.Fatal(err) } got := r.waitForEvent(r.deadline(), func(e Event) bool { return e.Path == child }) got = filterEventsForPaths(got, child) if len(got) == 0 { t.Fatalf("expected events for dir->file replacement at %s, got none", child) } }) } // TestRecreateSubdirAndModify verifies that after deleting and recreating // a subdirectory, changes inside it are still detected. func TestRecreateSubdirAndModify(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "sub") if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } inner := filepath.Join(sub, "file.txt") if err := os.WriteFile(inner, []byte("v1"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) // Delete the subdirectory tree. if err := os.RemoveAll(sub); err != nil { t.Fatal(err) } _ = r.drainQuiet(500 * time.Millisecond) // Recreate the same path. if err := os.Mkdir(sub, 0o755); err != nil { t.Fatal(err) } // Give the backend a chance to observe the FAN_CREATE / // IN_CREATE / compareDir on the parent and install its watch // on the new sub inode BEFORE we write inside it. Without // this beat the user's write can race the kernel's // enqueue-on-marked-inode and the event is never delivered. // The poll loop below then has nothing to wait for. time.Sleep(150 * time.Millisecond) // Keep nudging with a fresh file until one surfaces. Each // iteration uses a new filename so a missed event on attempt // N doesn't trap us waiting for it on attempt N+1. Short // per-attempt deadline + long total deadline = many retry // cycles, which is what kqueue needs when the kernel is slow // to deliver NOTE_WRITE on a freshly-watched sub inode. deadline := time.Now().Add(r.deadline() * 2) var allSeen []Event for attempt := 0; time.Now().Before(deadline); attempt++ { f := filepath.Join(sub, fmt.Sprintf("attempt-%d.txt", attempt)) if err := os.WriteFile(f, []byte("hi"), 0o644); err != nil { t.Fatal(err) } more := r.waitForEvent(750*time.Millisecond, func(e Event) bool { return e.Kind == EventUpdate && strings.HasPrefix(e.Path, sub+string(filepath.Separator)) }) allSeen = append(allSeen, more...) for _, e := range more { if e.Kind == EventUpdate && strings.HasPrefix(e.Path, sub+string(filepath.Separator)) { return } } } t.Fatalf("expected update for a file inside recreated sub (gave up after %s), got %v", r.deadline()*2, toWantEvents(allSeen)) }) } // TestReplaceParentDirWithDifferent verifies that replacing a subtree with // a different pre-populated directory of the same name still detects // changes inside the new tree. func TestReplaceParentDirWithDifferent(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) sub := filepath.Join(dir, "pkg") if err := os.MkdirAll(filepath.Join(sub, "old"), 0o755); err != nil { t.Fatal(err) } if err := os.WriteFile(filepath.Join(sub, "old", "a.txt"), []byte("a"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) // Replace: remove old tree, create new tree at same path. if err := os.RemoveAll(sub); err != nil { t.Fatal(err) } if err := os.MkdirAll(filepath.Join(sub, "new"), 0o755); err != nil { t.Fatal(err) } if err := os.WriteFile(filepath.Join(sub, "new", "b.txt"), []byte("b"), 0o644); err != nil { t.Fatal(err) } _ = r.drainQuiet(500 * time.Millisecond) // Now modify something inside the replaced tree. The same // "watch may not yet be armed on the replaced subtree" race // applies as TestRecreateSubdirAndModify; retry with fresh // filenames. newDir := filepath.Join(sub, "new") deadline := time.Now().Add(r.deadline()) var allSeen []Event for attempt := 0; time.Now().Before(deadline); attempt++ { f := filepath.Join(newDir, fmt.Sprintf("attempt-%d.txt", attempt)) if err := os.WriteFile(f, []byte("hi"), 0o644); err != nil { t.Fatal(err) } more := r.waitForEvent(750*time.Millisecond, func(e Event) bool { return e.Kind == EventUpdate && strings.HasPrefix(e.Path, newDir+string(filepath.Separator)) }) allSeen = append(allSeen, more...) for _, e := range more { if e.Kind == EventUpdate && strings.HasPrefix(e.Path, newDir+string(filepath.Separator)) { return } } } t.Fatalf("expected update for a file inside replaced tree (gave up after %s), got %v", r.deadline(), toWantEvents(allSeen)) }) } // TestRoundTripRename renames a file away and back within a short window. // The net result is that the file is unchanged, but we should see at least // some events (coalescing may merge them). func TestRoundTripRename(t *testing.T) { t.Parallel() runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { if watcherImpl == Kqueue() { t.Skip("kqueue fd-based tracking delivers stale delete before parent NOTE_WRITE reconciles") } dir := newTmpDir(t) orig := filepath.Join(dir, "data.txt") if err := os.WriteFile(orig, []byte("content"), 0o644); err != nil { t.Fatal(err) } r, _ := subscribeFor(t, dir, watcherImpl) tmp := filepath.Join(dir, "data.txt.bak") if err := os.Rename(orig, tmp); err != nil { t.Fatal(err) } if err := os.Rename(tmp, orig); err != nil { t.Fatal(err) } // Either some events or zero events (coalesced to no-op) are // both acceptable. On fd-based backends (kqueue), a transient // delete may appear if the debounce window fires between the // rename-away and rename-back; as long as a subsequent update // follows, the watcher correctly recovered. Use gatherUntilQuiet // here because we genuinely need to see "everything that arrives" // (the test asserts what coalesced, not a specific positive event). got := r.gatherUntilQuiet(r.deadline(), 500*time.Millisecond) got = filterEventsForPaths(got, orig) hasDelete := containsEvent(got, EventDelete, orig) hasUpdate := containsEvent(got, EventUpdate, orig) if hasDelete && !hasUpdate { t.Fatalf("round-trip rename left a stale delete without recovery for %s; events: %v", orig, toWantEvents(got)) } }) } // TestRecursiveWithDeniedSubdir verifies that a recursive watch succeeds // even when a child directory is unreadable. func TestRecursiveWithDeniedSubdir(t *testing.T) { t.Parallel() if runtime.GOOS == "windows" { t.Skip("chmod is not meaningful on Windows") } runForEachWatcher(t, func(t testingT, watcherImpl Watcher) { dir := newTmpDir(t) accessible := filepath.Join(dir, "ok") if err := os.Mkdir(accessible, 0o755); err != nil { t.Fatal(err) } denied := filepath.Join(dir, "denied") if err := os.Mkdir(denied, 0o755); err != nil { t.Fatal(err) } if err := os.Chmod(denied, 0); err != nil { t.Fatal(err) } t.Cleanup(func() { _ = os.Chmod(denied, 0o700) }) // Recursive watch should succeed despite the inaccessible child. r, _ := subscribeFor(t, dir, watcherImpl) // Events in the accessible sibling should still work. f := filepath.Join(accessible, "test.txt") if err := os.WriteFile(f, []byte("hello"), 0o644); err != nil { t.Fatal(err) } expectContains(t, r, EventUpdate, f) }) }