From a0b897cd0f0afb2fe0c7f2dc80dbc9b5badcf60d Mon Sep 17 00:00:00 2001 From: Gael Date: Mon, 2 Nov 2026 18:54:22 -0500 Subject: [PATCH 1/2] fix(watch): recover source watchers after build failures Publish the previously exercised owner implementation for draft review. --- agents/helpers/code/WATCHING.md | 43 ++ agents/helpers/code/watcher.go | 432 ++++++++++++------- agents/helpers/code/watcher_recovery_test.go | 182 ++++++++ agents/services/base.go | 32 +- agents/services/base_test.go | 50 +++ runners/golang/runner.go | 5 + 6 files changed, 591 insertions(+), 153 deletions(-) create mode 100644 agents/helpers/code/WATCHING.md create mode 100644 agents/helpers/code/watcher_recovery_test.go diff --git a/agents/helpers/code/WATCHING.md b/agents/helpers/code/WATCHING.md new file mode 100644 index 000000000..ddefb8b12 --- /dev/null +++ b/agents/helpers/code/WATCHING.md @@ -0,0 +1,43 @@ +# Source watching + +`NewWatcher` uses the shared filesystem observer. Existing builder dependencies +retain named-file selection and the legacy directory exclusions. Optional +`WatchTree` values declare directory interest even before a directory exists. +Their `Exclude` callback owns unconditional output exclusions. Editor-like names +such as `.tmp` and `.bak` are delivered for explicit trees because they can be +compiler inputs; only legacy directory dependency selection retains transient-name +filtering. An exact named-file dependency observes its explicitly selected file +even with such a suffix. A Git ignore rule +is not an output rule, because a compiler can consume ignored assets. + +A missing tree keeps a nonrecursive subscription on its nearest existing parent. +Created or replaced subtrees are registered recursively before invalidation is +sent. Remove and rename retain logical interest. Selected roots must be resolved +directories, not symlinks; symlink leaves are not traversed. Consumers needing +symlink targets must select those targets explicitly and own link-change policy. + +`Change.Rescan` requires full owner input discovery, including when no leaf +change was observed during a directory replacement. `Change.Err` means observation +is incomplete. `Watcher.Err()` retains the current filesystem coverage error; +it is safe to query concurrently. Neither a nil error nor successful subscription +recovery proves a valid build or application readiness. + +Registration errors retain previous successful subscriptions and retry with +backoff from 100 milliseconds to a 5-second ceiling. Overflow follows the same +resubscription and rescan path. A closed underlying event/error stream is terminal +and visible; a new watcher is required. The owner still validates its full input +selection after successful recovery. There is no source-build polling loop. + +`Start` owns channel closure and releases descriptors on cancellation. Base +owns the lifetime beyond an Init RPC, logs degraded observation and suppresses +pending rebuild callbacks while `Watcher.Err()` reports degraded coverage. +Ordinary sibling changes cannot resume callbacks. Successful registration clears +the error and emits a full rescan; Base checks coverage again after debounce. Base retains +rescan across its debounce window. Existing Pause/Resume behavior is unchanged: +ordinary events while paused are not queued; consumers must arrange rescan when +using that mode during a build. + +Real filesystem tests cover atomic save, missing/recreated roots, populated +rename-in, deletion, explicit exclusions, traversal permission failure and repair, +closed resources, and cancellation. Actual kernel overflow generation is not +portable and is not claimed by those tests. diff --git a/agents/helpers/code/watcher.go b/agents/helpers/code/watcher.go index 29f23ef7a..635917538 100644 --- a/agents/helpers/code/watcher.go +++ b/agents/helpers/code/watcher.go @@ -2,109 +2,208 @@ package code import ( "context" + "errors" + "fmt" + "io/fs" "os" - "path" "path/filepath" "strings" + "sync" "sync/atomic" - - "github.com/codefly-dev/core/shared" + "time" "github.com/codefly-dev/core/builders" - "github.com/codefly-dev/core/wool" "github.com/fsnotify/fsnotify" ) +// WatchTree is persistent directory interest, including a root which does not +// exist yet. Exclude receives absolute paths and must describe unconditional +// output exclusions, not ignore rules which may hide consumed inputs. +type WatchTree struct { + Root string + Exclude func(string) bool +} + +type watchRoot struct { + legacy bool // only legacy directory selection discards editor-like names + path string + tree bool + keep func(string) bool + exclude func(string) bool +} + type Watcher struct { - events chan<- Change - watcher *fsnotify.Watcher + directories map[string]bool // event-loop-owned prior directory identities - // internal - base string - dependencies *builders.Dependencies - paused atomic.Bool + events chan<- Change + watcher *fsnotify.Watcher + base string + roots []watchRoot + paused atomic.Bool + statusMu sync.RWMutex + watchErr error } type Change struct { Path string IsRelative bool + // Rescan requires owner input discovery even without a selected file event. + Rescan bool + // Err reports incomplete filesystem observation; it is not a source edit. + Err error +} + +// NewWatcher retains the existing dependency semantics. Explicit trees add +// recoverable directory interest and caller-owned traversal policy. +func NewWatcher(ctx context.Context, events chan<- Change, base string, dependencies *builders.Dependencies, trees ...WatchTree) (*Watcher, error) { + fw, err := fsnotify.NewWatcher() + if err != nil { + return nil, err + } + absolute, err := filepath.Abs(base) + if err != nil { + _ = fw.Close() + return nil, err + } + w := &Watcher{watcher: fw, events: events, base: absolute} + if dependencies != nil { + for _, dep := range dependencies.Components { + for _, component := range dep.Components() { + root := w.absolute(component) + info, _ := os.Stat(root) + w.roots = append(w.roots, watchRoot{path: root, tree: info != nil && info.IsDir(), keep: dep.Keep, exclude: legacyWatchExclusion, legacy: true}) + } + } + } + for _, tree := range trees { + w.roots = append(w.roots, watchRoot{path: w.absolute(tree.Root), tree: true, exclude: tree.Exclude}) + } + if err = w.reconcile(ctx); err != nil { + _ = fw.Close() + return nil, err + } + return w, nil +} + +func (w *Watcher) absolute(p string) string { + if !filepath.IsAbs(p) { + p = filepath.Join(w.base, p) + } + return filepath.Clean(p) +} + +func withinWatch(root, p string) bool { + relative, err := filepath.Rel(root, p) + return err == nil && (relative == "." || filepath.IsLocal(relative)) +} + +func legacyWatchExclusion(p string) bool { + switch filepath.Base(p) { + case ".next", "node_modules", "__pycache__", ".git": + return true + } + return false } -func addDependency(ctx context.Context, base string, dep *builders.Dependency, watcher *fsnotify.Watcher) error { - w := wool.Get(ctx).In("addDependency") - for _, p := range dep.Components() { - fullPath := path.Join(base, p) - // If path doesn't exist we skip - if exists, err := shared.Exists(ctx, fullPath); err != nil || !exists { - w.Trace("skipping", wool.Path(fullPath)) +// excluded applies directory policy to every ancestor, including deleted paths. +func (r watchRoot) excluded(p string) bool { + if r.exclude == nil { + return false + } + for withinWatch(r.path, p) { + if r.exclude(p) { + return true + } + if p == r.path { + break + } + p = filepath.Dir(p) + } + return false +} + +// reconcile keeps a nonrecursive parent sentinel for roots, then registers +// their full existing subtrees. Missing roots never broaden into a parent walk. +// It runs only on structural events or bounded recovery, not on every write. +func (w *Watcher) reconcile(ctx context.Context) error { + desired := map[string]bool{} + var failures []error + add := func(p string) { + desired[p] = true + if err := w.watcher.Add(p); err != nil { + failures = append(failures, fmt.Errorf("register watch %s: %w", p, err)) + } + } + for _, r := range w.roots { + if err := ctx.Err(); err != nil { + return err + } + parent := filepath.Dir(r.path) + for { + info, err := os.Stat(parent) + if err == nil && info.IsDir() { + add(parent) + break + } + if err != nil && !errors.Is(err, fs.ErrNotExist) { + failures = append(failures, fmt.Errorf("watch ancestor %s: %w", parent, err)) + break + } + next := filepath.Dir(parent) + if next == parent { + failures = append(failures, fmt.Errorf("no directory ancestor for %s", r.path)) + break + } + parent = next + } + if !r.tree { continue } - err := filepath.Walk(fullPath, func(path string, info os.FileInfo, err error) error { + err := filepath.WalkDir(r.path, func(p string, entry fs.DirEntry, err error) error { + if ctx.Err() != nil { + return ctx.Err() + } + if errors.Is(err, fs.ErrNotExist) { + return nil + } if err != nil { return err } - // Watch DIRECTORIES, not individual files. fsnotify file-watches - // die on atomic-rename saves — most editors write a temp file then - // rename it over the original, which destroys the watched inode, so - // the Write event is lost and hot-reload silently stops working - // (this is exactly why pkg/** edits didn't reload). A directory - // watch survives that and fires Create/Write/Rename for files - // within; the event loop filters by the dependency select. - if info.IsDir() { - base := filepath.Base(path) - if base == ".next" || base == "node_modules" || base == "__pycache__" || base == ".git" { - return filepath.SkipDir - } - w.Trace("watching dir", wool.Path(path)) - if err := watcher.Add(path); err != nil { - return w.Wrapf(err, "cannot add dir: %s", path) + if entry.Type()&os.ModeSymlink != 0 { + if p == r.path { + return fmt.Errorf("watch root %s is a symbolic link; select its resolved directory", p) } return nil } - // A single-file dependency (e.g. service.codefly.yaml) is never - // visited as a dir by Walk — watch its parent directory instead. - if dep.Keep(path) { - if err := watcher.Add(filepath.Dir(path)); err != nil { - return w.Wrapf(err, "cannot add file dir: %s", path) + if entry.IsDir() { + if r.excluded(p) { + return filepath.SkipDir } + add(p) } return nil }) if err != nil { - return w.Wrapf(err, "cannot walk path: %s", fullPath) + failures = append(failures, fmt.Errorf("watch tree %s: %w", r.path, err)) } } - return nil -} - -func NewWatcher(ctx context.Context, events chan<- Change, base string, dependencies *builders.Dependencies) (*Watcher, error) { - w := wool.Get(ctx).In("NewWatcher") - // Add new watcher. - fswatcher, err := fsnotify.NewWatcher() - if err != nil { - return nil, w.Wrapf(err, "cannot create fsnotify watcher") - } - - for _, dep := range dependencies.Components { - err = addDependency(ctx, base, dep, fswatcher) - if err != nil { - return nil, w.Wrapf(err, "cannot add dependency") + // Do not discard the previous successful coverage during a partial failure. + if len(failures) == 0 { + w.directories = desired + for _, p := range w.watcher.WatchList() { + if !desired[p] { + _ = w.watcher.Remove(p) + } } } - watcher := &Watcher{watcher: fswatcher, base: base, events: events, dependencies: dependencies} - - return watcher, nil + return errors.Join(failures...) } -// keep reports whether a changed path matches any watched dependency select. -// We watch whole directories now, so this filters out non-source churn -// (temp files, build output) that a directory watch also surfaces. -// isTransientFile reports whether a path is an editor temp / backup / lock -// file that fires filesystem events during a save but is not a real source -// change: generic temp/backup suffixes (.tmp/.bak/~), vim swap files -// (.swp/.swx/.swo/.swpx), and emacs lock/autosave (.#name, #name#). The -// atomic-rename TARGET (the real file) is not transient and still triggers a -// rebuild — so saves still hot-reload, but the temp churn around them doesn't. +// Err reports filesystem coverage only, not successful owner rediscovery or +// application readiness. Consumers must process Rescan before claiming either. +func (w *Watcher) Err() error { w.statusMu.RLock(); defer w.statusMu.RUnlock(); return w.watchErr } +func (w *Watcher) setError(err error) { w.statusMu.Lock(); w.watchErr = err; w.statusMu.Unlock() } + func isTransientFile(p string) bool { base := filepath.Base(p) if strings.HasPrefix(base, ".#") || (strings.HasPrefix(base, "#") && strings.HasSuffix(base, "#")) { @@ -118,130 +217,159 @@ func isTransientFile(p string) bool { return false } -func (watcher *Watcher) keep(absPath string) bool { - if watcher.dependencies == nil { - return true +func (w *Watcher) keep(p string) bool { + for _, r := range w.roots { + if (p == r.path || r.tree && withinWatch(r.path, p)) && !r.excluded(p) && (!r.legacy || !r.tree || !isTransientFile(p)) && (r.keep == nil || r.keep(p)) { + return true + } } - absPath = filepath.Clean(absPath) - for _, dep := range watcher.dependencies.Components { - for _, component := range dep.Components() { - root := component - if !filepath.IsAbs(root) { - root = filepath.Join(watcher.base, root) - } - root = filepath.Clean(root) - - // A dependency can name either one file (service.codefly.yaml) or - // a directory tree (code/). PathSelect only answers the second half - // of that question; applying it without first checking the component - // scope made every no-select file dependency match every event in its - // parent directory. - inScope := absPath == root - if info, err := os.Stat(root); err == nil && info.IsDir() { - rel, relErr := filepath.Rel(root, absPath) - inScope = relErr == nil && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator)) - } - if inScope && dep.Keep(absPath) { + return false +} + +func (w *Watcher) structural(p string) bool { + for _, r := range w.roots { + if r.legacy && r.tree && isTransientFile(p) { + continue + } + // Ancestor sentinels only admit the path leading to a selected root. + if withinWatch(p, r.path) { + return true + } + if r.tree && withinWatch(r.path, p) && !r.excluded(p) { + if info, err := os.Stat(p); err == nil && info.IsDir() { return true } + for watched := range w.directories { + if withinWatch(p, watched) { + return true + } + } } } return false } -func (watcher *Watcher) Start(ctx context.Context) { - w := wool.Get(ctx).In("Watcher.Start") - // Start listening for events. - defer watcher.watcher.Close() - // Start is the sole sender on events (via Handle, only from this loop) AND - // the sole closer: closing it here is the single close of the channel, which - // lets the consumer's receive drain and exit on teardown. Nothing else may - // close Events — a second closer (an earlier version had Runtime.Stop close - // it too) races this goroutine into a "close of closed channel" panic. To end - // this goroutine, cancel its context (see services.Base.StopWatcher). - defer close(watcher.events) +func (w *Watcher) Start(ctx context.Context) { + defer w.watcher.Close() + defer close(w.events) // sole closer, including cancellation while sending + var retry *time.Timer + var retryC <-chan time.Time + delay := 100 * time.Millisecond + defer func() { + if retry != nil { + retry.Stop() + } + }() + recoverWatch := func(cause error) bool { + w.setError(cause) + if !w.send(ctx, Change{Err: cause, Rescan: true}) { + return false + } + if retry == nil { + retry = time.NewTimer(delay) + } else { + retry.Reset(delay) + } + retryC = retry.C + if delay < 5*time.Second { + delay *= 2 + if delay > 5*time.Second { + delay = 5 * time.Second + } + } + return true + } for { select { case <-ctx.Done(): return - case event, ok := <-watcher.watcher.Events: + case <-retryC: + retryC = nil + if err := w.reconcile(ctx); err != nil { + if !recoverWatch(err) { + return + } + continue + } + w.setError(nil) + delay = 100 * time.Millisecond + if !w.send(ctx, Change{Rescan: true}) { + return + } + case event, ok := <-w.watcher.Events: if !ok { + failure := errors.New("filesystem event stream closed") + w.setError(failure) + w.send(ctx, Change{Err: failure, Rescan: true}) return } - // Write covers in-place saves; Create + Rename cover atomic-rename - // saves (write temp → rename over original), which is how most - // editors save — and which the old Write-only loop dropped. - if event.Has(fsnotify.Write) || event.Has(fsnotify.Create) || event.Has(fsnotify.Rename) { - // A newly-created directory (e.g. a new package) must be added - // to the watch set — fsnotify is not recursive. - if event.Has(fsnotify.Create) { - if info, statErr := os.Stat(event.Name); statErr == nil && info.IsDir() { - _ = watcher.watcher.Add(event.Name) - continue + if event.Op&(fsnotify.Write|fsnotify.Create|fsnotify.Rename|fsnotify.Remove) == 0 { + continue + } + structural := w.structural(event.Name) + selected := w.keep(event.Name) + if structural { + if event.Op&(fsnotify.Remove|fsnotify.Rename) != 0 { + // A replacement can already occupy the same pathname. Drop + // the old inode subscription before registering its successor. + for _, watched := range w.watcher.WatchList() { + if withinWatch(event.Name, watched) { + _ = w.watcher.Remove(watched) + } } } - // Editor temp/backup files fire Create/Rename during atomic - // saves (write temp → rename over the original) but are not real - // source changes. A named-file dependency with no path select - // (e.g. service.codefly.yaml) matches EVERY path, so without this - // skip a `foo.go.tmp` triggers a spurious rebuild — and rebuild - // churn races the restart and wedges hot-reload. The rename - // TARGET (the real file) is not transient and still triggers. - if isTransientFile(event.Name) { - continue - } - // We watch directories, so drop events for paths that don't - // match a watched dependency (build output, unrelated files). - if !watcher.keep(event.Name) { - continue - } - w.Debug("got event", wool.Field("event", event)) - rel, err := filepath.Rel(watcher.base, event.Name) - if err != nil { - w.Error("cannot get relative path", wool.Field("base", watcher.base), wool.Field("path", event.Name)) + if err := w.reconcile(ctx); err != nil { + if !recoverWatch(err) { + return + } continue } - change := Change{ - Path: rel, - IsRelative: true, - } - if !watcher.send(ctx, change) { + } + if !selected && !structural { + continue + } + relative, err := filepath.Rel(w.base, event.Name) + if err != nil { + if !recoverWatch(err) { return } - continue } - case _, ok := <-watcher.watcher.Errors: + if !w.send(ctx, Change{Path: relative, IsRelative: true, Rescan: structural}) { + return + } + case err, ok := <-w.watcher.Errors: if !ok { + failure := errors.New("filesystem error stream closed") + w.setError(failure) + w.send(ctx, Change{Err: failure, Rescan: true}) + return + } + if !recoverWatch(err) { return } } } } -func (watcher *Watcher) send(ctx context.Context, event Change) bool { - if watcher.paused.Load() { +func (w *Watcher) send(ctx context.Context, event Change) bool { + // Ordinary sibling traffic cannot stand in for restored full coverage. + // A successful recovery emits one full Rescan after clearing watchErr. + if event.Err == nil && (w.paused.Load() || w.Err() != nil) { return true } select { - case watcher.events <- event: + case w.events <- event: return true case <-ctx.Done(): return false } } -func (watcher *Watcher) Handle(event Change) { - if watcher.paused.Load() { - return +func (w *Watcher) Handle(event Change) { + if !w.paused.Load() { + w.events <- event } - watcher.events <- event -} - -func (watcher *Watcher) Pause() { - watcher.paused.Store(true) -} - -func (watcher *Watcher) Resume() { - watcher.paused.Store(false) } +func (w *Watcher) Pause() { w.paused.Store(true) } +func (w *Watcher) Resume() { w.paused.Store(false) } diff --git a/agents/helpers/code/watcher_recovery_test.go b/agents/helpers/code/watcher_recovery_test.go new file mode 100644 index 000000000..6b46b8e8d --- /dev/null +++ b/agents/helpers/code/watcher_recovery_test.go @@ -0,0 +1,182 @@ +package code + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" + + "github.com/codefly-dev/core/builders" + "github.com/codefly-dev/core/shared" + "github.com/stretchr/testify/require" +) + +func recoveryWatcher(t *testing.T, base string, trees ...WatchTree) (*Watcher, <-chan Change) { + t.Helper() + events := make(chan Change, 128) + ctx, cancel := context.WithCancel(context.Background()) + w, err := NewWatcher(ctx, events, base, nil, trees...) + require.NoError(t, err) + done := make(chan struct{}) + go func() { defer close(done); w.Start(ctx) }() + t.Cleanup(func() { + cancel() + select { + case <-done: + case <-time.After(3 * time.Second): + t.Error("watcher did not stop") + } + }) + return w, events +} + +func awaitWatch(t *testing.T, events <-chan Change, predicate func(Change) bool) Change { + t.Helper() + deadline := time.NewTimer(4 * time.Second) + defer deadline.Stop() + for { + select { + case change, ok := <-events: + require.True(t, ok, "watch closed before expected observation") + if predicate(change) { + return change + } + case <-deadline.C: + t.Fatal("missing filesystem observation") + return Change{} + } + } +} +func drainWatch(events <-chan Change) { + for { + select { + case _, ok := <-events: + if !ok { + return + } + default: + return + } + } +} + +func TestWatcherExplicitTreeRecoversMissingAndRecreatedRoot(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "missing", "source") + _, events := recoveryWatcher(t, base, WatchTree{Root: root}) + require.NoError(t, os.MkdirAll(root, 0755)) + awaitWatch(t, events, func(c Change) bool { return c.Rescan }) + target := filepath.Join(root, "schema.json") + require.NoError(t, os.WriteFile(target, []byte("first"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("missing", "source", "schema.json") }) + require.NoError(t, os.RemoveAll(filepath.Join(base, "missing"))) + awaitWatch(t, events, func(c Change) bool { return c.Rescan }) + require.NoError(t, os.MkdirAll(root, 0755)) + awaitWatch(t, events, func(c Change) bool { return c.Rescan }) + require.NoError(t, os.WriteFile(target, []byte("repaired"), 0644)) + // A recreated populated tree is a full invalidation; leaf events before + // registration need not be replayed. The owner reads the current input. + awaitWatch(t, events, func(c Change) bool { return c.Rescan || c.Path == filepath.Join("missing", "source", "schema.json") }) + actual, err := os.ReadFile(target) + require.NoError(t, err) + require.Equal(t, "repaired", string(actual)) + require.NoError(t, os.WriteFile(target, []byte("subsequent"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("missing", "source", "schema.json") }) +} + +func TestWatcherRegistersPopulatedIncomingTreeAndDeletion(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + incoming := filepath.Join(base, "incoming") + require.NoError(t, os.MkdirAll(root, 0755)) + require.NoError(t, os.MkdirAll(filepath.Join(incoming, "deep"), 0755)) + require.NoError(t, os.WriteFile(filepath.Join(incoming, "deep", "schema.json"), []byte("one"), 0644)) + _, events := recoveryWatcher(t, base, WatchTree{Root: root}) + require.NoError(t, os.Rename(incoming, filepath.Join(root, "new"))) + awaitWatch(t, events, func(c Change) bool { return c.Rescan && c.Path == filepath.Join("source", "new") }) + target := filepath.Join(root, "new", "deep", "schema.json") + require.NoError(t, os.WriteFile(target, []byte("two"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", "new", "deep", "schema.json") }) + require.NoError(t, os.Remove(target)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", "new", "deep", "schema.json") }) +} + +func TestWatcherExplicitExclusionsDoNotHideIgnoredInputs(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + for _, dir := range []string{"node_modules", "ignored", "dist"} { + require.NoError(t, os.MkdirAll(filepath.Join(root, dir), 0755)) + } + require.NoError(t, os.WriteFile(filepath.Join(root, ".gitignore"), []byte("ignored/\n"), 0644)) + _, events := recoveryWatcher(t, base, WatchTree{Root: root, Exclude: func(p string) bool { return filepath.Base(p) == "dist" }}) + for _, name := range []string{"node_modules/embedded.json", "ignored/named.tgz"} { + require.NoError(t, os.WriteFile(filepath.Join(root, name), []byte("input"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", name) }) + } + drainWatch(events) + require.NoError(t, os.WriteFile(filepath.Join(root, "dist", "output.js"), []byte("output"), 0644)) + select { + case c := <-events: + t.Fatalf("excluded output produced %#v", c) + case <-time.After(150 * time.Millisecond): + } +} + +func TestWatcherLegacyTreeKeepsScopeAfterRemoval(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + require.NoError(t, os.MkdirAll(root, 0755)) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + events := make(chan Change, 64) + w, err := NewWatcher(ctx, events, base, builders.NewDependencies("go", builders.NewDependency("source").WithPathSelect(shared.NewSelect("*.go")))) + require.NoError(t, err) + go w.Start(ctx) + require.NoError(t, os.RemoveAll(root)) + awaitWatch(t, events, func(c Change) bool { return c.Rescan }) + require.NoError(t, os.MkdirAll(root, 0755)) + awaitWatch(t, events, func(c Change) bool { return c.Rescan }) + require.NoError(t, os.WriteFile(filepath.Join(root, "main.go"), []byte("package main"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", "main.go") }) +} + +func TestWatcherClosedFilesystemReportsDegradedState(t *testing.T) { + base := t.TempDir() + w, events := recoveryWatcher(t, base, WatchTree{Root: base}) + require.NoError(t, w.watcher.Close()) // a real underlying resource failure, not a fake error stream + awaitWatch(t, events, func(c Change) bool { return c.Err != nil && c.Rescan }) + require.Error(t, w.Err()) +} + +func TestWatcherRegistrationFailureRecoversAfterPermissionsRepair(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + incoming := filepath.Join(base, "incoming") + require.NoError(t, os.MkdirAll(root, 0755)) + require.NoError(t, os.MkdirAll(incoming, 0755)) + require.NoError(t, os.Chmod(incoming, 0300)) + t.Cleanup(func() { _ = os.Chmod(filepath.Join(root, "incoming"), 0755) }) + if _, err := os.ReadDir(incoming); err == nil { + t.Skip("process can bypass directory permissions") + } + w, events := recoveryWatcher(t, base, WatchTree{Root: root}) + require.NoError(t, os.Rename(incoming, filepath.Join(root, "incoming"))) + awaitWatch(t, events, func(c Change) bool { return c.Err != nil }) + require.Error(t, w.Err()) + require.NoError(t, os.Chmod(filepath.Join(root, "incoming"), 0755)) + awaitWatch(t, events, func(c Change) bool { return c.Rescan && c.Err == nil }) + require.NoError(t, w.Err()) + require.NoError(t, os.WriteFile(filepath.Join(root, "incoming", "repaired.json"), []byte("ok"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", "incoming", "repaired.json") }) +} + +func TestWatcherExplicitSymlinkRootRefusesIncompleteCoverage(t *testing.T) { + base := t.TempDir() + target := filepath.Join(base, "target") + link := filepath.Join(base, "link") + require.NoError(t, os.Mkdir(target, 0755)) + require.NoError(t, os.Symlink(target, link)) + _, err := NewWatcher(t.Context(), make(chan Change), base, nil, WatchTree{Root: link}) + require.ErrorContains(t, err, "symbolic link") +} diff --git a/agents/services/base.go b/agents/services/base.go index ba189f4c0..0ba81d06b 100644 --- a/agents/services/base.go +++ b/agents/services/base.go @@ -255,6 +255,7 @@ func cloneDockerImage(image *resources.DockerImage) *resources.DockerImage { type WatchConfiguration struct { dependencies *builders.Dependencies + trees []code.WatchTree } func NewWatchConfiguration(dependencies *builders.Dependencies) *WatchConfiguration { @@ -263,6 +264,12 @@ func NewWatchConfiguration(dependencies *builders.Dependencies) *WatchConfigurat } } +// WithTrees adds explicit recoverable directory interest. The caller owns exclusions. +func (c *WatchConfiguration) WithTrees(trees ...code.WatchTree) *WatchConfiguration { + c.trees = append(c.trees, trees...) + return c +} + func (s *Base) SetupWatcher(ctx context.Context, conf *WatchConfiguration, handler func(event code.Change) error) error { s.Wool.Debug("watching for changes", wool.Field("dependencies", builders.MakeDependenciesSummary(conf.dependencies))) // Init is an RPC, so its context is cancelled as soon as the handler @@ -272,7 +279,7 @@ func (s *Base) SetupWatcher(ctx context.Context, conf *WatchConfiguration, handl watcherCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) events := make(chan code.Change) var err error - watcher, err := code.NewWatcher(ctx, events, s.Location, conf.dependencies) + watcher, err := code.NewWatcher(ctx, events, s.Location, conf.dependencies, conf.trees...) if err != nil { cancel() return err @@ -298,6 +305,11 @@ func (s *Base) SetupWatcher(ctx context.Context, conf *WatchConfiguration, handl go func() { const quiet = WatchDebounce var timer *time.Timer + defer func() { + if timer != nil { + timer.Stop() + } + }() var timerC <-chan time.Time var pending bool var last code.Change @@ -307,6 +319,19 @@ func (s *Base) SetupWatcher(ctx context.Context, conf *WatchConfiguration, handl if !ok { return } + if event.Err != nil { + s.Wool.Error("source watch incomplete; recovery pending", wool.ErrField(event.Err)) + pending = false + timerC = nil + if timer != nil { + timer.Stop() + } + continue + } + if watcher.Err() != nil { + continue + } + event.Rescan = event.Rescan || pending && last.Rescan last = event pending = true if timer == nil { @@ -325,6 +350,11 @@ func (s *Base) SetupWatcher(ctx context.Context, conf *WatchConfiguration, handl timerC = nil if pending { pending = false + // Coverage may have degraded since this event entered debounce. + // Keep the recovery watcher alive; only its rescan may resume. + if watcher.Err() != nil { + continue + } if err := handler(last); err != nil { s.Wool.Error("got", wool.ErrField(err)) } diff --git a/agents/services/base_test.go b/agents/services/base_test.go index 256363d2c..385af3fad 100644 --- a/agents/services/base_test.go +++ b/agents/services/base_test.go @@ -219,3 +219,53 @@ func save(t *testing.T, target, content string) { require.NoError(t, os.WriteFile(tmp, []byte(content), 0o644)) require.NoError(t, os.Rename(tmp, target)) } + +// A healthy sibling cannot tear down the watcher while a different subtree +// remains unreadable. The pending recovery must survive to observe its repair. +func TestSetupWatcherDoesNotRestartWhileCoverageDegraded(t *testing.T) { + root := t.TempDir() + source := filepath.Join(root, "source") + incoming := filepath.Join(root, "incoming") + require.NoError(t, os.Mkdir(source, 0755)) + require.NoError(t, os.Mkdir(incoming, 0755)) + require.NoError(t, os.Chmod(incoming, 0300)) + t.Cleanup(func() { _ = os.Chmod(filepath.Join(source, "incoming"), 0755); _ = os.Chmod(incoming, 0755) }) + if _, err := os.ReadDir(incoming); err == nil { + t.Skip("process bypasses directory permissions") + } + require.NoError(t, os.WriteFile(filepath.Join(source, "sibling.go"), []byte("package sibling"), 0644)) + base := &Base{Wool: wool.Get(t.Context()), Location: root} + callbacks := make(chan code.Change, 8) + require.NoError(t, base.SetupWatcher(t.Context(), NewWatchConfiguration(builders.NewDependencies("recovery")).WithTrees(code.WatchTree{Root: source}), func(c code.Change) error { callbacks <- c; return nil })) + t.Cleanup(base.StopWatcher) + require.NoError(t, os.Rename(incoming, filepath.Join(source, "incoming"))) + require.Eventually(t, func() bool { return base.Watcher.Err() != nil }, time.Second, 10*time.Millisecond) + // Observe four further real failed registrations. The next bounded backoff + // exceeds debounce, making the sibling callback race deterministic. + for attempt := 0; attempt < 4; attempt++ { + previous := base.Watcher.Err() + require.Eventually(t, func() bool { current := base.Watcher.Err(); return current != nil && current != previous }, 6*time.Second, 5*time.Millisecond) + } + require.NoError(t, os.WriteFile(filepath.Join(source, "sibling.go"), []byte("package sibling\n// changed"), 0644)) + select { + case c := <-callbacks: + t.Fatalf("premature restart while coverage is degraded: %#v", c) + case <-time.After(WatchDebounce + 150*time.Millisecond): + } + require.Error(t, base.Watcher.Err()) + require.NoError(t, os.Chmod(filepath.Join(source, "incoming"), 0755)) + select { + case c := <-callbacks: + require.True(t, c.Rescan, "repair must rediscover the whole input set") + case <-time.After(7 * time.Second): + t.Fatal("repair did not resume callback") + } + require.NoError(t, base.Watcher.Err()) + require.NoError(t, os.WriteFile(filepath.Join(source, "incoming", "next.go"), []byte("package next"), 0644)) + select { + case c := <-callbacks: + require.Equal(t, filepath.Join("source", "incoming", "next.go"), c.Path) + case <-time.After(2 * time.Second): + t.Fatal("repaired subtree is not watched") + } +} diff --git a/runners/golang/runner.go b/runners/golang/runner.go index 92cf686f8..d609b7dab 100644 --- a/runners/golang/runner.go +++ b/runners/golang/runner.go @@ -392,6 +392,11 @@ func (r *GoRunnerEnvironment) WithWorkspace(b bool) { r.withGoWorkspace = b || r.ownsGoWorkspace } +// WorkspaceEnabled reports the graph policy selected by this runner, including +// a source-owned go.work even when ambient workspace inheritance is disabled. +// Discovery must use this policy with Env(), just like native compilation. +func (r *GoRunnerEnvironment) WorkspaceEnabled() bool { return r.withGoWorkspace } + func (r *GoRunnerEnvironment) Init(ctx context.Context) error { w := wool.Get(ctx).In("init") From c1a62d1b555fc416bb0974b3900e38695b9103d5 Mon Sep 17 00:00:00 2001 From: Gael Date: Tue, 3 Nov 2026 00:07:16 -0500 Subject: [PATCH 2/2] test(watch): isolate exclusion checks from earlier filesystem events --- agents/helpers/code/watcher_recovery_test.go | 59 ++++++++++---------- 1 file changed, 28 insertions(+), 31 deletions(-) diff --git a/agents/helpers/code/watcher_recovery_test.go b/agents/helpers/code/watcher_recovery_test.go index 6b46b8e8d..313a07699 100644 --- a/agents/helpers/code/watcher_recovery_test.go +++ b/agents/helpers/code/watcher_recovery_test.go @@ -48,19 +48,6 @@ func awaitWatch(t *testing.T, events <-chan Change, predicate func(Change) bool) } } } -func drainWatch(events <-chan Change) { - for { - select { - case _, ok := <-events: - if !ok { - return - } - default: - return - } - } -} - func TestWatcherExplicitTreeRecoversMissingAndRecreatedRoot(t *testing.T) { base := t.TempDir() root := filepath.Join(base, "missing", "source") @@ -103,24 +90,34 @@ func TestWatcherRegistersPopulatedIncomingTreeAndDeletion(t *testing.T) { } func TestWatcherExplicitExclusionsDoNotHideIgnoredInputs(t *testing.T) { - base := t.TempDir() - root := filepath.Join(base, "source") - for _, dir := range []string{"node_modules", "ignored", "dist"} { - require.NoError(t, os.MkdirAll(filepath.Join(root, dir), 0755)) - } - require.NoError(t, os.WriteFile(filepath.Join(root, ".gitignore"), []byte("ignored/\n"), 0644)) - _, events := recoveryWatcher(t, base, WatchTree{Root: root, Exclude: func(p string) bool { return filepath.Base(p) == "dist" }}) - for _, name := range []string{"node_modules/embedded.json", "ignored/named.tgz"} { - require.NoError(t, os.WriteFile(filepath.Join(root, name), []byte("input"), 0644)) - awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", name) }) - } - drainWatch(events) - require.NoError(t, os.WriteFile(filepath.Join(root, "dist", "output.js"), []byte("output"), 0644)) - select { - case c := <-events: - t.Fatalf("excluded output produced %#v", c) - case <-time.After(150 * time.Millisecond): - } + t.Run("ignored inputs are observed", func(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + for _, dir := range []string{"node_modules", "ignored", "dist"} { + require.NoError(t, os.MkdirAll(filepath.Join(root, dir), 0755)) + } + require.NoError(t, os.WriteFile(filepath.Join(root, ".gitignore"), []byte("ignored/\n"), 0644)) + _, events := recoveryWatcher(t, base, WatchTree{Root: root, Exclude: func(p string) bool { return filepath.Base(p) == "dist" }}) + for _, name := range []string{"node_modules/embedded.json", "ignored/named.tgz"} { + require.NoError(t, os.WriteFile(filepath.Join(root, name), []byte("input"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", name) }) + } + }) + + t.Run("excluded output is not observed", func(t *testing.T) { + base := t.TempDir() + root := filepath.Join(base, "source") + require.NoError(t, os.MkdirAll(filepath.Join(root, "dist"), 0755)) + _, events := recoveryWatcher(t, base, WatchTree{Root: root, Exclude: func(p string) bool { return filepath.Base(p) == "dist" }}) + require.NoError(t, os.WriteFile(filepath.Join(root, "dist", "output.js"), []byte("output"), 0644)) + select { + case c := <-events: + t.Fatalf("excluded output produced %#v", c) + case <-time.After(150 * time.Millisecond): + } + require.NoError(t, os.WriteFile(filepath.Join(root, "input.json"), []byte("input"), 0644)) + awaitWatch(t, events, func(c Change) bool { return c.Path == filepath.Join("source", "input.json") }) + }) } func TestWatcherLegacyTreeKeepsScopeAfterRemoval(t *testing.T) {