feat(opensearchtransport): make DiscoverNodes blocking, add *bool DiscoverNodesOnStart

DiscoverNodes now waits for an in-flight discovery to complete (or for
the context to be cancelled) instead of returning nil immediately. This
lets callers block until topology data is available after client
construction.

- Auto-enable DiscoverNodesOnStart when OPENSEARCH_GO_ROUTER=true and
  the caller did not set the field

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

Author: sean-

opensearch_internal_test.go

@@ -50,6 +50,8 @@ import (
 
 var called int
 
+func boolPtr(v bool) *bool { return &v }
+
 var defaultRoundTripFunc = func(req *http.Request) (*http.Response, error) {
 	response := &http.Response{Header: http.Header{}}
 
@@ -78,8 +80,6 @@ type testReq struct {
 	Headers http.Header
 }
 
-func boolPtr(v bool) *bool { return &v }
-
 func (r testReq) GetRequest(method string) (*http.Request, error) {
 	if r.Error {
 		return nil, fmt.Errorf("test error")

opensearchtransport/discovery.go

@@ -252,34 +252,90 @@ func (m *_NodesMeta) formatFailures() string {
 	return string(b)
 }
 
-// DiscoverNodes reloads the client connections by fetching information from the cluster.
+// DiscoverNodes reloads the client connections by fetching information from
+// the cluster. If another discovery is already in progress, DiscoverNodes
+// blocks until that discovery completes (or ctx is cancelled) and returns
+// its result.
 func (c *Client) DiscoverNodes(ctx context.Context) error {
-	// Bail out early if the context is already cancelled (e.g. client shutting down).
 	if ctx.Err() != nil {
 		return ctx.Err()
 	}
 
-	// Prevent concurrent discovery operations
-	c.mu.Lock()
-	if c.mu.discoveryInProgress {
-		c.mu.Unlock()
+	c.discoverMu.Lock()
+
+	if c.discoverMu.inProgress {
+		// Another goroutine is running discovery. Wait for it using
+		// sync.Cond + context.AfterFunc so that context cancellation
+		// wakes us even though Cond.Wait is not context-aware.
+		stopf := context.AfterFunc(ctx, func() {
+			c.discoverMu.Lock()
+			defer c.discoverMu.Unlock()
+			c.discoverMu.cond.Broadcast()
+		})
+		defer stopf()
+
+		for c.discoverMu.inProgress {
+			c.discoverMu.cond.Wait()
+			if ctx.Err() != nil {
+				c.discoverMu.Unlock()
+				return ctx.Err()
+			}
+		}
+		err := c.discoverMu.lastErr
+		c.discoverMu.Unlock()
+		return err
+	}
+
+	// We won the race: start discovery.
+	// Lock is held — doDiscoverNodes takes ownership and releases it.
+	return c.doDiscoverNodes(ctx)
+}
+
+// tryDiscoverNodes attempts to start a discovery cycle. If discovery is
+// already in progress it returns nil immediately without waiting.
+//
+// This is used by the internal discoveryLoop, which must never block on
+// another discovery. It could be exported in the future if callers need
+// fire-and-forget semantics on the public API.
+func (c *Client) tryDiscoverNodes(ctx context.Context) error {
+	if ctx.Err() != nil {
+		return ctx.Err()
+	}
+
+	c.discoverMu.Lock()
+	if c.discoverMu.inProgress {
+		c.discoverMu.Unlock()
 		return nil
 	}
-	c.mu.discoveryInProgress = true
-	c.mu.Unlock()
+	// Lock is held — doDiscoverNodes takes ownership and releases it.
+	return c.doDiscoverNodes(ctx)
+}
 
+// doDiscoverNodes performs the discovery work.
+//
+// Called with c.discoverMu held. It sets inProgress = true, releases the
+// lock for I/O, then re-acquires it on completion to clear inProgress,
+// store the result, and wake any waiters.
+func (c *Client) doDiscoverNodes(ctx context.Context) error {
+	c.discoverMu.inProgress = true
+	c.discoverMu.Unlock()
+
+	var discoverErr error
 	defer func() {
-		c.mu.Lock()
-		c.mu.discoveryInProgress = false
-		c.mu.Unlock()
+		c.discoverMu.Lock()
+		c.discoverMu.inProgress = false
+		c.discoverMu.lastErr = discoverErr
+		c.discoverMu.cond.Broadcast()
+		c.discoverMu.Unlock()
 	}()
 
 	discovered, err := c.getNodesInfo(ctx)
 	if err != nil {
 		if dl := loadDebugLogger(); dl != nil {
 			dl.Logf("Error getting nodes info: %s\n", err)
 		}
-		return fmt.Errorf("discovery: get nodes: %w", err)
+		discoverErr = fmt.Errorf("discovery: get nodes: %w", err)
+		return discoverErr
 	}
 
 	c.mu.RLock()
@@ -289,11 +345,13 @@ func (c *Client) DiscoverNodes(ctx context.Context) error {
 
 	if isColdStart {
 		if err := c.nodeDiscoveryAsyncStart(ctx, discovered); err != nil {
-			return err
+			discoverErr = err
+			return discoverErr
 		}
 	} else {
 		if err := c.nodeDiscovery(ctx, discovered); err != nil {
-			return err
+			discoverErr = err
+			return discoverErr
 		}
 	}
 
@@ -1217,7 +1275,7 @@ func (c *Client) discoveryLoop() {
 		switch {
 		case !now.Before(nextNodes):
 			// Full node + shard discovery.
-			c.DiscoverNodes(c.ctx) //nolint:errcheck // errors logged inside
+			c.tryDiscoverNodes(c.ctx) //nolint:errcheck // errors logged inside
 
 			nextNodes = time.Now().Add(c.discoverNodesInterval)
 			nextCat = time.Time{}

opensearchtransport/discovery_internal_test.go

@@ -42,6 +42,7 @@ import (
 	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -2259,3 +2260,198 @@ func TestGetNodesInfoNodesMeta(t *testing.T) {
 		})
 	}
 }
+
+// gatedNodesHandler returns a /_nodes/http handler that signals entered when
+// the request arrives, then blocks until gate is closed before responding with
+// a single data node. The handler also selects on t.Context().Done() so that
+// test cleanup can unblock it.
+func gatedNodesHandler(t *testing.T, entered chan<- struct{}, gate <-chan struct{}) http.HandlerFunc {
+	t.Helper()
+	return func(w http.ResponseWriter, _ *http.Request) {
+		select {
+		case entered <- struct{}{}:
+		default:
+		}
+		select {
+		case <-gate:
+		case <-t.Context().Done():
+			http.Error(w, "test context cancelled", http.StatusServiceUnavailable)
+			return
+		}
+		w.Header().Set("Content-Type", "application/json")
+		fmt.Fprint(w, `{
+			"_nodes":{"total":1,"successful":1,"failed":0},
+			"cluster_name":"test",
+			"nodes":{
+				"n1":{
+					"name":"n1",
+					"roles":["data","ingest"],
+					"http":{"publish_address":"127.0.0.1:9200"}
+				}
+			}
+		}`)
+	}
+}
+
+// gatedErrorNodesHandler returns a /_nodes/http handler that signals entered
+// when the request arrives, then blocks until gate is closed before responding
+// with an HTTP 503 to trigger a discovery error. The handler also selects on
+// t.Context().Done() so that test cleanup can unblock it.
+func gatedErrorNodesHandler(t *testing.T, entered chan<- struct{}, gate <-chan struct{}) http.HandlerFunc {
+	t.Helper()
+	return func(w http.ResponseWriter, _ *http.Request) {
+		select {
+		case entered <- struct{}{}:
+		default:
+		}
+		select {
+		case <-gate:
+		case <-t.Context().Done():
+		}
+		http.Error(w, "unavailable", http.StatusServiceUnavailable)
+	}
+}
+
+// newGatedDiscoverClient creates a transport Client wired to the given
+// handler routes, with discoverMu.cond properly initialised.
+func newGatedDiscoverClient(t *testing.T, routes mockhttp.HandlerMap) *Client {
+	t.Helper()
+	transport := mockhttp.NewTransportFromRoutes(t, routes)
+	u, _ := url.Parse("http://127.0.0.1:9200")
+	tp, err := New(Config{URLs: []*url.URL{u}, Transport: transport})
+	require.NoError(t, err)
+	tp.discoverMu.cond = sync.NewCond(&tp.discoverMu)
+	return tp
+}
+
+func TestDiscoverNodesBlocking(t *testing.T) {
+	entered := make(chan struct{}, 1)
+	gate := make(chan struct{})
+
+	routes := mockhttp.GetDefaultHandlers(t)
+	routes["/_nodes/http"] = gatedNodesHandler(t, entered, gate)
+	tp := newGatedDiscoverClient(t, routes)
+
+	// Goroutine A: start discovery (blocks in handler on gate).
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		tp.DiscoverNodes(t.Context())
+	}()
+
+	// Wait for handler to be entered — discovery is now in-flight.
+	<-entered
+
+	// Goroutine B: should block in DiscoverNodes until A finishes.
+	bDone := make(chan error, 1)
+	go func() {
+		bDone <- tp.DiscoverNodes(t.Context())
+	}()
+
+	// B should not have returned yet (gate still closed, A still blocked).
+	select {
+	case <-bDone:
+		t.Fatal("goroutine B returned before discovery finished")
+	default:
+	}
+
+	// Release A — A finishes, B wakes up.
+	close(gate)
+	wg.Wait()
+
+	err := <-bDone
+	require.NoError(t, err, "goroutine B should succeed after waiting")
+}
+
+func TestDiscoverNodesBlockingPropagatesError(t *testing.T) {
+	entered := make(chan struct{}, 1)
+	gate := make(chan struct{})
+
+	routes := mockhttp.GetDefaultHandlers(t)
+	routes["/_nodes/http"] = gatedErrorNodesHandler(t, entered, gate)
+	tp := newGatedDiscoverClient(t, routes)
+
+	// Goroutine A: start discovery that will fail.
+	aDone := make(chan error, 1)
+	go func() {
+		aDone <- tp.DiscoverNodes(t.Context())
+	}()
+
+	<-entered // A is in handler, discovery in-flight.
+
+	// Goroutine B: waits for A, receives the same error via lastErr.
+	bDone := make(chan error, 1)
+	go func() {
+		bDone <- tp.DiscoverNodes(t.Context())
+	}()
+
+	close(gate)
+
+	errA := <-aDone
+	errB := <-bDone
+
+	require.Error(t, errA, "runner should report discovery error")
+	require.Error(t, errB, "waiter should receive the same error")
+	require.Equal(t, errA, errB)
+}
+
+func TestDiscoverNodesBlockingContextCancel(t *testing.T) {
+	entered := make(chan struct{}, 1)
+	gate := make(chan struct{})
+	defer close(gate) // prevent goroutine leak
+
+	routes := mockhttp.GetDefaultHandlers(t)
+	routes["/_nodes/http"] = gatedNodesHandler(t, entered, gate)
+	tp := newGatedDiscoverClient(t, routes)
+
+	// Goroutine A: start slow discovery.
+	go func() {
+		tp.DiscoverNodes(t.Context())
+	}()
+
+	<-entered // A is in handler, discovery in-flight.
+
+	// B: call DiscoverNodes with an already-cancelled context.
+	ctx, cancel := context.WithCancel(t.Context())
+	cancel()
+
+	err := tp.DiscoverNodes(ctx)
+	require.ErrorIs(t, err, context.Canceled)
+}
+
+func TestTryDiscoverNodesNonBlocking(t *testing.T) {
+	entered := make(chan struct{}, 1)
+	gate := make(chan struct{})
+	defer close(gate) // prevent goroutine leak
+
+	routes := mockhttp.GetDefaultHandlers(t)
+	routes["/_nodes/http"] = gatedNodesHandler(t, entered, gate)
+	tp := newGatedDiscoverClient(t, routes)
+
+	// Start discovery in a goroutine.
+	go func() {
+		tp.DiscoverNodes(t.Context())
+	}()
+
+	<-entered // discovery is in-flight.
+
+	// tryDiscoverNodes should return nil immediately without blocking.
+	err := tp.tryDiscoverNodes(t.Context())
+	require.NoError(t, err)
+}
+
+func TestDiscoverNodesSequential(t *testing.T) {
+	routes := mockhttp.GetDefaultHandlersWithNodes(t, map[string][]string{
+		"node1": {"data", "ingest"},
+	})
+	tp := newGatedDiscoverClient(t, routes)
+
+	// First call succeeds.
+	err := tp.DiscoverNodes(t.Context())
+	require.NoError(t, err)
+
+	// Second call also succeeds (starts a fresh discovery).
+	err = tp.DiscoverNodes(t.Context())
+	require.NoError(t, err)
+}

opensearchtransport/opensearchtransport.go

@@ -472,8 +472,17 @@ type Client struct {
 
 	mu struct {
 		sync.RWMutex
-		connectionPool      ConnectionPool // Used for both single-node and multi-node
-		discoveryInProgress bool           // Prevents concurrent discovery operations
+		connectionPool ConnectionPool // Used for both single-node and multi-node
+	}
+
+	// discoverMu serializes discovery and lets callers block until an
+	// in-flight discovery completes.  Separate from mu so that
+	// discovery waiters never contend with normal request-path readers.
+	discoverMu struct {
+		sync.Mutex
+		cond       *sync.Cond // signaled when inProgress transitions to false
+		lastErr    error      // result of the most recent completed discovery
+		inProgress bool       // true while a discovery cycle is running
 	}
 }
 
@@ -845,6 +854,7 @@ func New(cfg Config) (*Client, error) {
 	}
 
 	client.userAgent = initUserAgent()
+	client.discoverMu.cond = sync.NewCond(&client.discoverMu)
 
 	// Parse discovery feature config from environment variable.
 	// Controls which server calls are made during the discovery cycle.

opensearchtransport/pool_congestion_internal_test.go

@@ -891,7 +891,7 @@ type testDebugLogger struct{}
 func (l *testDebugLogger) Log(_ ...any) error            { return nil }
 func (l *testDebugLogger) Logf(_ string, _ ...any) error { return nil }
 
-// enableTestDebugLogger sets debugLogger to a no-op testDebugLogger exactly
+// enableTestDebugLogger sets the debug logger to a no-op testDebugLogger exactly
 // once for the lifetime of the test process. This avoids data races that
 // arise when individual tests save/restore the package-level global while
 // background goroutines from parallel tests are still reading it.