One reader goroutine per voice conn, so a client can send and listen (V-671)

SendRequest and RunPushReceiver each read the conn, so a client that wanted both raced for every frame. A second listening conn is not the fix: it never sends a request, so its lastActive never moves and PushToMostRecent never picks it. mavwaked needs both on one conn. The reader now owns the socket for the life of the conn. It hands each Response to whichever SendRequest waits on that id, and each Push to the handler. SendRequest waits on its own channel, on the conn dying, on its context, or on a timeout, and forgets its slot on every path that leaves without an answer. RunPushReceiver just wires the handler and blocks. Connect opens the conn without sending anything, for a client that must hold a session before it has spoken. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013ptwopxyo3Z2kwFckHkLvN
2026-08-09 13:44:52 +04:00
parent ce91d20ac8
commit 1f1e002789
2 changed files with 228 additions and 65 deletions
@@ -9,15 +9,18 @@
 //
 // The wire is symmetric: a Request from the client is answered by a
 // Response with a matching ID, OR a server-initiated Push frame (no ID)
-// may arrive interleaved. SendRequest loops reading frames, drops Push
-// frames to the harness if a receiver is running (or silently if not),
-// and returns the first Response with the matching ID.
+// may arrive interleaved. One reader goroutine per connection owns the
+// socket. It hands each Response to whichever SendRequest is waiting on
+// that ID and each Push to the handler, so a client may send and listen
+// at the same time on one conn. mavwaked needs exactly that: it speaks
+// utterances and it must hear nudges, and the server routes a nudge to
+// the session that spoke most recently, so a second listening conn would
+// never be picked (V-671).
 package voice

 import (
 	"context"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -41,13 +44,22 @@ type PushHandler interface {
 // Client — one connection to the voice.Server.
 type Client struct {
 	addr   string
-	mu     sync.Mutex
-	c      net.Conn
 	nextID atomic.Uint64

-	// pushCh fan-out: a reader goroutine (started by RunPushReceiver)
-	// writes Push frames here; SendRequest also drains it when no reader
-	// is running (drops the frame in that case).
+	mu sync.Mutex
+	c  net.Conn
+	// pending holds one channel per in-flight request, keyed by frame id.
+	// The reader goroutine delivers the Response here and deletes the entry.
+	pending map[uint64]chan *Response
+	// dead is closed by the reader goroutine when this conn ends, so a
+	// waiting SendRequest fails at once instead of at its own deadline.
+	dead chan struct{}
+
+	// wmu serialises writes. Frames must not interleave on the wire.
+	wmu sync.Mutex
+
+	// pushH is set by RunPushReceiver and survives a reconnect, because the
+	// client that wants pushes wants them on whatever conn it ends up with.
 	pushMu sync.Mutex
 	pushH  PushHandler
 }
@@ -55,6 +67,15 @@ type Client struct {
 // Dial returns a Client that will connect to addr on first use.
 func Dial(addr string) *Client { return &Client{addr: addr} }

+// Connect opens the conn now rather than on the first request. mavwaked calls
+// it at startup: the server registers a session on accept, and a client that
+// has never connected cannot be sent a nudge.
+func (c *Client) Connect(ctx context.Context) error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.ensureConnLocked(ctx)
+}
+
 // Close releases the conn. Idempotent.
 func (c *Client) Close() error {
 	c.mu.Lock()
@@ -75,12 +96,14 @@ func (c *Client) PushToTalk(ctx context.Context, a audio.Audio, lang string) (Pu
 	return out, err
 }

+// requestTimeout bounds a round-trip with no deadline on its context. It is
+// generous because the far end runs speech-to-text, a router and a voice.
+const requestTimeout = 120 * time.Second
+
 // SendRequest sends one Request frame and waits for the matching Response.
-// Push frames received while waiting are dropped on the floor UNLESS a
-// PushHandler has been wired via RunPushReceiver, in which case the handler
-// is invoked inline (still synchronous with the SendRequest caller's
-// read). For sanity, the reference client runs either one-shot (no
-// receiver) or interactive (RunPushReceiver, no concurrent SendRequest).
+// Push frames arriving meanwhile go to the handler on the reader goroutine,
+// so listening and sending on one Client is supported rather than merely
+// tolerated.
 func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any) error {
 	body, err := marshalParams(params)
 	if err != nil {
@@ -94,33 +117,38 @@ func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any)
 		c.mu.Unlock()
 		return err
 	}
-	conn := c.c
+	conn, dead := c.c, c.dead
+	ch := make(chan *Response, 1)
+	c.pending[id] = ch
 	c.mu.Unlock()

-	if dl, ok := ctx.Deadline(); ok {
-		_ = conn.SetDeadline(dl)
-	} else {
-		_ = conn.SetDeadline(time.Now().Add(120 * time.Second))
-	}
-	defer conn.SetDeadline(time.Time{})
-
-	if err := writeFrame(conn, &req); err != nil {
-		c.teardown()
-		return err
-	}
-	for {
-		resp, push, err := readOneFrame(conn)
+	c.wmu.Lock()
+	err = writeFrame(conn, &req)
+	c.wmu.Unlock()
 	if err != nil {
-			c.teardown()
+		c.forget(id)
+		c.teardownConn(conn)
 		return err
 	}
-		if push != nil {
-			c.deliverPush(*push)
-			continue
-		}
-		if resp.ID != id {
-			continue // not ours; ignore (singleplex ⇒ shouldn't happen)
+
+	timer := time.NewTimer(requestTimeout)
+	defer timer.Stop()
+
+	var resp *Response
+	select {
+	case resp = <-ch:
+	case <-dead:
+		c.forget(id)
+		return fmt.Errorf("voice: connection closed before reply")
+	case <-ctx.Done():
+		c.forget(id)
+		return ctx.Err()
+	case <-timer.C:
+		c.forget(id)
+		c.teardownConn(conn)
+		return fmt.Errorf("voice: no reply within %s", requestTimeout)
 	}
+
 	if resp.Error != nil {
 		return hydrate(resp.Error)
 	}
@@ -131,21 +159,27 @@ func (c *Client) SendRequest(ctx context.Context, m Method, params any, out any)
 	}
 	return nil
 }
+
+// forget drops an abandoned request so a late Response is discarded rather
+// than delivered to nobody.
+func (c *Client) forget(id uint64) {
+	c.mu.Lock()
+	delete(c.pending, id)
+	c.mu.Unlock()
 }

-// RunPushReceiver spawns a reader goroutine that delivers Push frames to h
-// until the conn closes or Close is called. Today's reference client uses
-// this in -listen mode (proactive voice playback). SendRequest and
-// RunPushReceiver SHOULD NOT be used concurrently on the same Client — the
-// wire is singleplex at the reference client's scale; production picks one
-// mode per conn. Returns when the goroutine ends (ctx cancel or conn close).
+// RunPushReceiver wires h and blocks until the conn ends or ctx is
+// cancelled. Frames are read by the per-conn reader goroutine, so a client
+// may call SendRequest on the same Client while this is running. Returns nil
+// when the conn ended, so a caller that wants to stay reachable reconnects
+// and calls it again.
 func (c *Client) RunPushReceiver(ctx context.Context, h PushHandler) error {
 	c.mu.Lock()
 	if err := c.ensureConnLocked(ctx); err != nil {
 		c.mu.Unlock()
 		return err
 	}
-	conn := c.c
+	dead := c.dead
 	c.mu.Unlock()

 	c.pushMu.Lock()
@@ -158,21 +192,34 @@ func (c *Client) RunPushReceiver(ctx context.Context, h PushHandler) error {
 		c.pushMu.Unlock()
 	}()

-	for {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-		default:
-		}
-		_, push, err := readOneFrame(conn)
-		if err != nil {
-			if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) {
+	case <-dead:
 		return nil
 	}
-			return err
+}
+
+// readLoop owns conn for its whole life. It ends on any read error, which is
+// how a closed conn, a killed server and a cancelled dial all arrive here.
+func (c *Client) readLoop(conn net.Conn, dead chan struct{}) {
+	defer close(dead)
+	for {
+		resp, push, err := readOneFrame(conn)
+		if err != nil {
+			c.teardownConn(conn)
+			return
 		}
 		if push != nil {
 			c.deliverPush(*push)
+			continue
+		}
+		c.mu.Lock()
+		ch := c.pending[resp.ID]
+		delete(c.pending, resp.ID)
+		c.mu.Unlock()
+		if ch != nil {
+			ch <- resp
 		}
 	}
 }
@@ -196,13 +243,19 @@ func (c *Client) ensureConnLocked(ctx context.Context) error {
 		return fmt.Errorf("voice: dial %s: %w", c.addr, err)
 	}
 	c.c = conn
+	c.pending = make(map[uint64]chan *Response)
+	c.dead = make(chan struct{})
+	go c.readLoop(conn, c.dead)
 	return nil
 }

-func (c *Client) teardown() {
+// teardownConn closes conn and forgets it, but only if it is still the live
+// one. A reconnect may already have replaced it, and closing the new conn
+// because the old one died takes the client down on every hiccup.
+func (c *Client) teardownConn(conn net.Conn) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
-	if c.c != nil {
+	if c.c != nil && c.c == conn {
 		_ = c.c.Close()
 		c.c = nil
 	}
@@ -221,3 +221,113 @@ func TestClientListenModeReceivesPush(t *testing.T) {
 type pushHandlerFunc func(Push)

 func (f pushHandlerFunc) OnPush(p Push) { f(p) }
+
+// The whole point of the per-conn reader (V-671): mavwaked speaks utterances
+// and must hear nudges, and the server routes a nudge to the session that
+// spoke most recently. A second listening conn would never be picked, so both
+// directions have to share one conn.
+func TestClientSendsAndListensOnOneConn(t *testing.T) {
+	l := newTestListener(t)
+	sess := NewSessions()
+	h := &stubHandler{}
+	srv := NewServer(l.Addr().String(), h, sess)
+	if err := srv.Listen(); err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	defer func() {
+		_ = srv.Close()
+		waitPort()
+	}()
+	go func() { _ = srv.Serve() }()
+
+	c := Dial(l.Addr().String())
+	defer c.Close()
+
+	got := make(chan audio.Audio, 4)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		_ = c.RunPushReceiver(ctx, pushHandlerFunc(func(p Push) {
+			var ap AudioNudgePush
+			if err := json.Unmarshal(p.Params, &ap); err == nil {
+				got <- ap.Audio
+			}
+		}))
+	}()
+
+	for i := 0; i < 100 && sess.Active() < 1; i++ {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if sess.Active() != 1 {
+		t.Fatalf("active sessions = %d, want exactly 1", sess.Active())
+	}
+
+	// A round-trip while the receiver is running. Before the reader owned the
+	// conn, this and the receiver raced for every frame.
+	resp, err := c.PushToTalk(context.Background(), audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("hello")}, "ru")
+	if err != nil {
+		t.Fatalf("PushToTalk with a receiver running: %v", err)
+	}
+	if resp.ReplyText != "got it" {
+		t.Fatalf("ReplyText = %q, want %q", resp.ReplyText, "got it")
+	}
+
+	// And the nudge still arrives, on the session that just spoke.
+	err = sess.PushToMostRecent(context.Background(), AudioNudgePush{
+		RuleName: "after-speaking",
+		Audio:    audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("proactive")},
+	})
+	if err != nil {
+		t.Fatalf("PushToMostRecent: %v", err)
+	}
+	select {
+	case a := <-got:
+		if string(a.Bytes) != "proactive" {
+			t.Fatalf("received %q, want the nudge audio", string(a.Bytes))
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("nudge never reached the handler after the client had spoken")
+	}
+}
+
+// A request abandoned by its context must not leave its slot behind, or a
+// long-running client leaks one channel per timeout.
+func TestClientForgetsAbandonedRequests(t *testing.T) {
+	l := newTestListener(t)
+	sess := NewSessions()
+	srv := NewServer(l.Addr().String(), &blockingHandler{}, sess)
+	if err := srv.Listen(); err != nil {
+		t.Fatalf("listen: %v", err)
+	}
+	defer func() {
+		_ = srv.Close()
+		waitPort()
+	}()
+	go func() { _ = srv.Serve() }()
+
+	c := Dial(l.Addr().String())
+	defer c.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	defer cancel()
+	_, err := c.PushToTalk(ctx, audio.Audio{Format: audio.PCM16kMono, Bytes: []byte("x")}, "ru")
+	if err == nil {
+		t.Fatal("expected the round-trip to fail on its context")
+	}
+
+	c.mu.Lock()
+	n := len(c.pending)
+	c.mu.Unlock()
+	if n != 0 {
+		t.Fatalf("pending = %d after an abandoned request, want 0", n)
+	}
+}
+
+// blockingHandler never answers, so the client's context is what ends the
+// round-trip.
+type blockingHandler struct{}
+
+func (blockingHandler) HandlePushToTalk(ctx context.Context, _ PushToTalkReq, _ uint64) (PushToTalkResp, error) {
+	<-ctx.Done()
+	return PushToTalkResp{}, ctx.Err()
+}