Merge branch 'main' into fix/admin-rolling-env-forward

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

docs/admin_ui_key_visualizer_design.md

@@ -319,11 +319,13 @@ Because writes are recorded by Raft leaders and follower-local reads are recorde
 |---|---|---|
 | 0 | `cmd/elastickv-admin` skeleton, token-protected `Admin` gRPC service stub, empty SPA shell, CI wiring. | Binary builds, `/api/cluster/overview` returns live data from a real node only when the configured admin token is supplied. |
 | 1 | Overview, Routes, Raft Groups, Adapters pages. `LiveSummary` added. No sampler. | All read-only pages match `grpcurl` ground truth. |
-| 2 | Key Visualizer MVP: in-memory sampler with adaptive sub-sampling, leader writes, leader/follower reads, fan-out across nodes, static matrix API with virtual-bucket metadata. | Benchmark gate green; heatmap shows synthetic hotspot within 2 s of load; ±5% / 95%-CI accuracy SLO holds under synthetic bursts; fan-out returns complete view with 1 node down. |
-| 3 | Bytes series, drill-down, split/merge continuity, namespace-isolated persistence of compacted columns distributed **per owning Raft group**, lineage recovery, and retention GC. | Heatmap remains continuous across a live `SplitRange`; restart preserves last 7 days; expired data and stale lineage records are collected; no single Raft group sees more than its share of KeyViz writes. |
+| 2-A | Key Visualizer MVP server side: in-memory sampler with adaptive sub-sampling, leader writes, leader/follower reads, static matrix API with virtual-bucket metadata. | Benchmark gate green; ±5% / 95%-CI accuracy SLO holds under synthetic bursts; matrix endpoint returns the local node's view. |
+| 2-B | KeyViz SPA integration into `web/admin/`: heatmap page, series picker, row budget, manual + auto refresh. See `docs/design/2026_04_27_proposed_keyviz_spa_integration.md`. | Heatmap shows synthetic hotspot within ~5 s of `make client` driving traffic against `make run`; type check (`tsc -b --noEmit`) clean. |
+| 2-C | Cluster fan-out: admin RPC that aggregates each node's local sampler view so the SPA shows a cluster-wide heatmap rather than the local node's slice. | Fan-out returns complete view with 1 node down; SPA renders aggregate within the §10 budget. |
+| 3 | Drill-down, split/merge continuity, namespace-isolated persistence of compacted columns distributed **per owning Raft group**, lineage recovery, and retention GC. | Heatmap remains continuous across a live `SplitRange`; restart preserves last 7 days; expired data and stale lineage records are collected; no single Raft group sees more than its share of KeyViz writes. |
 | 4 (deferred) | Mutating admin operations (`SplitRange` from UI), browser login, RBAC, and identity-provider integration. Out of scope for this design; a follow-up design will cover it. | — |
 
-Phases 0–2 are the minimum operationally useful product; Phase 3 is the "ship-quality" target.
+Phases 0–2 (A/B/C) together are the minimum operationally useful product; Phase 3 is the "ship-quality" target. As of 2026-04-27, Phase 2-A is shipped (PRs #639/#645/#646/#647/#651/#660/#661/#672), Phase 2-B lands with this proposal, and Phase 2-C is open. Bytes series, originally listed under Phase 3, was rolled forward into 2-A and is already on the wire.
 
 ## 13. Open Questions