fix(redis): address review feedback for keyTypeAtExpect

Address findings from claude review on 79a6989:

- [P0] Add hasHigherPriorityStringEncoding guard on the fast-path
  hit in keyTypeAtExpect, consistent with the wide-column fast-path
  callers (hashFieldFastLookup, zsetMemberFastScore,
  setMemberFastExists, hashFieldFastExists). Without this guard the
  rawKeyTypeAt "string wins" tiebreaker is bypassed when a
  redisStrKey row exists alongside a collection encoding for the
  same user key, returning the collection-specific answer instead
  of the WRONGTYPE / nil that the slow path produces. Cost: one
  extra ExistsAt seek per fast-path hit on non-string types; the
  19->3/4 reduction is preserved.
- [P1] Replace context.Background() with the in-scope ctx in eight
  keyTypeAtExpect call sites whose enclosing function (or
  retryRedisWrite closure) carries a real context with timeout /
  cancellation:
    applyHashFieldPairs (closure ctx from L2040 WithTimeout)
    hdelTxn (function param)
    hincrbyTxn (function param)
    zaddTxn (function param)
    zincrbyTxn (function param)
    zrem closure (closure ctx from L3540 WithTimeout)
    zremrangebyrank closure (closure ctx from L3588 WithTimeout)
    tryBZPopMin closure (closure ctx from L3625 WithTimeout)
  Pre-existing bug on the keyTypeAt callers, now fixed at the same
  lines this PR already touches.

Callers without ctx in scope (hmget, hlen, hgetall, zrangeRead,
xlen, rangeStream) are pre-existing and out of scope.

Author: bootjp

adapter/redis_compat_commands.go

@@ -2042,7 +2042,7 @@ func (r *RedisServer) applyHashFieldPairs(key []byte, args [][]byte) (int, error
 	var added int
 	err := r.retryRedisWrite(ctx, func() error {
 		readTS := r.readTS()
-		typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeHash)
+		typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeHash)
 		if err != nil {
 			return err
 		}
@@ -2669,7 +2669,7 @@ func (r *RedisServer) resolveHashFieldDelElems(ctx context.Context, key []byte,
 
 func (r *RedisServer) hdelTxn(ctx context.Context, key []byte, fields [][]byte) (int, error) {
 	readTS := r.readTS()
-	typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeHash)
+	typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeHash)
 	if err != nil {
 		return 0, err
 	}
@@ -2941,7 +2941,7 @@ func (r *RedisServer) hincrbyWithMigration(ctx context.Context, key, fieldKey []
 
 func (r *RedisServer) hincrbyTxn(ctx context.Context, key, field []byte, increment int64) (int64, error) {
 	readTS := r.readTS()
-	typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeHash)
+	typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeHash)
 	if err != nil {
 		return 0, err
 	}
@@ -3262,7 +3262,7 @@ func (r *RedisServer) applyZAddPair(ctx context.Context, key []byte, p zaddPair,
 
 func (r *RedisServer) zaddTxn(ctx context.Context, key []byte, flags zaddFlags, pairs []zaddPair) (int, error) {
 	readTS := r.readTS()
-	typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeZSet)
+	typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeZSet)
 	if err != nil {
 		return 0, err
 	}
@@ -3332,7 +3332,7 @@ func (r *RedisServer) zaddTxn(ctx context.Context, key []byte, flags zaddFlags,
 // Returns the new score after applying increment.
 func (r *RedisServer) zincrbyTxn(ctx context.Context, key []byte, member string, increment float64) (float64, error) {
 	readTS := r.readTS()
-	typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeZSet)
+	typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeZSet)
 	if err != nil {
 		return 0, err
 	}
@@ -3542,7 +3542,7 @@ func (r *RedisServer) zrem(conn redcon.Conn, cmd redcon.Command) {
 	var removed int
 	if err := r.retryRedisWrite(ctx, func() error {
 		readTS := r.readTS()
-		typ, err := r.keyTypeAtExpect(context.Background(), cmd.Args[1], readTS, redisTypeZSet)
+		typ, err := r.keyTypeAtExpect(ctx, cmd.Args[1], readTS, redisTypeZSet)
 		if err != nil {
 			return err
 		}
@@ -3590,7 +3590,7 @@ func (r *RedisServer) zremrangebyrank(conn redcon.Conn, cmd redcon.Command) {
 	var removed int
 	if err := r.retryRedisWrite(ctx, func() error {
 		readTS := r.readTS()
-		typ, err := r.keyTypeAtExpect(context.Background(), cmd.Args[1], readTS, redisTypeZSet)
+		typ, err := r.keyTypeAtExpect(ctx, cmd.Args[1], readTS, redisTypeZSet)
 		if err != nil {
 			return err
 		}
@@ -3627,7 +3627,7 @@ func (r *RedisServer) tryBZPopMin(key []byte) (*bzpopminResult, error) {
 	var result *bzpopminResult
 	err := r.retryRedisWrite(ctx, func() error {
 		readTS := r.readTS()
-		typ, err := r.keyTypeAtExpect(context.Background(), key, readTS, redisTypeZSet)
+		typ, err := r.keyTypeAtExpect(ctx, key, readTS, redisTypeZSet)
 		if err != nil {
 			return err
 		}

adapter/redis_compat_helpers.go

@@ -319,16 +319,27 @@ func (r *RedisServer) keyTypeAt(ctx context.Context, key []byte, readTS uint64)
 // seeks across every collection family before returning. The fast path:
 //
 //  1. Probe only the prefixes for `expected` (typically 2-3 seeks).
-//  2. On hit, apply the TTL filter and return `expected`.
+//  2. On hit, run the same string-priority guard the wide-column
+//     fast-path callers use (hashFieldFastLookup, zsetMemberFastScore,
+//     setMemberFastExists, hashFieldFastExists). When a redisStrKey
+//     row also exists at the same user key, fall back to the slow
+//     path so the rawKeyTypeAt "string wins" tiebreaker fires and
+//     the caller gets WRONGTYPE / nil instead of the
+//     collection-specific answer. The guard is the narrow form (see
+//     hasHigherPriorityStringEncoding's doc comment): only redisStrKey
+//     is checked, the rarer HLL / legacy-bare-key dual-encoding cases
+//     remain a known residual risk shared with the other fast-path
+//     callers.
 //  3. On miss, fall back to the full keyTypeAt slow path so that
-//     wrongType collisions (the key exists under a different type) still
-//     surface as the correct redisValueType.
+//     wrongType collisions (the key exists under a different type)
+//     still surface as the correct redisValueType.
 //
 // Steady-state production: most XADD/XREAD/HSET/etc. calls are on a key
 // of the expected type, so step 1 hits and the slow-path 19 seeks shrink
-// to 2-3. The slow path stays in place for first-write and wrongType
-// cases, which keep their existing semantics — wrongTypeError detection
-// is preserved by the fall-through.
+// to 2-3 (plus the priority-guard ExistsAt). The slow path stays in
+// place for first-write and wrongType cases, which keep their existing
+// semantics — wrongTypeError detection is preserved by the
+// fall-through.
 func (r *RedisServer) keyTypeAtExpect(ctx context.Context, key []byte, readTS uint64, expected redisValueType) (redisValueType, error) {
 	if expected == redisTypeNone {
 		return r.keyTypeAt(ctx, key, readTS)
@@ -337,10 +348,19 @@ func (r *RedisServer) keyTypeAtExpect(ctx context.Context, key []byte, readTS ui
 	if err != nil {
 		return redisTypeNone, err
 	}
-	if found {
-		return r.applyTTLFilter(ctx, key, readTS, expected)
+	if !found {
+		return r.keyTypeAt(ctx, key, readTS)
+	}
+	if expected != redisTypeString {
+		higher, hErr := r.hasHigherPriorityStringEncoding(ctx, key, readTS)
+		if hErr != nil {
+			return redisTypeNone, hErr
+		}
+		if higher {
+			return r.keyTypeAt(ctx, key, readTS)
+		}
 	}
-	return r.keyTypeAt(ctx, key, readTS)
+	return r.applyTTLFilter(ctx, key, readTS, expected)
 }
 
 // probeExpectedType issues only the prefix probes for the given type.