From 861e418669f0c283a41e6db873897d7b9135277f Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 3 Jul 2026 11:03:14 +0200 Subject: [PATCH] worker+voice: per-conn context, store error hygiene, StateDir wiring, pipe leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit worker/server.go: dispatch now receives a per-connection context instead of context.Background(), so handler cancellation propagates on conn close. voice/server.go: same — per-conn context fed through safeDispatch into HandlePushToTalk instead of context.Background(). store/reminders.go: propagate LastInsertId error. store/nudges.go: propagate LastInsertId and RowsAffected errors. store/tools.go: propagate RowsAffected error. config/config.go: applyDefaults now respects StateDir when set, using it as the base for empty DBPath/SocketPath instead of silently ignoring it. phraser/llmphraser.go: close stderr pipe fd when cmd.Start() fails. --- internal/config/config.go | 22 +++++++++++++++++----- internal/phraser/llmphraser.go | 1 + internal/store/nudges.go | 10 ++++++++-- internal/store/reminders.go | 5 ++++- internal/store/tools.go | 5 ++++- internal/voice/server.go | 8 +++++--- internal/worker/server.go | 14 ++++++++------ 7 files changed, 47 insertions(+), 18 deletions(-) diff --git a/internal/config/config.go b/internal/config/config.go index 18ca515..7cfa417 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -256,11 +256,23 @@ func (c *Config) applyDefaults() { if c.AutotuneInterval == 0 { c.AutotuneInterval = Duration(DefaultAutotuneInterval) } - if c.DBPath == "" { - c.DBPath = filepath.Join(defaultDataDir(), "maven.db") - } - if c.SocketPath == "" { - c.SocketPath = filepath.Join(defaultRuntimeDir(), "mavend.sock") + // StateDir — when set, use it as the base for both db and socket if their + // paths are still relative (empty). If StateDir is empty, fall back to the + // XDG-style defaults (data dir for db, runtime dir for socket). + if c.StateDir != "" { + if c.DBPath == "" { + c.DBPath = filepath.Join(c.StateDir, "maven.db") + } + if c.SocketPath == "" { + c.SocketPath = filepath.Join(c.StateDir, "mavend.sock") + } + } else { + if c.DBPath == "" { + c.DBPath = filepath.Join(defaultDataDir(), "maven.db") + } + if c.SocketPath == "" { + c.SocketPath = filepath.Join(defaultRuntimeDir(), "mavend.sock") + } } } diff --git a/internal/phraser/llmphraser.go b/internal/phraser/llmphraser.go index 2b56ccc..40b00be 100644 --- a/internal/phraser/llmphraser.go +++ b/internal/phraser/llmphraser.go @@ -89,6 +89,7 @@ func (p *LLMPhraser) start(ctx context.Context) error { } if err := cmd.Start(); err != nil { + stderr.Close() return fmt.Errorf("llm: start: %w", err) } diff --git a/internal/store/nudges.go b/internal/store/nudges.go index bbf02ee..2d4d65c 100644 --- a/internal/store/nudges.go +++ b/internal/store/nudges.go @@ -42,7 +42,10 @@ func (s *Store) RecordNudge(ctx context.Context, rule, channel, message string, if err != nil { return 0, fmt.Errorf("record nudge: %w", err) } - id, _ := res.LastInsertId() + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("record nudge: last insert id: %w", err) + } return id, nil } @@ -62,7 +65,10 @@ func (s *Store) ResolveNudge(ctx context.Context, id int64, outcome string, ts t if err != nil { return fmt.Errorf("resolve nudge: %w", err) } - n, _ := res.RowsAffected() + n, err := res.RowsAffected() + if err != nil { + return fmt.Errorf("resolve nudge: rows affected: %w", err) + } if n == 0 { // either no such row, or it was already resolved — distinguish so callers // can tell a bug from a race. diff --git a/internal/store/reminders.go b/internal/store/reminders.go index a8f296f..8f0dcf0 100644 --- a/internal/store/reminders.go +++ b/internal/store/reminders.go @@ -34,7 +34,10 @@ func (s *Store) CreateReminder(ctx context.Context, fire time.Time, payload stri if err != nil { return 0, fmt.Errorf("create reminder: %w", err) } - id, _ := res.LastInsertId() + id, err := res.LastInsertId() + if err != nil { + return 0, fmt.Errorf("create reminder: last insert id: %w", err) + } return id, nil } diff --git a/internal/store/tools.go b/internal/store/tools.go index b391965..3cc47da 100644 --- a/internal/store/tools.go +++ b/internal/store/tools.go @@ -43,7 +43,10 @@ func (s *Store) ProposeTool(ctx context.Context, name, utterance string, ts time if err != nil { return false, fmt.Errorf("propose tool: %w", err) } - n, _ := res.RowsAffected() + n, err := res.RowsAffected() + if err != nil { + return false, fmt.Errorf("propose tool: rows affected: %w", err) + } return n > 0, nil } diff --git a/internal/voice/server.go b/internal/voice/server.go index 00c9c04..7bbf111 100644 --- a/internal/voice/server.go +++ b/internal/voice/server.go @@ -123,6 +123,8 @@ func (s *Server) Serve() error { // on EOF / read error / ctx cancel. func (s *Server) serveConn(c net.Conn) { defer c.Close() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() // TODO(step-up): the auth handshake populates the surface from mTLS / // passkey enrollment. Today the floor sets SurfacePCClient (the // reference client's surface, capped at L3 per auth.MaxLayer). The @@ -145,7 +147,7 @@ func (s *Server) serveConn(c net.Conn) { } s.sessions.Touch(sess.ID, time.Now()) - result, err := s.safeDispatch(c.RemoteAddr(), sess.ID, req) + result, err := s.safeDispatch(ctx, c.RemoteAddr(), sess.ID, req) resp := Response{ID: req.ID} if err != nil { resp.Error = rpcErr(err) @@ -159,13 +161,13 @@ func (s *Server) serveConn(c net.Conn) { } } -func (s *Server) safeDispatch(addr net.Addr, sid uint64, req Request) (result json.RawMessage, err error) { +func (s *Server) safeDispatch(ctx context.Context, addr net.Addr, sid uint64, req Request) (result json.RawMessage, err error) { defer func() { if r := recover(); r != nil { err = fmt.Errorf("voice: panic dispatching %s (sid %d): %v", req.Method, sid, r) } }() - return s.dispatch(context.Background(), sid, req) + return s.dispatch(ctx, sid, req) } func (s *Server) dispatch(ctx context.Context, sid uint64, req Request) (json.RawMessage, error) { diff --git a/internal/worker/server.go b/internal/worker/server.go index 58f2074..2682fcc 100644 --- a/internal/worker/server.go +++ b/internal/worker/server.go @@ -115,12 +115,14 @@ func (srv *Server) Serve() error { func (srv *Server) serveConn(c net.Conn) { id := srv.connCount.Add(1) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() for { var req Request if err := readFrame(c, &req); err != nil { return // EOF / malformed ⇒ end this conn } - result, err := srv.safeDispatch(c.RemoteAddr(), id, req) + result, err := srv.safeDispatch(ctx, c.RemoteAddr(), id, req) resp := Response{} if err != nil { resp.Error = rpcErr(err) @@ -133,16 +135,16 @@ func (srv *Server) serveConn(c net.Conn) { } } -func (srv *Server) safeDispatch(addr net.Addr, id uint64, req Request) (result json.RawMessage, err error) { +func (srv *Server) safeDispatch(ctx context.Context, addr net.Addr, id uint64, req Request) (result json.RawMessage, err error) { defer func() { if r := recover(); r != nil { err = fmt.Errorf("worker: panic dispatching %s (conn %d): %v", req.Method, id, r) } }() - return srv.dispatch(req) + return srv.dispatch(ctx, req) } -func (srv *Server) dispatch(req Request) (json.RawMessage, error) { +func (srv *Server) dispatch(ctx context.Context, req Request) (json.RawMessage, error) { switch req.Method { case MethodTranscribe: if srv.t == nil { @@ -152,7 +154,7 @@ func (srv *Server) dispatch(req Request) (json.RawMessage, error) { if err := unmarshalParams(req.Params, &p); err != nil { return nil, err } - resp, err := srv.t.Transcribe(context.Background(), p) + resp, err := srv.t.Transcribe(ctx, p) if err != nil { return nil, err } @@ -166,7 +168,7 @@ func (srv *Server) dispatch(req Request) (json.RawMessage, error) { if err := unmarshalParams(req.Params, &p); err != nil { return nil, err } - resp, err := srv.s.Synthesize(context.Background(), p) + resp, err := srv.s.Synthesize(ctx, p) if err != nil { return nil, err }