Fix Wayland VA-API capture of DMA-BUF frames

internal/airplay/capture.go

@@ -804,6 +804,10 @@ func buildGstVideoPipeline(source gstStage, beforeConvert, afterScale []gstStage
 	if encoder.needsVulkan {
 		args = appendGstStage(args, gstStage{"vulkanupload"})
 	}
+	return appendGstVideoEncoding(args, encoder, timestampedOutput)
+}
+
+func appendGstVideoEncoding(args []string, encoder encoderResult, timestampedOutput bool) []string {
 	args = appendGstStage(args, encoder.parts)
 	parser, mediaType, payloader := "h264parse", "video/x-h264", "rtph264pay"
 	if encoder.codec == VideoCodecHEVC {
@@ -827,6 +831,30 @@ func buildGstVideoPipeline(source gstStage, beforeConvert, afterScale []gstStage
 	return appendGstStage(args, gstStage{"fdsink", "fd=1", "sync=false", "async=false"})
 }
 
+// buildVAWaylandVideoPipeline imports portal buffers without a CPU copy, then
+// keeps conversion, scaling, and encoding in VA memory. PipeWire keepalives
+// supply idle frames; videorate limits them to the requested output rate.
+func buildVAWaylandVideoPipeline(fd int, nodeID uint32, fps int, encoder encoderResult, maxWidth, maxHeight int, timestampedOutput bool) []string {
+	source := gstStage{"pipewiresrc", fmt.Sprintf("fd=%d", fd), fmt.Sprintf("path=%d", nodeID),
+		"do-timestamp=true", fmt.Sprintf("keepalive-time=%d", frameIntervalMillis(fps)), "always-copy=false"}
+	args := append([]string{"--quiet"}, source...)
+	// Desktop pixels are square. Without this constraint, the VA transform can
+	// negotiate the minimum of its PAR range and fail to calculate borders.
+	args = appendGstStage(args, gstStage{"video/x-raw(ANY),pixel-aspect-ratio=1/1"})
+	// Force a fresh surface even when the source already matches the output.
+	// Downstream retains the converted surface instead of another portal buffer.
+	args = appendGstStage(args, gstStage{"vapostproc", "disable-passthrough=true", "add-borders=true"})
+	caps := "video/x-raw(memory:VAMemory),format=NV12"
+	if maxWidth > 1 && maxHeight > 1 {
+		caps += fmt.Sprintf(",width=%d,height=%d,pixel-aspect-ratio=1/1", maxWidth&^1, maxHeight&^1)
+	}
+	args = appendGstStage(args, gstStage{caps})
+	args = appendGstStage(args, gstStage{"videorate", "drop-only=true", "skip-to-first=true"})
+	args = appendGstStage(args, gstStage{caps + fmt.Sprintf(",framerate=%d/1", fps)})
+	args = appendGstStage(args, lowLatencyVideoQueueStage())
+	return appendGstVideoEncoding(args, encoder, timestampedOutput)
+}
+
 func startPreparedWaylandCapture(ctx context.Context, cfg CaptureConfig, encoderParts encoderResult, nodeID uint32, pwFd *os.File, dbusConn *dbus.Conn, streamSize [2]int, timestampedOutput bool) (*ScreenCapture, error) {
 	if pwFd == nil || dbusConn == nil {
 		if pwFd != nil {
@@ -853,36 +881,41 @@ func startPreparedWaylandCapture(ctx context.Context, cfg CaptureConfig, encoder
 	// The encoded dimensions are capped to the receiver's advertised display size
 	// when available. The actual result is read back from the codec SPS downstream.
 	const pwFdNum = 3
-	source := pipeWireVideoSourceStage(pwFdNum, nodeID, fps)
+	var gstArgs []string
+	if len(encoderParts.parts) > 0 && encoderParts.parts[0] == "vah264enc" && hasGstElement("vapostproc") {
+		gstArgs = buildVAWaylandVideoPipeline(pwFdNum, nodeID, fps, encoderParts, cfg.MaxWidth, cfg.MaxHeight, timestampedOutput)
+	} else {
+		source := pipeWireVideoSourceStage(pwFdNum, nodeID, fps)
 
-	hasCompositor := streamSize[0] > 0 && streamSize[1] > 0 && hasGstElement("compositor")
+		hasCompositor := streamSize[0] > 0 && streamSize[1] > 0 && hasGstElement("compositor")
 
-	var beforeConvert []gstStage
-	if hasGstElement("vapostproc") {
-		beforeConvert = append(beforeConvert, gstStage{"vapostproc"})
-	} else {
-		log.Printf("[CAPTURE] vapostproc unavailable, using software conversion")
-	}
+		var beforeConvert []gstStage
+		if hasGstElement("vapostproc") {
+			beforeConvert = append(beforeConvert, gstStage{"vapostproc"})
+		} else {
+			log.Printf("[CAPTURE] vapostproc unavailable, using software conversion")
+		}
 
-	var afterScale []gstStage
-	if hasCompositor {
-		beforeConvert = append(beforeConvert,
-			gstStage{"compositor", "force-live=true", "ignore-inactive-pads=true", "background=black"},
-			gstStage{fmt.Sprintf("video/x-raw,width=%d,height=%d,framerate=%d/1", streamSize[0], streamSize[1], fps)},
-		)
-	} else {
-		log.Printf("[CAPTURE] idle-frame compositor unavailable; using portal frame timing")
-	}
-	if hasCompositor {
-		afterScale = append(afterScale, lowLatencyVideoQueueStage())
-	} else {
-		afterScale = append(afterScale,
-			gstStage{"videorate", "drop-only=true", "skip-to-first=true"},
-			frameRateStage(fps),
-			lowLatencyVideoQueueStage(),
-		)
+		var afterScale []gstStage
+		if hasCompositor {
+			beforeConvert = append(beforeConvert,
+				gstStage{"compositor", "force-live=true", "ignore-inactive-pads=true", "background=black"},
+				gstStage{fmt.Sprintf("video/x-raw,width=%d,height=%d,framerate=%d/1", streamSize[0], streamSize[1], fps)},
+			)
+		} else {
+			log.Printf("[CAPTURE] idle-frame compositor unavailable; using portal frame timing")
+		}
+		if hasCompositor {
+			afterScale = append(afterScale, lowLatencyVideoQueueStage())
+		} else {
+			afterScale = append(afterScale,
+				gstStage{"videorate", "drop-only=true", "skip-to-first=true"},
+				frameRateStage(fps),
+				lowLatencyVideoQueueStage(),
+			)
+		}
+		gstArgs = buildGstVideoPipeline(source, beforeConvert, afterScale, encoderParts, cfg.MaxWidth, cfg.MaxHeight, timestampedOutput)
 	}
-	gstArgs := buildGstVideoPipeline(source, beforeConvert, afterScale, encoderParts, cfg.MaxWidth, cfg.MaxHeight, timestampedOutput)
 
 	dbg("[CAPTURE] gst-launch-1.0 (wayland) %s", strings.Join(gstArgs, " "))
 	cmd := exec.CommandContext(captureCtx, "gst-launch-1.0", gstArgs...)

internal/airplay/capture_test.go

@@ -3,6 +3,7 @@ package airplay
 import (
 	"context"
 	"errors"
+	"fmt"
 	"os"
 	"os/exec"
 	"reflect"
@@ -810,3 +811,35 @@ func TestDetectGstEncoderSelectionContract(t *testing.T) {
 		})
 	}
 }
+
+func TestVAWaylandPipelineKeepsFramesInVAMemory(t *testing.T) {
+	encoder := encoderResult{parts: gstStage{"vah264enc"}, rawFormat: "NV12", codec: VideoCodecH264}
+	for _, size := range [][2]int{{0, 0}, {1920, 1080}, {1279, 719}, {1, 1}} {
+		pipeline := buildVAWaylandVideoPipeline(3, 42, 30, encoder, size[0], size[1], true)
+		joined := strings.Join(pipeline, " ")
+		for _, forbidden := range []string{"always-copy=true", "videoconvert", "videoscale", "compositor"} {
+			if strings.Contains(joined, forbidden) {
+				t.Errorf("VA pipeline must not copy or process portal frames on the CPU: %s", joined)
+			}
+		}
+		for _, required := range []string{"keepalive-time=33", "always-copy=false", "disable-passthrough=true", "add-borders=true", "video/x-raw(ANY),pixel-aspect-ratio=1/1", "video/x-raw(memory:VAMemory),format=NV12"} {
+			if !strings.Contains(joined, required) {
+				t.Errorf("VA pipeline is missing %q: %s", required, joined)
+			}
+		}
+		if size[0] > 1 && size[1] > 1 {
+			want := fmt.Sprintf("width=%d,height=%d,pixel-aspect-ratio=1/1", size[0]&^1, size[1]&^1)
+			if !strings.Contains(joined, want) {
+				t.Errorf("VA scaling must use an even receiver canvas: %s", joined)
+			}
+		} else if strings.Contains(joined, "width=") || strings.Contains(joined, "height=") {
+			t.Errorf("invalid receiver size must not constrain the capture: %s", joined)
+		}
+		// The VA path must retain the same timestamp-preserving output as other
+		// sources, since the sender schedules video from the encoded buffer PTS.
+		suffix := appendGstVideoEncoding(nil, encoder, true)
+		if !reflect.DeepEqual(pipeline[len(pipeline)-len(suffix):], suffix) {
+			t.Errorf("VA pipeline changed the shared encoding/output suffix: %s", joined)
+		}
+	}
+}