fix(connpool): back off between failed dials in OneTimePool.Run

conn/connpool/onetimepool.go

@@ -2,8 +2,17 @@ package connpool
 
 import (
 	"context"
+	"fmt"
 	"net"
 	"sync"
+	"time"
+)
+
+// Bounds of the wait between failed dials in OneTimePool.Run.
+const (
+	minDialBackoff = 50 * time.Millisecond
+	maxDialBackoff = time.Second
+	backoffFactor  = 2
 )
 
 type Dialer interface {
@@ -71,35 +80,49 @@ func (p *OneTimePool) Acquire(c context.Context) (net.Conn, error) {
 	}
 }
 
+// Run dials connections and keeps the pool filled until the pool is closed or
+// its context is done, then returns an error wrapping the context's error
+// (match it with errors.Is). Failed dials are retried with an exponentially
+// growing wait, capped at maxDialBackoff and reset by the next successful
+// dial, so an unreachable target does not consume a CPU core. Dial errors are
+// not reported; callers observe them as Acquire blocking until its own
+// context is done.
 func (p *OneTimePool) Run() error {
+	backoff := minDialBackoff
 	for {
+		if p.ctx.Err() != nil {
+			return p.stopped()
+		}
+
+		conn := p.dial()
+		if conn == nil {
+			if !p.sleep(backoff) {
+				return p.stopped()
+			}
+
+			backoff = min(backoff*backoffFactor, maxDialBackoff)
+
+			continue
+		}
+
+		backoff = minDialBackoff
+
+		p.InFlightConns.Add(conn)
+		// Close cancels the context before it closes the in-flight
+		// connections. A connection added after that sweep would never be
+		// closed, so a done context here means this goroutine must close it.
+		if p.ctx.Err() != nil {
+			p.discard(conn)
+
+			return p.stopped()
+		}
+
 		select {
 		case <-p.ctx.Done():
-			return p.ctx.Err()
-		default:
-			var (
-				conn net.Conn
-				err  error
-			)
-			p.mx.RLock()
-			hasDialer := p.Dialer != nil
-			p.mx.RUnlock()
-
-			if hasDialer {
-				p.mx.RLock()
-				conn, err = p.Dialer.Dial(p.ctx, "tcp", p.address)
-				p.mx.RUnlock()
-			} else {
-				conn, err = net.Dial("tcp", p.address)
-			}
-			if err == nil {
-				p.InFlightConns.Add(conn)
-				select {
-				case <-p.ctx.Done():
-					return p.ctx.Err()
-				case p.idleConnections <- conn:
-				}
-			}
+			p.discard(conn)
+
+			return p.stopped()
+		case p.idleConnections <- conn:
 		}
 	}
 }
@@ -114,3 +137,55 @@ func (p *OneTimePool) Close() error {
 
 	return p.InFlightConns.Close()
 }
+
+// stopped is the error Run returns once the pool context is done.
+func (p *OneTimePool) stopped() error {
+	return fmt.Errorf("connpool: run stopped: %w", p.ctx.Err())
+}
+
+// dial opens one connection and returns nil if the dial failed; Run retries
+// and does not report the cause, so the error is not built into a value. The
+// dialer is read under the lock but invoked outside it, so Close never waits
+// for a dial in progress.
+func (p *OneTimePool) dial() net.Conn {
+	p.mx.RLock()
+	dialer := p.Dialer
+	p.mx.RUnlock()
+
+	var (
+		conn net.Conn
+		err  error
+	)
+
+	if dialer != nil {
+		conn, err = dialer.Dial(p.ctx, "tcp", p.address)
+	} else {
+		var netDialer net.Dialer
+
+		conn, err = netDialer.DialContext(p.ctx, "tcp", p.address)
+	}
+
+	if err != nil {
+		return nil
+	}
+
+	return conn
+}
+
+// sleep waits for d and reports whether the pool context is still live.
+func (p *OneTimePool) sleep(d time.Duration) bool {
+	timer := time.NewTimer(d)
+	defer timer.Stop()
+
+	select {
+	case <-p.ctx.Done():
+		return false
+	case <-timer.C:
+		return true
+	}
+}
+
+func (p *OneTimePool) discard(conn net.Conn) {
+	p.InFlightConns.Remove(conn)
+	_ = conn.Close()
+}