diff --git a/cmd/context_routing.go b/cmd/context_routing.go index f18f392..f1bf370 100644 --- a/cmd/context_routing.go +++ b/cmd/context_routing.go @@ -17,6 +17,7 @@ type contextFileIndex struct { exact map[string][]string basenames map[string][]string sortedPaths []contextIndexedPath + prefixPaths map[string][]string } type contextIndexedPath struct { @@ -121,9 +122,6 @@ func newContextFileIndex(files []scanner.FileInfo, caseInsensitive bool) context if path == "" { continue } - if _, duplicate := index.exact[path]; duplicate && !caseInsensitive { - continue - } pathKey := index.key(path) duplicate := false for _, existing := range index.exact[pathKey] { @@ -147,6 +145,17 @@ func newContextFileIndex(files []scanner.FileInfo, caseInsensitive bool) context for order := range index.sortedPaths { index.sortedPaths[order].order = order } + index.prefixPaths = make(map[string][]string) + for _, indexed := range index.sortedPaths { + for prefix := indexed.path; prefix != "."; prefix = pathpkg.Dir(prefix) { + prefixKey := index.key(prefix) + index.prefixPaths[prefixKey] = append(index.prefixPaths[prefixKey], indexed.path) + dir := pathpkg.Dir(prefix) + if dir == prefix { + break + } + } + } } sort.Slice(index.sortedPaths, func(i, j int) bool { if index.sortedPaths[i].key == index.sortedPaths[j].key { @@ -158,31 +167,18 @@ func newContextFileIndex(files []scanner.FileInfo, caseInsensitive bool) context } func (i contextFileIndex) forPrefix(prefixKey string, visit func(string) bool) bool { - exactStart := sort.Search(len(i.sortedPaths), func(index int) bool { - return i.sortedPaths[index].key >= prefixKey - }) if i.caseInsensitive { - matches := make([]contextIndexedPath, 0) - for index := exactStart; index < len(i.sortedPaths) && i.sortedPaths[index].key == prefixKey; index++ { - matches = append(matches, i.sortedPaths[index]) - } - descendantPrefix := prefixKey + "/" - descendantStart := sort.Search(len(i.sortedPaths), func(index int) bool { - return i.sortedPaths[index].key >= descendantPrefix - }) - for index := descendantStart; index < len(i.sortedPaths) && strings.HasPrefix(i.sortedPaths[index].key, descendantPrefix); index++ { - matches = append(matches, i.sortedPaths[index]) - } - sort.Slice(matches, func(left, right int) bool { - return matches[left].order < matches[right].order - }) - for _, match := range matches { - if visit(match.path) { + for _, path := range i.prefixPaths[prefixKey] { + if visit(path) { return true } } return false } + + exactStart := sort.Search(len(i.sortedPaths), func(index int) bool { + return i.sortedPaths[index].key >= prefixKey + }) for index := exactStart; index < len(i.sortedPaths) && i.sortedPaths[index].key == prefixKey; index++ { if visit(i.sortedPaths[index].path) { return true diff --git a/cmd/context_routing_test.go b/cmd/context_routing_test.go index ad3a36a..14dc340 100644 --- a/cmd/context_routing_test.go +++ b/cmd/context_routing_test.go @@ -230,6 +230,9 @@ func TestContextFileIndexPrefixesRespectBoundaries(t *testing.T) { func TestContextFileIndexPrefixesRespectCaseFolding(t *testing.T) { files := routingFiles("Src/Build/z.go", "src/build/a.go", "src/building/b.go") index := newContextFileIndex(files, true) + if want := []string{"Src/Build/z.go", "src/build/a.go"}; !reflect.DeepEqual(index.prefixPaths["src/build"], want) { + t.Fatalf("case-folded prefix index = %#v, want %#v", index.prefixPaths["src/build"], want) + } var got []string index.forPrefix(index.key("src/build"), func(path string) bool { got = append(got, path) diff --git a/watch/daemon.go b/watch/daemon.go index e9d3eba..12f83bd 100644 --- a/watch/daemon.go +++ b/watch/daemon.go @@ -41,6 +41,29 @@ type Daemon struct { eventLoopWG sync.WaitGroup publisher *statePublisher closeWatcher func() error + + dependencyRequests chan dependencyGraphSnapshot + dependencyResults chan dependencyGraphResult + dependencyOnce sync.Once + dependencyCancel context.CancelFunc + dependencyWorkerWG sync.WaitGroup + // These flags are owned by eventLoop; the worker only exchanges snapshots + // and results through the channels above. + dependencyBusy bool + dependencyPending bool +} + +type dependencyGraphSnapshot struct { + configured []string + config config.ProjectConfig + generation uint64 +} + +type dependencyGraphResult struct { + snapshot dependencyGraphSnapshot + graph *scanner.FileGraph + err error + started time.Time } func (d *Daemon) runtimeStateDir() (string, error) { @@ -166,6 +189,7 @@ func (d *Daemon) Start() error { if err := d.publisher.publish(); err != nil { return fmt.Errorf("publish initial state: %w", err) } + d.startDependencyWorker() // Start event loop d.eventLoopWG.Add(1) @@ -227,7 +251,11 @@ func (d *Daemon) computeTopology() { // Stop gracefully shuts down the daemon func (d *Daemon) Stop() { close(d.done) + if d.dependencyCancel != nil { + d.dependencyCancel() + } d.eventLoopWG.Wait() + d.dependencyWorkerWG.Wait() _ = d.closeWatcher() } @@ -344,17 +372,15 @@ func (d *Daemon) refreshConfiguredFiles(resetIgnoreCache bool) error { d.graph.ConfiguredFiles = configured // Filters define dependency membership too, so the previous graph must not // be published under a new configured-file count. - d.graph.FileGraph = nil - d.graph.DepCtx = make(map[string]*DepContext) - d.graph.HasDeps = false + d.markGraphLifecycleLocked(newGraphState(d.root, config.Load(d.root), graphLifecycleStale, time.Time{}, nil)) d.graph.mu.Unlock() // Invalidation alone would leave the daemon serving no hub or importer // intelligence until it restarts, so every hook reading daemon state would - // silently degrade after one config edit. Rebuild under the same size guard - // Start uses. computeDeps takes the lock itself, so call it unlocked. + // silently degrade after one config edit. Queue the rebuild under the same + // size guard Start uses. if shouldComputeDependencyGraph(len(configured)) { - d.computeDeps() + d.refreshDependencies() } d.computeTopology() return nil @@ -362,18 +388,99 @@ func (d *Daemon) refreshConfiguredFiles(resetIgnoreCache bool) error { var daemonRefreshDependencies = (*Daemon).refreshDependencies +// refreshDependencies is called by eventLoop and owns the worker state flags. func (d *Daemon) refreshDependencies() { d.graph.mu.RLock() stale := d.graph.GraphState.Status == graphLifecycleStale configuredCount := len(d.graph.ConfiguredFiles) + configured := make([]string, 0, len(d.graph.ConfiguredFiles)) + for file := range d.graph.ConfiguredFiles { + configured = append(configured, file) + } + snapshot := dependencyGraphSnapshot{ + configured: configured, + config: config.Load(d.root), + generation: d.graph.graphGeneration, + } d.graph.mu.RUnlock() if !stale || !shouldComputeDependencyGraph(configuredCount) { return } - d.computeDeps() + d.startDependencyWorker() + if d.dependencyBusy { + d.dependencyPending = true + return + } + d.dependencyBusy = true + select { + case d.dependencyRequests <- snapshot: + case <-d.done: + d.dependencyBusy = false + } +} + +// buildDependencyGraph converts a worker panic into the existing failed-build +// path so a background scan cannot terminate the daemon process. +func buildDependencyGraph(ctx context.Context, root string, build func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error)) (graph *scanner.FileGraph, err error) { + defer func() { + if recovered := recover(); recovered != nil { + err = fmt.Errorf("dependency graph build panicked: %v", recovered) + } + }() + return build(ctx, root, scanner.ConfiguredFilters(root)) +} + +func (d *Daemon) startDependencyWorker() { + d.dependencyOnce.Do(func() { + if d.dependencyRequests == nil { + d.dependencyRequests = make(chan dependencyGraphSnapshot, 1) + } + if d.dependencyResults == nil { + d.dependencyResults = make(chan dependencyGraphResult, 1) + } + ctx, cancel := context.WithCancel(context.Background()) + d.dependencyCancel = cancel + d.dependencyWorkerWG.Add(1) + go func() { + defer d.dependencyWorkerWG.Done() + for { + select { + case <-ctx.Done(): + return + case snapshot := <-d.dependencyRequests: + started := time.Now() + graph, err := buildDependencyGraph(ctx, d.root, buildFileGraph) + result := dependencyGraphResult{snapshot: snapshot, graph: graph, err: err, started: started} + select { + case d.dependencyResults <- result: + case <-ctx.Done(): + return + } + } + } + }() + }) +} + +// handleDependencyGraphResult is called by eventLoop and owns the worker state flags. +func (d *Daemon) handleDependencyGraphResult(result dependencyGraphResult) { + d.dependencyBusy = false + d.applyDependencyGraph(result.snapshot, result.graph, result.err, result.started) + retry := d.dependencyPending + d.dependencyPending = false + if retry { + d.graph.mu.Lock() + if d.graph.GraphState.Status != graphLifecycleStale { + d.markGraphLifecycleLocked(newGraphState(d.root, config.Load(d.root), graphLifecycleStale, time.Time{}, nil)) + } + d.graph.mu.Unlock() + } if d.publisher != nil { d.reportPublicationError(d.writeState()) } + if retry { + d.refreshDependencies() + } } // computeDeps builds the file-to-file dependency graph @@ -389,28 +496,39 @@ func (d *Daemon) computeDepsWith(build func(context.Context, string) (*scanner.F func (d *Daemon) computeDepsWithBeforePublish(build func(context.Context, string) (*scanner.FileGraph, error), beforePublish func()) { start := time.Now() - buildConfig := config.Load(d.root) - d.graph.mu.RLock() - configuredBefore := make([]string, 0, len(d.graph.ConfiguredFiles)) - for file := range d.graph.ConfiguredFiles { - configuredBefore = append(configuredBefore, file) - } - generationBefore := d.graph.graphGeneration - d.graph.mu.RUnlock() + snapshot := d.dependencyGraphSnapshot() // Build the file graph. Unavailable coverage provides no usable dependency // evidence, so do not publish an authoritative empty graph. fg, err := build(context.Background(), d.root) - if err != nil || fg == nil || (len(configuredBefore) > 0 && fg.Coverage.Status == analysis.CoverageUnavailable) { + if beforePublish != nil { + beforePublish() + } + d.applyDependencyGraph(snapshot, fg, err, start) +} + +func (d *Daemon) dependencyGraphSnapshot() dependencyGraphSnapshot { + d.graph.mu.RLock() + defer d.graph.mu.RUnlock() + configured := make([]string, 0, len(d.graph.ConfiguredFiles)) + for file := range d.graph.ConfiguredFiles { + configured = append(configured, file) + } + return dependencyGraphSnapshot{ + configured: configured, + config: config.Load(d.root), + generation: d.graph.graphGeneration, + } +} + +func (d *Daemon) applyDependencyGraph(snapshot dependencyGraphSnapshot, fg *scanner.FileGraph, err error, start time.Time) { + if err != nil || fg == nil || (len(snapshot.configured) > 0 && fg.Coverage.Status == analysis.CoverageUnavailable) { d.markGraphLifecycle(graphLifecycleFailed) if d.verbose { fmt.Printf("[watch] File graph unavailable: %v\n", err) } return } - if beforePublish != nil { - beforePublish() - } d.graph.mu.Lock() defer d.graph.mu.Unlock() @@ -419,13 +537,13 @@ func (d *Daemon) computeDepsWithBeforePublish(build func(context.Context, string configuredAfter = append(configuredAfter, file) } currentConfig := config.Load(d.root) - if d.graph.graphGeneration != generationBefore || - ConfiguredInventoryFingerprint(configuredBefore) != ConfiguredInventoryFingerprint(configuredAfter) || - graphFilterFingerprint(buildConfig) != graphFilterFingerprint(currentConfig) { + if d.graph.graphGeneration != snapshot.generation || + ConfiguredInventoryFingerprint(snapshot.configured) != ConfiguredInventoryFingerprint(configuredAfter) || + graphFilterFingerprint(snapshot.config) != graphFilterFingerprint(currentConfig) { d.markGraphLifecycleLocked(newGraphState(d.root, currentConfig, graphLifecycleStale, time.Time{}, nil)) return } - state := newGraphState(d.root, buildConfig, graphLifecycleAvailable, time.Now(), configuredBefore) + state := newGraphState(d.root, snapshot.config, graphLifecycleAvailable, time.Now(), snapshot.configured) // Convert FileGraph to DepContext map d.graph.DepCtx = make(map[string]*DepContext) diff --git a/watch/events.go b/watch/events.go index aae9320..d482bad 100644 --- a/watch/events.go +++ b/watch/events.go @@ -240,6 +240,10 @@ func (d *Daemon) eventLoop() { case <-controlTimerC: refreshConfigured() + case result := <-d.dependencyResults: + d.handleDependencyGraphResult(result) + armTimer(time.Now()) + case event, ok := <-d.watcher.Events: if !ok { return diff --git a/watch/graph_state_test.go b/watch/graph_state_test.go index e1e4dfd..45ed698 100644 --- a/watch/graph_state_test.go +++ b/watch/graph_state_test.go @@ -5,6 +5,7 @@ import ( "errors" "os" "path/filepath" + "sync/atomic" "testing" "time" @@ -103,6 +104,13 @@ func TestConfiguredEventRefreshesDependencyGraph(t *testing.T) { original := buildFileGraph t.Cleanup(func() { buildFileGraph = original }) + t.Cleanup(func() { + close(d.done) + if d.dependencyCancel != nil { + d.dependencyCancel() + } + d.dependencyWorkerWG.Wait() + }) calls := 0 buildFileGraph = func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error) { calls++ @@ -119,6 +127,8 @@ func TestConfiguredEventRefreshesDependencyGraph(t *testing.T) { t.Fatalf("after write graph state = %#v, want stale", d.graph.GraphState) } d.refreshDependencies() + result := <-d.dependencyResults + d.handleDependencyGraphResult(result) if calls != 1 { t.Fatalf("dependency builds = %d, want 1", calls) @@ -128,6 +138,131 @@ func TestConfiguredEventRefreshesDependencyGraph(t *testing.T) { } } +func TestDependencyRefreshBuildsOffEventLoop(t *testing.T) { + root := t.TempDir() + d := testGraphStateDaemon(root) + started := make(chan struct{}) + release := make(chan struct{}) + original := buildFileGraph + t.Cleanup(func() { buildFileGraph = original }) + t.Cleanup(func() { + close(d.done) + if d.dependencyCancel != nil { + d.dependencyCancel() + } + d.dependencyWorkerWG.Wait() + }) + buildFileGraph = func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error) { + close(started) + <-release + return &scanner.FileGraph{}, nil + } + + d.markGraphLifecycle(graphLifecycleStale) + finished := make(chan struct{}) + go func() { + d.refreshDependencies() + close(finished) + }() + select { + case <-finished: + case <-time.After(time.Second): + t.Fatal("refreshDependencies blocked on graph construction") + } + select { + case <-started: + case <-time.After(time.Second): + t.Fatal("dependency worker did not start") + } + close(release) +} + +func TestDependencyGraphBuildRecoversPanics(t *testing.T) { + graph, err := buildDependencyGraph(context.Background(), t.TempDir(), func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error) { + panic("test panic") + }) + if graph != nil || err == nil || err.Error() != "dependency graph build panicked: test panic" { + t.Fatalf("panic result = graph=%#v err=%v", graph, err) + } +} + +func TestDependencyRefreshRejectsResultAfterInvalidation(t *testing.T) { + root := t.TempDir() + d := testGraphStateDaemon(root) + started := make(chan struct{}) + release := make(chan struct{}) + original := buildFileGraph + t.Cleanup(func() { buildFileGraph = original }) + t.Cleanup(func() { + close(d.done) + if d.dependencyCancel != nil { + d.dependencyCancel() + } + d.dependencyWorkerWG.Wait() + }) + buildFileGraph = func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error) { + close(started) + <-release + return &scanner.FileGraph{Importers: map[string][]string{"dep.go": {"main.go"}}}, nil + } + + d.markGraphLifecycle(graphLifecycleStale) + d.refreshDependencies() + <-started + d.markGraphLifecycle(graphLifecycleStale) + close(release) + d.handleDependencyGraphResult(<-d.dependencyResults) + + d.graph.mu.RLock() + defer d.graph.mu.RUnlock() + if d.graph.HasDeps || d.graph.FileGraph != nil || d.graph.GraphState.Status != graphLifecycleStale { + t.Fatalf("stale worker result was applied: state=%#v hasDeps=%t graph=%#v", d.graph.GraphState, d.graph.HasDeps, d.graph.FileGraph) + } +} + +func TestDependencyRefreshCoalescesInvalidations(t *testing.T) { + root := t.TempDir() + d := testGraphStateDaemon(root) + var calls atomic.Int32 + secondStarted := make(chan struct{}) + release := make(chan struct{}) + original := buildFileGraph + t.Cleanup(func() { buildFileGraph = original }) + t.Cleanup(func() { + close(d.done) + if d.dependencyCancel != nil { + d.dependencyCancel() + } + d.dependencyWorkerWG.Wait() + }) + buildFileGraph = func(context.Context, string, scanner.Filters) (*scanner.FileGraph, error) { + if calls.Add(1) == 2 { + close(secondStarted) + } + if calls.Load() == 1 { + <-release + return nil, errors.New("first dependency build failed") + } + return &scanner.FileGraph{}, nil + } + + d.markGraphLifecycle(graphLifecycleStale) + d.refreshDependencies() + d.markGraphLifecycle(graphLifecycleStale) + d.refreshDependencies() + close(release) + d.handleDependencyGraphResult(<-d.dependencyResults) + select { + case <-secondStarted: + case <-time.After(time.Second): + t.Fatal("pending invalidation did not schedule a second build") + } + d.handleDependencyGraphResult(<-d.dependencyResults) + if got := calls.Load(); got != 2 { + t.Fatalf("dependency builds = %d, want 2", got) + } +} + func TestGraphFilterFingerprintIsDeterministic(t *testing.T) { left := config.ProjectConfig{Only: []string{"rust", "go"}, Exclude: []string{"vendor", "generated"}} right := config.ProjectConfig{Only: []string{"go", "rust"}, Exclude: []string{"generated", "vendor"}} @@ -278,6 +413,7 @@ func testGraphStateDaemon(root string) *Daemon { state := newGraphState(root, config.ProjectConfig{}, graphLifecycleAvailable, time.Now(), []string{"main.go"}) return &Daemon{ root: root, + done: make(chan struct{}), graph: &Graph{ Root: root, Files: map[string]*scanner.FileInfo{"main.go": {Path: "main.go", Ext: ".go"}},