diff --git a/cmd/analyze/analyze_test.go b/cmd/analyze/analyze_test.go index 538790081..0cb060e2a 100644 --- a/cmd/analyze/analyze_test.go +++ b/cmd/analyze/analyze_test.go @@ -6,6 +6,7 @@ import ( "context" "encoding/gob" "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -18,6 +19,10 @@ import ( tea "github.com/charmbracelet/bubbletea" ) +// Navigation starts a replacement scan immediately, so abandoned scan work +// must release its subprocesses and workers before it can compete for I/O. +const liveScanCancellationBudget = 250 * time.Millisecond + func resetOverviewSnapshotForTest() { overviewSnapshotMu.Lock() overviewSnapshotCache = nil @@ -76,8 +81,6 @@ func drainLiveScanToResultMsg(t *testing.T, start liveScanStartMsg) scanResultMs return scanResultMsg{path: start.path, result: event.result} case liveScanFailed: return scanResultMsg{path: start.path, err: event.err} - case liveScanCanceled: - return scanResultMsg{path: start.path, err: event.err} } case <-deadline: if start.cancel != nil { @@ -96,6 +99,50 @@ func cancelAndDrainLiveScan(start liveScanStartMsg) { } } +func installBlockingDuProbe(t *testing.T) string { + t.Helper() + + binDir := t.TempDir() + started := filepath.Join(binDir, "du-started") + duStub := filepath.Join(binDir, "du") + stub := "#!/bin/sh\n" + + "printf started > \"$MOLE_TEST_DU_STARTED\"\n" + + "exec /usr/bin/tail -f /dev/null\n" + if err := os.WriteFile(duStub, []byte(stub), 0o755); err != nil { + t.Fatalf("write du stub: %v", err) + } + t.Setenv("PATH", binDir+string(os.PathListSeparator)+os.Getenv("PATH")) + t.Setenv("MOLE_TEST_DU_STARTED", started) + return started +} + +func waitForTestPath(t *testing.T, path string) { + t.Helper() + + deadline := time.Now().Add(5 * time.Second) + for { + if _, err := os.Stat(path); err == nil { + return + } + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", path) + } + time.Sleep(5 * time.Millisecond) + } +} + +func waitForTestCondition(t *testing.T, description string, condition func() bool) { + t.Helper() + + deadline := time.Now().Add(5 * time.Second) + for !condition() { + if time.Now().After(deadline) { + t.Fatalf("timed out waiting for %s", description) + } + time.Sleep(5 * time.Millisecond) + } +} + func rowContaining(view, needle string) string { for line := range strings.SplitSeq(view, "\n") { if strings.Contains(line, needle) { @@ -140,7 +187,7 @@ func TestScanPathConcurrentBasic(t *testing.T) { current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current) + result, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current) if err != nil { t.Fatalf("scanPathConcurrent returned error: %v", err) } @@ -220,7 +267,7 @@ func TestScanPathConcurrentDedupsHardlinks(t *testing.T) { current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current) + result, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current) if err != nil { t.Fatalf("scanPathConcurrent returned error: %v", err) } @@ -1231,7 +1278,7 @@ func TestScanPathConcurrentWarmsChildDirectoryCache(t *testing.T) { current := &atomic.Value{} current.Store("") - if _, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current); err != nil { + if _, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current); err != nil { t.Fatalf("scanPathConcurrent(root): %v", err) } @@ -1276,7 +1323,7 @@ func TestScanPathConcurrentSkipsCacheForCheapSubdir(t *testing.T) { current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current) + result, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current) if err != nil { t.Fatalf("scanPathConcurrent(root): %v", err) } @@ -1320,7 +1367,7 @@ func TestAnalyzeIncludesParallelsVMStorageButKeepsOtherVirtualizationSkips(t *te var filesScanned, dirsScanned, bytesScanned int64 current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrentWithOptions(root, &filesScanned, &dirsScanned, &bytesScanned, current, false, 0) + result, err := scanPathConcurrentWithOptions(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current, false, 0) if err != nil { t.Fatalf("scan root: %v", err) } @@ -1418,7 +1465,7 @@ func TestScanPathConcurrentUsesChildCacheLargeFiles(t *testing.T) { var childFiles, childDirs, childBytes int64 childCurrent := &atomic.Value{} childCurrent.Store("") - childResult, err := scanPathConcurrent(child, &childFiles, &childDirs, &childBytes, childCurrent) + childResult, err := scanPathConcurrent(context.Background(), child, &childFiles, &childDirs, &childBytes, childCurrent) if err != nil { t.Fatalf("scanPathConcurrent(child): %v", err) } @@ -1437,7 +1484,7 @@ func TestScanPathConcurrentUsesChildCacheLargeFiles(t *testing.T) { current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current) + result, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current) if err != nil { t.Fatalf("scanPathConcurrent(root): %v", err) } @@ -1498,7 +1545,7 @@ func TestScanPathConcurrentWarmsChildCachesWithoutRecursiveSpotlight(t *testing. current := &atomic.Value{} current.Store("") - if _, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current); err != nil { + if _, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current); err != nil { t.Fatalf("scanPathConcurrent(root): %v", err) } @@ -1507,6 +1554,32 @@ func TestScanPathConcurrentWarmsChildCachesWithoutRecursiveSpotlight(t *testing. } } +func TestSpotlightConsumerStopsWhenScanIsCanceled(t *testing.T) { + root := t.TempDir() + file := filepath.Join(root, "large.bin") + if err := os.WriteFile(file, []byte("large"), 0o644); err != nil { + t.Fatalf("write spotlight result: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + originalRunner := spotlightQueryRunner + spotlightQueryRunner = func(_ context.Context, _, _ string) ([]byte, error) { + cancel() + return []byte(strings.Repeat(file+"\n", 10_000)), nil + } + t.Cleanup(func() { + spotlightQueryRunner = originalRunner + }) + + files, err := findLargeFilesWithSpotlight(ctx, root, 1) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled Spotlight consumer, got %v", err) + } + if len(files) != 0 { + t.Fatalf("canceled Spotlight consumer returned %d files", len(files)) + } +} + func TestScanCmdTreatsWarmedCacheAsStale(t *testing.T) { home := t.TempDir() t.Setenv("HOME", home) @@ -1522,7 +1595,8 @@ func TestScanCmdTreatsWarmedCacheAsStale(t *testing.T) { TotalSize: 42, TotalFiles: 1, } - if err := saveCacheToDiskWithOptions(target, result, true); err != nil { + ctx := context.Background() + if err := saveCacheToDiskWithOptions(newScanPublication(ctx, nil), target, result, true); err != nil { t.Fatalf("saveCacheToDiskWithOptions: %v", err) } @@ -1540,6 +1614,97 @@ func TestScanCmdTreatsWarmedCacheAsStale(t *testing.T) { } } +func TestCanceledCacheSaveDoesNotPublish(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + target := filepath.Join(home, "target") + if err := os.Mkdir(target, 0o755); err != nil { + t.Fatalf("create target: %v", err) + } + + ctx, cancelContext := context.WithCancel(context.Background()) + publication := newScanPublication(ctx, cancelContext) + publication.cancel() + err := saveCacheToDiskWithOptions(publication, target, scanResult{TotalSize: 42, TotalFiles: 1}, true) + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled cache save, got %v", err) + } + + cachePath, err := getCachePath(target) + if err != nil { + t.Fatalf("resolve cache path: %v", err) + } + if _, err := os.Stat(cachePath); !os.IsNotExist(err) { + t.Fatalf("canceled cache save published %s", cachePath) + } +} + +func TestCanceledCacheMutationsDoNotPublish(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + target := filepath.Join(home, "target") + if err := os.Mkdir(target, 0o755); err != nil { + t.Fatalf("create target: %v", err) + } + cachePath, err := getCachePath(target) + if err != nil { + t.Fatalf("resolve cache path: %v", err) + } + + ctx, cancelContext := context.WithCancel(context.Background()) + publication := newScanPublication(ctx, cancelContext) + publication.mu.Lock() + saveDone := make(chan error, 1) + go func() { + saveDone <- saveCacheToDiskWithOptions(publication, target, scanResult{TotalSize: 42, TotalFiles: 1}, true) + }() + waitForTestCondition(t, "cache save to reach its temporary file", func() bool { + matches, globErr := filepath.Glob(filepath.Join(filepath.Dir(cachePath), "entry-*.tmp")) + return globErr == nil && len(matches) > 0 + }) + cancelDone := make(chan struct{}) + go func() { + publication.cancel() + close(cancelDone) + }() + waitForTestCondition(t, "cache-save cancellation to start", publication.canceling.Load) + publication.mu.Unlock() + + if err := <-saveDone; !errors.Is(err, context.Canceled) { + t.Fatalf("expected cache save to lose publication order, got %v", err) + } + <-cancelDone + if _, err := os.Stat(cachePath); !os.IsNotExist(err) { + t.Fatalf("canceled cache save published %s", cachePath) + } + + if err := saveCacheToDisk(target, scanResult{TotalSize: 84, TotalFiles: 2}); err != nil { + t.Fatalf("seed cache entry: %v", err) + } + removeCtx, removeCancel := context.WithCancel(context.Background()) + removePublication := newScanPublication(removeCtx, removeCancel) + removePublication.mu.Lock() + removeDone := make(chan error, 1) + go func() { + removeDone <- removeCacheEntryForScan(removePublication, target) + }() + removeCancelDone := make(chan struct{}) + go func() { + removePublication.cancel() + close(removeCancelDone) + }() + waitForTestCondition(t, "cache-removal cancellation to start", removePublication.canceling.Load) + removePublication.mu.Unlock() + + if err := <-removeDone; !errors.Is(err, context.Canceled) { + t.Fatalf("expected cache removal to lose publication order, got %v", err) + } + <-removeCancelDone + if _, err := os.Stat(cachePath); err != nil { + t.Fatalf("canceled cache removal changed published state: %v", err) + } +} + func TestLiveScanSortConfigFromEnv(t *testing.T) { t.Run("defaults to freeze on move", func(t *testing.T) { t.Setenv(liveSortModeEnv, "") @@ -1603,6 +1768,221 @@ func TestLiveScanInitialListingShowsImmediateChildren(t *testing.T) { } } +func TestLiveScanCancellationStopsFoldedDirectoryProbe(t *testing.T) { + root := t.TempDir() + target := filepath.Join(root, "folded") + if err := os.Mkdir(target, 0o755); err != nil { + t.Fatalf("create folded directory: %v", err) + } + + started := installBlockingDuProbe(t) + + ctx, cancelContext := context.WithCancel(context.Background()) + publication := newScanPublication(ctx, cancelContext) + defer publication.cancel() + limiter := newScanLimiter(1) + largeFileMinSize := int64(largeFileWarmupMinSize) + var filesScanned, dirsScanned, bytesScanned int64 + currentPath := &atomic.Value{} + currentPath.Store("") + + done := make(chan error, 1) + go func() { + _, err := scanLiveTarget( + ctx, + liveScanTarget{name: "folded", path: target, kind: liveScanTargetFoldedDirectory}, + make(chan fileEntry, maxLargeFiles*2), + &largeFileMinSize, + limiter, + &filesScanned, + &dirsScanned, + &bytesScanned, + currentPath, + scanCacheBypass, + publication, + ) + done <- err + }() + + waitForTestPath(t, started) + + publication.cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled scan, got %v", err) + } + case <-time.After(liveScanCancellationBudget): + t.Fatal("folded-directory probe kept running after live scan cancellation") + } +} + +func TestLiveScanCancellationStopsNestedFoldedDirectoryProbe(t *testing.T) { + root := t.TempDir() + target := filepath.Join(root, "target") + if err := os.MkdirAll(filepath.Join(target, ".git"), 0o755); err != nil { + t.Fatalf("create nested folded directory: %v", err) + } + started := installBlockingDuProbe(t) + + ctx, cancelContext := context.WithCancel(context.Background()) + publication := newScanPublication(ctx, cancelContext) + defer publication.cancel() + limiter := newScanLimiter(1) + largeFileMinSize := int64(largeFileWarmupMinSize) + var filesScanned, dirsScanned, bytesScanned int64 + currentPath := &atomic.Value{} + currentPath.Store("") + + done := make(chan error, 1) + go func() { + _, err := scanLiveTarget( + ctx, + liveScanTarget{name: "target", path: target, kind: liveScanTargetDirectory}, + make(chan fileEntry, maxLargeFiles*2), + &largeFileMinSize, + limiter, + &filesScanned, + &dirsScanned, + &bytesScanned, + currentPath, + scanCacheBypass, + publication, + ) + done <- err + }() + + waitForTestPath(t, started) + publication.cancel() + select { + case err := <-done: + if !errors.Is(err, context.Canceled) { + t.Fatalf("expected canceled scan, got %v", err) + } + case <-time.After(liveScanCancellationBudget): + t.Fatal("nested folded-directory probe kept running after live scan cancellation") + } +} + +func TestLiveScanEventStreamRejectsCompletionAfterCancellation(t *testing.T) { + ctx, cancelContext := context.WithCancel(context.Background()) + stream := newLiveScanEventStream(newScanPublication(ctx, cancelContext), 1) + + stream.publishProgress(liveScanEventMsg{kind: liveScanChildProgress}) + stream.publishProgress(liveScanEventMsg{kind: liveScanChildProgress}) + stream.publish(liveScanEventMsg{kind: liveScanChildDone}) + + canceled := make(chan struct{}) + go func() { + stream.cancel() + close(canceled) + }() + select { + case <-canceled: + case <-time.After(5 * time.Second): + t.Fatal("cancel blocked behind queued live-scan events") + } + + stream.publish(liveScanEventMsg{kind: liveScanComplete}) + stream.close() + for event := range stream.events { + if event.kind == liveScanComplete { + t.Fatal("stream published completion after cancellation") + } + } +} + +func TestLiveScanEventStreamReservesRequiredCapacity(t *testing.T) { + ctx, cancelContext := context.WithCancel(context.Background()) + const targetCount = 3 + stream := newLiveScanEventStream(newScanPublication(ctx, cancelContext), targetCount) + + done := make(chan struct{}) + go func() { + for range cap(stream.events) - stream.requiredSlots { + stream.publishProgress(liveScanEventMsg{kind: liveScanChildProgress}) + } + for range targetCount { + stream.publish(liveScanEventMsg{kind: liveScanChildDone}) + } + stream.publish(liveScanEventMsg{kind: liveScanComplete}) + stream.close() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("required live-scan events exhausted their reserved capacity") + } + if !errors.Is(ctx.Err(), context.Canceled) { + t.Fatalf("closed stream did not release its context: %v", ctx.Err()) + } + + var required int + for event := range stream.events { + if event.kind != liveScanChildProgress { + required++ + } + } + if required != targetCount+1 { + t.Fatalf("got %d required events, want %d", required, targetCount+1) + } +} + +func TestCanceledLiveScanPublishesNoResultsOrCache(t *testing.T) { + home := t.TempDir() + t.Setenv("HOME", home) + + root := filepath.Join(home, "root") + project := filepath.Join(root, "project") + if err := os.MkdirAll(filepath.Join(project, "node_modules"), 0o755); err != nil { + t.Fatalf("create folded subtree: %v", err) + } + for i := range subdirCacheMinFiles { + path := filepath.Join(project, fmt.Sprintf("file-%03d.bin", i)) + if err := os.WriteFile(path, []byte("cacheable"), 0o644); err != nil { + t.Fatalf("write cacheable file: %v", err) + } + } + started := installBlockingDuProbe(t) + + var filesScanned, dirsScanned, bytesScanned int64 + currentPath := &atomic.Value{} + currentPath.Store("") + start, ok := startLiveScanCmd(root, &filesScanned, &dirsScanned, &bytesScanned, currentPath)().(liveScanStartMsg) + if !ok { + t.Fatal("expected live scan start message") + } + + waitForTestPath(t, started) + start.cancel() + + deadline := time.NewTimer(liveScanCancellationBudget) + defer deadline.Stop() + for { + select { + case event, open := <-start.events: + if !open { + cachePath, err := getCachePath(project) + if err != nil { + t.Fatalf("resolve project cache path: %v", err) + } + if _, err := os.Stat(cachePath); !os.IsNotExist(err) { + t.Fatalf("canceled scan persisted partial cache at %s", cachePath) + } + return + } + switch event.kind { + case liveScanChildDone, liveScanComplete: + t.Fatalf("canceled scan published stale event kind %v", event.kind) + } + case <-deadline.C: + t.Fatal("canceled live scan did not close promptly") + } + } +} + func TestLiveScanStartDoesNotAddSecondSpinnerTick(t *testing.T) { root := t.TempDir() child := filepath.Join(root, "child") @@ -1831,13 +2211,14 @@ func TestCacheBypassSkipsHomeLibraryOverviewSnapshot(t *testing.T) { scanTarget := func(policy scanCachePolicy) scanResult { t.Helper() + ctx := context.Background() var filesScanned, dirsScanned, bytesScanned int64 current := &atomic.Value{} current.Store("") limiter := newScanLimiter(1) largeFileMinSize := int64(largeFileWarmupMinSize) result, err := scanLiveTarget( - context.Background(), + ctx, liveScanTarget{name: "Library", path: library, kind: liveScanTargetHomeLibrary}, make(chan fileEntry, maxLargeFiles*2), &largeFileMinSize, @@ -1847,6 +2228,7 @@ func TestCacheBypassSkipsHomeLibraryOverviewSnapshot(t *testing.T) { &bytesScanned, current, policy, + newScanPublication(ctx, nil), ) if err != nil { t.Fatalf("scan Home Library: %v", err) @@ -1863,10 +2245,11 @@ func TestCacheBypassSkipsHomeLibraryOverviewSnapshot(t *testing.T) { scanHome := func(policy scanCachePolicy) int64 { t.Helper() + ctx := context.Background() var filesScanned, dirsScanned, bytesScanned int64 current := &atomic.Value{} current.Store("") - result, err := scanPathConcurrentWithLimiter(home, &filesScanned, &dirsScanned, &bytesScanned, current, false, maxEntries, nil, policy) + result, err := scanPathConcurrentWithLimiter(ctx, home, &filesScanned, &dirsScanned, &bytesScanned, current, false, maxEntries, nil, policy, newScanPublication(ctx, nil)) if err != nil { t.Fatalf("scan Home: %v", err) } @@ -2255,7 +2638,8 @@ func TestEnterSelectedDirRefreshesStaleInMemoryCache(t *testing.T) { TotalSize: 1, TotalFiles: 1, } - if err := saveCacheToDiskWithOptions(child, warmed, true); err != nil { + ctx := context.Background() + if err := saveCacheToDiskWithOptions(newScanPublication(ctx, nil), child, warmed, true); err != nil { t.Fatalf("saveCacheToDiskWithOptions: %v", err) } @@ -2321,7 +2705,8 @@ func TestGoBackRefreshesHistoryEntryNeedingRefresh(t *testing.T) { TotalSize: 2, TotalFiles: 1, } - if err := saveCacheToDiskWithOptions(child, warmed, true); err != nil { + ctx := context.Background() + if err := saveCacheToDiskWithOptions(newScanPublication(ctx, nil), child, warmed, true); err != nil { t.Fatalf("saveCacheToDiskWithOptions: %v", err) } @@ -2394,7 +2779,7 @@ func TestScanPathConcurrentWarmsChildCacheWithLiveProgress(t *testing.T) { done := make(chan struct{}) errCh := make(chan error, 1) go func() { - _, err := scanPathConcurrent(root, &filesScanned, &dirsScanned, &bytesScanned, current) + _, err := scanPathConcurrent(context.Background(), root, &filesScanned, &dirsScanned, &bytesScanned, current) errCh <- err close(done) }() @@ -2873,7 +3258,7 @@ func TestScanPathPermissionError(t *testing.T) { current.Store("") // Scanning the locked dir itself should fail. - _, err := scanPathConcurrent(lockedDir, &files, &dirs, &bytes, current) + _, err := scanPathConcurrent(context.Background(), lockedDir, &files, &dirs, &bytes, current) if err == nil { t.Fatalf("expected error scanning locked directory, got nil") } @@ -2903,7 +3288,7 @@ func TestCalculateDirSizeFastHighFanoutCompletes(t *testing.T) { done := make(chan int64, 1) go func() { - done <- calculateDirSizeFast(root, &files, &dirs, &bytes, current) + done <- calculateDirSizeFast(context.Background(), root, &files, &dirs, &bytes, current) }() select { diff --git a/cmd/analyze/cache.go b/cmd/analyze/cache.go index cb698ad26..9cac8283e 100644 --- a/cmd/analyze/cache.go +++ b/cmd/analyze/cache.go @@ -644,10 +644,14 @@ func loadStaleCacheFromDisk(path string) (*cacheEntry, error) { } func saveCacheToDisk(path string, result scanResult) error { - return saveCacheToDiskWithOptions(path, result, false) + ctx := context.Background() + return saveCacheToDiskWithOptions(newScanPublication(ctx, nil), path, result, false) } -func saveCacheToDiskWithOptions(path string, result scanResult, needsRefresh bool) error { +func saveCacheToDiskWithOptions(publication *scanPublication, path string, result scanResult, needsRefresh bool) error { + if err := publication.ctx.Err(); err != nil { + return err + } cachePath, err := getCachePath(path) if err != nil { return err @@ -691,13 +695,23 @@ func saveCacheToDiskWithOptions(path string, result scanResult, needsRefresh boo _ = os.Remove(tmpPath) return err } - if err := os.Rename(tmpPath, cachePath); err != nil { + err = publication.commit(func() error { + return os.Rename(tmpPath, cachePath) + }) + if err != nil { _ = os.Remove(tmpPath) return err } return nil } +func removeCacheEntryForScan(publication *scanPublication, path string) error { + return publication.commit(func() error { + removeCacheEntry(path) + return nil + }) +} + // peekCacheTotalFiles reads the total file count from cache, ignoring // expiration, for initial scan progress estimates. It shares // loadRawCacheFromDisk so a schema-stale or corrupt entry is rejected and diff --git a/cmd/analyze/json.go b/cmd/analyze/json.go index 60d593f3d..2f9d7c2ee 100644 --- a/cmd/analyze/json.go +++ b/cmd/analyze/json.go @@ -3,6 +3,7 @@ package main import ( + "context" "encoding/json" "fmt" "os" @@ -60,7 +61,7 @@ func performDirectoryScanForJSON(path string) jsonOutput { currentPath := &atomic.Value{} currentPath.Store("") - result, err := scanPathConcurrentAllEntries(path, &filesScanned, &dirsScanned, &bytesScanned, currentPath) + result, err := scanPathConcurrentAllEntries(context.Background(), path, &filesScanned, &dirsScanned, &bytesScanned, currentPath) if err != nil { fmt.Fprintf(os.Stderr, "failed to scan directory: %v\n", err) os.Exit(1) diff --git a/cmd/analyze/live_scan.go b/cmd/analyze/live_scan.go index ba8f88643..90f5ec850 100644 --- a/cmd/analyze/live_scan.go +++ b/cmd/analyze/live_scan.go @@ -34,6 +34,60 @@ type liveScanTarget struct { kind liveScanTargetKind } +// liveScanEventStream uses the scan publication boundary for event delivery. +// Progress is lossy and may use only the non-reserved portion of the buffer; +// one slot per target plus the final completion slot stays available. +type liveScanEventStream struct { + publication *scanPublication + events chan liveScanEventMsg + requiredSlots int + closed bool +} + +func newLiveScanEventStream(publication *scanPublication, targetCount int) *liveScanEventStream { + return &liveScanEventStream{ + publication: publication, + events: make(chan liveScanEventMsg, max(targetCount*4, 1)), + requiredSlots: targetCount + 1, + } +} + +func (s *liveScanEventStream) cancel() { + s.publication.cancel() +} + +func (s *liveScanEventStream) publish(msg liveScanEventMsg) { + _ = s.publication.commit(func() error { + if s.closed { + return nil + } + // Progress publication reserves enough capacity that every target can + // emit one result or failure and the coordinator can emit completion. + s.events <- msg + return nil + }) +} + +func (s *liveScanEventStream) publishProgress(msg liveScanEventMsg) { + _ = s.publication.commit(func() error { + if s.closed || len(s.events) >= cap(s.events)-s.requiredSlots { + return nil + } + s.events <- msg + return nil + }) +} + +func (s *liveScanEventStream) close() { + s.publication.finish(func() { + if s.closed { + return + } + s.closed = true + close(s.events) + }) +} + func startLiveScanCmd(path string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) tea.Cmd { return startLiveScanCmdWithPolicy(path, filesScanned, dirsScanned, bytesScanned, currentPath, scanCacheReuse) } @@ -41,12 +95,12 @@ func startLiveScanCmd(path string, filesScanned, dirsScanned, bytesScanned *int6 func startLiveScanCmdWithPolicy(path string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, cachePolicy scanCachePolicy) tea.Cmd { return func() tea.Msg { id := nextLiveScanID.Add(1) - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancelContext := context.WithCancel(context.Background()) limiter := newScanLimiter(0) entries, targets, totalSize, totalFiles, largeFiles, err := readLiveScanInitialEntries(path, limiter) if err != nil { - cancel() + cancelContext() return liveScanStartMsg{id: id, path: path, err: err} } @@ -57,8 +111,9 @@ func startLiveScanCmdWithPolicy(path string, filesScanned, dirsScanned, bytesSca atomic.AddInt64(bytesScanned, totalSize) } - events := make(chan liveScanEventMsg, max(len(targets)*4, 1)) - go runLiveScan(ctx, id, path, entries, targets, totalSize, totalFiles, largeFiles, limiter, filesScanned, dirsScanned, bytesScanned, currentPath, events, cachePolicy) + publication := newScanPublication(ctx, cancelContext) + stream := newLiveScanEventStream(publication, len(targets)) + go runLiveScan(ctx, id, path, entries, targets, totalSize, totalFiles, largeFiles, limiter, filesScanned, dirsScanned, bytesScanned, currentPath, stream, cachePolicy) scanningPaths := make([]string, 0, len(targets)) for _, target := range targets { @@ -73,8 +128,8 @@ func startLiveScanCmdWithPolicy(path string, filesScanned, dirsScanned, bytesSca totalFiles: totalFiles, largeFiles: largeFiles, scanningPaths: scanningPaths, - events: events, - cancel: cancel, + events: stream.events, + cancel: stream.cancel, } } } @@ -188,10 +243,10 @@ func runLiveScan( limiter *scanLimiter, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, - events chan<- liveScanEventMsg, + stream *liveScanEventStream, cachePolicy scanCachePolicy, ) { - defer close(events) + defer stream.close() entriesByPath := make(map[string]dirEntry, len(initialEntries)) for _, entry := range initialEntries { @@ -219,9 +274,9 @@ func runLiveScan( target := target scanTarget := func() { defer wg.Done() - result, err := scanLiveTargetWithProgress(ctx, id, root, target, largeFileChan, &largeFileMinSize, limiter, currentPath, events, cachePolicy) + result, err := scanLiveTargetWithProgress(ctx, id, root, target, largeFileChan, &largeFileMinSize, limiter, currentPath, stream, cachePolicy) if err != nil && !errors.Is(err, context.Canceled) { - sendLiveScanEvent(ctx, events, liveScanEventMsg{id: id, path: root, kind: liveScanFailed, entry: dirEntry{Name: target.name, Path: target.path, IsDir: true}, err: err}) + stream.publish(liveScanEventMsg{id: id, path: root, kind: liveScanFailed, entry: dirEntry{Name: target.name, Path: target.path, IsDir: true}, err: err}) return } if ctx.Err() != nil { @@ -253,7 +308,7 @@ func runLiveScan( atomic.AddInt64(bytesScanned, result.TotalSize) } - sendLiveScanEvent(ctx, events, liveScanEventMsg{ + stream.publish(liveScanEventMsg{ id: id, path: root, kind: liveScanChildDone, @@ -278,7 +333,6 @@ func runLiveScan( largeFiles := <-largeFilesDone if ctx.Err() != nil { - sendLiveScanEvent(context.Background(), events, liveScanEventMsg{id: id, path: root, kind: liveScanCanceled, err: ctx.Err()}) return } @@ -301,10 +355,10 @@ func runLiveScan( dedupedHardlink: dedupedHardlink.Load(), } - sendLiveScanEvent(ctx, events, liveScanEventMsg{id: id, path: root, kind: liveScanComplete, result: result}) + stream.publish(liveScanEventMsg{id: id, path: root, kind: liveScanComplete, result: result}) } -func scanLiveTargetWithProgress(ctx context.Context, id int64, root string, target liveScanTarget, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, currentPath *atomic.Value, events chan<- liveScanEventMsg, cachePolicy scanCachePolicy) (scanResult, error) { +func scanLiveTargetWithProgress(ctx context.Context, id int64, root string, target liveScanTarget, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, currentPath *atomic.Value, stream *liveScanEventStream, cachePolicy scanCachePolicy) (scanResult, error) { var filesScanned int64 var dirsScanned int64 var bytesScanned int64 @@ -336,7 +390,7 @@ func scanLiveTargetWithProgress(ctx context.Context, id int64, root string, targ currentPath.Store(path) } } - sendLiveScanProgress(ctx, events, liveScanEventMsg{ + stream.publishProgress(liveScanEventMsg{ id: id, path: root, kind: liveScanChildProgress, @@ -351,7 +405,7 @@ func scanLiveTargetWithProgress(ctx context.Context, id int64, root string, targ } }() - result, err := scanLiveTarget(ctx, target, largeFileChan, largeFileMinSize, limiter, &filesScanned, &dirsScanned, &bytesScanned, localCurrentPath, cachePolicy) + result, err := scanLiveTarget(ctx, target, largeFileChan, largeFileMinSize, limiter, &filesScanned, &dirsScanned, &bytesScanned, localCurrentPath, cachePolicy, stream.publication) close(done) <-progressDone if result.TotalFiles == 0 { @@ -363,7 +417,7 @@ func scanLiveTargetWithProgress(ctx context.Context, id int64, root string, targ return result, err } -func scanLiveTarget(ctx context.Context, target liveScanTarget, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, cachePolicy scanCachePolicy) (scanResult, error) { +func scanLiveTarget(ctx context.Context, target liveScanTarget, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, cachePolicy scanCachePolicy, publication *scanPublication) (scanResult, error) { if err := ctx.Err(); err != nil { return scanResult{}, err } @@ -376,12 +430,18 @@ func scanLiveTarget(ctx context.Context, target liveScanTarget, largeFileChan ch } } case liveScanTargetFoldedDirectory: - size, err := getDirectorySizeFromDu(target.path) + size, err := getDirectorySizeFromDu(ctx, target.path) + if ctx.Err() != nil { + return scanResult{}, ctx.Err() + } if err != nil || size <= 0 { - size = calculateDirSizeFastWithLimiter(target.path, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) + size = calculateDirSizeFastWithLimiter(ctx, target.path, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) } else { atomic.AddInt64(bytesScanned, size) } + if ctx.Err() != nil { + return scanResult{}, ctx.Err() + } return scanResult{TotalSize: size}, nil } @@ -389,7 +449,7 @@ func scanLiveTarget(ctx context.Context, target liveScanTarget, largeFileChan ch return scanResult{}, err } - result := scanSubdirWithCache(target.path, largeFileChan, largeFileMinSize, limiter, limiter.dirSem, limiter.duSem, limiter.duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy) + result := scanSubdirWithCache(ctx, target.path, largeFileChan, largeFileMinSize, limiter, limiter.dirSem, limiter.duSem, limiter.duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy, publication) return result, ctx.Err() } @@ -424,21 +484,6 @@ func pushLiveLargeFile(h *largeFileHeap, file fileEntry, largeFileMinSize *int64 } } -func sendLiveScanEvent(ctx context.Context, events chan<- liveScanEventMsg, msg liveScanEventMsg) { - select { - case <-ctx.Done(): - case events <- msg: - } -} - -func sendLiveScanProgress(ctx context.Context, events chan<- liveScanEventMsg, msg liveScanEventMsg) { - select { - case <-ctx.Done(): - case events <- msg: - default: - } -} - func waitLiveScanEventCmd(events <-chan liveScanEventMsg) tea.Cmd { return func() tea.Msg { msg, ok := <-events diff --git a/cmd/analyze/model.go b/cmd/analyze/model.go index 6911574d9..227c5cbf1 100644 --- a/cmd/analyze/model.go +++ b/cmd/analyze/model.go @@ -91,7 +91,6 @@ const ( liveScanChildDone liveScanComplete liveScanFailed - liveScanCanceled ) type liveScanEventMsg struct { diff --git a/cmd/analyze/scanner.go b/cmd/analyze/scanner.go index 983041cf1..d8a7d4090 100644 --- a/cmd/analyze/scanner.go +++ b/cmd/analyze/scanner.go @@ -26,6 +26,61 @@ var spotlightQueryRunner = func(ctx context.Context, root, query string) ([]byte return exec.CommandContext(ctx, "mdfind", "-onlyin", root, query).Output() } +// scanPublication gives cancellation a linearizable boundary with externally +// visible scan side effects. A publication either completes before cancel +// returns, or observes the canceled scan and is rejected. +type scanPublication struct { + ctx context.Context + cancelContext context.CancelFunc + + mu sync.Mutex + canceling atomic.Bool + canceled bool +} + +func newScanPublication(ctx context.Context, cancel context.CancelFunc) *scanPublication { + return &scanPublication{ctx: ctx, cancelContext: cancel} +} + +func (p *scanPublication) cancel() { + p.canceling.Store(true) + p.mu.Lock() + defer p.mu.Unlock() + if p.canceled { + return + } + p.canceled = true + if p.cancelContext != nil { + p.cancelContext() + } +} + +func (p *scanPublication) commit(action func() error) error { + if p.canceling.Load() { + return context.Canceled + } + p.mu.Lock() + defer p.mu.Unlock() + if p.canceling.Load() || p.canceled { + return context.Canceled + } + if err := p.ctx.Err(); err != nil { + return err + } + return action() +} + +func (p *scanPublication) finish(action func()) { + p.canceling.Store(true) + p.mu.Lock() + defer p.mu.Unlock() + p.canceled = true + if p.cancelContext != nil { + p.cancelContext() + } + action() +} + // scanLimiter bundles the concurrency budgets used by a single scan pass. // // There are five separate semaphores on purpose: each protects a different @@ -99,9 +154,14 @@ func (l *scanLimiter) releaseEntry() { // trySend attempts to send an item to a channel with a timeout. // Returns true if the item was sent, false if the timeout was reached. -func trySend[T any](ch chan<- T, item T, timeout time.Duration) bool { +func trySend[T any](ctx context.Context, ch chan<- T, item T, timeout time.Duration) bool { + if ctx.Err() != nil { + return false + } if timeout <= 0 { select { + case <-ctx.Done(): + return false case ch <- item: return true default: @@ -110,6 +170,8 @@ func trySend[T any](ch chan<- T, item T, timeout time.Duration) bool { } select { + case <-ctx.Done(): + return false case ch <- item: return true default: @@ -126,6 +188,8 @@ func trySend[T any](ch chan<- T, item T, timeout time.Duration) bool { }() select { + case <-ctx.Done(): + return false case ch <- item: return true case <-timer.C: @@ -133,16 +197,25 @@ func trySend[T any](ch chan<- T, item T, timeout time.Duration) bool { } } -func scanPathConcurrent(root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) (scanResult, error) { - return scanPathConcurrentWithOptions(root, filesScanned, dirsScanned, bytesScanned, currentPath, true, maxEntries) +func acquireScanPermit(ctx context.Context, sem chan struct{}) error { + select { + case sem <- struct{}{}: + return nil + case <-ctx.Done(): + return ctx.Err() + } } -func scanPathConcurrentAllEntries(root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) (scanResult, error) { - return scanPathConcurrentWithOptions(root, filesScanned, dirsScanned, bytesScanned, currentPath, true, 0) +func scanPathConcurrent(ctx context.Context, root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) (scanResult, error) { + return scanPathConcurrentWithOptions(ctx, root, filesScanned, dirsScanned, bytesScanned, currentPath, true, maxEntries) } -func scanPathConcurrentWithOptions(root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, useSpotlight bool, entryLimit int) (scanResult, error) { - return scanPathConcurrentWithLimiter(root, filesScanned, dirsScanned, bytesScanned, currentPath, useSpotlight, entryLimit, nil, scanCacheReuse) +func scanPathConcurrentAllEntries(ctx context.Context, root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) (scanResult, error) { + return scanPathConcurrentWithOptions(ctx, root, filesScanned, dirsScanned, bytesScanned, currentPath, true, 0) +} + +func scanPathConcurrentWithOptions(ctx context.Context, root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, useSpotlight bool, entryLimit int) (scanResult, error) { + return scanPathConcurrentWithLimiter(ctx, root, filesScanned, dirsScanned, bytesScanned, currentPath, useSpotlight, entryLimit, nil, scanCacheReuse, newScanPublication(ctx, nil)) } type scanCachePolicy uint8 @@ -152,11 +225,17 @@ const ( scanCacheBypass ) -func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, useSpotlight bool, entryLimit int, limiter *scanLimiter, cachePolicy scanCachePolicy) (scanResult, error) { +func scanPathConcurrentWithLimiter(ctx context.Context, root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, useSpotlight bool, entryLimit int, limiter *scanLimiter, cachePolicy scanCachePolicy, publication *scanPublication) (scanResult, error) { + if err := ctx.Err(); err != nil { + return scanResult{}, err + } children, err := os.ReadDir(root) if err != nil { return scanResult{}, err } + if err := ctx.Err(); err != nil { + return scanResult{}, err + } if limiter == nil { limiter = newScanLimiter(len(children)) } @@ -226,7 +305,11 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes home := os.Getenv("HOME") isHomeDir := home != "" && root == home +scanChildren: for _, child := range children { + if ctx.Err() != nil { + break + } fullPath := filepath.Join(root, child.Name()) // Skip symlinks to avoid following unexpected targets. @@ -245,7 +328,7 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes size := getActualFileSize(fullPath, info) atomic.AddInt64(&total, size) - trySend(entryChan, dirEntry{ + trySend(ctx, entryChan, dirEntry{ Name: child.Name() + " →", Path: fullPath, Size: size, @@ -269,6 +352,9 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes // ~/Library is scanned separately; reuse cache when possible. if isHomeDir && child.Name() == "Library" { processDir := func(name, path string) { + if ctx.Err() != nil { + return + } result := scanResult{} if cachePolicy == scanCacheReuse { if cached, err := loadStoredOverviewSize(path); err == nil && cached > 0 { @@ -276,7 +362,10 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes } } if result.TotalSize <= 0 { - result = scanSubdirWithCache(path, largeFileChan, &largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy) + result = scanSubdirWithCache(ctx, path, largeFileChan, &largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy, publication) + } + if ctx.Err() != nil { + return } atomic.AddInt64(&total, result.TotalSize) if result.TotalFiles > 0 { @@ -287,7 +376,7 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes } atomic.AddInt64(dirsScanned, 1) - trySend(entryChan, dirEntry{ + trySend(ctx, entryChan, dirEntry{ Name: name, Path: path, Size: result.TotalSize, @@ -308,22 +397,35 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes // Folded dirs: fast size without expanding. if shouldFoldDirWithPath(child.Name(), fullPath) { - duQueueSem <- struct{}{} + if acquireScanPermit(ctx, duQueueSem) != nil { + break scanChildren + } wg.Go(func() { defer func() { <-duQueueSem }() + if ctx.Err() != nil { + return + } size, err := func() (int64, error) { - duSem <- struct{}{} + if err := acquireScanPermit(ctx, duSem); err != nil { + return 0, err + } defer func() { <-duSem }() - return getDirectorySizeFromDu(fullPath) + return getDirectorySizeFromDu(ctx, fullPath) }() + if ctx.Err() != nil { + return + } if err != nil || size <= 0 { - size = calculateDirSizeFastWithLimiter(fullPath, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) + size = calculateDirSizeFastWithLimiter(ctx, fullPath, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) + } + if ctx.Err() != nil { + return } atomic.AddInt64(&total, size) atomic.AddInt64(dirsScanned, 1) - trySend(entryChan, dirEntry{ + trySend(ctx, entryChan, dirEntry{ Name: child.Name(), Path: fullPath, Size: size, @@ -335,7 +437,13 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes } processDir := func(name, path string) { - result := scanSubdirWithCache(path, largeFileChan, &largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy) + if ctx.Err() != nil { + return + } + result := scanSubdirWithCache(ctx, path, largeFileChan, &largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath, cachePolicy, publication) + if ctx.Err() != nil { + return + } atomic.AddInt64(&total, result.TotalSize) if result.TotalFiles > 0 { subtreeFilesScanned.Add(result.TotalFiles) @@ -345,7 +453,7 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes } atomic.AddInt64(dirsScanned, 1) - trySend(entryChan, dirEntry{ + trySend(ctx, entryChan, dirEntry{ Name: name, Path: path, Size: result.TotalSize, @@ -377,7 +485,7 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes localFilesScanned++ localBytesScanned += size - trySend(entryChan, dirEntry{ + trySend(ctx, entryChan, dirEntry{ Name: child.Name(), Path: fullPath, Size: size, @@ -389,7 +497,7 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes if !shouldSkipFileForLargeTracking(fullPath) { minSize := atomic.LoadInt64(&largeFileMinSize) if size >= minSize { - trySend(largeFileChan, fileEntry{Name: child.Name(), Path: fullPath, Size: size}, scanSendTimeout) + trySend(ctx, largeFileChan, fileEntry{Name: child.Name(), Path: fullPath, Size: size}, scanSendTimeout) } } } @@ -407,6 +515,9 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes close(entryChan) close(largeFileChan) collectorWg.Wait() + if err := ctx.Err(); err != nil { + return scanResult{}, err + } // Convert heaps to sorted slices (descending). var entries []dirEntry @@ -429,7 +540,11 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes // Use Spotlight for large files when it expands the list. if useSpotlight { - if spotlightFiles := findLargeFilesWithSpotlight(root, spotlightMinFileSize); len(spotlightFiles) > len(largeFiles) { + spotlightFiles, _ := findLargeFilesWithSpotlight(ctx, root, spotlightMinFileSize) + if err := ctx.Err(); err != nil { + return scanResult{}, err + } + if len(spotlightFiles) > len(largeFiles) { largeFiles = spotlightFiles } } @@ -443,13 +558,18 @@ func scanPathConcurrentWithLimiter(root string, filesScanned, dirsScanned, bytes }, nil } -func publishLargeFiles(files []fileEntry, largeFileChan chan<- fileEntry) { +func publishLargeFiles(ctx context.Context, files []fileEntry, largeFileChan chan<- fileEntry) { for _, file := range files { - trySend(largeFileChan, file, scanSendTimeout) + if !trySend(ctx, largeFileChan, file, scanSendTimeout) && ctx.Err() != nil { + return + } } } -func loadCachedSubdirResult(path string, largeFileChan chan<- fileEntry) (scanResult, bool) { +func loadCachedSubdirResult(ctx context.Context, path string, largeFileChan chan<- fileEntry) (scanResult, bool) { + if ctx.Err() != nil { + return scanResult{}, false + } cached, err := loadCacheFromDisk(path) if err != nil { return scanResult{}, false @@ -461,13 +581,19 @@ func loadCachedSubdirResult(path string, largeFileChan chan<- fileEntry) (scanRe TotalSize: cached.TotalSize, TotalFiles: cached.TotalFiles, } - publishLargeFiles(result.LargeFiles, largeFileChan) + publishLargeFiles(ctx, result.LargeFiles, largeFileChan) return result, true } -func scanSubdirWithCache(root string, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, dirSem, duSem, duQueueSem chan struct{}, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, cachePolicy scanCachePolicy) scanResult { +func scanSubdirWithCache(ctx context.Context, root string, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, dirSem, duSem, duQueueSem chan struct{}, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value, cachePolicy scanCachePolicy, publication *scanPublication) scanResult { + if ctx.Err() != nil { + return scanResult{} + } if cachePolicy == scanCacheReuse { - if cached, ok := loadCachedSubdirResult(root, largeFileChan); ok { + if cached, ok := loadCachedSubdirResult(ctx, root, largeFileChan); ok { + if ctx.Err() != nil { + return scanResult{} + } if cached.TotalFiles > 0 { atomic.AddInt64(filesScanned, cached.TotalFiles) } @@ -478,21 +604,30 @@ func scanSubdirWithCache(root string, largeFileChan chan<- fileEntry, largeFileM } } - result, err := scanPathConcurrentWithLimiter(root, filesScanned, dirsScanned, bytesScanned, currentPath, false, maxEntries, limiter, cachePolicy) + result, err := scanPathConcurrentWithLimiter(ctx, root, filesScanned, dirsScanned, bytesScanned, currentPath, false, maxEntries, limiter, cachePolicy, publication) if err == nil { - publishLargeFiles(result.LargeFiles, largeFileChan) + if ctx.Err() != nil { + return scanResult{} + } + publishLargeFiles(ctx, result.LargeFiles, largeFileChan) + if ctx.Err() != nil { + return scanResult{} + } // A subtree whose size depended on hardlink dedup is scan-order // dependent; caching it would poison standalone re-scans. Cheap // subtrees are not persisted at all: see shouldPersistSubdirCache. if !result.dedupedHardlink && shouldPersistSubdirCache(result) { - _ = saveCacheToDiskWithOptions(root, result, true) + _ = saveCacheToDiskWithOptions(publication, root, result, true) } else if cachePolicy == scanCacheBypass { - removeCacheEntry(root) + _ = removeCacheEntryForScan(publication, root) } return result } + if ctx.Err() != nil { + return scanResult{} + } - return scanResult{TotalSize: calculateDirSizeConcurrent(root, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath)} + return scanResult{TotalSize: calculateDirSizeConcurrent(ctx, root, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath)} } func shouldFoldDirWithPath(name, path string) bool { @@ -520,15 +655,15 @@ func shouldSkipFileForLargeTracking(path string) bool { } // calculateDirSizeFast performs concurrent dir sizing using os.ReadDir. -func calculateDirSizeFast(root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { - return calculateDirSizeFastWithLimiter(root, newScanLimiter(0), filesScanned, dirsScanned, bytesScanned, currentPath) +func calculateDirSizeFast(ctx context.Context, root string, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { + return calculateDirSizeFastWithLimiter(ctx, root, newScanLimiter(0), filesScanned, dirsScanned, bytesScanned, currentPath) } -func calculateDirSizeFastWithLimiter(root string, limiter *scanLimiter, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { +func calculateDirSizeFastWithLimiter(ctx context.Context, root string, limiter *scanLimiter, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { var total atomic.Int64 var wg sync.WaitGroup - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + ctx, cancel := context.WithTimeout(ctx, 5*time.Minute) defer cancel() concurrency := min(runtime.NumCPU()*cpuMultiplier, maxWorkers) @@ -557,6 +692,9 @@ func calculateDirSizeFastWithLimiter(root string, limiter *scanLimiter, filesSca var localBytes, localFiles int64 for _, entry := range entries { + if ctx.Err() != nil { + return + } if entry.IsDir() { subDir := filepath.Join(dirPath, entry.Name()) atomic.AddInt64(dirsScanned, 1) @@ -597,31 +735,34 @@ func calculateDirSizeFastWithLimiter(root string, limiter *scanLimiter, filesSca } // Use Spotlight (mdfind) to quickly find large files. -func findLargeFilesWithSpotlight(root string, minSize int64) []fileEntry { +func findLargeFilesWithSpotlight(ctx context.Context, root string, minSize int64) ([]fileEntry, error) { // Validate root path. if err := validatePath(root); err != nil { - return nil + return nil, nil } // Validate minSize is reasonable (non-negative and not excessively large). if minSize < 0 || minSize > 1<<50 { // 1 PB max - return nil + return nil, nil } query := fmt.Sprintf("kMDItemFSSize >= %d", minSize) - ctx, cancel := context.WithTimeout(context.Background(), mdlsTimeout) + ctx, cancel := context.WithTimeout(ctx, mdlsTimeout) defer cancel() output, err := spotlightQueryRunner(ctx, root, query) if err != nil { - return nil + return nil, err } h := &largeFileHeap{} heap.Init(h) for line := range strings.Lines(strings.TrimSpace(string(output))) { + if err := ctx.Err(); err != nil { + return nil, err + } if line == "" { continue } @@ -666,7 +807,10 @@ func findLargeFilesWithSpotlight(root string, minSize int64) []fileEntry { files[i] = heap.Pop(h).(fileEntry) } - return files + if err := ctx.Err(); err != nil { + return nil, err + } + return files, nil } // isInFoldedDir checks if a path is inside a folded directory. @@ -680,7 +824,10 @@ func isInFoldedDir(path string) bool { return false } -func calculateDirSizeConcurrent(root string, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, dirSem, duSem, duQueueSem chan struct{}, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { +func calculateDirSizeConcurrent(ctx context.Context, root string, largeFileChan chan<- fileEntry, largeFileMinSize *int64, limiter *scanLimiter, dirSem, duSem, duQueueSem chan struct{}, filesScanned, dirsScanned, bytesScanned *int64, currentPath *atomic.Value) int64 { + if ctx.Err() != nil { + return 0 + } children, err := os.ReadDir(root) if err != nil { return 0 @@ -693,7 +840,11 @@ func calculateDirSizeConcurrent(root string, largeFileChan chan<- fileEntry, lar var localBytesScanned int64 var wg sync.WaitGroup +scanChildren: for _, child := range children { + if ctx.Err() != nil { + break + } fullPath := filepath.Join(root, child.Name()) if child.Type()&fs.ModeSymlink != 0 { @@ -712,20 +863,33 @@ func calculateDirSizeConcurrent(root string, largeFileChan chan<- fileEntry, lar localDirsScanned++ if shouldFoldDirWithPath(child.Name(), fullPath) { - duQueueSem <- struct{}{} + if acquireScanPermit(ctx, duQueueSem) != nil { + break scanChildren + } wg.Go(func() { defer func() { <-duQueueSem }() + if ctx.Err() != nil { + return + } size, err := func() (int64, error) { - duSem <- struct{}{} + if err := acquireScanPermit(ctx, duSem); err != nil { + return 0, err + } defer func() { <-duSem }() - return getDirectorySizeFromDu(fullPath) + return getDirectorySizeFromDu(ctx, fullPath) }() + if ctx.Err() != nil { + return + } if err != nil || size <= 0 { - size = calculateDirSizeFastWithLimiter(fullPath, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) + size = calculateDirSizeFastWithLimiter(ctx, fullPath, limiter, filesScanned, dirsScanned, bytesScanned, currentPath) } else { atomic.AddInt64(bytesScanned, size) } + if ctx.Err() != nil { + return + } total.Add(size) }) continue @@ -736,11 +900,13 @@ func calculateDirSizeConcurrent(root string, largeFileChan chan<- fileEntry, lar wg.Go(func() { defer func() { <-dirSem }() - size := calculateDirSizeConcurrent(fullPath, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath) + size := calculateDirSizeConcurrent(ctx, fullPath, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath) total.Add(size) }) + case <-ctx.Done(): + break scanChildren default: - size := calculateDirSizeConcurrent(fullPath, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath) + size := calculateDirSizeConcurrent(ctx, fullPath, largeFileChan, largeFileMinSize, limiter, dirSem, duSem, duQueueSem, filesScanned, dirsScanned, bytesScanned, currentPath) localTotal += size } continue @@ -759,7 +925,7 @@ func calculateDirSizeConcurrent(root string, largeFileChan chan<- fileEntry, lar if !shouldSkipFileForLargeTracking(fullPath) && largeFileMinSize != nil { minSize := atomic.LoadInt64(largeFileMinSize) if size >= minSize { - trySend(largeFileChan, fileEntry{Name: child.Name(), Path: fullPath, Size: size}, scanSendTimeout) + trySend(ctx, largeFileChan, fileEntry{Name: child.Name(), Path: fullPath, Size: size}, scanSendTimeout) } } @@ -811,7 +977,7 @@ func measureOverviewSize(path string) (int64, error) { excludePath = filepath.Join(home, "Library") } - if duSize, err := getDirectorySizeFromDuWithExcludeAndIgnores(path, excludePath, overviewIgnoreNamesForPath(path)); err == nil { + if duSize, err := getDirectorySizeFromDuWithExcludeAndIgnores(context.Background(), path, excludePath, overviewIgnoreNamesForPath(path)); err == nil { _ = storeOverviewSize(path, duSize) return duSize, nil } @@ -829,15 +995,15 @@ func measureOverviewSize(path string) (int64, error) { return 0, fmt.Errorf("unable to measure directory size with fast methods") } -func getDirectorySizeFromDu(path string) (int64, error) { - return getDirectorySizeFromDuWithExclude(path, "") +func getDirectorySizeFromDu(ctx context.Context, path string) (int64, error) { + return getDirectorySizeFromDuWithExclude(ctx, path, "") } -func getDirectorySizeFromDuWithExclude(path string, excludePath string) (int64, error) { - return getDirectorySizeFromDuWithExcludeAndIgnores(path, excludePath, nil) +func getDirectorySizeFromDuWithExclude(ctx context.Context, path string, excludePath string) (int64, error) { + return getDirectorySizeFromDuWithExcludeAndIgnores(ctx, path, excludePath, nil) } -func getDirectorySizeFromDuWithExcludeAndIgnores(path string, excludePath string, ignoreNames []string) (int64, error) { +func getDirectorySizeFromDuWithExcludeAndIgnores(ctx context.Context, path string, excludePath string, ignoreNames []string) (int64, error) { // Validate paths. if err := validatePath(path); err != nil { return 0, err @@ -858,7 +1024,7 @@ func getDirectorySizeFromDuWithExcludeAndIgnores(path string, excludePath string return 0, err } - ctx, cancel := context.WithTimeout(context.Background(), duTimeout) + ctx, cancel := context.WithTimeout(ctx, duTimeout) defer cancel() args := []string{"-skPx"} diff --git a/cmd/analyze/scanner_test.go b/cmd/analyze/scanner_test.go index 8ebe372e3..7cedbe3e9 100644 --- a/cmd/analyze/scanner_test.go +++ b/cmd/analyze/scanner_test.go @@ -3,6 +3,7 @@ package main import ( + "context" "fmt" "os" "path/filepath" @@ -79,11 +80,11 @@ func TestGetDirectorySizeFromDuWithIgnoresSkipsCloudPlaceholderTree(t *testing.T writeFileWithSize(t, filepath.Join(base, "Application Support", "state.dat"), 4096) writeFileWithSize(t, filepath.Join(base, "Mobile Documents", "cloud.dat"), 1024*1024) - withoutIgnore, err := getDirectorySizeFromDuWithExcludeAndIgnores(base, "", nil) + withoutIgnore, err := getDirectorySizeFromDuWithExcludeAndIgnores(context.Background(), base, "", nil) if err != nil { t.Fatalf("getDirectorySizeFromDuWithExcludeAndIgnores without ignore: %v", err) } - withIgnore, err := getDirectorySizeFromDuWithExcludeAndIgnores(base, "", []string{"Mobile Documents"}) + withIgnore, err := getDirectorySizeFromDuWithExcludeAndIgnores(context.Background(), base, "", []string{"Mobile Documents"}) if err != nil { t.Fatalf("getDirectorySizeFromDuWithExcludeAndIgnores with ignore: %v", err) } @@ -125,7 +126,7 @@ func BenchmarkGetDirectorySizeFromDuWithExcludeHomeLibrary(b *testing.B) { b.ResetTimer() for b.Loop() { - size, err := getDirectorySizeFromDuWithExclude(base, excludePath) + size, err := getDirectorySizeFromDuWithExclude(context.Background(), base, excludePath) if err != nil { b.Fatalf("getDirectorySizeFromDuWithExclude: %v", err) } diff --git a/cmd/analyze/update.go b/cmd/analyze/update.go index 5c5aa575b..7a80ed122 100644 --- a/cmd/analyze/update.go +++ b/cmd/analyze/update.go @@ -239,16 +239,6 @@ func (m *model) finishLiveScan(result scanResult) { m.status = fmt.Sprintf("Scanned %s", humanizeBytes(m.totalSize)) } -func (m *model) finishCanceledLiveScan() { - m.scanning = false - m.liveScanID = 0 - m.liveScanCancel = nil - m.liveScanEvents = nil - m.liveScanningPaths = nil - m.autoSortLiveEntries = false - m.status = "Scan cancelled" -} - func (m *model) sortLiveEntriesForActiveMode() { m.ensureLiveEntryBacking() selectedPath := "" @@ -464,9 +454,6 @@ func (m model) Update(msg tea.Msg) (tea.Model, tea.Cmd) { case liveScanFailed: m.status = fmt.Sprintf("Scan failed: %v", msg.err) return m, waitLiveScanEventCmd(m.liveScanEvents) - case liveScanCanceled: - m.finishCanceledLiveScan() - return m, nil default: return m, waitLiveScanEventCmd(m.liveScanEvents) }