Close split capture pipes on startup failure

Ultraworked with [omo](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: sisyphus-dev-ai <sisyphus-dev-ai@users.noreply.github.com>

Author: minpeter

internal/airplay/capture.go

@@ -907,6 +907,76 @@ func buildSplitGstVideoPipeline(source gstStage, beforeConvert, afterScale []gst
 	return producer, consumer
 }
 
+type waylandSplitPipes struct {
+	rawFrames     *os.File
+	sourceOutput  *os.File
+	sourceStderr  *os.File
+	sourceErrOut  *os.File
+	stdout        *os.File
+	encoderOutput *os.File
+	encoderStderr *os.File
+	encoderErrOut *os.File
+}
+
+func openWaylandSplitPipes(sourceCmd, encoderCmd *exec.Cmd) (*waylandSplitPipes, error) {
+	pipes := &waylandSplitPipes{}
+	var err error
+	pipes.rawFrames, pipes.sourceOutput, err = os.Pipe()
+	if err != nil {
+		return nil, fmt.Errorf("capture serialization pipe: %w", err)
+	}
+	sourceCmd.Stdout = pipes.sourceOutput
+	pipes.sourceStderr, pipes.sourceErrOut, err = os.Pipe()
+	if err != nil {
+		pipes.close()
+		return nil, fmt.Errorf("capture stderr pipe: %w", err)
+	}
+	sourceCmd.Stderr = pipes.sourceErrOut
+	encoderCmd.Stdin = pipes.rawFrames
+	pipes.stdout, pipes.encoderOutput, err = os.Pipe()
+	if err != nil {
+		pipes.close()
+		return nil, fmt.Errorf("encoder stdout pipe: %w", err)
+	}
+	encoderCmd.Stdout = pipes.encoderOutput
+	pipes.encoderStderr, pipes.encoderErrOut, err = os.Pipe()
+	if err != nil {
+		pipes.close()
+		return nil, fmt.Errorf("encoder stderr pipe: %w", err)
+	}
+	encoderCmd.Stderr = pipes.encoderErrOut
+	return pipes, nil
+}
+
+func (pipes *waylandSplitPipes) close() {
+	if pipes == nil {
+		return
+	}
+	for _, closer := range []io.Closer{
+		pipes.rawFrames,
+		pipes.sourceOutput,
+		pipes.sourceStderr,
+		pipes.sourceErrOut,
+		pipes.stdout,
+		pipes.encoderOutput,
+		pipes.encoderStderr,
+		pipes.encoderErrOut,
+	} {
+		if closer != nil {
+			_ = closer.Close()
+		}
+	}
+}
+
+func startWaylandEncoder(cmd *exec.Cmd, pipes *waylandSplitPipes) (<-chan error, error) {
+	wait, err := startGStreamerCommand(cmd)
+	if err != nil {
+		pipes.close()
+		return nil, err
+	}
+	return wait, nil
+}
+
 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 {
@@ -977,60 +1047,58 @@ func startPreparedWaylandCapture(ctx context.Context, cfg CaptureConfig, encoder
 
 	sourceCmd := exec.CommandContext(captureCtx, "gst-launch-1.0", sourceArgs...)
 	sourceCmd.ExtraFiles = []*os.File{pwFd}
-	rawFrames, err := sourceCmd.StdoutPipe()
-	if err != nil {
-		cancel()
-		_ = pwFd.Close()
-		_ = dbusConn.Close()
-		return nil, fmt.Errorf("capture serialization pipe: %w", err)
-	}
-	sourceStderr, _ := sourceCmd.StderrPipe()
-
 	cmd := exec.CommandContext(captureCtx, "gst-launch-1.0", encoderArgs...)
-	cmd.Stdin = rawFrames
-	stdout, err := cmd.StdoutPipe()
+	pipes, err := openWaylandSplitPipes(sourceCmd, cmd)
 	if err != nil {
 		cancel()
-		_ = rawFrames.Close()
 		_ = pwFd.Close()
 		_ = dbusConn.Close()
-		return nil, fmt.Errorf("encoder stdout pipe: %w", err)
+		return nil, err
 	}
-	encoderStderr, _ := cmd.StderrPipe()
 
-	encoderWait, err := startGStreamerCommand(cmd)
+	encoderWait, err := startWaylandEncoder(cmd, pipes)
 	if err != nil {
 		cancel()
-		_ = rawFrames.Close()
 		_ = pwFd.Close()
 		_ = dbusConn.Close()
 		return nil, fmt.Errorf("start encoder gst-launch: %w", err)
 	}
+	_ = pipes.rawFrames.Close()
+	_ = pipes.encoderOutput.Close()
+	_ = pipes.encoderErrOut.Close()
 	sourceWait, err := startGStreamerCommand(sourceCmd)
 	if err != nil {
 		cancel()
-		_ = rawFrames.Close()
+		pipes.close()
 		_ = pwFd.Close()
 		_ = dbusConn.Close()
 		<-encoderWait
 		return nil, fmt.Errorf("start capture gst-launch: %w", err)
 	}
+	_ = pipes.sourceOutput.Close()
+	_ = pipes.sourceErrOut.Close()
 	_ = pwFd.Close() // source child inherited it
 
-	go logStderr("GST-SOURCE", sourceStderr)
-	go logStderr("GST-ENCODER", encoderStderr)
+	go func() {
+		defer pipes.sourceStderr.Close()
+		logStderr("GST-SOURCE", pipes.sourceStderr)
+	}()
+	go func() {
+		defer pipes.encoderStderr.Close()
+		logStderr("GST-ENCODER", pipes.encoderStderr)
+	}()
 
 	capture := &ScreenCapture{
 		cmd:       cmd,
 		sourceCmd: sourceCmd,
-		stdout:    stdout,
+		stdout:    pipes.stdout,
 		cancel:    cancel,
 		pwNodeID:  nodeID,
 		dbusConn:  dbusConn,
 		waitCh:    make(chan struct{}),
 	}
 	if timestampedOutput {
-		capture.frames = newRTPVideoAccessUnitReader(stdout, encoderParts.codec)
+		capture.frames = newRTPVideoAccessUnitReader(pipes.stdout, encoderParts.codec)
 	}
 	go func() {
 		type processResult struct {

internal/airplay/capture_test.go

@@ -14,6 +14,52 @@ import (
 	"github.com/godbus/dbus/v5"
 )
 
+func TestWaylandEncoderStartFailureClosesEveryPipeDescriptor(t *testing.T) {
+	before := openDescriptorCount(t)
+	for i := 0; i < 20; i++ {
+		source := exec.Command("true")
+		encoder := exec.Command("/definitely/not/a/doubletake-test-command")
+		pipes, err := openWaylandSplitPipes(source, encoder)
+		if err != nil {
+			t.Fatalf("openWaylandSplitPipes: %v", err)
+		}
+		if _, err := startWaylandEncoder(encoder, pipes); err == nil {
+			t.Fatal("startWaylandEncoder unexpectedly succeeded")
+		}
+	}
+	after := openDescriptorCount(t)
+	if after > before {
+		t.Fatalf("stderr setup failures leaked descriptors: before=%d after=%d", before, after)
+	}
+}
+
+func TestOpenWaylandSplitPipesCloseReleasesEveryDescriptor(t *testing.T) {
+	before := openDescriptorCount(t)
+	source := exec.Command("true")
+	encoder := exec.Command("true")
+	pipes, err := openWaylandSplitPipes(source, encoder)
+	if err != nil {
+		t.Fatalf("openWaylandSplitPipes: %v", err)
+	}
+	pipes.close()
+	after := openDescriptorCount(t)
+	if after > before {
+		t.Fatalf("pipe close leaked descriptors: before=%d after=%d", before, after)
+	}
+}
+
+func openDescriptorCount(t *testing.T) int {
+	t.Helper()
+	entries, err := os.ReadDir("/proc/self/fd")
+	if errors.Is(err, os.ErrNotExist) {
+		t.Skip("/proc/self/fd is unavailable")
+	}
+	if err != nil {
+		t.Fatalf("read /proc/self/fd: %v", err)
+	}
+	return len(entries)
+}
+
 func TestCapturePreparationCloseReleasesUnstartedResources(t *testing.T) {
 	portalFD, peerFD, err := os.Pipe()
 	if err != nil {