adapter: Fence Redis reads before snapshots #1167
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -7,6 +7,7 @@ import ( | |
| "sort" | ||
| "strconv" | ||
| "strings" | ||
| "sync" | ||
| "time" | ||
|
|
||
| "github.com/bootjp/elastickv/kv" | ||
|
|
@@ -21,6 +22,10 @@ var redisTxnWideSetFencePrefix = []byte("!redis|txn-wide-set|") | |
| var redisTxnWideListFencePrefix = []byte("!redis|txn-wide-list|") | ||
| var redisTxnWideZSetFencePrefix = []byte("!redis|txn-wide-zset|") | ||
|
|
||
| const redisReadFenceLocalLeaderTarget = "\x00redis-read-fence-local-leader" | ||
|
|
||
| var errRedisExecSplitShardLeaders = errors.New("ERR EXEC read fence spans multiple shard leaders") | ||
|
|
||
| type txnCommandHandler func(*txnContext, redcon.Command) (redisResult, error) | ||
|
|
||
| var txnApplyHandlers = map[string]txnCommandHandler{ | ||
|
|
@@ -38,6 +43,211 @@ var txnApplyHandlers = map[string]txnCommandHandler{ | |
| cmdPExpire: (*txnContext).applyExpireMilliseconds, | ||
| } | ||
|
|
||
| func redisTxnReadFenceKeys(userKey []byte) [][]byte { | ||
| keys := [][]byte{ | ||
| redisStrKey(userKey), | ||
| redisHLLKey(userKey), | ||
| redisTTLKey(userKey), | ||
| listMetaKey(userKey), | ||
| store.ListMetaDeltaScanPrefix(userKey), | ||
| redisTxnWideListFenceKey(userKey), | ||
| redisHashKey(userKey), | ||
| store.HashMetaKey(userKey), | ||
| store.HashFieldScanPrefix(userKey), | ||
|
There was a problem hiding this comment.
When a hash, set, zset, or stream wide-column range is split after its scan-prefix key, this representative fences only the group owning the prefix start, although AGENTS.md reference: AGENTS.md:L24-L25 Useful? React with 👍 / 👎. |
||
| store.HashMetaDeltaScanPrefix(userKey), | ||
| redisTxnWideHashFenceKey(userKey), | ||
| redisSetKey(userKey), | ||
| store.SetMetaKey(userKey), | ||
| store.SetMemberScanPrefix(userKey), | ||
| store.SetMetaDeltaScanPrefix(userKey), | ||
| redisTxnWideSetFenceKey(userKey), | ||
| redisZSetKey(userKey), | ||
| store.ZSetMetaKey(userKey), | ||
| store.ZSetMemberScanPrefix(userKey), | ||
| store.ZSetScoreScanPrefix(userKey), | ||
| store.ZSetMetaDeltaScanPrefix(userKey), | ||
| redisTxnWideZSetFenceKey(userKey), | ||
| redisStreamKey(userKey), | ||
| store.StreamMetaKey(userKey), | ||
| store.StreamEntryScanPrefix(userKey), | ||
| } | ||
| if redisLegacyBareReadFenceAllowed(userKey) { | ||
| keys = append(keys, userKey) | ||
| } | ||
| return keys | ||
| } | ||
|
|
||
| func redisLegacyBareReadFenceAllowed(userKey []byte) bool { | ||
| if isKnownInternalKey(userKey) { | ||
| return false | ||
| } | ||
| return !bytes.HasPrefix(userKey, []byte("!sqs|")) | ||
| } | ||
|
|
||
| func redisQueuedCommandReadFenceKeys(queue []redcon.Command) [][]byte { | ||
| seen := make(map[string]struct{}, len(queue)) | ||
| keys := make([][]byte, 0, len(queue)) | ||
| appendKey := func(key []byte) { | ||
| keyID := string(key) | ||
| if _, ok := seen[keyID]; ok { | ||
| return | ||
| } | ||
| seen[keyID] = struct{}{} | ||
| keys = append(keys, key) | ||
| } | ||
| for _, cmd := range queue { | ||
| if len(cmd.Args) == 0 { | ||
| continue | ||
| } | ||
| meta, ok := redisCommandTable[strings.ToUpper(string(cmd.Args[0]))] | ||
| if !ok { | ||
| continue | ||
| } | ||
| for _, userKey := range redisCommandGetKeys(meta, cmd.Args) { | ||
| for _, fenceKey := range redisTxnReadFenceKeys(userKey) { | ||
| appendKey(fenceKey) | ||
| } | ||
| } | ||
| } | ||
| return keys | ||
| } | ||
|
|
||
| type redisTxnProxyRoute struct { | ||
| defaultLeader bool | ||
| key []byte | ||
| } | ||
|
|
||
| func (r *RedisServer) redisReadFenceGroupKeys(keys [][]byte) [][]byte { | ||
| if r == nil || r.coordinator == nil { | ||
| return nil | ||
| } | ||
| return kv.LeaseReadGroupKeys(r.coordinator, keys) | ||
| } | ||
|
|
||
| func (r *RedisServer) queuedCommandReadFenceGroupKeys(queue []redcon.Command) [][]byte { | ||
| return r.redisReadFenceGroupKeys(redisQueuedCommandReadFenceKeys(queue)) | ||
| } | ||
|
|
||
| func (r *RedisServer) readFenceProxyKey(groupKeys [][]byte) ([]byte, bool, error) { | ||
| if r == nil || r.coordinator == nil || len(groupKeys) == 0 { | ||
| return nil, false, nil | ||
| } | ||
|
|
||
| var targetLeader string | ||
| var proxyKey []byte | ||
| for _, key := range groupKeys { | ||
| leader, localLeader, err := r.readFenceLeader(key) | ||
| if err != nil { | ||
| return nil, false, err | ||
| } | ||
| if err := recordReadFenceTargetLeader(&targetLeader, leader); err != nil { | ||
| return nil, false, err | ||
| } | ||
| if localLeader || len(proxyKey) > 0 { | ||
| continue | ||
| } | ||
| proxyKey = key | ||
| } | ||
| if len(proxyKey) == 0 { | ||
| return nil, false, nil | ||
| } | ||
| return proxyKey, true, nil | ||
| } | ||
|
|
||
| func (r *RedisServer) readFenceLeader(key []byte) (string, bool, error) { | ||
| localLeader := r.coordinator.IsLeaderForKey(key) | ||
| if localLeader { | ||
| return redisReadFenceLocalLeaderTarget, true, nil | ||
| } | ||
| leader := r.coordinator.RaftLeaderForKey(key) | ||
| if leader == "" { | ||
| return "", false, ErrLeaderNotFound | ||
| } | ||
| return leader, false, nil | ||
| } | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| func recordReadFenceTargetLeader(target *string, leader string) error { | ||
| if leader == "" { | ||
| return nil | ||
| } | ||
| if *target == "" { | ||
| *target = leader | ||
| return nil | ||
| } | ||
| if *target != leader { | ||
| return errRedisExecSplitShardLeaders | ||
|
Comment on lines
+410
to
+411
There was a problem hiding this comment.
When a request reaches a node that is not the leader of either involved group, two groups led by the same remote process still normally report different Raft endpoints because each group has its own listener, while AGENTS.md reference: AGENTS.md:L25-L27 Useful? React with 👍 / 👎. |
||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (r *RedisServer) transactionProxyRoute(queue []redcon.Command) (redisTxnProxyRoute, error) { | ||
| if r == nil || r.coordinator == nil { | ||
| return redisTxnProxyRoute{}, nil | ||
| } | ||
|
|
||
| groupKeys := r.queuedCommandReadFenceGroupKeys(queue) | ||
| if len(groupKeys) == 0 { | ||
| if !r.coordinator.IsLeader() { | ||
| return redisTxnProxyRoute{defaultLeader: true}, nil | ||
| } | ||
| return redisTxnProxyRoute{}, nil | ||
| } | ||
|
|
||
| proxyKey, ok, err := r.readFenceProxyKey(groupKeys) | ||
| if err != nil { | ||
| return redisTxnProxyRoute{}, err | ||
| } | ||
| if ok { | ||
| return redisTxnProxyRoute{key: proxyKey}, nil | ||
| } | ||
| return redisTxnProxyRoute{}, nil | ||
| } | ||
|
|
||
| func (r *RedisServer) leaseRedisReadFenceGroups(ctx context.Context, groupKeys [][]byte) error { | ||
| if r == nil || r.coordinator == nil { | ||
| return nil | ||
| } | ||
| if len(groupKeys) == 0 { | ||
| return nil | ||
| } | ||
| if len(groupKeys) == 1 { | ||
| _, err := kv.LeaseReadForKeyThrough(r.coordinator, ctx, groupKeys[0]) | ||
| return errors.WithStack(err) | ||
| } | ||
|
|
||
| leaseCtx, cancel := context.WithCancel(ctx) | ||
| defer cancel() | ||
|
|
||
| errCh := make(chan error, len(groupKeys)) | ||
| var wg sync.WaitGroup | ||
| var cancelOnce sync.Once | ||
| for _, key := range groupKeys { | ||
| wg.Add(1) | ||
| go func(k []byte) { | ||
| defer wg.Done() | ||
| if _, err := kv.LeaseReadForKeyThrough(r.coordinator, leaseCtx, k); err != nil { | ||
| errCh <- errors.WithStack(err) | ||
| cancelOnce.Do(cancel) | ||
| } | ||
| }(key) | ||
| } | ||
| wg.Wait() | ||
| close(errCh) | ||
| for err := range errCh { | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (r *RedisServer) leaseQueuedCommandReadGroups(ctx context.Context, queue []redcon.Command) error { | ||
| if r.coordinator == nil { | ||
| return nil | ||
| } | ||
| return r.leaseRedisReadFenceGroups(ctx, r.queuedCommandReadFenceGroupKeys(queue)) | ||
| } | ||
|
|
||
| // MULTI/EXEC/DISCARD handling | ||
| func (r *RedisServer) multi(conn redcon.Conn, _ redcon.Command) { | ||
| state := getConnState(conn) | ||
|
|
@@ -72,13 +282,19 @@ func (r *RedisServer) exec(conn redcon.Conn, _ redcon.Command) { | |
| state.inTxn = false | ||
| state.queue = nil | ||
|
|
||
| route, err := r.transactionProxyRoute(queue) | ||
| if err != nil { | ||
| writeRedisError(conn, err) | ||
| return | ||
| } | ||
| if route.defaultLeader { | ||
| r.proxyTransactionToLeader(conn, queue) | ||
| return | ||
| } | ||
| if len(route.key) > 0 { | ||
| r.proxyTransactionToLeaderForKey(conn, route.key, queue) | ||
|
There was a problem hiding this comment.
When Useful? React with 👍 / 👎. |
||
| return | ||
| } | ||
|
|
||
| results, err := r.runTransaction(queue) | ||
| if err != nil { | ||
|
|
@@ -2378,7 +2594,12 @@ func (r *RedisServer) runTransactionDirect(queue []redcon.Command) ([]redisResul | |
| defer cancel() | ||
|
|
||
| var results []redisResult | ||
| fenceGroupKeys := r.queuedCommandReadFenceGroupKeys(queue) | ||
| err := r.retryRedisWrite(dispatchCtx, func() error { | ||
| if err := r.leaseRedisReadFenceGroups(dispatchCtx, fenceGroupKeys); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| startTS := r.txnStartTS() | ||
|
There was a problem hiding this comment.
When an EXEC spans at least two shard groups led by the same node, this fence-before-snapshot ordering can still produce stale OCC reads: after group A's lease completes, an A write can obtain T1 and remain unapplied while group B applies T2 > T1; AGENTS.md reference: AGENTS.md:L24-L25 Useful? React with 👍 / 👎. |
||
| readPin := r.pinReadTS(startTS) | ||
| defer readPin.Release() | ||
|
|
@@ -2593,6 +2814,10 @@ func (r *RedisServer) runTransactionWithDedup(queue []redcon.Command) ([]redisRe | |
| // from runTransactionWithDedup to keep that loop under the cyclop | ||
| // budget; the dedup rationale lives there. | ||
| func (r *RedisServer) firstExecAttempt(dispatchCtx context.Context, queue []redcon.Command) ([]redisResult, *reusableExecTxn, error) { | ||
| if err := r.leaseQueuedCommandReadGroups(dispatchCtx, queue); err != nil { | ||
| return nil, nil, err | ||
| } | ||
|
|
||
| startTS := r.txnStartTS() | ||
| readPin := r.pinReadTS(startTS) | ||
| defer readPin.Release() | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
When this new per-shard proxy path targets a node whose EXEC returns the newly introduced
errRedisExecRouteChangedAfterAmbiguousAttempt, go-redis reports the top-level non-array EXEC error as the pipeline error and copies it onto the queued command handles. Neither proxy error handler recognizes this error, soproxyTransactionToLeaderAddrwrites an EXEC result array containing per-command errors instead of preserving the top-level ambiguous-transaction failure; clients can therefore treat the transaction as having executed with ordinary command failures. Promote this EXEC-level error in the proxy handlers before writing the result array.Useful? React with 👍 / 👎.