From 49e34ed33ca3217ff6c01808321c9c2b2532fed3 Mon Sep 17 00:00:00 2001 From: Chen Kai <281165273grape@gmail.com> Date: Tue, 9 Jun 2026 15:52:45 +0800 Subject: [PATCH] sync: never close the per-request handler channel (fix responsesWorker panic) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit responsesWorker reads the handler channel under handlersMu, releases the lock, then sends `ch <- res` unlocked. makeRequest's ctx-cancellation goroutine, at teardown, took handlersMu and close()d that same channel — so a late response made responsesWorker send on a closed channel and panic, crashing the process (hit reliably by the testground daemon at multi-instance run teardown). A select{ case ch<-res: case <-done: } would NOT fix it: a send case on a closed channel is still selectable and panics. The only race-free fix is to never close ch from the cancel goroutine. Refactor: - Each handler is a pendingRequest{ ch, done }. The cancel goroutine deletes the handler and close(done); it never closes ch (responsesWorker is ch's only sender). - responsesWorker does select{ case h.ch<-res: case <-h.done: } — never panics (ch is never closed) and never blocks forever (done fires at teardown). - Since ch is no longer closed, one-shot receivers (publish, SignalEntry, Barrier) unblock via context instead of channel-close, via a new awaitResponse helper that selects on the response, the request ctx, and the client ctx. - subscribe already selects on its contexts; unchanged. go build / go vet / gofmt clean. --- sync/client.go | 13 +++++++++++-- sync/client_conn.go | 33 +++++++++++++++++++++++++++------ sync/client_pubsub.go | 6 +++--- sync/client_state.go | 12 ++++++------ 4 files changed, 47 insertions(+), 17 deletions(-) diff --git a/sync/client.go b/sync/client.go index cca05fe..3bd5b68 100644 --- a/sync/client.go +++ b/sync/client.go @@ -33,10 +33,19 @@ type DefaultClient struct { nextMu sync.Mutex next int handlersMu sync.Mutex - handlers map[string]chan *tgsync.Response + handlers map[string]*pendingRequest socket *websocket.Conn } +// pendingRequest is the per-request response handler. done is closed by the +// request's cancellation goroutine to tell responsesWorker to stop delivering; +// the channel ch is NEVER closed (responsesWorker is its only sender), so a +// late response can never panic on send-to-closed. +type pendingRequest struct { + ch chan *tgsync.Response + done chan struct{} +} + // NewBoundClient returns a new sync DefaultClient that is bound to the provided // RunEnv. All operations will be automatically scoped to the keyspace of that // run. @@ -99,7 +108,7 @@ func newClient(ctx context.Context, log *zap.SugaredLogger, extractor func(ctx c cancel: cancel, log: log, extractor: extractor, - handlers: map[string]chan *tgsync.Response{}, + handlers: map[string]*pendingRequest{}, } c.sugarOperations = &sugarOperations{c} diff --git a/sync/client_conn.go b/sync/client_conn.go index b1cc22d..584be88 100644 --- a/sync/client_conn.go +++ b/sync/client_conn.go @@ -32,15 +32,21 @@ func (c *DefaultClient) responsesWorker() { c.log.Fatalw("error while reading socket", "error", err) } - var ch chan *sync.Response + var h *pendingRequest c.handlersMu.Lock() - ch = c.handlers[res.ID] + h = c.handlers[res.ID] c.handlersMu.Unlock() - if ch == nil { + if h == nil { c.log.Warnf("no handler available for response: %s", res.ID) } else { - ch <- res + // Deliver, or drop if the request was torn down. h.ch is never closed, + // so this can never panic on send-to-closed; h.done guarantees we never + // block forever on a receiver that has gone away. + select { + case h.ch <- res: + case <-h.done: + } } } @@ -57,9 +63,10 @@ func (c *DefaultClient) makeRequest(ctx context.Context, req *sync.Request) (cha } ch := make(chan *sync.Response) + done := make(chan struct{}) c.handlersMu.Lock() - c.handlers[req.ID] = ch + c.handlers[req.ID] = &pendingRequest{ch: ch, done: done} c.handlersMu.Unlock() err := c.writeSocket(req) @@ -77,15 +84,29 @@ func (c *DefaultClient) makeRequest(ctx context.Context, req *sync.Request) (cha } c.handlersMu.Lock() - close(c.handlers[req.ID]) delete(c.handlers, req.ID) c.handlersMu.Unlock() + close(done) // tell responsesWorker to stop; ch is intentionally NOT closed c.wg.Done() }() return ch, nil } +// awaitResponse waits for the single response to a one-shot request, or for the +// request or client context to be cancelled. The handler channel is never closed, +// so the context is the only unblock. +func (c *DefaultClient) awaitResponse(ctx context.Context, ch chan *sync.Response) (*sync.Response, error) { + select { + case res := <-ch: + return res, nil + case <-ctx.Done(): + return nil, ctx.Err() + case <-c.ctx.Done(): + return nil, errors.New("client closed before getting response") + } +} + func (c *DefaultClient) readSocket() (*sync.Response, error) { // After one hour without receiving information from the sync service, // the test will inevitably fail. Note(hacdias): consider changing diff --git a/sync/client_pubsub.go b/sync/client_pubsub.go index c5696fe..dfcf3e7 100644 --- a/sync/client_pubsub.go +++ b/sync/client_pubsub.go @@ -22,9 +22,9 @@ func (c *DefaultClient) publish(ctx context.Context, topic string, payload inter return -1, err } - res, ok := <-ch - if !ok { - return -1, errors.New("channel closed before getting response") + res, err := c.awaitResponse(ctx, ch) + if err != nil { + return -1, err } if res.Error != "" { return -1, errors.New(res.Error) diff --git a/sync/client_state.go b/sync/client_state.go index 06ffa8c..15ceecf 100644 --- a/sync/client_state.go +++ b/sync/client_state.go @@ -59,9 +59,9 @@ func (c *DefaultClient) Barrier(ctx context.Context, state State, target int) (* } go func() { - res, ok := <-ch - if !ok { - b.C <- errors.New("channel closed before getting response") + res, err := c.awaitResponse(ctx, ch) + if err != nil { + b.C <- err } else if res.Error == "" { b.C <- nil } else { @@ -99,9 +99,9 @@ func (c *DefaultClient) SignalEntry(ctx context.Context, state State) (int64, er return -1, err } - res, ok := <-ch - if !ok { - return -1, errors.New("channel closed before getting response") + res, err := c.awaitResponse(ctx, ch) + if err != nil { + return -1, err } if res.Error != "" { return -1, errors.New(res.Error)