mirror of
https://github.com/navidrome/navidrome.git
synced 2026-08-31 07:30:32 +00:00
Merge branch 'master' into jsdoc
This commit is contained in:
commit
9b92d2a542
@ -8,7 +8,6 @@ import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
@ -187,7 +186,7 @@ func (s *playlists) updatePlaylist(ctx context.Context, newPls *model.Playlist,
|
||||
newPls.OwnerID = pls.OwnerID
|
||||
newPls.Public = pls.Public
|
||||
newPls.UploadedImage = pls.UploadedImage // Preserve manual upload
|
||||
newPls.EvaluatedAt = &time.Time{}
|
||||
newPls.EvaluatedAt = nil // force re-evaluation on next read
|
||||
} else {
|
||||
log.Info(ctx, "Adding synced playlist", "playlist", newPls.Name, "path", newPls.Path, "owner", owner.UserName)
|
||||
newPls.OwnerID = owner.ID
|
||||
|
||||
@ -135,6 +135,7 @@ func (s *playlists) applyContentUpdate(ctx context.Context, current, entity *mod
|
||||
}
|
||||
if rulesChanged {
|
||||
current.Rules = entity.Rules
|
||||
current.EvaluatedAt = nil // force re-evaluation on next read
|
||||
}
|
||||
if sent("sync") && current.Path != "" && current.Sync != entity.Sync {
|
||||
current.Sync = entity.Sync
|
||||
|
||||
@ -314,6 +314,38 @@ var _ = Describe("REST Adapter", func() {
|
||||
Expect(mockPlsRepo.Last.Public).To(BeTrue())
|
||||
})
|
||||
|
||||
It("resets EvaluatedAt when rules change", func() {
|
||||
evaluatedAt := time.Now().Add(-1 * time.Hour)
|
||||
mockPlsRepo.Data["smart-reset"] = &model.Playlist{
|
||||
ID: "smart-reset",
|
||||
Name: "Smart",
|
||||
OwnerID: "user-1",
|
||||
Rules: &criteria.Criteria{Expression: criteria.Is{"genre": "Rock"}},
|
||||
EvaluatedAt: &evaluatedAt,
|
||||
}
|
||||
repo = ps.NewRepository(ctx).(rest.Persistable)
|
||||
newRules := &criteria.Criteria{Expression: criteria.Is{"genre": "Jazz"}}
|
||||
err := repo.Update("smart-reset", &model.Playlist{Rules: newRules}, "rules")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(mockPlsRepo.Last.EvaluatedAt).To(BeNil())
|
||||
})
|
||||
|
||||
It("keeps EvaluatedAt when rules are not changed", func() {
|
||||
evaluatedAt := time.Now().Add(-1 * time.Hour)
|
||||
mockPlsRepo.Data["smart-keep"] = &model.Playlist{
|
||||
ID: "smart-keep",
|
||||
Name: "Smart",
|
||||
OwnerID: "user-1",
|
||||
Rules: &criteria.Criteria{Expression: criteria.Is{"genre": "Rock"}},
|
||||
EvaluatedAt: &evaluatedAt,
|
||||
}
|
||||
repo = ps.NewRepository(ctx).(rest.Persistable)
|
||||
err := repo.Update("smart-keep", &model.Playlist{Name: "Renamed Smart"}, "name")
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(mockPlsRepo.Last.EvaluatedAt).ToNot(BeNil())
|
||||
Expect(*mockPlsRepo.Last.EvaluatedAt).To(BeTemporally("~", evaluatedAt, time.Second))
|
||||
})
|
||||
|
||||
It("updates name and rules together (smart-playlist Edit form)", func() {
|
||||
mockPlsRepo.Data["smart-edit"] = &model.Playlist{
|
||||
ID: "smart-edit",
|
||||
|
||||
@ -89,6 +89,7 @@ type playTracker struct {
|
||||
ds model.DataStore
|
||||
broker events.Broker
|
||||
playMap cache.SimpleCache[string, PlaybackSession]
|
||||
sessionsMu sync.Mutex // serializes playMap check-then-write across concurrent reports
|
||||
builtinScrobblers map[string]Scrobbler
|
||||
pluginScrobblers map[string]Scrobbler
|
||||
pluginLoader PluginLoader
|
||||
@ -249,6 +250,12 @@ func (p *playTracker) getActiveScrobblers() map[string]Scrobbler {
|
||||
return combined
|
||||
}
|
||||
|
||||
// hasPlayingSession reports whether clientId's current session is already playing mediaId.
|
||||
func (p *playTracker) hasPlayingSession(clientId, mediaId string) bool {
|
||||
cur, err := p.playMap.Get(clientId)
|
||||
return err == nil && cur.MediaFile.ID == mediaId && cur.State == StatePlaying
|
||||
}
|
||||
|
||||
func remainingTTL(durationSec float32, positionMs int64, rate float64) time.Duration {
|
||||
if rate <= 0 {
|
||||
rate = 1.0
|
||||
@ -268,6 +275,12 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
|
||||
switch params.State {
|
||||
case StateStarting:
|
||||
// Clients may send starting/playing unordered; a late "starting" must not downgrade
|
||||
// a playing session, or position estimation freezes until the next report.
|
||||
if p.hasPlayingSession(clientId, params.MediaId) {
|
||||
log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
|
||||
return nil
|
||||
}
|
||||
mf, err := p.ds.MediaFile(ctx).GetWithParticipants(params.MediaId)
|
||||
if err != nil {
|
||||
return err
|
||||
@ -284,7 +297,15 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
PlaybackRate: params.PlaybackRate,
|
||||
LastReport: now,
|
||||
}
|
||||
p.sessionsMu.Lock()
|
||||
// re-check: a concurrent "playing" report may have created the session during the load above
|
||||
if p.hasPlayingSession(clientId, params.MediaId) {
|
||||
p.sessionsMu.Unlock()
|
||||
log.Trace(ctx, "Ignoring out-of-order starting report for playing session", "clientId", clientId, "mediaId", params.MediaId)
|
||||
return nil
|
||||
}
|
||||
err = p.playMap.AddWithTTL(clientId, info, remainingTTL(mf.Duration, params.PositionMs, params.PlaybackRate))
|
||||
p.sessionsMu.Unlock()
|
||||
if err != nil {
|
||||
log.Warn(ctx, "Error adding PlaybackSession to cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
|
||||
}
|
||||
@ -315,7 +336,9 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
ttl = remainingTTL(info.MediaFile.Duration, params.PositionMs, params.PlaybackRate)
|
||||
}
|
||||
log.Trace(ctx, "Updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, "positionMs", params.PositionMs, "playbackRate", params.PlaybackRate, "ttl", ttl)
|
||||
p.sessionsMu.Lock()
|
||||
err := p.playMap.AddWithTTL(clientId, info, ttl)
|
||||
p.sessionsMu.Unlock()
|
||||
if err != nil {
|
||||
log.Warn(ctx, "Error updating PlaybackSession in cache", "clientId", clientId, "mediaId", params.MediaId, "state", params.State, err)
|
||||
}
|
||||
@ -339,6 +362,17 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
p.dispatchScrobble(ctx, mf, now)
|
||||
}
|
||||
}
|
||||
p.sessionsMu.Lock()
|
||||
info, getErr := p.playMap.Get(clientId)
|
||||
// A late stop for a previous track must not end the current session nor reach
|
||||
// playback reporters, or presence-style plugins would clear the active track.
|
||||
if getErr == nil && info.MediaFile.ID != params.MediaId {
|
||||
p.sessionsMu.Unlock()
|
||||
log.Trace(ctx, "Ignoring out-of-order stopped report for different track", "clientId", clientId, "stoppedMediaId", params.MediaId, "currentMediaId", info.MediaFile.ID)
|
||||
return nil
|
||||
}
|
||||
p.playMap.Remove(clientId)
|
||||
p.sessionsMu.Unlock()
|
||||
stoppedInfo := PlaybackSession{
|
||||
UserId: user.ID,
|
||||
Username: user.UserName,
|
||||
@ -349,7 +383,7 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
PlaybackRate: params.PlaybackRate,
|
||||
LastReport: now,
|
||||
}
|
||||
if info, getErr := p.playMap.Get(clientId); getErr == nil {
|
||||
if getErr == nil {
|
||||
stoppedInfo.MediaFile = info.MediaFile
|
||||
stoppedInfo.Start = info.Start
|
||||
} else {
|
||||
@ -364,7 +398,6 @@ func (p *playTracker) ReportPlayback(ctx context.Context, params ReportPlaybackP
|
||||
stoppedInfo.MediaFile = *mf
|
||||
}
|
||||
p.enqueuePlaybackReport(ctx, stoppedInfo)
|
||||
p.playMap.Remove(clientId)
|
||||
}
|
||||
|
||||
if conf.Server.EnableNowPlaying {
|
||||
|
||||
@ -3,6 +3,7 @@ package scrobbler
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
@ -45,6 +46,17 @@ func (m *mockPluginLoader) LoadScrobbler(name string) (Scrobbler, bool) {
|
||||
return s, ok
|
||||
}
|
||||
|
||||
// slowMediaFileRepo widens the window between a report's session check and its
|
||||
// write, making check-then-write races reproducible.
|
||||
type slowMediaFileRepo struct {
|
||||
model.MediaFileRepository
|
||||
}
|
||||
|
||||
func (s *slowMediaFileRepo) GetWithParticipants(id string) (*model.MediaFile, error) {
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
return s.MediaFileRepository.GetWithParticipants(id)
|
||||
}
|
||||
|
||||
var _ = Describe("PlayTracker", func() {
|
||||
var ctx context.Context
|
||||
var ds model.DataStore
|
||||
@ -376,18 +388,23 @@ var _ = Describe("PlayTracker", func() {
|
||||
Expect(playing).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("starting replaces existing entry for same player", func() {
|
||||
It("starting replaces existing entry when switching tracks on same player", func() {
|
||||
track2 := track
|
||||
track2.ID = "456"
|
||||
_ = ds.MediaFile(ctx).Put(&track2)
|
||||
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 50000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
MediaId: "456", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
playing, err := tracker.GetNowPlaying(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(playing).To(HaveLen(1))
|
||||
Expect(playing[0].MediaFile.ID).To(Equal("456"))
|
||||
Expect(playing[0].State).To(Equal("starting"))
|
||||
Expect(playing[0].PositionMs).To(Equal(int64(0)))
|
||||
})
|
||||
@ -696,6 +713,119 @@ var _ = Describe("PlayTracker", func() {
|
||||
})
|
||||
})
|
||||
|
||||
Describe("resilience (out-of-order reports)", func() {
|
||||
BeforeEach(func() {
|
||||
track2 := track
|
||||
track2.ID = "456"
|
||||
_ = ds.MediaFile(ctx).Put(&track2)
|
||||
})
|
||||
|
||||
It("does not downgrade an actively playing session when a late starting report arrives for the same track", func() {
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 1000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
playing, err := tracker.GetNowPlaying(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(playing).To(HaveLen(1))
|
||||
Expect(playing[0].State).To(Equal("playing"))
|
||||
Expect(playing[0].PositionMs).To(BeNumerically(">=", int64(1000)))
|
||||
})
|
||||
|
||||
It("keeps the current session when a stopped report arrives for a different track", func() {
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 90000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
playing, err := tracker.GetNowPlaying(ctx)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(playing).To(HaveLen(1))
|
||||
Expect(playing[0].MediaFile.ID).To(Equal("456"))
|
||||
Expect(playing[0].State).To(Equal("playing"))
|
||||
})
|
||||
|
||||
It("still auto-scrobbles the stopped track when the current session is for a different track", func() {
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 90000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(track.PlayCount).To(Equal(int64(1)))
|
||||
})
|
||||
|
||||
It("does not dispatch NowPlaying from an ignored out-of-order starting report", func() {
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 60000, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Eventually(func() bool { return fake.GetNowPlayingCalled() }).Should(BeTrue())
|
||||
fake.nowPlayingCalled.Store(false)
|
||||
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Consistently(func() bool { return fake.GetNowPlayingCalled() }).Should(BeFalse())
|
||||
})
|
||||
|
||||
It("never lets a concurrent starting report downgrade the playing session", func() {
|
||||
ds.(*tests.MockDataStore).MockedMediaFile = &slowMediaFileRepo{MediaFileRepository: ds.MediaFile(ctx)}
|
||||
for i := range 20 {
|
||||
raceClientId := fmt.Sprintf("race-client-%d", i)
|
||||
var wg sync.WaitGroup
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer GinkgoRecover()
|
||||
_ = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 0, State: "starting", PlaybackRate: 1.0, ClientId: raceClientId,
|
||||
})
|
||||
}()
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
defer GinkgoRecover()
|
||||
_ = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: raceClientId,
|
||||
})
|
||||
}()
|
||||
wg.Wait()
|
||||
info, err := tracker.playMap.Get(raceClientId)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(info.State).To(Equal("playing"), "iteration %d", i)
|
||||
}
|
||||
})
|
||||
|
||||
It("does NOT forward a stopped report for a different track to playback reporters", func() {
|
||||
err := tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "456", PositionMs: 0, State: "playing", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Eventually(func() bool { return fake.PlaybackReportCalled.Load() }).Should(BeTrue())
|
||||
fake.PlaybackReportCalled.Store(false)
|
||||
fake.LastPlaybackReport.Store(nil)
|
||||
|
||||
err = tracker.ReportPlayback(ctx, ReportPlaybackParams{
|
||||
MediaId: "123", PositionMs: 100000, State: "stopped", PlaybackRate: 1.0, ClientId: defaultClientId,
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
|
||||
Consistently(func() bool { return fake.PlaybackReportCalled.Load() }).Should(BeFalse())
|
||||
})
|
||||
})
|
||||
|
||||
Describe("external scrobbler dispatch", func() {
|
||||
It("dispatches NowPlaying on starting", func() {
|
||||
fake.nowPlayingCalled.Store(false)
|
||||
|
||||
@ -4,9 +4,12 @@ package criteria
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/utils"
|
||||
)
|
||||
|
||||
type Expression interface {
|
||||
@ -20,6 +23,7 @@ type Criteria struct {
|
||||
Limit int
|
||||
LimitPercent int
|
||||
Offset int
|
||||
RefreshDelay time.Duration // 0 = use conf.Server.SmartPlaylistRefreshDelay
|
||||
}
|
||||
|
||||
// EffectiveLimit resolves the effective limit for a query. If a fixed Limit is
|
||||
@ -83,6 +87,7 @@ func (c Criteria) MarshalJSON() ([]byte, error) {
|
||||
Limit int `json:"limit,omitempty"`
|
||||
LimitPercent int `json:"limitPercent,omitempty"`
|
||||
Offset int `json:"offset,omitempty"`
|
||||
RefreshDelay string `json:"refreshDelay,omitempty"`
|
||||
}{
|
||||
Sort: c.Sort,
|
||||
Order: c.Order,
|
||||
@ -90,6 +95,9 @@ func (c Criteria) MarshalJSON() ([]byte, error) {
|
||||
LimitPercent: c.LimitPercent,
|
||||
Offset: c.Offset,
|
||||
}
|
||||
if c.RefreshDelay > 0 {
|
||||
aux.RefreshDelay = utils.FormatDuration(c.RefreshDelay)
|
||||
}
|
||||
switch rules := c.Expression.(type) {
|
||||
case Any:
|
||||
aux.Any = rules
|
||||
@ -110,6 +118,7 @@ func (c *Criteria) UnmarshalJSON(data []byte) error {
|
||||
Limit int `json:"limit"`
|
||||
LimitPercent int `json:"limitPercent"`
|
||||
Offset int `json:"offset"`
|
||||
RefreshDelay string `json:"refreshDelay"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &aux); err != nil {
|
||||
return err
|
||||
@ -131,6 +140,14 @@ func (c *Criteria) UnmarshalJSON(data []byte) error {
|
||||
c.Limit = aux.Limit
|
||||
c.Offset = aux.Offset
|
||||
|
||||
if aux.RefreshDelay != "" {
|
||||
d, err := utils.ParseDuration(aux.RefreshDelay)
|
||||
if err != nil {
|
||||
return fmt.Errorf("invalid refreshDelay: %w", err)
|
||||
}
|
||||
c.RefreshDelay = d
|
||||
}
|
||||
|
||||
// Clamp LimitPercent to [0, 100]
|
||||
if aux.LimitPercent < 0 {
|
||||
log.Warn("limitPercent value out of range, clamping to 0", "value", aux.LimitPercent)
|
||||
|
||||
@ -3,6 +3,7 @@ package criteria
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/google/uuid"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@ -255,6 +256,71 @@ var _ = Describe("Criteria", func() {
|
||||
})
|
||||
})
|
||||
|
||||
Describe("refreshDelay", func() {
|
||||
newCriteria := func(extra string) []byte {
|
||||
return []byte(`{"all":[{"is":{"loved":true}}]` + extra + `}`)
|
||||
}
|
||||
|
||||
It("unmarshals a valid refreshDelay", func() {
|
||||
var c Criteria
|
||||
gomega.Expect(json.Unmarshal(newCriteria(`,"refreshDelay":"1d"`), &c)).To(gomega.Succeed())
|
||||
gomega.Expect(c.RefreshDelay).To(gomega.Equal(24 * time.Hour))
|
||||
})
|
||||
|
||||
It("supports week units", func() {
|
||||
var c Criteria
|
||||
gomega.Expect(json.Unmarshal(newCriteria(`,"refreshDelay":"1w"`), &c)).To(gomega.Succeed())
|
||||
gomega.Expect(c.RefreshDelay).To(gomega.Equal(7 * 24 * time.Hour))
|
||||
})
|
||||
|
||||
It("leaves RefreshDelay zero when absent", func() {
|
||||
var c Criteria
|
||||
gomega.Expect(json.Unmarshal(newCriteria(``), &c)).To(gomega.Succeed())
|
||||
gomega.Expect(c.RefreshDelay).To(gomega.BeZero())
|
||||
})
|
||||
|
||||
It("rejects an invalid refreshDelay", func() {
|
||||
var c Criteria
|
||||
err := json.Unmarshal(newCriteria(`,"refreshDelay":"tomorrow"`), &c)
|
||||
gomega.Expect(err).To(gomega.MatchError(gomega.ContainSubstring("refreshDelay")))
|
||||
})
|
||||
|
||||
It("rejects a negative refreshDelay", func() {
|
||||
var c Criteria
|
||||
err := json.Unmarshal(newCriteria(`,"refreshDelay":"-1h"`), &c)
|
||||
gomega.Expect(err).To(gomega.MatchError(gomega.ContainSubstring("refreshDelay")))
|
||||
})
|
||||
|
||||
It("marshals RefreshDelay back as a duration string", func() {
|
||||
c := Criteria{
|
||||
Expression: All{Is{"loved": true}},
|
||||
RefreshDelay: 24 * time.Hour,
|
||||
}
|
||||
j, err := json.Marshal(c)
|
||||
gomega.Expect(err).ToNot(gomega.HaveOccurred())
|
||||
gomega.Expect(string(j)).To(gomega.ContainSubstring(`"refreshDelay":"1d"`))
|
||||
})
|
||||
|
||||
It("omits refreshDelay from JSON when zero", func() {
|
||||
c := Criteria{Expression: All{Is{"loved": true}}}
|
||||
j, err := json.Marshal(c)
|
||||
gomega.Expect(err).ToNot(gomega.HaveOccurred())
|
||||
gomega.Expect(string(j)).ToNot(gomega.ContainSubstring("refreshDelay"))
|
||||
})
|
||||
|
||||
It("round-trips through marshal and unmarshal", func() {
|
||||
c := Criteria{
|
||||
Expression: All{Is{"loved": true}},
|
||||
RefreshDelay: 36 * time.Hour,
|
||||
}
|
||||
j, err := json.Marshal(c)
|
||||
gomega.Expect(err).ToNot(gomega.HaveOccurred())
|
||||
var c2 Criteria
|
||||
gomega.Expect(json.Unmarshal(j, &c2)).To(gomega.Succeed())
|
||||
gomega.Expect(c2.RefreshDelay).To(gomega.Equal(36 * time.Hour))
|
||||
})
|
||||
})
|
||||
|
||||
Context("with child playlists", func() {
|
||||
var (
|
||||
topLevelInPlaylistID string
|
||||
|
||||
@ -6,6 +6,7 @@ import (
|
||||
"strconv"
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/consts"
|
||||
"github.com/navidrome/navidrome/model/criteria"
|
||||
)
|
||||
@ -39,6 +40,15 @@ func (pls Playlist) IsSmartPlaylist() bool {
|
||||
return pls.Rules != nil && pls.Rules.Expression != nil
|
||||
}
|
||||
|
||||
// RefreshDelay returns the playlist's own refresh window when set, falling
|
||||
// back to the global SmartPlaylistRefreshDelay.
|
||||
func (pls Playlist) RefreshDelay() time.Duration {
|
||||
if pls.IsSmartPlaylist() && pls.Rules.RefreshDelay > 0 {
|
||||
return pls.Rules.RefreshDelay
|
||||
}
|
||||
return conf.Server.SmartPlaylistRefreshDelay
|
||||
}
|
||||
|
||||
func (pls Playlist) MediaFiles() MediaFiles {
|
||||
if len(pls.Tracks) == 0 {
|
||||
return nil
|
||||
|
||||
@ -1,7 +1,12 @@
|
||||
package model_test
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/conf/configtest"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
"github.com/navidrome/navidrome/model/criteria"
|
||||
"github.com/navidrome/navidrome/tests"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
@ -43,4 +48,29 @@ var _ = Describe("Playlist", func() {
|
||||
Expect(pls.ToM3U8()).To(Equal(expected))
|
||||
})
|
||||
})
|
||||
|
||||
Describe("RefreshDelay", func() {
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
conf.Server.SmartPlaylistRefreshDelay = 5 * time.Second
|
||||
})
|
||||
|
||||
It("returns the global config value when rules have no refreshDelay", func() {
|
||||
pls := model.Playlist{Rules: &criteria.Criteria{Expression: criteria.All{criteria.Is{"loved": true}}}}
|
||||
Expect(pls.RefreshDelay()).To(Equal(5 * time.Second))
|
||||
})
|
||||
|
||||
It("returns the per-playlist value when set", func() {
|
||||
pls := model.Playlist{Rules: &criteria.Criteria{
|
||||
Expression: criteria.All{criteria.Is{"loved": true}},
|
||||
RefreshDelay: 24 * time.Hour,
|
||||
}}
|
||||
Expect(pls.RefreshDelay()).To(Equal(24 * time.Hour))
|
||||
})
|
||||
|
||||
It("returns the global value for non-smart playlists", func() {
|
||||
pls := model.Playlist{}
|
||||
Expect(pls.RefreshDelay()).To(Equal(5 * time.Second))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
@ -4,7 +4,6 @@ import (
|
||||
"time"
|
||||
|
||||
. "github.com/Masterminds/squirrel"
|
||||
"github.com/navidrome/navidrome/conf"
|
||||
"github.com/navidrome/navidrome/log"
|
||||
"github.com/navidrome/navidrome/model"
|
||||
)
|
||||
@ -77,7 +76,7 @@ func (r *playlistRepository) shouldRefreshSmartPlaylist(pls *model.Playlist, usr
|
||||
if !pls.IsSmartPlaylist() {
|
||||
return false
|
||||
}
|
||||
if pls.EvaluatedAt != nil && time.Since(*pls.EvaluatedAt) < conf.Server.SmartPlaylistRefreshDelay {
|
||||
if pls.EvaluatedAt != nil && time.Since(*pls.EvaluatedAt) < pls.RefreshDelay() {
|
||||
return false
|
||||
}
|
||||
if pls.OwnerID != usr.ID {
|
||||
|
||||
@ -147,6 +147,50 @@ var _ = Describe("PlaylistRepository - Smart Playlists", func() {
|
||||
Expect(*nestedPlsAfterParentGet.EvaluatedAt).To(Equal(*nestedPlsRead.EvaluatedAt))
|
||||
})
|
||||
})
|
||||
|
||||
Context("per-playlist refreshDelay", func() {
|
||||
BeforeEach(func() {
|
||||
DeferCleanup(configtest.SetupConfig())
|
||||
})
|
||||
|
||||
It("does NOT refresh when the per-playlist delay has not elapsed, even if global has", func() {
|
||||
conf.Server.SmartPlaylistRefreshDelay = -1 * time.Second
|
||||
evaluatedAt := time.Now().Add(-1 * time.Hour)
|
||||
|
||||
rules := &criteria.Criteria{
|
||||
Expression: criteria.All{criteria.Contains{"title": "Day"}},
|
||||
RefreshDelay: 24 * time.Hour,
|
||||
}
|
||||
pls := model.Playlist{Name: "Frozen Daily", OwnerID: "userid", Rules: rules, EvaluatedAt: &evaluatedAt}
|
||||
Expect(repo.Put(&pls)).To(Succeed())
|
||||
DeferCleanup(func() { _ = repo.Delete(pls.ID) })
|
||||
|
||||
got, err := repo.GetWithTracks(pls.ID, true, false)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
// Not re-evaluated: EvaluatedAt unchanged, no tracks materialized
|
||||
Expect(*got.EvaluatedAt).To(BeTemporally("~", evaluatedAt, time.Second))
|
||||
Expect(got.Tracks).To(BeEmpty())
|
||||
})
|
||||
|
||||
It("refreshes when the per-playlist delay has elapsed, even if global has not", func() {
|
||||
conf.Server.SmartPlaylistRefreshDelay = 1 * time.Hour
|
||||
evaluatedAt := time.Now().Add(-10 * time.Minute)
|
||||
|
||||
rules := &criteria.Criteria{
|
||||
Expression: criteria.All{criteria.Contains{"title": "Day"}},
|
||||
RefreshDelay: 5 * time.Minute,
|
||||
}
|
||||
pls := model.Playlist{Name: "Fast Refresh", OwnerID: "userid", Rules: rules, EvaluatedAt: &evaluatedAt}
|
||||
Expect(repo.Put(&pls)).To(Succeed())
|
||||
DeferCleanup(func() { _ = repo.Delete(pls.ID) })
|
||||
|
||||
got, err := repo.GetWithTracks(pls.ID, true, false)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(*got.EvaluatedAt).To(BeTemporally("~", time.Now(), 2*time.Second))
|
||||
Expect(got.Tracks).To(HaveLen(1))
|
||||
Expect(got.Tracks[0].MediaFileID).To(Equal(songDayInALife.ID))
|
||||
})
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
|
||||
@ -14,6 +14,10 @@ const (
|
||||
FuncLyricsGetLyrics = "nd_lyrics_get_lyrics"
|
||||
)
|
||||
|
||||
// maxConcurrentLyricsCalls caps in-flight lyrics calls per plugin: clients prefetch
|
||||
// lyrics for whole queues, and the resulting burst can rate-limit upstream providers.
|
||||
const maxConcurrentLyricsCalls = 2
|
||||
|
||||
func init() {
|
||||
registerCapability(
|
||||
CapabilityLyrics,
|
||||
@ -34,6 +38,12 @@ type LyricsPlugin struct {
|
||||
// GetLyrics calls the plugin to fetch lyrics, then content-sniffs each response
|
||||
// via model.ParseLyrics (TTML/SRT/YAML/LRC/plain).
|
||||
func (l *LyricsPlugin) GetLyrics(ctx context.Context, mf *model.MediaFile) (model.LyricList, error) {
|
||||
select {
|
||||
case l.plugin.lyricsSem <- struct{}{}:
|
||||
defer func() { <-l.plugin.lyricsSem }()
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
req := capabilities.GetLyricsRequest{
|
||||
Track: mediaFileToTrackInfo(l.plugin, mf),
|
||||
}
|
||||
|
||||
@ -3,6 +3,8 @@
|
||||
package plugins
|
||||
|
||||
import (
|
||||
"context"
|
||||
|
||||
"github.com/navidrome/navidrome/model"
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
. "github.com/onsi/gomega"
|
||||
@ -71,6 +73,45 @@ var _ = Describe("LyricsPlugin", Ordered, func() {
|
||||
Expect(result[0].Lang).To(Equal("xxx"))
|
||||
})
|
||||
|
||||
It("blocks new calls while the per-plugin concurrency cap is saturated", func() {
|
||||
sem := provider.plugin.lyricsSem
|
||||
for range cap(sem) {
|
||||
sem <- struct{}{}
|
||||
}
|
||||
|
||||
ctx := GinkgoT().Context()
|
||||
track := &model.MediaFile{ID: "track-1", Title: "Test Song", Artist: "Test Artist"}
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := provider.GetLyrics(ctx, track)
|
||||
done <- err
|
||||
}()
|
||||
|
||||
Consistently(done, "500ms").ShouldNot(Receive())
|
||||
<-sem // free one slot; the pending call should now proceed
|
||||
Eventually(done).Should(Receive(BeNil()))
|
||||
for range cap(sem) - 1 {
|
||||
<-sem
|
||||
}
|
||||
})
|
||||
|
||||
It("gives up waiting for a slot when the context is cancelled", func() {
|
||||
sem := provider.plugin.lyricsSem
|
||||
for range cap(sem) {
|
||||
sem <- struct{}{}
|
||||
}
|
||||
defer func() {
|
||||
for range cap(sem) {
|
||||
<-sem
|
||||
}
|
||||
}()
|
||||
|
||||
ctx, cancel := context.WithCancel(GinkgoT().Context())
|
||||
cancel()
|
||||
_, err := provider.GetLyrics(ctx, &model.MediaFile{ID: "track-1"})
|
||||
Expect(err).To(MatchError(context.Canceled))
|
||||
})
|
||||
|
||||
It("returns error when plugin returns error", func() {
|
||||
manager, _ := createTestManagerWithPlugins(map[string]map[string]string{
|
||||
"test-lyrics": {"error": "service unavailable"},
|
||||
|
||||
@ -421,6 +421,7 @@ func (m *Manager) loadPluginWithConfig(p *model.Plugin) error {
|
||||
allowedUserIDs: allowedUsers,
|
||||
allUsers: p.AllUsers,
|
||||
libraries: newLibraryAccess(allowedLibraries, p.AllLibraries),
|
||||
lyricsSem: make(chan struct{}, maxConcurrentLyricsCalls),
|
||||
}
|
||||
m.mu.Unlock()
|
||||
loaded = true
|
||||
|
||||
@ -24,6 +24,7 @@ type plugin struct {
|
||||
allowedUserIDs []string // User IDs this plugin can access (from DB configuration)
|
||||
allUsers bool // If true, plugin can access all users
|
||||
libraries libraryAccess
|
||||
lyricsSem chan struct{} // Caps concurrent lyrics calls (see LyricsPlugin.GetLyrics)
|
||||
}
|
||||
|
||||
// instance creates a new plugin instance for the given context.
|
||||
|
||||
@ -22,6 +22,10 @@ Enabled = true
|
||||
ServerName = "My Music Server"
|
||||
# Optional: usernames to show in the client login user-picker (default: none). See "Public user list".
|
||||
ExposedPublicUsers = "alice, bob"
|
||||
# Optional: max collection responses streaming at once (default: half the DB connection pool,
|
||||
# min 2). Each streaming response holds a DB connection for its whole duration; excess requests
|
||||
# queue rather than fail.
|
||||
MaxConcurrentStreams = 4
|
||||
```
|
||||
|
||||
or via environment variables:
|
||||
@ -104,12 +108,14 @@ authenticated user has access to; a library (or item within it) the user cannot
|
||||
`GET /Items` accepts the filter params clients use to build screens: `ParentId` (a library view id
|
||||
for scoping, an artist id when browsing into an artist's albums, or an album id when browsing into
|
||||
an album's tracks); `AlbumArtistIds`/`ArtistIds`/`contributingArtistIds` (an artist's albums or
|
||||
tracks — Finamp's artist screen sends these *alongside* `ParentId=<libraryId>`); `GenreIds` (a
|
||||
tracks — Finamp's artist screen sends these *alongside* `ParentId=<libraryId>`); `AlbumIds` (an
|
||||
album's tracks — Feishin fetches them this way instead of `ParentId`); `GenreIds` (a
|
||||
genre's albums or tracks — Finamp's genre screen sends it the same way; `/Artists/AlbumArtists`
|
||||
and `MusicArtist` queries accept it too, matching artists credited on an album of that genre);
|
||||
`SearchTerm`;
|
||||
favorites-only (`Filters=IsFavorite` or the standalone `isFavorite=true`); `SortBy`/`SortOrder`;
|
||||
`StartIndex`/`Limit`; and `Ids` (batch fetch by id).
|
||||
`StartIndex`/`Limit`; and `Ids` (batch fetch by id). `Recursive=false` with a library `ParentId`
|
||||
returns direct children only (no tracks — no track is a library's direct child).
|
||||
|
||||
## Implemented endpoints
|
||||
|
||||
@ -124,9 +130,11 @@ favorites-only (`Filters=IsFavorite` or the standalone `isFavorite=true`); `Sort
|
||||
| Images | `GET Items/{itemId}/Images/{type}[/{index}]` (public), `POST`/`DELETE Items/{itemId}/Images/{type}` (playlist cover, authenticated) |
|
||||
| Favorites / ratings for songs, albums, artists, and playlists | `POST`/`DELETE UserFavoriteItems/{itemId}`, `POST`/`DELETE Users/{userId}/FavoriteItems/{itemId}`, `POST`/`DELETE Users/{userId}/Items/{itemId}/Rating`, `GET UserItems/{itemId}/UserData`, `GET Users/{userId}/Items/{itemId}/UserData` |
|
||||
| Streaming | `GET Audio/{itemId}/stream[.{container}]`, `GET Audio/{itemId}/universal`, `GET Audio/{itemId}/main.m3u8`, `GET Items/{itemId}/File`, `GET Items/{itemId}/Download`, `GET`/`POST Items/{itemId}/PlaybackInfo` |
|
||||
| Lyrics | `GET Audio/{itemId}/Lyrics` |
|
||||
| Playback reporting | `POST Sessions/Playing`, `POST Sessions/Playing/Progress`, `POST Sessions/Playing/Stopped`, `POST Sessions/Capabilities[/Full]` |
|
||||
| Playlists | `POST Playlists`, `GET Playlists/{playlistId}`, `POST Playlists/{playlistId}` (rename / visibility / replace tracks), `GET Playlists/{playlistId}/Items`, `POST`/`DELETE Playlists/{playlistId}/Items`, `GET Playlists/{playlistId}/Users[/{userId}]` |
|
||||
| Real-time | `GET socket` (WebSocket; keeps clients like Finamp from 404-loop-reconnecting) |
|
||||
| AudioMuse-AI (see below) | `GET AudioMuseAI/info`, `GET AudioMuseAI/health`, `GET AudioMuseAI/similar_tracks`, `GET AudioMuseAI/find_path` |
|
||||
|
||||
Any other path returns a `404` with a `{}` JSON body, and is logged server-side at `Debug` level
|
||||
as `Jellyfin API: unhandled route` (method + path). If a client you're testing needs an endpoint
|
||||
@ -217,7 +225,9 @@ The stream endpoints reuse the same transcode-decision pipeline as the Subsonic
|
||||
|
||||
## AudioMuse-AI compatible endpoints
|
||||
|
||||
Compatibility shim for Jellyfin front-ends that integrate [AudioMuse-AI](https://github.com/NeptuneHub/audiomuse-ai-plugin).
|
||||
Compatibility shim for Jellyfin front-ends that integrate [AudioMuse-AI](https://github.com/NeptuneHub/audiomuse-ai-plugin)
|
||||
— e.g. [Symfonium](https://symfonium.app/) can use these endpoints for sonic mixes when
|
||||
connected as a Jellyfin client.
|
||||
Backed natively by Navidrome's `core/sonic` engine (the `SonicSimilarity` plugin capability) — no
|
||||
external AudioMuse-AI backend or proxy is involved. The endpoints are gated on a `SonicSimilarity`
|
||||
plugin being loaded, like the Subsonic `sonicSimilarity` OpenSubsonic extension.
|
||||
@ -320,9 +330,6 @@ make test PKG=./server/jellyfin/...
|
||||
Access control for artists is enforced by scoping the `Artists`/`Items?IncludeItemTypes=MusicArtist`
|
||||
*list* to the user's libraries, plus the persistence layer's own defense-in-depth; a client
|
||||
that already has an artist id from elsewhere is not re-checked against library membership.
|
||||
- **MD5-hash ids from old migrated libraries.** The hex id codec assumes ids are opaque; a raw
|
||||
32-char MD5 id is itself valid hex and so must be encoded/decoded symmetrically like any other.
|
||||
This is handled, but is the most fragile id case — see the note in `dto/ids.go`.
|
||||
- **Blurhashes are synthetic, not computed from the artwork (follow-up).** `ImageBlurHashes` is
|
||||
populated by `dto/blurhash.go`, which derives a well-formed **1-component (solid color)**
|
||||
blurhash by hashing the item id — it never looks at the actual image. Real Jellyfin computes a
|
||||
@ -348,15 +355,8 @@ make test PKG=./server/jellyfin/...
|
||||
check — the column is never `""` post-scan), while `PlaybackInfo` runs the full pipeline per
|
||||
track so sidecar/plugin lyrics also light up. Feishin additionally requires server version
|
||||
≥ 10.9 — the reason `jellyfinVersion` is 10.9.11.
|
||||
Follow-ups: the lyrics cache loader is not singleflighted, so concurrent misses on the same
|
||||
track can double-invoke the plugin pipeline (fix belongs in `utils/cache.SimpleCache` via
|
||||
ttlcache's `SuppressedLoader`, affecting all callers — separate change); tracks whose only
|
||||
lyrics are sidecar/plugin-sourced show no `HasLyrics` badge in lists (request-time sources
|
||||
can't be known at list time without per-row I/O).
|
||||
- **No sonic similarity (follow-up).** `Items/{id}/InstantMix` and the `/Similar` endpoints are
|
||||
backed only by external metadata agents (Last.fm), not sonic analysis: an instant mix is the seed
|
||||
track followed by the provider's similar songs (with agents disabled it degrades to a seed-only
|
||||
mix). A follow-up would back them with Navidrome's `core/sonic` provider — the same one behind
|
||||
the OpenSubsonic `sonicSimilarity` extension (`getSonicSimilarTracks`) that AudioMuse-AI feeds
|
||||
via its Navidrome plugin, and the exact endpoint AudioMuse's own Jellyfin plugin overrides.
|
||||
Needs the `core/sonic.Sonic` service injected into the `Router` (wire change).
|
||||
Concurrent misses on the same track share one pipeline invocation (`SimpleCache.GetWithLoader`
|
||||
is singleflighted), and the load runs detached from the request context with a one-minute bound,
|
||||
so a cancelled request or hung plugin can't fail or pin the load for other waiters.
|
||||
Follow-up: tracks whose only lyrics are sidecar/plugin-sourced show no `HasLyrics` badge in
|
||||
lists (request-time sources can't be known at list time without per-row I/O).
|
||||
|
||||
@ -10,11 +10,18 @@ import (
|
||||
"github.com/navidrome/navidrome/server/jellyfin/dto"
|
||||
)
|
||||
|
||||
const lyricsLoadTimeout = time.Minute
|
||||
|
||||
// cachedLyrics resolves lyrics through the full source pipeline (embedded, sidecar, plugins),
|
||||
// caching results — including empty: clients poll per played track, so misses are the hot path.
|
||||
func (api *Router) cachedLyrics(ctx context.Context, mf *model.MediaFile) model.LyricList {
|
||||
// The load is shared across requests (singleflight) and cached, so don't let one
|
||||
// cancelled request abort it for everybody — detach it from the request's lifetime,
|
||||
// keeping a bound so a hung plugin can't pin the fetch (and its plugin slot) forever.
|
||||
loadCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), lyricsLoadTimeout)
|
||||
defer cancel()
|
||||
list, err := api.lyricsCache.GetWithLoader(mf.ID, func(string) (model.LyricList, time.Duration, error) {
|
||||
l, err := api.lyrics.GetLyrics(ctx, mf)
|
||||
l, err := api.lyrics.GetLyrics(loadCtx, mf)
|
||||
return l, 0, err // 0 → cache DefaultTTL
|
||||
})
|
||||
if err != nil {
|
||||
|
||||
@ -17,13 +17,18 @@ import (
|
||||
|
||||
// fakeLyricsService returns canned lyrics per media-file ID and counts calls.
|
||||
type fakeLyricsService struct {
|
||||
lyrics map[string]model.LyricList
|
||||
err error
|
||||
calls int
|
||||
lyrics map[string]model.LyricList
|
||||
err error
|
||||
calls int
|
||||
hadDeadline bool
|
||||
}
|
||||
|
||||
func (f *fakeLyricsService) GetLyrics(_ context.Context, mf *model.MediaFile) (model.LyricList, error) {
|
||||
func (f *fakeLyricsService) GetLyrics(ctx context.Context, mf *model.MediaFile) (model.LyricList, error) {
|
||||
f.calls++
|
||||
_, f.hadDeadline = ctx.Deadline()
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
@ -129,4 +134,22 @@ var _ = Describe("getLyrics", func() {
|
||||
Expect(doRequest("s2").Code).To(Equal(http.StatusNotFound))
|
||||
Expect(fake.calls).To(Equal(1))
|
||||
})
|
||||
|
||||
It("completes and caches the fetch even when the request context is cancelled", func() {
|
||||
fake.lyrics["s1"] = model.LyricList{
|
||||
{Kind: "main", Synced: true, Line: []model.Line{{Start: p(1000), Value: "hello"}}},
|
||||
}
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancel()
|
||||
|
||||
list := api.cachedLyrics(ctx, &model.MediaFile{ID: "s1"})
|
||||
Expect(list).ToNot(BeEmpty())
|
||||
Expect(doRequest("s1").Code).To(Equal(http.StatusOK))
|
||||
Expect(fake.calls).To(Equal(1))
|
||||
})
|
||||
|
||||
It("bounds the detached fetch with a timeout", func() {
|
||||
Expect(doRequest("s2").Code).To(Equal(http.StatusNotFound))
|
||||
Expect(fake.hadDeadline).To(BeTrue())
|
||||
})
|
||||
})
|
||||
|
||||
@ -168,7 +168,7 @@ func buildOSPlaylist(ctx context.Context, p model.Playlist) *responses.OpenSubso
|
||||
pls.Readonly = true
|
||||
|
||||
if p.EvaluatedAt != nil {
|
||||
pls.ValidUntil = new(p.EvaluatedAt.Add(conf.Server.SmartPlaylistRefreshDelay))
|
||||
pls.ValidUntil = new(p.EvaluatedAt.Add(p.RefreshDelay()))
|
||||
}
|
||||
} else {
|
||||
user, ok := request.UserFrom(ctx)
|
||||
|
||||
@ -248,6 +248,20 @@ var _ = Describe("buildPlaylist", func() {
|
||||
Expect(result.OpenSubsonicPlaylist).To(BeNil())
|
||||
})
|
||||
})
|
||||
|
||||
Context("with a per-playlist refreshDelay", func() {
|
||||
BeforeEach(func() {
|
||||
playlist.Rules.RefreshDelay = 24 * time.Hour
|
||||
player := model.Player{Client: "regular-client"}
|
||||
ctx = request.WithPlayer(ctx, player)
|
||||
})
|
||||
|
||||
It("computes validUntil from the playlist's own delay", func() {
|
||||
result := router.buildPlaylist(ctx, playlist)
|
||||
expected := evaluatedAt.Add(24 * time.Hour)
|
||||
Expect(result.ValidUntil).To(Equal(&expected))
|
||||
})
|
||||
})
|
||||
})
|
||||
|
||||
Describe("annotation leakage", func() {
|
||||
|
||||
79
utils/cache/simple_cache.go
vendored
79
utils/cache/simple_cache.go
vendored
@ -5,6 +5,7 @@ import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"runtime"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
@ -44,7 +45,8 @@ func NewSimpleCache[K comparable, V any](options ...Options) SimpleCache[K, V] {
|
||||
|
||||
c := ttlcache.New[K, V](opts...)
|
||||
cache := &simpleCache[K, V]{
|
||||
data: c,
|
||||
data: c,
|
||||
loads: make(map[K]*flight[V]),
|
||||
}
|
||||
go cache.data.Start()
|
||||
|
||||
@ -61,6 +63,23 @@ const evictionTimeout = 1 * time.Hour
|
||||
type simpleCache[K comparable, V any] struct {
|
||||
data *ttlcache.Cache[K, V]
|
||||
evictionDeadline atomic.Pointer[time.Time]
|
||||
loadsMu sync.Mutex
|
||||
loads map[K]*flight[V]
|
||||
}
|
||||
|
||||
// flight tracks an in-progress load so concurrent misses of the same key share it.
|
||||
type flight[V any] struct {
|
||||
done chan struct{}
|
||||
val V
|
||||
err error
|
||||
}
|
||||
|
||||
func (f *flight[V]) result() (V, error) {
|
||||
if f.err != nil {
|
||||
var zero V
|
||||
return zero, fmt.Errorf("cache error: loader returned %w", f.err)
|
||||
}
|
||||
return f.val, nil
|
||||
}
|
||||
|
||||
func (c *simpleCache[K, V]) Add(key K, value V) error {
|
||||
@ -90,31 +109,47 @@ func (c *simpleCache[K, V]) Get(key K) (V, error) {
|
||||
return item.Value(), nil
|
||||
}
|
||||
|
||||
// GetWithLoader loads misses via the loader, deduplicating concurrent loads of
|
||||
// the same key: one loader call runs, and every waiter shares its result (or error).
|
||||
func (c *simpleCache[K, V]) GetWithLoader(key K, loader func(key K) (V, time.Duration, error)) (V, error) {
|
||||
var err error
|
||||
loaderWrapper := ttlcache.LoaderFunc[K, V](
|
||||
func(t *ttlcache.Cache[K, V], key K) *ttlcache.Item[K, V] {
|
||||
c.evictExpired()
|
||||
var value V
|
||||
var ttl time.Duration
|
||||
value, ttl, err = loader(key)
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
return t.Set(key, value, ttl)
|
||||
},
|
||||
)
|
||||
item := c.data.Get(key, ttlcache.WithLoader[K, V](loaderWrapper))
|
||||
if item == nil {
|
||||
var zero V
|
||||
if err != nil {
|
||||
return zero, fmt.Errorf("cache error: loader returned %w", err)
|
||||
}
|
||||
return zero, errors.New("item not found")
|
||||
if item := c.data.Get(key); item != nil {
|
||||
return item.Value(), nil
|
||||
}
|
||||
return item.Value(), nil
|
||||
|
||||
c.loadsMu.Lock()
|
||||
if f, ok := c.loads[key]; ok {
|
||||
c.loadsMu.Unlock()
|
||||
<-f.done
|
||||
return f.result()
|
||||
}
|
||||
f := &flight[V]{done: make(chan struct{}), err: errLoaderPanicked}
|
||||
c.loads[key] = f
|
||||
c.loadsMu.Unlock()
|
||||
|
||||
// Deregister even if the loader panics, so waiters get an error instead of
|
||||
// blocking forever on a flight that will never complete.
|
||||
defer func() {
|
||||
close(f.done)
|
||||
c.loadsMu.Lock()
|
||||
delete(c.loads, key)
|
||||
c.loadsMu.Unlock()
|
||||
}()
|
||||
|
||||
if item := c.data.Get(key); item != nil { // a flight may have completed since the miss
|
||||
f.val, f.err = item.Value(), nil
|
||||
} else {
|
||||
c.evictExpired()
|
||||
var ttl time.Duration
|
||||
f.val, ttl, f.err = loader(key)
|
||||
if f.err == nil {
|
||||
c.data.Set(key, f.val, ttl)
|
||||
}
|
||||
}
|
||||
return f.result()
|
||||
}
|
||||
|
||||
var errLoaderPanicked = errors.New("loader panicked")
|
||||
|
||||
func (c *simpleCache[K, V]) evictExpired() {
|
||||
if c.evictionDeadline.Load() == nil || c.evictionDeadline.Load().Before(time.Now()) {
|
||||
c.data.DeleteExpired()
|
||||
|
||||
112
utils/cache/simple_cache_test.go
vendored
112
utils/cache/simple_cache_test.go
vendored
@ -3,6 +3,8 @@ package cache
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
. "github.com/onsi/ginkgo/v2"
|
||||
@ -69,6 +71,116 @@ var _ = Describe("SimpleCache", func() {
|
||||
_, err := cache.GetWithLoader("key", loader)
|
||||
Expect(err).To(HaveOccurred())
|
||||
})
|
||||
|
||||
It("suppresses concurrent loads for the same key", func() {
|
||||
var calls atomic.Int32
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 10)
|
||||
loader := func(key string) (string, time.Duration, error) {
|
||||
calls.Add(1)
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return "shared", time.Minute, nil
|
||||
}
|
||||
|
||||
const n = 5
|
||||
var wg sync.WaitGroup
|
||||
results := make([]string, n)
|
||||
errs := make([]error, n)
|
||||
for i := range n {
|
||||
wg.Go(func() {
|
||||
results[i], errs[i] = cache.GetWithLoader("key", loader)
|
||||
})
|
||||
}
|
||||
|
||||
Eventually(started).Should(Receive())
|
||||
Consistently(started).ShouldNot(Receive())
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
Expect(calls.Load()).To(Equal(int32(1)))
|
||||
for i := range n {
|
||||
Expect(errs[i]).ToNot(HaveOccurred())
|
||||
Expect(results[i]).To(Equal("shared"))
|
||||
}
|
||||
})
|
||||
|
||||
It("returns the loader error to all concurrent callers", func() {
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 10)
|
||||
loader := func(key string) (string, time.Duration, error) {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return "", 0, errors.New("load failed")
|
||||
}
|
||||
|
||||
const n = 3
|
||||
var wg sync.WaitGroup
|
||||
errs := make([]error, n)
|
||||
for i := range n {
|
||||
wg.Go(func() {
|
||||
_, errs[i] = cache.GetWithLoader("key", loader)
|
||||
})
|
||||
}
|
||||
|
||||
Eventually(started).Should(Receive())
|
||||
Consistently(started).ShouldNot(Receive())
|
||||
close(release)
|
||||
wg.Wait()
|
||||
|
||||
for i := range n {
|
||||
Expect(errs[i]).To(MatchError(ContainSubstring("load failed")))
|
||||
}
|
||||
})
|
||||
|
||||
It("supports interface value types with nil results", func() {
|
||||
c := NewSimpleCache[string, any]()
|
||||
v, err := c.GetWithLoader("key", func(string) (any, time.Duration, error) {
|
||||
return nil, time.Minute, nil
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(v).To(BeNil())
|
||||
})
|
||||
|
||||
It("cleans up the in-flight registration when the loader panics", func() {
|
||||
Expect(func() {
|
||||
_, _ = cache.GetWithLoader("key", func(string) (string, time.Duration, error) {
|
||||
panic("boom")
|
||||
})
|
||||
}).To(PanicWith("boom"))
|
||||
|
||||
// Without cleanup this would deadlock on the never-completed flight
|
||||
v, err := cache.GetWithLoader("key", func(string) (string, time.Duration, error) {
|
||||
return "ok", 0, nil
|
||||
})
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(v).To(Equal("ok"))
|
||||
})
|
||||
|
||||
It("loads different keys independently", func() {
|
||||
release := make(chan struct{})
|
||||
started := make(chan struct{}, 10)
|
||||
loader := func(key string) (string, time.Duration, error) {
|
||||
started <- struct{}{}
|
||||
<-release
|
||||
return key + "=value", time.Minute, nil
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
for _, key := range []string{"key1", "key2"} {
|
||||
wg.Go(func() {
|
||||
value, err := cache.GetWithLoader(key, loader)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(value).To(Equal(key + "=value"))
|
||||
})
|
||||
}
|
||||
|
||||
// Both loaders must be in flight at once: distinct keys are not suppressed
|
||||
Eventually(started).Should(Receive())
|
||||
Eventually(started).Should(Receive())
|
||||
close(release)
|
||||
wg.Wait()
|
||||
})
|
||||
})
|
||||
|
||||
Describe("Keys and Values", func() {
|
||||
|
||||
@ -1,6 +1,12 @@
|
||||
package utils
|
||||
|
||||
import "time"
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TimeNewest(times ...time.Time) time.Time {
|
||||
newest := time.Time{}
|
||||
@ -11,3 +17,59 @@ func TimeNewest(times ...time.Time) time.Time {
|
||||
}
|
||||
return newest
|
||||
}
|
||||
|
||||
var durationDayWeekRe = regexp.MustCompile(`-?\d+(?:\.\d+)?[dw]`)
|
||||
|
||||
// ParseDuration is time.ParseDuration extended with d (24h) and w (168h) units.
|
||||
// Negative durations are rejected.
|
||||
func ParseDuration(s string) (time.Duration, error) {
|
||||
expanded := durationDayWeekRe.ReplaceAllStringFunc(s, func(match string) string {
|
||||
value, err := strconv.ParseFloat(match[:len(match)-1], 64)
|
||||
if err != nil {
|
||||
return match
|
||||
}
|
||||
hours := value * 24
|
||||
if match[len(match)-1] == 'w' {
|
||||
hours = value * 24 * 7
|
||||
}
|
||||
return strconv.FormatFloat(hours, 'f', -1, 64) + "h"
|
||||
})
|
||||
d, err := time.ParseDuration(expanded)
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("invalid duration %q: %w", s, err)
|
||||
}
|
||||
if d < 0 {
|
||||
return 0, fmt.Errorf("negative duration not allowed: %q", s)
|
||||
}
|
||||
return d, nil
|
||||
}
|
||||
|
||||
// FormatDuration renders whole w/d multiples with those units, falling back to
|
||||
// time.Duration.String for the sub-day remainder, so ParseDuration round-trips.
|
||||
func FormatDuration(d time.Duration) string {
|
||||
if d < 24*time.Hour {
|
||||
return formatSubDay(d)
|
||||
}
|
||||
var b strings.Builder
|
||||
weekDuration := 7 * 24 * time.Hour
|
||||
if weeks := d / weekDuration; weeks > 0 {
|
||||
b.WriteString(strconv.Itoa(int(weeks)) + "w")
|
||||
d %= weekDuration
|
||||
}
|
||||
dayDuration := 24 * time.Hour
|
||||
if days := d / dayDuration; days > 0 {
|
||||
b.WriteString(strconv.Itoa(int(days)) + "d")
|
||||
d %= dayDuration
|
||||
}
|
||||
if d > 0 {
|
||||
b.WriteString(formatSubDay(d))
|
||||
}
|
||||
return b.String()
|
||||
}
|
||||
|
||||
func formatSubDay(d time.Duration) string {
|
||||
if d >= time.Hour && d%time.Hour == 0 {
|
||||
return strconv.Itoa(int(d/time.Hour)) + "h"
|
||||
}
|
||||
return d.String()
|
||||
}
|
||||
|
||||
@ -26,3 +26,74 @@ var _ = Describe("TimeNewest", func() {
|
||||
Expect(utils.TimeNewest(t1, t2, t3)).To(Equal(t2))
|
||||
})
|
||||
})
|
||||
|
||||
var _ = Describe("ParseDuration", func() {
|
||||
DescribeTable("parses valid durations",
|
||||
func(input string, expected time.Duration) {
|
||||
d, err := utils.ParseDuration(input)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(d).To(Equal(expected))
|
||||
},
|
||||
Entry("standard Go units", "90m", 90*time.Minute),
|
||||
Entry("hours", "12h", 12*time.Hour),
|
||||
Entry("days", "1d", 24*time.Hour),
|
||||
Entry("weeks", "1w", 7*24*time.Hour),
|
||||
Entry("multiple days", "3d", 72*time.Hour),
|
||||
Entry("mixed day and hours", "1d12h", 36*time.Hour),
|
||||
Entry("mixed week, day and hours", "1w2d3h", (7*24+2*24+3)*time.Hour),
|
||||
Entry("fractional days", "0.5d", 12*time.Hour),
|
||||
)
|
||||
|
||||
DescribeTable("rejects invalid durations",
|
||||
func(input string) {
|
||||
_, err := utils.ParseDuration(input)
|
||||
Expect(err).To(HaveOccurred())
|
||||
},
|
||||
Entry("empty string", ""),
|
||||
Entry("not a duration", "tomorrow"),
|
||||
Entry("bare number", "42"),
|
||||
Entry("unit only", "d"),
|
||||
Entry("unknown unit", "5y"),
|
||||
)
|
||||
|
||||
DescribeTable("rejects negative durations",
|
||||
func(input string) {
|
||||
_, err := utils.ParseDuration(input)
|
||||
Expect(err).To(MatchError(ContainSubstring("negative duration")))
|
||||
},
|
||||
Entry("negative days", "-1d"),
|
||||
Entry("negative weeks", "-0.5w"),
|
||||
Entry("negative Go units", "-30m"),
|
||||
)
|
||||
})
|
||||
|
||||
var _ = Describe("FormatDuration", func() {
|
||||
DescribeTable("formats durations using the largest whole units",
|
||||
func(input time.Duration, expected string) {
|
||||
Expect(utils.FormatDuration(input)).To(Equal(expected))
|
||||
},
|
||||
Entry("whole weeks", 7*24*time.Hour, "1w"),
|
||||
Entry("whole days", 24*time.Hour, "1d"),
|
||||
Entry("multiple days", 72*time.Hour, "3d"),
|
||||
Entry("day and hours", 36*time.Hour, "1d12h"),
|
||||
Entry("week, day and hours", (7*24+2*24+3)*time.Hour, "1w2d3h"),
|
||||
Entry("hours only", 12*time.Hour, "12h"),
|
||||
Entry("sub-hour", 90*time.Minute, "1h30m0s"),
|
||||
Entry("zero", time.Duration(0), "0s"),
|
||||
)
|
||||
|
||||
DescribeTable("round-trips through ParseDuration",
|
||||
func(input string) {
|
||||
d, err := utils.ParseDuration(input)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
formatted := utils.FormatDuration(d)
|
||||
d2, err := utils.ParseDuration(formatted)
|
||||
Expect(err).ToNot(HaveOccurred())
|
||||
Expect(d2).To(Equal(d))
|
||||
},
|
||||
Entry("1d", "1d"),
|
||||
Entry("1w", "1w"),
|
||||
Entry("1d12h", "1d12h"),
|
||||
Entry("90m", "90m"),
|
||||
)
|
||||
})
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user