From 0bb4173fb2cf9307aa3987f16dd37f798e688ff1 Mon Sep 17 00:00:00 2001 From: Samuel K Date: Sat, 25 Apr 2026 19:23:06 -0500 Subject: [PATCH] refactor(tunnel): migrate container and direct tunnels to PipeBridge Replace inlined pipe creation + goroutine coordination in container.go and direct.go with the PipeBridge and ClassifyTunnelErrors primitives from PR #137. - container.go Run(): use PipeBridge.RunPair for host tunnel + handler - container.go runInContainer(): use NewPipeBridge for pipe creation, keep manual coordination for the SSH-client-between-launch pattern - direct.go NewTunnel(): use PipeBridge.RunPair, fixing the race where ctx was passed instead of cancelCtx to the outer tunnel goroutine - Remove inlined awaitGoroutines, classifyTunnelErrors, isEOFError - Remove unused imports (errors, strings, io in container.go; os, fmt in direct.go) --- pkg/tunnel/container.go | 146 ++++++++++------------------------------ pkg/tunnel/direct.go | 58 +++++----------- 2 files changed, 51 insertions(+), 153 deletions(-) diff --git a/pkg/tunnel/container.go b/pkg/tunnel/container.go index cb11e3440..c9ebab42c 100644 --- a/pkg/tunnel/container.go +++ b/pkg/tunnel/container.go @@ -6,11 +6,9 @@ package tunnel import ( "bytes" "context" - "errors" "fmt" "io" "os" - "strings" "time" "github.com/devsy-org/devsy/pkg/agent" @@ -22,10 +20,14 @@ import ( "golang.org/x/crypto/ssh" ) +// ContainerTunnel manages the state of the tunnel to the container. +type ContainerTunnel struct { + client client.WorkspaceClient + updateConfigInterval time.Duration +} + // NewContainerTunnel constructs a ContainerTunnel using the workspace client, if proxy is True then // the workspace's agent config is not periodically updated. -// -//nolint:funcorder func NewContainerTunnel(client client.WorkspaceClient) *ContainerTunnel { updateConfigInterval := time.Second * 30 return &ContainerTunnel{ @@ -34,12 +36,6 @@ func NewContainerTunnel(client client.WorkspaceClient) *ContainerTunnel { } } -// ContainerTunnel manages the state of the tunnel to the container. -type ContainerTunnel struct { - client client.WorkspaceClient - updateConfigInterval time.Duration -} - // Handler defines what to do once the tunnel has a client established. type Handler func(ctx context.Context, containerClient *ssh.Client) error @@ -55,56 +51,37 @@ func (c *ContainerTunnel) Run( return nil } - cancelCtx, cancel := context.WithCancel(ctx) - defer cancel() + timeout := config.ParseTimeOption(cfg, config.ContextOptionAgentInjectTimeout) - stdoutReader, stdoutWriter, err := os.Pipe() + pb, err := NewPipeBridge() if err != nil { return err } - stdinReader, stdinWriter, err := os.Pipe() - if err != nil { - return err - } - defer func() { _ = stdoutWriter.Close() }() - defer func() { _ = stdinWriter.Close() }() - - timeout := config.ParseTimeOption(cfg, config.ContextOptionAgentInjectTimeout) - - // tunnel to host - tunnelChan := make(chan error, 1) - go func() { - tunnelChan <- c.runHostTunnel(cancelCtx, stdinReader, stdoutWriter, timeout) - }() + defer pb.Close() - // connect to container - containerChan := make(chan error, 1) - go func() { - sshClient, err := devssh.StdioClient(stdoutReader, stdinWriter, false) - if err != nil { - containerChan <- fmt.Errorf("create ssh client: %w", err) - return - } - - defer func() { _ = sshClient.Close() }() - defer cancel() - defer log.Debugf("connection to container closed") - log.Debugf("connected to host") - - if c.updateConfigInterval > 0 { - go func() { - c.updateConfig(cancelCtx, sshClient) - }() - } + return pb.RunPair(ctx, + func(ctx context.Context, stdin, stdout *os.File) error { + return c.runHostTunnel(ctx, stdin, stdout, timeout) + }, + func(ctx context.Context, stdout, stdin *os.File) error { + sshClient, err := devssh.StdioClient(stdout, stdin, false) + if err != nil { + return fmt.Errorf("create ssh client: %w", err) + } + defer func() { _ = sshClient.Close() }() + defer log.Debugf("connection to container closed") + log.Debugf("connected to host") - if err := c.runInContainer(cancelCtx, sshClient, handler, envVars); err != nil { - containerChan <- fmt.Errorf("run in container: %w", err) - } else { - containerChan <- nil - } - }() + if c.updateConfigInterval > 0 { + go c.updateConfig(ctx, sshClient) + } - return awaitGoroutines(tunnelChan, containerChan, stdoutWriter, stdinWriter) + if err := c.runInContainer(ctx, sshClient, handler, envVars); err != nil { + return fmt.Errorf("run in container: %w", err) + } + return nil + }, + ) } // runHostTunnel injects the devsy agent onto the host and starts the SSH server, @@ -143,54 +120,6 @@ func (c *ContainerTunnel) runHostTunnel( }) } -// awaitGoroutines waits for both the container and tunnel goroutines to report -// and returns the appropriate error. The container result is the primary result; -// the tunnel result provides root-cause context when the container gets EOF. -func awaitGoroutines( - tunnelChan, containerChan <-chan error, - stdoutWriter, stdinWriter *os.File, -) error { - var tunnelErr, containerErr error - - select { - case containerErr = <-containerChan: - select { - case tunnelErr = <-tunnelChan: - default: - } - case tunnelErr = <-tunnelChan: - // Host tunnel exited before container finished. Close pipes to unblock - // the container goroutine (may be blocked in SSH handshake). - _ = stdoutWriter.Close() - _ = stdinWriter.Close() - containerErr = <-containerChan - } - - return classifyTunnelErrors(tunnelErr, containerErr) -} - -// classifyTunnelErrors determines which error to report when the tunnel and/or -// container goroutines fail. EOF errors from the container are suppressed when -// the tunnel error is the root cause. -func classifyTunnelErrors(tunnelErr, containerErr error) error { - if containerErr == nil { - return nil - } - if isEOFError(containerErr) { - if tunnelErr != nil { - return fmt.Errorf("connect to server: %w", tunnelErr) - } - return nil - } - return fmt.Errorf("tunnel to container: %w", containerErr) -} - -// isEOFError reports whether an error is caused by an EOF condition, -// including wrapped SSH handshake failures from closed pipes. -func isEOFError(err error) bool { - return errors.Is(err, io.EOF) || strings.Contains(err.Error(), ": EOF") -} - // updateConfig is called periodically to keep the workspace agent config up to date. func (c *ContainerTunnel) updateConfig(ctx context.Context, sshClient *ssh.Client) { for { @@ -253,16 +182,11 @@ func (c *ContainerTunnel) runInContainer( return err } - stdoutReader, stdoutWriter, err := os.Pipe() - if err != nil { - return err - } - stdinReader, stdinWriter, err := os.Pipe() + pb, err := NewPipeBridge() if err != nil { return err } - defer func() { _ = stdoutWriter.Close() }() - defer func() { _ = stdinWriter.Close() }() + defer pb.Close() cancelCtx, cancel := context.WithCancel(ctx) defer cancel() @@ -273,13 +197,13 @@ func (c *ContainerTunnel) runInContainer( tunnelDone <- c.runContainerTunnel(cancelCtx, containerTunnelOpts{ sshClient: sshClient, workspaceInfo: workspaceInfo, - stdinReader: stdinReader, - stdoutWriter: stdoutWriter, + stdinReader: pb.StdinReader, + stdoutWriter: pb.StdoutWriter, envVars: envVars, }) }() - containerClient, err := devssh.StdioClient(stdoutReader, stdinWriter, false) + containerClient, err := devssh.StdioClient(pb.StdoutReader, pb.StdinWriter, false) if err != nil { // StdioClient failed — check if the tunnel goroutine already exited // with an error. If so, the tunnel error is the root cause. diff --git a/pkg/tunnel/direct.go b/pkg/tunnel/direct.go index cecb04961..9ad60546a 100644 --- a/pkg/tunnel/direct.go +++ b/pkg/tunnel/direct.go @@ -2,7 +2,6 @@ package tunnel import ( "context" - "fmt" "io" "os" @@ -18,48 +17,23 @@ type Tunnel func(ctx context.Context, stdin io.Reader, stdout io.Writer) error // the handler will be the function to execute the command using the // connected SSH client. func NewTunnel(ctx context.Context, tunnel Tunnel, handler Handler) error { - // create context - cancelCtx, cancel := context.WithCancel(ctx) - defer cancel() - - // create readers - stdoutReader, stdoutWriter, err := os.Pipe() + pb, err := NewPipeBridge() if err != nil { return err } - stdinReader, stdinWriter, err := os.Pipe() - if err != nil { - return err - } - defer func() { _ = stdoutWriter.Close() }() - defer func() { _ = stdinWriter.Close() }() - - // start ssh proxy - outerTunnelChan := make(chan error, 1) - go func() { - outerTunnelChan <- tunnel(ctx, stdinReader, stdoutWriter) - }() - - // start ssh client as root / default user - innerTunnelChan := make(chan error, 1) - go func() { - sshClient, err := devssh.StdioClient(stdoutReader, stdinWriter, false) - if err != nil { - innerTunnelChan <- err - return - } - defer func() { _ = sshClient.Close() }() - defer cancel() - - // start ssh tunnel - innerTunnelChan <- handler(cancelCtx, sshClient) - }() - - // wait for result - select { - case err := <-innerTunnelChan: - return fmt.Errorf("inner tunnel: %w", err) - case err := <-outerTunnelChan: - return fmt.Errorf("outer tunnel: %w", err) - } + defer pb.Close() + + return pb.RunPair(ctx, + func(ctx context.Context, stdin, stdout *os.File) error { + return tunnel(ctx, stdin, stdout) + }, + func(ctx context.Context, stdout, stdin *os.File) error { + sshClient, err := devssh.StdioClient(stdout, stdin, false) + if err != nil { + return err + } + defer func() { _ = sshClient.Close() }() + return handler(ctx, sshClient) + }, + ) }