diff --git a/packages/shared/pkg/proxy/grpc_proxy_test.go b/packages/shared/pkg/proxy/grpc_proxy_test.go new file mode 100644 index 0000000000..9502fdb0c8 --- /dev/null +++ b/packages/shared/pkg/proxy/grpc_proxy_test.go @@ -0,0 +1,200 @@ +package proxy + +import ( + "context" + "crypto/tls" + "fmt" + "io" + "net" + "net/http" + "net/url" + "strings" + "sync/atomic" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "golang.org/x/net/http2" + + "github.com/e2b-dev/infra/packages/shared/pkg/httpserver" + "github.com/e2b-dev/infra/packages/shared/pkg/logger" + "github.com/e2b-dev/infra/packages/shared/pkg/proxy/pool" +) + +func TestProxyForwardsNativeGRPCOverH2C(t *testing.T) { + t.Parallel() + + backendURL, proto, requests := startH2COnlyBackend(t) + + proxy, port, err := newTestProxy(t, func(*http.Request) (*pool.Destination, error) { + return &pool.Destination{ + Url: backendURL, + SandboxId: "test-sandbox", + RequestLogger: logger.NewNopLogger(), + ConnectionKey: "grpc-backend", + }, nil + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, proxy.Close()) }) + + req, err := http.NewRequestWithContext( + t.Context(), + http.MethodPost, + fmt.Sprintf("http://127.0.0.1:%d/pkg.Service/Method", port), + strings.NewReader("frame"), + ) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/grpc") + req.Header.Set("Te", "trailers") + + resp, err := h2cClient().Do(req) + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + assert.Equal(t, http.StatusOK, resp.StatusCode) + assert.Equal(t, "grpc-ok", string(body)) + assert.Equal(t, "0", resp.Trailer.Get("Grpc-Status")) + assert.Equal(t, uint64(1), requests.Load()) + gotProto, ok := proto.Load().(string) + require.True(t, ok) + assert.Equal(t, "HTTP/2.0", gotProto) +} + +func TestProxySendsHTTP1GRPCOverH2CToTheBackend(t *testing.T) { + t.Parallel() + + backendURL, proto, requests := startH2COnlyBackend(t) + + proxy, port, err := newTestProxy(t, func(*http.Request) (*pool.Destination, error) { + return &pool.Destination{ + Url: backendURL, + SandboxId: "test-sandbox", + RequestLogger: logger.NewNopLogger(), + ConnectionKey: "grpc-http1-frontend", + }, nil + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, proxy.Close()) }) + + req, err := http.NewRequestWithContext( + t.Context(), + http.MethodPost, + fmt.Sprintf("http://127.0.0.1:%d/pkg.Service/Method", port), + strings.NewReader("frame"), + ) + require.NoError(t, err) + req.Header.Set("Content-Type", "application/grpc") + req.Header.Set("Te", "trailers") + + resp, err := (&http.Client{}).Do(req) + require.NoError(t, err) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.NoError(t, resp.Body.Close()) + + assert.Equal(t, http.StatusOK, resp.StatusCode) + assert.Equal(t, "grpc-ok", string(body)) + assert.Equal(t, "0", resp.Trailer.Get("Grpc-Status")) + assert.Equal(t, uint64(1), requests.Load()) + gotProto, ok := proto.Load().(string) + require.True(t, ok) + assert.Equal(t, "HTTP/2.0", gotProto) +} + +func TestProxyKeepsHTTP1BackendWhenFrontendIsHTTP2(t *testing.T) { + t.Parallel() + + var lisCfg net.ListenConfig + listener, err := lisCfg.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + + backend, err := newTestBackend(listener, "http1-backend") + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, backend.Close()) }) + + proxy, port, err := newTestProxy(t, func(*http.Request) (*pool.Destination, error) { + return &pool.Destination{ + Url: backend.url, + SandboxId: "test-sandbox", + RequestLogger: logger.NewNopLogger(), + ConnectionKey: backend.id, + }, nil + }) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, proxy.Close()) }) + + req, err := http.NewRequestWithContext( + t.Context(), + http.MethodGet, + fmt.Sprintf("http://127.0.0.1:%d/hello", port), + nil, + ) + require.NoError(t, err) + + resp, err := h2cClient().Do(req) + require.NoError(t, err) + t.Cleanup(func() { require.NoError(t, resp.Body.Close()) }) + + assertBackendOutput(t, backend, resp) + assert.Equal(t, uint64(1), backend.RequestCount()) +} + +func h2cClient() *http.Client { + return &http.Client{ + Transport: &http2.Transport{ + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { + return (&net.Dialer{}).DialContext(ctx, network, addr) + }, + }, + } +} + +func startH2COnlyBackend(t *testing.T) (*url.URL, *atomic.Value, *atomic.Uint64) { + t.Helper() + + var proto atomic.Value + var requests atomic.Uint64 + + handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + requests.Add(1) + proto.Store(r.Proto) + + if r.ProtoMajor != 2 { + http.Error(w, "backend requires HTTP/2", http.StatusHTTPVersionNotSupported) + + return + } + + if !strings.HasPrefix(r.Header.Get("Content-Type"), "application/grpc") { + http.Error(w, "missing grpc content type", http.StatusUnsupportedMediaType) + + return + } + + w.Header().Set("Content-Type", "application/grpc") + w.Header().Set("Trailer", "Grpc-Status") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("grpc-ok")) + w.Header().Set("Grpc-Status", "0") + }) + + server := &http.Server{Handler: handler} + httpserver.ConfigureH2C(server) + + var lisCfg net.ListenConfig + listener, err := lisCfg.Listen(t.Context(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + + go func() { _ = server.Serve(listener) }() + t.Cleanup(func() { _ = server.Close() }) + + backendURL, err := url.Parse("http://" + listener.Addr().String()) + require.NoError(t, err) + + return backendURL, &proto, &requests +} diff --git a/packages/shared/pkg/proxy/pool/client.go b/packages/shared/pkg/proxy/pool/client.go index 3fad1a73d9..505436d2f8 100644 --- a/packages/shared/pkg/proxy/pool/client.go +++ b/packages/shared/pkg/proxy/pool/client.go @@ -12,6 +12,7 @@ import ( "time" "go.uber.org/zap" + "golang.org/x/net/http2" "github.com/e2b-dev/infra/packages/shared/pkg/consts" "github.com/e2b-dev/infra/packages/shared/pkg/logger" @@ -28,7 +29,8 @@ const tlsHandshakeTimeout = 10 * time.Second type ProxyClient struct { httputil.ReverseProxy - transport *http.Transport + transport *http.Transport + h2cTransport *http2.Transport activeConnections *smap.Map[*tracking.Connection] } @@ -46,6 +48,18 @@ func newProxyClient( ) *ProxyClient { activeConnections := smap.New[*tracking.Connection]() + dial := func(ctx context.Context, network, addr string) (net.Conn, error) { + return dialUpstream( + ctx, + network, + addr, + maxConnectionAttempts, + totalConnsCounter, + currentConnsCounter, + activeConnections, + ) + } + transport := &http.Transport{ Proxy: http.ProxyFromEnvironment, // Limit the max connection per host to avoid exhausting the number of available ports to one host. @@ -55,60 +69,43 @@ func newProxyClient( TLSHandshakeTimeout: tlsHandshakeTimeout, ResponseHeaderTimeout: 0, DisableKeepAlives: disableKeepAlives, - ForceAttemptHTTP2: false, - // TCP configuration - DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { - var conn net.Conn - var err error - - // Retry connection attempts to handle port forwarding delays in sandbox envd. - // When a process binds to localhost inside the sandbox, it can take up to 1s (delay is 1s + socat startup delay) - // for the port scanner to detect it and start socat forwarding to the host IP. - maxAttempts := max(maxConnectionAttempts, 1) - for attempt := range maxAttempts { - conn, err = (&net.Dialer{ - Timeout: 30 * time.Second, - KeepAlive: 20 * time.Second, - }).DialContext(ctx, network, addr) - - if err == nil { - totalConnsCounter.Add(1) - - return tracking.NewConnection(conn, currentConnsCounter, activeConnections), nil - } - - if ctx.Err() != nil { - return nil, ctx.Err() - } - - // Don't sleep on the last attempt - if attempt < maxAttempts-1 { - // Linear backoff: 100ms, 200ms, 300ms, 400ms - backoff := time.Duration(100*(attempt+1)) * time.Millisecond - select { - case <-time.After(backoff): - // Continue to next attempt - case <-ctx.Done(): - return nil, ctx.Err() - } - } - } - - return nil, err - }, + // HTTP/1.1 stays the default: browsers already speak HTTP/2 to the edge, + // and user HTTP/WebSocket servers inside the sandbox are HTTP/1.1. + ForceAttemptHTTP2: false, + DialContext: dial, DisableCompression: true, // No need to request or manipulate compression } if insecureSkipTLSVerify { transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true} //nolint:gosec // Sandbox services commonly use self-signed certificates. } + h2IdleTimeout := idleTimeout + if disableKeepAlives { + // http2.Transport has no DisableKeepAlives; a short idle timeout avoids + // pinning a multiplexed connection to a sandbox process that just restarted. + h2IdleTimeout = time.Second + } + + h2cTransport := &http2.Transport{ + AllowHTTP: true, + // DialTLSContext is the plaintext dial when AllowHTTP is set. + DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { + return dial(ctx, network, addr) + }, + IdleConnTimeout: h2IdleTimeout, + DisableCompression: true, + ReadIdleTimeout: 30 * time.Second, + PingTimeout: 15 * time.Second, + } + pc := &ProxyClient{ transport: transport, + h2cTransport: h2cTransport, activeConnections: activeConnections, } pc.ReverseProxy = httputil.ReverseProxy{ - Transport: transport, + Transport: &protocolSwitchTransport{http1: transport, h2c: h2cTransport}, Rewrite: func(r *httputil.ProxyRequest) { t, ok := pc.getDestination(r.In) if !ok { @@ -182,6 +179,18 @@ func newProxyClient( cors.Error(w, "Failed to route request to sandbox", http.StatusBadGateway) }, ModifyResponse: func(r *http.Response) error { + if IsGRPCRequest(r.Request) { + // Native gRPC status lives in trailers. HTTP/2 backends often + // set Content-Length, and HTTP/1.1 cannot attach trailers to a + // length-delimited body. Do not replace r.Trailer: the HTTP/2 + // transport fills that map after the body is closed. + r.ContentLength = -1 + r.Header.Del("Content-Length") + if r.Header.Get("Trailer") == "" { + r.Header.Set("Trailer", "Grpc-Status, Grpc-Message, Grpc-Status-Details-Bin") + } + } + ctx := r.Request.Context() t, ok := pc.getDestination(r.Request) if !ok { @@ -225,6 +234,9 @@ func (p *ProxyClient) getDestination(r *http.Request) (*Destination, bool) { func (p *ProxyClient) closeIdleConnections() { p.transport.CloseIdleConnections() + if p.h2cTransport != nil { + p.h2cTransport.CloseIdleConnections() + } } func (p *ProxyClient) resetAllConnections() error { @@ -239,3 +251,48 @@ func (p *ProxyClient) resetAllConnections() error { return errors.Join(errs...) } + +// dialUpstream retries the TCP dial so a request that arrives before envd's +// socat has published the localhost port still succeeds. When a process binds +// to localhost inside the sandbox, it can take up to 1s (scanner interval plus +// socat startup) before the host IP accepts connections. +func dialUpstream( + ctx context.Context, + network, addr string, + maxConnectionAttempts int, + totalConnsCounter *atomic.Uint64, + currentConnsCounter *atomic.Int64, + activeConnections *smap.Map[*tracking.Connection], +) (net.Conn, error) { + var conn net.Conn + var err error + + maxAttempts := max(maxConnectionAttempts, 1) + for attempt := range maxAttempts { + conn, err = (&net.Dialer{ + Timeout: 30 * time.Second, + KeepAlive: 20 * time.Second, + }).DialContext(ctx, network, addr) + + if err == nil { + totalConnsCounter.Add(1) + + return tracking.NewConnection(conn, currentConnsCounter, activeConnections), nil + } + + if ctx.Err() != nil { + return nil, ctx.Err() + } + + if attempt < maxAttempts-1 { + backoff := time.Duration(100*(attempt+1)) * time.Millisecond + select { + case <-time.After(backoff): + case <-ctx.Done(): + return nil, ctx.Err() + } + } + } + + return nil, err +} diff --git a/packages/shared/pkg/proxy/pool/grpc.go b/packages/shared/pkg/proxy/pool/grpc.go new file mode 100644 index 0000000000..2755578dda --- /dev/null +++ b/packages/shared/pkg/proxy/pool/grpc.go @@ -0,0 +1,78 @@ +package pool + +import ( + "mime" + "net/http" + "strings" +) + +// IsGRPCRequest reports whether *r* is native gRPC (Content-Type +// application/grpc), as opposed to HTTP, WebSocket, Connect-RPC, or gRPC-Web. +// +// The public sandbox proxy must keep those other protocols on HTTP/1.1: browsers +// already speak HTTP/2 to the edge, and blindly forwarding HTTP/2 to a Python +// http.server or a VNC upgrade would break them. Native gRPC is identified by +// Content-Type so only those requests use an h2c backend transport. +// +// If an edge LB ever protocol-selects on Content-Type, it must use this same +// cutoff and must not prefix-match "application/grpc" (that would include +// grpc-web). Suggested matchers, kept in tests: +// +// regexp: (?i)^application/grpc($|[+;].*) +// AWS-style: application/grpc, application/grpc;*, application/grpc+* +func IsGRPCRequest(r *http.Request) bool { + if r == nil { + return false + } + + return isGRPCContentType(r.Header.Get("Content-Type")) +} + +func isGRPCContentType(contentType string) bool { + if contentType == "" { + return false + } + + mediaType, _, err := mime.ParseMediaType(contentType) + if err != nil { + mediaType, _, _ = strings.Cut(contentType, ";") + mediaType = strings.TrimSpace(mediaType) + } + + mediaType = strings.ToLower(mediaType) + if mediaType == "application/grpc" { + return true + } + + // application/grpc+proto, application/grpc+json, … — not application/grpc-web. + return strings.HasPrefix(mediaType, "application/grpc+") +} + +// protocolSwitchTransport sends native gRPC over unencrypted HTTP/2 (h2c) and +// everything else over the existing HTTP/1.1 transport. +type protocolSwitchTransport struct { + http1 http.RoundTripper + h2c http.RoundTripper +} + +func (t *protocolSwitchTransport) RoundTrip(req *http.Request) (*http.Response, error) { + if IsGRPCRequest(req) { + return t.h2c.RoundTrip(req) + } + + return t.http1.RoundTrip(req) +} + +func (t *protocolSwitchTransport) CloseIdleConnections() { + type idleCloser interface { + CloseIdleConnections() + } + + if closer, ok := t.http1.(idleCloser); ok { + closer.CloseIdleConnections() + } + + if closer, ok := t.h2c.(idleCloser); ok { + closer.CloseIdleConnections() + } +} diff --git a/packages/shared/pkg/proxy/pool/grpc_test.go b/packages/shared/pkg/proxy/pool/grpc_test.go new file mode 100644 index 0000000000..4425ef72e2 --- /dev/null +++ b/packages/shared/pkg/proxy/pool/grpc_test.go @@ -0,0 +1,85 @@ +package pool + +import ( + "net/http" + "regexp" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Must stay aligned with the suggested edge Content-Type matchers in grpc.go. +var edgeNativeGRPCContentType = regexp.MustCompile(`(?i)^application/grpc($|[+;].*)`) + +func TestIsGRPCRequest(t *testing.T) { + t.Parallel() + + tests := []struct { + contentType string + want bool + }{ + {contentType: "application/grpc", want: true}, + {contentType: "application/grpc+proto", want: true}, + {contentType: "application/grpc+json", want: true}, + {contentType: "application/grpc; charset=utf-8", want: true}, + {contentType: "Application/Grpc", want: true}, + {contentType: "application/grpc-web", want: false}, + {contentType: "application/grpc-web+proto", want: false}, + {contentType: "application/connect+proto", want: false}, + {contentType: "application/json", want: false}, + {contentType: "text/event-stream", want: false}, + {contentType: "", want: false}, + } + + for _, test := range tests { + t.Run(test.contentType, func(t *testing.T) { + t.Parallel() + + req, err := http.NewRequest(http.MethodPost, "http://sandbox/service/Method", nil) + require.NoError(t, err) + if test.contentType != "" { + req.Header.Set("Content-Type", test.contentType) + } + + assert.Equal(t, test.want, IsGRPCRequest(req)) + }) + } +} + +func TestIsGRPCRequestNil(t *testing.T) { + t.Parallel() + + assert.False(t, IsGRPCRequest(nil)) +} + +func TestEdgeGRPCContentTypeRegexpAgreesWithProxy(t *testing.T) { + t.Parallel() + + tests := []string{ + "application/grpc", + "application/grpc+proto", + "application/grpc+json", + "application/grpc; charset=utf-8", + "Application/Grpc", + "application/grpc-web", + "application/grpc-web+proto", + "application/connect+proto", + "application/json", + "", + } + + for _, contentType := range tests { + t.Run(contentType, func(t *testing.T) { + t.Parallel() + + req, err := http.NewRequest(http.MethodPost, "http://sandbox/service/Method", nil) + require.NoError(t, err) + if contentType != "" { + req.Header.Set("Content-Type", contentType) + } + + assert.Equal(t, IsGRPCRequest(req), edgeNativeGRPCContentType.MatchString(contentType)) + }) + } +} diff --git a/packages/shared/pkg/proxy/proxy_retries_unix_test.go b/packages/shared/pkg/proxy/proxy_retries_unix_test.go new file mode 100644 index 0000000000..1e78fc50af --- /dev/null +++ b/packages/shared/pkg/proxy/proxy_retries_unix_test.go @@ -0,0 +1,148 @@ +//go:build unix + +package proxy + +import ( + "fmt" + "net" + "net/http" + "net/url" + "os" + "syscall" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/e2b-dev/infra/packages/shared/pkg/logger" + "github.com/e2b-dev/infra/packages/shared/pkg/proxy/pool" +) + +// reservedPort holds a TCP socket that is bound but not yet listening: +// connections are refused while the port stays reserved for this test, and +// listen turns the same socket into a live listener. This avoids the +// close-and-rebind race where another process grabs the port in between. +type reservedPort struct { + fd int + addr string + closed bool +} + +func reserveTCPPort(t *testing.T) *reservedPort { + t.Helper() + + fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0) + require.NoError(t, err) + r := &reservedPort{fd: fd} + t.Cleanup(r.close) + + require.NoError(t, syscall.Bind(fd, &syscall.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}})) + sa, err := syscall.Getsockname(fd) + require.NoError(t, err) + r.addr = fmt.Sprintf("127.0.0.1:%d", sa.(*syscall.SockaddrInet4).Port) + + return r +} + +// close releases the raw fd unless listen already handed it off. +func (r *reservedPort) close() { + if r.closed { + return + } + r.closed = true + _ = syscall.Close(r.fd) +} + +func (r *reservedPort) listen() (net.Listener, error) { + if err := syscall.Listen(r.fd, 128); err != nil { + return nil, fmt.Errorf("listen on reserved port: %w", err) + } + if err := syscall.SetNonblock(r.fd, true); err != nil { + return nil, fmt.Errorf("set nonblock: %w", err) + } + r.closed = true // fd ownership moves to the os.File below + f := os.NewFile(uintptr(r.fd), "reserved-port") + defer f.Close() // net.FileListener dups the fd + + return net.FileListener(f) +} + +// TestProxyRetriesOnDelayedBackendStartup simulates the scenario where a backend +// server starts up after the initial connection attempt (like envd port forwarding delay). +func TestProxyRetriesOnDelayedBackendStartup(t *testing.T) { + t.Parallel() + reserved := reserveTCPPort(t) + backendAddr := reserved.addr + + backendURL, err := url.Parse(fmt.Sprintf("http://%s", backendAddr)) + require.NoError(t, err) + + getDestination := func(_ *http.Request) (*pool.Destination, error) { + return &pool.Destination{ + Url: backendURL, + SandboxId: "test-sandbox", + RequestLogger: logger.NewNopLogger(), + ConnectionKey: "delayed-backend", + }, nil + } + + proxy, port, err := newTestProxy(t, getDestination) + require.NoError(t, err) + defer proxy.Close() + + type backendResult struct { + backend *testBackend + err error + } + backendReady := make(chan backendResult, 1) + + // Start backend after a delay (simulating envd port forwarding) + go func() { + // Wait 300ms before starting the backend (should succeed on retry 2 or 3) + time.Sleep(300 * time.Millisecond) + + listener, err := reserved.listen() + if err != nil { + backendReady <- backendResult{nil, fmt.Errorf("failed to create delayed backend listener: %w", err)} + + return + } + + backend, err := newTestBackend(listener, "delayed-backend") + if err != nil { + listener.Close() + backendReady <- backendResult{nil, fmt.Errorf("failed to create delayed backend: %w", err)} + + return + } + + backendReady <- backendResult{backend, nil} + }() + + // Make request - this should retry and eventually succeed + proxyURL := fmt.Sprintf("http://127.0.0.1:%d/hello", port) + start := time.Now() + + resp, err := httpGet(t, proxyURL) + require.NoError(t, err) + defer resp.Body.Close() + + elapsed := time.Since(start) + + // Wait for backend to be ready before checking + result := <-backendReady + require.NoError(t, result.err) + backend := result.backend + defer backend.Close() + + assertBackendOutput(t, backend, resp) + + // Verify that it took at least the delay time (proving retries happened) + assert.GreaterOrEqual(t, elapsed, 300*time.Millisecond, "request should have waited for backend to start") + assert.Less(t, elapsed, 2*time.Second, "request should have succeeded before all retries exhausted") + + // Verify the connection was established + assert.Equal(t, uint64(1), backend.RequestCount(), "backend should have been called once") + assert.Equal(t, uint64(1), proxy.TotalPoolConnections(), "proxy should have established one connection") +} diff --git a/packages/shared/pkg/proxy/proxy_test.go b/packages/shared/pkg/proxy/proxy_test.go index 2450fd2209..97445b8fa6 100644 --- a/packages/shared/pkg/proxy/proxy_test.go +++ b/packages/shared/pkg/proxy/proxy_test.go @@ -10,7 +10,6 @@ import ( "net/http" "net/http/httptest" "net/url" - "os" "sync" "sync/atomic" "syscall" @@ -938,134 +937,6 @@ func TestProxyDoesNotReuseConnectionsWhenBackendChanges(t *testing.T) { assert.Equal(t, uint64(2), proxy.TotalPoolConnections(), "proxy should not have reused the connection") } -// reservedPort holds a TCP socket that is bound but not yet listening: -// connections are refused while the port stays reserved for this test, and -// listen turns the same socket into a live listener. This avoids the -// close-and-rebind race where another process grabs the port in between. -type reservedPort struct { - fd int - addr string - closed bool -} - -func reserveTCPPort(t *testing.T) *reservedPort { - t.Helper() - - fd, err := syscall.Socket(syscall.AF_INET, syscall.SOCK_STREAM, 0) - require.NoError(t, err) - r := &reservedPort{fd: fd} - t.Cleanup(r.close) - - require.NoError(t, syscall.Bind(fd, &syscall.SockaddrInet4{Addr: [4]byte{127, 0, 0, 1}})) - sa, err := syscall.Getsockname(fd) - require.NoError(t, err) - r.addr = fmt.Sprintf("127.0.0.1:%d", sa.(*syscall.SockaddrInet4).Port) - - return r -} - -// close releases the raw fd unless listen already handed it off. -func (r *reservedPort) close() { - if r.closed { - return - } - r.closed = true - _ = syscall.Close(r.fd) -} - -func (r *reservedPort) listen() (net.Listener, error) { - if err := syscall.Listen(r.fd, 128); err != nil { - return nil, fmt.Errorf("listen on reserved port: %w", err) - } - if err := syscall.SetNonblock(r.fd, true); err != nil { - return nil, fmt.Errorf("set nonblock: %w", err) - } - r.closed = true // fd ownership moves to the os.File below - f := os.NewFile(uintptr(r.fd), "reserved-port") - defer f.Close() // net.FileListener dups the fd - - return net.FileListener(f) -} - -// TestProxyRetriesOnDelayedBackendStartup simulates the scenario where a backend -// server starts up after the initial connection attempt (like envd port forwarding delay). -func TestProxyRetriesOnDelayedBackendStartup(t *testing.T) { - t.Parallel() - reserved := reserveTCPPort(t) - backendAddr := reserved.addr - - backendURL, err := url.Parse(fmt.Sprintf("http://%s", backendAddr)) - require.NoError(t, err) - - getDestination := func(_ *http.Request) (*pool.Destination, error) { - return &pool.Destination{ - Url: backendURL, - SandboxId: "test-sandbox", - RequestLogger: logger.NewNopLogger(), - ConnectionKey: "delayed-backend", - }, nil - } - - proxy, port, err := newTestProxy(t, getDestination) - require.NoError(t, err) - defer proxy.Close() - - type backendResult struct { - backend *testBackend - err error - } - backendReady := make(chan backendResult, 1) - - // Start backend after a delay (simulating envd port forwarding) - go func() { - // Wait 300ms before starting the backend (should succeed on retry 2 or 3) - time.Sleep(300 * time.Millisecond) - - listener, err := reserved.listen() - if err != nil { - backendReady <- backendResult{nil, fmt.Errorf("failed to create delayed backend listener: %w", err)} - - return - } - - backend, err := newTestBackend(listener, "delayed-backend") - if err != nil { - listener.Close() - backendReady <- backendResult{nil, fmt.Errorf("failed to create delayed backend: %w", err)} - - return - } - - backendReady <- backendResult{backend, nil} - }() - - // Make request - this should retry and eventually succeed - proxyURL := fmt.Sprintf("http://127.0.0.1:%d/hello", port) - start := time.Now() - - resp, err := httpGet(t, proxyURL) - require.NoError(t, err) - defer resp.Body.Close() - - elapsed := time.Since(start) - - // Wait for backend to be ready before checking - result := <-backendReady - require.NoError(t, result.err) - backend := result.backend - defer backend.Close() - - assertBackendOutput(t, backend, resp) - - // Verify that it took at least the delay time (proving retries happened) - assert.GreaterOrEqual(t, elapsed, 300*time.Millisecond, "request should have waited for backend to start") - assert.Less(t, elapsed, 2*time.Second, "request should have succeeded before all retries exhausted") - - // Verify the connection was established - assert.Equal(t, uint64(1), backend.RequestCount(), "backend should have been called once") - assert.Equal(t, uint64(1), proxy.TotalPoolConnections(), "proxy should have established one connection") -} - type data struct { Tag string `json:"tag"` Host string `json:"host"`