From 124b5384a97f151b45bbdeed9add213713202c96 Mon Sep 17 00:00:00 2001 From: sadd amr3e Date: Mon, 29 Jun 2026 17:20:17 +0530 Subject: [PATCH] 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 * test(hooks): cover withProcessHook slice readers in race test Signed-off-by: Saddam --------- Signed-off-by: Saddam Co-authored-by: Nedyalko Dyakov <1547186+ndyakov@users.noreply.github.com> --- hooks_current_race_test.go | 72 +++++++++++++++++++++ redis.go | 129 +++++++++++++++++++++++-------------- 2 files changed, 151 insertions(+), 50 deletions(-) create mode 100644 hooks_current_race_test.go diff --git a/hooks_current_race_test.go b/hooks_current_race_test.go new file mode 100644 index 00000000..06338a49 --- /dev/null +++ b/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() +} diff --git a/redis.go b/redis.go index 0dcaac23..6548216f 100644 --- a/redis.go +++ b/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) } //------------------------------------------------------------------------------