Avoid async IPC panic on peer close

tsc/internal/ipc/conn_async.go

@@ -64,7 +64,26 @@ func (c *AsyncConn) SetCollectTiming(enabled bool) {
 // Run starts processing messages on the connection.
 // It blocks until the context is cancelled or an error occurs.
 func (c *AsyncConn) Run(ctx context.Context) (err error) {
-	defer func() { c.closePendingCalls(err) }()
+	ctx, cancel := context.WithCancel(ctx)
+	requestErrors := make(chan error, 1)
+	reportRequestError := func(requestErr error) {
+		select {
+		case requestErrors <- requestErr:
+			return
+		default:
+			return
+		}
+	}
+	defer func() {
+		cancel()
+		c.closePendingCalls(err)
+		select {
+		case requestErr := <-requestErrors:
+			err = errors.Join(err, requestErr)
+		default:
+			return
+		}
+	}()
 	for {
 		if ctx.Err() != nil {
 			return ctx.Err()
@@ -81,7 +100,12 @@ func (c *AsyncConn) Run(ctx context.Context) (err error) {
 		if msg.IsResponse() {
 			c.handleResponse(msg)
 		} else if msg.IsRequest() {
-			go c.handleRequest(ctx, msg)
+			go func() {
+				if requestErr := c.handleRequest(ctx, msg); requestErr != nil {
+					reportRequestError(requestErr)
+					_ = c.rwc.Close()
+				}
+			}()
 		} else if msg.IsNotification() {
 			go c.handleNotification(ctx, msg)
 		}
@@ -120,7 +144,7 @@ func (c *AsyncConn) handleResponse(msg *Message) {
 }
 
 // handleRequest processes an incoming request.
-func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) {
+func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) (retErr error) {
 	// Intercept the meta-requests for collected server timing before dispatching
 	// to the handler, so they are answered directly and not themselves recorded.
 	switch msg.Method {
@@ -129,9 +153,9 @@ func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) {
 		writeErr := c.protocol.WriteResponse(msg.ID, serverTimingSnapshot(c.timing))
 		c.writeMu.Unlock()
 		if writeErr != nil {
-			panic(fmt.Sprintf("ipc: failed to write server timing response: %v", writeErr))
+			return fmt.Errorf("ipc: failed to write server timing response: %w", writeErr)
 		}
-		return
+		return nil
 	case string(MethodResetServerTiming):
 		if c.timing != nil {
 			c.timing.reset()
@@ -140,9 +164,9 @@ func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) {
 		writeErr := c.protocol.WriteResponse(msg.ID, nil)
 		c.writeMu.Unlock()
 		if writeErr != nil {
-			panic(fmt.Sprintf("ipc: failed to write reset server timing response: %v", writeErr))
+			return fmt.Errorf("ipc: failed to write reset server timing response: %w", writeErr)
 		}
-		return
+		return nil
 	}
 
 	var result any
@@ -167,7 +191,7 @@ func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) {
 			c.writeMu.Unlock()
 
 			if writeErr != nil {
-				panic(fmt.Sprintf("ipc: failed to write panic error response: %v (original panic: %v)", writeErr, r))
+				retErr = fmt.Errorf("ipc: failed to write panic error response: %w (original panic: %v)", writeErr, r)
 			}
 		}
 	}()
@@ -192,8 +216,9 @@ func (c *AsyncConn) handleRequest(ctx context.Context, msg *Message) {
 	}
 
 	if writeErr != nil {
-		panic(fmt.Sprintf("ipc: failed to write response: %v", writeErr))
+		return fmt.Errorf("ipc: failed to write response: %w", writeErr)
 	}
+	return nil
 }
 
 // handleNotification processes an incoming notification.

tsc/internal/ipc/conn_async_test.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/microsoft/TypeScript/tsc/internal/ipc"
 	"github.com/microsoft/TypeScript/tsc/internal/json"
+	"github.com/microsoft/TypeScript/tsc/internal/jsonrpc"
 	"gotest.tools/v3/assert"
 )
 
@@ -23,6 +24,23 @@ func (noOpHandler) HandleNotification(context.Context, string, json.Value) error
 	return nil
 }
 
+type blockingHandler struct {
+	started chan struct{}
+	release chan struct{}
+	done    chan struct{}
+}
+
+func (h blockingHandler) HandleRequest(context.Context, string, json.Value) (any, error) {
+	close(h.started)
+	<-h.release
+	close(h.done)
+	return nil, nil
+}
+
+func (blockingHandler) HandleNotification(context.Context, string, json.Value) error {
+	return nil
+}
+
 func TestAsyncConnCallReturnsWhenPeerCloses(t *testing.T) {
 	t.Parallel()
 	client, server := net.Pipe()
@@ -67,3 +85,50 @@ func TestAsyncConnCallAfterReadLoopFailureReturnsImmediately(t *testing.T) {
 	err = conn.Notify(ctx, "changed", nil)
 	assert.Assert(t, errors.Is(err, ipc.ErrConnClosed), "expected ErrConnClosed, got %v", err)
 }
+
+func TestAsyncConnRunReturnsWhenPeerClosesDuringRequest(t *testing.T) {
+	t.Parallel()
+	client, server := net.Pipe()
+	defer server.Close()
+	handler := blockingHandler{
+		started: make(chan struct{}),
+		release: make(chan struct{}),
+		done:    make(chan struct{}),
+	}
+	defer func() {
+		select {
+		case <-handler.release:
+			return
+		default:
+			close(handler.release)
+		}
+	}()
+	conn := ipc.NewAsyncConn(server, handler)
+	runDone := make(chan error, 1)
+	go func() { runDone <- conn.Run(t.Context()) }()
+
+	protocol := ipc.NewJSONRPCProtocol(client)
+	assert.NilError(t, protocol.WriteRequest(jsonrpc.NewIDInt(1), "transform", nil))
+	select {
+	case <-handler.started:
+		break
+	case <-time.After(time.Second):
+		t.Fatal("request handler did not start")
+	}
+	assert.NilError(t, client.Close())
+
+	select {
+	case <-runDone:
+		break
+	case <-time.After(time.Second):
+		t.Fatal("connection did not stop while request handler was blocked")
+	}
+
+	close(handler.release)
+	select {
+	case <-handler.done:
+		break
+	case <-time.After(time.Second):
+		t.Fatal("request handler did not stop")
+	}
+}