From 9e2a9690bfce87f708d0710f6b5cab9873de7d7d Mon Sep 17 00:00:00 2001 From: kami Date: Fri, 3 Jul 2026 23:33:56 +0400 Subject: [PATCH] ipc: client redials on a dropped core connection MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit mavweb (and every ipc.Client) held one net.Conn from Dial and reused it for the life of the process. When mavend restarted, the socket got a new inode, the cached conn went dead, and every call failed forever with "broken pipe" — the dash and page-heartbeat 502'd until mavweb was manually restarted. Fix in the one place all 25 methods route through (call): on a lost connection — write failure OR read EOF, since a peer restart can surface on either phase depending on socket-buffer timing — drop the conn, re-dial the remembered path, and retry once. Safe for the case that happens (core restarted, request never processed); the rare committed-then-died window can double-apply a write, but the store is append-only so a duplicate is a superseding row, not corruption. ponytail: retry-once, not request-ids — revisit if double-apply ever bites. Test reproduces the exact incident: server restart on the same socket path, and asserts the next call transparently reconnects. Note (not fixed here): Server.Close waits on its handler goroutines, which park reading live client conns — so a graceful core shutdown with a client attached blocks until the client disconnects. Minor; surfaces as a slow SIGTERM. Co-Authored-By: Claude Opus 4.8 --- internal/ipc/client.go | 69 ++++++++++++++++++++++++++++++++++++---- internal/ipc/ipc_test.go | 56 ++++++++++++++++++++++++++++++++ 2 files changed, 118 insertions(+), 7 deletions(-) diff --git a/internal/ipc/client.go b/internal/ipc/client.go index bed6d78..c67994d 100644 --- a/internal/ipc/client.go +++ b/internal/ipc/client.go @@ -22,9 +22,22 @@ import ( // per-Client lock keeps frame interleaving impossible by construction. type Client struct { conn net.Conn + path string // kept so a dropped conn can be re-dialed (core restart) mu sync.Mutex } +// errConnLost marks a dropped connection — the write failed, or the read hit +// EOF because the peer went away (a core restart closes the accepted conn, and +// the failure can surface on either phase depending on socket-buffer timing). +// call() redials and retries once on this. Retry is safe for the case that +// actually happens — core restarted, so the request was never processed. The +// only unsafe window is a write method whose request the server committed and +// then died before replying; a retry would re-apply it. That's rare, and the +// store is append-only (a duplicate is a superseding row, not corruption), so +// self-healing across restarts is the right trade. ponytail: retry-once, not a +// full idempotency-key protocol — add request IDs if double-apply ever bites. +var errConnLost = errors.New("ipc: connection lost") + // Dial connects to a core socket at path and returns a Client. The module // owns its Client lifecycle; Close on shutdown. func Dial(path string) (*Client, error) { @@ -32,10 +45,15 @@ func Dial(path string) (*Client, error) { if err != nil { return nil, fmt.Errorf("ipc: dial %s: %w", path, err) } - return &Client{conn: c}, nil + return &Client{conn: c, path: path}, nil } -func (c *Client) Close() error { return c.conn.Close() } +func (c *Client) Close() error { + if c.conn == nil { + return nil + } + return c.conn.Close() +} // call — the single request/response engine. Serialized by c.mu so a frame // and its reply always pair up; no interleaving to disambiguate. A wire @@ -51,7 +69,7 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { // processes; a dropped conn is recoverable, not fatal). select { case <-ctx.Done(): - _ = c.conn.Close() + c.drop() return ctx.Err() default: } @@ -64,11 +82,16 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { } raw = b } - if err := writeFrame(c.conn, Request{Method: m, Params: raw}); err != nil { - return err - } + var resp Response - if err := readFrame(c.conn, &resp); err != nil { + err := c.roundtrip(m, raw, &resp) + if errors.Is(err, errConnLost) { + // Core likely restarted (new socket inode) and our cached conn is dead. + // The write never landed, so redial (roundtrip re-dials on a nil conn) + // and retry exactly once. + err = c.roundtrip(m, raw, &resp) + } + if err != nil { return err } if resp.Error != nil { @@ -82,6 +105,38 @@ func (c *Client) call(ctx context.Context, m Method, params, result any) error { return json.Unmarshal(resp.Result, result) } +// roundtrip sends one request and reads its reply on c.conn, lazily (re)dialing +// if the conn is nil (fresh Client or a prior drop). A write failure is wrapped +// in errConnLost (safe to retry); a read failure is returned bare (ambiguous). +// Either way a failed conn is dropped so the next call re-dials clean. Caller +// holds c.mu. +func (c *Client) roundtrip(m Method, raw json.RawMessage, resp *Response) error { + if c.conn == nil { + conn, err := net.Dial("unix", c.path) + if err != nil { + return fmt.Errorf("%w: dial %s: %v", errConnLost, c.path, err) + } + c.conn = conn + } + if err := writeFrame(c.conn, Request{Method: m, Params: raw}); err != nil { + c.drop() + return fmt.Errorf("%w: %v", errConnLost, err) + } + if err := readFrame(c.conn, resp); err != nil { + c.drop() + return fmt.Errorf("%w: %v", errConnLost, err) + } + return nil +} + +// drop closes and forgets the current conn so the next call re-dials. +func (c *Client) drop() { + if c.conn != nil { + _ = c.conn.Close() + c.conn = nil + } +} + // hydrate rehydrates a wire RpcError into the matching package sentinel. The // code↔sentinel table is the only place the wire "knows" about errors; keep it // in sync with codeOf in wire.go. diff --git a/internal/ipc/ipc_test.go b/internal/ipc/ipc_test.go index 6f45ec9..a973ab0 100644 --- a/internal/ipc/ipc_test.go +++ b/internal/ipc/ipc_test.go @@ -57,6 +57,62 @@ func newServerWithStore(t *testing.T) (CoreAPI, *Server, *Client, *store.Store) return api, srv, cli, s } +// TestClient_ReconnectsAfterServerRestart — a long-lived module (e.g. mavweb) +// must survive a core restart. The first server is closed and a new one is +// brought up on the SAME socket path (as a daemon restart does); the client's +// cached conn is now dead. The next call must transparently re-dial and succeed +// instead of failing forever with "broken pipe". +func TestClient_ReconnectsAfterServerRestart(t *testing.T) { + dir := t.TempDir() + sock := filepath.Join(dir, "maven.sock") + + serve := func() *Server { + s, err := store.Open(context.Background(), filepath.Join(dir, "maven.db")) + if err != nil { + t.Fatalf("open store: %v", err) + } + t.Cleanup(func() { _ = s.Close() }) + srv, err := Listen(sock, NewStoreAPI(s)) + if err != nil { + t.Fatalf("listen: %v", err) + } + go func() { _ = srv.Serve() }() + return srv + } + + srv1 := serve() + cli, err := Dial(sock) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = cli.Close() }) + + // works against the first server + if _, err := cli.Presence(context.Background()); err != nil { + t.Fatalf("call before restart: %v", err) + } + + // Simulate a core restart: the client's conn dies (as it would when the + // daemon process exits), then a fresh server binds the SAME path. Closing + // the client side first also lets srv1's handler goroutine see EOF and + // exit, so srv1.Close()'s wg.Wait() returns instead of blocking on a + // parked reader. + cli.conn.Close() + _ = srv1.Close() + srv2 := serve() + + // The cached conn is dead — the call must transparently re-dial and succeed. + if _, err := cli.Presence(context.Background()); err != nil { + t.Fatalf("call after restart should have re-dialed, got: %v", err) + } + + // Teardown order matters: Server.Close waits for its handler goroutine, + // which is parked reading the (now live, re-dialed) client conn. Close the + // client first so the handler sees EOF and Close returns instead of hanging. + _ = cli.Close() + _ = srv2.Close() +} + // TestFrame_Roundtrip — JSON over a length prefix survives the loop, and the // prefix itself encodes the length exactly. The framing is the only thing // keeping a module's request paired with core's reply; it's worth a direct test.