Merge pull request 'Stop the stdio channel holding its lock across a pipe write (issue 1286)' (#41) from fix/1286-refused-subscribe-deadlock into main

go/stdio.go

@@ -57,7 +57,11 @@ type message struct {
 // channel is the one conversation with the runtime: what this bundle writes, serialised, and what
 // it asked and waits for.
 type channel struct {
-	mu        sync.Mutex
+	mu sync.Mutex
+	// wmu orders the writes to the runtime and nothing else: a write waits on the pipe for as long as
+	// the runtime does not read, and mu must never wait with it, or the loop that reads the runtime's
+	// answers stops (issue 1286).
+	wmu       sync.Mutex
 	enc       *json.Encoder
 	next      int
 	waiting   map[string]chan message
@@ -90,8 +94,8 @@ type Envelope struct {
 
 func (c *channel) say(m map[string]any) {
 	m["jsonrpc"] = "2.0"
-	c.mu.Lock()
+	c.wmu.Lock()
-	defer c.mu.Unlock()
+	defer c.wmu.Unlock()
 	_ = c.enc.Encode(m)
 }
 

go/stdio_test.go

@@ -224,3 +224,54 @@ func TestARefusedSubscribeLeavesNoHandlerBehind(t *testing.T) {
 		t.Fatalf("one event was handled %d times", handled)
 	}
 }
+
+// A message the runtime has not read yet does not stop the channel taking its answers: with one tool
+// answer blocked on the pipe, a refused subscribe is still delivered to Subscribe, which returns the
+// refusal — no lock of the channel is held across a write to the pipe (novox/hq issue 1286/AC1, 1286/AC2).
+func TestARefusedSubscribeIsDeliveredWhileAMessageWaitsOnThePipe(t *testing.T) {
+	t.Setenv(ServedAs, "plex")
+	inR, inW := io.Pipe()
+	outR, outW := io.Pipe()
+	defer func() { _ = inW.Close() }()
+	// Not the channel an earlier run left behind: Subscribe goes to the one this run serves.
+	currentMu.Lock()
+	current = nil
+	currentMu.Unlock()
+	go func() {
+		_ = ServeOn("", []Tool{{Name: "quick", Run: func(map[string]any) (any, error) { return "ok", nil }}}, inR, outW)
+	}()
+	subscribed := make(chan error, 1)
+	go func() {
+		for {
+			if _, err := live(); err == nil {
+				break
+			}
+			time.Sleep(time.Millisecond)
+		}
+		subscribed <- Subscribe("*.download.completed", func(Envelope) error { return nil })
+	}()
+	go func() {
+		lines := bufio.NewScanner(outR)
+		if !lines.Scan() {
+			return
+		}
+		var req map[string]any
+		_ = json.Unmarshal(lines.Bytes(), &req)
+		send := func(m map[string]any) { b, _ := json.Marshal(m); _, _ = inW.Write(append(b, '\n')) }
+		// A tool answer the runtime has not read: its write waits on the pipe, and no one reads further.
+		send(map[string]any{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": map[string]any{"name": "quick"}})
+		time.Sleep(100 * time.Millisecond)
+		send(map[string]any{"jsonrpc": "2.0", "id": req["id"], "error": map[string]any{"code": -32000, "message": "not yet"}})
+	}()
+	deadline := time.After(2 * time.Second)
+	select {
+	case err := <-subscribed:
+		if err == nil || err.Error() != "not yet" {
+			t.Fatalf("the refused subscribe said %v", err)
+		}
+	case <-deadline:
+		t.Fatal("Subscribe did not return in 2 s: the channel is stuck on a write to a pipe nobody reads")
+	}
+	// Let the waiting answer go, so nothing is left blocked.
+	go func() { _, _ = io.Copy(io.Discard, outR) }()
+}