From f1e10274afde92999c9bf79585b5ff7d6433c81b Mon Sep 17 00:00:00 2001
From: bootjp <contact@bootjp.me>
Date: Mon, 14 Sep 2026 17:40:48 +0900
Subject: [PATCH] adapter: stop the XREAD wrong-type test sharing one budget
 with registration

TestRedis_StreamXReadBlockChecksWrongTypeAtDeadline blocked for 2s and
waited up to 2s for the stream waiter to register before overwriting the
key. The wrong-type check runs when the block deadline fires, so a SET
that lands after it gets the ordinary block-timeout nil instead -- which
the test reports as

  "redis: nil" does not contain "WRONGTYPE"

On a loaded machine the registration wait can consume the whole window
before the SET is sent. Reproduced exactly by delaying the SET past the
deadline.

The block window is now 6s and the registration wait 1s, so the two no
longer share a budget: a slow registration fails as "the waiter never
appeared", with its own message, rather than surfacing later as a
confusing nil read. The registration timeout is a parameter now, since
any caller whose assertion depends on acting before the reader's deadline
has to bound it well under that deadline.

Claude-Session: https://claude.ai/code/session_013rNHooj7NF3giihWVba8QE
---
 adapter/redis_compat_commands_stream_test.go | 34 +++++++++++++++++---
 1 file changed, 29 insertions(+), 5 deletions(-)

diff --git a/adapter/redis_compat_commands_stream_test.go b/adapter/redis_compat_commands_stream_test.go
index 714159dd8..ca1445a9a 100644
--- a/adapter/redis_compat_commands_stream_test.go
+++ b/adapter/redis_compat_commands_stream_test.go
@@ -208,17 +208,32 @@ func TestRedis_StreamXReadBlockChecksWrongTypeAtDeadline(t *testing.T) {
 		streams []redis.XStream
 		err     error
 	}
+	// The block window has to be long enough that waiter registration plus the
+	// SET both land well inside it. The wrong-type check runs when the
+	// deadline fires, so a SET that arrives after it gets the ordinary
+	// block-timeout nil instead -- which is what this test then reports as
+	// `"redis: nil" does not contain "WRONGTYPE"`. At a 2s block that budget
+	// was shared with a registration wait also bounded at 2s, so a loaded
+	// machine could consume the whole window before the SET was sent.
+	const (
+		blockWindow       = 6 * time.Second
+		waiterRegistering = time.Second
+	)
+
 	resultCh := make(chan readResult, 1)
 	go func() {
 		streams, err := rdbReader.XRead(ctx, &redis.XReadArgs{
 			Streams: []string{key, "$"},
 			Count:   1,
-			Block:   2 * time.Second,
+			Block:   blockWindow,
 		}).Result()
 		resultCh <- readResult{streams: streams, err: err}
 	}()
 
-	requireStreamWaiterRegistered(t, nodes[0].redisServer.streamWaiters, key)
+	// Bounded well below the block window, and fatal on its own, so a slow
+	// registration fails as "the waiter never appeared" rather than surfacing
+	// later as a confusing nil read.
+	requireStreamWaiterRegistered(t, nodes[0].redisServer.streamWaiters, key, waiterRegistering)
 	require.NoError(t, rdbWriter.Set(ctx, key, "now-a-string", 0).Err())
 
 	select {
@@ -226,12 +241,20 @@ func TestRedis_StreamXReadBlockChecksWrongTypeAtDeadline(t *testing.T) {
 		require.Error(t, res.err)
 		require.Contains(t, res.err.Error(), "WRONGTYPE")
 		require.Empty(t, res.streams)
-	case <-time.After(4 * time.Second):
+	case <-time.After(blockWindow + 2*time.Second):
 		t.Fatal("XREAD BLOCK did not return after wrong-type overwrite")
 	}
 }
 
-func requireStreamWaiterRegistered(t *testing.T, reg *keyWaiterRegistry, key string) {
+// requireStreamWaiterRegistered waits for key's blocking reader to appear.
+//
+// The timeout is a parameter because a caller whose assertion depends on acting
+// BEFORE the reader's block deadline must bound this well under that deadline;
+// sharing one budget between the two is how a slow machine turned a wrong-type
+// assertion into a plain block timeout.
+func requireStreamWaiterRegistered(
+	t *testing.T, reg *keyWaiterRegistry, key string, timeout time.Duration,
+) {
 	t.Helper()
 	require.Eventually(t, func() bool {
 		if reg == nil {
@@ -240,7 +263,8 @@ func requireStreamWaiterRegistered(t *testing.T, reg *keyWaiterRegistry, key str
 		reg.mu.Lock()
 		defer reg.mu.Unlock()
 		return len(reg.waiters[key]) > 0
-	}, 2*time.Second, 10*time.Millisecond)
+	}, timeout, 10*time.Millisecond,
+		"stream waiter for %q never registered within %s", key, timeout)
 }
 
 // TestRedis_StreamCommandsRejectWrongType locks down the wrongType