fix(hooks): data race on hook state during AddHook (#3868)

* fix(hooks): data race on hooksMixin.current during AddHook

hooksMixin.current holds the composed hook chain. AddHook rewrote all
four current fields in place while command execution read them
unsynchronized, so calling AddHook on a live client raced on those
two-word func values. The unlocked slice append in AddHook was racy
across concurrent AddHook callers too.

Fold slice + current into an immutable hooksState published through a
single atomic.Pointer. The read paths (dial/process/pipeline/txPipeline)
become a lock-free Load, dropping the RWMutex from the hot path entirely.
AddHook is copy-on-write under a plain Mutex used only by writers, which
also makes concurrent AddHook safe. Covered by a -race test exercising
concurrent AddHook writers against the command-path readers.

Signed-off-by: Saddam <saddamr3e@gmail.com>

* test(hooks): cover withProcessHook slice readers in race test

Signed-off-by: Saddam <saddamr3e@gmail.com>

---------

Signed-off-by: Saddam <saddamr3e@gmail.com>
Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com>

hooks_current_race_test.go

@@ -0,0 +1,72 @@
+package redis
+
+import (
+	"context"
+	"sync"
+	"testing"
+)
+
+type noopRaceHook struct{}
+
+func (noopRaceHook) DialHook(next DialHook) DialHook          { return next }
+func (noopRaceHook) ProcessHook(next ProcessHook) ProcessHook { return next }
+func (noopRaceHook) ProcessPipelineHook(next ProcessPipelineHook) ProcessPipelineHook {
+	return next
+}
+
+// AddHook publishes a fresh hooks snapshot via hs.state, while the command-path
+// getters (processHook, processPipelineHook, processTxPipelineHook), the slice
+// readers (withProcessHook, withProcessPipelineHook) and the AddHook slice
+// append all touch the same hook state. Exercise concurrent AddHook writers
+// against both reader groups; run with -race to catch a torn read of the
+// published snapshot or an unsynchronized slice append.
+func TestHooksMixinCurrentConcurrent(t *testing.T) {
+	var hs hooksMixin
+	hs.initHooks(hooks{})
+
+	ctx := context.Background()
+	cmd := NewCmd(ctx, "ping")
+	cmds := []Cmder{cmd}
+
+	const writers = 2
+
+	var wg sync.WaitGroup
+	wg.Add(writers + 1)
+
+	done := make(chan struct{})
+
+	// Writers: concurrent AddHook calls, each replacing hs.state copy-on-write.
+	for i := 0; i < writers; i++ {
+		go func() {
+			defer wg.Done()
+			for {
+				select {
+				case <-done:
+					return
+				default:
+					hs.AddHook(noopRaceHook{})
+				}
+			}
+		}()
+	}
+
+	noopProcess := func(context.Context, Cmder) error { return nil }
+	noopPipeline := func(context.Context, []Cmder) error { return nil }
+
+	// Reader: the command-path getters Load hs.state lock-free; the slice
+	// readers (withProcessHook, withProcessPipelineHook) iterate the published
+	// snapshot's hook slice.
+	go func() {
+		defer wg.Done()
+		defer close(done)
+		for i := 0; i < 3000; i++ {
+			_ = hs.processHook(ctx, cmd)
+			_ = hs.processPipelineHook(ctx, cmds)
+			_ = hs.processTxPipelineHook(ctx, cmds)
+			_ = hs.withProcessHook(ctx, cmd, noopProcess)
+			_ = hs.withProcessPipelineHook(ctx, cmds, noopPipeline)
+		}
+	}()
+
+	wg.Wait()
+}

redis.go

@@ -64,17 +64,61 @@ type (
 )
 
 type hooksMixin struct {
-	hooksMu *sync.RWMutex
+	// hooksMu serializes writers (AddHook); readers never take it.
+	hooksMu *sync.Mutex
+	// state holds the immutable hook snapshot. Readers Load it lock-free;
+	// writers publish a replacement copy-on-write under hooksMu.
+	state *atomic.Pointer[hooksState]
+}
 
+// hooksState is an immutable snapshot of the hook configuration. Once stored
+// in hooksMixin.state it is never mutated; AddHook builds a fresh copy.
+type hooksState struct {
 	slice   []Hook
 	initial hooks
 	current hooks
 }
 
+// rebuild recomputes current from initial + slice. It mutates the receiver, so
+// it must only run on a state that has not yet been published.
+func (s *hooksState) rebuild() {
+	s.initial.setDefaults()
+
+	s.current.dial = s.initial.dial
+	s.current.process = s.initial.process
+	s.current.pipeline = s.initial.pipeline
+	s.current.txPipeline = s.initial.txPipeline
+
+	for i := len(s.slice) - 1; i >= 0; i-- {
+		if wrapped := s.slice[i].DialHook(s.current.dial); wrapped != nil {
+			s.current.dial = wrapped
+		}
+		if wrapped := s.slice[i].ProcessHook(s.current.process); wrapped != nil {
+			s.current.process = wrapped
+		}
+		if wrapped := s.slice[i].ProcessPipelineHook(s.current.pipeline); wrapped != nil {
+			s.current.pipeline = wrapped
+		}
+		if wrapped := s.slice[i].ProcessPipelineHook(s.current.txPipeline); wrapped != nil {
+			s.current.txPipeline = wrapped
+		}
+	}
+}
+
 func (hs *hooksMixin) initHooks(hooks hooks) {
-	hs.hooksMu = new(sync.RWMutex)
-	hs.initial = hooks
-	hs.chain()
+	var slice []Hook
+	if hs.state != nil {
+		if old := hs.state.Load(); old != nil {
+			slice = old.slice
+		}
+	}
+
+	hs.hooksMu = new(sync.Mutex)
+	hs.state = new(atomic.Pointer[hooksState])
+
+	state := &hooksState{slice: slice, initial: hooks}
+	state.rebuild()
+	hs.state.Store(state)
 }
 
 type hooks struct {
@@ -136,51 +180,42 @@ func (h *hooks) setDefaults() {
 // Please note: "next(ctx, cmd)" is very important, it will call the next hook,
 // if "next(ctx, cmd)" is not executed, the redis command will not be executed.
 func (hs *hooksMixin) AddHook(hook Hook) {
-	hs.slice = append(hs.slice, hook)
-	hs.chain()
-}
-
-func (hs *hooksMixin) chain() {
-	hs.initial.setDefaults()
-
 	hs.hooksMu.Lock()
 	defer hs.hooksMu.Unlock()
 
-	hs.current.dial = hs.initial.dial
-	hs.current.process = hs.initial.process
-	hs.current.pipeline = hs.initial.pipeline
-	hs.current.txPipeline = hs.initial.txPipeline
-
-	for i := len(hs.slice) - 1; i >= 0; i-- {
-		if wrapped := hs.slice[i].DialHook(hs.current.dial); wrapped != nil {
-			hs.current.dial = wrapped
-		}
-		if wrapped := hs.slice[i].ProcessHook(hs.current.process); wrapped != nil {
-			hs.current.process = wrapped
-		}
-		if wrapped := hs.slice[i].ProcessPipelineHook(hs.current.pipeline); wrapped != nil {
-			hs.current.pipeline = wrapped
-		}
-		if wrapped := hs.slice[i].ProcessPipelineHook(hs.current.txPipeline); wrapped != nil {
-			hs.current.txPipeline = wrapped
-		}
+	old := hs.state.Load()
+	state := &hooksState{
+		slice:   make([]Hook, len(old.slice)+1),
+		initial: old.initial,
 	}
+	copy(state.slice, old.slice)
+	state.slice[len(old.slice)] = hook
+	state.rebuild()
+
+	hs.state.Store(state)
 }
 
 func (hs *hooksMixin) clone() hooksMixin {
-	hs.hooksMu.Lock()
-	defer hs.hooksMu.Unlock()
+	old := hs.state.Load()
+	l := len(old.slice)
+	state := &hooksState{
+		slice:   old.slice[:l:l],
+		initial: old.initial,
+		current: old.current,
+	}
 
-	clone := *hs
-	l := len(clone.slice)
-	clone.slice = clone.slice[:l:l]
-	clone.hooksMu = new(sync.RWMutex)
+	clone := hooksMixin{
+		hooksMu: new(sync.Mutex),
+		state:   new(atomic.Pointer[hooksState]),
+	}
+	clone.state.Store(state)
 	return clone
 }
 
 func (hs *hooksMixin) withProcessHook(ctx context.Context, cmd Cmder, hook ProcessHook) error {
-	for i := len(hs.slice) - 1; i >= 0; i-- {
-		if wrapped := hs.slice[i].ProcessHook(hook); wrapped != nil {
+	slice := hs.state.Load().slice
+	for i := len(slice) - 1; i >= 0; i-- {
+		if wrapped := slice[i].ProcessHook(hook); wrapped != nil {
 			hook = wrapped
 		}
 	}
@@ -190,8 +225,9 @@ func (hs *hooksMixin) withProcessHook(ctx context.Context, cmd Cmder, hook Proce
 func (hs *hooksMixin) withProcessPipelineHook(
 	ctx context.Context, cmds []Cmder, hook ProcessPipelineHook,
 ) error {
-	for i := len(hs.slice) - 1; i >= 0; i-- {
-		if wrapped := hs.slice[i].ProcessPipelineHook(hook); wrapped != nil {
+	slice := hs.state.Load().slice
+	for i := len(slice) - 1; i >= 0; i-- {
+		if wrapped := slice[i].ProcessPipelineHook(hook); wrapped != nil {
 			hook = wrapped
 		}
 	}
@@ -199,26 +235,19 @@ func (hs *hooksMixin) withProcessPipelineHook(
 }
 
 func (hs *hooksMixin) dialHook(ctx context.Context, network, addr string) (net.Conn, error) {
-	// Access to hs.current is guarded by a read-only lock since it may be mutated by AddHook(...)
-	// while this dialer is concurrently accessed by the background connection pool population
-	// routine when MinIdleConns > 0.
-	hs.hooksMu.RLock()
-	current := hs.current
-	hs.hooksMu.RUnlock()
-
-	return current.dial(ctx, network, addr)
+	return hs.state.Load().current.dial(ctx, network, addr)
 }
 
 func (hs *hooksMixin) processHook(ctx context.Context, cmd Cmder) error {
-	return hs.current.process(ctx, cmd)
+	return hs.state.Load().current.process(ctx, cmd)
 }
 
 func (hs *hooksMixin) processPipelineHook(ctx context.Context, cmds []Cmder) error {
-	return hs.current.pipeline(ctx, cmds)
+	return hs.state.Load().current.pipeline(ctx, cmds)
 }
 
 func (hs *hooksMixin) processTxPipelineHook(ctx context.Context, cmds []Cmder) error {
-	return hs.current.txPipeline(ctx, cmds)
+	return hs.state.Load().current.txPipeline(ctx, cmds)
 }
 
 //------------------------------------------------------------------------------