fix(ssh): wait for reverse-forward listener before running setup-gpg

cmd/workspace/ssh.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"net"
 	"os"
 	"path"
 	"strings"
@@ -476,6 +477,91 @@ func (cmd *SSHCmd) reverseForwardPorts(
 	return <-errChan
 }
 
+type boundReverseForward struct {
+	portMapping string
+	mapping     port.Mapping
+	listener    net.Listener
+}
+
+// startReverseForwardsAndWait blocks until every forward's listener is bound,
+// unlike reverseForwardPorts which blocks for the forward's lifetime.
+func (cmd *SSHCmd) startReverseForwardsAndWait(
+	ctx context.Context,
+	containerClient *ssh.Client,
+) error {
+	timeout, err := cmd.forwardTimeout()
+	if err != nil {
+		return err
+	}
+
+	bound, err := bindReverseForwards(containerClient, cmd.ReverseForwardPorts)
+	if err != nil {
+		return err
+	}
+
+	for _, b := range bound {
+		log.Infof(
+			"Reverse forwarding local %s/%s to remote %s/%s",
+			b.mapping.Host.Protocol,
+			b.mapping.Host.Address,
+			b.mapping.Container.Protocol,
+			b.mapping.Container.Address,
+		)
+		go runReverseForwardInBackground(ctx, containerClient, b, timeout)
+	}
+
+	return nil
+}
+
+func bindReverseForwards(
+	containerClient *ssh.Client,
+	portMappings []string,
+) ([]boundReverseForward, error) {
+	var bound []boundReverseForward
+	closeBound := func() {
+		for _, b := range bound {
+			_ = b.listener.Close()
+		}
+	}
+	for _, portMapping := range portMappings {
+		mapping, err := port.ParsePortSpec(portMapping)
+		if err != nil {
+			closeBound()
+			return nil, fmt.Errorf("parse port mapping: %w", err)
+		}
+
+		listener, err := devssh.ReverseListen(
+			containerClient,
+			mapping.Host.Protocol,
+			mapping.Host.Address,
+		)
+		if err != nil {
+			closeBound()
+			return nil, fmt.Errorf("listen for reverse forward %s: %w", portMapping, err)
+		}
+		bound = append(bound, boundReverseForward{portMapping, mapping, listener})
+	}
+	return bound, nil
+}
+
+func runReverseForwardInBackground(
+	ctx context.Context,
+	containerClient *ssh.Client,
+	b boundReverseForward,
+	timeout time.Duration,
+) {
+	err := devssh.RunReverseForward(ctx, containerClient, devssh.ReverseForwardOpts{
+		Listener:         b.listener,
+		RemoteAddr:       b.mapping.Host.Address,
+		LocalNetwork:     b.mapping.Container.Protocol,
+		LocalAddr:        b.mapping.Container.Address,
+		ExitAfterTimeout: timeout,
+	})
+	if err != nil && !errors.Is(err, devssh.ErrIdleTimeout) && !errors.Is(err, io.EOF) {
+		log.Errorf("error forwarding %s: %v", b.portMapping, err)
+	}
+}
+
 func (cmd *SSHCmd) forwardPorts(
 	ctx context.Context,
 	containerClient *ssh.Client,
@@ -782,9 +868,9 @@ func (cmd *SSHCmd) setupGPGAgent(
 		gpgExtraSocketPath,
 	)
 
-	go func() {
-		log.Error(cmd.reverseForwardPorts(ctx, containerClient))
-	}()
+	if err := cmd.startReverseForwardsAndWait(ctx, containerClient); err != nil {
+		return fmt.Errorf("start gpg-agent reverse forward: %w", err)
+	}
 
 	writer := log.Writer(log.LevelInfo)
 	defer func() { _ = writer.Close() }()

pkg/ssh/forward.go

@@ -79,16 +79,41 @@ func ReversePortForward(
 	remoteNetwork, remoteAddr, localNetwork, localAddr string,
 	exitAfterTimeout time.Duration,
 ) error {
-	listener, err := client.Listen(remoteNetwork, remoteAddr)
+	listener, err := ReverseListen(client, remoteNetwork, remoteAddr)
 	if err != nil {
 		return err
 	}
-	defer func() { _ = listener.Close() }()
+	return RunReverseForward(ctx, client, ReverseForwardOpts{
+		Listener:         listener,
+		RemoteAddr:       remoteAddr,
+		LocalNetwork:     localNetwork,
+		LocalAddr:        localAddr,
+		ExitAfterTimeout: exitAfterTimeout,
+	})
+}
 
+// ReverseListen binds the remote listener; pair with RunReverseForward.
+func ReverseListen(client *ssh.Client, remoteNetwork, remoteAddr string) (net.Listener, error) {
+	return client.Listen(remoteNetwork, remoteAddr)
+}
+
+// ReverseForwardOpts groups the parameters for RunReverseForward.
+type ReverseForwardOpts struct {
+	Listener         net.Listener
+	RemoteAddr       string
+	LocalNetwork     string
+	LocalAddr        string
+	ExitAfterTimeout time.Duration
+}
+
+// RunReverseForward runs the forwarding loop for a listener obtained via
+// ReverseListen, closing it on return.
+func RunReverseForward(ctx context.Context, client *ssh.Client, opts ReverseForwardOpts) error {
+	defer func() { _ = opts.Listener.Close() }()
 	return portForwarding(
-		ctx, client, listener,
-		remoteAddr, localNetwork, localAddr,
-		exitAfterTimeout, reverseForward,
+		ctx, client, opts.Listener,
+		opts.RemoteAddr, opts.LocalNetwork, opts.LocalAddr,
+		opts.ExitAfterTimeout, reverseForward,
 	)
 }