diff --git a/core/stream/media_streamer.go b/core/stream/media_streamer.go index b09d9bab8..3bdcac44d 100644 --- a/core/stream/media_streamer.go +++ b/core/stream/media_streamer.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "hash/fnv" "io" "mime" "net/http" @@ -60,6 +61,20 @@ func (j *streamJob) Key() string { return fmt.Sprintf("%s.%s.%d.%d.%d.%d.%s.%d", j.mf.ID, j.mf.UpdatedAt.Format(time.RFC3339Nano), j.bitRate, j.sampleRate, j.bitDepth, j.channels, j.format, j.offset) } +// weakETag derives a weak HTTP entity tag from a transcode identity (the cache +// key). It is weak (W/) because ffmpeg output is not guaranteed to be +// byte-for-byte identical across runs (e.g. embedded timestamps/metadata), but +// it is a stable validator for a given representation: the same media file at +// the same bitrate/format/offset always yields the same tag, and it changes +// whenever any of those inputs change. Serving it lets a client revalidate a +// repeated request for the same transcoded track with a cheap 304 Not Modified +// instead of re-downloading the full audio. +func weakETag(identity string) string { + h := fnv.New64a() + _, _ = io.WriteString(h, identity) + return fmt.Sprintf(`W/"%x"`, h.Sum64()) +} + // NewStream creates a Stream for the given MediaFile and Request. It handles both raw streaming (no transcoding) // and transcoded streaming based on the requested format and bitrate. It also logs detailed information about // the streaming request and whether the transcoding result was served from cache or not. @@ -109,6 +124,9 @@ func (ms *mediaStreamer) NewStream(ctx context.Context, mf *model.MediaFile, req channels: req.Channels, offset: req.Offset, } + // Tag the transcoded representation so repeated requests for the same + // track within a playback can be revalidated (304) instead of re-fetched. + s.etag = weakETag(job.Key()) r, err := ms.cache.Get(ctx, job) if err != nil { // Rate-limit rejections are already logged at warn level by the @@ -137,6 +155,7 @@ type Stream struct { mf *model.MediaFile bitRate int format string + etag string io.ReadCloser io.Seeker } @@ -156,6 +175,14 @@ func (s *Stream) EstimatedContentLength() int { // (meaning the HTTP 200 status has not been flushed yet and the caller can still send an error response). // Empty output (0 bytes, no error) is logged but not treated as an error. func (s *Stream) Serve(ctx context.Context, w http.ResponseWriter, r *http.Request) (int64, error) { + // Advertise a validator for the transcoded representation so clients can + // revalidate a repeated request with a conditional GET. On the seekable + // (fully cached) path http.ServeContent handles If-None-Match/If-Range + // against it, answering repeats with 304 instead of re-sending the audio. + if s.etag != "" { + w.Header().Set("ETag", s.etag) + } + if s.Seekable() { http.ServeContent(w, r, s.Name(), s.ModTime(), s) return -1, nil diff --git a/core/stream/media_streamer_test.go b/core/stream/media_streamer_test.go index f5ca16d3f..8e802d544 100644 --- a/core/stream/media_streamer_test.go +++ b/core/stream/media_streamer_test.go @@ -4,6 +4,8 @@ import ( "context" "errors" "io" + "net/http" + "net/http/httptest" "os" "github.com/navidrome/navidrome/conf" @@ -138,5 +140,64 @@ var _ = Describe("MediaStreamer", func() { Expect(err).To(BeNil()) Expect(s.Seekable()).To(BeTrue()) }) + + It("revalidates a repeated transcoded request with 304 instead of re-sending the audio", func() { + // Dedicated ffmpeg/cache so the transcoded payload is deterministic + // (the suite-level mock's reader is single-use and shared). + freshFFmpeg := tests.NewMockFFmpeg("some transcoded audio payload") + localCache := stream.NewTranscodingCache() + Eventually(func() bool { return localCache.Available(context.TODO()) }).Should(BeTrue()) + localStreamer := stream.NewMediaStreamer(ds, freshFFmpeg, localCache) + + // Prime the cache so the transcode is complete and seekable. + s, err := localStreamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 32}) + Expect(err).ToNot(HaveOccurred()) + _, _ = io.ReadAll(s) + _ = s.Close() + Eventually(func() bool { return freshFFmpeg.IsClosed() }, "3s").Should(BeTrue()) + + // First request: serve fully and capture the advertised ETag. + s, err = localStreamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 32}) + Expect(err).ToNot(HaveOccurred()) + Expect(s.Seekable()).To(BeTrue()) + rec := httptest.NewRecorder() + _, err = s.Serve(ctx, rec, httptest.NewRequest(http.MethodGet, "/stream", nil)) + Expect(err).ToNot(HaveOccurred()) + _ = s.Close() + Expect(rec.Code).To(Equal(http.StatusOK)) + Expect(rec.Body.Len()).To(BeNumerically(">", 0)) + etag := rec.Header().Get("ETag") + Expect(etag).To(HavePrefix(`W/"`)) + + // Repeat with a matching If-None-Match: must be a bodiless 304. + s, err = localStreamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 32}) + Expect(err).ToNot(HaveOccurred()) + rec2 := httptest.NewRecorder() + r2 := httptest.NewRequest(http.MethodGet, "/stream", nil) + r2.Header.Set("If-None-Match", etag) + _, err = s.Serve(ctx, rec2, r2) + Expect(err).ToNot(HaveOccurred()) + _ = s.Close() + Expect(rec2.Code).To(Equal(http.StatusNotModified)) + Expect(rec2.Body.Len()).To(Equal(0)) + }) + + It("changes the ETag when the requested bitrate changes", func() { + s1, err := streamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 32}) + Expect(err).ToNot(HaveOccurred()) + rec1 := httptest.NewRecorder() + _, _ = s1.Serve(ctx, rec1, httptest.NewRequest(http.MethodGet, "/stream", nil)) + _ = s1.Close() + + s2, err := streamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 64}) + Expect(err).ToNot(HaveOccurred()) + rec2 := httptest.NewRecorder() + _, _ = s2.Serve(ctx, rec2, httptest.NewRequest(http.MethodGet, "/stream", nil)) + _ = s2.Close() + + Expect(rec1.Header().Get("ETag")).ToNot(BeEmpty()) + Expect(rec2.Header().Get("ETag")).ToNot(BeEmpty()) + Expect(rec1.Header().Get("ETag")).ToNot(Equal(rec2.Header().Get("ETag"))) + }) }) })