fix(registry/client): idempotent Close; pool reconnect keeps its cause, logs at Debug, backs off, retries on a healthy conn

registry/client/binary_client.go

@@ -5,6 +5,7 @@ package client
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"net"
@@ -23,6 +24,13 @@ type BinaryClient struct {
 	addr   string
 	closed bool
 	dial   DialContextFunc // optional; see WithDialer
+
+	// closeOnce makes Close idempotent; done (created lazily by doneCh)
+	// is closed by Close to wake a reconnect backoff, which runs with mu
+	// held.
+	closeOnce sync.Once
+	doneOnce  sync.Once
+	done      chan struct{}
 }
 
 // DialBinary connects to a registry server and negotiates the binary wire protocol.
@@ -48,46 +56,82 @@ func DialBinary(addr string, opts ...DialOption) (*BinaryClient, error) {
 	return &BinaryClient{conn: conn, addr: addr, dial: o.dial}, nil
 }
 
-// Close shuts down the binary client connection.
+// Close shuts down the binary client connection. It is idempotent: the
+// teardown runs once, later calls return nil, and every call after Close
+// returns an error matching ErrClosed.
 func (c *BinaryClient) Close() error {
-	c.mu.Lock()
-	c.closed = true
-	conn := c.conn
-	c.mu.Unlock()
-	if conn != nil {
-		return conn.Close()
+	if c == nil {
+		return nil
+	}
+	var err error
+	c.closeOnce.Do(func() {
+		// Wake a reconnect backoff first: it holds c.mu.
+		close(c.doneCh())
+		c.mu.Lock()
+		c.closed = true
+		conn := c.conn
+		c.mu.Unlock()
+		if conn != nil {
+			if cerr := conn.Close(); cerr != nil && !errors.Is(cerr, net.ErrClosed) {
+				err = cerr
+			}
+		}
+	})
+	return err
+}
+
+func (c *BinaryClient) doneCh() chan struct{} {
+	c.doneOnce.Do(func() { c.done = make(chan struct{}) })
+	return c.done
+}
+
+// closedLocked reports whether Close has started. Must be called with c.mu held.
+func (c *BinaryClient) closedLocked() bool {
+	if c.closed {
+		return true
+	}
+	select {
+	case <-c.doneCh():
+		return true
+	default:
+		return false
 	}
-	return nil
 }
 
 // Addr returns the registry address this client is connected to.
 func (c *BinaryClient) Addr() string {
 	return c.addr
 }
 
-// reconnect re-establishes the binary connection. Must be called with c.mu held.
-func (c *BinaryClient) reconnect() error {
-	if c.closed {
-		return fmt.Errorf("client closed")
+// reconnect re-establishes the binary connection after cause broke it.
+// Must be called with c.mu held.
+func (c *BinaryClient) reconnect(cause error) error {
+	if c.closedLocked() {
+		return ErrClosed
 	}
 	if c.conn != nil {
 		c.conn.Close()
+		// Retired: a failed redial leaves no conn for Close to close twice.
+		c.conn = nil
 	}
 
-	backoff := 500 * time.Millisecond
-	maxBackoff := 10 * time.Second
+	backoff := defaultReconnectPolicy.dialBase
 	var lastErr error
 
-	for attempts := 0; attempts < 5; attempts++ {
+	for attempts := 0; attempts < maxReconnectAttempts; attempts++ {
+		if attempts > 0 {
+			// Close wakes this wait (it cannot take c.mu while we sleep).
+			if err := sleepCtx(context.Background(), c.doneCh(), jitter(backoff)); err != nil {
+				return err
+			}
+			if backoff *= 2; backoff > defaultReconnectPolicy.dialMax {
+				backoff = defaultReconnectPolicy.dialMax
+			}
+		}
 		conn, err := dialConn(context.Background(), c.dial, c.addr, nil)
 		if err != nil {
 			lastErr = err
-			slog.Warn("binary client reconnect failed", "attempt", attempts+1, "err", err)
-			time.Sleep(backoff)
-			backoff *= 2
-			if backoff > maxBackoff {
-				backoff = maxBackoff
-			}
+			slog.Warn("binary client reconnect failed", "attempt", attempts+1, "err", err, "cause", errText(cause))
 			continue
 		}
 
@@ -101,30 +145,50 @@ func (c *BinaryClient) reconnect() error {
 			continue
 		}
 
+		if c.closedLocked() {
+			// Close started while we were dialing and waits for c.mu.
+			conn.Close()
+			return ErrClosed
+		}
 		c.conn = conn
-		slog.Info("binary client reconnected", "addr", c.addr)
+		slog.Info("binary client reconnected", "addr", c.addr, "cause", errText(cause))
 		return nil
 	}
-	return fmt.Errorf("reconnect failed after 5 attempts: %w", lastErr)
+	return fmt.Errorf("reconnect failed after %d attempts: %w", maxReconnectAttempts, lastErr)
 }
 
 // Heartbeat sends a binary heartbeat and returns the server time and key expiry warning.
 func (c *BinaryClient) Heartbeat(nodeID uint32, sig []byte) (unixTime int64, keyExpiryWarning bool, err error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	if c.closedLocked() {
+		return 0, false, ErrClosed
+	}
 
 	unixTime, keyExpiryWarning, err = c.heartbeatLocked(nodeID, sig)
-	if err != nil && !c.closed {
+	if err != nil && !c.closedLocked() {
 		// Connection-level failure — reconnect and retry once
-		if reconnErr := c.reconnect(); reconnErr != nil {
-			return 0, false, fmt.Errorf("heartbeat failed and reconnect failed: %w", err)
+		if reconnErr := c.reconnect(err); reconnErr != nil {
+			return 0, false, binaryReconnectFailed("heartbeat", err, reconnErr)
 		}
 		unixTime, keyExpiryWarning, err = c.heartbeatLocked(nodeID, sig)
 	}
 	return
 }
 
+// binaryReconnectFailed keeps both the request error and the reconnect
+// error in the chain.
+func binaryReconnectFailed(op string, err, reconnErr error) error {
+	if errors.Is(reconnErr, ErrClosed) {
+		return closedDuring(err)
+	}
+	return fmt.Errorf("%s failed and reconnect failed: %w (reconnect: %w)", op, err, reconnErr)
+}
+
 func (c *BinaryClient) heartbeatLocked(nodeID uint32, sig []byte) (int64, bool, error) {
+	if c.conn == nil {
+		return 0, false, errNotConnected
+	}
 	if err := wire.WriteFrame(c.conn, wire.MsgHeartbeat, wire.EncodeHeartbeatReq(nodeID, sig)); err != nil {
 		return 0, false, fmt.Errorf("send heartbeat: %w", err)
 	}
@@ -150,18 +214,24 @@ func (c *BinaryClient) heartbeatLocked(nodeID uint32, sig []byte) (int64, bool,
 func (c *BinaryClient) Lookup(nodeID uint32) (*wire.LookupResult, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	if c.closedLocked() {
+		return nil, ErrClosed
+	}
 
 	result, err := c.lookupLocked(nodeID)
-	if err != nil && !c.closed {
-		if reconnErr := c.reconnect(); reconnErr != nil {
-			return nil, fmt.Errorf("lookup failed and reconnect failed: %w", err)
+	if err != nil && !c.closedLocked() {
+		if reconnErr := c.reconnect(err); reconnErr != nil {
+			return nil, binaryReconnectFailed("lookup", err, reconnErr)
 		}
 		result, err = c.lookupLocked(nodeID)
 	}
 	return result, err
 }
 
 func (c *BinaryClient) lookupLocked(nodeID uint32) (*wire.LookupResult, error) {
+	if c.conn == nil {
+		return nil, errNotConnected
+	}
 	if err := wire.WriteFrame(c.conn, wire.MsgLookup, wire.EncodeLookupReq(nodeID)); err != nil {
 		return nil, fmt.Errorf("send lookup: %w", err)
 	}
@@ -191,18 +261,24 @@ func (c *BinaryClient) lookupLocked(nodeID uint32) (*wire.LookupResult, error) {
 func (c *BinaryClient) Resolve(nodeID, requesterID uint32, sig []byte) (*wire.ResolveResult, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	if c.closedLocked() {
+		return nil, ErrClosed
+	}
 
 	result, err := c.resolveLocked(nodeID, requesterID, sig)
-	if err != nil && !c.closed {
-		if reconnErr := c.reconnect(); reconnErr != nil {
-			return nil, fmt.Errorf("resolve failed and reconnect failed: %w", err)
+	if err != nil && !c.closedLocked() {
+		if reconnErr := c.reconnect(err); reconnErr != nil {
+			return nil, binaryReconnectFailed("resolve", err, reconnErr)
 		}
 		result, err = c.resolveLocked(nodeID, requesterID, sig)
 	}
 	return result, err
 }
 
 func (c *BinaryClient) resolveLocked(nodeID, requesterID uint32, sig []byte) (*wire.ResolveResult, error) {
+	if c.conn == nil {
+		return nil, errNotConnected
+	}
 	if err := wire.WriteFrame(c.conn, wire.MsgResolve, wire.EncodeResolveReq(nodeID, requesterID, sig)); err != nil {
 		return nil, fmt.Errorf("send resolve: %w", err)
 	}
@@ -233,11 +309,14 @@ func (c *BinaryClient) resolveLocked(nodeID, requesterID uint32, sig []byte) (*w
 func (c *BinaryClient) SendJSON(msg map[string]interface{}) (map[string]interface{}, error) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
+	if c.closedLocked() {
+		return nil, ErrClosed
+	}
 
 	resp, err := c.sendJSONLocked(msg)
-	if err != nil && resp == nil && !c.closed {
-		if reconnErr := c.reconnect(); reconnErr != nil {
-			return nil, fmt.Errorf("send failed and reconnect failed: %w", err)
+	if err != nil && resp == nil && !c.closedLocked() {
+		if reconnErr := c.reconnect(err); reconnErr != nil {
+			return nil, binaryReconnectFailed("send", err, reconnErr)
 		}
 		resp, err = c.sendJSONLocked(msg)
 	}
@@ -249,6 +328,9 @@ func (c *BinaryClient) sendJSONLocked(msg map[string]interface{}) (map[string]in
 	if err != nil {
 		return nil, fmt.Errorf("json encode: %w", err)
 	}
+	if c.conn == nil {
+		return nil, errNotConnected
+	}
 
 	if err := wire.WriteFrame(c.conn, wire.MsgJSON, body); err != nil {
 		return nil, fmt.Errorf("send: %w", err)