diff --git a/core/artwork/artwork.go b/core/artwork/artwork.go index 9b89230b9..a22cc471d 100644 --- a/core/artwork/artwork.go +++ b/core/artwork/artwork.go @@ -39,19 +39,10 @@ type Artwork interface { func NewArtwork(ds model.DataStore, cache cache.FileCache, ffmpeg ffmpeg.FFmpeg, provider external.Provider) Artwork { a := &artwork{ds: ds, cache: cache, ffmpeg: ffmpeg, provider: provider} - a.blurHashes = newBlurHashUpdater(a) + a.blurHashes = newBlurHashUpdater(ds) 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 @@ -72,10 +63,10 @@ func (a *artwork) GetOrPlaceholder(ctx context.Context, id string, size int, squ reader, lastUpdate, err = a.Get(ctx, artID, size, square) } if errors.Is(err, ErrUnavailable) { - if a.blurHashes != nil && eligibleKind(artID) { - // No bytes flowed through the tee; a real deletion must still clear the stored hash. The - // worker re-checks so a transient fetch failure doesn't clobber a valid hash. - a.blurHashes.EnqueueClearIfGone(artID, capAtNow(consts.ServerStart)) + if a.blurHashes != nil { + // The client is receiving the placeholder, so a stored hash describing the old cover must + // clear — hash-what-you-serve applies to the fallback too. + a.blurHashes.clearIfStored(ctx, artID, time.Now()) } if artID.Kind == model.KindArtistArtwork { reader, _ = resources.FS().Open(consts.PlaceholderArtistArt) @@ -107,7 +98,7 @@ func (a *artwork) Get(ctx context.Context, artID model.ArtworkID, size int, squa // The tee wraps r directly, so Close reaches the underlying stream (no fd leak). version := capAtNow(artReader.LastUpdated()) reader = newTeeReader(r, maxTeeBytes, - func(data []byte) { a.blurHashes.EnqueueBytes(artID, data, version) }) + func(data []byte) { a.blurHashes.update(ctx, artID, data, version) }) } return reader, artReader.LastUpdated(), nil } diff --git a/core/artwork/artwork_internal_test.go b/core/artwork/artwork_internal_test.go index cdb3d3204..c95371959 100644 --- a/core/artwork/artwork_internal_test.go +++ b/core/artwork/artwork_internal_test.go @@ -67,8 +67,6 @@ 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 b3e7b9421..adddd0dc3 100644 --- a/core/artwork/artwork_test.go +++ b/core/artwork/artwork_test.go @@ -26,8 +26,6 @@ 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 69a04a7c3..bf3d435a8 100644 --- a/core/artwork/benchmark_e2e_test.go +++ b/core/artwork/benchmark_e2e_test.go @@ -95,7 +95,6 @@ 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 67c76dbd7..0e05206fb 100644 --- a/core/artwork/blurhash_updater.go +++ b/core/artwork/blurhash_updater.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "hash/fnv" "image" "io" "sync" @@ -13,42 +14,28 @@ import ( "github.com/navidrome/navidrome/core/artwork/blurhash" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" - "github.com/navidrome/navidrome/model/request" "github.com/navidrome/navidrome/resources" ) -// blurHashJob is a unit of work: either bytes to hash (data != nil) or a deletion check (checkGone). -type blurHashJob struct { - data []byte - version time.Time - checkGone bool +// blurHashState remembers what was last persisted for an artwork, keyed by a checksum of the served +// bytes, so repeated serves of the same image skip the decode and the write entirely. +type blurHashState struct { + sum uint64 + version time.Time + hash string } -// blurHashUpdater keeps stored blurhashes in sync with the bytes actually served. The serve path tees -// the served image and hands it here; there is no change-detection proxy — the hash is a pure function -// of the captured bytes. A single worker decodes, encodes, and writes (dedup'd in memory). +// blurHashUpdater keeps stored blurhashes in sync with the bytes actually served. It runs inline in +// the serving goroutine after the response is fully written (decode+encode is a few ms): the hash is +// a pure function of the captured bytes — no change-detection proxy, no background worker. type blurHashUpdater struct { - a *artwork - mutex sync.Mutex - buffer map[model.ArtworkID]blurHashJob - last map[model.ArtworkID]string // last hash written this process; avoids redundant writes - wake chan struct{} - done chan struct{} - runDone chan struct{} - runCancel context.CancelFunc - started bool - stopped bool + ds model.DataStore + mutex sync.Mutex + seen map[model.ArtworkID]blurHashState } -func newBlurHashUpdater(a *artwork) *blurHashUpdater { - return &blurHashUpdater{ - a: a, - buffer: make(map[model.ArtworkID]blurHashJob), - last: make(map[model.ArtworkID]string), - wake: make(chan struct{}, 1), - done: make(chan struct{}), - runDone: make(chan struct{}), - } +func newBlurHashUpdater(ds model.DataStore) *blurHashUpdater { + return &blurHashUpdater{ds: ds, seen: make(map[model.ArtworkID]blurHashState)} } func eligibleKind(artID model.ArtworkID) bool { @@ -59,122 +46,35 @@ func eligibleKind(artID model.ArtworkID) bool { return false } -// EnqueueBytes schedules a blurhash update computed from the exact bytes served for artID. -func (u *blurHashUpdater) EnqueueBytes(artID model.ArtworkID, data []byte, version time.Time) { - u.enqueue(artID, blurHashJob{data: data, version: version}) -} - -// EnqueueClearIfGone schedules a deletion check: the worker re-reads the source once and clears the -// stored hash only if it still fails/serves a placeholder, so a transient failure won't clobber it. -func (u *blurHashUpdater) EnqueueClearIfGone(artID model.ArtworkID, version time.Time) { - u.enqueue(artID, blurHashJob{checkGone: true, version: version}) -} - -func (u *blurHashUpdater) enqueue(artID model.ArtworkID, job blurHashJob) { - if !eligibleKind(artID) { - return - } - u.mutex.Lock() - if u.stopped { - u.mutex.Unlock() - return - } - if !u.started { - u.started = true - // Admin context: playlist artwork readers require a user. Lazy start keeps idle Artwork - // instances goroutine-free; stop() ends the worker (tests call it, the server never does). - ctx, cancel := context.WithCancel(request.WithUser(context.Background(), model.User{IsAdmin: true})) - u.runCancel = cancel - go u.run(ctx) - } - // A bytes job supersedes a pending gone-check (a successful serve proves the artwork exists); - // otherwise keep whichever is newer. - prev, ok := u.buffer[artID] - if !ok || job.data != nil || (prev.checkGone && job.version.After(prev.version)) { - u.buffer[artID] = job - } - u.mutex.Unlock() - select { - case u.wake <- struct{}{}: - default: - } -} - -// 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 - cancel := u.runCancel - u.mutex.Unlock() - close(u.done) - if started { - cancel() - <-u.runDone - } -} - -func (u *blurHashUpdater) run(ctx context.Context) { - defer close(u.runDone) - for { - select { - case <-u.done: - return - case <-u.wake: - } - for { - select { - case <-u.done: - return - default: - } - artID, job, ok := u.next() - if !ok { - break - } - u.processJob(ctx, artID, job) - } - } -} - -func (u *blurHashUpdater) next() (model.ArtworkID, blurHashJob, bool) { - u.mutex.Lock() - defer u.mutex.Unlock() - for artID, job := range u.buffer { - delete(u.buffer, artID) - return artID, job, true - } - return model.ArtworkID{}, blurHashJob{}, false -} - -// processTimeout bounds one computation: readers can call external agents, and a hung call must not -// stall the worker forever. -const processTimeout = 30 * time.Second - -func (u *blurHashUpdater) processJob(ctx context.Context, artID model.ArtworkID, job blurHashJob) { - // Artwork readers can touch storage, agents and plugins; a panic here must not kill the server. +// update hashes the exact bytes served for artID and persists the result. Placeholder bytes mean the +// entity has no artwork anymore, so they clear a stored hash instead. +func (u *blurHashUpdater) update(ctx context.Context, artID model.ArtworkID, data []byte, version time.Time) { + // Decoding arbitrary image bytes can panic; the serve already succeeded, so just log it. defer func() { if r := recover(); r != nil { log.Error(ctx, "BlurHash: recovered from panic", "artID", artID, "panic", r) } }() - ctx, cancel := context.WithTimeout(ctx, processTimeout) - defer cancel() - - if job.checkGone { - u.processGone(ctx, artID, job.version) + // The response is already written when the tee fires; a client abort must not lose the write. + ctx = context.WithoutCancel(ctx) + if isPlaceholder(data) { + u.clearIfStored(ctx, artID, version) return } - if isPlaceholder(job.data) { - u.clear(ctx, artID, job.version) + sum := checksum(data) + u.mutex.Lock() + prev, ok := u.seen[artID] + u.mutex.Unlock() + if ok && prev.hash != "" && prev.sum == sum { + if !version.After(prev.version) { + return + } + // Same bytes under a newer artwork version: re-persist so blur_hash_updated_at keeps pace with + // the entity version, or the DTO staleness gate would emit the fake after any routine scan. + u.persistAndRemember(ctx, artID, prev.hash, sum, version) return } - img, _, err := image.Decode(bytes.NewReader(job.data)) + img, _, err := image.Decode(bytes.NewReader(data)) if err != nil { // Undecodable served bytes are not proof of change; leave the stored hash intact. log.Trace(ctx, "BlurHash: served bytes not decodable, keeping stored hash", "artID", artID, err) @@ -186,54 +86,83 @@ func (u *blurHashUpdater) processJob(ctx context.Context, artID model.ArtworkID, if err != nil || hash == "" { return } - u.write(ctx, artID, hash, job.version) + u.persistAndRemember(ctx, artID, hash, sum, version) } -// processGone re-reads the source once; if it still yields a placeholder or fails, the artwork is -// really gone and the stored hash is cleared. A transient failure recovers by now and is left alone. -func (u *blurHashUpdater) processGone(ctx context.Context, artID model.ArtworkID, version time.Time) { - artReader, err := u.a.getArtworkReader(ctx, artID, 0, false) - if err == nil { - r, gErr := u.a.cache.Get(ctx, artReader) - if gErr == nil { - data, rErr := io.ReadAll(r) - _ = r.Close() - if rErr == nil && !isPlaceholder(data) { - return // source came back (or never really failed): keep the hash - } +// clearIfStored clears the persisted hash after a placeholder was served (a cold map costs one row +// read to skip never-hashed entities); a failed read clears nothing — unknown state is not deletion. +func (u *blurHashUpdater) clearIfStored(ctx context.Context, artID model.ArtworkID, version time.Time) { + if !eligibleKind(artID) { + return + } + ctx = context.WithoutCancel(ctx) + u.mutex.Lock() + prev, ok := u.seen[artID] + u.mutex.Unlock() + if ok && prev.hash == "" { + return + } + if !ok { + stored, err := u.loadStoredHash(ctx, artID) + if err != nil { + return + } + if stored == "" { + u.remember(artID, blurHashState{version: version}) + return } } - u.clear(ctx, artID, version) -} - -func (u *blurHashUpdater) write(ctx context.Context, artID model.ArtworkID, hash string, version time.Time) { - u.mutex.Lock() - if u.last[artID] == hash { - u.mutex.Unlock() - return - } - u.mutex.Unlock() - if err := u.persist(ctx, artID, hash, version); err != nil { - log.Warn(ctx, "BlurHash: error persisting", "artID", artID, err) - return - } - u.mutex.Lock() - u.last[artID] = hash - u.mutex.Unlock() -} - -func (u *blurHashUpdater) clear(ctx context.Context, artID model.ArtworkID, version time.Time) { - // No cold-map dedup: an empty u.last[artID] means "never written" as easily as "already cleared", - // so skipping would leave a previous process's DB hash describing gone artwork. Clears are rare. if err := u.persist(ctx, artID, "", version); err != nil { log.Warn(ctx, "BlurHash: error clearing hash", "artID", artID, err) return } + u.remember(artID, blurHashState{version: version}) +} + +func (u *blurHashUpdater) persistAndRemember(ctx context.Context, artID model.ArtworkID, hash string, sum uint64, version time.Time) { + if err := u.persist(ctx, artID, hash, version); err != nil { + log.Warn(ctx, "BlurHash: error persisting", "artID", artID, err) + return + } + u.remember(artID, blurHashState{sum: sum, version: version, hash: hash}) +} + +func (u *blurHashUpdater) remember(artID model.ArtworkID, s blurHashState) { u.mutex.Lock() - u.last[artID] = "" + u.seen[artID] = s u.mutex.Unlock() } +func checksum(data []byte) uint64 { + h := fnv.New64a() + _, _ = h.Write(data) + return h.Sum64() +} + +func (u *blurHashUpdater) loadStoredHash(ctx context.Context, artID model.ArtworkID) (string, error) { + switch artID.Kind { + case model.KindAlbumArtwork: + al, err := u.ds.Album(ctx).Get(artID.ID) + if err != nil { + return "", err + } + return al.BlurHash, nil + case model.KindArtistArtwork: + ar, err := u.ds.Artist(ctx).Get(artID.ID) + if err != nil { + return "", err + } + return ar.BlurHash, nil + case model.KindPlaylistArtwork: + pl, err := u.ds.Playlist(ctx).Get(artID.ID) + if err != nil { + return "", err + } + return pl.BlurHash, nil + } + return "", model.ErrNotFound +} + // isPlaceholder byte-compares against the embedded placeholder assets: placeholder artwork must never // be persisted as an entity's blurhash, and captured bytes carry no source path to check. func isPlaceholder(data []byte) bool { @@ -261,11 +190,11 @@ var placeholderImages = sync.OnceValue(func() [][]byte { func (u *blurHashUpdater) persist(ctx context.Context, artID model.ArtworkID, hash string, version time.Time) error { switch artID.Kind { case model.KindAlbumArtwork: - return u.a.ds.Album(ctx).UpdateBlurHash(artID.ID, hash, version) + return u.ds.Album(ctx).UpdateBlurHash(artID.ID, hash, version) case model.KindArtistArtwork: - return u.a.ds.Artist(ctx).UpdateBlurHash(artID.ID, hash, version) + return u.ds.Artist(ctx).UpdateBlurHash(artID.ID, hash, version) case model.KindPlaylistArtwork: - return u.a.ds.Playlist(ctx).UpdateBlurHash(artID.ID, hash, version) + return u.ds.Playlist(ctx).UpdateBlurHash(artID.ID, hash, version) } return fmt.Errorf("blurhash: no persister for artwork kind %q", artID.Kind) } diff --git a/core/artwork/blurhash_updater_internal_test.go b/core/artwork/blurhash_updater_internal_test.go index 9f95a064c..0fa42eb08 100644 --- a/core/artwork/blurhash_updater_internal_test.go +++ b/core/artwork/blurhash_updater_internal_test.go @@ -38,78 +38,81 @@ func realPNGBytes(label string) []byte { var _ = Describe("blurHashUpdater", func() { var u *blurHashUpdater var ds *tests.MockDataStore + var repo *tests.MockAlbumRepo var version time.Time + album := func(al model.Album) model.ArtworkID { + repo = tests.CreateMockAlbumRepo() + repo.SetData(model.Albums{al}) + ds.MockedAlbum = repo + return al.CoverArtID() + } + stored := func(id string) model.Album { + al, err := ds.Album(GinkgoT().Context()).Get(id) + Expect(err).ToNot(HaveOccurred()) + return *al + } + BeforeEach(func() { ds = &tests.MockDataStore{} - u = &blurHashUpdater{ - a: &artwork{ds: ds}, - buffer: make(map[model.ArtworkID]blurHashJob), - wake: make(chan struct{}, 1), - last: make(map[model.ArtworkID]string), - started: true, - } + u = newBlurHashUpdater(ds) version = time.Date(2024, 1, 1, 0, 0, 0, 0, time.UTC) }) - It("persists a hash computed from the given bytes", func() { - al := model.Album{ID: "al-1", UpdatedAt: version} - repo := tests.CreateMockAlbumRepo() - repo.SetData(model.Albums{al}) - ds.MockedAlbum = repo - - u.processJob(GinkgoT().Context(), al.CoverArtID(), blurHashJob{data: realPNGBytes("x"), version: version}) - stored, err := ds.Album(GinkgoT().Context()).Get("al-1") - Expect(err).ToNot(HaveOccurred()) - Expect(stored.BlurHash).ToNot(BeEmpty()) + It("persists a hash computed from the served bytes", func() { + id := album(model.Album{ID: "al-1", UpdatedAt: version}) + u.update(GinkgoT().Context(), id, realPNGBytes("x"), version) + al := stored("al-1") + Expect(al.BlurHash).ToNot(BeEmpty()) + Expect(al.BlurHashUpdatedAt).To(HaveValue(Equal(version))) }) - It("clears the hash when the bytes are a placeholder", func() { - al := model.Album{ID: "al-1", UpdatedAt: version, BlurHash: "OLD"} - repo := tests.CreateMockAlbumRepo() - repo.SetData(model.Albums{al}) - ds.MockedAlbum = repo - - u.processJob(GinkgoT().Context(), al.CoverArtID(), blurHashJob{data: placeholderImages()[0], version: version}) - stored, _ := ds.Album(GinkgoT().Context()).Get("al-1") - Expect(stored.BlurHash).To(BeEmpty()) // cleared: placeholder means gone + It("clears the stored hash when the served bytes are a placeholder", func() { + id := album(model.Album{ID: "al-1", UpdatedAt: version, BlurHash: "OLD"}) + u.update(GinkgoT().Context(), id, placeholderImages()[0], version) + Expect(stored("al-1").BlurHash).To(BeEmpty()) }) It("leaves the hash untouched on undecodable bytes", func() { - al := model.Album{ID: "al-1", UpdatedAt: version, BlurHash: "KEEP"} - repo := tests.CreateMockAlbumRepo() - repo.SetData(model.Albums{al}) - ds.MockedAlbum = repo - - u.processJob(GinkgoT().Context(), al.CoverArtID(), blurHashJob{data: []byte("not an image"), version: version}) - stored, _ := ds.Album(GinkgoT().Context()).Get("al-1") - Expect(stored.BlurHash).To(Equal("KEEP")) + id := album(model.Album{ID: "al-1", UpdatedAt: version, BlurHash: "KEEP"}) + u.update(GinkgoT().Context(), id, []byte("not an image"), version) + Expect(stored("al-1").BlurHash).To(Equal("KEEP")) }) - It("skips a redundant write when the hash is unchanged (in-memory dedup)", func() { - al := model.Album{ID: "al-1", UpdatedAt: version} - repo := tests.CreateMockAlbumRepo() - repo.SetData(model.Albums{al}) - ds.MockedAlbum = repo + It("skips the write when the same bytes are served again under the same version", func() { + id := album(model.Album{ID: "al-1", UpdatedAt: version}) data := realPNGBytes("dedup") + u.update(GinkgoT().Context(), id, data, version) + // Tamper with the stored value: a second identical serve must not touch the row. + Expect(repo.UpdateBlurHash("al-1", "TAMPERED", version)).To(Succeed()) + u.update(GinkgoT().Context(), id, data, version) + Expect(stored("al-1").BlurHash).To(Equal("TAMPERED")) + }) - u.processJob(GinkgoT().Context(), al.CoverArtID(), blurHashJob{data: data, version: version}) - first, _ := ds.Album(GinkgoT().Context()).Get("al-1") - u.processJob(GinkgoT().Context(), al.CoverArtID(), blurHashJob{data: data, version: version.Add(time.Hour)}) - second, _ := ds.Album(GinkgoT().Context()).Get("al-1") + It("re-persists the same hash when the artwork version advances", func() { + // A scan can bump the entity version without changing the cover; blur_hash_updated_at must + // follow, or the DTO's staleness gate would emit the fake hash forever after. + id := album(model.Album{ID: "al-1", UpdatedAt: version}) + data := realPNGBytes("same-bytes") + u.update(GinkgoT().Context(), id, data, version) + first := stored("al-1") + newer := version.Add(time.Hour) + u.update(GinkgoT().Context(), id, data, newer) + second := stored("al-1") Expect(second.BlurHash).To(Equal(first.BlurHash)) + Expect(second.BlurHashUpdatedAt).To(HaveValue(Equal(newer))) }) - It("ignores non-eligible artwork kinds on enqueue", func() { - u.EnqueueBytes(model.ArtworkID{Kind: model.KindMediaFileArtwork, ID: "mf-1"}, realPNGBytes("x"), version) - Expect(u.buffer).To(BeEmpty()) + It("does not write when a placeholder is served and nothing was ever stored", func() { + id := album(model.Album{ID: "al-1", UpdatedAt: version}) + u.update(GinkgoT().Context(), id, placeholderImages()[0], version) + Expect(stored("al-1").BlurHashUpdatedAt).To(BeNil()) }) - It("supersedes a pending gone-check with a bytes job", func() { - id := model.Album{ID: "al-1"}.CoverArtID() - u.EnqueueClearIfGone(id, version) - u.EnqueueBytes(id, realPNGBytes("x"), version.Add(time.Hour)) - Expect(u.buffer[id].checkGone).To(BeFalse()) - Expect(u.buffer[id].data).ToNot(BeNil()) + It("ignores non-eligible artwork kinds", func() { + Expect(func() { + u.clearIfStored(GinkgoT().Context(), model.ArtworkID{Kind: model.KindMediaFileArtwork, ID: "mf-1"}, version) + }).ToNot(Panic()) + Expect(u.seen).To(BeEmpty()) }) }) diff --git a/core/artwork/e2e/suite_test.go b/core/artwork/e2e/suite_test.go index 81bec7f52..06cc05b6f 100644 --- a/core/artwork/e2e/suite_test.go +++ b/core/artwork/e2e/suite_test.go @@ -3,7 +3,6 @@ package artworke2e_test import ( "context" "fmt" - "io" "path/filepath" "testing" @@ -58,18 +57,6 @@ var _ = AfterSuite(func() { db.Close(GinkgoT().Context()) }) -// AfterEach runs before any DeferCleanup a spec body registered, so it stops the blurhash worker -// before spec-local TempDirs are removed. On Windows a still-running worker can hold an artwork -// file open, and TempDir removal cannot unlink an open file. -var _ = AfterEach(func() { - if aw == nil { - return - } - if c, ok := aw.(io.Closer); ok { - Expect(c.Close()).To(Succeed()) - } -}) - func setupHarness() { DeferCleanup(configtest.SetupConfig()) @@ -101,8 +88,6 @@ func setupHarness() { storagetest.Register(fakeLibScheme, fakeFS) aw = artwork.NewArtwork(ds, artwork.GetImageCache(), newNoopFFmpeg(), &noopProvider{}) - // The worker is stopped by the suite-level AfterEach (which runs before spec-local TempDir - // cleanups), so it can't outlive the spec or hold a file open past TempDir removal. } func scan() {