diff --git a/core/artwork/artwork.go b/core/artwork/artwork.go index 3b403ae6c..a021e7e46 100644 --- a/core/artwork/artwork.go +++ b/core/artwork/artwork.go @@ -30,6 +30,15 @@ func NewArtwork(ds model.DataStore, cache cache.FileCache, ffmpeg ffmpeg.FFmpeg, return a } +// Close stops the background blurhash worker. The server never calls it; tests must, so a leaked +// worker can't touch mocks/filesystems being torn down by the next spec. +func (a *artwork) Close() error { + if a.blurHashes != nil { + a.blurHashes.stop() + } + return nil +} + type artwork struct { ds model.DataStore cache cache.FileCache diff --git a/core/artwork/artwork_internal_test.go b/core/artwork/artwork_internal_test.go index c95371959..cdb3d3204 100644 --- a/core/artwork/artwork_internal_test.go +++ b/core/artwork/artwork_internal_test.go @@ -67,6 +67,8 @@ var _ = Describe("Artwork", func() { cache := GetImageCache() ffmpeg = tests.NewMockFFmpeg("content from ffmpeg") aw = NewArtwork(ds, cache, ffmpeg, nil).(*artwork) + // Stop the blurhash worker up front: it would mutate the non-thread-safe mocks mid-spec. + Expect(aw.Close()).To(Succeed()) }) Describe("albumArtworkReader", func() { diff --git a/core/artwork/artwork_test.go b/core/artwork/artwork_test.go index adddd0dc3..b3e7b9421 100644 --- a/core/artwork/artwork_test.go +++ b/core/artwork/artwork_test.go @@ -26,6 +26,8 @@ var _ = Describe("Artwork", func() { cache := artwork.GetImageCache() ffmpeg = tests.NewMockFFmpeg("content from ffmpeg") aw = artwork.NewArtwork(ds, cache, ffmpeg, nil) + // Stop the blurhash worker up front: it would mutate the non-thread-safe mocks mid-spec. + Expect(aw.(io.Closer).Close()).To(Succeed()) }) Context("GetOrPlaceholder", func() { diff --git a/core/artwork/benchmark_e2e_test.go b/core/artwork/benchmark_e2e_test.go index bf3d435a8..69a04a7c3 100644 --- a/core/artwork/benchmark_e2e_test.go +++ b/core/artwork/benchmark_e2e_test.go @@ -95,6 +95,7 @@ func setupE2EBenchmark(b *testing.B, cacheSize string) (Artwork, model.ArtworkID aw := NewArtwork(ds, imgCache, ffmpeg, nil) cleanupAll := func() { + _ = aw.(*artwork).Close() os.RemoveAll(tmpDir) } return aw, artID, cleanupAll diff --git a/core/artwork/blurhash_updater.go b/core/artwork/blurhash_updater.go index 735256baa..4457ff725 100644 --- a/core/artwork/blurhash_updater.go +++ b/core/artwork/blurhash_updater.go @@ -31,7 +31,10 @@ type blurHashUpdater struct { buffer map[model.ArtworkID]enqueueRequest noResult map[model.ArtworkID]time.Time wake chan struct{} - start sync.Once + done chan struct{} + runDone chan struct{} + started bool + stopped bool } func newBlurHashUpdater(a *artwork) *blurHashUpdater { @@ -40,6 +43,8 @@ func newBlurHashUpdater(a *artwork) *blurHashUpdater { buffer: make(map[model.ArtworkID]enqueueRequest), noResult: make(map[model.ArtworkID]time.Time), wake: make(chan struct{}, 1), + done: make(chan struct{}), + runDone: make(chan struct{}), } } @@ -49,12 +54,17 @@ func (u *blurHashUpdater) Enqueue(artID model.ArtworkID, imageUpdatedAt time.Tim default: return } - u.start.Do(func() { - // Playlist artwork readers require a user in the context. Like the cacheWarmer, the worker - // lives for the rest of the process; lazy-starting keeps idle Artwork instances goroutine-free. - go u.run(request.WithUser(context.Background(), model.User{IsAdmin: true})) - }) u.mutex.Lock() + if u.stopped { + u.mutex.Unlock() + return + } + if !u.started { + u.started = true + // Playlist artwork readers require a user in the context. Lazy-starting keeps idle Artwork + // instances goroutine-free; stop() ends the worker (tests must call it, the server never does). + go u.run(request.WithUser(context.Background(), model.User{IsAdmin: true})) + } req := u.buffer[artID] req.force = req.force || force if imageUpdatedAt.After(req.imageUpdatedAt) { @@ -68,8 +78,31 @@ func (u *blurHashUpdater) Enqueue(artID model.ArtworkID, imageUpdatedAt time.Tim } } +// stop ends the worker and waits for any in-flight computation, so callers can safely tear down +// the resources (DataStore, filesystems) the worker touches. +func (u *blurHashUpdater) stop() { + u.mutex.Lock() + if u.stopped { + u.mutex.Unlock() + return + } + u.stopped = true + started := u.started + u.mutex.Unlock() + close(u.done) + if started { + <-u.runDone + } +} + func (u *blurHashUpdater) run(ctx context.Context) { - for range u.wake { + defer close(u.runDone) + for { + select { + case <-u.done: + return + case <-u.wake: + } for { artID, req, ok := u.next() if !ok { diff --git a/core/artwork/blurhash_updater_internal_test.go b/core/artwork/blurhash_updater_internal_test.go index ec099eb91..e996e2abe 100644 --- a/core/artwork/blurhash_updater_internal_test.go +++ b/core/artwork/blurhash_updater_internal_test.go @@ -15,12 +15,13 @@ var _ = Describe("blurHashUpdater", func() { BeforeEach(func() { ds = &tests.MockDataStore{} - // No run() goroutine: tests drive next()/process() directly. + // started is pre-set so Enqueue never spawns run(): tests drive next()/process() directly. u = &blurHashUpdater{ a: &artwork{ds: ds}, buffer: make(map[model.ArtworkID]enqueueRequest), noResult: make(map[model.ArtworkID]time.Time), wake: make(chan struct{}, 1), + started: true, } }) diff --git a/core/artwork/e2e/suite_test.go b/core/artwork/e2e/suite_test.go index 06cc05b6f..1ca4f2e2a 100644 --- a/core/artwork/e2e/suite_test.go +++ b/core/artwork/e2e/suite_test.go @@ -3,6 +3,7 @@ package artworke2e_test import ( "context" "fmt" + "io" "path/filepath" "testing" @@ -88,6 +89,8 @@ func setupHarness() { storagetest.Register(fakeLibScheme, fakeFS) aw = artwork.NewArtwork(ds, artwork.GetImageCache(), newNoopFFmpeg(), &noopProvider{}) + // The worker must not outlive the spec: it would race the next spec's fakeFS/DB swaps. + DeferCleanup(aw.(io.Closer).Close) } func scan() {