From 8c7c7982c4b27557cb6bc33378cba6c09c5e8b5c Mon Sep 17 00:00:00 2001 From: Vulcan Date: Mon, 28 Sep 2026 17:45:52 +0800 Subject: [PATCH] =?UTF-8?q?feat(worker):=20Worker=20capabilities=20?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=8D=95=E8=A1=8C=E6=89=93=E5=8D=B0=20131=20?= =?UTF-8?q?=E7=BC=96=E7=A0=81=E5=99=A8=E6=B8=85=E5=8D=95=E8=BF=87=E9=95=BF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- pkg/audit/notifier.go | 50 ++++++++++- pkg/audit/notifier_test.go | 67 +++++++++++++++ pkg/audit/types.go | 8 +- pkg/worker/rewrite_adapter.go | 25 ++++-- pkg/worker/rewrite_adapter_test.go | 128 +++++++++++++++++++++++++++++ 5 files changed, 269 insertions(+), 9 deletions(-) diff --git a/pkg/audit/notifier.go b/pkg/audit/notifier.go index a2fb539..8098e18 100644 --- a/pkg/audit/notifier.go +++ b/pkg/audit/notifier.go @@ -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. 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)) } @@ -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)) @@ -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 { diff --git a/pkg/audit/notifier_test.go b/pkg/audit/notifier_test.go index 9516abe..b1a1feb 100644 --- a/pkg/audit/notifier_test.go +++ b/pkg/audit/notifier_test.go @@ -2,6 +2,7 @@ package audit import ( "bytes" + "fmt" "strings" "sync" "testing" @@ -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() diff --git a/pkg/audit/types.go b/pkg/audit/types.go index d694779..ed11663 100644 --- a/pkg/audit/types.go +++ b/pkg/audit/types.go @@ -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. diff --git a/pkg/worker/rewrite_adapter.go b/pkg/worker/rewrite_adapter.go index e2bf628..1f8df62 100644 --- a/pkg/worker/rewrite_adapter.go +++ b/pkg/worker/rewrite_adapter.go @@ -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" @@ -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" diff --git a/pkg/worker/rewrite_adapter_test.go b/pkg/worker/rewrite_adapter_test.go index e2e1d34..2617142 100644 --- a/pkg/worker/rewrite_adapter_test.go +++ b/pkg/worker/rewrite_adapter_test.go @@ -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() @@ -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) + } +}