fix(opensearchtransport): serialize discovery warmup recalc under pool lock

Backport of the opensearch-project#981 follow-up race fixed on v5 (opensearch-project#989).

In createOrUpdateMultiNodePoolWithLock, the warmup recalculation, ready-list
partitioning, and mu.activeCount write ran after the pool lock was released.
Those touch mu-guarded fields (activeListCap, warmupRounds, warmupSkipCount,
activeCount) that resurrectWithLock also reads/writes under pool.mu. The caller
holds c.mu(W), which serializes them against metrics.snapshot() but not against
resurrection -- a data race. Hold allConnsPool.mu across the whole section; the
per-connection conn.mu is taken inside the loop, matching the pool.mu -> conn.mu
ordering already used by deferredStandbyPromotion.

Rename recalculateWarmupParams/getWarmupParams to *WithLock: every caller now
holds the pool lock (the discovery path was the last one that didn't), so the
suffix documents the invariant, consistent with the pool's other WithLock
methods. Comment-only intent; no behavior change from the rename.

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

Author: sean-

CHANGELOG.md

@@ -215,6 +215,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 
+- Fix a data race on the multi-server pool's warmup fields in the node-discovery path: `createOrUpdateMultiNodePoolWithLock` recalculated warmup parameters, partitioned the ready list, and wrote `mu.activeCount` after releasing the pool write lock, touching `mu`-guarded fields (`activeListCap`, `warmupRounds`, `warmupSkipCount`, `activeCount`) that `resurrectWithLock` reads and writes under `pool.mu`. The caller holds the transport lock, which serializes this against `metrics.snapshot()` but not against resurrection. The whole section now runs under `allConnsPool.mu`, with per-connection `conn.mu` taken inside the loop (the `pool.mu` -> `conn.mu` ordering already used by `deferredStandbyPromotion`). `recalculateWarmupParams`/`getWarmupParams` are renamed `*WithLock` to document that every caller now holds the pool lock (the discovery path was the last that did not)
 - Fix node discovery hijacking the request stream with unverified, unreachable discovered nodes and masking the user-supplied seed-URL fallback. When discovered `publish_address` values are unroutable from the client (NAT'd or misconfigured clusters, e.g. a Kubernetes stack cluster in CI), a freshly discovered but never-health-checked node could be served to requests as a zombie -- failing every request with `no route to host` -- instead of returning `ErrNoConnections` and cascading to the reachable seed URL. Connections are now considered available for routing only when they are a user-supplied seed (assumed reachable) or a discovered node confirmed reachable, and every routing policy and pool (round-robin, role, coordinator, index/doc router, single-server, and multi-server pools) consistently honors that gate on both the enabled-bit and connection-selection paths, so the seed fallback serves requests until a discovered node health-checks clean ([#952](https://github.com/opensearch-project/opensearch-go/pull/952), [#954](https://github.com/opensearch-project/opensearch-go/pull/954), [#956](https://github.com/opensearch-project/opensearch-go/pull/956))
 - Fix a data race (reported by the race detector in `TestClientCustomTransport`) between `multiServerPool.snapshot()` and node discovery: `snapshot()` read `activeListCap` after releasing the pool read lock, while `recalculateWarmupParams` writes it under the write lock during `DiscoveryUpdate`. `activeListCap`, `warmupRounds`, and `warmupSkipCount` were guarded by the pool lock only by convention (declared at the top level of the struct), which let the unlocked read look correct; they are now nested inside the pool's lock-guarded `mu` struct so every access is spelled `cp.mu.<field>` and the guard is structural, and `snapshot()` reads `activeListCap` while holding the read lock. For the same reason `healthCheck` is moved under `mu` (it is rewritten by `updateConnectionPool` on pool reuse); this also surfaced one discovery-path read of `healthCheck` that had escaped the lock, now taken under the read lock
 - Fix `BulkIndexerStats.NumAdded` overcounting items rejected by `Add()` when the caller's context is cancelled before the item could be enqueued: increment `NumAdded` only after the queue accepts the item, and add a new `BulkAddFailCount` counter for items dropped on the `<-ctx.Done()` branch. Migrate `bulkIndexerStats` fields to `sync/atomic.Uint64` typed values so future direct access is a compile-time error rather than a `-race`-only finding ([#783](https://github.com/opensearch-project/opensearch-go/issues/783))

opensearchtransport/discovery.go

@@ -1031,9 +1031,19 @@ func (c *Client) createOrUpdateMultiNodePoolWithLock(readyConnections, deadConne
 		}
 	}
 
+	// Recalculate warmup parameters and partition the ready list under the pool
+	// write lock. recalculateWarmupParamsWithLock writes mu.activeListCap/warmupRounds/
+	// warmupSkipCount, getWarmupParamsWithLock reads them, and the final assignment sets
+	// mu.activeCount -- all mu-guarded fields that resurrectWithLock also touches
+	// under pool.mu. Holding the lock across the whole section serializes it
+	// against resurrection (c.mu, held by the caller, only serializes it against
+	// metrics.snapshot). Per-connection conn.mu is taken inside the loop, matching
+	// the pool.mu -> conn.mu ordering used by deferredStandbyPromotion.
+	allConnsPool.mu.Lock()
+
 	// Recalculate activeListCap and warmup parameters for the allConns pool before
 	// partitioning so startWarmup calls use the correctly-scaled values.
-	allConnsPool.recalculateWarmupParams(len(allReadyConns) + len(allDeadConns))
+	allConnsPool.recalculateWarmupParamsWithLock(len(allReadyConns) + len(allDeadConns))
 
 	// Partition ready connections by their current lifecycle state.
 	// Reused connections (unchanged in discovery) may already be in standby
@@ -1065,7 +1075,7 @@ func (c *Client) createOrUpdateMultiNodePoolWithLock(readyConnections, deadConne
 			conn.mu.Lock()
 			conn.casLifecycle(conn.loadConnState(), 0, lcActive, lcUnknown|lcStandby) //nolint:errcheck // lock held; only errLifecycleNoop possible
 			conn.mu.Unlock()
-			rounds, skip := allConnsPool.getWarmupParams()
+			rounds, skip := allConnsPool.getWarmupParamsWithLock()
 			conn.startWarmup(rounds, skip)
 			if i != activeCount {
 				allReadyConns[i], allReadyConns[activeCount] = allReadyConns[activeCount], allReadyConns[i]
@@ -1074,6 +1084,7 @@ func (c *Client) createOrUpdateMultiNodePoolWithLock(readyConnections, deadConne
 		}
 	}
 	allConnsPool.mu.activeCount = activeCount
+	allConnsPool.mu.Unlock()
 
 	// NOTE: enforceActiveCapWithLock() is intentionally NOT called here.
 	// The allConns pool is a transport-level container for discovery bookkeeping.

opensearchtransport/policy_cluster_coordinator.go

@@ -130,7 +130,7 @@ func (p *CoordinatorPolicy) DiscoveryUpdate(added, removed, unchanged []*Connect
 			targetPoolSize--
 		}
 	}
-	p.pool.recalculateWarmupParams(targetPoolSize)
+	p.pool.recalculateWarmupParamsWithLock(targetPoolSize)
 
 	// Add new coordinating-only connections
 	for _, newConn := range added {
@@ -152,7 +152,7 @@ func (p *CoordinatorPolicy) DiscoveryUpdate(added, removed, unchanged []*Connect
 			newConn.mu.Lock()
 			newConn.casLifecycle(newConn.loadConnState(), 0, lcActive, lcUnknown|lcStandby) //nolint:errcheck,lll // lock held; only errLifecycleNoop possible
 			newConn.mu.Unlock()
-			rounds, skip := p.pool.getWarmupParams()
+			rounds, skip := p.pool.getWarmupParamsWithLock()
 			newConn.startWarmup(rounds, skip)
 			p.pool.appendToReadyActiveWithLock(newConn)
 		} else {

opensearchtransport/policy_role.go

@@ -164,8 +164,8 @@ func (p *RolePolicy) DiscoveryUpdate(added, removed, unchanged []*Connection) er
 
 	// Compute projected pool size for warmup/activeListCap scaling and
 	// recalculate the warmup parameters under the pool write lock. The lock is
-	// required because recalculateWarmupParams writes the pool's warmupRounds,
-	// warmupSkipCount, and activeListCap fields, which getWarmupParams and the
+	// required because recalculateWarmupParamsWithLock writes the pool's warmupRounds,
+	// warmupSkipCount, and activeListCap fields, which getWarmupParamsWithLock and the
 	// other DiscoveryUpdate callers (roundrobin, cluster_coordinator) read and
 	// write under the same lock. Done before the mutations below so startWarmup
 	// calls during discoveryUpdateAdd use the new values. The lock is released
@@ -183,7 +183,7 @@ func (p *RolePolicy) DiscoveryUpdate(added, removed, unchanged []*Connection) er
 			targetPoolSize--
 		}
 	}
-	p.pool.recalculateWarmupParams(targetPoolSize)
+	p.pool.recalculateWarmupParamsWithLock(targetPoolSize)
 	p.pool.Unlock()
 
 	if added != nil {
@@ -248,7 +248,7 @@ func (p *RolePolicy) discoveryUpdateAdd(added []*Connection) {
 				conn.mu.Lock()
 				conn.casLifecycle(conn.loadConnState(), 0, lcActive, lcUnknown|lcStandby) //nolint:errcheck // lock held; only errLifecycleNoop possible
 				conn.mu.Unlock()
-				rounds, skip := p.pool.getWarmupParams()
+				rounds, skip := p.pool.getWarmupParamsWithLock()
 				conn.startWarmup(rounds, skip)
 				p.pool.appendToReadyActiveWithLock(conn)
 				p.pool.shuffleActiveWithLock()

opensearchtransport/policy_role_internal_test.go

@@ -354,7 +354,7 @@ func TestInvalidRoleError(t *testing.T) {
 
 // TestRolePolicyDiscoveryUpdateConcurrent guards against the data race that
 // occurs when two DiscoverNodes calls drive DiscoveryUpdate on a shared pool
-// simultaneously. recalculateWarmupParams writes the pool's warmupRounds,
+// simultaneously. recalculateWarmupParamsWithLock writes the pool's warmupRounds,
 // warmupSkipCount, and activeListCap fields; those writes must happen under the
 // pool write lock (as the roundrobin and cluster_coordinator policies already
 // do). Without the lock, concurrent updates race on those fields. Run under
@@ -367,7 +367,7 @@ func TestRolePolicyDiscoveryUpdateConcurrent(t *testing.T) {
 	require.NoError(t, rolePolicy.configurePolicySettings(createTestConfig()))
 
 	// Two connections that alternate in and out of the pool so each goroutine
-	// drives a real add/remove pass through recalculateWarmupParams.
+	// drives a real add/remove pass through recalculateWarmupParamsWithLock.
 	connA := createTestConnection("http://localhost:9200", RoleData)
 	connB := createTestConnection("http://localhost:9201", RoleData)
 

opensearchtransport/policy_roundrobin.go

@@ -114,7 +114,7 @@ func (p *RoundRobinPolicy) DiscoveryUpdate(added, removed, unchanged []*Connecti
 	// Recalculate activeListCap and warmup parameters based on projected pool size.
 	// Done before adds/removes so startWarmup calls use the correctly-scaled values.
 	targetPoolSize := len(p.pool.mu.ready) + len(p.pool.mu.dead) + len(added) - len(removed)
-	p.pool.recalculateWarmupParams(targetPoolSize)
+	p.pool.recalculateWarmupParamsWithLock(targetPoolSize)
 
 	// Add new connections based on their health status
 	for _, conn := range added {
@@ -132,7 +132,7 @@ func (p *RoundRobinPolicy) DiscoveryUpdate(added, removed, unchanged []*Connecti
 			conn.mu.Lock()
 			conn.casLifecycle(conn.loadConnState(), 0, lcActive, lcUnknown|lcStandby) //nolint:errcheck // lock held; only errLifecycleNoop possible
 			conn.mu.Unlock()
-			rounds, skip := p.pool.getWarmupParams()
+			rounds, skip := p.pool.getWarmupParamsWithLock()
 			conn.startWarmup(rounds, skip)
 			p.pool.appendToReadyActiveWithLock(conn)
 

opensearchtransport/pool_coverage_internal_test.go

@@ -325,7 +325,7 @@ func TestRecalculateWarmupParams(t *testing.T) {
 	t.Run("auto-scales activeListCap", func(t *testing.T) {
 		t.Parallel()
 		pool := &multiServerPool{}
-		pool.recalculateWarmupParams(5)
+		pool.recalculateWarmupParamsWithLock(5)
 
 		require.Equal(t, 5, pool.mu.activeListCap)
 		require.Positive(t, pool.mu.warmupRounds)
@@ -337,7 +337,7 @@ func TestRecalculateWarmupParams(t *testing.T) {
 		explicitCap := 2
 		pool := &multiServerPool{activeListCapConfig: &explicitCap}
 		pool.mu.activeListCap = 2
-		pool.recalculateWarmupParams(5)
+		pool.recalculateWarmupParamsWithLock(5)
 
 		// activeListCap should not change when explicit
 		require.Equal(t, 2, pool.mu.activeListCap)

opensearchtransport/pool_multi_server.go

@@ -47,7 +47,7 @@ type multiServerPool struct {
 		// instead of active when the ready list's active partition is at capacity.
 		activeListCap int // 0 = disabled (all connections go to active)
 
-		// Dynamic warmup parameters, scaled by recalculateWarmupParams().
+		// Dynamic warmup parameters, scaled by recalculateWarmupParamsWithLock().
 		// Small pools get lighter warmup (fewer rounds, fewer skips) so connections
 		// ramp up quickly. Large pools get heavier warmup to avoid traffic spikes.
 		warmupRounds    int // 0 = use defaultWarmupRounds
@@ -314,7 +314,7 @@ func (cp *multiServerPool) hasAvailableConnsWithLock() bool {
 	return false
 }
 
-// recalculateWarmupParams recalculates activeListCap (when auto-scaling) and sets
+// recalculateWarmupParamsWithLock recalculates activeListCap (when auto-scaling) and sets
 // warmupRounds/warmupSkipCount based on effective pool size.
 //
 // poolSize is the projected total number of connections in the pool (ready + dead)
@@ -326,7 +326,10 @@ func (cp *multiServerPool) hasAvailableConnsWithLock() bool {
 //	n = poolSize                   when activeListCap <= 0
 //	rounds = clamp(n, minWarmupRounds, maxWarmupRounds)
 //	skipCount = rounds * warmupSkipMultiple
-func (cp *multiServerPool) recalculateWarmupParams(poolSize int) {
+//
+// Caller must hold the pool write lock: it reads and writes the mu-guarded
+// activeListCap, warmupRounds, and warmupSkipCount fields.
+func (cp *multiServerPool) recalculateWarmupParamsWithLock(poolSize int) {
 	// Auto-scale activeListCap when the user didn't specify an explicit value.
 	if cp.activeListCapConfig == nil && poolSize > 0 {
 		cp.mu.activeListCap = poolSize
@@ -345,9 +348,12 @@ func (cp *multiServerPool) recalculateWarmupParams(poolSize int) {
 	cp.mu.warmupSkipCount = rounds * warmupSkipMultiple
 }
 
-// getWarmupParams returns the effective warmup parameters for this pool.
+// getWarmupParamsWithLock returns the effective warmup parameters for this pool.
 // Returns pool-specific values if set, otherwise falls back to defaults.
-func (cp *multiServerPool) getWarmupParams() (int, int) {
+//
+// Caller must hold the pool lock: it reads the mu-guarded warmupRounds and
+// warmupSkipCount fields.
+func (cp *multiServerPool) getWarmupParamsWithLock() (int, int) {
 	rounds := cp.mu.warmupRounds
 	if rounds <= 0 {
 		rounds = defaultWarmupRounds
@@ -602,7 +608,7 @@ func (cp *multiServerPool) resurrectWithLock(c *Connection) {
 	if cp.mu.activeListCap <= 0 || cp.mu.activeCount < cp.mu.activeListCap {
 		// Transition state: dead -> active with warmup (lcNeedsWarmup preserved if set)
 		c.casLifecycle(c.loadConnState(), 0, lcActive, lcUnknown|lcStandby) //nolint:errcheck // lock held; only errLifecycleNoop possible
-		rounds, skip := cp.getWarmupParams()
+		rounds, skip := cp.getWarmupParamsWithLock()
 		c.startWarmup(rounds, skip)
 		cp.appendToReadyActiveWithLock(c)
 		cp.shuffleActiveWithLock()

opensearchtransport/pool_standby.go

@@ -302,7 +302,7 @@ func (cp *multiServerPool) promoteStandbyWithLock(c *Connection) bool {
 	c.mu.Lock()
 	c.casLifecycle(c.loadConnState(), 0, lcActive, lcStandby) //nolint:errcheck // lock held; only errLifecycleNoop possible
 	c.mu.Unlock()
-	rounds, skip := cp.getWarmupParams()
+	rounds, skip := cp.getWarmupParamsWithLock()
 	c.startWarmup(rounds, skip)
 	cp.mu.activeCount++