diff --git a/core/artwork/artwork_test.go b/core/artwork/artwork_test.go index c6b848101..7de5475d6 100644 --- a/core/artwork/artwork_test.go +++ b/core/artwork/artwork_test.go @@ -106,7 +106,7 @@ var _ = Describe("Artwork", func() { func(ctx context.Context, arg cache.Item) (io.Reader, error) { return arg.(artworkReader).Reader(ctx) }) - Eventually(func() bool { return imgCache.Available(ctx) }).Should(BeTrue()) + Eventually(func() bool { return imgCache.Available(ctx) }, 10*time.Second).Should(BeTrue()) svc = NewArtwork(ds, imgCache, store, ffm) }) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 3042c47b0..3ded52629 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -170,8 +170,11 @@ func (w *Worker) drain(ctx context.Context, concurrency int, kinds ...string) (i select { case sem <- struct{}{}: case <-ctx.Done(): + } + // select picks randomly when both cases are ready, so re-check to never dispatch after cancellation. + if ctx.Err() != nil { wg.Wait() - return len(items), nil + return len(items), nil //nolint:nilerr // a cancelled drain is a clean stop, not an error } wg.Go(func() { defer func() { <-sem }() diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 1f3b52840..5b6e42885 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -151,7 +151,8 @@ var _ = Describe("Worker", func() { func(ctx context.Context, arg cache.Item) (io.Reader, error) { return arg.(artworkReader).Reader(ctx) })} - Eventually(func() bool { return imgCache.Available(ctx) }).Should(BeTrue()) + // Init walks the cache dir on a goroutine; loaded CI runners can take >1s. + Eventually(func() bool { return imgCache.Available(ctx) }, 10*time.Second).Should(BeTrue()) w = NewWorker(ds, store, ag, ffm, broker, imgCache) }) @@ -706,13 +707,16 @@ var _ = Describe("Worker", func() { Describe("batching", func() { It("leaves undispatched items queued when cancelled mid-batch", func() { + // Every album must resolve, so any dispatched item deletes its row regardless of dequeue order. + albums := model.Albums{} for i := range 8 { id := fmt.Sprintf("alc%d", i) - ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(model.Albums{{ID: id, Name: "Album"}}) + albums = append(albums, model.Album{ID: id, Name: "Album"}) Expect(queueRepo.Enqueue(model.ArtworkQueueItem{ ItemKind: "al", ItemID: id, Priority: model.ArtworkPriorityScan, })).To(Succeed()) } + ds.MockedAlbum.(*tests.MockAlbumRepo).SetData(albums) cancelledCtx, cancel := context.WithCancel(ctx) cancel() diff --git a/scanner/watcher_test.go b/scanner/watcher_test.go index 15e49e195..12c94602c 100644 --- a/scanner/watcher_test.go +++ b/scanner/watcher_test.go @@ -26,7 +26,8 @@ var _ = Describe("Watcher", func() { BeforeEach(func() { DeferCleanup(configtest.SetupConfig()) - conf.Server.Scanner.WatcherWait = 50 * time.Millisecond // Short wait for tests + // Must dwarf the 20ms Consistently windows below, or a loaded runner's delayed timer reset flakes the debouncing spec. + conf.Server.Scanner.WatcherWait = 200 * time.Millisecond ctx, cancel = context.WithCancel(GinkgoT().Context()) DeferCleanup(cancel) @@ -91,7 +92,7 @@ var _ = Describe("Watcher", func() { return nil } return calls[0].Targets - }, 500*time.Millisecond, 10*time.Millisecond).Should(HaveLen(2)) + }, 2*time.Second, 10*time.Millisecond).Should(HaveLen(2)) // Verify targets calls := mockScanner.GetScanFoldersCalls() @@ -111,7 +112,7 @@ var _ = Describe("Watcher", func() { // Wait for watcher to process and trigger scan Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) // Verify the target calls := mockScanner.GetScanFoldersCalls() @@ -129,7 +130,7 @@ var _ = Describe("Watcher", func() { // Wait for watcher to process and trigger scan Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) // Verify only one target despite multiple file/folder changes calls := mockScanner.GetScanFoldersCalls() @@ -170,7 +171,7 @@ var _ = Describe("Watcher", func() { // Now wait for the debounce timer to expire and trigger scan Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) }) It("triggers scan after quiet period", func() { @@ -183,7 +184,7 @@ var _ = Describe("Watcher", func() { // Wait for quiet period Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) }) }) @@ -205,7 +206,7 @@ var _ = Describe("Watcher", func() { // Wait for scan Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) // Should scan the library root calls := mockScanner.GetScanFoldersCalls() @@ -222,7 +223,7 @@ var _ = Describe("Watcher", func() { // Wait for scan Eventually(func() int { return mockScanner.GetScanFoldersCallCount() - }, 500*time.Millisecond, 10*time.Millisecond).Should(Equal(1)) + }, 2*time.Second, 10*time.Millisecond).Should(Equal(1)) // Should have only one target calls := mockScanner.GetScanFoldersCalls() @@ -266,7 +267,7 @@ var _ = Describe("Watcher", func() { return nil } return calls[0].Targets - }, 500*time.Millisecond, 10*time.Millisecond).Should(HaveLen(2)) + }, 2*time.Second, 10*time.Millisecond).Should(HaveLen(2)) // Verify library IDs are different calls := mockScanner.GetScanFoldersCalls() diff --git a/utils/cache/file_caches.go b/utils/cache/file_caches.go index 33fa046df..dff9e4e7a 100644 --- a/utils/cache/file_caches.go +++ b/utils/cache/file_caches.go @@ -183,12 +183,11 @@ func (fc *fileCache) Get(ctx context.Context, arg Item) (*CachedStream, error) { return nil, err } go func() { - if err := copyAndClose(w, reader); err != nil { + if err := fc.copyAndClose(ctx, key, w, reader); err != nil { log.Debug(ctx, "Error storing file in cache", "cache", fc.name, "key", key, err) _ = fc.invalidate(ctx, key) } else { log.Trace(ctx, "File successfully stored in cache", "cache", fc.name, "key", key) - fc.markComplete(ctx, key) } }() } @@ -243,7 +242,8 @@ func getFinalCachedSize(r fscache.ReadAtCloser) int64 { return -1 } -func copyAndClose(w io.WriteCloser, r io.Reader) error { +// copyAndClose marks the entry complete before closing w, so EOF implies the entry is settled on disk. +func (fc *fileCache) copyAndClose(ctx context.Context, key string, w io.WriteCloser, r io.Reader) error { _, err := io.Copy(w, r) if err != nil { err = fmt.Errorf("copying data to cache: %w", err) @@ -253,7 +253,9 @@ func copyAndClose(w io.WriteCloser, r io.Reader) error { err = multierror.Append(err, fmt.Errorf("closing source stream: %w", cErr)) } } - + if err == nil { + fc.markComplete(ctx, key) + } if cErr := w.Close(); cErr != nil { err = multierror.Append(err, fmt.Errorf("closing cache writer: %w", cErr)) } diff --git a/utils/cache/file_caches_test.go b/utils/cache/file_caches_test.go index 7ce1b0393..974200656 100644 --- a/utils/cache/file_caches_test.go +++ b/utils/cache/file_caches_test.go @@ -8,6 +8,7 @@ import ( "path/filepath" "strings" "sync/atomic" + "time" "github.com/navidrome/navidrome/conf" "github.com/navidrome/navidrome/conf/configtest" @@ -18,7 +19,7 @@ import ( // Call NewFileCache and wait for it to be ready func callNewFileCache(name, cacheSize, cacheFolder string, maxItems int, getReader ReadFunc) *fileCache { fc := NewFileCache(name, cacheSize, cacheFolder, maxItems, getReader).(*fileCache) - Eventually(func() bool { return fc.ready.Load() }).Should(BeTrue()) + Eventually(func() bool { return fc.ready.Load() }, 10*time.Second).Should(BeTrue()) return fc } @@ -114,11 +115,10 @@ var _ = Describe("File Caches", func() { _, _ = io.ReadAll(s) _ = s.Close() + // EOF must imply the entry is settled on disk (Windows temp-dir cleanups rely on it). dataPath := fcSpreadFS(fc).KeyMapper((&testArg{"markme"}).Key()) - Eventually(func() bool { - _, statErr := os.Stat(dataPath + ".complete") - return statErr == nil - }).Should(BeTrue()) + _, statErr := os.Stat(dataPath + ".complete") + Expect(statErr).ToNot(HaveOccurred()) }) It("serves a concurrent reader from an in-progress write and marks complete once", func() {