Skip to content

refactor(redis): bound rangeList / readValueAt VerifyLeaderForKey to per-call ctx #752

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bootjp wants to merge 1 commit into
base: main
Choose a base branch
from
+25 −8
Open

refactor(redis): bound rangeList / readValueAt VerifyLeaderForKey to per-call ctx #752

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 24 additions & 7 deletions adapter/redis.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1761,7 +1761,13 @@ type txnValue struct {
}

type txnContext struct {
server *RedisServer
server *RedisServer
// ctx is the per-EXEC dispatch context (redisDispatchTimeout-bounded
// at the call site in runTransaction). Plumbed through so reads
// inside the EXEC such as load() → readValueAt() respect the
// caller's deadline rather than falling back to handlerContext +
// the verifyLeaderEngineCtx safety net.
ctx context.Context //nolint:containedctx // EXEC is a long-lived value type that wraps a single client command, ctx must travel with it.
Comment on lines +1764 to +1770
Copy link
Copy Markdown

@coderabbitai coderabbitai Bot May 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Find every txnContext composite literal in the repo and dump 12 lines after
# so we can see which fields are initialised.
rg -nPU --type=go -A 12 '(?s)&?txnContext\s*\{' adapter
echo '---'
# Cross-check call sites of stageKeyDeletion and load(internalKey) reachable
# from those fixtures.
rg -n --type=go '\bstageKeyDeletion\s*\(|\.load\s*\(' adapter

Repository: bootjp/elastickv

Length of output: 4263

🏁 Script executed:

# Check if loadListState calls into stageKeyDeletion or load
rg -n --type=go -A 15 'func \(t \*txnContext\) loadListState' adapter/redis.go | head -30

# Also check stageKeyDeletion to see what it does
rg -n --type=go -A 20 'func \(t \*txnContext\) stageKeyDeletion' adapter/redis.go | head -40

Repository: bootjp/elastickv

Length of output: 1429

🏁 Script executed:

rg -n --type=go -A 50 'func \(t \*txnContext\) stageKeyDeletion' adapter/redis.go | head -60

Repository: bootjp/elastickv

Length of output: 1953

🏁 Script executed:

rg -n --type=go -A 8 'func \(t \*txnContext\) load\(key \[\]byte\)' adapter/redis.go | head -20

Repository: bootjp/elastickv

Length of output: 477

🏁 Script executed:

rg -n --type=go -A 25 'func \(t \*txnContext\) load\(key \[\]byte\)' adapter/redis.go | grep -A 25 "1847:"

Repository: bootjp/elastickv

Length of output: 1009

🏁 Script executed:

rg -n --type=go -A 40 'func \(t \*txnContext\) load\(key \[\]byte\)' adapter/redis.go | tail -20

Repository: bootjp/elastickv

Length of output: 639

🏁 Script executed:

rg -n --type=go 'func.*readValueAt' adapter/redis.go

Repository: bootjp/elastickv

Length of output: 165

🏁 Script executed:

sed -n '1877,1883p' adapter/redis.go | cat -n

Repository: bootjp/elastickv

Length of output: 302

Add a nil-safe fallback for t.ctx in load() to support test fixtures that omit the field.

Test fixtures in redis_txn_test.go (lines 37, 73, 108, 146) construct txnContext literals without ctx. When stageKeyDeletion() is called—for example, during key deletion in a transaction—it invokes t.load(internalKey) in a loop (lines 2380–2392), which passes t.ctx to readValueAt() at line 1879. If ctx is nil, any coordinator implementation or mock that uses context.WithTimeout(ctx, …) would panic.

Currently, test fixtures only call loadListState(), which uses context.Background() internally, so the issue is latent. However, this matches the existing pattern used for streamDeletions (line 2399): "test fixtures that build a minimal txnContext literal without this field still work." A one-line defensive fallback in load() preserves that design principle and adds robustness:

Suggested fix 
 	} else {
 		var err error
-		val, err = t.server.readValueAt(t.ctx, storageKey, t.startTS)
+		ctx := t.ctx
+		if ctx == nil {
+			ctx = context.Background()
+		}
+		val, err = t.server.readValueAt(ctx, storageKey, t.startTS)
 		if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
 			return nil, errors.WithStack(err)
 		}
 	}
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@adapter/redis.go` around lines 1764 - 1770, The load() method should
defensively handle a nil txnContext.ctx to avoid panics in tests that construct
txnContext literals without ctx; inside load() (the caller used by
stageKeyDeletion() which loops and calls readValueAt()), check if t.ctx is nil
and substitute context.Background() (or another appropriate base context) before
passing it into readValueAt(); this mirrors the pattern used by
streamDeletions/loadListState and ensures mocks that call context.WithTimeout
won't panic.

working map[string]*txnValue
listStates map[string]*listTxnState
zsetStates map[string]*zsetTxnState
Expand Down Expand Up @@ -1870,7 +1876,7 @@ func (t *txnContext) load(key []byte) (*txnValue, error) {
tv.ttl = ttl
} else {
var err error
val, err = t.server.readValueAt(storageKey, t.startTS)
val, err = t.server.readValueAt(t.ctx, storageKey, t.startTS)
if err != nil && !errors.Is(err, store.ErrKeyNotFound) {
return nil, errors.WithStack(err)
}
Expand Down Expand Up @@ -2797,6 +2803,7 @@ func (r *RedisServer) runTransaction(queue []redcon.Command) ([]redisResult, err

txn := &txnContext{
server: r,
ctx: dispatchCtx,
working: map[string]*txnValue{},
listStates: map[string]*listTxnState{},
zsetStates: map[string]*zsetTxnState{},
Expand Down Expand Up @@ -3244,7 +3251,7 @@ func (r *RedisServer) fetchListRange(ctx context.Context, key []byte, meta store
return out, nil
}

func (r *RedisServer) rangeList(key []byte, startRaw, endRaw []byte) ([]string, error) {
func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRaw []byte) ([]string, error) {
Copy link
Copy Markdown
Contributor

@gemini-code-assist gemini-code-assist Bot May 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

While ctx is now passed to rangeList, it is only used for the VerifyLeaderForKey call. Other context-aware operations in this method, such as keyTypeAt (line 3260), resolveListMeta (line 3279), and fetchListRange (line 3292), still use context.Background(). To fully adhere to the goal of bounding the command execution and ensuring prompt cancellation, these calls should also use the provided ctx. Additionally, per the rules for linearizable reads in MVCC, ensure the timestamp for the read is acquired after the linearizable read fence (VerifyLeaderForKey) to guarantee a consistent snapshot and prevent stale data reads.

References
  1. When designing interfaces, use context.Context for managing deadlines and cancellation instead of separate timeout parameters.
  2. When implementing linearizable reads in an MVCC system, ensure that the timestamp used for the MVCC read is acquired after the linearizable read fence has completed to guarantee a consistent snapshot view, especially for rangeList.

if !r.coordinator.IsLeaderForKey(key) {
return r.proxyLRange(key, startRaw, endRaw)
}
Expand All @@ -3261,7 +3268,11 @@ func (r *RedisServer) rangeList(key []byte, startRaw, endRaw []byte) ([]string,
return nil, wrongTypeError()
}

if err := r.coordinator.VerifyLeaderForKey(r.handlerContext(), key); err != nil {
// PR #749 follow-up: pass the per-call dispatch ctx so a stalled
// VerifyLeaderForKey honours the caller's deadline rather than the
// long-lived handlerContext + verifyLeaderEngineCtx fallback. Same
// shape as keys() / FLUSHDB.
if err := r.coordinator.VerifyLeaderForKey(ctx, key); err != nil {
return nil, errors.WithStack(err)
}

Expand Down Expand Up @@ -3498,7 +3509,7 @@ func (r *RedisServer) tryLeaderGetAt(key []byte, ts uint64) ([]byte, error) {
return resp.Value, nil
}

func (r *RedisServer) readValueAt(key []byte, readTS uint64) ([]byte, error) {
func (r *RedisServer) readValueAt(ctx context.Context, key []byte, readTS uint64) ([]byte, error) {
Copy link
Copy Markdown
Contributor

@gemini-code-assist gemini-code-assist Bot May 10, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Similar to rangeList, readValueAt now receives a ctx but continues to use context.Background() for hasExpired (line 3522) and store.GetAt (line 3538). These should be updated to use the passed ctx to ensure the entire read path respects the dispatch deadline. Furthermore, ensure that the timestamp used for the MVCC read is acquired after the linearizable read fence has completed to guarantee a consistent snapshot view and prevent stale data reads, as required for readValueAt.

References
  1. When designing interfaces, use context.Context for managing deadlines and cancellation instead of separate timeout parameters.
  2. When implementing linearizable reads in an MVCC system, ensure that the timestamp used for the MVCC read is acquired after the linearizable read fence has completed to guarantee a consistent snapshot view, especially for readValueAt.

ttlKey := key
nonStringInternal := false
if userKey := extractRedisInternalUserKey(key); userKey != nil {
Expand All @@ -3517,7 +3528,11 @@ func (r *RedisServer) readValueAt(key []byte, readTS uint64) ([]byte, error) {
}

if r.coordinator.IsLeaderForKey(key) {
if err := r.coordinator.VerifyLeaderForKey(r.handlerContext(), key); err != nil {
// PR #749 follow-up: caller-supplied ctx (with
// redisDispatchTimeout from the dispatch handler) replaces
// r.handlerContext() so VerifyLeaderForKey honours the
// per-command deadline. Same shape as keys() / FLUSHDB.
if err := r.coordinator.VerifyLeaderForKey(ctx, key); err != nil {
return nil, errors.WithStack(err)
}
v, err := r.store.GetAt(context.Background(), key, readTS)
Expand Down Expand Up @@ -3567,7 +3582,9 @@ func (r *RedisServer) rpush(conn redcon.Conn, cmd redcon.Command) {
}

func (r *RedisServer) lrange(conn redcon.Conn, cmd redcon.Command) {
items, err := r.rangeList(cmd.Args[1], cmd.Args[2], cmd.Args[3])
ctx, cancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout)
defer cancel()
items, err := r.rangeList(ctx, cmd.Args[1], cmd.Args[2], cmd.Args[3])
if err != nil {
conn.WriteError(err.Error())
return
Expand Down
2 changes: 1 addition & 1 deletion adapter/redis_retry_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -240,7 +240,7 @@ func TestRedisEvalRetriesWriteConflict(t *testing.T) {
require.Equal(t, "v1", string(conn.bulk))
require.Equal(t, 2, coord.dispatches)

rawValue, err := srv.readValueAt(redisStrKey([]byte("retry:lua")), snapshotTS(coord.clock, st))
rawValue, err := srv.readValueAt(context.Background(), redisStrKey([]byte("retry:lua")), snapshotTS(coord.clock, st))
require.NoError(t, err)
value, _, err := decodeRedisStr(rawValue)
require.NoError(t, err)
Expand Down
Loading