From 97da9993d76126cadc45f54159ed6e5ce0932230 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Deluan=20Quint=C3=A3o?= Date: Tue, 25 Aug 2026 18:48:43 -0400 Subject: [PATCH] fix(stream): abort the response when a transcoded stream is truncated (#6035) * fix(stream): abort the response when a transcoded stream is truncated When a transcode failed after some audio had already been sent, Serve logged the error and returned nil, so Go finished the chunked body normally and the client received an apparently complete, silently short file. Symfonium users hit this on large offline syncs, and the worst path, ffmpeg dying mid-write behind the transcoding cache, produced no error and nothing in the log above Debug: the cache writer was closed plainly, so readers drained the truncated entry to a clean EOF. The root cause of that silence is an fscache limitation: Close is the only way to end a cache write, and Close always means "complete". This adopts the deluan/fscache fork, which adds CloseWithError: on failure copyAndClose now cancels the entry with the cause, so every attached reader fails mid-read with the real error instead of EOF, a late Get for the entry is refused, and the entry never reports a final size. The error travels inside the entry each reader holds, which makes per-generation delivery automatic and needs no bookkeeping on our side. With the failure arriving in-band, one change in Serve covers every mode: an io.Copy error after bytes are on the wire panics with http.ErrAbortHandler. Go aborts the response without the terminating chunk (RST_STREAM on HTTP/2), chi's Recoverer re-panics that value, and the deferred stream.Close() still runs, so the transcode limiter slot is released as before. Two behaviors improve as side effects. A transcoder that dies before its first byte now yields a Subsonic error response instead of a 200 with an empty body, since the failure reaches Serve as an error while the status is still unsent; genuinely empty output (clean EOF, exit 0) keeps the 200. And a failed entry's invalidation no longer defers its unlink past a replacement entry re-creating the same file, because canceling already closed its readers. * fix(cache): warn when the cache writer cannot report failures to readers The CloseWithError capability comes from the fscache fork via a go.mod replace directive, and a type assertion picks it up. If that directive is ever lost, the assertion fails silently, readers of a dead writer go back to draining a truncated entry to a clean EOF, and nothing says so. Two layers against that: a warning on the failure path when the writer lacks the capability, and a test that asserts the writer fscache returns carries it, so losing the fork fails CI instead of a listener's download. * build: point the fscache replace at the fork's master deluan/fscache#1 is merged; pin the merge commit instead of the review branch. Pinned by sha because the module proxy still resolves the fork's master ref to its pre-merge commit. * build: reference the upstream fscache PR in the replace comment The replace itself must keep pointing at the fork: the commit only exists in djherbis/fscache under refs/pull/22/head, which the Go module fetcher cannot resolve (verified: unknown revision for both short and full sha). The same commit is advertised on the fork's master, so that is the fetchable source. --- core/stream/media_streamer.go | 6 ++- core/stream/media_streamer_test.go | 51 ++++++++++++++++++++++++ go.mod | 3 ++ go.sum | 4 +- utils/cache/file_caches.go | 11 ++++++ utils/cache/file_caches_test.go | 62 ++++++++++++++++++++++++++---- 6 files changed, 126 insertions(+), 11 deletions(-) diff --git a/core/stream/media_streamer.go b/core/stream/media_streamer.go index b09d9bab8..aaa3126b4 100644 --- a/core/stream/media_streamer.go +++ b/core/stream/media_streamer.go @@ -152,8 +152,9 @@ func (s *Stream) EstimatedContentLength() int { // Serve writes the stream to the HTTP response. For seekable streams it uses http.ServeContent // (supporting range requests). For non-seekable streams it writes directly and logs any errors. -// Returns the number of bytes written and an error only when io.Copy fails with 0 bytes written +// Returns the number of bytes written and an error only when it fails with 0 bytes written // (meaning the HTTP 200 status has not been flushed yet and the caller can still send an error response). +// Once bytes are on the wire it panics with http.ErrAbortHandler instead, aborting the 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) { if s.Seekable() { @@ -183,7 +184,8 @@ func (s *Stream) Serve(ctx context.Context, w http.ResponseWriter, r *http.Reque w.Header().Del("Content-Length") return 0, fmt.Errorf("sending transcoded file: %w", err) } - return c, nil + // The 200 is already sent, so dropping the connection is the only way to say "truncated". + panic(http.ErrAbortHandler) } if c == 0 { log.Error(ctx, "Transcoding returned empty output, ffmpeg may have failed. "+ diff --git a/core/stream/media_streamer_test.go b/core/stream/media_streamer_test.go index fb1c59a60..e06599208 100644 --- a/core/stream/media_streamer_test.go +++ b/core/stream/media_streamer_test.go @@ -1,12 +1,18 @@ package stream_test import ( + "bytes" "context" "errors" "io" + "net/http" + "net/http/httptest" "os" + "testing/iotest" "time" + "github.com/go-chi/chi/v5/middleware" + "github.com/navidrome/navidrome/conf" "github.com/navidrome/navidrome/conf/configtest" "github.com/navidrome/navidrome/core/stream" @@ -140,4 +146,49 @@ var _ = Describe("MediaStreamer", func() { Expect(s.Seekable()).To(BeTrue()) }) }) + + Context("Serve", func() { + var mf *model.MediaFile + BeforeEach(func() { + var err error + mf, err = ds.MediaFile(ctx).Get("123") + Expect(err).ToNot(HaveOccurred()) + }) + + It("keeps empty output a non-error, so callers still reply 200 with an empty body", func() { + s := stream.NewStream(mf, "mp3", 128, io.NopCloser(bytes.NewReader(nil))) + w := httptest.NewRecorder() + r := httptest.NewRequest(http.MethodGet, "/", nil) + + n, err := s.Serve(ctx, w, r) + + Expect(err).ToNot(HaveOccurred()) + Expect(n).To(BeZero()) + Expect(w.Code).To(Equal(http.StatusOK)) + }) + + It("aborts the response when the source fails after sending data", func() { + src := io.NopCloser(io.MultiReader( + bytes.NewReader(bytes.Repeat([]byte("a"), 64*1024)), + iotest.ErrReader(errors.New("transcoder died")), + )) + server := httptest.NewServer(serveHandler(stream.NewStream(mf, "mp3", 128, src))) + DeferCleanup(server.Close) + + resp, err := http.Get(server.URL) + Expect(err).ToNot(HaveOccurred()) + defer resp.Body.Close() + + // A client-side read failure is the only observable proof the response was aborted. + _, err = io.ReadAll(resp.Body) + Expect(err).To(HaveOccurred()) + }) + }) }) + +// Serve runs behind the real server's Recoverer, which must let ErrAbortHandler through. +func serveHandler(s *stream.Stream) http.Handler { + return middleware.Recoverer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = s.Serve(r.Context(), w, r) + })) +} diff --git a/go.mod b/go.mod index 65a01a9b5..c0809f7b5 100644 --- a/go.mod +++ b/go.mod @@ -5,6 +5,9 @@ go 1.26 // Fork to implement raw tags support replace go.senan.xyz/taglib => github.com/deluan/go-taglib v0.0.0-20260720134629-a133b9719ea3 +// Fork to implement CloseWithError, proposed upstream in https://github.com/djherbis/fscache/pull/22 +replace github.com/djherbis/fscache => github.com/deluan/fscache v0.9.1-0.20260825221051-a07d597526e2 + require ( github.com/Masterminds/squirrel v1.5.4 github.com/andybalholm/cascadia v1.3.4 diff --git a/go.sum b/go.sum index c11f4bb20..8be5a910c 100644 --- a/go.sum +++ b/go.sum @@ -29,6 +29,8 @@ github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSs github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1 h1:5RVFMOWjMyRy8cARdy79nAmgYw3hK/4HUq48LQ6Wwqo= github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.1/go.mod h1:ZXNYxsqcloTdSy/rNShjYzMhyjf0LaoftYK0p+A3h40= +github.com/deluan/fscache v0.9.1-0.20260825221051-a07d597526e2 h1:s254V2hsrrCJXYtAn9WPG/5p4QHenfL9E+j6Tiq5MW4= +github.com/deluan/fscache v0.9.1-0.20260825221051-a07d597526e2/go.mod h1:eNFa48vJrse+8ysT4IJnnUeXwLZNcR0JQumU/W/QoUI= github.com/deluan/go-taglib v0.0.0-20260720134629-a133b9719ea3 h1:j7eSXqgtjhlNfwnMEzRdXnJGZTEw4I7J9TeQAll83bU= github.com/deluan/go-taglib v0.0.0-20260720134629-a133b9719ea3/go.mod h1:QGxQ4Z1IWyY9w56xNEFjYAaWE8uSxA/gneQ7RPcFJrY= github.com/deluan/rest v0.0.0-20211102003136-6260bc399cbf h1:tb246l2Zmpt/GpF9EcHCKTtwzrd0HGfEmoODFA/qnk4= @@ -39,8 +41,6 @@ github.com/dexterlb/mpvipc v0.0.0-20260722094525-0cf47d745b36 h1:KtPfdSST6e0vJbM github.com/dexterlb/mpvipc v0.0.0-20260722094525-0cf47d745b36/go.mod h1:RkQWLNITKkXHLP7LXxZSgEq+uFWU25M5qW7qfEhL9Wc= github.com/djherbis/atime v1.1.0 h1:rgwVbP/5by8BvvjBNrbh64Qz33idKT3pSnMSJsxhi0g= github.com/djherbis/atime v1.1.0/go.mod h1:28OF6Y8s3NQWwacXc5eZTsEsiMzp7LF8MbXE+XJPdBE= -github.com/djherbis/fscache v0.10.2-0.20231127215153-442a07e326c4 h1:wdZllsLrDJtYfHiAKogB4PNHSDeO+v+5S3eqSWHGDlc= -github.com/djherbis/fscache v0.10.2-0.20231127215153-442a07e326c4/go.mod h1:dHWjlanKIxaHVH1xJOTb4kzP800XdcXlgJ6JYlR2DPU= github.com/djherbis/stream v1.4.0 h1:aVD46WZUiq5kJk55yxJAyw6Kuera6kmC3i2vEQyW/AE= github.com/djherbis/stream v1.4.0/go.mod h1:cqjC1ZRq3FFwkGmUtHwcldbnW8f0Q4YuVsGW1eAFtOk= github.com/djherbis/times v1.6.0 h1:w2ctJ92J8fBvWPxugmXIv7Nz7Q3iDMKNx9v5ocVH20c= diff --git a/utils/cache/file_caches.go b/utils/cache/file_caches.go index dff9e4e7a..48cd135cb 100644 --- a/utils/cache/file_caches.go +++ b/utils/cache/file_caches.go @@ -255,6 +255,17 @@ func (fc *fileCache) copyAndClose(ctx context.Context, key string, w io.WriteClo } if err == nil { fc.markComplete(ctx, key) + } else if cw, ok := w.(interface{ CloseWithError(error) error }); ok { + // Cancel instead of close, so readers fail with the cause rather than + // draining a truncated entry to a clean EOF. + if cErr := cw.CloseWithError(err); cErr != nil { + // Join, not Append: err is now shared with readers and must not be mutated. + return errors.Join(err, fmt.Errorf("closing cache writer: %w", cErr)) + } + return err + } else { + log.Warn(ctx, "Cache writer cannot report failures; readers will see a truncated entry as a clean EOF", + "cache", fc.name, "key", key, err) } if cErr := w.Close(); cErr != nil { err = multierror.Append(err, fmt.Errorf("closing cache writer: %w", cErr)) diff --git a/utils/cache/file_caches_test.go b/utils/cache/file_caches_test.go index 974200656..3189de6b2 100644 --- a/utils/cache/file_caches_test.go +++ b/utils/cache/file_caches_test.go @@ -259,6 +259,54 @@ var _ = Describe("File Caches", func() { }).Should(BeTrue()) }) + It("gets a writer that can report failures to readers", func() { + // Guards the fork adoption: if the fscache replace directive is ever lost, + // this fails in CI instead of silently reviving the truncation bug. + fc := callNewFileCache("test", "10MB", "test", 0, nil) + _, w, err := fc.cache.Get("capability") + Expect(err).To(BeNil()) + DeferCleanup(func() { _ = w.Close() }) + + _, ok := w.(interface{ CloseWithError(error) error }) + Expect(ok).To(BeTrue(), "fscache writer lost CloseWithError; check the go.mod replace directive") + }) + + It("fails the reader with the cause instead of a clean EOF", func() { + fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) { + return &partialThenErrReader{data: []byte("PARTIAL"), err: errors.New("transcoder died")}, nil + }) + s, err := fc.Get(context.Background(), &testArg{"inband"}) + Expect(err).To(BeNil()) + DeferCleanup(func() { _ = s.Close() }) + + _, err = io.ReadAll(s) + Expect(err).To(MatchError(ContainSubstring("transcoder died"))) + }) + + It("fails a reader that joined mid-write with the same cause", func() { + pr, pw := io.Pipe() + fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) { + return pr, nil + }) + s1, err := fc.Get(context.Background(), &testArg{"joined"}) + Expect(err).To(BeNil()) + DeferCleanup(func() { _ = s1.Close() }) + + // The blocking pipe write gives a happens-before: the entry is in flight. + _, err = pw.Write([]byte("PARTIAL")) + Expect(err).To(BeNil()) + + s2, err := fc.Get(context.Background(), &testArg{"joined"}) + Expect(err).To(BeNil()) + DeferCleanup(func() { _ = s2.Close() }) + Expect(s2.Cached).To(BeTrue()) + + Expect(pw.CloseWithError(errors.New("transcoder died"))).To(Succeed()) + + _, err = io.ReadAll(s2) + Expect(err).To(MatchError(ContainSubstring("transcoder died"))) + }) + It("does not write a completion marker when the write fails after partial bytes", func() { // Mimics a transcode that produces real output and then dies: // the bytes land on disk, but the entry must NOT be marked complete. @@ -304,9 +352,9 @@ var _ = Describe("File Caches", func() { Expect(calls.Load()).To(BeNumerically("==", 2)) }) - It("survives an invalidated entry's deferred file removal", func() { - // invalidate() drops the map entry but defers the unlink until readers close; - // a Get in that window re-creates the file, which the deferred unlink then eats. + It("removes a failed entry promptly, without eating its replacement", func() { + // Cancel closes the failed entry's readers, so its removal no longer defers + // past the point where a new entry re-creates the same file. var n atomic.Int32 fc := callNewFileCache("test", "10MB", "test", 0, func(ctx context.Context, arg Item) (io.Reader, error) { if n.Add(1) == 1 { @@ -319,7 +367,6 @@ var _ = Describe("File Caches", func() { s1, err := fc.Get(context.Background(), &testArg{"deferred"}) Expect(err).To(BeNil()) - // The failed write invalidates the entry; the removal now waits on s1. Eventually(func() bool { return fc.cache.Exists(key) }).Should(BeFalse()) s2, err := fc.Get(context.Background(), &testArg{"deferred"}) @@ -330,15 +377,16 @@ var _ = Describe("File Caches", func() { Expect(s1.Close()).To(Succeed()) dataPath := fcSpreadFS(fc).KeyMapper(key) - Eventually(func() bool { + Consistently(func() error { _, e := os.Stat(dataPath) - return os.IsNotExist(e) - }).Should(BeTrue(), "expected the deferred removal to take the re-created file") + return e + }).Should(Succeed(), "the replacement entry's file must survive the failed entry's cleanup") s3, err := fc.Get(context.Background(), &testArg{"deferred"}) Expect(err).ToNot(HaveOccurred()) Expect(io.ReadAll(s3)).To(Equal([]byte("GOOD"))) _ = s3.Close() + Expect(n.Load()).To(Equal(int32(2)), "the third Get must be served from cache") }) It("re-fetches when an adopted entry's data file vanished", func() {