test(adapter): widen raft election budget + retry gRPC on leader churn

PR #602 (leader-stability window) fixed setup-side flake in the redis
tests but left Test_consistency_satisfy_write_after_read_sequence,
Test_consistency_satisfy_write_after_read_for_parallel, and
Test_grpc_transaction flaky on CI. Those tests drive 9,999 (or 1,000×
parallel) raw/txn RPCs against the leader in a loop; if the leader
loses quorum mid-run -- easy under `-race` on a loaded GHA runner --
the first Put/Get after step-down returns "etcd raft engine is not
leader" or "leader not found" and the test fails (with a bonus nil
panic when assert.NoError let a nil response flow into resp.Value).

Root cause of the churn: the test raft config used ElectionTick=10
(100 ms election timeout) with HeartbeatTick=1. etcd/raft's CheckQuorum
only holds ~1 ElectionTick of heartbeat-response history, so any
goroutine-scheduler pause > 100 ms under -race made the leader
conclude quorum was lost and step down.

Fix:

1. Raise testEngineHeartbeatTick / testEngineElectionTick to 5/50
   (50 ms heartbeat, 500-1000 ms randomised election timeout). Still
   well within the etcd/raft recommended 10x ratio, fast enough for
   tests, and generous enough to survive -race scheduling jitter.
   As a bonus, LeaseDuration is now 200 ms (500-300 margin) instead
   of 0, so the lease-read fast path is actually exercised.

2. Extend isTransientNotLeaderErr to also match "leader not found"
   (ErrLeaderNotFound) -- the coordinator emits that variant during
   the re-election window, distinct from "not leader".

3. Add rawPut/rawGet/txnPut/txnGet/txnDeleteEventually wrappers that
   reuse doEventually to retry on transient leader-unavailable errors,
   mirroring rpushEventually/lpushEventually from #596 for the gRPC
   paths.

4. Rewrite the three 9999- / 1000-iteration consistency tests to use
   the *Eventually helpers. This also removes the nil-pointer crash:
   rawGetEventually / txnGetEventually only return a non-nil response.

Verified: all four originally-flaky tests run clean 3 out of 3 with
go test -race -count=1, and the full adapter package passes with
go test -race ./adapter/... -count=1.

Author: claude

adapter/grpc_test.go

@@ -151,24 +151,23 @@ func Test_consistency_satisfy_write_after_read_for_parallel(t *testing.T) {
 	nodes, adders, _ := createNode(t, 3)
 	c := rawKVClient(t, adders)
 
+	// 1000 concurrent clients × 3 RPCs saturates the single raft leader
+	// hard enough to provoke brief quorum checks to fail on CI, so use
+	// the *Eventually helpers that retry transient leader-unavailable
+	// errors. See Test_consistency_satisfy_write_after_read_sequence.
+	ctx := context.Background()
 	wg := sync.WaitGroup{}
 	wg.Add(1000)
 	for i := range 1000 {
 		go func(i int) {
+			defer wg.Done()
 			key := []byte("test-key-parallel" + strconv.Itoa(i))
 			want := []byte(strconv.Itoa(i))
-			_, err := c.RawPut(
-				context.Background(),
-				&pb.RawPutRequest{Key: key, Value: want},
-			)
-			assert.NoError(t, err, "Put RPC failed")
-			_, err = c.RawPut(context.TODO(), &pb.RawPutRequest{Key: key, Value: want})
-			assert.NoError(t, err, "Put RPC failed")
-
-			resp, err := c.RawGet(context.TODO(), &pb.RawGetRequest{Key: key})
-			assert.NoError(t, err, "Get RPC failed")
+			rawPutEventually(t, ctx, c, &pb.RawPutRequest{Key: key, Value: want})
+			rawPutEventually(t, ctx, c, &pb.RawPutRequest{Key: key, Value: want})
+
+			resp := rawGetEventually(t, ctx, c, &pb.RawGetRequest{Key: key})
 			assert.Equal(t, want, resp.Value, "consistency check failed")
-			wg.Done()
 		}(i)
 	}
 	wg.Wait()
@@ -183,20 +182,18 @@ func Test_consistency_satisfy_write_after_read_sequence(t *testing.T) {
 
 	key := []byte("test-key-sequence")
 
+	// Use *Eventually helpers because a 9999-iteration loop across three
+	// t.Parallel adapter tests loads CI enough that the raft leader can
+	// briefly lose quorum and step down mid-run, surfacing as transient
+	// "not leader" / "leader not found" RPC errors. The helpers retry
+	// only those transient errors; any other error still fails the test.
+	ctx := context.Background()
 	for i := range 9999 {
 		want := []byte("sequence" + strconv.Itoa(i))
-		_, err := c.RawPut(
-			context.Background(),
-			&pb.RawPutRequest{Key: key, Value: want},
-		)
-		assert.NoError(t, err, "Put RPC failed")
-
-		_, err = c.RawPut(context.TODO(), &pb.RawPutRequest{Key: key, Value: want})
-		assert.NoError(t, err, "Put RPC failed")
-
-		resp, err := c.RawGet(context.TODO(), &pb.RawGetRequest{Key: key})
-		assert.NoError(t, err, "Get RPC failed")
+		rawPutEventually(t, ctx, c, &pb.RawPutRequest{Key: key, Value: want})
+		rawPutEventually(t, ctx, c, &pb.RawPutRequest{Key: key, Value: want})
 
+		resp := rawGetEventually(t, ctx, c, &pb.RawGetRequest{Key: key})
 		assert.Equal(t, want, resp.Value, "consistency check failed")
 	}
 }
@@ -209,22 +206,18 @@ func Test_grpc_transaction(t *testing.T) {
 
 	key := []byte("test-key-sequence")
 
+	// See Test_consistency_satisfy_write_after_read_sequence for why the
+	// *Eventually helpers are necessary here.
+	ctx := context.Background()
 	for i := range 9999 {
 		want := []byte("sequence" + strconv.Itoa(i))
-		_, err := c.Put(
-			context.Background(),
-			&pb.PutRequest{Key: key, Value: want},
-		)
-		assert.NoError(t, err, "Put RPC failed")
-		resp, err := c.Get(context.TODO(), &pb.GetRequest{Key: key})
-		assert.NoError(t, err, "Get RPC failed")
+		txnPutEventually(t, ctx, c, &pb.PutRequest{Key: key, Value: want})
+		resp := txnGetEventually(t, ctx, c, &pb.GetRequest{Key: key})
 		assert.Equal(t, want, resp.Value, "consistency check failed")
 
-		_, err = c.Delete(context.TODO(), &pb.DeleteRequest{Key: key})
-		assert.NoError(t, err, "Delete RPC failed")
+		txnDeleteEventually(t, ctx, c, &pb.DeleteRequest{Key: key})
 
-		resp, err = c.Get(context.TODO(), &pb.GetRequest{Key: key})
-		assert.NoError(t, err, "Get RPC failed")
+		resp = txnGetEventually(t, ctx, c, &pb.GetRequest{Key: key})
 		assert.Nil(t, resp.Value, "consistency check failed")
 	}
 }

adapter/test_util.go

@@ -27,14 +27,29 @@ import (
 )
 
 const (
-	testEngineTickInterval  = 10 * time.Millisecond
-	testEngineHeartbeatTick = 1
-	testEngineElectionTick  = 10
+	testEngineTickInterval = 10 * time.Millisecond
+	// testEngineHeartbeatTick / testEngineElectionTick preserve etcd/raft's
+	// recommended 10x ratio (ElectionTick = 10 × HeartbeatTick). Previous
+	// values (1 / 10 → 100 ms election timeout) were too tight for `-race`
+	// on CI: CheckQuorum only holds ~100 ms of heartbeat-response history,
+	// so any scheduler pause on a loaded runner drops the leader's view of
+	// quorum, and it steps down with "quorum is not active", bouncing
+	// writes to "etcd raft engine is not leader" / "leader not found" in
+	// the middle of tests like Test_consistency_satisfy_write_after_read_sequence.
+	// 5 / 50 gives a 500 ms election timeout (500-1000 ms randomised) and
+	// a 50 ms heartbeat, absorbing goroutine-scheduler jitter while still
+	// keeping tests fast. Combined with leaseSafetyMargin = 300 ms, this
+	// also yields a non-zero 200 ms LeaseDuration so the lease-read fast
+	// path gets exercised instead of always falling through to ReadIndex.
+	testEngineHeartbeatTick = 5
+	testEngineElectionTick  = 50
 	testEngineMaxSizePerMsg = 1 << 20
 	testEngineMaxInflight   = 256
 
 	// leaderChurnRetryTimeout bounds how long doEventually keeps retrying a
-	// write that fails with "not leader" right after createNode returns.
+	// write that fails with a transient leader-unavailable error. It covers
+	// both startup churn right after createNode returns and mid-test churn
+	// when the leader briefly steps down under CI load.
 	leaderChurnRetryTimeout = 5 * time.Second
 	// leaderChurnRetryInterval is the poll interval between retries.
 	leaderChurnRetryInterval = 50 * time.Millisecond
@@ -519,18 +534,23 @@ func setupNodes(t *testing.T, ctx context.Context, n int, ports []portsAdress) (
 	return nodes, grpcAdders, redisAdders, peers
 }
 
-// isTransientNotLeaderErr reports whether err is a transient "not leader"
-// error that can happen right after createNode returns if the newly elected
-// leader briefly steps down due to a missed heartbeat quorum (common on slow
-// CI runners under -race). Callers should retry the write in that case.
+// isTransientNotLeaderErr reports whether err is a transient
+// leader-unavailable error that can happen right after createNode returns
+// (freshly-elected leader briefly steps down due to a missed heartbeat
+// quorum) or in the middle of a long-running test when CI load causes the
+// leader to miss quorum momentarily.
 //
-// The match is case-insensitive because Redis protocol error bodies and
-// other layers may capitalise the phrase differently (e.g. "ERR Not Leader").
+// Both "not leader" (ErrNotLeader, etcd/raft step errors) and
+// "leader not found" (ErrLeaderNotFound, emitted while the cluster is
+// re-electing) are treated as retryable. The match is case-insensitive
+// because Redis protocol error bodies and other layers may capitalise the
+// phrase differently (e.g. "ERR Not Leader").
 func isTransientNotLeaderErr(err error) bool {
 	if err == nil {
 		return false
 	}
-	return strings.Contains(strings.ToLower(err.Error()), "not leader")
+	s := strings.ToLower(err.Error())
+	return strings.Contains(s, "not leader") || strings.Contains(s, "leader not found")
 }
 
 // doEventually retries do() while it returns a transient "not leader" error,
@@ -568,3 +588,64 @@ func lpushEventually(t *testing.T, ctx context.Context, rdb *redis.Client, key s
 		return rdb.LPush(ctx, key, vals...).Err()
 	})
 }
+
+// rawPutEventually wraps RawKV.RawPut in doEventually so transient leader
+// churn (either at startup or in the middle of a long-running loop) does
+// not fail the test with "not leader" / "leader not found".
+func rawPutEventually(t *testing.T, ctx context.Context, c pb.RawKVClient, req *pb.RawPutRequest) {
+	t.Helper()
+	doEventually(t, func() error {
+		_, err := c.RawPut(ctx, req)
+		return err
+	})
+}
+
+// rawGetEventually wraps RawKV.RawGet in doEventually and returns the
+// response only after a successful (non-"not leader") call.
+func rawGetEventually(t *testing.T, ctx context.Context, c pb.RawKVClient, req *pb.RawGetRequest) *pb.RawGetResponse {
+	t.Helper()
+	var resp *pb.RawGetResponse
+	doEventually(t, func() error {
+		r, err := c.RawGet(ctx, req)
+		if err != nil {
+			return err
+		}
+		resp = r
+		return nil
+	})
+	return resp
+}
+
+// txnPutEventually wraps TransactionalKV.Put in doEventually.
+func txnPutEventually(t *testing.T, ctx context.Context, c pb.TransactionalKVClient, req *pb.PutRequest) {
+	t.Helper()
+	doEventually(t, func() error {
+		_, err := c.Put(ctx, req)
+		return err
+	})
+}
+
+// txnGetEventually wraps TransactionalKV.Get in doEventually and returns the
+// response only after a successful (non-"not leader") call.
+func txnGetEventually(t *testing.T, ctx context.Context, c pb.TransactionalKVClient, req *pb.GetRequest) *pb.GetResponse {
+	t.Helper()
+	var resp *pb.GetResponse
+	doEventually(t, func() error {
+		r, err := c.Get(ctx, req)
+		if err != nil {
+			return err
+		}
+		resp = r
+		return nil
+	})
+	return resp
+}
+
+// txnDeleteEventually wraps TransactionalKV.Delete in doEventually.
+func txnDeleteEventually(t *testing.T, ctx context.Context, c pb.TransactionalKVClient, req *pb.DeleteRequest) {
+	t.Helper()
+	doEventually(t, func() error {
+		_, err := c.Delete(ctx, req)
+		return err
+	})
+}