From 205fddb7aae47014580938d9950f528b54bd96c6 Mon Sep 17 00:00:00 2001 From: cplieger <917744+cplieger@users.noreply.github.com> Date: Mon, 5 Oct 2026 12:28:19 +0200 Subject: [PATCH 1/2] [api] Answer nested requests on the sync connection in stack order --- tsc/internal/ipc/conn_sync.go | 42 +++++++- tsc/internal/ipc/conn_sync_test.go | 149 +++++++++++++++++++++++++++++ 2 files changed, 190 insertions(+), 1 deletion(-) diff --git a/tsc/internal/ipc/conn_sync.go b/tsc/internal/ipc/conn_sync.go index a0b0cecb3c3d1..1b1a8033f6360 100644 --- a/tsc/internal/ipc/conn_sync.go +++ b/tsc/internal/ipc/conn_sync.go @@ -28,15 +28,38 @@ type SyncConn struct { // This ensures that concurrent calls from handler goroutines (e.g., project code // spawning goroutines that invoke filesystem callbacks) don't corrupt the stream. mu sync.Mutex + + // inFlight holds the exchanges in progress, innermost last: true for a call to the client, false for a request from it. + inFlight []bool + changed sync.Cond } // NewSyncConn creates a new sync connection with the given transport and handler. func NewSyncConn(rwc io.ReadWriteCloser, protocol Protocol, handler Handler) *SyncConn { - return &SyncConn{ + c := &SyncConn{ rwc: rwc, protocol: protocol, handler: handler, } + c.changed.L = &c.mu + return c +} + +func (c *SyncConn) push(call bool) int { + for call && len(c.inFlight) > 0 && c.inFlight[len(c.inFlight)-1] { + c.changed.Wait() + } + c.inFlight = append(c.inFlight, call) + c.changed.Broadcast() + return len(c.inFlight) +} + +func (c *SyncConn) pop(depth int) { + for len(c.inFlight) != depth { + c.changed.Wait() + } + c.inFlight = c.inFlight[:depth-1] + c.changed.Broadcast() } // SetCollectTiming enables or disables per-request server processing-time @@ -84,11 +107,23 @@ func (c *SyncConn) Run(ctx context.Context) error { // handleRequest processes an incoming request. func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr error) { + c.mu.Lock() + depth := c.push(false) + c.mu.Unlock() + answered := false + answer := func() { + if !answered { + answered = true + c.pop(depth) + } + } + // 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 { case string(MethodGetServerTiming): c.mu.Lock() + answer() writeErr := c.protocol.WriteResponse(msg.ID, serverTimingSnapshot(c.timing)) c.mu.Unlock() if writeErr != nil { @@ -100,6 +135,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro c.timing.reset() } c.mu.Lock() + answer() writeErr := c.protocol.WriteResponse(msg.ID, nil) c.mu.Unlock() if writeErr != nil { @@ -123,6 +159,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro err = fmt.Errorf("panic: %v\n%s", r, stack) c.mu.Lock() + answer() writeErr := c.protocol.WriteError(msg.ID, &jsonrpc.ResponseError{ Code: jsonrpc.CodeInternalError, Message: err.Error(), @@ -143,6 +180,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro c.mu.Lock() defer c.mu.Unlock() + answer() var writeErr error if err != nil { @@ -175,6 +213,8 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va // 3. We need to ensure write/read pairs are atomic c.mu.Lock() defer c.mu.Unlock() + call := c.push(true) + defer c.pop(call) id := jsonrpc.NewIDString(method) diff --git a/tsc/internal/ipc/conn_sync_test.go b/tsc/internal/ipc/conn_sync_test.go index cc10ce1523861..f4c4ef745d49e 100644 --- a/tsc/internal/ipc/conn_sync_test.go +++ b/tsc/internal/ipc/conn_sync_test.go @@ -4,7 +4,9 @@ import ( "context" "errors" "io" + "sync" "testing" + "testing/synctest" "github.com/microsoft/TypeScript/tsc/internal/ipc" "github.com/microsoft/TypeScript/tsc/internal/json" @@ -78,3 +80,150 @@ func TestSyncConnRunReturnsPanicResponseWriteFailure(t *testing.T) { assert.Assert(t, errors.Is(err, responseErr), "expected panic response write error, got %v", err) assert.ErrorContains(t, err, "original panic: handler panic") } + +type syncPanickingResponseProtocol struct { + syncFailingResponseProtocol +} + +func (*syncPanickingResponseProtocol) WriteResponse(*jsonrpc.ID, any) error { + panic("response write panic") +} + +func TestSyncConnRunReturnsResponseWritePanic(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + responseErr := errors.New("error write failed") + protocol := &syncPanickingResponseProtocol{syncFailingResponseProtocol{ + message: &ipc.Message{ID: jsonrpc.NewIDInt(1), Method: "transform"}, + responseErr: responseErr, + }} + conn := ipc.NewSyncConn(nil, protocol, noOpHandler{}) + + err := conn.Run(t.Context()) + assert.Assert(t, errors.Is(err, responseErr), "expected panic response write error, got %v", err) + assert.ErrorContains(t, err, "original panic: response write panic") + }) +} + +type stackClient struct { + toServer chan *ipc.Message + toClient chan *ipc.Message + waiting map[string]chan struct{} + answers map[string]string +} + +func (c *stackClient) ReadMessage() (*ipc.Message, error) { + return <-c.toServer, nil +} + +func (c *stackClient) WriteRequest(id *jsonrpc.ID, method string, params any) error { + data, err := json.Marshal(params) + c.toClient <- &ipc.Message{ID: id, Method: method, Params: data} + return err +} + +func (*stackClient) WriteNotification(string, any) error { + return nil +} + +func (c *stackClient) WriteResponse(id *jsonrpc.ID, result any) error { + data, err := json.Marshal(result) + c.toClient <- &ipc.Message{ID: id, Result: data} + return err +} + +func (c *stackClient) WriteError(id *jsonrpc.ID, responseErr *jsonrpc.ResponseError) error { + c.toClient <- &ipc.Message{ID: id, Error: responseErr} + return nil +} + +func (c *stackClient) run() { + for msg := range c.toClient { + c.handle(msg) + } +} + +func (c *stackClient) handle(callback *ipc.Message) { + c.toServer <- &ipc.Message{ID: jsonrpc.NewIDString("resolve"), Method: "resolve", Params: callback.Params} + if waiting := c.waiting[string(callback.Params)]; waiting != nil { + close(waiting) + } + c.answers[string(callback.Params)] = string(c.await()) + c.toServer <- &ipc.Message{ID: callback.ID, Result: callback.Params} +} + +func (c *stackClient) await() json.Value { + for msg := range c.toClient { + if msg.IsRequest() { + c.handle(msg) + continue + } + return msg.Result + } + return nil +} + +type nestedRequestHandler struct { + aServing chan struct{} + bAtClient chan struct{} + releaseB chan struct{} +} + +func (h *nestedRequestHandler) HandleRequest(_ context.Context, _ string, params json.Value) (any, error) { + switch string(params) { + case `"a"`: + close(h.aServing) + <-h.bAtClient + case `"b"`: + <-h.releaseB + } + return params, nil +} + +func (*nestedRequestHandler) HandleNotification(context.Context, string, json.Value) error { + return nil +} + +func TestSyncConnNestedRequestsAnswerInStackOrder(t *testing.T) { + t.Parallel() + synctest.Test(t, func(t *testing.T) { + handler := &nestedRequestHandler{ + aServing: make(chan struct{}), + bAtClient: make(chan struct{}), + releaseB: make(chan struct{}), + } + client := &stackClient{ + toServer: make(chan *ipc.Message), + toClient: make(chan *ipc.Message), + waiting: map[string]chan struct{}{`"b"`: handler.bAtClient}, + answers: map[string]string{}, + } + go client.run() + conn := ipc.NewSyncConn(nil, client, handler) + + var wg sync.WaitGroup + results := map[string]string{} + var resultsMu sync.Mutex + call := func(name string) { + result, err := conn.Call(t.Context(), "callback", json.Value(name)) + if err != nil { + t.Errorf("Call(callback, %s) = %v", name, err) + } + resultsMu.Lock() + results[name] = string(result) + resultsMu.Unlock() + } + wg.Go(func() { call(`"a"`) }) + <-handler.aServing + wg.Go(func() { call(`"b"`) }) + <-handler.bAtClient + synctest.Wait() + close(handler.releaseB) + wg.Wait() + close(client.toClient) + + want := map[string]string{`"a"`: `"a"`, `"b"`: `"b"`} + assert.DeepEqual(t, client.answers, want) + assert.DeepEqual(t, results, want) + }) +} From 55d745a4ab23187754622752068d78a3d12260be Mon Sep 17 00:00:00 2001 From: cplieger <917744+cplieger@users.noreply.github.com> Date: Wed, 7 Oct 2026 09:47:57 +0200 Subject: [PATCH 2/2] fix: track the sync connection stack with a call count and a flag --- tsc/internal/ipc/conn_sync.go | 77 +++++------- tsc/internal/ipc/conn_sync_test.go | 194 ++++++++++------------------- 2 files changed, 100 insertions(+), 171 deletions(-) diff --git a/tsc/internal/ipc/conn_sync.go b/tsc/internal/ipc/conn_sync.go index 1b1a8033f6360..cdbebc3a0338f 100644 --- a/tsc/internal/ipc/conn_sync.go +++ b/tsc/internal/ipc/conn_sync.go @@ -29,9 +29,9 @@ type SyncConn struct { // spawning goroutines that invoke filesystem callbacks) don't corrupt the stream. mu sync.Mutex - // inFlight holds the exchanges in progress, innermost last: true for a call to the client, false for a request from it. - inFlight []bool - changed sync.Cond + calls int + reading bool + turn sync.Cond } // NewSyncConn creates a new sync connection with the given transport and handler. @@ -41,27 +41,10 @@ func NewSyncConn(rwc io.ReadWriteCloser, protocol Protocol, handler Handler) *Sy protocol: protocol, handler: handler, } - c.changed.L = &c.mu + c.turn.L = &c.mu return c } -func (c *SyncConn) push(call bool) int { - for call && len(c.inFlight) > 0 && c.inFlight[len(c.inFlight)-1] { - c.changed.Wait() - } - c.inFlight = append(c.inFlight, call) - c.changed.Broadcast() - return len(c.inFlight) -} - -func (c *SyncConn) pop(depth int) { - for len(c.inFlight) != depth { - c.changed.Wait() - } - c.inFlight = c.inFlight[:depth-1] - c.changed.Broadcast() -} - // SetCollectTiming enables or disables per-request server processing-time // measurement. When enabled, the connection accumulates timing that clients can // retrieve via a getServerTiming request. @@ -93,7 +76,7 @@ func (c *SyncConn) Run(ctx context.Context) error { } if msg.IsRequest() { - if err := c.handleRequest(ctx, msg); err != nil { + if err := c.handleRequest(ctx, msg, 0); err != nil { return err } } else if msg.IsNotification() { @@ -105,25 +88,21 @@ func (c *SyncConn) Run(ctx context.Context) error { } } -// handleRequest processes an incoming request. -func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr error) { +func (c *SyncConn) lockTurn(depth int) { c.mu.Lock() - depth := c.push(false) - c.mu.Unlock() - answered := false - answer := func() { - if !answered { - answered = true - c.pop(depth) - } + for c.calls != depth { + c.turn.Wait() } + c.reading = depth > 0 +} +// handleRequest processes an incoming request. +func (c *SyncConn) handleRequest(ctx context.Context, msg *Message, depth int) (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 { case string(MethodGetServerTiming): - c.mu.Lock() - answer() + c.lockTurn(depth) writeErr := c.protocol.WriteResponse(msg.ID, serverTimingSnapshot(c.timing)) c.mu.Unlock() if writeErr != nil { @@ -134,8 +113,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro if c.timing != nil { c.timing.reset() } - c.mu.Lock() - answer() + c.lockTurn(depth) writeErr := c.protocol.WriteResponse(msg.ID, nil) c.mu.Unlock() if writeErr != nil { @@ -158,8 +136,7 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro stack := string(debug.Stack()) err = fmt.Errorf("panic: %v\n%s", r, stack) - c.mu.Lock() - answer() + c.lockTurn(depth) writeErr := c.protocol.WriteError(msg.ID, &jsonrpc.ResponseError{ Code: jsonrpc.CodeInternalError, Message: err.Error(), @@ -178,9 +155,8 @@ func (c *SyncConn) handleRequest(ctx context.Context, msg *Message) (retErr erro c.timing.record(msg.Method, time.Since(start)) } - c.mu.Lock() + c.lockTurn(depth) defer c.mu.Unlock() - answer() var writeErr error if err != nil { @@ -213,8 +189,17 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va // 3. We need to ensure write/read pairs are atomic c.mu.Lock() defer c.mu.Unlock() - call := c.push(true) - defer c.pop(call) + for c.reading { + c.turn.Wait() + } + c.calls++ + c.reading = true + depth := c.calls + defer func() { + c.calls-- + c.reading = false + c.turn.Broadcast() + }() id := jsonrpc.NewIDString(method) @@ -242,8 +227,10 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va if msg.IsRequest() { // A synchronous client callback may make a nested API request. Release // the protocol lock while handling it so nested callbacks can proceed. + c.reading = false + c.turn.Broadcast() c.mu.Unlock() - err := c.handleRequest(ctx, msg) + err := c.handleRequest(ctx, msg, depth) c.mu.Lock() if err != nil { return nil, err @@ -251,9 +238,11 @@ func (c *SyncConn) Call(ctx context.Context, method string, params any) (json.Va continue } if msg.IsNotification() { + c.reading = false + c.turn.Broadcast() c.mu.Unlock() c.handleNotification(ctx, msg) - c.mu.Lock() + c.lockTurn(depth) continue } return nil, fmt.Errorf("ipc: unexpected message while waiting for %q response", method) diff --git a/tsc/internal/ipc/conn_sync_test.go b/tsc/internal/ipc/conn_sync_test.go index f4c4ef745d49e..750798c20b42e 100644 --- a/tsc/internal/ipc/conn_sync_test.go +++ b/tsc/internal/ipc/conn_sync_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "io" + "net" "sync" "testing" "testing/synctest" @@ -81,149 +82,88 @@ func TestSyncConnRunReturnsPanicResponseWriteFailure(t *testing.T) { assert.ErrorContains(t, err, "original panic: handler panic") } -type syncPanickingResponseProtocol struct { - syncFailingResponseProtocol +type gatedHandler map[string]chan struct{} + +func (h gatedHandler) HandleRequest(_ context.Context, _ string, params json.Value) (any, error) { + <-h[string(params)] + return params, nil } -func (*syncPanickingResponseProtocol) WriteResponse(*jsonrpc.ID, any) error { - panic("response write panic") +func (gatedHandler) HandleNotification(context.Context, string, json.Value) error { + return nil } -func TestSyncConnRunReturnsResponseWritePanic(t *testing.T) { +func TestSyncConnAnswersNestedRequestsInStackOrder(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { - responseErr := errors.New("error write failed") - protocol := &syncPanickingResponseProtocol{syncFailingResponseProtocol{ - message: &ipc.Message{ID: jsonrpc.NewIDInt(1), Method: "transform"}, - responseErr: responseErr, - }} - conn := ipc.NewSyncConn(nil, protocol, noOpHandler{}) - - err := conn.Run(t.Context()) - assert.Assert(t, errors.Is(err, responseErr), "expected panic response write error, got %v", err) - assert.ErrorContains(t, err, "original panic: response write panic") - }) -} - -type stackClient struct { - toServer chan *ipc.Message - toClient chan *ipc.Message - waiting map[string]chan struct{} - answers map[string]string -} + server, client := net.Pipe() + handler := gatedHandler{`"a"`: make(chan struct{}), `"b"`: make(chan struct{})} + conn := ipc.NewSyncConn(server, ipc.NewJSONRPCProtocol(server), handler) + peer := ipc.NewJSONRPCProtocol(client) + answers := map[string]string{} + var serve func() json.Value + serve = func() json.Value { + for { + msg, err := peer.ReadMessage() + if err != nil { + return nil + } + if msg.IsResponse() { + return msg.Result + } + _ = peer.WriteRequest(jsonrpc.NewIDString("resolve"), "resolve", msg.Params) + answers[string(msg.Params)] = string(serve()) + _ = peer.WriteResponse(msg.ID, msg.Params) + } + } + go serve() -func (c *stackClient) ReadMessage() (*ipc.Message, error) { - return <-c.toServer, nil -} + var wg sync.WaitGroup + var a, b json.Value + wg.Go(func() { a, _ = conn.Call(t.Context(), "callback", json.Value(`"a"`)) }) + synctest.Wait() + wg.Go(func() { b, _ = conn.Call(t.Context(), "callback", json.Value(`"b"`)) }) + synctest.Wait() + close(handler[`"a"`]) + synctest.Wait() + close(handler[`"b"`]) + wg.Wait() + assert.NilError(t, client.Close()) -func (c *stackClient) WriteRequest(id *jsonrpc.ID, method string, params any) error { - data, err := json.Marshal(params) - c.toClient <- &ipc.Message{ID: id, Method: method, Params: data} - return err + assert.DeepEqual(t, answers, map[string]string{`"a"`: `"a"`, `"b"`: `"b"`}) + assert.Equal(t, string(a), `"a"`) + assert.Equal(t, string(b), `"b"`) + }) } -func (*stackClient) WriteNotification(string, any) error { - return nil +type callingNotificationHandler struct { + noOpHandler + conn *ipc.SyncConn } -func (c *stackClient) WriteResponse(id *jsonrpc.ID, result any) error { - data, err := json.Marshal(result) - c.toClient <- &ipc.Message{ID: id, Result: data} +func (h *callingNotificationHandler) HandleNotification(ctx context.Context, _ string, _ json.Value) error { + _, err := h.conn.Call(ctx, "inner", nil) return err } -func (c *stackClient) WriteError(id *jsonrpc.ID, responseErr *jsonrpc.ResponseError) error { - c.toClient <- &ipc.Message{ID: id, Error: responseErr} - return nil -} - -func (c *stackClient) run() { - for msg := range c.toClient { - c.handle(msg) - } -} - -func (c *stackClient) handle(callback *ipc.Message) { - c.toServer <- &ipc.Message{ID: jsonrpc.NewIDString("resolve"), Method: "resolve", Params: callback.Params} - if waiting := c.waiting[string(callback.Params)]; waiting != nil { - close(waiting) - } - c.answers[string(callback.Params)] = string(c.await()) - c.toServer <- &ipc.Message{ID: callback.ID, Result: callback.Params} -} - -func (c *stackClient) await() json.Value { - for msg := range c.toClient { - if msg.IsRequest() { - c.handle(msg) - continue - } - return msg.Result - } - return nil -} - -type nestedRequestHandler struct { - aServing chan struct{} - bAtClient chan struct{} - releaseB chan struct{} -} - -func (h *nestedRequestHandler) HandleRequest(_ context.Context, _ string, params json.Value) (any, error) { - switch string(params) { - case `"a"`: - close(h.aServing) - <-h.bAtClient - case `"b"`: - <-h.releaseB - } - return params, nil -} - -func (*nestedRequestHandler) HandleNotification(context.Context, string, json.Value) error { - return nil -} - -func TestSyncConnNestedRequestsAnswerInStackOrder(t *testing.T) { +func TestSyncConnNotificationHandlerCanCall(t *testing.T) { t.Parallel() synctest.Test(t, func(t *testing.T) { - handler := &nestedRequestHandler{ - aServing: make(chan struct{}), - bAtClient: make(chan struct{}), - releaseB: make(chan struct{}), - } - client := &stackClient{ - toServer: make(chan *ipc.Message), - toClient: make(chan *ipc.Message), - waiting: map[string]chan struct{}{`"b"`: handler.bAtClient}, - answers: map[string]string{}, - } - go client.run() - conn := ipc.NewSyncConn(nil, client, handler) - - var wg sync.WaitGroup - results := map[string]string{} - var resultsMu sync.Mutex - call := func(name string) { - result, err := conn.Call(t.Context(), "callback", json.Value(name)) - if err != nil { - t.Errorf("Call(callback, %s) = %v", name, err) - } - resultsMu.Lock() - results[name] = string(result) - resultsMu.Unlock() - } - wg.Go(func() { call(`"a"`) }) - <-handler.aServing - wg.Go(func() { call(`"b"`) }) - <-handler.bAtClient - synctest.Wait() - close(handler.releaseB) - wg.Wait() - close(client.toClient) - - want := map[string]string{`"a"`: `"a"`, `"b"`: `"b"`} - assert.DeepEqual(t, client.answers, want) - assert.DeepEqual(t, results, want) + server, client := net.Pipe() + handler := &callingNotificationHandler{} + conn := ipc.NewSyncConn(server, ipc.NewJSONRPCProtocol(server), handler) + handler.conn = conn + peer := ipc.NewJSONRPCProtocol(client) + go func() { + outer, _ := peer.ReadMessage() + _ = peer.WriteNotification("notify", nil) + inner, _ := peer.ReadMessage() + _ = peer.WriteResponse(inner.ID, inner.Method) + _ = peer.WriteResponse(outer.ID, outer.Method) + }() + + result, err := conn.Call(t.Context(), "outer", nil) + assert.NilError(t, err) + assert.Equal(t, string(result), `"outer"`) }) }