* docs(playback): add v3 neutral-contract finalization plan Supersedes the wire-contract sections of the 2026-07-12 v3 plan: server-owned attempt keys, delivery-keyed negotiation without Media3 engine names, tiered capability evidence, neutral device/output context, track/quality replan operations, audio-only planning, and coordinated no-back-compat rollout across server, Android, Apple, and web. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(playback): make v3 attempt keys server-owned and replace engines with deliveries Contract core of the platform-neutral v3 finalization (plan sections 3.1 and 3.2), breaking on purpose — v3 is dark and all clients move together: - Every PlanV3 now carries plan_attempt_key, an opaque server-computed token clients store and echo in attempted_plan_keys; ReplanRequestV3 gains bounded local_mutations that the replan handler folds into the failed plan's key. Clients never hash anything. - KotlinName() is deleted from DeliveryV3, StreamProtocolV3 and SubtitleModeV3; the attempt-key canonical string now uses lowercase wire tokens, and PlanRecipeVersionV3 bumps to v3.3 so no key or plan ID computed under the old canonicalization can collide. - EngineV3 leaves the wire: ClientPlaybackContextV3.Engines (media3_*) becomes Deliveries keyed original_http|progressive|hls, with EngineCapabilityV3 renamed DeliveryCapabilityV3. PlanV3.Engine is removed; the planner, subtitle policy and quirk registry re-key on delivery class, and the media3_only feature token is deleted. - Validated-claim strings drop the prefix: media3_h264_decode -> h264_decode, media3_audio_decode -> audio_decode. - Golden fixtures in testdata/protocol_v3 are regenerated by Go and are now the cross-repo source of truth. Part of the playback protocol v3 neutral-contract train (steps 2-3 of docs/superpowers/plans/2026-07-30-playback-protocol-v3-neutral-contract.md). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(playback): add v3 evidence tiers and neutral device/output context Implement plan sections 3.3 and 3.4 of the v3 neutral-contract pass: - ClientCodecCapabilitiesV3 gains required video_evidence and audio_evidence closed enums (exact | platform_attested | declared). Planner strictness follows the tier: exact keeps the strict decode-entry validation, platform_attested validates codec/resolution/bit-depth/ frame-rate but skips profile/level matching, declared grants copy routes from the flat codec lists. Only exact audio evidence earns passthrough claims. The detailed_decode_capabilities feature token is deleted (subsumed by video_evidence=exact), and evidence-blocked direct routes carry the new evidence_insufficient_for_direct reason/warning. - DeviceContextV3 is now platform/os_version/manufacturer/model plus a bounded platform_details map (<=16 entries, <=128 chars); the Android Build dump fields are gone. Fire TV quirks keep matching on manufacturer/model (brand fallback removed with the field). - output_route_generation (int64, dual-location) becomes an optional opaque output_context_id string on the output context; the dual-location consistency validation is deleted. Attempt keys, plan invalidation, route events, and the planstore column follow (new Goose migration). - Feature advertisement collapses to the top-level client_features list only; ClientPlaybackContextV3.Features is deleted and ReplanRequestV3 gains an optional client_features refresh. - PlanRecipeVersionV3 bumped v3.3 -> v3.4; fixtures re-keyed. Part of the playback protocol v3 neutral-contract finalization plan (docs/superpowers/plans/2026-07-30-playback-protocol-v3-neutral-contract.md). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * feat(playback): add v3 intent replans, quality menu, and audio-only routes Protocol v3 could only replan after a failure, so changing the audio track or the quality still required the legacy audio PATCH and the client-recipe transcode start — the two endpoints v3 is meant to replace. Clients also had to own a resolution ladder to render a quality menu, and a source with no video track was terminaled by the video/HDR gates, keeping audiobooks on the legacy path. Add track_change and quality_change replan operations. They carry no failure classification and route through the existing replan transaction, so they inherit its idempotency, capacity reservation, and staged-successor commit for free. Because nothing failed, the previous route stays eligible: neither the attempted-key history nor the failed-plan exclusion applies to them. Publish the server ladder on the plan as available_qualities so the quality menu is server-owned; the rungs come from the same resolutionLabelV3 and ladderBitrateKbpsV3 helpers the planner itself uses, not a parallel table. Plan audio-only sources through their own reduced route family: original_http when the client decodes the codec, otherwise a progressive AAC conversion. The plan advertises audio/mp4 for that remux and the transport now serves the same value, because a declared-tier client probes the advertised MIME with isTypeSupported before attaching a source buffer, and "video/mp4" on a stream with no video track is exactly the mismatch that makes the probe lie. Name the protocol's string vocabulary (dynamic ranges, transformations, executors, validated claims, terminal reasons) as constants while touching these lines, so the wire values have one definition. Part of #135 * docs(playback): publish the v3 protocol contract and fix subtitle ordinals Protocol v3 exists only as Go code today, so the Android and Apple ports have no authority to implement against other than reading this repository. Publish the contract as a normative document, machine-checkable schemas, and generated golden fixtures, and fix the one place where the server's own wire output disagreed with the ordinal space it publishes. - docs/architecture/playback-protocol-v3.md is self-contained enough for a third-party client: endpoints and status codes, evidence tiers and their bound-matching rules, delivery classes, the timeline model, replan semantics, registries, track identity, plan identity, quality, and transformations. - docs/design/schemas/playback-v3/ carries JSON Schemas for the five wire shapes plus valid and invalid fixtures, following the client-diagnostics layout. internal/playback/contract validates every fixture against its schema, so a schema that drifts from the Go types fails the Go suite. - cmd/playbackfixtures generates internal/playback/testdata/protocol_v3 from the production planner. `make playback-fixtures` writes them and `make verify-playback-fixtures` (wired into CI) fails when they are stale. These files are what the client ports consume, so drift would otherwise surface as a playback bug on three platforms at once. The subtitle fix: combined ordinals are one dense space over externals, then embedded tracks, then downloaded ones, but the legacy URL builder skipped burn-in-only tracks while assigning indices, so every track after a DVD/DVB track was numbered one too low and resolved to its neighbour. Ordinal assignment now lives in playback.BuildSubtitleInventoryV3 and both the plan inventory and the legacy `subtitle_urls` shape project from it; the legacy shape still filters burn-in-only entries but keeps each track's real index. Part of #135 * feat(web): migrate the players to the neutral playback v3 contract The web player was the last client still speaking the legacy start protocol: it picked its own file version from a codec probe, posted an ffmpeg recipe to start a transcode, PATCHed an endpoint to change audio tracks, and derived its own quality ladder. None of that survives a server-owned plan, and none of it produced telemetry the apps could be compared against. Video player: starts with a v3 request that advertises `declared` evidence from `isTypeSupported` probes and the three delivery classes, then consumes the returned plan for its URL, timeline, tracks and warnings. Quality and track changes become replans (`quality_change`, `track_change`), the quality menu renders `available_qualities` instead of computing rungs, and playback failures emit `route-events` so web failures land in the same diagnostics as Android and Apple. The duration comes from `source.duration_seconds` rather than the playback engine, and the "how was this delivered" overlay reads the plan's delivery and server transformations instead of comparing codec strings. Audiobook player: starts against the audio-only planner path with a single `original` rung, and takes its seek anchor from `timeline.player_start_seconds` so the progressive-remux route (which anchors the stream and restarts the player clock at zero) does not seek twice. Server side, `disable_progress_persistence` left the wire, so the rule it encoded is now derived. Resume state is keyed on the item, but every part of a multipart presentation shares that key while carrying its own file-local clock — persisting part 4's position would store "12 minutes in" as the book's resume point. `PresentationPartTotal > 1` expresses that directly and generalizes to multipart movies and split episodes, and a client can no longer forget to ask or lie about it. `useTranscodeQuality` and the legacy response types are deleted, and `WEBTEST_KNOWN_FAILURES` loses the audiobook entry along with its fix. Part of #135 * feat(playback)!: make v3 the only playback protocol Protocol v3 shipped behind a flag, alongside the legacy start path it was designed to replace. Running both meant every planner change had to be made twice, in two shapes that disagree about who decides the route: the legacy body carried a decision the client had already made, while v3 asks the server to make it. This deletes the legacy half. Removed: - `handleStartPlaybackLegacy` and its request/response bodies. The `POST /playback/start` route stays, but the protocol-version dispatch envelope is now a strict v3 decode — a body that does not declare `protocol_version: 3` gets `426 client_upgrade_required` so an outdated app can render a clear "update required" state instead of misreading a plan. Deliberately not a `400`: the request may be well-formed for the protocol it was written against. - `POST /playback/transcode/start`, superseded by the `quality_change` replan operation, and `PATCH /playback/{session_id}/audio`, superseded by `track_change`. Both mutated a session without re-planning. - The shadow planner and both rollout settings rows. With v3 the only protocol, `playback.protocol_v3_enabled` would mean "no playback at all"; `playback.protocol_v3_shadow_enabled` gated a comparison against a path that no longer exists. `409 protocol_disabled` on route-events goes with them, and capability `enabled` is now constant `true` (the field stays — clients feature-detect against it). - Version-selection helpers in `internal/playback/resolver.go` that only legacy start reached. `Resolve`/`ClientCapabilities`/`PlayDecision` stay: downloads consumes them. `internal/jellycompat` has its own resolution surface and is untouched. Behaviour the legacy handlers owned and v3 now owns explicitly: series version and audio-track preferences are persisted on start and on a `track_change` replan (not on failure recovery, whose forced route is not a user choice); an omitted `start_position` resolves to the profile's saved resume point; and an omitted audio track resolves through the series preference, the profile audio language, then the library override. Both are settled before planning, because the plan's timeline is cut at the start position. Spec §2.2 documents this as "omission is a request, not a default". The encode-target clamp that lived in the deleted transcode handler is already enforced in the planner, twice — `availableQualitiesV3` omits rungs at or above the source height, and the encode path clamps `targetHeight` to it. Unchanged: progress, stop, HLS manifest and segment delivery, the realtime control socket, stream tokens and restart reconstruction, watch together, downloads, jellycompat. Every removal is recorded in the pre-lock removals table in docs/architecture/v1-scope.md. Part of #135 * fix(scanner): stop recording embedded cover art as a video track ffprobe reports embedded cover art as a video stream carrying disposition.attached_pic. convertProbeData appended every "video" stream to VideoTracks without consulting isMainVideoStream, the predicate that already existed for duration decisions, so the picture was persisted as a playable track. That misreports the file twice: - An audio file with a cover picks up a video track, so it no longer satisfies MediaFile.IsAudioOnly and the v3 planner routes an audiobook through the video path instead of planAudioOnlyV3. - When the picture is ordered ahead of the real stream, the flat codec_video/resolution/hdr columns describe the poster: a 954x720 h264 episode was stored as mjpeg 480x480. Filter attached_pic streams out of the track loop. The guard is the disposition flag, not the codec name, so a genuine MJPEG video is still probed as video — the library has one. Already-probed rows self-heal on the next playback: NeedsCriticalProbeRepair already reprobes tracks missing color_range, which covers 21 of the 23 affected rows, and applyProbeData overwrites VideoTracks wholesale. The remaining two need a rescan; nothing persisted records attached_pic, and keying repair off still-image codec names would reprobe the genuine MJPEG file on every playback forever. Part of the playback v3 neutral-contract work: it is what lets Android drop AUDIOBOOK_COVER_ART_CODECS, which fabricated decode support the client cannot honestly claim under video_evidence: "exact". Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(playback): publish subtitle URLs even when playback starts with subtitles off The v3 plan's subtitle inventory is the authoritative track list a client builds its subtitle menu from, but the handler only rewrote it with session-scoped URLs when a track was actually selected. A start or replan that resolved to `subtitle.mode: "off"` therefore returned the planner's URL-less inventory, so a client whose picker reads the inventory had a menu it could not fetch anything from. The Cast path hits this every time: it starts with subtitles off and needs the receiver's text tracks up front. attachSubtitleArtifactV3 now scopes and publishes the inventory unconditionally and gates only the artifact stamping on the selection. Spec §8 records that the `url` on a sidecar entry does not depend on the current selection. Part of the v3 neutral-contract finalization. * chore(playback): reconcile neutral v3 with main * fix(playback): preserve subtitle intent across replans * fix(playback): retain subtitle inventory on adapted routes * fix(playback): software-decode High10 AVC for QSV * fix(playback): scale High10 frames before QSV upload * fix(playback): preserve empty subtitle inventories * fix(playback): freeze terminal attempt contract * chore(playback): name fixture contract tokens * fix(playback): close v3 conformance review gaps * chore(playback): name conformance category * fix(playback): complete v3 conformance contract * fix(playback): keep schema fixtures generated * fix(playback): emit schema-valid conformance arrays * fix(playback): omit empty replan failures * fix(web): omit empty replan failures * fix(playback): close neutral v3 contract gaps * fix(playback): harden v3 replan, transcode, and quality-ladder edge cases Review remediation for the neutral v3 cutover, server side: - A failed replan no longer overwrites the durable StartResponse with a terminal or advances the replan request ID; an idempotent start replay of a still-healthy session returns the original plan. - SoftwareVideoDecode is now derived inside the transcode layer from source facts (codec/profile/bit depth) carried on TranscodeOpts, so jellycompat, downloads, recipe-card reconstruction, and transcode nodes get the High10 software-decode fix, not just the v3 handler. video_to_h264 recipe version bumps to 2 so mixed-version node pools that would silently drop the flag fail validation instead. - Local transport startup shares the 30s ManifestStartupTimeout; a timeout with the process still running stays retryable and is no longer persisted as a durable terminal against the attempt. - Sparse replan bodies (failure_recovery et al) no longer reset a user-selected quality preference to auto; the empty-value guard now covers every operation. - availableQualitiesV3 publishes no fixed rungs when the source height is unknown, keeping the no-upscaling ladder contract. - The proxy remux path serves audio-only fMP4 as audio/mp4 via a new additive AudioOnly token claim, matching the integrated path. - Plain text subtitle sidecars accept any requested extension again (served as VTT), restoring the permissive v1 behavior; ASS and bitmap handling is unchanged. - The 4K-disallowed terminal message discloses when a lower-resolution alternate exists but was pinned away by quality "original". Part of #135. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(web): keep playback alive through failed replans and honest audio claims Review remediation for the neutral v3 cutover, web player: - A failed or refused replan no longer unmounts the player: the fatal error screen is reserved for loads with no adopted plan, and replan failures surface through the existing non-fatal replanError path. - changeQuality rolls its optimistic preference back when the replan is refused or errors, so a failed switch is not silently applied by the next unrelated replan and the menu shows the real active rung. - The capability probe now tests mp3/vorbis codecs and mp3/flac/ogg containers (MediaSource with a canPlayType fallback), restoring direct play for mp3 audiobooks instead of per-part AAC re-encodes. - Reanchor seeks issued while a replan is in flight coalesce and run when it settles instead of being silently dropped with the scrubber pinned to a phantom position. - Subtitle refresh/translation replans use the resume anchor while the media element has no metadata, so a subtitle_ready broadcast during startup no longer restarts a resumed stream at 0:00. - An exhausted failure-recovery chain sets a visible error instead of returning silently. Part of #135. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(playback): accept video-only and VP9 probe metadata Treat audio and video probe completeness independently so legitimate video-only assets converge without repeated ffprobe repair. Allow unknown codec profile/level metadata to fall through to server adaptation while preserving exact direct-decode constraints. Fixes #574 * fix(playback): address protocol v3 review findings * fix(playback): harden lease and probe repair decisions * fix(playback): close remaining v3 review gaps * fix(playback): recover failed transcode starts * fix(playback): address remaining review-bot findings on v3 replan and audio planning Server: - The deferred replan lease release is bounded by a 3s timeout so a saturated pool or DB outage cannot wedge a handler goroutine that holds the per-session store lock on an uncancellable context. - planAudioOnlyV3 honors the request bandwidth cap: an over-cap source skips the original_http direct route and converts to AAC with the same bandwidth_cap_applied warning and decision reason the video ladder uses. Unknown source bitrate never triggers the cap. - A copy-audio progressive plan rejected only by a per-delivery audio_decode_codecs subset retries as an AAC conversion instead of returning adaptation_unavailable, and the AAC recipe respects the delivery's max_channels. Web: - failure_recovery replans issued while another replan is in flight queue (superseding a pending seek reanchor) instead of being silently dropped with the fatal overlay already suppressed. - A terminal response to a fresh non-preserving start clears the previous plan and stops its session, so episode navigation cannot keep rendering the prior item under the new title. - A refused recovery replan for a transport-dead plan surfaces the error and re-arms the plan failure key, so transient recovery failures no longer strand an endless spinner; the audiobook player gets the same guard reset. - A track-less subtitle_translation_completed hands off to the refreshed persisted track once the inventory settles, clearing the live overlay, instead of pinning the synthetic live track forever. Part of #135. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(playback): reuse HLS transport for sidecar replans * fix(playback): stabilize copy HLS remount timeline * fix(playback): address v3 review findings * fix(playback): satisfy player contract types --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2608 lines
88 KiB
Go
2608 lines
88 KiB
Go
package playback
|
||
|
||
import (
|
||
"bytes"
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"log"
|
||
"log/slog"
|
||
"math"
|
||
"mime"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"syscall"
|
||
"time"
|
||
|
||
"github.com/Silo-Server/silo-server/internal/models"
|
||
)
|
||
|
||
func init() {
|
||
// Register .m4s so http.ServeFile sets the correct Content-Type for
|
||
// fMP4 HLS segments. Go's default MIME database does not include it.
|
||
_ = mime.AddExtensionType(".m4s", "video/mp4")
|
||
}
|
||
|
||
// TranscodeOpts holds configuration for an HLS transcode session.
|
||
type TranscodeOpts struct {
|
||
InputPath string
|
||
OutputDir string // e.g., /tmp/silo-transcode/{session_id}/
|
||
// subtitleFilterInputPath is a parser-safe local alias used only by the
|
||
// libass subtitles filter. FFmpeg still opens InputPath as the media input.
|
||
subtitleFilterInputPath string
|
||
// OutputSubdir is the signed, root-relative reconstruction directory. Empty
|
||
// retains the legacy flat {session_id} layout.
|
||
OutputSubdir string
|
||
TranscodeTransportID string
|
||
SessionID string
|
||
SourceVideoCodec string
|
||
SourceVideoProfile string
|
||
SourceVideoBitDepth int
|
||
VideoBitstreamFilter string // validated copy-mode BSF, e.g. dovi_rpu=strip=1
|
||
SeekSeconds float64
|
||
// StreamOriginSeconds is the keyframe timestamp at which a copy-video
|
||
// stream actually begins. SeekSeconds remains the client-requested -ss so
|
||
// FFmpeg performs exactly one demuxer seek; this origin keeps response and
|
||
// reconstruction timelines aligned with the resulting media pre-roll.
|
||
StreamOriginSeconds float64
|
||
// CopySeekAnchorResolved distinguishes a valid zero-second origin from
|
||
// older/shared recipes that never resolved a copy seek anchor.
|
||
CopySeekAnchorResolved bool
|
||
TargetResolution string // e.g., 1080p, 720p
|
||
TargetCodecVideo string // e.g., h264 (or hevc if allowed)
|
||
TargetCodecAudio string // e.g., aac
|
||
SegmentDuration int // seconds, default 6
|
||
StartSegmentNumber int // -hls_segment_start_number, default 0
|
||
FFmpegPath string // optional explicit ffmpeg binary path
|
||
HWAccel string // auto, qsv, vaapi, nvenc, none
|
||
HWDevice string // e.g., /dev/dri/renderD128 (default if empty)
|
||
// AvoidHWDevice asks the initial multi-device allocator to prefer any other
|
||
// present render device. It is a process-local startup hint used after an
|
||
// early GPU failure; the selected concrete device remains fully reserved and
|
||
// is the value frozen into the session recipe.
|
||
AvoidHWDevice string
|
||
// SoftwareVideoDecode keeps a hardware encoder while decoding the source
|
||
// on the CPU. Intel's VAAPI/QSV decoders cannot accept 10-bit AVC, but the
|
||
// decoded frames can still be converted to NV12, uploaded, and encoded by
|
||
// QSV/VAAPI. The flag is frozen into recipe cards so restarts do not put the
|
||
// unsupported hardware decoder back.
|
||
SoftwareVideoDecode bool
|
||
SubtitleTrackIndex int // -1 = no subtitles
|
||
SubtitleBurnIn bool
|
||
// SubtitleCodec is the probed codec of the burn-in track (e.g. "subrip",
|
||
// "hdmv_pgs_subtitle"). Bitmap codecs (PGS/DVD/DVB) select the overlay
|
||
// filter_complex pipeline; text codecs use the libass subtitles filter.
|
||
// Empty preserves the legacy text path for callers minted before the field.
|
||
SubtitleCodec string
|
||
AudioTrackIndex int // -1 = default (first track), >= 0 = specific track
|
||
// TargetAudioChannels selects mono (1), stereo (2/default), or 5.1 (6+)
|
||
// output. Ignored for copy/passthrough audio targets.
|
||
TargetAudioChannels int
|
||
// TargetAudioBitrateKbps caps an encoded audio stream. Zero selects the
|
||
// layout-appropriate AAC default. Ignored for copy/passthrough targets.
|
||
TargetAudioBitrateKbps int
|
||
TargetBitrateKbps int // max video bitrate in kbps; 0 = CRF-only (no cap)
|
||
TotalDuration float64 // total media duration in seconds (for VOD manifest)
|
||
FastStart bool // use superfast preset for faster first-segment production
|
||
NodeType string
|
||
ExecutionMode string
|
||
FFmpegLogSink FFmpegLogSink
|
||
}
|
||
|
||
// DV7ToHDR10BitstreamFilter strips Dolby Vision RPU metadata during a
|
||
// copy-mode HLS remux; the enhancement layer is dropped by stream mapping.
|
||
const DV7ToHDR10BitstreamFilter = "dovi_rpu=strip=1"
|
||
|
||
const (
|
||
transcodeCodecH264 = "h264"
|
||
transcodeHWQSV = "qsv"
|
||
transcodeHWVAAPI = "vaapi"
|
||
transcodeHWNVENC = "nvenc"
|
||
)
|
||
|
||
// TranscodeSession manages a running ffmpeg HLS transcode process.
|
||
type TranscodeSession struct {
|
||
cmd *exec.Cmd
|
||
cancel context.CancelFunc
|
||
opts TranscodeOpts
|
||
outputDir string
|
||
running bool
|
||
restarting bool
|
||
waitErr error
|
||
stderr *boundedTailBuffer
|
||
mu sync.Mutex
|
||
done chan struct{} // closed when the monitor goroutine finishes
|
||
stdinPipe io.WriteCloser
|
||
lastRequestedSegment int
|
||
throttler *TranscodeThrottler
|
||
stderrLinesLogged int
|
||
stderrBytesLogged int
|
||
stderrDroppedLines int
|
||
stderrCapLogged bool
|
||
restartCount int
|
||
stderrLineIndex int
|
||
stderrWriter *ffmpegStderrWriter
|
||
restartHook func(context.Context)
|
||
// reserveHWDeviceOnRestart is true when StartTranscode selected and reserved
|
||
// one device from a multi-device QSV/VAAPI setting. Each replacement ffmpeg
|
||
// process reacquires that same concrete device.
|
||
reserveHWDeviceOnRestart bool
|
||
}
|
||
|
||
// SetRestartHook registers a callback fired after every successful Restart.
|
||
// The owning handler uses it to re-arm the transcode throttler and the exit
|
||
// monitor; firing it from Restart itself keeps every restart caller (web
|
||
// segment recovery, audio switch, jellycompat seek) consistent.
|
||
func (s *TranscodeSession) SetRestartHook(fn func(context.Context)) {
|
||
s.mu.Lock()
|
||
s.restartHook = fn
|
||
s.mu.Unlock()
|
||
}
|
||
|
||
// SegmentProgress describes the media ffmpeg has actually produced on disk.
|
||
type SegmentProgress struct {
|
||
ProducedHead int
|
||
ProducedCount int
|
||
LastProducedAt time.Time
|
||
ManifestModTime time.Time
|
||
HasManifest bool
|
||
Running bool
|
||
Restarting bool
|
||
StartSegmentNumber int
|
||
SegmentDuration int
|
||
LastRequestedSegment int
|
||
}
|
||
|
||
// SegmentRecoveryDecision tells the segment handler whether to briefly wait
|
||
// for ffmpeg or seek-restart immediately.
|
||
type SegmentRecoveryDecision struct {
|
||
Wait bool
|
||
WaitTimeout time.Duration
|
||
RestartOnTimeout bool
|
||
Reason string
|
||
Progress SegmentProgress
|
||
}
|
||
|
||
// defaultSegmentDuration is the segment length when not specified. Short
|
||
// segments (2s) allow the player to start quickly while still maintaining
|
||
// efficient HTTP delivery. This matches the approach used by Plex.
|
||
const defaultSegmentDuration = 2
|
||
|
||
// DefaultSegmentDuration is the exported segment length used when a transcode
|
||
// request does not specify one. Callers minting a reconstruct recipe must embed
|
||
// a concrete (>0) value so the token passes the node's completeness gate and the
|
||
// embedded length matches what the node actually produces.
|
||
const DefaultSegmentDuration = defaultSegmentDuration
|
||
|
||
// maxSyntheticManifestSegments preserves the historical worst-case playlist
|
||
// size (100,000 seconds at two-second segments). Longer media uses FFmpeg's
|
||
// real sliding playlist instead of allocating a complete synthetic manifest.
|
||
const maxSyntheticManifestSegments = 50_000
|
||
|
||
// remountStartOffsetSeconds is a positive, effectively-zero HLS start offset.
|
||
// Media3 suppresses live-edge position projection for EVENT playlists only
|
||
// when EXT-X-START is positive. Anchoring one millisecond after the generation
|
||
// origin keeps an initial start indistinguishable from zero while giving a
|
||
// rebuilt MediaSource a stable timeline on which to restore its playhead.
|
||
const remountStartOffsetSeconds = 0.001
|
||
|
||
const maxPersistedFFmpegLines = 2000
|
||
const maxPersistedFFmpegBytes = 256 * 1024
|
||
const maxPersistedFFmpegChars = 2000
|
||
|
||
// ManifestStartupTimeout is the maximum wait for FFmpeg's first safe playback
|
||
// window before the caller reports a retryable startup timeout.
|
||
const ManifestStartupTimeout = 30 * time.Second
|
||
|
||
const (
|
||
maxSequentialMissingSegments = 2
|
||
activeSegmentWait = 12 * time.Second
|
||
segmentWaitGrace = 1500 * time.Millisecond
|
||
maxSegmentWait = 6 * time.Second
|
||
minSegmentWait = 3 * time.Second
|
||
minStaleProducedWindow = 5 * time.Second
|
||
)
|
||
|
||
// StartTranscode launches an ffmpeg process that produces HLS segments.
|
||
func StartTranscode(ctx context.Context, opts TranscodeOpts) (*TranscodeSession, error) {
|
||
if opts.VideoBitstreamFilter != "" &&
|
||
(opts.VideoBitstreamFilter != DV7ToHDR10BitstreamFilter || !strings.EqualFold(opts.TargetCodecVideo, "copy")) {
|
||
return nil, fmt.Errorf("unsupported video bitstream filter recipe")
|
||
}
|
||
if opts.SegmentDuration <= 0 {
|
||
opts.SegmentDuration = defaultSegmentDuration
|
||
}
|
||
opts = resolveSoftwareVideoDecode(opts)
|
||
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
|
||
configuredHWDevices := ParseHWDeviceSet(opts.HWDevice)
|
||
reserveHWDeviceOnRestart := configuredHWDevices.Multi() && hwAccelBalancesRenderDevices(opts.HWAccel)
|
||
// Resolve a multi-device hw_device list to one concrete GPU. Restarts reuse
|
||
// the selected device, but each ffmpeg process owns its own reservation.
|
||
hwDevice, releaseHWDevice := acquireHWDevice(opts.HWDevice, opts.HWAccel, opts.AvoidHWDevice)
|
||
opts.HWDevice = hwDevice
|
||
opts.AvoidHWDevice = ""
|
||
|
||
// Ensure output directory exists.
|
||
if err := os.MkdirAll(opts.OutputDir, 0o755); err != nil {
|
||
releaseHWDevice()
|
||
return nil, fmt.Errorf("create output dir: %w", err)
|
||
}
|
||
if err := prepareSubtitleFilterInput(&opts); err != nil {
|
||
releaseHWDevice()
|
||
return nil, err
|
||
}
|
||
|
||
ctx, cancel := context.WithCancel(ctx)
|
||
s := &TranscodeSession{
|
||
cancel: cancel,
|
||
opts: opts,
|
||
outputDir: opts.OutputDir,
|
||
running: true,
|
||
done: make(chan struct{}),
|
||
stderr: newBoundedTailBuffer(stderrTailMaxBytes),
|
||
lastRequestedSegment: opts.StartSegmentNumber,
|
||
reserveHWDeviceOnRestart: reserveHWDeviceOnRestart,
|
||
}
|
||
|
||
args := buildFFmpegArgs(opts)
|
||
bin := opts.FFmpegPath
|
||
if bin == "" {
|
||
bin = ffmpegBinary()
|
||
}
|
||
|
||
log.Printf("playback: ffmpeg cmd: %s %s", bin, strings.Join(args, " "))
|
||
s.logFFmpegEvent(ctx, "ffmpeg process starting", "")
|
||
|
||
cmd := exec.CommandContext(ctx, bin, args...)
|
||
stdinPipe, err := cmd.StdinPipe()
|
||
if err != nil {
|
||
cancel()
|
||
releaseHWDevice()
|
||
return nil, fmt.Errorf("create stdin pipe: %w", err)
|
||
}
|
||
cmd.Dir = opts.OutputDir
|
||
cmd.Stderr = s.newStderrWriter(ctx)
|
||
cmd.WaitDelay = 3 * time.Second
|
||
|
||
if err := cmd.Start(); err != nil {
|
||
cancel()
|
||
releaseHWDevice()
|
||
s.logFFmpegEvent(ctx, "ffmpeg process exit error", err.Error())
|
||
return nil, fmt.Errorf("start ffmpeg: %w", err)
|
||
}
|
||
s.cmd = cmd
|
||
s.stdinPipe = stdinPipe
|
||
s.logFFmpegEvent(ctx, "ffmpeg process started", "")
|
||
|
||
// Monitor ffmpeg in background. The process-specific reservation is released
|
||
// before done closes, so waiters can safely launch a replacement process.
|
||
go s.monitorFFmpeg(ctx, cmd, s.done, releaseHWDevice)
|
||
|
||
return s, nil
|
||
}
|
||
|
||
func (s *TranscodeSession) monitorFFmpeg(ctx context.Context, cmd *exec.Cmd, done chan struct{}, releaseHWDevice func()) {
|
||
waitErr := cmd.Wait()
|
||
// The child no longer owns the device once Wait returns. Release before
|
||
// flushing/logging so slow diagnostics cannot make the allocator count a
|
||
// process that has already exited.
|
||
releaseHWDevice()
|
||
s.flushStderr(ctx)
|
||
s.mu.Lock()
|
||
s.running = false
|
||
s.waitErr = waitErr
|
||
s.mu.Unlock()
|
||
s.logWaitResult(ctx, waitErr)
|
||
close(done)
|
||
}
|
||
|
||
// IsMPEG2VideoCodec reports whether a probed video codec name identifies
|
||
// MPEG-2 video. It accepts common FFmpeg aliases because codec strings can
|
||
// come from scan metadata, direct probes, or client capability lists.
|
||
func IsMPEG2VideoCodec(codec string) bool {
|
||
normalized := strings.NewReplacer(
|
||
" ", "",
|
||
"-", "",
|
||
"_", "",
|
||
".", "",
|
||
).Replace(strings.ToLower(strings.TrimSpace(codec)))
|
||
switch normalized {
|
||
case "mpeg2video", "mpeg2", "mp2v":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// IsMPEG4Part2VideoCodec reports whether a codec name identifies MPEG-4 Part 2
|
||
// video, commonly found in older XviD/DivX AVI files.
|
||
func IsMPEG4Part2VideoCodec(codec string) bool {
|
||
normalized := strings.NewReplacer(
|
||
" ", "",
|
||
"-", "",
|
||
"_", "",
|
||
".", "",
|
||
).Replace(strings.ToLower(strings.TrimSpace(codec)))
|
||
switch normalized {
|
||
case "mpeg4", "mp4v", "xvid", "divx", "dx50":
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// RequiresSoftwareVideoDecode reports sources that the installed hardware
|
||
// execution recipes must not feed to their video decoders. AVC High 10 is not
|
||
// supported by Intel VAAPI/QSV decode; scanner rows are not perfectly uniform,
|
||
// so either the probed bit depth or the normalized profile is sufficient.
|
||
func RequiresSoftwareVideoDecode(codec, profile string, bitDepth int) bool {
|
||
if normalizeCodecV3(codec) != transcodeCodecH264 {
|
||
return false
|
||
}
|
||
normalizedProfile := strings.NewReplacer(" ", "", "-", "", "_", "").Replace(strings.ToLower(strings.TrimSpace(profile)))
|
||
return bitDepth > 8 || normalizedProfile == "high10" || normalizedProfile == "high10intra" || normalizedProfile == "hi10p"
|
||
}
|
||
|
||
// SourceVideoTranscodeFacts returns the primary source facts needed to choose
|
||
// a safe FFmpeg decoder. Track metadata fills gaps in legacy media-file rows.
|
||
func SourceVideoTranscodeFacts(file *models.MediaFile) (codec, profile string, bitDepth int) {
|
||
if file == nil {
|
||
return "", "", 0
|
||
}
|
||
codec = file.CodecVideo
|
||
if len(file.VideoTracks) == 0 {
|
||
return codec, "", 0
|
||
}
|
||
track := file.VideoTracks[0]
|
||
if strings.TrimSpace(codec) == "" {
|
||
codec = track.Codec
|
||
}
|
||
return codec, track.Profile, models.NormalizeVideoBitDepth(track.BitDepth, track.PixelFormat, track.Profile)
|
||
}
|
||
|
||
func resolveSoftwareVideoDecode(opts TranscodeOpts) TranscodeOpts {
|
||
if RequiresSoftwareVideoDecode(opts.SourceVideoCodec, opts.SourceVideoProfile, opts.SourceVideoBitDepth) {
|
||
opts.SoftwareVideoDecode = true
|
||
}
|
||
return opts
|
||
}
|
||
|
||
// buildFFmpegArgs constructs the full ffmpeg argument list from TranscodeOpts.
|
||
func buildFFmpegArgs(opts TranscodeOpts) []string {
|
||
opts = resolveSoftwareVideoDecode(opts)
|
||
// Resolve "auto" into a concrete accel method once so all downstream
|
||
// helpers (appendHWAccelArgs, appendVideoArgs, etc.) see the real value.
|
||
opts.HWAccel = resolveEffectiveTranscodeHWAccel(opts)
|
||
|
||
isVideoCopy := opts.TargetCodecVideo == "copy"
|
||
isAudioCopy := opts.TargetCodecAudio == "copy"
|
||
|
||
args := []string{
|
||
"-hide_banner",
|
||
"-loglevel", "error",
|
||
}
|
||
|
||
// Hardware acceleration — skip when copying video (no encoding needed).
|
||
if !isVideoCopy {
|
||
args = appendHWAccelArgs(args, opts)
|
||
}
|
||
|
||
// Limit input probing to speed up startup, especially on network storage.
|
||
// -fflags +genpts generates PTS for files with missing timestamps;
|
||
// +fastseek enables faster input seeking (matches Jellyfin).
|
||
args = append(args,
|
||
"-fflags", "+genpts+fastseek",
|
||
"-analyzeduration", "3000000", // 3 seconds (default 5s)
|
||
"-probesize", "5000000", // 5 MB (default 5MB, explicit for clarity)
|
||
)
|
||
|
||
// Seek before input for fast seeking.
|
||
if opts.SeekSeconds > 0 {
|
||
args = append(args, "-ss", fmt.Sprintf("%.3f", opts.SeekSeconds))
|
||
// When video is copied but audio is transcoded, accurate_seek causes
|
||
// A/V desync: video must start at a keyframe but audio is trimmed to
|
||
// the exact seek point. Disabling it keeps both streams aligned.
|
||
if isVideoCopy && !isAudioCopy {
|
||
args = append(args, "-noaccurate_seek")
|
||
}
|
||
}
|
||
|
||
// Input file.
|
||
args = append(args, "-i", opts.InputPath)
|
||
args = append(args, "-map_metadata", "-1")
|
||
args = append(args, "-map_chapters", "-1")
|
||
args = appendStreamSelectionArgs(args, opts)
|
||
args = appendTimestampNormalizationArgs(args, opts)
|
||
|
||
// Video codec and encoding settings.
|
||
if isVideoCopy {
|
||
args = append(args, "-c:v", "copy")
|
||
if opts.VideoBitstreamFilter == DV7ToHDR10BitstreamFilter {
|
||
args = append(args, "-bsf:v", opts.VideoBitstreamFilter)
|
||
}
|
||
} else {
|
||
args = appendVideoArgs(args, opts)
|
||
}
|
||
|
||
// Copy-video sessions only do audio work on the filter/encode side.
|
||
// ffmpeg's default thread selection spawns one filter thread per CPU
|
||
// (observed 14 idle `af#0:1` threads for a 5.1→2.0 downmix), so pin
|
||
// audio filter + encode to a single thread.
|
||
if isVideoCopy && !isAudioCopy {
|
||
args = append(args, "-threads", "1", "-filter_threads", "1", "-filter_complex_threads", "1")
|
||
}
|
||
|
||
// Audio codec.
|
||
args = appendAudioArgs(args, opts)
|
||
|
||
// Subtitle burn-in and resolution scaling — only when encoding video.
|
||
if !isVideoCopy {
|
||
args = appendVideoFilterArgs(args, opts)
|
||
args = appendSegmentBoundaryArgs(args, opts)
|
||
}
|
||
|
||
// HLS output options.
|
||
// Codec-copy sessions usually use fMP4 segments — no transmuxing needed in
|
||
// hls.js, which avoids Safari MSE compatibility issues with certain codecs
|
||
// in TS. MPEG-2 video is the exception: Apple consumes it as compatibility
|
||
// HLS, so package it in MPEG-TS while still copying the video stream.
|
||
// Actual transcoding uses MPEG-TS segments to avoid the hls.js endOfStream()
|
||
// race with fMP4 (hls.js #6337).
|
||
var segmentPattern string
|
||
segmentType := "mpegts"
|
||
copyVideoUsesFMP4 := isVideoCopy && !IsMPEG2VideoCodec(opts.SourceVideoCodec)
|
||
if copyVideoUsesFMP4 {
|
||
segmentType = "fmp4"
|
||
segmentPattern = filepath.Join(opts.OutputDir, "seg_%05d.m4s")
|
||
} else {
|
||
segmentPattern = filepath.Join(opts.OutputDir, "seg_%05d.ts")
|
||
}
|
||
manifestPath := filepath.Join(opts.OutputDir, "stream.m3u8")
|
||
|
||
args = append(args,
|
||
"-max_muxing_queue_size", "2048",
|
||
"-max_delay", "5000000",
|
||
"-f", "hls",
|
||
"-hls_time", fmt.Sprintf("%d", opts.SegmentDuration),
|
||
// Bound real playlists as well as synthetic ones. Segment files remain on
|
||
// disk because delete_segments is not enabled, while the manifest itself
|
||
// cannot grow without limit during multi-day sessions.
|
||
"-hls_list_size", strconv.Itoa(maxSyntheticManifestSegments),
|
||
"-hls_segment_type", segmentType,
|
||
// Write segments to temp files first so the player never fetches a
|
||
// partially-written segment during a quality switch.
|
||
"-hls_flags", "independent_segments+temp_file",
|
||
"-hls_segment_filename", segmentPattern,
|
||
)
|
||
// fMP4 segments need movflags=+frag_discont so each fragment writes
|
||
// audio DTS/PTS including the initial delay into MOOF→TRAF→TFDT.
|
||
// Without this, some browsers (notably Chromium on macOS) can experience
|
||
// A/V sync issues during copy-mode HLS playback. Matches Jellyfin's
|
||
// proven fMP4 HLS pipeline.
|
||
if copyVideoUsesFMP4 {
|
||
args = append(args, "-hls_segment_options", "movflags=+frag_discont")
|
||
}
|
||
if opts.StartSegmentNumber > 0 {
|
||
args = append(args, "-start_number", fmt.Sprintf("%d", opts.StartSegmentNumber))
|
||
}
|
||
args = append(args, manifestPath)
|
||
|
||
return args
|
||
}
|
||
|
||
func resolveEffectiveTranscodeHWAccel(opts TranscodeOpts) string {
|
||
hwAccel := ResolveHWAccelWithFFmpeg(opts.HWAccel, opts.FFmpegPath)
|
||
if hwAccel == "" {
|
||
return ""
|
||
}
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
|
||
return "none"
|
||
}
|
||
if IsMPEG4Part2VideoCodec(opts.SourceVideoCodec) {
|
||
return "none"
|
||
}
|
||
// The bundled CUDA software-decode upload path has not been validated.
|
||
// Prefer the established libx264 fallback over selecting a decoder known
|
||
// not to accept this source. Intel QSV/VAAPI have the explicit upload paths
|
||
// below and retain hardware encoding.
|
||
if opts.SoftwareVideoDecode && hwAccel == transcodeHWNVENC {
|
||
return "none"
|
||
}
|
||
return hwAccel
|
||
}
|
||
|
||
// bitmapBurnInActive reports whether this transcode composites a bitmap
|
||
// subtitle track (PGS/VOBSUB/DVB) into the video via the overlay
|
||
// filter_complex pipeline. Bitmap burn-in requires a video encode: copy-video
|
||
// sessions never activate it (the API layer forces an encoding recipe before
|
||
// starting a burn-in transcode, this is a defensive backstop).
|
||
func bitmapBurnInActive(opts TranscodeOpts) bool {
|
||
return opts.SubtitleBurnIn &&
|
||
opts.SubtitleTrackIndex >= 0 &&
|
||
NeedsBurnIn(opts.SubtitleCodec) &&
|
||
!strings.EqualFold(opts.TargetCodecVideo, "copy")
|
||
}
|
||
|
||
// appendStreamSelectionArgs limits output to primary video/audio streams.
|
||
// When bitmap burn-in is active the video output comes from the overlay
|
||
// filter_complex graph's labeled pad instead of the raw input stream —
|
||
// mapping both would emit two video streams into the HLS mux.
|
||
func appendStreamSelectionArgs(args []string, opts TranscodeOpts) []string {
|
||
audioTrackIndex := opts.AudioTrackIndex
|
||
if bitmapBurnInActive(opts) {
|
||
args = append(args, "-map", "[vout]")
|
||
} else {
|
||
args = append(args, "-map", "0:v:0")
|
||
}
|
||
if audioTrackIndex >= 0 {
|
||
args = append(args, "-map", fmt.Sprintf("0:a:%d?", audioTrackIndex))
|
||
} else {
|
||
args = append(args, "-map", "0:a:0?")
|
||
}
|
||
args = append(args, "-sn")
|
||
args = append(args, "-dn")
|
||
return args
|
||
}
|
||
|
||
// appendTimestampNormalizationArgs selects timestamp handling based on the
|
||
// playback mode. Copy-video full-file starts use zero-based timestamps so
|
||
// fMP4 fragments always have sane local durations. Copy-video resumes
|
||
// preserve source timestamps so each fragment's TFDT matches its playlist
|
||
// position (segment K sits at playlist-time K*segDur); zero-basing here
|
||
// makes seg_K carry TFDT=0, and strict players (Jellyfin Android TV /
|
||
// ExoPlayer) read EXT-X-START, jump to seg_K expecting media at K*segDur,
|
||
// see TFDT=0, treat the gap as a discontinuity, reload init.mp4, and
|
||
// eventually abort — the symptom that crashes ATV on a second resume.
|
||
// Encoded transcodes keep the source-timestamp policy unconditionally.
|
||
func appendTimestampNormalizationArgs(args []string, opts TranscodeOpts) []string {
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
|
||
if opts.SeekSeconds > 0 {
|
||
return append(args,
|
||
"-copyts",
|
||
"-avoid_negative_ts", "disabled",
|
||
)
|
||
}
|
||
return append(args,
|
||
"-avoid_negative_ts", "make_zero",
|
||
)
|
||
}
|
||
return append(args,
|
||
"-copyts",
|
||
"-avoid_negative_ts", "disabled",
|
||
)
|
||
}
|
||
|
||
// appendSegmentBoundaryArgs forces keyframes on segment boundaries so each HLS
|
||
// fragment starts cleanly and can be appended independently by the player.
|
||
//
|
||
// With -copyts, the output timestamp t starts at the seek position rather than
|
||
// 0. Subtracting SeekSeconds prevents a "catch-up storm" where n_forced races
|
||
// from 0 to seek_position/segment_duration, making every frame an I-frame and
|
||
// grinding encoding to a halt for large seeks.
|
||
func appendSegmentBoundaryArgs(args []string, opts TranscodeOpts) []string {
|
||
args = append(args, "-sc_threshold", "0")
|
||
if opts.SeekSeconds > 0 {
|
||
args = append(args, "-force_key_frames",
|
||
fmt.Sprintf("expr:gte(t-%.3f,n_forced*%d)", opts.SeekSeconds, opts.SegmentDuration))
|
||
} else {
|
||
args = append(args, "-force_key_frames",
|
||
fmt.Sprintf("expr:gte(t,n_forced*%d)", opts.SegmentDuration))
|
||
}
|
||
|
||
// Hardware encoders (QSV, VAAPI, NVENC) may not reliably honor
|
||
// force_key_frames expressions. Set explicit GOP size so segment
|
||
// boundaries always start with an intra frame. We assume 30 fps as a
|
||
// safe ceiling — the GOP will be at most segmentDuration * 30 frames.
|
||
// Matches Jellyfin's approach for hardware encoders.
|
||
if opts.HWAccel == transcodeHWQSV || opts.HWAccel == transcodeHWVAAPI || opts.HWAccel == transcodeHWNVENC {
|
||
gopSize := fmt.Sprintf("%d", opts.SegmentDuration*30)
|
||
args = append(args, "-g", gopSize, "-keyint_min", gopSize)
|
||
}
|
||
// QSV otherwise encodes force_key_frames requests as non-IDR intra frames.
|
||
// The HLS muxer cannot split independent segments on those frames, so a
|
||
// 23.976 fps source with a 60-frame GOP produces 2.5-5 second fragments and
|
||
// breaks the fixed-duration VOD manifest/restart timeline. Promote requested
|
||
// keyframes to IDRs so the muxer cuts at the requested boundaries.
|
||
if opts.HWAccel == transcodeHWQSV {
|
||
args = append(args, "-forced_idr", "1")
|
||
}
|
||
|
||
return args
|
||
}
|
||
|
||
// appendHWAccelArgs adds hardware acceleration flags based on the HWAccel setting.
|
||
// The caller must resolve "auto" via ResolveHWAccel before calling this.
|
||
func appendHWAccelArgs(args []string, opts TranscodeOpts) []string {
|
||
switch opts.HWAccel {
|
||
case "qsv":
|
||
hwDevice := PickRenderDevice(opts.HWDevice)
|
||
if hwDevice == "" {
|
||
slog.Warn("no GPU render device found, QSV transcode may fail")
|
||
hwDevice = "/dev/dri/renderD128" // last-resort fallback
|
||
}
|
||
// VAAPI→QSV hardware pipeline: derive QSV from VAAPI device.
|
||
args = append(args,
|
||
"-init_hw_device", fmt.Sprintf("vaapi=va:%s,driver=iHD,kernel_driver=i915,vendor_id=0x8086", hwDevice),
|
||
"-init_hw_device", "qsv=qs@va",
|
||
"-filter_hw_device", "va",
|
||
)
|
||
if !opts.SoftwareVideoDecode {
|
||
args = append(args, "-hwaccel", "vaapi", "-hwaccel_output_format", "vaapi")
|
||
}
|
||
args = append(args, "-noautorotate")
|
||
case "vaapi":
|
||
vaapiDevice := PickRenderDevice(opts.HWDevice)
|
||
if vaapiDevice == "" {
|
||
vaapiDevice = "/dev/dri/renderD128" // last-resort fallback
|
||
}
|
||
args = append(args,
|
||
"-init_hw_device", fmt.Sprintf("vaapi=hw:%s", vaapiDevice),
|
||
"-filter_hw_device", "hw",
|
||
)
|
||
if !opts.SoftwareVideoDecode {
|
||
args = append(args, "-hwaccel", "vaapi", "-hwaccel_output_format", "vaapi")
|
||
}
|
||
case transcodeHWNVENC:
|
||
args = append(args,
|
||
"-hwaccel", "cuda",
|
||
"-hwaccel_output_format", "cuda",
|
||
"-noautorotate",
|
||
)
|
||
if hwDevice := strings.TrimSpace(opts.HWDevice); hwDevice != "" {
|
||
args = append(args, "-hwaccel_device", hwDevice)
|
||
}
|
||
}
|
||
return args
|
||
}
|
||
|
||
// videoPreset returns an encoder-compatible preset. CPU encoders use a faster
|
||
// fast-start preset for initial playback, while QSV stays on the fastest
|
||
// preset family it supports.
|
||
func videoPreset(opts TranscodeOpts, hwAccel string) string {
|
||
if hwAccel == "qsv" {
|
||
return "veryfast"
|
||
}
|
||
if opts.FastStart {
|
||
return "superfast"
|
||
}
|
||
return "veryfast"
|
||
}
|
||
|
||
// appendVideoArgs adds video codec arguments.
|
||
func appendVideoArgs(args []string, opts TranscodeOpts) []string {
|
||
codec := opts.TargetCodecVideo
|
||
if codec == "" {
|
||
codec = transcodeCodecH264
|
||
}
|
||
|
||
if codec == "copy" {
|
||
return append(args, "-c:v", "copy")
|
||
}
|
||
|
||
preset := videoPreset(opts, opts.HWAccel)
|
||
hasBitrateCap := opts.TargetBitrateKbps > 0
|
||
|
||
switch {
|
||
case opts.HWAccel == "qsv" && codec == transcodeCodecH264:
|
||
if hasBitrateCap {
|
||
// VBR mode with bitrate cap instead of global_quality.
|
||
args = append(args, "-c:v", "h264_qsv", "-preset", preset,
|
||
"-b:v", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
} else {
|
||
args = append(args, "-c:v", "h264_qsv", "-preset", preset, "-global_quality", "23")
|
||
}
|
||
case opts.HWAccel == "qsv" && codec == "hevc":
|
||
if hasBitrateCap {
|
||
args = append(args, "-c:v", "hevc_qsv", "-preset", preset,
|
||
"-b:v", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
} else {
|
||
args = append(args, "-c:v", "hevc_qsv", "-preset", preset, "-global_quality", "28")
|
||
}
|
||
case opts.HWAccel == "vaapi" && codec == transcodeCodecH264:
|
||
args = append(args, "-c:v", "h264_vaapi", "-qp", "23")
|
||
if hasBitrateCap {
|
||
args = append(args,
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
}
|
||
case opts.HWAccel == "vaapi" && codec == "hevc":
|
||
args = append(args, "-c:v", "hevc_vaapi", "-qp", "28")
|
||
if hasBitrateCap {
|
||
args = append(args,
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
}
|
||
case opts.HWAccel == transcodeHWNVENC && codec == transcodeCodecH264:
|
||
args = append(args, "-c:v", "h264_nvenc", "-rc:v", "vbr")
|
||
if hasBitrateCap {
|
||
args = append(args,
|
||
"-b:v", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
} else {
|
||
args = append(args, "-cq:v", "23", "-b:v", "0")
|
||
}
|
||
case opts.HWAccel == transcodeHWNVENC && codec == "hevc":
|
||
args = append(args, "-c:v", "hevc_nvenc", "-rc:v", "vbr")
|
||
if hasBitrateCap {
|
||
args = append(args,
|
||
"-b:v", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
} else {
|
||
args = append(args, "-cq:v", "28", "-b:v", "0")
|
||
}
|
||
default:
|
||
// CPU fallback — match Jellyfin's proven browser-compatible settings.
|
||
// Force yuv420p to ensure 8-bit output (10-bit sources produce High 10
|
||
// Profile which browsers cannot decode via MSE).
|
||
if codec == "hevc" {
|
||
args = append(args, "-c:v", "libx265", "-preset", preset, "-crf", "28", "-pix_fmt", "yuv420p")
|
||
} else {
|
||
args = append(args, "-c:v", "libx264", "-preset", preset, "-crf", "23",
|
||
"-pix_fmt", "yuv420p", "-profile:v", "high", "-level", "4.1")
|
||
}
|
||
if hasBitrateCap {
|
||
args = append(args,
|
||
"-maxrate", fmt.Sprintf("%dk", opts.TargetBitrateKbps),
|
||
"-bufsize", fmt.Sprintf("%dk", opts.TargetBitrateKbps*2))
|
||
}
|
||
}
|
||
|
||
return args
|
||
}
|
||
|
||
// appendVideoFilterArgs appends the -vf selection for an encoding (non-copy)
|
||
// video stream: the subtitle burn-in chain (which includes scaling and hw
|
||
// download/upload for QSV/VAAPI) or the hwaccel-appropriate standalone scale
|
||
// filter. The ONE home of this decision — the HLS builder and the single-file
|
||
// prepare builder must always produce identical filter chains (a fix landing
|
||
// in only one of them silently ships wrong cached artifacts).
|
||
func appendVideoFilterArgs(args []string, opts TranscodeOpts) []string {
|
||
switch {
|
||
case bitmapBurnInActive(opts):
|
||
return appendBitmapSubtitleBurnInArgs(args, opts)
|
||
case opts.SubtitleBurnIn && opts.SubtitleTrackIndex >= 0:
|
||
return appendSubtitleBurnInArgs(args, opts)
|
||
case opts.HWAccel == "qsv" && opts.SoftwareVideoDecode:
|
||
return append(args, "-vf", qsvSoftwareDecodeFilter(opts.TargetResolution))
|
||
case opts.HWAccel == "vaapi" && opts.SoftwareVideoDecode:
|
||
return append(args, "-vf", vaapiSoftwareDecodeFilter(opts.TargetResolution))
|
||
case opts.HWAccel == "qsv":
|
||
return append(args, "-vf", qsvScaleFilter(opts.TargetResolution))
|
||
case opts.HWAccel == "vaapi":
|
||
return append(args, "-vf", vaapiScaleFilter(opts.TargetResolution))
|
||
case opts.HWAccel == transcodeHWNVENC:
|
||
return append(args, "-vf", nvencScaleFilter(opts.TargetResolution))
|
||
case opts.TargetResolution != "":
|
||
if scale := resolutionToScale(opts.TargetResolution); scale != "" {
|
||
return append(args, "-vf", scale)
|
||
}
|
||
}
|
||
return args
|
||
}
|
||
|
||
// TranscodesAudio reports whether a transcode with the given target audio
|
||
// codec re-encodes the audio stream. Only an explicit "copy" passes audio
|
||
// through; an empty codec runs ffmpeg's AAC default (see appendAudioArgs), so
|
||
// every consumer of the session's audio decision — live stream state, recipe
|
||
// cards, and the compat mirror — must share this predicate or the activity
|
||
// bucket flips between remux and audio across restarts.
|
||
func TranscodesAudio(targetCodecAudio string) bool {
|
||
return !strings.EqualFold(targetCodecAudio, "copy")
|
||
}
|
||
|
||
// appendAudioArgs adds audio codec arguments. Supports "copy" for passthrough,
|
||
// plus opus / aac / eac3 / ac3 as re-encode targets. EAC3 and AC3 are useful
|
||
// when we must transcode video but want to preserve surround channels for an
|
||
// HDMI receiver — both are legal in HLS fMP4 (not MPEG-TS; ensure the HLS
|
||
// packager is fMP4 when emitting these).
|
||
func appendAudioArgs(args []string, opts TranscodeOpts) []string {
|
||
// Case-insensitive so the switch agrees with TranscodesAudio for any
|
||
// client-supplied spelling.
|
||
codec := strings.ToLower(opts.TargetCodecAudio)
|
||
if codec == "" {
|
||
codec = "aac"
|
||
}
|
||
|
||
switch codec {
|
||
case "copy":
|
||
args = append(args, "-c:a", "copy")
|
||
case "opus":
|
||
args = append(args, "-c:a", "libopus", "-b:a", "192k", "-ac", "2")
|
||
case "eac3":
|
||
// Typical Dolby Digital Plus 5.1 bitrate; let the source dictate channel
|
||
// count so we preserve surround when possible.
|
||
args = append(args, "-c:a", "eac3", "-b:a", "384k")
|
||
case "ac3":
|
||
// Legacy Dolby Digital; universal AVR support.
|
||
args = append(args, "-c:a", "ac3", "-b:a", "448k")
|
||
default:
|
||
channels, bitrateKbps := resolvedAACOutputV3(opts.TargetAudioChannels, opts.TargetAudioBitrateKbps)
|
||
args = append(args, "-c:a", "aac", "-b:a", strconv.Itoa(bitrateKbps)+"k", "-ac", strconv.Itoa(channels))
|
||
}
|
||
|
||
return args
|
||
}
|
||
|
||
func resolvedAACOutputV3(targetChannels, targetBitrateKbps int) (int, int) {
|
||
channels := 2
|
||
if targetChannels == 1 {
|
||
channels = 1
|
||
} else if targetChannels >= 6 {
|
||
channels = 6
|
||
}
|
||
if targetBitrateKbps > 0 {
|
||
return channels, targetBitrateKbps
|
||
}
|
||
switch channels {
|
||
case 1:
|
||
return channels, 128
|
||
case 6:
|
||
return channels, 384
|
||
default:
|
||
return channels, 192
|
||
}
|
||
}
|
||
|
||
// appendBitmapSubtitleBurnInArgs adds burn-in arguments for BITMAP subtitle
|
||
// codecs (PGS/VOBSUB/DVB). libass's subtitles= filter cannot render bitmap
|
||
// tracks, so the decoded subtitle stream is composited onto the video with an
|
||
// overlay in a -filter_complex graph (the "Plex route"). The graph's output
|
||
// pad [vout] replaces the raw video stream in stream mapping (see
|
||
// appendStreamSelectionArgs), so -vf must never be emitted alongside this.
|
||
//
|
||
// Overlay runs at the source's native resolution FIRST and any target scaling
|
||
// happens after, so bitmap subtitle geometry is never distorted by a
|
||
// pre-scaling mismatch between the video and subtitle planes.
|
||
// eof_action=pass keeps the video flowing untouched once the subtitle stream
|
||
// ends instead of freezing the last overlay frame on screen.
|
||
//
|
||
// QSV/VAAPI composite ON the GPU via overlay_vaapi: the decoded video never
|
||
// leaves its VAAPI surface, and only the small, low-frequency subtitle bitmap is
|
||
// uploaded. This avoids the full-frame GPU→CPU→GPU roundtrip a software overlay
|
||
// forces — that roundtrip runs below realtime on 1080p sources (~0.7x), starving
|
||
// the client, and can crash the QSV buffer path with SIGBUS. See
|
||
// appendSubtitleBurnInArgs for the TEXT path, which must stay on CPU because
|
||
// libass is a software renderer. NVENC and CPU encodes keep the software overlay:
|
||
// overlay_cuda is unverified on this build, so the CUDA path retains the safe
|
||
// (if slower) roundtrip rather than risk a broken graph.
|
||
func appendBitmapSubtitleBurnInArgs(args []string, opts TranscodeOpts) []string {
|
||
// [0:s:N] indexes subtitle streams only, matching the si=N semantics of
|
||
// the text path — SubtitleTrackIndex is the embedded subtitle ordinal.
|
||
subInput := fmt.Sprintf("[0:s:%d]", opts.SubtitleTrackIndex)
|
||
|
||
var graph string
|
||
switch opts.HWAccel {
|
||
case "qsv":
|
||
if opts.SoftwareVideoDecode {
|
||
graph = softwareDecodedBitmapBurnInGraph(opts, subInput, true)
|
||
break
|
||
}
|
||
// GPU composite: upload only the subtitle bitmap, overlay it onto the
|
||
// VAAPI video surface, scale, then map to QSV for the encoder. The scale
|
||
// helper already appends the hwmap=derive_device=qsv tail.
|
||
graph = subInput + "format=bgra,hwupload[sub];" +
|
||
"[0:v:0][sub]overlay_vaapi=eof_action=pass," + qsvScaleFilter(opts.TargetResolution) + "[vout]"
|
||
case "vaapi":
|
||
if opts.SoftwareVideoDecode {
|
||
graph = softwareDecodedBitmapBurnInGraph(opts, subInput, false)
|
||
break
|
||
}
|
||
// GPU composite: same as QSV but the frames stay on VAAPI through the
|
||
// encoder, so no cross-device map is needed.
|
||
graph = subInput + "format=bgra,hwupload[sub];" +
|
||
"[0:v:0][sub]overlay_vaapi=eof_action=pass," + vaapiScaleFilter(opts.TargetResolution) + "[vout]"
|
||
default:
|
||
// NVENC and CPU: software overlay on CPU frames. Build the overlay
|
||
// fragment (subtitle input + optional post-scale) once, then wire it into
|
||
// the encode-specific pipeline.
|
||
cpuFilters := subInput + "overlay=eof_action=pass"
|
||
if scale := resolutionToScale(opts.TargetResolution); scale != "" {
|
||
cpuFilters += "," + scale
|
||
}
|
||
if opts.HWAccel == transcodeHWNVENC {
|
||
// Download to CPU for the overlay, then re-upload to CUDA.
|
||
graph = "[0:v:0]hwdownload,format=yuv420p[vmain];[vmain]" + cpuFilters +
|
||
",format=nv12,hwupload_cuda[vout]"
|
||
} else {
|
||
// CPU encoding: overlay directly on decoded frames.
|
||
graph = "[0:v:0]" + cpuFilters + "[vout]"
|
||
}
|
||
}
|
||
|
||
return append(args, "-filter_complex", graph)
|
||
}
|
||
|
||
// softwareDecodedBitmapBurnInGraph composites decoded CPU frames and uploads
|
||
// the finished NV12 frames for the hardware encoder. It is the Hi10 AVC
|
||
// counterpart of the all-hardware overlay_vaapi graph above.
|
||
func softwareDecodedBitmapBurnInGraph(opts TranscodeOpts, subInput string, qsv bool) string {
|
||
filters := subInput + "overlay=eof_action=pass"
|
||
if scale := resolutionToScale(opts.TargetResolution); scale != "" {
|
||
filters += "," + scale
|
||
}
|
||
filters += ",format=nv12,hwupload"
|
||
if qsv {
|
||
filters += ",hwmap=derive_device=qsv,format=qsv"
|
||
}
|
||
return "[0:v:0]format=yuv420p[vmain];[vmain]" + filters + "[vout]"
|
||
}
|
||
|
||
// appendSubtitleBurnInArgs adds subtitle burn-in filter arguments for TEXT
|
||
// subtitle codecs (SRT/ASS/…) via the libass-based subtitles= filter; bitmap
|
||
// codecs take the overlay path in appendBitmapSubtitleBurnInArgs.
|
||
// For CPU encoding, the filter chain is: [scale,]subtitles.
|
||
// For QSV/VAAPI, frames must be downloaded from hardware, processed on CPU,
|
||
// then re-uploaded: hwdownload → format=yuv420p → [scale,] subtitles → hwupload → hwmap.
|
||
func appendSubtitleBurnInArgs(args []string, opts TranscodeOpts) []string {
|
||
scale := resolutionToScale(opts.TargetResolution)
|
||
subtitleInputPath := opts.InputPath
|
||
if opts.subtitleFilterInputPath != "" {
|
||
subtitleInputPath = opts.subtitleFilterInputPath
|
||
}
|
||
subFilter := fmt.Sprintf("subtitles='%s':si=%d",
|
||
escapeFilterPath(subtitleInputPath), opts.SubtitleTrackIndex)
|
||
|
||
// Build the CPU filter portion: scale (if any) then subtitle overlay.
|
||
// Scale must come before subtitles so text is rendered at target resolution.
|
||
var cpuFilters string
|
||
if scale != "" {
|
||
cpuFilters = scale + "," + subFilter
|
||
} else {
|
||
cpuFilters = subFilter
|
||
}
|
||
|
||
switch opts.HWAccel {
|
||
case "qsv":
|
||
if opts.SoftwareVideoDecode {
|
||
vf := "format=yuv420p," + cpuFilters + ",format=nv12,hwupload,hwmap=derive_device=qsv,format=qsv"
|
||
return append(args, "-vf", vf)
|
||
}
|
||
// VAAPI→QSV pipeline: download from VAAPI surface to CPU, apply subtitle
|
||
// and scale filters, convert to nv12 (required by hwupload for VAAPI
|
||
// surfaces), upload back to VAAPI, then map to QSV for the encoder.
|
||
vf := "hwdownload,format=yuv420p," + cpuFilters + ",format=nv12,hwupload,hwmap=derive_device=qsv,format=qsv"
|
||
args = append(args, "-vf", vf)
|
||
case "vaapi":
|
||
if opts.SoftwareVideoDecode {
|
||
vf := "format=yuv420p," + cpuFilters + ",format=nv12,hwupload"
|
||
return append(args, "-vf", vf)
|
||
}
|
||
// VAAPI-only: download, apply CPU filters, convert to nv12, upload back.
|
||
vf := "hwdownload,format=yuv420p," + cpuFilters + ",format=nv12,hwupload"
|
||
args = append(args, "-vf", vf)
|
||
case transcodeHWNVENC:
|
||
// NVENC/CUDA: download to CPU for subtitle rendering, then upload back.
|
||
vf := "hwdownload,format=yuv420p," + cpuFilters + ",format=nv12,hwupload_cuda"
|
||
args = append(args, "-vf", vf)
|
||
default:
|
||
// CPU encoding: filters run directly on decoded frames.
|
||
args = append(args, "-vf", cpuFilters)
|
||
}
|
||
|
||
return args
|
||
}
|
||
|
||
// resolutionToScale returns an ffmpeg scale filter string for the target resolution.
|
||
func resolutionToScale(res string) string {
|
||
switch res {
|
||
case "2160p":
|
||
return "scale=-2:2160"
|
||
case "1080p":
|
||
return "scale=-2:1080"
|
||
case "720p":
|
||
return "scale=-2:720"
|
||
case "480p":
|
||
return "scale=-2:480"
|
||
case "420p":
|
||
return "scale=-2:420"
|
||
case "328p":
|
||
return "scale=-2:328"
|
||
default:
|
||
return ""
|
||
}
|
||
}
|
||
|
||
// qsvScaleFilter returns the VAAPI→QSV filter chain with optional resolution scaling.
|
||
func qsvScaleFilter(res string) string {
|
||
switch res {
|
||
case "2160p":
|
||
return "scale_vaapi=w=-2:h=2160:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
case "1080p":
|
||
return "scale_vaapi=w=-2:h=1080:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
case "720p":
|
||
return "scale_vaapi=w=-2:h=720:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
case "480p":
|
||
return "scale_vaapi=w=-2:h=480:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
case "420p":
|
||
return "scale_vaapi=w=-2:h=420:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
case "328p":
|
||
return "scale_vaapi=w=-2:h=328:format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
default:
|
||
return "scale_vaapi=format=nv12,hwmap=derive_device=qsv,format=qsv"
|
||
}
|
||
}
|
||
|
||
func qsvSoftwareDecodeFilter(res string) string {
|
||
cpuFilters := ""
|
||
if scale := resolutionToScale(res); scale != "" {
|
||
cpuFilters = scale + ","
|
||
}
|
||
// High 10 AVC is decoded on the CPU. Scale those software frames before
|
||
// upload, matching the proven text-subtitle path; uploading first and then
|
||
// invoking scale_vaapi can leave the VAAPI/QSV graph alive without ever
|
||
// producing its initial HLS window.
|
||
return cpuFilters + "format=nv12,hwupload,hwmap=derive_device=qsv,format=qsv"
|
||
}
|
||
|
||
// vaapiScaleFilter keeps VAAPI frames in hardware and converts them to a
|
||
// browser-compatible encoder format. Using the CPU scale filter on VAAPI frames
|
||
// causes FFmpeg auto_scale format-negotiation failures.
|
||
func vaapiScaleFilter(res string) string {
|
||
switch res {
|
||
case "2160p":
|
||
return "scale_vaapi=w=-2:h=2160:format=nv12"
|
||
case "1080p":
|
||
return "scale_vaapi=w=-2:h=1080:format=nv12"
|
||
case "720p":
|
||
return "scale_vaapi=w=-2:h=720:format=nv12"
|
||
case "480p":
|
||
return "scale_vaapi=w=-2:h=480:format=nv12"
|
||
case "420p":
|
||
return "scale_vaapi=w=-2:h=420:format=nv12"
|
||
case "328p":
|
||
return "scale_vaapi=w=-2:h=328:format=nv12"
|
||
default:
|
||
return "scale_vaapi=format=nv12"
|
||
}
|
||
}
|
||
|
||
func vaapiSoftwareDecodeFilter(res string) string {
|
||
cpuFilters := ""
|
||
if scale := resolutionToScale(res); scale != "" {
|
||
cpuFilters = scale + ","
|
||
}
|
||
return cpuFilters + "format=nv12,hwupload"
|
||
}
|
||
|
||
func nvencScaleFilter(res string) string {
|
||
switch res {
|
||
case "2160p":
|
||
return "scale_cuda=w=-2:h=2160:format=nv12"
|
||
case "1080p":
|
||
return "scale_cuda=w=-2:h=1080:format=nv12"
|
||
case "720p":
|
||
return "scale_cuda=w=-2:h=720:format=nv12"
|
||
case "480p":
|
||
return "scale_cuda=w=-2:h=480:format=nv12"
|
||
case "420p":
|
||
return "scale_cuda=w=-2:h=420:format=nv12"
|
||
case "328p":
|
||
return "scale_cuda=w=-2:h=328:format=nv12"
|
||
default:
|
||
return "scale_cuda=format=nv12"
|
||
}
|
||
}
|
||
|
||
// filterPathReplacer escapes special characters in file paths for ffmpeg filter syntax.
|
||
var filterPathReplacer = strings.NewReplacer(
|
||
"'", "'\\''",
|
||
"[", "\\[",
|
||
"]", "\\]",
|
||
";", "\\;",
|
||
",", "\\,",
|
||
)
|
||
|
||
// escapeFilterPath escapes special characters in file paths for ffmpeg filter syntax.
|
||
func escapeFilterPath(path string) string {
|
||
return filterPathReplacer.Replace(path)
|
||
}
|
||
|
||
const subtitleFilterAliasName = "subtitle-source.media"
|
||
|
||
// prepareSubtitleFilterInput avoids passing the library filename through
|
||
// FFmpeg's nested filter parsers. In particular, an apostrophe can be consumed
|
||
// by the filtergraph parser even after it was escaped for the subtitles filter.
|
||
// A stable alias in the session directory keeps the media input unchanged and
|
||
// is retained for seek restarts with the rest of TranscodeOpts.
|
||
func prepareSubtitleFilterInput(opts *TranscodeOpts) error {
|
||
if !opts.SubtitleBurnIn || opts.SubtitleTrackIndex < 0 || NeedsBurnIn(opts.SubtitleCodec) {
|
||
return nil
|
||
}
|
||
|
||
aliasPath := filepath.Join(opts.OutputDir, subtitleFilterAliasName)
|
||
target, err := os.Readlink(aliasPath)
|
||
switch {
|
||
case err == nil:
|
||
if target != opts.InputPath {
|
||
return fmt.Errorf("prepare subtitle filter input: alias targets unexpected source")
|
||
}
|
||
case errors.Is(err, os.ErrNotExist):
|
||
if err := os.Symlink(opts.InputPath, aliasPath); err != nil {
|
||
return fmt.Errorf("prepare subtitle filter input: %w", err)
|
||
}
|
||
default:
|
||
return fmt.Errorf("prepare subtitle filter input: inspect alias: %w", err)
|
||
}
|
||
|
||
opts.subtitleFilterInputPath = aliasPath
|
||
return nil
|
||
}
|
||
|
||
// minManifestSegments is the standard startup lead for actively encoded HLS.
|
||
// True video transcodes benefit from a larger cushion so playback does not
|
||
// outrun the encoder immediately after the first frame appears.
|
||
const minManifestSegments = 3
|
||
|
||
// minCopyManifestSegments is the startup lead for codec-copy sessions.
|
||
// Copying video while transcoding only audio can produce startup files far
|
||
// faster than real-time encoding, so waiting for 3 full segments adds
|
||
// unnecessary latency at playback start.
|
||
const minCopyManifestSegments = 2
|
||
|
||
func startupSegmentRequirement(opts TranscodeOpts) int {
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
|
||
return minCopyManifestSegments
|
||
}
|
||
return minManifestSegments
|
||
}
|
||
|
||
// GetManifest returns the HLS m3u8 manifest content.
|
||
// It returns ErrManifestNotReady if the manifest does not yet contain enough
|
||
// segments for reliable HLS playback (see minManifestSegments).
|
||
func (s *TranscodeSession) GetManifest() ([]byte, error) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
manifestPath := filepath.Join(s.outputDir, "stream.m3u8")
|
||
data, err := os.ReadFile(manifestPath)
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
if !s.running {
|
||
if s.restarting {
|
||
return nil, ErrManifestNotReady
|
||
}
|
||
if s.waitErr != nil {
|
||
stderr := truncateStderr(s.stderr.String())
|
||
if stderr != "" {
|
||
return nil, fmt.Errorf("%w: %v (stderr: %s)", ErrTranscodeFailed, s.waitErr, stderr)
|
||
}
|
||
return nil, fmt.Errorf("%w: %v", ErrTranscodeFailed, s.waitErr)
|
||
}
|
||
return nil, ErrTranscodeFailed
|
||
}
|
||
return nil, ErrManifestNotReady
|
||
}
|
||
return nil, fmt.Errorf("read manifest: %w", err)
|
||
}
|
||
|
||
requiredSegments := startupSegmentRequirement(s.opts)
|
||
|
||
// Wait until enough startup media exists before serving. Counting #EXTINF
|
||
// lines alone is not enough for FFmpeg's live-written manifest because the
|
||
// playlist can reference copy-mode segments before the files are fully
|
||
// flushed to disk, especially on resumed sessions with a non-zero media
|
||
// sequence. Requiring the referenced startup files prevents the browser from
|
||
// stalling on its very first segment fetch.
|
||
if s.running && !startupFilesReady(data, s.outputDir, requiredSegments) {
|
||
return nil, ErrManifestNotReady
|
||
}
|
||
if strings.EqualFold(s.opts.TargetCodecVideo, "copy") {
|
||
if err := validateCopyPlaybackManifest(data); err != nil {
|
||
return nil, fmt.Errorf("invalid copy playback manifest: %w", err)
|
||
}
|
||
}
|
||
|
||
return data, nil
|
||
}
|
||
|
||
// WaitForManifest polls until the manifest is ready for playback or the timeout
|
||
// expires. It keeps the initial request open long enough for FFmpeg to write
|
||
// the first safe playback window instead of forcing the client to race a 503.
|
||
func (s *TranscodeSession) WaitForManifest(timeout time.Duration) ([]byte, error) {
|
||
deadline := time.After(timeout)
|
||
for {
|
||
manifest, err := s.GetManifest()
|
||
if err == nil {
|
||
return manifest, nil
|
||
}
|
||
if err != nil && err != ErrManifestNotReady {
|
||
return nil, err
|
||
}
|
||
|
||
select {
|
||
case <-deadline:
|
||
return nil, s.manifestTimeoutError(timeout)
|
||
case <-time.After(100 * time.Millisecond):
|
||
}
|
||
}
|
||
}
|
||
|
||
// BuildPlaybackManifest returns the manifest we should expose to clients.
|
||
//
|
||
// Copy-video sessions always expose FFmpeg's real manifest so the playlist
|
||
// timing matches the variable-length fragments FFmpeg actually writes and the
|
||
// seekable window reflects what FFmpeg has produced so far. Encoded transcodes
|
||
// use the synthetic full VOD manifest only while its segment count is bounded;
|
||
// longer media uses FFmpeg's real sliding playlist.
|
||
func (s *TranscodeSession) BuildPlaybackManifest(segPrefix, rawQuery string) ([]byte, error) {
|
||
opts := s.Opts()
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") ||
|
||
!CanGenerateSyntheticManifest(opts.TotalDuration, opts.SegmentDuration) {
|
||
// Copy-video, unknown-duration, or oversized sessions must use FFmpeg's
|
||
// real manifest.
|
||
manifest, err := s.WaitForManifest(ManifestStartupTimeout)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
manifest = stabilizeCopyHLSRemountTimeline(manifest, opts)
|
||
return RewriteManifestPaths(manifest, segPrefix, rawQuery)
|
||
}
|
||
|
||
return s.GenerateFullManifest(segPrefix, rawQuery), nil
|
||
}
|
||
|
||
// stabilizeCopyHLSRemountTimeline marks a bounded copy-HLS playlist as an
|
||
// append-only EVENT and gives it a precise start at the generation origin.
|
||
//
|
||
// FFmpeg's real playlist has no end tag while it is growing, so players treat
|
||
// it as live. Media3 consequently chooses the production edge whenever a track
|
||
// change rebuilds its MediaSource, even though protocol v3 deliberately reused
|
||
// the same A/V generation and the requested historical position is still in
|
||
// the playlist. EVENT plus a positive EXT-X-START disables that projection and
|
||
// leaves the client free to restore the source position on the stable timeline.
|
||
//
|
||
// The EVENT promise is made only when the known complete media fits within the
|
||
// same 50,000-segment bound passed to FFmpeg. Such a playlist never evicts an
|
||
// entry, and segment files are retained because delete_segments is not enabled.
|
||
// Unknown or oversized real playlists remain ordinary sliding live playlists.
|
||
func stabilizeCopyHLSRemountTimeline(manifest []byte, opts TranscodeOpts) []byte {
|
||
if !strings.EqualFold(opts.TargetCodecVideo, "copy") ||
|
||
opts.TotalDuration <= 0 ||
|
||
!CanGenerateSyntheticManifest(opts.TotalDuration, opts.SegmentDuration) ||
|
||
validateManifestHeader(manifest) != nil {
|
||
return manifest
|
||
}
|
||
|
||
lines := bytes.Split(manifest, []byte("\n"))
|
||
hasPlaylistType := false
|
||
hasStart := false
|
||
insertAfter := 0
|
||
for i, line := range lines {
|
||
trimmed := bytes.TrimSpace(line)
|
||
switch {
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-PLAYLIST-TYPE:")):
|
||
hasPlaylistType = true
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-START:")):
|
||
hasStart = true
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-VERSION:")):
|
||
insertAfter = i + 1
|
||
}
|
||
}
|
||
if hasPlaylistType && hasStart {
|
||
return manifest
|
||
}
|
||
|
||
tags := make([][]byte, 0, 2)
|
||
if !hasPlaylistType {
|
||
tags = append(tags, []byte("#EXT-X-PLAYLIST-TYPE:EVENT"))
|
||
}
|
||
if !hasStart {
|
||
tags = append(tags, []byte(fmt.Sprintf("#EXT-X-START:TIME-OFFSET=%.3f,PRECISE=YES", remountStartOffsetSeconds)))
|
||
}
|
||
result := make([][]byte, 0, len(lines)+len(tags))
|
||
result = append(result, lines[:insertAfter]...)
|
||
result = append(result, tags...)
|
||
result = append(result, lines[insertAfter:]...)
|
||
return bytes.Join(result, []byte("\n"))
|
||
}
|
||
|
||
// SourceTimelineQueryParam opts a real transcode manifest into source-time
|
||
// alignment for compatibility clients that apply their resume position to the
|
||
// HLS timeline themselves.
|
||
const SourceTimelineQueryParam = "source_timeline"
|
||
|
||
// BuildSourceAlignedPlaybackManifest builds the normal playback manifest and,
|
||
// when it is a seeked real playlist, prepends a virtual unavailable span so the
|
||
// first produced segment retains its source-time position. Synthetic manifests
|
||
// already cover the full source timeline and need no adjustment.
|
||
func (s *TranscodeSession) BuildSourceAlignedPlaybackManifest(segPrefix, rawQuery string) ([]byte, error) {
|
||
manifest, err := s.BuildPlaybackManifest(segPrefix, rawQuery)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
opts := s.Opts()
|
||
usesRealManifest := strings.EqualFold(opts.TargetCodecVideo, "copy") ||
|
||
!CanGenerateSyntheticManifest(opts.TotalDuration, opts.SegmentDuration)
|
||
if opts.SeekSeconds <= 0 || !usesRealManifest {
|
||
return manifest, nil
|
||
}
|
||
|
||
gapURI := segPrefix + "source_timeline_gap" + hlsSegmentExtension(opts)
|
||
if rawQuery != "" {
|
||
gapURI += "?" + rawQuery
|
||
}
|
||
return AlignRealManifestToSourceTimeline(manifest, opts, gapURI)
|
||
}
|
||
|
||
// AlignRealManifestToSourceTimeline prepends bounded EXT-X-GAP segments to a
|
||
// seeked FFmpeg playlist. The gaps contribute the omitted source time while
|
||
// allowing clients to seek to the original source position.
|
||
func AlignRealManifestToSourceTimeline(manifest []byte, opts TranscodeOpts, gapURI string) ([]byte, error) {
|
||
if opts.SeekSeconds <= 0 {
|
||
return manifest, nil
|
||
}
|
||
timeline, err := parseManifestTimeline(manifest)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
if len(timeline.entries) == 0 {
|
||
return nil, fmt.Errorf("manifest contains no media segments")
|
||
}
|
||
|
||
segmentDuration := opts.SegmentDuration
|
||
if segmentDuration <= 0 {
|
||
segmentDuration = defaultSegmentDuration
|
||
}
|
||
firstSegment := timeline.entries[0].number
|
||
advancedSegments := max(0, firstSegment-opts.StartSegmentNumber)
|
||
gapDuration := opts.SeekSeconds + float64(advancedSegments*segmentDuration)
|
||
if gapDuration <= 0 {
|
||
return manifest, nil
|
||
}
|
||
if gapURI == "" {
|
||
gapURI = "source_timeline_gap" + hlsSegmentExtension(opts)
|
||
}
|
||
|
||
lines := bytes.Split(manifest, []byte("\n"))
|
||
targetDuration := segmentDuration
|
||
for _, line := range lines {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if !bytes.HasPrefix(trimmed, []byte("#EXT-X-TARGETDURATION:")) {
|
||
continue
|
||
}
|
||
value := strings.TrimSpace(strings.TrimPrefix(string(trimmed), "#EXT-X-TARGETDURATION:"))
|
||
parsed, parseErr := strconv.Atoi(value)
|
||
if parseErr != nil || parsed <= 0 {
|
||
return nil, fmt.Errorf("parse manifest target duration %q", value)
|
||
}
|
||
targetDuration = parsed
|
||
break
|
||
}
|
||
|
||
// Keep the real segment's media sequence aligned with its FFmpeg segment
|
||
// number when possible. Very large seeks remain bounded by the same limit as
|
||
// generated VOD manifests; their gap durations grow instead of their count.
|
||
gapCount := max(1, firstSegment)
|
||
if gapCount > maxSyntheticManifestSegments {
|
||
gapCount = maxSyntheticManifestSegments
|
||
}
|
||
gapSegmentDuration := gapDuration / float64(gapCount)
|
||
requiredTargetDuration := int(math.Ceil(gapSegmentDuration))
|
||
if requiredTargetDuration > targetDuration {
|
||
targetDuration = requiredTargetDuration
|
||
}
|
||
mediaSequence := max(0, firstSegment-gapCount)
|
||
|
||
result := make([][]byte, 0, len(lines)+gapCount*3)
|
||
insertedGap := false
|
||
foundSequence := false
|
||
foundVersion := false
|
||
for _, line := range lines {
|
||
trimmed := bytes.TrimSpace(line)
|
||
switch {
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-VERSION:")):
|
||
foundVersion = true
|
||
value := strings.TrimSpace(strings.TrimPrefix(string(trimmed), "#EXT-X-VERSION:"))
|
||
version, parseErr := strconv.Atoi(value)
|
||
if parseErr != nil {
|
||
return nil, fmt.Errorf("parse manifest version %q: %w", value, parseErr)
|
||
}
|
||
if version < 8 {
|
||
line = []byte("#EXT-X-VERSION:8")
|
||
}
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-TARGETDURATION:")):
|
||
line = []byte(fmt.Sprintf("#EXT-X-TARGETDURATION:%d", targetDuration))
|
||
case bytes.HasPrefix(trimmed, []byte("#EXT-X-MEDIA-SEQUENCE:")):
|
||
foundSequence = true
|
||
line = []byte(fmt.Sprintf("#EXT-X-MEDIA-SEQUENCE:%d", mediaSequence))
|
||
case bytes.HasPrefix(trimmed, []byte("#EXTINF:")) && !insertedGap:
|
||
if !foundVersion {
|
||
result = append(result, []byte("#EXT-X-VERSION:8"))
|
||
foundVersion = true
|
||
}
|
||
if !foundSequence {
|
||
result = append(result, []byte(fmt.Sprintf("#EXT-X-MEDIA-SEQUENCE:%d", mediaSequence)))
|
||
foundSequence = true
|
||
}
|
||
for range gapCount {
|
||
result = append(result,
|
||
[]byte("#EXT-X-GAP"),
|
||
[]byte(fmt.Sprintf("#EXTINF:%.6f,", gapSegmentDuration)),
|
||
[]byte(gapURI),
|
||
)
|
||
}
|
||
insertedGap = true
|
||
}
|
||
result = append(result, line)
|
||
}
|
||
if !insertedGap {
|
||
return nil, fmt.Errorf("manifest contains no segment duration")
|
||
}
|
||
return bytes.Join(result, []byte("\n")), nil
|
||
}
|
||
|
||
// CanGenerateSyntheticManifest reports whether a complete VOD playlist fits
|
||
// within the shared segment-count bound. Callers outside playback use the same
|
||
// decision so native and compatibility manifests cannot drift.
|
||
func CanGenerateSyntheticManifest(totalDuration float64, segmentDuration int) bool {
|
||
if totalDuration <= 0 || math.IsNaN(totalDuration) || math.IsInf(totalDuration, 0) {
|
||
return false
|
||
}
|
||
if segmentDuration <= 0 {
|
||
segmentDuration = defaultSegmentDuration
|
||
}
|
||
return totalDuration <= float64(segmentDuration)*maxSyntheticManifestSegments
|
||
}
|
||
|
||
func firstNonEmptyManifestLine(manifest []byte) []byte {
|
||
for line := range bytes.SplitSeq(manifest, []byte("\n")) {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if len(trimmed) > 0 {
|
||
return trimmed
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func validateManifestHeader(manifest []byte) error {
|
||
if len(bytes.TrimSpace(manifest)) == 0 {
|
||
return fmt.Errorf("manifest is empty")
|
||
}
|
||
if line := firstNonEmptyManifestLine(manifest); !bytes.Equal(line, []byte("#EXTM3U")) {
|
||
return fmt.Errorf("manifest missing #EXTM3U header")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func parseTargetDuration(manifest []byte) (int, error) {
|
||
for line := range bytes.SplitSeq(manifest, []byte("\n")) {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if bytes.HasPrefix(trimmed, []byte("#EXT-X-TARGETDURATION:")) {
|
||
value := strings.TrimSpace(strings.TrimPrefix(string(trimmed), "#EXT-X-TARGETDURATION:"))
|
||
targetDuration, err := strconv.Atoi(value)
|
||
if err != nil {
|
||
return 0, fmt.Errorf("parse target duration %q: %w", value, err)
|
||
}
|
||
return targetDuration, nil
|
||
}
|
||
}
|
||
return 0, fmt.Errorf("manifest missing #EXT-X-TARGETDURATION")
|
||
}
|
||
|
||
func validateCopyPlaybackManifest(manifest []byte) error {
|
||
if err := validateManifestHeader(manifest); err != nil {
|
||
return err
|
||
}
|
||
|
||
targetDuration, err := parseTargetDuration(manifest)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if targetDuration <= 0 {
|
||
return fmt.Errorf("manifest target duration must be positive, got %d", targetDuration)
|
||
}
|
||
|
||
timeline, err := parseManifestTimeline(manifest)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
if len(timeline.entries) == 0 {
|
||
return fmt.Errorf("manifest contains no playable media segments")
|
||
}
|
||
for _, entry := range timeline.entries {
|
||
if entry.duration <= 0 {
|
||
return fmt.Errorf("segment %d has non-positive duration %.6f", entry.number, entry.duration)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func extractMapURI(line []byte) string {
|
||
const marker = `URI="`
|
||
text := string(line)
|
||
start := strings.Index(text, marker)
|
||
if start < 0 {
|
||
return ""
|
||
}
|
||
start += len(marker)
|
||
end := strings.Index(text[start:], `"`)
|
||
if end < 0 {
|
||
return ""
|
||
}
|
||
return text[start : start+end]
|
||
}
|
||
|
||
func manifestURIToFilename(uri string) string {
|
||
base, _, _ := strings.Cut(uri, "?")
|
||
return filepath.Base(base)
|
||
}
|
||
|
||
func manifestStartupFiles(manifest []byte, maxSegments int) ([]string, int) {
|
||
files := make([]string, 0, maxSegments+1)
|
||
segmentCount := 0
|
||
|
||
for line := range bytes.SplitSeq(manifest, []byte("\n")) {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if len(trimmed) == 0 {
|
||
continue
|
||
}
|
||
if bytes.HasPrefix(trimmed, []byte("#EXT-X-MAP:")) {
|
||
if uri := extractMapURI(trimmed); uri != "" {
|
||
files = append(files, manifestURIToFilename(uri))
|
||
}
|
||
continue
|
||
}
|
||
if trimmed[0] == '#' {
|
||
continue
|
||
}
|
||
|
||
files = append(files, manifestURIToFilename(string(trimmed)))
|
||
segmentCount++
|
||
if segmentCount >= maxSegments {
|
||
break
|
||
}
|
||
}
|
||
|
||
return files, segmentCount
|
||
}
|
||
|
||
func startupFilesReady(manifest []byte, outputDir string, requiredSegments int) bool {
|
||
files, segmentCount := manifestStartupFiles(manifest, requiredSegments)
|
||
if segmentCount < requiredSegments {
|
||
return false
|
||
}
|
||
|
||
for _, name := range files {
|
||
info, err := os.Stat(filepath.Join(outputDir, name))
|
||
if err != nil || info.Size() <= 0 {
|
||
return false
|
||
}
|
||
}
|
||
|
||
return true
|
||
}
|
||
|
||
type manifestSegmentEntry struct {
|
||
number int
|
||
duration float64
|
||
}
|
||
|
||
type manifestTimeline struct {
|
||
mediaSequence int
|
||
entries []manifestSegmentEntry
|
||
}
|
||
|
||
func parseManifestTimeline(manifest []byte) (manifestTimeline, error) {
|
||
if err := validateManifestHeader(manifest); err != nil {
|
||
return manifestTimeline{}, err
|
||
}
|
||
|
||
timeline := manifestTimeline{}
|
||
currentNumber := 0
|
||
var pendingDuration float64
|
||
var haveDuration bool
|
||
|
||
for line := range bytes.SplitSeq(manifest, []byte("\n")) {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if len(trimmed) == 0 {
|
||
continue
|
||
}
|
||
|
||
if bytes.HasPrefix(trimmed, []byte("#EXT-X-MEDIA-SEQUENCE:")) {
|
||
value := strings.TrimSpace(strings.TrimPrefix(string(trimmed), "#EXT-X-MEDIA-SEQUENCE:"))
|
||
sequence, err := strconv.Atoi(value)
|
||
if err != nil {
|
||
return manifestTimeline{}, fmt.Errorf("parse media sequence %q: %w", value, err)
|
||
}
|
||
timeline.mediaSequence = sequence
|
||
currentNumber = sequence
|
||
continue
|
||
}
|
||
|
||
if bytes.HasPrefix(trimmed, []byte("#EXTINF:")) {
|
||
value := strings.TrimSpace(strings.TrimSuffix(strings.TrimPrefix(string(trimmed), "#EXTINF:"), ","))
|
||
duration, err := strconv.ParseFloat(value, 64)
|
||
if err != nil {
|
||
return manifestTimeline{}, fmt.Errorf("parse segment duration %q: %w", value, err)
|
||
}
|
||
pendingDuration = duration
|
||
haveDuration = true
|
||
continue
|
||
}
|
||
|
||
if trimmed[0] == '#' {
|
||
continue
|
||
}
|
||
|
||
if !haveDuration {
|
||
continue
|
||
}
|
||
|
||
segmentNumber := currentNumber
|
||
if parsed, err := ParseSegmentNumber(filepath.Base(string(trimmed))); err == nil {
|
||
segmentNumber = parsed
|
||
}
|
||
|
||
timeline.entries = append(timeline.entries, manifestSegmentEntry{
|
||
number: segmentNumber,
|
||
duration: pendingDuration,
|
||
})
|
||
currentNumber = segmentNumber + 1
|
||
haveDuration = false
|
||
}
|
||
|
||
return timeline, nil
|
||
}
|
||
|
||
func hlsSegmentExtension(opts TranscodeOpts) string {
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") && !IsMPEG2VideoCodec(opts.SourceVideoCodec) {
|
||
return ".m4s"
|
||
}
|
||
return ".ts"
|
||
}
|
||
|
||
func segmentFilename(segNum int, opts TranscodeOpts) string {
|
||
return fmt.Sprintf("seg_%05d%s", segNum, hlsSegmentExtension(opts))
|
||
}
|
||
|
||
func segmentWaitTimeout(segmentDuration int) time.Duration {
|
||
if segmentDuration <= 0 {
|
||
segmentDuration = defaultSegmentDuration
|
||
}
|
||
timeout := time.Duration(segmentDuration)*time.Second + segmentWaitGrace
|
||
if timeout < minSegmentWait {
|
||
timeout = minSegmentWait
|
||
}
|
||
if timeout > maxSegmentWait {
|
||
timeout = maxSegmentWait
|
||
}
|
||
return timeout
|
||
}
|
||
|
||
func staleProducedWindow(segmentDuration int) time.Duration {
|
||
if segmentDuration <= 0 {
|
||
segmentDuration = defaultSegmentDuration
|
||
}
|
||
window := 2*time.Duration(segmentDuration)*time.Second + segmentWaitGrace
|
||
if window < minStaleProducedWindow {
|
||
window = minStaleProducedWindow
|
||
}
|
||
return window
|
||
}
|
||
|
||
// SegmentProgress reports the highest manifest-referenced segment that exists
|
||
// on disk with data. This is the produced media source of truth.
|
||
func (s *TranscodeSession) SegmentProgress(time.Time) SegmentProgress {
|
||
s.mu.Lock()
|
||
opts := s.opts
|
||
progress := SegmentProgress{
|
||
ProducedHead: opts.StartSegmentNumber - 1,
|
||
Running: s.running,
|
||
Restarting: s.restarting,
|
||
StartSegmentNumber: opts.StartSegmentNumber,
|
||
SegmentDuration: opts.SegmentDuration,
|
||
LastRequestedSegment: s.lastRequestedSegment,
|
||
}
|
||
s.mu.Unlock()
|
||
|
||
if progress.SegmentDuration <= 0 {
|
||
progress.SegmentDuration = defaultSegmentDuration
|
||
}
|
||
|
||
manifestPath := filepath.Join(s.outputDir, "stream.m3u8")
|
||
manifestInfo, statErr := os.Stat(manifestPath)
|
||
if statErr != nil {
|
||
return progress
|
||
}
|
||
progress.HasManifest = true
|
||
progress.ManifestModTime = manifestInfo.ModTime()
|
||
|
||
manifest, err := os.ReadFile(manifestPath)
|
||
if err != nil {
|
||
return progress
|
||
}
|
||
timeline, err := parseManifestTimeline(manifest)
|
||
if err != nil {
|
||
return progress
|
||
}
|
||
|
||
for _, entry := range timeline.entries {
|
||
segmentPath := filepath.Join(s.outputDir, segmentFilename(entry.number, opts))
|
||
info, err := os.Stat(segmentPath)
|
||
if err != nil || info.Size() <= 0 {
|
||
continue
|
||
}
|
||
progress.ProducedCount++
|
||
if entry.number > progress.ProducedHead {
|
||
progress.ProducedHead = entry.number
|
||
}
|
||
if info.ModTime().After(progress.LastProducedAt) {
|
||
progress.LastProducedAt = info.ModTime()
|
||
}
|
||
}
|
||
|
||
return progress
|
||
}
|
||
|
||
// SegmentRecoveryDecision determines whether a missing segment should briefly
|
||
// wait for ffmpeg or immediately use the seek-restart path.
|
||
func (s *TranscodeSession) SegmentRecoveryDecision(segNum int, now time.Time) SegmentRecoveryDecision {
|
||
progress := s.SegmentProgress(now)
|
||
decision := SegmentRecoveryDecision{
|
||
WaitTimeout: segmentWaitTimeout(progress.SegmentDuration),
|
||
RestartOnTimeout: true,
|
||
Progress: progress,
|
||
}
|
||
|
||
switch {
|
||
// Restarting must be checked before Running: the restart window runs
|
||
// with running=false, and a concurrent segment request must wait out
|
||
// the in-flight restart rather than trigger another one. Dueling
|
||
// restarts keep preempting the segment the player is blocked on,
|
||
// which surfaces as a seek/intro-skip freeze (issue #243).
|
||
case progress.Restarting:
|
||
decision.Wait = true
|
||
decision.WaitTimeout = activeSegmentWait
|
||
decision.RestartOnTimeout = false
|
||
decision.Reason = "transcode_restarting"
|
||
case !progress.Running:
|
||
decision.Reason = "transcode_not_running"
|
||
case segNum < progress.StartSegmentNumber:
|
||
decision.Reason = "before_start_segment"
|
||
case segNum <= progress.ProducedHead:
|
||
decision.Reason = "segment_missing_behind_produced_head"
|
||
case !progress.HasManifest:
|
||
if segNum <= progress.StartSegmentNumber+1 {
|
||
decision.Wait = true
|
||
decision.WaitTimeout = activeSegmentWait
|
||
decision.RestartOnTimeout = false
|
||
decision.Reason = "startup_manifest_not_ready"
|
||
} else {
|
||
decision.Reason = "startup_request_beyond_window"
|
||
}
|
||
case segNum > progress.ProducedHead+maxSequentialMissingSegments:
|
||
decision.Reason = "request_beyond_produced_window"
|
||
case progress.ProducedHead >= progress.StartSegmentNumber && now.Sub(progress.LastProducedAt) > staleProducedWindow(progress.SegmentDuration):
|
||
decision.Reason = "produced_output_stale"
|
||
default:
|
||
decision.Wait = true
|
||
decision.WaitTimeout = activeSegmentWait
|
||
decision.RestartOnTimeout = false
|
||
decision.Reason = "near_produced_head"
|
||
}
|
||
|
||
return decision
|
||
}
|
||
|
||
// GenerateFullManifest builds a complete VOD-style HLS manifest that lists
|
||
// every segment for the full media duration, matching Jellyfin's approach.
|
||
// The player can seek to any position immediately; the backend produces
|
||
// segments on demand when they are requested via HandleGetTranscodeSegment.
|
||
//
|
||
// segPrefix is prepended to each segment filename (e.g. "segment/") and
|
||
// rawQuery is appended as a query string (e.g. auth tokens).
|
||
func (s *TranscodeSession) GenerateFullManifest(segPrefix, rawQuery string) []byte {
|
||
opts := s.Opts()
|
||
totalDur := opts.TotalDuration
|
||
segDur := opts.SegmentDuration
|
||
if segDur <= 0 {
|
||
segDur = defaultSegmentDuration
|
||
}
|
||
if totalDur <= 0 {
|
||
totalDur = float64(segDur) // fallback: single segment
|
||
}
|
||
|
||
segCount := int(math.Ceil(totalDur / float64(segDur)))
|
||
if segCount < 1 {
|
||
segCount = 1
|
||
}
|
||
|
||
var suffix string
|
||
if rawQuery != "" {
|
||
suffix = "?" + rawQuery
|
||
}
|
||
|
||
segExt := hlsSegmentExtension(opts)
|
||
hlsVersion := 3
|
||
if segExt == ".m4s" {
|
||
hlsVersion = 7
|
||
}
|
||
|
||
var buf bytes.Buffer
|
||
buf.WriteString("#EXTM3U\n")
|
||
buf.WriteString(fmt.Sprintf("#EXT-X-VERSION:%d\n", hlsVersion))
|
||
buf.WriteString(fmt.Sprintf("#EXT-X-TARGETDURATION:%d\n", segDur))
|
||
buf.WriteString("#EXT-X-MEDIA-SEQUENCE:0\n")
|
||
buf.WriteString("#EXT-X-PLAYLIST-TYPE:VOD\n")
|
||
|
||
if segExt == ".m4s" {
|
||
buf.WriteString(fmt.Sprintf("#EXT-X-MAP:URI=\"%sinit.mp4%s\"\n", segPrefix, suffix))
|
||
}
|
||
|
||
for i := range segCount {
|
||
dur := float64(segDur)
|
||
if i == segCount-1 {
|
||
// Last segment covers the remainder.
|
||
dur = totalDur - float64(i)*float64(segDur)
|
||
if dur <= 0 {
|
||
dur = float64(segDur)
|
||
}
|
||
}
|
||
buf.WriteString(fmt.Sprintf("#EXTINF:%.6f,\n", dur))
|
||
buf.WriteString(fmt.Sprintf("%sseg_%05d%s%s\n", segPrefix, i, segExt, suffix))
|
||
}
|
||
|
||
buf.WriteString("#EXT-X-ENDLIST\n")
|
||
return buf.Bytes()
|
||
}
|
||
|
||
// GetSegment returns the file path of a named segment if it exists.
|
||
func (s *TranscodeSession) GetSegment(name string) (string, error) {
|
||
// Sanitize the name to prevent directory traversal.
|
||
clean := filepath.Base(name)
|
||
segPath := filepath.Join(s.outputDir, clean)
|
||
|
||
info, err := os.Stat(segPath)
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
return "", ErrSegmentNotFound
|
||
}
|
||
return "", fmt.Errorf("stat segment: %w", err)
|
||
}
|
||
if info.Size() <= 0 {
|
||
// ffmpeg can create init.mp4 before it has written any bytes. Treat
|
||
// zero-byte files as not-ready so callers fall back to WaitForSegment.
|
||
return "", ErrSegmentNotFound
|
||
}
|
||
return segPath, nil
|
||
}
|
||
|
||
// Close terminates the ffmpeg process and removes the temporary output directory.
|
||
func (s *TranscodeSession) Close() error {
|
||
return s.shutdown(true)
|
||
}
|
||
|
||
// CloseProcess terminates the ffmpeg process but leaves the output directory in
|
||
// place. It is used when another session owns the same output directory (e.g. a
|
||
// concurrent reconstruct race loser): tearing down this duplicate must not wipe
|
||
// the segments and init.mp4 the winning session is actively serving.
|
||
func (s *TranscodeSession) CloseProcess() error {
|
||
return s.shutdown(false)
|
||
}
|
||
|
||
// shutdown kills the ffmpeg process and, when removeOutput is true, removes the
|
||
// temporary output directory.
|
||
func (s *TranscodeSession) shutdown(removeOutput bool) error {
|
||
s.StopThrottler()
|
||
// Cancel the context to kill the process (no mutex needed for cancel).
|
||
if s.cancel != nil {
|
||
s.cancel()
|
||
}
|
||
|
||
// Wait for the monitor goroutine to finish reaping the process.
|
||
// This avoids a deadlock: the goroutine needs s.mu to mark running=false,
|
||
// so we must not hold s.mu while waiting.
|
||
// done is nil when no process was started (e.g. test-only sessions).
|
||
if s.done != nil {
|
||
<-s.done
|
||
}
|
||
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
s.running = false
|
||
|
||
// Clean up temporary directory.
|
||
if removeOutput && s.outputDir != "" {
|
||
if err := os.RemoveAll(s.outputDir); err != nil {
|
||
return fmt.Errorf("remove output dir: %w", err)
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// Done returns a channel that closes when the current ffmpeg process exits.
|
||
func (s *TranscodeSession) Done() <-chan struct{} {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.done
|
||
}
|
||
|
||
// IsRunning reports whether the ffmpeg process is still running.
|
||
func (s *TranscodeSession) IsRunning() bool {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.running
|
||
}
|
||
|
||
// WaitError returns the error from the last ffmpeg process exit, or nil if
|
||
// the process exited cleanly. A nil return means all output was written
|
||
// successfully and segments should remain servable.
|
||
func (s *TranscodeSession) WaitError() error {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.waitErr
|
||
}
|
||
|
||
// Opts returns the TranscodeOpts used to create this session (for testing).
|
||
func (s *TranscodeSession) Opts() TranscodeOpts {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.opts
|
||
}
|
||
|
||
// SetAudioTrackIndex updates the audio track index in the session's opts.
|
||
// Must be called before Restart() to take effect on the new ffmpeg process.
|
||
func (s *TranscodeSession) SetAudioTrackIndex(index int) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
s.opts.AudioTrackIndex = index
|
||
}
|
||
|
||
// cleanStaleSegments removes segment files at or after startSegment and the
|
||
// old manifest so a restarted copy-mode FFmpeg process writes clean output.
|
||
// The init.mp4 is preserved — its codec configuration is derived from the
|
||
// source file and is identical across restarts.
|
||
func (s *TranscodeSession) cleanStaleSegments(startSegment int) {
|
||
entries, err := os.ReadDir(s.outputDir)
|
||
if err != nil {
|
||
return
|
||
}
|
||
for _, entry := range entries {
|
||
name := entry.Name()
|
||
if name == "stream.m3u8" {
|
||
os.Remove(filepath.Join(s.outputDir, name))
|
||
continue
|
||
}
|
||
if name == "init.mp4" {
|
||
continue
|
||
}
|
||
segNum, parseErr := ParseSegmentNumber(name)
|
||
if parseErr != nil {
|
||
continue
|
||
}
|
||
if segNum >= startSegment {
|
||
os.Remove(filepath.Join(s.outputDir, name))
|
||
}
|
||
}
|
||
}
|
||
|
||
// Restart kills the current ffmpeg process and starts a new one seeking to
|
||
// the given position. startSegment sets -hls_segment_start_number so that
|
||
// output filenames align with the expected segment numbering. Existing
|
||
// segment files are preserved so backward seeks can reuse them; for
|
||
// copy-mode sessions, stale segments at or after the restart point are
|
||
// cleaned to prevent serving data from the wrong timeline position.
|
||
func (s *TranscodeSession) Restart(ctx context.Context, seekSeconds float64, startSegment int) error {
|
||
return s.restart(ctx, seekSeconds, startSegment, 0, false)
|
||
}
|
||
|
||
// RestartWithCopySeekAnchor restarts a copy-video stream with the keyframe
|
||
// origin resolved for this specific seek. Keeping this metadata explicit
|
||
// prevents a prior seek's origin from being reused after an audio switch.
|
||
func (s *TranscodeSession) RestartWithCopySeekAnchor(
|
||
ctx context.Context,
|
||
seekSeconds float64,
|
||
startSegment int,
|
||
streamOriginSeconds float64,
|
||
) error {
|
||
return s.restart(ctx, seekSeconds, startSegment, streamOriginSeconds, true)
|
||
}
|
||
|
||
func (s *TranscodeSession) restart(
|
||
ctx context.Context,
|
||
seekSeconds float64,
|
||
startSegment int,
|
||
streamOriginSeconds float64,
|
||
copySeekAnchorResolved bool,
|
||
) error {
|
||
s.mu.Lock()
|
||
// Single-flight: a second caller arriving while a restart is in
|
||
// progress must not kill the process the first restart just started.
|
||
// It returns immediately and the caller falls through to
|
||
// WaitForSegment, which polls through the in-flight restart.
|
||
if s.restarting {
|
||
s.mu.Unlock()
|
||
return nil
|
||
}
|
||
s.restarting = true
|
||
cancelCurrent := s.cancel
|
||
done := s.done
|
||
s.mu.Unlock()
|
||
s.StopThrottler()
|
||
|
||
// Kill current process without removing output directory.
|
||
if cancelCurrent != nil {
|
||
cancelCurrent()
|
||
}
|
||
if done != nil {
|
||
<-done
|
||
}
|
||
|
||
s.mu.Lock()
|
||
s.running = false
|
||
s.waitErr = nil
|
||
if s.stderr != nil {
|
||
s.stderr.Reset()
|
||
}
|
||
s.restartCount++
|
||
opts := s.opts
|
||
reserveHWDevice := s.reserveHWDeviceOnRestart
|
||
s.mu.Unlock()
|
||
|
||
// Copy-mode restarts must clean stale segments so ffmpeg writes fresh
|
||
// output. Encoded transcodes keep old segments for backward seek reuse.
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
|
||
s.cleanStaleSegments(startSegment)
|
||
}
|
||
|
||
opts.SeekSeconds = seekSeconds
|
||
opts.StartSegmentNumber = startSegment
|
||
if strings.EqualFold(opts.TargetCodecVideo, "copy") {
|
||
// A seek restart describes a new emitted timeline. Never retain the
|
||
// previous copy seek's keyframe origin when the caller has not resolved
|
||
// a replacement; falling back to SeekSeconds is conservative and matches
|
||
// the behavior of recipes created before copy anchors were introduced.
|
||
opts.StreamOriginSeconds = streamOriginSeconds
|
||
opts.CopySeekAnchorResolved = copySeekAnchorResolved
|
||
}
|
||
opts.FastStart = false // seek-restarts use veryfast for better quality
|
||
|
||
args := buildFFmpegArgs(opts)
|
||
bin := opts.FFmpegPath
|
||
if bin == "" {
|
||
bin = ffmpegBinary()
|
||
}
|
||
|
||
log.Printf("playback: ffmpeg restart cmd: %s %s", bin, strings.Join(args, " "))
|
||
s.logFFmpegEvent(ctx, "ffmpeg process restart", "")
|
||
|
||
ctx, cancel := context.WithCancel(ctx)
|
||
cmd := exec.CommandContext(ctx, bin, args...)
|
||
stdinPipe, err := cmd.StdinPipe()
|
||
if err != nil {
|
||
cancel()
|
||
s.mu.Lock()
|
||
s.restarting = false
|
||
s.waitErr = err
|
||
s.mu.Unlock()
|
||
return fmt.Errorf("create stdin pipe: %w", err)
|
||
}
|
||
cmd.Dir = opts.OutputDir
|
||
cmd.Stderr = s.newStderrWriter(ctx)
|
||
cmd.WaitDelay = 3 * time.Second
|
||
|
||
// The previous process released its reservation before closing done. Keep
|
||
// this session on the same concrete GPU while accounting for the replacement
|
||
// process as a new active workload.
|
||
releaseHWDevice := func() {}
|
||
if reserveHWDevice {
|
||
releaseHWDevice = reserveConcreteHWDevice(opts.HWDevice)
|
||
}
|
||
|
||
if err := cmd.Start(); err != nil {
|
||
cancel()
|
||
releaseHWDevice()
|
||
s.mu.Lock()
|
||
s.restarting = false
|
||
s.waitErr = err
|
||
s.mu.Unlock()
|
||
s.logFFmpegEvent(ctx, "ffmpeg process exit error", err.Error())
|
||
return fmt.Errorf("restart ffmpeg: %w", err)
|
||
}
|
||
|
||
s.mu.Lock()
|
||
if s.stdinPipe != nil {
|
||
s.stdinPipe.Close()
|
||
}
|
||
s.cmd = cmd
|
||
s.cancel = cancel
|
||
s.opts = opts
|
||
s.running = true
|
||
s.restarting = false
|
||
s.stdinPipe = stdinPipe
|
||
s.lastRequestedSegment = startSegment
|
||
s.done = make(chan struct{})
|
||
hook := s.restartHook
|
||
s.mu.Unlock()
|
||
|
||
go s.monitorFFmpeg(ctx, cmd, s.done, releaseHWDevice)
|
||
|
||
if hook != nil {
|
||
hook(ctx)
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// WaitForSegment polls until the named segment file exists on disk or the
|
||
// timeout expires. Returns the segment file path on success.
|
||
//
|
||
// Segments are served as soon as they appear on disk. The -hls_flags temp_file
|
||
// flag ensures ffmpeg writes to a .tmp file and atomically renames on completion,
|
||
// so a successful stat means the segment is fully written.
|
||
func (s *TranscodeSession) WaitForSegment(name string, timeout time.Duration) (string, error) {
|
||
clean := filepath.Base(name)
|
||
segPath := filepath.Join(s.outputDir, clean)
|
||
|
||
deadline := time.After(timeout)
|
||
for {
|
||
info, statErr := os.Stat(segPath)
|
||
segReady := statErr == nil && info.Size() > 0
|
||
if segReady {
|
||
return segPath, nil
|
||
}
|
||
|
||
s.mu.Lock()
|
||
running := s.running
|
||
restarting := s.restarting
|
||
waitErr := s.waitErr
|
||
s.mu.Unlock()
|
||
|
||
if restarting {
|
||
select {
|
||
case <-deadline:
|
||
return "", ErrSegmentNotFound
|
||
case <-time.After(100 * time.Millisecond):
|
||
continue
|
||
}
|
||
}
|
||
|
||
if !running && waitErr != nil {
|
||
return "", fmt.Errorf("%w: %v", ErrTranscodeFailed, waitErr)
|
||
}
|
||
// If ffmpeg finished cleanly but the segment doesn't exist,
|
||
// it won't appear later — fail fast.
|
||
if !running {
|
||
return "", ErrSegmentNotFound
|
||
}
|
||
|
||
select {
|
||
case <-deadline:
|
||
return "", ErrSegmentNotFound
|
||
case <-time.After(100 * time.Millisecond):
|
||
}
|
||
}
|
||
}
|
||
|
||
// RewriteManifestPaths prefixes relative segment references in an HLS manifest
|
||
// with segPrefix (e.g. "segment/") and optionally appends rawQuery as a query
|
||
// string. This ensures the HLS player's segment requests match server routes
|
||
// and preserve any auth or cache-busting parameters from the manifest URL.
|
||
func RewriteManifestPaths(manifest []byte, segPrefix, rawQuery string) ([]byte, error) {
|
||
if err := validateManifestHeader(manifest); err != nil {
|
||
return nil, fmt.Errorf("invalid manifest: %w", err)
|
||
}
|
||
|
||
var suffix string
|
||
if rawQuery != "" {
|
||
suffix = "?" + rawQuery
|
||
}
|
||
|
||
lines := bytes.Split(manifest, []byte("\n"))
|
||
for i, line := range lines {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if len(trimmed) == 0 {
|
||
continue
|
||
}
|
||
|
||
// Rewrite #EXT-X-MAP:URI="filename"
|
||
if bytes.HasPrefix(trimmed, []byte("#EXT-X-MAP:")) {
|
||
rewritten, err := rewriteMapURI(trimmed, segPrefix, suffix)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
lines[i] = rewritten
|
||
continue
|
||
}
|
||
|
||
// Skip other tags/comments.
|
||
if trimmed[0] == '#' {
|
||
continue
|
||
}
|
||
|
||
// Segment filename line.
|
||
lines[i] = []byte(segPrefix + string(trimmed) + suffix)
|
||
}
|
||
return bytes.Join(lines, []byte("\n")), nil
|
||
}
|
||
|
||
// rewriteMapURI rewrites the URI value inside an #EXT-X-MAP tag.
|
||
func rewriteMapURI(line []byte, segPrefix, suffix string) ([]byte, error) {
|
||
uriStart := bytes.Index(line, []byte(`URI="`))
|
||
if uriStart < 0 {
|
||
return nil, fmt.Errorf("invalid #EXT-X-MAP line: missing URI attribute")
|
||
}
|
||
uriStart += 5 // skip past URI="
|
||
uriEnd := bytes.IndexByte(line[uriStart:], '"')
|
||
if uriEnd < 0 {
|
||
return nil, fmt.Errorf("invalid #EXT-X-MAP line: unterminated URI attribute")
|
||
}
|
||
uriEnd += uriStart
|
||
|
||
oldURI := string(line[uriStart:uriEnd])
|
||
newURI := segPrefix + oldURI + suffix
|
||
|
||
result := make([]byte, 0, len(line)+len(newURI)-len(oldURI))
|
||
result = append(result, line[:uriStart]...)
|
||
result = append(result, []byte(newURI)...)
|
||
result = append(result, line[uriEnd:]...)
|
||
return result, nil
|
||
}
|
||
|
||
// AppendManifestQueryParam appends a single "key=value" query parameter to every
|
||
// segment and #EXT-X-MAP init URI in an HLS media playlist, preserving any query
|
||
// the URI already carries (using "?" or "&" as appropriate). The value is
|
||
// appended verbatim — callers must supply a URL-safe value (a signed stream token
|
||
// is base64url and safe). A manifest without a valid #EXTM3U header is returned
|
||
// unchanged so a non-manifest body is never corrupted.
|
||
//
|
||
// It exists for the API proxy boundary: a transcode node builds its manifest from
|
||
// the forwarded request query, which deliberately omits the signed stream token
|
||
// ("st") so the token never reaches the node URL or its logs. That leaves the
|
||
// node-built segment URIs token-less, so a later segment fetch after a node/API
|
||
// restart cannot reconstruct the session. Re-injecting the client-facing token
|
||
// here keeps reconstruction working without ever exposing it to the node.
|
||
func AppendManifestQueryParam(manifest []byte, key, value string) []byte {
|
||
if key == "" || validateManifestHeader(manifest) != nil {
|
||
return manifest
|
||
}
|
||
param := key + "=" + value
|
||
lines := bytes.Split(manifest, []byte("\n"))
|
||
for i, line := range lines {
|
||
trimmed := bytes.TrimSpace(line)
|
||
if len(trimmed) == 0 {
|
||
continue
|
||
}
|
||
if bytes.HasPrefix(trimmed, []byte("#EXT-X-MAP:")) {
|
||
lines[i] = appendMapURIQueryParam(trimmed, param)
|
||
continue
|
||
}
|
||
if trimmed[0] == '#' {
|
||
continue
|
||
}
|
||
lines[i] = appendURIQueryParam(trimmed, param)
|
||
}
|
||
return bytes.Join(lines, []byte("\n"))
|
||
}
|
||
|
||
// appendURIQueryParam appends a "key=value" param to a bare URI line.
|
||
func appendURIQueryParam(uri []byte, param string) []byte {
|
||
sep := []byte("?")
|
||
if bytes.IndexByte(uri, '?') >= 0 {
|
||
sep = []byte("&")
|
||
}
|
||
out := make([]byte, 0, len(uri)+len(sep)+len(param))
|
||
out = append(out, uri...)
|
||
out = append(out, sep...)
|
||
out = append(out, param...)
|
||
return out
|
||
}
|
||
|
||
// appendMapURIQueryParam appends a "key=value" param to the URI inside an
|
||
// #EXT-X-MAP tag. A line without a well-formed URI attribute is returned
|
||
// unchanged.
|
||
func appendMapURIQueryParam(line []byte, param string) []byte {
|
||
uriStart := bytes.Index(line, []byte(`URI="`))
|
||
if uriStart < 0 {
|
||
return line
|
||
}
|
||
uriStart += 5 // skip past URI="
|
||
uriEnd := bytes.IndexByte(line[uriStart:], '"')
|
||
if uriEnd < 0 {
|
||
return line
|
||
}
|
||
uriEnd += uriStart
|
||
|
||
sep := []byte("?")
|
||
if bytes.IndexByte(line[uriStart:uriEnd], '?') >= 0 {
|
||
sep = []byte("&")
|
||
}
|
||
result := make([]byte, 0, len(line)+len(sep)+len(param))
|
||
result = append(result, line[:uriEnd]...)
|
||
result = append(result, sep...)
|
||
result = append(result, param...)
|
||
result = append(result, line[uriEnd:]...)
|
||
return result
|
||
}
|
||
|
||
func (s *TranscodeSession) manifestTimeoutError(timeout time.Duration) error {
|
||
s.mu.Lock()
|
||
running := s.running
|
||
waitErr := s.waitErr
|
||
stderr := ""
|
||
if s.stderr != nil {
|
||
stderr = truncateStderr(s.stderr.String())
|
||
}
|
||
s.mu.Unlock()
|
||
|
||
switch {
|
||
case waitErr != nil && stderr != "":
|
||
return fmt.Errorf("%w after %s: ffmpeg exited: %v (stderr: %s)", ErrManifestNotReady, timeout, waitErr, stderr)
|
||
case waitErr != nil:
|
||
return fmt.Errorf("%w after %s: ffmpeg exited: %v", ErrManifestNotReady, timeout, waitErr)
|
||
case running:
|
||
return fmt.Errorf("%w after %s: ffmpeg still running", ErrManifestNotReady, timeout)
|
||
default:
|
||
return fmt.Errorf("%w after %s: ffmpeg is no longer running", ErrManifestNotReady, timeout)
|
||
}
|
||
}
|
||
|
||
// IsCopyVideo reports whether this session is repackaging video without
|
||
// re-encoding. Copy-mode manifests must reflect FFmpeg's real fragment timing.
|
||
func (s *TranscodeSession) IsCopyVideo() bool {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return strings.EqualFold(s.opts.TargetCodecVideo, "copy")
|
||
}
|
||
|
||
// SegmentStartTime reports the source-timeline start time of the requested
|
||
// segment using the current on-disk manifest. The bool return is false when
|
||
// the segment is not present in the manifest yet.
|
||
func (s *TranscodeSession) SegmentStartTime(segNum int) (float64, bool, error) {
|
||
s.mu.Lock()
|
||
manifestPath := filepath.Join(s.outputDir, "stream.m3u8")
|
||
baseSeekSeconds := s.opts.SeekSeconds
|
||
if strings.EqualFold(s.opts.TargetCodecVideo, "copy") && s.opts.CopySeekAnchorResolved {
|
||
baseSeekSeconds = s.opts.StreamOriginSeconds
|
||
}
|
||
s.mu.Unlock()
|
||
|
||
manifest, err := os.ReadFile(manifestPath)
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
return 0, false, ErrManifestNotReady
|
||
}
|
||
return 0, false, fmt.Errorf("read manifest: %w", err)
|
||
}
|
||
|
||
timeline, err := parseManifestTimeline(manifest)
|
||
if err != nil {
|
||
return 0, false, fmt.Errorf("parse manifest timeline: %w", err)
|
||
}
|
||
|
||
if len(timeline.entries) == 0 {
|
||
return 0, false, ErrManifestNotReady
|
||
}
|
||
|
||
currentTime := baseSeekSeconds
|
||
for _, entry := range timeline.entries {
|
||
if entry.number == segNum {
|
||
return currentTime, true, nil
|
||
}
|
||
currentTime += entry.duration
|
||
}
|
||
|
||
return 0, false, nil
|
||
}
|
||
|
||
// RestartSeekTarget resolves the source-timeline time to restart FFmpeg for
|
||
// the requested segment. Copy-mode sessions prefer the current manifest's real
|
||
// timing when available; encoded sessions use fixed-duration seek math
|
||
// matching the synthetic VOD manifest.
|
||
func (s *TranscodeSession) RestartSeekTarget(segNum int) (float64, bool, error) {
|
||
if strings.EqualFold(s.Opts().TargetCodecVideo, "copy") {
|
||
seekSeconds, ok, err := s.SegmentStartTime(segNum)
|
||
switch {
|
||
case err == nil && ok:
|
||
return seekSeconds, true, nil
|
||
case err != nil && !errors.Is(err, ErrManifestNotReady):
|
||
return 0, false, err
|
||
}
|
||
// Copy-mode fragments have variable durations, so the encoded
|
||
// `seg×dur` math would seek FFmpeg to the wrong source time and
|
||
// desync A/V after restart. When the manifest can't resolve this
|
||
// segment yet (ok=false with no error, including ErrManifestNotReady
|
||
// in a freshly reconstructed window), report the seek target as
|
||
// unresolved (0, false, nil) rather than guessing. The caller treats
|
||
// this as a retryable miss so the session keeps producing manifest
|
||
// until real timing is available.
|
||
return 0, false, nil
|
||
}
|
||
|
||
segDuration := defaultSegmentDuration
|
||
if opts := s.Opts(); opts.SegmentDuration > 0 {
|
||
segDuration = opts.SegmentDuration
|
||
}
|
||
return float64(segNum * segDuration), true, nil
|
||
}
|
||
|
||
// ReportSegmentDownloaded records that the client has downloaded the given
|
||
// segment number. Only updates if segNum exceeds the current high-water mark.
|
||
func (s *TranscodeSession) ReportSegmentDownloaded(segNum int) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if segNum > s.lastRequestedSegment {
|
||
s.lastRequestedSegment = segNum
|
||
}
|
||
}
|
||
|
||
// LastRequestedSegment returns the highest segment number downloaded by the client.
|
||
func (s *TranscodeSession) LastRequestedSegment() int {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.lastRequestedSegment
|
||
}
|
||
|
||
// StartThrottler creates and starts a throttler for this session.
|
||
// No-op if thresholdSeconds <= 0 or stdinPipe is nil.
|
||
func (s *TranscodeSession) StartThrottler(thresholdSeconds int) {
|
||
s.mu.Lock()
|
||
if s.stdinPipe == nil || thresholdSeconds <= 0 {
|
||
s.mu.Unlock()
|
||
return
|
||
}
|
||
t := NewTranscodeThrottler(s, s.stdinPipe, thresholdSeconds, s.opts.SegmentDuration)
|
||
s.throttler = t
|
||
s.mu.Unlock()
|
||
t.Start()
|
||
}
|
||
|
||
// StopThrottler stops the throttler if one is running.
|
||
func (s *TranscodeSession) StopThrottler() {
|
||
s.mu.Lock()
|
||
t := s.throttler
|
||
s.throttler = nil
|
||
s.mu.Unlock()
|
||
if t != nil {
|
||
t.Stop()
|
||
}
|
||
}
|
||
|
||
type ffmpegStderrWriter struct {
|
||
session *TranscodeSession
|
||
ctx context.Context
|
||
partial []byte
|
||
}
|
||
|
||
func (w *ffmpegStderrWriter) Write(p []byte) (int, error) {
|
||
if len(p) == 0 {
|
||
return 0, nil
|
||
}
|
||
w.partial = append(w.partial, p...)
|
||
for {
|
||
idx := bytes.IndexByte(w.partial, '\n')
|
||
if idx < 0 {
|
||
break
|
||
}
|
||
line := strings.TrimRight(string(w.partial[:idx]), "\r")
|
||
w.session.logFFmpegLine(w.ctx, line)
|
||
w.partial = append([]byte(nil), w.partial[idx+1:]...)
|
||
}
|
||
return len(p), nil
|
||
}
|
||
|
||
func (w *ffmpegStderrWriter) Flush() {
|
||
if len(w.partial) == 0 {
|
||
return
|
||
}
|
||
w.session.logFFmpegLine(w.ctx, strings.TrimRight(string(w.partial), "\r"))
|
||
w.partial = nil
|
||
}
|
||
|
||
func (s *TranscodeSession) newStderrWriter(ctx context.Context) io.Writer {
|
||
lineWriter := &ffmpegStderrWriter{session: s, ctx: ctx}
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if s.stderr == nil {
|
||
s.stderr = newBoundedTailBuffer(stderrTailMaxBytes)
|
||
}
|
||
s.stderrWriter = lineWriter
|
||
return io.MultiWriter(s.stderr, lineWriter)
|
||
}
|
||
|
||
func (s *TranscodeSession) flushStderr(ctx context.Context) {
|
||
s.mu.Lock()
|
||
writer := s.stderrWriter
|
||
s.stderrWriter = nil
|
||
s.mu.Unlock()
|
||
if writer != nil {
|
||
writer.Flush()
|
||
}
|
||
}
|
||
|
||
func (s *TranscodeSession) logFFmpegLine(ctx context.Context, line string) {
|
||
if s == nil || s.opts.FFmpegLogSink == nil {
|
||
return
|
||
}
|
||
|
||
line = strings.ToValidUTF8(line, "\uFFFD")
|
||
if strings.TrimSpace(line) == "" {
|
||
return
|
||
}
|
||
trimmed, truncated := truncateUTF8String(line, maxPersistedFFmpegChars)
|
||
if truncated {
|
||
trimmed += "...[truncated]"
|
||
}
|
||
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
|
||
if s.stderrLinesLogged >= maxPersistedFFmpegLines || s.stderrBytesLogged+len(trimmed) > maxPersistedFFmpegBytes {
|
||
s.stderrDroppedLines++
|
||
if !s.stderrCapLogged {
|
||
s.stderrCapLogged = true
|
||
attrs := s.ffmpegAttrsLocked()
|
||
attrs.DroppedLines = s.stderrDroppedLines
|
||
s.opts.FFmpegLogSink.WriteEvent(ctx, s.opts.SessionID, attrs, "ffmpeg stderr logging capped")
|
||
}
|
||
return
|
||
}
|
||
|
||
s.stderrLinesLogged++
|
||
s.stderrBytesLogged += len(trimmed)
|
||
s.stderrLineIndex++
|
||
attrs := s.ffmpegAttrsLocked()
|
||
attrs.LineIndex = s.stderrLineIndex
|
||
s.opts.FFmpegLogSink.WriteLine(ctx, s.opts.SessionID, attrs, trimmed)
|
||
}
|
||
|
||
func (s *TranscodeSession) logFFmpegEvent(ctx context.Context, message, exitError string) {
|
||
if s == nil || s.opts.FFmpegLogSink == nil {
|
||
return
|
||
}
|
||
s.mu.Lock()
|
||
attrs := s.ffmpegAttrsLocked()
|
||
attrs.ExitError = exitError
|
||
s.mu.Unlock()
|
||
s.opts.FFmpegLogSink.WriteEvent(ctx, s.opts.SessionID, attrs, message)
|
||
}
|
||
|
||
func (s *TranscodeSession) logWaitResult(ctx context.Context, waitErr error) {
|
||
if waitErr == nil {
|
||
s.logFFmpegEvent(ctx, "ffmpeg process exited", "")
|
||
return
|
||
}
|
||
s.logFFmpegEvent(ctx, "ffmpeg process exit error", formatWaitError(waitErr))
|
||
}
|
||
|
||
func (s *TranscodeSession) ffmpegAttrsLocked() FFmpegLogAttrs {
|
||
return FFmpegLogAttrs{
|
||
NodeType: s.opts.NodeType,
|
||
ExecutionMode: s.opts.ExecutionMode,
|
||
InputPath: s.opts.InputPath,
|
||
OutputDir: s.opts.OutputDir,
|
||
TargetResolution: s.opts.TargetResolution,
|
||
TargetVideoCodec: s.opts.TargetCodecVideo,
|
||
TargetAudioCodec: s.opts.TargetCodecAudio,
|
||
HWAccel: s.opts.HWAccel,
|
||
SeekSeconds: s.opts.SeekSeconds,
|
||
StartSegmentNumber: s.opts.StartSegmentNumber,
|
||
RestartCount: s.restartCount,
|
||
DroppedLines: s.stderrDroppedLines,
|
||
}
|
||
}
|
||
|
||
func formatWaitError(err error) string {
|
||
if err == nil {
|
||
return ""
|
||
}
|
||
if exitErr, ok := err.(*exec.ExitError); ok {
|
||
if status, ok := exitErr.Sys().(syscall.WaitStatus); ok {
|
||
return fmt.Sprintf("exit_code=%d: %v", status.ExitStatus(), err)
|
||
}
|
||
}
|
||
return err.Error()
|
||
}
|