Two concurrent Next() calls could each spawn a deferredCapEnforcement

 goroutine. Between spawning and lock acquisition, Metrics() and
 discovery could read stale lifecycle bits, observing inconsistent
 active/standby counts.

 Replace go deferredCapEnforcement() with triggerCapEnforcement(), which
 acquires the write lock via TryLock before launching the goroutine. If
 the lock is held, enforcement is skipped and self-heals on the next
 Next() call. RLock callers (snapshot, Metrics) block until the
 goroutine releases the write lock, ensuring consistent observations.

Signed-off-by: Sean Chittenden <sean.chittenden@crowdstrike.com>

Author: sean-

opensearchtransport/coverage_critical_path_internal_test.go

@@ -169,7 +169,7 @@ func TestGetNextActiveConnWithLock(t *testing.T) {
 		pool := makeTestPool("test", conns, 3, sel)
 
 		pool.mu.RLock()
-		got := pool.getNextActiveConnWithLock()
+		got, _ := pool.getNextActiveConnWithLock()
 		pool.mu.RUnlock()
 
 		require.Same(t, conns[1], got)
@@ -182,7 +182,7 @@ func TestGetNextActiveConnWithLock(t *testing.T) {
 		pool := makeTestPool("test", conns, 2, sel)
 
 		pool.mu.RLock()
-		got := pool.getNextActiveConnWithLock()
+		got, _ := pool.getNextActiveConnWithLock()
 		pool.mu.RUnlock()
 
 		require.Nil(t, got)
@@ -198,7 +198,7 @@ func TestGetNextActiveConnWithLock(t *testing.T) {
 		pool := makeTestPool("test", conns, 2, sel)
 
 		pool.mu.RLock()
-		got := pool.getNextActiveConnWithLock()
+		got, _ := pool.getNextActiveConnWithLock()
 		pool.mu.RUnlock()
 
 		require.Same(t, conns[0], got)
@@ -216,9 +216,9 @@ func TestGetNextActiveConnWithLock(t *testing.T) {
 		pool := makeTestPool("test", conns, 3, nil)
 
 		pool.mu.RLock()
-		c1 := pool.getNextActiveConnWithLock()
-		c2 := pool.getNextActiveConnWithLock()
-		c3 := pool.getNextActiveConnWithLock()
+		c1, _ := pool.getNextActiveConnWithLock()
+		c2, _ := pool.getNextActiveConnWithLock()
+		c3, _ := pool.getNextActiveConnWithLock()
 		pool.mu.RUnlock()
 
 		// Round-robin should cycle through all connections

opensearchtransport/policy_chain_internal_test.go

@@ -302,7 +302,6 @@ func TestPolicyChain(t *testing.T) {
 		err := chain.OnFailure(conn)
 		require.NoError(t, err)
 	})
-
 }
 
 // testPolicyWithError is a mock policy that returns errors for testing

opensearchtransport/pool_coverage_internal_test.go

@@ -29,8 +29,8 @@ func TestGetNextActiveConnWithLock_Legacy(t *testing.T) {
 		pool.mu.activeCount = 2
 
 		pool.mu.RLock()
-		first := pool.getNextActiveConnWithLock()
-		second := pool.getNextActiveConnWithLock()
+		first, _ := pool.getNextActiveConnWithLock()
+		second, _ := pool.getNextActiveConnWithLock()
 		pool.mu.RUnlock()
 
 		// With round-robin, two calls should return different connections

opensearchtransport/pool_multi_server.go

@@ -101,32 +101,40 @@ func (cp *multiServerPool) poolCtx() context.Context {
 }
 
 // getNextActiveConnWithLock returns the next active connection using the pool's
-// selector strategy. Falls back to the legacy round-robin counter when no
-// selector is configured.
+// selector strategy and a boolean indicating whether cap enforcement is needed.
+// Falls back to the legacy round-robin counter when no selector is configured.
+//
+// When needsCapEnforce is true, the caller must arrange for cap enforcement
+// after releasing the read lock (triggerCapEnforcement uses TryLock, which
+// fails while any read lock is held). Callers that already hold the write lock
+// can call enforceActiveCapWithLock directly.
 //
 // CALLER RESPONSIBILITIES:
 //   - Caller must hold pool read or write lock
 //   - Caller must ensure cp.mu.activeCount > 0 before calling
-func (cp *multiServerPool) getNextActiveConnWithLock() *Connection {
+func (cp *multiServerPool) getNextActiveConnWithLock() (*Connection, bool) {
 	if cp.selector != nil {
 		conn, activeCap, _, err := cp.selector.selectNext(cp.mu.ready, cp.mu.activeCount)
 		if err != nil {
-			return nil
+			return nil, false
 		}
-		// Handle cap adjustment signal asynchronously (don't block the read path).
+		// Handle cap adjustment signals from the selector.
+		// capGrow (standby promotion) stays async -- adding a connection is safe.
+		// capShrink is signaled back to the caller, which must call
+		// triggerCapEnforcement after releasing the read lock.
 		switch activeCap {
 		case capGrow:
 			go cp.deferredStandbyPromotion()
 		case capShrink:
-			go cp.deferredCapEnforcement()
+			return conn, true
 		}
-		return conn
+		return conn, false
 	}
 
 	// Legacy round-robin fallback.
 	next := cp.nextReady.Add(1)
 	idx := int(next-1) % cp.mu.activeCount
-	return cp.mu.ready[idx]
+	return cp.mu.ready[idx], false
 }
 
 // deferredStandbyPromotion acquires the pool write lock and promotes one
@@ -161,6 +169,29 @@ func (cp *multiServerPool) deferredStandbyPromotion() {
 	}
 }
 
+// triggerCapEnforcement attempts to start asynchronous cap enforcement.
+//
+// Cap enforcement trims the active partition when activeCount > activeListCap
+// by demoting fully-warmed connections to standby. Triggered by warmup
+// completion in Next() and selector capShrink signals.
+//
+// TryLock keeps enforcement off the Next() hot path: if the write lock is
+// available, we acquire it and launch a goroutine that runs enforcement and
+// releases the lock on exit. If the lock is held (by another enforcement
+// goroutine, discovery, etc.), the call is a no-op. Self-heals because
+// Next() re-checks activeCount > cap on every warmup completion and
+// capShrink signal.
+func (cp *multiServerPool) triggerCapEnforcement() {
+	if !cp.mu.TryLock() {
+		return
+	}
+
+	go func() {
+		defer cp.mu.Unlock()
+		cp.enforceActiveCapWithLock()
+	}()
+}
+
 // snapshot returns a point-in-time PolicySnapshot of this pool's partitions and counters.
 func (cp *multiServerPool) snapshot() PolicySnapshot {
 	cp.mu.RLock()

opensearchtransport/pool_selection.go

@@ -49,9 +49,15 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 	if cp.mu.activeCount > 0 { //nolint:nestif // warmup skip/accept/starvation requires nested branching
 		var bestWarmingConn *Connection
 		bestWarmingRemSkip := int(^uint(0) >> 1) // max int
+		var needsCapEnforce bool
 
 		for attempt := range cp.mu.activeCount {
-			conn := cp.getNextActiveConnWithLock()
+			conn, selectorCapEnforce := cp.getNextActiveConnWithLock()
+			needsCapEnforce = needsCapEnforce || selectorCapEnforce
+
+			if conn == nil {
+				continue // selector error
+			}
 			state := conn.loadConnState()
 
 			if state.lifecycle()&(lcActive|lcStandby) == 0 {
@@ -67,6 +73,9 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 			if !state.isWarmingUp() {
 				cp.mu.RUnlock()
 				cp.poolRequests.Add(1)
+				if needsCapEnforce {
+					cp.triggerCapEnforcement()
+				}
 				return conn, nil
 			}
 
@@ -77,11 +86,11 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 				// Check if warmup finished and cap enforcement is needed.
 				warmupDone := !conn.loadConnState().isWarmingUp()
 				if warmupDone && cp.activeListCap > 0 && cp.mu.activeCount > cp.activeListCap {
+					needsCapEnforce = true
 					if debugLogger != nil {
 						debugLogger.Logf("[%s] Next: warmup complete for %s, triggering cap enforcement (active=%d, cap=%d)\n",
 							cp.name, conn.URL, cp.mu.activeCount, cp.activeListCap)
 					}
-					go cp.deferredCapEnforcement()
 				} else if warmupDone && debugLogger != nil {
 					debugLogger.Logf("[%s] Next: warmup complete for %s, no cap enforcement (active=%d, cap=%d)\n",
 						cp.name, conn.URL, cp.mu.activeCount, cp.activeListCap)
@@ -95,12 +104,18 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 
 				cp.mu.RUnlock()
 				cp.poolRequests.Add(1)
+				if needsCapEnforce {
+					cp.triggerCapEnforcement()
+				}
 				return conn, nil
 
 			case warmupInactive:
 				// Warmup completed between our isWarmingUp() check and tryWarmupSkip() call.
 				cp.mu.RUnlock()
 				cp.poolRequests.Add(1)
+				if needsCapEnforce {
+					cp.triggerCapEnforcement()
+				}
 				return conn, nil
 
 			case warmupSkipped:
@@ -126,6 +141,9 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 			}
 			cp.mu.RUnlock()
 			cp.poolRequests.Add(1)
+			if needsCapEnforce {
+				cp.triggerCapEnforcement()
+			}
 			return bestWarmingConn, nil
 		}
 	}
@@ -141,15 +159,6 @@ func (cp *multiServerPool) Next() (*Connection, error) {
 	return cp.nextFallback()
 }
 
-// deferredCapEnforcement acquires the pool write lock and trims the active
-// partition if it exceeds activeListCap. Called as a goroutine when warmup
-// completes and the active partition is temporarily over capacity.
-func (cp *multiServerPool) deferredCapEnforcement() {
-	cp.mu.Lock()
-	defer cp.mu.Unlock()
-	cp.enforceActiveCapWithLock()
-}
-
 // nextWithEviction acquires a write lock and iterates active connections,
 // evicting any that were externally demoted (lifecycle != lcActive) by another
 // pool's stats poller. Returns the first healthy connection found, or falls
@@ -168,10 +177,16 @@ func (cp *multiServerPool) nextWithEviction() (*Connection, error) {
 			break
 		}
 
-		conn := cp.getNextActiveConnWithLock()
+		conn, needsCapEnforce := cp.getNextActiveConnWithLock()
+		if conn == nil {
+			continue
+		}
 		state := conn.loadConnState()
 
 		if state.lifecycle()&(lcActive|lcStandby) != 0 {
+			if needsCapEnforce {
+				cp.enforceActiveCapWithLock()
+			}
 			cp.poolRequests.Add(1)
 			return conn, nil
 		}
@@ -192,23 +207,34 @@ func (cp *multiServerPool) nextFallback() (*Connection, error) {
 	// Double-check active after acquiring write lock
 	if cp.mu.activeCount > 0 {
 		// Re-check state on the selected connection under write lock
-		conn := cp.getNextActiveConnWithLock()
-		state := conn.loadConnState()
-		if state.lifecycle()&(lcActive|lcStandby) != 0 {
-			cp.poolRequests.Add(1)
-			return conn, nil
+		conn, needsCapEnforce := cp.getNextActiveConnWithLock()
+		if conn != nil {
+			state := conn.loadConnState()
+			if state.lifecycle()&(lcActive|lcStandby) != 0 {
+				if needsCapEnforce {
+					cp.enforceActiveCapWithLock()
+				}
+				cp.poolRequests.Add(1)
+				return conn, nil
+			}
+			// Externally killed (no position bits) -- evict and continue
+			cp.evictExternallyDemotedWithLock(conn, state)
 		}
-		// Externally killed (no position bits) -- evict and continue
-		cp.evictExternallyDemotedWithLock(conn, state)
 		// Try remaining active connections
 		maxSkips := cp.mu.activeCount
 		for range maxSkips {
 			if cp.mu.activeCount <= 0 {
 				break
 			}
-			conn = cp.getNextActiveConnWithLock()
-			state = conn.loadConnState()
+			conn, needsCapEnforce = cp.getNextActiveConnWithLock()
+			if conn == nil {
+				continue
+			}
+			state := conn.loadConnState()
 			if state.lifecycle()&(lcActive|lcStandby) != 0 {
+				if needsCapEnforce {
+					cp.enforceActiveCapWithLock()
+				}
 				cp.poolRequests.Add(1)
 				return conn, nil
 			}

opensearchtransport/pool_selection_internal_test.go

@@ -311,15 +311,17 @@ func TestEvictExternallyDemotedWithLock(t *testing.T) {
 	})
 }
 
-func TestDeferredCapEnforcement(t *testing.T) {
+func TestEnforceActiveCap(t *testing.T) {
 	t.Run("reduces active to cap", func(t *testing.T) {
 		a1 := newActiveConn("a1")
 		a2 := newActiveConn("a2")
 		a3 := newActiveConn("a3")
 		pool := newStandbyPool([]*Connection{a1, a2, a3}, nil)
 		pool.activeListCap = 2
 
-		pool.deferredCapEnforcement()
+		pool.mu.Lock()
+		pool.enforceActiveCapWithLock()
+		pool.mu.Unlock()
 
 		require.Equal(t, 2, pool.mu.activeCount)
 		require.Len(t, pool.mu.ready, 3)

opensearchtransport/pool_standby.go

@@ -152,7 +152,7 @@ func (cp *multiServerPool) promoteFromOverloaded(c *Connection) {
 // always allowed in the active partition alongside the capped warmed ones.
 // This ensures proven active connections aren't evicted to make room for a
 // warming connection that hasn't yet integrated into the traffic mix.
-// When a warming connection finishes warmup (detected in Next()), deferredCapEnforcement
+// When a warming connection finishes warmup (detected in Next()), triggerCapEnforcement
 // fires and this function evicts the excess fully-warmed connection.
 //
 // No-op when activeListCap <= 0 (disabled) or when the fully-warmed active count

opensearchtransport/standby_rotation_integration_test.go

@@ -169,6 +169,7 @@ func discoverWithStandby(t *testing.T, transport *opensearchtransport.Client) op
 
 	var m opensearchtransport.Metrics
 	var lastErr error
+	var lastActive, lastStandby, lastDead int
 
 	// On slower clusters (e.g. v2.0.1 with security plugin), health checks
 	// during discovery can take several seconds per cycle. Allow up to 30s
@@ -194,8 +195,13 @@ func discoverWithStandby(t *testing.T, transport *opensearchtransport.Client) op
 		m, err = transport.Metrics()
 		require.NoError(t, err)
 
-		t.Logf("Discovery attempt %d: active=%d, standby=%d, dead=%d",
-			attempt, m.LiveConnections-m.StandbyConnections, m.StandbyConnections, m.DeadConnections)
+		active := m.LiveConnections - m.StandbyConnections
+		// Only log when state changes to reduce CI noise.
+		if active != lastActive || m.StandbyConnections != lastStandby || m.DeadConnections != lastDead {
+			t.Logf("Discovery attempt %d: active=%d, standby=%d, dead=%d",
+				attempt, active, m.StandbyConnections, m.DeadConnections)
+			lastActive, lastStandby, lastDead = active, m.StandbyConnections, m.DeadConnections
+		}
 
 		if m.StandbyConnections >= 2 {
 			return m
@@ -205,7 +211,10 @@ func discoverWithStandby(t *testing.T, transport *opensearchtransport.Client) op
 	if lastErr != nil {
 		require.NoError(t, lastErr, "all discovery attempts failed")
 	}
-	return m
+	require.FailNowf(t, "discoverWithStandby timed out",
+		"pool did not reach 2 standby after %d attempts (active=%d, standby=%d, dead=%d)",
+		attempt, lastActive, lastStandby, lastDead)
+	return m // unreachable
 }
 
 // TestStandbyRotation verifies that standby rotation works end-to-end against
@@ -316,12 +325,12 @@ func TestStandbyRotation(t *testing.T) {
 		// Each DiscoverNodes call triggers rotateStandby at its end. When the
 		// pool was just rebuilt from new Connection objects (all lcActive+warming),
 		// rotation finds no standbys and is a no-op. drainWarmup then completes
-		// warmup and fires deferredCapEnforcement, re-establishing the standby
+		// warmup and fires triggerCapEnforcement, re-establishing the standby
 		// partition. On the *next* DiscoverNodes, connections are reused with
 		// their lcStandby state intact, so rotation finds standbys and promotes one.
 		//
 		// We wait for both promotion AND demotion before breaking because
-		// deferredCapEnforcement runs asynchronously (goroutine spawned by
+		// triggerCapEnforcement runs asynchronously (goroutine spawned by
 		// Next() when warmup completes). By requiring the demotion count to
 		// advance, we ensure cap enforcement has finished before we assert
 		// pool state. If the goroutine hasn't run yet, the next iteration's
@@ -406,7 +415,7 @@ func TestStandbyRotation(t *testing.T) {
 		//     clusters like 2.1.0 can move standbys to dead, so we need to
 		//     wait for resurrection + cap enforcement before the next rotation).
 		//  2. Call DiscoverNodes which triggers rotateStandbyConnections at the end.
-		//  3. Drain warmup so deferredCapEnforcement can fire.
+		//  3. Drain warmup so triggerCapEnforcement can fire.
 		//  4. Check if the observer recorded a new promotion.
 		for cycle := range 12 {
 			prevPromotions := obs.promotionCount()