From eea0e6a2ac62960d98b06bc05ff97586fa75a88a Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Thu, 24 Sep 2026 02:16:34 +0300 Subject: [PATCH 1/3] fix(registry/client): idempotent Close; pool reconnect keeps its cause, backs off, retries on a healthy conn Close (sweep item 8) - Client.Close was not idempotent: it always ran close(c.pool.done), so a second Close panicked with "close of closed channel". Every nightly since 07-26 hits this when the daemon's forceReconnectRegistry `go old.Close()` races Stop's Close. - Close now runs its teardown once (sync.Once). Later and concurrent calls return nil. The teardown closes every pool connection exactly once and interrupts in-flight requests instead of waiting on them. - A redial that finishes after Close closes its own connection, so it does not leak into the closed client. - Close wakes requests parked on the free list or in a reconnect backoff. The single-conn path runs its backoff with c.mu held, so without this wake-up Close would wait out the backoff. - Every request after Close, and every request that Close interrupted, fails with the new ErrClosed. errors.Is(ErrClosed, ErrNoRegistry) is true, so callers that already treat a missing registry as recoverable need no change. - BinaryClient.Close gets the same treatment. Pool reconnect (client side of sweep item 5) - The error that triggered a reconnect is no longer thrown away. It is logged with the drop and the redial (cause=...). When the redial fails, the returned error wraps both errors: "send failed and reconnect failed: (reconnect: )". When the retry fails too, its error names the first failure. - Per-reconnect success lines ("registry pool conn reconnected", "registry reconnected") move to Debug. They were 113,804 lines, 57% of daemon.log, in 5.5 days. In their place is one INFO line, "registry reconnect summary", at most once a minute and flushed on Close. It carries the counts, rate_limit_suspected and last_cause. - Jittered exponential backoff: - A client-wide cooldown before a redial grows with consecutive rapid closes and with reconnects that fail every attempt. A rapid close is the peer closing or resetting a connection younger than 30s with no error frame, which is the registry rate limiter's signature. - The cooldown starts at 250ms and is capped at 2s, well under the daemon's 8s registry deadline. - It resets after an answer on a connection at least 30s old, after a successful redial (failure streak only), or after a quiet minute. - The attempts inside one reconnect are now jittered and no longer sleep after the final attempt. - After a closed-connection error on one pool entry, the request is retried once on another idle healthy entry before any redial. The dropped entry is marked broken and its fd closed at once. Healthy entries are preferred, and a broken entry is redialed lazily when it is next needed. Timeouts still go straight to the redial. - SendContext now also honours ctx while it waits for a free pool entry. Public API: additive only (ErrClosed). Error texts keep their old substrings ("client closed", "send failed and reconnect failed", "reconnect failed after 5 attempts"). Tests (zz_client_close_reconnect_test.go) use a scripted fake registry that closes connections after N requests or after a grace period. - Double and concurrent Close under -race, with concurrent Sends. - Each pool connection is closed exactly once. - Close during a redial, and Close waking a reconnect backoff. - Debug logging with the cause, plus the periodic INFO summary. - Retry on another entry, and the fallback to a redial. - Backoff growth, jitter bounds, cap and reset, with timing checked against lower bounds only. - Both errors kept on failure, and error classification. Two proxy tests now hold the other pool entry so their dropped request still has to take the redial path they check. Co-Authored-By: Claude Opus 5.5 (1M context) --- registry/client/binary_client.go | 152 +- registry/client/client.go | 591 ++++++-- registry/client/reconnect.go | 323 ++++ registry/client/zz_client_branch_test.go | 4 +- .../client/zz_client_close_reconnect_test.go | 1350 +++++++++++++++++ registry/client/zz_client_pool_test.go | 4 +- registry/client/zz_client_proxy_test.go | 6 + 7 files changed, 2236 insertions(+), 194 deletions(-) create mode 100644 registry/client/reconnect.go create mode 100644 registry/client/zz_client_close_reconnect_test.go 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..debeba7 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,197 @@ 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...)...) +} - for attempts := 0; attempts < 5; attempts++ { +// noteConnError records a connection-level request failure (the triggering +// error of the reconnect that follows) and logs it at Debug. +func (c *Client) noteConnError(err error, age time.Duration) { + rapid := c.reconn.connError(err, age, c.logSummary) + c.log().Debug("registry conn dropped", "addr", c.addr, "err", err, + "conn_age", age.Round(time.Millisecond).String(), "rapid_close", rapid) +} + +// 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) { + 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): + 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 } - backoff *= 2 - if backoff > maxBackoff { - backoff = maxBackoff + if backoff *= 2; backoff > p.dialMax { + backoff = p.dialMax } } - return fmt.Errorf("reconnect failed after 5 attempts: %w", err) + 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) + } + 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 @@ -382,6 +530,11 @@ func (c *Client) Send(msg map[string]interface{}) (map[string]interface{}, error // 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. +// +// A connection-level failure (no response) is retried once: on a pooled +// client first on another idle, healthy connection, then after redialing. +// 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 +554,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)) + 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 +588,208 @@ 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() + + // 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) + } + } + + resp, err := c.sendOnEntry(entry, msg) + 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. + // 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(msg); served { + return resp, err + } + if c.isClosed() { + return nil, closedDuring(cause) + } } - var entry *pooledConn + // Redial this entry and retry once. + if rerr := c.reconnectEntry(ctx, entry, cause); rerr != nil { + return nil, reconnectFailed(cause, rerr) + } + resp, err = c.sendOnEntry(entry, msg) + 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) +// sendOnOtherEntry retries msg once on another idle, healthy entry. +// served is false when no such entry is free or the retry also failed at +// the connection level (that entry is then marked broken too). +func (c *Client) sendOnOtherEntry(msg map[string]interface{}) (resp map[string]interface{}, served bool, err error) { + other := c.tryAcquireHealthy() + if other == nil { + return nil, false, nil } - return resp, err + defer c.releaseEntry(other) + other.mu.Lock() + defer other.mu.Unlock() + resp, err = c.sendOnEntry(other, msg) + 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. +// 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{}) (map[string]interface{}, error) { - if err := wire.WriteMessage(entry.conn, msg); err != nil { - return nil, fmt.Errorf("send: %w", err) + resp, err := roundTrip(entry.conn, msg) + if err != nil && resp == nil { + entry.broken = err + if !c.isClosed() { + c.noteConnError(err, time.Since(entry.dialedAt)) + } + 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 } - return fmt.Errorf("reconnect failed after 5 attempts: %w", 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() + } + 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 +801,22 @@ 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{}) 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..dc1fb03 --- /dev/null +++ b/registry/client/reconnect.go @@ -0,0 +1,323 @@ +// 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 +} + +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, +} + +// 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 request failure on a connection of +// the given age and reports whether it is a rapid close. +func (t *reconnectTracker) connError(err error, age time.Duration, 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.quietResetLocked(now, p) + t.rapidStreak++ + t.stats.rapidCloses++ + t.lastEvent = now + t.countedLocked(now, emit) + } + 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..6c5dbce --- /dev/null +++ b/registry/client/zz_client_close_reconnect_test.go @@ -0,0 +1,1350 @@ +// 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 + served atomic.Int32 // requests answered + dropped atomic.Int32 // requests answered by closing the conn + + mu sync.Mutex + drop func(connSeq, reqSeq int, age time.Duration) 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() +} + +// 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.mu.Lock() + drop := s.drop + s.mu.Unlock() + 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, + } +} + +// 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") + } +} + +// --- 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, 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, nil) { + t.Fatal("a timeout classified as a rapid close") + } + if tr.connError(fmt.Errorf("send: %w", net.ErrClosed), time.Second, 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, 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, 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..f1f6d7f 100644 --- a/registry/client/zz_client_pool_test.go +++ b/registry/client/zz_client_pool_test.go @@ -480,7 +480,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 +515,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) } From df4fc719b309cdbfc0d64478fa679934357f172c Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Thu, 24 Sep 2026 02:16:46 +0300 Subject: [PATCH 2/3] test(registry/client): make the DialPool partial-failure test deterministic TestDialPoolPartialSecondaryFailureClosesPrimary closed the listener after the first accept to make a secondary dial fail. A secondary dial can land in the listen backlog before that close, and then DialPool succeeds. On origin/main this failed 13 times in 600 runs (-race -cpu 1,4). The test now uses WithDialer to fail the third dial. It also checks that the primary and the secondary that did open are each closed exactly once. 0 failures in 600 runs. Co-Authored-By: Claude Opus 5.5 (1M context) --- registry/client/zz_client_pool_test.go | 52 +++++++++++++------------- 1 file changed, 27 insertions(+), 25 deletions(-) diff --git a/registry/client/zz_client_pool_test.go b/registry/client/zz_client_pool_test.go index f1f6d7f..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) + } } } From 3eb5b6b7e6a68091200f684f6474bf108a62a535 Mon Sep 17 00:00:00 2001 From: Teodor Calin Date: Thu, 24 Sep 2026 02:51:16 +0300 Subject: [PATCH 3/3] fix(registry/client): bound the other-conn retry; one backoff step per request Review fixes for the pool retry on another connection: - F1: the retry on another idle pooled conn after a drop ran with the 30s read deadline and ignored ctx, so a silently hung other conn stalled a request that had failed fast (30.2s vs 1ms on main, past the daemon's 8s registry deadline). The retry's whole exchange is now bounded by otherConnTimeout (2s) and by ctx, both its deadline and its cancellation. A conn that times out or is abandoned is retired, since a late reply could still arrive on it, and the request falls back to the redial. dialWithBackoff returns at once when ctx is already done, and a dial abandoned for ctx is neither logged as a failed dial nor counted toward the failure streak. The reviewer's repro now returns in 2.2s. - F2: a pooled request can reach the registry three times (own conn, other conn, redialed conn), but SendContext said "retried once". The doc now states the limits (twice single-conn, three times pooled) and that an operation not safe to repeat may have been applied. Each failed transmission also extended the rapid-close streak, so one shed request doubled the next cooldown. A logical request now extends the streak at most once; every close still shows in the summary. Tests: bounded retry on a hung other conn, ctx deadline and cancel ending it with no redial, at most three sends and one streak step per pooled request, at most two sends single-conn, a cancellation racing a bounded exchange never leaves a deadline behind, and the default timeout budget. Co-Authored-By: Claude Opus 5.5 (1M context) --- registry/client/client.go | 151 ++++++-- registry/client/reconnect.go | 30 +- .../client/zz_client_close_reconnect_test.go | 324 +++++++++++++++++- 3 files changed, 466 insertions(+), 39 deletions(-) diff --git a/registry/client/client.go b/registry/client/client.go index debeba7..31219a2 100644 --- a/registry/client/client.go +++ b/registry/client/client.go @@ -412,12 +412,16 @@ func (c *Client) logSummary(attrs []any) { c.log().Info("registry reconnect summary", append([]any{"addr", c.addr}, attrs...)...) } -// noteConnError records a connection-level request failure (the triggering -// error of the reconnect that follows) and logs it at Debug. -func (c *Client) noteConnError(err error, age time.Duration) { - rapid := c.reconn.connError(err, age, c.logSummary) +// 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 } // dialWithBackoff opens a replacement connection after cause broke the old @@ -429,6 +433,12 @@ func (c *Client) noteConnError(err error, age time.Duration) { // // 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 } @@ -444,6 +454,9 @@ func (c *Client) dialWithBackoff(ctx context.Context, what string, cause error) if err == nil { return conn, nil } + if ctx.Err() != nil { + return nil, ctx.Err() + } c.log().Warn(what+" reconnect failed", "addr", c.addr, "attempt", attempt, "err", err, "cause", errText(cause)) c.reconn.dialFailed(c.logSummary) if attempt == maxReconnectAttempts { @@ -529,12 +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 once: on a pooled -// client first on another idle, healthy connection, then after redialing. -// An error response from the registry is returned as is, never retried. -// After Close every call returns an error matching ErrClosed. +// 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 @@ -566,7 +595,7 @@ func (c *Client) SendContext(ctx context.Context, msg map[string]interface{}) (m return nil, closedDuring(err) } cause := err - c.noteConnError(cause, time.Since(c.connDialedAt)) + c.noteConnError(cause, time.Since(c.connDialedAt), false) if reconnErr := c.reconnect(ctx, cause); reconnErr != nil { return nil, reconnectFailed(cause, reconnErr) } @@ -607,7 +636,8 @@ func (c *Client) sendPool(ctx context.Context, msg map[string]interface{}) (map[ } } - resp, err := c.sendOnEntry(entry, msg) + var st sendState + resp, err := c.sendOnEntry(entry, msg, &st) if err == nil || resp != nil { return resp, err } @@ -619,10 +649,12 @@ func (c *Client) sendPool(ctx context.Context, msg map[string]interface{}) (map[ // 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. - // Timeouts are not retried this way: a half-open conn already cost - // the caller the full read deadline. + // 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(msg); served { + if resp, served, err := c.sendOnOtherEntry(ctx, msg, &st); served { return resp, err } if c.isClosed() { @@ -630,11 +662,12 @@ func (c *Client) sendPool(ctx context.Context, msg map[string]interface{}) (map[ } } - // Redial this entry and retry once. + // 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) + resp, err = c.sendOnEntry(entry, msg, &st) if err != nil && resp == nil { if c.isClosed() { return nil, closedDuring(err) @@ -697,10 +730,23 @@ func (c *Client) releaseEntry(e *pooledConn) { } } -// sendOnOtherEntry retries msg once on another idle, healthy entry. -// served is false when no such entry is free or the retry also failed at -// the connection level (that entry is then marked broken too). -func (c *Client) sendOnOtherEntry(msg map[string]interface{}) (resp map[string]interface{}, served bool, err error) { +// 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 + } other := c.tryAcquireHealthy() if other == nil { return nil, false, nil @@ -708,7 +754,8 @@ func (c *Client) sendOnOtherEntry(msg map[string]interface{}) (resp map[string]i defer c.releaseEntry(other) other.mu.Lock() defer other.mu.Unlock() - resp, err = c.sendOnEntry(other, msg) + 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 } @@ -721,12 +768,20 @@ func (c *Client) sendOnOtherEntry(msg map[string]interface{}) (resp map[string]i // 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{}) (map[string]interface{}, error) { +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) 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() { - c.noteConnError(err, time.Since(entry.dialedAt)) + st.streakNoted = c.noteConnError(err, time.Since(entry.dialedAt), st.streakNoted) } c.retireEntryConn(entry) return nil, err @@ -817,6 +872,56 @@ func roundTrip(conn net.Conn, msg map[string]interface{}) (map[string]interface{ 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 index dc1fb03..558d168 100644 --- a/registry/client/reconnect.go +++ b/registry/client/reconnect.go @@ -85,6 +85,13 @@ type reconnectPolicy struct { 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{ @@ -97,6 +104,10 @@ var defaultReconnectPolicy = reconnectPolicy{ 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 @@ -175,9 +186,14 @@ func (t *reconnectTracker) countedLocked(now time.Time, emit func([]any)) { }) } -// connError records a connection-level request failure on a connection of -// the given age and reports whether it is a rapid close. -func (t *reconnectTracker) connError(err error, age time.Duration, emit func([]any)) (rapid bool) { +// 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() @@ -185,11 +201,13 @@ func (t *reconnectTracker) connError(err error, age time.Duration, emit func([]a defer t.mu.Unlock() t.stats.lastCause = err.Error() if rapid { - t.quietResetLocked(now, p) - t.rapidStreak++ t.stats.rapidCloses++ - t.lastEvent = now t.countedLocked(now, emit) + if extendStreak { + t.quietResetLocked(now, p) + t.rapidStreak++ + t.lastEvent = now + } } return rapid } diff --git a/registry/client/zz_client_close_reconnect_test.go b/registry/client/zz_client_close_reconnect_test.go index 6c5dbce..4665556 100644 --- a/registry/client/zz_client_close_reconnect_test.go +++ b/registry/client/zz_client_close_reconnect_test.go @@ -38,13 +38,18 @@ import ( // 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 - served atomic.Int32 // requests answered - dropped atomic.Int32 // requests answered by closing the conn + 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 } @@ -97,6 +102,12 @@ func (s *scriptedRegistry) setDrop(fn func(int, int, time.Duration) bool) { 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() @@ -151,9 +162,15 @@ func (s *scriptedRegistry) serve(conn net.Conn, connSeq int) { if err := json.Unmarshal(body, &req); err != nil { return } + s.received.Add(1) s.mu.Lock() - drop := s.drop + 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 @@ -331,6 +348,8 @@ func testPolicy() *reconnectPolicy { rapidWindow: 30 * time.Second, quietReset: time.Hour, summaryEvery: time.Hour, + // Generous: tests that need a hung other conn shrink it. + otherConnTimeout: 5 * time.Second, } } @@ -1018,6 +1037,291 @@ func TestPoolDoesNotRetryTimeoutOnOtherEntry(t *testing.T) { } } +// 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 @@ -1126,16 +1430,16 @@ func TestIdleCloseDoesNotBackOff(t *testing.T) { var tr reconnectTracker p := testPolicy() tr.policy = p - if tr.connError(fmt.Errorf("recv: %w", io.EOF), time.Minute, nil) { + 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, nil) { + 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, nil) { + if tr.connError(fmt.Errorf("send: %w", net.ErrClosed), time.Second, true, nil) { t.Fatal("a locally closed conn classified as a rapid close") } } @@ -1207,7 +1511,7 @@ func TestCooldownGrowsExponentiallyWithJitterAndCaps(t *testing.T) { t.Fatalf("initial cooldown = %v, want 0", d) } for n := 1; n <= 8; n++ { - tr.connError(io.EOF, time.Second, nil) // rapid close + tr.connError(io.EOF, time.Second, true, nil) // rapid close want := p.cooldownBase << (n - 1) if want > p.cooldownMax { want = p.cooldownMax @@ -1236,7 +1540,7 @@ func TestStreaksResetAfterQuietPeriod(t *testing.T) { p := testPolicy() p.quietReset = 30 * time.Millisecond tr.policy = p - tr.connError(io.EOF, time.Second, nil) + tr.connError(io.EOF, time.Second, true, nil) tr.exhausted() if tr.cooldown() == 0 { t.Fatal("expected a cooldown right after the events")