Skip to content

refactor(tunnel): migrate container and direct tunnels to PipeBridge #138

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
skevetter merged 1 commit into from
Apr 26, 2026
Merged

refactor(tunnel): migrate container and direct tunnels to PipeBridge #138

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 35 additions & 111 deletions pkg/tunnel/container.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,11 +6,9 @@ package tunnel
import (
"bytes"
"context"
"errors"
"fmt"
"io"
"os"
"strings"
"time"

"github.com/devsy-org/devsy/pkg/agent"
Expand All @@ -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{
Expand All @@ -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

Expand All @@ -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,
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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()
Expand All @@ -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.
Expand Down
58 changes: 16 additions & 42 deletions pkg/tunnel/direct.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,6 @@ package tunnel

import (
"context"
"fmt"
"io"
"os"

Expand All @@ -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)
},
)
Comment on lines +26 to +38

@coderabbitai coderabbitai Bot Apr 26, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Confirm ClassifyTunnelErrors message and all callers to assess blast radius.
rg -nP -C2 'ClassifyTunnelErrors|tunnel to container|connect to server'
rg -nP -C3 'NewTunnel\s*\(' --type=go

Repository: devsy-org/devsy

Length of output: 5884

ClassifyTunnelErrors hardcodes "tunnel to container" prefix for all tunnel flows — misleading for direct SSH and browser tunnels.

pb.RunPair routes all non-EOF handler errors through ClassifyTunnelErrors in connerror.go (line 49), which unconditionally wraps with "tunnel to container: %w". However, NewTunnel in direct.go is used for direct SSH proxy tunnels (cmd/ssh.go:308) and browser tunnels (pkg/tunnel/browser.go:52) — not container tunnels. This prefix is misleading for these call sites and creates inconsistent user-facing error messages (visible in stderr and e2e tests confirm this wrapping occurs).

Consider:

  • Parameterizing ClassifyTunnelErrors with a tunnel type label (e.g., "direct tunnel" vs. "tunnel to container"), or
  • Genericizing the message to work for both container and non-container flows, or
  • Post-processing errors in direct.go before propagation.
🤖 Prompt for AI Agents 
Verify each finding against the current code and only fix it if needed.

In `@pkg/tunnel/direct.go` around lines 26 - 38, The error wrapper in
ClassifyTunnelErrors is hardcoded to "tunnel to container" which mislabels
errors from NewTunnel used by direct.go (direct SSH/browser flows) via
pb.RunPair; change ClassifyTunnelErrors to accept a tunnelType string (or a
generic message) and update its callers to pass an appropriate label (e.g.,
"direct tunnel" for NewTunnel in direct.go and "tunnel to container" where
container flows are used), or alternatively adjust direct.go to detect and
replace the container prefix before returning; update the connerror.go function
signature for ClassifyTunnelErrors and all call sites (including where
pb.RunPair routes errors) to use the new parameter so user-facing error messages
are accurate.

}
Loading