diff --git a/registry/client/binary_client.go b/registry/client/binary_client.go index 056e8b7..8d7ec6a 100644 --- a/registry/client/binary_client.go +++ b/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,16 +56,46 @@ 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. @@ -65,29 +103,35 @@ 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,11 +214,14 @@ 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) } @@ -162,6 +229,9 @@ func (c *BinaryClient) Lookup(nodeID uint32) (*wire.LookupResult, error) { } 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,11 +261,14 @@ 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) } @@ -203,6 +276,9 @@ func (c *BinaryClient) Resolve(nodeID, requesterID uint32, sig []byte) (*wire.Re } 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) diff --git a/registry/client/client.go b/registry/client/client.go index 1568520..31219a2 100644 --- a/registry/client/client.go +++ b/registry/client/client.go @@ -57,8 +57,9 @@ const dialTimeout = 5 * time.Second // client's connections (initial, pool and reconnect) through a custom // dialer, e.g. an HTTP CONNECT proxy from the netproxy package. type Client struct { - // Primary connection. Always present; tests in this package read - // c.conn / c.mu / c.closed directly so the field set must stay stable. + // Primary connection. nil only while a dropped connection waits for + // its redial. Tests in this package read c.conn / c.mu / c.closed + // directly so the field set must stay stable. conn net.Conn mu sync.Mutex addr string // registry address for reconnection @@ -69,15 +70,33 @@ type Client struct { // instead of a direct net.Dialer — see WithDialer. Immutable after // construction. dial DialContextFunc + // connDialedAt is when c.conn was established (single-conn path; guarded + // by c.mu). Its age tells a rate-limit close from an idle one. + connDialedAt time.Time // Optional pool of secondary connections used to parallelise Send. // nil / empty when DialPool was not used. pool poolState + + // closeOnce makes Close idempotent: the teardown (closing done and every + // connection) runs exactly once however many goroutines call Close. + closeOnce sync.Once + // done is closed by Close. It wakes goroutines parked on the pool's + // free list or in a reconnect backoff. Created lazily by doneCh so the + // zero-value Client works. + doneOnce sync.Once + done chan struct{} + + // reconn holds the reconnect backoff streaks and log counters shared by + // every connection of this client. + reconn reconnectTracker + // logger overrides slog.Default() (tests only). + logger *slog.Logger } // poolState holds the secondary-conn pool. The primary slot (c.conn / c.mu) -// is also represented here as the first entry, so acquireConn / releaseConn -// can pick uniformly across all conns. +// is also represented here as the first entry, so acquireEntry / +// releaseEntry can pick uniformly across all conns. type poolState struct { // entries is the full set of conns including the primary at index 0. // Each entry has its own mu — taking entry.mu lets one Send proceed @@ -87,20 +106,31 @@ type poolState struct { // free is a buffered channel of pointers to entries currently free. // Capacity equals len(entries). Send: <-free; defer free<-entry. // nil means "no pool" (legacy single-conn path via c.mu). + // It is never closed: Close closes c.done instead, which avoids the + // race between close(free) and concurrent sends on free. free chan *pooledConn - // done is closed by Close() to wake goroutines blocked on <-free and to - // signal the deferred pool-return in sendPool to drop its entry instead - // of sending on free. Using a separate done channel avoids the race - // between close(free) and concurrent sends on free. - done chan struct{} } -// pooledConn wraps one TCP connection plus its own mutex. The mutex -// guards both the conn pointer and any reconnect that happens through -// it; sendOnEntry takes it for the full write/read round-trip. +// pooledConn wraps one registry connection plus its own mutex. The Send +// that took the entry off the free list holds mu for its whole round trip +// (including any redial). conn is written only with both mu and the +// client's c.mu held, so Close can snapshot it under c.mu alone. type pooledConn struct { mu sync.Mutex conn net.Conn + // dialedAt is when conn was established (guarded by mu). + dialedAt time.Time + // broken, when non-nil, is the error that dropped conn; the entry must + // be redialed before reuse (guarded by mu). + broken error +} + +// healthy reports whether the entry can be used without a redial. The +// caller must own the entry (it came off the free list). +func (e *pooledConn) healthy() bool { + e.mu.Lock() + defer e.mu.Unlock() + return e.broken == nil && e.conn != nil } // SetSigner sets a signing function for authenticated registry operations (H3 fix). @@ -148,7 +178,12 @@ func Dial(addr string, opts ...DialOption) (*Client, error) { if err != nil { return nil, fmt.Errorf("dial registry: %w", err) } - return &Client{conn: conn, addr: addr, dial: o.dial}, nil + return newClient(conn, addr, nil, o.dial), nil +} + +// newClient wraps an established primary connection. +func newClient(conn net.Conn, addr string, tlsConfig *tls.Config, dial DialContextFunc) *Client { + return &Client{conn: conn, addr: addr, tlsConfig: tlsConfig, dial: dial, connDialedAt: time.Now()} } // DialPool connects to a registry server over plain TCP and pre-warms a @@ -174,7 +209,7 @@ func DialPool(addr string, size int, opts ...DialOption) (*Client, error) { if err != nil { return nil, fmt.Errorf("dial registry: %w", err) } - c := &Client{conn: primary, addr: addr, dial: o.dial} + c := newClient(primary, addr, nil, o.dial) if err := c.initPool(size, nil); err != nil { primary.Close() return nil, err @@ -193,7 +228,7 @@ func DialTLS(addr string, tlsConfig *tls.Config, opts ...DialOption) (*Client, e if err != nil { return nil, fmt.Errorf("dial registry TLS: %w", err) } - return &Client{conn: conn, addr: addr, tlsConfig: tlsConfig, dial: o.dial}, nil + return newClient(conn, addr, tlsConfig, o.dial), nil } // DialTLSPool is the TLS variant of DialPool. @@ -212,7 +247,7 @@ func dialTLSPool(addr string, tlsConfig *tls.Config, size int, errPrefix string, if err != nil { return nil, fmt.Errorf("%s: %w", errPrefix, err) } - c := &Client{conn: primary, addr: addr, tlsConfig: tlsConfig, dial: o.dial} + c := newClient(primary, addr, tlsConfig, o.dial) if err := c.initPool(size, tlsConfig); err != nil { primary.Close() return nil, err @@ -230,7 +265,7 @@ func (c *Client) initPool(size int, tlsCfg *tls.Config) error { return nil } entries := make([]*pooledConn, 0, size) - entries = append(entries, &pooledConn{conn: c.conn}) + entries = append(entries, &pooledConn{conn: c.conn, dialedAt: c.connDialedAt}) for i := 1; i < size; i++ { conn, err := dialConn(context.Background(), c.dial, c.addr, tlsCfg) if err != nil { @@ -241,7 +276,7 @@ func (c *Client) initPool(size int, tlsCfg *tls.Config) error { } return fmt.Errorf("dial pool conn %d: %w", i, err) } - entries = append(entries, &pooledConn{conn: conn}) + entries = append(entries, &pooledConn{conn: conn, dialedAt: time.Now()}) } free := make(chan *pooledConn, len(entries)) for _, e := range entries { @@ -249,7 +284,6 @@ func (c *Client) initPool(size int, tlsCfg *tls.Config) error { } c.pool.entries = entries c.pool.free = free - c.pool.done = make(chan struct{}) return nil } @@ -262,7 +296,7 @@ func DialTLSPinned(addr, fingerprint string, opts ...DialOption) (*Client, error if err != nil { return nil, fmt.Errorf("dial registry TLS pinned: %w", err) } - return &Client{conn: conn, addr: addr, tlsConfig: tlsConfig, dial: o.dial}, nil + return newClient(conn, addr, tlsConfig, o.dial), nil } // DialTLSPinnedPool is the pooled variant of DialTLSPinned (see DialPool). @@ -293,83 +327,210 @@ func pinnedTLSConfig(fingerprint string) *tls.Config { } } +// Close closes every registry connection of the client. It is idempotent +// and safe to call concurrently with itself and with in-flight requests: +// the teardown runs exactly once, later calls return nil, and every request +// that is in flight or starts afterwards fails with an error matching +// ErrClosed (and ErrNoRegistry) instead of panicking. func (c *Client) Close() error { if c == nil { return nil } + var err error + c.closeOnce.Do(func() { err = c.shutdown() }) + return err +} + +// shutdown is Close's one-time teardown. +func (c *Client) shutdown() error { + // Wake goroutines parked on the pool's free list or in a reconnect + // backoff before taking c.mu: the single-conn path holds c.mu for its + // whole round trip, reconnect waits included. + close(c.doneCh()) + c.mu.Lock() c.closed = true - conn := c.conn - pool := c.pool.entries + // Snapshot every live conn. Entry conns are only replaced under c.mu, + // and nothing closes a conn once closed is set, so each conn is closed + // exactly once, here. In pool mode entries[0] shares the primary conn + // with c.conn, so c.conn is not closed separately. + var conns []net.Conn + if len(c.pool.entries) > 0 { + for _, e := range c.pool.entries { + if e.conn != nil { + conns = append(conns, e.conn) + } + } + } else if c.conn != nil { + conns = append(conns, c.conn) + } c.mu.Unlock() - // Close the conn after releasing the lock; conn is captured by value - // so reconnect() can't see it after we set c.closed=true (M7 fix) + + // Closing a conn in use interrupts its Read/Write; that request then + // sees the client closed and returns ErrClosed. Close does not wait + // for in-flight pooled requests. var firstErr error - if conn != nil { - if err := conn.Close(); err != nil { + for _, conn := range conns { + if err := conn.Close(); err != nil && !errors.Is(err, net.ErrClosed) && firstErr == nil { firstErr = err } } - // Close every secondary pooled conn. The primary is already closed - // above (entries[0] holds the same fd as c.conn). Skip index 0 so - // we don't double-close. - for i := 1; i < len(pool); i++ { - e := pool[i] - // Take e.mu to coordinate with any in-flight sendOnEntry that - // holds it; once we release the mutex it'll see a closed conn - // on its next Read/Write and return an error to the caller. - e.mu.Lock() - if e.conn != nil { - if err := e.conn.Close(); err != nil && firstErr == nil { - firstErr = err - } - } - e.mu.Unlock() - } - // Close pool.done to wake any goroutine blocked on <-c.pool.free and to - // signal the deferred pool-return in sendPool to drop its entry. We never - // close pool.free itself because that would race with concurrent sends on - // it from the sendPool defer. - if c.pool.done != nil { - close(c.pool.done) - } + c.logSummary(c.reconn.stop()) return firstErr } -// reconnect re-establishes the TCP connection to the registry. -// Must be called with c.mu held. -func (c *Client) reconnect(ctx context.Context) error { - if c.closed { - return fmt.Errorf("client closed") +// doneCh returns the channel Close closes, creating it on first use. +func (c *Client) doneCh() chan struct{} { + c.doneOnce.Do(func() { c.done = make(chan struct{}) }) + return c.done +} + +// closing reports whether Close has started. Unlike isClosed it takes no +// lock, so it is safe on the single-conn path, which holds c.mu. +func (c *Client) closing() bool { + select { + case <-c.doneCh(): + return true + default: + return false } - if c.conn != nil { - c.conn.Close() +} + +func (c *Client) log() *slog.Logger { + if c.logger != nil { + return c.logger } + return slog.Default() +} - var conn net.Conn - var err error - backoff := 500 * time.Millisecond - maxBackoff := 10 * time.Second +// logSummary emits the periodic reconnect summary (attrs from the +// tracker; nil means nothing to report). +func (c *Client) logSummary(attrs []any) { + if attrs == nil { + return + } + c.log().Info("registry reconnect summary", append([]any{"addr", c.addr}, attrs...)...) +} + +// noteConnError records a connection-level failure of one transmission (the +// triggering error of the reconnect that follows) and logs it at Debug. A +// logical request extends the rapid-close backoff streak at most once: +// streakNoted says whether an earlier transmission of it already did, and +// the result says whether one has now. +func (c *Client) noteConnError(err error, age time.Duration, streakNoted bool) bool { + rapid := c.reconn.connError(err, age, !streakNoted, c.logSummary) + c.log().Debug("registry conn dropped", "addr", c.addr, "err", err, + "conn_age", age.Round(time.Millisecond).String(), "rapid_close", rapid) + return streakNoted || rapid +} - for attempts := 0; attempts < 5; attempts++ { +// dialWithBackoff opens a replacement connection after cause broke the old +// one. It first waits out the client-wide cooldown, which is zero until the +// registry keeps closing connections shortly after they are used (its rate +// limiter's signature) or whole reconnects keep failing. Then it makes up +// to maxReconnectAttempts dials with jittered exponential backoff between +// them. It returns early when ctx is done or the client is closed. +// +// It takes no lock, so the single-conn path may call it with c.mu held. +func (c *Client) dialWithBackoff(ctx context.Context, what string, cause error) (net.Conn, error) { + // A caller that has given up (for example while the request was being + // retried on another pooled conn) gets no redial, and its abandoned + // dials are not counted against the registry. + if err := ctx.Err(); err != nil { + return nil, err + } + if err := sleepCtx(ctx, c.doneCh(), c.reconn.cooldown()); err != nil { + return nil, err + } + p := c.reconn.pol() + backoff := p.dialBase + var err error + for attempt := 1; attempt <= maxReconnectAttempts; attempt++ { + if c.closing() { + return nil, ErrClosed + } + var conn net.Conn conn, err = dialConn(ctx, c.dial, c.addr, c.tlsConfig) if err == nil { - c.conn = conn - slog.Info("registry reconnected", "addr", c.addr) - return nil + return conn, nil } - slog.Warn("registry reconnect failed", "attempt", attempts+1, "err", err) - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(backoff): + if ctx.Err() != nil { + return nil, ctx.Err() } - backoff *= 2 - if backoff > maxBackoff { - backoff = maxBackoff + c.log().Warn(what+" reconnect failed", "addr", c.addr, "attempt", attempt, "err", err, "cause", errText(cause)) + c.reconn.dialFailed(c.logSummary) + if attempt == maxReconnectAttempts { + break } + if werr := sleepCtx(ctx, c.doneCh(), jitter(backoff)); werr != nil { + return nil, werr + } + if backoff *= 2; backoff > p.dialMax { + backoff = p.dialMax + } + } + c.reconn.exhausted() + return nil, fmt.Errorf("reconnect failed after %d attempts: %w", maxReconnectAttempts, err) +} + +// noteReconnected logs one successful redial at Debug (with the error that +// triggered it) and feeds the periodic INFO summary. +func (c *Client) noteReconnected(what string, cause error) { + c.reconn.reconnected(cause, c.logSummary) + c.log().Debug(what+" reconnected", "addr", c.addr, "cause", errText(cause)) +} + +func errText(err error) string { + if err == nil { + return "" + } + return err.Error() +} + +// reconnectFailed builds the error for a request whose connection broke +// (cause) and could not be redialed (reconnErr). Both stay in the chain. +func reconnectFailed(cause, reconnErr error) error { + if errors.Is(reconnErr, ErrClosed) { + return closedDuring(cause) } - return fmt.Errorf("reconnect failed after 5 attempts: %w", err) + if cause == nil { + cause = errNotConnected + } + return fmt.Errorf("send failed and reconnect failed: %w (reconnect: %w)", cause, reconnErr) +} + +// retriedAfter annotates the error of a retry that also failed at the +// connection level with the error that triggered the reconnect. +func retriedAfter(err, cause error) error { + return fmt.Errorf("%w (retried after reconnect; first failure: %v)", err, cause) +} + +// reconnect re-establishes the single-conn path's connection after cause +// broke it. Must be called with c.mu held. +func (c *Client) reconnect(ctx context.Context, cause error) error { + if c.closed || c.closing() { + return ErrClosed + } + if c.conn != nil { + c.conn.Close() + // Retired: a failed redial leaves no conn behind for Close to + // close a second time. + c.conn = nil + } + conn, err := c.dialWithBackoff(ctx, "registry", cause) + if err != nil { + return err + } + if c.closing() { + // Close started while we were dialing and is waiting for c.mu; + // do not hand it a conn to close, and do not retry on it. + conn.Close() + return ErrClosed + } + c.conn = conn + c.connDialedAt = time.Now() + c.noteReconnected("registry", cause) + return nil } // Send sends a registry message without a deadline. For shutdown-safe use @@ -381,7 +542,28 @@ func (c *Client) Send(msg map[string]interface{}) (map[string]interface{}, error // SendContext sends a registry message with context propagation through // reconnect retries. Callers should pass a context with deadline or // cancellation (e.g. daemon shutdown context) so that reconnect backoff -// does not block graceful stop. +// does not block graceful stop. ctx bounds the wait for a free pooled +// connection, the retry on another pooled connection and the reconnect; +// it does not interrupt a round trip on the request's own connection, +// which the 30 s read deadline bounds. +// +// A connection-level failure (no response) is retried: +// - On a single-connection client the connection is redialed and the +// request sent once more, so it reaches the registry at most twice. +// - On a pooled client, when the connection was dropped (EOF, reset, +// closed) the request is first retried on another idle, healthy +// pooled connection, bounded by a short deadline and ctx. Only if that +// fails too is the request's own connection redialed and the request +// sent a third time. Other connection-level failures (timeouts) go +// straight to the redial. +// +// So a request whose connections drop may reach the registry more than +// once (up to three times on a pooled client); an operation that is not +// safe to repeat may have been applied even when an error is returned. +// However many attempts fail, one request extends the rapid-close backoff +// streak at most once. An error response from the registry is returned as +// is, never retried. After Close every call returns an error matching +// ErrClosed. func (c *Client) SendContext(ctx context.Context, msg map[string]interface{}) (map[string]interface{}, error) { // Nil receiver — return a sentinel rather than panicking. Every // exported wrapper method (Register, Lookup, Resolve, …) funnels @@ -401,16 +583,31 @@ func (c *Client) SendContext(ctx context.Context, msg map[string]interface{}) (m c.mu.Lock() defer c.mu.Unlock() + if c.closed || c.closing() { + return nil, ErrClosed + } resp, err := c.sendLocked(msg) - if err != nil && resp == nil && !c.closed { + if err != nil && resp == nil { // Connection-level failure (no response received) — reconnect and retry once. // Server error responses (resp != nil) do NOT trigger reconnection. - if reconnErr := c.reconnect(ctx); reconnErr != nil { - return nil, fmt.Errorf("send failed and reconnect failed: %w", err) + if c.closing() { + return nil, closedDuring(err) + } + cause := err + c.noteConnError(cause, time.Since(c.connDialedAt), false) + if reconnErr := c.reconnect(ctx, cause); reconnErr != nil { + return nil, reconnectFailed(cause, reconnErr) } resp, err = c.sendLocked(msg) + if err != nil && resp == nil { + if c.closing() { + return nil, closedDuring(err) + } + return nil, retriedAfter(err, cause) + } } + c.reconn.healthy(time.Since(c.connDialedAt)) return resp, err } @@ -420,105 +617,234 @@ func (c *Client) SendContext(ctx context.Context, msg map[string]interface{}) (m func (c *Client) sendPool(ctx context.Context, msg map[string]interface{}) (map[string]interface{}, error) { // Cheap closed check — avoids a wedged caller waiting on a free // channel that nobody will ever return to once Close has run. - c.mu.Lock() - closed := c.closed - c.mu.Unlock() - if closed { - return nil, fmt.Errorf("client closed") + if c.isClosed() { + return nil, ErrClosed } + entry, err := c.acquireEntry(ctx) + if err != nil { + return nil, err + } + defer c.releaseEntry(entry) + entry.mu.Lock() + defer entry.mu.Unlock() - var entry *pooledConn + // An earlier request dropped this entry's conn (and was served + // elsewhere, or its redial failed): redial before use. + if entry.broken != nil || entry.conn == nil { + if rerr := c.reconnectEntry(ctx, entry, entry.broken); rerr != nil { + return nil, reconnectFailed(entry.broken, rerr) + } + } + + var st sendState + resp, err := c.sendOnEntry(entry, msg, &st) + if err == nil || resp != nil { + return resp, err + } + if c.isClosed() { + return nil, closedDuring(err) + } + cause := err + + // The connection was dropped (EOF, reset, closed). The other entries + // were dialed at other times and may still be good, so retry once on + // an idle healthy one before paying for a redial and any cooldown. + // That retry is bounded by otherConnTimeout and ctx: the other conn + // may be silently hung, and the redial below must still fit in the + // caller's deadline. Timeouts are not retried this way: a half-open + // conn already cost the caller the full read deadline. + if isConnDropped(cause) { + if resp, served, err := c.sendOnOtherEntry(ctx, msg, &st); served { + return resp, err + } + if c.isClosed() { + return nil, closedDuring(cause) + } + } + + // Redial this entry and retry once. dialWithBackoff gives up at once + // when ctx ended during the retry above. + if rerr := c.reconnectEntry(ctx, entry, cause); rerr != nil { + return nil, reconnectFailed(cause, rerr) + } + resp, err = c.sendOnEntry(entry, msg, &st) + if err != nil && resp == nil { + if c.isClosed() { + return nil, closedDuring(err) + } + return nil, retriedAfter(err, cause) + } + return resp, err +} + +// acquireEntry takes a pool entry off the free list, preferring an idle +// healthy one and otherwise blocking until any entry is free, ctx is done +// or the client is closed. +func (c *Client) acquireEntry(ctx context.Context) (*pooledConn, error) { + if e := c.tryAcquireHealthy(); e != nil { + return e, nil + } select { - case entry = <-c.pool.free: - case <-c.pool.done: - return nil, fmt.Errorf("client closed") + case e := <-c.pool.free: + if c.closing() { + c.releaseEntry(e) + return nil, ErrClosed + } + return e, nil + case <-c.doneCh(): + return nil, ErrClosed + case <-ctx.Done(): + return nil, ctx.Err() } +} + +// tryAcquireHealthy takes an idle entry that needs no redial off the free +// list without blocking. It returns nil when none is free right now. +func (c *Client) tryAcquireHealthy() *pooledConn { + var skipped []*pooledConn defer func() { - select { - case c.pool.free <- entry: - case <-c.pool.done: - // pool is torn down; drop the entry + for _, e := range skipped { + c.releaseEntry(e) } }() + for range len(c.pool.entries) { + select { + case e := <-c.pool.free: + if e.healthy() { + return e + } + skipped = append(skipped, e) + default: + return nil + } + } + return nil +} - entry.mu.Lock() - defer entry.mu.Unlock() +// releaseEntry returns an entry to the free list. +func (c *Client) releaseEntry(e *pooledConn) { + select { + case c.pool.free <- e: + case <-c.doneCh(): + // pool is torn down; drop the entry + } +} - resp, err := c.sendOnEntry(entry, msg) - if err != nil && resp == nil && !c.isClosed() { - // Connection-level failure on this entry — reconnect THIS entry - // only (other pool entries are unaffected) and retry once. - if reconnErr := c.reconnectEntry(ctx, entry); reconnErr != nil { - return nil, fmt.Errorf("send failed and reconnect failed: %w", err) - } - resp, err = c.sendOnEntry(entry, msg) +// sendState is the bookkeeping of one logical request across its attempts +// on a pooled client. +type sendState struct { + // streakNoted is set once a failed attempt of the request extended the + // rapid-close streak; later failures only feed the summary. + streakNoted bool +} + +// sendOnOtherEntry retries msg once on another idle, healthy entry, with +// the exchange bounded by otherConnTimeout and ctx. served is false when +// no such entry is free or the retry also failed at the connection level, +// timed out or was abandoned for ctx (that entry is then marked broken and +// its conn retired, since a late reply could still arrive on it). +func (c *Client) sendOnOtherEntry(ctx context.Context, msg map[string]interface{}, st *sendState) (resp map[string]interface{}, served bool, err error) { + if ctx.Err() != nil { + return nil, false, nil } - return resp, err + other := c.tryAcquireHealthy() + if other == nil { + return nil, false, nil + } + defer c.releaseEntry(other) + other.mu.Lock() + defer other.mu.Unlock() + resp, err = boundedRoundTrip(ctx, other.conn, msg, c.reconn.pol().otherConnTimeout) + resp, err = c.entryResult(other, st, resp, err) + if err != nil && resp == nil { + return nil, false, nil + } + if err == nil { + c.reconn.servedByOther(c.logSummary) + } + return resp, true, err } // sendOnEntry writes the request and reads the response on entry.conn. +// Caller must hold entry.mu. A connection-level failure (no response) +// marks the entry broken and retires its conn. +func (c *Client) sendOnEntry(entry *pooledConn, msg map[string]interface{}, st *sendState) (map[string]interface{}, error) { + resp, err := roundTrip(entry.conn, msg) + return c.entryResult(entry, st, resp, err) +} + +// entryResult records the outcome of one attempt on entry: a +// connection-level failure (no response) marks the entry broken and +// retires its conn; a response counts toward the rapid-close streak reset. // Caller must hold entry.mu. -func (c *Client) sendOnEntry(entry *pooledConn, msg map[string]interface{}) (map[string]interface{}, error) { - if err := wire.WriteMessage(entry.conn, msg); err != nil { - return nil, fmt.Errorf("send: %w", err) +func (c *Client) entryResult(entry *pooledConn, st *sendState, resp map[string]interface{}, err error) (map[string]interface{}, error) { + if err != nil && resp == nil { + entry.broken = err + if !c.isClosed() { + st.streakNoted = c.noteConnError(err, time.Since(entry.dialedAt), st.streakNoted) + } + c.retireEntryConn(entry) + return nil, err } - entry.conn.SetReadDeadline(time.Now().Add(30 * time.Second)) - resp, err := wire.ReadMessage(entry.conn) - entry.conn.SetReadDeadline(time.Time{}) - if err != nil { - return nil, fmt.Errorf("recv: %w", err) + c.reconn.healthy(time.Since(entry.dialedAt)) + return resp, err +} + +// retireEntryConn closes a dropped entry's conn right away (rather than +// holding the fd until the entry is next used) and clears it, so Close +// never closes it a second time. Once Close has started, the conn belongs +// to Close and is left alone. Caller must hold entry.mu. +func (c *Client) retireEntryConn(entry *pooledConn) { + c.mu.Lock() + if c.closed { + c.mu.Unlock() + return } - if errVal, ok := resp["error"]; ok { - return resp, fmt.Errorf("registry: %v", errVal) + old := entry.conn + entry.conn = nil + if entry == c.pool.entries[0] { + c.conn = nil } - // PILOT-132: reject valid JSON that lacks the expected "type" envelope key. - if _, hasType := resp["type"]; !hasType && len(resp) > 0 { - return resp, fmt.Errorf("registry: malformed response (missing %q field)", "type") + c.mu.Unlock() + if old != nil { + old.Close() } - return resp, nil } -// reconnectEntry redials a single pool entry. Caller must hold entry.mu. -// This is the per-entry analogue of Client.reconnect. -func (c *Client) reconnectEntry(ctx context.Context, entry *pooledConn) error { +// reconnectEntry redials a single pool entry after cause broke it (cause +// may be nil). Caller must hold entry.mu. This is the per-entry analogue +// of Client.reconnect. +func (c *Client) reconnectEntry(ctx context.Context, entry *pooledConn, cause error) error { if c.isClosed() { - return fmt.Errorf("client closed") - } - if entry.conn != nil { - entry.conn.Close() + return ErrClosed } + c.retireEntryConn(entry) - var conn net.Conn - var err error - backoff := 500 * time.Millisecond - maxBackoff := 10 * time.Second - for attempts := 0; attempts < 5; attempts++ { - conn, err = dialConn(ctx, c.dial, c.addr, c.tlsConfig) - if err == nil { - entry.conn = conn - // Keep c.conn (primary) in sync if this is the primary entry. - // Tests in this package read c.conn directly, so we must not - // leave it pointing at a closed fd. - if entry == c.pool.entries[0] { - c.mu.Lock() - c.conn = conn - c.mu.Unlock() - } - slog.Info("registry pool conn reconnected", "addr", c.addr) - return nil - } - slog.Warn("registry pool conn reconnect failed", "attempt", attempts+1, "err", err) - select { - case <-ctx.Done(): - return ctx.Err() - case <-time.After(backoff): - } - backoff *= 2 - if backoff > maxBackoff { - backoff = maxBackoff - } + conn, err := c.dialWithBackoff(ctx, "registry pool conn", cause) + if err != nil { + return err + } + c.mu.Lock() + if c.closed { + // Close ran while we were dialing and has already closed every + // conn it could see; this one is ours to close. + c.mu.Unlock() + conn.Close() + return ErrClosed + } + entry.conn = conn + // Keep c.conn (primary) in sync if this is the primary entry. + // Tests in this package read c.conn directly, so we must not + // leave it pointing at a closed fd. + if entry == c.pool.entries[0] { + c.conn = conn + c.connDialedAt = time.Now() } - return fmt.Errorf("reconnect failed after 5 attempts: %w", err) + c.mu.Unlock() + entry.dialedAt = time.Now() + entry.broken = nil + c.noteReconnected("registry pool conn", cause) + return nil } // isClosed returns whether Close has been called. Cheap, lock-protected. @@ -530,12 +856,72 @@ func (c *Client) isClosed() bool { // sendLocked sends a message and reads the response. Must be called with c.mu held. func (c *Client) sendLocked(msg map[string]interface{}) (map[string]interface{}, error) { - if err := wire.WriteMessage(c.conn, msg); err != nil { + return roundTrip(c.conn, msg) +} + +// roundTrip writes msg on conn and reads the response. A nil response +// with a non-nil error is a connection-level failure; a non-nil response +// with an error is the registry's error reply. +func roundTrip(conn net.Conn, msg map[string]interface{}) (map[string]interface{}, error) { + if conn == nil { + return nil, errNotConnected + } + if err := wire.WriteMessage(conn, msg); err != nil { return nil, fmt.Errorf("send: %w", err) } - c.conn.SetReadDeadline(time.Now().Add(30 * time.Second)) - resp, err := wire.ReadMessage(c.conn) - c.conn.SetReadDeadline(time.Time{}) + conn.SetReadDeadline(time.Now().Add(30 * time.Second)) + resp, err := wire.ReadMessage(conn) + conn.SetReadDeadline(time.Time{}) + return checkResponse(resp, err) +} + +// boundedRoundTrip is roundTrip with the whole exchange (write and read) +// limited to timeout and to ctx, both its deadline and its cancellation. +// When it gives up, the request may still be answered later on conn, so +// the caller must retire the conn. +func boundedRoundTrip(ctx context.Context, conn net.Conn, msg map[string]interface{}, timeout time.Duration) (map[string]interface{}, error) { + if conn == nil { + return nil, errNotConnected + } + deadline := time.Now().Add(timeout) + if d, ok := ctx.Deadline(); ok && d.Before(deadline) { + deadline = d + } + conn.SetDeadline(deadline) + // Cancelling ctx interrupts the exchange by pulling the deadline in. + // The watcher is joined before the deadline is cleared, so a + // cancellation that races the end of the exchange cannot leave a + // deadline on a conn that goes back to the pool. + finished := make(chan struct{}) + watcherDone := make(chan struct{}) + go func() { + defer close(watcherDone) + select { + case <-ctx.Done(): + conn.SetDeadline(time.Now()) + case <-finished: + } + }() + var resp map[string]interface{} + err := wire.WriteMessage(conn, msg) + if err != nil { + err = fmt.Errorf("send: %w", err) + } else { + resp, err = checkResponse(wire.ReadMessage(conn)) + } + close(finished) + <-watcherDone + conn.SetDeadline(time.Time{}) + if err != nil && resp == nil && ctx.Err() != nil { + err = fmt.Errorf("%w (%w)", err, ctx.Err()) + } + return resp, err +} + +// checkResponse turns the result of reading one response frame into +// roundTrip's result: a read error is a connection-level failure (nil +// response); an error reply or a malformed one keeps the response. +func checkResponse(resp map[string]interface{}, err error) (map[string]interface{}, error) { if err != nil { return nil, fmt.Errorf("recv: %w", err) } diff --git a/registry/client/reconnect.go b/registry/client/reconnect.go new file mode 100644 index 0000000..558d168 --- /dev/null +++ b/registry/client/reconnect.go @@ -0,0 +1,341 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package client + +import ( + "context" + "errors" + "fmt" + "io" + "math/rand/v2" + "net" + "sync" + "syscall" + "time" +) + +// ErrClosed is returned by every request method of a closed *Client or +// *BinaryClient, including requests that were still in flight when Close +// ran. errors.Is(ErrClosed, ErrNoRegistry) is true, so callers that already +// treat "no registry" as recoverable handle a closed client (for example one +// retired by a forced reconnect) the same way. +var ErrClosed error = closedError{} + +type closedError struct{} + +func (closedError) Error() string { return "registry client closed" } + +// Is makes errors.Is(ErrClosed, ErrNoRegistry) true. +func (closedError) Is(target error) bool { return target == ErrNoRegistry } + +// errNotConnected is the connection-level error for a slot whose connection +// was retired (dropped, or a redial failed) and has not been replaced yet. +var errNotConnected = errors.New("registry connection not established") + +// closedDuring reports a request that was aborted because Close ran while it +// was in flight. The result matches ErrClosed (and so ErrNoRegistry) and +// keeps the I/O error that Close caused. +func closedDuring(err error) error { + if err == nil || errors.Is(err, ErrClosed) { + return ErrClosed + } + return fmt.Errorf("%w: request aborted: %w", ErrClosed, err) +} + +// isPeerClose reports whether err means the registry (or something on the +// path) closed or reset the connection: EOF with no response frame, a +// connection reset, or a broken pipe. +func isPeerClose(err error) bool { + return errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || + errors.Is(err, syscall.ECONNRESET) || + errors.Is(err, syscall.ECONNABORTED) || + errors.Is(err, syscall.EPIPE) +} + +// isConnDropped reports whether err means the connection itself is gone +// (closed by the peer or locally), as opposed to a timeout on a half-open +// connection, a malformed frame, or an error response from the registry. +func isConnDropped(err error) bool { + return isPeerClose(err) || errors.Is(err, net.ErrClosed) || errors.Is(err, errNotConnected) +} + +// maxReconnectAttempts bounds the dials one reconnect makes. +const maxReconnectAttempts = 5 + +// reconnectPolicy holds the reconnect timing knobs. Tests shrink them through +// reconnectTracker.policy; production always uses defaultReconnectPolicy. +type reconnectPolicy struct { + // dialBase and dialMax shape the jittered exponential backoff between + // the dial attempts of one reconnect. + dialBase, dialMax time.Duration + // cooldownBase and cooldownMax shape the client-wide wait before a + // reconnect's first dial. It is zero until the registry starts closing + // fresh connections (see rapidWindow) or whole reconnects fail, then + // doubles with each such event. + cooldownBase, cooldownMax time.Duration + // rapidWindow: a connection the peer closes or resets while younger + // than this counts as a rapid close, the registry rate limiter's + // signature (it closes, with no error frame, the first request it + // denies after a connection's grace period). A success on a connection + // at least this old ends the rapid-close streak. + rapidWindow time.Duration + // quietReset ends both streaks when no rapid close or failed reconnect + // happened for this long. + quietReset time.Duration + // summaryEvery is the minimum interval between INFO summaries. + summaryEvery time.Duration + // otherConnTimeout bounds the whole exchange (write and read) of the + // retry on another pooled connection after a drop. Nothing shows that + // connection is still alive, and the request has already failed once, + // so it gets far less than the 30 s read deadline: a silently hung + // connection is retired after this long and the request falls back to + // a redial. The caller's ctx can end the retry sooner. + otherConnTimeout time.Duration +} + +var defaultReconnectPolicy = reconnectPolicy{ + dialBase: 500 * time.Millisecond, + dialMax: 8 * time.Second, + cooldownBase: 250 * time.Millisecond, + // Kept well below the 8 s deadline the daemon puts on registry calls, + // so the cooldown alone never turns a shed request into a timeout. + cooldownMax: 2 * time.Second, + rapidWindow: 30 * time.Second, + quietReset: time.Minute, + summaryEvery: time.Minute, + // Together with cooldownMax this stays well below the daemon's 8 s + // deadline, leaving room for the redial and the final attempt, while + // being several registry round trips long. + otherConnTimeout: 2 * time.Second, +} + +// jitter returns a random duration in [d/2, d] ("equal jitter"), so clients +// shed by the same limiter do not redial in lockstep. +func jitter(d time.Duration) time.Duration { + if d <= 1 { + return d + } + half := d / 2 + return half + rand.N(d-half+1) //nolint:gosec // backoff jitter, not a security value +} + +// reconnectTracker is the reconnect state shared by every connection of one +// Client: the backoff streaks and the counters behind the periodic INFO +// summary. The registry's limiter works per source IP and globally, so one +// connection being shed is evidence about all of them. The zero value is +// ready to use. +type reconnectTracker struct { + mu sync.Mutex + // policy overrides defaultReconnectPolicy when non-nil (tests only). + policy *reconnectPolicy + + rapidStreak int // consecutive rapid closes (rate-limit signature) + failStreak int // consecutive reconnects that exhausted every dial attempt + lastEvent time.Time // last rapid close or failed reconnect + + // The INFO summary: the first event of a window arms timer, which + // emits the window's counters summaryEvery later. So at most one + // summary per summaryEvery, and a burst is reported even if nothing + // follows it. stopped (set by Close) disarms it for good. + windowStart time.Time + stats reconnectStats + timer *time.Timer + stopped bool +} + +// reconnectStats are the counters reported by one INFO summary. +type reconnectStats struct { + reconnects int // successful redials + failedDials int // failed dial attempts + rapidCloses int // connections closed by the peer shortly after they were used + otherConn int // requests served by another pooled connection after a drop + lastCause string // the connection error behind the latest drop or reconnect +} + +func (t *reconnectTracker) pol() reconnectPolicy { + if t.policy != nil { + return *t.policy + } + return defaultReconnectPolicy +} + +// quietResetLocked ends both streaks after a quiet period. +func (t *reconnectTracker) quietResetLocked(now time.Time, p reconnectPolicy) { + if !t.lastEvent.IsZero() && now.Sub(t.lastEvent) >= p.quietReset { + t.rapidStreak = 0 + t.failStreak = 0 + } +} + +// countedLocked is called after a summary counter changed: it opens the +// summary window and arms the timer that will emit it through emit. +func (t *reconnectTracker) countedLocked(now time.Time, emit func([]any)) { + if t.windowStart.IsZero() { + t.windowStart = now + } + if t.timer != nil || t.stopped || emit == nil { + return + } + t.timer = time.AfterFunc(t.pol().summaryEvery-now.Sub(t.windowStart), func() { + t.mu.Lock() + t.timer = nil + summary := t.summaryLocked(time.Now()) + t.mu.Unlock() + emit(summary) + }) +} + +// connError records a connection-level failure of one transmission on a +// connection of the given age and reports whether it is a rapid close. +// Every rapid close is counted in the summary, but it extends the backoff +// streak only when extendStreak is set: the caller passes false once an +// earlier transmission of the same logical request already did, so one +// shed request grows the cooldown by one step however many connections it +// was tried on. +func (t *reconnectTracker) connError(err error, age time.Duration, extendStreak bool, emit func([]any)) (rapid bool) { + p := t.pol() + rapid = isPeerClose(err) && age > 0 && age < p.rapidWindow + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + t.stats.lastCause = err.Error() + if rapid { + t.stats.rapidCloses++ + t.countedLocked(now, emit) + if extendStreak { + t.quietResetLocked(now, p) + t.rapidStreak++ + t.lastEvent = now + } + } + return rapid +} + +// healthy records a request answered on a connection of the given age. An +// answer on a connection older than the rapid-close window shows the +// registry is no longer shedding it, which ends the rapid-close streak. +func (t *reconnectTracker) healthy(age time.Duration) { + if age < t.pol().rapidWindow { + return + } + t.mu.Lock() + t.rapidStreak = 0 + t.mu.Unlock() +} + +// cooldown returns the wait before a reconnect's first dial: zero while +// nothing suspicious happened, otherwise a jittered exponential of the +// combined streak, capped at cooldownMax. +func (t *reconnectTracker) cooldown() time.Duration { + p := t.pol() + t.mu.Lock() + t.quietResetLocked(time.Now(), p) + n := t.rapidStreak + t.failStreak + t.mu.Unlock() + if n <= 0 || p.cooldownBase <= 0 { + return 0 + } + d := p.cooldownMax + if n <= 16 { + if v := p.cooldownBase << (n - 1); v < d { + d = v + } + } + return jitter(d) +} + +func (t *reconnectTracker) dialFailed(emit func([]any)) { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + t.stats.failedDials++ + t.countedLocked(now, emit) +} + +// exhausted records a reconnect that failed every dial attempt. +func (t *reconnectTracker) exhausted() { + now := time.Now() + t.mu.Lock() + t.quietResetLocked(now, t.pol()) + t.failStreak++ + t.lastEvent = now + t.mu.Unlock() +} + +func (t *reconnectTracker) reconnected(cause error, emit func([]any)) { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + t.failStreak = 0 + t.stats.reconnects++ + if cause != nil { + t.stats.lastCause = cause.Error() + } + t.countedLocked(now, emit) +} + +func (t *reconnectTracker) servedByOther(emit func([]any)) { + now := time.Now() + t.mu.Lock() + defer t.mu.Unlock() + t.stats.otherConn++ + t.countedLocked(now, emit) +} + +// stop disarms the summary timer for good and returns the summary of the +// current window, or nil if nothing happened in it (used by Close so the +// last window is not lost). +func (t *reconnectTracker) stop() []any { + t.mu.Lock() + defer t.mu.Unlock() + t.stopped = true + if t.timer != nil { + t.timer.Stop() + t.timer = nil + } + return t.summaryLocked(time.Now()) +} + +// summaryLocked returns the slog attributes summarising the current window +// and starts a new one, or nil when nothing happened in it. +func (t *reconnectTracker) summaryLocked(now time.Time) []any { + s := t.stats + window := time.Duration(0) + if !t.windowStart.IsZero() { + window = now.Sub(t.windowStart) + } + t.stats = reconnectStats{} + t.windowStart = time.Time{} + if s.reconnects == 0 && s.failedDials == 0 && s.rapidCloses == 0 && s.otherConn == 0 { + return nil + } + return []any{ + "window", window.Round(time.Second).String(), + "reconnects", s.reconnects, + "failed_dials", s.failedDials, + "rapid_closes", s.rapidCloses, + "served_by_other_conn", s.otherConn, + "rate_limit_suspected", s.rapidCloses > 0, + "backoff_streak", t.rapidStreak + t.failStreak, + "last_cause", s.lastCause, + } +} + +// sleepCtx waits for d, returning early with ctx's error when ctx is done or +// with ErrClosed when done is closed. +func sleepCtx(ctx context.Context, done <-chan struct{}, d time.Duration) error { + if d <= 0 { + return nil + } + timer := time.NewTimer(d) + defer timer.Stop() + select { + case <-timer.C: + return nil + case <-ctx.Done(): + return ctx.Err() + case <-done: + return ErrClosed + } +} diff --git a/registry/client/zz_client_branch_test.go b/registry/client/zz_client_branch_test.go index b0dacde..93ecf6d 100644 --- a/registry/client/zz_client_branch_test.go +++ b/registry/client/zz_client_branch_test.go @@ -279,7 +279,7 @@ func TestBinaryReconnectShortCircuitsWhenClosed(t *testing.T) { c.mu.Lock() _ = c.conn.Close() c.closed = true - err = c.reconnect() + err = c.reconnect(nil) c.mu.Unlock() if err == nil { t.Fatalf("reconnect after Close should fail") @@ -354,7 +354,7 @@ func TestClientReconnectExhaustsAttempts(t *testing.T) { c := &Client{addr: addr} c.mu.Lock() - err := c.reconnect(context.Background()) + err := c.reconnect(context.Background(), nil) c.mu.Unlock() if err == nil { t.Fatalf("expected reconnect failure") diff --git a/registry/client/zz_client_close_reconnect_test.go b/registry/client/zz_client_close_reconnect_test.go new file mode 100644 index 0000000..4665556 --- /dev/null +++ b/registry/client/zz_client_close_reconnect_test.go @@ -0,0 +1,1654 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package client + +import ( + "context" + "encoding/binary" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "os" + "strings" + "sync" + "sync/atomic" + "syscall" + "testing" + "time" + + "github.com/pilot-protocol/common/registry/wire" +) + +// Regression and behaviour tests for: +// +// - Close idempotency (nightly "close of closed channel" panic when the +// daemon's forceReconnectRegistry `go old.Close()` raced Stop's Close); +// - pool reconnect behaviour: the triggering error is kept, per-reconnect +// success logging is Debug with a periodic INFO summary, jittered +// exponential backoff on rapid server closes (the registry rate +// limiter's EOF-without-error-frame signature) and on repeated reconnect +// failures, and a retry on another healthy pool conn before redialing. + +// --- helpers ---------------------------------------------------------------- + +// scriptedRegistry is a JSON-wire registry stand-in whose drop hook decides, +// per request, whether to close the connection without answering (what the +// real registry does when its rate limiter denies a request). +type scriptedRegistry struct { + ln net.Listener + conns atomic.Int32 // connections accepted + received atomic.Int32 // requests read (answered, dropped or hung) + served atomic.Int32 // requests answered + dropped atomic.Int32 // requests answered by closing the conn + hung atomic.Int32 // requests never answered, conn left open + + mu sync.Mutex + drop func(connSeq, reqSeq int, age time.Duration) bool + // hang, when set, picks requests that are read but never answered + // while the conn stays open: a silently hung registry connection. + hang func(connSeq, reqSeq int) bool + live []net.Conn + wg sync.WaitGroup +} + +func newScriptedRegistry(t *testing.T, drop func(connSeq, reqSeq int, age time.Duration) bool) *scriptedRegistry { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + s := &scriptedRegistry{ln: ln, drop: drop} + s.wg.Add(1) + go s.accept() + t.Cleanup(s.stop) + return s +} + +// closeAfter answers n requests per connection, then closes the connection +// on the next request without replying. +func closeAfter(n int) func(int, int, time.Duration) bool { + return func(_, reqSeq int, _ time.Duration) bool { return reqSeq > n } +} + +// afterGrace mimics the production registry: the first request that +// arrives on a connection older than grace is closed without a reply. +func afterGrace(grace time.Duration) func(int, int, time.Duration) bool { + return func(_, _ int, age time.Duration) bool { return age >= grace } +} + +func neverDrop(int, int, time.Duration) bool { return false } + +func (s *scriptedRegistry) addr() string { return s.ln.Addr().String() } + +// waitConns waits until the server has accepted n connections. Closing the +// listener before that would reset connections still in the accept queue. +func (s *scriptedRegistry) waitConns(t *testing.T, n int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for int(s.conns.Load()) < n { + if time.Now().After(deadline) { + t.Fatalf("server accepted %d conns, want %d", s.conns.Load(), n) + } + time.Sleep(time.Millisecond) + } +} + +func (s *scriptedRegistry) setDrop(fn func(int, int, time.Duration) bool) { + s.mu.Lock() + s.drop = fn + s.mu.Unlock() +} + +func (s *scriptedRegistry) setHang(fn func(connSeq, reqSeq int) bool) { + s.mu.Lock() + s.hang = fn + s.mu.Unlock() +} + +// shutdown stops accepting and closes every live connection, so redials fail. +func (s *scriptedRegistry) shutdown() { + s.ln.Close() + s.mu.Lock() + for _, c := range s.live { + c.Close() + } + s.mu.Unlock() +} + +func (s *scriptedRegistry) stop() { + s.shutdown() + s.wg.Wait() +} + +func (s *scriptedRegistry) accept() { + defer s.wg.Done() + for { + conn, err := s.ln.Accept() + if err != nil { + return + } + seq := int(s.conns.Add(1)) + s.mu.Lock() + s.live = append(s.live, conn) + s.mu.Unlock() + s.wg.Add(1) + go func() { + defer s.wg.Done() + s.serve(conn, seq) + }() + } +} + +func (s *scriptedRegistry) serve(conn net.Conn, connSeq int) { + defer conn.Close() + start := time.Now() + for reqSeq := 1; ; reqSeq++ { + var lenBuf [4]byte + if _, err := io.ReadFull(conn, lenBuf[:]); err != nil { + return + } + n := binary.BigEndian.Uint32(lenBuf[:]) + if n > 1<<20 { + return + } + body := make([]byte, n) + if _, err := io.ReadFull(conn, body); err != nil { + return + } + var req map[string]interface{} + if err := json.Unmarshal(body, &req); err != nil { + return + } + s.received.Add(1) + s.mu.Lock() + drop, hang := s.drop, s.hang + s.mu.Unlock() + if hang != nil && hang(connSeq, reqSeq) { + s.hung.Add(1) + _, _ = io.Copy(io.Discard, conn) // until the client gives up + return + } + if drop(connSeq, reqSeq, time.Since(start)) { + s.dropped.Add(1) + return // close without an error frame + } + out, _ := json.Marshal(map[string]interface{}{"type": "ok", "echo": req}) + var outLen [4]byte + binary.BigEndian.PutUint32(outLen[:], uint32(len(out))) + if _, err := conn.Write(append(outLen[:], out...)); err != nil { + return + } + s.served.Add(1) + } +} + +// holdIdleEntries takes n entries off the pool's free list so requests +// cannot use them, and returns an idempotent function that gives them back. +func holdIdleEntries(t *testing.T, c *Client, n int) func() { + t.Helper() + held := make([]*pooledConn, 0, n) + for i := 0; i < n; i++ { + select { + case e := <-c.pool.free: + held = append(held, e) + case <-time.After(2 * time.Second): + t.Fatalf("could not take pool entry %d", i) + } + } + var once sync.Once + release := func() { + once.Do(func() { + for _, e := range held { + c.releaseEntry(e) + } + }) + } + t.Cleanup(release) + return release +} + +// holdAllBut takes every pool entry except keep off the free list (all +// entries must be idle), so the next request is forced onto keep. It +// returns an idempotent function that gives them back. +func holdAllBut(t *testing.T, c *Client, keep *pooledConn) func() { + t.Helper() + var held []*pooledConn + for range len(c.pool.entries) { + select { + case e := <-c.pool.free: + if e != keep { + held = append(held, e) + } + case <-time.After(2 * time.Second): + t.Fatal("pool entry not idle") + } + } + c.releaseEntry(keep) + var once sync.Once + release := func() { + once.Do(func() { + for _, e := range held { + c.releaseEntry(e) + } + }) + } + t.Cleanup(release) + return release +} + +// countingConn counts Close calls on one dialed connection. +type countingConn struct { + net.Conn + closes atomic.Int32 +} + +func (c *countingConn) Close() error { + c.closes.Add(1) + return c.Conn.Close() +} + +// connRecorder is a WithDialer dialer that records every connection it +// opens. When gate is non-nil it is called with the 1-based dial number +// and may return a channel the dial then waits on. +type connRecorder struct { + mu sync.Mutex + conns []*countingConn + gate func(n int) <-chan struct{} +} + +func (r *connRecorder) dial(ctx context.Context, network, addr string) (net.Conn, error) { + r.mu.Lock() + n := len(r.conns) + 1 + gate := r.gate + r.mu.Unlock() + if gate != nil { + if ch := gate(n); ch != nil { + <-ch + } + } + var d net.Dialer + raw, err := d.DialContext(ctx, network, addr) + if err != nil { + return nil, err + } + cc := &countingConn{Conn: raw} + r.mu.Lock() + r.conns = append(r.conns, cc) + r.mu.Unlock() + return cc, nil +} + +func (r *connRecorder) all() []*countingConn { + r.mu.Lock() + defer r.mu.Unlock() + return append([]*countingConn(nil), r.conns...) +} + +// logCapture is a slog.Handler that records every record at every level. +type logCapture struct { + mu sync.Mutex + recs []capturedLog +} + +type capturedLog struct { + level slog.Level + msg string + attrs map[string]any +} + +func (h *logCapture) Enabled(context.Context, slog.Level) bool { return true } +func (h *logCapture) WithAttrs([]slog.Attr) slog.Handler { return h } +func (h *logCapture) WithGroup(string) slog.Handler { return h } +func (h *logCapture) Handle(_ context.Context, r slog.Record) error { + rec := capturedLog{level: r.Level, msg: r.Message, attrs: map[string]any{}} + r.Attrs(func(a slog.Attr) bool { + rec.attrs[a.Key] = a.Value.Resolve().Any() + return true + }) + h.mu.Lock() + h.recs = append(h.recs, rec) + h.mu.Unlock() + return nil +} + +func (h *logCapture) find(level slog.Level, msg string) []capturedLog { + h.mu.Lock() + defer h.mu.Unlock() + var out []capturedLog + for _, r := range h.recs { + if r.level == level && r.msg == msg { + out = append(out, r) + } + } + return out +} + +func (h *logCapture) count(level slog.Level) int { + h.mu.Lock() + defer h.mu.Unlock() + n := 0 + for _, r := range h.recs { + if r.level == level { + n++ + } + } + return n +} + +// testPolicy is a reconnect policy with test-sized timings. +func testPolicy() *reconnectPolicy { + return &reconnectPolicy{ + dialBase: 5 * time.Millisecond, + dialMax: 20 * time.Millisecond, + cooldownBase: 10 * time.Millisecond, + cooldownMax: 80 * time.Millisecond, + rapidWindow: 30 * time.Second, + quietReset: time.Hour, + summaryEvery: time.Hour, + // Generous: tests that need a hung other conn shrink it. + otherConnTimeout: 5 * time.Second, + } +} + +// instrument points c at a log capture and a test policy. It must run +// before c is used concurrently. +func instrument(c *Client, p *reconnectPolicy) *logCapture { + h := &logCapture{} + c.logger = slog.New(h) + c.reconn.policy = p + return h +} + +func send(c *Client, typ string) (map[string]interface{}, error) { + return c.Send(map[string]interface{}{"type": typ}) +} + +func mustOK(t *testing.T, c *Client, typ string) { + t.Helper() + resp, err := send(c, typ) + if err != nil { + t.Fatalf("send %q: %v", typ, err) + } + if resp["type"] != "ok" { + t.Fatalf("send %q: resp %v", typ, resp) + } +} + +func assertClosedErr(t *testing.T, err error) { + t.Helper() + if err == nil { + t.Fatal("expected an error after Close, got nil") + } + if !errors.Is(err, ErrClosed) { + t.Fatalf("err = %v, want errors.Is(err, ErrClosed)", err) + } + if !errors.Is(err, ErrNoRegistry) { + t.Fatalf("err = %v, want errors.Is(err, ErrNoRegistry)", err) + } + if !strings.Contains(err.Error(), "closed") { + t.Fatalf("err = %q, want it to mention closed", err) + } +} + +// --- Close: idempotent, race-free --------------------------------------------- + +func TestErrClosedMatchesErrNoRegistry(t *testing.T) { + t.Parallel() + if !errors.Is(ErrClosed, ErrNoRegistry) { + t.Fatal("errors.Is(ErrClosed, ErrNoRegistry) = false") + } + if errors.Is(ErrNoRegistry, ErrClosed) { + t.Fatal("ErrNoRegistry must not match ErrClosed") + } + wrapped := closedDuring(io.EOF) + if !errors.Is(wrapped, ErrClosed) || !errors.Is(wrapped, io.EOF) { + t.Fatalf("closedDuring(EOF) = %v, want both ErrClosed and io.EOF in the chain", wrapped) + } + if closedDuring(nil) != ErrClosed || closedDuring(ErrClosed) != ErrClosed { + t.Fatal("closedDuring(nil / ErrClosed) must return ErrClosed itself") + } +} + +// TestCloseTwiceDoesNotPanic is the direct regression for the nightly +// "panic: close of closed channel" at Client.Close: a pooled client closed +// once by forceReconnectRegistry's `go old.Close()` and once by Stop. +func TestCloseTwiceDoesNotPanic(t *testing.T) { + t.Parallel() + for _, size := range []int{1, 4} { + srv := newScriptedRegistry(t, neverDrop) + c, err := DialPool(srv.addr(), size) + if err != nil { + t.Fatalf("DialPool(%d): %v", size, err) + } + mustOK(t, c, "warm") + if err := c.Close(); err != nil { + t.Fatalf("size %d: first Close: %v", size, err) + } + if err := c.Close(); err != nil { + t.Fatalf("size %d: second Close = %v, want nil", size, err) + } + _, err = send(c, "after-close") + assertClosedErr(t, err) + } +} + +// TestCloseRacesCloseAndSend reproduces the daemon shape under -race: many +// goroutines hammer Send while two goroutines Close the same client (Stop +// and a forced reconnect retiring it). Nothing may panic, every failed Send +// must report the client closed, and Send after Close must keep doing so. +func TestCloseRacesCloseAndSend(t *testing.T) { + t.Parallel() + ctors := map[string]func(addr string) (*Client, error){ + "Dial": func(a string) (*Client, error) { return Dial(a) }, + "DialPool": func(a string) (*Client, error) { return DialPool(a, 4) }, + } + for name, dial := range ctors { + t.Run(name, func(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, neverDrop) + for iter := 0; iter < 20; iter++ { + c, err := dial(srv.addr()) + if err != nil { + t.Fatalf("dial: %v", err) + } + instrument(c, testPolicy()) + + var wg sync.WaitGroup + start := make(chan struct{}) + errs := make(chan error, 64) + for g := 0; g < 8; g++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + for { + if _, err := send(c, "x"); err != nil { + errs <- err + return + } + } + }() + } + closeErrs := make(chan error, 3) + for g := 0; g < 3; g++ { + wg.Add(1) + go func() { + defer wg.Done() + <-start + time.Sleep(time.Millisecond) + closeErrs <- c.Close() + }() + } + close(start) + wg.Wait() + close(errs) + close(closeErrs) + for err := range errs { + assertClosedErr(t, err) + } + for err := range closeErrs { + if err != nil { + t.Fatalf("Close: %v", err) + } + } + _, err = send(c, "after") + assertClosedErr(t, err) + } + }) + } +} + +// TestCloseClosesEveryPoolConnExactlyOnce checks that Close (called several +// times, concurrently) closes each pooled connection exactly once, including +// a connection that replaced a dropped one, and that the dropped connection +// itself was closed exactly once. +func TestCloseClosesEveryPoolConnExactlyOnce(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq == 1 && reqSeq == 1 // drop the primary's first request + }) + rec := &connRecorder{} + c, err := DialPool(srv.addr(), 3, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + instrument(c, testPolicy()) + release := holdAllBut(t, c, c.pool.entries[0]) // only the primary is free + mustOK(t, c, "drop-then-redial") + release() + if got := len(rec.all()); got != 4 { + t.Fatalf("dials = %d, want 4 (3 pool conns + 1 redial)", got) + } + + var wg sync.WaitGroup + for i := 0; i < 4; i++ { + wg.Add(1) + go func() { defer wg.Done(); _ = c.Close() }() + } + wg.Wait() + _ = c.Close() + + for i, cc := range rec.all() { + if n := cc.closes.Load(); n != 1 { + t.Errorf("conn %d closed %d times, want exactly 1", i, n) + } + } +} + +// TestCloseDuringRedialClosesLateConn: Close runs while a pool entry is being +// redialed. The redial completes after Close; its connection must be closed +// (not leaked into a closed client) and the request must report ErrClosed. +func TestCloseDuringRedialClosesLateConn(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq == 1 && reqSeq == 1 + }) + gate := make(chan struct{}) + entered := make(chan struct{}) + rec := &connRecorder{gate: func(n int) <-chan struct{} { + if n == 3 { // the redial after the two initial pool conns + close(entered) + return gate + } + return nil + }} + c, err := DialPool(srv.addr(), 2, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + instrument(c, testPolicy()) + holdAllBut(t, c, c.pool.entries[0]) // the primary is the conn the server drops + + done := make(chan error, 1) + go func() { + _, err := send(c, "x") + done <- err + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("redial never started") + } + + closed := make(chan error, 1) + go func() { closed <- c.Close() }() + select { + case err := <-closed: + if err != nil { + t.Fatalf("Close: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Close blocked on an in-flight redial") + } + close(gate) + + select { + case err := <-done: + assertClosedErr(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Send did not return after Close") + } + conns := rec.all() + if len(conns) != 3 { + t.Fatalf("dials = %d, want 3", len(conns)) + } + for i, cc := range conns { + if n := cc.closes.Load(); n != 1 { + t.Errorf("conn %d closed %d times, want exactly 1", i, n) + } + } +} + +// TestCloseWakesReconnectBackoff: a request stuck in reconnect backoff +// against a dead registry returns promptly with ErrClosed when the client is +// closed, on both the pooled and the single-conn path (where the backoff +// runs with c.mu held and would otherwise block Close). +func TestCloseWakesReconnectBackoff(t *testing.T) { + t.Parallel() + for _, size := range []int{1, 2} { + t.Run(fmt.Sprintf("size=%d", size), func(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(0)) + c, err := DialPool(srv.addr(), size) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + p := testPolicy() + p.dialBase, p.dialMax = time.Minute, time.Minute + instrument(c, p) + srv.waitConns(t, size) + srv.shutdown() // every request now fails and every redial is refused + + done := make(chan error, 1) + go func() { + _, err := send(c, "x") + done <- err + }() + // Let the request fail, dial once and park in the backoff. + time.Sleep(100 * time.Millisecond) + start := time.Now() + closed := make(chan error, 1) + go func() { closed <- c.Close() }() + select { + case err := <-done: + assertClosedErr(t, err) + case <-time.After(5 * time.Second): + t.Fatal("Send stayed in reconnect backoff after Close") + } + select { + case <-closed: + case <-time.After(5 * time.Second): + t.Fatal("Close blocked behind the reconnect backoff") + } + if d := time.Since(start); d > 5*time.Second { + t.Fatalf("Close/Send took %v", d) + } + }) + } +} + +// TestSendWaitingForFreeEntryHonoursContext: a pooled Send blocked because +// every entry is busy returns when its context is done. +func TestSendWaitingForFreeEntryHonoursContext(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, neverDrop) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + holdIdleEntries(t, c, 2) + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + if _, err := c.SendContext(ctx, map[string]interface{}{"type": "x"}); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("err = %v, want context.DeadlineExceeded", err) + } +} + +func TestBinaryClientCloseIdempotentAndErrClosed(t *testing.T) { + t.Parallel() + srv := newFakeBinaryServer(t, func(byte, []byte) (byte, []byte) { + return wire.MsgHeartbeatOK, wire.EncodeHeartbeatResp(1, false) + }) + c, err := DialBinary(srv.addr()) + if err != nil { + t.Fatalf("DialBinary: %v", err) + } + if _, _, err := c.Heartbeat(1, nil); err != nil { + t.Fatalf("Heartbeat: %v", err) + } + var wg sync.WaitGroup + for i := 0; i < 3; i++ { + wg.Add(1) + go func() { defer wg.Done(); _ = c.Close() }() + } + wg.Wait() + if err := c.Close(); err != nil { + t.Fatalf("second Close = %v, want nil", err) + } + _, _, err = c.Heartbeat(1, nil) + assertClosedErr(t, err) + _, err = c.Lookup(1) + assertClosedErr(t, err) + _, err = c.Resolve(1, 2, nil) + assertClosedErr(t, err) + _, err = c.SendJSON(map[string]interface{}{"type": "x"}) + assertClosedErr(t, err) + var nilClient *BinaryClient + if err := nilClient.Close(); err != nil { + t.Fatalf("nil BinaryClient Close = %v", err) + } +} + +// TestBinaryClientCloseWakesReconnectBackoff: BinaryClient.reconnect sleeps +// with c.mu held; Close must wake it instead of waiting out the backoff. +func TestBinaryClientCloseWakesReconnectBackoff(t *testing.T) { + t.Parallel() + srv := newFakeBinaryServer(t, func(byte, []byte) (byte, []byte) { + return wire.MsgHeartbeatOK, wire.EncodeHeartbeatResp(1, false) + }) + c, err := DialBinary(srv.addr()) + if err != nil { + t.Fatalf("DialBinary: %v", err) + } + srv.Close() + c.mu.Lock() + _ = c.conn.Close() + c.mu.Unlock() + + done := make(chan error, 1) + go func() { + _, _, err := c.Heartbeat(1, nil) + done <- err + }() + time.Sleep(100 * time.Millisecond) // first redial refused; now in backoff + closed := make(chan error, 1) + go func() { closed <- c.Close() }() + select { + case err := <-done: + assertClosedErr(t, err) + case <-time.After(3 * time.Second): + t.Fatal("Heartbeat stayed in reconnect backoff after Close") + } + select { + case <-closed: + case <-time.After(3 * time.Second): + t.Fatal("Close blocked behind the reconnect backoff") + } +} + +// --- Reconnect: triggering error, logging, retry on another entry ------------- + +// TestPoolReconnectLogsCauseAtDebugWithInfoSummary: a server that closes +// every connection after one request makes each request redial. Each +// reconnect is logged at Debug with the error that triggered it; nothing is +// logged per reconnect at INFO; one INFO summary (flushed by Close) carries +// the counts and the last cause. +func TestPoolReconnectLogsCauseAtDebugWithInfoSummary(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(1)) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + logs := instrument(c, testPolicy()) + release := holdIdleEntries(t, c, 1) + + const sends = 6 + for i := 0; i < sends; i++ { + mustOK(t, c, "x") + } + release() + + if got := logs.count(slog.LevelInfo); got != 0 { + t.Fatalf("%d INFO records before the summary interval, want 0: %+v", got, logs.find(slog.LevelInfo, "registry reconnect summary")) + } + reconnected := logs.find(slog.LevelDebug, "registry pool conn reconnected") + if len(reconnected) != sends-1 { + t.Fatalf("Debug reconnect records = %d, want %d", len(reconnected), sends-1) + } + for _, r := range reconnected { + if cause, _ := r.attrs["cause"].(string); !strings.Contains(cause, "EOF") { + t.Fatalf("reconnect record cause = %q, want the triggering EOF", cause) + } + } + dropped := logs.find(slog.LevelDebug, "registry conn dropped") + if len(dropped) != sends-1 { + t.Fatalf("Debug drop records = %d, want %d", len(dropped), sends-1) + } + + if err := c.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + summaries := logs.find(slog.LevelInfo, "registry reconnect summary") + if len(summaries) != 1 { + t.Fatalf("INFO summaries = %d, want 1 (flushed by Close)", len(summaries)) + } + s := summaries[0].attrs + if s["reconnects"] != int64(sends-1) || s["rapid_closes"] != int64(sends-1) { + t.Fatalf("summary counts = %v", s) + } + if s["rate_limit_suspected"] != true { + t.Fatalf("summary rate_limit_suspected = %v, want true", s["rate_limit_suspected"]) + } + if cause, _ := s["last_cause"].(string); !strings.Contains(cause, "EOF") { + t.Fatalf("summary last_cause = %q", cause) + } +} + +// TestReconnectSummaryIsPeriodic: with a short summary interval, INFO +// summaries appear at most once per interval, however many reconnects +// happen. +func TestReconnectSummaryIsPeriodic(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(1)) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.summaryEvery = 150 * time.Millisecond + p.cooldownBase = 0 // keep the loop fast; backoff is tested separately + logs := instrument(c, p) + holdIdleEntries(t, c, 1) + + start := time.Now() + sends := 0 + for time.Since(start) < 500*time.Millisecond { + mustOK(t, c, "x") + sends++ + } + elapsed := time.Since(start) + summaries := logs.find(slog.LevelInfo, "registry reconnect summary") + maxWant := int(elapsed/p.summaryEvery) + 1 + if len(summaries) == 0 || len(summaries) > maxWant { + t.Fatalf("%d summaries for %d reconnects in %v, want 1..%d", len(summaries), sends-1, elapsed, maxWant) + } + if got := len(logs.find(slog.LevelInfo, "registry pool conn reconnected")); got != 0 { + t.Fatalf("%d per-reconnect INFO lines, want 0", got) + } + total := 0 + for _, s := range summaries { + total += int(s.attrs["reconnects"].(int64)) + } + if total > sends-1 || total == 0 { + t.Fatalf("summaries report %d reconnects, %d happened", total, sends-1) + } +} + +// TestSinglePathReconnectLogsCauseAtDebug covers the non-pooled client. +func TestSinglePathReconnectLogsCauseAtDebug(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(1)) + c, err := Dial(srv.addr()) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer c.Close() + logs := instrument(c, testPolicy()) + mustOK(t, c, "a") + mustOK(t, c, "b") + recs := logs.find(slog.LevelDebug, "registry reconnected") + if len(recs) != 1 { + t.Fatalf("Debug reconnect records = %d, want 1", len(recs)) + } + if cause, _ := recs[0].attrs["cause"].(string); !strings.Contains(cause, "EOF") { + t.Fatalf("cause = %q, want EOF", cause) + } + if got := logs.count(slog.LevelInfo); got != 0 { + t.Fatalf("%d INFO records, want 0", got) + } +} + +// TestReconnectFailureKeepsTriggeringAndReconnectErrors: when the redial +// fails too, the returned error carries both the error that triggered the +// reconnect and the reconnect's own error. +func TestReconnectFailureKeepsTriggeringAndReconnectErrors(t *testing.T) { + t.Parallel() + for _, size := range []int{1, 2} { + t.Run(fmt.Sprintf("size=%d", size), func(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(0)) + c, err := DialPool(srv.addr(), size) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + logs := instrument(c, testPolicy()) + if size > 1 { + holdIdleEntries(t, c, size-1) + } + srv.waitConns(t, size) + srv.ln.Close() // the drop still happens; the redial is refused + + _, err = send(c, "x") + if err == nil { + t.Fatal("expected an error") + } + if !errors.Is(err, io.EOF) { + t.Fatalf("err = %v, want the triggering io.EOF in the chain", err) + } + if !errors.Is(err, syscall.ECONNREFUSED) { + t.Fatalf("err = %v, want the reconnect's ECONNREFUSED in the chain", err) + } + for _, want := range []string{"send failed and reconnect failed", "recv: EOF", "reconnect failed after 5 attempts"} { + if !strings.Contains(err.Error(), want) { + t.Fatalf("err = %q, want it to contain %q", err, want) + } + } + warns := 0 + for _, lvl := range []slog.Level{slog.LevelWarn} { + for _, msg := range []string{"registry reconnect failed", "registry pool conn reconnect failed"} { + for _, r := range logs.find(lvl, msg) { + warns++ + if cause, _ := r.attrs["cause"].(string); !strings.Contains(cause, "EOF") { + t.Fatalf("reconnect-failed WARN cause = %q, want EOF", cause) + } + } + } + } + if warns != maxReconnectAttempts { + t.Fatalf("reconnect-failed WARN records = %d, want %d", warns, maxReconnectAttempts) + } + }) + } +} + +// TestRetryAfterReconnectFailureKeepsTriggeringError: the redial succeeds +// but the retry is dropped too; the error names both failures. +func TestRetryAfterReconnectFailureKeepsTriggeringError(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(0)) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + instrument(c, testPolicy()) + holdIdleEntries(t, c, 1) + _, err = send(c, "x") + if err == nil || !strings.Contains(err.Error(), "retried after reconnect; first failure: recv: EOF") { + t.Fatalf("err = %v, want the retry error annotated with the first failure", err) + } + if !errors.Is(err, io.EOF) { + t.Fatalf("err = %v, want io.EOF in the chain", err) + } +} + +// TestPoolRetriesOnOtherHealthyEntryBeforeRedial: after a closed-connection +// error on one entry, the request is retried on another idle healthy entry +// with no redial. The dropped entry is skipped while healthy entries are +// free and redialed lazily when it is next needed. +func TestPoolRetriesOnOtherHealthyEntryBeforeRedial(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq == 1 && reqSeq == 1 + }) + rec := &connRecorder{} + c, err := DialPool(srv.addr(), 2, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + logs := instrument(c, testPolicy()) + + // Make sure the first request lands on the primary (the conn the + // server drops): drain the free list and put the primary back first. + <-c.pool.free + <-c.pool.free + c.releaseEntry(c.pool.entries[0]) + c.releaseEntry(c.pool.entries[1]) + + mustOK(t, c, "dropped-then-served-by-other") + if got := len(rec.all()); got != 2 { + t.Fatalf("dials = %d, want 2 (served by the other entry, no redial)", got) + } + if got := srv.dropped.Load(); got != 1 { + t.Fatalf("server drops = %d, want 1", got) + } + if n := rec.all()[0].closes.Load(); n != 1 { + t.Fatalf("dropped conn closed %d times, want 1 (retired right away)", n) + } + + // The healthy entry is preferred: still no redial. + mustOK(t, c, "healthy-preferred") + if got := len(rec.all()); got != 2 { + t.Fatalf("dials = %d after a second request, want 2", got) + } + + // With the healthy entry busy, the broken one is redialed lazily. + release := holdAllBut(t, c, c.pool.entries[0]) + mustOK(t, c, "lazy-redial") + release() + if got := len(rec.all()); got != 3 { + t.Fatalf("dials = %d, want 3 after the broken entry was needed", got) + } + if !c.pool.entries[0].healthy() || !c.pool.entries[1].healthy() { + t.Fatal("both entries should be healthy again") + } + c.mu.Lock() + primarySynced := c.conn == c.pool.entries[0].conn + c.mu.Unlock() + if !primarySynced { + t.Fatal("c.conn out of sync with the redialed primary entry") + } + + if err := c.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + s := logs.find(slog.LevelInfo, "registry reconnect summary") + if len(s) != 1 || s[0].attrs["served_by_other_conn"] != int64(1) { + t.Fatalf("summary = %+v, want served_by_other_conn=1", s) + } +} + +// TestPoolFallsBackToRedialWhenOtherEntryAlsoDropped: the other entry is +// dead too, so after one retry on it the client redials and succeeds. +func TestPoolFallsBackToRedialWhenOtherEntryAlsoDropped(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq <= 2 && reqSeq == 1 // both initial conns die on first use + }) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + instrument(c, testPolicy()) + mustOK(t, c, "x") + if got := srv.dropped.Load(); got != 2 { + t.Fatalf("server drops = %d, want 2 (first entry, then the other)", got) + } + if got := srv.conns.Load(); got != 3 { + t.Fatalf("server conns = %d, want 3 (one redial)", got) + } +} + +// TestPoolDoesNotRetryTimeoutOnOtherEntry: a timeout is not a +// closed-connection error, so it goes straight to the redial. +func TestPoolDoesNotRetryTimeoutOnOtherEntry(t *testing.T) { + t.Parallel() + if isConnDropped(fmt.Errorf("recv: %w", os.ErrDeadlineExceeded)) { + t.Fatal("a read timeout must not count as a dropped connection") + } +} + +// primaryFirst reorders a two-entry pool's free list so the next request +// lands on the primary entry (connSeq 1 on the server). +func primaryFirst(t *testing.T, c *Client) { + t.Helper() + if len(c.pool.entries) != 2 { + t.Fatalf("pool size %d, want 2", len(c.pool.entries)) + } + <-c.pool.free + <-c.pool.free + c.releaseEntry(c.pool.entries[0]) + c.releaseEntry(c.pool.entries[1]) +} + +// hangsOn returns a hang hook for every request on connection connSeq. +func hangsOn(connSeq int) func(int, int) bool { + return func(cs, _ int) bool { return cs == connSeq } +} + +// TestPoolOtherEntryRetryIsBoundedWhenOtherConnHangs: the request's conn +// is dropped (a limiter denial) and the other idle conn is silently hung. +// The retry on the other conn gives up after otherConnTimeout instead of +// the 30 s read deadline, retires that conn (a late reply could still +// arrive on it), and the request is served on the redialed conn. +func TestPoolOtherEntryRetryIsBoundedWhenOtherConnHangs(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq == 1 && reqSeq == 1 + }) + srv.setHang(hangsOn(2)) + rec := &connRecorder{} + c, err := DialPool(srv.addr(), 2, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.otherConnTimeout = 150 * time.Millisecond + logs := instrument(c, p) + primaryFirst(t, c) + + start := time.Now() + mustOK(t, c, "x") + elapsed := time.Since(start) + if elapsed < p.otherConnTimeout { + t.Fatalf("Send took %v, want >= otherConnTimeout %v (the other conn was not tried)", elapsed, p.otherConnTimeout) + } + if elapsed > 5*time.Second { + t.Fatalf("Send took %v: the retry on the hung conn was not bounded by otherConnTimeout", elapsed) + } + if got := srv.received.Load(); got != 3 { + t.Fatalf("server received the request %d times, want 3 (dropped, hung, redialed)", got) + } + if got := srv.hung.Load(); got != 1 { + t.Fatalf("hung requests = %d, want 1", got) + } + if c.pool.entries[1].healthy() { + t.Fatal("the hung entry is still marked healthy; a late reply could reach the next request") + } + conns := rec.all() + if len(conns) != 3 { + t.Fatalf("dials = %d, want 3 (one redial)", len(conns)) + } + if n := conns[1].closes.Load(); n != 1 { + t.Fatalf("hung conn closed %d times, want 1 (retired)", n) + } + timedOut := false + for _, r := range logs.find(slog.LevelDebug, "registry conn dropped") { + if e, ok := r.attrs["err"].(error); ok && errors.Is(e, os.ErrDeadlineExceeded) { + timedOut = true + } + } + if !timedOut { + t.Fatal("no Debug record for the other conn's timeout") + } + + // The redialed primary is healthy and preferred: no further dial. + mustOK(t, c, "after") + if got := len(rec.all()); got != 3 { + t.Fatalf("dials = %d after a second request, want 3", got) + } +} + +// TestPoolOtherEntryRetryHonoursContext: with a long otherConnTimeout, the +// caller's ctx (its deadline or its cancellation) ends the retry on a hung +// other conn, and no redial is attempted once ctx is done. The error keeps +// the triggering drop and matches the ctx error. +func TestPoolOtherEntryRetryHonoursContext(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + ctx func() (context.Context, context.CancelFunc) + want error + }{ + {"deadline", func() (context.Context, context.CancelFunc) { + return context.WithTimeout(context.Background(), 150*time.Millisecond) + }, context.DeadlineExceeded}, + {"cancel", func() (context.Context, context.CancelFunc) { + ctx, cancel := context.WithCancel(context.Background()) + time.AfterFunc(150*time.Millisecond, cancel) + return ctx, cancel + }, context.Canceled}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq == 1 && reqSeq == 1 + }) + srv.setHang(hangsOn(2)) + rec := &connRecorder{} + c, err := DialPool(srv.addr(), 2, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.otherConnTimeout = time.Minute // only ctx can end the retry + logs := instrument(c, p) + primaryFirst(t, c) + + ctx, cancel := tc.ctx() + defer cancel() + start := time.Now() + _, err = c.SendContext(ctx, map[string]interface{}{"type": "x"}) + elapsed := time.Since(start) + if err == nil { + t.Fatal("expected an error once ctx ended") + } + if elapsed > 5*time.Second { + t.Fatalf("SendContext took %v: ctx did not end the retry on the hung conn", elapsed) + } + if !errors.Is(err, tc.want) { + t.Fatalf("err = %v, want errors.Is(err, %v)", err, tc.want) + } + if !errors.Is(err, io.EOF) { + t.Fatalf("err = %v, want the triggering io.EOF in the chain", err) + } + if got := len(rec.all()); got != 2 { + t.Fatalf("dials = %d, want 2 (no redial after ctx ended)", got) + } + if c.pool.entries[1].healthy() { + t.Fatal("the abandoned entry is still marked healthy") + } + if w := logs.find(slog.LevelWarn, "registry pool conn reconnect failed"); len(w) != 0 { + t.Fatalf("%d reconnect-failed WARNs for a caller that gave up, want 0", len(w)) + } + c.reconn.mu.Lock() + fs := c.reconn.failStreak + c.reconn.mu.Unlock() + if fs != 0 { + t.Fatalf("failStreak = %d, want 0: ctx ending is not a registry failure", fs) + } + }) + } +} + +// TestPoolRequestSentAtMostThreeTimesAndExtendsStreakOnce: every conn a +// request is tried on drops it. The request reaches the registry at most +// three times (its conn, another idle conn, its redialed conn), and the +// rapid-close streak grows by one for the whole request, so one shed +// request does not double the next cooldown. Every close still shows in +// the summary. +func TestPoolRequestSentAtMostThreeTimesAndExtendsStreakOnce(t *testing.T) { + t.Parallel() + for _, tc := range []struct { + name string + dropping int // conns (by accept order) that drop their first request + wantOK bool + }{ + {"served-after-redial", 2, true}, + {"every-attempt-dropped", 3, false}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, func(connSeq, reqSeq int, _ time.Duration) bool { + return connSeq <= tc.dropping && reqSeq == 1 + }) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + logs := instrument(c, testPolicy()) + _, err = send(c, "register") + if tc.wantOK && err != nil { + t.Fatalf("send: %v", err) + } + if !tc.wantOK && (err == nil || !strings.Contains(err.Error(), "retried after reconnect")) { + t.Fatalf("err = %v, want the redialed retry's failure", err) + } + if got := srv.received.Load(); got != 3 { + t.Fatalf("server received the request %d times, want 3", got) + } + c.reconn.mu.Lock() + streak := c.reconn.rapidStreak + c.reconn.mu.Unlock() + if streak != 1 { + t.Fatalf("rapid-close streak = %d after one request, want 1", streak) + } + if err := c.Close(); err != nil { + t.Fatalf("Close: %v", err) + } + s := logs.find(slog.LevelInfo, "registry reconnect summary") + if len(s) != 1 || s[0].attrs["rapid_closes"] != int64(tc.dropping) { + t.Fatalf("summary = %+v, want rapid_closes=%d", s, tc.dropping) + } + }) + } +} + +// TestSinglePathRequestSentAtMostTwice: the single-conn client redials +// and retries once, so a request reaches the registry at most twice. +func TestSinglePathRequestSentAtMostTwice(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(0)) + c, err := Dial(srv.addr()) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer c.Close() + instrument(c, testPolicy()) + if _, err := send(c, "register"); err == nil { + t.Fatal("expected an error: every conn drops") + } + if got := srv.received.Load(); got != 2 { + t.Fatalf("server received the request %d times, want 2", got) + } +} + +// TestBoundedRoundTripLeavesNoDeadline: a ctx cancelled at varying points +// around a bounded exchange either fails it with the cancellation or +// leaves no deadline on the conn that answered, so its next plain round +// trip works. (Both outcomes occur across the iterations.) +func TestBoundedRoundTripLeavesNoDeadline(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, neverDrop) + dial := func() net.Conn { + conn, err := net.Dial("tcp", srv.addr()) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { conn.Close() }) + return conn + } + conn := dial() + msg := map[string]interface{}{"type": "x"} + answered := 0 + for i := 0; i < 200; i++ { + ctx, cancel := context.WithCancel(context.Background()) + go func(d time.Duration) { + time.Sleep(d) + cancel() + }(time.Duration(i%50) * 10 * time.Microsecond) + _, err := boundedRoundTrip(ctx, conn, msg, time.Second) + cancel() + if err != nil { + if !errors.Is(err, context.Canceled) { + t.Fatalf("iteration %d: err = %v, want a cancellation", i, err) + } + conn = dial() // interrupted mid-exchange: retire it + continue + } + answered++ + time.Sleep(50 * time.Microsecond) // let a late cancellation run + if _, err := roundTrip(conn, msg); err != nil { + t.Fatalf("iteration %d: plain round trip after a bounded one: %v", i, err) + } + } + if answered == 0 { + t.Fatal("no bounded exchange completed; the test did not exercise the race") + } +} + +// TestDefaultOtherConnTimeoutFitsDaemonDeadline: the other-conn retry +// plus the longest cooldown leaves most of the daemon's 8 s registry call +// deadline for the redial and the final attempt. +func TestDefaultOtherConnTimeoutFitsDaemonDeadline(t *testing.T) { + t.Parallel() + p := defaultReconnectPolicy + if p.otherConnTimeout <= 0 { + t.Fatalf("otherConnTimeout = %v, want > 0", p.otherConnTimeout) + } + if budget := p.otherConnTimeout + p.cooldownMax; budget > 4*time.Second { + t.Fatalf("otherConnTimeout + cooldownMax = %v, want <= 4s (half the daemon's 8s deadline)", budget) + } +} + +// --- Backoff ------------------------------------------------------------------- + +// TestRapidClosesBackOffExponentiallyWithJitter: a server that closes every +// connection after one request (the rate-limit signature: EOF, no error +// frame, on a young connection) makes the client wait a growing, jittered +// cooldown before each redial. Timing is checked against lower bounds only, +// which a slow machine cannot break. +func TestRapidClosesBackOffExponentiallyWithJitter(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(1)) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.cooldownBase, p.cooldownMax = 20*time.Millisecond, 160*time.Millisecond + instrument(c, p) + holdIdleEntries(t, c, 1) + + mustOK(t, c, "first") // uses up the primary's one request + start := time.Now() + const redials = 5 + for i := 0; i < redials; i++ { + mustOK(t, c, "x") + } + elapsed := time.Since(start) + + // Streak n waits jitter(min(base<<(n-1), max)) >= half of that. + var minWait time.Duration + for n := 1; n <= redials; n++ { + d := p.cooldownBase << (n - 1) + if d > p.cooldownMax { + d = p.cooldownMax + } + minWait += d / 2 + } + if elapsed < minWait { + t.Fatalf("%d rapid-close redials took %v, want >= %v of backoff", redials, elapsed, minWait) + } + c.reconn.mu.Lock() + streak := c.reconn.rapidStreak + c.reconn.mu.Unlock() + if streak != redials { + t.Fatalf("rapid-close streak = %d, want %d", streak, redials) + } +} + +// TestGracePeriodClosesAreDetectedAndStreakResets mirrors production: the +// registry closes the first request on a connection older than its grace +// period. Requests keep succeeding (retry), the closes are classified as +// rapid, and once the server stops shedding and a connection outlives the +// rapid-close window, the streak resets. +func TestGracePeriodClosesAreDetectedAndStreakResets(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, afterGrace(60*time.Millisecond)) + c, err := DialPool(srv.addr(), 2) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.rapidWindow = 250 * time.Millisecond + logs := instrument(c, p) + + deadline := time.Now().Add(400 * time.Millisecond) + for time.Now().Before(deadline) { + mustOK(t, c, "x") + time.Sleep(20 * time.Millisecond) + } + if srv.dropped.Load() == 0 { + t.Fatal("server never shed a request; test did not exercise the grace path") + } + rapid := 0 + for _, r := range logs.find(slog.LevelDebug, "registry conn dropped") { + if r.attrs["rapid_close"] == true { + rapid++ + } + } + if rapid == 0 { + t.Fatal("no drop was classified as a rapid close") + } + + // Stop shedding; a request on a connection older than the rapid-close + // window proves the registry is healthy again. + srv.setDrop(neverDrop) + time.Sleep(p.rapidWindow + 50*time.Millisecond) + mustOK(t, c, "healthy") + mustOK(t, c, "healthy") + c.reconn.mu.Lock() + streak := c.reconn.rapidStreak + c.reconn.mu.Unlock() + if streak != 0 { + t.Fatalf("rapid-close streak = %d after a healthy old connection, want 0", streak) + } + if d := c.reconn.cooldown(); d != 0 { + t.Fatalf("cooldown = %v after reset, want 0", d) + } +} + +// TestIdleCloseDoesNotBackOff: a connection closed by the peer after it +// outlived the rapid-close window (an idle timeout, a NAT drop) is redialed +// with no cooldown. +func TestIdleCloseDoesNotBackOff(t *testing.T) { + t.Parallel() + var tr reconnectTracker + p := testPolicy() + tr.policy = p + if tr.connError(fmt.Errorf("recv: %w", io.EOF), time.Minute, true, nil) { + t.Fatal("EOF on a minute-old conn classified as rapid") + } + if d := tr.cooldown(); d != 0 { + t.Fatalf("cooldown after an idle close = %v, want 0", d) + } + if tr.connError(fmt.Errorf("recv: %w", os.ErrDeadlineExceeded), time.Second, true, nil) { + t.Fatal("a timeout classified as a rapid close") + } + if tr.connError(fmt.Errorf("send: %w", net.ErrClosed), time.Second, true, nil) { + t.Fatal("a locally closed conn classified as a rapid close") + } +} + +// TestRepeatedReconnectFailuresBackOff: each reconnect that exhausts its +// dial attempts grows the cooldown before the next one; a successful redial +// ends that streak. +func TestRepeatedReconnectFailuresBackOff(t *testing.T) { + t.Parallel() + srv := newScriptedRegistry(t, closeAfter(0)) + rec := &connRecorder{} + c, err := DialPool(srv.addr(), 2, WithDialer(rec.dial)) + if err != nil { + t.Fatalf("DialPool: %v", err) + } + defer c.Close() + p := testPolicy() + p.rapidWindow = time.Nanosecond // isolate the failure streak + instrument(c, p) + holdIdleEntries(t, c, 1) + realAddr := srv.addr() + srv.waitConns(t, 2) + srv.ln.Close() + + for i := 1; i <= 3; i++ { + if _, err := send(c, "x"); err == nil { + t.Fatal("expected a failure while the registry is down") + } + c.reconn.mu.Lock() + fs := c.reconn.failStreak + c.reconn.mu.Unlock() + if fs != i { + t.Fatalf("after %d failed reconnects failStreak = %d", i, fs) + } + } + for i := 0; i < 20; i++ { + if d := c.reconn.cooldown(); d < p.cooldownBase<<2/2 || d > p.cooldownBase<<2 { + t.Fatalf("cooldown at failStreak 3 = %v, want in [%v, %v]", d, p.cooldownBase<<2/2, p.cooldownBase<<2) + } + } + + // Bring a registry back on the same address and redial successfully. + ln, err := net.Listen("tcp", realAddr) + if err != nil { + t.Skipf("could not rebind %s: %v", realAddr, err) + } + srv2 := &scriptedRegistry{ln: ln, drop: neverDrop} + srv2.wg.Add(1) + go srv2.accept() + t.Cleanup(srv2.stop) + mustOK(t, c, "back") + c.reconn.mu.Lock() + fs := c.reconn.failStreak + c.reconn.mu.Unlock() + if fs != 0 { + t.Fatalf("failStreak = %d after a successful redial, want 0", fs) + } +} + +// --- Unit: tracker and classification ----------------------------------------- + +func TestCooldownGrowsExponentiallyWithJitterAndCaps(t *testing.T) { + t.Parallel() + var tr reconnectTracker + p := testPolicy() + p.cooldownBase, p.cooldownMax = 100*time.Millisecond, 800*time.Millisecond + tr.policy = p + if d := tr.cooldown(); d != 0 { + t.Fatalf("initial cooldown = %v, want 0", d) + } + for n := 1; n <= 8; n++ { + tr.connError(io.EOF, time.Second, true, nil) // rapid close + want := p.cooldownBase << (n - 1) + if want > p.cooldownMax { + want = p.cooldownMax + } + seen := map[time.Duration]bool{} + for i := 0; i < 50; i++ { + d := tr.cooldown() + if d < want/2 || d > want { + t.Fatalf("streak %d: cooldown %v outside [%v, %v]", n, d, want/2, want) + } + seen[d] = true + } + if len(seen) < 2 { + t.Fatalf("streak %d: cooldown not jittered (always %v)", n, want) + } + } + tr.healthy(p.rapidWindow) // an answer on an old conn ends the streak + if d := tr.cooldown(); d != 0 { + t.Fatalf("cooldown after healthy = %v, want 0", d) + } +} + +func TestStreaksResetAfterQuietPeriod(t *testing.T) { + t.Parallel() + var tr reconnectTracker + p := testPolicy() + p.quietReset = 30 * time.Millisecond + tr.policy = p + tr.connError(io.EOF, time.Second, true, nil) + tr.exhausted() + if tr.cooldown() == 0 { + t.Fatal("expected a cooldown right after the events") + } + time.Sleep(50 * time.Millisecond) + if d := tr.cooldown(); d != 0 { + t.Fatalf("cooldown after the quiet period = %v, want 0", d) + } +} + +// TestSummaryTimer: the first event of a window arms a timer that emits +// one summary summaryEvery later, even if nothing else happens; events +// inside a pending window do not emit; stop flushes and disarms. +func TestSummaryTimer(t *testing.T) { + t.Parallel() + var tr reconnectTracker + p := testPolicy() + p.summaryEvery = 60 * time.Millisecond + tr.policy = p + got := make(chan []any, 8) + emit := func(s []any) { got <- s } + asMap := func(s []any) map[string]any { + m := map[string]any{} + for i := 0; i+1 < len(s); i += 2 { + m[s[i].(string)] = s[i+1] + } + return m + } + + if s := tr.stop(); s != nil { + t.Fatalf("stop with no events = %v, want nil", s) + } + tr = reconnectTracker{policy: p} + + start := time.Now() + tr.reconnected(io.EOF, emit) + tr.reconnected(io.EOF, emit) + tr.servedByOther(emit) + var first []any + select { + case first = <-got: + case <-time.After(5 * time.Second): + t.Fatal("no summary after the interval") + } + if waited := time.Since(start); waited < p.summaryEvery { + t.Fatalf("summary after %v, want >= %v", waited, p.summaryEvery) + } + m := asMap(first) + if m["reconnects"] != 2 || m["served_by_other_conn"] != 1 || m["last_cause"] != "EOF" { + t.Fatalf("summary attrs = %v", m) + } + select { + case s := <-got: + t.Fatalf("second summary with no new events: %v", s) + case <-time.After(3 * p.summaryEvery): + } + + // A new event opens a new window; stop flushes it before the timer. + tr.dialFailed(emit) + s := tr.stop() + if m := asMap(s); m["failed_dials"] != 1 { + t.Fatalf("stop summary = %v", s) + } + tr.reconnected(io.EOF, emit) // after stop: counted, never emitted + select { + case s := <-got: + t.Fatalf("summary emitted after stop: %v", s) + case <-time.After(3 * p.summaryEvery): + } +} + +func TestConnErrorClassification(t *testing.T) { + t.Parallel() + opErr := func(errno syscall.Errno) error { + return fmt.Errorf("recv: %w", &net.OpError{Op: "read", Net: "tcp", Err: os.NewSyscallError("read", errno)}) + } + cases := []struct { + name string + err error + peer, dropped bool + }{ + {"eof", fmt.Errorf("recv: %w", io.EOF), true, true}, + {"unexpected eof", fmt.Errorf("recv: %w", io.ErrUnexpectedEOF), true, true}, + {"reset", opErr(syscall.ECONNRESET), true, true}, + {"broken pipe", opErr(syscall.EPIPE), true, true}, + {"local close", fmt.Errorf("send: %w", net.ErrClosed), false, true}, + {"not connected", errNotConnected, false, true}, + {"timeout", fmt.Errorf("recv: %w", os.ErrDeadlineExceeded), false, false}, + {"decode", errors.New("recv: json decode: bad"), false, false}, + } + for _, tc := range cases { + if got := isPeerClose(tc.err); got != tc.peer { + t.Errorf("%s: isPeerClose = %v, want %v", tc.name, got, tc.peer) + } + if got := isConnDropped(tc.err); got != tc.dropped { + t.Errorf("%s: isConnDropped = %v, want %v", tc.name, got, tc.dropped) + } + } +} + +func TestJitterBounds(t *testing.T) { + t.Parallel() + for _, d := range []time.Duration{0, 1, 2, 3, time.Millisecond, time.Second} { + for i := 0; i < 100; i++ { + j := jitter(d) + if j < d/2 || j > d { + t.Fatalf("jitter(%v) = %v outside [%v, %v]", d, j, d/2, d) + } + } + } +} diff --git a/registry/client/zz_client_pool_test.go b/registry/client/zz_client_pool_test.go index 04c314e..e1ea7ee 100644 --- a/registry/client/zz_client_pool_test.go +++ b/registry/client/zz_client_pool_test.go @@ -14,6 +14,7 @@ import ( "encoding/hex" "encoding/json" "encoding/pem" + "errors" "math/big" "net" "strings" @@ -287,41 +288,42 @@ func TestDialPoolErrorOnUnreachable(t *testing.T) { // TestDialPoolPartialSecondaryFailureClosesPrimary covers the // "primary dialed, secondary dial failed → close primary, return err" branch -// inside initPool. We accept the primary conn then close the listener to make -// secondary dials fail. +// inside initPool. The dialer fails the third dial (the second secondary), +// so the failure is deterministic: closing the listener after the first +// accept raced the secondary dials, which could still land in the backlog. func TestDialPoolPartialSecondaryFailureClosesPrimary(t *testing.T) { t.Parallel() - ln, err := net.Listen("tcp", "127.0.0.1:0") - if err != nil { - t.Fatalf("listen: %v", err) - } - addr := ln.Addr().String() + srv := newFakeJSONServer(t, echoHandler()) + defer srv.close() - accepted := make(chan net.Conn, 1) - go func() { - conn, err := ln.Accept() + boom := errors.New("secondary dial refused") + var opened []*countingConn + dial := func(ctx context.Context, network, addr string) (net.Conn, error) { + if len(opened) == 2 { + return nil, boom + } + var d net.Dialer + raw, err := d.DialContext(ctx, network, addr) if err != nil { - return + return nil, err } - accepted <- conn - // Close listener immediately so the next net.Dial inside initPool - // fails with ECONNREFUSED. - ln.Close() - }() + cc := &countingConn{Conn: raw} + opened = append(opened, cc) + return cc, nil + } - _, err = DialPool(addr, 4) + _, err := DialPool(srv.addr(), 4, WithDialer(dial)) if err == nil { t.Fatalf("expected DialPool to fail when secondary dial errors") } - if !strings.Contains(err.Error(), "dial pool conn") { - t.Fatalf("error should mention dial pool conn, got: %v", err) + if !strings.Contains(err.Error(), "dial pool conn 2") || !errors.Is(err, boom) { + t.Fatalf("error should wrap the secondary dial failure, got: %v", err) } - - // Clean up the primary conn that the server accepted. - select { - case c := <-accepted: - c.Close() - case <-time.After(time.Second): + // The primary and the secondary that did open are both closed. + for i, cc := range opened { + if n := cc.closes.Load(); n != 1 { + t.Fatalf("conn %d closed %d times, want 1", i, n) + } } } @@ -480,7 +482,7 @@ func TestReconnectEntrySyncsPrimary(t *testing.T) { oldConn := c.conn primary.mu.Lock() _ = primary.conn.Close() - if err := c.reconnectEntry(context.Background(), primary); err != nil { + if err := c.reconnectEntry(context.Background(), primary, nil); err != nil { primary.mu.Unlock() t.Fatalf("reconnectEntry: %v", err) } @@ -515,7 +517,7 @@ func TestReconnectEntryFailsWhenClosed(t *testing.T) { } c.Close() - if err := c.reconnectEntry(context.Background(), c.pool.entries[0]); err == nil { + if err := c.reconnectEntry(context.Background(), c.pool.entries[0], nil); err == nil { t.Fatalf("reconnectEntry on closed client should fail") } } diff --git a/registry/client/zz_client_proxy_test.go b/registry/client/zz_client_proxy_test.go index 9488f3e..4876446 100644 --- a/registry/client/zz_client_proxy_test.go +++ b/registry/client/zz_client_proxy_test.go @@ -480,7 +480,11 @@ func TestPoolReconnectTLSThroughProxy(t *testing.T) { } defer c.Close() // reconnectEntry redials only the broken entry, through the proxy. + // Hold the other entry so the dropped request cannot be served by it + // and must redial. + release := holdIdleEntries(t, c, 1) mustSendOK(t, c, "drop") + release() mustSendOK(t, c, "after") assertConnects(t, proxy, srv, 3) if got := srv.conns.Load(); got != 3 { @@ -579,7 +583,9 @@ func TestWithDialerReceivesUnresolvedAddrAndBoundedContext(t *testing.T) { t.Fatalf("DialPool: %v", err) } defer c.Close() + release := holdIdleEntries(t, c, 1) // force the redial path mustSendOK(t, c, "drop") + release() if got := calls.Load(); got != 3 { t.Fatalf("dialer called %d times, want 3 (2 pool conns + 1 reconnect)", got) }