test(redis): fix flaky TestRedis_LuaRPopLPushBullMQLikeLists (#596)

## Summary

- `TestRedis_LuaRPopLPushBullMQLikeLists` flaked on CI with `etcd raft
engine is not leader` on the first `RPush` right after `createNode(t,
3)` returned. The freshly-elected raft leader briefly stepped down
("stepped down to follower since quorum is not active") before the
test's first write reached it, so the write raced a re-election.
- Root cause is benign startup leader churn on slow GitHub Actions
runners under `-race` — not a product bug. `waitForRaftReadiness` only
confirms that a leader exists at one instant, which is not enough to
guarantee stability for the very next write.
- Add small `rpushEventually` / `lpushEventually` helpers (backed by a
shared `doEventually` that retries only on `strings.Contains(err, "not
leader")` for 5s at 50ms intervals) and use them for the first two
writes in the flaky test. `waitForRaftReadiness` is intentionally left
untouched because other tests rely on its exact semantics.

Failing CI run:
https://github.com/bootjp/elastickv/actions/runs/24842581631/job/72720509753

## Test plan

- [x] `go test -race -run TestRedis_LuaRPopLPushBullMQLikeLists
./adapter/ -count=5` — 5/5 pass locally.
- [x] `go test -race ./adapter/... -timeout 900s` — only unrelated
pre-existing flake `Test_grpc_transaction` fires (nil-pointer panic in
teardown); explicitly out of scope per the fix request.
- [x] `golangci-lint run ./adapter/...` — 0 issues.
- [ ] Re-run CI job linked above to confirm the flake is gone.

Author: bootjp

adapter/redis_lua_compat_test.go

@@ -121,8 +121,12 @@ func TestRedis_LuaRPopLPushBullMQLikeLists(t *testing.T) {
 	rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
 	defer func() { _ = rdb.Close() }()
 
-	require.NoError(t, rdb.RPush(ctx, "bull:test:wait", "job-1", "job-2", "job-3").Err())
-	require.NoError(t, rdb.LPush(ctx, "bull:test:active", "job-0").Err())
+	// Raft leadership can briefly churn right after createNode returns — the
+	// freshly-elected leader can step down if the initial heartbeats miss
+	// quorum on a slow CI runner (see waitForRaftReadiness). Retry the first
+	// writes so they ride out that re-election instead of failing the test.
+	rpushEventually(t, ctx, rdb, "bull:test:wait", "job-1", "job-2", "job-3")
+	lpushEventually(t, ctx, rdb, "bull:test:active", "job-0")
 
 	result, err := rdb.Eval(ctx, `
 local moved = redis.call("RPOPLPUSH", KEYS[1], KEYS[2])

adapter/test_util.go

@@ -5,6 +5,7 @@ import (
 	"log"
 	"net"
 	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -18,6 +19,7 @@ import (
 	pb "github.com/bootjp/elastickv/proto"
 	"github.com/bootjp/elastickv/store"
 	"github.com/cockroachdb/errors"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sys/unix"
@@ -30,6 +32,12 @@ const (
 	testEngineElectionTick  = 10
 	testEngineMaxSizePerMsg = 1 << 20
 	testEngineMaxInflight   = 256
+
+	// leaderChurnRetryTimeout bounds how long doEventually keeps retrying a
+	// write that fails with "not leader" right after createNode returns.
+	leaderChurnRetryTimeout = 5 * time.Second
+	// leaderChurnRetryInterval is the poll interval between retries.
+	leaderChurnRetryInterval = 50 * time.Millisecond
 )
 
 func newTestFactory() raftengine.Factory {
@@ -442,3 +450,53 @@ 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.
+//
+// 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")
+}
+
+// doEventually retries do() while it returns a transient "not leader" error,
+// giving the cluster a few seconds to re-settle leadership after startup.
+// Non-"not leader" errors fail the test immediately.
+//
+// Note: we use assert.Eventually (not require.Eventually) so we can capture
+// the last observed error in lastErr and surface it via require.NoError after
+// the loop. Otherwise a timeout would only report "condition never met in 5s"
+// and swallow the underlying error.
+func doEventually(t *testing.T, do func() error) {
+	t.Helper()
+	var lastErr error
+	_ = assert.Eventually(t, func() bool {
+		lastErr = do()
+		return lastErr == nil || !isTransientNotLeaderErr(lastErr)
+	}, leaderChurnRetryTimeout, leaderChurnRetryInterval)
+	require.NoError(t, lastErr)
+}
+
+// rpushEventually wraps RPUSH in doEventually so transient leader churn
+// immediately after createNode doesn't fail the test.
+func rpushEventually(t *testing.T, ctx context.Context, rdb *redis.Client, key string, vals ...any) {
+	t.Helper()
+	doEventually(t, func() error {
+		return rdb.RPush(ctx, key, vals...).Err()
+	})
+}
+
+// lpushEventually wraps LPUSH in doEventually so transient leader churn
+// immediately after createNode doesn't fail the test.
+func lpushEventually(t *testing.T, ctx context.Context, rdb *redis.Client, key string, vals ...any) {
+	t.Helper()
+	doEventually(t, func() error {
+		return rdb.LPush(ctx, key, vals...).Err()
+	})
+}