2781 lines
93 KiB
Go
2781 lines
93 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"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
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
|
||
SegmentRetentionSeconds int // downloaded media retained behind the client; 0 disables pruning
|
||
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
|
||
lastCompletedSegment int
|
||
lastPruneFloor int
|
||
lastPruneHighWater int
|
||
segmentPruneRunning bool
|
||
pruneBeforeStart bool
|
||
segmentGeneration uint64
|
||
segmentIncarnation string
|
||
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,
|
||
lastCompletedSegment: opts.StartSegmentNumber - 1,
|
||
lastPruneFloor: opts.StartSegmentNumber - 1,
|
||
lastPruneHighWater: opts.StartSegmentNumber - 1,
|
||
segmentIncarnation: uuid.NewString(),
|
||
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
|
||
}
|
||
|
||
// SegmentLease binds an opened segment descriptor to the FFmpeg generation it
|
||
// came from. Keeping the descriptor open makes it a filesystem-backed lease: a
|
||
// concurrent prune may unlink the directory entry, but the response can still
|
||
// read the complete file on POSIX filesystems.
|
||
type SegmentLease struct {
|
||
File *os.File
|
||
Info os.FileInfo
|
||
Generation uint64
|
||
GenerationToken string
|
||
}
|
||
|
||
// Close releases the opened segment descriptor.
|
||
func (l *SegmentLease) Close() error {
|
||
return l.File.Close()
|
||
}
|
||
|
||
// OpenSegment opens a completed segment for serving and captures its FFmpeg
|
||
// generation atomically with the open.
|
||
func (s *TranscodeSession) OpenSegment(name string) (*SegmentLease, error) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if s.restarting {
|
||
return nil, ErrSegmentNotFound
|
||
}
|
||
|
||
clean := filepath.Base(name)
|
||
segment, err := os.Open(filepath.Join(s.outputDir, clean))
|
||
if err != nil {
|
||
if os.IsNotExist(err) {
|
||
return nil, ErrSegmentNotFound
|
||
}
|
||
return nil, fmt.Errorf("open segment: %w", err)
|
||
}
|
||
info, err := segment.Stat()
|
||
if err != nil {
|
||
_ = segment.Close()
|
||
return nil, fmt.Errorf("stat segment: %w", err)
|
||
}
|
||
if info.Size() <= 0 {
|
||
_ = segment.Close()
|
||
return nil, ErrSegmentNotFound
|
||
}
|
||
if s.segmentIncarnation == "" {
|
||
s.segmentIncarnation = uuid.NewString()
|
||
}
|
||
return &SegmentLease{
|
||
File: segment,
|
||
Info: info,
|
||
Generation: s.segmentGeneration,
|
||
GenerationToken: s.segmentGenerationTokenLocked(),
|
||
}, 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
|
||
s.segmentGeneration++
|
||
s.segmentPruneRunning = 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
|
||
s.segmentGeneration++
|
||
s.segmentPruneRunning = false
|
||
previousRequestedSegment := s.lastRequestedSegment
|
||
forwardRestart := startSegment > previousRequestedSegment
|
||
s.lastRequestedSegment = startSegment
|
||
s.lastCompletedSegment = startSegment - 1
|
||
if forwardRestart && s.opts.SegmentRetentionSeconds > 0 {
|
||
// Keep the old prune cursor reachable until the replacement process has
|
||
// produced a complete back buffer. Otherwise every forward restart leaves
|
||
// one retained window permanently behind startSegment.
|
||
s.pruneBeforeStart = true
|
||
} else {
|
||
s.lastPruneFloor = startSegment - 1
|
||
s.pruneBeforeStart = false
|
||
}
|
||
s.lastPruneHighWater = startSegment - 1
|
||
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.lastCompletedSegment = startSegment - 1
|
||
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):
|
||
}
|
||
}
|
||
}
|
||
|
||
// WaitForOpenSegment is the opened-file counterpart to WaitForSegment. It
|
||
// closes the stat-to-open race for HTTP serving while preserving the same
|
||
// restart and ffmpeg-exit error contract.
|
||
func (s *TranscodeSession) WaitForOpenSegment(name string, timeout time.Duration) (*SegmentLease, error) {
|
||
deadline := time.After(timeout)
|
||
for {
|
||
segment, err := s.OpenSegment(name)
|
||
if err == nil {
|
||
return segment, nil
|
||
}
|
||
if !errors.Is(err, ErrSegmentNotFound) {
|
||
return nil, err
|
||
}
|
||
|
||
s.mu.Lock()
|
||
running := s.running
|
||
restarting := s.restarting
|
||
waitErr := s.waitErr
|
||
s.mu.Unlock()
|
||
|
||
if restarting {
|
||
select {
|
||
case <-deadline:
|
||
return nil, ErrSegmentNotFound
|
||
case <-time.After(100 * time.Millisecond):
|
||
continue
|
||
}
|
||
}
|
||
if !running && waitErr != nil {
|
||
return nil, fmt.Errorf("%w: %w", ErrTranscodeFailed, waitErr)
|
||
}
|
||
if !running {
|
||
return nil, ErrSegmentNotFound
|
||
}
|
||
|
||
select {
|
||
case <-deadline:
|
||
return nil, 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()
|
||
s.reportSegmentDownloadedLocked(segNum)
|
||
}
|
||
|
||
// SegmentGeneration identifies the current ffmpeg timeline. Callers that hold
|
||
// an open segment across a restart use it to prevent the old response from
|
||
// advancing the replacement process's download position.
|
||
func (s *TranscodeSession) SegmentGeneration() uint64 {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
return s.segmentGeneration
|
||
}
|
||
|
||
// ReportSegmentDownloadedForGeneration records completion only if the served
|
||
// file belongs to the current ffmpeg timeline.
|
||
func (s *TranscodeSession) ReportSegmentDownloadedForGeneration(segNum int, generation uint64) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if generation != s.segmentGeneration {
|
||
return
|
||
}
|
||
s.reportSegmentDownloadedLocked(segNum)
|
||
}
|
||
|
||
// ReportSegmentDownloadedForGenerationToken records a completion relayed over
|
||
// HTTP only when it belongs to this exact session object and FFmpeg timeline.
|
||
// The opaque incarnation prevents a delayed proxy acknowledgement from matching
|
||
// a reconstructed session whose numeric generation restarted at zero.
|
||
func (s *TranscodeSession) ReportSegmentDownloadedForGenerationToken(segNum int, generationToken string) {
|
||
s.mu.Lock()
|
||
defer s.mu.Unlock()
|
||
if generationToken == "" || generationToken != s.segmentGenerationTokenLocked() {
|
||
return
|
||
}
|
||
s.reportSegmentDownloadedLocked(segNum)
|
||
}
|
||
|
||
func (s *TranscodeSession) segmentGenerationTokenLocked() string {
|
||
return s.segmentIncarnation + ":" + strconv.FormatUint(s.segmentGeneration, 10)
|
||
}
|
||
|
||
func (s *TranscodeSession) reportSegmentDownloadedLocked(segNum int) {
|
||
if segNum > s.lastRequestedSegment {
|
||
s.lastRequestedSegment = segNum
|
||
}
|
||
if segNum > s.lastCompletedSegment {
|
||
s.lastCompletedSegment = segNum
|
||
}
|
||
s.scheduleSegmentPruneLocked()
|
||
}
|
||
|
||
// 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()
|
||
}
|