fix: Resolve dispatcher tunnel response hang and connection handoff issues.

The dispatcher TunnelClient waited for connection close before resolving
HTTP responses, causing indefinite hangs with HTTP/1.1 keep-alive. Now
detects complete responses via Content-Length and chunked encoding.

On the interceptor side, new dispatcher function instances would connect
via WebSocket but the tunnel kept reading from the old connection. The
interceptor now closes the previous active connection when a new one
arrives, forcing a stream refresh. The WakeableWSStream also checks for
pending connections before calling wake to avoid redundant reconnects.

Increases dispatcher e2e sandbox timeout from 1m to 5m and suite context
from 1m to 3m to accommodate deployment wait times.

🐘 Generated with Crush

Assisted-by: Claude Opus 4.6 via Crush <crush@charm.land>

Author: vercel-eddie

e2esandbox/dispatcher_test.go

@@ -46,7 +46,7 @@ func TestDispatcherSuite(t *testing.T) {
 }
 
 func (s *DispatcherSuite) SetupSuite() {
-	s.ctx, s.cancel = context.WithTimeout(context.Background(), 1*time.Minute)
+	s.ctx, s.cancel = context.WithTimeout(context.Background(), 3*time.Minute)
 	t := s.T()
 
 	token := resolveVercelToken(t)
@@ -83,7 +83,7 @@ func (s *DispatcherSuite) SetupSuite() {
 		ProjectID: projectID,
 		Ports:     []int{8081},
 		Runtime:   "node24",
-		Timeout:   1 * time.Minute,
+		Timeout:   5 * time.Minute,
 	})
 	require.NoError(t, err, "failed to create sandbox")
 
@@ -186,7 +186,9 @@ func (s *DispatcherSuite) TestDispatcherPreviewFlow() {
 		}
 
 		return payload.Status == "ok"
-	}, 30*time.Second, 2*time.Second, "dispatcher preview did not forward /api/health to the sandbox userservice")
+	}, 60*time.Second, 3*time.Second, "dispatcher preview did not forward /api/health to the sandbox userservice")
+
+	s.dumpSandboxDiagnostics()
 
 	out, err := s.execInSandbox("curl", "-sf", "--max-time", "20", "--proxy", "socks5h://127.0.0.1:1080", "http://example.com")
 	require.NoError(t, err, "curl through dispatcher-backed SOCKS proxy failed")

flow/dispatcher/src/tunnel.ts

@@ -228,6 +228,23 @@ export class TunnelClient {
       if (message.data.length > 0) {
         pending.chunks.push(message.data);
       }
+
+      if (this.isCompleteHttpResponse(pending.chunks)) {
+        const fullData = this.concatenateChunks(pending.chunks);
+        const finalMessage = create(TunnelNetworkMessageSchema, {
+          connectionId,
+          data: fullData,
+        });
+        pending.resolve(finalMessage);
+        this.pendingRequests.delete(connectionId);
+
+        const closeMsg = create(TunnelNetworkMessageSchema, {
+          connectionId,
+          error: "dispatcher: response complete",
+          data: new Uint8Array(),
+        });
+        this.sendMessageDirect(closeMsg);
+      }
       return;
     }
 
@@ -401,6 +418,30 @@ export class TunnelClient {
     });
   }
 
+  private isCompleteHttpResponse(chunks: Uint8Array[]): boolean {
+    const data = this.concatenateChunks(chunks);
+    const text = new TextDecoder().decode(data);
+
+    const headerEnd = text.indexOf("\r\n\r\n");
+    if (headerEnd === -1) return false;
+
+    const headerSection = text.slice(0, headerEnd).toLowerCase();
+    const bodyStartIdx = headerEnd + 4;
+
+    const clMatch = headerSection.match(/content-length:\s*(\d+)/);
+    if (clMatch) {
+      const expectedLen = parseInt(clMatch[1], 10);
+      const bodyBytes = data.slice(new TextEncoder().encode(text.slice(0, bodyStartIdx)).length);
+      return bodyBytes.length >= expectedLen;
+    }
+
+    if (headerSection.includes("transfer-encoding: chunked")) {
+      return text.endsWith("0\r\n\r\n") || text.includes("\r\n0\r\n\r\n");
+    }
+
+    return false;
+  }
+
   private concatenateChunks(chunks: Uint8Array[]): Uint8Array {
     const totalLength = chunks.reduce((sum, chunk) => sum + chunk.length, 0);
     const result = new Uint8Array(totalLength);

pkg/commands/intercept_health.go

@@ -6,16 +6,18 @@ import (
 	"log/slog"
 	"net"
 	"net/http"
+	"sync"
 
 	"github.com/gorilla/websocket"
 )
 
-// interceptServer accepts inbound dispatcher tunnel connections and hands them
-// to the interceptor stream via a channel.
 type interceptServer struct {
 	httpServer   *http.Server
 	tunnelConnCh chan *websocket.Conn
 	upgrader     websocket.Upgrader
+
+	mu         sync.Mutex
+	activeConn *websocket.Conn
 }
 
 func newInterceptServer(addr string) (*interceptServer, error) {
@@ -64,16 +66,22 @@ func (s *interceptServer) handleTunnelWS(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	s.mu.Lock()
+	oldActive := s.activeConn
+	s.activeConn = conn
+	s.mu.Unlock()
+
+	if oldActive != nil {
+		slog.Info("New tunnel connection arrived, closing previous active connection")
+		_ = oldActive.Close()
+	}
+
 	select {
-	case s.tunnelConnCh <- conn:
-	default:
-		select {
-		case staleConn := <-s.tunnelConnCh:
-			if staleConn != nil {
-				_ = staleConn.Close()
-			}
-		default:
+	case old := <-s.tunnelConnCh:
+		if old != nil {
+			_ = old.Close()
 		}
-		s.tunnelConnCh <- conn
+	default:
 	}
+	s.tunnelConnCh <- conn
 }

pkg/tunnel/stream_ws.go

@@ -43,12 +43,21 @@ func NewReconnectableWSStream(conn *websocket.Conn, refresher func(context.Conte
 func NewWakeableWSStream(connCh <-chan *websocket.Conn, wake func(context.Context) error) *WSStream {
 	return &WSStream{
 		refresher: func(ctx context.Context) (*websocket.Conn, error) {
-			if wake == nil {
-				return nil, fmt.Errorf("ws stream: wake not configured")
-			}
 			if connCh == nil {
 				return nil, fmt.Errorf("ws stream: tunnel connection channel not configured")
 			}
+
+			select {
+			case conn := <-connCh:
+				if conn != nil {
+					return conn, nil
+				}
+			default:
+			}
+
+			if wake == nil {
+				return nil, fmt.Errorf("ws stream: wake not configured")
+			}
 			if err := wake(ctx); err != nil {
 				return nil, err
 			}