From 07825450131cde9fa0b9fcbab3b1df42b9d04ddf Mon Sep 17 00:00:00 2001 From: ponzischeme89 Date: Wed, 19 Aug 2026 18:08:00 +1200 Subject: [PATCH] 0.2.79 - Slow api fixes --- .env.example | 7 + CLAUDE.md | 115 +++++++ admin-ui/package.json | 2 +- admin-ui/src/api/types.ts | 2 + admin-ui/src/lib/logmodel.ts | 15 +- admin-ui/src/pages/Settings.tsx | 19 +- app/build.gradle.kts | 2 +- server/go.mod | 2 +- server/internal/api/admin_gateway_settings.go | 18 ++ server/internal/api/api.go | 31 ++ server/internal/api/coalesce.go | 105 +++++++ server/internal/api/extras.go | 96 +++--- server/internal/api/features.go | 18 +- server/internal/api/features_cache.go | 61 ++++ server/internal/api/gateway_settings.go | 16 + server/internal/api/home.go | 78 +++-- server/internal/api/items.go | 217 +++++++------ server/internal/api/people.go | 93 +++--- server/internal/api/playback.go | 13 +- server/internal/api/playback_follow.go | 54 ++++ server/internal/api/playback_follow_test.go | 87 ++++++ server/internal/api/ranking.go | 291 ++++++++++++++---- server/internal/api/related.go | 54 ++-- server/internal/bazarr/client.go | 6 +- server/internal/buildinfo/VERSION | 2 +- server/internal/cache/cache.go | 96 +++++- server/internal/cache/cache_test.go | 29 ++ server/internal/config/config.go | 7 + server/internal/emby/client.go | 5 +- server/internal/integrations/integrations.go | 3 +- server/internal/mdblist/client.go | 6 +- server/internal/opensubtitles/client.go | 6 +- server/internal/radarr/client.go | 6 +- server/internal/sonarr/client.go | 6 +- server/internal/store/gateway_settings.go | 10 + server/internal/store/store.go | 33 +- server/internal/store/tracing.go | 39 +++ server/internal/timing/timing.go | 268 ++++++++++++++++ server/internal/timing/timing_test.go | 135 ++++++++ server/internal/timing/transport.go | 80 +++++ server/internal/tracearr/client.go | 4 +- 41 files changed, 1820 insertions(+), 317 deletions(-) create mode 100644 server/internal/api/coalesce.go create mode 100644 server/internal/api/features_cache.go create mode 100644 server/internal/api/playback_follow.go create mode 100644 server/internal/api/playback_follow_test.go create mode 100644 server/internal/cache/cache_test.go create mode 100644 server/internal/store/tracing.go create mode 100644 server/internal/timing/timing.go create mode 100644 server/internal/timing/timing_test.go create mode 100644 server/internal/timing/transport.go diff --git a/.env.example b/.env.example index 47314e0..168615e 100644 --- a/.env.example +++ b/.env.example @@ -143,6 +143,13 @@ MEMBY_BAZARR_TIMEOUT=45s # raises the matching "back online" one. 0 turns the probe and both banners off. MEMBY_EMBY_HEALTH_INTERVAL=60s +# How slow a request has to be before its log line carries a breakdown of where the time +# went: Emby, Postgres, Redis, each *arr, and the gateway's own row assembly and encoding. +# Fast requests carry none deliberately — a breakdown on every /v1/status poll would bury +# the one it exists to reveal. Lower it while chasing one slow screen. The gateway's +# settings page overrides it without a redeployment. +MEMBY_SLOW_REQUEST_THRESHOLD=500ms + # Optional Tracearr-powered For You signals. Create a read-only public API key in # Tracearr Settings. Server ID is optional unless Tracearr monitors multiple servers. MEMBY_TRACEARR_URL=https://tracearr.sublogue.com/ diff --git a/CLAUDE.md b/CLAUDE.md index fab50b3..f5e1217 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -869,6 +869,121 @@ Ten-second progress reports and per-keystroke searches are DEBUG on purpose. the admin rail — bump it with a meaningful server change; nothing else identifies which tree a container was deployed from. +**A slow request says where its time went.** `internal/timing` is a per-request trace +carried in the context by pointer, the `requestIdentity` arrangement, and every layer +under the handler records against it: an `http.RoundTripper` wrapper on all eight upstream +clients (Emby, both *arrs, Tracearr, MDBList, Bazarr, OpenSubtitles, the integrations +dispatcher), a `pgx.QueryTracer` on the pool, and the cache's own `Get`/`Set`/`Delete`. +`withLogging` attaches the result as a `breakdown` field once the request passes +`slowRequestThreshold` — an operator override over `MEMBY_SLOW_REQUEST_THRESHOLD`, 500ms +deployed — and the console gives it a heading of its own in the log drawer and a second +line on the row. A duration column could only ever say that something was slow; this is +what makes it actionable: + +``` +GET /v1/home 9.51s breakdown=emby.favourites=7.80s fanout=7.81s rows=1.40s db=18ms encode=31ms miss=1 +``` + +Things to preserve: + +- **It measures stages, not spans in a tree.** Most of what matters here is concurrent — + Home fans five Emby queries out at once — so a total of wall-clock spans would exceed + the request's own duration and mean nothing. What a stage reports is summed busy time + and a **call count**, and the 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. +- **The transport times the body too**, not only `RoundTrip`. Headers arrive while several + hundred kilobytes of row JSON is still on the wire, so timing the round trip alone would + under-report exactly the largest responses. +- **`timing.WithLabel` is what makes a fan-out legible.** Five concurrent Emby queries + under one `emby=` stage say the launcher waited on Emby without saying which of them it + waited on, which is the whole of the next question. Each home row, and each leg of the + season-finale chain, names its own stage. A label is only ever a finer name for a stage + the client would have recorded anyway, so its absence changes nothing. +- **`elsewhere` is the duration less the *largest* stage**, never less their sum, and is + clamped at zero. It is deliberately not called "gateway processing": with overlapping + stages no such figure exists, and what this is instead is a lower bound on time nothing + accounted for. +- **Everything is inert without a trace.** A scheduled sync, a health probe and every unit + test call the same helpers and pay one nil check. + +**And the fixes the trace found.** They are all one shape — work on the critical path that +did not have to be there — and each is easy to reintroduce: + +- **The automatic My Shows check ran a hundred times per episode.** It hangs off playback + reports, one arrives every ten seconds, and its own guard (half watched) is true for + every one of those after the halfway mark — so the second half of every episode ran two + serial Emby lookups, a Sonarr catalogue read and a Postgres write per tick and threw all + but the first away at the durable insert. `followChecks.claim` is a map lookup that comes + *first*, before the feature check and long before Emby. Deduplicating on the insert was + correct and was never the problem; what it could not do is prevent the work leading up to + it. +- **`InvalidateUser` walked the whole keyspace.** Redis SCAN's cost is a property of the + keyspace rather than of the pattern, so dropping one viewer's dozen cached rows meant + paging through every item lookup the household had ever cached — on the playback stop + report, which is the request a television makes as somebody presses Back out of a film. + `Cache.Set` now indexes user-scoped keys into `idx:u:` in the same pipeline as the + write, and invalidation reads that set. The SCAN survives as the fallback for a user with + no index, which is the honest answer for a gateway upgraded with older keys still live. +- **Stable metadata lived where invalidation could reach it.** A person's biography, a + film's extras and whether an episode closes its season carry no user data and cannot + change when somebody finishes an episode, yet all three sat under `u::` and were + destroyed several times an evening by playback stops — then re-read from Emby by the next + card anybody looked at. `cache.MetadataKey` is where they live now: outside the + invalidated namespace and unkeyed by viewer, so one lookup answers for the household. The + rule for putting something there is narrow and worth stating: **the value must contain + nothing derived from a viewer.** Anything carrying `UserData` belongs in `UserKey`, where + invalidation can reach it — which is why the *filmography* stayed behind while the + biography moved. +- **Identical concurrent misses each asked upstream.** `s.cachedRead` / `buildCached` + (`coalesce.go`) put a `singleflight` between the cache miss and the upstream call, keyed + by the cache key. The log showed one person lookup answering in 2.2s, then 4.6s, then + 7.7s within a few seconds; nothing was getting slower, three televisions had missed the + same key at the same moment and Emby was serving three copies of one expensive query. + The cache could not help — nothing had been written when the second and third arrived. + 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 a D-pad + moving off a card would otherwise fail everybody queued behind it), and **a joiner is + counted** on the breakdown, because "slow" and "queued behind itself" are different + problems with different fixes. It is applied to the person biography and filmography, + related titles, extras, series episodes and item detail — every route the launcher warms + on focus and then asks for again on the press. +- **Serial chains that were never chains.** The season-finale route fetched Sonarr's series + catalogue *after* both Emby lookups it does not depend on; the extras route made its two + lookups in turn; `filterRecommendationPermissions` walked its hundred-id batches one at a + time on the home tail. All three now run together, and the permission check keeps its + original refusal — one failed batch hides every candidate, because a partial answer is + not evidence that the rest is permitted. +- **Home read its ranking evidence after the rows arrived.** `rankingContext` is seven + Postgres reads and `UserRowStats` an eighth, all of them functions of the viewer's id and + nothing else — so they ran with nothing else in flight, on the critical path, for answers + that were available before Emby was asked anything. They are issued *beside* the Emby + fan-out now (`rankingInputs`, `personalizeTitlesWith`), and the seven inside + `rankingContext` are issued together. The two that are genuinely a chain stay one, but + their *writes* are ordered rather than their reads: `ApplyOnboarding` and + `ApplyExplicitPreference` are not commutative and the ranking must never depend on which + query answered first. +- **Two documents were queried per request and could not change per request.** The feature + policy is read from sixteen places plus `/v1/status`, which every open television polls + every ten seconds; the household completion scores are a six-month aggregate whose answer + is *the same for every viewer in the house*. Both are cached in memory — five seconds and + a minute — and the operator's own write invalidates the first outright, so the window is + "how long until another instance notices", not "how long until my change takes effect". + A failed household read returns the previous reading rather than caching an empty one. +- **The connection pool was four.** `pgxpool`'s default is `max(4, GOMAXPROCS)`, which on + the NAS is fewer than one household's televisions — and a query waiting for a connection + is invisible, because the query time stays honest while the request is slow anyway. + `store.Open` floors it at 16 with 4 kept warm, and leaves alone any deployment that + stated `pool_max_conns` for itself. + +Two things the trace found and did **not** change, because both are behaviour rather than +waste. `fieldsRow` asks Emby for `People` and `RecursiveItemCount`, which are the two +expensive fields on a row query — but the weighted ranker scores person affinity from the +first and a series card prints its episode count from the second, so dropping either is a +feature decision. And the home cache key carries the device id, which multiplies the +rebuild by the number of televisions; that is there because a recommendation's reason names +"this TV", although the recommendations it decorates come from a cache that is not +device-keyed. Both are worth revisiting deliberately. + **Emby's own version rides the reachability probe**, and the television prints it beside the gateway's on Settings → About: `0.1.50 (4.10.0.21)`. `emby.Client.Ping` already asks `/System/Info/Public`, which carries `Version`, so reading it costs nothing — a second call diff --git a/admin-ui/package.json b/admin-ui/package.json index 18f313e..6aaa631 100644 --- a/admin-ui/package.json +++ b/admin-ui/package.json @@ -1,7 +1,7 @@ { "name": "memby-admin", "private": true, - "version": "0.1.39", + "version": "0.1.40", "type": "module", "description": "The Memby operations console. Built by Vite, served by the memby-admin container, proxied by the gateway at /admin.", "scripts": { diff --git a/admin-ui/src/api/types.ts b/admin-ui/src/api/types.ts index 4638a47..133215a 100644 --- a/admin-ui/src/api/types.ts +++ b/admin-ui/src/api/types.ts @@ -752,6 +752,7 @@ export interface GatewaySettings { sonarrAlertMinutes: number; radarrAlertMinutes: number; embyHealthSeconds: number; + slowRequestMillis: number; librarySyncMinutes: number; updatedAt?: string; updatedBy?: string; @@ -766,6 +767,7 @@ export interface GatewaySettingValues { sonarrAlertMinutes: number; radarrAlertMinutes: number; embyHealthSeconds: number; + slowRequestMillis: number; librarySyncMinutes: number; } diff --git a/admin-ui/src/lib/logmodel.ts b/admin-ui/src/lib/logmodel.ts index 81387cf..37ffe9f 100644 --- a/admin-ui/src/lib/logmodel.ts +++ b/admin-ui/src/lib/logmodel.ts @@ -373,7 +373,13 @@ export const DRAWER_SECTIONS: { title: string; keys: string[] }[] = [ 'watched', 'subtitles', 'subtitle_track', 'subtitle_language', 'event_name', ], }, - { title: 'Diagnostics', keys: ['error', 'stack', 'correlation', 'version', 'gateway_version', 'duration'] }, + // The breakdown is the answer to the question the duration column can only ask, so it + // gets a heading of its own rather than sitting among the diagnostics. `elsewhere` is + // the request's own time less its largest stage: not "gateway processing" — the stages + // overlap, so no such figure exists — but the honest answer to whether any of it is + // unaccounted for. + { title: 'Where the time went', keys: ['duration', 'breakdown', 'elsewhere'] }, + { title: 'Diagnostics', keys: ['error', 'stack', 'correlation', 'version', 'gateway_version'] }, ]; const SECTIONED = new Set(DRAWER_SECTIONS.flatMap((section) => section.keys)); @@ -447,7 +453,12 @@ function derive(event: LogEvent): Shaped { // most of a log — keeps its identity at the end of its own line instead, because density // is the whole reason this page is worth watching and a second line on every request // would halve what an operator can see at once. - const secondary = detail || (isRequest ? '' : context); + // A slow request earns the second line every other request is denied. It is exactly + // the case density was being protected for: these are rare by definition, and the whole + // point of measuring where the time went is that an operator scanning the log can see it + // without opening anything. + const breakdown = text(attributes.breakdown); + const secondary = detail || (isRequest ? breakdown : context); const secondaryTone: Shaped['secondaryTone'] = detail ? 'error' : secondary ? 'context' : ''; const trail = !detail && isRequest ? context : ''; diff --git a/admin-ui/src/pages/Settings.tsx b/admin-ui/src/pages/Settings.tsx index dd0b7e5..76de479 100644 --- a/admin-ui/src/pages/Settings.tsx +++ b/admin-ui/src/pages/Settings.tsx @@ -53,6 +53,7 @@ interface Draft { sonarrAlertMinutes: string; radarrAlertMinutes: string; embyHealthSeconds: string; + slowRequestMillis: string; librarySyncMinutes: string; } @@ -64,6 +65,7 @@ function draftFrom(settings: GatewaySettings): Draft { sonarrAlertMinutes: numberFieldValue(settings.sonarrAlertMinutes), radarrAlertMinutes: numberFieldValue(settings.radarrAlertMinutes), embyHealthSeconds: numberFieldValue(settings.embyHealthSeconds), + slowRequestMillis: numberFieldValue(settings.slowRequestMillis), librarySyncMinutes: numberFieldValue(settings.librarySyncMinutes), }; } @@ -94,6 +96,7 @@ export function SettingsPage() { sonarrAlertMinutes: parseNumberField(draft.sonarrAlertMinutes, true), radarrAlertMinutes: parseNumberField(draft.radarrAlertMinutes, true), embyHealthSeconds: parseNumberField(draft.embyHealthSeconds, true), + slowRequestMillis: parseNumberField(draft.slowRequestMillis, true), librarySyncMinutes: parseNumberField(draft.librarySyncMinutes, true), }; const saved = await wrap( @@ -113,7 +116,7 @@ export function SettingsPage() { const saved = await wrap( () => api.post('/admin/api/gateway-settings', { timezone: '', logLevel: '', sessionIdleDays: 0, - sonarrAlertMinutes: 0, radarrAlertMinutes: 0, embyHealthSeconds: 0, + sonarrAlertMinutes: 0, radarrAlertMinutes: 0, embyHealthSeconds: 0, slowRequestMillis: 0, librarySyncMinutes: 0, }), 'Every setting is back to what this container was deployed with.', @@ -151,6 +154,7 @@ export function SettingsPage() { { label: 'Log level', value: effective.logLevel }, { label: 'Sign-in expiry', value: describe(effective.sessionIdleDays, 'day') }, { label: 'Emby health probe', value: describe(effective.embyHealthSeconds, 'second') }, + { label: 'Slow-request breakdown', value: describe(effective.slowRequestMillis, 'millisecond') }, { label: 'Catalogue sweep', value: describe(effective.librarySyncMinutes, 'minute') }, { label: 'Episode alert window', value: describe(effective.sonarrAlertMinutes, 'minute') }, { label: 'Film alert window', value: describe(effective.radarrAlertMinutes, 'minute') }, @@ -216,6 +220,19 @@ export function SettingsPage() { /> + + set('slowRequestMillis', event.target.value)} + /> + + 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 "" diff --git a/server/internal/api/api.go b/server/internal/api/api.go index 39e1d47..361d71e 100644 --- a/server/internal/api/api.go +++ b/server/internal/api/api.go @@ -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...) }) } diff --git a/server/internal/api/coalesce.go b/server/internal/api/coalesce.go new file mode 100644 index 0000000..4d3c669 --- /dev/null +++ b/server/internal/api/coalesce.go @@ -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) +} diff --git a/server/internal/api/extras.go b/server/internal/api/extras.go index 6ee0392..55cc1ed 100644 --- a/server/internal/api/extras.go +++ b/server/internal/api/extras.go @@ -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. diff --git a/server/internal/api/features.go b/server/internal/api/features.go index 553cf5a..d55d6c3 100644 --- a/server/internal/api/features.go +++ b/server/internal/api/features.go @@ -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) { diff --git a/server/internal/api/features_cache.go b/server/internal/api/features_cache.go new file mode 100644 index 0000000..a9e39ac --- /dev/null +++ b/server/internal/api/features_cache.go @@ -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 +} diff --git a/server/internal/api/gateway_settings.go b/server/internal/api/gateway_settings.go index c6002e6..682d5c8 100644 --- a/server/internal/api/gateway_settings.go +++ b/server/internal/api/gateway_settings.go @@ -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), } } diff --git a/server/internal/api/home.go b/server/internal/api/home.go index 9302fce..a4ce09d 100644 --- a/server/internal/api/home.go +++ b/server/internal/api/home.go @@ -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") diff --git a/server/internal/api/items.go b/server/internal/api/items.go index 1a39246..9b191ea 100644 --- a/server/internal/api/items.go +++ b/server/internal/api/items.go @@ -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, ¤t) != 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. diff --git a/server/internal/api/people.go b/server/internal/api/people.go index a733ff2..8d06789 100644 --- a/server/internal/api/people.go +++ b/server/internal/api/people.go @@ -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) } diff --git a/server/internal/api/playback.go b/server/internal/api/playback.go index 7b6a5ce..f94add3 100644 --- a/server/internal/api/playback.go +++ b/server/internal/api/playback.go @@ -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) } diff --git a/server/internal/api/playback_follow.go b/server/internal/api/playback_follow.go new file mode 100644 index 0000000..1d94aa1 --- /dev/null +++ b/server/internal/api/playback_follow.go @@ -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 +} diff --git a/server/internal/api/playback_follow_test.go b/server/internal/api/playback_follow_test.go new file mode 100644 index 0000000..6029c13 --- /dev/null +++ b/server/internal/api/playback_follow_test.go @@ -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) + } +} diff --git a/server/internal/api/ranking.go b/server/internal/api/ranking.go index 6b16eb1..23588f5 100644 --- a/server/internal/api/ranking.go +++ b/server/internal/api/ranking.go @@ -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() diff --git a/server/internal/api/related.go b/server/internal/api/related.go index 64f06e9..1230e9c 100644 --- a/server/internal/api/related.go +++ b/server/internal/api/related.go @@ -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 diff --git a/server/internal/bazarr/client.go b/server/internal/bazarr/client.go index dd9bcdc..c856d8e 100644 --- a/server/internal/bazarr/client.go +++ b/server/internal/bazarr/client.go @@ -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), } } diff --git a/server/internal/buildinfo/VERSION b/server/internal/buildinfo/VERSION index 09d8256..1ec9b0b 100644 --- a/server/internal/buildinfo/VERSION +++ b/server/internal/buildinfo/VERSION @@ -1 +1 @@ -0.1.59 +0.1.60 diff --git a/server/internal/cache/cache.go b/server/internal/cache/cache.go index ed30b39..b393f2a 100644 --- a/server/internal/cache/cache.go +++ b/server/internal/cache/cache.go @@ -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::*`, 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::*` 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 } diff --git a/server/internal/cache/cache_test.go b/server/internal/cache/cache_test.go new file mode 100644 index 0000000..7c25711 --- /dev/null +++ b/server/internal/cache/cache_test.go @@ -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) + } + }) + } +} diff --git a/server/internal/config/config.go b/server/internal/config/config.go index 045cbcd..f6c3117 100644 --- a/server/internal/config/config.go +++ b/server/internal/config/config.go @@ -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")), diff --git a/server/internal/emby/client.go b/server/internal/emby/client.go index 9019911..bc95fcc 100644 --- a/server/internal/emby/client.go +++ b/server/internal/emby/client.go @@ -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), } } diff --git a/server/internal/integrations/integrations.go b/server/internal/integrations/integrations.go index 10e376e..03d1816 100644 --- a/server/internal/integrations/integrations.go +++ b/server/internal/integrations/integrations.go @@ -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), } diff --git a/server/internal/mdblist/client.go b/server/internal/mdblist/client.go index 3e3464f..d79f3f2 100644 --- a/server/internal/mdblist/client.go +++ b/server/internal/mdblist/client.go @@ -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), } } diff --git a/server/internal/opensubtitles/client.go b/server/internal/opensubtitles/client.go index 0a4c080..26d188d 100644 --- a/server/internal/opensubtitles/client.go +++ b/server/internal/opensubtitles/client.go @@ -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), } } diff --git a/server/internal/radarr/client.go b/server/internal/radarr/client.go index bb72884..0de1baf 100644 --- a/server/internal/radarr/client.go +++ b/server/internal/radarr/client.go @@ -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), } } diff --git a/server/internal/sonarr/client.go b/server/internal/sonarr/client.go index ca40a13..df78391 100644 --- a/server/internal/sonarr/client.go +++ b/server/internal/sonarr/client.go @@ -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), } } diff --git a/server/internal/store/gateway_settings.go b/server/internal/store/gateway_settings.go index 12d8ad3..e1871c3 100644 --- a/server/internal/store/gateway_settings.go +++ b/server/internal/store/gateway_settings.go @@ -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) diff --git a/server/internal/store/store.go b/server/internal/store/store.go index a23546c..ad303e9 100644 --- a/server/internal/store/store.go +++ b/server/internal/store/store.go @@ -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) } diff --git a/server/internal/store/tracing.go b/server/internal/store/tracing.go new file mode 100644 index 0000000..5e9c1c3 --- /dev/null +++ b/server/internal/store/tracing.go @@ -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)) +} diff --git a/server/internal/timing/timing.go b/server/internal/timing/timing.go new file mode 100644 index 0000000..f6d2e53 --- /dev/null +++ b/server/internal/timing/timing.go @@ -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 +} diff --git a/server/internal/timing/timing_test.go b/server/internal/timing/timing_test.go new file mode 100644 index 0000000..f5e598f --- /dev/null +++ b/server/internal/timing/timing_test.go @@ -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) + } + } +} diff --git a/server/internal/timing/transport.go b/server/internal/timing/transport.go new file mode 100644 index 0000000..1250177 --- /dev/null +++ b/server/internal/timing/transport.go @@ -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 +} diff --git a/server/internal/tracearr/client.go b/server/internal/tracearr/client.go index 9211fd0..5993593 100644 --- a/server/internal/tracearr/client.go +++ b/server/internal/tracearr/client.go @@ -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), } }