Restore end-to-end probe in shard routing readiness gates

The suite refactor (f571e4a) rewrote the standalone
TestShardExactRouting_FullPipeline_Integration into a testify suite but
dropped the end-to-end probe search from both readiness gates. The
original fix (7b43d50) replaced structural-only checks with a probe
that issues a routed search and verifies ShardExactMatch=true through
the observer, catching cases where the shard map looks complete but
shardExactCandidates still returns nil (connection Names not converged,
needsCatUpdate still set, etc.).

Both SetupSuite and TestFullPipelineRouting gates now check all three
layers: structural completeness, needsCatUpdate flags, and an actual
probe search confirming the full pipeline works.

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

Author: sean-

opensearchtransport/shard_routing_integration_test.go

@@ -181,12 +181,14 @@ func (s *ShardRoutingSuite) SetupSuite() {
 	require.NoError(t, err)
 	warmResp.Body.Close()
 
-	// --- Discover shard mappings for the new index ---
+	// --- Discover shard mappings and verify end-to-end shard-exact routing ---
 	// DiscoverNodes fetches /_cat/shards and populates the router's shard
-	// catalog. Poll until:
+	// catalog. Poll until all three conditions hold:
 	//   1. The shard map for the test index has all shards populated
-	//   2. No connections have lcNeedsCatUpdate set (which would cause
+	//   2. No connections have needsCatUpdate set (which would cause
 	//      shardExactCandidates to filter them out)
+	//   3. An actual probe search gets ShardExactMatch=true through the
+	//      observer, proving the full pipeline has converged
 	routerPolicy, ok := s.transport.router.(Policy)
 	require.True(t, ok, "router does not implement Policy")
 	s.cache = findRouterCache(routerPolicy)
@@ -218,9 +220,26 @@ func (s *ShardRoutingSuite) SetupSuite() {
 				}
 			}
 		}
-		return true
+
+		// End-to-end probe: issue a routed search and verify ShardExactMatch.
+		// Structural checks above can pass while shardExactCandidates still
+		// returns nil (e.g. connection Names haven't converged into sorted
+		// snapshots). The probe tests the full pipeline in a single assertion.
+		s.obs.reset()
+		probeURL := apiURL([]string{"/", s.index, "_search"}, map[string]string{"routing": "probe-canary"})
+		probeReq, _ := http.NewRequestWithContext(s.ctx, http.MethodPost,
+			probeURL,
+			bytes.NewReader([]byte(`{"query":{"match_all":{}},"size":0}`)))
+		probeReq.Header.Set("Content-Type", "application/json")
+		probeResp, probeErr := s.transport.Perform(probeReq)
+		if probeErr != nil {
+			return false
+		}
+		probeResp.Body.Close()
+		probeEvent := s.obs.lastEvent()
+		return probeEvent != nil && probeEvent.ShardExactMatch
 	}, 30*time.Second, 500*time.Millisecond,
-		"shard catalog for index %q not ready after repeated DiscoverNodes", s.index)
+		"shard-exact routing for index %q not working after repeated DiscoverNodes", s.index)
 	s.obs.reset()
 }
 
@@ -310,13 +329,26 @@ func (s *ShardRoutingSuite) TestDocIDAsDefaultRouting() {
 func (s *ShardRoutingSuite) TestFullPipelineRouting() {
 	t := s.T()
 
-	// Refresh the shard catalog and clear any needsCatUpdate flags that
-	// previous test methods (e.g., TestDocIDAsDefaultRouting) may have set
-	// via transport errors.
+	// Refresh the shard catalog and verify end-to-end shard-exact routing
+	// works before running assertions. Previous test methods (e.g.,
+	// TestDocIDAsDefaultRouting) may have set needsCatUpdate flags via
+	// transport errors, and the resulting DiscoverNodes can replace the
+	// shard map with incomplete data if shards are still settling. Poll
+	// until an actual probe search gets ShardExactMatch=true through the
+	// observer — this tests the full pipeline (shard map populated,
+	// routing_num_shards fetched, connection names converged, no
+	// connections filtered by needsCatUpdate) in a single assertion.
 	require.Eventually(t, func() bool {
 		if err := s.transport.DiscoverNodes(s.ctx); err != nil {
 			return false
 		}
+		// Quick structural check: shard map must have all shards populated.
+		slot := s.cache.getOrCreate(s.index)
+		sm := slot.shardMap.Load()
+		if sm == nil || sm.RoutingNumShards == 0 || len(sm.Shards) < s.numShards {
+			return false
+		}
+		// Check no connections have needsCatUpdate set.
 		s.transport.mu.RLock()
 		pool := s.transport.mu.connectionPool
 		s.transport.mu.RUnlock()
@@ -327,8 +359,22 @@ func (s *ShardRoutingSuite) TestFullPipelineRouting() {
 				}
 			}
 		}
-		return true
-	}, 30*time.Second, 500*time.Millisecond, "connections still have needsCatUpdate after DiscoverNodes")
+		// End-to-end probe: issue a routed search and verify ShardExactMatch.
+		s.obs.reset()
+		probeURL := apiURL([]string{"/", s.index, "_search"}, map[string]string{"routing": "probe-canary"})
+		probeReq, _ := http.NewRequestWithContext(s.ctx, http.MethodPost,
+			probeURL,
+			bytes.NewReader([]byte(`{"query":{"match_all":{}},"size":0}`)))
+		probeReq.Header.Set("Content-Type", "application/json")
+		probeResp, probeErr := s.transport.Perform(probeReq)
+		if probeErr != nil {
+			return false
+		}
+		probeResp.Body.Close()
+		probeEvent := s.obs.lastEvent()
+		return probeEvent != nil && probeEvent.ShardExactMatch
+	}, 30*time.Second, 500*time.Millisecond,
+		"shard-exact routing for index %q not working after repeated DiscoverNodes", s.index)
 	s.obs.reset()
 
 	type groundTruth struct {