From 4fb8e887c276096f17dd3cd97e9535d950e94877 Mon Sep 17 00:00:00 2001 From: yetone Date: Tue, 6 Oct 2026 17:06:38 +0800 Subject: [PATCH] =?UTF-8?q?claude:=20a=20reply=20that=20fills=20its=20segm?= =?UTF-8?q?ent=20before=20the=20request=20reads=20it=20doesn't=20hold=20th?= =?UTF-8?q?e=20run's=20lock=20(CI=20Test=2037434913069=20on=208fbd39f8:=20?= =?UTF-8?q?TestClaudeLastWordsAreRead=20hung=2020m=20on=20macOS=20-race)?= =?UTF-8?q?=20=E2=80=94=20emit=20sent=20each=20event=20to=20the=20segment,?= =?UTF-8?q?=20a=2064-event=20channel,=20with=20r.mu=20held,=20and=20the=20?= =?UTF-8?q?request=20calls=20run.heard,=20which=20takes=20r.mu,=20before?= =?UTF-8?q?=20it=20starts=20reading=20the=20segment.=20A=20Claude=20Code?= =?UTF-8?q?=20that=20wrote=20more=20than=2064=20events=20first=20(the=20te?= =?UTF-8?q?st's=20300=20deltas,=20or=20a=20quick=20long=20reply)=20left=20?= =?UTF-8?q?emit=20blocked=20on=20the=20send=20holding=20the=20lock=20and?= =?UTF-8?q?=20heard=20waiting=20on=20it:=20the=20run=20hung=20for=20good,?= =?UTF-8?q?=20and=20abort,=20stderr=20and=20every=20other=20r.mu=20user=20?= =?UTF-8?q?with=20it.=20emit=20now=20takes=20the=20segment=20and=20updates?= =?UTF-8?q?=20shown=20under=20r.mu,=20lets=20it=20go,=20and=20sends=20unde?= =?UTF-8?q?r=20a=20lock=20of=20its=20own=20(sending);=20the=20two=20places?= =?UTF-8?q?=20that=20close=20a=20segment=20(endSegment,=20finish)=20take?= =?UTF-8?q?=20that=20lock=20first,=20so=20a=20send=20never=20meets=20a=20c?= =?UTF-8?q?losed=20channel.=20Events=20keep=20their=20order,=20emit=20bein?= =?UTF-8?q?g=20serialised=20by=20sending.=20TestEmitOnFullSegmentLeavesThe?= =?UTF-8?q?RunFree=20(200=20events=20emitted=20before=20anyone=20reads,=20?= =?UTF-8?q?then=20heard)=20fails=20on=20the=20old=20code=20(heard=20waits)?= =?UTF-8?q?=20and=20passes;=20it,=20TestClaudeLastWordsAreRead,=20TestClie?= =?UTF-8?q?ntGoneRecordedAsCanceled=20and=20Keepalive=20pass=20-race=20-co?= =?UTF-8?q?unt=3D20;=20the=20whole=20gateway=20package=20passes=20-race;?= =?UTF-8?q?=20go=20vet,=20linux(nogui)/windows=20builds=20pass.=20The=20ot?= =?UTF-8?q?her=20sends=20in=20the=20file=20(answered,=20the=20MCP=20waiter?= =?UTF-8?q?)=20already=20send=20after=20unlocking.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/gateway/claude_emit_test.go | 37 +++++++++++++++++++++++++ internal/gateway/claude_subscription.go | 29 +++++++++++++++---- 2 files changed, 60 insertions(+), 6 deletions(-) create mode 100644 internal/gateway/claude_emit_test.go diff --git a/internal/gateway/claude_emit_test.go b/internal/gateway/claude_emit_test.go new file mode 100644 index 00000000..6072205e --- /dev/null +++ b/internal/gateway/claude_emit_test.go @@ -0,0 +1,37 @@ +package gateway + +import ( + "testing" + "time" +) + +// A reply that fills the segment before the request reads it doesn't hold +// the run: the request's own heard, and so its reading, still go on (CI on +// 8fbd39f8: TestClaudeLastWordsAreRead hung 20m on macOS -race, emit +// blocked on a full segment with the run's lock held, heard waiting for it). +func TestEmitOnFullSegmentLeavesTheRunFree(t *testing.T) { + r := &subscriptionRun{} + ch := r.attach() + const n = 200 + go func() { + for i := 0; i < n; i++ { + r.emit(Event{Kind: KText, Text: "x"}) + } + r.endSegment() + }() + time.Sleep(50 * time.Millisecond) // the segment is full by now + heard := make(chan struct{}) + go func() { r.heard(nil); close(heard) }() + select { + case <-heard: + case <-time.After(5 * time.Second): + t.Fatal("heard waited on a send to a segment no one reads yet") + } + got := 0 + for range ch { + got++ + } + if got != n { + t.Fatalf("read %d events of %d", got, n) + } +} diff --git a/internal/gateway/claude_subscription.go b/internal/gateway/claude_subscription.go index b4265b46..e0c3f311 100644 --- a/internal/gateway/claude_subscription.go +++ b/internal/gateway/claude_subscription.go @@ -133,7 +133,11 @@ type subscriptionRun struct { // the tools its agent was told of, as it started and since tools map[string]bool - mu sync.Mutex + mu sync.Mutex + // sending is held across a send to the segment and taken before it is + // closed, so a send blocked on a full segment no one reads yet holds + // this, not mu: the request's heard, stderr and abort go on meanwhile + sending sync.Mutex segment chan Event pending map[string]chan mcpToolResult closed bool @@ -1717,9 +1721,11 @@ func (r *subscriptionRun) attach() chan Event { } func (r *subscriptionRun) emit(ev Event) { + r.sending.Lock() + defer r.sending.Unlock() r.mu.Lock() - defer r.mu.Unlock() - if r.segment != nil { + ch := r.segment + if ch != nil { switch ev.Kind { case KStart: // a reply begins r.shown = nil @@ -1730,8 +1736,19 @@ func (r *subscriptionRun) emit(ev Event) { r.shown[n-1].args += ev.Text } } - r.segment <- ev } + r.mu.Unlock() + if ch != nil { + ch <- ev + } +} + +// closeSegment closes a segment taken off the run, once no send to it is +// under way. +func (r *subscriptionRun) closeSegment(ch chan Event) { + r.sending.Lock() + close(ch) + r.sending.Unlock() } func (r *subscriptionRun) endSegment() { @@ -1743,7 +1760,7 @@ func (r *subscriptionRun) endSegment() { } r.mu.Unlock() if ch != nil { - close(ch) + r.closeSegment(ch) } } @@ -2796,7 +2813,7 @@ func (r *subscriptionRun) finish() { close(waiter) } if ch != nil { - close(ch) + r.closeSegment(ch) } if r.bridge != nil { // a run made in a test may have none r.bridge.removeRun(r)