0.2.64 update
This commit is contained in:
ponzischeme89
@@ -0,0 +1,525 @@
|
||||
package credits
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The service: one worker, one queue, and the rules about when it is allowed to run.
|
||||
//
|
||||
// Everything expensive in this package funnels through Process, and the shape of that
|
||||
// function is the performance claim in miniature — resolve, check the marker, and in the
|
||||
// common case stop there having touched no media at all. Only a candidate that survives the
|
||||
// cache check reaches a decoder.
|
||||
|
||||
const (
|
||||
// livePlaybackDelay is how long a playback has to survive before it is worth scanning
|
||||
// for. Somebody browsing the launcher starts and abandons episodes constantly, and
|
||||
// scanning on the Play press would turn every one of those into disk activity. Waiting
|
||||
// costs nothing: the credits are forty minutes away.
|
||||
livePlaybackDelay = 45 * time.Second
|
||||
|
||||
// idlePoll is how often the worker looks for work when the queue is empty. Coarse on
|
||||
// purpose — nothing here is urgent, and a tight loop on an idle NAS is exactly the sort
|
||||
// of background cost this package is supposed not to have.
|
||||
idlePoll = 30 * time.Second
|
||||
|
||||
// busyBackoff is how long the worker stands down when the server is under load.
|
||||
busyBackoff = 2 * time.Minute
|
||||
|
||||
// scanBudget bounds one candidate end to end. Past it the answer is not worth the
|
||||
// resources, and a scan that overruns is far more likely to be a pathological file than
|
||||
// a slow one.
|
||||
scanBudget = 90 * time.Second
|
||||
)
|
||||
|
||||
// ResolvedMedia is everything the service needs to turn a candidate into a scan.
|
||||
type ResolvedMedia struct {
|
||||
Version MediaVersion
|
||||
// URL is where the bytes are — Emby's direct stream route, since the gateway has no
|
||||
// filesystem access to the media.
|
||||
URL string
|
||||
SeriesID string
|
||||
Season int
|
||||
Episode int
|
||||
RuntimeMs int64
|
||||
// EmbyCreditsMs is a chapter marker Emby found itself, if any. When Emby already knows,
|
||||
// this subsystem must do nothing at all.
|
||||
EmbyCreditsMs int64
|
||||
}
|
||||
|
||||
// MediaResolver turns an item id into something scannable. Backed by Emby; kept an interface
|
||||
// so the scheduler, the queue and every rule above them can be tested with no server.
|
||||
type MediaResolver interface {
|
||||
Resolve(ctx context.Context, itemID string) (ResolvedMedia, error)
|
||||
}
|
||||
|
||||
// BehaviourSource reads the stops a household already made. Backed by the tracearr_sessions
|
||||
// table, which is written by the For You import and by nothing here.
|
||||
type BehaviourSource interface {
|
||||
Stops(ctx context.Context, itemID string) ([]StopEvent, error)
|
||||
}
|
||||
|
||||
// LoadGauge answers whether the server is too busy for speculative work. Deliberately a
|
||||
// single boolean: building a telemetry subsystem to answer it would cost more than the
|
||||
// scanning it is meant to defer.
|
||||
type LoadGauge interface {
|
||||
Busy(ctx context.Context) bool
|
||||
}
|
||||
|
||||
// Deps is the service's wiring. Only Repository and MediaResolver are required; a service
|
||||
// with no detector runs on behavioural evidence alone, which is what happens when ffmpeg is
|
||||
// missing from the image.
|
||||
type Deps struct {
|
||||
Repository Repository
|
||||
Resolver MediaResolver
|
||||
Source CandidateSource
|
||||
Detector Detector
|
||||
Behaviour BehaviourSource
|
||||
Load LoadGauge
|
||||
Log *slog.Logger
|
||||
Config Config
|
||||
}
|
||||
|
||||
type Service struct {
|
||||
repo Repository
|
||||
resolver MediaResolver
|
||||
source CandidateSource
|
||||
detector Detector
|
||||
behaviour BehaviourSource
|
||||
load LoadGauge
|
||||
log *slog.Logger
|
||||
cfg Config
|
||||
|
||||
queue *Queue
|
||||
flight *flightGroup
|
||||
|
||||
mu sync.Mutex
|
||||
pending map[string]*pending
|
||||
|
||||
// liveDelay is a field rather than the constant so a test does not have to wait
|
||||
// three quarters of a minute to prove that abandonment cancels a scan.
|
||||
liveDelay time.Duration
|
||||
}
|
||||
|
||||
func New(deps Deps) *Service {
|
||||
cfg := deps.Config
|
||||
if cfg.QueueLimit <= 0 {
|
||||
cfg = DefaultConfig()
|
||||
}
|
||||
detector := deps.Detector
|
||||
if detector == nil {
|
||||
detector = noopDetector{}
|
||||
}
|
||||
log := deps.Log
|
||||
if log == nil {
|
||||
log = slog.Default()
|
||||
}
|
||||
return &Service{
|
||||
repo: deps.Repository,
|
||||
resolver: deps.Resolver,
|
||||
source: deps.Source,
|
||||
detector: detector,
|
||||
behaviour: deps.Behaviour,
|
||||
load: deps.Load,
|
||||
log: log.With("component", "credits"),
|
||||
cfg: cfg,
|
||||
queue: NewQueue(cfg.QueueLimit),
|
||||
flight: newFlightGroup(),
|
||||
pending: map[string]*pending{},
|
||||
liveDelay: livePlaybackDelay,
|
||||
}
|
||||
}
|
||||
|
||||
// Marker is the read path, and the one the API calls. It is a single indexed lookup and it
|
||||
// is what makes the steady state free: once a household's viewing has settled, almost every
|
||||
// call to this subsystem is this function returning a row.
|
||||
func (s *Service) Marker(ctx context.Context, itemID string) (Marker, bool, error) {
|
||||
if s == nil || s.repo == nil || itemID == "" {
|
||||
return Marker{}, false, nil
|
||||
}
|
||||
resolved, err := s.resolver.Resolve(ctx, itemID)
|
||||
if err != nil {
|
||||
return Marker{}, false, err
|
||||
}
|
||||
return s.repo.GetMarker(ctx, itemID, Fingerprint(resolved.Version))
|
||||
}
|
||||
|
||||
// Refresh rebuilds the speculative queue from demand. Registered as a scheduled task, so its
|
||||
// interval is the operator's to change and its last run is visible in the console.
|
||||
//
|
||||
// The returned detail is the scheduler's one-line summary and is empty when nothing changed,
|
||||
// which is what stops a task running every ten minutes announcing itself every ten minutes.
|
||||
func (s *Service) Refresh(ctx context.Context) (string, error) {
|
||||
if s.source == nil {
|
||||
return "", nil
|
||||
}
|
||||
candidates, err := s.source.Candidates(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
// The cache check happens here as well as in the worker, and that is the point: an
|
||||
// episode whose marker already exists must never occupy a queue slot, or a household
|
||||
// that has been watching one show for a week ends up with a permanently full queue of
|
||||
// work that will all turn out to be unnecessary.
|
||||
wanted := make([]Candidate, 0, len(candidates))
|
||||
skipped := 0
|
||||
for _, candidate := range candidates {
|
||||
known, err := s.known(ctx, candidate.ItemID)
|
||||
if err != nil {
|
||||
// Trouble reading a marker is not a reason to drop the candidate; the worker
|
||||
// will check again and is the one that can afford to be wrong.
|
||||
s.log.Debug("marker lookup failed", "item_id", candidate.ItemID, "error", err)
|
||||
}
|
||||
if known {
|
||||
skipped++
|
||||
continue
|
||||
}
|
||||
wanted = append(wanted, candidate)
|
||||
}
|
||||
s.queue.Replace(wanted)
|
||||
|
||||
if len(wanted) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
for _, candidate := range wanted {
|
||||
s.log.Debug("credits_candidate",
|
||||
"item", candidate.ItemID, "reason", candidate.Reason,
|
||||
"priority", candidate.Priority, "users", candidate.UserCount)
|
||||
}
|
||||
return fmt.Sprintf("%d candidate%s queued, %d already known",
|
||||
len(wanted), plural(len(wanted)), skipped), nil
|
||||
}
|
||||
|
||||
// NotePlayback is the live signal, and the strongest one there is: somebody is watching this
|
||||
// episode now, and if there is no marker they will reach the credits without a button.
|
||||
//
|
||||
// It does not scan immediately. A Play press is not yet a viewing — the launcher makes it
|
||||
// trivially easy to start something and change your mind — so the candidate is held for
|
||||
// livePlaybackDelay and raised only if the playback is still going. That is the difference
|
||||
// between a subsystem that reads a file per curious button press and one that reads a file
|
||||
// per episode actually watched.
|
||||
func (s *Service) NotePlayback(ctx context.Context, itemID string) {
|
||||
if s == nil || itemID == "" {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
if _, waiting := s.pending[itemID]; waiting {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
// Detached from the request context deliberately: the HTTP request that reported the
|
||||
// playback is over in milliseconds, and hanging the timer off it would cancel every
|
||||
// delayed scan the instant it returned.
|
||||
timerCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
|
||||
s.pending[itemID] = &pending{
|
||||
candidate: Candidate{
|
||||
ItemID: itemID,
|
||||
Priority: PriorityLive,
|
||||
Reason: ReasonLivePlayback,
|
||||
UserCount: 1,
|
||||
LastViewed: time.Now().UTC(),
|
||||
},
|
||||
due: time.Now().Add(s.liveDelay),
|
||||
cancel: cancel,
|
||||
}
|
||||
delay := s.liveDelay
|
||||
s.mu.Unlock()
|
||||
|
||||
go func() {
|
||||
defer cancel()
|
||||
select {
|
||||
case <-timerCtx.Done():
|
||||
return
|
||||
case <-time.After(delay):
|
||||
}
|
||||
s.mu.Lock()
|
||||
entry, waiting := s.pending[itemID]
|
||||
delete(s.pending, itemID)
|
||||
s.mu.Unlock()
|
||||
if !waiting {
|
||||
return
|
||||
}
|
||||
// One last cache check before queueing. In the settled case the marker was already
|
||||
// there and this costs one indexed read instead of a queue slot.
|
||||
if known, err := s.known(timerCtx, itemID); err == nil && known {
|
||||
return
|
||||
}
|
||||
if s.queue.Push(entry.candidate) {
|
||||
s.log.Debug("credits_candidate",
|
||||
"item", itemID, "reason", ReasonLivePlayback, "priority", PriorityLive)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// AbandonPlayback withdraws a playback that stopped before its delay elapsed. This is what
|
||||
// makes the delay worth having: without it the timer would fire regardless and the episode
|
||||
// somebody looked at for ten seconds would be scanned anyway.
|
||||
func (s *Service) AbandonPlayback(itemID string) {
|
||||
if s == nil || itemID == "" {
|
||||
return
|
||||
}
|
||||
s.mu.Lock()
|
||||
entry, waiting := s.pending[itemID]
|
||||
delete(s.pending, itemID)
|
||||
s.mu.Unlock()
|
||||
if waiting && entry.cancel != nil {
|
||||
entry.cancel()
|
||||
}
|
||||
}
|
||||
|
||||
// Run is the worker. One goroutine, for the whole gateway, for ever.
|
||||
//
|
||||
// A single worker is not a placeholder for a pool. Concurrent scans would multiply exactly
|
||||
// the two costs this package is built to minimise — disk reads and decoder CPU — on a machine
|
||||
// whose primary job is streaming video to televisions. Depth here buys nothing: the queue is
|
||||
// a handful of episodes, and there are hours before anybody reaches them.
|
||||
func (s *Service) Run(ctx context.Context) {
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
candidate, found := s.queue.Claim()
|
||||
if !found {
|
||||
if !sleep(ctx, idlePoll) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
// Live playback goes ahead regardless of load: somebody is waiting, and the scan is
|
||||
// a couple of minutes of ranged reads against a file the server is already serving.
|
||||
// Everything else stands down.
|
||||
if candidate.Priority < PriorityLive && s.busy(ctx) {
|
||||
s.queue.Release(candidate.ItemID)
|
||||
s.queue.Push(candidate)
|
||||
s.log.Debug("credits scan deferred; server busy", "item", candidate.ItemID)
|
||||
if !sleep(ctx, busyBackoff) {
|
||||
return
|
||||
}
|
||||
continue
|
||||
}
|
||||
|
||||
scanCtx, cancel := context.WithTimeout(ctx, scanBudget)
|
||||
detection, stored, err := s.Process(scanCtx, candidate.ItemID)
|
||||
cancel()
|
||||
s.queue.Release(candidate.ItemID)
|
||||
|
||||
switch {
|
||||
case err != nil && errors.Is(err, context.Canceled):
|
||||
return
|
||||
case err != nil:
|
||||
s.log.Debug("credits scan failed",
|
||||
"item", candidate.ItemID, "reason", candidate.Reason, "error", err)
|
||||
case stored:
|
||||
s.log.Info("credits_detected",
|
||||
"item", candidate.ItemID,
|
||||
"marker_ms", detection.StartMs,
|
||||
"confidence", round2(detection.Confidence),
|
||||
"method", detection.Method,
|
||||
"frames", detection.FramesSampled,
|
||||
"duration_ms", detection.Elapsed.Milliseconds())
|
||||
default:
|
||||
s.log.Debug("credits not detected",
|
||||
"item", candidate.ItemID, "frames", detection.FramesSampled)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Process is one candidate, start to finish. Single-flighted on the media version, so the
|
||||
// four ways an episode can be asked for — live playback, a predicted next, another viewer's
|
||||
// prediction, a refresh cycle — collapse into one scan.
|
||||
//
|
||||
// Returns whether anything was stored, which is not the same as whether anything was found:
|
||||
// a detection below the confidence threshold, or one too close to an existing marker to be
|
||||
// worth a write, is a successful scan that deliberately produces no database activity.
|
||||
func (s *Service) Process(ctx context.Context, itemID string) (Detection, bool, error) {
|
||||
resolved, err := s.resolver.Resolve(ctx, itemID)
|
||||
if err != nil {
|
||||
return Detection{}, false, err
|
||||
}
|
||||
fingerprint := Fingerprint(resolved.Version)
|
||||
|
||||
result, err := s.flight.Do(itemID+":"+fingerprint, func() (any, error) {
|
||||
detection, stored, err := s.process(ctx, itemID, fingerprint, resolved)
|
||||
return processResult{detection: detection, stored: stored}, err
|
||||
})
|
||||
outcome, _ := result.(processResult)
|
||||
return outcome.detection, outcome.stored, err
|
||||
}
|
||||
|
||||
type processResult struct {
|
||||
detection Detection
|
||||
stored bool
|
||||
}
|
||||
|
||||
func (s *Service) process(
|
||||
ctx context.Context, itemID, fingerprint string, resolved ResolvedMedia,
|
||||
) (Detection, bool, error) {
|
||||
// Emby already knows. Nothing to do, and nothing to store: the live path reads Emby's
|
||||
// chapters directly, so duplicating the answer here would be a row that could only ever
|
||||
// go stale.
|
||||
if resolved.EmbyCreditsMs > 0 {
|
||||
return Detection{Found: true, StartMs: resolved.EmbyCreditsMs, Method: MethodEmby}, false, nil
|
||||
}
|
||||
existing, found, err := s.repo.GetMarker(ctx, itemID, fingerprint)
|
||||
if err != nil {
|
||||
return Detection{}, false, err
|
||||
}
|
||||
// The common path, and the one everything is optimised for: already decided, no media
|
||||
// access, no decoder, no write.
|
||||
if found && existing.Confidence >= ConfidenceThreshold {
|
||||
return Detection{
|
||||
Found: true, StartMs: existing.CreditsStartMs,
|
||||
Confidence: existing.Confidence, Method: existing.DetectionMethod,
|
||||
}, false, nil
|
||||
}
|
||||
if resolved.RuntimeMs <= 0 {
|
||||
return Detection{}, false, nil
|
||||
}
|
||||
|
||||
// Behaviour first, because it is free. On a well-watched show it can settle the question
|
||||
// without any media being opened at all, and where it cannot it still narrows the window
|
||||
// the decoder has to read.
|
||||
evidence := s.evidence(ctx, itemID, resolved.RuntimeMs)
|
||||
|
||||
var visual Detection
|
||||
if !evidence.StandaloneMarker() || !evidence.Usable() {
|
||||
history, runtimes := s.seasonHistory(ctx, resolved)
|
||||
window := NarrowWindow(resolved.RuntimeMs, history, runtimes, evidence)
|
||||
visual, err = s.detector.Detect(ctx, MediaInfo{
|
||||
URL: resolved.URL, RuntimeMs: resolved.RuntimeMs, Window: window,
|
||||
})
|
||||
if err != nil {
|
||||
if errors.Is(err, ErrNoFFmpeg) {
|
||||
// Run on behaviour alone rather than failing. An image without a decoder is
|
||||
// a deliberate deployment, not a fault.
|
||||
s.log.Debug("visual credits detection unavailable")
|
||||
} else {
|
||||
return Detection{}, false, err
|
||||
}
|
||||
}
|
||||
visual.Window = window
|
||||
}
|
||||
|
||||
detection, acceptable := Combine(visual, evidence)
|
||||
if !acceptable {
|
||||
return visual, false, nil
|
||||
}
|
||||
if found && !ShouldRewrite(existing, detection) {
|
||||
// Stability: the same answer, or a marginally different one, is not worth a write.
|
||||
return detection, false, nil
|
||||
}
|
||||
|
||||
// A fingerprint too weak to detect a file replacement is one whose marker could outlive
|
||||
// the file it describes. Better to keep re-deriving it than to store something nothing
|
||||
// will ever invalidate.
|
||||
if resolved.Version.Weak() {
|
||||
s.log.Debug("credits marker withheld; weak fingerprint", "item", itemID)
|
||||
return detection, false, nil
|
||||
}
|
||||
|
||||
now := time.Now().UTC()
|
||||
marker := Marker{
|
||||
ItemID: itemID,
|
||||
MediaFingerprint: fingerprint,
|
||||
CreditsStartMs: detection.StartMs,
|
||||
Confidence: detection.Confidence,
|
||||
DetectionMethod: detection.Method,
|
||||
SeriesID: resolved.SeriesID,
|
||||
Season: resolved.Season,
|
||||
CreatedAt: now,
|
||||
UpdatedAt: now,
|
||||
}
|
||||
if err := s.repo.SaveMarker(ctx, marker); err != nil {
|
||||
return detection, false, err
|
||||
}
|
||||
return detection, true, nil
|
||||
}
|
||||
|
||||
// evidence reads the household's stops. Failures are swallowed: behaviour is an optimisation
|
||||
// and an enhancement, never a precondition.
|
||||
func (s *Service) evidence(ctx context.Context, itemID string, runtimeMs int64) BehaviourEvidence {
|
||||
if s.behaviour == nil {
|
||||
return BehaviourEvidence{}
|
||||
}
|
||||
stops, err := s.behaviour.Stops(ctx, itemID)
|
||||
if err != nil {
|
||||
s.log.Debug("behavioural evidence unavailable", "item", itemID, "error", err)
|
||||
return BehaviourEvidence{}
|
||||
}
|
||||
return AnalyseStops(stops, runtimeMs)
|
||||
}
|
||||
|
||||
// seasonHistory reads what is already known about this episode's neighbours, and the runtime
|
||||
// of each so the tail offsets can be compared. Failures degrade to a generic tail window.
|
||||
func (s *Service) seasonHistory(
|
||||
ctx context.Context, resolved ResolvedMedia,
|
||||
) ([]Marker, func(Marker) int64) {
|
||||
if resolved.SeriesID == "" || resolved.Season <= 0 {
|
||||
return nil, nil
|
||||
}
|
||||
const historyLimit = 6
|
||||
history, err := s.repo.SeasonMarkers(ctx, resolved.SeriesID, resolved.Season, historyLimit)
|
||||
if err != nil || len(history) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
// The neighbours' runtimes come from resolving them, which would be a request each. The
|
||||
// episodes of one season are within a minute or two of each other, so this episode's own
|
||||
// runtime is a good enough stand-in — and being slightly wrong here only widens or
|
||||
// narrows a window that carries a ninety-second margin either way.
|
||||
runtime := resolved.RuntimeMs
|
||||
return history, func(Marker) int64 { return runtime }
|
||||
}
|
||||
|
||||
// known is the cache check in its cheapest form: does a marker exist for the current version
|
||||
// of this item. One resolve, one indexed read.
|
||||
func (s *Service) known(ctx context.Context, itemID string) (bool, error) {
|
||||
resolved, err := s.resolver.Resolve(ctx, itemID)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
if resolved.EmbyCreditsMs > 0 {
|
||||
return true, nil
|
||||
}
|
||||
marker, found, err := s.repo.GetMarker(ctx, itemID, Fingerprint(resolved.Version))
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return found && marker.Confidence >= ConfidenceThreshold, nil
|
||||
}
|
||||
|
||||
func (s *Service) busy(ctx context.Context) bool {
|
||||
return s.load != nil && s.load.Busy(ctx)
|
||||
}
|
||||
|
||||
// QueueDepth is what the console reads.
|
||||
func (s *Service) QueueDepth() int { return s.queue.Len() }
|
||||
|
||||
// Pending is the queue in worker order, for the admin console.
|
||||
func (s *Service) Pending() []Candidate { return s.queue.Snapshot() }
|
||||
|
||||
func sleep(ctx context.Context, duration time.Duration) bool {
|
||||
timer := time.NewTimer(duration)
|
||||
defer timer.Stop()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return false
|
||||
case <-timer.C:
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
func plural(count int) string {
|
||||
if count == 1 {
|
||||
return ""
|
||||
}
|
||||
return "s"
|
||||
}
|
||||
|
||||
func round2(value float64) float64 {
|
||||
return float64(int(value*100+0.5)) / 100
|
||||
}
|
||||
Reference in New Issue
Block a user