ipc: client redials on a dropped core connection
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 <noreply@anthropic.com>
This commit is contained in:
+62
-7
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user