From 53d54baef04254c6243b5dd842db239fd0378cc1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Deluan=20Quint=C3=A3o?= Date: Wed, 15 Jul 2026 14:07:00 -0400 Subject: [PATCH] fix(jellyfin): stream collection responses to prevent OOM on large libraries (#5783) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(jellyfin): stream /Items responses to prevent OOM on large libraries Finamp's library sync issues an unbounded GET /Items?IncludeItemTypes=Audio with Fields=MediaSources and no Limit. On a large library this built the whole result set — every MediaFile and every BaseItemDto, each fat with MediaSources — in memory, and json.Encoder then buffered the entire ~200MB response before writing a byte. Measured on a 96k-track library, one such request peaked near 1.6GB RSS; in a memory-limited container the resulting slowness made clients retry, stacking concurrent full-library builds until the process was OOM-killed. Stream the song listing straight from a DB cursor (MediaFileRepository.GetCursor), mapping and encoding one row at a time, so peak memory is bounded to about one item regardless of library size. The cursor is opened lazily, when streaming begins, so it doesn't hold a DB connection across the CountAll and ServerId lookups that run first (ServerId can write on first use and would deadlock against an open reader). Pagination still works — the cursor query carries LIMIT/OFFSET/ORDER BY from StartIndex/Limit/SortBy — and TotalRecordCount stays the full count. Other materialized responses stream their JSON too, avoiding the encoder's buffer-the-whole-output cost. Also bound artwork image concurrency with the same server.ThrottleBacklog that Subsonic's getCoverArt uses, so a burst of image requests during sync can't exhaust memory through concurrent decode/resize. Verified against a copy of a 96k-track production database: byte-for-byte identical response, peak RSS 1593MB -> 53MB, time-to-first-byte 8.1s -> 0.2s. * fix(jellyfin): fail loudly on /Items streaming errors instead of a truncated 200 Addresses code review: a streamed /Items response commits HTTP 200 before an error can occur, so failures were surfacing as misleadingly successful bodies. - A cursor-open failure (e.g. a busy DB under connection pressure) is now surfaced before the first byte is written: the cursor open is deferred into the item source and run in writeItems after the ServerId lookup but before the envelope, returning a clean 500 the client can retry — not a 200 with an empty item list. - A mid-stream (row-scan) error now aborts without closing the JSON envelope, leaving the body malformed. A truncated-but-valid response would let a sync client (Finamp) treat the short list as the whole library and prune local tracks; malformed JSON forces the client's parser to fail and retry. * refactor(jellyfin): stream every collection endpoint from a DB cursor Streaming was applied only to the song listing, which left two write paths, two ServerId stamping mechanisms and a listXxx return-type split ("songs is special") that made the code hard to follow. Add GetCursor to the album, artist, genre and playlist repositories, mirroring MediaFileRepository.GetCursor: same select builder as GetAll, so each cursor yields identical, fully-hydrated rows (hydration happens per-row in PostScan, and toModels is only a deref loop). The playlist cursor keeps GetAll's owner/public visibility filter. Each repository test asserts the cursor yields exactly what GetAll returns, including Max/Offset. With cursors everywhere, every listXxx returns itemsResult and every collection streams through one writer: - listAlbums, listArtists and listPlaylists now stream from their cursors; search paths stay materialized (Search returns a slice). - listGenres stays materialized: its total is the length of the full list and it paginates in memory, so there is nothing for a cursor to page over. - api.ok's QueryResult case delegates to writeItems, so the ~9 inline callers funnel into the same writer; streamQueryResult and wrapResult are gone. - /Items/Latest streams as a bare JSON array (its Jellyfin wire shape) via writeItemsArray, which shares the item loop and stamping. That also closes the same unbounded hole /Items had: limit=0 disabled its LIMIT. - ServerId is now stamped in exactly one place, so api.ok only handles single, non-collection payloads. Verified against a copy of a 96k-track production database: every endpoint byte-for-byte identical to master. Unbounded /Items peak RSS 1691MB -> 54MB; time-to-first-byte 6.3s -> 0.19s, /Artists 1.14s -> 0.13s. * refactor(jellyfin): route every response through api.ok Handlers were split between api.ok and api.writeItems with no clear rule for which to call. api.ok now accepts itemsResult too, so it is the single entry point: callers hand it whatever they have and it routes collections (cursor -backed or materialized) to the streaming writer. writeItems is reached only through api.ok now. The one exception is /Items/Latest, which returns a bare JSON array rather than a QueryResult envelope and so writes directly — noted in both doc comments. Also drop GenreRepository.GetCursor: listGenres derives its total from the length of the full list and paginates in memory, so there is nothing for a cursor to page over, leaving the method unused. * fix(jellyfin): stream the unbounded multi-type /Items merge The multi-type merge capped each per-type query at offset+limit only when a Limit was given. Without one (Finamp's favorites screen sends multi-type), every type ran an unbounded query and collect() drained each cursor into a slice — so /Items?IncludeItemTypes=MusicAlbum,Audio with no Limit materialized every album and every song, the same OOM class this branch set out to fix. The comment claiming the set was "capped at offset+limit per type" was wrong for limit=0. Without a limit the merged page is just each type's rows in order minus the first offset, which is exactly what chaining the per-type cursors yields, so stream that instead of merging in memory. The bounded path is unchanged: with a Limit each type holds at most offset+limit rows, so merging and paginating across the combined list is safe. Cursors are opened one at a time (each pins a DB connection until drained), with the first opened eagerly so the usual failure is still a clean error before any byte is written. Measured on a 96k-track library, /Items?IncludeItemTypes=MusicAlbum,Audio with no Limit (103,393 items): peak RSS 612MB -> 53MB, time-to-first-byte 4.9s -> 0.6s, response byte-for-byte identical. * refactor(jellyfin): parse /Items params into a struct The /Items dispatcher was a single 90-line block that parsed a dozen params and then threaded them positionally through three layers — queryItemsOfType took 11 arguments, listSongs and listAlbums 9 each — so reading any one of them meant decoding a long argument list. Parse once into an itemsQuery and pass that instead. queryItems now reads as the four things it actually does: parse, the id/playlists-folder/playlist-parent special cases, single-type dispatch, multi-type merge — with the playlist-parent and merge bodies moved to playlistTracks and mergeTypes. Every listXxx takes (ctx, opts, q). No behavior change: parsing, ordering and the entityParent rule are unchanged, and /Artists still passes favOnly=false explicitly by building the subset of the query it uses. * docs(jellyfin): trim comments on the streaming path Cut the comments back to the non-obvious why: the deferred cursor open (the ServerId write would deadlock against an open reader), the deliberate malformed JSON on a mid-stream error, why chained opens one cursor at a time, why listGenres stays materialized, and why the cursor helpers take the underlying func type. Dropped the rest — restatements of the code, doc comments that repeated a test's own name, and the GetCursor interface comments that the existing MediaFileRepository.GetCursor does without. * refactor(persistence): fold the cursor wrappers into a generic wrapCursor wrapAlbumCursor, wrapArtistCursor, wrapMediaFileCursor and wrapFolderCursor were the same twelve lines four times over, differing only in the type name, and the playlist cursor had its own inline copy of the loop. Add one generic wrapCursor. It takes an extractor func rather than a method on an interface: a type parameter can't reach an embedded field, and methods that only satisfy a generic constraint are reported by the unused linter. The extractor also lets both type parameters be inferred, so call sites need no explicit instantiation. Each entity keeps its named wrapper as a one-liner, since the model cursor types are defined types and need the conversion — and the existing wrapper tests call them directly. The nil-row error now names the model type via %T ("unexpected nil model.Album") instead of a hand-written per-entity string; the three tests asserting that message are updated. * fix(jellyfin): bound concurrent collection streams A streamed collection holds a DB cursor — and its pooled connection — for the whole client-paced response, where the old materialize-then-write path released it as soon as the query finished. The pool is shared with the scanner, Subsonic, the native API and the UI, so enough slow clients take every connection and everything else blocks waiting for one. Measured against a copy of a 96k-track production DB, 20 concurrent unbounded streams against a pool of 16 stalled a write for 16.2s (the other 14 writes in the run took ~2ms — the signature of connection starvation, not lock contention). Cap concurrent streams at half the pool. Excess requests queue rather than fail, so no client is rejected: the same 20 streams now all complete and the worst write is 2.5ms. - conf.MaxOpenConns() now owns the pool sizing (db calls it). It belongs in conf: the pool is a tunable, db already imports conf, and putting it in db would force server/jellyfin to import db just to size the cap against it. Expressing the cap as MaxOpenConns()/2 also keeps the two from drifting apart. - Uses chi's ThrottleBacklog, not server.ThrottleBacklog: the latter buffers the whole response to release its token early, which is right for artwork but would undo the streaming. chi's panics on a non-positive limit, so throttleStreams guards it — setting MaxConcurrentStreams=0 disables the cap instead of crashing the server at startup. * refactor(jellyfin): move throttleStreams to middlewares.go It's a middleware, so it belongs beside normalizeQueryKeys, authenticate and withPlayer rather than in api.go. Its tests move to middlewares_test.go with it, keeping one test file per production file. * fix(jellyfin): abandon the scan when the client stops reading encodeItems discarded its write errors and relied on the final Flush to report them, so once bufio's buffer filled and the flush failed, the loop still pulled every remaining row through the cursor and serialized it for a client that was gone. A test with a failing writer confirms it: all 20k items were drained. That wastes CPU, and holds the cursor's pooled DB connection and a stream slot (now a capped resource) for the length of a full scan nobody is reading. Check the per-item writes so the first failure ends the scan; the fixed envelope writes stay unchecked, since bufio latches for them anyway. --- conf/configuration.go | 18 + db/db.go | 3 +- model/album.go | 1 + model/artist.go | 4 + model/playlist.go | 4 + persistence/album_repository.go | 21 +- persistence/album_repository_test.go | 18 +- persistence/artist_repository.go | 14 + persistence/artist_repository_test.go | 16 + persistence/folder_repository.go | 12 +- persistence/folder_repository_test.go | 2 +- persistence/mediafile_repository.go | 12 +- persistence/mediafile_repository_test.go | 18 +- persistence/persistence_suite_test.go | 13 + persistence/playlist_repository.go | 16 + persistence/playlist_repository_test.go | 9 + persistence/sql_base_repository.go | 18 + server/jellyfin/api.go | 52 ++- server/jellyfin/browsing.go | 9 +- server/jellyfin/e2e/browsing_test.go | 28 ++ server/jellyfin/items.go | 537 ++++++++++++++++------- server/jellyfin/items_test.go | 8 + server/jellyfin/middlewares.go | 16 + server/jellyfin/middlewares_test.go | 69 +++ server/jellyfin/response.go | 87 ++++ server/jellyfin/response_test.go | 116 +++++ tests/mock_album_repo.go | 14 + tests/mock_artist_repo.go | 14 + tests/mock_mediafile_repo.go | 14 + tests/mock_playlist_repo.go | 14 + 30 files changed, 957 insertions(+), 220 deletions(-) create mode 100644 server/jellyfin/response.go create mode 100644 server/jellyfin/response_test.go diff --git a/conf/configuration.go b/conf/configuration.go index 562f52465..83793bd43 100644 --- a/conf/configuration.go +++ b/conf/configuration.go @@ -225,6 +225,10 @@ type jellyfinOptions struct { // ExposedPublicUsers is a comma-separated list of usernames to advertise on the unauthenticated // GET /Users/Public, so Jellyfin clients can show a login user-picker. Empty exposes no users. ExposedPublicUsers string + // MaxConcurrentStreams bounds how many collection responses can stream at once. Each holds a DB + // cursor — and its pooled connection — for the whole client-paced response, so without a bound + // enough slow clients would take the entire pool and stall the scanner, scrobbles and the UI. + MaxConcurrentStreams int } type httpHeaderOptions struct { @@ -889,6 +893,9 @@ func setViperDefaults() { viper.SetDefault("devuishowconfig", true) viper.SetDefault("devneweventstream", true) viper.SetDefault("devoffsetoptimize", 50000) + // Half the pool: streams may take up to this many connections, leaving the rest for the scanner, + // scrobbles and the UI. See MaxOpenConns. + viper.SetDefault("jellyfin.maxconcurrentstreams", max(2, MaxOpenConns()/2)) viper.SetDefault("devartworkmaxrequests", max(2, runtime.NumCPU()/2)) viper.SetDefault("devartworkthrottlebackloglimit", consts.RequestThrottleBacklogLimit) viper.SetDefault("devartworkthrottlebacklogtimeout", consts.RequestThrottleBacklogTimeout) @@ -959,3 +966,14 @@ func getConfigFile(cfgFile string) string { } return "" } + +// MaxOpenConns is the size of the shared SQLite connection pool, used by every subsystem (scanner, +// Subsonic, Jellyfin, native API, UI). +// +// It bounds concurrent *readers*: SQLite serializes writers on a single database-wide write lock, so +// more connections buy no write parallelism. A connection is held while blocked on disk I/O or on a +// slow HTTP client, neither of which is CPU-bound — the CPU-bound knob is DevScannerThreads — so the +// count is only loosely related to core count, and the floor is what matters on small machines. +func MaxOpenConns() int { + return max(4, runtime.NumCPU()) +} diff --git a/db/db.go b/db/db.go index 3f3f61d71..4ca996fe5 100644 --- a/db/db.go +++ b/db/db.go @@ -5,7 +5,6 @@ import ( "database/sql" "embed" "fmt" - "runtime" "time" "github.com/mattn/go-sqlite3" @@ -44,7 +43,7 @@ func Db() *sql.DB { } log.Debug("Opening DataBase", "dbPath", Path, "driver", Driver) db, err := sql.Open(Driver, Path) - db.SetMaxOpenConns(max(4, runtime.NumCPU())) + db.SetMaxOpenConns(conf.MaxOpenConns()) if err != nil { log.Fatal("Error opening database", err) } diff --git a/model/album.go b/model/album.go index 667f4695b..ade7f6ee0 100644 --- a/model/album.go +++ b/model/album.go @@ -141,6 +141,7 @@ type AlbumRepository interface { UpdateExternalInfo(*Album) error Get(id string) (*Album, error) GetAll(...QueryOptions) (Albums, error) + GetCursor(...QueryOptions) (AlbumCursor, error) // The following methods are used exclusively by the scanner: Touch(ids ...string) error diff --git a/model/artist.go b/model/artist.go index 2085f0051..f9c4bffd5 100644 --- a/model/artist.go +++ b/model/artist.go @@ -1,6 +1,7 @@ package model import ( + "iter" "maps" "slices" "time" @@ -79,6 +80,8 @@ type ArtistIndex struct { } type ArtistIndexes []ArtistIndex +type ArtistCursor iter.Seq2[Artist, error] + type ArtistRepository interface { CountAll(options ...QueryOptions) (int64, error) Exists(id string) (bool, error) @@ -86,6 +89,7 @@ type ArtistRepository interface { UpdateExternalInfo(a *Artist) error Get(id string) (*Artist, error) GetAll(options ...QueryOptions) (Artists, error) + GetCursor(options ...QueryOptions) (ArtistCursor, error) GetIndex(includeMissing bool, libraryIds []int, roles ...Role) (ArtistIndexes, error) // The following methods are used exclusively by the scanner: diff --git a/model/playlist.go b/model/playlist.go index 262774aa7..f2586f52d 100644 --- a/model/playlist.go +++ b/model/playlist.go @@ -1,6 +1,7 @@ package model import ( + "iter" "slices" "strconv" "time" @@ -121,6 +122,8 @@ func (pls Playlist) UploadedImagePath() string { type Playlists []Playlist +type PlaylistCursor iter.Seq2[Playlist, error] + type PlaylistRepository interface { ResourceRepository AnnotatedRepository @@ -130,6 +133,7 @@ type PlaylistRepository interface { Get(id string) (*Playlist, error) GetWithTracks(id string, refreshSmartPlaylist, includeMissing bool) (*Playlist, error) GetAll(options ...QueryOptions) (Playlists, error) + GetCursor(options ...QueryOptions) (PlaylistCursor, error) FindByPath(path string) (*Playlist, error) Delete(id string) error Tracks(playlistId string, refreshSmartPlaylist bool) PlaylistTrackRepository diff --git a/persistence/album_repository.go b/persistence/album_repository.go index 34845be15..6ebbd9202 100644 --- a/persistence/album_repository.go +++ b/persistence/album_repository.go @@ -247,6 +247,15 @@ func (r *albumRepository) GetAll(options ...model.QueryOptions) (model.Albums, e return res.toModels(), nil } +func (r *albumRepository) GetCursor(options ...model.QueryOptions) (model.AlbumCursor, error) { + sq := r.selectAlbum(options...) + cursor, err := queryWithStableResults[dbAlbum](r.sqlRepository, sq) + if err != nil { + return nil, err + } + return wrapAlbumCursor(cursor), nil +} + func (r *albumRepository) CopyAttributes(fromID, toID string, columns ...string) error { var from dbx.NullStringMap err := r.queryOne(Select(columns...).From(r.tableName).Where(Eq{"id": fromID}), &from) @@ -319,17 +328,7 @@ func (r *albumRepository) GetTouchedAlbums(libID int) (model.AlbumCursor, error) } func wrapAlbumCursor(cursor iter.Seq2[dbAlbum, error]) model.AlbumCursor { - return func(yield func(model.Album, error) bool) { - for a, err := range cursor { - if a.Album == nil { - yield(model.Album{}, fmt.Errorf("unexpected nil album (%v): %w", a, err)) - return - } - if !yield(*a.Album, err) || err != nil { - return - } - } - } + return model.AlbumCursor(wrapCursor(cursor, func(a dbAlbum) *model.Album { return a.Album })) } // RefreshPlayCounts updates the play count and last play date annotations for all albums, based diff --git a/persistence/album_repository_test.go b/persistence/album_repository_test.go index f72f778db..64ff0095e 100644 --- a/persistence/album_repository_test.go +++ b/persistence/album_repository_test.go @@ -67,6 +67,22 @@ var _ = Describe("AlbumRepository", func() { }) }) + Describe("GetCursor", func() { + It("yields the same albums as GetAll", func() { + opts := model.QueryOptions{Sort: "name"} + want, err := albumRepo.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(albumRepo.GetCursor(opts))).To(Equal([]model.Album(want))) + }) + + It("honors Max/Offset like GetAll", func() { + opts := model.QueryOptions{Sort: "name", Max: 2, Offset: 1} + want, err := albumRepo.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(albumRepo.GetCursor(opts))).To(Equal([]model.Album(want))) + }) + }) + Describe("GetAll", func() { var GetAll = func(opts ...model.QueryOptions) (model.Albums, error) { albums, err := albumRepo.GetAll(opts...) @@ -854,7 +870,7 @@ var _ = Describe("AlbumRepository", func() { } }).ToNot(Panic()) Expect(gotErr).To(HaveOccurred()) - Expect(gotErr.Error()).To(ContainSubstring("unexpected nil album")) + Expect(gotErr.Error()).To(ContainSubstring("unexpected nil model.Album")) Expect(errors.Is(gotErr, dbErr)).To(BeTrue(), "should wrap the original cursor error") }) diff --git a/persistence/artist_repository.go b/persistence/artist_repository.go index f84f410e9..b542dedb4 100644 --- a/persistence/artist_repository.go +++ b/persistence/artist_repository.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "iter" "os" "slices" "strings" @@ -263,6 +264,19 @@ func (r *artistRepository) GetAll(options ...model.QueryOptions) (model.Artists, return res, err } +func (r *artistRepository) GetCursor(options ...model.QueryOptions) (model.ArtistCursor, error) { + sel := r.selectArtist(options...) + cursor, err := queryWithStableResults[dbArtist](r.sqlRepository, sel) + if err != nil { + return nil, err + } + return wrapArtistCursor(cursor), nil +} + +func wrapArtistCursor(cursor iter.Seq2[dbArtist, error]) model.ArtistCursor { + return model.ArtistCursor(wrapCursor(cursor, func(a dbArtist) *model.Artist { return a.Artist })) +} + func (r *artistRepository) getIndexKey(a model.Artist) string { source := a.OrderArtistName if conf.Server.PreferSortTags { diff --git a/persistence/artist_repository_test.go b/persistence/artist_repository_test.go index d7b695ade..dc11ede36 100644 --- a/persistence/artist_repository_test.go +++ b/persistence/artist_repository_test.go @@ -268,6 +268,22 @@ var _ = Describe("ArtistRepository", func() { repo = NewArtistRepository(ctx, GetDBXBuilder()) }) + Describe("GetCursor", func() { + It("yields the same artists as GetAll", func() { + opts := model.QueryOptions{Sort: "name"} + want, err := repo.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(repo.GetCursor(opts))).To(Equal([]model.Artist(want))) + }) + + It("honors Max/Offset like GetAll", func() { + opts := model.QueryOptions{Sort: "name", Max: 2, Offset: 1} + want, err := repo.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(repo.GetCursor(opts))).To(Equal([]model.Artist(want))) + }) + }) + Describe("Basic Operations", func() { Describe("Count", func() { It("returns the number of artists in the DB", func() { diff --git a/persistence/folder_repository.go b/persistence/folder_repository.go index 8fb7f0296..5da395a74 100644 --- a/persistence/folder_repository.go +++ b/persistence/folder_repository.go @@ -263,17 +263,7 @@ func (r folderRepository) GetAllWithPlaylists() (model.FolderCursor, error) { } func wrapFolderCursor(cursor iter.Seq2[dbFolder, error]) model.FolderCursor { - return func(yield func(model.Folder, error) bool) { - for f, err := range cursor { - if f.Folder == nil { - yield(model.Folder{}, fmt.Errorf("unexpected nil folder (%v): %w", f, err)) - return - } - if !yield(*f.Folder, err) || err != nil { - return - } - } - } + return model.FolderCursor(wrapCursor(cursor, func(f dbFolder) *model.Folder { return f.Folder })) } func (r folderRepository) purgeEmpty(libraryIDs ...int) error { diff --git a/persistence/folder_repository_test.go b/persistence/folder_repository_test.go index a8945dfee..8cd45f16b 100644 --- a/persistence/folder_repository_test.go +++ b/persistence/folder_repository_test.go @@ -297,7 +297,7 @@ var _ = Describe("FolderRepository", func() { } }).ToNot(Panic()) Expect(gotErr).To(HaveOccurred()) - Expect(gotErr.Error()).To(ContainSubstring("unexpected nil folder")) + Expect(gotErr.Error()).To(ContainSubstring("unexpected nil model.Folder")) Expect(errors.Is(gotErr, dbErr)).To(BeTrue(), "should wrap the original cursor error") }) diff --git a/persistence/mediafile_repository.go b/persistence/mediafile_repository.go index b4979ca77..ace61610c 100644 --- a/persistence/mediafile_repository.go +++ b/persistence/mediafile_repository.go @@ -420,17 +420,7 @@ func (r *mediaFileRepository) GetMissingAndMatching(libId int) (model.MediaFileC } func wrapMediaFileCursor(cursor iter.Seq2[dbMediaFile, error]) model.MediaFileCursor { - return func(yield func(model.MediaFile, error) bool) { - for m, err := range cursor { - if m.MediaFile == nil { - yield(model.MediaFile{}, fmt.Errorf("unexpected nil mediafile (%v): %w", m, err)) - return - } - if !yield(*m.MediaFile, err) || err != nil { - return - } - } - } + return model.MediaFileCursor(wrapCursor(cursor, func(m dbMediaFile) *model.MediaFile { return m.MediaFile })) } // FindRecentFilesByMBZTrackID finds recently added files by MusicBrainz Track ID in other libraries diff --git a/persistence/mediafile_repository_test.go b/persistence/mediafile_repository_test.go index 80d440c41..f6a744d8d 100644 --- a/persistence/mediafile_repository_test.go +++ b/persistence/mediafile_repository_test.go @@ -29,6 +29,22 @@ var _ = Describe("MediaRepository", func() { mr = NewMediaFileRepository(ctx, GetDBXBuilder()) }) + Describe("GetCursor", func() { + It("yields the same media files as GetAll", func() { + opts := model.QueryOptions{Sort: "title"} + want, err := mr.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(mr.GetCursor(opts))).To(Equal([]model.MediaFile(want))) + }) + + It("honors Max/Offset like GetAll", func() { + opts := model.QueryOptions{Sort: "title", Max: 2, Offset: 1} + want, err := mr.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(mr.GetCursor(opts))).To(Equal([]model.MediaFile(want))) + }) + }) + It("gets mediafile from the DB", func() { actual, err := mr.Get("1004") Expect(err).ToNot(HaveOccurred()) @@ -1012,7 +1028,7 @@ var _ = Describe("MediaRepository", func() { } }).ToNot(Panic()) Expect(gotErr).To(HaveOccurred()) - Expect(gotErr.Error()).To(ContainSubstring("unexpected nil mediafile")) + Expect(gotErr.Error()).To(ContainSubstring("unexpected nil model.MediaFile")) Expect(errors.Is(gotErr, dbErr)).To(BeTrue(), "should wrap the original cursor error") }) diff --git a/persistence/persistence_suite_test.go b/persistence/persistence_suite_test.go index 4f2fd7fe2..f146cb06b 100644 --- a/persistence/persistence_suite_test.go +++ b/persistence/persistence_suite_test.go @@ -329,3 +329,16 @@ var _ = BeforeSuite(func() { func GetDBXBuilder() *dbx.DB { return dbx.NewFromDB(db.Db(), db.Dialect) } + +// collectCursor takes the cursor's underlying func type so the named cursor types +// (model.AlbumCursor, ...) infer T. +func collectCursor[T any](cursor func(func(T, error) bool), err error) []T { + GinkgoHelper() + Expect(err).ToNot(HaveOccurred()) + var out []T + for item, err := range cursor { + Expect(err).ToNot(HaveOccurred()) + out = append(out, item) + } + return out +} diff --git a/persistence/playlist_repository.go b/persistence/playlist_repository.go index 573f43c10..9626aad6a 100644 --- a/persistence/playlist_repository.go +++ b/persistence/playlist_repository.go @@ -6,6 +6,7 @@ import ( "encoding/json" "errors" "fmt" + "iter" "slices" "time" @@ -186,6 +187,21 @@ func (r *playlistRepository) GetAll(options ...model.QueryOptions) (model.Playli return playlists, err } +func (r *playlistRepository) GetCursor(options ...model.QueryOptions) (model.PlaylistCursor, error) { + // Same userFilter as GetAll: a cursor must not widen visibility beyond public/owned playlists. + sel := r.selectPlaylist(options...).Where(r.userFilter()) + cursor, err := queryWithStableResults[dbPlaylist](r.sqlRepository, sel) + if err != nil { + return nil, err + } + return wrapPlaylistCursor(cursor), nil +} + +// dbPlaylist embeds a value, not a pointer, so its model is never nil. +func wrapPlaylistCursor(cursor iter.Seq2[dbPlaylist, error]) model.PlaylistCursor { + return model.PlaylistCursor(wrapCursor(cursor, func(p dbPlaylist) *model.Playlist { return &p.Playlist })) +} + func (r *playlistRepository) GetPlaylists(mediaFileId string) (model.Playlists, error) { sel := r.selectPlaylist(model.QueryOptions{Sort: "name"}). Join("playlist_tracks on playlist.id = playlist_tracks.playlist_id"). diff --git a/persistence/playlist_repository_test.go b/persistence/playlist_repository_test.go index 831e24453..c51ff6222 100644 --- a/persistence/playlist_repository_test.go +++ b/persistence/playlist_repository_test.go @@ -27,6 +27,15 @@ var _ = Describe("PlaylistRepository", func() { }) }) + Describe("GetCursor", func() { + It("yields the same playlists as GetAll", func() { + opts := model.QueryOptions{Sort: "name"} + want, err := repo.GetAll(opts) + Expect(err).ToNot(HaveOccurred()) + Expect(collectCursor(repo.GetCursor(opts))).To(Equal([]model.Playlist(want))) + }) + }) + Describe("Exists", func() { It("returns true for an existing playlist", func() { Expect(repo.Exists(plsCool.ID)).To(BeTrue()) diff --git a/persistence/sql_base_repository.go b/persistence/sql_base_repository.go index ce5221d19..d0cbb2946 100644 --- a/persistence/sql_base_repository.go +++ b/persistence/sql_base_repository.go @@ -347,6 +347,24 @@ func (r sqlRepository) queryOne(sq Sqlizer, response any) error { return err } +// wrapCursor adapts a cursor over db rows into one over their models. toModel pulls out the row's +// embedded model, which a type parameter can't reach on its own. +func wrapCursor[D, T any](cursor iter.Seq2[D, error], toModel func(D) *T) iter.Seq2[T, error] { + return func(yield func(T, error) bool) { + for row, err := range cursor { + m := toModel(row) + if m == nil { + var zero T + yield(zero, fmt.Errorf("unexpected nil %T (%v): %w", zero, row, err)) + return + } + if !yield(*m, err) || err != nil { + return + } + } + } +} + // queryWithStableResults is a helper function to execute a query and return an iterator that will yield its results // from a cursor, guaranteeing that the results will be stable, even if the underlying data changes. func queryWithStableResults[T any](r sqlRepository, sq SelectBuilder, options ...model.QueryOptions) (iter.Seq2[T, error], error) { diff --git a/server/jellyfin/api.go b/server/jellyfin/api.go index fd523d154..d169a4c83 100644 --- a/server/jellyfin/api.go +++ b/server/jellyfin/api.go @@ -1,7 +1,6 @@ package jellyfin import ( - "context" "encoding/json" "net/http" "sync" @@ -19,6 +18,7 @@ import ( "github.com/navidrome/navidrome/core/stream" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" + "github.com/navidrome/navidrome/server" "github.com/navidrome/navidrome/server/jellyfin/dto" ) @@ -71,8 +71,14 @@ func (api *Router) routes() http.Handler { inner.Get("/Users/Public", api.getPublicUsers) // Images are intentionally public: artwork isn't sensitive, matching Jellyfin's image handling. - inner.Get("/Items/{itemId}/Images/{type}", api.getItemImage) - inner.Get("/Items/{itemId}/Images/{type}/{index}", api.getItemImage) + // Bound concurrency like Subsonic's getCoverArt: image decode/resize is CPU- and memory-heavy, + // and an unbounded burst (a client fetching artwork across a large library) can exhaust memory. + inner.Group(func(r chi.Router) { + r.Use(server.ThrottleBacklog(conf.Server.DevArtworkMaxRequests, conf.Server.DevArtworkThrottleBacklogLimit, + conf.Server.DevArtworkThrottleBacklogTimeout)) + r.Get("/Items/{itemId}/Images/{type}", api.getItemImage) + r.Get("/Items/{itemId}/Images/{type}/{index}", api.getItemImage) + }) inner.Group(func(r chi.Router) { r.Use(api.authenticate) @@ -85,12 +91,22 @@ func (api *Router) routes() http.Handler { r.Get("/Users/Me", api.getCurrentUser) r.Get("/Users/{userId}", api.getCurrentUser) - r.Get("/Items", api.getItems) - r.Get("/Users/{userId}/Items", api.getItems) + // Cursor-backed collections: each streams straight from the DB, holding a connection for the + // whole client-paced response, so enough slow clients would take the entire pool and stall the + // scanner, scrobbles and the UI. Cap them at half the pool (see conf.MaxOpenConns); excess + // requests queue rather than fail. + r.Group(func(r chi.Router) { + r.Use(throttleStreams(conf.Server.Jellyfin.MaxConcurrentStreams)) + r.Get("/Items", api.getItems) + r.Get("/Users/{userId}/Items", api.getItems) + r.Get("/Users/{userId}/Items/Latest", api.getLatest) + r.Get("/Artists", api.getArtists) + r.Get("/Artists/AlbumArtists", api.getAlbumArtists) + }) + r.Get("/Items/{itemId}", api.getItem) r.Get("/Users/{userId}/Items/{itemId}", api.getItem) r.Delete("/Items/{itemId}", api.deleteItem) - r.Get("/Users/{userId}/Items/Latest", api.getLatest) // /UserFavoriteItems is the current @jellyfin/sdk spelling (Jellify); the // /Users/{userId}/FavoriteItems form is the legacy one Finamp still uses. @@ -106,8 +122,6 @@ func (api *Router) routes() http.Handler { r.Get("/UserItems/{itemId}/UserData", api.getUserItemData) r.Get("/Users/{userId}/Items/{itemId}/UserData", api.getUserItemData) - r.Get("/Artists", api.getArtists) - r.Get("/Artists/AlbumArtists", api.getAlbumArtists) r.Get("/Artists/{itemId}/Similar", api.getSimilarArtists) r.Get("/Items/{itemId}/Similar", api.getSimilarItems) r.Get("/Items/{itemId}/InstantMix", api.getInstantMix) @@ -158,14 +172,19 @@ func (api *Router) routes() http.Handler { return caseInsensitivePaths(inner) } -// ok writes payload as JSON, stamping ServerId on any item(s) in it — real Jellyfin always sets it, -// and it's the same value for every item, so it's applied here rather than threaded through mappers. +// ok writes payload as JSON — the single entry point for every handler. Collections are routed to +// the streaming writer, so callers needn't know whether theirs is cursor-backed. ServerId is stamped +// on any item(s): real Jellyfin always sets it, and it's constant per request. +// +// Only /Items/Latest bypasses this, for its bare-array shape (see writeItemsArray). func (api *Router) ok(w http.ResponseWriter, r *http.Request, payload any) { switch p := payload.(type) { + case itemsResult: + api.writeItems(w, r, p) + return case dto.QueryResult: - api.stampServerID(r.Context(), p.Items) - case []dto.BaseItemDto: - api.stampServerID(r.Context(), p) + api.writeItems(w, r, materialized(p)) + return case dto.BaseItemDto: p.ServerId = api.serverID(r.Context()) payload = p @@ -176,13 +195,6 @@ func (api *Router) ok(w http.ResponseWriter, r *http.Request, payload any) { } } -func (api *Router) stampServerID(ctx context.Context, items []dto.BaseItemDto) { - sid := api.serverID(ctx) - for i := range items { - items[i].ServerId = sid - } -} - // notFound handles unmatched routes and unsupported methods, logging them so unimplemented // endpoints surface instead of returning chi's default plain-text 404/405. func (api *Router) notFound(w http.ResponseWriter, r *http.Request) { diff --git a/server/jellyfin/browsing.go b/server/jellyfin/browsing.go index bf77c4594..d5a00e492 100644 --- a/server/jellyfin/browsing.go +++ b/server/jellyfin/browsing.go @@ -28,10 +28,15 @@ func (api *Router) listArtistsByRole(w http.ResponseWriter, r *http.Request, rol applySort(&opts, "MusicArtist", p.StringOr("sortby", ""), p.StringOr("sortorder", "")) scopeIDs, _ := resolveLibraryScope(ctx, dto.DecodeID(p.StringOr("parentid", ""))) + // Only the fields listArtists reads; /Artists has no favorites filter, so favOnly stays false. // Finamp's artist tab sends GenreIds when a genre filter is active. - genreIds := decodedQueryIDs(r, "genreids") + q := itemsQuery{ + scopeIDs: scopeIDs, + genreIds: decodedQueryIDs(r, "genreids"), + search: p.StringOr("searchterm", ""), + } - res, err := api.listArtists(ctx, opts, genreIds, scopeIDs, p.StringOr("searchterm", ""), false, role) + res, err := api.listArtists(ctx, opts, q, role) if err != nil { api.internalError(w, r, err) return diff --git a/server/jellyfin/e2e/browsing_test.go b/server/jellyfin/e2e/browsing_test.go index af639ed29..646de4497 100644 --- a/server/jellyfin/e2e/browsing_test.go +++ b/server/jellyfin/e2e/browsing_test.go @@ -300,6 +300,34 @@ var _ = Describe("Browsing", func() { q := queryResult(get("/Items?IncludeItemTypes=MusicAlbum,Audio&Recursive=true")) Expect(q.TotalRecordCount).To(Equal(12)) // 5 albums + 7 songs }) + + // Chaining the per-type cursors must preserve the merged order. + It("streams an unbounded multi-type merge, honoring StartIndex", func() { + all := queryResult(get("/Items?IncludeItemTypes=MusicAlbum,Audio&Recursive=true")) + Expect(all.Items).To(HaveLen(12)) + + skipped := queryResult(get("/Items?IncludeItemTypes=MusicAlbum,Audio&Recursive=true&StartIndex=2")) + Expect(skipped.Items).To(HaveLen(10)) + Expect(skipped.TotalRecordCount).To(Equal(12)) + Expect(skipped.StartIndex).To(Equal(2)) + Expect(names(skipped.Items)).To(Equal(names(all.Items)[2:])) + }) + + // Paging must ride on the cursor query's LIMIT/OFFSET, not be applied after materializing. + It("pages songs via StartIndex/Limit while reporting the full total", func() { + all := queryResult(get("/Items?IncludeItemTypes=Audio&Recursive=true&SortBy=SortName")) + Expect(all.TotalRecordCount).To(Equal(7)) + Expect(all.Items).To(HaveLen(7)) + + p1 := queryResult(get("/Items?IncludeItemTypes=Audio&Recursive=true&SortBy=SortName&Limit=3&StartIndex=0")) + p2 := queryResult(get("/Items?IncludeItemTypes=Audio&Recursive=true&SortBy=SortName&Limit=3&StartIndex=3")) + Expect(p1.Items).To(HaveLen(3)) + Expect(p2.Items).To(HaveLen(3)) + Expect(p1.TotalRecordCount).To(Equal(7)) + // The two pages are distinct and match the head of the unpaged, identically-sorted list. + Expect(names(p1.Items)).ToNot(ContainElement(BeElementOf(names(p2.Items)))) + Expect(append(names(p1.Items), names(p2.Items)...)).To(Equal(names(all.Items)[:6])) + }) }) Describe("GET /Items/{id}", func() { diff --git a/server/jellyfin/items.go b/server/jellyfin/items.go index f8158bca4..0dca48938 100644 --- a/server/jellyfin/items.go +++ b/server/jellyfin/items.go @@ -2,6 +2,8 @@ package jellyfin import ( "context" + "io" + "iter" "net/http" "slices" "strconv" @@ -31,112 +33,328 @@ func (api *Router) getItems(w http.ResponseWriter, r *http.Request) { api.ok(w, r, res) } -// queryItems is the /Items dispatcher: it parses entity types from IncludeItemTypes (defaulting to -// MusicAlbum), queries each via the matching listXxx, and merges multi-type results into one -// paginated list (as Finamp's favorites screen requests). -func (api *Router) queryItems(ctx context.Context, r *http.Request) (dto.QueryResult, error) { - p := req.Params(r) - // Query keys are read lowercase because normalizeQueryKeys folded them (Jellyfin binds - // case-insensitively). /Items?ids= is a batch-fetch-by-id that bypasses the type dispatch below. - fields := dto.ParseFields(p.StringOr("fields", "")) - if ids := decodedQueryIDs(r, "ids"); len(ids) > 0 { - return api.itemsByIDs(ctx, ids, fields), nil - } - parentId := dto.DecodeID(p.StringOr("parentid", "")) - search := p.StringOr("searchterm", "") - // Clients express "favorites only" two ways: Filters=IsFavorite and the standalone - // isFavorite=true param (Finamp's "Favourite tracks" widget uses the latter). - favOnly := strings.Contains(p.StringOr("filters", ""), "IsFavorite") || p.BoolOr("isfavorite", false) - sortBy := p.StringOr("sortby", "") - sortOrder := p.StringOr("sortorder", "") - offset := p.IntOr("startindex", 0) - limit := p.IntOr("limit", 0) - rawTypes := p.StringOr("includeitemtypes", "") - // A ManualPlaylistsFolder query asks for the synthetic "playlists library" container, not real items. - if strings.Contains(rawTypes, "ManualPlaylistsFolder") { - return result([]dto.BaseItemDto{playlistsFolder()}, 1, 0), nil - } - types := parseTypes(rawTypes) - // An artist's page filters by artist, not ParentId: Finamp sends ParentId= for scoping - // plus AlbumArtistIds/ArtistIds/contributingArtistIds for the artist. albumArtistIds/artistIds - // select the artist's own discography; contributingArtistIds alone means albums they merely appear - // on (Jellyfin's "Featured On"), which must exclude that discography. - albumArtistScope := firstNonEmpty(p.StringOr("albumartistids", ""), p.StringOr("artistids", "")) - contributingScope := p.StringOr("contributingartistids", "") - artistId := firstDecodedID(firstNonEmpty(albumArtistScope, contributingScope)) - contributingOnly := albumArtistScope == "" && contributingScope != "" - // Finamp's genre screen sends ParentId= for scoping plus GenreIds for the genre. - genreIds := decodedQueryIDs(r, "genreids") +// itemsResult is the outcome of a collection query: a materialized page, or a cursor opener so a +// full-library response never builds every DTO at once. Exactly one of items/openCursor is set. +// +// openCursor is deferred rather than opened here: it must run after the ServerId lookup, which +// writes to the DB on first use and would deadlock against an open reader, but before the first +// response byte, so a failed open is still a clean error rather than a truncated 200. +type itemsResult struct { + items []dto.BaseItemDto + openCursor func() (iter.Seq2[dto.BaseItemDto, error], error) + total int + start int +} - scopeIDs, isLibraryParent := resolveLibraryScope(ctx, parentId) - // A playlist parent always resolves to its tracks, whatever IncludeItemTypes says. Jellify opens - // a playlist with ParentId=&IncludeItemTypes=Audio; routing that through listSongs would - // treat the playlist id as an album id and return nothing. - if parentId != "" && !isLibraryParent && parentId != playlistsFolderID { - if pls, err := api.playlists.GetWithTracks(ctx, parentId); err == nil { - // GetWithTracks enforces visibility (public or owned by the current user). - items := slice.Map(pls.Tracks, func(t model.PlaylistTrack) dto.BaseItemDto { return trackToBaseItem(t, fields) }) - return result(paginate(items, offset, limit), len(items), offset), nil +func materialized(q dto.QueryResult) itemsResult { + return itemsResult{items: q.Items, total: q.TotalRecordCount, start: q.StartIndex} +} + +func streamed(open func() (iter.Seq2[dto.BaseItemDto, error], error), total, start int) itemsResult { + return itemsResult{openCursor: open, total: total, start: start} +} + +// chained streams several results back to back, skipping the first skip items — the unbounded +// multi-type merge, where paginate(items, offset, 0) is just the concatenation minus its head. +func chained(results []itemsResult, total, skip int) itemsResult { + open := func() (iter.Seq2[dto.BaseItemDto, error], error) { + if len(results) == 0 { + return sliceItems(nil), nil + } + // Only the first opens eagerly (so the usual failure is still a clean error); the rest open as + // the stream reaches them, so only one cursor pins a DB connection at a time. + first, err := results[0].seq() + if err != nil { + return nil, err + } + return func(yield func(dto.BaseItemDto, error) bool) { + n := 0 + emit := func(seq iter.Seq2[dto.BaseItemDto, error]) bool { + for it, err := range seq { + if err != nil { + yield(dto.BaseItemDto{}, err) + return false + } + if n < skip { + n++ + continue + } + if !yield(it, nil) { + return false + } + } + return true + } + if !emit(first) { + return + } + for _, res := range results[1:] { + seq, err := res.seq() + if err != nil { + yield(dto.BaseItemDto{}, err) + return + } + if !emit(seq) { + return + } + } + }, nil + } + return streamed(open, total, skip) +} + +// streamCursor builds a deferred opener that maps each row as it's yielded. It takes the cursor's +// underlying func type, so callers wrap repo.GetCursor for the named type to infer T. +func streamCursor[T any](openCursor func() (func(func(T, error) bool), error), toItem func(T) dto.BaseItemDto) func() (iter.Seq2[dto.BaseItemDto, error], error) { + return func() (iter.Seq2[dto.BaseItemDto, error], error) { + cursor, err := openCursor() + if err != nil { + return nil, err + } + return func(yield func(dto.BaseItemDto, error) bool) { + for row, err := range cursor { + if err != nil { + yield(dto.BaseItemDto{}, err) + return + } + if !yield(toItem(row), nil) { + return + } + } + }, nil + } +} + +// seq returns the items as one sequence, opening the cursor if there is one. +func (ir itemsResult) seq() (iter.Seq2[dto.BaseItemDto, error], error) { + if ir.openCursor != nil { + return ir.openCursor() + } + return sliceItems(ir.items), nil +} + +// collect drains the result into a slice, for the merge that combines types before paginating. +func (ir itemsResult) collect() ([]dto.BaseItemDto, error) { + if ir.openCursor == nil { + return ir.items, nil + } + seq, err := ir.openCursor() + if err != nil { + return nil, err + } + var out []dto.BaseItemDto + for it, err := range seq { + if err != nil { + return nil, err + } + out = append(out, it) + } + return out, nil +} + +func (api *Router) writeItems(w http.ResponseWriter, r *http.Request, res itemsResult) { + api.streamResult(w, r, res, func(w io.Writer, items iter.Seq2[dto.BaseItemDto, error]) error { + return streamItemsEnvelope(w, items, res.total, res.start) + }) +} + +// writeItemsArray writes the bare-array shape (/Items/Latest), which has no QueryResult envelope. +func (api *Router) writeItemsArray(w http.ResponseWriter, r *http.Request, res itemsResult) { + api.streamResult(w, r, res, streamItemsArray) +} + +// streamResult stamps every item's ServerId (constant per request, so it's set here rather than in +// each mapper). The cursor opens before the first byte, so a failed open is still a clean 500. +func (api *Router) streamResult(w http.ResponseWriter, r *http.Request, res itemsResult, + write func(io.Writer, iter.Seq2[dto.BaseItemDto, error]) error) { + sid := api.serverID(r.Context()) + seq, err := res.seq() + if err != nil { + api.internalError(w, r, err) + return + } + stamped := func(yield func(dto.BaseItemDto, error) bool) { + for it, err := range seq { + if err != nil { + yield(dto.BaseItemDto{}, err) + return + } + it.ServerId = sid + if !yield(it, nil) { + return + } } } + w.Header().Set("Content-Type", "application/json; charset=utf-8") + if err := write(w, stamped); err != nil { + log.Error(r.Context(), "Jellyfin API: error streaming response", err) + } +} + +// itemsQuery is a parsed /Items request, so the dispatch and every listXxx take one value instead +// of a long positional parameter list. +type itemsQuery struct { + fields dto.Fields + ids []string + rawTypes string + types []string + search string + sortBy string + sortOrder string + offset int + limit int + favOnly bool + // parentId scopes the query. entityParent is the same id only when it names an entity (an artist + // for MusicAlbum, an album for Audio) rather than a library. + parentId string + entityParent string + isLibraryParent bool + scopeIDs []int + // artistId selects that artist's own discography; contributingOnly means albums they merely + // appear on (Jellyfin's "Featured On"), which must exclude that discography. + artistId string + contributingOnly bool + genreIds []string +} + +// parseItemsQuery also resolves the entity types (inferring them from the parent when +// IncludeItemTypes is absent) and the library scope. Query keys are read lowercase because +// normalizeQueryKeys folded them (Jellyfin binds case-insensitively). +func (api *Router) parseItemsQuery(ctx context.Context, r *http.Request) itemsQuery { + p := req.Params(r) + q := itemsQuery{ + fields: dto.ParseFields(p.StringOr("fields", "")), + ids: decodedQueryIDs(r, "ids"), + rawTypes: p.StringOr("includeitemtypes", ""), + search: p.StringOr("searchterm", ""), + sortBy: p.StringOr("sortby", ""), + sortOrder: p.StringOr("sortorder", ""), + offset: p.IntOr("startindex", 0), + limit: p.IntOr("limit", 0), + // Clients express "favorites only" two ways: Filters=IsFavorite and the standalone + // isFavorite=true param (Finamp's "Favourite tracks" widget uses the latter). + favOnly: strings.Contains(p.StringOr("filters", ""), "IsFavorite") || p.BoolOr("isfavorite", false), + parentId: dto.DecodeID(p.StringOr("parentid", "")), + // Finamp's genre screen sends ParentId= for scoping plus GenreIds for the genre. + genreIds: decodedQueryIDs(r, "genreids"), + } + // An artist's page filters by artist, not ParentId: Finamp sends ParentId= for scoping + // plus AlbumArtistIds/ArtistIds/contributingArtistIds for the artist. + albumArtistScope := firstNonEmpty(p.StringOr("albumartistids", ""), p.StringOr("artistids", "")) + contributingScope := p.StringOr("contributingartistids", "") + q.artistId = firstDecodedID(firstNonEmpty(albumArtistScope, contributingScope)) + q.contributingOnly = albumArtistScope == "" && contributingScope != "" + + q.types = parseTypes(q.rawTypes) + q.scopeIDs, q.isLibraryParent = resolveLibraryScope(ctx, q.parentId) + // With no item type, Jellyfin infers the child type from the parent: album parent -> its tracks // (Jellify opens albums this way). An artist parent keeps parseTypes' MusicAlbum default (browse // its albums). - if rawTypes == "" && parentId != "" && !isLibraryParent { - if parentId == playlistsFolderID { + if q.rawTypes == "" && q.parentId != "" && !q.isLibraryParent { + if q.parentId == playlistsFolderID { // Browsing into the synthetic playlists folder lists the user's playlists. - types = []string{"Playlist"} - } else if _, err := api.ds.Album(ctx).Get(parentId); err == nil { - types = []string{"Audio"} + q.types = []string{"Playlist"} + } else if _, err := api.ds.Album(ctx).Get(q.parentId); err == nil { + q.types = []string{"Audio"} } } - entityParent := parentId - // ParentId-as-entity-id (artist for MusicAlbum, album for Audio) only makes sense for a single - // type; a multi-type query has no natural parent entity, so ParentId is only library scoping there. - if isLibraryParent || len(types) > 1 { - entityParent = "" + // ParentId-as-entity-id only makes sense for a single type; a multi-type query has no natural + // parent entity, so there ParentId is only library scoping. + q.entityParent = q.parentId + if q.isLibraryParent || len(q.types) > 1 { + q.entityParent = "" } - - if len(types) == 1 { - opts := model.QueryOptions{Offset: offset, Max: limit} - applySort(&opts, types[0], sortBy, sortOrder) - return api.queryItemsOfType(ctx, types[0], opts, entityParent, artistId, contributingOnly, genreIds, scopeIDs, search, favOnly, fields) - } - - var items []dto.BaseItemDto - total := 0 - for _, itemType := range types { - var opts model.QueryOptions - // Each per-type query needs at most offset+limit rows (the worst case where one type fills the - // whole [offset, offset+limit) window); without this cap each would fetch its whole table. - // Totals are unaffected — they come from CountAll. - if limit > 0 { - opts.Max = offset + limit - } - applySort(&opts, itemType, sortBy, sortOrder) - res, err := api.queryItemsOfType(ctx, itemType, opts, entityParent, artistId, contributingOnly, genreIds, scopeIDs, search, favOnly, fields) - if err != nil { - return dto.QueryResult{}, err - } - items = append(items, res.Items...) - total += res.TotalRecordCount - } - return result(paginate(items, offset, limit), total, offset), nil + return q } -func (api *Router) queryItemsOfType(ctx context.Context, itemType string, opts model.QueryOptions, entityParent, artistId string, contributingOnly bool, genreIds []string, scopeIDs []int, search string, favOnly bool, fields dto.Fields) (dto.QueryResult, error) { +// queryItems is the /Items dispatcher: it resolves the request to entity types and queries each via +// the matching listXxx, merging multi-type results into one paginated list (as Finamp's favorites +// screen requests). +func (api *Router) queryItems(ctx context.Context, r *http.Request) (itemsResult, error) { + q := api.parseItemsQuery(ctx, r) + switch { + // /Items?ids= is a batch-fetch-by-id that bypasses the type dispatch. + case len(q.ids) > 0: + return materialized(api.itemsByIDs(ctx, q.ids, q.fields)), nil + // A ManualPlaylistsFolder query asks for the synthetic "playlists library" container, not real items. + case strings.Contains(q.rawTypes, "ManualPlaylistsFolder"): + return materialized(result([]dto.BaseItemDto{playlistsFolder()}, 1, 0)), nil + } + if res, ok := api.playlistTracks(ctx, q); ok { + return res, nil + } + if len(q.types) == 1 { + opts := model.QueryOptions{Offset: q.offset, Max: q.limit} + applySort(&opts, q.types[0], q.sortBy, q.sortOrder) + return api.queryItemsOfType(ctx, q.types[0], opts, q) + } + return api.mergeTypes(ctx, q) +} + +// playlistTracks resolves a playlist parent to its tracks, whatever IncludeItemTypes says: Jellify +// opens a playlist with ParentId=&IncludeItemTypes=Audio, and routing that through +// listSongs would treat the playlist id as an album id and return nothing. +func (api *Router) playlistTracks(ctx context.Context, q itemsQuery) (itemsResult, bool) { + if q.parentId == "" || q.isLibraryParent || q.parentId == playlistsFolderID { + return itemsResult{}, false + } + pls, err := api.playlists.GetWithTracks(ctx, q.parentId) + if err != nil { + return itemsResult{}, false + } + // GetWithTracks enforces visibility (public or owned by the current user). + items := slice.Map(pls.Tracks, func(t model.PlaylistTrack) dto.BaseItemDto { return trackToBaseItem(t, q.fields) }) + return materialized(result(paginate(items, q.offset, q.limit), len(items), q.offset)), true +} + +func (api *Router) mergeTypes(ctx context.Context, q itemsQuery) (itemsResult, error) { + // Each per-type query needs at most offset+limit rows (the worst case where one type fills the + // whole [offset, offset+limit) window). Totals are unaffected — they come from CountAll. + var results []itemsResult + total := 0 + for _, itemType := range q.types { + var opts model.QueryOptions + if q.limit > 0 { + opts.Max = q.offset + q.limit + } + applySort(&opts, itemType, q.sortBy, q.sortOrder) + res, err := api.queryItemsOfType(ctx, itemType, opts, q) + if err != nil { + return itemsResult{}, err + } + results = append(results, res) + total += res.total + } + if q.limit == 0 { + // No cap above, so merging in memory would pull every row of every type. The merged page is + // just their rows in order minus the first offset — what chaining the cursors yields. + return chained(results, total, q.offset), nil + } + var items []dto.BaseItemDto + for _, res := range results { + typeItems, err := res.collect() + if err != nil { + return itemsResult{}, err + } + items = append(items, typeItems...) + } + return materialized(result(paginate(items, q.offset, q.limit), total, q.offset)), nil +} + +func (api *Router) queryItemsOfType(ctx context.Context, itemType string, opts model.QueryOptions, q itemsQuery) (itemsResult, error) { switch itemType { case "Audio": - return api.listSongs(ctx, opts, entityParent, artistId, genreIds, scopeIDs, search, favOnly, fields) + return api.listSongs(ctx, opts, q) case "MusicArtist": // The MusicArtist browse hierarchy (UserViews -> artists -> albums) means album artists. - return api.listArtists(ctx, opts, genreIds, scopeIDs, search, favOnly, model.RoleAlbumArtist) + return api.listArtists(ctx, opts, q, model.RoleAlbumArtist) case "MusicGenre": return api.listGenres(ctx, opts) case "Playlist": - return api.listPlaylists(ctx, opts, favOnly) + return api.listPlaylists(ctx, opts, q) default: // MusicAlbum - return api.listAlbums(ctx, opts, entityParent, artistId, contributingOnly, genreIds, scopeIDs, search, favOnly) + return api.listAlbums(ctx, opts, q) } } @@ -213,145 +431,144 @@ func searchPage[S ~[]E, E any](opts model.QueryOptions, search func(model.QueryO return rows, total, nil } -func (api *Router) listAlbums(ctx context.Context, opts model.QueryOptions, parentId, artistId string, contributingOnly bool, genreIds []string, scopeIDs []int, search string, fav bool) (dto.QueryResult, error) { +func (api *Router) listAlbums(ctx context.Context, opts model.QueryOptions, q itemsQuery) (itemsResult, error) { repo := api.ds.Album(ctx) filters := squirrel.And{} // For albums, ParentId (browse an artist) and AlbumArtistIds/ArtistIds both mean "this artist's // albums"; contributingArtistIds means "albums they only appear on" (Featured On). switch { - case contributingOnly && artistId != "": - filters = append(filters, filter.AlbumsByContributingArtistID(artistId).Filters) - case firstNonEmpty(artistId, parentId) != "": - filters = append(filters, filter.AlbumsByArtistID(firstNonEmpty(artistId, parentId)).Filters) + case q.contributingOnly && q.artistId != "": + filters = append(filters, filter.AlbumsByContributingArtistID(q.artistId).Filters) + case firstNonEmpty(q.artistId, q.entityParent) != "": + filters = append(filters, filter.AlbumsByArtistID(firstNonEmpty(q.artistId, q.entityParent)).Filters) default: filters = append(filters, notMissing) } - if len(genreIds) > 0 { - filters = append(filters, filter.ByGenreID(genreIds)) + if len(q.genreIds) > 0 { + filters = append(filters, filter.ByGenreID(q.genreIds)) } - if fav { + if q.favOnly { filters = append(filters, filter.ByStarred().Filters) } opts.Filters = filters - opts = filter.ApplyLibraryFilter(opts, scopeIDs) + opts = filter.ApplyLibraryFilter(opts, q.scopeIDs) - if search != "" { + if q.search != "" { albums, total, err := searchPage(opts, func(o model.QueryOptions) (model.Albums, error) { - return repo.Search(search, o) + return repo.Search(q.search, o) }) if err != nil { - return dto.QueryResult{}, err + return itemsResult{}, err } - return result(slice.Map(albums, dto.AlbumToBaseItem), total, opts.Offset), nil - } - albums, err := repo.GetAll(opts) - if err != nil { - return dto.QueryResult{}, err + return materialized(result(slice.Map(albums, dto.AlbumToBaseItem), total, opts.Offset)), nil } total, _ := repo.CountAll(model.QueryOptions{Filters: opts.Filters}) - return result(slice.Map(albums, dto.AlbumToBaseItem), int(total), opts.Offset), nil + open := streamCursor(func() (func(func(model.Album, error) bool), error) { + return repo.GetCursor(opts) + }, dto.AlbumToBaseItem) + return streamed(open, int(total), opts.Offset), nil } -func (api *Router) listSongs(ctx context.Context, opts model.QueryOptions, parentId, artistId string, genreIds []string, scopeIDs []int, search string, fav bool, fields dto.Fields) (dto.QueryResult, error) { - toItem := func(mf model.MediaFile) dto.BaseItemDto { return dto.SongToBaseItem(mf, fields) } +func (api *Router) listSongs(ctx context.Context, opts model.QueryOptions, q itemsQuery) (itemsResult, error) { + toItem := func(mf model.MediaFile) dto.BaseItemDto { return dto.SongToBaseItem(mf, q.fields) } repo := api.ds.MediaFile(ctx) filters := squirrel.And{} // For songs, ArtistIds/AlbumArtistIds selects an artist's tracks; ParentId selects an album's. switch { - case artistId != "": - filters = append(filters, filter.SongsByArtistID(artistId).Filters) - case parentId != "": - filters = append(filters, filter.SongsByAlbum(parentId).Filters) + case q.artistId != "": + filters = append(filters, filter.SongsByArtistID(q.artistId).Filters) + case q.entityParent != "": + filters = append(filters, filter.SongsByAlbum(q.entityParent).Filters) default: filters = append(filters, notMissing) } - if len(genreIds) > 0 { - filters = append(filters, filter.ByGenreID(genreIds)) + if len(q.genreIds) > 0 { + filters = append(filters, filter.ByGenreID(q.genreIds)) } - if fav { + if q.favOnly { filters = append(filters, filter.ByStarred().Filters) } opts.Filters = filters - opts = filter.ApplyLibraryFilter(opts, scopeIDs) + opts = filter.ApplyLibraryFilter(opts, q.scopeIDs) - if search != "" { + if q.search != "" { mfs, total, err := searchPage(opts, func(o model.QueryOptions) (model.MediaFiles, error) { - return repo.Search(search, o) + return repo.Search(q.search, o) }) if err != nil { - return dto.QueryResult{}, err + return itemsResult{}, err } - return result(slice.Map(mfs, toItem), total, opts.Offset), nil + return materialized(result(slice.Map(mfs, toItem), total, opts.Offset)), nil } // When browsing an album's tracks, default to disc+track order (like Subsonic's GetAlbum); an // explicit client SortBy still wins, since applySort already set opts.Sort. - if artistId == "" && parentId != "" && opts.Sort == "" { - opts.Sort = filter.SongsByAlbum(parentId).Sort - } - mfs, err := repo.GetAll(opts) - if err != nil { - return dto.QueryResult{}, err + if q.artistId == "" && q.entityParent != "" && opts.Sort == "" { + opts.Sort = filter.SongsByAlbum(q.entityParent).Sort } + // A full-library request (Finamp's sync, with MediaSources) is tens of thousands of fat rows. total, _ := repo.CountAll(model.QueryOptions{Filters: opts.Filters}) - return result(slice.Map(mfs, toItem), int(total), opts.Offset), nil + open := streamCursor(func() (func(func(model.MediaFile, error) bool), error) { + return repo.GetCursor(opts) + }, toItem) + return streamed(open, int(total), opts.Offset), nil } // listArtists lists artists in the given role: RoleAlbumArtist for the "album artists" views, // RoleArtist for performing artists (/Artists). Without the role filter both lists would be identical. // genreIds isn't applied to search — a name lookup, like role (see below). -func (api *Router) listArtists(ctx context.Context, opts model.QueryOptions, genreIds []string, scopeIDs []int, search string, fav bool, role model.Role) (dto.QueryResult, error) { +func (api *Router) listArtists(ctx context.Context, opts model.QueryOptions, q itemsQuery, role model.Role) (itemsResult, error) { repo := api.ds.Artist(ctx) // Artist Search does its own library scoping: it consumes a sole Eq{"library_id": ...} filter as a // search scope (artists have no library_id column). A compound or join-based filter // (ApplyArtistLibraryFilter) would leak into the FTS query and 500, so search and browse build // filters differently. Role isn't applied to search for the same reason — it's a name lookup. - if search != "" { - if len(scopeIDs) > 0 { - opts.Filters = squirrel.Eq{"library_id": scopeIDs} + if q.search != "" { + if len(q.scopeIDs) > 0 { + opts.Filters = squirrel.Eq{"library_id": q.scopeIDs} } artists, total, err := searchPage(opts, func(o model.QueryOptions) (model.Artists, error) { - return repo.Search(search, o) + return repo.Search(q.search, o) }) if err != nil { - return dto.QueryResult{}, err + return itemsResult{}, err } - return result(slice.Map(artists, dto.ArtistToBaseItem), total, opts.Offset), nil + return materialized(result(slice.Map(artists, dto.ArtistToBaseItem), total, opts.Offset)), nil } - if fav { + if q.favOnly { opts.Filters = filter.ArtistsByStarred().Filters } else { opts.Filters = notMissing } - if len(genreIds) > 0 { - opts.Filters = squirrel.And{opts.Filters, filter.ArtistsByGenreID(genreIds)} + if len(q.genreIds) > 0 { + opts.Filters = squirrel.And{opts.Filters, filter.ArtistsByGenreID(q.genreIds)} } opts = filter.ArtistsByRole(opts, role) - opts = filter.ApplyArtistLibraryFilter(opts, scopeIDs) - artists, err := repo.GetAll(opts) - if err != nil { - return dto.QueryResult{}, err - } + opts = filter.ApplyArtistLibraryFilter(opts, q.scopeIDs) total, _ := repo.CountAll(model.QueryOptions{Filters: opts.Filters}) - return result(slice.Map(artists, dto.ArtistToBaseItem), int(total), opts.Offset), nil + open := streamCursor(func() (func(func(model.Artist, error) bool), error) { + return repo.GetCursor(opts) + }, dto.ArtistToBaseItem) + return streamed(open, int(total), opts.Offset), nil } -// listGenres is intentionally unscoped: genres are global tags, not per-library entities. Paging is -// in-memory (GenreRepository has no CountAll, lists are small) so TotalRecordCount is the real total. -func (api *Router) listGenres(ctx context.Context, opts model.QueryOptions) (dto.QueryResult, error) { +// listGenres is intentionally unscoped: genres are global tags, not per-library entities. It's also +// the one listXxx that stays materialized: GenreRepository has no CountAll, so the total is the +// length of the full list and paging is in-memory — nothing for a cursor to page over. +func (api *Router) listGenres(ctx context.Context, opts model.QueryOptions) (itemsResult, error) { genres, err := api.ds.Genre(ctx).GetAll(model.QueryOptions{Sort: opts.Sort, Order: opts.Order}) if err != nil { - return dto.QueryResult{}, err + return itemsResult{}, err } items := slice.Map(genres, dto.GenreToBaseItem) - return result(paginate(items, opts.Offset, opts.Max), len(items), opts.Offset), nil + return materialized(result(paginate(items, opts.Offset, opts.Max), len(items), opts.Offset)), nil } // listPlaylists lists playlists visible to the current user. Visibility (public or owned) is // enforced by playlistRepository, not scopeIDs. -func (api *Router) listPlaylists(ctx context.Context, opts model.QueryOptions, favOnly bool) (dto.QueryResult, error) { - if favOnly { +func (api *Router) listPlaylists(ctx context.Context, opts model.QueryOptions, q itemsQuery) (itemsResult, error) { + if q.favOnly { starred := squirrel.Eq{"starred": true} if opts.Filters == nil { opts.Filters = starred @@ -360,15 +577,14 @@ func (api *Router) listPlaylists(ctx context.Context, opts model.QueryOptions, f } } repo := api.ds.Playlist(ctx) - playlists, err := repo.GetAll(opts) - if err != nil { - return dto.QueryResult{}, err - } total, err := repo.CountAll(model.QueryOptions{Filters: opts.Filters}) if err != nil { - return dto.QueryResult{}, err + return itemsResult{}, err } - return result(slice.Map(playlists, dto.PlaylistToBaseItem), int(total), opts.Offset), nil + open := streamCursor(func() (func(func(model.Playlist, error) bool), error) { + return repo.GetCursor(opts) + }, dto.PlaylistToBaseItem) + return streamed(open, int(total), opts.Offset), nil } // resolveItemByID resolves a decoded navidrome id to its BaseItemDto, trying library view, album, @@ -482,17 +698,18 @@ func (api *Router) deleteItem(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusNoContent) } +// getLatest returns a bare array, not a QueryResult envelope — real Jellyfin's shape for +// /Items/Latest, and why it writes directly instead of going through api.ok. func (api *Router) getLatest(w http.ResponseWriter, r *http.Request) { ctx := r.Context() opts := filter.AlbumsByNewest() opts.Max = req.Params(r).IntOr("limit", 20) opts = filter.ApplyLibraryFilter(opts, accessibleLibraryIDs(ctx)) - albums, err := api.ds.Album(ctx).GetAll(opts) - if err != nil { - api.internalError(w, r, err) - return - } - api.ok(w, r, slice.Map(albums, dto.AlbumToBaseItem)) // /Latest returns a bare array + repo := api.ds.Album(ctx) + open := streamCursor(func() (func(func(model.Album, error) bool), error) { + return repo.GetCursor(opts) + }, dto.AlbumToBaseItem) + api.writeItemsArray(w, r, streamed(open, 0, 0)) } func result(items []dto.BaseItemDto, total, start int) dto.QueryResult { diff --git a/server/jellyfin/items_test.go b/server/jellyfin/items_test.go index 8bd1bf052..801db9e3a 100644 --- a/server/jellyfin/items_test.go +++ b/server/jellyfin/items_test.go @@ -71,6 +71,14 @@ var _ = Describe("Items", func() { Expect(res.Items[0].Id).To(Equal(dto.EncodeID("s1"))) }) + It("returns 500 when the song cursor fails to open, instead of a truncated 200", func() { + ds.MediaFile(context.Background()).(*tests.MockMediaFileRepo).SetError(true) + w := httptest.NewRecorder() + r := httptest.NewRequest("GET", "/Items?IncludeItemTypes=Audio&Recursive=true", nil).WithContext(ctxUser()) + invoke(api.getItems, w, r) + Expect(w.Code).To(Equal(http.StatusInternalServerError)) + }) + It("lists an artist's albums when ParentId is an artist and type is MusicAlbum", func() { ds.Album(context.Background()).(*tests.MockAlbumRepo).SetData(model.Albums{{ID: "a1", Name: "One", AlbumArtistID: "ar1"}}) w := httptest.NewRecorder() diff --git a/server/jellyfin/middlewares.go b/server/jellyfin/middlewares.go index 2941d3c32..d9e41d252 100644 --- a/server/jellyfin/middlewares.go +++ b/server/jellyfin/middlewares.go @@ -7,12 +7,28 @@ import ( "regexp" "strings" + "github.com/go-chi/chi/v5/middleware" + "github.com/navidrome/navidrome/consts" "github.com/navidrome/navidrome/core/auth" "github.com/navidrome/navidrome/log" "github.com/navidrome/navidrome/model" "github.com/navidrome/navidrome/model/request" ) +// throttleStreams bounds how many collection responses stream concurrently, so they can't take every +// connection in the shared DB pool: each holds a cursor, and its connection, for the whole +// client-paced response. Excess requests queue rather than fail. limit <= 0 disables it. +// +// Deliberately chi's ThrottleBacklog and not server.ThrottleBacklog: the latter buffers the entire +// response to release its token early, which is right for artwork but would undo the streaming here. +// chi's panics on a non-positive limit, hence the guard. +func throttleStreams(limit int) func(http.Handler) http.Handler { + if limit <= 0 { + return func(next http.Handler) http.Handler { return next } + } + return middleware.ThrottleBacklog(limit, consts.RequestThrottleBacklogLimit, consts.RequestThrottleBacklogTimeout) +} + // normalizeQueryKeys folds query-parameter keys to lowercase so handlers can read params // case-insensitively, matching real Jellyfin. Clients disagree on casing (Finamp sends PascalCase, // Jellify and the Jellyfin TypeScript SDK camelCase), so a case-sensitive read would drop one diff --git a/server/jellyfin/middlewares_test.go b/server/jellyfin/middlewares_test.go index 2c0761834..c766bff70 100644 --- a/server/jellyfin/middlewares_test.go +++ b/server/jellyfin/middlewares_test.go @@ -4,6 +4,9 @@ import ( "context" "net/http" "net/http/httptest" + "sync" + "sync/atomic" + "time" "github.com/navidrome/navidrome/core/auth" "github.com/navidrome/navidrome/model" @@ -252,3 +255,69 @@ var _ = Describe("normalizeQueryKeys", func() { Expect(got).To(ConsistOf("aaa", "bbb")) }) }) + +var _ = Describe("throttleStreams", func() { + // serve fires n concurrent requests through the middleware and reports the highest number that + // were ever inside the handler at once. + serve := func(limit, n int) int32 { + var inFlight, peak int32 + release := make(chan struct{}) + h := throttleStreams(limit)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cur := atomic.AddInt32(&inFlight, 1) + for { + old := atomic.LoadInt32(&peak) + if cur <= old || atomic.CompareAndSwapInt32(&peak, old, cur) { + break + } + } + <-release // hold the slot until every request has had a chance to enter + atomic.AddInt32(&inFlight, -1) + })) + + var wg sync.WaitGroup + for range n { + wg.Add(1) + go func() { + defer wg.Done() + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/Items", nil)) + }() + } + // Give the admitted requests time to pile up before letting them finish. + time.Sleep(100 * time.Millisecond) + close(release) + wg.Wait() + return atomic.LoadInt32(&peak) + } + + It("admits no more than the limit at once", func() { + Expect(serve(2, 8)).To(Equal(int32(2))) + }) + + It("queues the excess rather than rejecting it", func() { + // All 8 still complete — they wait for a slot instead of getting a 429. + var served int32 + release := make(chan struct{}) + h := throttleStreams(2)(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-release + atomic.AddInt32(&served, 1) + })) + var wg sync.WaitGroup + for range 8 { + wg.Add(1) + go func() { + defer wg.Done() + h.ServeHTTP(httptest.NewRecorder(), httptest.NewRequest("GET", "/Items", nil)) + }() + } + close(release) + wg.Wait() + Expect(served).To(Equal(int32(8))) + }) + + // chi's ThrottleBacklog panics on a non-positive limit, so a user disabling the cap must not + // crash the server at startup. + It("is disabled, not panicking, when the limit is zero", func() { + Expect(func() { serve(0, 4) }).ToNot(Panic()) + Expect(serve(0, 4)).To(BeNumerically(">", int32(1))) + }) +}) diff --git a/server/jellyfin/response.go b/server/jellyfin/response.go new file mode 100644 index 000000000..f4b96eda3 --- /dev/null +++ b/server/jellyfin/response.go @@ -0,0 +1,87 @@ +package jellyfin + +import ( + "bufio" + "bytes" + "encoding/json" + "io" + "iter" + "strconv" + + "github.com/navidrome/navidrome/server/jellyfin/dto" +) + +// streamItemsEnvelope writes a QueryResult, byte-identical to json.NewEncoder(w).Encode(q). +// +// A mid-stream error aborts without closing the envelope: the 200 is already committed, so a +// truncated-but-valid body would let a sync client treat the short list as the whole library and +// prune local tracks. Malformed JSON forces its parser to fail instead. Callers open the cursor +// before the first byte, so this only fires on a rare mid-iteration failure. +func streamItemsEnvelope(w io.Writer, items iter.Seq2[dto.BaseItemDto, error], total, start int) error { + bw := bufio.NewWriterSize(w, 64*1024) + _, _ = bw.WriteString(`{"Items":[`) + if err := encodeItems(bw, items); err != nil { + _ = bw.Flush() + return err + } + _, _ = bw.WriteString(`],"TotalRecordCount":`) + _, _ = bw.WriteString(strconv.Itoa(total)) + _, _ = bw.WriteString(`,"StartIndex":`) + _, _ = bw.WriteString(strconv.Itoa(start)) + _, _ = bw.WriteString("}\n") + return bw.Flush() +} + +// streamItemsArray writes a bare JSON array — the shape /Items/Latest returns, with no envelope. +func streamItemsArray(w io.Writer, items iter.Seq2[dto.BaseItemDto, error]) error { + bw := bufio.NewWriterSize(w, 64*1024) + _, _ = bw.WriteString("[") + if err := encodeItems(bw, items); err != nil { + _ = bw.Flush() + return err + } + _, _ = bw.WriteString("]\n") + return bw.Flush() +} + +// encodeItems writes items comma-separated. Unlike the fixed envelope writes, these are checked: +// bufio surfaces a latched write error here once a flush fails, and a client that has gone away must +// abandon the scan rather than pull the rest of the library through the cursor — which would hold its +// pooled DB connection and stream slot for a response nobody is reading. +func encodeItems(bw *bufio.Writer, items iter.Seq2[dto.BaseItemDto, error]) error { + // One reused buffer+encoder, so per-item JSON doesn't allocate. Encode HTML-escapes like + // json.Marshal, and appends a newline that's dropped below. + var itemBuf bytes.Buffer + enc := json.NewEncoder(&itemBuf) + first := true + for item, err := range items { + if err != nil { + return err + } + if !first { + if _, err := bw.WriteString(","); err != nil { + return err + } + } + first = false + itemBuf.Reset() + if err := enc.Encode(item); err != nil { + return err + } + b := itemBuf.Bytes() + if _, err := bw.Write(b[:len(b)-1]); err != nil { + return err + } + } + return nil +} + +func sliceItems(items []dto.BaseItemDto) iter.Seq2[dto.BaseItemDto, error] { + return func(yield func(dto.BaseItemDto, error) bool) { + for i := range items { + if !yield(items[i], nil) { + return + } + } + } +} diff --git a/server/jellyfin/response_test.go b/server/jellyfin/response_test.go new file mode 100644 index 000000000..c32c566fc --- /dev/null +++ b/server/jellyfin/response_test.go @@ -0,0 +1,116 @@ +package jellyfin + +import ( + "bytes" + "encoding/json" + "errors" + "iter" + "strings" + + "github.com/navidrome/navidrome/server/jellyfin/dto" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +// deadWriter stands in for a client that went away mid-response. +type deadWriter struct{} + +func (deadWriter) Write([]byte) (int, error) { return 0, errors.New("connection reset by peer") } + +var _ = Describe("streaming a materialized QueryResult", func() { + // The materialized path (api.ok -> writeItems -> sliceItems) must stay byte-for-byte identical to + // what json.Encoder.Encode produced before, so no client sees a different response. + assertIdenticalToEncoder := func(q dto.QueryResult) { + var got bytes.Buffer + Expect(streamItemsEnvelope(&got, sliceItems(q.Items), q.TotalRecordCount, q.StartIndex)).To(Succeed()) + + var want bytes.Buffer + Expect(json.NewEncoder(&want).Encode(q)).To(Succeed()) + + Expect(got.String()).To(Equal(want.String())) + } + + It("encodes an empty item list", func() { + assertIdenticalToEncoder(dto.QueryResult{Items: []dto.BaseItemDto{}}) + }) + + It("encodes a single item", func() { + assertIdenticalToEncoder(dto.QueryResult{ + Items: []dto.BaseItemDto{{Id: "a", Name: "One"}}, + TotalRecordCount: 1, + }) + }) + + It("encodes multiple items, honoring HTML escaping and StartIndex", func() { + assertIdenticalToEncoder(dto.QueryResult{ + Items: []dto.BaseItemDto{ + {Id: "a", Name: "One"}, + {Id: "b", Name: "Two & "}, + }, + TotalRecordCount: 500, + StartIndex: 100, + }) + }) +}) + +var _ = Describe("streamItemsEnvelope", func() { + seqOf := func(items ...dto.BaseItemDto) iter.Seq2[dto.BaseItemDto, error] { + return func(yield func(dto.BaseItemDto, error) bool) { + for _, it := range items { + if !yield(it, nil) { + return + } + } + } + } + + It("produces the same bytes as encoding an equivalent QueryResult", func() { + items := []dto.BaseItemDto{{Id: "a", Name: "One"}, {Id: "b", Name: "Two & "}} + var got bytes.Buffer + Expect(streamItemsEnvelope(&got, seqOf(items...), 500, 100)).To(Succeed()) + + var want bytes.Buffer + Expect(json.NewEncoder(&want).Encode(dto.QueryResult{Items: items, TotalRecordCount: 500, StartIndex: 100})).To(Succeed()) + Expect(got.String()).To(Equal(want.String())) + }) + + It("emits an empty array (not null) for a sequence that yields nothing", func() { + var got bytes.Buffer + Expect(streamItemsEnvelope(&got, seqOf(), 0, 0)).To(Succeed()) + Expect(got.String()).To(Equal("{\"Items\":[],\"TotalRecordCount\":0,\"StartIndex\":0}\n")) + }) + + // A client that goes away must not keep the source (a DB cursor, holding its pooled connection + // and a stream slot) running to the end of the library. + It("stops pulling from the source once writing fails", func() { + const total = 20000 + pulled := 0 + seq := func(yield func(dto.BaseItemDto, error) bool) { + for range total { + pulled++ + if !yield(dto.BaseItemDto{Id: "a", Name: strings.Repeat("x", 200)}, nil) { + return + } + } + } + err := streamItemsEnvelope(deadWriter{}, seq, total, 0) + Expect(err).To(HaveOccurred()) + Expect(pulled).To(BeNumerically("<", total), "should abandon the scan, not drain it") + }) + + It("aborts on a mid-stream error, leaving the envelope open (malformed) so the client fails loudly", func() { + boom := errors.New("scan failed") + first := dto.BaseItemDto{Id: "a", Name: "One"} + seq := func(yield func(dto.BaseItemDto, error) bool) { + if !yield(first, nil) { + return + } + yield(dto.BaseItemDto{}, boom) + } + var got bytes.Buffer + err := streamItemsEnvelope(&got, seq, 7, 0) + Expect(err).To(MatchError(boom)) + firstJSON, _ := json.Marshal(first) + Expect(got.String()).To(Equal("{\"Items\":[" + string(firstJSON))) + }) +}) diff --git a/tests/mock_album_repo.go b/tests/mock_album_repo.go index 85765abf8..6635881b7 100644 --- a/tests/mock_album_repo.go +++ b/tests/mock_album_repo.go @@ -75,6 +75,20 @@ func (m *MockAlbumRepo) GetAll(qo ...model.QueryOptions) (model.Albums, error) { return m.All, nil } +func (m *MockAlbumRepo) GetCursor(qo ...model.QueryOptions) (model.AlbumCursor, error) { + res, err := m.GetAll(qo...) + if err != nil { + return nil, err + } + return func(yield func(model.Album, error) bool) { + for _, a := range res { + if !yield(a, nil) { + return + } + } + }, nil +} + func (m *MockAlbumRepo) IncPlayCount(id string, timestamp time.Time) error { if m.Err { return errors.New("unexpected error") diff --git a/tests/mock_artist_repo.go b/tests/mock_artist_repo.go index 748002882..e6ea7aea4 100644 --- a/tests/mock_artist_repo.go +++ b/tests/mock_artist_repo.go @@ -113,6 +113,20 @@ func (m *MockArtistRepo) GetAll(options ...model.QueryOptions) (model.Artists, e return allArtists, nil } +func (m *MockArtistRepo) GetCursor(options ...model.QueryOptions) (model.ArtistCursor, error) { + res, err := m.GetAll(options...) + if err != nil { + return nil, err + } + return func(yield func(model.Artist, error) bool) { + for _, a := range res { + if !yield(a, nil) { + return + } + } + }, nil +} + func (m *MockArtistRepo) UpdateExternalInfo(artist *model.Artist) error { if m.Err { return errors.New("mock repo error") diff --git a/tests/mock_mediafile_repo.go b/tests/mock_mediafile_repo.go index 6ddd77f14..990b91d7c 100644 --- a/tests/mock_mediafile_repo.go +++ b/tests/mock_mediafile_repo.go @@ -109,6 +109,20 @@ func (m *MockMediaFileRepo) GetRandom(qo ...model.QueryOptions) (model.MediaFile return res, nil } +func (m *MockMediaFileRepo) GetCursor(qo ...model.QueryOptions) (model.MediaFileCursor, error) { + res, err := m.GetAll(qo...) + if err != nil { + return nil, err + } + return func(yield func(model.MediaFile, error) bool) { + for _, mf := range res { + if !yield(mf, nil) { + return + } + } + }, nil +} + func (m *MockMediaFileRepo) Put(mf *model.MediaFile) error { if m.Err { return errors.New("error") diff --git a/tests/mock_playlist_repo.go b/tests/mock_playlist_repo.go index 908d6aab5..8f8842c8e 100644 --- a/tests/mock_playlist_repo.go +++ b/tests/mock_playlist_repo.go @@ -52,6 +52,20 @@ func (m *MockPlaylistRepo) GetAll(options ...model.QueryOptions) (model.Playlist return m.All, nil } +func (m *MockPlaylistRepo) GetCursor(options ...model.QueryOptions) (model.PlaylistCursor, error) { + res, err := m.GetAll(options...) + if err != nil { + return nil, err + } + return func(yield func(model.Playlist, error) bool) { + for _, p := range res { + if !yield(p, nil) { + return + } + } + }, nil +} + func (m *MockPlaylistRepo) Get(id string) (*model.Playlist, error) { if m.Err { return nil, errors.New("error")