106 lines
4.3 KiB
Go
106 lines
4.3 KiB
Go
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)
|
|
}
|