perf(redis): empty-zset fast path for BZPOPMIN signal-driven wakes

Production profile after #666 deploy showed BZPOPMIN still taking
47.76 percent of n1 leader CPU (down from 86.56 percent pre-deploy
but not eliminated). Per-wake breakdown: each tryBZPopMin runs
keyTypeAtExpect, the zset prefix probe misses on the empty queue,
so the slow-path fallback re-issues all ~19 keyTypeAt seeks just
to confirm "still empty". On a busy queue with ZADD signals firing
at the producer rate, this baseline grows linearly with the wake
rate.

Add a fast variant for the signal-driven-wake case:

- keyTypeAtExpectFast: probes only the expected type's prefixes
  (3 seeks for zset). On a miss, returns redisTypeNone immediately
  without rawKeyTypeAt's 16 follow-up seeks and without the
  hasHigherPriorityStringEncoding guard.
- tryBZPopMinFast: tryBZPopMin variant that uses keyTypeAtExpectFast.
  No wrongType detection on the type-check step (the load step
  itself still surfaces obvious corruption).
- waitForBlockedCommandUpdate now returns true iff the wake came
  from a Signal channel (false on fallback timer / handlerCtx
  cancellation).
- bzpopminWaitLoop tracks the wake reason and threads it as a
  fast bool into bzpopminTryAllKeys / tryBZPopMin{,Fast}. First
  iteration is always full; signal-driven wakes flip to fast;
  fallback-timer wakes flip back to full.

The wrongType-detection invariant is preserved because
zsetWaiters.Signal only fires on ZADD / ZINCRBY (the only writes
that produce zset rows visible to the fast probe). A
wrongType-introducing write (HSET, SET, XADD on the same user key)
does not fire Signal, so the fallback timer's full check is the
safety net that detects it within ~one redisBlockWaitFallback
(100 ms). Initial-iteration wrongType detection is unchanged.

Behavior change: a HSET / SET / XADD that arrives mid-BLOCK on a
key being BZPOPMINed surfaces WRONGTYPE within ~100 ms instead of
immediately. Previous behavior surfaced WRONGTYPE within ~10 ms
under the busy-poll, ~100 ms under #666's fallback (this PR keeps
the same 100 ms ceiling). No correctness regression for existing
contracts (initial wrongType, BLOCK-timeout-returns-nil).

Test coverage:

- TestRedis_BZPopMinRejectsInitialWrongType (new): pre-existing
  string at the BZPOPMIN target key surfaces WRONGTYPE on the very
  first iteration's full check, well under the BLOCK timeout.
- TestRedis_BZPopMinDetectsMidBlockWrongType (new): SET arriving
  during a BZPOPMIN BLOCK surfaces WRONGTYPE through the fallback-
  timer full check before the BLOCK budget elapses. Locks down the
  fallback safety net.
- Existing TestRedis_BZPopMinWakesOnZAdd / WakesOnZIncrBy /
  TimesOutOnEmptyKey continue to lock down the signal path and
  the deadline contract.

Self-review (CLAUDE.md 5 lenses):
1. Data loss -- None. Read-only fast path; persistBZPopMinResult
   is unchanged.
2. Concurrency -- waitForBlockedCommandUpdate's new bool return is
   set under the same select that already serialised the three
   wake sources; no new shared state.
3. Performance -- 16 fewer Pebble seeks per signal-driven wake on
   empty queues. The number of wakes per BZPOPMIN is unchanged --
   this is per-wake cost reduction, not wake-rate reduction.
4. Data consistency -- Signal-driven wake reads at a fresh readTS;
   the fast probe and the slow probe see the same MVCC snapshot.
   wrongType detection on the slow path is preserved on initial
   iteration and on every fallback wake.
5. Test coverage -- two new lockdown tests for the wrongType paths
   that cannot regress silently if the fast / full mode flag is
   wired incorrectly.

Author: bootjp

adapter/redis_bzpopmin_wake_test.go

@@ -119,6 +119,82 @@ func TestRedis_BZPopMinWakesOnZIncrBy(t *testing.T) {
 // 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
+// string), the very first tryBZPopMin must surface WRONGTYPE. The
+// signal-driven fast path must never run before the initial full
+// check has confirmed the type.
+func TestRedis_BZPopMinRejectsInitialWrongType(t *testing.T) {
+	t.Parallel()
+	nodes, _, _ := createNode(t, 3)
+	defer shutdown(nodes)
+
+	rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress})
+	defer func() { _ = rdb.Close() }()
+	ctx := context.Background()
+
+	// Pre-write a string at the BZPOPMIN target key.
+	require.NoError(t, rdb.Set(ctx, "bzpop-wrongtype", "I am a string", 0).Err())
+
+	// BZPOPMIN with a 1 s budget — full check on the first iteration
+	// must catch the wrongType immediately, well under the BLOCK
+	// timeout.
+	zwk, err := rdb.BZPopMin(ctx, time.Second, "bzpop-wrongtype").Result()
+	require.Error(t, err, "BZPOPMIN on a string key must return WRONGTYPE")
+	require.Contains(t, err.Error(), "WRONGTYPE")
+	require.Nil(t, zwk)
+}
+
+// TestRedis_BZPopMinDetectsMidBlockWrongType locks down the
+// fallback-timer-driven full check: when BZPOPMIN is blocked on a
+// non-existent key and a wrongType encoding (e.g. SET) is written
+// to that key during the BLOCK window, the next fallback-timer
+// wake must run the full check and surface WRONGTYPE within
+// ~redisBlockWaitFallback (100 ms). A pure signal-driven path
+// would miss this because SET / HSET / etc. do not fire
+// zsetWaiters.Signal.
+//
+// The 5 s BLOCK budget gives plenty of slack for CI scheduler
+// jitter; the assertion is "WRONGTYPE before BLOCK timeout", not a
+// tight latency gate.
+func TestRedis_BZPopMinDetectsMidBlockWrongType(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() }()
+	ctx := context.Background()
+
+	type popResult struct {
+		zwk *redis.ZWithKey
+		err error
+	}
+	resultCh := make(chan popResult, 1)
+	go func() {
+		zwk, err := rdbReader.BZPopMin(ctx, 5*time.Second, "bzpop-mid-wrongtype").Result()
+		resultCh <- popResult{zwk: zwk, err: err}
+	}()
+
+	// Let the reader enter the wait loop and exhaust its first
+	// (full) iteration on a missing key. Then SET a string at the
+	// same key. The next fallback-timer wake (~100 ms after the
+	// previous one) must run the full check and surface WRONGTYPE.
+	time.Sleep(50 * time.Millisecond)
+	require.NoError(t, rdbWriter.Set(ctx, "bzpop-mid-wrongtype", "I am a string", 0).Err())
+
+	select {
+	case res := <-resultCh:
+		require.Error(t, res.err, "BZPOPMIN must return WRONGTYPE after mid-block SET, got zwk=%v", res.zwk)
+		require.Contains(t, res.err.Error(), "WRONGTYPE")
+	case <-time.After(6 * time.Second):
+		t.Fatal("BZPOPMIN did not return WRONGTYPE within the BLOCK window after a mid-block SET")
+	}
+}
+
 func TestRedis_BZPopMinTimesOutOnEmptyKey(t *testing.T) {
 	t.Parallel()
 	nodes, _, _ := createNode(t, 3)

adapter/redis_compat_commands.go

@@ -3651,12 +3651,35 @@ func (r *RedisServer) zremrangebyrank(conn redcon.Conn, cmd redcon.Command) {
 }
 
 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)
+}
+
+func (r *RedisServer) tryBZPopMinWithMode(key []byte, fast bool) (*bzpopminResult, error) {
 	ctx, cancel := context.WithTimeout(context.Background(), redisDispatchTimeout)
 	defer cancel()
 	var result *bzpopminResult
 	err := r.retryRedisWrite(ctx, func() error {
 		readTS := r.readTS()
-		typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeZSet)
+		var typ redisValueType
+		var err error
+		if fast {
+			typ, err = r.keyTypeAtExpectFast(ctx, key, readTS, redisTypeZSet)
+		} else {
+			typ, err = r.keyTypeAtExpect(ctx, key, readTS, redisTypeZSet)
+		}
 		if err != nil {
 			return err
 		}
@@ -3763,28 +3786,47 @@ 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 := false
 	for {
 		if handlerCtx.Err() != nil {
 			conn.WriteNull()
 			return
 		}
-		if r.bzpopminTryAllKeys(conn, keys) {
+		if r.bzpopminTryAllKeys(conn, keys, fast) {
 			return
 		}
 		if !time.Now().Before(deadline) {
 			conn.WriteNull()
 			return
 		}
-		waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
+		fast = waitForBlockedCommandUpdate(handlerCtx, w.C, deadline)
 	}
 }
 
 // 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.
-func (r *RedisServer) bzpopminTryAllKeys(conn redcon.Conn, keys [][]byte) bool {
+// caller should stop the loop, false to continue waiting. The fast
+// flag selects tryBZPopMinFast (signal-driven wake, skips wrongType
+// detection) over tryBZPopMin (full check).
+func (r *RedisServer) bzpopminTryAllKeys(conn redcon.Conn, keys [][]byte, fast bool) bool {
 	for _, key := range keys {
-		result, err := r.tryBZPopMin(key)
+		var result *bzpopminResult
+		var err error
+		if fast {
+			result, err = r.tryBZPopMinFast(key)
+		} else {
+			result, err = r.tryBZPopMin(key)
+		}
 		if err != nil {
 			conn.WriteError(err.Error())
 			return true
@@ -3812,7 +3854,16 @@ func (r *RedisServer) bzpopminTryAllKeys(conn redcon.Conn, keys [][]byte) bool {
 // BRPOP / BLMOVE in follow-ups) — the keyWaiterRegistry that produces
 // waiterC is per-domain (streamWaiters vs zsetWaiters), but the
 // timer-and-select shape is identical.
-func waitForBlockedCommandUpdate(handlerCtx context.Context, waiterC <-chan struct{}, deadline time.Time) {
+//
+// Returns true iff the wake came from waiterC (i.e., a producer
+// Signal). False on fallback-timer fire or handlerCtx cancellation.
+// Callers that have a signal-implied invariant (e.g., "only ZADD /
+// ZINCRBY fires zsetWaiters.Signal") can use the return value to
+// pick a faster re-check on the next iteration; fallback wakes
+// always need the full check because writes that bypass Signal
+// (Lua flush, follower-applied entries, wrongType-introducing
+// commands) only become observable through the timer branch.
+func waitForBlockedCommandUpdate(handlerCtx context.Context, waiterC <-chan struct{}, deadline time.Time) bool {
 	fallback := redisBlockWaitFallback
 	if remaining := time.Until(deadline); remaining < fallback {
 		fallback = remaining
@@ -3832,8 +3883,11 @@ func waitForBlockedCommandUpdate(handlerCtx context.Context, waiterC <-chan stru
 	}()
 	select {
 	case <-waiterC:
+		return true
 	case <-timer.C:
+		return false
 	case <-handlerCtx.Done():
+		return false
 	}
 }
 

adapter/redis_compat_helpers.go

@@ -363,6 +363,44 @@ func (r *RedisServer) keyTypeAtExpect(ctx context.Context, key []byte, readTS ui
 	return r.applyTTLFilter(ctx, key, readTS, expected)
 }
 
+// keyTypeAtExpectFast is the signal-driven-wake variant of
+// keyTypeAtExpect. On a fast-probe miss it returns redisTypeNone
+// directly (no rawKeyTypeAt slow-path fallback, no
+// hasHigherPriorityStringEncoding guard). Callers MUST have an
+// invariant that the only mutation since the last full check could
+// have produced a row visible to probeExpectedType — typically a
+// blocking-command wait loop after a Signal-driven wake, where the
+// only writes that fire keyWaiterRegistry.Signal are
+// expected-type-creating writes (ZADD/ZINCRBY for zsets,
+// XADD-and-friends for streams). A wrongType-introducing write
+// (HSET, SET, etc.) does NOT signal, so a non-zset key that
+// appeared between iterations is invisible to this fast path; the
+// blocking command's fallback-timer wake (which uses the slow
+// keyTypeAtExpect) is the safety net that detects it within
+// ~redisBlockWaitFallback (100ms).
+//
+// Compared to keyTypeAtExpect on the empty-key case
+// (probeExpectedType -> false -> rawKeyTypeAt slow path = ~19
+// seeks), the fast variant returns after the 3-seek probe. For a
+// BZPOPMIN waiting on an empty zset and being woken by Signal at
+// the ZADD rate, this turns each wake from "19 seeks just to
+// confirm still-empty" into "0 seeks because the probe found the
+// new ZADD's row" or "3 seeks to confirm the ZADD raced and the
+// queue is empty again".
+func (r *RedisServer) keyTypeAtExpectFast(ctx context.Context, key []byte, readTS uint64, expected redisValueType) (redisValueType, error) {
+	if expected == redisTypeNone {
+		return redisTypeNone, nil
+	}
+	found, err := r.probeExpectedType(ctx, key, readTS, expected)
+	if err != nil {
+		return redisTypeNone, err
+	}
+	if !found {
+		return redisTypeNone, nil
+	}
+	return r.applyTTLFilter(ctx, key, readTS, expected)
+}
+
 // probeExpectedType issues only the prefix probes for the given type.
 // It is intentionally conservative: returning false here means "no row
 // of the expected type was visible at readTS", not "the key does not