diff --git a/internal/api/handlers/playback.go b/internal/api/handlers/playback.go index 2d01acea..2d974791 100644 --- a/internal/api/handlers/playback.go +++ b/internal/api/handlers/playback.go @@ -2609,8 +2609,12 @@ func (h *PlaybackHandler) HandleChangeAudioTrack(w http.ResponseWriter, r *http. } // A v3 DV strip remux carries its bitstream filter in the // durable session route; dropping it here would hand the node - // a DV7 copy recipe that leaves dangling RPUs. - if updatedSession.RemuxDVMode == playback.RemuxDVStripToHDR10V3 && strings.EqualFold(nodeReq.TargetCodecVideo, "copy") { + // a DV7 copy recipe that leaves dangling RPUs. Sources that + // fail the RPU probe are the exception — for them the filter + // rejects every packet, so re-adding it on an audio switch + // would hang a session that was playing a moment ago. + if updatedSession.RemuxDVMode == playback.RemuxDVStripToHDR10V3 && strings.EqualFold(nodeReq.TargetCodecVideo, "copy") && + playback.DVRPUStrippable(r.Context(), h.playbackConfig().FFmpegPath, file.FilePath) { nodeReq.VideoBitstreamFilter = playback.DV7ToHDR10BitstreamFilter } @@ -3100,10 +3104,21 @@ func (h *PlaybackHandler) HandleStartTranscode(w http.ResponseWriter, r *http.Re // for "copy" has no way to know the source needs the strip. Derived after // the burn-in guard above so a copy request it rewrites to h264 never carries // a copy-only bitstream filter. + // + // Gated on the per-source probe for the same reason the planner is: a + // source whose RPU ffmpeg cannot parse turns the filter into a per-packet + // rejection that never produces a segment, and this endpoint would + // otherwise put it back on every quality change, seek and burn-in restart + // of a session the planner had already routed away from it. videoBitstreamFilter := "" if strings.EqualFold(req.TargetCodecVideo, "copy") && (session.RemuxDVMode == playback.RemuxDVStripToHDR10V3 || file.PrimaryDVProfile() == 7) { - videoBitstreamFilter = playback.DV7ToHDR10BitstreamFilter + if playback.DVRPUStrippable(r.Context(), h.playbackConfig().FFmpegPath, file.FilePath) { + videoBitstreamFilter = playback.DV7ToHDR10BitstreamFilter + } else { + slog.WarnContext(r.Context(), "restart dropped the dolby vision rpu strip: source cannot be stripped", + "component", "api", "playback_session_id", req.SessionID, "file_id", file.ID) + } } // The request-level permission check above intentionally runs before the diff --git a/internal/api/handlers/playback_v3.go b/internal/api/handlers/playback_v3.go index c1f026ee..1c8ca19c 100644 --- a/internal/api/handlers/playback_v3.go +++ b/internal/api/handlers/playback_v3.go @@ -425,7 +425,7 @@ func (h *PlaybackHandler) handleStartPlaybackV3(w http.ResponseWriter, r *http.R result := playback.PlanPlaybackV3(playback.PlannerInputV3{ Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, - Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), + Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile), }) if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && shouldTryAlternateFileV3(req.QualityPreference) { @@ -440,7 +440,7 @@ func (h *PlaybackHandler) handleStartPlaybackV3(w http.ResponseWriter, r *http.R writePlaybackFilePreflightError(w, err) return } - result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) + result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: req, RequestedFile: requestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: settings, Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) } } if result.Terminal != nil { @@ -1212,7 +1212,7 @@ func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.Atte attemptedKeys = append(attemptedKeys, currentKey) } } - result := playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) + result := playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) if result.Terminal != nil && result.Terminal.Reason == "no_alternate_version" && replanAllowsAlternateFileV3(operation, start.QualityPreference) { if alternate, alternateErr := h.findAlternateFile(r.Context(), requestedFile); alternateErr == nil && alternate != nil { alternate = h.ensurePlaybackProbe(r.Context(), alternate) @@ -1222,7 +1222,7 @@ func (h *PlaybackHandler) executeReplanV3(r *http.Request, record *playback.Atte if err := preflightPlaybackFile(r.Context(), alternate, h.MissingMarker, h.EventsHub); err == nil { effectiveFile = alternate audioIndex = remappedAudio - result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) + result = playback.PlanPlaybackV3(playback.PlannerInputV3{Request: start, RequestedFile: plannerRequestedFile, EffectiveFile: effectiveFile, AudioTrackIndex: audioIndex, Settings: h.plannerSettingsV3(r.Context()), Registry: h.transformationRegistryV3(r.Context()), HLSRegistry: h.lazyHLSPlanningRegistryV3(r.Context()), DVRPUStrippable: h.lazyDVRPUStrippableV3(r.Context(), effectiveFile), Now: time.Now(), AttemptedKeys: attemptedKeys, AdditionalSubtitles: h.downloadedSubtitleInventoryV3(r.Context(), effectiveFile)}) } } } @@ -1990,6 +1990,29 @@ func videoBitstreamFilterForPlanV3(plan *playback.PlanV3) string { return "" } +// lazyDVRPUStrippableV3 defers (and memoizes) the per-source RPU probe so the +// planner only shells out to ffmpeg when a Dolby Vision strip route is +// genuinely on the table; every other start never touches it. +// +// The probe belongs to planning, not to the transport: the plan's HDR10 promise +// and the durable session's RemuxDVMode are both derived from the strip +// decision and are re-read by the restart and audio-switch paths, so +// suppressing the filter downstream would leave those claims describing a +// stream the server is no longer producing. +func (h *PlaybackHandler) lazyDVRPUStrippableV3(ctx context.Context, file *models.MediaFile) func() bool { + if file == nil || strings.TrimSpace(file.FilePath) == "" { + return nil + } + var once sync.Once + strippable := true + return func() bool { + once.Do(func() { + strippable = playback.DVRPUStrippable(ctx, h.playbackConfig().FFmpegPath, file.FilePath) + }) + return strippable + } +} + func configureHLSTimelineV3(plan *playback.PlanV3, videoCodec string, segmentDuration int, durationSeconds float64) (float64, int) { if plan == nil { return 0, 0 diff --git a/internal/playback/dovi_rpu_probe.go b/internal/playback/dovi_rpu_probe.go new file mode 100644 index 00000000..7ee97e59 --- /dev/null +++ b/internal/playback/dovi_rpu_probe.go @@ -0,0 +1,275 @@ +package playback + +import ( + "context" + "log/slog" + "os" + "os/exec" + "strconv" + "strings" + "sync" + "time" +) + +// Some Dolby Vision sources carry an RPU that ffmpeg's dovi_rpu bitstream +// filter cannot parse. The filter does not fail cleanly: it rejects every +// packet, and ffmpeg keeps going, emitting a pair of errors per frame — one +// observed session produced 376,316 stderr lines before the process was +// killed. Playback never starts, the manifest build fails, and the client is +// handed a 503 after ~10 seconds, which it shows as an endless spinner: +// +// [dovi_rpu] Failed to read unit 1 (type 39). +// [vost#0:0/copy] Error applying bitstream filters to a packet: +// Invalid data found ... Invalid SEI message: payload_size too large +// +// Whether a given file survives the strip is a property of that file, not of +// ffmpeg, so it cannot be answered by supportsDoviRPUFilter. It can be answered +// in about a second by asking ffmpeg to strip a couple of seconds to nowhere. +// +// The probe is a planning input, not a transport patch: a source that fails it +// must not be planned onto a strip route at all, because the plan's HDR10 +// promise and the durable session's RemuxDVMode are both derived from that +// decision and are re-read on every later restart. +const ( + // Enough packets to reach the RPU. This catches the observed failure, in + // which the filter rejects the very first access unit and every one after + // it. It does not certify the whole title: a source that parses cleanly at + // the head and breaks an hour in still fails mid-stream, which is the + // separate "bail out on repeated per-packet filter errors" problem. + dvRPUProbeSeconds = 2 + // A copy of two seconds of video decodes nothing, so this is generous even + // for a cold spinning disk or a network mount. A source that cannot hand + // over two seconds within it cannot feed a live transcode either, and the + // timeout is inconclusive rather than fatal: the strip is kept and nothing + // is cached. Sized to stay well inside the ~10s a client waits on the + // manifest, so a slow probe cannot itself become the spinner. + dvRPUProbeTimeout = 6 * time.Second + // The rejection markers appear in the first few lines; a broken source + // killed at the timeout can otherwise pile up hundreds of thousands more. + maxDVRPUProbeOutput = 64 << 10 + // Same reasoning as the letterbox cache: bounded so a long-lived server + // with a large library cannot accumulate an entry per file forever. + maxDVRPUProbeEntries = 4096 +) + +// dvRPUVerdict separates "this source rejects the strip" from "the probe did +// not find out". Only the former may be cached: a probe cancelled with the +// client's request, or one that timed out on a cold mount, says nothing about +// the file and must not disable the strip for every later viewer. +type dvRPUVerdict int + +const ( + dvRPUUnknown dvRPUVerdict = iota + dvRPUStrippable + dvRPUBroken +) + +// DVRPUProbe records, per source file, whether the Dolby Vision RPU strip +// works. The zero value is not usable; call NewDVRPUProbe. +type DVRPUProbe struct { + mu sync.Mutex + results map[string]bool + order []string + inflight map[string]*dvRPUCall +} + +// dvRPUCall is one in-flight probe that concurrent callers for the same source +// share instead of each spawning their own ffmpeg. +type dvRPUCall struct { + done chan struct{} + strippable bool +} + +func NewDVRPUProbe() *DVRPUProbe { + return &DVRPUProbe{ + results: make(map[string]bool), + inflight: make(map[string]*dvRPUCall), + } +} + +// sharedDVRPUProbe backs DVRPUStrippable. One cache per process, like +// doviRPUCache: the planner, the progressive remux and the restart endpoints +// all ask the same question about the same files and must agree, and none of +// them should pay for a probe another already ran. +var sharedDVRPUProbe = NewDVRPUProbe() + +// DVRPUStrippable reports whether the RPU strip should be attempted for this +// source. ffmpegPath is the configured playback binary (empty selects the +// process-global discovery), matching every other capability probe here. +// +// Unknown sources are probed inline — this runs on the playback-start path, +// where a second of certainty is worth far more than handing the client a +// stream that cannot start. Only sources the planner would actually strip ever +// reach here, so the cost lands on a small minority of titles, once each. +func DVRPUStrippable(ctx context.Context, ffmpegPath, inputPath string) bool { + return sharedDVRPUProbe.CanStrip(ctx, ResolveFFmpegPath(ffmpegPath), inputPath) +} + +// CanStrip answers for one source, probing it if the verdict is not cached. +// It fails open: anything that stops the probe from reaching a conclusion +// keeps the previous strip-always behaviour, because most Profile 7 sources +// genuinely need the strip and "we did not find out" must not silently +// disable it. +func (p *DVRPUProbe) CanStrip(ctx context.Context, bin, inputPath string) bool { + if p == nil || strings.TrimSpace(inputPath) == "" { + return true + } + key, ok := dvRPUProbeKey(bin, inputPath) + if !ok { + // The file cannot even be stat'd; leave the verdict to whatever + // actually tries to open it. + return true + } + + p.mu.Lock() + if strippable, cached := p.results[key]; cached { + p.mu.Unlock() + return strippable + } + if call, running := p.inflight[key]; running { + p.mu.Unlock() + select { + case <-call.done: + return call.strippable + case <-ctx.Done(): + return true + } + } + call := &dvRPUCall{done: make(chan struct{}), strippable: true} + p.inflight[key] = call + p.mu.Unlock() + + // Detached from the caller: the leader is probing on behalf of every + // follower queued behind it, so its own client giving up must not turn a + // verdict the others are waiting on into an inconclusive fail-open — that + // would hand a live session the strip this source cannot survive. The + // probe stays bounded by dvRPUProbeTimeout, and a verdict reached after + // the leader has left is still worth caching for the next start. + started := time.Now() + verdict := runDVRPUProbe(context.WithoutCancel(ctx), bin, inputPath) + call.strippable = verdict != dvRPUBroken + + p.mu.Lock() + delete(p.inflight, key) + if verdict != dvRPUUnknown { + if _, exists := p.results[key]; !exists { + p.order = append(p.order, key) + } + p.results[key] = call.strippable + p.trimLocked() + } + p.mu.Unlock() + close(call.done) + + slog.InfoContext(ctx, "dolby vision rpu strip probed", + "component", "playback", + "input", inputPath, + "can_strip", call.strippable, + "conclusive", verdict != dvRPUUnknown, + "took_ms", time.Since(started).Milliseconds(), + ) + return call.strippable +} + +// dvRPUProbeKey identifies a probe result. Keyed on size and modification time +// as well as path so a file replaced in place is re-probed rather than +// inheriting the old verdict, and on the binary because a different ffmpeg +// build can parse a different set of RPUs. +func dvRPUProbeKey(bin, inputPath string) (string, bool) { + info, err := os.Stat(inputPath) + if err != nil { + return "", false + } + return strings.Join([]string{ + bin, + inputPath, + strconv.FormatInt(info.Size(), 10), + strconv.FormatInt(info.ModTime().UnixNano(), 10), + }, "|"), true +} + +// trimLocked bounds the cache, oldest first. A wrong eviction costs one extra +// probe of a file nobody has played in a long time. +func (p *DVRPUProbe) trimLocked() { + for len(p.order) > maxDVRPUProbeEntries { + delete(p.results, p.order[0]) + p.order = p.order[1:] + } +} + +// runDVRPUProbe strips a short head of the file to the null muxer. +// +// stderr is the signal, not the exit code: ffmpeg treats a per-packet +// bitstream-filter error as non-fatal and exits 0, which is precisely how a +// stream that could never start reached a live session in the first place. A +// rejection on stderr is therefore checked first and is conclusive whether or +// not the process also died — a source killed at the timeout mid-flood is +// broken, not merely slow. +func runDVRPUProbe(ctx context.Context, bin, inputPath string) dvRPUVerdict { + probeCtx, cancel := context.WithTimeout(ctx, dvRPUProbeTimeout) + defer cancel() + + cmd := exec.CommandContext( + probeCtx, + bin, + "-hide_banner", + "-nostats", + "-v", "error", + "-i", inputPath, + "-t", strconv.Itoa(dvRPUProbeSeconds), + "-map", "0:v:0", + "-c:v", "copy", + "-bsf:v", DV7ToHDR10BitstreamFilter, + "-an", "-sn", "-dn", + "-f", "null", "-", + ) + output := &cappedBuffer{limit: maxDVRPUProbeOutput} + cmd.Stdout = output + cmd.Stderr = output + err := cmd.Run() + + if dvRPUOutputFailed(output.String()) { + return dvRPUBroken + } + if probeCtx.Err() != nil || ctx.Err() != nil { + // The run ended on the deadline rather than on ffmpeg's own verdict, + // with the filter having said nothing. Nothing was learned about the + // file. CanStrip detaches from the request, so in practice this is the + // timeout; the ctx check keeps the rule right for any other caller. + return dvRPUUnknown + } + if err != nil { + // ffmpeg could not run, or failed for a reason unrelated to the RPU + // (an unreadable mount, a missing binary). Not a verdict on the strip. + return dvRPUUnknown + } + return dvRPUStrippable +} + +// dvRPUOutputFailed spots the filter rejecting packets even when ffmpeg exits 0 +// (it treats per-packet filter errors as non-fatal and keeps running). +func dvRPUOutputFailed(output string) bool { + lowered := strings.ToLower(output) + return strings.Contains(lowered, "error applying bitstream filters") || + strings.Contains(lowered, "failed to read access unit") || + strings.Contains(lowered, "failed to read unit") +} + +// cappedBuffer keeps the head of a stream and discards the rest, so a filter +// failing once per frame cannot turn a diagnostic into a memory problem. +type cappedBuffer struct { + limit int + buf []byte +} + +func (c *cappedBuffer) Write(p []byte) (int, error) { + if room := c.limit - len(c.buf); room > 0 { + if len(p) < room { + room = len(p) + } + c.buf = append(c.buf, p[:room]...) + } + return len(p), nil +} + +func (c *cappedBuffer) String() string { return string(c.buf) } diff --git a/internal/playback/dovi_rpu_probe_test.go b/internal/playback/dovi_rpu_probe_test.go new file mode 100644 index 00000000..6211d6bb --- /dev/null +++ b/internal/playback/dovi_rpu_probe_test.go @@ -0,0 +1,292 @@ +package playback + +import ( + "context" + "os" + "path/filepath" + "strconv" + "strings" + "sync" + "testing" + "time" +) + +// The real failure exits 0: ffmpeg treats a per-packet bitstream-filter error +// as non-fatal and keeps running, so only stderr reveals that every packet was +// rejected. Trusting the exit code alone is what let a broken strip reach a +// live session. +func TestProbeOutputDetectsRejectedPackets(t *testing.T) { + rejected := `[dovi_rpu @ 0x55] Failed to read unit 1 (type 39). +[dovi_rpu @ 0x55] Failed to read access unit from packet. +[vost#0:0/copy @ 0x55] Error applying bitstream filters to a packet: Invalid data found when processing input` + + if !dvRPUOutputFailed(rejected) { + t.Fatal("a stream of rejected packets was read as success") + } + if dvRPUOutputFailed("") { + t.Fatal("clean output was read as a failure") + } + if dvRPUOutputFailed("[hevc @ 0x55] Stream #0:0: Video: hevc") { + t.Fatal("ordinary progress output was read as a failure") + } +} + +func writeProbeFile(t *testing.T, contents string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "film.mkv") + if err := os.WriteFile(path, []byte(contents), 0o600); err != nil { + t.Fatalf("write probe fixture: %v", err) + } + return path +} + +// A file replaced in place keeps its path; inheriting the old verdict would +// keep stripping an RPU the new file cannot survive (or refuse one it can). +// Size alone misses a same-length re-encode, so the modification time counts +// too. +func TestProbeKeyChangesWithTheFile(t *testing.T) { + path := writeProbeFile(t, "original") + first, ok := dvRPUProbeKey("ffmpeg", path) + if !ok { + t.Fatal("a readable file produced no key") + } + again, _ := dvRPUProbeKey("ffmpeg", path) + if again != first { + t.Fatal("the same file produced two keys") + } + + if err := os.WriteFile(path, []byte("replaced, same length"), 0o600); err != nil { + t.Fatalf("replace probe fixture: %v", err) + } + if resized, _ := dvRPUProbeKey("ffmpeg", path); resized == first { + t.Fatal("a replaced file reused the old verdict") + } + + if sameFileOtherBinary, _ := dvRPUProbeKey("/opt/other/ffmpeg", path); sameFileOtherBinary == first { + t.Fatal("a different ffmpeg build reused the old verdict") + } +} + +// A file that cannot be stat'd gets no key: the verdict belongs to whatever +// actually tries to open it, not to a probe that never ran. +func TestProbeKeyRefusesAnUnreadableFile(t *testing.T) { + if _, ok := dvRPUProbeKey("ffmpeg", filepath.Join(t.TempDir(), "absent.mkv")); ok { + t.Fatal("a missing file produced a cache key") + } +} + +// A nil probe must not change behaviour: most Profile 7 sources need the strip, +// so "no probe configured" has to mean "strip", not "never strip". +func TestNilProbeKeepsStripping(t *testing.T) { + var probe *DVRPUProbe + if !probe.CanStrip(context.Background(), "ffmpeg", "/media/film.mkv") { + t.Fatal("a nil probe suppressed the strip") + } +} + +func TestProbeRefusesAnEmptyPath(t *testing.T) { + probe := NewDVRPUProbe() + if !probe.CanStrip(context.Background(), "ffmpeg", " ") { + t.Fatal("an unprobeable input should fall back to stripping") + } +} + +// Anything that stops the probe from reaching a conclusion — a cancelled +// request, a cold mount that outruns the timeout, an ffmpeg that will not +// start — must leave the strip on and leave the cache empty. Caching it would +// disable the strip for that file for every later viewer, which is the +// opposite of the nil-probe rule above. +func TestInconclusiveProbeIsNeitherFatalNorCached(t *testing.T) { + path := writeProbeFile(t, "not really a movie") + probe := NewDVRPUProbe() + + // A binary that does not exist fails to start: no verdict on the file. + if !probe.CanStrip(context.Background(), filepath.Join(t.TempDir(), "no-such-ffmpeg"), path) { + t.Fatal("an ffmpeg that could not run suppressed the strip") + } + probe.mu.Lock() + cached := len(probe.results) + probe.mu.Unlock() + if cached != 0 { + t.Fatalf("an inconclusive probe was cached: %d entries", cached) + } +} + +// The leader probes on behalf of every follower queued behind it, so its own +// client giving up must not abandon the run: the verdict is still reached and +// still cached for the next start. While the probe rode the caller's context, +// one client disconnecting turned the answer its neighbours were waiting on +// into an inconclusive fail-open and hung them on the very source it had been +// about to reject. +func TestProbeSurvivesTheLeaderLeaving(t *testing.T) { + path := writeProbeFile(t, "not really a movie") + bin, runLog := writeRejectingFFmpeg(t) + probe := NewDVRPUProbe() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + if probe.CanStrip(ctx, bin, path) { + t.Fatal("a cancelled caller abandoned the probe and fell back to stripping") + } + if runs := countRuns(t, runLog); runs != 1 { + t.Fatalf("the probe ran %d times, want 1", runs) + } + probe.mu.Lock() + cached := len(probe.results) + probe.mu.Unlock() + if cached != 1 { + t.Fatalf("the verdict was not cached for the next start: %d entries", cached) + } +} + +// A stderr-confirmed rejection is the one verdict worth remembering. +func TestConclusiveVerdictsAreCachedAndReused(t *testing.T) { + path := writeProbeFile(t, "not really a movie") + probe := NewDVRPUProbe() + key, ok := dvRPUProbeKey("ffmpeg", path) + if !ok { + t.Fatal("a readable file produced no key") + } + probe.mu.Lock() + probe.results[key] = false + probe.order = append(probe.order, key) + probe.mu.Unlock() + + if probe.CanStrip(context.Background(), "ffmpeg", path) { + t.Fatal("a cached rejection was ignored") + } +} + +// writeRejectingFFmpeg stands in for the real failure: it exits 0 while +// printing the dovi_rpu rejection, and records every invocation so a test can +// tell how many probes actually ran. +func writeRejectingFFmpeg(t *testing.T) (bin, runLog string) { + t.Helper() + dir := t.TempDir() + bin = filepath.Join(dir, "ffmpeg") + runLog = filepath.Join(dir, "runs") + script := "#!/bin/sh\n" + + "echo run >> " + runLog + "\n" + + "sleep 1\n" + + "echo '[dovi_rpu @ 0x55] Failed to read unit 1 (type 39).' >&2\n" + + "exit 0\n" + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatalf("write fake ffmpeg: %v", err) + } + return bin, runLog +} + +func countRuns(t *testing.T, runLog string) int { + t.Helper() + contents, err := os.ReadFile(runLog) + if os.IsNotExist(err) { + return 0 + } + if err != nil { + t.Fatalf("read run log: %v", err) + } + return len(strings.Fields(string(contents))) +} + +// Concurrent starts of the same title (a retry, a second household profile) +// must share one ffmpeg rather than each spawning a full probe, and all of them +// must come back with the one verdict it reached. +func TestConcurrentProbesShareOneRun(t *testing.T) { + path := writeProbeFile(t, "not really a movie") + bin, runLog := writeRejectingFFmpeg(t) + probe := NewDVRPUProbe() + + var wg sync.WaitGroup + results := make([]bool, 8) + for i := range results { + wg.Add(1) + go func(i int) { + defer wg.Done() + results[i] = probe.CanStrip(context.Background(), bin, path) + }(i) + } + wg.Wait() + + for i, strippable := range results { + if strippable { + t.Fatalf("caller %d missed the rejection the probe found", i) + } + } + if runs := countRuns(t, runLog); runs != 1 { + t.Fatalf("the same source was probed %d times, want 1", runs) + } + + // The verdict is conclusive, so a later start must reuse it rather than + // pay for the probe again. + if probe.CanStrip(context.Background(), bin, path) { + t.Fatal("the cached rejection was ignored") + } + if runs := countRuns(t, runLog); runs != 1 { + t.Fatalf("a cached verdict still spawned ffmpeg: %d runs", runs) + } +} + +// A follower whose own request is cancelled must not block on the leader. +func TestFollowerLeavesWhenItsRequestIsCancelled(t *testing.T) { + path := writeProbeFile(t, "not really a movie") + probe := NewDVRPUProbe() + key, _ := dvRPUProbeKey("ffmpeg", path) + probe.mu.Lock() + probe.inflight[key] = &dvRPUCall{done: make(chan struct{}), strippable: false} + probe.mu.Unlock() + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + done := make(chan bool, 1) + go func() { done <- probe.CanStrip(ctx, "ffmpeg", path) }() + + select { + case strippable := <-done: + if !strippable { + t.Fatal("a cancelled follower reported a verdict it never waited for") + } + case <-time.After(2 * time.Second): + t.Fatal("a cancelled follower blocked on the leader's probe") + } +} + +// A long-lived server must not accumulate one entry per file it has ever +// played. +func TestProbeCacheIsBounded(t *testing.T) { + probe := NewDVRPUProbe() + oldest := "ffmpeg|/media/film.mkv|0|0" + for i := range maxDVRPUProbeEntries + 5 { + key := "ffmpeg|/media/film.mkv|" + strconv.Itoa(i) + "|0" + probe.results[key] = true + probe.order = append(probe.order, key) + } + probe.trimLocked() + + if len(probe.results) != maxDVRPUProbeEntries || len(probe.order) != maxDVRPUProbeEntries { + t.Fatalf("probe cache grew unbounded: %d", len(probe.results)) + } + if _, ok := probe.results[oldest]; ok { + t.Fatal("the oldest entry survived eviction") + } +} + +// The filter failing once per frame produced 376,316 stderr lines in the +// observed session; the markers are all in the first few, so the capture is +// bounded and the detection still works on the truncated head. +func TestProbeOutputCaptureIsBounded(t *testing.T) { + buf := &cappedBuffer{limit: 64} + head := "[dovi_rpu] Failed to read unit 1 (type 39).\n" + if n, err := buf.Write([]byte(head)); n != len(head) || err != nil { + t.Fatalf("short write reported to ffmpeg: %d %v", n, err) + } + flood := make([]byte, 1<<20) + if n, err := buf.Write(flood); n != len(flood) || err != nil { + t.Fatalf("short write reported to ffmpeg: %d %v", n, err) + } + if len(buf.buf) != 64 { + t.Fatalf("capture exceeded its limit: %d bytes", len(buf.buf)) + } + if !dvRPUOutputFailed(buf.String()) { + t.Fatal("truncation lost the rejection the probe exists to spot") + } +} diff --git a/internal/playback/plan_v3.go b/internal/playback/plan_v3.go index f6981d40..ef25b7e4 100644 --- a/internal/playback/plan_v3.go +++ b/internal/playback/plan_v3.go @@ -34,12 +34,27 @@ type PlannerInputV3 struct { // return a superset of Registry (local ∪ node capabilities) and should // memoize; the transport layer re-validates whichever executor is // actually selected. - HLSRegistry func() *TransformationRegistryV3 + HLSRegistry func() *TransformationRegistryV3 + // DVRPUStrippable reports whether this particular source survives the + // Dolby Vision RPU strip. The registries answer whether the executor + // carries the transformation; this answers whether the file does, which + // no capability probe can. Nil means "assume it does", preserving the + // pre-probe behaviour for callers that cannot run one (the shadow + // planner, tests). Lazy for the same reason as HLSRegistry: it shells out + // to ffmpeg, so it is consulted only once every cheap eligibility gate + // has already passed and a strip route is genuinely on the table. + DVRPUStrippable func() bool Now time.Time AttemptedKeys []string AdditionalSubtitles []SubtitleInventoryEntryV3 } +// dvRPUStrippable resolves the per-source strip verdict, defaulting to true +// when no probe is wired in. +func (input PlannerInputV3) dvRPUStrippable() bool { + return input.DVRPUStrippable == nil || input.DVRPUStrippable() +} + // hlsRegistry resolves the registry HLS deliveries gate on: the widened // local∪node registry when provided, otherwise the local one. Callers must // keep it behind short-circuits so transformation-free routes never force @@ -129,8 +144,30 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 { if !dvStripEligible && hlsEngineOK && source.DynamicRange == "dolby_vision" { dvStripEligible = canStripDolbyVisionToHDR10V3(source, input.Request, input.hlsRegistry()) } + // A source whose RPU ffmpeg cannot parse must lose the strip here rather + // than at the transport, so that the plan's HDR10 promise, the durable + // session's RemuxDVMode and every restart derived from it stay consistent + // with what the pipeline can actually produce. Ordered last: the probe + // only runs once an executor has been found for a strip this client wants. + dvStripUnsupportedBySource := false + if dvStripEligible && !input.dvRPUStrippable() { + dvStripUnsupportedBySource = true + dvStripEligible = false + dvStripEligibleLocal = false + } clientDV81Eligible := canClientTransformDV7ToDV81V3(source, input.Request) clientHDR10Eligible := canClientTransformDV7ToHDR10V3(source, input.Request) + // With the server strip gone, a client that cannot take the source range + // and cannot run its own DV transformation has no route left: this + // codebase has no tone-map recipe, so every remaining branch funnels into + // planVideoTranscodeV3's hdr_transcode_unsupported. Terminate here instead + // so the client is told the actual cause — a source whose Dolby Vision + // metadata cannot be removed — rather than a generic HDR message that + // sends the user looking for a missing encoder. + if dvStripUnsupportedBySource && !rangeOK && !clientDV81Eligible && !clientHDR10Eligible { + return terminalPlannerResultV3("dv_conversion_unsupported", + "This source's Dolby Vision metadata cannot be removed cleanly, and this device cannot play the source as it is.", false) + } base := PlanV3{ ProtocolVersion: ProtocolV3, @@ -156,6 +193,14 @@ func PlanPlaybackV3(input PlannerInputV3) PlannerResultV3 { Message: "The source is flagged HDR without precise range metadata and is delivered as HDR10.", }) } + if dvStripUnsupportedBySource { + // Say why the HDR10 route this client is capable of was not taken; + // otherwise the fallback looks like an unexplained quality drop. + base.DegradationWarnings = append(base.DegradationWarnings, DegradationWarningV3{ + Code: "dolby_vision_strip_unsupported_by_source", + Message: "This source's Dolby Vision metadata cannot be removed cleanly, so the validated HDR10 route is unavailable for it.", + }) + } if !detailedVideoEvidenceCompleteV3(source) { return terminalPlannerResultV3("source_metadata_incomplete", "The source is missing video metadata required for a validated playback route.", true) } diff --git a/internal/playback/protocol_v3_test.go b/internal/playback/protocol_v3_test.go index 65f757ea..1970b607 100644 --- a/internal/playback/protocol_v3_test.go +++ b/internal/playback/protocol_v3_test.go @@ -1004,3 +1004,106 @@ func testTransformationRegistryV3() *TransformationRegistryV3 { {Name: "server_dv7_to_hdr10", Available: true}, }) } + +// A source whose RPU ffmpeg cannot parse must lose the strip in the plan, not +// at the transport. The plan is what promises HDR10, and the durable session's +// RemuxDVMode — re-read by every later restart, seek and audio switch — is +// derived from it, so a plan that still names server_dv7_to_hdr10 puts the +// hanging filter back no matter what the start path did with it. With no +// tone-map recipe in the tree there is no route left for an HDR10-only client, +// and the terminal has to name the real cause. +func TestPlanPlaybackV3AbandonsStripForAnUnstrippableSource(t *testing.T) { + file := unstrippableProfile7FixtureV3() + req := hdr10OnlyProfile7RequestV3() + registry := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "server_dv7_to_hdr10", Available: true}}) + input := PlannerInputV3{Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, Settings: PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}, Registry: registry} + + // Baseline: with a parseable RPU this is the validated HDR10 remux. + if healthy := PlanPlaybackV3(input); healthy.Plan == nil || len(healthy.Plan.Transformations) != 1 || healthy.Plan.Transformations[0].Name != "server_dv7_to_hdr10" { + t.Fatalf("the strip route regressed for a healthy source: %#v", healthy) + } + + input.DVRPUStrippable = func() bool { return false } + result := PlanPlaybackV3(input) + if result.Plan != nil { + t.Fatalf("a source that cannot be stripped was still planned onto a route: %#v", result.Plan) + } + if result.Terminal == nil || result.Terminal.Reason != "dv_conversion_unsupported" { + t.Fatalf("terminal = %#v, want the Dolby Vision cause rather than a generic HDR message", result.Terminal) + } +} + +// The strip is a server capability, not the only one: a client that can do the +// conversion itself must still get its route, with the reason the server route +// was dropped attached. +func TestPlanPlaybackV3KeepsTheClientTransformWhenTheSourceCannotBeStripped(t *testing.T) { + file := unstrippableProfile7FixtureV3() + req := hdr10OnlyProfile7RequestV3() + req.ClientFeatures = append(req.ClientFeatures, FeatureClientVideoTransforms) + req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureClientVideoTransforms) + direct := req.ClientPlaybackContext.Engines[string(EngineMedia3DirectV3)] + direct.Transformations = []TransformationV3{{Name: ClientDV7ToHDR10V3, Executor: "client", RecipeVersion: ClientDVTransformVersionV3}} + req.ClientPlaybackContext.Engines[string(EngineMedia3DirectV3)] = direct + registry := NewTransformationRegistryV3([]TransformationSpecV3{{Name: "server_dv7_to_hdr10", Available: true}}) + + result := PlanPlaybackV3(PlannerInputV3{ + Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, + Settings: PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}, Registry: registry, + DVRPUStrippable: func() bool { return false }, + }) + if result.Plan == nil { + t.Fatalf("terminal = %#v, want the client-side transformation route", result.Terminal) + } + for _, transformation := range result.Plan.Transformations { + if transformation.Name == "server_dv7_to_hdr10" { + t.Fatalf("the unusable server strip survived: %#v", result.Plan.Transformations) + } + } + if !hasDegradationWarningV3(result.Plan.DegradationWarnings, "dolby_vision_strip_unsupported_by_source") { + t.Fatalf("the client was not told why the server route was dropped: %#v", result.Plan.DegradationWarnings) + } +} + +func unstrippableProfile7FixtureV3() *models.MediaFile { + file := detailedFixtureFileV3() + file.VideoTracks[0].DVProfile = 7 + file.VideoTracks[0].DVBLCompatID = 6 + file.VideoTracks[0].DVELPresent = false + file.VideoTracks[0].DVEnhancementLayer = "" + file.VideoTracks[0].VideoRange = "DolbyVision" + file.VideoTracks[0].VideoRangeType = "DOVIWithEL" + return file +} + +func hdr10OnlyProfile7RequestV3() StartRequestV3 { + req := validStartRequestV3() + req.ClientFeatures = append(req.ClientFeatures, FeatureDetailedDecodeV3) + req.ClientPlaybackContext.Features = append(req.ClientPlaybackContext.Features, FeatureDetailedDecodeV3) + req.Capabilities.VideoDecode = []VideoDecodeCapabilityV3{{Codec: "hevc", Profiles: []string{"main 10"}, Levels: []int{153}, BitDepths: []int{10}, MaxWidth: 3840, MaxHeight: 2160, MaxFrameRate: 60, MaxBitrateKbps: 80_000, Hardware: true}} + req.Capabilities.HDRDetails = &HDRCapabilitiesV3{HDR10: true, DolbyVisionProfiles: []int{5, 8}} + req.ClientPlaybackContext.Output.HDRDetails = req.Capabilities.HDRDetails + return req +} + +// The probe is expensive, so it must sit behind every cheap gate: a source +// nobody would strip anyway must never spawn one. +func TestPlanPlaybackV3DoesNotProbeWhenNoStripIsOnTheTable(t *testing.T) { + file := detailedFixtureFileV3() + file.VideoTracks[0].VideoRange = "SDR" + file.VideoTracks[0].VideoRangeType = "SDR" + file.VideoTracks[0].BitDepth = 8 + req := validStartRequestV3() + probed := false + result := PlanPlaybackV3(PlannerInputV3{ + Request: req, RequestedFile: file, EffectiveFile: file, AudioTrackIndex: 0, + Settings: PlannerSettingsV3{TranscodeEnabled: true, Allow4KTranscode: true}, + Registry: testTransformationRegistryV3(), + DVRPUStrippable: func() bool { probed = true; return true }, + }) + if result.Plan == nil { + t.Fatalf("terminal = %#v", result.Terminal) + } + if probed { + t.Fatal("an ordinary SDR source paid for a Dolby Vision RPU probe") + } +} diff --git a/internal/playback/remux.go b/internal/playback/remux.go index bdc0d7c6..4f5d330e 100644 --- a/internal/playback/remux.go +++ b/internal/playback/remux.go @@ -74,9 +74,13 @@ func supportsDoviRPUFilter(bin string) bool { // remuxDVProfile neutralizes a Dolby Vision profile the local ffmpeg cannot // handle. Profile 7 is the only profile that triggers an RPU strip in -// buildRemuxArgs; when the dovi_rpu filter is unavailable the remux must -// still start (an unknown bitstream filter aborts ffmpeg immediately), so -// fall back to the pre-strip behavior instead of failing playback. +// buildRemuxArgs; when the strip is unavailable — the dovi_rpu filter is +// missing, or this source's RPU cannot be parsed — the remux must still start +// (an unknown bitstream filter aborts ffmpeg immediately, and a filter that +// rejects every packet hangs the session), so fall back to the pre-strip +// behavior instead of failing playback. Only the legacy/auto mode takes this +// route; the explicit v3 strip recipe fails loudly instead, because it has +// promised the client an HDR10 output it could not then produce. func remuxDVProfile(dvProfile int, canStripRPU bool) int { if dvProfile == 7 && !canStripRPU { return 0 @@ -223,7 +227,8 @@ func StartRemuxWithDVMode(ctx context.Context, filePath, outputFormat string, se tagDVSampleEntry := false switch mode { case "", RemuxDVLegacyAutoV3: - effectiveProfile = remuxDVProfile(dvProfile, supportsDoviRPUFilter(bin)) + effectiveProfile = remuxDVProfile(dvProfile, supportsDoviRPUFilter(bin) && + (dvProfile != 7 || sharedDVRPUProbe.CanStrip(ctx, bin, filePath))) case RemuxDVStripToHDR10V3: if dvProfile != 7 && dvProfile != 8 { cancel() @@ -233,6 +238,17 @@ func StartRemuxWithDVMode(ctx context.Context, filePath, outputFormat string, se cancel() return nil, fmt.Errorf("Dolby Vision HDR10 remux requires the dovi_rpu bitstream filter") } + // The planner refuses this recipe for a source that fails the probe, + // so reaching here means a session or stream token minted before the + // verdict was known. Fail definitively: copying the base layer without + // the strip would leave dangling RPUs (the decoder stall this recipe + // exists to prevent) while still claiming HDR10, and attempting the + // strip anyway is the per-packet rejection that hangs the session. The + // next start re-plans against the now-cached verdict. + if !sharedDVRPUProbe.CanStrip(ctx, bin, filePath) { + cancel() + return nil, fmt.Errorf("this source's Dolby Vision RPU cannot be stripped to HDR10") + } // buildRemuxArgs uses profile 7 as the explicit strip sentinel; the // filter is equally required for a compatible profile 8 base layer. effectiveProfile = 7 diff --git a/internal/playback/remux_dv_test.go b/internal/playback/remux_dv_test.go index a1e19e20..bb082b9a 100644 --- a/internal/playback/remux_dv_test.go +++ b/internal/playback/remux_dv_test.go @@ -1,6 +1,10 @@ package playback import ( + "context" + "io" + "os" + "path/filepath" "strings" "testing" ) @@ -97,3 +101,79 @@ func TestBuildRemuxArgsDelaysMoovForCopiedAtmosConfiguration(t *testing.T) { t.Fatalf("remux must delay moov until copied audio is parsed, args=%v", strings.Join(args, " ")) } } + +// writeProbeAwareFFmpeg stands in for an ffmpeg that carries the dovi_rpu +// filter but cannot parse this source's RPU: it advertises the filter, fails +// the probe the way the real one does (rejecting packets while exiting 0), and +// records the arguments of every non-probe invocation. +func writeProbeAwareFFmpeg(t *testing.T) (bin, argLog string) { + t.Helper() + dir := t.TempDir() + bin = filepath.Join(dir, "ffmpeg") + argLog = filepath.Join(dir, "args") + script := "#!/bin/sh\n" + + "case \"$*\" in\n" + + " *-bsfs*) echo dovi_rpu; exit 0;;\n" + + " *'-f null'*) echo '[dovi_rpu @ 0x55] Failed to read unit 1 (type 39).' >&2; exit 0;;\n" + + "esac\n" + + "echo \"$*\" >> " + argLog + "\n" + + "exit 0\n" + if err := os.WriteFile(bin, []byte(script), 0o755); err != nil { + t.Fatalf("write fake ffmpeg: %v", err) + } + return bin, argLog +} + +func remuxSourceFile(t *testing.T) string { + t.Helper() + path := filepath.Join(t.TempDir(), "movie.mkv") + if err := os.WriteFile(path, []byte("not really a movie"), 0o600); err != nil { + t.Fatalf("write remux fixture: %v", err) + } + return path +} + +// The legacy/auto mode derives the strip from the profile alone, with no plan +// to consult — the web player, jellycompat and pre-v3 stream tokens all arrive +// here. An unparseable RPU has to neutralize the profile the same way a missing +// dovi_rpu filter does, or the filter rejects every packet and the response +// never produces a byte. +func TestLegacyRemuxDropsTheStripForAnUnstrippableSource(t *testing.T) { + bin, argLog := writeProbeAwareFFmpeg(t) + path := remuxSourceFile(t) + + session, err := StartRemuxWithDVMode(context.Background(), path, "mp4", 0, false, -1, 7, RemuxDVLegacyAutoV3, bin) + if err != nil { + t.Fatalf("legacy remux refused to start: %v", err) + } + // Drain to EOF so the stand-in has finished recording before it is killed. + _, _ = io.ReadAll(session) + session.Close() + + recorded, err := os.ReadFile(argLog) + if err != nil { + t.Fatalf("the remux never ran: %v", err) + } + if strings.Contains(string(recorded), DV7ToHDR10BitstreamFilter) { + t.Fatalf("the hanging filter survived into the remux: %s", recorded) + } +} + +// The explicit v3 recipe has already promised the client HDR10. Reaching it +// with an unstrippable source means a session or stream token minted before +// the verdict was known, and neither honouring nor silently dropping the strip +// is right: fail so the request gets a definite error instead of a stalled +// stream, and so the next start re-plans onto a route that works. +func TestExplicitStripRecipeRefusesAnUnstrippableSource(t *testing.T) { + bin, _ := writeProbeAwareFFmpeg(t) + path := remuxSourceFile(t) + + session, err := StartRemuxWithDVMode(context.Background(), path, "mp4", 0, false, -1, 7, RemuxDVStripToHDR10V3, bin) + if err == nil { + session.Close() + t.Fatal("the explicit HDR10 strip accepted a source it cannot strip") + } + if !strings.Contains(err.Error(), "cannot be stripped") { + t.Fatalf("error = %v, want it to name the unstrippable source", err) + } +}