Documentation
¶
Overview ¶
Package stream is Ether's stream engine: it turns an NZB into a seekable core.MediaSource without writing the full file anywhere.
Open parses the NZB, picks the main video (a plain post, or the file inside a store-mode RAR/7z set or a plain split), and builds a byte-offset index from the yEnc part headers so any byte range maps to exact articles. ReadAt fetches those articles in parallel through the core.ArticleFetcher, reads ahead of sequential readers, coalesces concurrent requests for the same article, keeps a bounded in-memory hot buffer and spills decoded segments to the core.SegmentCache.
Open touches as few articles as possible: the first article of the video (plain posts), or the first article of the first, second and last volumes of an archive set, assuming the uniform volume layout that archivers produce. That assumption is checked against the archive's own totals at Open and verified per volume, lazily, before any byte of that volume is served; a mismatch rebuilds the layout from every volume's headers.
Index ¶
- Variables
- func IsRejection(err error) bool
- type Engine
- func (e *Engine) Check(ctx context.Context, nzbData []byte) (healthy bool, err error)
- func (e *Engine) CheckRelease(ctx context.Context, releaseID string, nzbData []byte) (healthy bool, err error)
- func (e *Engine) Close() error
- func (e *Engine) DurablePrefetch(ctx context.Context, id string, payload []byte, off, n int64) error
- func (e *Engine) HealthTimings() HealthTrace
- func (e *Engine) InitRepair(ctx context.Context, releaseID string, nzbData []byte) error
- func (e *Engine) Open(ctx context.Context, releaseID string, nzbData []byte) (core.MediaSource, error)
- func (e *Engine) PostedFileInfo(ctx context.Context, id string, payload []byte, index int) (core.ArticlePart, error)
- func (e *Engine) Prefetch(ctx context.Context, releaseID string, nzbData []byte, offset, length int64) error
- func (e *Engine) ReadPostedFile(ctx context.Context, releaseID string, nzbData []byte, fileIndex int, ...) (int, error)
- func (e *Engine) ReadPostedRange(ctx context.Context, id string, payload []byte, index int, off int64, ...) (int, error)
- func (e *Engine) Stats() Stats
- func (e *Engine) UpdateRepair(cfg config.Repair)
- type HealthTrace
- type Options
- type Stats
Constants ¶
This section is empty.
Variables ¶
var ( // ErrNoPlayableFile: the release has no video file the engine can serve // (only samples/extras, unsupported containers, or nothing recognisable). ErrNoPlayableFile = errors.New("stream: no playable video file in release") // ErrCompressed: the video sits in a compressed (non-store) archive. ErrCompressed = archive.ErrCompressed // ErrEncrypted: the archive is password protected (headers or data). ErrEncrypted = archive.ErrEncrypted // ErrSolid: the archive is solid, so a file cannot be read in isolation. ErrSolid = archive.ErrSolid // ErrUnsupported: an archive feature the engine cannot stream. ErrUnsupported = archive.ErrUnsupported // ErrCorrupt: archive headers or the segment layout are inconsistent. ErrCorrupt = archive.ErrCorrupt // ErrInvalidNZB: the NZB could not be parsed. ErrInvalidNZB = nzb.ErrInvalid // ErrIncomplete: the NZB itself lacks segments or volumes of the main // file. It wraps core.ErrArticleMissing (see incompleteError). ErrIncomplete = errors.New("stream: release is incomplete") // ErrPar2Mismatch: the content does not match the checksums in the // release's own PAR2 index (a wrong or damaged upload that yEnc CRCs // cannot catch). It wraps ErrCorrupt, so it is a rejection. ErrPar2Mismatch = fmt.Errorf("stream: content does not match the release's par2 checksums: %w", archive.ErrCorrupt) // ErrClosed is returned by ReadAt after Close. ErrClosed = errors.New("stream: source closed") )
Typed errors returned by Open, Prefetch and Check. They are re-exported from the packages that detect them so callers (the resolver, playback) only need to import stream. Errors are wrapped with context; use errors.Is.
Functions ¶
func IsRejection ¶
IsRejection reports whether err means the release itself can never be streamed (as opposed to a transient network or context error), so a resolver should move on and may mark the release bad.
Types ¶
type Engine ¶
type Engine struct {
// contains filtered or unexported fields
}
Engine implements core.Engine and core.HealthChecker.
func New ¶
func New(fetcher core.ArticleFetcher, cache core.SegmentCache, opts Options, logger *slog.Logger) *Engine
New returns an engine. cache may be nil (memory only). logger may be nil.
func (*Engine) Check ¶
Check implements core.HealthChecker. It parses the NZB, inspects the archive headers of the main video (rejecting compressed, solid and unsupported encryption and releases without a playable file) and STATs a spread sample of the main video's articles in parallel: the first, the last and evenly spaced ones across every volume.
healthy=false comes with an error saying why: one wrapping core.ErrArticleMissing (sampled articles are gone or the NZB is incomplete), one of the package's rejection errors (see IsRejection), or a transient error (network, context).
When the release carries a small PAR2 index it is read alongside, and the first article of the main video (or first archive volume) is checked against the recorded length, first-16 KiB MD5 and covered slice checksums; a mismatch is reported as unhealthy with an error wrapping ErrPar2Mismatch (a rejection). A missing or unusable PAR2 is ignored.
The index built here is kept, so an Open of the same NZB right after a successful Check does not repeat the header fetches.
func (*Engine) CheckRelease ¶
func (*Engine) Close ¶
Close cancels every in-flight fetch, makes open sources fail and waits for background work (fetches, cache writes) to stop.
func (*Engine) DurablePrefetch ¶ added in v0.2.4
func (e *Engine) DurablePrefetch(ctx context.Context, id string, payload []byte, off, n int64) error
DurablePrefetch is for managed preloading: unlike best-effort startup warming, success requires every segment covering the requested span to remain on disk.
func (*Engine) HealthTimings ¶
func (e *Engine) HealthTimings() HealthTrace
HealthTimings returns timing for the most recent Check call.
func (*Engine) InitRepair ¶ added in v0.2.8
InitRepair primes bounded metadata and summaries without repairing anything. It is also used by the explicit live benchmark. Normal Open calls it with a short context; missing PAR2 remains optional.
func (*Engine) Open ¶
func (e *Engine) Open(ctx context.Context, releaseID string, nzbData []byte) (core.MediaSource, error)
Open builds the byte-offset index for the main video file of an NZB.
func (*Engine) PostedFileInfo ¶ added in v0.2.8
func (e *Engine) PostedFileInfo(ctx context.Context, id string, payload []byte, index int) (core.ArticlePart, error)
PostedFileInfo reads a header through the shared cache and repair path. Bulk diagnostics must not refetch the same first article through a separate client.
func (*Engine) Prefetch ¶
func (e *Engine) Prefetch(ctx context.Context, releaseID string, nzbData []byte, offset, length int64) error
Prefetch warms [offset, offset+length) of the main video into the segment cache at background priority.
func (*Engine) ReadPostedFile ¶ added in v0.2.8
func (e *Engine) ReadPostedFile(ctx context.Context, releaseID string, nzbData []byte, fileIndex int, off int64, dst []byte) (int, error)
ReadPostedFile exposes the same verified article/repair path for diagnostics and full-download comparisons, before archive extraction. fileIndex is the NZB's file index. It does not create a Plex session or write watch history.
func (*Engine) ReadPostedRange ¶ added in v0.2.8
func (e *Engine) ReadPostedRange(ctx context.Context, id string, payload []byte, index int, off int64, dst []byte) (int, error)
ReadPostedRange is the explicit bulk reader entry point. Its window is independent of the streaming player's demand/read-ahead window.
func (*Engine) UpdateRepair ¶ added in v0.2.8
UpdateRepair applies owner settings to new work without mutating snapshots held by concurrent readers. Existing dependency reads observe disablement.
type HealthTrace ¶
type HealthTrace struct {
Warm time.Duration `json:"warm"`
Parse time.Duration `json:"parse"`
Plan time.Duration `json:"plan"`
STAT time.Duration `json:"stat"`
Samples int `json:"samples"`
// BODY probe results: the mid-file segments verified with a real
// BODY (yEnc header) read, since STAT 223 does not prove a BODY
// would not be taken down (451).
Body time.Duration `json:"body"`
BodyProbes int `json:"bodyProbes"`
BodyOK int `json:"bodyOk"`
// FailFast is set when the check bailed out on the first confirmed
// missing article instead of waiting for every sample; FailReason
// says which probe decided it.
FailFast bool `json:"failFast"`
FailReason string `json:"failReason,omitempty"`
}
HealthTrace describes the most recent release validation phases.
type Options ¶
type Options struct {
Repair config.Repair
// HotBufferBytes bounds decoded segments kept in memory across all
// sources (default 256 MiB).
HotBufferBytes int64
// ReadAheadMin is the initial read-ahead window of a sequential reader
// (default 4 MiB); it grows with the bytes read up to ReadAheadMax
// (default 64 MiB).
ReadAheadMin int64
ReadAheadMax int64
// ReadAheadParallel caps concurrent read-ahead fetches per source
// (default 8).
ReadAheadParallel int
// ReadParallel caps concurrent fetches serving one ReadAt (default 16).
ReadParallel int
// PrefetchParallel caps concurrent fetches of one Prefetch call
// (default 4).
PrefetchParallel int
// HealthSamples is how many articles Check STATs (default 12, from
// config Resolver.HealthSamples).
HealthSamples int
// HealthParallel caps concurrent STATs of one Check (default 16).
HealthParallel int
// OpenParallel caps concurrent header fetches while opening (default 16).
OpenParallel int
// ReadTimeout bounds one ReadAt (default 90 s); ReadAt has no context.
ReadTimeout time.Duration
// MaxReleases is how many parsed release indexes are kept for reuse
// (default 32). Releases with open sources are never dropped.
MaxReleases int
// Retries is how often a failed article fetch (other than missing or
// cancelled) is retried (default 2).
Retries int
// MaxProbes bounds articles fetched to identify obfuscated files
// (default 64).
MaxProbes int
}
Options tune the engine. Zero values take the defaults noted.
func OptionsFromConfig ¶
OptionsFromConfig derives options from the owner settings.
type Stats ¶
type Stats struct {
RepairLastReason string
RepairAttempts int64
RepairSuccesses int64
RepairFailures int64
RepairInputBytes int64
RepairSummaryBytes int64
Fetches int64 // articles fetched from Usenet
FetchedBytes int64 // decoded bytes fetched from Usenet
HotHits int64 // segments served from memory
CacheHits int64 // segments served from the disk cache
Missing int64 // articles missing on every provider
Coalesced int64 // requests that joined an in-flight fetch
HotBytes int64 // bytes currently in the hot buffer
OpenReleases int // release indexes held
ActiveSources int64
// PAR2 (see par2.go): index files parsed, unusable (fetched but not
// parseable) and skipped as oversize; obfuscated files named from PAR2;
// head probes of obfuscated files avoided thanks to it; first volumes
// verified against it and found not matching.
Par2Loaded int64
Par2Failed int64
Par2Skipped int64
Par2Named int64
Par2ProbesSaved int64
Par2Verified int64
Par2Mismatches int64
}
Stats are engine counters for diagnostics.