From e42418db77a24a99f237427c0d574cae870da156 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 27 Sep 2026 15:25:19 -0400 Subject: [PATCH 1/5] fix(windows): avoid filepath.Rel hang on UNC share roots On Windows, filepath.Rel never returns when the base is a UNC share root (\\host\share) and the target is the same root with a trailing separator (golang/go#79784, fixed only in Go 1.28 by CL 786100). A user on Windows Server with MusicFolder='\\server\Music' (discussion #6232) sent CPU profiles showing 96% of CPU time in filepath.Rel, called from fromArtistFolder <- resolveArtistFolderPattern <- resolver.resolveArtist. For an album directly under the library root, loadArtistFolder's filepath.Dir returns '\\server\Music\', so the artwork worker spins forever and, because Worker.drain waits for the whole batch, the artist artwork pool stops. Add utils.RelPath, which normalizes a UNC volume root followed only by separators to the volume plus one separator before calling filepath.Rel. Both sides then compare equal and Rel returns '.' before reaching the loop. A bare '\\host\share' target under a deeper base now also gives '..' instead of an error. Drive-letter paths are left unchanged, and on non-Windows systems the helper behaves exactly like filepath.Rel. Use it at every production filepath.Rel call site: artist folder artwork, scanner library-relative paths, the watcher, playlist path resolution and PlaylistsPath matching, and the cache spread FS. Windows-only tests cover the UNC cases, with a timeout so a regression fails instead of hanging CI. Run under Wine, 6 of them fail with plain filepath.Rel (4 by timeout) and all pass with RelPath. The helper can be removed after moving to Go 1.28. --- core/artwork/folders_artist.go | 2 +- core/playlists/parse_m3u.go | 3 ++- core/playlists/playlists.go | 3 ++- scanner/scanner.go | 3 ++- scanner/watcher.go | 3 ++- utils/cache/spread_fs.go | 3 ++- utils/files.go | 13 +++++++++++ utils/files_test.go | 23 +++++++++++++++++++ utils/files_windows_test.go | 40 ++++++++++++++++++++++++++++++++++ 9 files changed, 87 insertions(+), 6 deletions(-) create mode 100644 utils/files_windows_test.go diff --git a/core/artwork/folders_artist.go b/core/artwork/folders_artist.go index 3403935db..40972f71d 100644 --- a/core/artwork/folders_artist.go +++ b/core/artwork/folders_artist.go @@ -30,7 +30,7 @@ func fromArtistFolder(ctx context.Context, libFS fs.FS, libPath, artistFolder, p if libFS == nil { return nil, "", fmt.Errorf("artist folder lookup unavailable") } - rel, err := filepath.Rel(libPath, artistFolder) + rel, err := utils.RelPath(libPath, artistFolder) if err != nil || rel == ".." || strings.HasPrefix(rel, ".."+string(filepath.Separator)) { return nil, "", fmt.Errorf(`artist folder '%s' is outside library '%s'`, artistFolder, libPath) } diff --git a/core/playlists/parse_m3u.go b/core/playlists/parse_m3u.go index 9610e9dbb..9ca1c9b6c 100644 --- a/core/playlists/parse_m3u.go +++ b/core/playlists/parse_m3u.go @@ -15,6 +15,7 @@ import ( "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/model/request" + "github.com/navidrome/navidrome/utils" "github.com/navidrome/navidrome/utils/slice" "golang.org/x/text/unicode/norm" ) @@ -148,7 +149,7 @@ func (r pathResolution) ToQualifiedString() (string, error) { if !r.valid { return "", fmt.Errorf("invalid path resolution") } - relativePath, err := filepath.Rel(r.libraryPath, r.absolutePath) + relativePath, err := utils.RelPath(r.libraryPath, r.absolutePath) if err != nil { return "", err } diff --git a/core/playlists/playlists.go b/core/playlists/playlists.go index c9bc03b97..8e7ccf93c 100644 --- a/core/playlists/playlists.go +++ b/core/playlists/playlists.go @@ -15,6 +15,7 @@ import ( "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/model/request" + "github.com/navidrome/navidrome/utils" ) type Playlists interface { @@ -77,7 +78,7 @@ func InPath(folder model.Folder) bool { if conf.Server.PlaylistsPath == "" { return true } - rel, _ := filepath.Rel(folder.LibraryPath, folder.AbsolutePath()) + rel, _ := utils.RelPath(folder.LibraryPath, folder.AbsolutePath()) for path := range strings.SplitSeq(conf.Server.PlaylistsPath, string(filepath.ListSeparator)) { if match, _ := doublestar.Match(path, rel); match { return true diff --git a/scanner/scanner.go b/scanner/scanner.go index cd2fe3c8d..f156d15fd 100644 --- a/scanner/scanner.go +++ b/scanner/scanner.go @@ -15,6 +15,7 @@ import ( "github.com/navidrome/navidrome/core/playlists" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/utils" "github.com/navidrome/navidrome/utils/run" "github.com/navidrome/navidrome/utils/slice" ) @@ -65,7 +66,7 @@ func libraryRelativePath(libPath, folderPath string) string { if err != nil { return folderPath } - rel, err := filepath.Rel(absLib, folderPath) + rel, err := utils.RelPath(absLib, folderPath) if err != nil || !filepath.IsLocal(rel) { return folderPath } diff --git a/scanner/watcher.go b/scanner/watcher.go index 1ac5468f0..3c0fbdbe7 100644 --- a/scanner/watcher.go +++ b/scanner/watcher.go @@ -13,6 +13,7 @@ import ( "github.com/navidrome/navidrome/core/storage" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/utils" "github.com/navidrome/navidrome/utils/singleton" ) @@ -245,7 +246,7 @@ func (w *watcher) processLibraryEvents(ctx context.Context, lib *model.Library, log.Debug(ctx, "Watcher stopped due to context cancellation", "libraryID", lib.ID, "name", lib.Name) return nil case path := <-events: - path, err := filepath.Rel(absLibPath, path) + path, err := utils.RelPath(absLibPath, path) if err != nil { log.Error(ctx, "Error getting relative path", "libraryID", lib.ID, "absolutePath", absLibPath, "path", path, err) continue diff --git a/utils/cache/spread_fs.go b/utils/cache/spread_fs.go index 11801b324..3ebd00e1a 100644 --- a/utils/cache/spread_fs.go +++ b/utils/cache/spread_fs.go @@ -12,6 +12,7 @@ import ( "github.com/djherbis/fscache" "github.com/djherbis/stream" "github.com/navidrome/navidrome/log" + "github.com/navidrome/navidrome/utils" ) const completeMarkerSuffix = ".complete" @@ -97,7 +98,7 @@ func (sfs *spreadFS) walkDataFiles(visit func(absoluteFilePath string)) error { log.Error("Error loading cache", "dir", sfs.root, err) return nil } - path, err := filepath.Rel(sfs.root, absoluteFilePath) + path, err := utils.RelPath(sfs.root, absoluteFilePath) if err != nil { return nil //nolint:nilerr } diff --git a/utils/files.go b/utils/files.go index 2fce307ea..8ef522e5e 100644 --- a/utils/files.go +++ b/utils/files.go @@ -40,3 +40,16 @@ func FileExists(path string) bool { _, err := os.Stat(path) return err == nil || !os.IsNotExist(err) } + +// RelPath is filepath.Rel without the UNC share root hang (golang/go#79784). Remove it after moving to Go 1.28. +func RelPath(basePath, targPath string) (string, error) { + return filepath.Rel(normalizeUNCRoot(basePath), normalizeUNCRoot(targPath)) +} + +func normalizeUNCRoot(p string) string { + vol := filepath.VolumeName(p) + if len(vol) > 2 && strings.Trim(p[len(vol):], `\/`) == "" { + return vol + string(filepath.Separator) + } + return p +} diff --git a/utils/files_test.go b/utils/files_test.go index c6e578f05..04b4d4e0b 100644 --- a/utils/files_test.go +++ b/utils/files_test.go @@ -223,3 +223,26 @@ var _ = Describe("FileExists", func() { Expect(result).To(Or(BeTrue(), BeFalse())) // Should not panic }) }) + +var _ = Describe("RelPath", func() { + DescribeTable("returns the same result as filepath.Rel", + func(base, target string) { + base, target = filepath.FromSlash(base), filepath.FromSlash(target) + expected, expectedErr := filepath.Rel(base, target) + rel, err := utils.RelPath(base, target) + Expect(rel).To(Equal(expected)) + if expectedErr == nil { + Expect(err).ToNot(HaveOccurred()) + } else { + Expect(err).To(MatchError(expectedErr.Error())) + } + }, + Entry("same path", "/music", "/music"), + Entry("same path with trailing separator", "/music", "/music/"), + Entry("child path", "/music", "/music/artist/album"), + Entry("sibling path", "/music/a", "/music/b"), + Entry("parent path", "/music/artist", "/music"), + Entry("relative paths", "music", "music/artist"), + Entry("absolute and relative paths", "/music", "music"), + ) +}) diff --git a/utils/files_windows_test.go b/utils/files_windows_test.go new file mode 100644 index 000000000..419291b98 --- /dev/null +++ b/utils/files_windows_test.go @@ -0,0 +1,40 @@ +package utils_test + +import ( + "time" + + "github.com/navidrome/navidrome/utils" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +var _ = Describe("RelPath on Windows", func() { + type result struct { + rel string + err error + } + + DescribeTable("handles UNC share roots without hanging", + func(base, target, expected string) { + done := make(chan result, 1) + go func() { + rel, err := utils.RelPath(base, target) + done <- result{rel, err} + }() + var res result + Eventually(done).WithTimeout(5 * time.Second).Should(Receive(&res)) + Expect(res.err).ToNot(HaveOccurred()) + Expect(res.rel).To(Equal(expected)) + }, + Entry("root and root with trailing separator", `\\server\Music`, `\\server\Music\`, "."), + Entry("root with trailing separator and root", `\\server\Music\`, `\\server\Music`, "."), + Entry("root and root with many separators", `\\server\Music`, `\\server\Music\\`, "."), + Entry("root with forward slashes", `//server/Music`, `//server/Music/`, "."), + Entry("root and child folder", `\\server\Music`, `\\server\Music\Artist`, "Artist"), + Entry("root with trailing separator and child folder", `\\server\Music\`, `\\server\Music\Artist\Album`, `Artist\Album`), + Entry("child folder and root", `\\server\Music\Artist`, `\\server\Music\`, ".."), + Entry("child folder and root without trailing separator", `\\server\Music\Artist`, `\\server\Music`, ".."), + Entry("root and root with different case", `\\server\Music`, `\\SERVER\music\`, "."), + Entry("drive root and child folder", `C:\`, `C:\Music`, "Music"), + ) +}) From d513b1ea4857db8d7adf80eb51abf7424c03659d Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 27 Sep 2026 15:39:16 -0400 Subject: [PATCH 2/5] fix(artwork): stop a stuck item from stalling its drain pool Worker.drain waits for every item in a batch before it dequeues the next one, so a single item that never returns stops the whole pool. This is what happened in discussion #6232: filepath.Rel looped forever on a UNC share root, and all artist artwork stopped. Go can't stop a goroutine that spins, so the only way to keep the pool going is to stop waiting. process now runs acquire in its own goroutine and waits at most itemTimeout (5 minutes). On timeout it logs an error, adds a trace step, and settles the row as failed, so the normal backoff and give-up budget apply. A per-worker busy set keeps one item from running twice at the same time: if the row comes due again while the old run is still stuck, it is settled as failed without starting a second run. This keeps the existing guarantee that the drain resolves each item serially. Shutdown still waits for in-flight items, but now at most itemTimeout for one that is stuck. --- core/artwork/worker.go | 65 +++++++++++++++++++++++++++++- core/artwork/worker_timing_test.go | 54 +++++++++++++++++++++++++ 2 files changed, 117 insertions(+), 2 deletions(-) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 28e51958c..198fcc87a 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -26,6 +26,8 @@ const ( // giveUpAfter bounds the retry budget from enqueue; past it the item settles and only an // explicit reprocess retries it. giveUpAfter = 12 * time.Hour + // itemTimeout is how long a drain waits on one item before rescheduling it. + itemTimeout = 5 * time.Minute ) // drainPool drains one class of work with its own slot budget, so a blocking kind cannot @@ -50,8 +52,14 @@ type Worker struct { gatesMu sync.Mutex gates map[string]*extGate + + // busy holds the items still being acquired, including ones a drain stopped waiting for. + busyMu sync.Mutex + busy map[itemKey]struct{} } +type itemKey struct{ kind, id, imageType string } + func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg ffmpeg.FFmpeg, broker events.Broker, imgCache cache.FileCache) *Worker { w := &Worker{ proc: &processor{ds: ds, store: store}, @@ -62,6 +70,7 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg runCtx: context.Background(), paused: func() bool { return false }, gates: map[string]*extGate{}, + busy: map[itemKey]struct{}{}, } w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate) w.proc.pruneLock = w.pruneMu.RLocker() @@ -244,8 +253,61 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary) trace := &ChainTrace{} ctx = withTrace(ctx, trace) - out, got, retryIn := w.proc.acquire(ctx, item) + out, got, retryIn := w.acquireWithTimeout(ctx, item) + w.settle(ctx, item, out, retryIn, trace) + return out, got +} +// acquireWithTimeout stops waiting on a stuck item, which Go can't stop, and never runs one item twice at once. +func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired, time.Duration) { + key := itemKey{item.ItemKind, item.ItemID, item.ImageType} + if !w.markBusy(key) { + log.Debug(ctx, "Artwork: Item still running from an earlier drain", "kind", item.ItemKind, "id", item.ItemID) + traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "previous attempt still running"}) + return outcomeFailed, nil, 0 + } + type result struct { + out outcome + got *acquired + retryIn time.Duration + } + done := make(chan result, 1) + go func() { + out, got, retryIn := w.proc.acquire(ctx, item) + // Cleared before the send, so the next drain never sees a finished item as busy. + w.clearBusy(key) + done <- result{out, got, retryIn} + }() + timer := time.NewTimer(itemTimeout) + defer timer.Stop() + select { + case r := <-done: + return r.out, r.got, r.retryIn + case <-timer.C: + log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout) + traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "timed out after " + itemTimeout.String()}) + return outcomeFailed, nil, 0 + } +} + +func (w *Worker) markBusy(key itemKey) bool { + w.busyMu.Lock() + defer w.busyMu.Unlock() + if _, ok := w.busy[key]; ok { + return false + } + w.busy[key] = struct{}{} + return true +} + +func (w *Worker) clearBusy(key itemKey) { + w.busyMu.Lock() + defer w.busyMu.Unlock() + delete(w.busy, key) +} + +// settle deletes or reschedules the queue row for an acquire outcome. +func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration, trace *ChainTrace) { queue := w.proc.ds.ArtworkQueue() switch out { case outcomeFound, outcomeAbsent: @@ -283,7 +345,6 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc log.Warn(ctx, "Artwork: Could not remove exhausted queue item", "kind", item.ItemKind, "id", item.ItemID, err) } } - return out, got } // recordGiveUp keeps the last failure on the state row after the queue row is deleted. An item diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index 63f12d03c..d991fc38d 100644 --- a/core/artwork/worker_timing_test.go +++ b/core/artwork/worker_timing_test.go @@ -7,7 +7,10 @@ import ( "testing/synctest" "time" + "github.com/navidrome/navidrome/conf" + "github.com/navidrome/navidrome/conf/configtest" "github.com/navidrome/navidrome/core/agents" + "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/tests" . "github.com/onsi/gomega" ) @@ -113,3 +116,54 @@ func TestArtworkGatePerAgentBreakerIsolation(t *testing.T) { g.Expect(aCalls).To(Equal(1)) }) } + +// Go can't stop a stuck item, so the drain must stop waiting for it, or the whole pool stalls. +func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + defer configtest.SetupConfig()() + conf.Server.ArtistArtPriority = "external" + conf.Server.DevArtworkExternalMaxRPS = 1000 + + block := make(chan struct{}) + agent := &fakeImageAgent{name: "stuckAgent", block: block} + ag := imageAgents(agent) + artists := tests.CreateMockArtistRepo() + artists.SetData(model.Artists{{ID: "ar1", Name: "Artist"}}) + queue := tests.CreateMockArtworkQueueRepo() + ds := &tests.MockDataStore{MockedArtist: artists, MockedFolder: &fakeFolderRepo{}, MockedArtwork: tests.CreateMockArtworkRepo(), MockedArtworkQueue: queue} + w := NewWorker(ds, NewImageStore(t.TempDir()), ag, tests.NewMockFFmpeg(""), &fakeEventBroker{}, nil) + ctx := t.Context() + g.Expect(queue.Enqueue(ctx, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"})).To(Succeed()) + + start := time.Now() + n, err := w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(n).To(Equal(1)) + g.Expect(time.Since(start)).To(Equal(itemTimeout), "the drain stops waiting once the item times out") + row := findQueued(queue, "ar", "ar1") + g.Expect(row).ToNot(BeNil(), "a timed-out item is rescheduled, not dropped") + g.Expect(row.Attempts).To(Equal(1)) + g.Expect(row.RetryAt).To(BeTemporally(">", time.Now())) + g.Expect(row.Trace).To(ContainSubstring("timed out")) + + // Due again while the first run is still stuck: it must not start a second one. + time.Sleep(time.Until(row.RetryAt)) + start = time.Now() + _, err = w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(time.Since(start)).To(BeZero(), "a still-running item is rescheduled without waiting") + row = findQueued(queue, "ar", "ar1") + g.Expect(row.Attempts).To(Equal(2)) + g.Expect(row.Trace).To(ContainSubstring("still running")) + + // Once the stuck run returns, the next attempt runs normally and settles the row. + close(block) + synctest.Wait() + time.Sleep(time.Until(row.RetryAt)) + _, err = w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(findQueued(queue, "ar", "ar1")).To(BeNil()) + g.Expect(agent.artistCalls).To(Equal(2), "one stuck call and one retry, never two at once") + }) +} From 89e17f7ed888bbc67d1aa8fea9e3903af16d1e62 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 27 Sep 2026 15:51:50 -0400 Subject: [PATCH 3/5] refactor(artwork): cancel a timed-out item's context and simplify the busy set Give acquire a context with the same itemTimeout deadline, so network and ffmpeg work that honors ctx stops when the drain moves on. The select still covers work that ignores ctx, like the filepath.Rel loop. Replace the busy mutex, map and helpers with a sync.Map, use time.After, record the timeout and still-running steps with traceStage, and let settle read the trace from ctx. --- core/artwork/worker.go | 50 ++++++++++++++---------------------------- 1 file changed, 17 insertions(+), 33 deletions(-) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 198fcc87a..8c2f2bd40 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -4,6 +4,8 @@ import ( "bytes" "cmp" "context" + "errors" + "fmt" "io" "math" "math/rand/v2" @@ -53,9 +55,8 @@ type Worker struct { gatesMu sync.Mutex gates map[string]*extGate - // busy holds the items still being acquired, including ones a drain stopped waiting for. - busyMu sync.Mutex - busy map[itemKey]struct{} + // busy holds the itemKeys still being acquired, including ones a drain stopped waiting for. + busy sync.Map } type itemKey struct{ kind, id, imageType string } @@ -70,7 +71,6 @@ func NewWorker(ds model.DataStore, store *ImageStore, ag *agents.Agents, ffmpeg runCtx: context.Background(), paused: func() bool { return false }, gates: map[string]*extGate{}, - busy: map[itemKey]struct{}{}, } w.proc.resolver = newResolver(ds, ag, ffmpeg, w.gate) w.proc.pruneLock = w.pruneMu.RLocker() @@ -251,19 +251,18 @@ func (w *Worker) broadcastRefresh(ctx context.Context, found []model.ArtworkQueu func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired) { item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary) - trace := &ChainTrace{} - ctx = withTrace(ctx, trace) + ctx = withTrace(ctx, &ChainTrace{}) out, got, retryIn := w.acquireWithTimeout(ctx, item) - w.settle(ctx, item, out, retryIn, trace) + w.settle(ctx, item, out, retryIn) return out, got } // acquireWithTimeout stops waiting on a stuck item, which Go can't stop, and never runs one item twice at once. func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueueItem) (outcome, *acquired, time.Duration) { key := itemKey{item.ItemKind, item.ItemID, item.ImageType} - if !w.markBusy(key) { + if _, running := w.busy.LoadOrStore(key, struct{}{}); running { log.Debug(ctx, "Artwork: Item still running from an earlier drain", "kind", item.ItemKind, "id", item.ItemID) - traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "previous attempt still running"}) + traceStage(ctx, "worker", errors.New("previous attempt still running")) return outcomeFailed, nil, 0 } type result struct { @@ -273,41 +272,26 @@ func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueue } done := make(chan result, 1) go func() { - out, got, retryIn := w.proc.acquire(ctx, item) + // The deadline stops work that honors ctx; the select below covers work that doesn't. + actx, cancel := context.WithTimeout(ctx, itemTimeout) + defer cancel() + out, got, retryIn := w.proc.acquire(actx, item) // Cleared before the send, so the next drain never sees a finished item as busy. - w.clearBusy(key) + w.busy.Delete(key) done <- result{out, got, retryIn} }() - timer := time.NewTimer(itemTimeout) - defer timer.Stop() select { case r := <-done: return r.out, r.got, r.retryIn - case <-timer.C: + case <-time.After(itemTimeout): log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout) - traceFrom(ctx).add(TraceStep{Candidate: "worker", Outcome: OutcomeError, Detail: "timed out after " + itemTimeout.String()}) + traceStage(ctx, "worker", fmt.Errorf("timed out after %s", itemTimeout)) return outcomeFailed, nil, 0 } } -func (w *Worker) markBusy(key itemKey) bool { - w.busyMu.Lock() - defer w.busyMu.Unlock() - if _, ok := w.busy[key]; ok { - return false - } - w.busy[key] = struct{}{} - return true -} - -func (w *Worker) clearBusy(key itemKey) { - w.busyMu.Lock() - defer w.busyMu.Unlock() - delete(w.busy, key) -} - // settle deletes or reschedules the queue row for an acquire outcome. -func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration, trace *ChainTrace) { +func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out outcome, retryIn time.Duration) { queue := w.proc.ds.ArtworkQueue() switch out { case outcomeFound, outcomeAbsent: @@ -318,7 +302,7 @@ func (w *Worker) settle(ctx context.Context, item model.ArtworkQueueItem, out ou } case outcomeFoundStale, outcomeFailed: retryAt := time.Now().Add(retryDelay(item.Attempts, retryIn)) - encoded := trace.encode("") + encoded := traceFrom(ctx).encode("") if retryAt.Before(item.EnqueuedAt.Add(giveUpAfter)) { // A mid-flight re-enqueue reset retry_at; stale backoff must not stomp its // fresh, immediate eligibility. From 3817637207f15b7f172e3012cf77a2bfee78ec71 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 27 Sep 2026 18:07:27 -0400 Subject: [PATCH 4/5] fix(artwork): lower the item timeout and stop draining at once on shutdown Lower itemTimeout from 5 to 3 minutes. The slowest healthy item calls each image agent in turn: a short rate-limiter wait, the agent call (10s HTTP timeout for built-in agents, 30s for plugins) and a 5s image download. With the default agents plus a plugin that is about 90s, so 3 minutes keeps a safe margin without cancelling slow but healthy lookups. When the worker context is cancelled, process now stops waiting at once and leaves the queue row untouched for the next run, instead of holding shutdown for up to itemTimeout on a stuck item. --- core/artwork/worker.go | 9 +++-- core/artwork/worker_test.go | 5 +++ core/artwork/worker_timing_test.go | 58 +++++++++++++++++++++++------- 3 files changed, 57 insertions(+), 15 deletions(-) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 8c2f2bd40..62c6a7ea6 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -28,8 +28,8 @@ const ( // giveUpAfter bounds the retry budget from enqueue; past it the item settles and only an // explicit reprocess retries it. giveUpAfter = 12 * time.Hour - // itemTimeout is how long a drain waits on one item before rescheduling it. - itemTimeout = 5 * time.Minute + // itemTimeout must outlast a slow but healthy item: each image agent's call plus its download, in turn. + itemTimeout = 3 * time.Minute ) // drainPool drains one class of work with its own slot budget, so a blocking kind cannot @@ -253,6 +253,9 @@ func (w *Worker) process(ctx context.Context, item model.ArtworkQueueItem) (outc item.ImageType = cmp.Or(item.ImageType, model.ImageTypePrimary) ctx = withTrace(ctx, &ChainTrace{}) out, got, retryIn := w.acquireWithTimeout(ctx, item) + if ctx.Err() != nil { + return outcomeFailed, nil // shutting down: leave the row for the next run + } w.settle(ctx, item, out, retryIn) return out, got } @@ -283,6 +286,8 @@ func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueue select { case r := <-done: return r.out, r.got, r.retryIn + case <-ctx.Done(): + return outcomeFailed, nil, 0 case <-time.After(itemTimeout): log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout) traceStage(ctx, "worker", fmt.Errorf("timed out after %s", itemTimeout)) diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index a6c07b763..95b2df0a6 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -838,6 +838,11 @@ var _ = Describe("Worker", func() { cancel() close(block) // unpark the blocked lookups so the pools can unwind <-done + // Run returns without waiting on the lookups, so wait for them before config restores. + Eventually(func() (n int) { + w.busy.Range(func(_, _ any) bool { n++; return true }) + return n + }).Should(BeZero()) }) Eventually(func() bool { diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index d991fc38d..d4aa79d26 100644 --- a/core/artwork/worker_timing_test.go +++ b/core/artwork/worker_timing_test.go @@ -1,6 +1,7 @@ package artwork import ( + "context" "errors" "io" "testing" @@ -117,24 +118,32 @@ func TestArtworkGatePerAgentBreakerIsolation(t *testing.T) { }) } +// newStuckItemWorker queues one artist whose agent blocks until the returned channel is closed. +func newStuckItemWorker(t *testing.T) (*Worker, *tests.MockArtworkQueueRepo, *fakeImageAgent, chan struct{}) { + t.Cleanup(configtest.SetupConfig()) + conf.Server.ArtistArtPriority = "external" + conf.Server.DevArtworkExternalMaxRPS = 1000 + + block := make(chan struct{}) + agent := &fakeImageAgent{name: "stuckAgent", block: block} + ag := imageAgents(agent) + artists := tests.CreateMockArtistRepo() + artists.SetData(model.Artists{{ID: "ar1", Name: "Artist"}}) + queue := tests.CreateMockArtworkQueueRepo() + ds := &tests.MockDataStore{MockedArtist: artists, MockedFolder: &fakeFolderRepo{}, MockedArtwork: tests.CreateMockArtworkRepo(), MockedArtworkQueue: queue} + w := NewWorker(ds, NewImageStore(t.TempDir()), ag, tests.NewMockFFmpeg(""), &fakeEventBroker{}, nil) + if err := queue.Enqueue(t.Context(), model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"}); err != nil { + t.Fatal(err) + } + return w, queue, agent, block +} + // Go can't stop a stuck item, so the drain must stop waiting for it, or the whole pool stalls. func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) { synctest.Test(t, func(t *testing.T) { g := NewWithT(t) - defer configtest.SetupConfig()() - conf.Server.ArtistArtPriority = "external" - conf.Server.DevArtworkExternalMaxRPS = 1000 - - block := make(chan struct{}) - agent := &fakeImageAgent{name: "stuckAgent", block: block} - ag := imageAgents(agent) - artists := tests.CreateMockArtistRepo() - artists.SetData(model.Artists{{ID: "ar1", Name: "Artist"}}) - queue := tests.CreateMockArtworkQueueRepo() - ds := &tests.MockDataStore{MockedArtist: artists, MockedFolder: &fakeFolderRepo{}, MockedArtwork: tests.CreateMockArtworkRepo(), MockedArtworkQueue: queue} - w := NewWorker(ds, NewImageStore(t.TempDir()), ag, tests.NewMockFFmpeg(""), &fakeEventBroker{}, nil) + w, queue, agent, block := newStuckItemWorker(t) ctx := t.Context() - g.Expect(queue.Enqueue(ctx, model.ArtworkQueueItem{ItemKind: "ar", ItemID: "ar1"})).To(Succeed()) start := time.Now() n, err := w.drain(ctx, 1) @@ -167,3 +176,26 @@ func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) { g.Expect(agent.artistCalls).To(Equal(2), "one stuck call and one retry, never two at once") }) } + +// Shutdown must not wait on a stuck item, and must leave its row for the next run. +func TestArtworkDrainStopsAtOnceOnShutdown(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + w, queue, _, block := newStuckItemWorker(t) + before := *findQueued(queue, "ar", "ar1") + ctx, cancel := context.WithCancel(t.Context()) + go func() { + time.Sleep(time.Second) + cancel() + }() + + start := time.Now() + _, err := w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(time.Since(start)).To(Equal(time.Second), "the drain returns as soon as the context is cancelled") + g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left as it was") + + close(block) + synctest.Wait() + }) +} From 83d1aa4c3bb1caa6fa03fa5379c9f34579d38ab6 Mon Sep 17 00:00:00 2001 From: Deluan Date: Sun, 27 Sep 2026 18:23:14 -0400 Subject: [PATCH 5/5] fix(artwork): let cancelled work unwind before the worker stops On shutdown the drain returned while cancelled acquires were still running, so Run could return, and the DB close, before they finished. The race detector caught one of them reading config while the e2e suite restored it. When the context is cancelled, process now waits up to shutdownGrace (5s) for the cancelled acquire to return. Work that honors ctx unwinds in milliseconds, so Run joins it; only a stuck item adds the grace period. --- core/artwork/worker.go | 6 ++++++ core/artwork/worker_test.go | 5 ----- core/artwork/worker_timing_test.go | 32 ++++++++++++++++++++++++++---- 3 files changed, 34 insertions(+), 9 deletions(-) diff --git a/core/artwork/worker.go b/core/artwork/worker.go index 62c6a7ea6..6122f88f2 100644 --- a/core/artwork/worker.go +++ b/core/artwork/worker.go @@ -30,6 +30,8 @@ const ( giveUpAfter = 12 * time.Hour // itemTimeout must outlast a slow but healthy item: each image agent's call plus its download, in turn. itemTimeout = 3 * time.Minute + // shutdownGrace lets cancelled work unwind before Run returns and the DB closes, without waiting on a stuck item. + shutdownGrace = 5 * time.Second ) // drainPool drains one class of work with its own slot budget, so a blocking kind cannot @@ -287,6 +289,10 @@ func (w *Worker) acquireWithTimeout(ctx context.Context, item model.ArtworkQueue case r := <-done: return r.out, r.got, r.retryIn case <-ctx.Done(): + select { + case <-done: + case <-time.After(shutdownGrace): + } return outcomeFailed, nil, 0 case <-time.After(itemTimeout): log.Error(ctx, "Artwork: Item timed out, moving on", "kind", item.ItemKind, "id", item.ItemID, "timeout", itemTimeout) diff --git a/core/artwork/worker_test.go b/core/artwork/worker_test.go index 95b2df0a6..a6c07b763 100644 --- a/core/artwork/worker_test.go +++ b/core/artwork/worker_test.go @@ -838,11 +838,6 @@ var _ = Describe("Worker", func() { cancel() close(block) // unpark the blocked lookups so the pools can unwind <-done - // Run returns without waiting on the lookups, so wait for them before config restores. - Eventually(func() (n int) { - w.busy.Range(func(_, _ any) bool { n++; return true }) - return n - }).Should(BeZero()) }) Eventually(func() bool { diff --git a/core/artwork/worker_timing_test.go b/core/artwork/worker_timing_test.go index d4aa79d26..7896f6f09 100644 --- a/core/artwork/worker_timing_test.go +++ b/core/artwork/worker_timing_test.go @@ -177,8 +177,32 @@ func TestArtworkDrainMovesOnFromAStuckItem(t *testing.T) { }) } -// Shutdown must not wait on a stuck item, and must leave its row for the next run. -func TestArtworkDrainStopsAtOnceOnShutdown(t *testing.T) { +// Shutdown joins cancelled work that unwinds, so nothing touches the DB after Run returns. +func TestArtworkDrainShutdownWaitsForWorkToUnwind(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + g := NewWithT(t) + w, queue, _, block := newStuckItemWorker(t) + before := *findQueued(queue, "ar", "ar1") + ctx, cancel := context.WithCancel(t.Context()) + go func() { + time.Sleep(time.Second) + cancel() + time.Sleep(100 * time.Millisecond) + close(block) + }() + + start := time.Now() + _, err := w.drain(ctx, 1) + g.Expect(err).ToNot(HaveOccurred()) + g.Expect(time.Since(start)).To(Equal(time.Second+100*time.Millisecond), "the drain waits for the cancelled item to return") + _, running := w.busy.Load(itemKey{"ar", "ar1", model.ImageTypePrimary}) + g.Expect(running).To(BeFalse()) + g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left for the next run") + }) +} + +// Shutdown must not wait on a stuck item for longer than the grace period. +func TestArtworkDrainShutdownDoesNotWaitOnAStuckItem(t *testing.T) { synctest.Test(t, func(t *testing.T) { g := NewWithT(t) w, queue, _, block := newStuckItemWorker(t) @@ -192,8 +216,8 @@ func TestArtworkDrainStopsAtOnceOnShutdown(t *testing.T) { start := time.Now() _, err := w.drain(ctx, 1) g.Expect(err).ToNot(HaveOccurred()) - g.Expect(time.Since(start)).To(Equal(time.Second), "the drain returns as soon as the context is cancelled") - g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left as it was") + g.Expect(time.Since(start)).To(Equal(time.Second+shutdownGrace), "the drain gives up on the stuck item after the grace period") + g.Expect(*findQueued(queue, "ar", "ar1")).To(Equal(before), "the row is left for the next run") close(block) synctest.Wait()