0.2.79 - Slow api fixes

This commit is contained in:
ponzischeme89
2026-08-19 18:08:00 +12:00
parent 590e069366
commit 0782545013
41 changed files with 1820 additions and 317 deletions
@@ -40,12 +40,24 @@ func (s *Server) gatewaySettingsResponse() adminGatewaySettingsResponse {
SonarrAlertMinutes: int(s.sonarrAlertWindow() / time.Minute),
RadarrAlertMinutes: int(s.radarrAlertWindow() / time.Minute),
EmbyHealthSeconds: int(s.embyHealthInterval() / time.Second),
SlowRequestMillis: effectiveSlowRequestMillis(s.slowRequestThreshold()),
LibrarySyncMinutes: int(s.LibrarySyncInterval() / time.Minute),
},
LogLevels: store.GatewayLogLevels,
Version: buildinfo.Version(),
}
}
// effectiveSlowRequestMillis reports the switched-off threshold as zero rather than as
// the impossible duration the reader uses, because the console draws the effective value
// beside the field an operator cleared and "0" is what its own vocabulary means by off.
func effectiveSlowRequestMillis(threshold time.Duration) int {
if threshold <= 0 || threshold > time.Hour {
return 0
}
return int(threshold / time.Millisecond)
}
func (s *Server) effectiveLogLevel() string {
if s.logLevel != nil {
return levelName(s.logLevel.Level())
@@ -121,6 +133,12 @@ func gatewaySettingsChanges(before, after store.GatewaySettings) string {
if before.EmbyHealthSeconds != after.EmbyHealthSeconds {
changes = append(changes, "Emby health probe")
}
if before.SlowRequestMillis != after.SlowRequestMillis {
changes = append(changes, "slow-request threshold")
}
if before.LibrarySyncMinutes != after.LibrarySyncMinutes {
changes = append(changes, "library sweep interval")
}
switch len(changes) {
case 0:
return ""
+31
View File
@@ -40,6 +40,7 @@ import (
"github.com/ponzischeme89/memby/server/internal/scheduler"
"github.com/ponzischeme89/memby/server/internal/sonarr"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Server struct {
@@ -107,6 +108,19 @@ type Server struct {
// playbackTitles lets a progress or stop report, which carries only an item id, be
// logged by name.
playbackTitles playbackTitles
// featurePolicy keeps the operator's feature switches out of the request path. See
// features_cache.go.
featurePolicy featurePolicyCache
// upstream deduplicates concurrent cache misses for the same key, so two televisions
// asking for the same expensive answer at the same moment cost one upstream call
// rather than two. See coalesce.go.
upstream upstreamGroup
// household caches the completion scores every viewer's ranking reads and none of
// them can get a different answer to. See ranking.go.
household householdScores
// followChecks stops the automatic My Shows check repeating for the length of an
// episode. See playback_follow.go.
followChecks followChecks
// ingestRuns collapses a season pack's worth of finished scans into one banner.
ingestRuns ingestRuns
@@ -505,6 +519,13 @@ func (s *Server) withLogging(next http.Handler) http.Handler {
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
start := time.Now()
r, identity := withRequestIdentity(r)
// Every layer under this one — the Emby client, the *arr clients, Postgres,
// Redis — records against this. It is installed for every request rather than
// only for ones that turn out to be slow, because nothing knows a request is
// slow until it has finished and the evidence has to have been collected on the
// way through. What it costs when the request is fast is a map nobody reads.
ctx, trace := timing.New(r.Context())
r = r.WithContext(ctx)
w.Header().Set("X-Memby-Correlation", identity.correlation)
s.loggerFor(r.Context()).Log(r.Context(), serverlogging.LevelTrace, "request started",
"method", r.Method, "path", r.URL.Path, "query_keys", queryKeys(r))
@@ -539,6 +560,16 @@ func (s *Server) withLogging(next http.Handler) http.Handler {
if cached := rec.Header().Get("X-Memby-Cache"); cached != "" {
fields = append(fields, "cache", cached)
}
// A duration on its own says something is slow and nothing about why, which is
// the state every latency investigation here started from. The breakdown is
// attached only past the threshold: on a fast request it would be a second
// column of noise on every line, and on a slow one it is the answer.
if elapsed := time.Since(start); elapsed >= s.slowRequestThreshold() && !trace.Empty() {
fields = append(fields, "breakdown", trace.Breakdown())
if rest := trace.Unattributed(elapsed); rest > 0 {
fields = append(fields, "elsewhere", rest.Round(time.Millisecond))
}
}
s.log.Log(r.Context(), level, "request", fields...)
})
}
+105
View File
@@ -0,0 +1,105 @@
package api
import (
"context"
"encoding/json"
"net/http"
"time"
"golang.org/x/sync/singleflight"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// coalesceTimeout bounds a shared upstream read. It is longer than any single call the
// gateway makes because it covers the whole miss — read, decorate, cache — and shorter
// than a television's patience, so a joiner is never held past the point where the
// answer would have been useful.
const coalesceTimeout = 30 * time.Second
// cachedRead is the shape of nearly every read-through route here: look in Redis, ask
// upstream on a miss, store what came back. What it adds is the thing missing from every
// hand-written copy of that shape — only one request at a time actually asks.
//
// The failure this exists for is a real one and it is visible in the log as a sequence:
// the same person lookup answering in 2.2s, then 4.6s, then 7.7s within a few seconds.
// Nothing was getting slower. Three televisions — or one television warming three cast
// cards — missed the same key at the same moment, all three asked Emby, and Emby served
// three copies of one expensive query while each of them waited for the other two's work
// to finish. The cache could not help: nothing had been written yet when the second and
// third arrived. Coalescing is what turns that into one upstream call and three fast
// answers, and it gets better rather than worse as the household grows.
//
// Two properties are load-bearing:
//
// - **The shared work does not inherit the caller's cancellation.** singleflight hands
// every joiner the first caller's result, so if that context is the one that goes
// away — a D-pad moving off a card, a television giving up — everybody waiting behind
// it is failed by a request none of them made. The work runs on a detached context
// with its own deadline for that reason.
// - **A joiner is counted.** `coalesced` on the breakdown is what distinguishes "this
// route is slow" from "this route is slow because it is queued behind itself", which
// are different problems with different fixes.
func (s *Server) cachedRead(
ctx context.Context,
key string,
ttl time.Duration,
load func(context.Context) (json.RawMessage, error),
) (json.RawMessage, bool, error) {
if raw, err := s.cache.Get(ctx, key); err == nil {
return raw, true, nil
}
body, err := s.buildCached(ctx, key, func(json.RawMessage) time.Duration { return ttl }, load)
return body, false, err
}
// buildCached is cachedRead without the lookup, for a route that has to do some of its
// own work before it knows what to build — and for one whose time to live depends on
// what came back. The lookup is not optional there, only earlier.
func (s *Server) buildCached(
ctx context.Context,
key string,
ttlFor func(json.RawMessage) time.Duration,
load func(context.Context) (json.RawMessage, error),
) (json.RawMessage, error) {
shared, err, joined := s.upstream.Do(key, func() (any, error) {
// Detached deliberately — see above. The parent's values (the trace, the request
// identity) are kept so the work is still attributable to the request that
// triggered it.
work, cancel := context.WithTimeout(context.WithoutCancel(ctx), coalesceTimeout)
defer cancel()
body, err := load(work)
if err != nil {
return nil, err
}
if err := s.cache.Set(work, key, body, ttlFor(body)); err != nil {
s.loggerFor(work).Warn("cache write failed", "key", key, "error", err)
}
return body, nil
})
if joined {
timing.Count(ctx, "coalesced")
}
if err != nil {
return nil, err
}
body, _ := shared.(json.RawMessage)
return body, nil
}
// upstreamGroup is the deduplicator itself. One per server: the keys are already the
// cache keys, which are namespaced per user where the answer is per user, so nothing in
// one household's traffic can join another's.
type upstreamGroup = singleflight.Group
// writeCached is the one place the cache header and the body are written together, so a
// route cannot answer from cache and report a miss. Two lines, and it was wrong at least
// once before it was one function.
func writeCached(w http.ResponseWriter, hit bool, body json.RawMessage) {
if hit {
w.Header().Set("X-Memby-Cache", "hit")
} else {
w.Header().Set("X-Memby-Cache", "miss")
}
writeRaw(w, http.StatusOK, body)
}
+58 -38
View File
@@ -1,11 +1,15 @@
package api
import (
"context"
"encoding/json"
"net/http"
"sync"
"github.com/ponzischeme89/memby/server/internal/cache"
"github.com/ponzischeme89/memby/server/internal/emby"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// extrasResponse is the Extras tab: featurettes, deleted scenes, interviews and the local
@@ -31,48 +35,64 @@ func (s *Server) handleExtras(w http.ResponseWriter, r *http.Request, sess store
writeError(w, http.StatusBadRequest, "item id is required")
return
}
key := cache.UserKey(sess.EmbyUserID, "extras:v1:"+itemID)
if raw, err := s.cache.Get(ctx, key); err == nil {
w.Header().Set("X-Memby-Cache", "hit")
writeRaw(w, http.StatusOK, raw)
return
}
cred := credentials(sess)
// Neither list failing is fatal, and only both failing is a failure worth reporting: a
// title can perfectly well have a trailer and no special features, or the reverse, and
// an Extras tab withheld because one of two lookups was unwell is the worse outcome.
var items []json.RawMessage
trailerErr := error(nil)
if result, err := s.emby.LocalTrailers(ctx, cred, itemID); err == nil {
items = append(items, result.Items...)
} else {
trailerErr = err
}
featureErr := error(nil)
if result, err := s.emby.SpecialFeatures(ctx, cred, itemID); err == nil {
items = append(items, result.Items...)
} else {
featureErr = err
}
if trailerErr != nil && featureErr != nil {
s.writeUpstreamError(ctx, w, featureErr, "could not load extras")
return
}
items = dedupeExtras(items)
body, err := json.Marshal(extrasResponse{Items: nonNilRaws(items)})
// Cached for the household rather than per viewer, and outside the namespace a
// playback stop invalidates: what a film ships with is a fact about the film, and
// nothing in this response carries user data. Empty answers included — most of a
// library has no extras at all and every detail page asks, so the "no" is the entry
// worth keeping.
key := cache.MetadataKey("extras:v2:" + itemID)
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
return s.extras(ctx, credentials(sess), itemID)
})
if err != nil {
writeError(w, http.StatusInternalServerError, "could not encode extras")
s.writeUpstreamError(ctx, w, err, "could not load extras")
return
}
// Cached for the ordinary item lifetime, empty answers included. Most of a library has
// no extras at all and every detail page asks, so the "no" is the entry worth keeping.
if err := s.cache.Set(ctx, key, body, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("extras cache write failed", "error", err)
writeCached(w, hit, body)
}
// extras joins the two places Emby keeps them.
//
// The two lookups are independent and are made together. They used to run in turn, which
// on a detail page open — where this is one of five requests the television makes at once
// — was one round trip of pure waiting for no reason at all.
//
// Neither list failing is fatal, and only both failing is a failure worth reporting: a
// title can perfectly well have a trailer and no special features, or the reverse, and an
// Extras tab withheld because one of two lookups was unwell is the worse outcome.
func (s *Server) extras(
ctx context.Context, cred emby.Credentials, itemID string,
) (json.RawMessage, error) {
var (
trailers *emby.ItemsResult
trailerErr error
features *emby.ItemsResult
featureErr error
wg sync.WaitGroup
)
wg.Add(2)
go func() {
defer wg.Done()
trailers, trailerErr = s.emby.LocalTrailers(timing.WithLabel(ctx, "emby.trailers"), cred, itemID)
}()
go func() {
defer wg.Done()
features, featureErr = s.emby.SpecialFeatures(timing.WithLabel(ctx, "emby.features"), cred, itemID)
}()
wg.Wait()
if trailerErr != nil && featureErr != nil {
return nil, featureErr
}
w.Header().Set("X-Memby-Cache", "miss")
writeRaw(w, http.StatusOK, body)
var items []json.RawMessage
if trailerErr == nil && trailers != nil {
items = append(items, trailers.Items...)
}
if featureErr == nil && features != nil {
items = append(items, features.Items...)
}
return json.Marshal(extrasResponse{Items: nonNilRaws(dedupeExtras(items))})
}
// dedupeExtras drops the same file appearing in both lists, in first-seen order.
+12 -6
View File
@@ -195,16 +195,21 @@ func evaluateFeature(policy store.FeaturePolicy, definition featureDefinition, p
return evaluatedFeature{featureDefinition: definition, Enabled: enabled, Source: source, Compatible: compatible}
}
// currentFeaturePolicy is read on the request path from sixteen places and by the status
// poll every open television makes, so it is cached rather than queried — see
// featurePolicyCache for how long and why that is safe.
func (s *Server) currentFeaturePolicy(ctx context.Context) store.FeaturePolicy {
if s.store == nil {
return store.DefaultFeaturePolicy()
}
policy, err := s.store.FeaturePolicy(ctx)
if err != nil {
s.loggerFor(ctx).Warn("feature policy unavailable; using safe defaults", "error", err)
return store.DefaultFeaturePolicy()
}
return policy
return s.featurePolicy.read(ctx, func(ctx context.Context) store.FeaturePolicy {
policy, err := s.store.FeaturePolicy(ctx)
if err != nil {
s.loggerFor(ctx).Warn("feature policy unavailable; using safe defaults", "error", err)
return store.DefaultFeaturePolicy()
}
return policy
})
}
func (s *Server) featureEnabled(ctx context.Context, key string) bool {
@@ -296,6 +301,7 @@ func (s *Server) handleAdminFeaturePolicy(w http.ResponseWriter, r *http.Request
writeError(w, http.StatusBadRequest, "unknown feature policy action")
return
}
s.featurePolicy.invalidate()
stored, err := s.store.SetFeaturePolicy(r.Context(), next, req.ExpectedRevision)
if err != nil {
if errors.Is(err, store.ErrFeaturePolicyConflict) {
+61
View File
@@ -0,0 +1,61 @@
package api
import (
"context"
"sync"
"time"
"github.com/ponzischeme89/memby/server/internal/store"
)
// featurePolicyTTL is how stale the cached feature policy may be.
//
// The document is read on the request path from sixteen places and by /v1/status, which
// every open television polls every ten seconds — so on a four-set household the policy
// was fetched from Postgres somewhere over twenty times a minute to answer a question
// whose answer changes when an operator presses a switch. Each read is a millisecond and
// none of them was ever the reason a screen was slow; what they were is a standing draw
// on a connection pool that the requests which *are* slow have to queue behind.
//
// Five seconds because the operator is the only writer and their own write clears this
// instance's copy outright: the window is not "how long until my change takes effect" but
// "how long until an instance that did not make the change notices", which is the same
// question WatchMaintenance answers with thirty.
const featurePolicyTTL = 5 * time.Second
// featurePolicyCache is a read-through cache with a deliberately simple concurrency
// story: a stale read takes the lock, refreshes, and every other caller waits for that
// one refresh rather than starting its own. That is worth stating because the opposite —
// releasing the lock to query — is what turns one expired entry into a thundering herd of
// identical queries, which is the failure this exists to prevent.
type featurePolicyCache struct {
mu sync.Mutex
value store.FeaturePolicy
valid bool
fetched time.Time
}
// read returns the cached value, refreshing through load when it has expired. load is
// only ever called with the lock held, so it must not itself read the cache.
func (c *featurePolicyCache) read(
ctx context.Context, load func(context.Context) store.FeaturePolicy,
) store.FeaturePolicy {
c.mu.Lock()
defer c.mu.Unlock()
if c.valid && time.Since(c.fetched) < featurePolicyTTL {
return c.value
}
c.value = load(ctx)
c.valid = true
c.fetched = time.Now()
return c.value
}
// invalidate drops the copy so the next read goes to Postgres. Called by the operator's
// own write: a console that showed the old answer back to the person who had just
// changed it would read as a save that failed.
func (c *featurePolicyCache) invalidate() {
c.mu.Lock()
defer c.mu.Unlock()
c.valid = false
}
+16
View File
@@ -3,6 +3,7 @@ package api
import (
"context"
"log/slog"
"math"
"sync"
"time"
@@ -124,6 +125,19 @@ func (s *Server) radarrAlertWindow() time.Duration {
s.cfg.RadarrAlertWindow)
}
// slowRequestThreshold is the line between a request that logs a breakdown and one that
// does not. Switched off it returns a duration no request can exceed rather than zero,
// because zero would mean *every* request carried one — the opposite of what the operator
// asked for.
func (s *Server) slowRequestThreshold() time.Duration {
threshold := overrideWindow(s.gatewaySettings.get().SlowRequestMillis, time.Millisecond,
s.cfg.SlowRequestThreshold)
if threshold <= 0 {
return math.MaxInt64
}
return threshold
}
func (s *Server) embyHealthInterval() time.Duration {
return overrideWindow(s.gatewaySettings.get().EmbyHealthSeconds, time.Second,
s.cfg.EmbyHealthInterval)
@@ -162,6 +176,7 @@ type deployedGatewaySettings struct {
SonarrAlertMinutes int `json:"sonarrAlertMinutes"`
RadarrAlertMinutes int `json:"radarrAlertMinutes"`
EmbyHealthSeconds int `json:"embyHealthSeconds"`
SlowRequestMillis int `json:"slowRequestMillis"`
LibrarySyncMinutes int `json:"librarySyncMinutes"`
}
@@ -177,6 +192,7 @@ func (s *Server) deployedSettings() deployedGatewaySettings {
SonarrAlertMinutes: int(s.cfg.SonarrAlertWindow / time.Minute),
RadarrAlertMinutes: int(s.cfg.RadarrAlertWindow / time.Minute),
EmbyHealthSeconds: int(s.cfg.EmbyHealthInterval / time.Second),
SlowRequestMillis: int(s.cfg.SlowRequestThreshold / time.Millisecond),
LibrarySyncMinutes: int(s.cfg.SyncInterval / time.Minute),
}
}
+60 -18
View File
@@ -18,6 +18,7 @@ import (
"github.com/ponzischeme89/memby/server/internal/emby"
"github.com/ponzischeme89/memby/server/internal/recommend"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// Field sets mirror what each TV row actually renders. Asking Emby for less is the
@@ -88,6 +89,11 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
}
cred := credentials(sess)
// The fan-out's wall clock, which is a different number from the summed Emby time
// beside it and the more useful of the two: the sum says how much work Emby did, this
// says how long the launcher waited for it. They diverge exactly when the calls stop
// running concurrently, which is the failure this instrumentation exists to catch.
upstream := timing.Start(ctx, "fanout")
var (
mu sync.Mutex
failures int
@@ -97,33 +103,39 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
forYouRows []recommend.Row
forYouRowStale bool
seriesPlayed map[string]time.Time
ranking rankingInputs
rowStats []store.RowStat
rowStatsOK bool
wg sync.WaitGroup
)
run := func(dest *[]json.RawMessage, fetch func() (*emby.ItemsResult, error)) {
// Each row names its own stage, so a slow launcher says which query it was waiting
// on rather than only that it was waiting on Emby. They are concurrent, so a single
// summed `emby=` would be the one number that could not answer that.
run := func(name string, dest *[]json.RawMessage, fetch func(context.Context) (*emby.ItemsResult, error)) {
wg.Add(1)
go func() {
defer wg.Done()
result, err := fetch()
result, err := fetch(timing.WithLabel(ctx, "emby."+name))
mu.Lock()
defer mu.Unlock()
if err != nil {
failures++
s.loggerFor(ctx).Warn("home row failed", "error", err)
s.loggerFor(ctx).Warn("home row failed", "row", name, "error", err)
return
}
*dest = result.Items
}()
}
run(&out.ContinueWatching, func() (*emby.ItemsResult, error) {
run("resume", &out.ContinueWatching, func(ctx context.Context) (*emby.ItemsResult, error) {
return s.emby.ResumeItems(ctx, cred, rowParams(url.Values{
"Recursive": {"true"},
"MediaTypes": {"Video"},
"Limit": {itoa(limit)},
}, fieldsContinue))
})
run(&out.Favorites, func() (*emby.ItemsResult, error) {
run("favourites", &out.Favorites, func(ctx context.Context) (*emby.ItemsResult, error) {
return s.emby.Items(ctx, cred, rowParams(url.Values{
"Filters": {"IsFavorite"},
"IncludeItemTypes": {"Movie,Series"},
@@ -133,7 +145,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
"Limit": {itoa(limit)},
}, fieldsRow))
})
run(&out.NextUp, func() (*emby.ItemsResult, error) {
run("nextup", &out.NextUp, func(ctx context.Context) (*emby.ItemsResult, error) {
return s.emby.NextUp(ctx, cred, rowParams(url.Values{
"Limit": {itoa(limit)},
}, fieldsContinue))
@@ -144,7 +156,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
wg.Add(1)
go func() {
defer wg.Done()
played, err := s.recentlyPlayedSeries(ctx, cred)
played, err := s.recentlyPlayedSeries(timing.WithLabel(ctx, "emby.recent"), cred)
if err != nil {
s.loggerFor(ctx).Warn("recently played lookup failed", "error", err)
return
@@ -153,7 +165,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
seriesPlayed = played
mu.Unlock()
}()
run(&out.LatestMovies, func() (*emby.ItemsResult, error) {
run("latest", &out.LatestMovies, func(ctx context.Context) (*emby.ItemsResult, error) {
newReleaseDays := s.weightedConfig().NewReleaseDays
if newReleaseDays < 1 {
newReleaseDays = recommend.DefaultWeightedConfig().NewReleaseDays
@@ -197,6 +209,30 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
mu.Unlock()
}()
}
// The ranking evidence and the row-order preferences depend on the viewer and on
// nothing that is being fetched around them, so they are read *beside* the Emby
// fan-out rather than after it. They used to be the first two things the response did
// once every row had arrived — eight Postgres reads with nothing else in flight,
// entirely on the critical path, for answers that were available before the request
// asked Emby anything.
wg.Add(1)
go func() {
defer wg.Done()
ranking = s.rankingInputs(ctx, sess.EmbyUserID)
}()
if s.store != nil {
wg.Add(1)
go func() {
defer wg.Done()
stats, err := s.store.UserRowStats(ctx, sess.EmbyUserID, time.Now().Add(-45*24*time.Hour))
if err != nil {
s.loggerFor(ctx).Warn("home row preferences unavailable",
"user", sess.EmbyUserID, "error", err)
return
}
rowStats, rowStatsOK = stats, true
}()
}
if s.forYou != nil {
// The prepared pool is one indexed PostgreSQL read. It runs beside the Emby
// calls and deliberately has no live-engine fallback, so Home can never inherit
@@ -226,6 +262,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
}
wg.Wait()
upstream()
if failures == 4 {
writeError(w, http.StatusBadGateway, "could not reach the emby server")
@@ -249,6 +286,7 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
if recommendations == nil {
s.refreshRecommendationsInBackground(sess)
}
assemble := timing.Start(ctx, timing.StageRows)
rows := baseRows(out)
nearContinue := make([]recommend.Row, 0, len(forYouRows)+2)
if len(forYouRows) > 0 {
@@ -271,34 +309,38 @@ func (s *Server) handleHome(w http.ResponseWriter, r *http.Request, sess store.S
}
out.Rows = append(rows, recommendations...)
out.Rows = s.filterRecommendationPermissions(ctx, sess, out.Rows)
if stats, err := s.store.UserRowStats(
ctx,
sess.EmbyUserID,
time.Now().Add(-45*24*time.Hour),
); err != nil {
s.loggerFor(ctx).Warn("home row preferences unavailable", "user", sess.EmbyUserID, "error", err)
} else {
out.Rows = personalizeHomeRows(out.Rows, stats)
if rowStatsOK {
out.Rows = personalizeHomeRows(out.Rows, rowStats)
}
out.Rows = s.personalizeTitles(ctx, sess, out.Rows)
assemble()
rank := timing.Start(ctx, timing.StageRank)
out.Rows = s.personalizeTitlesWith(out.Rows, ranking)
out.Rows = personalizeRowsByTitleScores(selectPersonalizedRows(out.Rows))
out.Rows = deduplicateRows(out.Rows)
rank()
// Ratings ride on the cards themselves. Only what is already stored is attached, so
// the launcher pays one indexed read rather than a request per poster, and a card
// shows its scores as it is drawn instead of when focus reaches it.
decorate := timing.Start(ctx, timing.StageRatings)
s.decorateHomeRatings(ctx, &out)
decorate()
// The hero is composed last, from the finished rows, because that is the only point
// at which the ratings it ranks by are already attached. It is prepended rather than
// inserted: the television consumes this row instead of drawing it, so its position
// among the shelves means nothing, and being first is what lets an older reader that
// does draw it put it somewhere sensible.
if hero {
if row := s.heroRow(ctx, out.Rows, sess.EmbyUserID, time.Now()); row != nil {
compose := timing.Start(ctx, timing.StageHero)
row := s.heroRow(ctx, out.Rows, sess.EmbyUserID, time.Now())
compose()
if row != nil {
out.Rows = append([]recommend.Row{*row}, out.Rows...)
}
}
encode := timing.Start(ctx, timing.StageEncode)
body, err := json.Marshal(out)
encode()
if err != nil {
s.loggerFor(ctx).Error("home encode failed", "error", err)
writeError(w, http.StatusInternalServerError, "could not build the home payload")
+119 -98
View File
@@ -7,10 +7,12 @@ import (
"net/url"
"strconv"
"strings"
"sync"
"github.com/ponzischeme89/memby/server/internal/cache"
"github.com/ponzischeme89/memby/server/internal/sonarr"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type seriesEpisodesResponse struct {
@@ -75,23 +77,21 @@ func itemDetailKey(userID, itemID string) string {
func (s *Server) detailItem(
ctx context.Context, sess store.Session, itemID string,
) (json.RawMessage, error) {
key := itemDetailKey(sess.EmbyUserID, itemID)
if raw, err := s.cache.Get(ctx, key); err == nil {
return raw, nil
}
item, err := s.emby.Item(ctx, credentials(sess), itemID, fieldsDetail)
if err != nil {
return nil, err
}
// A detail page can then draw its ratings with the rest of the hero rather than
// after a second request. Anything not yet stored still arrives on /ratings.
decorated := []json.RawMessage{item}
s.decorateItemRatings(ctx, decorated)
item = decorated[0]
if err := s.cache.Set(ctx, key, item, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("item cache write failed", "error", err)
}
return item, nil
item, _, err := s.cachedRead(ctx, itemDetailKey(sess.EmbyUserID, itemID), s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
raw, err := s.emby.Item(
timing.WithLabel(ctx, "emby.item"), credentials(sess), itemID, fieldsDetail)
if err != nil {
return nil, err
}
// A detail page can then draw its ratings with the rest of the hero rather
// than after a second request. Anything not yet stored still arrives on
// /ratings.
decorated := []json.RawMessage{raw}
s.decorateItemRatings(ctx, decorated)
return decorated[0], nil
})
return item, err
}
// handleSeasonFinale verifies an episode against Sonarr's complete season, including
@@ -109,92 +109,113 @@ func (s *Server) handleSeasonFinale(w http.ResponseWriter, r *http.Request, sess
writeJSON(w, http.StatusOK, empty)
return
}
key := cache.UserKey(sess.EmbyUserID, "season-finale:v1:"+itemID)
if raw, err := s.cache.Get(ctx, key); err == nil {
writeRaw(w, http.StatusOK, raw)
return
}
currentRaw, err := s.emby.Item(
ctx, credentials(sess), itemID,
"ProviderIds,SeriesId,SeriesName,ParentIndexNumber,IndexNumber",
)
// Cached for the household rather than per viewer, and outside the namespace a
// playback stop invalidates. Whether an episode closes its season is a fact about
// the season; it carries no user data, and it was previously being thrown away by
// the very event that most often precedes somebody asking for it — the stop report
// at the end of the episode before.
key := cache.MetadataKey("season-finale:v2:" + itemID)
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
return json.Marshal(s.seasonFinale(ctx, sess, itemID))
})
if err != nil {
s.writeUpstreamError(ctx, w, err, "could not inspect the playing episode")
return
}
writeCached(w, hit, body)
}
// seasonFinale is four upstream reads and only three of them are a chain.
//
// The episode has to be read before its series can be, and the series before Sonarr can
// be searched for it — but Sonarr's series catalogue depends on none of that, and used
// to be fetched fourth, in sequence, after both Emby lookups had returned. It is now
// fetched beside them, which takes a leg off the wait for the ordinary case and all of
// it for a household whose catalogue is still warm.
//
// Every failure below answers "not a finale" rather than an error. This decorates the
// end of an episode; being unable to say is the same outcome as saying no, and a viewer
// must never be shown a failure for it.
func (s *Server) seasonFinale(
ctx context.Context, sess store.Session, itemID string,
) seasonFinaleResponse {
empty := seasonFinaleResponse{}
var (
catalogue []sonarr.Series
catalogueErr error
wg sync.WaitGroup
)
wg.Add(1)
go func() {
defer wg.Done()
catalogue, catalogueErr = s.sonarrSeriesCatalogue(timing.WithLabel(ctx, "sonarr.series"))
}()
// Whatever happens below, the catalogue read has to be waited for: it is running on
// this request's context and returning without it would leave a goroutine writing to
// variables the caller has moved on from.
defer wg.Wait()
currentRaw, err := s.emby.Item(
timing.WithLabel(ctx, "emby.episode"), credentials(sess), itemID,
"ProviderIds,SeriesId,SeriesName,ParentIndexNumber,IndexNumber",
)
if err != nil {
s.loggerFor(ctx).Warn("season finale episode unavailable", "item", itemID, "error", err)
return empty
}
var current finaleEmbyItem
if json.Unmarshal(currentRaw, &current) != nil ||
!strings.EqualFold(current.Type, "Episode") || current.SeriesID == "" ||
current.ParentIndexNumber <= 0 || current.IndexNumber <= 0 {
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
seriesRaw, err := s.emby.Item(ctx, credentials(sess), current.SeriesID, "ProviderIds")
seriesRaw, err := s.emby.Item(
timing.WithLabel(ctx, "emby.series"), credentials(sess), current.SeriesID, "ProviderIds")
if err != nil {
s.loggerFor(ctx).Warn("season finale series metadata unavailable", "item", itemID, "error", err)
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
var seriesItem finaleEmbyItem
if json.Unmarshal(seriesRaw, &seriesItem) != nil {
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
tvdbID, err := strconv.Atoi(providerID(seriesItem.ProviderIDs, "tvdb"))
if err != nil || tvdbID <= 0 {
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
series, err := s.sonarrSeriesCatalogue(ctx)
if err != nil {
s.loggerFor(ctx).Warn("season finale Sonarr series unavailable", "item", itemID, "error", err)
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
wg.Wait()
if catalogueErr != nil {
s.loggerFor(ctx).Warn("season finale Sonarr series unavailable", "item", itemID, "error", catalogueErr)
return empty
}
sonarrSeriesID := 0
for _, candidate := range series {
for _, candidate := range catalogue {
if candidate.TVDBID == tvdbID {
sonarrSeriesID = candidate.ID
break
}
}
if sonarrSeriesID == 0 {
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
episodes, err := s.sonarr.Episodes(ctx, sonarrSeriesID)
episodes, err := s.sonarr.Episodes(timing.WithLabel(ctx, "sonarr.episodes"), sonarrSeriesID)
if err != nil {
s.loggerFor(ctx).Warn("season finale Sonarr episodes unavailable", "item", itemID, "error", err)
s.writeSeasonFinaleResponse(ctx, key, empty, w)
return
return empty
}
result := seasonFinaleResponse{
SeasonFinale: isSeasonFinale(current.ParentIndexNumber, current.IndexNumber, episodes),
if !isSeasonFinale(current.ParentIndexNumber, current.IndexNumber, episodes) {
return empty
}
return seasonFinaleResponse{
SeasonFinale: true,
SeriesName: current.SeriesName,
SeasonNumber: current.ParentIndexNumber,
EpisodeNumber: current.IndexNumber,
}
if !result.SeasonFinale {
result = empty
}
s.writeSeasonFinaleResponse(ctx, key, result, w)
}
func (s *Server) writeSeasonFinaleResponse(
ctx context.Context, key string, result seasonFinaleResponse, w http.ResponseWriter,
) {
body, err := json.Marshal(result)
if err != nil {
writeError(w, http.StatusInternalServerError, "could not encode finale status")
return
}
if err := s.cache.Set(ctx, key, body, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("season finale cache write failed", "error", err)
}
writeRaw(w, http.StatusOK, body)
}
func providerID(ids map[string]string, wanted string) string {
@@ -228,42 +249,42 @@ func (s *Server) handleSeriesEpisodes(w http.ResponseWriter, r *http.Request, se
writeError(w, http.StatusBadRequest, "series id is required")
return
}
// Per viewer, because the browser marks what has been watched — and coalesced,
// because this is the largest request the television makes and it is made twice for
// the same show as a matter of course: the launcher warms it while a Continue
// Watching card is focused, and the page asks again when somebody presses. A
// long-running show is a thousand records, so two of them is a real cost on the one
// press that must feel free.
key := cache.UserKey(sess.EmbyUserID, "series-episodes:"+seriesID)
if raw, err := s.cache.Get(ctx, key); err == nil {
w.Header().Set("X-Memby-Cache", "hit")
writeRaw(w, http.StatusOK, raw)
return
}
result, err := s.emby.Episodes(ctx, credentials(sess), seriesID, url.Values{
"UserId": {sess.EmbyUserID},
// PremiereDate is the one fact that tells two episodes of a list apart, so the
// episode browser asks for it by name — Emby does not return it otherwise.
"Fields": {"Overview,RunTimeTicks,SeriesName,PremiereDate,PrimaryImageAspectRatio"},
"EnableUserData": {"true"},
"EnableImages": {"true"},
"EnableImageTypes": {"Primary,Thumb,Backdrop"},
"ImageTypeLimit": {"1"},
"Limit": {"1000"},
})
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
result, err := s.emby.Episodes(
timing.WithLabel(ctx, "emby.episodes"), credentials(sess), seriesID, url.Values{
"UserId": {sess.EmbyUserID},
// PremiereDate is the one fact that tells two episodes of a list apart,
// so the episode browser asks for it by name — Emby does not return it
// otherwise.
"Fields": {"Overview,RunTimeTicks,SeriesName,PremiereDate,PrimaryImageAspectRatio"},
"EnableUserData": {"true"},
"EnableImages": {"true"},
"EnableImageTypes": {"Primary,Thumb,Backdrop"},
"ImageTypeLimit": {"1"},
"Limit": {"1000"},
})
if err != nil {
return nil, err
}
items := result.Items
if items == nil {
items = []json.RawMessage{}
}
return json.Marshal(seriesEpisodesResponse{Items: items})
})
if err != nil {
s.writeUpstreamError(ctx, w, err, "could not load series episodes")
return
}
items := result.Items
if items == nil {
items = []json.RawMessage{}
}
body, err := json.Marshal(seriesEpisodesResponse{Items: items})
if err != nil {
writeError(w, http.StatusInternalServerError, "could not encode series episodes")
return
}
if err := s.cache.Set(ctx, key, body, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("series episodes cache write failed", "error", err)
}
w.Header().Set("X-Memby-Cache", "miss")
writeRaw(w, http.StatusOK, body)
writeCached(w, hit, body)
}
// handleTrailer answers with the item's first local trailer, or 404 when it has none.
+44 -49
View File
@@ -1,6 +1,7 @@
package api
import (
"context"
"encoding/json"
"net/http"
"net/url"
@@ -15,6 +16,12 @@ type personFilmographyResponse struct {
// handlePerson returns Emby's person item. PremiereDate and EndDate are the person's
// birth and death dates; Overview is their biography.
//
// It is cached for the household rather than per viewer, and outside the namespace a
// playback stop invalidates. Nothing about who Cillian Murphy is depends on who is
// asking or on what they finished watching ten minutes ago, and filing it under the
// viewer meant a cast page paid Emby again after every episode anybody in the house
// completed.
func (s *Server) handlePerson(w http.ResponseWriter, r *http.Request, sess store.Session) {
ctx := r.Context()
personID := r.PathValue("id")
@@ -22,31 +29,26 @@ func (s *Server) handlePerson(w http.ResponseWriter, r *http.Request, sess store
writeError(w, http.StatusBadRequest, "person id is required")
return
}
key := cache.UserKey(sess.EmbyUserID, "person:v1:"+personID)
if raw, err := s.cache.Get(ctx, key); err == nil {
w.Header().Set("X-Memby-Cache", "hit")
writeRaw(w, http.StatusOK, raw)
return
}
person, err := s.emby.Item(
ctx,
credentials(sess),
personID,
"Overview,Genres,PrimaryImageAspectRatio",
)
person, hit, err := s.cachedRead(ctx, cache.MetadataKey("person:v2:"+personID), s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
return s.emby.Item(ctx, credentials(sess), personID,
"Overview,Genres,PrimaryImageAspectRatio")
})
if err != nil {
s.writeUpstreamError(ctx, w, err, "could not load the person")
return
}
if err := s.cache.Set(ctx, key, person, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("person cache write failed", "error", err)
}
w.Header().Set("X-Memby-Cache", "miss")
writeRaw(w, http.StatusOK, person)
writeCached(w, hit, person)
}
// Filmography is kept separate from the biography so life dates can decorate the cast
// row without also loading every cast member's credits.
//
// Unlike the biography this one is genuinely per viewer — it asks Emby for user data, so
// the grid can mark what has been watched — and therefore stays where invalidation can
// reach it. What it gains instead is coalescing: this is the most expensive query the
// gateway makes of Emby, a recursive scan of the whole library filtered by person, and
// three televisions opening the same cast card used to mean three of them.
func (s *Server) handlePersonFilmography(w http.ResponseWriter, r *http.Request, sess store.Session) {
ctx := r.Context()
personID := r.PathValue("id")
@@ -55,40 +57,33 @@ func (s *Server) handlePersonFilmography(w http.ResponseWriter, r *http.Request,
return
}
key := cache.UserKey(sess.EmbyUserID, "person-filmography:v1:"+personID)
if raw, err := s.cache.Get(ctx, key); err == nil {
w.Header().Set("X-Memby-Cache", "hit")
writeRaw(w, http.StatusOK, raw)
return
}
result, err := s.emby.Items(ctx, credentials(sess), url.Values{
"PersonIds": {personID},
"IncludeItemTypes": {"Movie,Series"},
"Recursive": {"true"},
"SortBy": {"ProductionYear,SortName"},
"SortOrder": {"Descending"},
"Limit": {"60"},
"Fields": {"Overview,ProductionYear,PrimaryImageAspectRatio"},
"EnableImages": {"true"},
"EnableImageTypes": {"Primary"},
"ImageTypeLimit": {"1"},
"EnableUserData": {"true"},
})
body, hit, err := s.cachedRead(ctx, key, s.cfg.ItemTTL,
func(ctx context.Context) (json.RawMessage, error) {
result, err := s.emby.Items(ctx, credentials(sess), url.Values{
"PersonIds": {personID},
"IncludeItemTypes": {"Movie,Series"},
"Recursive": {"true"},
"SortBy": {"ProductionYear,SortName"},
"SortOrder": {"Descending"},
"Limit": {"60"},
"Fields": {"Overview,ProductionYear,PrimaryImageAspectRatio"},
"EnableImages": {"true"},
"EnableImageTypes": {"Primary"},
"ImageTypeLimit": {"1"},
"EnableUserData": {"true"},
})
if err != nil {
return nil, err
}
items := result.Items
if items == nil {
items = []json.RawMessage{}
}
return json.Marshal(personFilmographyResponse{Items: items})
})
if err != nil {
s.writeUpstreamError(ctx, w, err, "could not load the person's filmography")
return
}
items := result.Items
if items == nil {
items = []json.RawMessage{}
}
body, err := json.Marshal(personFilmographyResponse{Items: items})
if err != nil {
writeError(w, http.StatusInternalServerError, "could not encode the filmography")
return
}
if err := s.cache.Set(ctx, key, body, s.cfg.ItemTTL); err != nil {
s.loggerFor(ctx).Warn("person filmography cache write failed", "error", err)
}
w.Header().Set("X-Memby-Cache", "miss")
writeRaw(w, http.StatusOK, body)
writeCached(w, hit, body)
}
+12 -1
View File
@@ -14,6 +14,7 @@ import (
serverlogging "github.com/ponzischeme89/memby/server/internal/logging"
"github.com/ponzischeme89/memby/server/internal/notify"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
const ticksPerMillisecond = 10_000
@@ -826,7 +827,8 @@ func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, se
)
err := s.emby.ReportPlayback(
r.Context(), credentials(sess), phase, report.ItemID, report.MediaSourceID,
timing.WithLabel(r.Context(), "emby.report"),
credentials(sess), phase, report.ItemID, report.MediaSourceID,
report.PlaySessionID, report.PlayMethod, report.EventName,
max64(report.PositionMs, 0)*ticksPerMillisecond, report.IsPaused,
)
@@ -873,18 +875,27 @@ func (s *Server) handlePlaybackReport(w http.ResponseWriter, r *http.Request, se
playbackSessionKey(sess.DeviceID, report.PlaySessionID))
if phase == "stopped" {
invalidate := timing.Start(r.Context(), "invalidate")
if err := s.cache.InvalidateUser(r.Context(), sess.EmbyUserID); err != nil {
s.loggerFor(r.Context()).Warn("cache invalidation failed", "error", err)
}
invalidate()
// Recommendation taste changes slowly. Tracearr marks this user's prepared
// profile dirty only when the session first becomes terminal; the daily builder
// then refreshes it without turning every player exit into catalogue-wide work.
}
response := playbackReportResponse{}
// The order of these three is the whole of what keeps a progress report cheap. The
// claim is a map lookup and comes first, so every report after the first for this
// episode stops here rather than at the durable insert four upstream calls later;
// the feature check is a cached read; and only then is anything asked of Emby.
if shouldAutoFollowShow(phase, report.PositionMs, report.DurationMs) &&
s.followChecks.claim(sess.EmbyUserID, report.ItemID) &&
s.featureEnabled(r.Context(), featureAutomaticMyShows) {
follow := timing.Start(r.Context(), "autofollow")
response.AutoFollowedShowTitle = s.autoFollowContinuingShow(r.Context(), sess, report.ItemID)
follow()
}
writeJSON(w, http.StatusOK, response)
}
+54
View File
@@ -0,0 +1,54 @@
package api
import "sync"
// consideredFollows bounds the memory below. A household has a handful of episodes in
// flight; this is generous enough that a whole evening's viewing fits and small enough
// that nothing a client can do grows it into a leak.
const consideredFollows = 512
// followChecks remembers which (viewer, episode) pairs have already been put through
// the automatic My Shows check.
//
// The check is hung off playback reports, and a report arrives every ten seconds for the
// length of an episode. Its own guard — half watched — is true for every one of those
// after the halfway point, so on a forty-minute episode the second half of the viewing
// ran the whole check a hundred times: two serial Emby lookups, the Sonarr catalogue and
// a Postgres write, all of it on the response path of a request the television makes
// while somebody is watching something. Every run after the first reached the same
// `inserted == false` and discarded its own work.
//
// Deduplicating on the durable insert is correct and was never the problem; what it
// could not do is prevent the work leading up to it. This is deliberately in memory and
// deliberately lossy, the playbackTitles arrangement: a restarted gateway does the check
// once more per episode, which is the cost of one lookup rather than a schema change.
type followChecks struct {
mu sync.Mutex
seen map[string]struct{}
order []string
}
// claim reports whether this is the first time the pair has been offered, and records it.
// One call does both because two would be a race between the check and the record, and
// the thing being protected is precisely a burst of concurrent reports.
func (f *followChecks) claim(userID, itemID string) bool {
if userID == "" || itemID == "" {
return false
}
key := userID + "|" + itemID
f.mu.Lock()
defer f.mu.Unlock()
if f.seen == nil {
f.seen = make(map[string]struct{}, consideredFollows)
}
if _, known := f.seen[key]; known {
return false
}
f.seen[key] = struct{}{}
f.order = append(f.order, key)
if len(f.order) > consideredFollows {
delete(f.seen, f.order[0])
f.order = f.order[1:]
}
return true
}
@@ -0,0 +1,87 @@
package api
import (
"fmt"
"sync"
"testing"
)
// The whole point of the claim is that it is true exactly once. A progress report arrives
// every ten seconds for the length of an episode and its own guard — half watched — is
// true for every one of those after the halfway mark, so without this the check behind it
// ran two Emby lookups, a Sonarr catalogue read and a Postgres write a hundred times per
// episode and discarded all but the first.
func TestFollowCheckIsClaimedOnce(t *testing.T) {
var checks followChecks
if !checks.claim("user-1", "episode-1") {
t.Fatal("the first report should claim the check")
}
for range 10 {
if checks.claim("user-1", "episode-1") {
t.Fatal("a repeat report must not claim the check again")
}
}
}
// Two people watching the same episode on two televisions are two viewers, and each of
// them has their own My Shows list.
func TestFollowCheckIsPerViewerAndPerEpisode(t *testing.T) {
var checks followChecks
checks.claim("user-1", "episode-1")
if !checks.claim("user-2", "episode-1") {
t.Fatal("another viewer must get their own check")
}
if !checks.claim("user-1", "episode-2") {
t.Fatal("the next episode must get its own check")
}
}
// An empty id is not a claim to record. The alternative is one entry that swallows the
// first real check for whichever viewer or episode arrives unnamed.
func TestFollowCheckRefusesEmptyIdentifiers(t *testing.T) {
var checks followChecks
if checks.claim("", "episode-1") || checks.claim("user-1", "") {
t.Fatal("an unnamed viewer or episode is not a check to claim")
}
}
// The memory is bounded, and what falls out of it falls out oldest first. A gateway that
// has been up for a month must not be holding every episode the house has ever watched.
func TestFollowCheckForgetsTheOldestFirst(t *testing.T) {
var checks followChecks
for index := range consideredFollows + 1 {
checks.claim("user-1", fmt.Sprintf("episode-%d", index))
}
if !checks.claim("user-1", "episode-0") {
t.Fatal("the oldest entry should have been evicted")
}
if checks.claim("user-1", fmt.Sprintf("episode-%d", consideredFollows)) {
t.Fatal("the newest entry should still be held")
}
}
// Checking and recording are one call because two would be a race, and the thing being
// protected is precisely a burst of concurrent reports for one playback.
func TestFollowCheckClaimsOnceUnderConcurrency(t *testing.T) {
var (
checks followChecks
claimed int
mu sync.Mutex
wg sync.WaitGroup
)
for range 32 {
wg.Add(1)
go func() {
defer wg.Done()
if checks.claim("user-1", "episode-1") {
mu.Lock()
claimed++
mu.Unlock()
}
}()
}
wg.Wait()
if claimed != 1 {
t.Fatalf("claimed %d times, want exactly 1", claimed)
}
}
+232 -59
View File
@@ -7,6 +7,7 @@ import (
"net/url"
"sort"
"strings"
"sync"
"time"
"github.com/ponzischeme89/memby/server/internal/recommend"
@@ -40,32 +41,60 @@ func (s *Server) filterRecommendationPermissions(
if len(ids) == 0 {
return rows
}
// The batches are independent questions about disjoint sets of ids, and there can be
// several of them on a launcher full of recommendations. Asking Emby them one after
// another put the whole of that latency on the tail of the home response; asking
// together costs Emby the same work and the viewer one batch's wait.
//
// There is deliberately no cap on how many run at once. The count is a hundredth of
// the recommendation candidate pool, which is itself bounded, and the Emby client's
// transport already keeps a warm pool of connections for them to share.
allowed := map[string]bool{}
cred := credentials(sess)
var (
mu sync.Mutex
failed bool
wg sync.WaitGroup
)
for start := 0; start < len(ids); start += 100 {
end := min(start+100, len(ids))
result, err := s.emby.Items(ctx, cred, rowParams(url.Values{
"Ids": {strings.Join(ids[start:end], ",")},
"Recursive": {"true"},
"IncludeItemTypes": {"Movie,Series"},
"Limit": {itoa(end - start)},
}, fieldsRow))
if err != nil {
s.log.Warn("recommendation permission check failed; hiding candidates",
"user", sess.EmbyUserID, "error", err)
for index := range rows {
if recommendationRow(rows[index].ID) {
rows[index].Items = []json.RawMessage{}
batch := ids[start:end]
wg.Add(1)
go func() {
defer wg.Done()
result, err := s.emby.Items(ctx, cred, rowParams(url.Values{
"Ids": {strings.Join(batch, ",")},
"Recursive": {"true"},
"IncludeItemTypes": {"Movie,Series"},
"Limit": {itoa(len(batch))},
}, fieldsRow))
mu.Lock()
defer mu.Unlock()
if err != nil {
s.loggerFor(ctx).Warn("recommendation permission check failed; hiding candidates",
"user", sess.EmbyUserID, "error", err)
failed = true
return
}
for _, raw := range result.Items {
decoded := recommend.Decode([]json.RawMessage{raw})
if len(decoded) == 1 {
allowed[decoded[0].ID] = true
}
}
return rows
}
for _, raw := range result.Items {
decoded := recommend.Decode([]json.RawMessage{raw})
if len(decoded) == 1 {
allowed[decoded[0].ID] = true
}()
}
wg.Wait()
// One failed batch hides every candidate, exactly as it did when the loop returned
// early: this is Emby being the final authority on what a viewer may see, and a
// partial answer is not evidence that the rest is permitted.
if failed {
for index := range rows {
if recommendationRow(rows[index].ID) {
rows[index].Items = []json.RawMessage{}
}
}
return rows
}
for index := range rows {
if !recommendationRow(rows[index].ID) {
@@ -82,6 +111,22 @@ func (s *Server) filterRecommendationPermissions(
return rows
}
// rankingContext gathers everything the weighted ranker needs about one viewer: their
// learned profile, what they have already been shown, and how the household as a whole
// has been getting on with the library.
//
// It is seven Postgres reads and it used to make all seven one after another, on the tail
// of the home response — after the Emby fan-out had finished, so nothing else was in
// flight and every one of them was pure added latency. Six of the seven depend on nothing
// but the viewer's id. They are now issued together, which turns the sum of seven round
// trips into roughly the slowest one.
//
// The two that are genuinely a chain stay a chain — an onboarding document names the
// items its ratings refer to, and the actions name theirs — but the two chains run beside
// each other, and their results are applied to the profile afterwards in the order they
// were applied before. Ordering the *writes* rather than the reads is the whole trick:
// ApplyOnboarding and ApplyExplicitPreference are not commutative and this must not
// become a place where the ranking depends on which query answered first.
func (s *Server) rankingContext(
ctx context.Context,
userID string,
@@ -90,10 +135,90 @@ func (s *Server) rankingContext(
if s.store == nil {
return profile, nil, nil
}
if raw, err := s.store.WeightedRecommendationProfile(ctx, userID); err == nil {
_ = json.Unmarshal(raw, &profile)
} else {
s.log.Warn("weighted profile unavailable", "user", userID, "error", err)
var (
wg sync.WaitGroup
profileRaw json.RawMessage
onboarding recommend.OnboardingPreferences
onboardingOK bool
onboardingItems []json.RawMessage
actions []store.RecommendationAction
actionItems []json.RawMessage
exposureValues []store.ItemExposureStat
household map[string]float64
)
run := func(fn func()) {
wg.Add(1)
go func() {
defer wg.Done()
fn()
}()
}
run(func() {
raw, err := s.store.WeightedRecommendationProfile(ctx, userID)
if err != nil {
s.loggerFor(ctx).Warn("weighted profile unavailable", "user", userID, "error", err)
return
}
profileRaw = raw
})
run(func() {
raw, err := s.store.RecommendationOnboarding(ctx, userID)
if err != nil {
return
}
var preferences recommend.OnboardingPreferences
if json.Unmarshal(raw, &preferences) != nil {
return
}
onboarding, onboardingOK = preferences, true
ids := make([]string, 0, len(preferences.Ratings))
for id, rating := range preferences.Ratings {
if strings.TrimSpace(id) != "" && rating >= 1 && rating <= 5 {
ids = append(ids, id)
}
}
if len(ids) == 0 {
return
}
if raws, err := s.store.LibraryItemsByID(ctx, ids); err == nil {
onboardingItems = raws
}
})
run(func() {
values, err := s.store.RecommendationActions(ctx, userID)
if err != nil {
return
}
actions = values
ids := make([]string, 0, len(values))
for _, action := range values {
ids = append(ids, action.ItemID)
}
if len(ids) == 0 {
return
}
if raws, err := s.store.LibraryItemsByID(ctx, ids); err == nil {
actionItems = raws
}
})
run(func() {
values, err := s.store.UserItemExposures(ctx, userID, time.Now().Add(-45*24*time.Hour))
if err == nil {
exposureValues = values
}
})
run(func() { household = s.householdCompletionScores(ctx) })
wg.Wait()
if len(profileRaw) > 0 {
_ = json.Unmarshal(profileRaw, &profile)
}
if profile.ExplicitPositive == nil {
profile.ExplicitPositive = map[string]bool{}
@@ -101,31 +226,16 @@ func (s *Server) rankingContext(
if profile.ExplicitNegative == nil {
profile.ExplicitNegative = map[string]bool{}
}
if raw, err := s.store.RecommendationOnboarding(ctx, userID); err == nil {
var preferences recommend.OnboardingPreferences
if json.Unmarshal(raw, &preferences) == nil {
profile.ApplyOnboarding(preferences, s.weightedConfig().MinimumEvidence)
ids := make([]string, 0, len(preferences.Ratings))
for id, rating := range preferences.Ratings {
if strings.TrimSpace(id) != "" && rating >= 1 && rating <= 5 {
ids = append(ids, id)
}
}
if raws, itemErr := s.store.LibraryItemsByID(ctx, ids); itemErr == nil {
for _, item := range recommend.Decode(raws) {
profile.ApplyOnboardingRating(
item, preferences.Ratings[item.ID],
s.weightedConfig().MinimumEvidence,
)
}
}
if onboardingOK {
evidence := s.weightedConfig().MinimumEvidence
profile.ApplyOnboarding(onboarding, evidence)
for _, item := range recommend.Decode(onboardingItems) {
profile.ApplyOnboardingRating(item, onboarding.Ratings[item.ID], evidence)
}
}
if actions, err := s.store.RecommendationActions(ctx, userID); err == nil {
ids := make([]string, 0, len(actions))
if len(actions) > 0 {
byID := make(map[string]string, len(actions))
for _, action := range actions {
ids = append(ids, action.ItemID)
byID[action.ItemID] = action.Action
switch action.Action {
case "more_like_this":
@@ -134,30 +244,82 @@ func (s *Server) rankingContext(
profile.ExplicitNegative[action.ItemID] = true
}
}
if raws, itemErr := s.store.LibraryItemsByID(ctx, ids); itemErr == nil {
for _, item := range recommend.Decode(raws) {
profile.ApplyExplicitPreference(item, byID[item.ID] == "more_like_this")
}
for _, item := range recommend.Decode(actionItems) {
profile.ApplyExplicitPreference(item, byID[item.ID] == "more_like_this")
}
}
exposures := map[string]recommend.ItemExposure{}
if values, err := s.store.UserItemExposures(
ctx, userID, time.Now().Add(-45*24*time.Hour),
); err == nil {
for _, value := range values {
exposures[value.ItemID] = recommend.ItemExposure{
Impressions: value.Impressions, Focuses: value.Focuses,
Selects: value.Selects, LastShown: value.LastShown,
}
exposures := make(map[string]recommend.ItemExposure, len(exposureValues))
for _, value := range exposureValues {
exposures[value.ItemID] = recommend.ItemExposure{
Impressions: value.Impressions, Focuses: value.Focuses,
Selects: value.Selects, LastShown: value.LastShown,
}
}
household, err := s.store.HouseholdCompletionScores(ctx, time.Now().Add(-180*24*time.Hour))
if err != nil {
if household == nil {
household = map[string]float64{}
}
return profile, exposures, household
}
// householdCompletionScoreTTL is how long one reading of the household's completion
// scores is reused.
//
// The query aggregates six months of viewing across everybody, it is the heaviest read
// in rankingContext, and its answer is *the same for every viewer in the house* — so
// four televisions refreshing their launchers were running four copies of one
// six-month aggregate, none of which could have produced a different number. A minute
// is chosen against what it measures: this moves when somebody finishes something, and
// nothing on a launcher is different for having learned that a minute sooner.
const householdCompletionScoreTTL = time.Minute
// householdScores caches that reading. Coalesced as well as cached, so the request that
// finds it expired is the only one that pays for the refresh rather than the first of
// several that all do.
type householdScores struct {
mu sync.Mutex
value map[string]float64
fetched time.Time
}
func (s *Server) householdCompletionScores(ctx context.Context) map[string]float64 {
s.household.mu.Lock()
defer s.household.mu.Unlock()
if s.household.value != nil && time.Since(s.household.fetched) < householdCompletionScoreTTL {
return s.household.value
}
scores, err := s.store.HouseholdCompletionScores(ctx, time.Now().Add(-180*24*time.Hour))
if err != nil {
// A failed read must not be cached as an empty household: that would suppress
// the signal for a whole minute on the strength of one timeout. The previous
// reading is the better answer where there is one.
if s.household.value != nil {
return s.household.value
}
return map[string]float64{}
}
s.household.value = scores
s.household.fetched = time.Now()
return scores
}
// rankingInputs is one viewer's ranking evidence, gathered.
//
// It exists so the gathering can happen somewhere other than immediately before the
// ranking. On the home path the evidence depends on nothing but the viewer's id, while
// the rows it will be applied to take seconds to arrive from Emby — so it is read
// beside them rather than after them, and by the time there is anything to rank it is
// already in hand. See handleHome.
type rankingInputs struct {
profile recommend.WeightedProfile
exposures map[string]recommend.ItemExposure
household map[string]float64
}
func (s *Server) rankingInputs(ctx context.Context, userID string) rankingInputs {
profile, exposures, household := s.rankingContext(ctx, userID)
return rankingInputs{profile: profile, exposures: exposures, household: household}
}
func (s *Server) personalizeTitles(
ctx context.Context,
sess store.Session,
@@ -166,7 +328,18 @@ func (s *Server) personalizeTitles(
if len(rows) == 0 {
return rows
}
profile, exposures, household := s.rankingContext(ctx, sess.EmbyUserID)
return s.personalizeTitlesWith(rows, s.rankingInputs(ctx, sess.EmbyUserID))
}
// personalizeTitlesWith is the ranking itself, over evidence somebody else gathered.
func (s *Server) personalizeTitlesWith(
rows []recommend.Row,
inputs rankingInputs,
) []recommend.Row {
if len(rows) == 0 {
return rows
}
profile, exposures, household := inputs.profile, inputs.exposures, inputs.household
cfg := s.weightedConfig()
now := time.Now()
location := s.householdLocation()
+32 -22
View File
@@ -1,6 +1,7 @@
package api
import (
"context"
"encoding/json"
"errors"
"net/http"
@@ -10,6 +11,7 @@ import (
"github.com/ponzischeme89/memby/server/internal/emby"
"github.com/ponzischeme89/memby/server/internal/recommend"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// fieldsRelated is the detail set. It once added Studios on its own account — the
@@ -50,7 +52,24 @@ func (s *Server) handleRelated(w http.ResponseWriter, r *http.Request, sess stor
return
}
reasons, related, err := s.recommender.RelatedTo(ctx, credentials(sess), item, relatedRowSize)
// Coalesced, because this route is asked for twice for the same title as a matter of
// course: the launcher warms it while a card is focused and the detail page asks for
// it again the moment somebody presses. Those two requests used to build the answer
// twice, and building it is the same Emby fan-out the home rows pay for.
body, err := s.buildCached(ctx, key, relatedTTL(s.cfg.ItemTTL),
func(ctx context.Context) (json.RawMessage, error) {
reasons, related, err := s.recommender.RelatedTo(
timing.WithLabel(ctx, "emby.related"), credentials(sess), item, relatedRowSize)
if err != nil {
return nil, err
}
items := nonNilRaws(recommend.Raws(related))
s.decorateItemRatings(ctx, items)
return json.Marshal(relatedResponse{
Reasons: nonNilStrings(reasons),
Items: items,
})
})
if err != nil {
// RelatedTo degrades rather than failing, so the only error it returns is the
// viewer having navigated on. That is the television saving work, not a fault.
@@ -61,29 +80,20 @@ func (s *Server) handleRelated(w http.ResponseWriter, r *http.Request, sess stor
s.writeUpstreamError(ctx, w, err, "could not load related titles")
return
}
writeCached(w, false, body)
}
items := nonNilRaws(recommend.Raws(related))
s.decorateItemRatings(ctx, items)
body, err := json.Marshal(relatedResponse{
Reasons: nonNilStrings(reasons),
Items: items,
})
if err != nil {
writeError(w, http.StatusInternalServerError, "could not encode related titles")
return
// relatedTTL keeps an empty carousel briefly rather than for the item lifetime: it
// usually means something upstream was unwell, and the ten-minute answer would otherwise
// outlive the minute of trouble that produced it.
func relatedTTL(itemTTL time.Duration) func(json.RawMessage) time.Duration {
return func(body json.RawMessage) time.Duration {
var decoded relatedResponse
if json.Unmarshal(body, &decoded) == nil && len(decoded.Items) == 0 {
return relatedEmptyTTL
}
return itemTTL
}
// An empty carousel is cached briefly rather than for the item lifetime: it usually
// means something upstream was unwell, and the ten-minute answer would otherwise
// outlive the minute of trouble that produced it.
ttl := s.cfg.ItemTTL
if len(items) == 0 {
ttl = relatedEmptyTTL
}
if err := s.cache.Set(ctx, key, body, ttl); err != nil {
s.loggerFor(ctx).Warn("related cache write failed", "error", err)
}
w.Header().Set("X-Memby-Cache", "miss")
writeRaw(w, http.StatusOK, body)
}
// relatedRowSize is a carousel's worth. The strip scrolls, but a viewer who reaches the
+4 -2
View File
@@ -26,6 +26,8 @@ import (
"strconv"
"strings"
"time"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Client struct {
@@ -168,14 +170,14 @@ func New(baseURL, apiKey string, timeout time.Duration) *Client {
return &Client{
baseURL: strings.TrimRight(baseURL, "/"),
apiKey: apiKey,
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 10,
MaxIdleConnsPerHost: 5,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageBazarr),
}
}
+1 -1
View File
@@ -1 +1 @@
0.1.59
0.1.60
+92 -4
View File
@@ -9,9 +9,12 @@ import (
"context"
"errors"
"fmt"
"strings"
"time"
"github.com/redis/go-redis/v9"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// ErrMiss means the key was absent — an ordinary outcome, not a failure.
@@ -33,37 +36,106 @@ func (c *Cache) Close() error { return c.rdb.Close() }
func (c *Cache) Ping(ctx context.Context) error { return c.rdb.Ping(ctx).Err() }
// Get reports its own outcome to the request's trace. Whether an answer came from
// Redis is the first thing anybody asks of a slow screen, and it is the one fact the
// duration alone can never carry: a 9-second home request that missed and a 9-second
// home request that hit are two entirely different investigations.
func (c *Cache) Get(ctx context.Context, key string) ([]byte, error) {
defer timing.Start(ctx, timing.StageRedis)()
b, err := c.rdb.Get(ctx, key).Bytes()
if errors.Is(err, redis.Nil) {
timing.Count(ctx, "miss")
return nil, ErrMiss
}
if err != nil {
timing.Count(ctx, "cache_error")
return nil, err
}
timing.Count(ctx, "hit")
return b, nil
}
// Set writes a value and, for a user-scoped key, records it in that user's index.
//
// The index is what makes invalidation proportional to one viewer's cached views rather
// than to the whole keyspace — see InvalidateUser. It is written in the same pipeline as
// the value, so it costs one round trip rather than two, and it is given a generous TTL
// of its own so a user who stops watching cannot leave a set name growing for ever.
func (c *Cache) Set(ctx context.Context, key string, value []byte, ttl time.Duration) error {
return c.rdb.Set(ctx, key, value, ttl).Err()
defer timing.Start(ctx, timing.StageRedis)()
index := userIndexFor(key)
if index == "" {
return c.rdb.Set(ctx, key, value, ttl).Err()
}
_, err := c.rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
pipe.Set(ctx, key, value, ttl)
pipe.SAdd(ctx, index, key)
pipe.Expire(ctx, index, userIndexTTL)
return nil
})
return err
}
// userIndexTTL outlives any cached view by a wide margin. It is a bound on the index's
// own lifetime, not a cache policy: the members it names may expire underneath it, which
// costs a deletion of keys that are already gone and nothing else.
const userIndexTTL = 30 * 24 * time.Hour
// userIndexFor names the set that tracks one viewer's cached views, or empty for a key
// that is not user-scoped. It parses rather than being told, so a caller cannot write a
// user key and forget to index it — which would be an invalidation that silently misses.
func userIndexFor(key string) string {
if !strings.HasPrefix(key, "u:") {
return ""
}
rest := key[len("u:"):]
end := strings.IndexByte(rest, ':')
if end <= 0 {
return ""
}
return "idx:u:" + rest[:end]
}
func (c *Cache) Delete(ctx context.Context, keys ...string) error {
if len(keys) == 0 {
return nil
}
defer timing.Start(ctx, timing.StageRedis)()
return c.rdb.Del(ctx, keys...).Err()
}
// InvalidateUser drops every cached view belonging to one Emby user.
//
// SCAN rather than KEYS so a large keyspace never blocks Redis; the key count here is
// small, but the habit costs nothing.
// It reads the viewer's own index and deletes what it names: two round trips, whatever
// the size of the keyspace. It used to SCAN for `u:<user>:*`, and SCAN's cost is a
// property of the *whole* keyspace rather than of the pattern — Redis walks every key it
// holds and filters, in pages, so a household with a few thousand cached item lookups
// paid a few dozen round trips to find the handful of rows it wanted gone. That cost
// landed on the playback stop report, which is a request the television makes as
// somebody presses Back out of a film.
//
// The SCAN survives as the fallback for a user with no index, which is the honest answer
// for two cases: a gateway upgraded while keys written by the previous build were still
// live, and the window between an invalidation and the next write. Both are rare and
// neither is any slower than the behaviour this replaced.
func (c *Cache) InvalidateUser(ctx context.Context, userID string) error {
defer timing.Start(ctx, timing.StageRedis)()
index := "idx:u:" + userID
keys, err := c.rdb.SMembers(ctx, index).Result()
if err != nil {
return err
}
if len(keys) == 0 {
return c.scanInvalidate(ctx, userID)
}
return c.rdb.Del(ctx, append(keys, index)...).Err()
}
func (c *Cache) scanInvalidate(ctx context.Context, userID string) error {
pattern := fmt.Sprintf("u:%s:*", userID)
var cursor uint64
for {
keys, next, err := c.rdb.Scan(ctx, cursor, pattern, 200).Result()
keys, next, err := c.rdb.Scan(ctx, cursor, pattern, 1000).Result()
if err != nil {
return err
}
@@ -94,5 +166,21 @@ func RecommendationsKey(userID string) string { return fmt.Sprintf("r:%s:rows:v3
// press would pay the full rebuild.
func MagicPoolKey(userID string) string { return fmt.Sprintf("m:%s:pool:v1", userID) }
// MetadataKey is for catalogue facts that belong to no one: a person's biography, the
// list of extras on a film, whether an episode ends a season.
//
// It sits outside the `u:` namespace for the reason RecommendationsKey does, and one
// more besides. Every playback stop and every favourite calls InvalidateUser, which
// drops `u:<user>:*` wholesale — so a person's biography, which cannot change when
// somebody finishes an episode, was being thrown away several times an evening and
// re-read from Emby by the next cast card anybody looked at. It is also unkeyed by
// viewer, which is the larger win in a household: one lookup answers for everybody
// rather than one per person.
//
// The rule for putting something here is narrow and worth stating: the value must
// contain nothing derived from a viewer. Anything carrying UserData — watched marks,
// resume positions, favourites — belongs in UserKey, where invalidation can reach it.
func MetadataKey(view string) string { return "meta:" + view }
// SessionKey caches a token→session lookup, keyed by token hash (never the token).
func SessionKey(tokenHashHex string) string { return "sess:" + tokenHashHex }
+29
View File
@@ -0,0 +1,29 @@
package cache
import "testing"
// The index name is derived from the key rather than passed in, so this is the whole of
// what stands between a user-scoped write and an invalidation that misses it.
func TestUserIndexFor(t *testing.T) {
cases := []struct {
name string
key string
want string
}{
{"user view", UserKey("abc123", "home:v4:24"), "idx:u:abc123"},
{"view containing colons", UserKey("abc123", "item:v6:9f:2"), "idx:u:abc123"},
{"recommendations are deliberately outside the namespace", RecommendationsKey("abc"), ""},
{"magic pool likewise", MagicPoolKey("abc"), ""},
{"household metadata belongs to nobody", MetadataKey("person:v2:44"), ""},
{"a session is not a view", SessionKey("deadbeef"), ""},
{"a malformed user key indexes nothing rather than guessing", "u:", ""},
{"a user key with no view is not a view either", "u:abc", ""},
}
for _, test := range cases {
t.Run(test.name, func(t *testing.T) {
if got := userIndexFor(test.key); got != test.want {
t.Fatalf("userIndexFor(%q) = %q, want %q", test.key, got, test.want)
}
})
}
}
+7
View File
@@ -82,6 +82,12 @@ type Config struct {
UpstreamTimeout time.Duration
// SlowRequestThreshold is how long a request has to take before its log line
// carries a stage breakdown. Fast requests deliberately carry none: it is the one
// field on the line that varies in width, and a column of them on every /v1/status
// poll would make the thing it exists to reveal harder to find, not easier.
SlowRequestThreshold time.Duration
// EmbyHealthInterval paces the reachability probe behind the "server not responding"
// and "back online" banners. One probe per gateway, not per TV. Zero disables it.
EmbyHealthInterval time.Duration
@@ -251,6 +257,7 @@ func Load() (Config, error) {
SyncAPIKey: strings.TrimSpace(os.Getenv("MEMBY_SYNC_API_KEY")),
AnalyticsRetention: duration("MEMBY_ANALYTICS_RETENTION", 90*24*time.Hour),
UpstreamTimeout: duration("MEMBY_UPSTREAM_TIMEOUT", 20*time.Second),
SlowRequestThreshold: duration("MEMBY_SLOW_REQUEST_THRESHOLD", 500*time.Millisecond),
EmbyHealthInterval: duration("MEMBY_EMBY_HEALTH_INTERVAL", 60*time.Second),
SonarrURL: strings.TrimRight(strings.TrimSpace(os.Getenv("MEMBY_SONARR_URL")), "/"),
SonarrAPIKey: strings.TrimSpace(os.Getenv("MEMBY_SONARR_API_KEY")),
+3 -2
View File
@@ -19,6 +19,7 @@ import (
"time"
"github.com/ponzischeme89/memby/server/internal/buildinfo"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Client struct {
@@ -174,14 +175,14 @@ func New(baseURL, publicURL, clientName, gatewayClientName string, timeout time.
clientName: clientName,
gatewayClientName: gatewayClientName,
gatewayVersion: buildinfo.Version(),
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 100,
MaxIdleConnsPerHost: 20,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageEmby),
}
}
+2 -1
View File
@@ -24,6 +24,7 @@ import (
"github.com/ponzischeme89/memby/server/internal/adminevents"
"github.com/ponzischeme89/memby/server/internal/notify"
"github.com/ponzischeme89/memby/server/internal/store"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// queueDepth bounds the work waiting to be delivered.
@@ -94,7 +95,7 @@ func New(
dispatcher := &Dispatcher{
store: st, log: log.With("component", "integrations"), events: events,
notify: notifier,
client: &http.Client{Timeout: requestTimeout},
client: timing.Instrument(&http.Client{Timeout: requestTimeout}, timing.StageIntegrations),
transports: map[string]Transport{},
queue: make(chan job, queueDepth),
}
+4 -2
View File
@@ -13,6 +13,8 @@ import (
"strconv"
"strings"
"time"
"github.com/ponzischeme89/memby/server/internal/timing"
)
const DefaultBaseURL = "https://api.mdblist.com"
@@ -50,14 +52,14 @@ func (e *APIError) Error() string {
func New(baseURL string, timeout time.Duration) *Client {
return &Client{
baseURL: strings.TrimRight(baseURL, "/"),
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 20,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageMDBList),
}
}
+4 -2
View File
@@ -36,6 +36,8 @@ import (
"strings"
"sync"
"time"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// DefaultBaseURL is the REST API's home. It is a field on the client so the tests can
@@ -136,14 +138,14 @@ func New(apiKey, userAgent, username, password string, timeout time.Duration) *C
userAgent: strings.TrimSpace(userAgent),
username: strings.TrimSpace(username),
password: password,
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 10,
MaxIdleConnsPerHost: 5,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageOpenSubtitles),
}
}
+4 -2
View File
@@ -11,6 +11,8 @@ import (
"strconv"
"strings"
"time"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Client struct {
@@ -117,14 +119,14 @@ func New(baseURL, apiKey string, timeout time.Duration) *Client {
return &Client{
baseURL: strings.TrimRight(baseURL, "/"),
apiKey: apiKey,
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 20,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageRadarr),
}
}
+4 -2
View File
@@ -11,6 +11,8 @@ import (
"strconv"
"strings"
"time"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Client struct {
@@ -159,14 +161,14 @@ func New(baseURL, apiKey string, timeout time.Duration) *Client {
return &Client{
baseURL: strings.TrimRight(baseURL, "/"),
apiKey: apiKey,
http: &http.Client{
http: timing.Instrument(&http.Client{
Timeout: timeout,
Transport: &http.Transport{
MaxIdleConns: 20,
MaxIdleConnsPerHost: 10,
IdleConnTimeout: 90 * time.Second,
},
},
}, timing.StageSonarr),
}
}
+10
View File
@@ -46,6 +46,12 @@ type GatewaySettings struct {
RadarrAlertMinutes int `json:"radarrAlertMinutes"`
// EmbyHealthSeconds is how often the reachability probe asks Emby whether it is there.
EmbyHealthSeconds int `json:"embyHealthSeconds"`
// SlowRequestMillis is how long a request must take before its log line carries a
// stage breakdown — Emby time, database time, cache outcome, gateway work. It is an
// override worth having because the useful threshold is a property of the household
// rather than of the build: an operator chasing one slow screen wants it at 100ms for
// an evening and back at 500 afterwards, and neither is a redeployment.
SlowRequestMillis int `json:"slowRequestMillis"`
// LibrarySyncMinutes is how often the catalogue sweep asks Emby what changed.
//
// It is an override worth having because the answer now depends on the household's
@@ -86,6 +92,10 @@ func normalizeGatewaySettings(settings GatewaySettings) GatewaySettings {
settings.SonarrAlertMinutes = clampOverride(settings.SonarrAlertMinutes, 1, 24*60, true)
settings.RadarrAlertMinutes = clampOverride(settings.RadarrAlertMinutes, 1, 7*24*60, true)
settings.EmbyHealthSeconds = clampOverride(settings.EmbyHealthSeconds, 10, 3600, true)
// The floor is 1ms rather than 0 because "breakdown on everything" is a real thing to
// want for a few minutes, and off is a real thing to want too — this is the one
// setting here whose switched-off state costs nothing but a column of text.
settings.SlowRequestMillis = clampOverride(settings.SlowRequestMillis, 1, 60000, true)
// A day is the ceiling rather than a week: however well the webhooks are working, the
// sweep is the only thing that ever notices a file somebody moved by hand.
settings.LibrarySyncMinutes = clampOverride(settings.LibrarySyncMinutes, 5, 24*60, true)
+32 -1
View File
@@ -294,8 +294,39 @@ type Store struct {
pool *pgxpool.Pool
}
// defaultMaxConns is the floor on the connection pool.
//
// pgxpool's own default is max(4, GOMAXPROCS), which on the NAS this runs on is four —
// and four is fewer than one household's televisions. Every one of them polls
// /v1/status every ten seconds, and that poll alone reads the feature policy, the
// notification preferences, the preference revision, the theme and the request
// allowlist. Queries that each take a millisecond then queue behind each other for
// hundreds, and the waiting is invisible: the query time is honest and the request is
// slow anyway. A deployment that has stated pool_max_conns for itself keeps its own
// answer — this replaces a default nobody chose, not a decision somebody made.
const defaultMaxConns = 16
// minConns keeps a few connections open. Establishing one costs a TCP connection, TLS
// where it is configured and Postgres' own backend fork, which is several milliseconds
// paid by whichever request happened to arrive after an idle period — most often the
// first thing somebody does after switching a television on.
const minConns = 4
func Open(ctx context.Context, databaseURL string) (*Store, error) {
pool, err := pgxpool.New(ctx, databaseURL)
config, err := pgxpool.ParseConfig(databaseURL)
if err != nil {
return nil, fmt.Errorf("store: parse database url: %w", err)
}
if !strings.Contains(databaseURL, "pool_max_conns") && config.MaxConns < defaultMaxConns {
config.MaxConns = defaultMaxConns
}
if !strings.Contains(databaseURL, "pool_min_conns") && config.MinConns < minConns {
config.MinConns = minConns
}
// Every query in the gateway passes through here, which is what makes a slow request
// able to say how much of itself was Postgres.
config.ConnConfig.Tracer = queryTracer{}
pool, err := pgxpool.NewWithConfig(ctx, config)
if err != nil {
return nil, fmt.Errorf("store: connect: %w", err)
}
+39
View File
@@ -0,0 +1,39 @@
package store
import (
"context"
"time"
"github.com/jackc/pgx/v5"
"github.com/ponzischeme89/memby/server/internal/timing"
)
// queryTracer attributes Postgres time to the request that asked for it.
//
// pgx's tracer interface is the only place every query in the gateway passes through —
// there are several hundred call sites across internal/store and instrumenting them by
// hand would guarantee the one that matters is the one nobody wrapped. The SQL itself
// is deliberately not recorded: a breakdown is read beside a request line in the admin
// console, and a query text there would put table and column names in front of anybody
// who can open the log.
type queryTracer struct{}
type queryStartKey struct{}
func (queryTracer) TraceQueryStart(
ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryStartData,
) context.Context {
if timing.From(ctx) == nil {
return ctx
}
return context.WithValue(ctx, queryStartKey{}, time.Now())
}
func (queryTracer) TraceQueryEnd(ctx context.Context, _ *pgx.Conn, _ pgx.TraceQueryEndData) {
began, ok := ctx.Value(queryStartKey{}).(time.Time)
if !ok {
return
}
timing.Record(ctx, timing.StageDB, time.Since(began))
}
+268
View File
@@ -0,0 +1,268 @@
// Package timing answers "where did the time go" for one request.
//
// "The home screen took 9.5 seconds" is not actionable; "Emby 7.8s over four calls,
// row assembly 1.4s, Postgres 18ms" is. The gateway's slow requests are almost never
// slow in the gateway — they are waiting on Emby, on an *arr, on Postgres, or on
// several of those in a row — and until the request line could separate those, every
// investigation started by guessing.
//
// A trace is carried in the request context by pointer, the requestIdentity
// arrangement, so a layer four calls down can record against a trace the middleware
// created without anything in between having to know it exists. Everything here is a
// no-op when no trace is installed: a scheduled task, a health probe and every test
// call the same helpers and pay an interface-nil check.
//
// It measures *stages*, not spans in a tree. Most of what matters here is concurrent —
// Home fans four Emby queries out at once — so a total of wall-clock spans would
// exceed the request's own duration and mean nothing. What each stage reports is the
// summed busy time and the number of calls, and the call count is the half that finds
// duplicate work: "emby=7.8s ×14" on a page that should make two lookups is the
// finding, whatever the seconds say.
package timing
import (
"context"
"fmt"
"sort"
"strings"
"sync"
"time"
)
// Stage names. They are constants rather than free strings so a typo cannot quietly
// create a second column that never lines up with the first.
const (
StageEmby = "emby"
StageDB = "db"
StageRedis = "redis"
StageSonarr = "sonarr"
StageRadarr = "radarr"
StageTracearr = "tracearr"
StageMDBList = "mdblist"
StageBazarr = "bazarr"
StageOpenSubtitles = "opensubtitles"
StageIntegrations = "integrations"
// Gateway-side work, named for what a reader would call it rather than for the
// function that does it.
StageRows = "rows"
StageRank = "rank"
StageRatings = "ratings"
StageHero = "hero"
StageEncode = "encode"
StageDecode = "decode"
)
type stage struct {
calls int
total time.Duration
}
// Trace collects one request's stage totals. It is written from every goroutine a
// handler fans out to, so it is locked; the lock is held for a map write and nothing
// else, which is nothing beside the work being measured.
type Trace struct {
mu sync.Mutex
stages map[string]*stage
counts map[string]int
started time.Time
}
type traceKey struct{}
// New installs a trace on ctx and returns both. The caller keeps the pointer so it can
// read the breakdown after the handler has returned.
func New(ctx context.Context) (context.Context, *Trace) {
t := &Trace{
stages: make(map[string]*stage, 8),
counts: make(map[string]int, 4),
started: time.Now(),
}
return context.WithValue(ctx, traceKey{}, t), t
}
// From returns the trace on ctx, or nil outside a traced request.
func From(ctx context.Context) *Trace {
if ctx == nil {
return nil
}
t, _ := ctx.Value(traceKey{}).(*Trace)
return t
}
// Record adds one call's worth of time to a stage. Safe on a nil trace and on an
// untraced context, which is what lets call sites record unconditionally.
func Record(ctx context.Context, name string, d time.Duration) {
From(ctx).Record(name, d)
}
func (t *Trace) Record(name string, d time.Duration) {
if t == nil {
return
}
t.mu.Lock()
defer t.mu.Unlock()
s := t.stages[name]
if s == nil {
s = &stage{}
t.stages[name] = s
}
s.calls++
s.total += d
}
// Start begins a stage and returns the function that ends it. The idiom at the call
// site is `defer timing.Start(ctx, timing.StageRows)()`, which is why the returned
// func takes no argument: a stop that needed a value would be a stop somebody forgets
// to call on the error path.
func Start(ctx context.Context, name string) func() {
t := From(ctx)
if t == nil {
return func() {}
}
began := time.Now()
return func() { t.Record(name, time.Since(began)) }
}
// Count records a tally with no duration — a cache hit, a cache miss, a deduplicated
// upstream call. "Cache MISS" is the first thing anybody wants to know about a slow
// screen and it has no time of its own to report.
func Count(ctx context.Context, name string) {
t := From(ctx)
if t == nil {
return
}
t.mu.Lock()
defer t.mu.Unlock()
t.counts[name]++
}
// Empty reports whether anything was recorded. A request that touched nothing
// measurable should print no breakdown rather than an empty one, which reads as a
// breakdown that failed.
func (t *Trace) Empty() bool {
if t == nil {
return true
}
t.mu.Lock()
defer t.mu.Unlock()
return len(t.stages) == 0 && len(t.counts) == 0
}
// Breakdown renders the trace as one scannable field:
//
// emby=7.81s×4 rows=1.40s db=18ms×3 redis=2ms×2 encode=31ms miss=1
//
// Ordered by time spent, largest first, because the first term is the answer in almost
// every case. Counts follow the stages, since they qualify the timings rather than
// competing with them.
func (t *Trace) Breakdown() string {
if t == nil {
return ""
}
t.mu.Lock()
defer t.mu.Unlock()
type entry struct {
name string
s stage
}
entries := make([]entry, 0, len(t.stages))
for name, s := range t.stages {
entries = append(entries, entry{name: name, s: *s})
}
sort.Slice(entries, func(i, j int) bool {
if entries[i].s.total != entries[j].s.total {
return entries[i].s.total > entries[j].s.total
}
return entries[i].name < entries[j].name
})
parts := make([]string, 0, len(entries)+len(t.counts))
for _, e := range entries {
part := e.name + "=" + formatDuration(e.s.total)
// One call is the ordinary case and saying so on every term would make the
// line harder to read, not easier. A repeat is the interesting number.
if e.s.calls > 1 {
part += fmt.Sprintf("×%d", e.s.calls)
}
parts = append(parts, part)
}
names := make([]string, 0, len(t.counts))
for name := range t.counts {
names = append(names, name)
}
sort.Strings(names)
for _, name := range names {
parts = append(parts, fmt.Sprintf("%s=%d", name, t.counts[name]))
}
return strings.Join(parts, " ")
}
// Unattributed is the request's own time less its largest stage.
//
// It is deliberately not called "gateway processing", and it is deliberately not the
// duration less the *sum* of the stages. Stages overlap — Home fans four Emby queries
// out at once — so a sum would routinely exceed the request's own duration and report a
// negative remainder. What this is instead is a lower bound on time nothing accounted
// for: zero means the evidence explains the request, and a large figure means something
// on the path is not being measured.
func (t *Trace) Unattributed(total time.Duration) time.Duration {
if t == nil {
return 0
}
t.mu.Lock()
defer t.mu.Unlock()
var largest time.Duration
for _, s := range t.stages {
if s.total > largest {
largest = s.total
}
}
if remainder := total - largest; remainder > 0 {
return remainder
}
return 0
}
// formatDuration keeps the column narrow: milliseconds under a second, two decimals
// above it. time.Duration's own String prints "7.812345678s", which is nine digits of
// precision nobody reading a request line has any use for.
func formatDuration(d time.Duration) string {
if d < time.Second {
return fmt.Sprintf("%dms", d.Round(time.Millisecond)/time.Millisecond)
}
return fmt.Sprintf("%.2fs", d.Seconds())
}
type labelKey struct{}
// WithLabel names the stage that upstream calls made under ctx record against, in place
// of the client's own.
//
// The reason it exists is Home: five Emby queries run concurrently, and a breakdown
// reading `emby=7.81s×5` says the launcher waited on Emby without saying which of the
// five it waited on — which is the whole of the next question. Labelled, the same
// request reports `emby.favourites=6.90s emby.resume=310ms emby.nextup=290ms …` and the
// answer is the first term.
//
// It is a context value rather than an argument because the thing being labelled is
// several layers below the thing that knows the name: the row's fetch closure knows it
// is the favourites row, and the transport that measures it is inside an HTTP client two
// packages away. A label is only ever a *finer* name for a stage the client would have
// recorded anyway, so nothing is lost where it is absent.
func WithLabel(ctx context.Context, label string) context.Context {
if label == "" {
return ctx
}
return context.WithValue(ctx, labelKey{}, label)
}
// LabelFrom returns the stage name in force for ctx, falling back to the client's own.
func LabelFrom(ctx context.Context, fallback string) string {
if label, ok := ctx.Value(labelKey{}).(string); ok && label != "" {
return label
}
return fallback
}
+135
View File
@@ -0,0 +1,135 @@
package timing
import (
"context"
"strings"
"testing"
"time"
)
// The breakdown is read at a glance beside a request line, so what matters about it is
// the order and the shape rather than any one figure: the term an operator needs is the
// first one.
func TestBreakdownLeadsWithTheLargestStage(t *testing.T) {
ctx, trace := New(context.Background())
Record(ctx, StageRedis, 2*time.Millisecond)
Record(ctx, StageEmby, 7810*time.Millisecond)
Record(ctx, StageEmby, 300*time.Millisecond)
Record(ctx, StageRows, 1400*time.Millisecond)
Record(ctx, StageDB, 18*time.Millisecond)
got := trace.Breakdown()
want := "emby=8.11s×2 rows=1.40s db=18ms redis=2ms"
if got != want {
t.Fatalf("breakdown = %q, want %q", got, want)
}
}
// A repeat is the interesting number and a single call is the ordinary case, so only the
// repeat is printed. A route that should make one lookup and reports fourteen is the
// finding, whatever its seconds say.
func TestBreakdownMarksRepeatsOnly(t *testing.T) {
ctx, trace := New(context.Background())
Record(ctx, StageEmby, time.Millisecond)
if got := trace.Breakdown(); strings.Contains(got, "×") {
t.Fatalf("a single call should carry no multiplier: %q", got)
}
Record(ctx, StageEmby, time.Millisecond)
if got := trace.Breakdown(); !strings.Contains(got, "×2") {
t.Fatalf("a repeated call should say so: %q", got)
}
}
// Counts qualify the timings rather than competing with them, so they follow the stages
// however large they get.
func TestBreakdownPutsCountsAfterStages(t *testing.T) {
ctx, trace := New(context.Background())
Count(ctx, "miss")
Record(ctx, StageEmby, 5*time.Millisecond)
got := trace.Breakdown()
if !strings.HasPrefix(got, "emby=") || !strings.HasSuffix(got, "miss=1") {
t.Fatalf("breakdown = %q, want a stage first and a count last", got)
}
}
// Everything here is called unconditionally from layers that have no idea whether they
// are inside a request. An untraced context must cost a nil check and nothing else.
func TestUntracedContextIsInert(t *testing.T) {
ctx := context.Background()
Record(ctx, StageEmby, time.Second)
Count(ctx, "miss")
Start(ctx, StageRows)()
if From(ctx) != nil {
t.Fatal("a plain context must carry no trace")
}
var absent *Trace
if !absent.Empty() || absent.Breakdown() != "" {
t.Fatal("a nil trace must report nothing rather than panicking")
}
}
// A request that touched nothing measurable prints no breakdown at all: an empty one
// reads as a breakdown that failed.
func TestEmptyTraceIsEmpty(t *testing.T) {
_, trace := New(context.Background())
if !trace.Empty() {
t.Fatal("a fresh trace should be empty")
}
trace.Record(StageDB, time.Millisecond)
if trace.Empty() {
t.Fatal("a trace with a stage is not empty")
}
}
// Unattributed subtracts the largest stage rather than the sum of all of them, because
// stages overlap — Home fans four Emby queries out at once — and a sum would routinely
// exceed the request's own duration and report a negative remainder.
func TestUnattributedSubtractsTheLargestStageOnly(t *testing.T) {
_, trace := New(context.Background())
trace.Record(StageEmby, 400*time.Millisecond)
trace.Record(StageDB, 50*time.Millisecond)
if got := trace.Unattributed(500 * time.Millisecond); got != 100*time.Millisecond {
t.Fatalf("unattributed = %v, want 100ms", got)
}
// Never negative. Concurrent calls summed into one stage can exceed the wall clock,
// and a negative "elsewhere" would read as a broken measurement rather than as the
// concurrency it actually is.
trace.Record(StageEmby, 400*time.Millisecond)
if got := trace.Unattributed(500 * time.Millisecond); got != 0 {
t.Fatalf("unattributed = %v, want 0", got)
}
}
// A label is only ever a finer name for a stage the client would have recorded anyway,
// so its absence must change nothing.
func TestLabelFallsBackToTheClientsOwnStage(t *testing.T) {
ctx := context.Background()
if got := LabelFrom(ctx, StageEmby); got != StageEmby {
t.Fatalf("unlabelled context = %q, want %q", got, StageEmby)
}
if got := LabelFrom(WithLabel(ctx, "emby.favourites"), StageEmby); got != "emby.favourites" {
t.Fatalf("labelled context = %q, want emby.favourites", got)
}
if got := LabelFrom(WithLabel(ctx, ""), StageEmby); got != StageEmby {
t.Fatalf("an empty label must not replace the stage, got %q", got)
}
}
// Sub-second times are read in milliseconds and longer ones in seconds. Duration's own
// String prints nine digits of precision, which is unreadable in a column.
func TestFormatDurationStaysNarrow(t *testing.T) {
cases := map[time.Duration]string{
0: "0ms",
1500 * time.Microsecond: "2ms",
999 * time.Millisecond: "999ms",
time.Second: "1.00s",
9512 * time.Millisecond: "9.51s",
}
for input, want := range cases {
if got := formatDuration(input); got != want {
t.Fatalf("formatDuration(%v) = %q, want %q", input, got, want)
}
}
}
+80
View File
@@ -0,0 +1,80 @@
package timing
import (
"io"
"net/http"
"time"
)
// Transport records every upstream HTTP call against a stage.
//
// It is a RoundTripper rather than a helper each client calls because there are eight
// upstream clients here and the one thing they all genuinely share is that they make
// HTTP requests. Wrapping the transport means a new integration is instrumented by
// being constructed rather than by somebody remembering to measure it.
//
// The context it reads is the request's, so a call made outside a traced request —
// a scheduled sync, a health probe — records nothing and costs one nil check.
type Transport struct {
Stage string
Base http.RoundTripper
}
func (t *Transport) RoundTrip(r *http.Request) (*http.Response, error) {
base := t.Base
if base == nil {
base = http.DefaultTransport
}
trace := From(r.Context())
if trace == nil {
return base.RoundTrip(r)
}
stage := LabelFrom(r.Context(), t.Stage)
began := time.Now()
resp, err := base.RoundTrip(r)
trace.Record(stage, time.Since(began))
if err != nil || resp == nil || resp.Body == nil {
return resp, err
}
// RoundTrip returns once the headers are in, and a home row is several hundred
// kilobytes of JSON still on the wire at that point. Reading the body is part of
// what the upstream cost, so the stage keeps accruing until the caller closes it —
// otherwise the largest responses are the ones the breakdown under-reports.
resp.Body = &timedBody{ReadCloser: resp.Body, trace: trace, stage: stage}
return resp, nil
}
type timedBody struct {
io.ReadCloser
trace *Trace
stage string
spent time.Duration
done bool
}
func (b *timedBody) Read(p []byte) (int, error) {
began := time.Now()
n, err := b.ReadCloser.Read(p)
b.spent += time.Since(began)
return n, err
}
func (b *timedBody) Close() error {
err := b.ReadCloser.Close()
if !b.done {
b.done = true
b.trace.Record(b.stage, b.spent)
}
return err
}
// Instrument wraps a client's transport in place. Called during construction, before
// anything is serving, so it needs no synchronisation — the stance the Emby client's
// other setters take.
func Instrument(client *http.Client, stage string) *http.Client {
if client == nil {
return nil
}
client.Transport = &Transport{Stage: stage, Base: client.Transport}
return client
}
+3 -1
View File
@@ -16,6 +16,8 @@ import (
"strings"
"time"
"unicode"
"github.com/ponzischeme89/memby/server/internal/timing"
)
type Client struct {
@@ -115,7 +117,7 @@ func New(baseURL, apiKey, serverID string, timeout time.Duration) *Client {
baseURL: strings.TrimRight(strings.TrimSpace(baseURL), "/"),
apiKey: strings.TrimSpace(apiKey),
serverID: strings.TrimSpace(serverID),
http: &http.Client{Timeout: timeout},
http: timing.Instrument(&http.Client{Timeout: timeout}, timing.StageTracearr),
}
}