vrr/navidrome
2
0
Fork 0
mirror of https://github.com/navidrome/navidrome.git synced 2026-07-09 17:18:45 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 4

fix(transcoding): cap concurrent transcodes to prevent ffmpeg DoS (#5522)

Browse source 
* feat(transcoding): add MaxConcurrent and MaxConcurrentPerUser config

Introduce Transcoding.MaxConcurrent (default NumCPU()*2) and
Transcoding.MaxConcurrentPerUser (default 3) to support upcoming
concurrency limits on the streaming pipeline. No behavior change yet.

Refs #5246

* feat(transcoding): add TranscodeLimiter with global and per-user caps

Introduce a non-blocking limiter that gates concurrent transcodes. Returns
ErrTooManyTranscodes immediately when the cap is reached so callers can
translate it into a 429 response, rather than queuing requests.

The per-user reservation is taken first to avoid burning a global slot
that would only be rolled back when the per-user cap rejects the caller.
Release is idempotent so wrapping the transcoder reader's Close is safe.

Refs #5246

* feat(transcoding): cap concurrent transcodes in media streamer

Acquire a TranscodeLimiter slot before spawning ffmpeg in the transcoding
cache's read function, and release it when the resulting reader is closed.
Raw streams and cache hits bypass the limiter so a single saturating client
cannot block ordinary playback.

When the cap is reached, ErrTooManyTranscodes bubbles up through cache.Get,
ready for the HTTP layer to translate into a 429 response.

Refs #5246

* feat(transcoding): return HTTP 429 with Retry-After when transcode cap is hit

Map stream.ErrTooManyTranscodes to HTTP 429 in both the Subsonic API
(/stream, /download) and the public share endpoint, including a 5s
Retry-After hint. The Subsonic response still carries a failed-status
envelope so clients that ignore HTTP codes also see the failure.

Refs #5246

* feat(transcoding): default MaxConcurrent to 0 (disabled)

Ship the limiter opt-in so existing installations are not affected by a
behavior change on upgrade. Users hitting the DoS reported in #5246 can
enable it by setting Transcoding.MaxConcurrent to a positive value
(NumCPU()*2 is a reasonable starting point).

Refs #5246

* fix(transcoding): make global and per-user caps independent

Previously the limiter short-circuited to a no-op whenever MaxConcurrent
was zero, silently ignoring a configured MaxConcurrentPerUser. Treat each
cap independently so an operator can throttle per-user without enforcing
a global ceiling (or vice versa), and only fall back to the no-op limiter
when both caps are disabled.

* fix(archiver): abort archive download when the transcode limiter rejects

The album/artist/playlist zip writers were silently producing zip entries
with headers but no data when ms.NewStream returned ErrTooManyTranscodes,
because the per-file error was discarded by `_ = a.addFileToZip(...)`.
The client received HTTP 200 with a corrupt zip and no indication that
the server was rate-limited.

Now the zip loop bails out as soon as it sees ErrTooManyTranscodes, and
the Download handler swallows the error (the response status and
Content-Disposition are already flushed by the time the limit is hit, so
no 429 can be sent). The truncated zip surfaces the problem to the
client; operators see a clear "transcode cap reached" warning in the
server logs.

Refs #5246

* fix(transcoding): release limiter slot on client close, not ffmpeg EOF

Previously the slot was wrapped around the ffmpeg source reader, so it
was only released by the cache's background copyAndClose goroutine when
ffmpeg finished producing the file — meaning a client that disconnected
after a single byte still held the slot for the full transcode duration.
Under MaxConcurrent=N this serialized fresh requests behind abandoned
encodes for minutes.

Hand the release function back from the cache producer via the streamJob
struct and wire it into the consumer-side Stream.Close. The HTTP handler
already runs `defer stream.Close()`, so disconnect now frees the slot
immediately. Cache hits never enter the producer and still pay no slot,
and singleflight waiters on the same key correctly inherit no release
(only the original producer's job holds the slot).

Refs #5246

* fix(transcoding): skip per-user cap for anonymous requests

Public share viewers have no user in context, so userName(ctx) returned
the literal string "UNKNOWN" and the limiter mapped every anonymous
viewer to the same bucket. With MaxConcurrentPerUser=N, only N
unrelated anonymous clients could stream a viral share at any time —
the opposite of the fairness the per-user cap is meant to provide.

Introduce a limiterKey(ctx) helper that returns "" for anonymous
callers (userName(ctx) is unchanged for logs), and teach Acquire to
skip the per-user reservation when the key is empty. The global cap is
still enforced for anonymous traffic and remains the protection against
runaway anonymous load.

Refs #5246

* refactor(transcoding): tidy limiter struct and centralize Retry-After

Per review feedback:

- Drop the redundant maxConcurrent field on transcodeLimiter; the channel
  capacity already enforces the global cap and the field was only used
  inside the constructor.
- Only allocate the perUser map when MaxConcurrentPerUser > 0.
- Move the Retry-After value into core/stream as RetryAfterSeconds so the
  Subsonic API and public-share handlers cannot drift if the window is
  later tuned.

* fix(transcoding): do not log limiter rejections as cache failures

NewStream was emitting an error-level "Error accessing transcoding cache"
log whenever cache.Get returned anything non-nil, including the limiter's
ErrTooManyTranscodes — even though the producer had already logged the
rejection at warn level. The result was double logging and a misleading
"cache failure" classification that buries real cache problems.

Skip the error log when the cause is ErrTooManyTranscodes; the warn line
from the producer is the canonical signal.

* fix(archiver): open stream before writing zip entry header

Per review: addFileToZip previously called z.CreateHeader before
NewStream, so when the limiter rejected a transcode the zip already
contained a 0-byte entry for that track. Open the source first and only
write the header once the read side is ready; rejections now skip the
entry entirely.

The truncation comment in handleArchiveErr was also misleading — z.Close
finalises the central directory, so the client receives a well-formed
zip containing only the tracks written before the rejection, not a
"truncated" archive. Reword to match reality.

* fix(transcoding): hold slot for ffmpeg lifetime, force cancellable ctx

The previous release-on-consumer-close design let a client open many
unique transcodes, disconnect immediately, and still spawn the
configured cap's worth of ffmpeg processes — the cache writer goroutine
continued draining ffmpeg to disk after the client disappeared, defeating
the DoS protection the limiter is meant to provide.

Move the release back onto the source reader so the slot is freed only
when ffmpeg actually exits (either EOF or context cancellation). To keep
disconnects from leaking slots for the full transcode duration, force
the request context into ffmpeg whenever the limiter is enabled — so
client disconnect cancels the process and frees the slot promptly.

When the limiter is disabled, the legacy EnableTranscodingCancellation
behavior is preserved unchanged.

Reported by codex and Copilot reviewers on #5522.
This commit is contained in:
Deluan Quintão 2026-05-24 00:24:30 -03:00 • committed by GitHub
parent 55a31f30b3
commit 945d0ba1e2
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 571 additions and 35 deletions
Download patch file Download diff file Expand all files Collapse all files

8
conf/configuration.go
View file

@ -113,6 +113,7 @@ type configOptions struct {
PID pidOptions `json:",omitzero"`
Inspect inspectOptions `json:",omitzero"`
Subsonic subsonicOptions `json:",omitzero"`
Transcoding transcodingOptions `json:",omitzero"`
LastFM lastfmOptions `json:",omitzero"`
Deezer deezerOptions `json:",omitzero"`
ListenBrainz listenBrainzOptions `json:",omitzero"`
@ -165,6 +166,11 @@ type scannerOptions struct {
PurgeMissing string // Values: "never", "always", "full"
}
type transcodingOptions struct {
MaxConcurrent int
MaxConcurrentPerUser int
}
type subsonicOptions struct {
AppendSubtitle bool
AppendAlbumVersion bool
@ -822,6 +828,8 @@ func setViperDefaults() {
viper.SetDefault("subsonic.enableaveragerating", true)
viper.SetDefault("subsonic.legacyclients", "DSub")
viper.SetDefault("subsonic.minimalclients", "SubMusic")
viper.SetDefault("transcoding.maxconcurrent", 0)
viper.SetDefault("transcoding.maxconcurrentperuser", 0)
viper.SetDefault("agents", "deezer,lastfm,listenbrainz")
viper.SetDefault("lastfm.enabled", true)
viper.SetDefault("lastfm.language", consts.DefaultInfoLanguage)

56
core/archiver.go
View file

@ -3,6 +3,7 @@ package core
import (
"archive/zip"
"context"
"errors"
"fmt"
"io"
"os"
@ -60,7 +61,15 @@ func (a *archiver) zipAlbums(ctx context.Context, id string, format string, bitr
"format", format, "bitrate", bitrate, "isMultiDisc", isMultiDisc, "numTracks", len(album))
for _, mf := range album {
file := a.albumFilename(mf, format, isMultiDisc)
_ = a.addFileToZip(ctx, z, mf, format, bitrate, file)
if addErr := a.addFileToZip(ctx, z, mf, format, bitrate, file); errors.Is(addErr, stream.ErrTooManyTranscodes) {
// Stop iterating: continuing would just rack up more
// rejections from the limiter. Close finalises whatever
// tracks were already written; the rejected one is not
// present in the archive (addFileToZip aborts before
// writing its entry header).
_ = z.Close()
return addErr
}
}
}
err = z.Close()
@ -120,7 +129,12 @@ func (a *archiver) zipMediaFiles(ctx context.Context, id, name string, format st
zippedMfs := make(model.MediaFiles, len(mfs))
for idx, mf := range mfs {
file := a.playlistFilename(mf, format, idx)
_ = a.addFileToZip(ctx, z, mf, format, bitrate, file)
if addErr := a.addFileToZip(ctx, z, mf, format, bitrate, file); errors.Is(addErr, stream.ErrTooManyTranscodes) {
// Abort the whole archive: continuing would silently emit
// empty zip entries since the headers are already written.
_ = z.Close()
return addErr
}
mf.Path = file
zippedMfs[idx] = mf
}
@ -162,6 +176,27 @@ func (a *archiver) playlistFilename(mf model.MediaFile, format string, idx int)
func (a *archiver) addFileToZip(ctx context.Context, z *zip.Writer, mf model.MediaFile, format string, bitrate int, filename string) error {
path := mf.AbsolutePath()
// Open the source before writing the zip entry header so a rejection
// (limiter, missing file, etc.) does not leave an empty entry in the
// archive.
var r io.ReadCloser
var err error
if format != "raw" && format != "" {
r, err = a.ms.NewStream(ctx, &mf, stream.Request{Format: format, BitRate: bitrate})
} else {
r, err = os.Open(path)
}
if err != nil {
log.Error(ctx, "Error opening file for zipping", "file", path, "format", format, err)
return err
}
defer func() {
if err := r.Close(); err != nil && log.IsGreaterOrEqualTo(log.LevelDebug) {
log.Error(ctx, "Error closing stream", "id", mf.ID, "file", path, err)
}
}()
w, err := z.CreateHeader(&zip.FileHeader{
Name: filename,
Modified: mf.UpdatedAt,
@ -172,23 +207,6 @@ func (a *archiver) addFileToZip(ctx context.Context, z *zip.Writer, mf model.Med
return err
}
var r io.ReadCloser
if format != "raw" && format != "" {
r, err = a.ms.NewStream(ctx, &mf, stream.Request{Format: format, BitRate: bitrate})
} else {
r, err = os.Open(path)
}
if err != nil {
log.Error(ctx, "Error opening file for zipping", "file", path, "format", format, err)
return err
}
defer func() {
if err := r.Close(); err != nil && log.IsGreaterOrEqualTo(log.LevelDebug) {
log.Error(ctx, "Error closing stream", "id", mf.ID, "file", path, err)
}
}()
_, err = io.Copy(w, r)
if err != nil {
log.Error(ctx, "Error zipping file", "file", path, err)

26
core/archiver_test.go
View file

@ -89,6 +89,32 @@ var _ = Describe("Archiver", func() {
})
})
Context("when the transcode limiter rejects a file", func() {
It("aborts the archive instead of continuing with empty entries", func() {
mfs := model.MediaFiles{
{Path: "test_data/01 - track1.mp3", Suffix: "mp3", AlbumID: "1", Album: "Album", DiscNumber: 1},
{Path: "test_data/02 - track2.mp3", Suffix: "mp3", AlbumID: "1", Album: "Album", DiscNumber: 1},
}
mfRepo := &mockMediaFileRepository{}
mfRepo.On("GetAll", []model.QueryOptions{{
Filters: squirrel.Eq{"album_id": "1"},
Sort: "album",
}}).Return(mfs, nil)
ds.On("MediaFile", mock.Anything).Return(mfRepo)
ms.On("NewStream", mock.Anything, mock.Anything, stream.Request{Format: "mp3", BitRate: 128}).
Return(nil, stream.ErrTooManyTranscodes).Once()
out := new(bytes.Buffer)
err := arch.ZipAlbum(context.Background(), "1", "mp3", 128, out)
Expect(err).To(MatchError(stream.ErrTooManyTranscodes))
// NewStream should only have been called once: the loop must bail
// out on the rejection instead of trying every remaining track.
ms.AssertNumberOfCalls(GinkgoT(), "NewStream", 1)
})
})
Context("ZipShare", func() {
It("zips a share correctly", func() {
mfs := model.MediaFiles{

135
core/stream/limiter.go Normal file
View file

@ -0,0 +1,135 @@
package stream
import (
"context"
"errors"
"io"
"sync"
"sync/atomic"
)
// ErrTooManyTranscodes is returned by TranscodeLimiter.Acquire when the
// configured concurrency cap has been reached. Callers should translate this
// into an HTTP 429 response so well-behaved clients back off and retry.
var ErrTooManyTranscodes = errors.New("too many concurrent transcodes")
// RetryAfterSeconds is the value returned in the HTTP Retry-After header when
// a request is rejected with ErrTooManyTranscodes. Most transcodes finish well
// within this window, so retrying after this delay typically succeeds.
const RetryAfterSeconds = 5
// TranscodeLimiter gates the number of concurrent ffmpeg transcodes. It enforces
// both a global cap (to protect the host from process exhaustion) and an optional
// per-user cap (to keep one client from starving the others). Acquire never
// blocks: it either reserves a slot or returns ErrTooManyTranscodes immediately.
type TranscodeLimiter interface {
// Acquire reserves a slot for the given user. On success it returns a release
// function that must be called exactly once when the transcode is done.
// Calling release more than once is safe and idempotent.
Acquire(ctx context.Context, user string) (release func(), err error)
// Enabled reports whether the limiter actually enforces any cap. Callers
// can use it to decide whether to bind ffmpeg's lifetime to the request
// context so disconnects free slots quickly, rather than letting the
// process drain to completion in the background.
Enabled() bool
}
// NewTranscodeLimiter returns a limiter enforcing the given caps. Each cap is
// independent: a value of zero or less disables that cap. When both caps are
// disabled the limiter is a no-op.
func NewTranscodeLimiter(maxConcurrent, maxPerUser int) TranscodeLimiter {
if maxConcurrent <= 0 && maxPerUser <= 0 {
return noopLimiter{}
}
l := &transcodeLimiter{maxPerUser: maxPerUser}
if maxConcurrent > 0 {
l.global = make(chan struct{}, maxConcurrent)
}
if maxPerUser > 0 {
l.perUser = make(map[string]int)
}
return l
}
// releasingReadCloser wraps an io.ReadCloser so that closing it also releases
// the limiter slot exactly once. release must be the function returned by
// TranscodeLimiter.Acquire; its own idempotency makes double-Close safe too.
type releasingReadCloser struct {
io.ReadCloser
release func()
}
func (r *releasingReadCloser) Close() error {
err := r.ReadCloser.Close()
r.release()
return err
}
type noopLimiter struct{}
func (noopLimiter) Acquire(context.Context, string) (func(), error) {
return func() {}, nil
}
func (noopLimiter) Enabled() bool { return false }
type transcodeLimiter struct {
maxPerUser int
global chan struct{}
mu sync.Mutex
perUser map[string]int
}
func (*transcodeLimiter) Enabled() bool { return true }
func (l *transcodeLimiter) Acquire(_ context.Context, user string) (func(), error) {
// Reserve a per-user slot first so a noisy user can't burn through
// global slots only to be rejected later. An empty user key means
// "anonymous" (e.g. public share viewers); we skip the per-user cap
// entirely so unrelated anonymous clients do not share a bucket.
perUserActive := l.maxPerUser > 0 && user != ""
if perUserActive {
l.mu.Lock()
if l.perUser[user] >= l.maxPerUser {
l.mu.Unlock()
return nil, ErrTooManyTranscodes
}
l.perUser[user]++
l.mu.Unlock()
}
if l.global != nil {
select {
case l.global <- struct{}{}:
default:
if perUserActive {
l.releasePerUser(user)
}
return nil, ErrTooManyTranscodes
}
}
var released atomic.Bool
return func() {
if !released.CompareAndSwap(false, true) {
return
}
if l.global != nil {
<-l.global
}
if perUserActive {
l.releasePerUser(user)
}
}, nil
}
func (l *transcodeLimiter) releasePerUser(user string) {
l.mu.Lock()
defer l.mu.Unlock()
l.perUser[user]--
if l.perUser[user] <= 0 {
delete(l.perUser, user)
}
}

186
core/stream/limiter_test.go Normal file
View file

@ -0,0 +1,186 @@
package stream_test
import (
"context"
"errors"
"sync"
"github.com/navidrome/navidrome/core/stream"
"github.com/navidrome/navidrome/log"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
)
var _ = Describe("TranscodeLimiter", func() {
ctx := log.NewContext(context.TODO())
Describe("Disabled (both caps <= 0)", func() {
It("never blocks and never returns ErrTooManyTranscodes", func() {
lim := stream.NewTranscodeLimiter(0, 0)
for range 100 {
rel, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
Expect(rel).ToNot(BeNil())
}
})
})
Describe("Per-user cap only (no global cap)", func() {
It("still enforces the per-user limit when MaxConcurrent is disabled", func() {
lim := stream.NewTranscodeLimiter(0, 2)
rel1, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel2, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
_, err = lim.Acquire(ctx, "alice")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
// Other users have their own buckets.
rel3, err := lim.Acquire(ctx, "bob")
Expect(err).ToNot(HaveOccurred())
rel1()
_, err = lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel2()
rel3()
})
})
Describe("Global cap", func() {
It("rejects requests beyond MaxConcurrent with ErrTooManyTranscodes", func() {
lim := stream.NewTranscodeLimiter(2, 0)
rel1, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel2, err := lim.Acquire(ctx, "bob")
Expect(err).ToNot(HaveOccurred())
_, err = lim.Acquire(ctx, "carol")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
rel1()
_, err = lim.Acquire(ctx, "carol")
Expect(err).ToNot(HaveOccurred())
rel2()
})
It("releases a slot only once even if release is called multiple times", func() {
lim := stream.NewTranscodeLimiter(1, 0)
rel, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel()
rel()
rel()
// After releases, exactly one slot should be available.
_, err = lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
_, err = lim.Acquire(ctx, "alice")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
})
})
Describe("Per-user cap", func() {
It("rejects a user beyond MaxConcurrentPerUser even if global slots remain", func() {
lim := stream.NewTranscodeLimiter(10, 2)
rel1, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel2, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
_, err = lim.Acquire(ctx, "alice")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
// A different user is unaffected.
rel3, err := lim.Acquire(ctx, "bob")
Expect(err).ToNot(HaveOccurred())
rel1()
_, err = lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
rel2()
rel3()
})
It("skips the per-user cap for anonymous users (empty key)", func() {
// Anonymous requests (e.g. public share viewers) deliberately
// bypass the per-user cap so unrelated anonymous clients are not
// collapsed into a single shared bucket. The global cap remains
// the only ceiling on anonymous traffic.
lim := stream.NewTranscodeLimiter(10, 1)
rels := make([]func(), 0, 5)
for range 5 {
rel, err := lim.Acquire(ctx, "")
Expect(err).ToNot(HaveOccurred())
rels = append(rels, rel)
}
for _, rel := range rels {
rel()
}
})
It("still applies the global cap to anonymous users", func() {
lim := stream.NewTranscodeLimiter(2, 1)
rel1, err := lim.Acquire(ctx, "")
Expect(err).ToNot(HaveOccurred())
rel2, err := lim.Acquire(ctx, "")
Expect(err).ToNot(HaveOccurred())
_, err = lim.Acquire(ctx, "")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
rel1()
rel2()
})
})
Describe("Concurrent safety", func() {
It("survives parallel Acquire/release with consistent counts", func() {
lim := stream.NewTranscodeLimiter(5, 0)
var wg sync.WaitGroup
var acquired int64
var rejected int64
var mu sync.Mutex
for i := range 50 {
wg.Add(1)
go func(i int) {
defer wg.Done()
rel, err := lim.Acquire(ctx, "alice")
mu.Lock()
if err == nil {
acquired++
mu.Unlock()
rel()
} else {
rejected++
mu.Unlock()
}
_ = i
}(i)
}
wg.Wait()
Expect(acquired + rejected).To(Equal(int64(50)))
// After all releases, all 5 slots should be free again.
for range 5 {
_, err := lim.Acquire(ctx, "alice")
Expect(err).ToNot(HaveOccurred())
}
_, err := lim.Acquire(ctx, "alice")
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
})
})
})

63
core/stream/media_streamer.go
View file

@ -2,6 +2,7 @@ package stream
import (
"context"
"errors"
"fmt"
"io"
"mime"
@ -28,13 +29,19 @@ type MediaStreamer interface {
type TranscodingCache cache.FileCache
func NewMediaStreamer(ds model.DataStore, t ffmpeg.FFmpeg, cache TranscodingCache) MediaStreamer {
return &mediaStreamer{ds: ds, transcoder: t, cache: cache}
return &mediaStreamer{
ds: ds,
transcoder: t,
cache: cache,
limiter: NewTranscodeLimiter(conf.Server.Transcoding.MaxConcurrent, conf.Server.Transcoding.MaxConcurrentPerUser),
}
}
type mediaStreamer struct {
ds model.DataStore
transcoder ffmpeg.FFmpeg
cache cache.FileCache
limiter TranscodeLimiter
}
type streamJob struct {
@ -104,7 +111,12 @@ func (ms *mediaStreamer) NewStream(ctx context.Context, mf *model.MediaFile, req
}
r, err := ms.cache.Get(ctx, job)
if err != nil {
log.Error(ctx, "Error accessing transcoding cache", "id", mf.ID, err)
// Rate-limit rejections are already logged at warn level by the
// producer; treating them as cache failures here would both
// double-log and mask actual cache problems.
if !errors.Is(err, ErrTooManyTranscodes) {
log.Error(ctx, "Error accessing transcoding cache", "id", mf.ID, err)
}
return nil, err
}
cached = r.Cached
@ -217,15 +229,31 @@ func NewTranscodingCache() TranscodingCache {
return nil, os.ErrInvalid
}
// Choose the appropriate context based on EnableTranscodingCancellation configuration.
// This is where we decide whether transcoding processes should be cancellable or not.
release, err := job.ms.limiter.Acquire(ctx, limiterKey(ctx))
if err != nil {
log.Warn(ctx, "Refusing transcode: concurrent transcode limit reached",
"id", job.mf.ID, "user", userName(ctx),
"maxConcurrent", conf.Server.Transcoding.MaxConcurrent,
"maxPerUser", conf.Server.Transcoding.MaxConcurrentPerUser)
return nil, err
}
// Choose the context that drives the ffmpeg process.
//
// When the limiter is enabled, force the request context so a
// client disconnect cancels ffmpeg and frees the slot promptly.
// Otherwise a client could open many transcodes, disconnect
// immediately, and still leave the configured cap's worth of
// ffmpeg processes draining in the background — which is exactly
// the DoS the limiter is meant to prevent.
//
// When the limiter is disabled, preserve the legacy behavior
// governed by EnableTranscodingCancellation so this PR does not
// change observable behavior for operators who have not opted in.
var transcodingCtx context.Context
if conf.Server.EnableTranscodingCancellation {
// Use the request context directly, allowing cancellation when client disconnects
if job.ms.limiter.Enabled() || conf.Server.EnableTranscodingCancellation {
transcodingCtx = ctx
} else {
// Use background context with request values preserved.
// This prevents cancellation but maintains request metadata (user, client, etc.)
transcodingCtx = request.AddValues(context.Background(), ctx)
}
@ -240,10 +268,14 @@ func NewTranscodingCache() TranscodingCache {
Offset: job.offset,
})
if err != nil {
release()
log.Error(ctx, "Error starting transcoder", "id", job.mf.ID, err)
return nil, os.ErrInvalid
}
return out, nil
// Tie the slot to the ffmpeg process: copyAndClose calls Close
// on this reader after io.Copy returns, which is exactly when
// ffmpeg has exited (either EOF or context cancellation).
return &releasingReadCloser{ReadCloser: out, release: release}, nil
})
}
@ -255,3 +287,16 @@ func userName(ctx context.Context) string {
return user.UserName
}
}
// limiterKey returns the per-user bucket key used by the transcode limiter.
// For anonymous requests (e.g. public shares) it returns the empty string,
// which signals the limiter to skip the per-user cap entirely — otherwise
// every anonymous viewer of a public share would collide on the same key
// and starve each other within MaxConcurrentPerUser slots. The global cap
// still applies and remains the protection against runaway anonymous load.
func limiterKey(ctx context.Context) string {
if user, ok := request.UserFrom(ctx); ok {
return user.UserName
}
return ""
}

54
core/stream/media_streamer_test.go
View file

@ -2,6 +2,7 @@ package stream_test
import (
"context"
"errors"
"io"
"os"
@ -10,6 +11,7 @@ import (
"github.com/navidrome/navidrome/core/stream"
"github.com/navidrome/navidrome/log"
"github.com/navidrome/navidrome/model"
"github.com/navidrome/navidrome/model/request"
"github.com/navidrome/navidrome/tests"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
@ -61,6 +63,58 @@ var _ = Describe("MediaStreamer", func() {
Expect(s.Seekable()).To(BeFalse())
Expect(s.Duration()).To(Equal(float32(257.0)))
})
It("rejects transcode requests beyond MaxConcurrent with ErrTooManyTranscodes", func() {
// Rebuild the streamer with a tight cap. The first request will hold the
// ffmpeg reader open (we don't read/close it), saturating the single slot.
conf.Server.Transcoding.MaxConcurrent = 1
conf.Server.Transcoding.MaxConcurrentPerUser = 0
tightStreamer := stream.NewMediaStreamer(ds, ffmpeg, stream.NewTranscodingCache())
userCtx := request.WithUsername(ctx, "alice")
s1, err := tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "mp3", BitRate: 64})
Expect(err).ToNot(HaveOccurred())
defer s1.Close()
// Different cache key so it doesn't dedupe with the first request.
_, err = tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "mp3", BitRate: 96})
Expect(errors.Is(err, stream.ErrTooManyTranscodes)).To(BeTrue())
})
It("releases the slot once the stream is closed", func() {
conf.Server.Transcoding.MaxConcurrent = 1
conf.Server.Transcoding.MaxConcurrentPerUser = 0
tightStreamer := stream.NewMediaStreamer(ds, ffmpeg, stream.NewTranscodingCache())
userCtx := request.WithUsername(ctx, "alice")
s1, err := tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "mp3", BitRate: 64})
Expect(err).ToNot(HaveOccurred())
_, _ = io.ReadAll(s1)
_ = s1.Close()
Eventually(func() bool { return ffmpeg.IsClosed() }, "3s").Should(BeTrue())
// Slot should now be free for a different transcode.
s2, err := tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "mp3", BitRate: 96})
Expect(err).ToNot(HaveOccurred())
defer s2.Close()
})
It("does not consume a slot for raw streams", func() {
conf.Server.Transcoding.MaxConcurrent = 1
conf.Server.Transcoding.MaxConcurrentPerUser = 0
tightStreamer := stream.NewMediaStreamer(ds, ffmpeg, stream.NewTranscodingCache())
userCtx := request.WithUsername(ctx, "alice")
// First, saturate the single transcode slot.
s1, err := tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "mp3", BitRate: 64})
Expect(err).ToNot(HaveOccurred())
defer s1.Close()
// Raw stream must still succeed.
s2, err := tightStreamer.NewStream(userCtx, mf, stream.Request{Format: "raw"})
Expect(err).ToNot(HaveOccurred())
defer s2.Close()
})
It("returns a seekable stream if the file is complete in the cache", func() {
s, err := streamer.NewStream(ctx, mf, stream.Request{Format: "mp3", BitRate: 32})
Expect(err).To(BeNil())

9
server/public/handle_streams.go
View file

@ -7,7 +7,7 @@ import (
"time"
"github.com/navidrome/navidrome/core/auth"
"github.com/navidrome/navidrome/core/stream"
streampkg "github.com/navidrome/navidrome/core/stream"
"github.com/navidrome/navidrome/log"
"github.com/navidrome/navidrome/model"
. "github.com/navidrome/navidrome/utils/gg"
@ -48,10 +48,15 @@ func (pub *Router) handleStream(w http.ResponseWriter, r *http.Request) {
return
}
stream, err := pub.streamer.NewStream(ctx, mf, stream.Request{
stream, err := pub.streamer.NewStream(ctx, mf, streampkg.Request{
Format: info.format, BitRate: info.bitrate,
})
if err != nil {
if errors.Is(err, streampkg.ErrTooManyTranscodes) {
w.Header().Set("Retry-After", strconv.Itoa(streampkg.RetryAfterSeconds))
http.Error(w, "too many concurrent transcodes, please retry shortly", http.StatusTooManyRequests)
return
}
log.Error(ctx, "Error starting shared stream", err)
http.Error(w, "invalid request", http.StatusInternalServerError)
return

26
server/subsonic/api.go
View file

@ -7,6 +7,7 @@ import (
"fmt"
"net/http"
"regexp"
"strconv"
"github.com/go-chi/chi/v5"
"github.com/navidrome/navidrome/conf"
@ -304,6 +305,8 @@ func mapToSubsonicError(err error) subError {
err = newError(responses.ErrorDataNotFound, "data not found")
case errors.Is(err, model.ErrNotAuthorized):
err = newError(responses.ErrorAuthorizationFail)
case errors.Is(err, stream.ErrTooManyTranscodes):
err = newError(responses.ErrorGeneric, "too many concurrent transcodes, please retry shortly")
default:
err = newError(responses.ErrorGeneric, fmt.Sprintf("Internal Server Error: %s", err))
}
@ -313,15 +316,31 @@ func mapToSubsonicError(err error) subError {
}
func sendError(w http.ResponseWriter, r *http.Request, err error) {
if errors.Is(err, stream.ErrTooManyTranscodes) {
w.Header().Set("Retry-After", strconv.Itoa(stream.RetryAfterSeconds))
sendResponseWithStatus(w, r, errorResponse(err), http.StatusTooManyRequests)
return
}
sendResponse(w, r, errorResponse(err))
}
func errorResponse(err error) *responses.Subsonic {
subErr := mapToSubsonicError(err)
response := newResponse()
response.Status = responses.StatusFailed
response.Error = &responses.Error{Code: subErr.code, Message: subErr.Error()}
sendResponse(w, r, response)
return response
}
func sendResponse(w http.ResponseWriter, r *http.Request, payload *responses.Subsonic) {
sendResponseWithStatus(w, r, payload, 0)
}
// sendResponseWithStatus writes the response body in the format requested by
// the client. When status is non-zero, WriteHeader is called with that code
// before the body is written; callers that need to set additional headers
// (e.g. Retry-After) must set them before calling.
func sendResponseWithStatus(w http.ResponseWriter, r *http.Request, payload *responses.Subsonic, status int) {
p := req.Params(r)
f, _ := p.String("f")
var response []byte
@ -356,6 +375,9 @@ func sendResponse(w http.ResponseWriter, r *http.Request, payload *responses.Sub
sendError(w, r, err)
return
}
if status != 0 {
w.WriteHeader(status)
}
if payload.Status == responses.StatusOK {
if log.IsGreaterOrEqualTo(log.LevelTrace) {

20
server/subsonic/api_test.go
View file

@ -3,11 +3,13 @@ package subsonic
import (
"encoding/json"
"encoding/xml"
"fmt"
"math"
"net/http"
"net/http/httptest"
"strings"
"github.com/navidrome/navidrome/core/stream"
"github.com/navidrome/navidrome/server/subsonic/responses"
. "github.com/onsi/ginkgo/v2"
. "github.com/onsi/gomega"
@ -152,6 +154,24 @@ var _ = Describe("sendResponse", func() {
})
})
It("responds with HTTP 429 and Retry-After when the transcode limiter rejects", func() {
w = httptest.NewRecorder()
r = httptest.NewRequest("GET", "/rest/stream", nil)
sendError(w, r, fmt.Errorf("rejected: %w", stream.ErrTooManyTranscodes))
Expect(w.Code).To(Equal(http.StatusTooManyRequests))
Expect(w.Header().Get("Retry-After")).ToNot(BeEmpty())
var subsonicResponse responses.Subsonic
err := xml.Unmarshal(w.Body.Bytes(), &subsonicResponse)
Expect(err).NotTo(HaveOccurred())
Expect(subsonicResponse.Status).To(Equal(responses.StatusFailed))
Expect(subsonicResponse.Error).ToNot(BeNil())
Expect(subsonicResponse.Error.Code).To(Equal(responses.ErrorGeneric))
Expect(subsonicResponse.Error.Message).To(ContainSubstring("transcode"))
})
It("updates status pointer when an error occurs", func() {
pointer := int32(0)

23
server/subsonic/stream.go
View file

@ -1,12 +1,15 @@
package subsonic
import (
"context"
"errors"
"fmt"
"net/http"
"strconv"
"strings"
"github.com/navidrome/navidrome/conf"
"github.com/navidrome/navidrome/core/stream"
"github.com/navidrome/navidrome/log"
"github.com/navidrome/navidrome/model"
"github.com/navidrome/navidrome/model/request"
@ -119,14 +122,28 @@ func (api *Router) Download(w http.ResponseWriter, r *http.Request) (*responses.
return nil, err
case *model.Album:
setHeaders(v.Name)
return nil, api.archiver.ZipAlbum(ctx, id, format, maxBitRate, w)
return nil, handleArchiveErr(ctx, id, api.archiver.ZipAlbum(ctx, id, format, maxBitRate, w))
case *model.Artist:
setHeaders(v.Name)
return nil, api.archiver.ZipArtist(ctx, id, format, maxBitRate, w)
return nil, handleArchiveErr(ctx, id, api.archiver.ZipArtist(ctx, id, format, maxBitRate, w))
case *model.Playlist:
setHeaders(v.Name)
return nil, api.archiver.ZipPlaylist(ctx, id, format, maxBitRate, w)
return nil, handleArchiveErr(ctx, id, api.archiver.ZipPlaylist(ctx, id, format, maxBitRate, w))
default:
return nil, model.ErrNotFound
}
}
// handleArchiveErr swallows ErrTooManyTranscodes from archive downloads so the
// outer error handler does not try to write a 429 onto a response whose status
// and Content-Disposition have already been flushed. The archive ends up with
// the tracks that were written before the rejection (the rejected track and
// any following ones are omitted); the server-side log is the unambiguous
// signal operators can act on.
func handleArchiveErr(ctx context.Context, id string, err error) error {
if errors.Is(err, stream.ErrTooManyTranscodes) {
log.Warn(ctx, "Archive download finalized early: transcode cap reached", "id", id, err)
return nil
}
return err
}