fix(redis): bound BZPOPMIN fast mode to redisBlockWaitFallback (#683)

Address Codex P1 / Claude bot review on 6173f58:

The previous bzpopminWaitLoop set `fast` directly from
waitForBlockedCommandUpdate's return value. Under sustained
zsetWaiters.Signal pressure (a ZADD/ZINCRBY hot key whose signals fire
faster than each bzpopminTryAllKeys round finishes) the buffered-1
waiterC stays perpetually full: every wait returns via the signal arm,
fast stays true forever, the 100 ms fallback timer never fires. A
wrongType write (HSET / SET / etc.) on a co- registered key (multi-key
BZPOPMIN) then goes undetected for the entire BLOCK window — a
regression vs #666's strict 100 ms ceiling.

Fix: track the wall time of the last full check inside bzpopminWaitLoop
and demote `fast` back to false once redisBlockWaitFallback has elapsed
since that check. Signal-driven wakes still take the fast path during
the first 100 ms after a full check; after that, the next wake (signal
or timer) runs the full check and resets the budget. The 100 ms
WRONGTYPE-detection ceiling is restored regardless of signal rate.

Per CLAUDE.md regression-test convention, the failing scenario is locked
down by TestRedis_BZPopMinDetectsWrongTypeUnderSignalLoad before the
fix:

- A signal-pressure goroutine drives zsetWaiters.Signal in a tight loop
on bzpop-press-hot. The reader's waiter, registered on both
bzpop-press-cold and bzpop-press-hot, wakes on every signal.
- After the reader settles into fast mode (~200 ms), a string is SET at
bzpop-press-cold.
- Pre-fix: the loop never demotes fast=false, the fast-path probe on
bzpop-press-cold never detects the wrongType, BZPOPMIN times out at the
3 s BLOCK deadline.
- Post-fix: the wall-time gate forces a full check within ~100 ms of the
SET, WRONGTYPE surfaces well before the BLOCK deadline.

Verified: the test fails on parent commit 6173f58 (loop ran for the
full 3 s without surfacing WRONGTYPE) and passes on this commit (returns
WRONGTYPE in ~100 ms).

Drive-by cleanups from the same review pass:

- Collapse tryBZPopMin / tryBZPopMinFast wrappers; bzpopminTryAllKeys
calls tryBZPopMinWithMode directly with the fast bool. Both wrappers had
no other callers (per grep) and only added one level of indirection.
- Move TestRedis_BZPopMinTimesOutOnEmptyKey's stranded doc comment back
over its own function; the previous PR mistakenly prepended it to
TestRedis_BZPopMinRejectsInitialWrongType.

Self-review (CLAUDE.md 5 lenses):
1. Data loss -- None. Loop control flow only; persistBZPopMinResult path
unchanged.
2. Concurrency -- The wall-time gate is a local variable; no shared
state added. waitForBlockedCommandUpdate's signaled-bool contract is
unchanged.
3. Performance -- One time.Since per loop iteration on top of the
existing ScanAt cost; negligible. Worst-case fast-path utilisation drops
from "always" to "first 100 ms of every full check window," but the
fast-mode CPU win is still claimed for the dominant signal-driven case
(most BZPOPMIN BLOCK windows complete within one full-check window
because the consumer pops the first matched ZADD).
4. Data consistency -- WRONGTYPE detection ceiling is back to 100 ms
regardless of signal rate. Fast-path correctness invariants from #677
(signal-driven wake guarantees no wrongType transition) are unchanged.
5. Test coverage -- regression-test-first per CLAUDE.md convention.
Existing TestRedis_BZPopMinRejectsInitialWrongType,
TestRedis_BZPopMinDetectsMidBlockWrongType, and
TestRedis_BZPopMinTimesOutOnEmptyKey continue to lock down the other
branches.

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->

## Summary by CodeRabbit

* **Bug Fixes**
* BZPOPMIN now correctly detects wrong-type errors even under sustained
signal-driven conditions, preventing timeouts in edge cases

* **Tests**
* Added test to verify BZPOPMIN wrong-type error detection under high
signal load

<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: bootjp

adapter/redis_bzpopmin_wake_test.go

@@ -2,6 +2,7 @@ package adapter
 
 import (
 	"context"
+	"runtime"
 	"testing"
 	"time"
 
@@ -113,12 +114,6 @@ func TestRedis_BZPopMinWakesOnZIncrBy(t *testing.T) {
 	}
 }
 
-// TestRedis_BZPopMinTimesOutOnEmptyKey locks down the BLOCK-timeout
-// contract: when no ZADD arrives within the BLOCK window, BZPOPMIN
-// returns redis.Nil rather than a protocol error. This guards a
-// regression in the wait-loop refactor where the new
-// waitForBlockedCommandUpdate timer or context-cancel branch could otherwise
-// leak a -ERR reply.
 // TestRedis_BZPopMinRejectsInitialWrongType locks down the
 // first-iteration full-check invariant: when BZPOPMIN is issued
 // against a key that already holds a wrongType encoding (e.g. a
@@ -195,6 +190,99 @@ func TestRedis_BZPopMinDetectsMidBlockWrongType(t *testing.T) {
 	}
 }
 
+// TestRedis_BZPopMinDetectsWrongTypeUnderSignalLoad locks down the
+// periodic full-check invariant under sustained signal pressure
+// (Codex P1 / Claude bot review on 6173f584).
+//
+// The fast-path optimisation skips the wrongType slow probe on
+// signal-driven wakes. If `fast` is set directly from
+// waitForBlockedCommandUpdate's return value, sustained ZADD/ZINCRBY
+// signals can keep `fast=true` indefinitely — the 100 ms fallback
+// timer never fires because waiterC is constantly refilled. A
+// mid-block wrongType write on one of the registered keys then goes
+// undetected for the entire BLOCK window.
+//
+// Trigger: directly drive zsetWaiters.Signal in a tight loop on
+// bzpop-press-hot while the reader BZPOPMINs on
+// [bzpop-press-cold, bzpop-press-hot]. The reader's waiter is
+// registered on both keys, so each signal wakes it; the empty hot
+// key fast-probe returns nil, the loop continues with fast=true.
+// Driving Signal directly (rather than through a real producer)
+// removes the OCC-race non-determinism a stealer goroutine would
+// introduce. After the reader settles into fast mode, SET a string
+// at bzpop-press-cold. Without a wall-time-based forced full check,
+// the reader's fast probe on bzpop-press-cold never detects the
+// wrongType and the command times out at the BLOCK deadline
+// instead of returning WRONGTYPE.
+//
+// The fix maintains a lastFullCheck wall time inside bzpopminWaitLoop
+// and demotes `fast` back to false when more than redisBlockWaitFallback
+// has elapsed since the last full check, restoring the #666 ceiling.
+func TestRedis_BZPopMinDetectsWrongTypeUnderSignalLoad(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdbReader := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbReader.Close() }()
+	rdbWriter := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdbWriter.Close() }()
+
+	stealerCtx, stealerCancel := context.WithCancel(context.Background())
+	defer stealerCancel()
+
+	// Synthetic signal generator: tight loop calling
+	// zsetWaiters.Signal on bzpop-press-hot. The reader's waiter is
+	// registered on both bzpop-press-cold and bzpop-press-hot, so
+	// any signal on either key wakes it. No actual zset rows are
+	// written — the fast-path probe on bzpop-press-hot finds
+	// nothing and the reader stays blocked. runtime.Gosched lets
+	// the reader's goroutine make forward progress between
+	// signals.
+	signalServer := nodes[0].redisServer
+	go func() {
+		key := []byte("bzpop-press-hot")
+		for stealerCtx.Err() == nil {
+			signalServer.zsetWaiters.Signal(key)
+			runtime.Gosched()
+		}
+	}()
+
+	type popResult struct {
+		zwk *redis.ZWithKey
+		err error
+	}
+	resultCh := make(chan popResult, 1)
+	go func() {
+		zwk, err := rdbReader.BZPopMin(context.Background(), 3*time.Second,
+			"bzpop-press-cold", "bzpop-press-hot").Result()
+		resultCh <- popResult{zwk: zwk, err: err}
+	}()
+
+	// Let the reader complete several signal-driven wakes before we
+	// introduce the wrongType. 200 ms is well above the 100 ms
+	// fallback budget; with the fix in place a forced full check
+	// has run during this window.
+	time.Sleep(200 * time.Millisecond)
+
+	require.NoError(t, rdbWriter.Set(context.Background(), "bzpop-press-cold",
+		"I am a string", 0).Err())
+
+	select {
+	case res := <-resultCh:
+		require.Error(t, res.err, "BZPOPMIN must surface WRONGTYPE on bzpop-press-cold under signal load (zwk=%v)", res.zwk)
+		require.Contains(t, res.err.Error(), "WRONGTYPE")
+	case <-time.After(3500 * time.Millisecond):
+		t.Fatal("BZPOPMIN did not return WRONGTYPE under sustained signal load — fast mode likely sticky")
+	}
+}
+
+// TestRedis_BZPopMinTimesOutOnEmptyKey locks down the BLOCK-timeout
+// contract: when no ZADD arrives within the BLOCK window, BZPOPMIN
+// returns redis.Nil rather than a protocol error. This guards a
+// regression in the wait-loop refactor where the new
+// waitForBlockedCommandUpdate timer or context-cancel branch could otherwise
+// leak a -ERR reply.
 func TestRedis_BZPopMinTimesOutOnEmptyKey(t *testing.T) {
 	t.Parallel()
 	nodes, _, _ := createNode(t, 3)

adapter/redis_compat_commands.go

@@ -3650,23 +3650,13 @@ func (r *RedisServer) zremrangebyrank(conn redcon.Conn, cmd redcon.Command) {
 	conn.WriteInt(removed)
 }
 
-func (r *RedisServer) tryBZPopMin(key []byte) (*bzpopminResult, error) {
-	return r.tryBZPopMinWithMode(key, false)
-}
-
-// tryBZPopMinFast is the signal-driven-wake variant of tryBZPopMin.
-// It uses keyTypeAtExpectFast (no slow-path fallback, no wrongType
-// detection) on the assumption that the only mutation since the
-// caller's previous full check is a ZADD/ZINCRBY — the only writes
-// that fire zsetWaiters.Signal. A wrongType write that arrived since
-// the last full check would not have signalled, so this fast path
-// would treat the key as "still empty" and return nil; the
-// fallback-timer wake in bzpopminWaitLoop, which uses the full
-// tryBZPopMin path, detects such cases within ~redisBlockWaitFallback.
-func (r *RedisServer) tryBZPopMinFast(key []byte) (*bzpopminResult, error) {
-	return r.tryBZPopMinWithMode(key, true)
-}
-
+// tryBZPopMinWithMode runs one BZPOPMIN attempt against key. The
+// fast flag selects keyTypeAtExpectFast (no slow-path fallback, no
+// wrongType detection) when true; the caller MUST guarantee that the
+// only mutations since the previous full check are signalling writes
+// (ZADD/ZINCRBY for zsetWaiters). bzpopminWaitLoop enforces this by
+// running fast=false on the first iteration and after every
+// fallback-timer wake or wall-time-bounded re-arm.
 func (r *RedisServer) tryBZPopMinWithMode(key []byte, fast bool) (*bzpopminResult, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), redisDispatchTimeout)
 	defer cancel()
@@ -3786,17 +3776,28 @@ func (r *RedisServer) bzpopminWaitLoop(conn redcon.Conn, keys [][]byte, deadline
 	handlerCtx := r.handlerContext()
 	w, release := r.zsetWaiters.Register(keys)
 	defer release()
-	// fast tracks whether the previous wake came from a Signal (true)
-	// or the fallback timer / shutdown (false). The first iteration
-	// always runs the full tryBZPopMin so an existing wrongType key
-	// surfaces an immediate WRONGTYPE; subsequent iterations after a
-	// signal-driven wake skip the wrongType detection because the
-	// only writes that fire zsetWaiters.Signal are ZADD / ZINCRBY,
-	// neither of which can introduce a wrongType. Fallback-timer
-	// wakes always re-run the full check so a HSET / SET / etc. that
-	// arrived between iterations is detected within ~one
-	// redisBlockWaitFallback (100ms).
+	// fast tracks whether the next iteration may skip the wrongType
+	// slow probe. The first iteration is always full so an existing
+	// wrongType key surfaces an immediate WRONGTYPE; subsequent
+	// iterations after a signal-driven wake skip the wrongType
+	// detection because zsetWaiters.Signal only fires for ZADD /
+	// ZINCRBY (neither of which can introduce a wrongType).
+	//
+	// lastFullCheck wall-time-bounds how long the fast mode can stay
+	// active under sustained signal pressure. Without this gate, a
+	// hot key whose zsetWaiters.Signal fires faster than each
+	// bzpopminTryAllKeys round finishes can keep waiterC perpetually
+	// full, starving the fallback timer and letting a wrongType
+	// write on a co-registered key (multi-key BZPOPMIN) go
+	// undetected for the entire BLOCK window. Demoting `fast` back
+	// to false after redisBlockWaitFallback elapses since the last
+	// full check restores the #666 ceiling: WRONGTYPE on any
+	// registered key surfaces within ~one fallback interval (100 ms)
+	// regardless of signal rate. See
+	// TestRedis_BZPopMinDetectsWrongTypeUnderSignalLoad for the
+	// regression scenario.
 	fast := false
+	lastFullCheck := time.Now()
 	for {
 		if handlerCtx.Err() != nil {
 			conn.WriteNull()
@@ -3805,28 +3806,27 @@ func (r *RedisServer) bzpopminWaitLoop(conn redcon.Conn, keys [][]byte, deadline
 		if r.bzpopminTryAllKeys(conn, keys, fast) {
 			return
 		}
+		if !fast {
+			lastFullCheck = time.Now()
+		}
 		if !time.Now().Before(deadline) {
 			conn.WriteNull()
 			return
 		}
-		fast = waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
+		signaled := waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
+		fast = signaled && time.Since(lastFullCheck) < redisBlockWaitFallback
 	}
 }
 
-// bzpopminTryAllKeys runs one tryBZPopMin pass across keys. Returns
-// true when a result was written (success or terminal error) and the
-// caller should stop the loop, false to continue waiting. The fast
-// flag selects tryBZPopMinFast (signal-driven wake, skips wrongType
-// detection) over tryBZPopMin (full check).
+// bzpopminTryAllKeys runs one tryBZPopMinWithMode pass across keys.
+// Returns true when a result was written (success or terminal error)
+// and the caller should stop the loop, false to continue waiting.
+// The fast flag is forwarded to tryBZPopMinWithMode: true selects
+// the signal-driven-wake path (skips wrongType detection); false
+// selects the full check.
 func (r *RedisServer) bzpopminTryAllKeys(conn redcon.Conn, keys [][]byte, fast bool) bool {
 	for _, key := range keys {
-		var result *bzpopminResult
-		var err error
-		if fast {
-			result, err = r.tryBZPopMinFast(key)
-		} else {
-			result, err = r.tryBZPopMin(key)
-		}
+		result, err := r.tryBZPopMinWithMode(key, fast)
 		if err != nil {
 			conn.WriteError(err.Error())
 			return true