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.

Author: sean-

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,26 @@ 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 once
+// confirmed reachable, keyed off the lcNeedsHardware lifecycle bit. That bit is
+// set at discovery (createConnection) and on every failure (OnFailure), and
+// cleared only by a successful hardware/health check. A discovered connection
+// still carrying lcNeedsHardware has never 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

@@ -253,9 +253,10 @@ 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)
+			// Update hasMatching state while holding the lock. Dead-but-never
+			// -verified connections do not count as serviceable so the request
+			// cascades to the seed fallback rather than being served as a zombie.
+			psSetEnabled(&p.policyState, p.pool.hasAvailableConnsWithLock())
 
 			p.pool.Unlock()
 		}
@@ -346,9 +347,10 @@ 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)
+	// Update hasMatching state while holding the lock. 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())
 	p.pool.Unlock()
 }
 

opensearchtransport/policy_roundrobin.go

@@ -196,8 +196,12 @@ 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
+	// that has been verified at least once. Dead-but-never-verified connections
+	// (freshly discovered, 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 a
+// discovered connection already confirmed reachable.
+//
+// A dead, never-verified discovered connection 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 never-verified 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())
+	})
+}