adapter: Retry forwarded transaction conflicts - #1121
Conversation
📝 WalkthroughWalkthroughgRPC経由で型情報が失われたRedisトランザクションエラーを解析し、書き込み競合・ロック中エラーとして復元する処理を追加しました。復元結果をリトライ判定、ポリシー選択、エラー正規化へ反映し、関連する再試行とキー秘匿のテストを追加しています。 ChangesRedis wireエラーリトライ対応
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Redis as Redis操作
participant Adapter as Redisリトライ処理
participant Parser as wireエラー解析
Redis->>Adapter: gRPCステータス付きトランザクションエラー
Adapter->>Parser: エラー種別と論理キーを解析
Parser-->>Adapter: 型付きエラーを返却
Adapter->>Adapter: リトライポリシーを選択
Adapter->>Redis: 再試行
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://cdn.github.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a mechanism to classify and retry transaction errors (such as write conflicts and transaction locks) that cross gRPC boundaries as untyped gRPC status errors. It adds wireRedisTxnErrKind to parse these errors and includes comprehensive unit tests. The review feedback highlights a potential security issue where internal keys could be leaked to clients if the retry limit is exceeded, and suggests parsing and normalizing these wire errors before returning them.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| func wireRedisTxnErrKind(err error) redisTxnWireErrKind { | ||
| type grpcStatusCarrier interface { | ||
| GRPCStatus() *status.Status | ||
| } | ||
| var carrier grpcStatusCarrier | ||
| if !errors.As(err, &carrier) { | ||
| return redisTxnWireErrNone | ||
| } | ||
| st := carrier.GRPCStatus() | ||
| if st == nil || (st.Code() != codes.Unknown && st.Code() != codes.Aborted) { | ||
| return redisTxnWireErrNone | ||
| } | ||
| msg := st.Message() | ||
| if !strings.HasPrefix(msg, "key: ") { | ||
| return redisTxnWireErrNone | ||
| } | ||
| switch { | ||
| case strings.HasSuffix(msg, ": "+store.ErrWriteConflict.Error()): | ||
| return redisTxnWireErrWriteConflict | ||
| case strings.HasSuffix(msg, ": "+kv.ErrTxnLocked.Error()): | ||
| return redisTxnWireErrLocked | ||
| default: | ||
| return redisTxnWireErrNone | ||
| } | ||
| } |
There was a problem hiding this comment.
Issue: Leaking Internal Keys to Clients on Retry Limit Exceeded
When a forwarded transaction conflict occurs, the error returned over gRPC is a codes.Unknown or codes.Aborted status containing the internal key (e.g., !txn|cmt|... or !redis|...).
Currently, normalizeRetryableRedisTxnErr only matches typed TxnLockedError and WriteConflictError using kv.TxnLockedDetails and store.WriteConflictKey. It does not match or normalize these gRPC status errors. As a result, if the retry limit is exceeded, the raw gRPC status error containing the internal key is returned directly to the client, leaking internal storage layout details.
Solution
We can implement a parser parseWireRedisTxnErr that extracts the key and detail from the gRPC status message, normalizes the key, and returns a typed, normalized error.
To fully resolve this, please also update normalizeRetryableRedisTxnErr (located around line 153) to call this parser at the beginning:
func normalizeRetryableRedisTxnErr(err error) error {
if parsed := parseWireRedisTxnErr(err); parsed != nil {
return parsed
}
// ... existing typed checks ...
}func parseWireRedisTxnErr(err error) error {
type grpcStatusCarrier interface {
GRPCStatus() *status.Status
}
var carrier grpcStatusCarrier
if !errors.As(err, &carrier) {
return nil
}
st := carrier.GRPCStatus()
if st == nil || (st.Code() != codes.Unknown && st.Code() != codes.Aborted) {
return nil
}
msg := st.Message()
if !strings.HasPrefix(msg, "key: ") {
return nil
}
if strings.HasSuffix(msg, ": "+store.ErrWriteConflict.Error()) {
keyStr := msg[len("key: ") : len(msg)-len(": "+store.ErrWriteConflict.Error())]
logicalKey := normalizeRetryableRedisTxnKey([]byte(keyStr))
return errors.WithStack(store.NewWriteConflictError(logicalKey))
}
if strings.HasSuffix(msg, ": "+kv.ErrTxnLocked.Error()) {
content := msg[len("key: ") : len(msg)-len(": "+kv.ErrTxnLocked.Error())]
var keyStr string
var detail string
if strings.HasSuffix(content, ")") {
if idx := strings.LastIndex(content, " ("); idx != -1 {
keyStr = content[:idx]
detail = content[idx+2 : len(content)-1]
} else {
keyStr = content
}
} else {
keyStr = content
}
logicalKey := normalizeRetryableRedisTxnKey([]byte(keyStr))
if detail != "" {
return errors.WithStack(kv.NewTxnLockedErrorWithDetail(logicalKey, detail))
}
return errors.WithStack(kv.NewTxnLockedError(logicalKey))
}
return nil
}
func wireRedisTxnErrKind(err error) redisTxnWireErrKind {
parsed := parseWireRedisTxnErr(err)
if parsed == nil {
return redisTxnWireErrNone
}
if errors.Is(parsed, store.ErrWriteConflict) {
return redisTxnWireErrWriteConflict
}
if errors.Is(parsed, kv.ErrTxnLocked) {
return redisTxnWireErrLocked
}
return redisTxnWireErrNone
}There was a problem hiding this comment.
Addressed in 962ea1f. Forwarded transaction statuses are now parsed into normalized typed errors before retry policy selection or terminal return, so retry-limit errors expose only the logical Redis key. I also audited every retryRedisWrite caller; they propagate the returned error and none depends on the original gRPC status or internal key text. Added direct retry-limit and TxnLocked-detail regression coverage; targeted race tests pass 10 times and adapter lint reports 0 issues.
![coderabbitai[bot]](https://cdn.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (2)
adapter/redis_retry_test.go (1)
370-389: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueサブテストの名前付け(
t.Run)がない点が他のテーブル駆動テストと不整合。同ファイルの他のテーブル駆動テスト（Line 327-368、410-455）は
t.Run(tt.name, ...)でケースを分離しているが、ここはfor _, wireErr := range testsのみでケース名がなく、失敗時にどのケースが落ちたか特定しにくい。As per coding guidelines,**/*_test.goにおいて「Unit tests must be co-located with packages in *_test.go files; prefer table-driven test cases」とされているため、他ケースと同様に名前付きサブテストにすると一貫性が高まる。♻️ 提案
- tests := []error{ - status.Error(codes.Unknown, "write conflict"), - status.Error(codes.Unknown, "key: user-visible failure"), - status.Error(codes.Internal, "key: k: write conflict"), - } - for _, wireErr := range tests { - attempts := 0 - srv := &RedisServer{} - err := srv.retryRedisWrite(context.Background(), func() error { - attempts++ - return wireErr - }) - - require.ErrorIs(t, err, wireErr) - require.Equal(t, 1, attempts) - } + tests := []struct { + name string + wireErr error + }{ + {"plain unknown", status.Error(codes.Unknown, "write conflict")}, + {"unknown non-matching envelope", status.Error(codes.Unknown, "key: user-visible failure")}, + {"internal code with matching message", status.Error(codes.Internal, "key: k: write conflict")}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + attempts := 0 + srv := &RedisServer{} + err := srv.retryRedisWrite(context.Background(), func() error { + attempts++ + return tt.wireErr + }) + + require.ErrorIs(t, err, tt.wireErr) + require.Equal(t, 1, attempts) + }) + }🤖 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_retry_test.go` around lines 370 - 389, Update TestRetryRedisWriteDoesNotRetryUnclassifiedWireErrors to use named table-driven cases and execute each with t.Run. Add descriptive names for each wire error scenario, then move the per-case setup, retry invocation, and assertions into the subtest so failures identify the specific case.Source: Coding guidelines
adapter/redis_retry.go (1)
59-133: 🎯 Functional Correctness | 🔵 Trivial | 💤 Low valuewire封筒の復元ロジックは妥当。
wireRedisTxnStatus/parseWireRedisTxnErr/wireRedisTxnErrKindはkv/txn_errors.goとstore/store.goのエラーフォーマット（key: %s: %v/key: %s (%s): %v）と正確に一致しており、コードもUnknown/Abortedに限定した厳密一致方式になっている。設計として妥当。1点、Line 107-114 の detail 分離ロジックは、論理キー自体が
" ("を含んだり")"で終わったりするケースを想定していない（稀な入力では detail 抽出を誤り、キーに丸括弧が残る可能性がある）。実害は小さく、クラッシュもしない安全なフォールバックだが、コメントで明記しておくと将来のメンテで気づきやすい。🤖 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_retry.go` around lines 59 - 133, Add a concise comment near the detail-splitting logic in parseWireRedisTxnErr documenting that keys containing " (" or ending with ")" can be ambiguously parsed, leaving parentheses in the key as a safe fallback. Preserve the current parsing behavior and error reconstruction.
🤖 Prompt for all review comments with 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.
Nitpick comments:
In `@adapter/redis_retry_test.go`:
- Around line 370-389: Update
TestRetryRedisWriteDoesNotRetryUnclassifiedWireErrors to use named table-driven
cases and execute each with t.Run. Add descriptive names for each wire error
scenario, then move the per-case setup, retry invocation, and assertions into
the subtest so failures identify the specific case.
In `@adapter/redis_retry.go`:
- Around line 59-133: Add a concise comment near the detail-splitting logic in
parseWireRedisTxnErr documenting that keys containing " (" or ending with ")"
can be ambiguously parsed, leaving parentheses in the key as a safe fallback.
Preserve the current parsing behavior and error reconstruction.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 8cefc1db-7613-4785-a2a0-01d1cc6baff0
📒 Files selected for processing (2)
adapter/redis_retry.goadapter/redis_retry_test.go
![chatgpt-codex-connector[bot]](https://cdn.github.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: b5b15b16f4
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| return errors.Is(err, store.ErrWriteConflict) || errors.Is(err, kv.ErrTxnLocked) | ||
| return errors.Is(err, store.ErrWriteConflict) || | ||
| errors.Is(err, kv.ErrTxnLocked) || | ||
| wireRedisTxnErrKind(err) != redisTxnWireErrNone |
There was a problem hiding this comment.
When a forwarded write conflict reaches the Redis transaction/list dedup paths, this new boolean classification lets firstExecAttempt/list-push save and retry the old pending write set, but the reuse handlers still test errors.Is(dispErr, store.ErrWriteConflict) before dropping stale state (dispatchExecReuse and dispatchListPushReuse). The wire status does not satisfy that typed check, so a genuine conflict on a follower-forwarded reuse attempt is treated as “ambiguous”, advances pending.commitTS, and keeps retrying the stale snapshot until the retry limit instead of recomputing from a fresh read.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 7c4e8a6. The EXEC and list-push first/reuse dispatch paths now normalize forwarded transaction errors before retry gating and before their typed WriteConflict branches. A wire conflict on reuse therefore runs the existing self-commit probe; a probe miss drops the stale pending write set and rebuilds from a fresh snapshot. The genuine-conflict EXEC/list regression tests now inject keyed gRPC wire statuses and still require the exact three-dispatch rebuild topology. I also audited every retryRedisWrite caller: only reuse-aware EXEC, list push, and XADD explicitly promote wire write conflicts to retryable typed errors; all other callers fail closed after key normalization.
|
|
||
| if content, ok := strings.CutSuffix(strings.TrimPrefix(msg, "key: "), ": "+store.ErrWriteConflict.Error()); ok { | ||
| logicalKey := normalizeRetryableRedisTxnKey([]byte(content)) | ||
| return errors.WithStack(store.NewWriteConflictError(logicalKey)) |
There was a problem hiding this comment.
When a follower-forwarded XADD hits the ambiguous path where kv.LeaderProxy.runForwardCycle loses a response after the leader applied the request and a later internal retry reports the self-conflict as a gRPC status, this reconstruction turns that final wire write conflict into a normal retryable ErrWriteConflict. retryRedisWrite then reruns xaddTxn, which resolves * from the advanced stream meta and appends a second entry instead of surfacing the uncertain first attempt; the list/EXEC paths have reusable pending state for this class, but XADD has no equivalent dedup.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Addressed in 7c4e8a6. Raw wire write conflicts are no longer replayed by the generic retry loop. With one-phase dedup enabled, XADD now captures its generated ID, exact write set, commitTS, and coordinator primary probe key; retries reuse that write set with PrevCommitTS, and a fresh self-conflict is probed before any recompute. The landed-then-wire-error regression returns the original ID after an exact-ts no-op and leaves stream length at one. The rollback path without dedup returns a normalized error after one dispatch instead of replaying. Targeted race tests passed 10 times, go test ./... passed, and repository lint reports 0 issues.
|
@codex review Please review the latest HEAD 7c4e8a6. The two P1 findings are addressed: forwarded write conflicts are fail-closed in the generic retry loop, while EXEC/list push normalize them inside exact write-set reuse, and XADD now retries through fixed-ID write-set reuse with PrevCommitTS and a self-conflict probe. All retryRedisWrite callers were audited. Validation: targeted race regressions passed 10 times, go test ./adapter passed, go test ./... passed, and repository lint reports 0 issues. |
|
Codex Review: Didn't find any major issues. 🚀 Reviewed commit: ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
If Codex has suggestions, it will comment; otherwise it will react with 👍. Codex can also answer questions or update the PR. Try commenting "@codex address that feedback". |
1 participant
Summary
Root cause
Internal leader forwarding converts typed transaction errors into gRPC statuses. The Redis retry layer originally recognized only the typed error chain, so a forwarded conflict could escape instead of being handled.
Blindly making every forwarded write conflict retryable is unsafe: if the leader applied an operation but its response was lost, a later forwarded self-conflict is ambiguous. Recomputing XADD with
*can then append a second entry, while EXEC and list-push reuse paths can retain stale pending state unless the wire error is normalized before their typed conflict handling.Behavior
TxnLockederrors remain retryable because the lock rejection did not apply the transaction.WriteConflicterrors are normalized and returned without replay.PrevCommitTS.Risk
Wire parsing remains restricted to
UnknownorAbortedstatuses with the exact server-generated keyed envelope. Retry promotion is limited to operations that retain an exact write set and use the one-phase exact-timestamp probe.Validation
go test -race ./adapter -run '^(TestRedisXAddRetriesWireWriteConflict|TestRedisXAddDedupsLandedWireWriteConflict|TestRedisXAddDedupDisabledDoesNotReplayLandedWireConflict|TestRetryRedisWriteRetriesWireTxnLocked|TestRetryRedisWriteDoesNotRetryUnclassifiedWireErrors|TestRetryRedisWriteDoesNotReplayOrLeakWireWriteConflict|TestNormalizeRetryableRedisTxnErrWireErrors|TestListPushDedup_GenuineConflictRecomputes|TestExecDedup_GenuineConflictRebuildsAndApplies)$' -count=10go test ./adaptergo test ./...golangci-lint run ./... --timeout=5m