Help us shape the future of Go! Take the 2026 Go Developer Survey

playback

package
v0.2.2 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Oct 20, 2026 License: GPL-3.0 Imports: 32 Imported by: 0

Documentation

Overview

Package playback implements core.Playback: it binds a user, a client and a title to a resolved, open release, enforces stream limits, serves the source to clients (direct play) and to ffmpeg (a loopback HTTP server), fails over to another release mid-stream when articles go missing, and protects the releases of live and in-progress titles in the segment cache.

Sessions are internal and mutable; every PlaySession handed out is a snapshot. Read media bytes through ServeRange or InputURL rather than a snapshot's Source, which is closed when a failover swaps it.

Index

Constants

View Source 
const (
	PinReasonLive       = "playback:live"
	PinReasonInProgress = "playback:inprogress"
)

Pin reasons used with core.SegmentCache.

Variables

This section is empty.

Functions

func AlignTime 

func AlignTime(at, from, to time.Duration) time.Duration

AlignTime maps a timestamp in a release of duration from to the matching timestamp in a release of duration to.

Releases of the same cut share timestamps, so the time is kept as is. The one systematic difference that can be detected from durations alone is PAL speedup (25 fps vs 23.976 fps, a 4.3% runtime difference): then the time is scaled by the ratio. The result is clamped to [0, to] when to is known. Different cuts or intros cannot be detected from durations alone.

func EstimateInfo 

func EstimateInfo(t *core.Title, caps core.ClientCaps, p core.QualityProfile) core.MediaInfo

EstimateInfo builds the estimated MediaInfo for a title, client and profile: Matroska, H.264 (HEVC at 4K or when the client lacks H.264) at the profile height, a typical bitrate for that height kept inside the profile's size bounds, and E-AC-3 5.1 or AAC stereo audio.

func MapOffset 

func MapOffset(off int64, from, to TimeMap) int64

MapOffset maps a byte offset in one release to the byte offset showing the same moment in another release: offset -> time (from), align, time -> offset (to).

func ServerStreamLimit 

func ServerStreamLimit(cfg config.Config, connsPerStream int) int

ServerStreamLimit is the server-wide stream cap: Server.MaxStreams, or the total connections of enabled primary (non-backfill) Usenet providers divided by connsPerStream, at least 1. 0 means unlimited (no providers).

func UserStreamLimit 

func UserStreamLimit(cfg config.Config, u *core.User) int

UserStreamLimit is the number of concurrent streams a user may hold: the user's own limit, else the server default; 0 means unlimited.

Types

type Deps 

type Deps struct {
	AugmentMedia func(context.Context, *core.Title, *core.User, *core.Resolution, core.MediaSource, *core.MediaInfo) // synchronous enrichment before publishing a generation
	Resolver     core.Resolver
	Engine       core.Engine
	Transcoder   core.Transcoder
	Catalog      core.Catalog
	// Cache, when set, receives pins for live and in-progress releases.
	Cache core.SegmentCache
	// State, when set, tells which titles are in progress (cache protection).
	State core.State
	// SelectProfile picks the quality profile for a client and user. It is
	// implemented by the resolver package.
	SelectProfile func(caps core.ClientCaps, user *core.User) core.QualityProfile
	// HLSLookup finds the HLS session fed by a playback session, for
	// mid-stream failover. Defaults to Transcoder.Session(sessionID).
	HLSLookup func(sessionID string) (core.HLSSession, bool)
	// FetchNZB optionally downloads an alternate's NZB so failover can try
	// Resolution.Alternates directly; without it failover asks the resolver
	// again with the tried releases excluded.
	FetchNZB func(ctx context.Context, r core.Release) ([]byte, error)
	// Aligner, when set, maps failover timestamps between releases by
	// dialogue timing; it gets a short timeout and any miss falls back to
	// duration-based AlignTime.
	Aligner core.TimelineAligner
	// DB holds the playback_resume table; nil disables resume records.
	DB     *sql.DB
	Logger *slog.Logger
	// Cluster, when set (clustered nodes only; requires DB), mirrors this
	// node's sessions into playback_live_sessions and applies stream limits
	// across the cluster. Nil keeps everything in this process.
	Cluster clusterbus.Node
	// DescribeGroup optionally describes a live TV viewing unit reserved
	// with BeginChannel for other nodes' status pages.
	DescribeGroup func(group string) *core.PlaySession
	// Availability, when set (clustered nodes), is the node's view of
	// database reachability: resume records the database cannot take are
	// held and written later, and admission skips the cluster-wide check
	// while the database is down (failing open) instead of waiting on it.
	Availability *store.Availability
}

Deps are the components playback orchestrates. Resolver, Engine, Transcoder, Catalog and SelectProfile are required.

type Event 

type Event struct {
	Kind    string
	Session *core.PlaySession
	At      time.Time
	Error   string
}

Event contains a playback snapshot; subscribers must not retain live Sources.

type IndexPoint 

type IndexPoint struct {
	Offset int64
	Time   time.Duration
}

IndexPoint ties a byte offset in a media file to a presentation timestamp, e.g. a Matroska cue point or an MP4 sync sample. Points let TimeMap follow variable bitrate content instead of assuming a constant bitrate.

type Manager 

type Manager struct {
	// contains filtered or unexported fields
}

Manager implements core.Playback.

func New 

func New(deps Deps, cfg config.Config, opts Options) (*Manager, error)

New starts a Manager: the loopback source server and its background loops (idle expiry, pin reconciliation). Call Close to stop it.

func (*Manager) ActiveFor 

func (m *Manager) ActiveFor(userID int64, key core.RatingKey) bool

ActiveFor reports whether the user has a live session for the title.

func (*Manager) BeginChannel 

func (m *Manager) BeginChannel(u *core.User, caps core.ClientCaps, group string) error

BeginChannel reserves a viewing unit in the same transaction as ordinary Prepare. This also counts passthrough IPTV, which has no on-demand source.

func (*Manager) ClientCaps 

func (m *Manager) ClientCaps(userID int64, key core.RatingKey) (caps core.ClientCaps, live bool, ok bool)

ClientCaps returns the capabilities to resolve related titles (the next episode) with: those of the user's live session for key (live=true), else of the user's most recent Prepare.

func (*Manager) Close 

func (m *Manager) Close() error

Close stops every session, the loopback server and background loops.

func (*Manager) Draining 

func (m *Manager) Draining() bool

Draining reports whether SetDraining(true) is in effect.

func (*Manager) EndChannel 

func (m *Manager) EndChannel(group string)

func (*Manager) Estimate 

func (m *Manager) Estimate(_ context.Context, t *core.Title, caps core.ClientCaps) core.MediaInfo

Estimate is the virtual media info for a title before it is resolved, derived from the quality profile chosen for the client and the title's runtime. Estimate has no user, so profile rules keyed on a user do not apply here.

func (*Manager) Failover 

func (m *Manager) Failover(ctx context.Context, sessionID string, at time.Duration) (*core.PlaySession, error)

Failover swaps the session to the next candidate release, keeping the timestamp at. The current release is marked bad. When an HLS session reads from this playback session it is restarted on the new input at the same position. Returns the updated session.

func (*Manager) InputURL 

func (m *Manager) InputURL(id string) string

InputURL is the loopback URL ffmpeg reads the session's current source from; "" for an unknown session. The URL changes on failover.

func (*Manager) InvalidateTitle 

func (m *Manager) InvalidateTitle(key core.RatingKey) int

InvalidateTitle drops the warm (pre-resolved) entries for a title, from both the detail and speculative budgets, so the next Play resolves afresh — after the owner pins, unpins, blocks or unblocks a release, or forgets a title's search. Stored source choices are invalidated too; watch progress and container indexes remain intact. A show or season key also drops its episodes' entries. A warm-up still in flight finishes into nothing. It returns how many warm entries were dropped.

func (*Manager) InvalidateTitleEverywhere 

func (m *Manager) InvalidateTitleEverywhere(key core.RatingKey) int

InvalidateTitleEverywhere is InvalidateTitle on this node and, in a cluster, on every other node. It returns how many local entries were dropped.

func (*Manager) PreResolve 

func (m *Manager) PreResolve(ctx context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps) bool

PreResolve schedules detail-page warming; it is the core.PrewarmDetail form of Prewarm, kept for existing callers.

func (*Manager) Prepare 

func (m *Manager) Prepare(ctx context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps) (*core.PlaySession, error)

Prepare resolves and opens a source for the user and client, enforcing stream limits, or returns the existing session for the same user, client and title. Concurrent calls for the same session share one resolution.

func (*Manager) PrepareChannel 

func (m *Manager) PrepareChannel(ctx context.Context, u *core.User, key core.RatingKey, caps core.ClientCaps, group string) (*core.PlaySession, error)

PrepareChannel is an internal playout operation. One viewing session may overlap at most three programme sources for continuity. The server and user limits count the group once, and ordinary Prepare cannot create groups.

func (*Manager) PrepareVariant 

func (m *Manager) PrepareVariant(ctx context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps, releaseID string) (*core.PlaySession, error)

PrepareVariant implements core.VariantPlayback: like Prepare, but the viewer picked releaseID ("Play Version"), so the resolver is asked to try it first (ResolveRequest.Prefer). An existing session for the same user, client and title on a different release is replaced, and a warm entry is only reused when it holds the picked release. When the picked release is unavailable, resolution falls back to the normal choice.

func (*Manager) Prewarm 

func (m *Manager) Prewarm(_ context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps, opts core.PrewarmOptions) bool

Prewarm implements core.Prewarmer: it resolves, opens, probes (including the container keyframe index) and pre-buffers a title ahead of Play, so the later Prepare for the same user, client and title needs no resolver or probe round trip. Request cancellation does not cancel a warm-up; manager shutdown always does. It neither reserves a live stream nor writes resume/history. False means skipped.

Budgets: two warm-ups in flight server-wide, one per user, six requests/minute/user. A detail warm-up runs 8s, prefetches 2 MiB and is skipped while anything is playing. Trailer/post-play warm-ups may run during playback (DuringPlayback), prefetch 48 MiB by default (hard cap 64 MiB) and get 45s. Prefetch goes through Engine.Prefetch, which runs at core.PriorityPrefetch and yields to live streams.

func (*Manager) ReconcilePins 

func (m *Manager) ReconcilePins(ctx context.Context) error

ReconcilePins makes the cache's in-progress pins match the resume records of titles in progress for any user: each such user's release is pinned, releases no longer in progress are unpinned, and records past Options.ResumeRetention that are neither in progress nor live are deleted. Only titles watched within Cache.InProgressPinMaxAge are pinned, and pins together may hold at most Cache.InProgressPinPercent of the cache budget (most recently watched first), so abandoned half-watched titles cannot hold the cache over budget forever. The first pass also unpins stale pins from a previous run. Resume expiry runs even without a cache.

func (*Manager) Remember 

func (m *Manager) Remember(ctx context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps, res *core.Resolution) error

Remember stores a resolution for a title the user has not played yet (a prefetched next episode) so the eventual Prepare opens the same, already warm release. An existing record for the same profile is kept.

func (*Manager) RemoteSessions 

func (m *Manager) RemoteSessions(ctx context.Context) []*core.PlaySession

RemoteSessions returns the sessions other cluster nodes are serving, as the owner's status page presents them: a live TV viewing unit stands for its programme sources. Each has Node set. Nil on a standalone server.

func (*Manager) RequiresRemux 

func (m *Manager) RequiresRemux(sessionID string) bool

RequiresRemux reports whether the session must be (re)started as a zero-encode remux instead of direct play (see core.RemuxRequirer).

func (*Manager) ResumeFor 

func (m *Manager) ResumeFor(ctx context.Context, user *core.User, key core.RatingKey, caps core.ClientCaps) (*ResumePoint, error)

ResumeFor is ResumePoint restricted to a record resolved for the quality profile this client and user would get, i.e. the release a Prepare from that client would open.

func (*Manager) ResumeOutboxStatus 

func (m *Manager) ResumeOutboxStatus() (st store.OutboxStatus, ok bool)

ResumeOutboxStatus reports resume records held through a database outage; ok is false when records are written synchronously.

func (*Manager) ResumePoint 

func (m *Manager) ResumePoint(ctx context.Context, userID int64, key core.RatingKey) (*ResumePoint, error)

ResumePoint returns the release that last served (or was prefetched for) a user's title, with its NZB; nil when there is none.

func (*Manager) ServeRange 

func (m *Manager) ServeRange(w http.ResponseWriter, r *http.Request, sessionID string)

ServeRange serves the session's media to a client (direct play) with Range and If-Range support. Offsets are in the coordinates of the file the session started with; after a failover they are mapped by timestamp into the current release. When articles go missing mid-response the session fails over: an MPEG-TS response continues from the new release at the same moment, any other container ends the response so the client re-requests (a brief rebuffer) and that request is served from the new release.

func (*Manager) Session 

func (m *Manager) Session(id string) (*core.PlaySession, bool)

Session returns a snapshot of a live session.

func (*Manager) SessionFor 

func (m *Manager) SessionFor(userID int64, clientID string, key core.RatingKey) (*core.PlaySession, bool)

SessionFor finds the live session of a client for a title.

func (*Manager) Sessions 

func (m *Manager) Sessions() []*core.PlaySession

Sessions returns snapshots of every live session, oldest first.

func (*Manager) SetConfig 

func (m *Manager) SetConfig(cfg config.Config)

SetConfig applies a new config snapshot (stream limits).

func (*Manager) SetDecision 

func (m *Manager) SetDecision(id string, d core.Decision)

SetDecision records the transcode decision the front end made for a session, shown in Sessions(). A remux/transcode decision satisfies a pending RequiresRemux.

func (*Manager) SetDraining 

func (m *Manager) SetDraining(draining bool)

SetDraining stops (true) or resumes (false) admitting new playback sessions on this node. A draining node — about to restart for an update — keeps serving its existing sessions and their seeks, but a new Prepare or live TV tune fails with core.ErrStreamLimit so the client retries, reaching another node through another published connection.

func (*Manager) Stop 

func (m *Manager) Stop(id string)

Stop ends a session: its source is closed once reads finish, its live pin is dropped and its HLS session, if any, is stopped.

func (*Manager) Subscribe 

func (m *Manager) Subscribe(fn func(Event)) func()

Subscribe observes prepared sessions, decisions, failovers, errors and stops. It runs after session locks have been released. Unsubscribe during shutdown.

func (*Manager) Timeline 

func (m *Manager) Timeline(userID int64, key core.RatingKey, state string)

Timeline renews a viewer's lease while playing or paused. Buffering alone cannot preserve a failed stream; stopped ends it after Options.StoppedGrace.

type Options 

type Options struct {
	// DataDir is where NZBs of resumable releases are kept (DataDir/playback/nzb).
	DataDir string
	// IdleTimeout ends sessions without viewer activity (default 2m).
	IdleTimeout time.Duration
	// WriteStallTimeout fails a direct-play write that makes no progress
	// for this long, so a client that stopped reading releases its source
	// (default IdleTimeout: a paused player is not cut off before its
	// session would be reaped anyway).
	WriteStallTimeout time.Duration
	// StoppedGrace ends a session this long after a "stopped" timeline (default 1m).
	StoppedGrace time.Duration
	// PrepareTimeout bounds resolve+open+probe (default 90s).
	PrepareTimeout time.Duration
	// ProbeTimeout bounds one Transcoder.Probe (default 30s).
	ProbeTimeout time.Duration
	// FailoverTimeout bounds a background (loopback-triggered) failover (default 90s).
	FailoverTimeout time.Duration
	// ConnectionsPerStream derives the server-wide stream cap from the total
	// Usenet connections when Server.MaxStreams is 0 (default 8).
	ConnectionsPerStream int
	// MaxAttempts is how many releases one Prepare or failover tries (default 4).
	MaxAttempts int
	// MaxFailovers per session before giving up (default 6).
	MaxFailovers int
	// ChunkSize is the copy buffer size (default 1 MiB); ReadAhead the number
	// of chunks read ahead of the writer (default 4).
	ChunkSize int
	ReadAhead int
	// PinInterval is how often in-progress pins are reconciled (default 5m).
	PinInterval time.Duration
	// ResumeRetention drops resume records of titles no longer in progress
	// after this long (default 30 days).
	ResumeRetention time.Duration
	// IndexTimeout bounds the background container keyframe-index read
	// after a generation is published (default 60s). Play never waits on it.
	IndexTimeout time.Duration
	// SpeedCheckInterval is how often source fetch speed is compared with
	// media bitrate (default 2s). SpeedGrace after a start, seek or
	// failover is ignored (default 20s). A session whose fetch rate stays
	// below SpeedThreshold (default 0.8) of its media bitrate for
	// SpeedSlowChecks consecutive checks (default 3) down-fails over, at
	// most MaxDownFailovers times (default 2; negative disables).
	SpeedCheckInterval time.Duration
	SpeedGrace         time.Duration
	SpeedThreshold     float64
	SpeedSlowChecks    int
	MaxDownFailovers   int
	// LoopbackAddr is the source server's listen address (default 127.0.0.1:0).
	LoopbackAddr string
	// SessionHeartbeat is how often a clustered node refreshes its shared
	// session rows (default 10s); SessionTTL is how long a row counts
	// without a refresh (default 45s).
	SessionHeartbeat time.Duration
	SessionTTL       time.Duration
	// Now is the clock (tests).
	Now func() time.Time
}

Options tune the manager. Zero values take the defaults noted.

type ResumePoint 

type ResumePoint struct {
	UserID    int64
	Key       core.RatingKey
	Release   core.Release
	NZB       []byte
	Profile   string        // quality profile it was resolved for
	Size      int64         // media file size, 0 if never opened
	Duration  time.Duration // probed duration, 0 if unknown
	Keyframes []core.KeyframePoint
	UpdatedAt time.Time
}

ResumePoint is the release that last served a user's title, so a resume (or a prefetched next episode) opens the same release and hits cached segments without asking the indexer again.

func (*ResumePoint) ReleaseID 

func (p *ResumePoint) ReleaseID() string

ReleaseID is the stable release identifier.

type TimeMap 

type TimeMap struct {
	Size     int64
	Duration time.Duration
	// Points are optional index entries sorted by Offset (and Time).
	Points []IndexPoint
}

TimeMap converts between byte offsets and timestamps inside one media file.

Without Points it assumes a constant bitrate: offset/Size == time/Duration. With Points it interpolates linearly between the surrounding points, with (0, 0) and (Size, Duration) as implicit end anchors.

func TimeMapFor 

func TimeMapFor(info core.MediaInfo) TimeMap

TimeMapFor builds the TimeMap of a probed source from its MediaInfo, including the container keyframe index when one was read, so callers outside playback (resume prefetch, trickplay) map with the same points.

func (TimeMap) KeyframeOffsetAt 

func (m TimeMap) KeyframeOffsetAt(t time.Duration) int64

KeyframeOffsetAt is OffsetAt snapped down to the keyframe at or before t — the offset a player or ffmpeg seek will actually start reading from — when index points are known; plain OffsetAt otherwise.

func (TimeMap) OffsetAt 

func (m TimeMap) OffsetAt(t time.Duration) int64

OffsetAt maps a timestamp to a byte offset, clamped to [0, Size].

func (TimeMap) TimeAt 

func (m TimeMap) TimeAt(off int64) time.Duration

TimeAt maps a byte offset to a timestamp, clamped to [0, Duration].

func (TimeMap) Valid 

func (m TimeMap) Valid() bool

Valid reports whether the map has enough information to convert.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.