resolve goroutine leak in serveOnPipe

internal/sambox/mesh.go

@@ -82,6 +82,7 @@ func (d *AgentDialer) dialMeshService(ctx context.Context, route Route) (net.Con
 // connection.
 func serveOnPipe(h http.Handler) net.Conn {
 	agentSide, boundarySide := net.Pipe()
+	ln := newSingleConnListener(boundarySide)
 	server := &http.Server{
 		Handler: h,
 		// Mirrors the sidecar: bound header reads, but let bodies and responses
@@ -90,7 +91,7 @@ func serveOnPipe(h http.Handler) net.Conn {
 		IdleTimeout:       120 * time.Second,
 	}
 	go func() {
-		_ = server.Serve(newSingleConnListener(boundarySide))
+		_ = server.Serve(ln)
 	}()
 	return agentSide
 }
@@ -153,8 +154,8 @@ func sidecarTransport(socket string) http.RoundTripper {
 }
 
 // singleConnListener hands one already-established connection to an
-// http.Server and then blocks, so the server lives exactly as long as the
-// agent's connection does.
+// http.Server and then blocks until Close, so the server lives exactly as long
+// as the agent's connection does.
 type singleConnListener struct {
 	conn net.Conn
 
@@ -163,8 +164,26 @@ type singleConnListener struct {
 	closed  chan struct{}
 }
 
+// closeNotifyConn closes the listener when the underlying connection is closed,
+// so http.Server.Serve unblocks from Accept instead of leaking a goroutine.
+type closeNotifyConn struct {
+	net.Conn
+	fn func()
+}
+
+func (c *closeNotifyConn) Close() error {
+	err := c.Conn.Close()
+	c.fn()
+	return err
+}
+
 func newSingleConnListener(conn net.Conn) *singleConnListener {
-	return &singleConnListener{conn: conn, closed: make(chan struct{})}
+	l := &singleConnListener{closed: make(chan struct{})}
+	l.conn = &closeNotifyConn{
+		Conn: conn,
+		fn:   func() { _ = l.Close() },
+	}
+	return l
 }
 
 func (l *singleConnListener) Accept() (net.Conn, error) {

internal/sambox/mesh_test.go

@@ -25,6 +25,7 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/google/sam/api"
 )
@@ -189,3 +190,37 @@ func TestUnreachableSidecarIsReportedAsUnreachable(t *testing.T) {
 		t.Fatalf("DialDestination = %v, want ErrHostUnreachable", err)
 	}
 }
+
+// TestSingleConnListenerServeExitsOnConnClose verifies that after the one
+// connection is finished and closed, http.Server.Serve returns instead of
+// blocking forever on a second Accept (goroutine leak in serveOnPipe).
+func TestSingleConnListenerServeExitsOnConnClose(t *testing.T) {
+	agentSide, boundarySide := net.Pipe()
+	ln := newSingleConnListener(boundarySide)
+	srv := &http.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+			_, _ = io.WriteString(w, "ok")
+		}),
+		ReadHeaderTimeout: 10 * time.Second,
+	}
+
+	exited := make(chan struct{})
+	go func() {
+		_ = srv.Serve(ln)
+		close(exited)
+	}()
+
+	resp, err := clientOver(agentSide).Get("http://mesh.example/")
+	if err != nil {
+		t.Fatalf("Get: %v", err)
+	}
+	_, _ = io.Copy(io.Discard, resp.Body)
+	_ = resp.Body.Close()
+	_ = agentSide.Close()
+
+	select {
+	case <-exited:
+	case <-time.After(2 * time.Second):
+		t.Fatal("http.Server.Serve did not return after pipe close (Accept leak)")
+	}
+}