fix(cache): write the completion marker before closing the cache writer (#5927)

* fix(cache): write the completion marker before closing the cache writer

Readers of an in-progress cache write see EOF the moment the writer closes,
but the .complete marker was created after the close, on the background
goroutine — so a fully-read stream did not mean the cache was done touching
disk. The new artwork precache spec ends right at EOF, and its
GinkgoT().TempDir() cleanup raced the marker creation, failing the Windows CI
job with 'unlinkat ...: The directory is not empty' (the race also reproduces
on macOS, 2 of 3 runs, with the tightened test).

Writing the marker after a clean copy but before Close makes reader-EOF imply
every on-disk write for the entry is finished. A failed writer Close still
invalidates the entry, which removes both the marker and the data file. The
existing marker test now asserts the marker exists immediately at EOF instead
of Eventually.

* fix(artwork): never dispatch queue items after the drain context is cancelled

The 10x Windows stress run for the previous commit surfaced a second flake in
the same package: 'leaves undispatched items queued when cancelled mid-batch'
lost row alc7 in 4 of 10 runs. In drain, when a semaphore slot is free and the
context is already cancelled, both cases of the blocking select are ready and
Go picks one at random — so a cancelled drain could still dispatch items. A
non-blocking Done check before the select gives cancellation priority.

The race was invisible on Linux/macOS only by accident: the spec seeded the
album repo with a single album (each SetData overwrote the last), so only the
final row (alc7) resolved to absent and got deleted when dispatched; the
others fell on the retry path and survived. Nanosecond enqueue timestamps
made alc0 always first out of the mock dequeue, masking the race, while
Windows' coarse clock ties the timestamps and randomizes the order. The spec
now seeds all eight albums, which made the race reproduce locally on the
first try (row alc0) and now guards the fix on every platform.

* test: give cache-init waits a 10s timeout for loaded CI runners

A 10x parallel Windows stress run timed out one artwork spec in BeforeEach:
the FileCache init goroutine (mkdir + reload walk) took over Gomega's default
1s Eventually timeout under shared-runner disk contention. Bump the three
identical init waits (two artwork suites and the utils/cache helper) to 10s.

* test(scanner): widen watcher debounce margins for loaded CI runners

The watcher debouncing spec asserts 'no scan yet' inside 20ms Consistently
windows while the debounce wait was only 50ms — a 2.5x margin that a loaded
Windows runner blows through by delaying the timer-reset notification, firing
the scan early (failed all three FlakeAttempts in a 10x stress run). Raise the
test debounce wait to 200ms (10x the observation windows) and the scan-fired
Eventually timeouts to 2s to match.

* refactor(artwork): collapse drain cancellation into a single exit path

Replace the non-blocking ctx pre-check plus duplicated select exit with one
select and a ctx.Err() check after it. Besides removing the duplication, this
closes the residual race: a cancellation landing between the two selects could
still let the blocking select randomly pick the free semaphore slot and
dispatch the item. Now a dispatch is only possible when the context was live
after slot acquisition.

* style: trim flaky-test fix comments to single lines

Compress each two-line comment added by this PR to the one line that carries
the invariant; drop the narration around it.

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)
 	})
 

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 }()

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()
 

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()

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))
 	}

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() {