fix(opensearchtransport): don't let unverified discovered nodes mask seed fallback

Node discovery inserts freshly discovered connections into a policy pool's
dead list before they are ever health-checked (createConnection sets
lcDead|lcNeedsWarmup|lcNeedsHardware). Each policy cached its "enabled" bit as
len(ready)+len(dead) > 0, so a policy holding only never-verified discovered
connections still reported enabled. Route then handed the request to the pool,
which served the never-verified connection as a zombie (tryZombieWithLock)
rather than returning ErrNoConnections -- so the request never cascaded to the
user-supplied seed-URL fallback (which only triggers on ErrNoConnections).

When discovered publish_addresses are unroutable from the client (NAT'd or
misconfigured clusters, as seen in CI against a k8s stack cluster), this
hijacks the request stream onto dead pod IPs and every request fails with
"no route to host", even though the seed URL is reachable.

Compute the enabled bit from availability instead of presence: a connection is
available for routing if it is a user-supplied seed (assumed reachable) or a
discovered connection confirmed reachable (lcNeedsHardware cleared). A dead,
never-verified discovered connection no longer counts, so the policy reports
not-enabled, Route returns ErrNoConnections, and the seed fallback serves the
request until a discovered node health-checks clean and takes over.

- Add Connection.seed (immutable, set at the two seed construction sites) and
  Connection.availableForRouting().
- Add multiServerPool.hasAvailableConnsWithLock(); use it for the enabled bit
  in the round-robin, role, and coordinator policies, and apply the same
  per-connection check in the index/doc router's flat active list.

Genuinely-once-healthy dead connections (OnFailure re-sets lcNeedsHardware, but
they are re-verified on the next health check) and steady-state zombie behavior
are unchanged; seeds are always available so the fallback path is preserved.

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

Author: sean-

CHANGELOG.md

@@ -229,6 +229,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.0.0/)
 
 ### Fixed
 
+- Fix node discovery hijacking the request stream with unverified, unreachable discovered nodes and masking the seed-URL fallback. Discovery inserts freshly discovered connections into a policy pool's dead list before they are health-checked (`lcDead|lcNeedsWarmup|lcNeedsHardware`), and each policy cached its enabled bit as `len(ready)+len(dead) > 0` -- so a policy holding only never-verified discovered connections still reported enabled. `Route` then handed the request to the pool, which served the unverified connection as a zombie instead of returning `ErrNoConnections`, so the request never cascaded to the user-supplied seed-URL fallback (which triggers only on `ErrNoConnections`). When discovered `publish_address` values are unroutable from the client (NAT'd or misconfigured clusters, e.g. a Kubernetes stack cluster in CI), this failed every request with `no route to host` even though the seed URL was reachable. A connection is now considered available for routing only if it is a user-supplied seed (assumed reachable) or a discovered connection confirmed reachable (`lcNeedsHardware` cleared); the round-robin, role, coordinator, and index/doc-router policies compute their enabled bit from availability rather than presence, so an all-unverified pool reports not-enabled and the seed fallback serves requests until a discovered node health-checks clean ([#952](https://github.com/opensearch-project/opensearch-go/pull/952))
 - Fix a data race on the multi-server pool's `warmupRounds`, `warmupSkipCount`, and `activeListCap` fields when two concurrent `DiscoverNodes` calls drive `RolePolicy.DiscoveryUpdate` on a shared transport. `RolePolicy` called `recalculateWarmupParams` (which writes those fields) without holding the pool write lock, while the `roundrobin` and `cluster_coordinator` policies took the lock for the identical call. `RolePolicy.DiscoveryUpdate` now computes the projected pool size and recalculates the warmup parameters under `pool.Lock()`, matching the other callers
 - Cache credentials in the `signer/awsv2` constructors. A raw `CredentialsProvider` is wrapped in an `aws.CredentialsCache` (an already-cached provider, such as one from `config.LoadDefaultConfig`, is left as-is), so SigV4 signing no longer calls `Credentials.Retrieve` on every request. For STS-backed providers (assume-role, web identity, IRSA) the previous behavior was a per-request STS call that could exhaust the account's STS rate limits under load. `signer/awsv2` shipped without this in v4.6.0.
 - Fix `cmd/osgen` silently dropping a response struct when a response schema has a `oneOf`/`anyOf` field whose parent-scoped union name collides with the parent struct's own Go name. The union registered first and the parent struct was then dropped by the type registry (its name already taken), degrading the response to raw `json.RawMessage`. Such a union is now re-keyed by its referenced schema so the parent struct survives. The generator also reports any remaining Go type name collisions to stderr at generation time instead of dropping types silently. Regenerating fixes two type families: `tasks.list`, `tasks.cancel`, and `delete_by_query_rethrottle` change from raw `Body json.RawMessage` to typed structs (`NodeFailures`, `TaskFailures`, `Nodes map[string]TasksTaskExecutingNode`, `Tasks *TasksTaskInfos`), and the `_common.mapping___DynamicTemplate.mapping` field becomes typed `*CommonMappingProperty` (accounting for the large `unions_gen.go`/`indices-put_mapping_gen.go` churn). ([#890](https://github.com/opensearch-project/opensearch-go/pull/890))

opensearchtransport/connection.go

@@ -233,6 +233,16 @@ type Connection struct {
 			lastAt time.Time
 		}
 	}
+
+	// seed marks a connection built from a user-supplied Config.URLs address
+	// (as opposed to one learned from node discovery). Immutable after
+	// construction; set only at the seed-connection construction sites in
+	// opensearchtransport.New. Seed connections are assumed reachable and are
+	// always available for routing (see availableForRouting) so they can serve
+	// as the last-resort fallback when no discovered node is verified. The zero
+	// value (false) means "not a seed", so any connection not explicitly minted
+	// as a seed is treated as needing verification.
+	seed bool
 }
 
 // timeToNano converts a time.Time to its Unix-nanosecond representation. The
@@ -338,6 +348,30 @@ func (c *Connection) markAsHealthyWithLock() {
 	c.failures.Store(0)
 }
 
+// availableForRouting reports whether this connection may be routed to when a
+// policy decides if it has anything worth serving. A seed connection (one built
+// from a user-supplied Config.URLs address) is always available: the user
+// asserted it, and it is the last-resort fallback target. Any other connection
+// -- discovered, or a zero-value/test connection -- is available only while
+// currently confirmed reachable, keyed off the lcNeedsHardware lifecycle bit.
+// That bit is set at discovery (createConnection) and on every failure
+// (OnFailure), and cleared whenever the node responds to a hardware/health
+// probe at all (including a non-2xx such as 403 -- it signals reachability, not
+// success). This is a current-state test, not an "ever verified" latch: a
+// connection that verified once and later failed is re-marked lcNeedsHardware
+// and is not available until it responds again. A connection still carrying the
+// bit has never (or not since its last failure) been proven reachable -- a node
+// whose publish_address may be unroutable, e.g. a NAT'd or misconfigured
+// cluster; counting it as available would let it mask the seed-URL fallback.
+// Because the bit is unset on a zero-value connection, connections not minted by
+// discovery default to available. Lock-free (single atomic load).
+func (c *Connection) availableForRouting() bool {
+	if c.seed {
+		return true // user-supplied seed: assumed reachable
+	}
+	return !c.loadConnState().lifecycle().has(lcNeedsHardware)
+}
+
 // RTTMedian returns the median health-check round-trip time for this connection.
 // Returns -1 if no RTT data is available (the connection has not completed
 // enough health checks for the ring buffer median to drop below the unknown

opensearchtransport/opensearchtransport.go

@@ -830,7 +830,7 @@ func New(cfg Config) (*Transport, error) {
 
 	conns := make([]*Connection, len(cfg.URLs))
 	for idx, u := range cfg.URLs {
-		conn := &Connection{URL: u, URLString: u.String()}
+		conn := &Connection{URL: u, URLString: u.String(), seed: true}
 		conn.estLoad.clock = realClock{}
 		conn.weight.Store(1)
 		conns[idx] = conn
@@ -982,7 +982,7 @@ func New(cfg Config) (*Transport, error) {
 	if !client.seedFallbackDisabled && len(cfg.URLs) > 0 {
 		seedConns := make([]*Connection, len(cfg.URLs))
 		for i, u := range cfg.URLs {
-			conn := &Connection{URL: u, URLString: u.String()}
+			conn := &Connection{URL: u, URLString: u.String(), seed: true}
 			conn.estLoad.clock = realClock{}
 			conn.weight.Store(1)
 			conn.mu.Lock()

opensearchtransport/policy_cluster_coordinator.go

@@ -216,9 +216,10 @@ func (p *CoordinatorPolicy) DiscoveryUpdate(added, removed, unchanged []*Connect
 		}
 	}
 
-	// Update cached state
-	hasCoords := len(p.pool.mu.ready) > 0 || len(p.pool.mu.dead) > 0
-	psSetEnabled(&p.policyState, hasCoords)
+	// Update cached state. Dead-but-never-verified connections do not count as
+	// available so the request cascades to the seed fallback rather than being
+	// served as a zombie.
+	psSetEnabled(&p.policyState, p.pool.hasAvailableConnsWithLock())
 
 	return nil
 }

opensearchtransport/policy_index_router.go

@@ -359,7 +359,18 @@ func routerDiscoveryUpdate(
 	*activeConns = all
 	mu.Unlock()
 
-	psSetEnabled(policyState, len(all) > 0)
+	// A router policy is enabled only when it has a connection available for
+	// routing -- a user-supplied seed, or a discovered connection confirmed
+	// reachable. A never-verified discovered connection does not count, so the
+	// request cascades to the seed fallback rather than being served as a zombie.
+	hasAvailable := false
+	for _, c := range all {
+		if c.availableForRouting() {
+			hasAvailable = true
+			break
+		}
+	}
+	psSetEnabled(policyState, hasAvailable)
 
 	return nil
 }

opensearchtransport/policy_role.go

@@ -194,6 +194,17 @@ func (p *RolePolicy) DiscoveryUpdate(added, removed, unchanged []*Connection) er
 		p.discoveryUpdateRemove(removed)
 	}
 
+	// Recompute the cached enabled bit unconditionally, mirroring RoundRobinPolicy
+	// and CoordinatorPolicy. The discoveryUpdateAdd/Remove helpers only recompute
+	// when they touch the pool, so an unchanged-only cycle (the steady state)
+	// would otherwise leave the bit stale after a health check flips a
+	// connection's routability (dead+lcNeedsHardware -> ready) off the discovery
+	// path. Without this, a freshly verified node never becomes routable and
+	// traffic pins to the seed fallback.
+	p.pool.Lock()
+	psSetEnabled(&p.policyState, p.pool.hasAvailableConnsWithLock())
+	p.pool.Unlock()
+
 	// unchanged connections don't need any action
 	return nil
 }
@@ -253,10 +264,6 @@ func (p *RolePolicy) discoveryUpdateAdd(added []*Connection) {
 				p.pool.appendToDeadWithLock(conn)
 			}
 
-			// Update hasMatching state while holding the lock
-			hasConnections := len(p.pool.mu.ready) > 0 || len(p.pool.mu.dead) > 0
-			psSetEnabled(&p.policyState, hasConnections)
-
 			p.pool.Unlock()
 		}
 	}
@@ -346,9 +353,6 @@ func (p *RolePolicy) discoveryUpdateRemove(removed []*Connection) {
 	gap := activeCountBefore - p.pool.mu.activeCount
 	p.pool.promoteStandbyGracefullyWithLock(p.pool.poolCtx(), gap)
 
-	// Update hasMatching state while holding the lock
-	hasConnections := len(p.pool.mu.ready) > 0 || len(p.pool.mu.dead) > 0
-	psSetEnabled(&p.policyState, hasConnections)
 	p.pool.Unlock()
 }
 

opensearchtransport/policy_roundrobin.go

@@ -196,8 +196,14 @@ func (p *RoundRobinPolicy) DiscoveryUpdate(added, removed, unchanged []*Connecti
 		}
 	}
 
-	// Update cached enabled state
-	psSetEnabled(&p.policyState, len(p.pool.mu.ready)+len(p.pool.mu.dead) > 0)
+	// Update cached enabled state. A policy is enabled only when it has a
+	// connection worth routing to -- a ready connection, or a dead connection
+	// currently confirmed reachable (lcNeedsHardware clear; see
+	// availableForRouting). Dead connections not currently confirmed reachable
+	// (freshly discovered, or verified-then-failed, possibly unroutable) do not
+	// count, so the request cascades to the seed fallback instead of being
+	// served as a zombie.
+	psSetEnabled(&p.policyState, p.pool.hasAvailableConnsWithLock())
 
 	return nil
 }

opensearchtransport/pool_multi_server.go

@@ -283,6 +283,33 @@ func (cp *multiServerPool) countByLifecycleWithLock() lifecycleCounts {
 	return counts
 }
 
+// hasAvailableConnsWithLock reports whether the pool holds any connection that
+// is available for routing (see Connection.availableForRouting): any ready
+// connection, or any dead connection that is either a user-supplied seed or
+// currently confirmed reachable.
+//
+// A dead connection not currently confirmed reachable does not count. Counting
+// it as available is what lets an unreachable discovered node mask the seed-URL
+// fallback: the policy reports "enabled", Route hands the request to the pool,
+// and the pool serves the unverified connection as a zombie instead of
+// returning ErrNoConnections. Excluding it here makes the policy report
+// "not enabled" so the request cascades to the seed fallback until a discovered
+// connection actually proves reachable.
+//
+// CALLER RESPONSIBILITIES:
+//   - Caller must hold the pool lock (read or write).
+func (cp *multiServerPool) hasAvailableConnsWithLock() bool {
+	if len(cp.mu.ready) > 0 {
+		return true
+	}
+	for _, c := range cp.mu.dead {
+		if c.availableForRouting() {
+			return true
+		}
+	}
+	return false
+}
+
 // recalculateWarmupParams recalculates activeListCap (when auto-scaling) and sets
 // warmupRounds/warmupSkipCount based on effective pool size.
 //

opensearchtransport/seed_cascade_internal_test.go

@@ -0,0 +1,120 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// The OpenSearch Contributors require contributions made to
+// this file be licensed under the Apache-2.0 license or a
+// compatible open source license.
+
+package opensearchtransport
+
+import (
+	"net/url"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+// TestAvailableForRouting covers the per-connection predicate that decides
+// whether a connection may satisfy a policy's "has anything to serve" check.
+// Seeds are always available; discovered connections are available only once
+// they have been verified (lcNeedsHardware cleared).
+func TestAvailableForRouting(t *testing.T) {
+	newConn := func(seed bool, lc connLifecycle) *Connection {
+		c := &Connection{URL: &url.URL{Scheme: "http", Host: "n:9200"}, seed: seed}
+		c.state.Store(int64(newConnState(lc)))
+		return c
+	}
+
+	tests := []struct {
+		name string
+		conn *Connection
+		want bool
+	}{
+		{
+			name: "seed dead and never verified is still available",
+			conn: newConn(true, lcDead|lcNeedsWarmup|lcNeedsHardware),
+			want: true,
+		},
+		{
+			name: "seed active is available",
+			conn: newConn(true, lcActive),
+			want: true,
+		},
+		{
+			name: "discovered never verified is not available",
+			conn: newConn(false, lcDead|lcNeedsWarmup|lcNeedsHardware),
+			want: false,
+		},
+		{
+			name: "discovered verified (hardware cleared) is available",
+			conn: newConn(false, lcDead|lcNeedsWarmup),
+			want: true,
+		},
+		{
+			name: "zero-value (non-seed, no hardware bit) defaults to available",
+			conn: &Connection{URL: &url.URL{Scheme: "http", Host: "n:9200"}},
+			want: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			require.Equal(t, tt.want, tt.conn.availableForRouting())
+		})
+	}
+}
+
+// TestHasAvailableConnsWithLock covers the pool-level predicate that policies
+// use to set their enabled bit. A pool whose only dead connections are
+// never-verified discovered nodes must report "no available connections" so the
+// request cascades to the seed fallback.
+func TestHasAvailableConnsWithLock(t *testing.T) {
+	seedURL := &url.URL{Scheme: "http", Host: "seed:9200"}
+	discURL := &url.URL{Scheme: "http", Host: "10.0.0.1:9200"}
+
+	deadDiscovered := func() *Connection {
+		c := &Connection{URL: discURL, URLString: discURL.String()}
+		c.state.Store(int64(newConnState(lcDead | lcNeedsWarmup | lcNeedsHardware)))
+		return c
+	}
+	deadSeed := func() *Connection {
+		c := &Connection{URL: seedURL, URLString: seedURL.String(), seed: true}
+		c.state.Store(int64(newConnState(lcDead | lcNeedsWarmup | lcNeedsHardware)))
+		return c
+	}
+	verifiedDiscovered := func() *Connection {
+		c := &Connection{URL: discURL, URLString: discURL.String()}
+		c.state.Store(int64(newConnState(lcDead | lcNeedsWarmup))) // hardware cleared
+		return c
+	}
+
+	t.Run("only never-verified discovered dead conns -> not available", func(t *testing.T) {
+		cp := &multiServerPool{name: "test"}
+		cp.mu.dead = []*Connection{deadDiscovered(), deadDiscovered()}
+		require.False(t, cp.hasAvailableConnsWithLock())
+	})
+
+	t.Run("a dead seed keeps the pool available", func(t *testing.T) {
+		cp := &multiServerPool{name: "test"}
+		cp.mu.dead = []*Connection{deadDiscovered(), deadSeed()}
+		require.True(t, cp.hasAvailableConnsWithLock())
+	})
+
+	t.Run("a verified discovered dead conn keeps the pool available", func(t *testing.T) {
+		cp := &multiServerPool{name: "test"}
+		cp.mu.dead = []*Connection{deadDiscovered(), verifiedDiscovered()}
+		require.True(t, cp.hasAvailableConnsWithLock())
+	})
+
+	t.Run("any ready conn keeps the pool available", func(t *testing.T) {
+		cp := &multiServerPool{name: "test"}
+		ready := &Connection{URL: discURL, URLString: discURL.String()}
+		ready.state.Store(int64(newConnState(lcActive)))
+		cp.mu.ready = []*Connection{ready}
+		require.True(t, cp.hasAvailableConnsWithLock())
+	})
+
+	t.Run("empty pool is not available", func(t *testing.T) {
+		cp := &multiServerPool{name: "test"}
+		require.False(t, cp.hasAvailableConnsWithLock())
+	})
+}