fix: host access port instability

docker.go

@@ -1185,6 +1185,18 @@ func (p *DockerProvider) CreateContainer(ctx context.Context, req ContainerReque
 			return nil, fmt.Errorf("expose host ports: %w", err)
 		}
 
+		defer func() {
+			if err != nil && con == nil {
+				// Container setup failed so ensure we clean up the sshd container too.
+				ctr := &DockerContainer{
+					provider:       p,
+					logger:         p.Logger,
+					lifecycleHooks: []ContainerLifecycleHooks{sshdForwardPortsHook},
+				}
+				err = errors.Join(ctr.terminatingHook(ctx))
+			}
+		}()
+
 		defaultHooks = append(defaultHooks, sshdForwardPortsHook)
 	}
 
@@ -1623,17 +1635,22 @@ func (p *DockerProvider) ensureDefaultNetwork(ctx context.Context) (string, erro
 		return "", fmt.Errorf("network list: %w", err)
 	}
 
+	// Prefer the default bridge network if it exists.
+	// This makes the results stable as network list order is not guaranteed.
 	for _, net := range networkResources {
 		switch net.Name {
 		case p.defaultBridgeNetworkName:
 			p.defaultNetwork = p.defaultBridgeNetworkName
 			return p.defaultNetwork, nil
 		case ReaperDefault:
 			p.defaultNetwork = ReaperDefault
-			return p.defaultNetwork, nil
 		}
 	}
 
+	if p.defaultNetwork != "" {
+		return p.defaultNetwork, nil
+	}
+
 	// Create a bridge network for the container communications.
 	_, err = p.client.NetworkCreate(ctx, ReaperDefault, network.CreateOptions{
 		Driver:     Bridge,

exec/processor.go

@@ -2,7 +2,9 @@ package exec
 
 import (
 	"bytes"
+	"fmt"
 	"io"
+	"sync"
 
 	"github.com/docker/docker/api/types/container"
 	"github.com/docker/docker/pkg/stdcopy"
@@ -60,6 +62,43 @@ func WithEnv(env []string) ProcessOption {
 	})
 }
 
+// safeBuffer is a goroutine safe buffer.
+type safeBuffer struct {
+	mtx sync.Mutex
+	buf bytes.Buffer
+	err error
+}
+
+// Error sets an error for the next read.
+func (sb *safeBuffer) Error(err error) {
+	sb.mtx.Lock()
+	defer sb.mtx.Unlock()
+
+	sb.err = err
+}
+
+// Write writes p to the buffer.
+// It is safe for concurrent use by multiple goroutines.
+func (sb *safeBuffer) Write(p []byte) (n int, err error) {
+	sb.mtx.Lock()
+	defer sb.mtx.Unlock()
+
+	return sb.buf.Write(p)
+}
+
+// Read reads up to len(p) bytes into p from the buffer.
+// It is safe for concurrent use by multiple goroutines.
+func (sb *safeBuffer) Read(p []byte) (n int, err error) {
+	sb.mtx.Lock()
+	defer sb.mtx.Unlock()
+
+	if sb.err != nil {
+		return 0, sb.err
+	}
+
+	return sb.buf.Read(p)
+}
+
 // Multiplexed returns a [ProcessOption] that configures the command execution
 // to combine stdout and stderr into a single stream without Docker's multiplexing headers.
 func Multiplexed() ProcessOption {
@@ -73,13 +112,14 @@ func Multiplexed() ProcessOption {
 
 		done := make(chan struct{})
 
-		var outBuff bytes.Buffer
-		var errBuff bytes.Buffer
+		var outBuff safeBuffer
+		var errBuff safeBuffer
 		go func() {
+			defer close(done)
 			if _, err := stdcopy.StdCopy(&outBuff, &errBuff, opts.Reader); err != nil {
+				outBuff.Error(fmt.Errorf("copying output: %w", err))
 				return
 			}
-			close(done)
 		}()
 
 		<-done

port_forwarding.go

@@ -99,10 +99,25 @@ func exposeHostPorts(ctx context.Context, req *ContainerRequest, ports ...int) (
 		return sshdConnectHook, fmt.Errorf("new sshd container: %w", err)
 	}
 
-	// IP in the first network of the container
-	sshdIP, err := sshdContainer.ContainerIP(context.Background())
+	// IP in the first network of the container.
+	inspect, err := sshdContainer.Inspect(ctx)
 	if err != nil {
-		return sshdConnectHook, fmt.Errorf("get sshd container IP: %w", err)
+		return sshdConnectHook, fmt.Errorf("inspect sshd container: %w", err)
+	}
+
+	sshdIP := inspect.NetworkSettings.IPAddress
+	if sshdIP == "" {
+		single := len(inspect.NetworkSettings.Networks) == 1
+		for name, network := range inspect.NetworkSettings.Networks {
+			if name == sshdFirstNetwork || single {
+				sshdIP = network.IPAddress
+				break
+			}
+		}
+	}
+
+	if sshdIP == "" {
+		return sshdConnectHook, errors.New("sshd container IP not found")
 	}
 
 	if req.HostConfigModifier == nil {
@@ -166,11 +181,10 @@ func exposeHostPorts(ctx context.Context, req *ContainerRequest, ports ...int) (
 func newSshdContainer(ctx context.Context, opts ...ContainerCustomizer) (*sshdContainer, error) {
 	req := GenericContainerRequest{
 		ContainerRequest: ContainerRequest{
-			Image:           sshdImage,
-			HostAccessPorts: []int{}, // empty list because it does not need any port
-			ExposedPorts:    []string{sshPort},
-			Env:             map[string]string{"PASSWORD": sshPassword},
-			WaitingFor:      wait.ForListeningPort(sshPort),
+			Image:        sshdImage,
+			ExposedPorts: []string{sshPort},
+			Env:          map[string]string{"PASSWORD": sshPassword},
+			WaitingFor:   wait.ForListeningPort(sshPort),
 		},
 		Started: true,
 	}

port_forwarding_test.go

@@ -22,101 +22,95 @@ const (
 )
 
 func TestExposeHostPorts(t *testing.T) {
-	tests := []struct {
-		name          string
-		numberOfPorts int
-		hasNetwork    bool
-		hasHostAccess bool
-	}{
-		{
-			name:          "single port",
-			numberOfPorts: 1,
-			hasHostAccess: true,
-		},
-		{
-			name:          "single port using a network",
-			numberOfPorts: 1,
-			hasNetwork:    true,
-			hasHostAccess: true,
-		},
-		{
-			name:          "multiple ports",
-			numberOfPorts: 3,
-			hasHostAccess: true,
-		},
-		{
-			name:          "single port with cancellation",
-			numberOfPorts: 1,
-			hasHostAccess: false,
+	t.Run("single-port", func(t *testing.T) {
+		testExposeHostPorts(t, 1, false, true)
+	})
+
+	t.Run("single-port-using-network", func(t *testing.T) {
+		testExposeHostPorts(t, 1, true, true)
+	})
+
+	t.Run("single-port-cancellation", func(t *testing.T) {
+		testExposeHostPorts(t, 1, false, false)
+	})
+
+	t.Run("multi-port", func(t *testing.T) {
+		testExposeHostPorts(t, 3, false, true)
+	})
+
+	t.Run("multi-port-using-network", func(t *testing.T) {
+		testExposeHostPorts(t, 3, false, true)
+	})
+}
+
+func testExposeHostPorts(t *testing.T, numberOfPorts int, hasNetwork, hasHostAccess bool) {
+	t.Helper()
+
+	freePorts := make([]int, numberOfPorts)
+	for i := range freePorts {
+		server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			fmt.Fprint(w, expectedResponse)
+		}))
+		freePorts[i] = server.Listener.Addr().(*net.TCPAddr).Port
+		t.Cleanup(func() {
+			server.Close()
+		})
+	}
+
+	req := testcontainers.GenericContainerRequest{
+		// hostAccessPorts {
+		ContainerRequest: testcontainers.ContainerRequest{
+			Image:           "alpine:3.17",
+			HostAccessPorts: freePorts,
+			Cmd:             []string{"top"},
 		},
+		// }
+		Started: true,
 	}
 
-	for _, tc := range tests {
-		t.Run(tc.name, func(tt *testing.T) {
-			freePorts := make([]int, tc.numberOfPorts)
-			for i := range freePorts {
-				server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
-					fmt.Fprint(w, expectedResponse)
-				}))
-				freePorts[i] = server.Listener.Addr().(*net.TCPAddr).Port
-				tt.Cleanup(func() {
-					server.Close()
-				})
-			}
-
-			req := testcontainers.GenericContainerRequest{
-				// hostAccessPorts {
-				ContainerRequest: testcontainers.ContainerRequest{
-					Image:           "alpine:3.17",
-					HostAccessPorts: freePorts,
-					Cmd:             []string{"top"},
-				},
-				// }
-				Started: true,
-			}
-
-			var nw *testcontainers.DockerNetwork
-			if tc.hasNetwork {
-				var err error
-				nw, err = network.New(context.Background())
-				require.NoError(tt, err)
-				testcontainers.CleanupNetwork(t, nw)
-
-				req.Networks = []string{nw.Name}
-				req.NetworkAliases = map[string][]string{nw.Name: {"myalpine"}}
-			}
-
-			ctx := context.Background()
-			if !tc.hasHostAccess {
-				var cancel context.CancelFunc
-				ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
-				defer cancel()
-			}
-
-			c, err := testcontainers.GenericContainer(ctx, req)
-			testcontainers.CleanupContainer(t, c)
-			require.NoError(tt, err)
-
-			if tc.hasHostAccess {
-				// create a container that has host access, which will
-				// automatically forward the port to the container
-				assertContainerHasHostAccess(tt, c, freePorts...)
-			} else {
-				// force cancellation because of timeout
-				time.Sleep(11 * time.Second)
-
-				assertContainerHasNoHostAccess(tt, c, freePorts...)
-			}
-		})
+	var nw *testcontainers.DockerNetwork
+	if hasNetwork {
+		var err error
+		nw, err = network.New(context.Background())
+		require.NoError(t, err)
+		testcontainers.CleanupNetwork(t, nw)
+
+		req.Networks = []string{nw.Name}
+		req.NetworkAliases = map[string][]string{nw.Name: {"myalpine"}}
+	}
+
+	ctx := context.Background()
+	if !hasHostAccess {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, 10*time.Second)
+		defer cancel()
+	}
+
+	c, err := testcontainers.GenericContainer(ctx, req)
+	testcontainers.CleanupContainer(t, c)
+	require.NoError(t, err)
+
+	if hasHostAccess {
+		// create a container that has host access, which will
+		// automatically forward the port to the container
+		assertContainerHasHostAccess(t, c, freePorts...)
+	} else {
+		// force cancellation because of timeout
+		time.Sleep(11 * time.Second)
+
+		assertContainerHasNoHostAccess(t, c, freePorts...)
 	}
 }
 
 func httpRequest(t *testing.T, c testcontainers.Container, port int) (int, string) {
 	t.Helper()
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
 	// wgetHostInternal {
 	code, reader, err := c.Exec(
-		context.Background(),
-		[]string{"wget", "-q", "-O", "-", fmt.Sprintf("http://%s:%d", testcontainers.HostInternal, port)},
+		ctx,
+		[]string{"wget", "-q", "-O", "-", "-T", "2", fmt.Sprintf("http://%s:%d", testcontainers.HostInternal, port)},
 		tcexec.Multiplexed(),
 	)
 	// }