Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 48 additions & 2 deletions pkg/audit/notifier.go
Original file line number Diff line number Diff line change
Expand Up @@ -119,11 +119,24 @@ func (n *StderrNotifier) NotifyOperation(op AuditOperation) error {
// DefaultNotifierPrefix is the standard notification line prefix.
const DefaultNotifierPrefix = "[rffmpeg]"

// HardwareEncoderMarker suffixes a hardware encoder in a capabilities summary
// (e.g. "h264_qsv[HW],libx264"). It is the contract between the summary the
// worker stores and formatCapabilitiesForDisplay below.
const HardwareEncoderMarker = "[HW]"

// maxSoftwareEncodersOnLine caps how many software encoders the rewrite chain
// line lists. A stock ffmpeg build reports well over a hundred encoders and
// this line reaches the user's terminal on every rewrite, so the software tail
// is replaced by a "[... +N more]" count. Hardware encoders are exempt: they
// are what the rewrite decision is about and they are a short, bounded set.
const maxSoftwareEncodersOnLine = 10

// FormatRewriteChainLine renders the detailed rewrite chain as a complete
// notification line, including the "<prefix> <LEVEL>: " prefix. It is the
// single source of truth for this format: the Worker's per-job stderr stream
// emits its output so CLI clients see exactly what worker-side
// logs show.
// logs show. The capabilities are summarized for display; the persisted audit
// record keeps the full summary.
func FormatRewriteChainLine(capabilitiesSummary string, requested string, rewritten string, reason string, level NotifyLevel) string {
return fmt.Sprintf("%s %s: %s\n", DefaultNotifierPrefix, level, formatRewriteChainSegments(capabilitiesSummary, requested, rewritten, reason))
}
Expand All @@ -134,7 +147,7 @@ func FormatRewriteChainLine(capabilitiesSummary string, requested string, rewrit
func formatRewriteChainSegments(capabilitiesSummary string, requested string, rewritten string, reason string) string {
var parts []string
if capabilitiesSummary != "" {
parts = append(parts, fmt.Sprintf("Worker capabilities: %s", capabilitiesSummary))
parts = append(parts, fmt.Sprintf("Worker capabilities: %s", formatCapabilitiesForDisplay(capabilitiesSummary)))
}
if requested != "" {
parts = append(parts, fmt.Sprintf("Requested: %s", requested))
Expand All @@ -148,6 +161,39 @@ func formatRewriteChainSegments(capabilitiesSummary string, requested string, re
return strings.Join(parts, " | ")
}

// formatCapabilitiesForDisplay renders a capabilities summary for the rewrite
// chain line: hardware encoders first, then the software encoders truncated to
// maxSoftwareEncodersOnLine. Only this line is bounded — the audit record
// stores the summary as produced.
func formatCapabilitiesForDisplay(capabilitiesSummary string) string {
var hardware, software []string
for _, name := range strings.Split(capabilitiesSummary, ",") {
if strings.HasSuffix(name, HardwareEncoderMarker) {
hardware = append(hardware, name)
continue
}
software = append(software, name)
}

groups := make([]string, 0, 2)
if len(hardware) > 0 {
groups = append(groups, strings.Join(hardware, ","))
}
if len(software) > 0 {
groups = append(groups, summarizeNames(software, maxSoftwareEncodersOnLine))
}
return strings.Join(groups, "; ")
}

// summarizeNames joins up to limit names and appends the count of the omitted
// tail, so the displayed line stays bounded for any ffmpeg build size.
func summarizeNames(names []string, limit int) string {
if len(names) <= limit {
return strings.Join(names, ",")
}
return fmt.Sprintf("%s [... +%d more]", strings.Join(names[:limit], ","), len(names)-limit)
}

// NotifyRewriteChain outputs the detailed rewrite chain notification.
// Format: [rffmpeg] Worker capabilities: {caps} | Requested: {req} | Rewritten: {rew} | Reason: {reason}
func (n *StderrNotifier) NotifyRewriteChain(capabilitiesSummary string, requested string, rewritten string, reason string, level NotifyLevel) error {
Expand Down
67 changes: 67 additions & 0 deletions pkg/audit/notifier_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package audit

import (
"bytes"
"fmt"
"strings"
"sync"
"testing"
Expand Down Expand Up @@ -260,6 +261,72 @@ func TestStderrNotifier_NotifyRewriteChain_EmptyFields(t *testing.T) {
}
}

// TestFormatRewriteChainLine_TruncatesCapabilities pins the display transform
// of the capabilities summary: hardware encoders (marked "[HW]") are listed in
// full ahead of the software tail, which is capped so a stock ffmpeg build's
// hundred-plus encoders cannot flood the terminal. Both display sinks must
// render the same line, and the summary handed in here is not modified — the
// audit record keeps it whole.
func TestFormatRewriteChainLine_TruncatesCapabilities(t *testing.T) {
software := func(count int) string {
names := make([]string, count)
for i := range count {
names[i] = fmt.Sprintf("sw-%02d", i)
}
return strings.Join(names, ",")
}

tests := []struct {
name string
summary string
wantCaps string
}{
{
name: "software list below the cap is kept whole",
summary: software(7),
wantCaps: software(7),
},
{
name: "software list at the cap is kept whole",
summary: software(10),
wantCaps: software(10),
},
{
name: "software list past the cap reports the omitted tail",
summary: software(12),
wantCaps: software(10) + " [... +2 more]",
},
{
name: "hardware leads, is never truncated and keeps its marker",
summary: software(12) + ",h264_qsv[HW],av1_amf[HW]",
wantCaps: "h264_qsv[HW],av1_amf[HW]; " + software(10) + " [... +2 more]",
},
{
name: "hardware-only summary is untouched",
summary: "h264_nvenc[HW],h264_qsv[HW]",
wantCaps: "h264_nvenc[HW],h264_qsv[HW]",
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
want := "[rffmpeg] INFO: Worker capabilities: " + tt.wantCaps + " | Requested: libx264 | Rewritten: h264_qsv | Reason: upgrade\n"

if got := FormatRewriteChainLine(tt.summary, "libx264", "h264_qsv", "upgrade", InfoLevel); got != want {
t.Errorf("FormatRewriteChainLine() = %q, want %q", got, want)
}

var buf bytes.Buffer
if err := NewNotifierWithOutput(&buf).NotifyRewriteChain(tt.summary, "libx264", "h264_qsv", "upgrade", InfoLevel); err != nil {
t.Fatalf("NotifyRewriteChain failed: %v", err)
}
if got := buf.String(); got != want {
t.Errorf("NotifyRewriteChain() = %q, want %q", got, want)
}
})
}
}

func TestSilentNotifier(t *testing.T) {
notifier := NewSilentNotifier()

Expand Down
8 changes: 6 additions & 2 deletions pkg/audit/types.go
Original file line number Diff line number Diff line change
Expand Up @@ -102,8 +102,12 @@ type AuditOperation struct {
// DecisionReason explains why this rewrite was applied.
DecisionReason string `json:"decision_reason"`

// CapabilitiesSummary provides a concise summary of the worker's hardware capabilities
// at the time of this operation (e.g., "h264_nvenc,h264_qsv,libx264").
// CapabilitiesSummary records the worker's capabilities at the time of
// this operation as a comma-separated encoder list, hardware encoders
// suffixed "[HW]" (e.g. "h264_qsv[HW],libx264"). It is the full list the
// worker reported, so an operator can reconstruct what was available at
// this moment; only the rewrite chain line shown to users is truncated
// (see FormatRewriteChainLine).
CapabilitiesSummary string `json:"capabilities_summary,omitempty"`

// Context contains additional contextual information.
Expand Down
25 changes: 20 additions & 5 deletions pkg/worker/rewrite_adapter.go
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@ import (
"strings"
"sync"

"github.com/tsic404/rffmpeg/pkg/audit"
"github.com/tsic404/rffmpeg/pkg/encoder"
"github.com/tsic404/rffmpeg/pkg/encoder/rewrite"
"github.com/tsic404/rffmpeg/pkg/ffmpegopts"
Expand Down Expand Up @@ -326,18 +327,32 @@ type RewriteResult struct {
FallbackUsed bool
}

// buildCapabilitiesSummary creates a concise string summary of worker hardware capabilities.
// buildCapabilitiesSummary creates a string summary of worker hardware capabilities.
// Every available encoder is listed in the order the worker reported it, with
// hardware encoders suffixed by audit.HardwareEncoderMarker. The summary is
// stored in the audit record in full; the rewrite chain line truncates it for
// display (see audit.FormatRewriteChainLine).
func (a *RewriteAdapter) buildCapabilitiesSummary(hwCaps *rewrite.HardwareCapabilities) string {
if hwCaps == nil {
return "unknown"
}
// Hardware membership comes from what the worker reported (IsHW, carried in
// HardwareEncoders), not from the static EncoderFamily classifier: the
// classifier only knows a fixed set of names, so a detected encoder such as
// av1_amf or mjpeg_qsv would be misfiled as software and could then be
// truncated away on the display line.
hardware := make(map[encoder.EncoderFamily]struct{}, len(hwCaps.HardwareEncoders))
for _, enc := range hwCaps.HardwareEncoders {
hardware[enc] = struct{}{}
}

encoders := make([]string, 0, len(hwCaps.AvailableEncoders))
for _, enc := range hwCaps.AvailableEncoders {
marker := string(enc)
if enc.IsHardware() {
marker += "[HW]"
if _, isHW := hardware[enc]; isHW {
encoders = append(encoders, string(enc)+audit.HardwareEncoderMarker)
continue
}
encoders = append(encoders, marker)
encoders = append(encoders, string(enc))
}
if len(encoders) == 0 {
return "none"
Expand Down
128 changes: 128 additions & 0 deletions pkg/worker/rewrite_adapter_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,15 +2,40 @@ package worker

import (
"context"
"fmt"
"slices"
"strings"
"testing"

"github.com/tsic404/rffmpeg/pkg/audit"
"github.com/tsic404/rffmpeg/pkg/encoder"
"github.com/tsic404/rffmpeg/pkg/encoder/rewrite"
"github.com/tsic404/rffmpeg/pkg/protocol"
)

// testHWEncoderNames lists every hardware encoder family the shared classifier
// recognises; the capabilities summary must mark all of them.
var testHWEncoderNames = []string{
"h264_nvenc", "h264_qsv", "h264_vaapi", "h264_amf", "h264_videotoolbox",
"hevc_nvenc", "hevc_qsv", "hevc_vaapi", "hevc_amf", "hevc_videotoolbox",
"vp9_nvenc", "vp9_qsv", "vp9_vaapi",
"av1_nvenc", "av1_qsv", "av1_vaapi",
}

// testEncoderInfos builds a capability set with the software encoders listed
// first, so a summary that puts hardware up front proves it reorders rather
// than echoing the input order.
func testEncoderInfos(hw, sw int) []protocol.EncoderInfo {
encoders := make([]protocol.EncoderInfo, 0, hw+sw)
for i := range sw {
encoders = append(encoders, protocol.EncoderInfo{Name: fmt.Sprintf("sw-%02d", i), Type: "video"})
}
for _, name := range testHWEncoderNames[:hw] {
encoders = append(encoders, protocol.EncoderInfo{Name: name, Type: "video", IsHW: true})
}
return encoders
}

func TestRewriteAdapter_RewriteArgs(t *testing.T) {
adapter := NewRewriteAdapter()

Expand Down Expand Up @@ -980,3 +1005,106 @@ func TestRewriteAdapter_AutoHWUpgradeReportsParamFate(t *testing.T) {
}
}
}

// TestRewriteAdapter_CapabilitiesSummaryMarksReportedHardware pins the summary
// the worker reports: every encoder it saw, in the reported order, with the
// hardware ones marked from the worker's own IsHW flag — not from the static
// EncoderFamily classifier, which only knows a fixed set of names. The summary
// stays complete; truncation for humans happens on the display line.
func TestRewriteAdapter_CapabilitiesSummaryMarksReportedHardware(t *testing.T) {
tests := []struct {
name string
encoders []protocol.EncoderInfo
wantSummary string
}{
{
name: "no encoders",
wantSummary: "none",
},
{
name: "reported hardware keeps its marker and order",
encoders: []protocol.EncoderInfo{
{Name: "libx264", Type: "video"},
{Name: "h264_qsv", Type: "video", IsHW: true},
{Name: "libx265", Type: "video"},
},
wantSummary: "libx264,h264_qsv[HW],libx265",
},
{
// The probe classifies hardware by name suffix, so it reports
// encoders the static family classifier does not know. Missing the
// marker on these let them sort into the truncatable software group
// and disappear from the display line behind "[... +N more]".
name: "reported hardware unknown to the static classifier is marked",
encoders: []protocol.EncoderInfo{
{Name: "av1_amf", Type: "video", IsHW: true},
{Name: "vp9_amf", Type: "video", IsHW: true},
{Name: "mpeg2_vaapi", Type: "video", IsHW: true},
{Name: "mjpeg_qsv", Type: "video", IsHW: true},
{Name: "libx264", Type: "video"},
},
wantSummary: "av1_amf[HW],vp9_amf[HW],mpeg2_vaapi[HW],mjpeg_qsv[HW],libx264",
},
}

for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
adapter := NewRewriteAdapter()
hwCaps := adapter.convertCapabilities(&protocol.WorkerCapabilities{VideoEncoders: tt.encoders})

if got := adapter.buildCapabilitiesSummary(hwCaps); got != tt.wantSummary {
t.Errorf("buildCapabilitiesSummary() = %q, want %q", got, tt.wantSummary)
}
})
}

if got := NewRewriteAdapter().buildCapabilitiesSummary(nil); got != "unknown" {
t.Errorf("buildCapabilitiesSummary(nil) = %q, want %q", got, "unknown")
}
}

// TestRewriteChainLineStaysBoundedForFullFFmpegBuild is the regression for the
// full-screen stderr dump: a worker reporting a stock ffmpeg build's video
// encoders must stream a single terminal-friendly line to CLI clients, with
// every hardware encoder still visible — including the ones the static
// classifier does not know.
func TestRewriteChainLineStaysBoundedForFullFFmpegBuild(t *testing.T) {
encoders := append([]protocol.EncoderInfo{
{Name: "av1_amf", Type: "video", IsHW: true},
{Name: "mjpeg_qsv", Type: "video", IsHW: true},
}, testEncoderInfos(4, 125)...)

adapter := NewRewriteAdapter()
adapter.SetHardwareCapabilities(&protocol.WorkerCapabilities{
VideoEncoders: encoders,
GPUDevices: []protocol.GPUDeviceInfo{{Type: "nvenc", Vendor: "NVIDIA", Accessible: true}},
})

_, result, err := adapter.RewriteArgs(
context.Background(),
[]string{"-i", "input.mp4", "-c:v", "libx264", "output.mp4"},
true,
)
if err != nil {
t.Fatalf("unexpected error: %v", err)
}

line := audit.FormatRewriteChainLine(
result.CapabilitiesSummary, "libx264", result.TargetEncoder, result.DecisionReason, audit.InfoLevel)

if strings.Count(line, "\n") != 1 {
t.Errorf("rewrite chain must stay one line, got %q", line)
}
// 125 software encoders push the tail behind [... +N more]; hardware
// encoders the classifier does not recognise must still be on the line.
for _, want := range []string{"av1_amf[HW]", "mjpeg_qsv[HW]", "h264_nvenc[HW]", "[... +115 more]"} {
if !strings.Contains(line, want) {
t.Errorf("capabilities line missing %q, got %q", want, line)
}
}
// 6 hardware names, 10 software names and one "[... +N more]" tail; listing
// all 131 entries produced a line over 1 KB, i.e. several terminal screens.
if len(line) > 512 {
t.Errorf("rewrite chain line too long (%d bytes): %q", len(line), line)
}
}
Loading