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

@@ -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. "+

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)
+	}))
+}

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

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=

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))

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() {