go-redis/autopipeline_test.go

// Code consolidated from per-topic autopipeline test files.
package redis_test

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"math/rand"
	"net"
	"os"
	"runtime"
	"strconv"
	"strings"
	"sync"
	"sync/atomic"
	"testing"
	"time"

	. "github.com/bsm/ginkgo/v2"
	. "github.com/bsm/gomega"

	"github.com/redis/go-redis/v9"
	"github.com/redis/go-redis/v9/internal/hashtag"
	"github.com/redis/go-redis/v9/internal/routing"
)

// apTestAddr is the redis address for the plain go tests in this file:
// REDIS_PORT if set, else the stack default :6379. The suite-level redisAddr
// variable is NOT safe here — it defaults to :6380 and is corrected only by
// the ginkgo BeforeSuite, which plain go tests may run without.
func apTestAddr() string {
	if p := os.Getenv("REDIS_PORT"); p != "" {
		return ":" + p
	}
	return ":6379"
}

// apRELoadOnce guards a lazy, one-time load of the Redis Enterprise connection
// for the plain go tests in this file. redisOptions() resolves reConn in the
// ginkgo BeforeSuite, but plain TestXxx funcs may run before it, so
// apTestOptions() loads it on first use instead.
var apRELoadOnce sync.Once

// apTestOptions returns the base options for the plain go tests: the resolved
// Redis Enterprise database (address, credentials, TLS) when running against an
// external RE cluster (RE_CLUSTER=true and REDIS_ENDPOINTS_CONFIG_PATH set),
// otherwise Addr: apTestAddr(). Unlike redisOptions() it does not depend on
// BeforeSuite, so it is safe to call from plain TestXxx funcs. Callers that need
// extra options mutate the returned struct before NewClient.
func apTestOptions() *redis.Options {
	apRELoadOnce.Do(func() {
		if reConn != nil {
			return
		}
		if re, _ := strconv.ParseBool(os.Getenv("RE_CLUSTER")); re && os.Getenv("REDIS_ENDPOINTS_CONFIG_PATH") != "" {
			reConn, _ = loadREConnection()
		}
	})
	opt := &redis.Options{Addr: apTestAddr()}
	applyREConnection(opt) // no-op when reConn is nil (non-RE runs)
	return opt
}

// apBaselineRTT returns a conservative baseline round-trip time to the test
// server so latency-bound assertions can scale to the endpoint (localhost vs a
// remote Redis Enterprise database) instead of assuming a fixed local bound. It
// warms the connection, then returns the minimum of a few PINGs (least noisy);
// on error it returns 0 and callers fall back to their fixed floor.
func apBaselineRTT(ctx context.Context, c *redis.Client) time.Duration {
	_ = c.Ping(ctx).Err() // warm dial + handshake
	best := time.Duration(0)
	for i := 0; i < 5; i++ {
		start := time.Now()
		if err := c.Ping(ctx).Err(); err != nil {
			return 0
		}
		if d := time.Since(start); best == 0 || d < best {
			best = d
		}
	}
	return best
}

// apLatencyBudget scales a per-round-trip time budget to the measured RTT, with
// a fixed floor so localhost (near-zero RTT) still enforces a tight ceiling.
// roundTrips is the number of sequential server round trips the timed section
// makes; the 2x factor plus 20ms leaves headroom for scheduling/jitter without
// masking a real per-command coalescing regression.
func apLatencyBudget(rtt time.Duration, roundTrips int, floor time.Duration) time.Duration {
	scaled := time.Duration(roundTrips)*rtt*2 + 20*time.Millisecond
	if scaled < floor {
		return floor
	}
	return scaled
}

// ===== from autopipeline_accessor_test.go =====
// TestAutoPipelineSecondaryAccessors guards a regression where the deferred
// await() was injected only on the canonical Val()/Result() of each command
// type, leaving secondary value-accessors (Int, Bytes, Bool, Float, the generic
// Cmd type, SliceCmd.Val, map getters, ...) reading cmd.val before the batch
// executed. Each accessor below must block until its command runs and return the
// real value, not the zero value. Run under -race to catch the missing barrier.
func TestAutoPipelineSecondaryAccessors(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)
	// The deferred face is where secondary accessors must block until the
	// batch executes (the blocking face waits inside the call itself, so it
	// could never catch a missing await barrier).
	ap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const G = 50
	var wg sync.WaitGroup
	var bad int64
	fail := func(cond bool) {
		if cond {
			atomic.AddInt64(&bad, 1)
		}
	}

	wg.Add(G)
	for g := 0; g < G; g++ {
		go func(id int) {
			defer wg.Done()
			ik := fmt.Sprintf("acc:int:%d", id)
			sk := fmt.Sprintf("acc:str:%d", id)
			hk := fmt.Sprintf("acc:hash:%d", id)

			// Seed values via the autopipeline (these return *StatusCmd/*IntCmd).
			ap.Set(ctx, sk, "12345", 0)
			ap.Set(ctx, ik, "0", 0)
			ap.HSet(ctx, hk, "f1", "v1", "f2", "v2")

			// StringCmd secondary converters — previously missing await().
			n, err := ap.Get(ctx, sk).Int()
			fail(err != nil || n != 12345)
			i64, err := ap.Get(ctx, sk).Int64()
			fail(err != nil || i64 != 12345)
			b, err := ap.Get(ctx, sk).Bytes()
			fail(err != nil || string(b) != "12345")

			// IntCmd path + IntCmd.Uint64 (previously missing await()).
			incr := ap.Incr(ctx, ik)
			u, err := incr.Uint64()
			fail(err != nil || u != 1)

			// Generic Cmd via Do() — the whole *Cmd type was missing await().
			doCmd := ap.Do(ctx, "GET", sk)
			v, err := doCmd.Int()
			fail(err != nil || v != 12345)
			txt, err := ap.Do(ctx, "GET", sk).Text()
			fail(err != nil || txt != "12345")

			// SliceCmd.Val()/Result() — previously missing await().
			mget := ap.MGet(ctx, sk, ik)
			vals := mget.Val()
			fail(len(vals) != 2 || vals[0] != "12345")

			// MapStringStringCmd via HGetAll (Val()/Result() are covered, but
			// exercise it as the typical map path).
			m, err := ap.HGetAll(ctx, hk).Result()
			fail(err != nil || m["f1"] != "v1" || m["f2"] != "v2")
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("secondary accessors: %d wrong/zero reads (await regression)", bad)
	}
}

// ===== from autopipeline_backpressure_test.go =====
// TestAutoPipelineWindowedNoDeadlock submits a large window of commands without
// waiting between them (the async producer pattern), then reads them all. It must
// not deadlock and every future must resolve with its correct result — exercises
// the enqueue/permit/flush path under a producer that outruns per-command waits.
func TestAutoPipelineWindowedNoDeadlock(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.PoolSize = 20
	client := redis.NewClient(opt)
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 300, MaxConcurrentBatches: 8, Unordered: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const n = 5000
	futs := make([]redis.AutoFuture, n)
	for i := 0; i < n; i++ {
		futs[i] = ap.Submit(ctx, redis.NewCmd(ctx, "set", fmt.Sprintf("bp:%d", i), i))
	}
	for i := range futs {
		if err := futs[i].Wait(); err != nil {
			t.Fatalf("future %d: %v", i, err)
		}
	}
	// spot-check a few landed correctly
	for _, i := range []int{0, n / 2, n - 1} {
		if v, _ := client.Get(ctx, fmt.Sprintf("bp:%d", i)).Result(); v != fmt.Sprintf("%d", i) {
			t.Fatalf("bp:%d = %q, want %d", i, v, i)
		}
	}
	for i := 0; i < n; i++ {
		client.Del(ctx, fmt.Sprintf("bp:%d", i))
	}
}

// TestAutoPipelineSoak drives continuous autopipelined traffic from many
// goroutines for a few seconds and checks that goroutine and connection counts
// return to a stable baseline afterward (no slow leak) and results stay correct.
// Skipped under -short; bump `dur` for a longer soak.
// settleGoroutines GCs and polls runtime.NumGoroutine until it drops to
// limit or a deadline passes, returning the last observed count. Replaces
// fixed post-GC sleeps, which false-positive on slow CI.
func settleGoroutines(limit int) int {
	deadline := time.Now().Add(5 * time.Second)
	got := 0
	for {
		runtime.GC()
		got = runtime.NumGoroutine()
		if got <= limit || time.Now().After(deadline) {
			return got
		}
		time.Sleep(50 * time.Millisecond)
	}
}

func TestAutoPipelineSoak(t *testing.T) {
	if testing.Short() {
		t.Skip("soak skipped in -short")
	}
	ctx := context.Background()
	opt := apTestOptions()
	opt.PoolSize = 20
	client := redis.NewClient(opt)
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 200, MaxConcurrentBatches: 16, Unordered: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	runtime.GC()
	time.Sleep(100 * time.Millisecond)
	baseGoroutines := runtime.NumGoroutine()

	const dur = 3 * time.Second
	deadline := time.Now().Add(dur)
	var wg sync.WaitGroup
	var mismatch int64
	var mu sync.Mutex
	for g := 0; g < 32; g++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			i := 0
			for time.Now().Before(deadline) {
				key := fmt.Sprintf("soak:%d:%d", id, i%16)
				val := fmt.Sprintf("%d", i)
				setF := ap.Set(ctx, key, val, 0)
				if err := setF.Err(); err != nil {
					mu.Lock()
					mismatch++
					mu.Unlock()
				}
				i++
			}
		}(g)
	}
	wg.Wait()

	if mismatch != 0 {
		t.Fatalf("%d errors during soak", mismatch)
	}
	got := settleGoroutines(baseGoroutines + 16)
	if got > baseGoroutines+16 {
		t.Fatalf("goroutine leak: baseline=%d after soak=%d", baseGoroutines, got)
	}
}

// ===== from autopipeline_blocking_test.go =====
var _ = Describe("AutoPipeline Blocking Commands", func() {
	ctx := context.Background()
	var client *redis.Client
	var ap *redis.AutoPipeliner

	BeforeEach(func() {
		client = redis.NewClient(redisOptions())
		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())

		var err error
		ap, err = client.AutoPipeline()
		Expect(err).NotTo(HaveOccurred())
	})

	AfterEach(func() {
		if ap != nil {
			Expect(ap.Close()).NotTo(HaveOccurred())
		}
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should not autopipeline blocking commands", func() {
		// Push a value to the list
		Expect(client.RPush(ctx, "list", "value").Err()).NotTo(HaveOccurred())

		// BLPOP should execute immediately without autopipelining.
		// ap.Do returns a generic *redis.Cmd; use its typed accessors.
		rtt := apBaselineRTT(ctx, client)
		start := time.Now()
		result := ap.Do(ctx, "BLPOP", "list", "1")
		val, err := result.StringSlice()
		elapsed := time.Since(start)

		Expect(err).NotTo(HaveOccurred())
		Expect(val).To(Equal([]string{"list", "value"}))
		// Value is available, so this is ~one round trip; bound scales to the
		// measured RTT (100ms floor for localhost).
		Expect(elapsed).To(BeNumerically("<", apLatencyBudget(rtt, 1, 100*time.Millisecond)))
	})

	It("should mix blocking and non-blocking commands", func() {
		// Push values
		Expect(client.RPush(ctx, "list3", "a", "b", "c").Err()).NotTo(HaveOccurred())
		Expect(client.Set(ctx, "key1", "value1", 0).Err()).NotTo(HaveOccurred())

		// Mix blocking and non-blocking commands
		blpopCmd := ap.Do(ctx, "BLPOP", "list3", "1")
		getCmd := ap.Do(ctx, "GET", "key1")
		brpopCmd := ap.Do(ctx, "BRPOP", "list3", "1")

		// Get results — ap.Do returns generic *redis.Cmd; use typed accessors.
		blpopVal, err := blpopCmd.StringSlice()
		Expect(err).NotTo(HaveOccurred())
		Expect(blpopVal).To(Equal([]string{"list3", "a"}))

		getVal, err := getCmd.Text()
		Expect(err).NotTo(HaveOccurred())
		Expect(getVal).To(Equal("value1"))

		brpopVal, err := brpopCmd.StringSlice()
		Expect(err).NotTo(HaveOccurred())
		Expect(brpopVal).To(Equal([]string{"list3", "c"}))
	})
})

// ===== from autopipeline_buffer_test.go =====
// Autopipeline + zero-copy GetToBuffer/SetFromBuffer, standalone, concurrent.
func TestAPZeroCopyBuffer(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{MaxBatchSize: 100, MaxConcurrentBatches: 30, Unordered: true}
	c := redis.NewClient(opt)
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const g = 100
	const perG = 30
	var wg sync.WaitGroup
	var bad int64
	var mu sync.Mutex
	var errs []string
	wg.Add(g)
	for x := 0; x < g; x++ {
		go func(id int) {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				key := fmt.Sprintf("buf:%d:%d", id, i)
				val := fmt.Sprintf("payload-%d-%d", id, i)
				// write from buffer via autopipeline
				if err := ap.SetFromBuffer(ctx, key, []byte(val)).Err(); err != nil {
					atomic.AddInt64(&bad, 1)
					mu.Lock()
					if len(errs) < 10 {
						errs = append(errs, "set:"+err.Error())
					}
					mu.Unlock()
					continue
				}
				// read into caller buffer via autopipeline
				rbuf := make([]byte, len(val))
				cmd := ap.GetToBuffer(ctx, key, rbuf)
				n, err := cmd.Result()
				if err != nil {
					atomic.AddInt64(&bad, 1)
					mu.Lock()
					if len(errs) < 10 {
						errs = append(errs, "get:"+err.Error())
					}
					mu.Unlock()
					continue
				}
				got := string(cmd.Bytes())
				if n != len(val) || got != val {
					atomic.AddInt64(&bad, 1)
					mu.Lock()
					if len(errs) < 10 {
						errs = append(errs, fmt.Sprintf("key %s n=%d got=%q want=%q", key, n, got, val))
					}
					mu.Unlock()
				}
			}
		}(x)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("zero-copy buffer via autopipeline failed: %d cases; sample=%v", bad, errs)
	}
}

// ===== from autopipeline_close_race_test.go =====
// TestAutoPipelineCloseRace stresses concurrent command submission racing with
// Close(). The hazard: a submit that reads closed==false, then stalls, then
// enqueues onto a shard whose flusher has already drained and exited during
// shutdown — leaving the command's completion signal never fired, so the
// caller's blocking accessor hangs forever. Every submitted command must
// resolve (success or ErrClosed), never block past the deadline.
//
// Best-effort: this is a stress test, not a deterministic reproducer. The exact
// interleaving (a submit preempted between the closed-check and the enqueue for
// the whole duration of Close) is narrow and may not trigger every run; the fix
// (re-checking closed under the shard lock) is justified by code analysis. The
// test guards against regressions of the obvious form and runs many iterations
// to widen the window.
func TestAutoPipelineCloseRace(t *testing.T) {
	for iter := 0; iter < 50; iter++ {
		ctx := context.Background()
		c := redis.NewClient(apTestOptions())
		ap, err := c.AutoPipeline()
		if err != nil {
			t.Fatal(err)
		}

		const G = 64
		var wg sync.WaitGroup
		wg.Add(G)
		start := make(chan struct{})
		for g := 0; g < G; g++ {
			go func() {
				defer wg.Done()
				<-start
				for i := 0; i < 20; i++ {
					// .Err() blocks until the command resolves. If the command
					// is lost (never signalled), this never returns and the
					// watchdog below fires.
					_ = ap.Set(ctx, "closerace", i, 0).Err()
				}
			}()
		}

		close(start)
		// Close concurrently with in-flight submits — the race window.
		time.Sleep(time.Millisecond)
		_ = ap.Close()

		// Watchdog: all submitters must finish promptly. A hung await() means a
		// lost command (the bug this guards).
		done := make(chan struct{})
		go func() { wg.Wait(); close(done) }()
		select {
		case <-done:
		case <-time.After(5 * time.Second):
			t.Fatalf("iter %d: submitters hung after Close — a command was lost", iter)
		}
		c.Close()
	}
}

// ===== from autopipeline_close_wiring_test.go =====
// TestClientCloseClosesAutoPipeliner verifies that closing the client also
// closes the shared AutoPipeliner, so its background flusher goroutines don't
// outlive the client. IsClosed() is the direct signal — checking it (rather than
// a downstream command error, which could just reflect the closed pool) proves
// Client.Close stopped the pipeliner itself. Without the wiring in Client.Close
// the autopipeliner would report open after the client is gone.
func TestClientCloseClosesAutoPipeliner(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	c.FlushDB(ctx)

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if err := ap.Set(ctx, "cw", "1", 0).Err(); err != nil {
		t.Fatalf("set via autopipeline before close: %v", err)
	}
	if ap.IsClosed() {
		t.Fatal("autopipeliner reported closed before client.Close")
	}

	if err := c.Close(); err != nil {
		t.Fatalf("client close: %v", err)
	}
	if !ap.IsClosed() {
		t.Fatal("client.Close did not close the shared autopipeliner")
	}

	// A second client Close must not panic; like the base client it reports the
	// already-closed error (the autopipeliner part is idempotent and nil here).
	if err := c.Close(); err != nil && err != redis.ErrClosed {
		t.Fatalf("second client close: unexpected error %v", err)
	}
}

// ===== from autopipeline_cmdable_test.go =====
var _ = Describe("AutoPipeline Cmdable Interface", func() {
	ctx := context.Background()
	var client *redis.Client
	var ap *redis.AutoPipeliner

	BeforeEach(func() {
		client = redis.NewClient(redisOptions())
		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())

		var err error
		ap, err = client.AutoPipeline()
		Expect(err).NotTo(HaveOccurred())
	})

	AfterEach(func() {
		if ap != nil {
			Expect(ap.Close()).NotTo(HaveOccurred())
		}
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should support string commands", func() {
		// Use autopipeline like a regular client
		setCmd := ap.Set(ctx, "key1", "value1", 0)
		getCmd := ap.Get(ctx, "key1")
		incrCmd := ap.Incr(ctx, "counter")
		decrCmd := ap.Decr(ctx, "counter")

		// Get results
		Expect(setCmd.Err()).NotTo(HaveOccurred())
		Expect(setCmd.Val()).To(Equal("OK"))

		val, err := getCmd.Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(val).To(Equal("value1"))

		Expect(incrCmd.Val()).To(Equal(int64(1)))
		Expect(decrCmd.Val()).To(Equal(int64(0)))
	})

	It("should support hash commands", func() {
		// Use hash commands
		hsetCmd := ap.HSet(ctx, "hash1", "field1", "value1", "field2", "value2")
		hgetCmd := ap.HGet(ctx, "hash1", "field1")
		hgetallCmd := ap.HGetAll(ctx, "hash1")

		// Get results
		Expect(hsetCmd.Val()).To(Equal(int64(2)))
		Expect(hgetCmd.Val()).To(Equal("value1"))
		Expect(hgetallCmd.Val()).To(Equal(map[string]string{
			"field1": "value1",
			"field2": "value2",
		}))
	})

	It("should support list commands", func() {
		// Use list commands
		rpushCmd := ap.RPush(ctx, "list1", "a", "b", "c")
		lrangeCmd := ap.LRange(ctx, "list1", 0, -1)
		lpopCmd := ap.LPop(ctx, "list1")

		// Get results
		Expect(rpushCmd.Val()).To(Equal(int64(3)))
		Expect(lrangeCmd.Val()).To(Equal([]string{"a", "b", "c"}))
		Expect(lpopCmd.Val()).To(Equal("a"))
	})

	It("should support set commands", func() {
		// Use set commands
		saddCmd := ap.SAdd(ctx, "set1", "member1", "member2", "member3")
		smembersCmd := ap.SMembers(ctx, "set1")
		sismemberCmd := ap.SIsMember(ctx, "set1", "member1")

		// Get results
		Expect(saddCmd.Val()).To(Equal(int64(3)))
		Expect(smembersCmd.Val()).To(ConsistOf("member1", "member2", "member3"))
		Expect(sismemberCmd.Val()).To(BeTrue())
	})

	It("should support sorted set commands", func() {
		// Use sorted set commands
		zaddCmd := ap.ZAdd(
			ctx, "zset1",
			redis.Z{Score: 1, Member: "one"},
			redis.Z{Score: 2, Member: "two"},
			redis.Z{Score: 3, Member: "three"},
		)
		zrangeCmd := ap.ZRange(ctx, "zset1", 0, -1)
		zscoreCmd := ap.ZScore(ctx, "zset1", "two")

		// Get results
		Expect(zaddCmd.Val()).To(Equal(int64(3)))
		Expect(zrangeCmd.Val()).To(Equal([]string{"one", "two", "three"}))
		Expect(zscoreCmd.Val()).To(Equal(float64(2)))
	})

	It("should support generic commands", func() {
		// Set some keys
		ap.Set(ctx, "key1", "value1", 0)
		ap.Set(ctx, "key2", "value2", 0)
		ap.Set(ctx, "key3", "value3", 0)

		// Use generic commands
		existsCmd := ap.Exists(ctx, "key1", "key2", "key3")
		delCmd := ap.Del(ctx, "key1")
		ttlCmd := ap.TTL(ctx, "key2")

		// Get results
		Expect(existsCmd.Val()).To(Equal(int64(3)))
		Expect(delCmd.Val()).To(Equal(int64(1)))
		Expect(ttlCmd.Val()).To(Equal(time.Duration(-1))) // No expiration
	})

	It("should support Do method for custom commands", func() {
		// Use Do for custom commands
		setCmd := ap.Do(ctx, "SET", "custom_key", "custom_value")
		getCmd := ap.Do(ctx, "GET", "custom_key")

		// Get results
		setVal, err := setCmd.Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(setVal).To(Equal("OK"))

		getVal, err := getCmd.Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(getVal).To(Equal("custom_value"))
	})

	It("should support Pipeline method", func() {
		// Get a traditional pipeline from autopipeliner
		pipe := ap.Pipeline()
		Expect(pipe).NotTo(BeNil())

		// Use the pipeline
		pipe.Set(ctx, "pipe_key", "pipe_value", 0)
		pipe.Get(ctx, "pipe_key")

		cmds, err := pipe.Exec(ctx)
		Expect(err).NotTo(HaveOccurred())
		Expect(cmds).To(HaveLen(2))
	})

	It("should support Pipelined method", func() {
		// Use Pipelined for convenience
		cmds, err := ap.Pipelined(ctx, func(pipe redis.Pipeliner) error {
			pipe.Set(ctx, "pipelined_key", "pipelined_value", 0)
			pipe.Get(ctx, "pipelined_key")
			return nil
		})

		Expect(err).NotTo(HaveOccurred())
		Expect(cmds).To(HaveLen(2))
		Expect(cmds[0].(*redis.StatusCmd).Val()).To(Equal("OK"))
		Expect(cmds[1].(*redis.StringCmd).Val()).To(Equal("pipelined_value"))
	})

	It("should mix autopipelined and direct commands", func() {
		// Use autopipeline commands
		ap.Set(ctx, "ap_key1", "ap_value1", 0)
		ap.Set(ctx, "ap_key2", "ap_value2", 0)

		// Use traditional pipeline
		pipe := ap.Pipeline()
		pipe.Set(ctx, "pipe_key1", "pipe_value1", 0)
		pipe.Set(ctx, "pipe_key2", "pipe_value2", 0)
		_, err := pipe.Exec(ctx)
		Expect(err).NotTo(HaveOccurred())

		// Verify all keys exist
		val1, _ := ap.Get(ctx, "ap_key1").Result()
		val2, _ := ap.Get(ctx, "ap_key2").Result()
		val3, _ := ap.Get(ctx, "pipe_key1").Result()
		val4, _ := ap.Get(ctx, "pipe_key2").Result()

		Expect(val1).To(Equal("ap_value1"))
		Expect(val2).To(Equal("ap_value2"))
		Expect(val3).To(Equal("pipe_value1"))
		Expect(val4).To(Equal("pipe_value2"))
	})
})

// ===== from autopipeline_config_test.go =====
// TestAutoPipelineOptionsNotMutated guards a regression where the autopipeliner
// filled zero-value defaults (MaxBatchSize, MaxConcurrentBatches) directly into
// the caller's *AutoPipelineOptions. A config shared across clients (or inspected
// later by the caller) must not be mutated by creating an AutoPipeliner.
func TestAutoPipelineOptionsNotMutated(t *testing.T) {
	cfg := &redis.AutoPipelineOptions{} // all zero -> defaults applied internally

	c := redis.NewClient(&redis.Options{Addr: apTestAddr(), AutoPipelineOptions: cfg})
	defer c.Close()
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	if cfg.MaxBatchSize != 0 {
		t.Fatalf("caller config MaxBatchSize mutated to %d (want 0)", cfg.MaxBatchSize)
	}
	if cfg.MaxConcurrentBatches != 0 {
		t.Fatalf("caller config MaxConcurrentBatches mutated to %d (want 0)", cfg.MaxConcurrentBatches)
	}

	// The same config reused for a second client must still be pristine and
	// usable (defaults applied to an internal copy, not the shared struct).
	c2 := redis.NewClient(&redis.Options{Addr: apTestAddr(), AutoPipelineOptions: cfg})
	defer c2.Close()
	ap2, err := c2.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap2.Close()
	if cfg.MaxBatchSize != 0 || cfg.MaxConcurrentBatches != 0 {
		t.Fatal("shared config mutated after second client created an AutoPipeliner")
	}
}

// ===== from autopipeline_conn_state_test.go =====
// TestAutoPipelineBlockingCommandIsolation verifies a blocking command (detected
// via readTimeout at the submit gate, autopipeline.go:679) runs outside the
// pipeline and does NOT corrupt batched commands interleaved around it: the
// batched SET/GET before and after the BLPOP all return their correct replies.
func TestAutoPipelineBlockingCommandIsolation(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	client.Del(ctx, "cs:list", "cs:a", "cs:b")
	if err := client.RPush(ctx, "cs:list", "item").Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline() // blocking face
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	if v := ap.Set(ctx, "cs:a", "1", 0).Val(); v != "OK" {
		t.Fatalf("set a before blocking: %q", v)
	}
	// Blocking command interleaved with batched work.
	blp, err := ap.BLPop(ctx, time.Second, "cs:list").Result()
	if err != nil {
		t.Fatalf("blpop: %v", err)
	}
	if len(blp) != 2 || blp[1] != "item" {
		t.Fatalf("blpop = %v, want [cs:list item]", blp)
	}
	// Batched work after a blocking command still works and stays uncorrupted.
	if v := ap.Set(ctx, "cs:b", "2", 0).Val(); v != "OK" {
		t.Fatalf("set b after blocking: %q", v)
	}
	if v := ap.Get(ctx, "cs:a").Val(); v != "1" {
		t.Fatalf("get a after blocking: %q, want 1", v)
	}
	if v := ap.Get(ctx, "cs:b").Val(); v != "2" {
		t.Fatalf("get b after blocking: %q, want 2", v)
	}
	client.Del(ctx, "cs:a", "cs:b")
}

// TestAutoPipelineDoubleClose verifies Close is idempotent — a second Close does
// not panic (it may return an error, but must not crash or hang).
func TestAutoPipelineDoubleClose(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if err := ap.Close(); err != nil {
		t.Fatalf("first close: %v", err)
	}
	// Second close must not panic; ErrClosed or nil are both acceptable.
	_ = ap.Close()
}

// ===== from autopipeline_context_test.go =====
// AutoPipeline context contract (previously untested; WaitContext had zero
// coverage). The documented contract:
//   - a caller's context is NOT honored once a command is enqueued (execution
//     runs on context.Background); it only gates the wait via WaitContext.
//   - WaitContext abandons the wait on ctx done, but the command still executes
//     and its result becomes readable via a later Wait.
//   - blocking commands (readTimeout != nil) run directly with the caller ctx
//     and DO honor it.

// AP-C1: a caller ctx cancelled after enqueue does not cancel the command; the
// result is still produced.
func TestAPContextCancelDoesNotCancelExecution(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	cctx, cancel := context.WithCancel(ctx)
	f := ap.Set(cctx, "apctx:k", "v", 0)
	cancel() // cancel after submit, before the batch flushes/completes

	if err := f.Err(); err != nil {
		t.Fatalf("command should still execute after caller ctx cancel, got %v", err)
	}
	if f.Val() != "OK" {
		t.Fatalf("val = %q, want OK", f.Val())
	}
	if v, _ := client.Get(ctx, "apctx:k").Result(); v != "v" {
		t.Fatalf("value not persisted: %q", v)
	}
	client.Del(ctx, "apctx:k")
}

// AP-C2: WaitContext returns ctx.Err while the batch is in flight, and a later
// Wait returns the real result.
func TestAPWaitContextAbandonsWaitOnly(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	// Hold the batch open long enough that the WaitContext deadline fires first.
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{MaxBatchSize: 300, MaxFlushDelay: 500 * time.Millisecond})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	f := ap.Submit(ctx, redis.NewCmd(ctx, "set", "apctx:wc", "v"))

	wctx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
	defer cancel()
	if err := f.WaitContext(wctx); err != context.DeadlineExceeded {
		t.Fatalf("WaitContext err = %v, want DeadlineExceeded (batch held open)", err)
	}
	// The command still completes; a plain Wait returns the real result.
	if err := f.Wait(); err != nil {
		t.Fatalf("Wait after WaitContext timeout: %v", err)
	}
	client.Del(ctx, "apctx:wc")
}

// AP-C3: WaitContext returns the real result when the batch completes first.
func TestAPWaitContextReturnsResult(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	f := ap.Submit(ctx, redis.NewCmd(ctx, "set", "apctx:ok", "v"))
	if err := f.WaitContext(ctx); err != nil {
		t.Fatalf("WaitContext: %v", err)
	}
	client.Del(ctx, "apctx:ok")
}

// AP-C4: a blocking command runs directly with the caller ctx (not batched on
// Background). With ContextTimeoutEnabled the ctx deadline reaches the read, so a
// BLPOP is cut short by the ctx — proving the caller ctx flows to the blocking
// path (unlike batched commands, which execute on Background).
func TestAPBlockingCommandHonorsContext(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.ContextTimeoutEnabled = true
	client := redis.NewClient(opt)
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AutoPipeline() // blocking face
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()
	client.Del(ctx, "apctx:blk")

	bctx, cancel := context.WithTimeout(ctx, 200*time.Millisecond)
	defer cancel()
	start := time.Now()
	// BLPOP on an empty list blocks; the ctx deadline must cut it short.
	err = ap.BLPop(bctx, 5*time.Second, "apctx:blk").Err()
	elapsed := time.Since(start)
	if err == nil {
		t.Fatal("expected BLPOP to fail on ctx deadline")
	}
	if elapsed > 2*time.Second {
		t.Fatalf("BLPOP did not honor ctx deadline (took %v)", elapsed)
	}
}

// AP-C5: submitting after Close returns ErrClosed.
func TestAPSubmitAfterCloseErrClosed(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if err := ap.Close(); err != nil {
		t.Fatalf("close: %v", err)
	}
	f := ap.Set(ctx, "apctx:closed", "v", 0)
	if err := f.Err(); err != redis.ErrClosed {
		t.Fatalf("submit after close err = %v, want ErrClosed", err)
	}
}

// AP-C6: WaitContext on a zero AutoFuture returns an error, not a panic.
func TestAPZeroFutureWaitContext(t *testing.T) {
	var f redis.AutoFuture
	if err := f.WaitContext(context.Background()); err == nil {
		t.Fatal("zero AutoFuture WaitContext should return an error")
	}
}

// ===== from autopipeline_correctness_test.go =====
// Validates: each command gets its OWN result (no cross-talk), under heavy
// concurrency. Each goroutine writes a unique value, reads it back, asserts it
// got exactly what it wrote.
func TestAPNoCrossTalk(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{MaxBatchSize: 200, MaxConcurrentBatches: 50, Unordered: true}
	c := redis.NewClient(opt)
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 200
	const perG = 50
	var wg sync.WaitGroup
	var mismatches int64
	var mu sync.Mutex
	errs := []string{}

	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				key := fmt.Sprintf("ct:%d:%d", id, i)
				want := fmt.Sprintf("v-%d-%d", id, i)
				// SET then GET via autopipeline; assert GET returns our value.
				if err := ap.Set(ctx, key, want, 0).Err(); err != nil {
					mu.Lock()
					errs = append(errs, fmt.Sprintf("set %s: %v", key, err))
					mu.Unlock()
					continue
				}
				got, err := ap.Get(ctx, key).Result()
				if err != nil {
					mu.Lock()
					errs = append(errs, fmt.Sprintf("get %s: %v", key, err))
					mu.Unlock()
					continue
				}
				if got != want {
					mu.Lock()
					mismatches++
					if len(errs) < 20 {
						errs = append(errs, fmt.Sprintf("key %s: got %q want %q", key, got, want))
					}
					mu.Unlock()
				}
			}
		}(g)
	}
	wg.Wait()
	if mismatches > 0 || len(errs) > 0 {
		t.Fatalf("cross-talk detected: %d mismatches, errs=%v", mismatches, errs)
	}
}

// Validates: per-goroutine ordering. Each goroutine does INCR on its own key N
// times; final value must equal N and intermediate results must be strictly
// 1,2,3,... in call order (blocking Do guarantees this only if results map to
// the right command).
func TestAPPerGoroutineOrder(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{MaxBatchSize: 100, MaxConcurrentBatches: 30, Unordered: true}
	c := redis.NewClient(opt)
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 100
	const incrs = 100
	var wg sync.WaitGroup
	var bad int64
	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			key := fmt.Sprintf("ord:%d", id)
			for i := 1; i <= incrs; i++ {
				v, err := ap.Incr(ctx, key).Result()
				if err != nil || v != int64(i) {
					atomicAdd(&bad)
				}
			}
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("ordering violated on %d INCR results (expected strict 1..N per goroutine)", bad)
	}
}

// Validates: no lost commands. Issue exactly N writes across goroutines, then
// count keys; must be exactly N.
func TestAPNoLostCommands(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{MaxBatchSize: 256, MaxConcurrentBatches: 64, Unordered: true}
	c := redis.NewClient(opt)
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 300
	const perG = 100
	var wg sync.WaitGroup
	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				ap.Set(ctx, fmt.Sprintf("lost:%d:%d", id, i), "1", 0)
			}
		}(g)
	}
	wg.Wait()

	n, err := c.DBSize(ctx).Result()
	if err != nil {
		t.Fatal(err)
	}
	want := int64(goroutines * perG)
	if n != want {
		t.Fatalf("lost commands: DBSize=%d want=%d (%d missing)", n, want, want-n)
	}
}

// Validates: error isolation. A WRONGTYPE error on one command must not
// corrupt sibling commands sharing the same batch.
func TestAPErrorIsolation(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{MaxBatchSize: 64, MaxConcurrentBatches: 16, Unordered: true}
	c := redis.NewClient(opt)
	defer c.Close()
	c.FlushDB(ctx)
	// Seed a list key so INCR on it errors with WRONGTYPE.
	c.RPush(ctx, "alist", "x")
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 50
	var wg sync.WaitGroup
	var goodFail, badNeighbor int64
	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			// Good neighbor command (own key)
			ok := fmt.Sprintf("iso:%d", id)
			// Interleave a guaranteed-error command (INCR on a list) with a good one.
			errCmd := ap.Incr(ctx, "alist")
			goodCmd := ap.Set(ctx, ok, strconv.Itoa(id), 0)
			if errCmd.Err() == nil {
				atomicAdd(&badNeighbor) // the bad command should have errored
			} else {
				atomicAdd(&goodFail)
			}
			if goodCmd.Err() != nil {
				atomicAdd(&badNeighbor) // the good command must NOT inherit the error
			}
		}(g)
	}
	wg.Wait()
	if badNeighbor > 0 {
		t.Fatalf("error isolation broken: %d cases (good cmd errored or bad cmd succeeded)", badNeighbor)
	}
	if goodFail != goroutines {
		t.Fatalf("expected all %d INCR-on-list to error, got %d", goroutines, goodFail)
	}
}

func atomicAdd(p *int64) {
	atomic.AddInt64(p, 1)
}

// ===== from autopipeline_faces_split_test.go =====
// TestBlockingFaceExecutesOnCall: AutoPipeline() command blocks until executed —
// the returned cmd already holds its result without an explicit further wait.
func TestBlockingFaceExecutesOnCall(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AutoPipeline() // blocking face
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const G, N = 50, 100
	var wg sync.WaitGroup
	var bad int
	var mu sync.Mutex
	wg.Add(G)
	for g := 0; g < G; g++ {
		go func(id int) {
			defer wg.Done()
			key := func() string { return "blk:" + string(rune('a'+id%26)) }()
			_ = key
			for i := 1; i <= N; i++ {
				// Incr blocks until executed; per-goroutine order must hold.
				cmd := ap.Incr(ctx, "blkkey:"+string(rune(id)))
				if v, err := cmd.Result(); err != nil || v != int64(i) {
					mu.Lock()
					bad++
					mu.Unlock()
				}
			}
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("blocking face: %d ordering/exec failures", bad)
	}
}

// TestAsyncFaceDeferred: AsyncAutoPipeline() command returns immediately; result
// is read later. Windowed submit then read works.
func TestAsyncFaceDeferred(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)
	ap, err := c.AsyncAutoPipeline() // deferred face, ordered default
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const N = 500
	cmds := make([]*redis.IntCmd, 0, N)
	for i := 0; i < N; i++ {
		cmds = append(cmds, ap.Incr(ctx, "asynckey")) // does not block
	}
	for i, cmd := range cmds {
		if v, err := cmd.Result(); err != nil || v != int64(i+1) {
			t.Fatalf("async windowed: cmd %d got v=%d err=%v", i, v, err)
		}
	}
}

// ===== from autopipeline_faces_test.go =====
// TestFutureFaceTyped verifies the future face exposes the typed command
// surface returning the usual *XxxCmd (same shape as the blocking client),
// with the result deferred until read. No cross-talk across goroutines.
func TestFutureFaceTyped(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)
	fap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer fap.Close()

	const G, N = 100, 50
	var wg sync.WaitGroup
	var bad int64
	wg.Add(G)
	for g := 0; g < G; g++ {
		go func(id int) {
			defer wg.Done()
			sets := make([]*redis.StatusCmd, N)
			for i := 0; i < N; i++ {
				sets[i] = fap.Set(ctx, fmt.Sprintf("ff:%d:%d", id, i), fmt.Sprintf("v%d-%d", id, i), 0)
			}
			for _, s := range sets {
				if s.Err() != nil {
					atomic.AddInt64(&bad, 1)
				}
			}
			for i := 0; i < N; i++ {
				want := fmt.Sprintf("v%d-%d", id, i)
				if v, _ := fap.Get(ctx, fmt.Sprintf("ff:%d:%d", id, i)).Result(); v != want {
					atomic.AddInt64(&bad, 1)
				}
			}
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("future face: %d failures", bad)
	}
}

// TestOrderedModeWindowed verifies that MaxConcurrentBatches=1 gives a single
// ordered command stream: a windowed caller (submit many, read later) sees
// strict per-key ordering even though it never blocks between submits.
func TestOrderedModeWindowed(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)
	// Deferred face: submits genuinely don't block, so the windowed claim is
	// real (on the blocking face every Incr would wait, proving nothing).
	fap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{MaxBatchSize: 500, MaxConcurrentBatches: 1})
	if err != nil {
		t.Fatal(err)
	}
	defer fap.Close()

	const G, N = 80, 150
	var wg sync.WaitGroup
	var bad int64
	wg.Add(G)
	for g := 0; g < G; g++ {
		go func(id int) {
			defer wg.Done()
			key := fmt.Sprintf("om:%d", id)
			cs := make([]*redis.IntCmd, 0, N)
			for i := 0; i < N; i++ {
				cs = append(cs, fap.Incr(ctx, key)) // windowed: no read between submits
			}
			for i, cc := range cs {
				if v, err := cc.Result(); err != nil || v != int64(i+1) {
					atomic.AddInt64(&bad, 1)
				}
			}
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("ordered windowed: %d ordering violations", bad)
	}
}

// TestAutoPipelineOptionsValidate verifies the ordering-safety rule: parallel
// batches (MaxConcurrentBatches>1) require an explicit Unordered:true opt-out.
func TestAutoPipelineOptionsValidate(t *testing.T) {
	// conc>1 without Unordered -> error.
	if err := (&redis.AutoPipelineOptions{MaxConcurrentBatches: 8}).Validate(); err == nil {
		t.Fatal("expected error for MaxConcurrentBatches>1 without Unordered")
	}
	// conc>1 with Unordered -> ok.
	if err := (&redis.AutoPipelineOptions{MaxConcurrentBatches: 8, Unordered: true}).Validate(); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	// ordered single stream -> ok.
	if err := (&redis.AutoPipelineOptions{MaxConcurrentBatches: 1}).Validate(); err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	// default is ordered and valid.
	if err := redis.DefaultAutoPipelineOptions().Validate(); err != nil {
		t.Fatalf("default config invalid: %v", err)
	}
	// negative values are rejected so a typo surfaces at construction.
	if err := (&redis.AutoPipelineOptions{MaxBatchSize: -1}).Validate(); err == nil {
		t.Fatal("expected error for negative MaxBatchSize")
	}
	if err := (&redis.AutoPipelineOptions{MaxConcurrentBatches: -1}).Validate(); err == nil {
		t.Fatal("expected error for negative MaxConcurrentBatches")
	}
	if err := (&redis.AutoPipelineOptions{MaxFlushDelay: -1}).Validate(); err == nil {
		t.Fatal("expected error for negative MaxFlushDelay")
	}
	// zero values are allowed (mean "use default" / "no delay").
	if err := (&redis.AutoPipelineOptions{}).Validate(); err != nil {
		t.Fatalf("zero-value config should be valid: %v", err)
	}
}

// TestAutoPipelineErrorsOnUnsafeConfig verifies the unsafe config is rejected
// with an error (not a panic): AsyncAutoPipeline runs from a post-init call, so a
// bad config surfaces as a returned error the caller can handle.
func TestAutoPipelineErrorsOnUnsafeConfig(t *testing.T) {
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	ap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{MaxConcurrentBatches: 8})
	if err == nil {
		t.Fatal("expected error for MaxConcurrentBatches>1 without Unordered")
	}
	if ap != nil {
		t.Fatal("expected nil AutoPipeliner on config error")
	}
}

// ===== from autopipeline_fuzz_test.go =====
// TestAutoPipelineDifferentialFuzz runs the same randomized, type-diverse command
// stream through a plain client and diffs every result against two autopipeline
// runs: the blocking face called sequentially (which exercises the lone-command
// fast path — each call flushes solo), and the async face driven windowed —
// every command submitted before any result is read — so the stream coalesces
// into real multi-command batches and the diff covers cross-talk, ordering, and
// reply-shape demux where they can actually break. A counting hook asserts the
// async phase really formed >=2-command batches (anti-vacuousness guard).
func TestAutoPipelineDifferentialFuzz(t *testing.T) {
	ctx := context.Background()
	// Dedicated DB so FlushDB between runs can't touch anything else.
	client := redis.NewClient(&redis.Options{Addr: apTestAddr(), DB: 15})
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	for _, seed := range []int64{1, 7, 42, 1337} {
		ops := genOps(rand.New(rand.NewSource(seed)), 400)

		if err := client.FlushDB(ctx).Err(); err != nil {
			t.Fatal(err)
		}
		plainRes := runOps(ctx, client, ops)

		if err := client.FlushDB(ctx).Err(); err != nil {
			t.Fatal(err)
		}
		ap, err := client.AutoPipeline()
		if err != nil {
			t.Fatal(err)
		}
		apRes := runOps(ctx, ap, ops)
		ap.Close()

		for i := range ops {
			if plainRes[i] != apRes[i] {
				t.Fatalf("seed=%d op#%d %s(%s): plain=%q autopipeline=%q",
					seed, i, ops[i].kind, ops[i].key, plainRes[i], apRes[i])
			}
		}

		// Async windowed phase: submit everything, then read. The ordered
		// default preserves submit order, so results must equal the
		// sequential plain run — but now via real multi-command batches.
		if err := client.FlushDB(ctx).Err(); err != nil {
			t.Fatal(err)
		}
		batchClient := redis.NewClient(&redis.Options{Addr: apTestAddr(), DB: 15})
		var realBatches atomic.Int32
		batchClient.AddHook(batchCountingHook{batches: &realBatches})
		asyncAP, err := batchClient.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
			MaxBatchSize:  100,
			MaxFlushDelay: 2 * time.Millisecond,
		})
		if err != nil {
			t.Fatal(err)
		}
		asyncRes := runOpsWindowed(ctx, asyncAP, ops)
		asyncAP.Close()
		batchClient.Close()

		for i := range ops {
			if plainRes[i] != asyncRes[i] {
				t.Fatalf("seed=%d op#%d %s(%s): plain=%q async-windowed=%q",
					seed, i, ops[i].kind, ops[i].key, plainRes[i], asyncRes[i])
			}
		}
		if realBatches.Load() == 0 {
			t.Fatalf("seed=%d: async windowed run formed no >=2-command batches — differential coverage is vacuous", seed)
		}
	}
	client.FlushDB(ctx)
}

// batchCountingHook counts pipeline dispatches of >=2 commands: proof the
// windowed fuzz phase exercised real batches rather than the solo fast path.
type batchCountingHook struct{ batches *atomic.Int32 }

func (batchCountingHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (batchCountingHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return next
}

func (h batchCountingHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if len(cmds) >= 2 {
			h.batches.Add(1)
		}
		return next(ctx, cmds)
	}
}

type fuzzOp struct {
	kind string
	key  string
	val  string
	n    int64
}

func genOps(rng *rand.Rand, n int) []fuzzOp {
	kinds := []string{
		"set", "get", "getdel", "incr", "incrby", "append", "strlen",
		"exists", "del", "hset", "hget", "hgetall", "lpush", "lrange", "llen",
		"expire", "ttl", "type", "setbig", "setbin",
	}
	keys := make([]string, 8)
	for i := range keys {
		keys[i] = fmt.Sprintf("fz:%d", i)
	}
	ops := make([]fuzzOp, n)
	for i := range ops {
		k := keys[rng.Intn(len(keys))]
		ops[i] = fuzzOp{
			kind: kinds[rng.Intn(len(kinds))], key: k,
			val: fmt.Sprintf("v%d", rng.Intn(1000)), n: int64(rng.Intn(50)),
		}
	}
	return ops
}

// runOps executes the op sequence via the Cmdable interface and returns a
// normalized "value|error" string per op for cross-run comparison.
func runOps(ctx context.Context, c redis.Cmdable, ops []fuzzOp) []string {
	out := make([]string, len(ops))
	res := func(v interface{}, err error) string {
		if err != nil {
			return "ERR:" + err.Error()
		}
		return fmt.Sprintf("%v", v)
	}
	for i, op := range ops {
		switch op.kind {
		case "set":
			out[i] = res(c.Set(ctx, op.key, op.val, 0).Result())
		case "get":
			out[i] = res(c.Get(ctx, op.key).Result())
		case "getdel":
			out[i] = res(c.GetDel(ctx, op.key).Result())
		case "incr":
			out[i] = res(c.Incr(ctx, op.key).Result())
		case "incrby":
			out[i] = res(c.IncrBy(ctx, op.key, op.n).Result())
		case "append":
			out[i] = res(c.Append(ctx, op.key, op.val).Result())
		case "strlen":
			out[i] = res(c.StrLen(ctx, op.key).Result())
		case "exists":
			out[i] = res(c.Exists(ctx, op.key).Result())
		case "del":
			out[i] = res(c.Del(ctx, op.key).Result())
		case "hset":
			out[i] = res(c.HSet(ctx, op.key, op.val, op.n).Result())
		case "hget":
			out[i] = res(c.HGet(ctx, op.key, op.val).Result())
		case "hgetall":
			out[i] = res(c.HGetAll(ctx, op.key).Result())
		case "lpush":
			out[i] = res(c.LPush(ctx, op.key, op.val).Result())
		case "lrange":
			out[i] = res(c.LRange(ctx, op.key, 0, -1).Result())
		case "llen":
			out[i] = res(c.LLen(ctx, op.key).Result())
		case "expire":
			out[i] = res(c.Expire(ctx, op.key, 100*time.Second).Result())
		case "ttl":
			// TTL varies with wall clock; normalize to "set"/"unset".
			d, err := c.TTL(ctx, op.key).Result()
			if err != nil {
				out[i] = "ERR:" + err.Error()
			} else if d > 0 {
				out[i] = "ttl>0"
			} else {
				out[i] = fmt.Sprintf("%v", d)
			}
		case "type":
			out[i] = res(c.Type(ctx, op.key).Result())
		case "setbig":
			out[i] = res(c.Set(ctx, op.key, strings.Repeat("Z", 70000), 0).Result())
		case "setbin":
			out[i] = res(c.Set(ctx, op.key, "b\x00\r\n\x01\xffX", 0).Result())
		}
	}
	return out
}

// runOpsWindowed issues the whole op sequence first and materializes results
// afterwards, in order — on the async face this coalesces the stream into
// real multi-command batches instead of lone-command flushes. Result strings
// are normalized identically to runOps.
func runOpsWindowed(ctx context.Context, c redis.Cmdable, ops []fuzzOp) []string {
	out := make([]string, len(ops))
	res := func(v interface{}, err error) string {
		if err != nil {
			return "ERR:" + err.Error()
		}
		return fmt.Sprintf("%v", v)
	}
	thunks := make([]func() string, len(ops))
	for i, op := range ops {
		switch op.kind {
		case "set":
			cmd := c.Set(ctx, op.key, op.val, 0)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "get":
			cmd := c.Get(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "getdel":
			cmd := c.GetDel(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "incr":
			cmd := c.Incr(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "incrby":
			cmd := c.IncrBy(ctx, op.key, op.n)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "append":
			cmd := c.Append(ctx, op.key, op.val)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "strlen":
			cmd := c.StrLen(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "exists":
			cmd := c.Exists(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "del":
			cmd := c.Del(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "hset":
			cmd := c.HSet(ctx, op.key, op.val, op.n)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "hget":
			cmd := c.HGet(ctx, op.key, op.val)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "hgetall":
			cmd := c.HGetAll(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "lpush":
			cmd := c.LPush(ctx, op.key, op.val)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "lrange":
			cmd := c.LRange(ctx, op.key, 0, -1)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "llen":
			cmd := c.LLen(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "expire":
			cmd := c.Expire(ctx, op.key, 100*time.Second)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "ttl":
			cmd := c.TTL(ctx, op.key)
			thunks[i] = func() string {
				d, err := cmd.Result()
				if err != nil {
					return "ERR:" + err.Error()
				}
				if d > 0 {
					return "ttl>0"
				}
				return fmt.Sprintf("%v", d)
			}
		case "type":
			cmd := c.Type(ctx, op.key)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "setbig":
			cmd := c.Set(ctx, op.key, strings.Repeat("Z", 70000), 0)
			thunks[i] = func() string { return res(cmd.Result()) }
		case "setbin":
			cmd := c.Set(ctx, op.key, "b\x00\r\n\x01\xffX", 0)
			thunks[i] = func() string { return res(cmd.Result()) }
		}
	}
	for i, th := range thunks {
		out[i] = th()
	}
	return out
}

// ===== from autopipeline_hook_deadlock_test.go =====
// resultPeekHook reads every command's result inside the hook chain — the
// redisotel span-status pattern (after next) and the stricter before-next
// variant from the #3867 deadlock report. Reading a deferred command's result
// on the dispatch goroutine used to block on the batch's done channel, which
// that same goroutine closes only after the hook chain returns: a
// deterministic deadlock on the async face.
type resultPeekHook struct{ before bool }

func (h resultPeekHook) DialHook(next redis.DialHook) redis.DialHook { return next }

func (h resultPeekHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		if h.before {
			_ = cmd.Err()
			_ = cmd.String()
		}
		err := next(ctx, cmd)
		if !h.before {
			_ = cmd.Err()
			_ = cmd.String()
		}
		return err
	}
}

func (h resultPeekHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if h.before {
			for _, c := range cmds {
				_ = c.Err()
				_ = c.String()
			}
		}
		err := next(ctx, cmds)
		if !h.before {
			for _, c := range cmds {
				_ = c.Err()
				_ = c.String()
			}
		}
		return err
	}
}

// runWithWatchdog fails the test instead of hanging the suite if fn deadlocks.
func runWithWatchdog(t *testing.T, timeout time.Duration, fn func()) {
	t.Helper()
	done := make(chan struct{})
	go func() { defer close(done); fn() }()
	select {
	case <-done:
	case <-time.After(timeout):
		t.Fatal("deadlock: hook reading deferred results blocked the dispatch")
	}
}

func testAsyncHookPeek(t *testing.T, before bool) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.AddHook(resultPeekHook{before: before})
	c.FlushDB(ctx)

	ap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	runWithWatchdog(t, 30*time.Second, func() {
		const N = 200
		sets := make([]*redis.StatusCmd, N)
		for i := 0; i < N; i++ {
			sets[i] = ap.Set(ctx, fmt.Sprintf("hookpeek:%d", i), i, 0)
		}
		for i, s := range sets {
			if s.Err() != nil {
				t.Errorf("set %d: %v", i, s.Err())
			}
		}
		for i := 0; i < N; i++ {
			v, err := ap.Get(ctx, fmt.Sprintf("hookpeek:%d", i)).Result()
			if err != nil || v != fmt.Sprint(i) {
				t.Errorf("get %d: v=%q err=%v", i, v, err)
			}
		}
	})
}

// TestAsyncAutoPipelineHookReadsAfterNext: redisotel-style hook (reads results
// after next()). Deadlocked before the innermost-close fix.
func TestAsyncAutoPipelineHookReadsAfterNext(t *testing.T) { testAsyncHookPeek(t, false) }

// TestAsyncAutoPipelineHookReadsBeforeNext: hook reads results before next().
// Deadlocked before the dispatcher-goroutine guard in await().
func TestAsyncAutoPipelineHookReadsBeforeNext(t *testing.T) { testAsyncHookPeek(t, true) }

// TestAsyncAutoPipelineHookSoloCommand drives the lone-command fast path
// (Process chain, not the pipeline chain) with both hook variants.
func TestAsyncAutoPipelineHookSoloCommand(t *testing.T) {
	for _, before := range []bool{false, true} {
		t.Run(fmt.Sprintf("before=%v", before), func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			c.AddHook(resultPeekHook{before: before})
			c.FlushDB(ctx)

			ap, err := c.AsyncAutoPipeline()
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				// One command, then an immediate read: the lone-caller flush
				// dispatches it through the solo Process path.
				cmd := ap.Set(ctx, "hookpeek:solo", "v", 0)
				if err := cmd.Err(); err != nil {
					t.Errorf("solo set: %v", err)
				}
				if v, err := ap.Get(ctx, "hookpeek:solo").Result(); err != nil || v != "v" {
					t.Errorf("solo get: v=%q err=%v", v, err)
				}
			})
		})
	}
}

// TestBlockingAutoPipelineHookReads: the blocking face was never affected;
// pin that with the same hooks.
func TestBlockingAutoPipelineHookReads(t *testing.T) {
	for _, before := range []bool{false, true} {
		t.Run(fmt.Sprintf("before=%v", before), func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			c.AddHook(resultPeekHook{before: before})
			c.FlushDB(ctx)

			ap, err := c.AutoPipeline()
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				for i := 0; i < 50; i++ {
					if err := ap.Set(ctx, fmt.Sprintf("hookpeek:b:%d", i), i, 0).Err(); err != nil {
						t.Errorf("set %d: %v", i, err)
					}
				}
			})
		})
	}
}

// TestAsyncAutoPipelineDoHookReads drives Do's background goroutine (its own
// close path) under a result-reading ProcessHook.
func TestAsyncAutoPipelineDoHookReads(t *testing.T) {
	for _, before := range []bool{false, true} {
		t.Run(fmt.Sprintf("before=%v", before), func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			c.AddHook(resultPeekHook{before: before})

			ap, err := c.AsyncAutoPipeline()
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				cmd := ap.Do(ctx, "PING")
				if v, err := cmd.Result(); err != nil || v != "PONG" {
					t.Errorf("do ping: v=%v err=%v", v, err)
				}
			})
		})
	}
}

// ===== from autopipeline_leak_test.go =====
// TestAutoPipelineNoGoroutineLeak opens and closes many autopipelines (both
// faces, ordered + unordered) and asserts goroutines return to baseline — i.e.
// flushers AND dispatcher workers all exit on Close, nothing leaks.
func TestAutoPipelineNoGoroutineLeak(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.PoolSize = 50
	client := redis.NewClient(opt)
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	cfgs := []*redis.AutoPipelineOptions{
		{MaxBatchSize: 300}, // ordered, 1 permit, 1 stripe
		{MaxBatchSize: 300, MaxConcurrentBatches: 8, Unordered: true},  // unordered, 8 permits, 8 stripes
		{MaxBatchSize: 300, MaxConcurrentBatches: 50, Unordered: true}, // many permits -> many workers
	}

	cycle := func() {
		for _, cfg := range cfgs {
			for _, async := range []bool{false, true} {
				var ap *redis.AutoPipeliner
				var err error
				if async {
					ap, err = client.AsyncAutoPipelineWithOptions(cfg)
				} else {
					ap, err = client.AutoPipelineWithOptions(cfg)
				}
				if err != nil {
					t.Fatal(err)
				}
				wg := make([]redis.AutoFuture, 0, 200)
				for i := 0; i < 200; i++ {
					if async {
						wg = append(wg, ap.Submit(ctx, redis.NewCmd(ctx, "set", "leak:k", i)))
					} else {
						_ = ap.Set(ctx, "leak:k", i, 0).Err()
					}
				}
				for i := range wg {
					_ = wg[i].Wait()
				}
				if err := ap.Close(); err != nil {
					t.Fatalf("close: %v", err)
				}
			}
		}
	}

	cycle() // warmup so lazy client/pool goroutines exist before we snapshot
	runtime.GC()
	time.Sleep(150 * time.Millisecond)
	base := runtime.NumGoroutine()

	const rounds = 40
	for r := 0; r < rounds; r++ {
		cycle()
	}
	got := settleGoroutines(base + 8) // poll: exiting goroutines wind down at their own pace

	// Allow small slack for pool/runtime background goroutines; a leak of
	// flushers/workers would be rounds*(shards+workers) ~ hundreds.
	if got > base+8 {
		t.Fatalf("goroutine leak: baseline=%d after %d open/close rounds=%d (delta %d)",
			base, rounds, got, got-base)
	}
	t.Logf("no leak: baseline=%d final=%d (delta %d) over %d rounds", base, got, got-base, rounds)
}

// ===== from autopipeline_retry_test.go =====
// TestAutoPipelineRetriesOnNetworkError verifies AutoPipeline inherits the
// pipeline retry/re-drive loop (it dispatches through the same execer as manual
// Pipeline). The pooled conn's next write is armed to fail at the wire (a
// seeded broken idle conn would be filtered by the pool health check and never
// carry an attempt); the batch's first attempt genuinely dies, the retry
// redials, and the futures must read correct results from the re-drive.
//
// MaxFlushDelay forces the two commands into one real (>=2) batch so they run
// through the pipeline execer; a lone command would take the single-command
// fast path (Process) instead.
func TestAutoPipelineRetriesOnNetworkError(t *testing.T) {
	ctx := context.Background()
	var failNextWrite atomic.Bool
	var dials atomic.Int32
	client := redis.NewClient(&redis.Options{
		Addr: apTestAddr(), MaxRetries: 2, PoolSize: 1,
		Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
			dials.Add(1)
			cn, err := (&net.Dialer{}).DialContext(ctx, network, addr)
			if err != nil {
				return nil, err
			}
			return &flakyWriteConn{Conn: cn, fail: &failNextWrite}, nil
		},
	})
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	if err := client.Set(ctx, "apr:n", 0, 0).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  300,
		MaxFlushDelay: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Warm the DEDICATED PIPELINE POOL connection first: creating the
	// autopipeliner created the pipeline pool (lazily), and BATCHES run there,
	// not on the main-pool conn the Ping dialed. Without this warm-up the
	// batch's first dispatch dials a fresh pipeline conn and the dial count
	// reads one high for a reason unrelated to the retry. The warm-up must be
	// a real multi-command batch: a lone command takes the solo fast path
	// (Process on the main pool) and would not touch the pipeline pool.
	w1 := ap.Set(ctx, "apr:warm", 1, 0)
	w2 := ap.Incr(ctx, "apr:warm2")
	if err := w1.Err(); err != nil {
		t.Fatal(err)
	}
	if err := w2.Err(); err != nil {
		t.Fatal(err)
	}

	// Arm after the handshake: the pipeline-pool conn is healthy, so it passes
	// the pool health check, and the batch's first write dies on the wire.
	dialsBefore := dials.Load()
	failNextWrite.Store(true)

	s := ap.Set(ctx, "apr:k", "v", 0)
	n := ap.Incr(ctx, "apr:n")
	if err := s.Err(); err != nil {
		t.Fatalf("set after retry: %v", err)
	}
	if err := n.Err(); err != nil {
		t.Fatalf("incr after retry: %v", err)
	}
	if s.Val() != "OK" {
		t.Fatalf("set = %q, want OK", s.Val())
	}
	if n.Val() != 1 {
		t.Fatalf("incr = %d, want 1", n.Val())
	}
	if failNextWrite.Load() {
		t.Fatal("armed write failure never consumed: the first attempt did not reach the wire")
	}
	if got := dials.Load(); got != dialsBefore+1 {
		t.Fatalf("dials after failure = %d, want %d (exactly one redial by the retry)", got, dialsBefore+1)
	}
}

// TestAutoPipelineRetryExhaustionSurfacesError checks that when the pipeline
// retry loop is EXHAUSTED (every attempt fails on a retryable error), the
// AutoPipeline commands surface the error to their futures.
//
// This matters more for AutoPipeline than for manual Pipeline: manual Pipeline
// returns the error from Exec, but AutoPipeline discards the pipeline error
// (dispatchCmds: `_ = ...processPipelineHook(...)`) and results are observable
// ONLY via each command's future. If generalProcessPipeline does not call
// setCmdsErr on exhaustion (it only does so on the early-exit branch), a caller
// sees a nil error and a zero value — silent loss.
func TestAutoPipelineRetryExhaustionSurfacesError(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{
		Addr:       apTestAddr(),
		MaxRetries: 2,
		PoolSize:   1,
		Dialer: func(ctx context.Context, network, addr string) (net.Conn, error) {
			return &badConn{writeErr: io.EOF}, nil // every conn fails all I/O
		},
	})
	defer client.Close()

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  300,
		MaxFlushDelay: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	s := ap.Set(ctx, "k", "v", 0)
	n := ap.Incr(ctx, "n")
	if s.Err() == nil || n.Err() == nil {
		t.Fatalf("SILENT LOSS: AutoPipeline commands returned nil error on retry exhaustion (set err=%v, incr err=%v) — every attempt failed but the futures report success",
			s.Err(), n.Err())
	}
}

// ===== from autopipeline_sequential_test.go =====
func TestAutoPipelineSequential(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
		MaxBatchSize:         10,
		MaxFlushDelay:        50 * time.Millisecond,
		MaxConcurrentBatches: 5,
		Unordered:            true,
	}
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AsyncAutoPipeline() // deferred face: submit-then-read (Do bypasses the engine)
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Sequential usage - no goroutines needed: typed calls on the deferred
	// face queue without blocking and are batched automatically
	cmds := make([]redis.Cmder, 20)
	for i := 0; i < 20; i++ {
		key := fmt.Sprintf("key%d", i)
		cmds[i] = ap.Set(ctx, key, i, 0)
	}

	// Now access results - this will block until commands execute
	for i, cmd := range cmds {
		if err := cmd.Err(); err != nil {
			t.Fatalf("Command %d failed: %v", i, err)
		}
	}

	// Verify all keys were set
	for i := 0; i < 20; i++ {
		key := fmt.Sprintf("key%d", i)
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get %s: %v", key, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d", i, val)
		}
	}
}

func TestAutoPipelineSequentialSmallBatches(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
		MaxBatchSize:         1000,                  // Large batch size
		MaxFlushDelay:        20 * time.Millisecond, // Rely on timer
		MaxConcurrentBatches: 5,
		Unordered:            true,
	}
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AsyncAutoPipeline() // deferred face: submit-then-read (Do bypasses the engine)
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Queue commands sequentially with small delays
	// They should be flushed by timer, not batch size
	for i := 0; i < 5; i++ {
		key := fmt.Sprintf("key%d", i)
		cmd := ap.Set(ctx, key, i, 0)

		// Access result immediately - should block until flush
		if err := cmd.Err(); err != nil {
			t.Fatalf("Command %d failed: %v", i, err)
		}

		time.Sleep(5 * time.Millisecond)
	}

	// Verify all keys were set
	for i := 0; i < 5; i++ {
		key := fmt.Sprintf("key%d", i)
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get %s: %v", key, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d", i, val)
		}
	}
}

func TestAutoPipelineSequentialMixed(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
		MaxBatchSize:         5,
		MaxFlushDelay:        50 * time.Millisecond,
		MaxConcurrentBatches: 5,
		Unordered:            true,
	}
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AsyncAutoPipeline() // deferred face: submit-then-read (Do bypasses the engine)
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Queue some commands
	cmd1 := ap.Set(ctx, "key1", "value1", 0)
	cmd2 := ap.Set(ctx, "key2", "value2", 0)
	cmd3 := ap.Set(ctx, "key3", "value3", 0)

	// Access first result - should trigger batch flush
	if err := cmd1.Err(); err != nil {
		t.Fatalf("cmd1 failed: %v", err)
	}

	// Other commands in same batch should also be done
	if err := cmd2.Err(); err != nil {
		t.Fatalf("cmd2 failed: %v", err)
	}
	if err := cmd3.Err(); err != nil {
		t.Fatalf("cmd3 failed: %v", err)
	}

	// Verify
	val, err := client.Get(ctx, "key1").Result()
	if err != nil || val != "value1" {
		t.Fatalf("key1: expected value1, got %v, err %v", val, err)
	}
}

// ===== from autopipeline_singleton_test.go =====
// TestAutoPipelineConcurrentFirstCall guards a data race where AutoPipeline()
// did its cache check-and-set on the client's autopipeliner field without a
// lock. Concurrent first calls could each build a separate AutoPipeliner —
// racing the field (caught by -race) and leaking the loser's flusher goroutines
// and pipeline pool. Every concurrent caller must observe the SAME instance.
func TestAutoPipelineConcurrentFirstCall(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	c.FlushDB(ctx)

	const G = 64
	results := make([]*redis.AutoPipeliner, G)
	errs := make([]error, G)
	var wg sync.WaitGroup
	start := make(chan struct{})
	wg.Add(G)
	for g := 0; g < G; g++ {
		go func(i int) {
			defer wg.Done()
			<-start // maximize the race on the first call
			results[i], errs[i] = c.AutoPipeline()
		}(g)
	}
	close(start)
	wg.Wait()

	for i, err := range errs {
		if err != nil {
			t.Fatalf("call %d: AutoPipeline() returned error: %v", i, err)
		}
	}

	first := results[0]
	if first == nil {
		t.Fatal("AutoPipeline() returned nil")
	}
	for i, ap := range results {
		if ap != first {
			t.Fatalf("call %d returned a different AutoPipeliner instance; concurrent first calls must share one", i)
		}
	}
	defer first.Close()

	// The shared instance must still work.
	if err := first.Set(ctx, "singleton", "1", 0).Err(); err != nil {
		t.Fatalf("shared autopipeliner unusable: %v", err)
	}
}

// ===== from autopipeline_test.go =====
var _ = Describe("AutoPipeline", func() {
	ctx := context.Background()
	var client *redis.Client
	var ap *redis.AutoPipeliner

	BeforeEach(func() {
		opt := redisOptions()
		opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
			MaxBatchSize:         10,
			MaxFlushDelay:        50 * time.Millisecond,
			MaxConcurrentBatches: 5,
			Unordered:            true,
		}
		client = redis.NewClient(opt)
		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())

		var err error
		ap, err = client.AutoPipeline()
		Expect(err).NotTo(HaveOccurred())
	})

	AfterEach(func() {
		if ap != nil {
			Expect(ap.Close()).NotTo(HaveOccurred())
		}
		Expect(client.Close()).NotTo(HaveOccurred())
	})

	It("should batch commands automatically", func() {
		// Issue multiple commands concurrently to enable batching. A typed call
		// without blocking, so each goroutine must await its own command's result
		// (cmd.Err) before reporting done — wg.Wait alone only waits for the
		// queueing, not the batch flush+execute, which would race the reads below.
		var wg sync.WaitGroup
		for i := 0; i < 5; i++ {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()
				key := fmt.Sprintf("key%d", idx)
				Expect(ap.Set(ctx, key, idx, 0).Err()).NotTo(HaveOccurred())
			}(i)
		}

		// Wait for all commands to execute (each goroutine awaited its result).
		wg.Wait()

		// Verify all commands were executed
		for i := 0; i < 5; i++ {
			key := fmt.Sprintf("key%d", i)
			val, err := client.Get(ctx, key).Int()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal(i))
		}
	})

	It("should flush when batch size is reached", func() {
		// Queue exactly MaxBatchSize commands concurrently. Each goroutine awaits
		// its command's result so wg.Wait reflects execution, not just queueing.
		var wg sync.WaitGroup
		for i := 0; i < 10; i++ {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()
				key := fmt.Sprintf("key%d", idx)
				Expect(ap.Set(ctx, key, idx, 0).Err()).NotTo(HaveOccurred())
			}(i)
		}

		// Wait for all commands to execute (each goroutine awaited its result).
		wg.Wait()

		// Verify all commands were executed
		for i := 0; i < 10; i++ {
			key := fmt.Sprintf("key%d", i)
			val, err := client.Get(ctx, key).Int()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal(i))
		}
	})

	It("should flush on timer expiry", func() {
		// Queue a single command from another goroutine; the blocking call
		// returns only after its batch flushed, so signal completion instead
		// of sleeping a fixed window.
		done := make(chan error, 1)
		go func() {
			done <- ap.Set(ctx, "key1", "value1", 0).Err()
		}()
		Eventually(done, "5s").Should(Receive(BeNil()))

		val, err := client.Get(ctx, "key1").Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(val).To(Equal("value1"))
	})

	It("should handle concurrent commands", func() {
		const numGoroutines = 10
		const commandsPerGoroutine = 100

		var wg sync.WaitGroup
		wg.Add(numGoroutines)

		for g := 0; g < numGoroutines; g++ {
			go func(goroutineID int) {
				defer wg.Done()
				for i := 0; i < commandsPerGoroutine; i++ {
					key := fmt.Sprintf("g%d:key%d", goroutineID, i)
					ap.Set(ctx, key, i, 0)
				}
			}(g)
		}

		wg.Wait()

		// All commands are now complete (Do() blocks until executed)
		// Verify all commands were executed
		for g := 0; g < numGoroutines; g++ {
			for i := 0; i < commandsPerGoroutine; i++ {
				key := fmt.Sprintf("g%d:key%d", g, i)
				val, err := client.Get(ctx, key).Int()
				Expect(err).NotTo(HaveOccurred())
				Expect(val).To(Equal(i))
			}
		}
	})

	It("should support manual flush", func() {
		// Queue commands concurrently
		var wg sync.WaitGroup
		for i := 0; i < 5; i++ {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()
				key := fmt.Sprintf("key%d", idx)
				ap.Set(ctx, key, idx, 0)
			}(i)
		}

		// Wait for all commands to complete
		wg.Wait()

		// Verify all commands were executed
		for i := 0; i < 5; i++ {
			key := fmt.Sprintf("key%d", i)
			val, err := client.Get(ctx, key).Int()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal(i))
		}
	})

	It("should handle errors in commands", func() {
		// Queue a mix of valid and invalid commands concurrently
		var wg sync.WaitGroup

		wg.Add(1)
		go func() {
			defer wg.Done()
			ap.Set(ctx, "key1", "value1", 0)
		}()

		wg.Add(1)
		go func() {
			defer wg.Done()
			// A command failing with a redis error (WRONGTYPE) inside a batch
			// must not affect its batch neighbours.
			client.Set(ctx, "not-a-counter", "str", 0)
			_ = ap.Incr(ctx, "not-a-counter").Err()
		}()

		wg.Add(1)
		go func() {
			defer wg.Done()
			ap.Set(ctx, "key2", "value2", 0)
		}()

		// Wait for all commands to complete
		wg.Wait()

		// Valid commands should still execute
		val, err := client.Get(ctx, "key1").Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(val).To(Equal("value1"))

		val, err = client.Get(ctx, "key2").Result()
		Expect(err).NotTo(HaveOccurred())
		Expect(val).To(Equal("value2"))
	})

	It("should close gracefully", func() {
		// Queue commands concurrently
		var wg sync.WaitGroup
		for i := 0; i < 5; i++ {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()
				key := fmt.Sprintf("key%d", idx)
				ap.Set(ctx, key, idx, 0)
			}(i)
		}

		// Wait for all commands to complete
		wg.Wait()

		// Close should flush pending commands
		Expect(ap.Close()).NotTo(HaveOccurred())
		ap = nil // Prevent double close in AfterEach

		// Verify all commands were executed
		for i := 0; i < 5; i++ {
			key := fmt.Sprintf("key%d", i)
			val, err := client.Get(ctx, key).Int()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal(i))
		}
	})

	It("should return error after close", func() {
		Expect(ap.Close()).NotTo(HaveOccurred())

		// Commands after close should fail
		cmd := ap.Set(ctx, "key", "value", 0)
		Expect(cmd.Err()).To(Equal(redis.ErrClosed))
	})

	It("should respect MaxConcurrentBatches", func() {
		// Create autopipeliner with low concurrency limit
		opt := redisOptions()
		opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
			MaxBatchSize:         5,
			MaxFlushDelay:        10 * time.Millisecond,
			MaxConcurrentBatches: 2,
			Unordered:            true,
		}
		client2 := redis.NewClient(opt)
		defer client2.Close()

		ap2, err := client2.AutoPipeline()
		Expect(err).NotTo(HaveOccurred())
		defer ap2.Close()

		// Queue many commands to trigger multiple batches concurrently
		var wg sync.WaitGroup
		for i := 0; i < 50; i++ {
			wg.Add(1)
			go func(idx int) {
				defer wg.Done()
				key := fmt.Sprintf("key%d", idx)
				ap2.Set(ctx, key, idx, 0)
			}(i)
		}

		// Wait for all commands to complete
		wg.Wait()

		// Verify all commands were executed
		for i := 0; i < 50; i++ {
			key := fmt.Sprintf("key%d", i)
			val, err := client2.Get(ctx, key).Int()
			Expect(err).NotTo(HaveOccurred())
			Expect(val).To(Equal(i))
		}
	})

	It("should report queue length", func() {
		// Deferred face with a wide explicit window: queued commands stay
		// queued until the window elapses, so Len is deterministic.
		aap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
			MaxBatchSize:  100,
			MaxFlushDelay: time.Second,
		})
		Expect(err).NotTo(HaveOccurred())
		defer aap.Close()

		Expect(aap.Len()).To(Equal(0))

		cmds := make([]*redis.StatusCmd, 0, 5)
		for i := 0; i < 5; i++ {
			cmds = append(cmds, aap.Set(ctx, fmt.Sprintf("key%d", i), i, 0)) // queued, non-blocking
		}
		Expect(aap.Len()).To(Equal(5))

		for _, c := range cmds {
			Expect(c.Err()).NotTo(HaveOccurred()) // blocks until the window flushes
		}
		Expect(aap.Len()).To(Equal(0))
	})
})

func TestAutoPipelineBasic(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
		MaxBatchSize:         10,
		MaxFlushDelay:        50 * time.Millisecond,
		MaxConcurrentBatches: 5,
		Unordered:            true,
	}
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Queue commands concurrently
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			key := fmt.Sprintf("key%d", idx)
			cmd := ap.Set(ctx, key, idx, 0)
			// Wait for command to complete
			_ = cmd.Err()
		}(i)
	}

	// Wait for all commands to complete
	wg.Wait()

	// Verify
	for i := 0; i < 100; i++ {
		key := fmt.Sprintf("key%d", i)
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get key%d: %v", i, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d", i, val)
		}
	}
}

func TestAutoPipelineMaxFlushDelay(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = &redis.AutoPipelineOptions{
		MaxBatchSize:         1000, // Large batch size so only timer triggers flush
		MaxFlushDelay:        50 * time.Millisecond,
		MaxConcurrentBatches: 5,
		Unordered:            true,
	}
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()
	rtt := apBaselineRTT(ctx, client)

	// Send commands slowly, one at a time
	// They should be flushed by the timer, not by batch size
	start := time.Now()

	var wg sync.WaitGroup
	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			key := fmt.Sprintf("key%d", idx)
			cmd := ap.Set(ctx, key, idx, 0)
			// Wait for command to complete
			_ = cmd.Err()
		}(i)

		// Wait a bit between commands to ensure they don't batch by size
		time.Sleep(10 * time.Millisecond)
	}

	wg.Wait()
	elapsed := time.Since(start)

	// Verify all keys were set
	for i := 0; i < 10; i++ {
		key := fmt.Sprintf("key%d", i)
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get %s: %v", key, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d", i, val)
		}
	}

	// Roughly 10*10ms sends + a few flush intervals; if the timer doesn't work,
	// commands would wait until Close(). Bound scales to RTT (500ms floor).
	bound := apLatencyBudget(rtt, 10, 500*time.Millisecond)
	if elapsed > bound {
		t.Fatalf("Commands took too long (%v, bound %v, rtt %v), flush interval may not be working", elapsed, bound, rtt)
	}

	t.Logf("Completed in %v (flush interval working correctly)", elapsed)
}

func TestAutoPipelineConcurrency(t *testing.T) {
	ctx := context.Background()

	opt := apTestOptions()
	opt.AutoPipelineOptions = redis.DefaultAutoPipelineOptions()
	client := redis.NewClient(opt)
	defer client.Close()

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const numGoroutines = 100
	const commandsPerGoroutine = 100

	var wg sync.WaitGroup
	var failures atomic.Int64

	wg.Add(numGoroutines)
	for g := 0; g < numGoroutines; g++ {
		go func(goroutineID int) {
			defer wg.Done()
			for i := 0; i < commandsPerGoroutine; i++ {
				key := fmt.Sprintf("g%d:key%d", goroutineID, i)
				if err := ap.Set(ctx, key, i, 0).Err(); err != nil {
					failures.Add(1)
				}
			}
		}(g)
	}

	wg.Wait()

	if n := failures.Load(); n != 0 {
		t.Fatalf("%d of %d Sets failed", n, numGoroutines*commandsPerGoroutine)
	}
	// Blocking-face Sets have executed by the time they return — verify the
	// server actually holds every key.
	size, err := client.DBSize(ctx).Result()
	if err != nil {
		t.Fatal(err)
	}
	if expected := int64(numGoroutines * commandsPerGoroutine); size != expected {
		t.Fatalf("DBSize = %d, want %d", size, expected)
	}
}

// TestAutoPipelineSingleCommandNoBlock verifies that single commands don't block
func TestAutoPipelineSingleCommandNoBlock(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = redis.DefaultAutoPipelineOptions()
	client := redis.NewClient(opt)
	defer client.Close()

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Warm up: the first-ever command pays dial + protocol handshake, which
	// is not what this test bounds.
	if err := ap.Ping(ctx).Err(); err != nil {
		t.Fatal(err)
	}
	rtt := apBaselineRTT(ctx, client)

	start := time.Now()
	cmd := ap.Ping(ctx)
	err = cmd.Err()
	elapsed := time.Since(start)

	if err != nil {
		t.Fatalf("Command failed: %v", err)
	}
	if cmd.Val() != "PONG" {
		t.Fatalf("Ping = %q, want PONG", cmd.Val())
	}

	// A warmed lone caller flushes immediately (no coalescing wait): one round
	// trip. The bound scales to the measured RTT (50ms floor for localhost).
	bound := apLatencyBudget(rtt, 1, 50*time.Millisecond)
	if elapsed > bound {
		t.Errorf("Single command took too long: %v (bound %v, rtt %v)", elapsed, bound, rtt)
	}

	t.Logf("Single command completed in %v", elapsed)
}

// TestAutoPipelineSequentialSingleThread verifies sequential single-threaded execution
func TestAutoPipelineSequentialSingleThread(t *testing.T) {
	ctx := context.Background()
	opt := apTestOptions()
	opt.AutoPipelineOptions = redis.DefaultAutoPipelineOptions()
	client := redis.NewClient(opt)
	defer client.Close()

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Warm up dial + handshake so the loop times steady-state round trips.
	if err := ap.Ping(ctx).Err(); err != nil {
		t.Fatal(err)
	}
	rtt := apBaselineRTT(ctx, client)

	// Execute 10 commands sequentially in a single thread
	start := time.Now()
	for i := 0; i < 10; i++ {
		key := fmt.Sprintf("test:key:%d", i)
		t.Logf("Sending command %d", i)
		cmd := ap.Set(ctx, key, i, 0)
		t.Logf("Waiting for command %d to complete", i)
		err := cmd.Err()
		if err != nil {
			t.Fatalf("Command %d failed: %v", i, err)
		}
		t.Logf("Command %d completed", i)
	}
	elapsed := time.Since(start)

	// Ten sequential round trips: the bound scales to the measured RTT so this
	// holds against a remote endpoint, with a 100ms floor to keep localhost tight.
	bound := apLatencyBudget(rtt, 10, 100*time.Millisecond)
	if elapsed > bound {
		t.Errorf("10 sequential commands took too long: %v (bound %v, rtt %v)", elapsed, bound, rtt)
	}

	t.Logf("10 sequential commands completed in %v (avg: %v per command)", elapsed, elapsed/10)
}

// ===== from autopipeline_typed_test.go =====
// TestAutoPipelineTypedCommands tests that typed commands like Get() and Set()
// work correctly with autopipelining and block until execution.
func TestAutoPipelineTypedCommands(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()

	// Skip if Redis is not available
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("Redis not available: %v", err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Test Set and Get
	setResult, setErr := ap.Set(ctx, "test_key", "test_value", 0).Result()
	if setErr != nil {
		t.Fatalf("Set failed: %v", setErr)
	}
	if setResult != "OK" {
		t.Errorf("Expected 'OK', got '%s'", setResult)
	}

	getResult, getErr := ap.Get(ctx, "test_key").Result()
	if getErr != nil {
		t.Fatalf("Get failed: %v", getErr)
	}
	if getResult != "test_value" {
		t.Errorf("Expected 'test_value', got '%s'", getResult)
	}

	// Clean up
	client.Del(ctx, "test_key")
}

// TestAutoPipelineTypedCommandsMultiple tests multiple typed commands
func TestAutoPipelineTypedCommandsMultiple(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()

	// Skip if Redis is not available
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("Redis not available: %v", err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Queue multiple Set commands
	ap.Set(ctx, "key1", "value1", 0)
	ap.Set(ctx, "key2", "value2", 0)
	ap.Set(ctx, "key3", "value3", 0)

	// Get the values - should block until execution
	val1, err1 := ap.Get(ctx, "key1").Result()
	if err1 != nil {
		t.Fatalf("Get key1 failed: %v", err1)
	}
	if val1 != "value1" {
		t.Errorf("Expected 'value1', got '%s'", val1)
	}

	val2, err2 := ap.Get(ctx, "key2").Result()
	if err2 != nil {
		t.Fatalf("Get key2 failed: %v", err2)
	}
	if val2 != "value2" {
		t.Errorf("Expected 'value2', got '%s'", val2)
	}

	val3, err3 := ap.Get(ctx, "key3").Result()
	if err3 != nil {
		t.Fatalf("Get key3 failed: %v", err3)
	}
	if val3 != "value3" {
		t.Errorf("Expected 'value3', got '%s'", val3)
	}

	// Clean up
	client.Del(ctx, "key1", "key2", "key3")
}

// TestAutoPipelineTypedCommandsVal tests the Val() method
func TestAutoPipelineTypedCommandsVal(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()

	// Skip if Redis is not available
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("Redis not available: %v", err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Set a value
	ap.Set(ctx, "test_val_key", "test_val_value", 0)

	// Get using Val() - should block until execution
	val := ap.Get(ctx, "test_val_key").Val()
	if val != "test_val_value" {
		t.Errorf("Expected 'test_val_value', got '%s'", val)
	}

	// Clean up
	client.Del(ctx, "test_val_key")
}

// ===== from autopipeline_universal_test.go =====
// TestUniversalClientAutoPipeline verifies AutoPipeline is reachable through the
// UniversalClient interface: a standalone/cluster-backed UniversalClient returns
// a working AutoPipeliner, while Ring (also a UniversalClient) returns an error
// rather than being silently absent.
func TestUniversalClientAutoPipeline(t *testing.T) {
	ctx := context.Background()

	var uc redis.UniversalClient = redis.NewUniversalClient(&redis.UniversalOptions{
		Addrs: []string{apTestAddr()},
	})
	defer uc.Close()
	if err := uc.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	ap, err := uc.AutoPipeline()
	if err != nil {
		t.Fatalf("UniversalClient.AutoPipeline: %v", err)
	}
	defer ap.Close()
	if v := ap.Set(ctx, "uc:k", "v", 0).Val(); v != "OK" {
		t.Fatalf("set via UniversalClient autopipeline = %q, want OK", v)
	}
	if v := ap.Get(ctx, "uc:k").Val(); v != "v" {
		t.Fatalf("get via UniversalClient autopipeline = %q, want v", v)
	}
	uc.Del(ctx, "uc:k")

	// Ring: AutoPipeline is unsupported and must return an error (directly and
	// through the UniversalClient interface).
	ring := redis.NewRing(&redis.RingOptions{Addrs: map[string]string{"s1": apTestAddr()}})
	defer ring.Close()
	if _, err := ring.AutoPipeline(); err == nil {
		t.Fatal("Ring.AutoPipeline should return an error (unsupported)")
	}
	if _, err := ring.AsyncAutoPipeline(); err == nil {
		t.Fatal("Ring.AsyncAutoPipeline should return an error (unsupported)")
	}
	var ru redis.UniversalClient = ring
	if _, err := ru.AutoPipeline(); err == nil {
		t.Fatal("Ring via UniversalClient.AutoPipeline should return an error")
	}
}

// TestAutoPipelinerAsUniversalClient verifies an *AutoPipeliner is a drop-in
// UniversalClient: data commands are batched through it, while the non-batched
// surface (PoolStats, Do, ...) delegates to the underlying client.
func TestAutoPipelinerAsUniversalClient(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	apc, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer apc.Close()

	// Use the autopipeliner strictly through the UniversalClient interface.
	var uc redis.UniversalClient = apc

	// Data commands: batched by the autopipeliner.
	if v := uc.Set(ctx, "ucap:k", "v", 0).Val(); v != "OK" {
		t.Fatalf("Set = %q, want OK", v)
	}
	if v := uc.Get(ctx, "ucap:k").Val(); v != "v" {
		t.Fatalf("Get = %q, want v", v)
	}
	// Non-batched surface delegates to the underlying client.
	if uc.PoolStats() == nil {
		t.Fatal("PoolStats() returned nil via UniversalClient")
	}
	if err := uc.Do(ctx, "PING").Err(); err != nil {
		t.Fatalf("Do(PING): %v", err)
	}
	uc.Del(ctx, "ucap:k")
}

// ===== from osscluster_autopipeline_buffer_test.go =====
// Autopipeline + zero-copy buffer on CLUSTER (cross-slot, concurrent).
func TestAPZeroCopyBufferCluster(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:               []string{":16600", ":16601", ":16602", ":16603", ":16604", ":16605"},
		AutoPipelineOptions: &redis.AutoPipelineOptions{MaxBatchSize: 100, MaxConcurrentBatches: 30, Unordered: true},
	})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)
	_ = c.ForEachMaster(ctx, func(ctx context.Context, m *redis.Client) error { return m.FlushAll(ctx).Err() })
	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const g = 80
	const perG = 30
	var wg sync.WaitGroup
	var bad int64
	var mu sync.Mutex
	var errs []string
	wg.Add(g)
	for x := 0; x < g; x++ {
		go func(id int) {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				key := fmt.Sprintf("clbuf:%d:%d:%d", id, i, id*7919+i) // spread slots
				val := fmt.Sprintf("p-%d-%d", id, i)
				if err := ap.SetFromBuffer(ctx, key, []byte(val)).Err(); err != nil {
					atomic.AddInt64(&bad, 1)
					mu.Lock()
					if len(errs) < 10 {
						errs = append(errs, "set:"+err.Error())
					}
					mu.Unlock()
					continue
				}
				rbuf := make([]byte, len(val))
				cmd := ap.GetToBuffer(ctx, key, rbuf)
				n, err := cmd.Result()
				if err != nil || n != len(val) || string(cmd.Bytes()) != val {
					atomic.AddInt64(&bad, 1)
					mu.Lock()
					if len(errs) < 10 {
						errs = append(errs, fmt.Sprintf("key %s n=%d got=%q want=%q err=%v", key, n, string(cmd.Bytes()), val, err))
					}
					mu.Unlock()
				}
			}
		}(x)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("cluster zero-copy buffer via autopipeline failed: %d; sample=%v", bad, errs)
	}
}

// ===== from osscluster_autopipeline_correctness_test.go =====
func newTestCluster() *redis.ClusterClient {
	return redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602", ":16603", ":16604", ":16605"},
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize: 200, MaxConcurrentBatches: 50, Unordered: true,
		},
	})
}

// skipIfClusterUnhealthy skips the test unless a real, formed cluster answers.
// Ping alone is insufficient: in CI's standalone test job a Redis may answer on
// the cluster ports while the cluster is not formed (CLUSTER INFO reports
// cluster_state:fail / CLUSTERDOWN, and commands return MOVED), which would turn
// "no cluster available" into spurious failures. Gate on cluster_state:ok.
func skipIfClusterUnhealthy(t *testing.T, c *redis.ClusterClient) {
	t.Helper()
	ctx := context.Background()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("cluster not reachable: %v", err)
	}
	info, err := c.ClusterInfo(ctx).Result()
	if err != nil {
		t.Skipf("cluster not reachable (CLUSTER INFO): %v", err)
	}
	if !strings.Contains(info, "cluster_state:ok") {
		t.Skipf("cluster not healthy (no cluster_state:ok)")
	}
}

// Validates cluster routing: a single autopipeline batch contains keys that
// hash to MANY different slots/shards. Each command must route to the correct
// node and return its own correct value. If routing were wrong, GETs would
// miss or return another key's value.
func TestAPClusterCrossSlotRouting(t *testing.T) {
	ctx := context.Background()
	c := newTestCluster()
	defer c.Close()
	skipIfClusterUnhealthy(t, c)
	_ = c.ForEachMaster(ctx, func(ctx context.Context, m *redis.Client) error {
		return m.FlushAll(ctx).Err()
	})

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 100
	const perG = 50
	var wg sync.WaitGroup
	var mismatch, slotsSeen sync.Map
	var bad int64
	var mu sync.Mutex

	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			for i := 0; i < perG; i++ {
				// Keys with no hashtag => spread across all 16384 slots.
				key := fmt.Sprintf("cl:%d:%d:%d", id, i, id*7919+i)
				want := fmt.Sprintf("V-%d-%d", id, i)
				slotsSeen.Store(hashtag.Slot(key), true)
				if err := ap.Set(ctx, key, want, 0).Err(); err != nil {
					mu.Lock()
					bad++
					mismatch.Store(key, "set:"+err.Error())
					mu.Unlock()
					continue
				}
				got, err := ap.Get(ctx, key).Result()
				if err != nil || got != want {
					mu.Lock()
					bad++
					mismatch.Store(key, fmt.Sprintf("got=%q want=%q err=%v", got, want, err))
					mu.Unlock()
				}
			}
		}(g)
	}
	wg.Wait()

	if bad > 0 {
		shown := 0
		mismatch.Range(func(k, v any) bool {
			t.Logf("MISMATCH %v: %v", k, v)
			shown++
			return shown < 15
		})
		t.Fatalf("cluster cross-slot routing failures: %d", bad)
	}

	// Sanity: we actually exercised many distinct slots (not all one shard).
	n := 0
	slotsSeen.Range(func(_, _ any) bool { n++; return true })
	if n < 100 {
		t.Fatalf("test too weak: only %d distinct slots exercised", n)
	}
	t.Logf("validated cross-slot routing across %d distinct slots", n)
}

// Validates per-key correctness on cluster with INCR ordering per goroutine,
// each goroutine on its own (randomly-slotted) key.
func TestAPClusterPerGoroutineOrder(t *testing.T) {
	ctx := context.Background()
	c := newTestCluster()
	defer c.Close()
	skipIfClusterUnhealthy(t, c)
	// Start from a clean slate so INCR counts are deterministic across runs.
	_ = c.ForEachMaster(ctx, func(ctx context.Context, m *redis.Client) error {
		return m.FlushAll(ctx).Err()
	})

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const goroutines = 80
	const incrs = 80
	var wg sync.WaitGroup
	var bad int64
	var mu sync.Mutex
	wg.Add(goroutines)
	for g := 0; g < goroutines; g++ {
		go func(id int) {
			defer wg.Done()
			key := fmt.Sprintf("clord:%d:%d", id, id*104729)
			for i := 1; i <= incrs; i++ {
				v, err := ap.Incr(ctx, key).Result()
				if err != nil || v != int64(i) {
					mu.Lock()
					bad++
					mu.Unlock()
				}
			}
		}(g)
	}
	wg.Wait()
	if bad > 0 {
		t.Fatalf("cluster ordering/correctness failures: %d", bad)
	}
}

// ===== from osscluster_autopipeline_moved_test.go =====
// TestAPClusterMovedRedirectFollowed verifies AutoPipeline inherits the cluster
// MOVED-redirect handling (it dispatches through ClusterClient.processPipeline,
// which runs the MaxRedirects loop + checkMovedErr). SwapNodes forces the client
// to believe a replica is the master for a slot; a write there is answered with
// MOVED and the client must follow it, so the value is still read back correctly.
func TestAPClusterMovedRedirectFollowed(t *testing.T) {
	ctx := context.Background()
	c := newTestCluster()
	defer c.Close()
	skipIfClusterUnhealthy(t, c)

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const key, val = "moved:key", "moved:val"
	if err := ap.Set(ctx, key, val, 0).Err(); err != nil {
		t.Fatalf("initial set: %v", err)
	}
	// Force the client's in-memory topology to point the slot at the wrong node,
	// so the next access is answered with MOVED and must be followed.
	if err := c.SwapNodes(ctx, key); err != nil {
		t.Fatalf("SwapNodes: %v", err)
	}

	got, err := ap.Get(ctx, key).Result()
	if err != nil {
		t.Fatalf("get after MOVED: %v", err)
	}
	if got != val {
		t.Fatalf("get after MOVED = %q, want %q (redirect not followed correctly)", got, val)
	}
}

// TestAPClusterMovedSurfacedWithoutRedirects anchors the positive test: with
// MaxRedirects disabled (-1 -> 0, one attempt), the forced MOVED is NOT followed
// and must surface as an error on the command's future — proving SwapNodes really
// induced a redirect (so TestAPClusterMovedRedirectFollowed exercised the
// redirect path rather than passing trivially).
func TestAPClusterMovedSurfacedWithoutRedirects(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        []string{":16600", ":16601", ":16602", ":16603", ":16604", ":16605"},
		MaxRedirects: -1, // normalized to 0: a single attempt, no follow-through
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize: 200, MaxConcurrentBatches: 50, Unordered: true,
		},
	})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const key = "moved:noredir:key"
	if err := ap.Set(ctx, key, "v", 0).Err(); err != nil {
		t.Fatalf("initial set: %v", err)
	}

	// SwapNodes is racy against the periodic LazyReload, so re-swap and retry
	// until we observe the MOVED (or give up and skip — the swap never stuck).
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if err := c.SwapNodes(ctx, key); err != nil {
			t.Fatalf("SwapNodes: %v", err)
		}
		err := ap.Get(ctx, key).Err()
		if err != nil && strings.Contains(err.Error(), "MOVED") {
			return // proven: the redirect was needed and, with no redirects, surfaced
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Skip("could not observe a MOVED (swap kept being reverted by topology reload)")
}

// ===== from osscluster_autopipeline_test.go =====
func TestClusterAutoPipelineBasic(t *testing.T) {
	ctx := context.Background()

	client := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602"},
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize:         10,
			MaxFlushDelay:        50 * time.Millisecond,
			MaxConcurrentBatches: 5,
			Unordered:            true,
		},
	})
	defer client.Close()

	skipIfClusterUnhealthy(t, client)

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Queue commands concurrently across different keys (different slots)
	var wg sync.WaitGroup
	for i := 0; i < 100; i++ {
		wg.Add(1)
		go func(idx int) {
			defer wg.Done()
			key := fmt.Sprintf("key%d", idx)
			cmd := ap.Set(ctx, key, idx, 0)
			// Wait for command to complete
			_ = cmd.Err()
		}(i)
	}

	// Wait for all commands to complete
	wg.Wait()

	// Verify all keys were set correctly
	for i := 0; i < 100; i++ {
		key := fmt.Sprintf("key%d", i)
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get %s: %v", key, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d", i, val)
		}
	}
}

func TestClusterAutoPipelineConcurrency(t *testing.T) {
	ctx := context.Background()

	client := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602"},
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize:         50,
			MaxFlushDelay:        10 * time.Millisecond,
			MaxConcurrentBatches: 10,
			Unordered:            true,
		},
	})
	defer client.Close()

	skipIfClusterUnhealthy(t, client)

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const numGoroutines = 10
	const commandsPerGoroutine = 100

	var wg sync.WaitGroup
	for g := 0; g < numGoroutines; g++ {
		wg.Add(1)
		go func(goroutineID int) {
			defer wg.Done()
			for i := 0; i < commandsPerGoroutine; i++ {
				key := fmt.Sprintf("g%d:key%d", goroutineID, i)
				cmd := ap.Set(ctx, key, i, 0)
				// Wait for command to complete
				_ = cmd.Err()
			}
		}(g)
	}

	// Wait for all commands to complete
	wg.Wait()

	// Verify all commands were executed
	for g := 0; g < numGoroutines; g++ {
		for i := 0; i < commandsPerGoroutine; i++ {
			key := fmt.Sprintf("g%d:key%d", g, i)
			val, err := client.Get(ctx, key).Int()
			if err != nil {
				t.Fatalf("Failed to get %s: %v", key, err)
			}
			if val != i {
				t.Fatalf("Expected %d, got %d for key %s", i, val, key)
			}
		}
	}
}

func TestClusterAutoPipelineCrossSlot(t *testing.T) {
	ctx := context.Background()

	client := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602"},
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize:         20,
			MaxFlushDelay:        10 * time.Millisecond,
			MaxConcurrentBatches: 5,
			Unordered:            true,
		},
	})
	defer client.Close()

	skipIfClusterUnhealthy(t, client)

	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Use keys that will hash to different slots
	keys := []string{
		"user:1000",
		"user:2000",
		"user:3000",
		"session:abc",
		"session:def",
		"cache:foo",
		"cache:bar",
	}

	var wg sync.WaitGroup
	for i, key := range keys {
		wg.Add(1)
		go func(k string, val int) {
			defer wg.Done()
			cmd := ap.Set(ctx, k, val, 0)
			// Wait for command to complete
			_ = cmd.Err()
		}(key, i)
	}

	wg.Wait()

	// Verify all keys were set
	for i, key := range keys {
		val, err := client.Get(ctx, key).Int()
		if err != nil {
			t.Fatalf("Failed to get %s: %v", key, err)
		}
		if val != i {
			t.Fatalf("Expected %d, got %d for key %s", i, val, key)
		}
	}
}

// ===== review-fix coverage additions (deep review 2026-07-13) =====

// shortCircuitHook returns from the chain WITHOUT calling next — the
// circuit-breaker hook shape the Hook contract explicitly allows. The engine
// must surface its error on every command instead of waking waiters with
// silent zero values.
type shortCircuitHook struct{ err error }

func (h shortCircuitHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h shortCircuitHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error { return h.err }
}

func (h shortCircuitHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error { return h.err }
}

// pipelineBreakerHook short-circuits the pipeline chain once armed. It stays
// open until then because connection init itself runs a handshake pipeline
// through the hook chain (initConn -> Pipeline.Exec), and plain commands
// (Ping skip-gates, topology discovery) always pass through.
type pipelineBreakerHook struct {
	err   error
	armed *atomic.Bool
}

func (h pipelineBreakerHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h pipelineBreakerHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error { return next(ctx, cmd) }
}

func (h pipelineBreakerHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if !h.armed.Load() {
			return next(ctx, cmds)
		}
		return h.err
	}
}

func TestAutoPipelineHookShortCircuit(t *testing.T) {
	errBreaker := errors.New("breaker open")
	for _, tc := range []struct {
		name  string
		async bool
	}{{"blocking", false}, {"async", true}} {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			if err := c.Ping(ctx).Err(); err != nil {
				t.Skipf("no redis: %v", err)
			}
			c.AddHook(shortCircuitHook{err: errBreaker})

			var ap *redis.AutoPipeliner
			var err error
			if tc.async {
				ap, err = c.AsyncAutoPipeline()
			} else {
				ap, err = c.AutoPipeline()
			}
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				// Batch path: several concurrent commands so they flush as a
				// pipeline batch through the short-circuiting chain.
				const N = 20
				var wg sync.WaitGroup
				errs := make([]error, N)
				wg.Add(N)
				for i := 0; i < N; i++ {
					go func(i int) {
						defer wg.Done()
						errs[i] = ap.Set(ctx, fmt.Sprintf("sc:%d", i), i, 0).Err()
					}(i)
				}
				wg.Wait()
				for i, err := range errs {
					if !errors.Is(err, errBreaker) {
						t.Errorf("cmd %d: err = %v, want the hook's error", i, err)
					}
				}
				// Solo path: a lone command takes the Process fast path.
				if err := ap.Set(ctx, "sc:solo", 1, 0).Err(); !errors.Is(err, errBreaker) {
					t.Errorf("solo: err = %v, want the hook's error", err)
				}
			})
		})
	}
}

// TestAutoPipelineHookAddedAfterCreate pins that hooks installed AFTER the
// autopipeliner exists (directly on the client and via the AutoPipeliner's
// delegating AddHook) take effect and may read results without deadlocking.
func TestAutoPipelineHookAddedAfterCreate(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	ap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	runWithWatchdog(t, 30*time.Second, func() {
		// Round 1: no hooks yet.
		if err := ap.Set(ctx, "late:0", "v0", 0).Err(); err != nil {
			t.Fatal(err)
		}
		// Round 2: result-reading hook added on the client after creation.
		c.AddHook(resultPeekHook{before: false})
		if v, err := ap.Get(ctx, "late:0").Result(); err != nil || v != "v0" {
			t.Fatalf("after client AddHook: v=%q err=%v", v, err)
		}
		// Round 3: another added through the AutoPipeliner's delegate.
		ap.AddHook(resultPeekHook{before: true})
		cmds := make([]*redis.StatusCmd, 0, 30)
		for i := 0; i < 30; i++ {
			cmds = append(cmds, ap.Set(ctx, fmt.Sprintf("late:%d", i), i, 0))
		}
		for i, cmd := range cmds {
			if err := cmd.Err(); err != nil {
				t.Fatalf("cmd %d after ap.AddHook: %v", i, err)
			}
		}
	})
}

// TestAutoPipelineCloseFlushesPendingFutures pins Close's contract: deferred
// commands submitted but not yet flushed are executed (not dropped, never
// hung) by the shutdown drain.
func TestAutoPipelineCloseFlushesPendingFutures(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	if err := c.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	// Wide window + deep batch: nothing flushes on its own before Close.
	ap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  1000,
		MaxFlushDelay: 500 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		const N = 50
		cmds := make([]*redis.StatusCmd, 0, N)
		for i := 0; i < N; i++ {
			cmds = append(cmds, ap.Set(ctx, fmt.Sprintf("pend:%d", i), i, 0))
		}
		if err := ap.Close(); err != nil {
			t.Fatal(err)
		}
		for i, cmd := range cmds {
			if err := cmd.Err(); err != nil {
				t.Fatalf("pending cmd %d after Close: %v", i, err)
			}
		}
		n, err := c.DBSize(ctx).Result()
		if err != nil {
			t.Fatal(err)
		}
		if n != N {
			t.Fatalf("DBSize = %d, want %d (Close must flush pending commands)", n, N)
		}
	})
}

// TestAutoPipelineClientLevelOptionsApply pins that Options.AutoPipelineOptions
// actually configures both faces (nothing previously failed if it were ignored).
func TestAutoPipelineClientLevelOptionsApply(t *testing.T) {
	c := redis.NewClient(&redis.Options{
		Addr:                apTestAddr(),
		AutoPipelineOptions: &redis.AutoPipelineOptions{MaxBatchSize: 123},
	})
	defer c.Close()

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if got := ap.Config().MaxBatchSize; got != 123 {
		t.Fatalf("blocking face MaxBatchSize = %d, want 123 (client-level options ignored)", got)
	}
	aap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if got := aap.Config().MaxBatchSize; got != 123 {
		t.Fatalf("async face MaxBatchSize = %d, want 123 (client-level options ignored)", got)
	}
}

// TestAutoPipelineMaxBatchBytesDefaultsToGuardrail pins that an unset (or
// explicitly zero) MaxBatchBytes resolves to the 128 KiB full-duplex deadlock
// guardrail (see the field's GoDoc), not to "unbounded" — the previous
// behavior. Checked via both preset constructors and the actual resolved
// config on a constructed autopipeliner, since MaxBatchSize's own default is
// applied redundantly in three places and MaxBatchBytes now mirrors that.
func TestAutoPipelineMaxBatchBytesDefaultsToGuardrail(t *testing.T) {
	const want = 128 * 1024

	if got := redis.DefaultAutoPipelineOptions().MaxBatchBytes; got != want {
		t.Errorf("DefaultAutoPipelineOptions().MaxBatchBytes = %d, want %d", got, want)
	}
	if got := redis.DefaultBlockingAutoPipelineOptions().MaxBatchBytes; got != want {
		t.Errorf("DefaultBlockingAutoPipelineOptions().MaxBatchBytes = %d, want %d", got, want)
	}

	c := redis.NewClient(&redis.Options{
		Addr: apTestAddr(),
		AutoPipelineOptions: &redis.AutoPipelineOptions{
			MaxBatchSize: 50, // leave MaxBatchBytes at its zero value
		},
	})
	defer c.Close()

	ap, err := c.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if got := ap.Config().MaxBatchBytes; got != want {
		t.Fatalf("resolved MaxBatchBytes = %d, want %d (zero must default to the guardrail, not stay unbounded)", got, want)
	}
}

// TestAsyncAutoPipelineBlockingCommand covers the readTimeout diversion on the
// DEFERRED face: BLPOP must bypass the batch (it would stall a whole pipeline)
// and still deliver its result through the usual accessors.
func TestAsyncAutoPipelineBlockingCommand(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	c.Del(ctx, "blk:list")

	ap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	runWithWatchdog(t, 30*time.Second, func() {
		// Value present: BLPOP returns it.
		if err := c.LPush(ctx, "blk:list", "v1").Err(); err != nil {
			t.Fatal(err)
		}
		vals, err := ap.BLPop(ctx, time.Second, "blk:list").Result()
		if err != nil || len(vals) != 2 || vals[1] != "v1" {
			t.Fatalf("BLPop = %v, %v; want [blk:list v1]", vals, err)
		}
		// Empty list: times out with redis.Nil, does not wedge the engine.
		if _, err := ap.BLPop(ctx, 50*time.Millisecond, "blk:list").Result(); err != redis.Nil {
			t.Fatalf("BLPop empty = %v, want redis.Nil", err)
		}
		// The engine still works afterwards.
		if err := ap.Set(ctx, "blk:after", "ok", 0).Err(); err != nil {
			t.Fatal(err)
		}
	})
}

// TestAdaptiveDelayEndToEnd runs real traffic through an autopipeliner
// constructed with AdaptiveDelay (previously only the pure delay math was
// tested; no command had ever executed under this configuration).
func TestAdaptiveDelayEndToEnd(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	if err := c.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  50,
		MaxFlushDelay: 5 * time.Millisecond,
		AdaptiveDelay: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	runWithWatchdog(t, 60*time.Second, func() {
		// Windowed bursts exercise the fill-level-dependent delay branches.
		const rounds, window = 5, 80
		for r := 0; r < rounds; r++ {
			cmds := make([]*redis.StatusCmd, 0, window)
			for i := 0; i < window; i++ {
				cmds = append(cmds, ap.Set(ctx, fmt.Sprintf("ad:%d:%d", r, i), i, 0))
			}
			for i, cmd := range cmds {
				if err := cmd.Err(); err != nil {
					t.Fatalf("round %d cmd %d: %v", r, i, err)
				}
			}
		}
		for r := 0; r < rounds; r++ {
			for i := 0; i < window; i += 17 {
				v, err := ap.Get(ctx, fmt.Sprintf("ad:%d:%d", r, i)).Result()
				if err != nil || v != fmt.Sprint(i) {
					t.Fatalf("round %d get %d: v=%q err=%v", r, i, v, err)
				}
			}
		}
	})
}

// TestClusterPipelineUsesDedicatedPool pins the fix for the review finding
// that cluster pipelines acquired connections via getConn (main pool),
// bypassing the dedicated pipeline pool entirely: node clients inherit the
// Pipeline* options, and processPipelineNode now routes through
// withPipelineConn. Covers the manual pipeline and the autopipeline batch
// path (both converge on processPipelineNode).
func TestClusterPipelineUsesDedicatedPool(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:                  []string{":16600", ":16601", ":16602"},
		PipelineReadBufferSize: 64 * 1024,
	})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)

	runWithWatchdog(t, 60*time.Second, func() {
		// Manual pipeline spanning many slots.
		pipe := c.Pipeline()
		for i := 0; i < 50; i++ {
			pipe.Set(ctx, fmt.Sprintf("cpp:%d", i), i, 0)
		}
		if _, err := pipe.Exec(ctx); err != nil {
			t.Fatalf("pipeline exec: %v", err)
		}

		// Autopipeline batch (same node dispatch seam underneath).
		ap, err := c.AsyncAutoPipeline()
		if err != nil {
			t.Fatal(err)
		}
		defer ap.Close()
		cmds := make([]*redis.StringCmd, 0, 50)
		for i := 0; i < 50; i++ {
			cmds = append(cmds, ap.Get(ctx, fmt.Sprintf("cpp:%d", i)))
		}
		for i, cmd := range cmds {
			if v, err := cmd.Result(); err != nil || v != fmt.Sprint(i) {
				t.Fatalf("get %d: v=%q err=%v", i, v, err)
			}
		}

		// The dedicated pipeline pool on the nodes must have carried traffic.
		var pipelineConns atomic.Uint64 // ForEachMaster runs callbacks concurrently
		err = c.ForEachMaster(ctx, func(ctx context.Context, cl *redis.Client) error {
			if ps := cl.PoolStats().PipelineStats; ps != nil {
				pipelineConns.Add(uint64(ps.TotalConns) + uint64(ps.Hits))
			}
			return nil
		})
		if err != nil {
			t.Fatal(err)
		}
		if pipelineConns.Load() == 0 {
			t.Fatal("dedicated pipeline pool unused: cluster pipelines still bypass it")
		}
	})
}

// TestAutoPipelineDoRawOutsidePipeline pins that DoRaw/DoRawWriteTo do NOT
// ride the batching engine (AutoPipeliner embeds cmdable, so without the
// overrides they would): they follow Do's escape-hatch shape on both faces,
// and hooks reading their results cannot deadlock.
func TestAutoPipelineDoRawOutsidePipeline(t *testing.T) {
	for _, tc := range []struct {
		name  string
		async bool
	}{{"blocking", false}, {"async", true}} {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			if err := c.Ping(ctx).Err(); err != nil {
				t.Skipf("no redis: %v", err)
			}
			c.AddHook(resultPeekHook{before: false}) // reads results in hooks

			var ap *redis.AutoPipeliner
			var err error
			if tc.async {
				ap, err = c.AsyncAutoPipeline()
			} else {
				ap, err = c.AutoPipeline()
			}
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				raw := ap.DoRaw(ctx, "PING")
				if err := raw.Err(); err != nil {
					t.Fatalf("DoRaw: %v", err)
				}
				if v, _ := raw.Result(); !strings.Contains(string(v), "PONG") {
					t.Fatalf("DoRaw PING raw = %q, want PONG in RESP", v)
				}

				var buf bytes.Buffer
				wcmd := ap.DoRawWriteTo(ctx, &buf, "PING")
				if err := wcmd.Err(); err != nil { // async face: blocks until executed
					t.Fatalf("DoRawWriteTo: %v", err)
				}
				if !strings.Contains(buf.String(), "PONG") {
					t.Fatalf("DoRawWriteTo buffer = %q, want PONG in RESP", buf.String())
				}

				// no-args guard mirrors Do
				if err := ap.DoRaw(ctx).Err(); err == nil {
					t.Fatal("DoRaw() with no args: want error")
				}
			})
		})
	}
}

// TestClusterNodeHookShortCircuit pins that a NODE-level hook returning
// without calling next (circuit-breaker shape, installed via OnNewNode) is
// surfaced on every command instead of the cluster pipeline reporting silent
// success for commands that were never sent.
func TestClusterNodeHookShortCircuit(t *testing.T) {
	ctx := context.Background()
	errBreaker := errors.New("node breaker open")
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602"},
	})
	defer c.Close()
	// Install BEFORE any command: OnNewNode only fires for newly created node
	// clients. The breaker is disarmed during warmup (conn-init handshakes
	// are pipelines too) and armed just before the asserted round.
	var armed atomic.Bool
	c.OnNewNode(func(cl *redis.Client) { cl.AddHook(pipelineBreakerHook{err: errBreaker, armed: &armed}) })

	skipIfClusterUnhealthy(t, c)

	runWithWatchdog(t, 60*time.Second, func() {
		// Warm pooled connections on every node so the asserted round doesn't
		// dial fresh ones (whose handshake the armed breaker would kill first).
		warm := c.Pipeline()
		for i := 0; i < 12; i++ {
			warm.Set(ctx, fmt.Sprintf("nsc:%d", i), i, 0)
		}
		if _, err := warm.Exec(ctx); err != nil {
			t.Fatalf("warmup: %v", err)
		}
		armed.Store(true)

		pipe := c.Pipeline()
		cmds := make([]*redis.StatusCmd, 0, 12)
		for i := 0; i < 12; i++ {
			cmds = append(cmds, pipe.Set(ctx, fmt.Sprintf("nsc:%d", i), i, 0))
		}
		_, execErr := pipe.Exec(ctx)
		if execErr == nil {
			t.Fatal("cluster pipeline reported success though node hooks short-circuited")
		}
		for i, cmd := range cmds {
			if cmd.Err() == nil {
				t.Fatalf("cmd %d: err = nil, want the node hook's error surfaced", i)
			}
		}
	})
}

// postNextErrorHook calls next and then returns its own error regardless —
// ofek's #3867 case: a hook's post-next verdict used to be discarded because
// the batch had already completed at the innermost seam.
type postNextErrorHook struct{ err error }

func (h postNextErrorHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h postNextErrorHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		_ = next(ctx, cmd)
		return h.err
	}
}

func (h postNextErrorHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		_ = next(ctx, cmds)
		return h.err
	}
}

// TestAutoPipelineHookPostNextError pins that an error a hook returns AFTER
// calling next (successful exec underneath) reaches every command on both
// faces, across the batch, solo and Do dispatch paths.
func TestAutoPipelineHookPostNextError(t *testing.T) {
	errInjected := errors.New("hook injected after next")
	for _, tc := range []struct {
		name  string
		async bool
	}{{"blocking", false}, {"async", true}} {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClient(apTestOptions())
			defer c.Close()
			if err := c.Ping(ctx).Err(); err != nil {
				t.Skipf("no redis: %v", err)
			}
			c.AddHook(postNextErrorHook{err: errInjected})

			var ap *redis.AutoPipeliner
			var err error
			if tc.async {
				ap, err = c.AsyncAutoPipeline()
			} else {
				ap, err = c.AutoPipeline()
			}
			if err != nil {
				t.Fatal(err)
			}
			defer ap.Close()

			runWithWatchdog(t, 30*time.Second, func() {
				// Batch path: concurrent commands flush as one pipeline.
				const N = 16
				var wg sync.WaitGroup
				errs := make([]error, N)
				wg.Add(N)
				for i := 0; i < N; i++ {
					go func(i int) {
						defer wg.Done()
						errs[i] = ap.Set(ctx, fmt.Sprintf("pn:%d", i), i, 0).Err()
					}(i)
				}
				wg.Wait()
				for i, err := range errs {
					if !errors.Is(err, errInjected) {
						t.Errorf("batch cmd %d: err = %v, want the hook's post-next error", i, err)
					}
				}
				// Solo path (lone command -> Process chain).
				if err := ap.Set(ctx, "pn:solo", 1, 0).Err(); !errors.Is(err, errInjected) {
					t.Errorf("solo: err = %v, want the hook's post-next error", err)
				}
				// Do (escape hatch).
				if err := ap.Do(ctx, "PING").Err(); !errors.Is(err, errInjected) {
					t.Errorf("Do: err = %v, want the hook's post-next error", err)
				}
			})
		})
	}
}

// TestAutoPipelineHookPostNextErrorPartialBatch pins the boundary of the
// post-next rule: when the exec recorded ANY per-command outcome-error
// (here a redis.Nil), those results are the source of truth — a hook's
// post-next error must not rewrite them, exactly as a plain pipeline never
// rewrites per-command results (the hook's verdict goes to the Exec caller,
// which does not exist here).
func TestAutoPipelineHookPostNextErrorPartialBatch(t *testing.T) {
	errInjected := errors.New("hook injected after next")
	ctx := context.Background()
	c := redis.NewClient(apTestOptions())
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	if err := c.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}
	if err := c.Set(ctx, "pnp:present", "v", 0).Err(); err != nil {
		t.Fatal(err)
	}

	// Wide flush window so all three commands deterministically land in ONE
	// pipeline batch (the rule is per-batch: hooks fire per batch).
	ap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  100,
		MaxFlushDelay: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	// Initialize the dedicated pipeline-pool connection BEFORE installing the
	// error-injecting hook: connection init runs its handshake pipeline through
	// the client's hook chain (newConn shares hooksMixin), so a hook that
	// unconditionally injects an error would fail the pipeline conn's init and
	// this test would measure init poisoning instead of the post-next rule.
	// Must be a real multi-command batch — a lone command takes the solo fast
	// path (Process on the main pool) and would not init the pipeline conn.
	w1 := ap.Set(ctx, "pnp:warm", 1, 0)
	w2 := ap.Set(ctx, "pnp:warm2", 1, 0)
	if err := w1.Err(); err != nil {
		t.Fatal(err)
	}
	if err := w2.Err(); err != nil {
		t.Fatal(err)
	}
	c.AddHook(postNextErrorHook{err: errInjected})

	runWithWatchdog(t, 30*time.Second, func() {
		// One batch: a hit, a miss (redis.Nil), and a write.
		hit := ap.Get(ctx, "pnp:present")
		miss := ap.Get(ctx, "pnp:absent")
		set := ap.Set(ctx, "pnp:new", 1, 0)

		if err := miss.Err(); err != redis.Nil {
			t.Errorf("miss: err = %v, want redis.Nil kept (exec outcome wins)", err)
		}
		if v, err := hit.Result(); err != nil || v != "v" {
			t.Errorf("hit: v=%q err=%v, want success kept", v, err)
		}
		if err := set.Err(); err != nil {
			t.Errorf("set: err = %v, want success kept", err)
		}
	})
}

// Regression: the async face with NO hooks must survive a multi-command
// batch whose first reply is an error. pipelineProcessCmds re-reads the
// first command's error after the himport re-issue pass; doing that via
// Err() awaited the command's own still-open batch and self-deadlocked the
// dispatcher (the dispGid guard is unarmed without hooks). A plain redis.Nil
// on the first command was enough to trigger it. rawErr() is the fix.
func TestAsyncAutoPipelineErrorFirstReplyNoHooks(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()

	// Forced single batch: everything lands in one flush, first reply is a
	// redis.Nil (missing key). No hooks anywhere — the default configuration.
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  100,
		MaxFlushDelay: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		cmds := make([]*redis.StringCmd, 10)
		for i := range cmds {
			cmds[i] = ap.Get(ctx, fmt.Sprintf("ap:errfirst:missing:%d", i))
		}
		for i, cmd := range cmds {
			if err := cmd.Err(); err != redis.Nil {
				t.Errorf("cmd %d: got %v, want redis.Nil", i, err)
			}
		}
	})
}

// Regression: HIMPORT commands inside an async batch. himportAfterBatch and
// himportRetryFailedSets run on the execution path (standalone and per-node
// cluster goroutines) and read the batch's commands; doing so via Err()
// awaited the still-open batch and self-deadlocked exactly like the case
// above — even when every command succeeded. Works against servers without
// HIMPORT support too: the reads happen regardless of the reply.
func TestAsyncAutoPipelineHImportBatchNoHooks(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  100,
		MaxFlushDelay: 50 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		set := ap.Set(ctx, "ap:himport:plain", "v", 0)
		h1 := ap.HImportSet(ctx, "ap:himport:h", "fieldset-a", "f1", "v1")
		h2 := ap.HImportSet(ctx, "ap:himport:h", "fieldset-b", "f2", "v2")
		disc := ap.HImportDiscard(ctx, "fieldset-a")

		// The HIMPORT replies depend on server support — only completion
		// matters here (pre-fix this blocked forever). The plain SET in the
		// same batch must succeed.
		_ = h1.Err()
		_ = h2.Err()
		_ = disc.Err()
		if err := set.Err(); err != nil {
			t.Errorf("plain SET in himport batch: %v", err)
		}
	})
}

// Regression: Publish (and XRead's lag metric) read the command result only
// to record OTel metrics; with no recorder installed that read awaited
// execution, silently turning fire-and-forget publishes on the async face
// into blocking calls. Submission of a window must complete well inside the
// flush delay — i.e. without waiting for any result.
func TestAsyncAutoPipelinePublishDoesNotAwait(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  1000,
		MaxFlushDelay: 100 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	start := time.Now()
	cmds := make([]*redis.IntCmd, 100)
	for i := range cmds {
		cmds[i] = ap.Publish(ctx, "ap:pub:chan", i)
	}
	submitted := time.Since(start)
	// Generous bound: submission is pure enqueueing (microseconds); awaiting
	// even one result would cost the 100ms flush delay.
	if submitted > 50*time.Millisecond {
		t.Fatalf("submitting 100 publishes took %v — a result read is awaiting execution", submitted)
	}
	for i, cmd := range cmds {
		if err := cmd.Err(); err != nil {
			t.Fatalf("publish %d: %v", i, err)
		}
	}
}

// routeRecordingHook records which dispatch path each command takes: names
// seen by the pipeline hook rode a batch; names seen by the process hook ran
// directly on their own connection.
type routeRecordingHook struct {
	mu       *sync.Mutex
	pipeline map[string]bool
	direct   map[string]bool
}

func (h routeRecordingHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h routeRecordingHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		h.mu.Lock()
		h.direct[cmd.Name()] = true
		h.mu.Unlock()
		return next(ctx, cmd)
	}
}

func (h routeRecordingHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		h.mu.Lock()
		for _, cmd := range cmds {
			h.pipeline[cmd.Name()] = true
		}
		h.mu.Unlock()
		return next(ctx, cmds)
	}
}

// Regression: SHUTDOWN must never ride a shared pipeline batch — the server
// exits before replying, so every batchmate would fail with EOF and the
// whole batch would be retried against a dead server (review finding by
// cxljs on the pool PR). It runs directly, like a blocking command. The
// client points at a closed port so nothing real is shut down: only the
// routing is asserted, via hooks that record which path each command took.
func TestAutoPipelineShutdownRunsOutsidePipeline(t *testing.T) {
	ln, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		t.Fatal(err)
	}
	deadAddr := ln.Addr().String()
	ln.Close() // guaranteed-refused port

	hook := routeRecordingHook{mu: &sync.Mutex{}, pipeline: map[string]bool{}, direct: map[string]bool{}}
	client := redis.NewClient(&redis.Options{Addr: deadAddr, MaxRetries: -1})
	defer client.Close()
	client.AddHook(hook)

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  10,
		MaxFlushDelay: 10 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	set := ap.Set(context.Background(), "k", "v", 0)
	down := ap.Shutdown(context.Background())
	mon := ap.Monitor(context.Background(), make(chan string, 1))
	_ = set.Err()  // dial refused — only routing matters
	_ = down.Err() // dial refused — only routing matters
	_ = mon.Err()  // dial refused — only routing matters

	hook.mu.Lock()
	defer hook.mu.Unlock()
	for _, name := range []string{"shutdown", "monitor"} {
		if !hook.direct[name] {
			t.Fatalf("%s did not take the direct (outside-pipeline) path", name)
		}
		if hook.pipeline[name] {
			t.Fatalf("%s rode a pipeline batch — it would poison its batchmates", name)
		}
	}
	if !hook.pipeline["set"] && !hook.direct["set"] {
		t.Fatal("SET was never dispatched at all")
	}
}

// armablePostNextPipelineHook calls next and then, when armed, returns its
// own error regardless of the exec outcome — the node-level twin of ofek's
// post-next-verdict case. Disarmed it is transparent, so conn-init
// handshakes (which pipeline through the hook chain) pass during warmup.
type armablePostNextPipelineHook struct {
	err   error
	armed *atomic.Bool
}

func (h armablePostNextPipelineHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h armablePostNextPipelineHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return next
}

func (h armablePostNextPipelineHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		err := next(ctx, cmds)
		if h.armed.Load() {
			return h.err
		}
		return err
	}
}

// TestClusterNodeHookPostNextError pins that a NODE-level hook's post-next
// verdict is applied to an all-clean sub-batch (mirroring dispatchCmds):
// the exec succeeds on the wire, the hook returns an error afterwards, and
// every command must carry that error instead of reporting success
// (suppressed Copilot finding on #3867: the verdict was dropped on the
// cluster node path).
func TestClusterNodeHookPostNextError(t *testing.T) {
	ctx := context.Background()
	errVerdict := errors.New("node post-next verdict")
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{":16600", ":16601", ":16602"},
	})
	defer c.Close()
	var armed atomic.Bool
	c.OnNewNode(func(cl *redis.Client) {
		cl.AddHook(armablePostNextPipelineHook{err: errVerdict, armed: &armed})
	})

	skipIfClusterUnhealthy(t, c)

	runWithWatchdog(t, 60*time.Second, func() {
		warm := c.Pipeline()
		for i := 0; i < 12; i++ {
			warm.Set(ctx, fmt.Sprintf("npn:%d", i), i, 0)
		}
		if _, err := warm.Exec(ctx); err != nil {
			t.Fatalf("warmup: %v", err)
		}
		armed.Store(true)
		defer armed.Store(false)

		pipe := c.Pipeline()
		cmds := make([]*redis.StatusCmd, 0, 12)
		for i := 0; i < 12; i++ {
			cmds = append(cmds, pipe.Set(ctx, fmt.Sprintf("npn:%d", i), i, 0))
		}
		_, _ = pipe.Exec(ctx)
		for i, cmd := range cmds {
			if !errors.Is(cmd.Err(), errVerdict) {
				t.Fatalf("cmd %d: err = %v, want the node hook's post-next verdict", i, cmd.Err())
			}
		}
	})
}

// Regression: ScanIterator on the deferred (async) face. Next() used to read
// the command's page/cursor fields without awaiting execution — the fetch
// only enqueues on this face — so it spun re-issuing SCANs with a stale
// cursor forever, flooding the engine with an unbounded batch and hanging
// the caller. Next() now awaits the fetch before reading.
func TestAsyncAutoPipelineScanIterator(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	if err := client.FlushDB(ctx).Err(); err != nil {
		t.Fatal(err)
	}

	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()

	const n = 33
	for i := 0; i < n; i++ {
		if err := ap.Set(ctx, fmt.Sprintf("apiter:%d", i), i, 0).Err(); err != nil {
			t.Fatal(err)
		}
	}

	runWithWatchdog(t, 30*time.Second, func() {
		var got int
		iter := ap.Scan(ctx, 0, "apiter:*", 5).Iterator()
		for iter.Next(ctx) {
			got++
		}
		if err := iter.Err(); err != nil {
			t.Fatalf("iterator: %v", err)
		}
		if got != n {
			t.Fatalf("scanned %d keys, want %d", got, n)
		}
	})
}

// markNonPipelinable installs a resolver marking one command name with a
// fan-out request policy (ReqMultiShard), which cannot ride a pipeline.
func markNonPipelinable(c *redis.ClusterClient, name string) {
	c.SetCommandInfoResolver(redis.NewCommandInfoResolver(func(ctx context.Context, cmd redis.Cmder) *routing.CommandPolicy {
		if cmd.Name() == name {
			return &routing.CommandPolicy{Request: routing.ReqMultiShard}
		}
		return nil
	}))
}

// An explicitly configured custom fan-out policy unsupported by Pipeline is
// rejected before batch dispatch. This validation does not make ordinary
// pipelines atomic; Redis command errors can still accompany successful writes.
func TestClusterPipelineNonPipelinableAllOrNothing(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
	defer c.Close()
	if err := c.Ping(ctx).Err(); err != nil {
		t.Skipf("no cluster: %v", err)
	}
	markNonPipelinable(c, "getdel")

	_ = c.Del(ctx, "aon:{a}:1", "aon:{a}:2").Err()

	pipe := c.Pipeline()
	set1 := pipe.Set(ctx, "aon:{a}:1", "v1", 0)
	bad := pipe.GetDel(ctx, "aon:{a}:marked")
	set2 := pipe.Set(ctx, "aon:{a}:2", "v2", 0)
	_, execErr := pipe.Exec(ctx)
	if execErr == nil {
		t.Fatal("Exec must fail when the batch contains a non-pipelinable command")
	}
	for i, cmd := range []redis.Cmder{set1, bad, set2} {
		if cmd.Err() == nil {
			t.Fatalf("cmd %d: err = nil, want the policy error on every command (all-or-nothing)", i)
		}
	}
	// Nothing may have executed.
	if n, _ := c.Exists(ctx, "aon:{a}:1", "aon:{a}:2").Result(); n != 0 {
		t.Fatalf("writes landed despite the failed batch: %d keys exist", n)
	}
}

// Fan-out is rejected on ordered async streams and remains available with an
// explicit Unordered opt-in. Neither outcome affects neighboring batches.
func TestAPClusterFanoutRespectsOrdering(t *testing.T) {
	for _, tc := range []struct {
		name       string
		fullDuplex bool
		unordered  bool
	}{
		{name: "ordered"},
		{name: "ordered-full-duplex", fullDuplex: true},
		{name: "unordered", unordered: true},
	} {
		t.Run(tc.name, func(t *testing.T) {
			ctx := context.Background()
			c := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
			defer c.Close()
			skipIfClusterUnhealthy(t, c)
			c.SetCommandInfoResolver(redis.NewCommandInfoResolver(func(_ context.Context, cmd redis.Cmder) *routing.CommandPolicy {
				if cmd.Name() == "ping" {
					return &routing.CommandPolicy{Request: routing.ReqAllShards, Response: routing.RespAllSucceeded}
				}
				return nil
			}))

			var hooksMu sync.Mutex
			var hooks []routeRecordingHook
			if err := c.ForEachMaster(ctx, func(_ context.Context, node *redis.Client) error {
				hook := routeRecordingHook{mu: &sync.Mutex{}, pipeline: map[string]bool{}, direct: map[string]bool{}}
				node.AddHook(hook)
				hooksMu.Lock()
				hooks = append(hooks, hook)
				hooksMu.Unlock()
				return nil
			}); err != nil {
				t.Fatal(err)
			}
			if len(hooks) < 2 {
				t.Fatal("fan-out test requires multiple masters")
			}

			ap, err := c.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
				MaxBatchSize:  100,
				MaxFlushDelay: 20 * time.Millisecond,
				FullDuplex:    tc.fullDuplex,
				Unordered:     tc.unordered,
			})
			if err != nil {
				t.Fatal(err)
			}
			key1 := fmt.Sprintf("apfanout:%s:{a}:1", tc.name)
			key2 := fmt.Sprintf("apfanout:%s:{a}:2", tc.name)
			defer c.Del(ctx, key1, key2)

			runWithWatchdog(t, 30*time.Second, func() {
				set1 := ap.Set(ctx, key1, "v1", 0)
				ping := ap.Ping(ctx)
				set2 := ap.Set(ctx, key2, "v2", 0)

				got, err := ping.Result()
				if tc.unordered {
					if err != nil || got != "PONG" {
						t.Fatalf("fan-out PING = %q, %v; want PONG", got, err)
					}
				} else if err == nil || !strings.Contains(err.Error(), "ordered AsyncAutoPipeline") {
					t.Fatalf("fan-out PING = %q, %v; want ordering rejection", got, err)
				}
				if err := set1.Err(); err != nil {
					t.Fatalf("neighboring set1 failed: %v", err)
				}
				if err := set2.Err(); err != nil {
					t.Fatalf("neighboring set2 failed: %v", err)
				}
			})
			for i, hook := range hooks {
				hook.mu.Lock()
				direct, pipelined := hook.direct["ping"], hook.pipeline["ping"]
				hook.mu.Unlock()
				if direct != tc.unordered || pipelined {
					t.Fatalf("master %d: PING direct=%v pipelined=%v; want direct=%v and no pipelining", i, direct, pipelined, tc.unordered)
				}
			}
			values, err := c.MGet(ctx, key1, key2).Result()
			if err != nil || len(values) != 2 || values[0] != "v1" || values[1] != "v2" {
				t.Fatalf("neighboring writes = %v, %v; want [v1 v2]", values, err)
			}
		})
	}
}

// Regression: reading a Submit()-ed command's result accessors BEFORE calling
// Wait() must be safe — Submit now marks the command ready like every other
// async entry point, so Err()/Val() self-gate through await(). Previously
// this was a silent data race with the dispatch goroutine (this test trips
// -race on the unfixed code).
func TestAutoPipelineSubmitReadBeforeWait(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  100,
		MaxFlushDelay: 10 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		for i := 0; i < 200; i++ {
			cmd := redis.NewStatusCmd(ctx, "set", fmt.Sprintf("subrace:%d", i), i)
			_ = ap.Submit(ctx, cmd) // future deliberately ignored
			// Misuse-but-must-be-safe: read the command directly, no Wait().
			if err := cmd.Err(); err != nil {
				t.Fatalf("cmd %d: %v", i, err)
			}
		}
	})
}

// batchSizeRecordingHook records the command count of every dispatch. A
// drain cycle that catches exactly one command takes the engine's SOLO fast
// path (ProcessHook, no pipeline) — timing-dependent on loaded runners — so
// solo dispatches are recorded as size-1 entries to keep the total exact.
type batchSizeRecordingHook struct {
	mu    *sync.Mutex
	sizes *[]int
}

func (h batchSizeRecordingHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h batchSizeRecordingHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		if cmd.Name() == "set" { // only the test's own commands, not handshakes
			h.mu.Lock()
			*h.sizes = append(*h.sizes, 1)
			h.mu.Unlock()
		}
		return next(ctx, cmd)
	}
}

func (h batchSizeRecordingHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		n := 0
		for _, cmd := range cmds {
			if cmd.Name() == "set" { // exclude conn-init handshake pipelines
				n++
			}
		}
		if n > 0 {
			h.mu.Lock()
			*h.sizes = append(*h.sizes, n)
			h.mu.Unlock()
		}
		return next(ctx, cmds)
	}
}

// TestAutoPipelineMaxBatchBytes pins the byte-volume flush cap: with large
// payloads and a small MaxBatchBytes, a window that would otherwise coalesce
// into one huge burst is split into several bounded batches (300 x 64KiB in
// one batch is ~19MB written down a single connection before any reply is
// read — enough to stall a constrained link past its write deadline).
func TestAutoPipelineMaxBatchBytes(t *testing.T) {
	ctx := context.Background()
	const payload = 64 * 1024
	value := strings.Repeat("x", payload)

	run := func(maxBytes int) []int {
		client := redis.NewClient(apTestOptions())
		defer client.Close()
		if err := client.Ping(ctx).Err(); err != nil {
			t.Skipf("no redis: %v", err)
		}
		var mu sync.Mutex
		var sizes []int
		client.AddHook(batchSizeRecordingHook{mu: &mu, sizes: &sizes})

		ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
			MaxBatchSize:  100,
			MaxFlushDelay: 30 * time.Millisecond, // would coalesce everything without the byte cap
			MaxBatchBytes: maxBytes,
		})
		if err != nil {
			t.Fatal(err)
		}

		cmds := make([]*redis.StatusCmd, 10)
		for i := range cmds {
			cmds[i] = ap.Set(ctx, fmt.Sprintf("mbb:%d", i), value, 0)
		}
		for i, cmd := range cmds {
			if err := cmd.Err(); err != nil {
				t.Fatalf("cmd %d: %v", i, err)
			}
		}
		mu.Lock()
		defer mu.Unlock()
		out := make([]int, len(sizes))
		copy(out, sizes)
		return out
	}

	runWithWatchdog(t, 60*time.Second, func() {
		capped := run(150 * 1024) // ~2 x 64KiB per batch
		total, maxBatch := 0, 0
		for _, n := range capped {
			total += n
			if n > maxBatch {
				maxBatch = n
			}
		}
		if total != 10 {
			t.Fatalf("capped run dispatched %d commands, want 10", total)
		}
		if len(capped) < 4 {
			t.Fatalf("capped run used %d batches (%v), want the byte cap to split into >= 4", len(capped), capped)
		}
		if maxBatch > 3 {
			t.Fatalf("capped run had a %d-command batch (%v) — cap not enforced", maxBatch, capped)
		}

		// 0 no longer means "uncapped" (it now gets the 128 KiB default
		// guardrail — see MaxBatchBytes's GoDoc); opt out explicitly with a
		// value comfortably above this run's total payload (10 x 64 KiB).
		uncapped := run(10 * 1024 * 1024)
		utotal := 0
		for _, n := range uncapped {
			utotal += n
		}
		if utotal != 10 {
			t.Fatalf("uncapped run dispatched %d commands, want 10", utotal)
		}
		if len(uncapped) > 4 {
			// A slow runner can split a solo straggler off the window; the
			// signal is "far fewer dispatches than the capped run", not an
			// exact count.
			t.Fatalf("uncapped control fragmented into %d batches (%v) — coalescing regressed", len(uncapped), uncapped)
		}
	})
}

// panicPipelineHook panics inside the pipeline hook chain when armed — a
// stand-in for a hook or command-encoder panic during dispatch.
type panicPipelineHook struct{ armed *atomic.Bool }

func (h panicPipelineHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h panicPipelineHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return next
}

func (h panicPipelineHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if h.armed.Load() {
			panic("boom: injected dispatch panic")
		}
		return next(ctx, cmds)
	}
}

// TestAutoPipelinePanicInDispatchWakesWaiters pins the engine's panic
// containment: a panic inside the dispatch (hook chain / exec) must not kill
// the process — on a plain client it unwinds into the caller, but the
// engine's dispatch goroutines have no caller. Every waiter must wake with a
// descriptive error, and the engine must stay usable afterwards.
func TestAutoPipelinePanicInDispatchWakesWaiters(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	var armed atomic.Bool
	client.AddHook(panicPipelineHook{armed: &armed})

	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize:  100,
		MaxFlushDelay: 20 * time.Millisecond,
	})
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		armed.Store(true)
		cmds := make([]*redis.StatusCmd, 5)
		for i := range cmds {
			cmds[i] = ap.Set(ctx, fmt.Sprintf("pan:%d", i), i, 0)
		}
		for i, cmd := range cmds {
			err := cmd.Err() // must wake, not hang, not crash
			if err == nil || !strings.Contains(err.Error(), "panic during dispatch") {
				t.Fatalf("cmd %d: err = %v, want the recovered-panic error", i, err)
			}
		}
		armed.Store(false)

		// Engine survived: the next round executes normally.
		if err := ap.Set(ctx, "pan:after", "ok", 0).Err(); err != nil {
			t.Fatalf("engine unusable after recovered panic: %v", err)
		}
	})
}

// TestAPBlockingContextCancelDoesNotCancelExecution is the blocking-face twin
// of TestAPContextCancelDoesNotCancelExecution: cancelling the caller's ctx
// concurrently with a blocking-face call must not cancel the command's
// execution — the batch runs on the engine's context, and the call returns
// the executed result.
func TestAPBlockingContextCancelDoesNotCancelExecution(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AutoPipeline()
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		cctx, cancel := context.WithCancel(ctx)
		go func() {
			// Cancel while the call is (likely) queued/flushing; the engine
			// dispatches on its own context, so this must be irrelevant.
			time.Sleep(time.Millisecond)
			cancel()
		}()
		if err := ap.Set(cctx, "apbctx:k", "v", 0).Err(); err != nil {
			t.Fatalf("blocking call should return the executed result after ctx cancel, got %v", err)
		}
		if v, _ := client.Get(ctx, "apbctx:k").Result(); v != "v" {
			t.Fatalf("value not persisted: %q", v)
		}
		client.Del(ctx, "apbctx:k")
	})
}

// Regression: a blocking command (BLPOP) submitted on the DEFERRED face must
// not stall the submitter — the call returns immediately and only the result
// accessors block (exactly like Do). Previously the diverted path executed
// inline, so an async ap.BLPop froze the caller for the full block duration.
func TestAsyncAutoPipelineBlockingCommandDoesNotStallSubmit(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		start := time.Now()
		cmd := ap.BLPop(ctx, 500*time.Millisecond, "apblk:empty")
		submitted := time.Since(start)
		if submitted > 100*time.Millisecond {
			t.Fatalf("async BLPop stalled the submitter for %v — must return immediately", submitted)
		}
		if err := cmd.Err(); err != redis.Nil { // times out empty -> Nil
			t.Fatalf("BLPop result: %v, want redis.Nil", err)
		}
		if elapsed := time.Since(start); elapsed < 400*time.Millisecond {
			t.Fatalf("BLPop completed in %v — did not actually block server-side", elapsed)
		}
	})
}

// TestWithTimeoutCloneCloseBlocksParentResurrection pins the shared-pool
// closed flag: closing an ordinary WithTimeout clone closes the SHARED pools,
// so the parent's AutoPipeline getters must return ErrClosed instead of
// building fresh flushers against dead pools.
func TestWithTimeoutCloneCloseBlocksParentResurrection(t *testing.T) {
	ctx := context.Background()
	parent := redis.NewClient(apTestOptions())
	if err := parent.Ping(ctx).Err(); err != nil {
		parent.Close()
		t.Skipf("no redis: %v", err)
	}

	clone := parent.WithTimeout(5 * time.Second)
	if err := clone.Close(); err != nil {
		t.Fatalf("clone close: %v", err)
	}

	if _, err := parent.AutoPipeline(); err != redis.ErrClosed {
		t.Fatalf("parent.AutoPipeline after clone Close = %v, want ErrClosed (shared pools are gone)", err)
	}
	if _, err := parent.AsyncAutoPipeline(); err != redis.ErrClosed {
		t.Fatalf("parent.AsyncAutoPipeline after clone Close = %v, want ErrClosed", err)
	}
	_ = parent.Close()
}

// TestWithTimeoutCloneCloseStopsCachedAutoPipeliner pins the second half of
// the shared-closed contract: an autopipeliner built and CACHED before a
// WithTimeout clone closes the shared pools must refuse new work at submit
// with ErrClosed — not accept enqueues that then fail against dead pools one
// dispatch at a time (codex on #3942).
func TestWithTimeoutCloneCloseStopsCachedAutoPipeliner(t *testing.T) {
	ctx := context.Background()
	parent := redis.NewClient(apTestOptions())
	if err := parent.Ping(ctx).Err(); err != nil {
		parent.Close()
		t.Skipf("no redis: %v", err)
	}
	defer parent.Close()

	ap, err := parent.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	if err := ap.Set(ctx, "cloneclose:pre", "v", 0).Err(); err != nil {
		t.Fatalf("pre-close set: %v", err)
	}

	clone := parent.WithTimeout(5 * time.Second)
	if err := clone.Close(); err != nil {
		t.Fatalf("clone close: %v", err)
	}

	if err := ap.Set(ctx, "cloneclose:post", "v", 0).Err(); err != redis.ErrClosed {
		t.Fatalf("cached autopipeliner Set after clone Close = %v, want ErrClosed", err)
	}
	if err := ap.Do(ctx, "get", "cloneclose:post").Err(); err != redis.ErrClosed {
		t.Fatalf("cached autopipeliner Do after clone Close = %v, want ErrClosed", err)
	}
	// Diverted commands must refuse too, not run against the closed pools.
	if err := ap.Do(ctx, "client", "id").Err(); err != redis.ErrClosed {
		t.Fatalf("diverted command after clone Close = %v, want ErrClosed", err)
	}
}

// TestAPClusterWideCommandsDelegate pins that cluster-wide commands issued on
// the autopipeliner use ClusterClient's fan-out overrides rather than the
// generic single-shard routing (DBSize must sum every master).
func TestAPClusterWideCommandsDelegate(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)
	ap, err := c.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}

	// Keys hashed to different slots so the cluster-wide count differs from
	// any single shard's.
	for i := 0; i < 30; i++ {
		if err := ap.Set(ctx, fmt.Sprintf("apwide:%d", i), i, 0).Err(); err != nil {
			t.Fatal(err)
		}
	}
	want, err := c.DBSize(ctx).Result()
	if err != nil {
		t.Fatal(err)
	}
	got, err := ap.DBSize(ctx).Result()
	if err != nil {
		t.Fatal(err)
	}
	if got != want {
		t.Fatalf("ap.DBSize = %d, cluster-wide DBSize = %d — the autopipeliner must delegate to the fan-out override", got, want)
	}
	sha, err := ap.ScriptLoad(ctx, "return 1").Result()
	if err != nil || sha == "" {
		t.Fatalf("ap.ScriptLoad: %q, %v", sha, err)
	}
	exists, err := ap.ScriptExists(ctx, sha).Result()
	if err != nil || len(exists) != 1 || !exists[0] {
		t.Fatalf("script must exist on EVERY shard after delegated ScriptLoad: %v %v", exists, err)
	}
}

// TestAsyncCommandStringAwaits pins String()'s accessor contract on the
// deferred face: formatting an in-flight command from another goroutine
// blocks until execution — reading the value field without that barrier is a
// data race with reply processing (review finding by ofek and cursor on
// #3942) — and then includes the executed result. The dispatch-goroutine
// case (a hook formatting its own batch) is covered by resultPeekHook, which
// also calls String and relies on the dispatch-gid guard in await.
func TestAsyncCommandStringAwaits(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(apTestOptions())
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	gate := make(chan struct{})
	var armed atomic.Bool
	client.AddHook(gateHook{gate: gate, armed: &armed})

	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}

	runWithWatchdog(t, 30*time.Second, func() {
		armed.Store(true)
		cmd := ap.Set(ctx, "strawait", "value-42", 0)
		sCh := make(chan string, 1)
		go func() { sCh <- cmd.String() }()

		select {
		case s := <-sCh:
			t.Fatalf("String returned %q before the command executed", s)
		case <-time.After(100 * time.Millisecond):
		}

		armed.Store(false)
		close(gate)
		s := <-sCh
		if !strings.Contains(s, "set strawait value-42") || !strings.Contains(s, "OK") {
			t.Fatalf("String after execution = %q, want args and the OK result", s)
		}
	})
}

// gateHook holds every dispatch (solo and pipelined) until gate closes,
// while armed.
type gateHook struct {
	gate  chan struct{}
	armed *atomic.Bool
}

func (h gateHook) DialHook(next redis.DialHook) redis.DialHook { return next }
func (h gateHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
	return func(ctx context.Context, cmd redis.Cmder) error {
		if h.armed.Load() {
			<-h.gate
		}
		return next(ctx, cmd)
	}
}

func (h gateHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if h.armed.Load() {
			<-h.gate
		}
		return next(ctx, cmds)
	}
}

// ===== from autopipeline_fd_shards_test.go =====

// With N full-duplex engines on one client, every reply must
// still be the reply to its own command. Mis-correlation is the failure mode
// that matters -- N engines means N independent FIFO streams, and a command
// routed to engine A whose reply is read off engine B would corrupt data
// silently rather than error.
//
// Requires a local server (apTestAddr: REDIS_PORT, default 6379).

func TestFDShardsReplyCorrelation(t *testing.T) {
	ctx := context.Background()
	cl := redis.NewClient(&redis.Options{Addr: apTestAddr()})
	defer cl.Close()
	if err := cl.Ping(ctx).Err(); err != nil {
		t.Skipf("no local redis: %v", err)
	}

	shards := 8
	if v := os.Getenv("FD_SHARDS"); v != "" {
		// Overridable so the same test can run at ONE engine, which is how the
		// silent half-duplex fallback was first caught: 1 passed, 8 failed.
		shards, _ = strconv.Atoi(v)
	}
	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		FullDuplex:           true,
		NumShards:            shards,
		MaxConcurrentBatches: 1,
	})
	if err != nil {
		t.Fatalf("construct: %v", err)
	}
	defer ap.Close()

	// Each key's value encodes the key, so a reply landing on the wrong command
	// is caught rather than merely suspected.
	const keys = 2000
	for i := 0; i < keys; i++ {
		k := fmt.Sprintf("fdshards:{%d}", i)
		if err := ap.Set(ctx, k, fmt.Sprintf("val-%d", i), 0).Err(); err != nil {
			t.Fatalf("set %d: %v", i, err)
		}
	}

	var wg sync.WaitGroup
	errs := make(chan error, keys)
	for w := 0; w < 64; w++ {
		wg.Add(1)
		go func(w int) {
			defer wg.Done()
			for i := w; i < keys; i += 64 {
				k := fmt.Sprintf("fdshards:{%d}", i)
				want := fmt.Sprintf("val-%d", i)
				got, err := ap.Get(ctx, k).Result()
				if err != nil {
					errs <- fmt.Errorf("get %s: %w", k, err)
					return
				}
				if got != want {
					errs <- fmt.Errorf("MIS-CORRELATED: %s got %q want %q", k, got, want)
					return
				}
			}
		}(w)
	}
	wg.Wait()
	close(errs)
	n := 0
	for e := range errs {
		if n < 5 {
			t.Error(e)
		}
		n++
	}
	if n > 0 {
		t.Fatalf("%d correlation failures across %d keys", n, keys)
	}

	// Per-caller ordering on the deferred face: write then read the SAME key
	// without waiting in between. Key routing must put both on one engine.
	for i := 0; i < 500; i++ {
		k := fmt.Sprintf("fdshards:order:{%d}", i)
		setF := ap.Set(ctx, k, "second", 0)
		getF := ap.Get(ctx, k)
		if err := setF.Err(); err != nil {
			t.Fatalf("ordered set: %v", err)
		}
		got, err := getF.Result()
		if err != nil {
			t.Fatalf("ordered get: %v", err)
		}
		if got != "second" {
			t.Fatalf("ORDER VIOLATION at %s: got %q, want the value written just before it", k, got)
		}
	}
	t.Logf("ok: %d keys correlated, 500 same-key write-then-read ordered, %d engines", keys, shards)
}

// The two-sited rule: NumShards>1 under FullDuplex must actually ENGAGE full
// duplex, not fall back to half-duplex shards. Config() reports the EFFECTIVE
// state, so a fallback surfaces as FullDuplex:false.
func TestFDShardsKeepsFullDuplexOn(t *testing.T) {
	ctx := context.Background()
	cl := redis.NewClient(&redis.Options{Addr: apTestAddr()})
	defer cl.Close()
	if err := cl.Ping(ctx).Err(); err != nil {
		t.Skipf("no local redis: %v", err)
	}
	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		FullDuplex:           true,
		NumShards:            8,
		MaxConcurrentBatches: 1,
	})
	if err != nil {
		t.Fatalf("construct: %v", err)
	}
	defer ap.Close()
	if got := ap.Config(); !got.FullDuplex {
		t.Fatal("FullDuplex reported false with NumShards=8: the engine fell back " +
			"to half-duplex shards, which are unordered")
	}
}

// Full duplex that cannot engage (no pipeline pool) falls back to round-robin
// half-duplex shards, which do not keep submit order. The NumShards>1 ordering
// check must then apply: the FullDuplex exemption follows the effective state,
// not the requested one. Construction only, so no server is needed.
func TestFDShardsFallbackStillNeedsUnordered(t *testing.T) {
	cl := redis.NewClient(&redis.Options{
		Addr:             apTestAddr(),
		PipelinePoolSize: -1,
	})
	defer cl.Close()
	ap, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		FullDuplex:           true,
		NumShards:            2,
		MaxConcurrentBatches: 1,
	})
	if err == nil {
		ap.Close()
		t.Fatal("FullDuplex without a pipeline pool fell back to round-robin shards " +
			"without the Unordered opt-in")
	}
	if !strings.Contains(err.Error(), "Unordered") {
		t.Fatalf("error should name the Unordered opt-in, got: %v", err)
	}
}

// More engines than the pipeline pool can hold must fail at construction rather
// than leave the surplus spilling to the main pool for the client's lifetime.
func TestFDShardsRejectsMoreEnginesThanPool(t *testing.T) {
	ctx := context.Background()
	cl := redis.NewClient(&redis.Options{
		Addr:             apTestAddr(),
		PipelinePoolSize: 4,
	})
	defer cl.Close()
	if err := cl.Ping(ctx).Err(); err != nil {
		t.Skipf("no local redis: %v", err)
	}
	_, err := cl.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		FullDuplex:           true,
		NumShards:            8,
		MaxConcurrentBatches: 1,
	})
	if err == nil {
		t.Fatal("expected NumShards=8 with PipelinePoolSize=4 to be rejected")
	}
	if !strings.Contains(err.Error(), "pipeline pool") {
		t.Fatalf("error should name the pipeline pool, got: %v", err)
	}
}

// ===== from autopipeline_truth_test.go =====

// Truth battery: adversarial end-to-end validation of the AutoPipeliner's
// core contracts against a live server — per-goroutine ORDER, exact RESPONSE
// demultiplexing, no lost/duplicated executions — under both faces, ordered
// and unordered configs, chunked dispatch, mid-flight cancellation and
// close-drain. Written for the PR #3942 validation pass; every test derives
// an independent ground truth (a model value computed client-side, or a
// differential run against a plain client) rather than trusting the engine's
// own accounting.

func truthAddr() string { return apTestAddr() }

func newTruthClient(t *testing.T) *redis.Client {
	t.Helper()
	c := redis.NewClient(&redis.Options{Addr: truthAddr(), PoolSize: 64})
	if err := c.Ping(context.Background()).Err(); err != nil {
		c.Close()
		t.Skipf("no redis: %v", err)
	}
	return c
}

// TestTruthBlockingPerGoroutineOrder pins the blocking face's ordering
// contract with a server-side proof: each goroutine RPUSHes its own strictly
// increasing sequence through a SHARED autopipeliner; the list on the server
// is the ground truth and must come back exactly 0..M-1 for every goroutine.
func TestTruthBlockingPerGoroutineOrder(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 64, MaxConcurrentBatches: 8, Unordered: true, NumShards: 4,
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 64, 200
	runWithWatchdog(t, 60*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				key := "order:blocking:" + strconv.Itoa(g)
				for i := 0; i < M; i++ {
					if err := ap.RPush(ctx, key, i).Err(); err != nil {
						t.Errorf("g%d rpush %d: %v", g, i, err)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
	for g := 0; g < G; g++ {
		vals, err := client.LRange(ctx, "order:blocking:"+strconv.Itoa(g), 0, -1).Result()
		if err != nil {
			t.Fatalf("lrange g%d: %v", g, err)
		}
		if len(vals) != M {
			t.Fatalf("g%d: %d elements, want %d (lost or duplicated commands)", g, len(vals), M)
		}
		for i, v := range vals {
			if v != strconv.Itoa(i) {
				t.Fatalf("g%d: position %d = %q, want %d (per-goroutine order violated)", g, i, v, i)
			}
		}
	}
}

// TestTruthAsyncOrderedSingleGoroutine pins the deferred face's ordering on
// the ORDERED default config: one goroutine submits an increasing sequence
// without awaiting; the server-side list must be exactly in submit order,
// and a read submitted after a write in the same window must observe it.
func TestTruthAsyncOrderedSingleGoroutine(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipeline() // ordered default
	if err != nil {
		t.Fatal(err)
	}
	const N = 3000
	runWithWatchdog(t, 60*time.Second, func() {
		cmds := make([]*redis.IntCmd, N)
		for i := 0; i < N; i++ {
			cmds[i] = ap.RPush(ctx, "order:async", i)
		}
		// read-your-write inside the window: SET then GET, same key, no await
		// between them.
		set := ap.Set(ctx, "order:ryw", "written", 0)
		get := ap.Get(ctx, "order:ryw")
		for i, c := range cmds {
			if err := c.Err(); err != nil {
				t.Fatalf("rpush %d: %v", i, err)
			}
		}
		if err := set.Err(); err != nil {
			t.Fatal(err)
		}
		if v, err := get.Result(); err != nil || v != "written" {
			t.Fatalf("read-your-write: v=%q err=%v, want %q", v, err, "written")
		}
	})
	vals, err := client.LRange(ctx, "order:async", 0, -1).Result()
	if err != nil {
		t.Fatal(err)
	}
	if len(vals) != N {
		t.Fatalf("%d elements, want %d", len(vals), N)
	}
	for i, v := range vals {
		if v != strconv.Itoa(i) {
			t.Fatalf("position %d = %q, want %d (submit order violated on ordered face)", i, v, i)
		}
	}
}

// TestTruthResponseDemux pins reply demultiplexing: every one of G*M GETs
// must return exactly the unique payload its OWN goroutine wrote — any
// cross-wiring of replies between commands or goroutines fails loudly.
// Run on the most adversarial config: unordered, striped, parallel batches.
func TestTruthResponseDemux(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 128, MaxConcurrentBatches: 32, Unordered: true, NumShards: 8,
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 100, 300
	// Unordered config: a GET may overtake its own SET by design, so the
	// phases are separated — all writes awaited, then all reads. The demux
	// property under test is untouched: every reply must match its OWN
	// command's unique payload.
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				sets := make([]*redis.StatusCmd, M)
				for i := 0; i < M; i++ {
					key := fmt.Sprintf("demux:%d:%d", g, i)
					want := fmt.Sprintf("payload|g=%d|i=%d|%x", g, i, g*1_000_003+i*7919)
					sets[i] = ap.Set(ctx, key, want, 0)
				}
				for i, s := range sets {
					if err := s.Err(); err != nil {
						t.Errorf("set g%d i%d: %v", g, i, err)
						return
					}
				}
				gets := make([]*redis.StringCmd, M)
				for i := 0; i < M; i++ {
					gets[i] = ap.Get(ctx, fmt.Sprintf("demux:%d:%d", g, i))
				}
				for i, gc := range gets {
					want := fmt.Sprintf("payload|g=%d|i=%d|%x", g, i, g*1_000_003+i*7919)
					got, err := gc.Result()
					if err != nil {
						t.Errorf("get g%d i%d: %v", g, i, err)
						return
					}
					if got != want {
						t.Errorf("DEMUX VIOLATION g%d i%d: got %q want %q", g, i, got, want)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
}

// TestTruthOrderedPairsAcrossGoroutines pins per-goroutine read-your-write
// on the ORDERED default config with many goroutines sharing the pipeliner:
// each goroutine interleaves SET/GET pairs on its own keys without any
// barrier — the ordered face must preserve each goroutine's submit order.
func TestTruthOrderedPairsAcrossGoroutines(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipeline() // ordered default
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 64, 150
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				for i := 0; i < M; i++ {
					key := fmt.Sprintf("pair:%d:%d", g, i)
					want := fmt.Sprintf("v|%d|%d", g, i)
					set := ap.Set(ctx, key, want, 0)
					get := ap.Get(ctx, key) // no await between: ordered face must sequence
					if err := set.Err(); err != nil {
						t.Errorf("set g%d i%d: %v", g, i, err)
						return
					}
					got, err := get.Result()
					if err != nil || got != want {
						t.Errorf("ORDER VIOLATION g%d i%d: got %q err=%v want %q", g, i, got, err, want)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
}

// TestTruthCountersExact pins no-lost-no-duplicated execution: G goroutines
// INCRBY random deltas on one shared counter; the client-side sum is the
// ground truth and the final server value must equal it exactly. A dropped
// command, a duplicated retry, or a misrouted reply cannot pass.
func TestTruthCountersExact(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	for _, cfg := range []struct {
		name string
		opts *redis.AutoPipelineOptions
	}{
		{"ordered-default", nil},
		{"unordered-parallel", &redis.AutoPipelineOptions{MaxBatchSize: 200, MaxConcurrentBatches: 40, Unordered: true, NumShards: 8}},
		{"tiny-batches", &redis.AutoPipelineOptions{MaxBatchSize: 2, MaxConcurrentBatches: 4, Unordered: true}},
	} {
		t.Run(cfg.name, func(t *testing.T) {
			var ap *redis.AutoPipeliner
			var err error
			if cfg.opts == nil {
				ap, err = client.AsyncAutoPipeline()
			} else {
				ap, err = client.AsyncAutoPipelineWithOptions(cfg.opts)
			}
			if err != nil {
				t.Fatal(err)
			}
			key := "counter:" + cfg.name
			client.Del(ctx, key)
			const G, M = 50, 200
			var wantTotal atomic.Int64
			runWithWatchdog(t, 90*time.Second, func() {
				var wg sync.WaitGroup
				for g := 0; g < G; g++ {
					wg.Add(1)
					go func(g int) {
						defer wg.Done()
						rng := rand.New(rand.NewSource(int64(g)*7919 + 1))
						futs := make([]*redis.IntCmd, M)
						deltas := make([]int64, M)
						for i := 0; i < M; i++ {
							deltas[i] = rng.Int63n(1000) - 500
							futs[i] = ap.IncrBy(ctx, key, deltas[i])
						}
						for i, f := range futs {
							if err := f.Err(); err != nil {
								t.Errorf("incrby: %v", err)
								return
							}
							wantTotal.Add(deltas[i])
						}
					}(g)
				}
				wg.Wait()
			})
			got, err := client.Get(ctx, key).Int64()
			if err != nil {
				t.Fatal(err)
			}
			if got != wantTotal.Load() {
				t.Fatalf("counter drift: server=%d client-sum=%d (lost or duplicated executions)", got, wantTotal.Load())
			}
		})
	}
}

// TestTruthDifferentialVsPlainClient runs an identical randomized mixed
// workload through a plain client (the model) and the autopipeliner (the
// subject) on disjoint keyspaces, then diffs the full keyspace dumps. Any
// divergence in values, hashes, or list contents is an engine correctness
// bug by construction.
func TestTruthDifferentialVsPlainClient(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 100, MaxConcurrentBatches: 16, Unordered: true, NumShards: 4,
	})
	if err != nil {
		t.Fatal(err)
	}
	type op struct {
		kind       int
		key, field string
		val        int64
	}
	const G, M = 32, 150
	plans := make([][]op, G)
	for g := 0; g < G; g++ {
		rng := rand.New(rand.NewSource(int64(g) + 42))
		plans[g] = make([]op, M)
		for i := range plans[g] {
			plans[g][i] = op{
				kind:  rng.Intn(4),
				key:   strconv.Itoa(rng.Intn(20)), // few keys -> heavy contention
				field: "f" + strconv.Itoa(rng.Intn(5)),
				val:   rng.Int63n(100),
			}
		}
	}
	run := func(prefix string, exec redis.Cmdable) {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				for _, o := range plans[g] {
					key := prefix + o.key
					var err error
					switch o.kind {
					case 0:
						err = exec.IncrBy(ctx, key+":c", o.val).Err()
					case 1:
						err = exec.HIncrBy(ctx, key+":h", o.field, o.val).Err()
					case 2:
						err = exec.SAdd(ctx, key+":s", o.val).Err()
					case 3:
						err = exec.RPush(ctx, key+":l:"+strconv.Itoa(g), o.val).Err()
					}
					if err != nil {
						t.Errorf("%s op: %v", prefix, err)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	}
	runWithWatchdog(t, 120*time.Second, func() {
		run("model:", client) // plain client = the model
		run("subject:", ap)   // autopipeliner = the subject
	})
	dump := func(prefix string) map[string]string {
		out := map[string]string{}
		keys, err := client.Keys(ctx, prefix+"*").Result()
		if err != nil {
			t.Fatal(err)
		}
		for _, k := range keys {
			short := strings.TrimPrefix(k, prefix)
			switch typ := client.Type(ctx, k).Val(); typ {
			case "string":
				out[short] = client.Get(ctx, k).Val()
			case "hash":
				out[short] = fmt.Sprint(client.HGetAll(ctx, k).Val())
			case "set":
				vals, _ := client.SMembers(ctx, k).Result()
				m := map[string]bool{}
				for _, v := range vals {
					m[v] = true
				}
				out[short] = fmt.Sprint(len(m)) // set cardinality (order-free)
			case "list":
				out[short] = strings.Join(client.LRange(ctx, k, 0, -1).Val(), ",")
			default:
				out[short] = "type:" + typ
			}
		}
		return out
	}
	model, subject := dump("model:"), dump("subject:")
	if len(model) != len(subject) {
		t.Fatalf("keyspace size differs: model=%d subject=%d", len(model), len(subject))
	}
	for k, mv := range model {
		if sv, ok := subject[k]; !ok || sv != mv {
			t.Errorf("DIVERGENCE key %q: model=%q subject=%q", k, mv, sv)
		}
	}
}

// TestTruthChunkedDispatchIntegrity drives the MaxBatchBytes chunked path
// with values large enough to split every drain into several chunks and
// verifies exact payload echo — chunk boundaries must not tear, reorder, or
// cross-wire replies.
func TestTruthChunkedDispatchIntegrity(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 500, MaxConcurrentBatches: 8, Unordered: true,
		MaxBatchBytes: 64 * 1024, // ~4 x 16KiB values per chunk
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 16, 40
	base := strings.Repeat("x", 16*1024)
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				sets := make([]*redis.StatusCmd, M)
				for i := 0; i < M; i++ {
					key := fmt.Sprintf("chunk:%d:%d", g, i)
					want := fmt.Sprintf("%s|tail=%d:%d", base, g, i)
					sets[i] = ap.Set(ctx, key, want, 0)
				}
				for i, s := range sets {
					if err := s.Err(); err != nil {
						t.Errorf("set g%d i%d: %v", g, i, err)
						return
					}
				}
				for i := 0; i < M; i++ {
					key := fmt.Sprintf("chunk:%d:%d", g, i)
					want := fmt.Sprintf("%s|tail=%d:%d", base, g, i)
					got, err := ap.Get(ctx, key).Result()
					if err != nil {
						t.Errorf("get %s: %v", key, err)
						return
					}
					if got != want {
						t.Errorf("CHUNK TEAR %s: got %d bytes tail %q, want %d bytes", key, len(got), got[len(got)-24:], len(want))
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
}

// TestTruthCancelNeverLies pins cancellation semantics: a context canceled
// mid-flight may fail the WAIT, but any command that reports success must
// have really executed, and any that reports an error must not have half a
// result. Verified by re-reading the keyspace with a fresh context.
func TestTruthCancelNeverLies(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	const N = 2000
	runWithWatchdog(t, 60*time.Second, func() {
		cctx, cancel := context.WithCancel(ctx)
		futs := make([]*redis.StatusCmd, N)
		for i := 0; i < N; i++ {
			futs[i] = ap.Set(cctx, "cancel:"+strconv.Itoa(i), "v"+strconv.Itoa(i), 0)
			if i == N/2 {
				cancel()
			}
		}
		okCount, errCount := 0, 0
		for i, f := range futs {
			if err := f.Err(); err == nil {
				okCount++
				// claimed success must be true on the server
				v, gerr := client.Get(ctx, "cancel:"+strconv.Itoa(i)).Result()
				if gerr != nil || v != "v"+strconv.Itoa(i) {
					t.Fatalf("cmd %d reported OK but server has %q/%v", i, v, gerr)
				}
			} else {
				errCount++
			}
		}
		t.Logf("cancel split: %d ok, %d error (both outcomes legal; lies are not)", okCount, errCount)
		cancel()
	})
}

// TestTruthCloseDrains pins Close's drain contract under load: every future
// submitted BEFORE Close returns a definite outcome (value or error, no
// hang), and successes are really on the server.
func TestTruthCloseDrains(t *testing.T) {
	ctx := context.Background()
	client := newTruthClient(t)
	defer client.Close()
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 64, MaxConcurrentBatches: 8, Unordered: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	const N = 5000
	runWithWatchdog(t, 60*time.Second, func() {
		futs := make([]*redis.StatusCmd, N)
		for i := 0; i < N; i++ {
			futs[i] = ap.Set(ctx, "drain:"+strconv.Itoa(i), "v", 0)
		}
		if err := ap.Close(); err != nil {
			t.Fatalf("close: %v", err)
		}
		okCount := 0
		for i, f := range futs {
			if err := f.Err(); err == nil {
				okCount++
			} else if err != redis.ErrClosed {
				t.Fatalf("future %d: unexpected error %v", i, err)
			}
		}
		// Every claimed success must exist server-side.
		var present int64
		for i := 0; i < N; i++ {
			if client.Exists(ctx, "drain:"+strconv.Itoa(i)).Val() == 1 {
				present++
			}
		}
		if int(present) < okCount {
			t.Fatalf("%d futures claimed OK but only %d keys present", okCount, present)
		}
		t.Logf("close drain: %d/%d confirmed executed, rest cleanly rejected", okCount, N)
	})
}

// TestTruthClusterOrderAndDemux repeats the order and demux proofs against a
// live cluster through the sharded autopipeliner: per-goroutine order on
// hashtagged keys (one slot per goroutine, so order is meaningful) and exact
// payload echo across slot-spread keys.
func TestTruthClusterOrderAndDemux(t *testing.T) {
	ctx := context.Background()
	cc := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
	defer cc.Close()
	skipIfClusterUnhealthy(t, cc)
	for _, n := range []string{":16600", ":16601", ":16602"} {
		nc := redis.NewClient(&redis.Options{Addr: n})
		nc.FlushDB(ctx)
		nc.Close()
	}
	ap, err := cc.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 32, 100
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				// order proof: hashtag pins the goroutine's list to one slot
				key := fmt.Sprintf("{ord%d}:list", g)
				futs := make([]*redis.IntCmd, M)
				for i := 0; i < M; i++ {
					futs[i] = ap.RPush(ctx, key, i)
				}
				for i, f := range futs {
					if err := f.Err(); err != nil {
						t.Errorf("g%d rpush %d: %v", g, i, err)
						return
					}
				}
				// demux proof: slot-spread unique echoes, phase-separated
				sets := make([]*redis.StatusCmd, M)
				for i := 0; i < M; i++ {
					sets[i] = ap.Set(ctx, fmt.Sprintf("cdemux:%d:%d", g, i), fmt.Sprintf("p|%d|%d", g, i), 0)
				}
				for i, s := range sets {
					if err := s.Err(); err != nil {
						t.Errorf("g%d set %d: %v", g, i, err)
						return
					}
				}
				for i := 0; i < M; i++ {
					want := fmt.Sprintf("p|%d|%d", g, i)
					got, err := ap.Get(ctx, fmt.Sprintf("cdemux:%d:%d", g, i)).Result()
					if err != nil || got != want {
						t.Errorf("CLUSTER DEMUX g%d i%d: got %q err=%v want %q", g, i, got, err, want)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
	for g := 0; g < G; g++ {
		vals, err := cc.LRange(ctx, fmt.Sprintf("{ord%d}:list", g), 0, -1).Result()
		if err != nil {
			t.Fatal(err)
		}
		if len(vals) != M {
			t.Fatalf("cluster g%d: %d elements want %d", g, len(vals), M)
		}
		for i, v := range vals {
			if v != strconv.Itoa(i) {
				t.Fatalf("CLUSTER ORDER g%d: pos %d = %q want %d", g, i, v, i)
			}
		}
	}
}

// TestTruthRESP2Demux repeats the demultiplexing proof on RESP2, where
// replies carry no push-frame framing and a desynchronized stream shows up
// as cross-wired values rather than a protocol error.
func TestTruthRESP2Demux(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{Addr: truthAddr(), Protocol: 2, PoolSize: 32})
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 100, MaxConcurrentBatches: 16, Unordered: true, NumShards: 4,
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 40, 150
	runWithWatchdog(t, 90*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				sets := make([]*redis.StatusCmd, M)
				for i := 0; i < M; i++ {
					sets[i] = ap.Set(ctx, fmt.Sprintf("r2:%d:%d", g, i), fmt.Sprintf("v|%d|%d", g, i), 0)
				}
				for _, s := range sets {
					if err := s.Err(); err != nil {
						t.Errorf("resp2 set: %v", err)
						return
					}
				}
				// mixed reply SHAPES in one batch: string, integer, array,
				// hash, nil — a demux slip shows up as a type/shape mismatch.
				for i := 0; i < M; i++ {
					key := fmt.Sprintf("r2:%d:%d", g, i)
					sv := ap.Get(ctx, key)
					iv := ap.StrLen(ctx, key)
					av := ap.LRange(ctx, "r2:missing:list", 0, -1)
					nv := ap.Get(ctx, "r2:missing:key")
					want := fmt.Sprintf("v|%d|%d", g, i)
					if got, err := sv.Result(); err != nil || got != want {
						t.Errorf("RESP2 string demux g%d i%d: %q %v", g, i, got, err)
						return
					}
					if got, err := iv.Result(); err != nil || got != int64(len(want)) {
						t.Errorf("RESP2 int demux g%d i%d: %d %v want %d", g, i, got, err, len(want))
						return
					}
					if got, err := av.Result(); err != nil || len(got) != 0 {
						t.Errorf("RESP2 array demux g%d i%d: %v %v", g, i, got, err)
						return
					}
					if _, err := nv.Result(); err != redis.Nil {
						t.Errorf("RESP2 nil demux g%d i%d: err=%v want redis.Nil", g, i, err)
						return
					}
				}
			}(g)
		}
		wg.Wait()
	})
}

// TestTruthPipelinePoolInterleaving runs the autopipeliner against a client
// configured with the DEDICATED pipeline pool while a plain client hammers
// the same server through the regular pool: batched traffic and single
// commands must not contaminate each other's connections.
func TestTruthPipelinePoolInterleaving(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{
		Addr: truthAddr(), PoolSize: 32,
		PipelineReadBufferSize: 128 << 10, PipelineWriteBufferSize: 128 << 10,
		PipelinePoolSize: 4,
	})
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 128, MaxConcurrentBatches: 16, Unordered: true,
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 24, 200
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		// plain-client noise on the regular pool
		stop := make(chan struct{})
		for n := 0; n < 8; n++ {
			wg.Add(1)
			go func(n int) {
				defer wg.Done()
				for i := 0; ; i++ {
					select {
					case <-stop:
						return
					default:
					}
					key := fmt.Sprintf("noise:%d", n)
					want := fmt.Sprintf("n|%d|%d", n, i)
					if err := client.Set(ctx, key, want, 0).Err(); err != nil {
						t.Errorf("noise set: %v", err)
						return
					}
					if got, err := client.Get(ctx, key).Result(); err != nil || got != want {
						t.Errorf("NOISE CONTAMINATION: got %q err=%v want %q", got, err, want)
						return
					}
				}
			}(n)
		}
		// autopipelined traffic on the pipeline pool
		var apWg sync.WaitGroup
		for g := 0; g < G; g++ {
			apWg.Add(1)
			go func(g int) {
				defer apWg.Done()
				sets := make([]*redis.StatusCmd, M)
				for i := 0; i < M; i++ {
					sets[i] = ap.Set(ctx, fmt.Sprintf("pp:%d:%d", g, i), fmt.Sprintf("p|%d|%d", g, i), 0)
				}
				for _, s := range sets {
					if err := s.Err(); err != nil {
						t.Errorf("pp set: %v", err)
						return
					}
				}
				for i := 0; i < M; i++ {
					want := fmt.Sprintf("p|%d|%d", g, i)
					got, err := ap.Get(ctx, fmt.Sprintf("pp:%d:%d", g, i)).Result()
					if err != nil || got != want {
						t.Errorf("PIPELINE POOL DEMUX g%d i%d: got %q err=%v want %q", g, i, got, err, want)
						return
					}
				}
			}(g)
		}
		apWg.Wait()
		close(stop)
		wg.Wait()
	})
}

// TestTruthRetriesDoNotDuplicate pins at-most-once execution across the
// engine's retry machinery: non-idempotent INCRs run through an
// autopipeliner whose client has retries ENABLED while the server is
// pounded concurrently; the final counter must equal the exact number of
// successful futures — a retried batch that re-executed a committed command
// would overshoot.
func TestTruthRetriesDoNotDuplicate(t *testing.T) {
	ctx := context.Background()
	client := redis.NewClient(&redis.Options{Addr: truthAddr(), PoolSize: 32, MaxRetries: 3})
	defer client.Close()
	if err := client.Ping(ctx).Err(); err != nil {
		t.Skipf("no redis: %v", err)
	}
	client.FlushDB(ctx)
	ap, err := client.AsyncAutoPipelineWithOptions(&redis.AutoPipelineOptions{
		MaxBatchSize: 64, MaxConcurrentBatches: 16, Unordered: true, NumShards: 4,
	})
	if err != nil {
		t.Fatal(err)
	}
	const G, M = 40, 250
	var okCount atomic.Int64
	runWithWatchdog(t, 120*time.Second, func() {
		var wg sync.WaitGroup
		for g := 0; g < G; g++ {
			wg.Add(1)
			go func(g int) {
				defer wg.Done()
				futs := make([]*redis.IntCmd, M)
				for i := 0; i < M; i++ {
					futs[i] = ap.Incr(ctx, "retry:counter")
				}
				for _, f := range futs {
					if err := f.Err(); err == nil {
						okCount.Add(1)
					}
				}
			}(g)
		}
		wg.Wait()
	})
	got, err := client.Get(ctx, "retry:counter").Int64()
	if err != nil {
		t.Fatal(err)
	}
	if got != okCount.Load() {
		t.Fatalf("at-most-once violated: counter=%d successful futures=%d", got, okCount.Load())
	}
	t.Logf("at-most-once: %d increments == %d successful futures", got, okCount.Load())
}

// ===== from osscluster_autopipeline_fd_moved_test.go =====

// clusterFDAddrs is the local 6-node cluster (3 masters + 3 replicas) the FD
// MOVED tests target; SwapNodes needs the replicas to swap a slot's master.
var clusterFDAddrs = []string{":16600", ":16601", ":16602", ":16603", ":16604", ":16605"}

// TestAPClusterFDMovedRedirectFollowed is the discriminating proof that native
// cluster full-duplex FOLLOWS a MOVED redirect. SwapNodes forces the client to
// believe a replica is the master for a slot; a command routed there is answered
// with MOVED. The FD child's reader diverts the redirect to the redirect-aware
// ClusterClient path, which re-routes to the real master and reloads topology, so
// the value is still read back correctly. Before the redirect fix this returned
// the MOVED error and the test failed.
func TestAPClusterFDMovedRedirectFollowed(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{Addrs: clusterFDAddrs})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)

	ap, err := c.AutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()
	if !ap.Config().FullDuplex {
		t.Fatalf("Config().FullDuplex = false, want true (cluster FD not active)")
	}

	const key, val = "fdmoved:key", "fdmoved:val"
	if err := ap.Set(ctx, key, val, 0).Err(); err != nil {
		t.Fatalf("initial set: %v", err)
	}
	// Point the slot at the wrong node so the next access is answered with MOVED
	// and must be followed.
	if err := c.SwapNodes(ctx, key); err != nil {
		t.Fatalf("SwapNodes: %v", err)
	}

	got, err := ap.Get(ctx, key).Result()
	if err != nil {
		t.Fatalf("get after MOVED: %v (FD redirect not followed)", err)
	}
	if got != val {
		t.Fatalf("get after MOVED = %q, want %q (FD redirect not followed correctly)", got, val)
	}
}

// TestAPClusterFDMovedSurfacedWithoutRedirects anchors the positive test: with
// MaxRedirects disabled (-1 -> 0, one attempt), the FD divert still routes through
// ClusterClient.process, which now honors the zero-redirect budget and surfaces
// the forced MOVED as an error. This proves SwapNodes really induced a redirect on
// the FD path (so the positive test exercised the redirect logic, not a trivial
// pass) and that the FD reprocess goes through the cluster budget.
func TestAPClusterFDMovedSurfacedWithoutRedirects(t *testing.T) {
	ctx := context.Background()
	c := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        clusterFDAddrs,
		MaxRedirects: -1, // normalized to 0: a single attempt, no follow-through
	})
	defer c.Close()
	skipIfClusterUnhealthy(t, c)

	ap, err := c.AutoPipelineWithOptions(&redis.AutoPipelineOptions{FullDuplex: true})
	if err != nil {
		t.Fatal(err)
	}
	defer ap.Close()
	if !ap.Config().FullDuplex {
		t.Fatalf("Config().FullDuplex = false, want true (cluster FD not active)")
	}

	const key = "fdmoved:noredir:key"
	if err := ap.Set(ctx, key, "v", 0).Err(); err != nil {
		t.Fatalf("initial set: %v", err)
	}

	// SwapNodes is racy against the periodic LazyReload, so re-swap and retry until
	// we observe the MOVED (or give up and skip — the swap never stuck).
	deadline := time.Now().Add(3 * time.Second)
	for time.Now().Before(deadline) {
		if err := c.SwapNodes(ctx, key); err != nil {
			t.Fatalf("SwapNodes: %v", err)
		}
		err := ap.Get(ctx, key).Err()
		if err != nil && strings.Contains(err.Error(), "MOVED") {
			return // proven: the redirect was needed and, with no redirects, surfaced
		}
		time.Sleep(100 * time.Millisecond)
	}
	t.Skip("could not observe a MOVED (swap kept being reverted by topology reload)")
}

// ===== from osscluster_autopipeline_nodehook_test.go =====

type nodePeekHook struct{ armed *atomic.Bool }

func (h nodePeekHook) DialHook(next redis.DialHook) redis.DialHook          { return next }
func (h nodePeekHook) ProcessHook(next redis.ProcessHook) redis.ProcessHook { return next }
func (h nodePeekHook) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
	return func(ctx context.Context, cmds []redis.Cmder) error {
		if h.armed.Load() {
			for _, c := range cmds {
				_ = c.Err() // redisotel via rediscmd.CmdsString does exactly this pre-next
				_ = c.String()
			}
		}
		err := next(ctx, cmds)
		if h.armed.Load() {
			for _, c := range cmds {
				_ = c.Err()
				_ = c.String()
			}
		}
		return err
	}
}

// TestClusterNodeHookResultPeekNoDeadlock pins the batch-executor guard for
// cluster per-node goroutines: redisotel installs tracing hooks on NODE
// clients via OnNewNode, and rediscmd.CmdsString reads every command's Err()
// BEFORE next() — on the deferred face those reads used to block on a batch
// that completes only after the node call returns, wedging the dispatch
// permanently (found while assessing the String review round on #3942).
func TestClusterNodeHookResultPeekNoDeadlock(t *testing.T) {
	ctx := context.Background()
	probe := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
	if err := probe.Ping(ctx).Err(); err != nil {
		probe.Close()
		t.Skipf("no cluster: %v", err)
	}
	probe.Close()

	cc := redis.NewClusterClient(&redis.ClusterOptions{Addrs: []string{":16600", ":16601", ":16602"}})
	defer cc.Close()
	var armed atomic.Bool
	var installs atomic.Int32
	cc.OnNewNode(func(nc *redis.Client) {
		installs.Add(1)
		nc.AddHook(nodePeekHook{armed: &armed})
	})
	cc.Ping(ctx)
	t.Logf("node hooks installed: %d", installs.Load())

	ap, err := cc.AsyncAutoPipeline()
	if err != nil {
		t.Fatal(err)
	}

	done := make(chan error, 1)
	go func() {
		armed.Store(true)
		defer armed.Store(false)
		cmds := make([]*redis.StatusCmd, 50)
		for i := range cmds {
			cmds[i] = ap.Set(ctx, "nodehook:"+string(rune('a'+i%26))+":"+string(rune('0'+i%10)), "v", 0)
		}
		var firstErr error
		for _, c := range cmds {
			if err := c.Err(); err != nil && firstErr == nil {
				firstErr = err
			}
		}
		done <- firstErr
	}()
	select {
	case err := <-done:
		t.Logf("no hang, err=%v", err)
	case <-time.After(10 * time.Second):
		t.Fatal("node-level hook result peek wedged the async cluster dispatch (batch executor registration regressed)")
	}
}