diff --git a/internal/voice/client.go b/internal/voice/client.go index ca7260b..c8bf8c0 100644 --- a/internal/voice/client.go +++ b/internal/voice/client.go @@ -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,58 +117,69 @@ 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() + c.wmu.Lock() + err = writeFrame(conn, &req) + c.wmu.Unlock() + if err != nil { + c.forget(id) + c.teardownConn(conn) return err } - for { - resp, push, err := readOneFrame(conn) - if err != nil { - c.teardown() - return err - } - if push != nil { - c.deliverPush(*push) - continue - } - if resp.ID != id { - continue // not ours; ignore (singleplex ⇒ shouldn't happen) - } - if resp.Error != nil { - return hydrate(resp.Error) - } - if out != nil { - if err := json.Unmarshal(resp.Result, out); err != nil { - return fmt.Errorf("voice: unmarshal result: %w", err) - } - } - return nil + + 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) + } + if out != nil { + if err := json.Unmarshal(resp.Result, out); err != nil { + return fmt.Errorf("voice: unmarshal result: %w", err) + } + } + return nil } -// 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). +// 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 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() }() + select { + case <-ctx.Done(): + return ctx.Err() + case <-dead: + return nil + } +} + +// 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 { - select { - case <-ctx.Done(): - return ctx.Err() - default: - } - _, push, err := readOneFrame(conn) + resp, push, err := readOneFrame(conn) if err != nil { - if errors.Is(err, io.EOF) || errors.Is(err, net.ErrClosed) { - return nil - } - return err + 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 } diff --git a/internal/voice/voice_test.go b/internal/voice/voice_test.go index 3483352..57b77e5 100644 --- a/internal/voice/voice_test.go +++ b/internal/voice/voice_test.go @@ -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() +}