fix(otel): count connections that predate Init (#4089)

* fix(otel): count connections that predate Init

Connections created before the metric callbacks were set never emitted
+1, but still emitted -1 when they left idle or were closed, so
db.client.connection.count went negative when a client was created
before redisotel-native's Init.

Existing connections are now counted when the callbacks are set and
uncounted when they are cleared.

Closes #4083

* fix(otel): hold pools weakly, uncount on shutdown

Keep weak pointers to live pools so a client that is dropped without
Close can still be collected.

SetGlobalRecorder(nil) now clears the pool callbacks before swapping in
the noop recorder, so the -1 for existing connections reaches the
recorder being removed.

* fix(otel): read count callback with the map change

Removal paths, Close and PubSub tracking now read the connection count
callback under the same lock as the map change, so a concurrent
SetAllMetricCallbacks can't land between them and leave the counter off
by one.

* fix(otel): send count deltas to the recorder that set them

* fix(otel): drop collected pools from the registry

* fix(otel): uncount the refreshed spare under connsMu

* fix(otel): close pubsub conns after releasing trackMu

* fix(otel): serialize SetGlobalRecorder

* test(otel): check per-state counts under churn

* docs(otel): note lock order, relax Init timing

* fix(otel): drop unpooled spare from idleConns

* fix(otel): drop unpooled idle conns in removeConn

* docs(otel): count callback must not re-enter

extra/redisotel-native/redisotel.go

@@ -73,7 +73,8 @@ func GetObservabilityInstance() *ObservabilityInstance {
 }
 
 // Init initializes OpenTelemetry observability globally for all Redis clients.
-// This should be called once at application startup, BEFORE creating any Redis clients.
+// It is usually called once at application startup. Clients created before Init
+// are covered too: their existing connections are counted when Init runs.
 // After initialization, all Redis clients will automatically collect and export
 // metrics without needing any additional configuration.
 func (o *ObservabilityInstance) Init(cfg *Config) error {

internal/otel/connection_count_test.go

@@ -0,0 +1,69 @@
+package otel
+
+import (
+	"context"
+	"net"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9/internal/pool"
+)
+
+type countRecorder struct {
+	noopRecorder
+	mu    sync.Mutex
+	count int
+}
+
+func (r *countRecorder) RecordConnectionCount(_ context.Context, delta int, cn *pool.Conn, _ string, _ bool) {
+	if cn.PoolName() != "shutdown-count" {
+		return
+	}
+	r.mu.Lock()
+	r.count += delta
+	r.mu.Unlock()
+}
+
+func (r *countRecorder) get() int {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.count
+}
+
+func TestShutdownUncountsExistingConns(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Name: "shutdown-count",
+		Dialer: func(context.Context) (net.Conn, error) {
+			c, _ := net.Pipe()
+			return c, nil
+		},
+		PoolSize:           2,
+		MinIdleConns:       2,
+		MaxConcurrentDials: 2,
+		PoolTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+
+	deadline := time.Now().Add(time.Second)
+	for connPool.Len() != 2 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+
+	if got := connPool.Len(); got != 2 {
+		t.Fatalf("MinIdleConns: got %d, want 2", got)
+	}
+
+	r := &countRecorder{}
+	SetGlobalRecorder(r)
+	if got := r.get(); got != 2 {
+		SetGlobalRecorder(nil)
+		t.Fatalf("after SetGlobalRecorder: got %d, want 2", got)
+	}
+
+	SetGlobalRecorder(nil)
+	if got := r.get(); got != 0 {
+		t.Fatalf("after SetGlobalRecorder(nil): got %d, want 0", got)
+	}
+}

internal/otel/metrics.go

@@ -120,6 +120,9 @@ type PoolRegistrar interface {
 }
 
 var (
+	// setRecorderMu serializes SetGlobalRecorder, so the recorder swap and the pool callbacks change together.
+	setRecorderMu sync.Mutex
+
 	// recorderMu protects globalRecorder and operation duration callbacks
 	recorderMu sync.RWMutex
 
@@ -166,16 +169,19 @@ func Enabled() bool {
 
 // SetGlobalRecorder sets the global recorder (called by Init() in extra/redisotel-native)
 func SetGlobalRecorder(r Recorder) {
-	recorderMu.Lock()
+	setRecorderMu.Lock()
+	defer setRecorderMu.Unlock()
 	if r == nil {
+		// Unregister pool callbacks first so their final deltas reach the current recorder.
+		pool.SetAllMetricCallbacks(nil)
+		recorderMu.Lock()
 		globalRecorder = noopRecorder{}
 		operationDurationCallback = nil
 		pipelineOperationDurationCallback = nil
 		recorderMu.Unlock()
-		// Unregister all pool metric callbacks atomically
-		pool.SetAllMetricCallbacks(nil)
 		return
 	}
+	recorderMu.Lock()
 	globalRecorder = r
 
 	// Register operation duration callbacks
@@ -213,8 +219,9 @@ func SetGlobalRecorder(r Recorder) {
 		ConnectionClosed: func(ctx context.Context, cn *pool.Conn, reason string, err error) {
 			getRecorder().RecordConnectionClosed(ctx, cn, reason, err)
 		},
+		// Uses r, not getRecorder(), so a delta captured before Shutdown reaches the recorder that counted it.
 		ConnectionCount: func(ctx context.Context, delta int, cn *pool.Conn, state string, isPubSub bool) {
-			getRecorder().RecordConnectionCount(ctx, delta, cn, state, isPubSub)
+			r.RecordConnectionCount(ctx, delta, cn, state, isPubSub)
 		},
 		PendingRequests: func(ctx context.Context, delta int, cn *pool.Conn, poolName string) {
 			getRecorder().RecordPendingRequests(ctx, delta, cn, poolName)

internal/pool/live_pools_test.go

@@ -0,0 +1,33 @@
+package pool
+
+import (
+	"context"
+	"net"
+	"runtime"
+	"testing"
+	"time"
+	"weak"
+)
+
+func TestLivePoolsDropCollectedPools(t *testing.T) {
+	dialer := func(context.Context) (net.Conn, error) {
+		c, _ := net.Pipe()
+		return c, nil
+	}
+	connPool := weak.Make(NewConnPool(&Options{Dialer: dialer, PoolSize: 1, PoolTimeout: time.Second}))
+	pubSubPool := weak.Make(NewPubSubPool(&Options{}, nil))
+
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		runtime.GC()
+		livePoolsMu.Lock()
+		_, connLive := liveConnPools[connPool]
+		_, pubSubLive := livePubSubPools[pubSubPool]
+		livePoolsMu.Unlock()
+		if !connLive && !pubSubLive {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatal("pools collected without Close are still registered")
+}

internal/pool/pool.go

@@ -5,9 +5,11 @@ import (
 	"errors"
 	"math/rand"
 	"net"
+	"runtime"
 	"sync"
 	"sync/atomic"
 	"time"
+	"weak"
 
 	"github.com/redis/go-redis/v9/internal"
 	"github.com/redis/go-redis/v9/internal/proto"
@@ -88,6 +90,13 @@ var (
 	// metricCallbackMu protects all global metric callback functions for thread-safe access.
 	metricCallbackMu sync.RWMutex
 
+	// livePoolsMu protects the pools whose connections are counted when metric callbacks change.
+	// Lock order: livePoolsMu, then ConnPool.connsMu, then PubSubPool.trackMu, then metricCallbackMu.
+	// SetAllMetricCallbacks holds every pool's connsMu at once; never hold two pools' connsMu elsewhere.
+	livePoolsMu     sync.Mutex
+	liveConnPools   = make(map[weak.Pointer[ConnPool]]struct{})
+	livePubSubPools = make(map[weak.Pointer[PubSubPool]]struct{})
+
 	// Global metric callbacks for connection state changes
 	metricConnectionStateChangeCallback func(ctx context.Context, cn *Conn, fromState, toState string)
 
@@ -152,6 +161,8 @@ var (
 	noExpiration = maxTime
 )
 
+type connectionCountCallback = func(ctx context.Context, delta int, cn *Conn, state string, isPubSub bool)
+
 // MetricCallbacks holds all metric callback functions.
 // Use SetAllMetricCallbacks to register all callbacks atomically.
 type MetricCallbacks struct {
@@ -181,6 +192,7 @@ type MetricCallbacks struct {
 	// delta: +1 when connection added, -1 when connection removed
 	// state: connection state (e.g., "idle", "used")
 	// isPubSub: true if this is a PubSub connection
+	// It may run with connsMu or trackMu held and must not re-enter the pool.
 	ConnectionCount func(ctx context.Context, delta int, cn *Conn, state string, isPubSub bool)
 
 	// PendingRequests is called when pending requests count changes (UpDownCounter)
@@ -199,7 +211,54 @@ type MetricCallbacks struct {
 // acceptable for metrics - at most one event may go to the old recorder
 // or be missed during the transition. The callbacks themselves are immutable
 // function pointers, so calling an "old" callback is safe.
+//
+// Connections that already exist are counted when callbacks are set and
+// uncounted when they are cleared, so db.client.connection.count stays balanced.
 func SetAllMetricCallbacks(callbacks *MetricCallbacks) {
+	livePoolsMu.Lock()
+	defer livePoolsMu.Unlock()
+	connPools := make([]*ConnPool, 0, len(liveConnPools))
+	for wp := range liveConnPools {
+		if p := wp.Value(); p != nil {
+			connPools = append(connPools, p)
+		} else {
+			delete(liveConnPools, wp)
+		}
+	}
+	pubSubPools := make([]*PubSubPool, 0, len(livePubSubPools))
+	for wp := range livePubSubPools {
+		if p := wp.Value(); p != nil {
+			pubSubPools = append(pubSubPools, p)
+		} else {
+			delete(livePubSubPools, wp)
+		}
+	}
+	for _, p := range connPools {
+		p.connsMu.Lock()
+	}
+	for _, p := range pubSubPools {
+		p.trackMu.Lock()
+	}
+	defer func() {
+		for _, p := range connPools {
+			p.connsMu.Unlock()
+		}
+		for _, p := range pubSubPools {
+			p.trackMu.Unlock()
+		}
+	}()
+
+	ctx := context.Background()
+	if cb := getMetricConnectionCountCallback(); cb != nil {
+		recordLiveConnCounts(ctx, cb, -1, connPools, pubSubPools)
+	}
+	setMetricCallbacks(callbacks)
+	if cb := getMetricConnectionCountCallback(); cb != nil {
+		recordLiveConnCounts(ctx, cb, 1, connPools, pubSubPools)
+	}
+}
+
+func setMetricCallbacks(callbacks *MetricCallbacks) {
 	metricCallbackMu.Lock()
 	defer metricCallbackMu.Unlock()
 
@@ -308,6 +367,30 @@ func getMetricConnectionCountCallback() func(ctx context.Context, delta int, cn
 	return cb
 }
 
+// recordLiveConnCounts records delta for every connection in the given pools.
+// The caller must hold the connsMu of every ConnPool and the trackMu of every PubSubPool.
+func recordLiveConnCounts(ctx context.Context, cb connectionCountCallback, delta int, connPools []*ConnPool, pubSubPools []*PubSubPool) {
+	for _, p := range connPools {
+		idle := make(map[uint64]struct{}, len(p.idleConns))
+		for _, cn := range p.idleConns {
+			idle[cn.GetID()] = struct{}{}
+		}
+		for id, cn := range p.conns {
+			state := MetricStateUsed
+			if _, ok := idle[id]; ok {
+				state = MetricStateIdle
+			}
+			cb(ctx, delta, cn, state, false)
+		}
+	}
+	for _, p := range pubSubPools {
+		p.activeConns.Range(func(_, value any) bool {
+			cb(ctx, delta, value.(*Conn), MetricStateUsed, true)
+			return true
+		})
+	}
+}
+
 // getMetricPendingRequestsCallback returns the metric callback for pending requests changes (UpDownCounter).
 func getMetricPendingRequestsCallback() func(ctx context.Context, delta int, cn *Conn, poolName string) {
 	metricCallbackMu.RLock()
@@ -471,6 +554,17 @@ func NewConnPool(opt *Options) *ConnPool {
 		idleConns:       make([]*Conn, 0, opt.PoolSize),
 	}
 
+	wp := weak.Make(p)
+	livePoolsMu.Lock()
+	liveConnPools[wp] = struct{}{}
+	livePoolsMu.Unlock()
+	// Drops the entry for a pool that is collected without Close.
+	runtime.AddCleanup(p, func(wp weak.Pointer[ConnPool]) {
+		livePoolsMu.Lock()
+		delete(liveConnPools, wp)
+		livePoolsMu.Unlock()
+	}, wp)
+
 	// Only create MinIdleConns if explicitly requested (> 0)
 	// This avoids creating connections during pool initialization for tests
 	if opt.MinIdleConns > 0 {
@@ -583,12 +677,17 @@ func (p *ConnPool) refreshStaleSpare(ctx context.Context, nowNs int64) {
 		p.connsMu.Unlock()
 		return
 	}
-	copy(p.idleConns, p.idleConns[1:])
-	p.idleConns = p.idleConns[:len(p.idleConns)-1]
-	p.idleConnsLen.Add(-1)
+	removed := p.removeConn(oldest)
+	countCb := getMetricConnectionCountCallback()
 	p.connsMu.Unlock()
 
-	_ = p.CloseConn(ctx, oldest, CloseReasonStale, MetricStateIdle)
+	if hookManager := p.hookManager.Load(); hookManager != nil {
+		hookManager.ProcessOnRemove(ctx, oldest, errors.New(CloseReasonStale))
+	}
+	if removed {
+		p.recordConnectionMetrics(ctx, oldest, CloseReasonStale, MetricStateIdle, countCb)
+	}
+	_ = p.closeConn(oldest)
 	p.checkMinIdleConns()
 }
 
@@ -1522,6 +1621,7 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 
 	var shouldCloseConn bool
 	var removedFromPool bool
+	var countCb connectionCountCallback
 
 	if p.cfg.MaxIdleConns == 0 || p.idleConnsLen.Load() < p.cfg.MaxIdleConns {
 		// Hot path optimization: try fast IN_USE → IDLE transition
@@ -1549,7 +1649,7 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 					hookManager.ProcessOnRemove(ctx, cn, errHookRequestedRemoval)
 				}
 				shouldCloseConn = true
-				removedFromPool = p.removeConnWithLock(cn)
+				removedFromPool, countCb = p.removeConnWithLock(cn)
 			default:
 				// Pool as-is
 				internal.Logger.Printf(ctx, "Unexpected conn[%d] state changed by hook to %v, pooling as-is", cn.GetID(), currentState)
@@ -1609,8 +1709,8 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 				if cb := getMetricConnectionStateChangeCallback(); cb != nil {
 					cb(ctx, cn, MetricStateUsed, "")
 				}
-				if cb := getMetricConnectionCountCallback(); cb != nil {
-					cb(ctx, -1, cn, "used", false)
+				if countCb != nil {
+					countCb(ctx, -1, cn, "used", false)
 				}
 			}
 		}
@@ -1619,7 +1719,7 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 		if hookManager != nil {
 			hookManager.ProcessOnRemove(ctx, cn, errConnEvictedIdle)
 		}
-		removedFromPool = p.removeConnWithLock(cn)
+		removedFromPool, countCb = p.removeConnWithLock(cn)
 
 		// Only emit if we actually removed it from the map (not already taken by Close()).
 		if removedFromPool {
@@ -1628,8 +1728,8 @@ func (p *ConnPool) putConn(ctx context.Context, cn *Conn, freeTurn bool) {
 				cb(ctx, cn, MetricStateUsed, "")
 			}
 			// Record connection count decrement (connection removed while in used state)
-			if cb := getMetricConnectionCountCallback(); cb != nil {
-				cb(ctx, -1, cn, "used", false)
+			if countCb != nil {
+				countCb(ctx, -1, cn, "used", false)
 			}
 		}
 	}
@@ -1677,7 +1777,7 @@ func (p *ConnPool) removeConnInternal(ctx context.Context, cn *Conn, reason erro
 		hookManager.ProcessOnRemove(ctx, cn, reason)
 	}
 
-	removed := p.removeConnWithLock(cn)
+	removed, countCb := p.removeConnWithLock(cn)
 
 	if freeTurn {
 		p.freeTurn()
@@ -1691,8 +1791,8 @@ func (p *ConnPool) removeConnInternal(ctx context.Context, cn *Conn, reason erro
 			cb(ctx, cn, MetricStateUsed, "")
 		}
 		// Record connection count decrement (connection removed, assume from used state)
-		if cb := getMetricConnectionCountCallback(); cb != nil {
-			cb(ctx, -1, cn, "used", false)
+		if countCb != nil {
+			countCb(ctx, -1, cn, "used", false)
 		}
 	}
 
@@ -1725,28 +1825,28 @@ func (p *ConnPool) CloseConn(ctx context.Context, cn *Conn, reason string, fromS
 		hookManager.ProcessOnRemove(ctx, cn, errors.New(reason))
 	}
 
-	removed := p.removeConnWithLock(cn)
+	removed, countCb := p.removeConnWithLock(cn)
 
 	// Only emit UpDownCounter decrements if we actually removed the connection.
 	// If removed is false, Close() already removed it and emitted the -1 delta.
 	// Only emit connection closed if we actually owned the removal.
 	// If removed is false, Close() already emitted connectionClosed for this conn.
 	if removed {
-		p.recordConnectionMetrics(ctx, cn, reason, fromState)
+		p.recordConnectionMetrics(ctx, cn, reason, fromState, countCb)
 	}
 
 	return p.closeConn(cn)
 }
 
-func (p *ConnPool) recordConnectionMetrics(ctx context.Context, cn *Conn, reason string, fromState string) {
+func (p *ConnPool) recordConnectionMetrics(ctx context.Context, cn *Conn, reason string, fromState string, countCb connectionCountCallback) {
 	// Record connection state change: connection is being removed from the specified state
 	if cb := getMetricConnectionStateChangeCallback(); cb != nil && fromState != "" {
 		cb(ctx, cn, fromState, "")
 	}
 
 	// Record connection count decrement (UpDownCounter) for the state the connection was in
-	if cb := getMetricConnectionCountCallback(); cb != nil && fromState != "" {
-		cb(ctx, -1, cn, fromState, false)
+	if countCb != nil && fromState != "" {
+		countCb(ctx, -1, cn, fromState, false)
 	}
 
 	if cb := getMetricConnectionClosedCallback(); cb != nil {
@@ -1758,11 +1858,12 @@ func (p *ConnPool) recordConnectionMetrics(ctx context.Context, cn *Conn, reason
 // Returns true if the connection was actually present in p.conns and was removed,
 // false if it was already gone (e.g., removed by Close()). Callers must use the
 // return value to decide whether to emit metric decrements — this eliminates the
-// shutdown race between Close() and concurrent removal paths.
-func (p *ConnPool) removeConnWithLock(cn *Conn) bool {
+// shutdown race between Close() and concurrent removal paths. The returned count
+// callback is read under the same lock, so it matches SetAllMetricCallbacks' snapshot.
+func (p *ConnPool) removeConnWithLock(cn *Conn) (bool, connectionCountCallback) {
 	p.connsMu.Lock()
 	defer p.connsMu.Unlock()
-	return p.removeConn(cn)
+	return p.removeConn(cn), getMetricConnectionCountCallback()
 }
 
 // removeConn removes a connection from the pool's internal data structures.
@@ -1778,13 +1879,14 @@ func (p *ConnPool) removeConn(cn *Conn) bool {
 	// Decrement pool size counter when removing a connection
 	if cn.pooled {
 		p.poolSize.Add(-1)
-		// this can be idle conn
-		for idx, ic := range p.idleConns {
-			if ic == cn {
-				p.idleConns = append(p.idleConns[:idx], p.idleConns[idx+1:]...)
-				p.idleConnsLen.Add(-1)
-				break
-			}
+	}
+	// An unpooled conn can be idle too: putIdleConn parks an undelivered
+	// overflow dial in idleConns.
+	for idx, ic := range p.idleConns {
+		if ic == cn {
+			p.idleConns = append(p.idleConns[:idx], p.idleConns[idx+1:]...)
+			p.idleConnsLen.Add(-1)
+			break
 		}
 	}
 	return true
@@ -1929,6 +2031,7 @@ func (p *ConnPool) RetireConns(ctx context.Context, conns []*Conn, reason string
 		}
 		cn.MarkCloseOnPut(reason)
 	}
+	countCb := getMetricConnectionCountCallback()
 	p.connsMu.Unlock()
 
 	if hookManager := p.hookManager.Load(); hookManager != nil {
@@ -1937,7 +2040,7 @@ func (p *ConnPool) RetireConns(ctx context.Context, conns []*Conn, reason string
 		}
 	}
 	for _, cn := range toClose {
-		p.recordConnectionMetrics(ctx, cn, reason, MetricStateIdle)
+		p.recordConnectionMetrics(ctx, cn, reason, MetricStateIdle, countCb)
 		_ = p.closeConn(cn)
 	}
 	p.checkMinIdleConns()
@@ -1961,7 +2064,7 @@ func (p *ConnPool) Filter(fn func(*Conn) bool) error {
 			if _, isIdle := idleConnSet[cn]; isIdle {
 				// Idle connection - remove from pool and close.
 				p.removeConn(cn)
-				p.recordConnectionMetrics(ctx, cn, CloseReasonFailover, MetricStateIdle)
+				p.recordConnectionMetrics(ctx, cn, CloseReasonFailover, MetricStateIdle, getMetricConnectionCountCallback())
 				err = p.closeConn(cn)
 			} else {
 				// Used connection - set closeReason and close the connection.
@@ -2119,11 +2222,14 @@ func (p *ConnPool) Close() error {
 
 	var firstErr error
 	nowNs := time.Now().UnixNano()
-	p.connsMu.Lock()
 
+	livePoolsMu.Lock()
+	delete(liveConnPools, weak.Make(p))
+	p.connsMu.Lock()
 	// Emit -1 for each connection. Since all idle↔used transitions happen
 	// under connsMu, the idleConns slice is the source of truth for state.
 	cb := getMetricConnectionCountCallback()
+	livePoolsMu.Unlock()
 	idleSet := make(map[uint64]struct{}, len(p.idleConns))
 	for _, cn := range p.idleConns {
 		idleSet[cn.GetID()] = struct{}{}

internal/pool/pool_test.go

@@ -352,7 +352,10 @@ func TestDrainIdleConnsReportsBorrowedConnectionStates(t *testing.T) {
 	var mu sync.Mutex
 	var events []event
 	pool.SetAllMetricCallbacks(&pool.MetricCallbacks{
-		ConnectionCount: func(_ context.Context, delta int, _ *pool.Conn, state string, _ bool) {
+		ConnectionCount: func(_ context.Context, delta int, c *pool.Conn, state string, _ bool) {
+			if c != cn {
+				return
+			}
 			mu.Lock()
 			events = append(events, event{delta: delta, state: state})
 			mu.Unlock()
@@ -366,6 +369,7 @@ func TestDrainIdleConnsReportsBorrowedConnectionStates(t *testing.T) {
 	mu.Lock()
 	defer mu.Unlock()
 	want := []event{
+		{delta: 1, state: pool.MetricStateIdle},
 		{delta: -1, state: pool.MetricStateIdle},
 		{delta: 1, state: pool.MetricStateUsed},
 		{delta: -1, state: pool.MetricStateUsed},
@@ -376,6 +380,170 @@ func TestDrainIdleConnsReportsBorrowedConnectionStates(t *testing.T) {
 	}
 }
 
+func TestConnectionCountIncludesExistingConns(t *testing.T) {
+	connPool := pool.NewConnPool(&pool.Options{
+		Name:               "existing-conns",
+		Dialer:             dummyDialer,
+		PoolSize:           4,
+		MinIdleConns:       4,
+		MaxConcurrentDials: 4,
+		PoolTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	})
+	t.Cleanup(func() { _ = connPool.Close() })
+
+	deadline := time.Now().Add(time.Second)
+	for connPool.Len() != 4 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if got := connPool.Len(); got != 4 {
+		t.Fatalf("MinIdleConns: got %d, want 4", got)
+	}
+
+	var mu sync.Mutex
+	counts := map[string]int{}
+	pool.SetAllMetricCallbacks(&pool.MetricCallbacks{
+		ConnectionCount: func(_ context.Context, delta int, cn *pool.Conn, state string, _ bool) {
+			if cn.PoolName() != "existing-conns" {
+				return
+			}
+			mu.Lock()
+			counts[state] += delta
+			mu.Unlock()
+		},
+	})
+
+	mu.Lock()
+	got := map[string]int{pool.MetricStateIdle: counts[pool.MetricStateIdle], pool.MetricStateUsed: counts[pool.MetricStateUsed]}
+	mu.Unlock()
+	want := map[string]int{pool.MetricStateIdle: 4, pool.MetricStateUsed: 0}
+	if !reflect.DeepEqual(got, want) {
+		pool.SetAllMetricCallbacks(nil)
+		t.Fatalf("after attach: got %v, want %v", got, want)
+	}
+
+	pool.SetAllMetricCallbacks(nil)
+
+	mu.Lock()
+	defer mu.Unlock()
+	if counts[pool.MetricStateIdle] != 0 || counts[pool.MetricStateUsed] != 0 {
+		t.Fatalf("after detach: got %v, want all zero", counts)
+	}
+}
+
+func TestConnectionCountBalancedWhileCallbacksChange(t *testing.T) {
+	opt := &pool.Options{
+		Name:               "callback-churn",
+		Dialer:             dummyDialer,
+		PoolSize:           8,
+		MinIdleConns:       2,
+		MaxConcurrentDials: 8,
+		PoolTimeout:        time.Second,
+		ConnMaxIdleTime:    -1,
+	}
+	connPool := pool.NewConnPool(opt)
+	t.Cleanup(func() { _ = connPool.Close() })
+	pubSubPool := pool.NewPubSubPool(opt, func(ctx context.Context, _, _ string) (net.Conn, error) {
+		return dummyDialer(ctx)
+	})
+	t.Cleanup(func() { _ = pubSubPool.Close() })
+
+	var mu sync.Mutex
+	counts := map[string]int{}
+	callbacks := &pool.MetricCallbacks{
+		ConnectionCount: func(_ context.Context, delta int, cn *pool.Conn, state string, _ bool) {
+			if cn.PoolName() != "callback-churn" {
+				return
+			}
+			mu.Lock()
+			counts[state] += delta
+			mu.Unlock()
+		},
+	}
+
+	ctx := context.Background()
+	stop := make(chan struct{})
+	var wg sync.WaitGroup
+	for i := 0; i < 4; i++ {
+		wg.Add(1)
+		go func(i int) {
+			defer wg.Done()
+			for n := 0; ; n++ {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				cn, err := connPool.Get(ctx)
+				if err != nil {
+					continue
+				}
+				if (n+i)%3 == 0 {
+					connPool.Remove(ctx, cn, nil)
+				} else {
+					connPool.Put(ctx, cn)
+				}
+			}
+		}(i)
+	}
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			cn, err := pubSubPool.NewConn(ctx, "tcp", "", nil)
+			if err != nil {
+				continue
+			}
+			pubSubPool.TrackConn(cn)
+			pubSubPool.UntrackConn(cn)
+			_ = cn.Close()
+		}
+	}()
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+			}
+			p := pool.NewConnPool(&pool.Options{
+				Name:               "callback-churn",
+				Dialer:             dummyDialer,
+				PoolSize:           1,
+				MaxConcurrentDials: 1,
+				PoolTimeout:        time.Second,
+				ConnMaxIdleTime:    -1,
+			})
+			if cn, err := p.Get(ctx); err == nil {
+				p.Put(ctx, cn)
+			}
+			_ = p.Close()
+		}
+	}()
+
+	for i := 0; i < 200; i++ {
+		pool.SetAllMetricCallbacks(callbacks)
+		pool.SetAllMetricCallbacks(nil)
+	}
+	close(stop)
+	wg.Wait()
+
+	mu.Lock()
+	defer mu.Unlock()
+	for state, count := range counts {
+		if count != 0 {
+			t.Fatalf("connection count for state %q after detach: got %d, want 0", state, count)
+		}
+	}
+}
+
 func TestDrainIdleConnsSkipsUninitializedMinIdleConns(t *testing.T) {
 	connPool := pool.NewConnPool(&pool.Options{
 		Dialer:             dummyDialer,

internal/pool/pubsub.go

@@ -3,8 +3,10 @@ package pool
 import (
 	"context"
 	"net"
+	"runtime"
 	"sync"
 	"sync/atomic"
+	"weak"
 )
 
 // PubSubStats contains pub/sub connection pool stats.
@@ -24,17 +26,30 @@ type PubSubPool struct {
 
 	// Map to track active PubSub connections
 	activeConns sync.Map // map[uint64]*Conn (connID -> conn)
-	closed      atomic.Bool
-	stats       PubSubStats
+	// trackMu pairs activeConns changes with the connection count callback they use.
+	trackMu sync.Mutex
+	closed  atomic.Bool
+	stats   PubSubStats
 }
 
 // NewPubSubPool implements a pool for PubSub connections.
 // It intentionally does not implement the Pooler interface
 func NewPubSubPool(opt *Options, netDialer func(ctx context.Context, network, addr string) (net.Conn, error)) *PubSubPool {
-	return &PubSubPool{
+	p := &PubSubPool{
 		opt:       opt,
 		netDialer: netDialer,
 	}
+	wp := weak.Make(p)
+	livePoolsMu.Lock()
+	livePubSubPools[wp] = struct{}{}
+	livePoolsMu.Unlock()
+	// Drops the entry for a pool that is collected without Close.
+	runtime.AddCleanup(p, func(wp weak.Pointer[PubSubPool]) {
+		livePoolsMu.Lock()
+		delete(livePubSubPools, wp)
+		livePoolsMu.Unlock()
+	}, wp)
+	return p
 }
 
 func (p *PubSubPool) NewConn(ctx context.Context, network string, addr string, channels []string) (*Conn, error) {
@@ -55,10 +70,13 @@ func (p *PubSubPool) NewConn(ctx context.Context, network string, addr string, c
 }
 
 func (p *PubSubPool) TrackConn(cn *Conn) {
+	p.trackMu.Lock()
 	atomic.AddUint32(&p.stats.Active, 1)
 	p.activeConns.Store(cn.GetID(), cn)
+	cb := getMetricConnectionCountCallback()
+	p.trackMu.Unlock()
 	// Emit +1 used for PubSub connection
-	if cb := getMetricConnectionCountCallback(); cb != nil {
+	if cb != nil {
 		cb(context.Background(), 1, cn, "used", true)
 	}
 }
@@ -66,20 +84,29 @@ func (p *PubSubPool) TrackConn(cn *Conn) {
 func (p *PubSubPool) UntrackConn(cn *Conn) {
 	// LoadAndDelete ensures each connection is only decremented once,
 	// guarding against double-decrement if Close() already untracked it.
+	p.trackMu.Lock()
 	if _, loaded := p.activeConns.LoadAndDelete(cn.GetID()); !loaded {
+		p.trackMu.Unlock()
 		return
 	}
+	cb := getMetricConnectionCountCallback()
+	p.trackMu.Unlock()
 	atomic.AddUint32(&p.stats.Active, ^uint32(0))
 	atomic.AddUint32(&p.stats.Untracked, 1)
 	// Emit -1 used for PubSub connection
-	if cb := getMetricConnectionCountCallback(); cb != nil {
+	if cb != nil {
 		cb(context.Background(), -1, cn, "used", true)
 	}
 }
 
 func (p *PubSubPool) Close() error {
 	p.closed.Store(true)
+	livePoolsMu.Lock()
+	delete(livePubSubPools, weak.Make(p))
+	p.trackMu.Lock()
 	cb := getMetricConnectionCountCallback()
+	livePoolsMu.Unlock()
+	var conns []*Conn
 	p.activeConns.Range(func(key, value interface{}) bool {
 		cn := value.(*Conn)
 		// Use LoadAndDelete to atomically claim ownership of this entry.
@@ -93,9 +120,15 @@ func (p *PubSubPool) Close() error {
 		if cb != nil {
 			cb(context.Background(), -1, cn, "used", true)
 		}
-		_ = cn.Close()
+		conns = append(conns, cn)
 		return true
 	})
+	p.trackMu.Unlock()
+
+	// Close outside trackMu: cn.Close runs the connection's onClose callback.
+	for _, cn := range conns {
+		_ = cn.Close()
+	}
 	return nil
 }
 

internal/pool/spare_refresh_test.go

@@ -0,0 +1,85 @@
+package pool
+
+import (
+	"context"
+	"net"
+	"testing"
+	"time"
+)
+
+func newUnpooledIdleConn(t *testing.T) (*ConnPool, *Conn) {
+	t.Helper()
+	p := NewConnPool(&Options{
+		Dialer: func(context.Context) (net.Conn, error) {
+			c, _ := net.Pipe()
+			return c, nil
+		},
+		PoolSize:        2,
+		PoolTimeout:     time.Second,
+		DialTimeout:     time.Second,
+		ConnMaxIdleTime: time.Minute,
+	})
+	t.Cleanup(func() { _ = p.Close() })
+
+	// An undelivered dial made while the pool was full is parked in
+	// idleConns by putIdleConn without being pooled.
+	cn, err := p.newConn(context.Background(), false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	p.putIdleConn(context.Background(), cn)
+	return p, cn
+}
+
+func assertNotIdle(t *testing.T, p *ConnPool, cn *Conn) {
+	t.Helper()
+	p.connsMu.Lock()
+	defer p.connsMu.Unlock()
+	for _, ic := range p.idleConns {
+		if ic == cn {
+			t.Fatal("removed unpooled conn is still in idleConns")
+		}
+	}
+}
+
+// TestRefreshStaleSpareDropsUnpooledConn checks an unpooled conn parked in
+// idleConns is dropped from idleConns when it is refreshed as the stale spare.
+func TestRefreshStaleSpareDropsUnpooledConn(t *testing.T) {
+	p, cn := newUnpooledIdleConn(t)
+	p.minIdleFloor.Store(1)
+
+	now := time.Now()
+	cn.SetUsedAt(now.Add(-time.Hour))
+	p.refreshStaleSpare(context.Background(), now.UnixNano())
+	assertNotIdle(t, p, cn)
+
+	// The floor is refilled with a fresh pooled spare.
+	deadline := time.Now().Add(5 * time.Second)
+	for p.Len() != 1 || p.IdleLen() != 1 {
+		if time.Now().After(deadline) {
+			t.Fatalf("Len=%d IdleLen=%d, want 1/1", p.Len(), p.IdleLen())
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+// TestRemoveConnDropsUnpooledIdleConn checks the other removeConn callers
+// that act on idle conns drop an unpooled one from idleConns too.
+func TestRemoveConnDropsUnpooledIdleConn(t *testing.T) {
+	t.Run("RetireConns", func(t *testing.T) {
+		p, cn := newUnpooledIdleConn(t)
+		p.RetireConns(context.Background(), []*Conn{cn}, CloseReasonStale)
+		assertNotIdle(t, p, cn)
+		if n := p.IdleLen(); n != 0 {
+			t.Fatalf("IdleLen=%d, want 0", n)
+		}
+	})
+	t.Run("Filter", func(t *testing.T) {
+		p, cn := newUnpooledIdleConn(t)
+		_ = p.Filter(func(c *Conn) bool { return c == cn })
+		assertNotIdle(t, p, cn)
+		if n := p.IdleLen(); n != 0 {
+			t.Fatalf("IdleLen=%d, want 0", n)
+		}
+	})
+}

otel.go

@@ -93,6 +93,7 @@ type OTelConnectionCounter interface {
 	// delta: +1 when connection added, -1 when connection removed
 	// state: connection state (e.g., "idle", "used")
 	// isPubSub: true if this is a PubSub connection
+	// It may be called with pool locks held, so it must not call back into go-redis.
 	RecordConnectionCount(ctx context.Context, delta int, cn ConnInfo, state string, isPubSub bool)
 
 	// RecordPendingRequests records a change in pending requests (UpDownCounter)