fix(test): flaky tests (#3815)

commands_test.go

@@ -7518,8 +7518,14 @@ var _ = Describe("Commands", func() {
 		It("should XRead LastEntry blocks", Label("NonRedisEnterprise"), func() {
 			SkipBeforeRedisVersion(7.4, "doesn't work with older redis stack images")
 			start := time.Now()
+			// Wait for the goroutine to finish before the spec returns so it
+			// can't outlive AfterEach's client.Close() and trip "use of closed
+			// network connection" during the next spec's BeforeEach.
+			done := make(chan struct{})
+			defer func() { Eventually(done, 5*time.Second).Should(BeClosed()) }()
 			go func() {
 				defer GinkgoRecover()
+				defer close(done)
 
 				time.Sleep(100 * time.Millisecond)
 				id, err := client.XAdd(ctx, &redis.XAddArgs{
@@ -9283,37 +9289,46 @@ var _ = Describe("Commands", func() {
 		It("reset all latencies", func() {
 			const key = "latency-monitor-threshold"
 
-			result, err := client.Latency(ctx).Result()
 			// reset all latencies
-			err = client.LatencyReset(ctx).Err()
+			err := client.LatencyReset(ctx).Err()
 			Expect(err).NotTo(HaveOccurred())
 
 			old := client.ConfigGet(ctx, key).Val()
-			client.ConfigSet(ctx, key, "1")
+			// Use a higher threshold (100ms) to avoid capturing normal operations
+			// that could cause flakiness due to timing variations on busy CI runners.
+			client.ConfigSet(ctx, key, "100")
 			defer client.ConfigSet(ctx, key, old[key])
 
 			// get latency after reset
-			result, err = client.Latency(ctx).Result()
+			result, err := client.Latency(ctx).Result()
 			Expect(err).NotTo(HaveOccurred())
 			Expect(len(result)).Should(Equal(0))
 
-			// create a new latency
-			err = client.Do(ctx, "DEBUG", "SLEEP", 0.01).Err()
+			// create a new latency event by sleeping past the threshold
+			err = client.Do(ctx, "DEBUG", "SLEEP", 0.15).Err()
 			Expect(err).NotTo(HaveOccurred())
 
 			// get latency after create a new latency
 			result, err = client.Latency(ctx).Result()
 			Expect(err).NotTo(HaveOccurred())
-			Expect(len(result)).Should(Equal(1))
+			Expect(len(result)).Should(BeNumerically(">=", 1))
+			triggered := result[0].Name
 
 			// reset all latencies again
 			err = client.LatencyReset(ctx).Err()
 			Expect(err).NotTo(HaveOccurred())
 
-			// get latency after reset again
+			// get latency after reset again. Don't assert total length is 0:
+			// other operations on the connection may briefly land in the
+			// latency log on a slow CI runner. Instead verify the specific
+			// event we triggered is gone.
 			result, err = client.Latency(ctx).Result()
 			Expect(err).NotTo(HaveOccurred())
-			Expect(len(result)).Should(Equal(0))
+			for _, event := range result {
+				if event.Name == triggered {
+					Fail("Event " + triggered + " should have been reset")
+				}
+			}
 		})
 
 		It("reset latencies by add event name args", func() {

internal/pool/dial_conn_retry_test.go

@@ -36,7 +36,10 @@ func TestDialConn_HangingDial_RetriesWithPerAttemptTimeout(t *testing.T) {
 			<-ctx.Done()
 			return nil, ctx.Err()
 		},
-		PoolSize:           1,
+		// PoolSize is set high so the pool's tryDial probe (triggered when
+		// dialErrorsNum == PoolSize) does not fire during this test and race
+		// the assertion on `calls`.
+		PoolSize:           100,
 		MaxConcurrentDials: 1,
 		DialTimeout:        dialTimeout,
 		DialerRetries:      retries,
@@ -92,7 +95,7 @@ func TestDialConn_DoesNotExtendEarlierParentDeadline(t *testing.T) {
 			<-ctx.Done()
 			return nil, ctx.Err()
 		},
-		PoolSize:           1,
+		PoolSize:           100,
 		MaxConcurrentDials: 1,
 		DialTimeout:        500 * time.Millisecond,
 		DialerRetries:      1,
@@ -123,7 +126,7 @@ func TestDialConn_ContextCancelStopsFurtherRetries(t *testing.T) {
 			}
 			return nil, errors.New("unexpected extra attempt")
 		},
-		PoolSize:           1,
+		PoolSize:           100,
 		MaxConcurrentDials: 1,
 		DialTimeout:        5 * time.Second,
 		DialerRetries:      5,
@@ -159,7 +162,7 @@ func TestDialConn_DialTimeoutDisabled_DoesNotSetDeadline(t *testing.T) {
 			}
 			return nil, errors.New("dial failed")
 		},
-		PoolSize:           1,
+		PoolSize:           100,
 		MaxConcurrentDials: 1,
 		DialTimeout:        0,
 		DialerRetries:      1,
@@ -184,7 +187,7 @@ func TestDialConn_NoBackoffAfterLastAttempt(t *testing.T) {
 			calls.Add(1)
 			return nil, errors.New("dial failed")
 		},
-		PoolSize:           1,
+		PoolSize:           100,
 		MaxConcurrentDials: 1,
 		DialTimeout:        5 * time.Second,
 		DialerRetries:      1,

osscluster_test.go

@@ -319,9 +319,12 @@ var _ = Describe("ClusterClient", func() {
 			err := client.Set(ctx, "A", "VALUE", 0).Err()
 			Expect(err).NotTo(HaveOccurred())
 
-			v, err := client.Get(ctx, "A").Result()
-			Expect(err).NotTo(HaveOccurred())
-			Expect(v).To(Equal("VALUE"))
+			// With RouteByLatency / RouteRandomly the GET may be served by a
+			// replica that has not yet replicated the SET. Poll until the
+			// value propagates instead of failing on the first attempt.
+			Eventually(func() string {
+				return client.Get(ctx, "A").Val()
+			}, 30*time.Second).Should(Equal("VALUE"))
 		})
 
 		It("should distribute keys", func() {
@@ -757,17 +760,23 @@ var _ = Describe("ClusterClient", func() {
 			err = client.Do(ctx, []byte("GET"), []byte("A")).Err()
 			Expect(err).To(Equal(redis.Nil))
 
-			Eventually(func() error {
-				return client.SwapNodes(ctx, "A")
-			}, 30*time.Second).ShouldNot(HaveOccurred())
-
-			err = client.Do(ctx, "GET", "A").Err()
-			Expect(err).To(HaveOccurred())
-			Expect(err.Error()).To(ContainSubstring("MOVED"))
-
-			err = client.Do(ctx, []byte("GET"), []byte("A")).Err()
-			Expect(err).To(HaveOccurred())
-			Expect(err.Error()).To(ContainSubstring("MOVED"))
+			// Re-swap and re-issue both GETs inside a single Eventually so a
+			// background topology reload that reverts the swap before our
+			// assertions just causes another iteration instead of a flake.
+			Eventually(func() string {
+				if err := client.SwapNodes(ctx, "A"); err != nil {
+					return err.Error()
+				}
+				strErr := client.Do(ctx, "GET", "A").Err()
+				bytesErr := client.Do(ctx, []byte("GET"), []byte("A")).Err()
+				if strErr == nil || bytesErr == nil {
+					return ""
+				}
+				if !strings.Contains(strErr.Error(), "MOVED") {
+					return strErr.Error()
+				}
+				return bytesErr.Error()
+			}, 30*time.Second).Should(ContainSubstring("MOVED"))
 
 			Expect(client.Close()).NotTo(HaveOccurred())
 		})
@@ -3217,8 +3226,12 @@ var _ = Describe("ClusterClient FailingTimeoutSeconds", func() {
 	})
 
 	It("should handle node failing timeout correctly", func() {
+		// Use a larger window than the original 2s. Failing() compares
+		// truncated unix seconds (failing < timeout), so a 1s sleep against
+		// a 2s timeout has < 1s of margin and flips false on slow CI runners
+		// where time.Sleep overshoots.
 		opt := redisClusterOptions()
-		opt.FailingTimeoutSeconds = 2 // Short timeout for testing
+		opt.FailingTimeoutSeconds = 5
 		client = cluster.newClusterClient(ctx, opt)
 
 		// Get a node and mark it as failing
@@ -3235,12 +3248,13 @@ var _ = Describe("ClusterClient FailingTimeoutSeconds", func() {
 		node.MarkAsFailing()
 		Expect(node.Failing()).To(BeTrue())
 
-		// Should still be failing after 1 second (less than timeout)
-		time.Sleep(1 * time.Second)
+		// Should still be failing well before the timeout expires.
+		time.Sleep(2 * time.Second)
 		Expect(node.Failing()).To(BeTrue())
 
-		// Should not be failing after timeout expires
-		time.Sleep(2 * time.Second) // Total 3 seconds > 2 second timeout
+		// Should not be failing after timeout expires (total 7s > 5s timeout,
+		// with > 1s margin on each side).
+		time.Sleep(5 * time.Second)
 		Expect(node.Failing()).To(BeFalse())
 	})
 

pubsub_test.go

@@ -636,6 +636,13 @@ var _ = Describe("PubSub", func() {
 		err = pubsub.ClientSetName(ctx, "my-subscriber")
 		Expect(err).NotTo(HaveOccurred())
 
+		// PubSub.ClientSetName only writes the command; the +OK reply is
+		// drained here. This also synchronizes us with the server so that
+		// the subsequent CLIENT LIST on a different connection observes
+		// the rename (CLIENT LIST may otherwise race the SETNAME packet).
+		_, err = pubsub.Receive(ctx)
+		Expect(err).NotTo(HaveOccurred())
+
 		// Verify the name is set via CLIENT LIST
 		clientList, err := client.ClientList(ctx).Result()
 		Expect(err).NotTo(HaveOccurred())
@@ -646,19 +653,24 @@ var _ = Describe("PubSub", func() {
 		pubsub := client.Subscribe(ctx, "mychannel")
 		defer pubsub.Close()
 
+		// Wait for the SUBSCRIBE confirmation before calling Channel(), so the
+		// server is guaranteed to have registered the subscription before we
+		// Publish. Once Channel() starts its pump goroutine, all subsequent
+		// reads happen there and Subscription frames are filtered out.
+		_, err := pubsub.ReceiveTimeout(ctx, 5*time.Second)
+		Expect(err).NotTo(HaveOccurred())
+
 		ch := pubsub.Channel(
 			redis.WithChannelSize(10),
 			redis.WithChannelHealthCheckInterval(time.Second),
 		)
 
 		text := "test channel message"
-		err := client.Publish(ctx, "mychannel", text).Err()
+		err = client.Publish(ctx, "mychannel", text).Err()
 		Expect(err).NotTo(HaveOccurred())
 
-		time.Sleep(10 * time.Millisecond)
-
 		var msg *redis.Message
-		Eventually(ch).Should(Receive(&msg))
+		Eventually(ch, 5*time.Second).Should(Receive(&msg))
 		Expect(msg.Channel).To(Equal("mychannel"))
 		Expect(msg.Payload).To(Equal(text))
 	})

search_test.go

@@ -15,18 +15,23 @@ import (
 	"github.com/redis/go-redis/v9/helper"
 )
 
+// WaitForIndexing polls FT.INFO until the index has finished both its
+// initial indexing pass and any background cleaning. Cleaning can briefly
+// report 1 right after index creation (e.g. for HNSW vector indexes) while
+// background setup is in progress, so callers that immediately assert on
+// FT.INFO fields would otherwise be flaky.
 func WaitForIndexing(c *redis.Client, index string) {
+	deadline := time.Now().Add(30 * time.Second)
 	for {
 		res, err := c.FTInfo(context.Background(), index).Result()
 		Expect(err).NotTo(HaveOccurred())
-		if c.Options().Protocol == 2 {
-			if res.Indexing == 0 {
-				return
-			}
-			time.Sleep(100 * time.Millisecond)
-		} else {
+		if res.Indexing == 0 && res.Cleaning == 0 {
 			return
 		}
+		if time.Now().After(deadline) {
+			Fail(fmt.Sprintf("WaitForIndexing(%q) timed out: indexing=%d cleaning=%d", index, res.Indexing, res.Cleaning))
+		}
+		time.Sleep(100 * time.Millisecond)
 	}
 }
 
@@ -3469,9 +3474,11 @@ var _ = Describe("RediSearch commands Resp 2", Label("search"), func() {
 	})
 
 	It("should parse FTInfo response with vector fields", Label("search", "ftinfo", "NonRedisEnterprise"), func() {
-		// Create an index with multiple field types including vector
+		// Scope to a prefix so that hashes left over from previous specs in
+		// this Describe (e.g. doc_resp3_*) cannot be matched by this index
+		// and skew NumTerms / NumRecords.
 		val, err := client.FTCreate(ctx, "idx_vector",
-			&redis.FTCreateOptions{},
+			&redis.FTCreateOptions{Prefix: []interface{}{"idx_vector:"}},
 			&redis.FieldSchema{
 				FieldName: "prompt",
 				FieldType: redis.SearchFieldTypeText,