diff --git a/adapter/redis_compat_helpers.go b/adapter/redis_compat_helpers.go index 016b423c6..96dee75ea 100644 --- a/adapter/redis_compat_helpers.go +++ b/adapter/redis_compat_helpers.go @@ -884,11 +884,11 @@ func (r *RedisServer) snapshotGetAt(key []byte, readTS uint64) ([]byte, error) { } func (r *RedisServer) doGetAt(key []byte, readTS uint64, verify bool) ([]byte, error) { - // Leadership is partitioned by the logical user key, so strip the internal - // prefix before asking the coordinator. + // Leadership is partitioned by the logical user key, but route through a + // Redis wrapper so list-internal-shaped user keys remain literal. routingKey := key if userKey := extractRedisInternalUserKey(key); userKey != nil { - routingKey = userKey + routingKey = redisUserRouteKey(userKey) } if r.coordinator.IsLeaderForKey(routingKey) { if verify { diff --git a/adapter/redis_compat_types.go b/adapter/redis_compat_types.go index c775cefd4..4d333a662 100644 --- a/adapter/redis_compat_types.go +++ b/adapter/redis_compat_types.go @@ -142,6 +142,7 @@ const ( redisHLLPrefix = "!redis|hll|" redisStreamPrefix = "!redis|stream|" redisTTLPrefix = "!redis|ttl|" + redisRoutePrefix = "!redis|route|" ) var redisInternalPrefixes = []string{ @@ -250,6 +251,12 @@ func redisTTLKey(userKey []byte) []byte { return append([]byte(redisTTLPrefix), userKey...) } +// redisUserRouteKey routes a literal Redis user key without recursively +// decoding bytes that happen to look like another internal storage key. +func redisUserRouteKey(userKey []byte) []byte { + return append([]byte(redisRoutePrefix), userKey...) +} + func redisExactSetStorageKey(kind string, userKey []byte) []byte { switch kind { case "set": diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index b90ed2fa0..0c6f998ec 100644 --- a/adapter/redis_delta_compactor.go +++ b/adapter/redis_delta_compactor.go @@ -214,7 +214,7 @@ func (c *DeltaCompactor) compactUrgentKey(ctx context.Context, req urgentCompact }() // Use per-key leadership so that in sharded deployments this node compacts // keys for the shards it leads, not just those of the default Raft group. - if !c.coord.IsLeaderForKey(req.userKey) { + if !c.coord.IsLeaderForKey(redisUserRouteKey(req.userKey)) { return } h := c.handlerByTypeName(req.typeName) @@ -574,7 +574,7 @@ func (c *DeltaCompactor) buildBatchElems(ctx context.Context, h collectionDeltaH // In sharded deployments IsLeaderForKey returns false for keys whose // shard this node does not lead. Skip those to avoid dispatching a // transaction that the responsible leader will reject. - if !c.coord.IsLeaderForKey(userKey) { + if !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { continue } elems, buildErr := h.buildElems(ctx, userKey, deltaKVs, readTS) diff --git a/adapter/redis_error_prefix_test.go b/adapter/redis_error_prefix_test.go index 242fad658..9ce29e0c7 100644 --- a/adapter/redis_error_prefix_test.go +++ b/adapter/redis_error_prefix_test.go @@ -124,7 +124,7 @@ func TestWriteRedisError(t *testing.T) { // instead of :prefix :leader. This covers proxyDBSize / proxyDel / // proxyFlushDatabase / proxyFlushLegacy in redis_proxy.go + // proxyKeys / proxyLRange / proxyRPush / proxyLPush / - // leaderClientForKey / resolveLeaderRedisAddr in redis.go — all + // leaderClientForRedisUserKey / resolveLeaderRedisAddr in redis.go — all // errors.Newf'd with the same "ERR leader redis address unknown // for %s" prefix. {"leader-address-unknown config-gap is already ERR-prefixed", @@ -178,6 +178,52 @@ func TestHandleProxyTxnError(t *testing.T) { } +func TestHandleProxyTxnTerminalExecError(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + err error + want string + }{ + { + name: "ambiguous route change typed", + err: errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt), + want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(), + }, + { + name: "ambiguous route change decoded RESP", + err: errors.New(errRedisExecRouteChangedAfterAmbiguousAttempt.Error()), + want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(), + }, + { + name: "split leader typed", + err: errors.WithStack(errRedisExecSplitShardLeaders), + want: errRedisExecSplitShardLeaders.Error(), + }, + { + name: "split leader decoded RESP", + err: errors.New(errRedisExecSplitShardLeaders.Error()), + want: errRedisExecSplitShardLeaders.Error(), + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + c := &captureConn{} + handled := handleProxyTxnError(c, tc.err) + if !handled { + t.Fatal("handleProxyTxnError returned false") + } + if c.lastErr != tc.want { + t.Fatalf("last error = %q", c.lastErr) + } + if c.wroteArray { + t.Fatalf("unexpected array reply %d", c.lastArray) + } + }) + } +} + func TestHandleProxyTxnHeavyCommandBusyError(t *testing.T) { t.Parallel() c := &captureConn{} @@ -222,6 +268,51 @@ func TestHandleProxyTxnCommandError(t *testing.T) { } }) + t.Run("ambiguous route change is promoted to top-level EXEC error", func(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + err error + want string + }{ + { + name: "ambiguous route change typed", + err: errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt), + want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(), + }, + { + name: "ambiguous route change decoded RESP", + err: errors.New(errRedisExecRouteChangedAfterAmbiguousAttempt.Error()), + want: errRedisExecRouteChangedAfterAmbiguousAttempt.Error(), + }, + { + name: "split leader typed", + err: errors.WithStack(errRedisExecSplitShardLeaders), + want: errRedisExecSplitShardLeaders.Error(), + }, + { + name: "split leader decoded RESP", + err: errors.New(errRedisExecSplitShardLeaders.Error()), + want: errRedisExecSplitShardLeaders.Error(), + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + cmd := redis.NewCmd(context.Background(), "SET", "k", "v") + cmd.SetErr(tc.err) + c := &captureConn{} + handled := handleProxyTxnCommandError(c, []*redis.Cmd{cmd}) + if !handled { + t.Fatal("handleProxyTxnCommandError returned false") + } + if c.lastErr != tc.want { + t.Fatalf("last error = %q", c.lastErr) + } + }) + } + }) + } func TestHandleProxyTxnCommandHeavyCommandBusyError(t *testing.T) { diff --git a/adapter/redis_lists.go b/adapter/redis_lists.go index 1f7a6fcf2..c4423b8ce 100644 --- a/adapter/redis_lists.go +++ b/adapter/redis_lists.go @@ -265,7 +265,7 @@ func (r *RedisServer) dispatchListPushReuse(ctx context.Context, key []byte, pen // (non-retryable errors escape to the client; pending is then // discarded with the goroutine, so the update is wasted and the // stale value would be misleading if some future caller reads it). - if isRetryableRedisTxnErr(dispErr) { + if isReusableRedisTxnErr(dispErr) { pending.commitTS = commitTS } return 0, false, errors.WithStack(dispErr) @@ -440,7 +440,7 @@ func (r *RedisServer) listPushCoreWithDedup(ctx context.Context, key []byte, val // retryRedisWrite's retry predicate; ambiguous errors that escape // to the client are a separate problem space (cross-request // idempotency cache) and out of scope for this design. - if isRetryableRedisTxnErr(dispErr) { + if isReusableRedisTxnErr(dispErr) { pending = &reusableListPush{ ops: ops, startTS: startTS, @@ -709,11 +709,40 @@ func (r *RedisServer) fetchListRange(ctx context.Context, key []byte, meta store } func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRaw []byte) ([]string, error) { - if !r.coordinator.IsLeaderForKey(key) { - return r.proxyLRange(key, startRaw, endRaw) + var out []string + err := r.retryRedisWrite(ctx, func() error { + routeVersion := r.redisReadFenceRouteVersion() + readTS, readPin, proxied, ok, err := r.fenceRangeListReadGroups(ctx, key, startRaw, endRaw) + if err != nil { + return err + } else if ok { + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } + out = proxied + return nil + } + defer readPin.Release() + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } + next, err := r.rangeListAt(ctx, key, startRaw, endRaw, readTS) + if err != nil { + return err + } + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } + out = next + return nil + }) + if err != nil { + return nil, err } + return out, nil +} - readTS := r.readTS() +func (r *RedisServer) rangeListAt(ctx context.Context, key []byte, startRaw, endRaw []byte, readTS uint64) ([]string, error) { typ, err := r.keyTypeAt(ctx, key, readTS) if err != nil { return nil, err @@ -725,14 +754,6 @@ func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRa return nil, wrongTypeError() } - // 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) - } - meta, exists, err := r.resolveListMeta(ctx, key, readTS) if err != nil { return nil, err @@ -749,12 +770,29 @@ func (r *RedisServer) rangeList(ctx context.Context, key []byte, startRaw, endRa return r.fetchListRange(ctx, key, meta, int64(s), int64(e), readTS) } +func (r *RedisServer) fenceRangeListReadGroups(ctx context.Context, key []byte, startRaw, endRaw []byte) (uint64, *kv.ActiveTimestampToken, []string, bool, error) { + groupKeys := r.redisReadFenceGroupKeys(r.redisTxnReadFenceKeysForRanges(key, redisListReadFenceRanges(key))) + proxyKey, ok, err := r.readFenceProxyKey(groupKeys) + if err != nil { + return 0, nil, nil, false, err + } + if ok { + proxied, err := r.proxyLRange(key, proxyKey, startRaw, endRaw) + return 0, nil, proxied, true, err + } + readTS, readPin, err := r.redisReadFencedTimestamp(ctx, groupKeys, r.readTS) + if err != nil { + return 0, nil, nil, false, err + } + return readTS, readPin, nil, false, nil +} + type listPushFunc func(ctx context.Context, key []byte, values [][]byte) (int64, error) type listProxyFunc func(key []byte, values [][]byte) (int64, error) func (r *RedisServer) listPushCmd(conn redcon.Conn, cmd redcon.Command, pushFn listPushFunc, proxyFn listProxyFunc) { key := cmd.Args[1] - if !r.coordinator.IsLeaderForKey(key) { + if !r.coordinator.IsLeaderForKey(redisUserRouteKey(key)) { length, err := proxyFn(key, cmd.Args[2:]) if err != nil { writeRedisError(conn, err) diff --git a/adapter/redis_proxy.go b/adapter/redis_proxy.go index 338a3b5c2..7b7c158cb 100644 --- a/adapter/redis_proxy.go +++ b/adapter/redis_proxy.go @@ -33,7 +33,7 @@ func (r *RedisServer) proxyDel(keys [][]byte) (int64, error) { // Group keys by leader Redis address. byAddr := make(map[string][]string) for _, k := range keys { - leader := r.coordinator.RaftLeaderForKey(k) + leader := r.coordinator.RaftLeaderForKey(redisUserRouteKey(k)) if leader == "" { return 0, ErrLeaderNotFound } diff --git a/adapter/redis_proxy_leader.go b/adapter/redis_proxy_leader.go index b7fa7498f..ce6703fc9 100644 --- a/adapter/redis_proxy_leader.go +++ b/adapter/redis_proxy_leader.go @@ -50,6 +50,18 @@ func (r *RedisServer) proxyTransactionToLeader(conn redcon.Conn, queue []redcon. if !ok { return } + r.proxyTransactionToLeaderAddr(conn, queue, leaderAddr) +} + +func (r *RedisServer) proxyTransactionToLeaderForKey(conn redcon.Conn, routingKey []byte, queue []redcon.Command) { + leaderAddr, ok := r.resolveLeaderRedisAddrForKey(conn, routingKey) + if !ok { + return + } + r.proxyTransactionToLeaderAddr(conn, queue, leaderAddr) +} + +func (r *RedisServer) proxyTransactionToLeaderAddr(conn redcon.Conn, queue []redcon.Command, leaderAddr string) { cli := r.getOrCreateLeaderClient(leaderAddr) ctx, cancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout) @@ -81,6 +93,20 @@ func (r *RedisServer) resolveLeaderRedisAddr(conn redcon.Conn) (string, bool) { return leaderAddr, true } +func (r *RedisServer) resolveLeaderRedisAddrForKey(conn redcon.Conn, key []byte) (string, bool) { + leader := r.coordinator.RaftLeaderForKey(key) + if leader == "" { + writeRedisError(conn, ErrLeaderNotFound) + return "", false + } + leaderAddr, ok := r.leaderRedis[leader] + if !ok || leaderAddr == "" { + conn.WriteError(fmt.Sprintf("ERR leader redis address unknown for raft address %s", leader)) + return "", false + } + return leaderAddr, true +} + // execTxPipeline sends queue as a single TxPipelined batch and returns the // per-command result handles together with any pipeline-level error. func (r *RedisServer) execTxPipeline(ctx context.Context, cli *redis.Client, queue []redcon.Command) ([]*redis.Cmd, error) { @@ -115,13 +141,7 @@ func handleProxyTxnError(conn redcon.Conn, err error) bool { conn.WriteError(errRedisHeavyCommandPoolFull.Error()) return true } - var netErr net.Error - if isTransientLeaderRedisError(err) || - errors.Is(err, context.DeadlineExceeded) || - errors.Is(err, context.Canceled) || - errors.Is(err, io.EOF) || - errors.Is(err, io.ErrUnexpectedEOF) || - errors.As(err, &netErr) { + if isTerminalProxyTxnError(err) { writeRedisError(conn, err) return true } @@ -143,6 +163,10 @@ func handleProxyTxnCommandError(conn redcon.Conn, cmds []*redis.Cmd) bool { conn.WriteError(errRedisHeavyCommandPoolFull.Error()) return true } + if isRedisExecTerminalProxyError(err) { + writeRedisError(conn, err) + return true + } if isTransientLeaderRedisError(err) { writeRedisError(conn, err) return true @@ -151,6 +175,27 @@ func handleProxyTxnCommandError(conn redcon.Conn, cmds []*redis.Cmd) bool { return false } +func isTerminalProxyTxnError(err error) bool { + var netErr net.Error + return isRedisExecTerminalProxyError(err) || + isTransientLeaderRedisError(err) || + errors.Is(err, context.DeadlineExceeded) || + errors.Is(err, context.Canceled) || + errors.Is(err, io.EOF) || + errors.Is(err, io.ErrUnexpectedEOF) || + errors.As(err, &netErr) +} + +func isRedisExecTerminalProxyError(err error) bool { + return isExactRedisExecTerminalProxyError(err, errRedisExecRouteChangedAfterAmbiguousAttempt) || + isExactRedisExecTerminalProxyError(err, errRedisExecSplitShardLeaders) +} + +func isExactRedisExecTerminalProxyError(err error, sentinel error) bool { + return err != nil && + (errors.Is(err, sentinel) || err.Error() == sentinel.Error()) +} + // writeProxyCmdsResult writes an EXEC-style array reply for the given pipeline // command handles. For any other non-nil per-command errors, each cmd carries // its own result, which is the correct Redis EXEC semantics. @@ -161,8 +206,8 @@ func writeProxyCmdsResult(conn redcon.Conn, cmds []*redis.Cmd) { } } -func (r *RedisServer) proxyLRange(key []byte, startRaw, endRaw []byte) ([]string, error) { - leader := r.coordinator.RaftLeaderForKey(key) +func (r *RedisServer) proxyLRange(key, routingKey []byte, startRaw, endRaw []byte) ([]string, error) { + leader := r.coordinator.RaftLeaderForKey(routingKey) if leader == "" { return nil, ErrLeaderNotFound } @@ -190,7 +235,7 @@ func (r *RedisServer) proxyLRange(key []byte, startRaw, endRaw []byte) ([]string } func (r *RedisServer) proxyRPush(key []byte, values [][]byte) (int64, error) { - leader := r.coordinator.RaftLeaderForKey(key) + leader := r.coordinator.RaftLeaderForKey(redisUserRouteKey(key)) if leader == "" { return 0, ErrLeaderNotFound } @@ -214,7 +259,7 @@ func (r *RedisServer) proxyRPush(key []byte, values [][]byte) (int64, error) { } func (r *RedisServer) proxyLPush(key []byte, values [][]byte) (int64, error) { - leader := r.coordinator.RaftLeaderForKey(key) + leader := r.coordinator.RaftLeaderForKey(redisUserRouteKey(key)) if leader == "" { return 0, ErrLeaderNotFound } @@ -319,10 +364,10 @@ func (r *RedisServer) leaderProxyPoolSizes() (normal, blocking int) { return normal, blocking } -// leaderClientForKey returns a cached go-redis client connected to the leader -// for the given key. -func (r *RedisServer) leaderClientForKey(key []byte) (*redis.Client, error) { - leader := r.coordinator.RaftLeaderForKey(key) +// leaderClientForRedisUserKey returns a cached go-redis client connected to +// the leader for the literal Redis user key. +func (r *RedisServer) leaderClientForRedisUserKey(key []byte) (*redis.Client, error) { + leader := r.coordinator.RaftLeaderForKey(redisUserRouteKey(key)) if leader == "" { return nil, ErrLeaderNotFound } @@ -337,10 +382,11 @@ func (r *RedisServer) leaderClientForKey(key []byte) (*redis.Client, error) { // response to conn. Returns true if the command was proxied (caller should // return immediately), false if this node is the leader. func (r *RedisServer) proxyToLeader(conn redcon.Conn, cmd redcon.Command, key []byte) bool { - if r.coordinator.IsLeaderForKey(key) { + routeKey := redisUserRouteKey(key) + if r.coordinator.IsLeaderForKey(routeKey) { return false } - cli, err := r.leaderClientForKey(key) + cli, err := r.leaderClientForRedisUserKey(key) if err != nil { writeRedisError(conn, err) return true @@ -358,10 +404,11 @@ func (r *RedisServer) proxyToLeader(conn redcon.Conn, cmd redcon.Command, key [] } func (r *RedisServer) proxyBlockingToLeader(conn redcon.Conn, cmd redcon.Command, key []byte) bool { - if r.coordinator.IsLeaderForKey(key) { + routeKey := redisUserRouteKey(key) + if r.coordinator.IsLeaderForKey(routeKey) { return false } - leader := r.coordinator.RaftLeaderForKey(key) + leader := r.coordinator.RaftLeaderForKey(routeKey) if leader == "" { writeRedisError(conn, ErrLeaderNotFound) return true diff --git a/adapter/redis_retry.go b/adapter/redis_retry.go index 77233cbbb..e6a6eafda 100644 --- a/adapter/redis_retry.go +++ b/adapter/redis_retry.go @@ -44,11 +44,25 @@ var ( ) func isRetryableRedisTxnErr(err error) bool { + return isReusableRedisTxnErr(err) || isRedisComposedRouteErr(err) +} + +func isReusableRedisTxnErr(err error) bool { return errors.Is(err, store.ErrWriteConflict) || errors.Is(err, kv.ErrTxnLocked) || wireRedisTxnErrKind(err) == redisTxnWireErrLocked } +func isRedisComposedRouteErr(err error) bool { + if errors.Is(err, kv.ErrComposed1Violation) || + errors.Is(err, kv.ErrComposed1VersionGCd) { + return true + } + parsed := parseWireRedisTxnErr(err) + return errors.Is(parsed, kv.ErrComposed1Violation) || + errors.Is(parsed, kv.ErrComposed1VersionGCd) +} + func retryPolicyForRedisTxnErr(err error) redisTxnRetryPolicy { if errors.Is(err, kv.ErrTxnLocked) || wireRedisTxnErrKind(err) == redisTxnWireErrLocked { return redisTxnLockedRetryPolicy @@ -66,13 +80,13 @@ const ( // parseWireRedisTxnErr restores transaction error typing after an internal // leader redirect crosses gRPC. Forward currently returns transaction failures -// as a status, which strips the typed ErrWriteConflict / ErrTxnLocked chain. -// Match only the exact server-generated key error envelope and normalize the -// storage key before rebuilding the typed error. Wire write conflicts are not -// generally retryable: a lost forwarding response can turn an already-applied -// write into a later self-conflict. Reuse-aware callers explicitly normalize -// them before retrying; all other callers return the normalized error without -// replaying the operation or exposing the internal key layout. +// as a status, which strips the typed ErrWriteConflict / ErrTxnLocked / +// Composed-1 sentinel chain. Match only known server-generated envelopes. +// Wire write conflicts are not generally retryable: a lost forwarding response +// can turn an already-applied write into a later self-conflict. Reuse-aware +// callers explicitly normalize them before retrying; all other callers return +// the normalized error without replaying the operation or exposing the internal +// key layout. func wireRedisTxnStatus(err error) (*status.Status, bool) { type grpcStatusCarrier interface { GRPCStatus() *status.Status @@ -94,6 +108,31 @@ func parseWireRedisTxnErr(err error) error { return nil } msg := st.Message() + if parsed := parseWireRedisTxnKeyErr(msg); parsed != nil { + return parsed + } + if isWireComposedRouteMessage(msg, kv.ErrComposed1Violation) { + return errors.WithStack(kv.ErrComposed1Violation) + } + if isWireComposedRouteMessage(msg, kv.ErrComposed1VersionGCd) { + return errors.WithStack(kv.ErrComposed1VersionGCd) + } + return nil +} + +func isWireComposedRouteMessage(msg string, sentinel error) bool { + sentinelMsg := sentinel.Error() + if msg == sentinelMsg { + return true + } + if !strings.HasSuffix(msg, ": "+sentinelMsg) { + return false + } + return strings.HasPrefix(msg, "observed-version v=") || + strings.HasPrefix(msg, "current-version v=") +} + +func parseWireRedisTxnKeyErr(msg string) error { if !strings.HasPrefix(msg, "key: ") { return nil } @@ -274,15 +313,15 @@ func normalizeRetryableRedisTxnKey(key []byte) []byte { if userKey := redisTxnWideFenceUserKey(key); userKey != nil { return userKey } - if store.IsListMetaKey(key) || store.IsListItemKey(key) { - return store.ExtractListUserKey(key) - } if store.IsListMetaDeltaKey(key) { return store.ExtractListUserKeyFromDelta(key) } if store.IsListClaimKey(key) { return store.ExtractListUserKeyFromClaim(key) } + if store.IsListMetaKey(key) || store.IsListItemKey(key) { + return store.ExtractListUserKey(key) + } if wideKey, ok := normalizeWideColumnKey(key); ok { return wideKey } diff --git a/adapter/redis_retry_test.go b/adapter/redis_retry_test.go index 7dff88b7e..48b54e155 100644 --- a/adapter/redis_retry_test.go +++ b/adapter/redis_retry_test.go @@ -399,6 +399,65 @@ func TestRetryRedisWriteRetriesWireTxnLocked(t *testing.T) { require.Equal(t, redisTxnLockedRetryPolicy, retryPolicyForRedisTxnErr(wireErr)) } +func TestRetryRedisWriteRetriesWireComposedRouteErrors(t *testing.T) { + t.Parallel() + + wireErr := errors.WithStack(status.Error(codes.Aborted, + "current-version v=2: key \"k\" owned by group 2: "+kv.ErrComposed1Violation.Error())) + attempts := 0 + srv := &RedisServer{} + err := srv.retryRedisWrite(context.Background(), func() error { + attempts++ + if attempts == 1 { + return wireErr + } + return nil + }) + + require.NoError(t, err) + require.Equal(t, 2, attempts) +} + +func TestRetryRedisWriteDoesNotTreatComposedSentinelInWireConflictKeyAsRouteError(t *testing.T) { + t.Parallel() + + for _, sentinel := range []error{kv.ErrComposed1Violation, kv.ErrComposed1VersionGCd} { + t.Run(sentinel.Error(), func(t *testing.T) { + t.Parallel() + wireErr := errors.WithStack(status.Error(codes.Unknown, + store.NewWriteConflictError([]byte("retry:"+sentinel.Error())).Error())) + attempts := 0 + srv := &RedisServer{} + + err := srv.retryRedisWrite(context.Background(), func() error { + attempts++ + return wireErr + }) + + require.ErrorIs(t, err, store.ErrWriteConflict) + require.NotErrorIs(t, err, sentinel) + require.Equal(t, 1, attempts) + }) + } +} + +func TestNormalizeRetryableRedisTxnErrPrefersWireTxnKeyEnvelope(t *testing.T) { + t.Parallel() + + for _, sentinel := range []error{kv.ErrComposed1Violation, kv.ErrComposed1VersionGCd} { + t.Run(sentinel.Error(), func(t *testing.T) { + t.Parallel() + wireErr := errors.WithStack(status.Error(codes.Aborted, + kv.NewTxnLockedError([]byte("locked:"+sentinel.Error())).Error())) + + err := normalizeRetryableRedisTxnErr(wireErr) + + require.ErrorIs(t, err, kv.ErrTxnLocked) + require.NotErrorIs(t, err, sentinel) + }) + } +} + func TestRetryRedisWriteDoesNotRetryUnclassifiedWireErrors(t *testing.T) { t.Parallel() diff --git a/adapter/redis_route_test.go b/adapter/redis_route_test.go new file mode 100644 index 000000000..37f518f49 --- /dev/null +++ b/adapter/redis_route_test.go @@ -0,0 +1,82 @@ +package adapter + +import ( + "bytes" + "context" + "testing" + + "github.com/bootjp/elastickv/kv" + "github.com/bootjp/elastickv/store" + "github.com/stretchr/testify/require" + "github.com/tidwall/redcon" +) + +type recordingRedisRouteCoordinator struct { + stubAdapterCoordinator + isLeaderKeys [][]byte + verifyKeys [][]byte +} + +func (c *recordingRedisRouteCoordinator) IsLeaderForKey(key []byte) bool { + c.isLeaderKeys = append(c.isLeaderKeys, bytes.Clone(key)) + return true +} + +func (c *recordingRedisRouteCoordinator) VerifyLeaderForKey(_ context.Context, key []byte) error { + c.verifyKeys = append(c.verifyKeys, bytes.Clone(key)) + return nil +} + +func TestRedisUserRouteKeyPreservesListStorageShapedUserKey(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + userKey []byte + }{ + {"empty", []byte{}}, + {"delta", store.ListMetaDeltaKey([]byte("other"), 1, 1)}, + {"claim", store.ListClaimKey([]byte("other"), 3)}, + {"meta", store.ListMetaKey([]byte("other"))}, + {"item", store.ListItemKey([]byte("other"), 5)}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + require.Equal(t, tc.userKey, kv.RouteKey(redisUserRouteKey(tc.userKey))) + }) + } +} + +func TestProxyToLeaderRoutesRawRedisKeyThroughLiteralWrapper(t *testing.T) { + t.Parallel() + + userKey := store.ListMetaDeltaKey([]byte("other"), 1, 1) + coord := &recordingRedisRouteCoordinator{} + server := &RedisServer{coordinator: coord} + + proxied := server.proxyToLeader(&recordingConn{}, redcon.Command{ + Args: [][]byte{[]byte("GET"), userKey}, + }, userKey) + + require.False(t, proxied) + require.Equal(t, [][]byte{redisUserRouteKey(userKey)}, coord.isLeaderKeys) +} + +func TestLeaderAwareGetAtRoutesRedisInternalKeyThroughLiteralWrapper(t *testing.T) { + t.Parallel() + + ctx := context.Background() + userKey := store.ListMetaDeltaKey([]byte("other"), 1, 1) + storageKey := redisStrKey(userKey) + st := store.NewMVCCStore() + require.NoError(t, st.PutAt(ctx, storageKey, []byte("value"), 1, 0)) + coord := &recordingRedisRouteCoordinator{} + server := &RedisServer{store: st, coordinator: coord} + + got, err := server.leaderAwareGetAt(storageKey, 1) + + require.NoError(t, err) + require.Equal(t, []byte("value"), got) + require.Equal(t, [][]byte{redisUserRouteKey(userKey)}, coord.isLeaderKeys) + require.Equal(t, [][]byte{redisUserRouteKey(userKey)}, coord.verifyKeys) +} diff --git a/adapter/redis_stream_cmds.go b/adapter/redis_stream_cmds.go index 52efed53d..1e1c252fb 100644 --- a/adapter/redis_stream_cmds.go +++ b/adapter/redis_stream_cmds.go @@ -386,7 +386,7 @@ func (r *RedisServer) firstXAddAttempt( // This path owns the exact ID and write set, so it can safely restore a // forwarded conflict to its typed form before entering retryRedisWrite. dispErr = normalizeRetryableRedisTxnErr(dispErr) - if !isRetryableRedisTxnErr(dispErr) { + if !isReusableRedisTxnErr(dispErr) { return "", nil, cockerrors.WithStack(dispErr) } return "", &reusableXAdd{ @@ -441,7 +441,7 @@ func (r *RedisServer) dispatchXAddReuse( } return "", true, cockerrors.WithStack(dispErr) } - if isRetryableRedisTxnErr(dispErr) { + if isReusableRedisTxnErr(dispErr) { // A lock response did not apply, but carrying the fresh commitTS keeps // the retry correct if a future retryable transport shape is added. pending.commitTS = commitTS diff --git a/adapter/redis_strings.go b/adapter/redis_strings.go index a01fa41a1..ba09cd995 100644 --- a/adapter/redis_strings.go +++ b/adapter/redis_strings.go @@ -186,7 +186,7 @@ func (r *RedisServer) trySetFastPath(conn redcon.Conn, ctx context.Context, key, // Only use the fast path when we are the leader for this key so the local // type check is authoritative. On followers, stale MVCC state could miss a // non-string type, leaving orphaned internal keys after overwrite. - if !r.coordinator.IsLeaderForKey(key) { + if !r.coordinator.IsLeaderForKey(redisUserRouteKey(key)) { return false } readTS := r.readTS() @@ -330,7 +330,7 @@ func (r *RedisServer) get(conn redcon.Conn, cmd redcon.Command) { // that can actually block on quorum / I/O. ctx, cancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout) defer cancel() - if _, err := kv.LeaseReadForKeyThrough(r.coordinator, ctx, key); err != nil { + if _, err := kv.LeaseReadForKeyThrough(r.coordinator, ctx, redisUserRouteKey(key)); err != nil { writeRedisError(conn, err) return } @@ -454,7 +454,7 @@ func (r *RedisServer) tryLeaderLogicalExists(key []byte) bool { // existence with ttlAt() semantics (including the in-memory TTL buffer). // If this path is unavailable we fall back to raw-KV probing, which is // best-effort and may lag unflushed buffer-only TTL updates. - if cli, err := r.leaderClientForKey(key); err == nil { + if cli, err := r.leaderClientForRedisUserKey(key); err == nil { ctx, cancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout) defer cancel() if count, existsErr := cli.Exists(ctx, string(key)).Result(); existsErr == nil { @@ -480,7 +480,7 @@ func (r *RedisServer) del(conn redcon.Conn, cmd redcon.Command) { localKeys := make([][]byte, 0, len(cmd.Args)-1) proxyKeys := make([][]byte, 0) for _, key := range cmd.Args[1:] { - if r.coordinator.IsLeaderForKey(key) { + if r.coordinator.IsLeaderForKey(redisUserRouteKey(key)) { localKeys = append(localKeys, key) } else { proxyKeys = append(proxyKeys, key) @@ -559,7 +559,7 @@ func (r *RedisServer) exists(conn redcon.Conn, cmd redcon.Command) { } if ok { count++ - } else if !r.coordinator.IsLeaderForKey(key) { + } else if !r.coordinator.IsLeaderForKey(redisUserRouteKey(key)) { // Local MVCC may be stale on a follower; proxy to the leader. if r.tryLeaderLogicalExists(key) { count++ diff --git a/adapter/redis_ttl_inline_migrator.go b/adapter/redis_ttl_inline_migrator.go index 4274f0e62..b9b2bbec0 100644 --- a/adapter/redis_ttl_inline_migrator.go +++ b/adapter/redis_ttl_inline_migrator.go @@ -218,7 +218,7 @@ func (c *DeltaCompactor) migrateTTLInlineScanPlan( ) for _, pair := range kvs { userKey := h.extractUserKey(pair.Key) - if userKey == nil || !c.coord.IsLeaderForKey(userKey) { + if userKey == nil || !c.coord.IsLeaderForKey(redisUserRouteKey(userKey)) { continue } built, buildErr := h.buildElems(ctx, pair, readTS) diff --git a/adapter/redis_txn.go b/adapter/redis_txn.go index a38001517..02de2c8a4 100644 --- a/adapter/redis_txn.go +++ b/adapter/redis_txn.go @@ -7,6 +7,7 @@ import ( "sort" "strconv" "strings" + "sync" "time" "github.com/bootjp/elastickv/kv" @@ -20,9 +21,40 @@ var redisTxnWideHashFencePrefix = []byte("!redis|txn-wide-hash|") var redisTxnWideSetFencePrefix = []byte("!redis|txn-wide-set|") var redisTxnWideListFencePrefix = []byte("!redis|txn-wide-list|") var redisTxnWideZSetFencePrefix = []byte("!redis|txn-wide-zset|") +var redisReadFenceRouteChangedKey = []byte("!redis|read-fence-route-changed") + +const redisReadFenceLocalLeaderTarget = "\x00redis-read-fence-local-leader" + +var errRedisExecSplitShardLeaders = errors.New("ERR EXEC read fence spans multiple shard leaders") +var errRedisExecRouteChangedAfterAmbiguousAttempt = errors.New("ERR EXEC read fence route changed after ambiguous dispatch") type txnCommandHandler func(*txnContext, redcon.Command) (redisResult, error) +const ( + redisCommandKeyArgCount = 2 + redisHashFirstFieldArg = 2 + redisZIncrByArgCount = 4 + redisZIncrByMemberArg = 3 +) + +type redisReadFenceRange struct { + start []byte + end []byte +} + +type redisReadFenceRangeGroupKeyProvider interface { + ReadFenceGroupKeysForRange(start []byte, end []byte) [][]byte +} + +type redisReadFenceRouteVersionProvider interface { + ReadFenceRouteVersion() uint64 +} + +type redisReadFenceRouteVersion struct { + tracked bool + version uint64 +} + var txnApplyHandlers = map[string]txnCommandHandler{ cmdSet: (*txnContext).applySet, cmdDel: (*txnContext).applyDel, @@ -38,6 +70,447 @@ var txnApplyHandlers = map[string]txnCommandHandler{ cmdPExpire: (*txnContext).applyExpireMilliseconds, } +func redisTxnReadFenceKeys(userKey []byte) [][]byte { + keys := append([][]byte{}, redisTxnReadFencePointKeys(userKey)...) + keys = append(keys, + store.ListMetaDeltaScanPrefix(userKey), + store.ListClaimScanPrefix(userKey), + store.HashFieldScanPrefix(userKey), + store.HashMetaDeltaScanPrefix(userKey), + store.SetMemberScanPrefix(userKey), + store.SetMetaDeltaScanPrefix(userKey), + store.ZSetMemberScanPrefix(userKey), + store.ZSetScoreScanPrefix(userKey), + store.ZSetMetaDeltaScanPrefix(userKey), + store.StreamEntryScanPrefix(userKey), + ) + return keys +} + +func redisTxnReadFencePointKeys(userKey []byte) [][]byte { + keys := [][]byte{ + redisStrKey(userKey), + redisHLLKey(userKey), + redisTTLKey(userKey), + listMetaKey(userKey), + listItemKey(userKey, 0), + redisTxnWideListFenceKey(userKey), + redisHashKey(userKey), + store.HashMetaKey(userKey), + redisTxnWideHashFenceKey(userKey), + redisSetKey(userKey), + store.SetMetaKey(userKey), + redisTxnWideSetFenceKey(userKey), + redisZSetKey(userKey), + store.ZSetMetaKey(userKey), + redisTxnWideZSetFenceKey(userKey), + redisStreamKey(userKey), + store.StreamMetaKey(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 (r *RedisServer) redisTxnReadFenceKeysForRanges(userKey []byte, ranges []redisReadFenceRange) [][]byte { + keys := append([][]byte{}, redisTxnReadFencePointKeys(userKey)...) + for _, readRange := range ranges { + rangeGroupKeys := r.redisReadFenceRangeGroupKeys(readRange.start, readRange.end) + if len(rangeGroupKeys) == 0 { + rangeGroupKeys = [][]byte{readRange.start} + } + keys = append(keys, rangeGroupKeys...) + } + return keys +} + +func redisTxnReadFenceRanges(userKey []byte) []redisReadFenceRange { + ranges := redisListReadFenceRanges(userKey) + ranges = append(ranges, redisHashReadFenceRanges(userKey)...) + ranges = append(ranges, redisSetReadFenceRanges(userKey)...) + ranges = append(ranges, redisZSetReadFenceRanges(userKey)...) + ranges = append(ranges, redisStreamReadFenceRanges(userKey)...) + return ranges +} + +func redisListReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.ListMetaDeltaScanPrefix(userKey), + store.ListClaimScanPrefix(userKey), + }) +} + +func redisTypeDetectionReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.ListMetaDeltaScanPrefix(userKey), + store.HashFieldScanPrefix(userKey), + store.HashMetaDeltaScanPrefix(userKey), + store.SetMemberScanPrefix(userKey), + store.SetMetaDeltaScanPrefix(userKey), + store.ZSetMemberScanPrefix(userKey), + store.ZSetMetaDeltaScanPrefix(userKey), + }) +} + +func redisHashReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.HashFieldScanPrefix(userKey), + store.HashMetaDeltaScanPrefix(userKey), + }) +} + +func redisSetReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.SetMemberScanPrefix(userKey), + store.SetMetaDeltaScanPrefix(userKey), + }) +} + +func redisZSetReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.ZSetMemberScanPrefix(userKey), + store.ZSetScoreScanPrefix(userKey), + store.ZSetMetaDeltaScanPrefix(userKey), + }) +} + +func redisStreamReadFenceRanges(userKey []byte) []redisReadFenceRange { + return redisReadFenceRangesForPrefixes([][]byte{ + store.StreamEntryScanPrefix(userKey), + }) +} + +func redisReadFenceRangesForPrefixes(prefixes [][]byte) []redisReadFenceRange { + ranges := make([]redisReadFenceRange, 0, len(prefixes)) + for _, prefix := range prefixes { + ranges = append(ranges, redisReadFenceRange{ + start: prefix, + end: store.PrefixScanEnd(prefix), + }) + } + return ranges +} + +func (r *RedisServer) redisReadFenceRangeGroupKeys(start []byte, end []byte) [][]byte { + if r == nil || r.store == nil { + return nil + } + provider, ok := r.store.(redisReadFenceRangeGroupKeyProvider) + if !ok { + return nil + } + return provider.ReadFenceGroupKeysForRange(start, end) +} + +func (r *RedisServer) redisReadFenceRouteVersion() redisReadFenceRouteVersion { + if r == nil || r.store == nil { + return redisReadFenceRouteVersion{} + } + provider, ok := r.store.(redisReadFenceRouteVersionProvider) + if !ok { + return redisReadFenceRouteVersion{} + } + return redisReadFenceRouteVersion{ + tracked: true, + version: provider.ReadFenceRouteVersion(), + } +} + +func (r *RedisServer) ensureRedisReadFenceRouteStable(observed redisReadFenceRouteVersion) error { + if !observed.tracked || r == nil || r.store == nil { + return nil + } + provider, ok := r.store.(redisReadFenceRouteVersionProvider) + if !ok || provider.ReadFenceRouteVersion() == observed.version { + return nil + } + return errors.WithStack(store.NewWriteConflictError(redisReadFenceRouteChangedKey)) +} + +func (v redisReadFenceRouteVersion) observedRouteVersion() uint64 { + if !v.tracked { + return 0 + } + return kv.EncodeObservedRouteVersion(v.version) +} + +func redisQueuedCommandReadFenceKeys(queue []redcon.Command) [][]byte { + return redisQueuedCommandReadFenceKeysForServer(nil, queue) +} + +func (r *RedisServer) redisQueuedCommandReadFenceKeys(queue []redcon.Command) [][]byte { + return redisQueuedCommandReadFenceKeysForServer(r, queue) +} + +func redisQueuedCommandReadFenceKeysForServer(r *RedisServer, 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 { + for _, fenceKey := range redisCommandReadFenceKeysForServer(r, cmd) { + appendKey(fenceKey) + } + } + return keys +} + +func redisCommandReadFenceKeysForServer(r *RedisServer, cmd redcon.Command) [][]byte { + if len(cmd.Args) == 0 { + return nil + } + cmdName := strings.ToUpper(string(cmd.Args[0])) + meta, ok := redisCommandTable[cmdName] + if !ok { + return nil + } + keys := make([][]byte, 0) + for _, userKey := range redisCommandGetKeys(meta, cmd.Args) { + if r == nil { + keys = append(keys, redisTxnReadFenceKeys(userKey)...) + continue + } + keys = append(keys, r.redisTxnReadFenceKeysForRanges(userKey, redisCommandReadFenceRanges(cmdName, userKey))...) + } + keys = append(keys, redisCommandExactReadFenceKeys(cmd)...) + return keys +} + +func redisCommandReadFenceRanges(cmdName string, userKey []byte) []redisReadFenceRange { + switch cmdName { + case cmdGet, cmdExists: + return redisTypeDetectionReadFenceRanges(userKey) + case cmdHSet, cmdHMSet: + return redisTypeCheckedReadFenceRanges(userKey, redisHashReadFenceRanges(userKey)) + case cmdRPush, cmdLRange: + return redisTypeCheckedReadFenceRanges(userKey, redisListReadFenceRanges(userKey)) + case cmdZIncrBy: + return redisTypeCheckedReadFenceRanges(userKey, redisZSetReadFenceRanges(userKey)) + case cmdSet, cmdDel, cmdIncr, cmdExpire, cmdPExpire: + return redisTxnReadFenceRanges(userKey) + default: + return nil + } +} + +func redisTypeCheckedReadFenceRanges(userKey []byte, ranges []redisReadFenceRange) []redisReadFenceRange { + typeRanges := redisTypeDetectionReadFenceRanges(userKey) + return append(typeRanges, ranges...) +} + +func redisCommandExactReadFenceKeys(cmd redcon.Command) [][]byte { + if len(cmd.Args) < redisCommandKeyArgCount { + return nil + } + key := cmd.Args[1] + switch strings.ToUpper(string(cmd.Args[0])) { + case cmdHSet, cmdHMSet: + fieldArgs := cmd.Args[redisHashFirstFieldArg:] + if len(fieldArgs) == 0 || len(fieldArgs)%redisPairWidth != 0 { + return nil + } + keys := make([][]byte, 0, len(fieldArgs)/redisPairWidth) + for i := redisHashFirstFieldArg; i < len(cmd.Args); i += redisPairWidth { + keys = append(keys, store.HashFieldKey(key, cmd.Args[i])) + } + return keys + case cmdZIncrBy: + if len(cmd.Args) < redisZIncrByArgCount { + return nil + } + return [][]byte{store.ZSetMemberKey(key, cmd.Args[redisZIncrByMemberArg])} + default: + return nil + } +} + +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(r.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 + } + if r.leaderRedis != nil { + leaderAddr, ok := r.leaderRedis[leader] + if !ok || leaderAddr == "" { + return "", false, errors.WithStack(errors.Newf("ERR leader redis address unknown for raft address %s", leader)) + } + return leaderAddr, false, nil + } + return leader, false, nil +} + +func recordReadFenceTargetLeader(target *string, leader string) error { + if leader == "" { + return nil + } + if *target == "" { + *target = leader + return nil + } + if *target != leader { + return errRedisExecSplitShardLeaders + } + return nil +} + +func (r *RedisServer) transactionProxyRoute(queue []redcon.Command) (redisTxnProxyRoute, error) { + if r == nil || r.coordinator == nil { + return redisTxnProxyRoute{}, nil + } + + routeVersion := r.redisReadFenceRouteVersion() + groupKeys := r.queuedCommandReadFenceGroupKeys(queue) + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return redisTxnProxyRoute{}, err + } + 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 err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return redisTxnProxyRoute{}, err + } + if ok { + return redisTxnProxyRoute{key: proxyKey}, nil + } + return redisTxnProxyRoute{}, nil +} + +func (r *RedisServer) retryTransactionProxyRoute(ctx context.Context, queue []redcon.Command) (redisTxnProxyRoute, error) { + var route redisTxnProxyRoute + err := r.retryRedisWrite(ctx, func() error { + next, err := r.transactionProxyRoute(queue) + if err != nil { + return err + } + route = next + return nil + }) + return route, err +} + +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) redisReadFencedTimestamp( + ctx context.Context, + groupKeys [][]byte, + selectTS func() uint64, +) (uint64, *kv.ActiveTimestampToken, error) { + if err := r.leaseRedisReadFenceGroups(ctx, groupKeys); err != nil { + return 0, nil, err + } + readTS := selectTS() + readPin := r.pinReadTS(readTS) + if err := r.leaseRedisReadFenceGroups(ctx, groupKeys); err != nil { + readPin.Release() + return 0, nil, err + } + return readTS, readPin, nil +} + // MULTI/EXEC/DISCARD handling func (r *RedisServer) multi(conn redcon.Conn, _ redcon.Command) { state := getConnState(conn) @@ -72,13 +545,21 @@ func (r *RedisServer) exec(conn redcon.Conn, _ redcon.Command) { state.inTxn = false state.queue = nil - // Always execute MULTI/EXEC on the leader so that reads and writes within - // the transaction see consistent, up-to-date data. Serving transactions - // on followers risks reading stale MVCC state and producing write cycles. - if !r.coordinator.IsLeader() { + routeCtx, routeCancel := context.WithTimeout(r.handlerContext(), redisDispatchTimeout) + route, err := r.retryTransactionProxyRoute(routeCtx, queue) + routeCancel() + 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) + return + } results, err := r.runTransaction(queue) if err != nil { @@ -1561,21 +2042,25 @@ func (t *txnContext) dispatchContext() (context.Context, context.CancelFunc) { return context.WithTimeout(parentCtx, redisDispatchTimeout) } -func (t *txnContext) commit() error { +func (t *txnContext) commit(routeVersion redisReadFenceRouteVersion) error { prepared, err := t.prepareDispatch() if err != nil { return err } defer prepared.cancel() + if err := t.server.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } if len(prepared.elems) == 0 { return nil } group := &kv.OperationGroup[kv.OP]{ - IsTxn: true, - Elems: prepared.elems, - StartTS: t.startTS, - CommitTS: prepared.commitTS, - ReadKeys: prepared.readKeys, + IsTxn: true, + Elems: prepared.elems, + StartTS: t.startTS, + CommitTS: prepared.commitTS, + ReadKeys: prepared.readKeys, + ObservedRouteVersion: routeVersion.observedRouteVersion(), } if _, err := t.server.coordinator.Dispatch(prepared.ctx, group); err != nil { return errors.WithStack(err) @@ -2369,6 +2854,42 @@ func (r *RedisServer) runTransaction(queue []redcon.Command) ([]redisResult, err return r.runTransactionDirect(queue) } +func (r *RedisServer) applyExecQueueAtSnapshot( + dispatchCtx context.Context, + queue []redcon.Command, + startTS uint64, +) (*txnContext, []redisResult, error) { + txn := &txnContext{ + server: r, + ctx: dispatchCtx, + working: map[string]*txnValue{}, + replacers: map[string]*stringReplacement{}, + listStates: map[string]*listTxnState{}, + hashStates: map[string]*hashTxnState{}, + zsetStates: map[string]*zsetTxnState{}, + ttlStates: map[string]*ttlTxnState{}, + readKeys: map[string][]byte{}, + deletedKeys: map[string]struct{}{}, + logicalDeletes: map[string][]byte{}, + hashDeletes: map[string][]byte{}, + setDeletes: map[string][]byte{}, + hashCreates: map[string]struct{}{}, + collectionExpireTypes: map[string]redisValueType{}, + streamDeletions: map[string][]byte{}, + startTS: startTS, + } + + nextResults := make([]redisResult, 0, len(queue)) + for _, cmd := range queue { + res, err := txn.apply(cmd) + if err != nil { + return nil, nil, err + } + nextResults = append(nextResults, res) + } + return txn, nextResults, nil +} + func (r *RedisServer) runTransactionDirect(queue []redcon.Command) ([]redisResult, error) { if r.onePhaseTxnDedup { return r.runTransactionWithDedup(queue) @@ -2379,43 +2900,29 @@ func (r *RedisServer) runTransactionDirect(queue []redcon.Command) ([]redisResul var results []redisResult err := r.retryRedisWrite(dispatchCtx, func() error { - startTS := r.txnStartTS() - readPin := r.pinReadTS(startTS) + routeVersion := r.redisReadFenceRouteVersion() + fenceGroupKeys := r.queuedCommandReadFenceGroupKeys(queue) + startTS, readPin, err := r.redisReadFencedTimestamp(dispatchCtx, fenceGroupKeys, r.txnStartTS) + if err != nil { + return err + } defer readPin.Release() + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } - txn := &txnContext{ - server: r, - ctx: dispatchCtx, - working: map[string]*txnValue{}, - replacers: map[string]*stringReplacement{}, - listStates: map[string]*listTxnState{}, - hashStates: map[string]*hashTxnState{}, - zsetStates: map[string]*zsetTxnState{}, - ttlStates: map[string]*ttlTxnState{}, - readKeys: map[string][]byte{}, - deletedKeys: map[string]struct{}{}, - logicalDeletes: map[string][]byte{}, - hashDeletes: map[string][]byte{}, - setDeletes: map[string][]byte{}, - hashCreates: map[string]struct{}{}, - collectionExpireTypes: map[string]redisValueType{}, - streamDeletions: map[string][]byte{}, - startTS: startTS, - } - - nextResults := make([]redisResult, 0, len(queue)) - for _, cmd := range queue { - res, err := txn.apply(cmd) - if err != nil { - return err - } - nextResults = append(nextResults, res) + txn, nextResults, err := r.applyExecQueueAtSnapshot(dispatchCtx, queue, startTS) + if err != nil { + return err } if err := txn.validateReadSet(dispatchCtx); err != nil { return err } - if err := txn.commit(); err != nil { + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return err + } + if err := txn.commit(routeVersion); err != nil { return err } results = nextResults @@ -2446,11 +2953,12 @@ func (r *RedisServer) runTransactionDirect(queue []redcon.Command) ([]redisResul // results are only returned when reuse actually represents the // outcome of attempt 1's intent. type reusableExecTxn struct { - elems []*kv.Elem[kv.OP] - startTS uint64 - commitTS uint64 - readKeys [][]byte - results []redisResult + elems []*kv.Elem[kv.OP] + startTS uint64 + commitTS uint64 + observedRouteVersion uint64 + readKeys [][]byte + results []redisResult } // dispatchExecReuse runs one iteration of the option-2 reuse path for @@ -2468,6 +2976,9 @@ type reusableExecTxn struct { // is the current length" question; the client-visible result IS the // cached results array. func (r *RedisServer) dispatchExecReuse(ctx context.Context, pending *reusableExecTxn) (results []redisResult, drop bool, err error) { + if err := r.ensureReusableExecRouteStable(pending); err != nil { + return nil, false, errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt) + } // gemini PR-A HIGH: persistence-grade commit_ts allocation must honor the // HLC-4 physical-ceiling fence (see kv/hlc.go NextFenced + the TLA proof // at tla/hlc/MCHLC_gap.cfg). Clock().Next() bypasses the ceiling and @@ -2479,12 +2990,13 @@ func (r *RedisServer) dispatchExecReuse(ctx context.Context, pending *reusableEx return nil, false, errors.WithStack(allocErr) } _, dispErr := r.coordinator.Dispatch(ctx, &kv.OperationGroup[kv.OP]{ - IsTxn: true, - StartTS: pending.startTS, - CommitTS: commitTS, - PrevCommitTS: pending.commitTS, - ReadKeys: pending.readKeys, - Elems: pending.elems, + IsTxn: true, + StartTS: pending.startTS, + CommitTS: commitTS, + PrevCommitTS: pending.commitTS, + ReadKeys: pending.readKeys, + Elems: pending.elems, + ObservedRouteVersion: pending.observedRouteVersion, }) if dispErr == nil { return pending.results, false, nil @@ -2494,6 +3006,9 @@ func (r *RedisServer) dispatchExecReuse(ctx context.Context, pending *reusableEx // Normalize before the typed conflict branch; the generic retry loop keeps // raw wire write conflicts fail-closed for callers without this protection. dispErr = normalizeRetryableRedisTxnErr(dispErr) + if isRedisComposedRouteErr(dispErr) { + return nil, false, errors.WithStack(errRedisExecRouteChangedAfterAmbiguousAttempt) + } if errors.Is(dispErr, store.ErrWriteConflict) { // Self-inflicted-conflict guard (mirrors dispatchListPushReuse): // the apply might have landed at this fresh commitTS but bubbled @@ -2516,17 +3031,27 @@ func (r *RedisServer) dispatchExecReuse(ctx context.Context, pending *reusableEx // iteration rebuilds from a fresh snapshot. return nil, true, errors.WithStack(dispErr) } - // Still ambiguous (lock / other retryable): the reuse may itself - // have landed, so the next retry must probe THIS commit_ts. Only - // advance pending.commitTS if retryRedisWrite will actually loop - // (non-retryable errors escape to the client; pending is then - // discarded with the goroutine). - if isRetryableRedisTxnErr(dispErr) { - pending.commitTS = commitTS - } + // TxnLocked did not apply the reuse attempt, and non-retryable errors will + // escape the retry loop. In both cases, keep pending.commitTS pointing at + // the last ambiguous dispatch so a later retry probes the only commit_ts + // that might already have landed. return nil, false, errors.WithStack(dispErr) } +func (r *RedisServer) ensureReusableExecRouteStable(pending *reusableExecTxn) error { + if pending == nil { + return nil + } + version, tracked := kv.DecodeObservedRouteVersion(pending.observedRouteVersion) + if !tracked { + return nil + } + return r.ensureRedisReadFenceRouteStable(redisReadFenceRouteVersion{ + tracked: true, + version: version, + }) +} + // runTransactionWithDedup is the option-2 retry loop for MULTI/EXEC. // The first attempt builds the txn write set + cached results from // the user's startTS snapshot; any retryable failure makes the next @@ -2583,6 +3108,21 @@ func (r *RedisServer) runTransactionWithDedup(queue []redcon.Command) ([]redisRe return results, nil } +func (r *RedisServer) prepareExecDispatchWithStableRoute( + txn *txnContext, + routeVersion redisReadFenceRouteVersion, +) (preparedTxnDispatch, error) { + prepared, err := txn.prepareDispatch() + if err != nil { + return preparedTxnDispatch{cancel: func() {}}, err + } + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + prepared.cancel() + return preparedTxnDispatch{cancel: func() {}}, err + } + return prepared, nil +} + // firstExecAttempt runs the initial (no-reuse) EXEC attempt: builds the // txn snapshot, applies each command to capture the client-visible // results, validates the read set, and dispatches. On success returns @@ -2593,44 +3133,30 @@ 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) { - startTS := r.txnStartTS() - readPin := r.pinReadTS(startTS) + routeVersion := r.redisReadFenceRouteVersion() + fenceGroupKeys := r.queuedCommandReadFenceGroupKeys(queue) + startTS, readPin, err := r.redisReadFencedTimestamp(dispatchCtx, fenceGroupKeys, r.txnStartTS) + if err != nil { + return nil, nil, err + } defer readPin.Release() - - txn := &txnContext{ - server: r, - ctx: dispatchCtx, - working: map[string]*txnValue{}, - replacers: map[string]*stringReplacement{}, - listStates: map[string]*listTxnState{}, - hashStates: map[string]*hashTxnState{}, - zsetStates: map[string]*zsetTxnState{}, - ttlStates: map[string]*ttlTxnState{}, - readKeys: map[string][]byte{}, - deletedKeys: map[string]struct{}{}, - logicalDeletes: map[string][]byte{}, - hashDeletes: map[string][]byte{}, - setDeletes: map[string][]byte{}, - hashCreates: map[string]struct{}{}, - collectionExpireTypes: map[string]redisValueType{}, - streamDeletions: map[string][]byte{}, - startTS: startTS, + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return nil, nil, err } - nextResults := make([]redisResult, 0, len(queue)) - for _, cmd := range queue { - res, err := txn.apply(cmd) - if err != nil { - return nil, nil, err - } - nextResults = append(nextResults, res) + txn, nextResults, err := r.applyExecQueueAtSnapshot(dispatchCtx, queue, startTS) + if err != nil { + return nil, nil, err } if err := txn.validateReadSet(dispatchCtx); err != nil { return nil, nil, err } + if err := r.ensureRedisReadFenceRouteStable(routeVersion); err != nil { + return nil, nil, err + } - prepared, err := txn.prepareDispatch() + prepared, err := r.prepareExecDispatchWithStableRoute(txn, routeVersion) if err != nil { return nil, nil, err } @@ -2641,30 +3167,29 @@ func (r *RedisServer) firstExecAttempt(dispatchCtx context.Context, queue []redc } group := &kv.OperationGroup[kv.OP]{ - IsTxn: true, - Elems: prepared.elems, - StartTS: txn.startTS, - CommitTS: prepared.commitTS, - ReadKeys: prepared.readKeys, + IsTxn: true, + Elems: prepared.elems, + StartTS: txn.startTS, + CommitTS: prepared.commitTS, + ReadKeys: prepared.readKeys, + ObservedRouteVersion: routeVersion.observedRouteVersion(), } if _, dispErr := r.coordinator.Dispatch(prepared.ctx, group); dispErr != nil { // Preserve the exact attempt for a forwarded conflict only after // restoring its typed form. runTransactionWithDedup can then reuse this // write set instead of replaying the EXEC body from a new snapshot. dispErr = normalizeRetryableRedisTxnErr(dispErr) - // Only remember the attempt for reuse if retryRedisWrite will - // actually loop. Mirrors listPushCoreWithDedup's gating - // rationale — errors that escape the loop (transient-leader, - // context deadline, FSM apply error) leave pending pointing at - // state wasted with the goroutine; ambiguous errors that - // escape to the client are out of scope for this loop. - if isRetryableRedisTxnErr(dispErr) { + // Only remember the attempt when the dispatch outcome is ambiguous. + // TxnLocked is retryable, but it did not apply; keeping no pending lets + // the next iteration rebuild against the current read-fence route. + if isAmbiguousRedisExecDispatchErr(dispErr) { return nil, &reusableExecTxn{ - elems: prepared.elems, - startTS: txn.startTS, - commitTS: prepared.commitTS, - readKeys: prepared.readKeys, - results: nextResults, + elems: prepared.elems, + startTS: txn.startTS, + commitTS: prepared.commitTS, + observedRouteVersion: routeVersion.observedRouteVersion(), + readKeys: prepared.readKeys, + results: nextResults, }, errors.WithStack(dispErr) } return nil, nil, errors.WithStack(dispErr) @@ -2672,6 +3197,10 @@ func (r *RedisServer) firstExecAttempt(dispatchCtx context.Context, queue []redc return nextResults, nil, nil } +func isAmbiguousRedisExecDispatchErr(err error) bool { + return errors.Is(err, store.ErrWriteConflict) +} + func (r *RedisServer) txnStartTS() uint64 { // store.LastCommitTS() is the authoritative safe-snapshot watermark: it is // updated atomically only AFTER the corresponding Pebble batch commit, so diff --git a/adapter/redis_txn_test.go b/adapter/redis_txn_test.go index 6eb0088c8..ae3cdbc33 100644 --- a/adapter/redis_txn_test.go +++ b/adapter/redis_txn_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "fmt" + "sync" "testing" "time" @@ -42,6 +43,130 @@ func newRedisTxnTestContext(server *RedisServer) *txnContext { } } +type verifyHookCoordinator struct { + *localAdapterCoordinator + leaseForKey func(context.Context, []byte) (uint64, error) + groupForKey func([]byte) uint64 + mu sync.Mutex + leaseCalls int + leaseKeys [][]byte +} + +func newVerifyHookCoordinator(st store.MVCCStore) *verifyHookCoordinator { + return &verifyHookCoordinator{localAdapterCoordinator: newLocalAdapterCoordinator(st)} +} + +func (c *verifyHookCoordinator) LeaseReadForKey(ctx context.Context, key []byte) (uint64, error) { + c.mu.Lock() + c.leaseCalls++ + c.leaseKeys = append(c.leaseKeys, bytes.Clone(key)) + c.mu.Unlock() + if c.leaseForKey != nil { + return c.leaseForKey(ctx, key) + } + return c.localAdapterCoordinator.LeaseReadForKey(ctx, key) +} + +func (c *verifyHookCoordinator) EngineGroupIDForKey(key []byte) uint64 { + if c.groupForKey != nil { + return c.groupForKey(key) + } + return 1 +} + +type observingLastCommitStore struct { + store.MVCCStore + onLastCommitTS func() +} + +func (s *observingLastCommitStore) LastCommitTS() uint64 { + if s.onLastCommitTS != nil { + s.onLastCommitTS() + } + return s.MVCCStore.LastCommitTS() +} + +type redisReadFenceRangeStore struct { + store.MVCCStore + rangeKeysByStart map[string][][]byte +} + +func (s *redisReadFenceRangeStore) ReadFenceGroupKeysForRange(start []byte, _ []byte) [][]byte { + if s == nil { + return nil + } + return cloneReadKeys(s.rangeKeysByStart[string(start)]) +} + +type redisReadFenceRouteVersionStore struct { + store.MVCCStore + mu sync.Mutex + routeVersion uint64 + rangeKeysByVersion map[uint64][][]byte + lastReadFenceGroupKeys [][]byte + readFenceGroupKeyCallCount int + onLastCommitTS func() + onScanAt func() + onReadFenceGroupKeys func() +} + +func (s *redisReadFenceRouteVersionStore) ReadFenceRouteVersion() uint64 { + s.mu.Lock() + defer s.mu.Unlock() + return s.routeVersion +} + +func (s *redisReadFenceRouteVersionStore) advanceRouteVersionForTest() { + s.mu.Lock() + defer s.mu.Unlock() + s.routeVersion = 2 +} + +func (s *redisReadFenceRouteVersionStore) ReadFenceGroupKeysForRange(_ []byte, _ []byte) [][]byte { + if s == nil { + return nil + } + if s.onReadFenceGroupKeys != nil { + s.onReadFenceGroupKeys() + } + s.mu.Lock() + defer s.mu.Unlock() + s.readFenceGroupKeyCallCount++ + keys := cloneReadKeys(s.rangeKeysByVersion[s.routeVersion]) + s.lastReadFenceGroupKeys = cloneReadKeys(keys) + return keys +} + +func (s *redisReadFenceRouteVersionStore) readFenceGroupKeyCalls() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.readFenceGroupKeyCallCount +} + +func (s *redisReadFenceRouteVersionStore) LastCommitTS() uint64 { + if s.onLastCommitTS != nil { + s.onLastCommitTS() + } + return s.MVCCStore.LastCommitTS() +} + +func (s *redisReadFenceRouteVersionStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) { + if s.onScanAt != nil { + s.onScanAt() + } + return s.MVCCStore.ScanAt(ctx, start, end, limit, ts) +} + +func seedRedisListAt(t *testing.T, st store.MVCCStore, key []byte, values ...string) { + t.Helper() + metaBytes, err := store.MarshalListMeta(store.ListMeta{Len: int64(len(values))}) + require.NoError(t, err) + require.NoError(t, st.PutAt(context.Background(), store.ListMetaKey(key), metaBytes, redisTxnTestStartTS, 0)) + for i, value := range values { + require.NoError(t, st.PutAt(context.Background(), listItemKey(key, int64(i)), []byte(value), redisTxnTestStartTS, 0)) + } +} + func elemKeysContain(elems []*kv.Elem[kv.OP], want []byte) bool { for _, elem := range elems { if elem != nil && string(elem.Key) == string(want) { @@ -84,7 +209,10 @@ func requireTTLNear(t *testing.T, raw []byte, want time.Time) { type readKeyRecordingCoordinator struct { *localAdapterCoordinator - lastReadKeys [][]byte + lastReadKeys [][]byte + lastObservedRouteVersion uint64 + dispatches int + prevCommitTS []uint64 } func newReadKeyRecordingCoordinator(st store.MVCCStore) *readKeyRecordingCoordinator { @@ -92,7 +220,50 @@ func newReadKeyRecordingCoordinator(st store.MVCCStore) *readKeyRecordingCoordin } func (c *readKeyRecordingCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + c.dispatches++ c.lastReadKeys = cloneReadKeys(req.ReadKeys) + c.lastObservedRouteVersion = req.ObservedRouteVersion + c.prevCommitTS = append(c.prevCommitTS, req.PrevCommitTS) + return c.localAdapterCoordinator.Dispatch(ctx, req) +} + +type composedRetryCoordinator struct { + *localAdapterCoordinator + dispatches int + prevCommitTS []uint64 + firstErr error +} + +func (c *composedRetryCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + c.dispatches++ + c.prevCommitTS = append(c.prevCommitTS, req.PrevCommitTS) + if c.dispatches == 1 { + if c.firstErr != nil { + return nil, c.firstErr + } + return nil, kv.ErrComposed1Violation + } + return c.localAdapterCoordinator.Dispatch(ctx, req) +} + +type routeChangingDispatchCoordinator struct { + *verifyHookCoordinator + dispatches int + beforeDispatch func(int) + firstErr error +} + +func (c *routeChangingDispatchCoordinator) Dispatch(ctx context.Context, req *kv.OperationGroup[kv.OP]) (*kv.CoordinateResponse, error) { + c.dispatches++ + if c.beforeDispatch != nil { + c.beforeDispatch(c.dispatches) + } + if c.dispatches == 1 { + if c.firstErr != nil { + return nil, c.firstErr + } + return nil, store.NewWriteConflictError([]byte("redis-route-version-retry")) + } return c.localAdapterCoordinator.Dispatch(ctx, req) } @@ -104,6 +275,29 @@ func cloneReadKeys(in [][]byte) [][]byte { return out } +func countReadKey(in [][]byte, want []byte) int { + count := 0 + for _, key := range in { + if bytes.Equal(key, want) { + count++ + } + } + return count +} + +func readFenceRouteVersionGroupForKey(routeA, routeB []byte) func([]byte) uint64 { + return func(got []byte) uint64 { + switch { + case bytes.Equal(got, routeA): + return 101 + case bytes.Equal(got, routeB): + return 102 + default: + return 1 + } + } +} + func requireReadKeysMatch(t *testing.T, got [][]byte, want [][]byte) { t.Helper() gotSet := make(map[string]struct{}, len(got)) @@ -117,6 +311,905 @@ func requireReadKeysMatch(t *testing.T, got [][]byte, want [][]byte) { require.Equal(t, wantSet, gotSet) } +type redisTxnFenceRoutingCoordinator struct { + stubAdapterCoordinator + defaultLeader bool + localLeader func([]byte) bool + raftLeader func([]byte) string + groupID func([]byte) uint64 +} + +func (c *redisTxnFenceRoutingCoordinator) IsLeader() bool { + return c.defaultLeader +} + +func (c *redisTxnFenceRoutingCoordinator) IsLeaderForKey(key []byte) bool { + if c.localLeader != nil { + return c.localLeader(key) + } + return true +} + +func (c *redisTxnFenceRoutingCoordinator) RaftLeaderForKey(key []byte) string { + if c.raftLeader != nil { + return c.raftLeader(key) + } + return "" +} + +func (c *redisTxnFenceRoutingCoordinator) EngineGroupIDForKey(key []byte) uint64 { + if c.groupID != nil { + return c.groupID(key) + } + return 1 +} + +func TestRedisRangeListVerifiesLeaderBeforeSnapshot(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + key := []byte("list:leader-fence-lrange") + coord := newVerifyHookCoordinator(st) + server := NewRedisServer(nil, "", st, coord, nil, nil) + + seeded := false + coord.leaseForKey = func(_ context.Context, got []byte) (uint64, error) { + if !seeded { + seedRedisListAt(t, st, key, "v1") + seeded = true + } + return 0, nil + } + + got, err := server.rangeList(context.Background(), key, []byte("0"), []byte("-1")) + require.NoError(t, err) + require.Equal(t, []string{"v1"}, got) + require.Equal(t, 2, coord.leaseCalls) +} + +func TestRedisRangeListFencesEveryStorageReadGroup(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + key := []byte("list:multi-fence-lrange") + coord := newVerifyHookCoordinator(st) + coord.groupForKey = func(got []byte) uint64 { + switch string(got) { + case string(redisStrKey(key)): + return 1 + case string(listMetaKey(key)): + return 2 + case string(store.ListMetaDeltaScanPrefix(key)): + return 3 + default: + return 1 + } + } + server := NewRedisServer(nil, "", st, coord, nil, nil) + + var seedOnce sync.Once + coord.leaseForKey = func(_ context.Context, _ []byte) (uint64, error) { + seedOnce.Do(func() { + seedRedisListAt(t, st, key, "v1") + }) + return 0, nil + } + + got, err := server.rangeList(context.Background(), key, []byte("0"), []byte("-1")) + require.NoError(t, err) + require.Equal(t, []string{"v1"}, got) + require.Equal(t, 6, coord.leaseCalls) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listMetaKey(key), + store.ListMetaDeltaScanPrefix(key), + redisStrKey(key), + listMetaKey(key), + store.ListMetaDeltaScanPrefix(key), + }, coord.leaseKeys) +} + +func TestRedisRangeListRetriesWhenReadFenceRouteVersionChangesDuringScan(t *testing.T) { + t.Parallel() + + ctx := context.Background() + key := []byte("list:route-version-retry") + routeA := []byte("list-route-a") + routeB := []byte("list-route-b") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + rangeKeysByVersion: map[uint64][][]byte{ + 1: {routeA}, + 2: {routeB}, + }, + } + seedRedisListAt(t, st, key, "v1") + + var scanOnce sync.Once + st.onScanAt = func() { + scanOnce.Do(func() { + require.NoError(t, st.PutAt(ctx, listItemKey(key, 0), []byte("v2"), 20, 0)) + st.advanceRouteVersionForTest() + }) + } + + coord := newVerifyHookCoordinator(st) + coord.groupForKey = readFenceRouteVersionGroupForKey(routeA, routeB) + server := NewRedisServer(nil, "", st, coord, nil, nil) + + got, err := server.rangeList(ctx, key, []byte("0"), []byte("-1")) + require.NoError(t, err) + require.Equal(t, []string{"v2"}, got) + require.GreaterOrEqual(t, countReadKey(coord.leaseKeys, routeA), 2) + require.GreaterOrEqual(t, countReadKey(coord.leaseKeys, routeB), 2) +} + +func TestRedisExecProxyRouteUsesShardLeaderInsteadOfDefaultLeader(t *testing.T) { + t.Parallel() + + key := []byte("txn:shard-local") + coord := &redisTxnFenceRoutingCoordinator{ + defaultLeader: false, + localLeader: func([]byte) bool { + return true + }, + raftLeader: func([]byte) string { + return "raft-local" + }, + } + server := &RedisServer{coordinator: coord, leaderRedis: map[string]string{"raft-remote": "redis-remote"}} + + route, err := server.transactionProxyRoute([]redcon.Command{{ + Args: [][]byte{[]byte(cmdGet), key}, + }}) + require.NoError(t, err) + require.False(t, route.defaultLeader) + require.Empty(t, route.key) +} + +func TestRedisExecProxyRouteTargetsRemoteShardLeader(t *testing.T) { + t.Parallel() + + key := []byte("txn:shard-remote") + coord := &redisTxnFenceRoutingCoordinator{ + defaultLeader: true, + localLeader: func([]byte) bool { + return false + }, + raftLeader: func([]byte) string { + return "raft-remote" + }, + } + server := &RedisServer{coordinator: coord, leaderRedis: map[string]string{"raft-remote": "redis-remote"}} + + route, err := server.transactionProxyRoute([]redcon.Command{{ + Args: [][]byte{[]byte(cmdGet), key}, + }}) + require.NoError(t, err) + require.False(t, route.defaultLeader) + require.Equal(t, redisStrKey(key), route.key) +} + +func TestRedisExecProxyRouteFailsClosedOnSplitShardLeaders(t *testing.T) { + t.Parallel() + + keyA := []byte("txn:split-a") + keyB := []byte("txn:split-b") + coord := &redisTxnFenceRoutingCoordinator{ + defaultLeader: true, + localLeader: func([]byte) bool { + return false + }, + raftLeader: func(key []byte) string { + if bytes.Contains(key, keyA) { + return "raft-a" + } + return "raft-b" + }, + groupID: func(key []byte) uint64 { + if bytes.Contains(key, keyA) { + return 1 + } + return 2 + }, + } + server := &RedisServer{coordinator: coord, leaderRedis: map[string]string{ + "raft-a": "redis-a", + "raft-b": "redis-b", + }} + + _, err := server.transactionProxyRoute([]redcon.Command{ + {Args: [][]byte{[]byte(cmdGet), keyA}}, + {Args: [][]byte{[]byte(cmdGet), keyB}}, + }) + require.ErrorIs(t, err, errRedisExecSplitShardLeaders) +} + +func TestRedisExecProxyRouteFailsClosedOnMixedLocalAndRemoteShardLeaders(t *testing.T) { + t.Parallel() + + localKey := []byte("txn:split-local") + remoteKey := []byte("txn:split-remote") + coord := &redisTxnFenceRoutingCoordinator{ + defaultLeader: true, + localLeader: func(key []byte) bool { + return bytes.Contains(key, localKey) + }, + raftLeader: func(key []byte) string { + if bytes.Contains(key, remoteKey) { + return "raft-remote" + } + return "" + }, + groupID: func(key []byte) uint64 { + if bytes.Contains(key, localKey) { + return 1 + } + return 2 + }, + } + server := &RedisServer{coordinator: coord} + + _, err := server.transactionProxyRoute([]redcon.Command{ + {Args: [][]byte{[]byte(cmdGet), localKey}}, + {Args: [][]byte{[]byte(cmdGet), remoteKey}}, + }) + require.ErrorIs(t, err, errRedisExecSplitShardLeaders) +} + +func TestRedisExecProxyRouteAllowsDistinctRaftEndpointsWithSameRedisTarget(t *testing.T) { + t.Parallel() + + keyA := []byte("txn:same-redis-a") + keyB := []byte("txn:same-redis-b") + coord := &redisTxnFenceRoutingCoordinator{ + defaultLeader: true, + localLeader: func([]byte) bool { + return false + }, + raftLeader: func(key []byte) string { + if bytes.Contains(key, keyA) { + return "raft-a" + } + return "raft-b" + }, + groupID: func(key []byte) uint64 { + if bytes.Contains(key, keyA) { + return 1 + } + return 2 + }, + } + server := &RedisServer{ + coordinator: coord, + leaderRedis: map[string]string{ + "raft-a": "redis-remote", + "raft-b": "redis-remote", + }, + } + + route, err := server.transactionProxyRoute([]redcon.Command{ + {Args: [][]byte{[]byte(cmdGet), keyA}}, + {Args: [][]byte{[]byte(cmdGet), keyB}}, + }) + require.NoError(t, err) + require.False(t, route.defaultLeader) + require.Equal(t, redisStrKey(keyA), route.key) +} + +func TestRedisExecReadFenceUsesRangeRoutesAndExactHashFields(t *testing.T) { + t.Parallel() + + key := []byte("txn:hash-fence-ranges") + field := []byte("field-b") + hashPrefix := store.HashFieldScanPrefix(key) + rangeKeyA := []byte("hash-route-a") + rangeKeyB := []byte("hash-route-b") + exactFieldKey := store.HashFieldKey(key, field) + st := &redisReadFenceRangeStore{ + MVCCStore: store.NewMVCCStore(), + rangeKeysByStart: map[string][][]byte{ + string(hashPrefix): {rangeKeyA, rangeKeyB}, + }, + } + coord := newVerifyHookCoordinator(st) + coord.groupForKey = func(got []byte) uint64 { + switch string(got) { + case string(rangeKeyA): + return 101 + case string(rangeKeyB): + return 102 + case string(exactFieldKey): + return 103 + default: + return 1 + } + } + server := &RedisServer{store: st, coordinator: coord} + + got := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdHSet), key, field, []byte("value")}, + }}) + + require.Contains(t, got, rangeKeyA) + require.Contains(t, got, rangeKeyB) + require.Contains(t, got, exactFieldKey) +} + +func TestRedisExecReadFenceUsesOnlyCommandRelevantRanges(t *testing.T) { + t.Parallel() + + key := []byte("txn:command-specific-fence-ranges") + otherKey := []byte("txn:command-specific-fence-ranges-other") + field := []byte("field-a") + member := []byte("member-a") + listTypeRoute := []byte("list-type-route") + listClaimRoute := []byte("list-claim-route") + hashFieldRoute := []byte("hash-field-route") + hashDeltaRoute := []byte("hash-delta-route") + setMemberRoute := []byte("set-member-route") + setDeltaRoute := []byte("set-delta-route") + zsetMemberRoute := []byte("zset-member-route") + zsetScoreRoute := []byte("zset-score-route") + zsetDeltaRoute := []byte("zset-delta-route") + streamEntryRoute := []byte("stream-entry-route") + otherListTypeRoute := []byte("other-list-type-route") + exactFieldKey := store.HashFieldKey(key, field) + exactZSetMemberKey := store.ZSetMemberKey(key, member) + st := &redisReadFenceRangeStore{ + MVCCStore: store.NewMVCCStore(), + rangeKeysByStart: map[string][][]byte{ + string(store.ListMetaDeltaScanPrefix(key)): {listTypeRoute}, + string(store.ListClaimScanPrefix(key)): {listClaimRoute}, + string(store.HashFieldScanPrefix(key)): {hashFieldRoute}, + string(store.HashMetaDeltaScanPrefix(key)): {hashDeltaRoute}, + string(store.SetMemberScanPrefix(key)): {setMemberRoute}, + string(store.SetMetaDeltaScanPrefix(key)): {setDeltaRoute}, + string(store.ZSetMemberScanPrefix(key)): {zsetMemberRoute}, + string(store.ZSetScoreScanPrefix(key)): {zsetScoreRoute}, + string(store.ZSetMetaDeltaScanPrefix(key)): {zsetDeltaRoute}, + string(store.StreamEntryScanPrefix(key)): {streamEntryRoute}, + string(store.ListMetaDeltaScanPrefix(otherKey)): {otherListTypeRoute}, + }, + } + coord := newVerifyHookCoordinator(st) + groupIDsByKey := map[string]uint64{ + string(listTypeRoute): 101, + string(listClaimRoute): 102, + string(hashFieldRoute): 103, + string(hashDeltaRoute): 104, + string(setMemberRoute): 105, + string(setDeltaRoute): 106, + string(zsetMemberRoute): 107, + string(zsetScoreRoute): 108, + string(zsetDeltaRoute): 109, + string(streamEntryRoute): 110, + string(otherListTypeRoute): 111, + string(exactFieldKey): 112, + string(exactZSetMemberKey): 113, + string(store.ListMetaDeltaScanPrefix(key)): 201, + string(store.ListClaimScanPrefix(key)): 202, + string(store.HashFieldScanPrefix(key)): 203, + string(store.HashMetaDeltaScanPrefix(key)): 204, + string(store.SetMemberScanPrefix(key)): 205, + string(store.SetMetaDeltaScanPrefix(key)): 206, + string(store.ZSetMemberScanPrefix(key)): 207, + string(store.ZSetScoreScanPrefix(key)): 208, + string(store.ZSetMetaDeltaScanPrefix(key)): 209, + string(store.StreamEntryScanPrefix(key)): 210, + } + coord.groupForKey = func(got []byte) uint64 { + if gid, ok := groupIDsByKey[string(got)]; ok { + return gid + } + return 1 + } + server := &RedisServer{store: st, coordinator: coord} + + getKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdGet), key}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetDeltaRoute, + }, getKeys) + + existsKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdExists), key, otherKey}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetDeltaRoute, + otherListTypeRoute, + }, existsKeys) + + lrangeKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdLRange), key, []byte("0"), []byte("-1")}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + listClaimRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetDeltaRoute, + }, lrangeKeys) + + rpushKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdRPush), key, []byte("value")}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + listClaimRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetDeltaRoute, + }, rpushKeys) + + zincrbyKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdZIncrBy), key, []byte("1"), member}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetScoreRoute, + zsetDeltaRoute, + exactZSetMemberKey, + }, zincrbyKeys) + + hsetKeys := server.queuedCommandReadFenceGroupKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdHSet), key, field, []byte("value")}, + }}) + require.ElementsMatch(t, [][]byte{ + redisStrKey(key), + listTypeRoute, + hashFieldRoute, + hashDeltaRoute, + setMemberRoute, + setDeltaRoute, + zsetMemberRoute, + zsetDeltaRoute, + exactFieldKey, + }, hsetKeys) +} + +func TestRedisReadFencedTimestampLeasesBeforeAndAfterSelectingTimestamp(t *testing.T) { + t.Parallel() + + base := store.NewMVCCStore() + coord := newVerifyHookCoordinator(base) + var leaseCallsAtLastCommitTS int + st := &observingLastCommitStore{ + MVCCStore: base, + onLastCommitTS: func() { + coord.mu.Lock() + defer coord.mu.Unlock() + leaseCallsAtLastCommitTS = coord.leaseCalls + }, + } + server := &RedisServer{store: st, coordinator: coord} + + startTS, readPin, err := server.redisReadFencedTimestamp( + context.Background(), + [][]byte{redisStrKey([]byte("txn:fence-order"))}, + server.txnStartTS, + ) + defer readPin.Release() + + require.NoError(t, err) + require.Equal(t, uint64(1), startTS) + coord.mu.Lock() + leaseCalls := coord.leaseCalls + coord.mu.Unlock() + require.Equal(t, 1, leaseCallsAtLastCommitTS) + require.Equal(t, 2, leaseCalls) +} + +func TestRedisExecRetriesWhenReadFenceRouteVersionChanges(t *testing.T) { + t.Parallel() + + key := []byte("txn:route-version-retry") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + } + + var bumpOnce sync.Once + lastCommitCalls := 0 + st.onLastCommitTS = func() { + lastCommitCalls++ + bumpOnce.Do(func() { + st.advanceRouteVersionForTest() + }) + } + + coord := newVerifyHookCoordinator(st) + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + } + + results, err := server.runTransactionDirect([]redcon.Command{{ + Args: [][]byte{[]byte(cmdGet), key}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, resultNil, results[0].typ) + require.Equal(t, 2, lastCommitCalls) +} + +func TestRedisExecDispatchCarriesReadFenceRouteVersion(t *testing.T) { + t.Parallel() + + key := []byte("txn:observed-route-version") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 7, + rangeKeysByVersion: map[uint64][][]byte{ + 7: {[]byte("txn-observed-route")}, + }, + } + coord := newReadKeyRecordingCoordinator(st) + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + } + + results, err := server.runTransactionDirect([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, uint64(7), coord.lastObservedRouteVersion) +} + +func TestRedisExecDispatchLeavesReadFenceRouteVersionZeroUnpinnedUntilCapabilityGate(t *testing.T) { + t.Parallel() + + key := []byte("txn:observed-route-version-zero") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 0, + rangeKeysByVersion: map[uint64][][]byte{ + 0: {[]byte("txn-observed-route-zero")}, + }, + } + coord := newReadKeyRecordingCoordinator(st) + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + } + + results, err := server.runTransactionDirect([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, uint64(0), coord.lastObservedRouteVersion) +} + +func TestRedisExecDedupFailsClosedWhenReadFenceRouteVersionChangesAfterAmbiguousAttempt(t *testing.T) { + t.Parallel() + + key := []byte("txn:route-version-dedup") + routeA := []byte("txn-dedup-route-a") + routeB := []byte("txn-dedup-route-b") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + rangeKeysByVersion: map[uint64][][]byte{ + 1: {routeA}, + 2: {routeB}, + }, + } + coord := &routeChangingDispatchCoordinator{ + verifyHookCoordinator: newVerifyHookCoordinator(st), + } + coord.groupForKey = readFenceRouteVersionGroupForKey(routeA, routeB) + coord.beforeDispatch = func(n int) { + if n == 1 { + st.advanceRouteVersionForTest() + } + } + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + onePhaseTxnDedup: true, + } + + results, err := server.runTransactionWithDedup([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.ErrorIs(t, err, errRedisExecRouteChangedAfterAmbiguousAttempt) + require.Nil(t, results) + require.Equal(t, 1, coord.dispatches) + require.GreaterOrEqual(t, countReadKey(coord.leaseKeys, routeA), 2) + require.Zero(t, countReadKey(coord.leaseKeys, routeB)) +} + +func TestRedisExecDedupRebuildsLockedAttemptWhenReadFenceRouteVersionChanges(t *testing.T) { + t.Parallel() + + key := []byte("txn:route-version-locked-rebuild") + routeA := []byte("txn-locked-route-a") + routeB := []byte("txn-locked-route-b") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + rangeKeysByVersion: map[uint64][][]byte{ + 1: {routeA}, + 2: {routeB}, + }, + } + coord := &routeChangingDispatchCoordinator{ + verifyHookCoordinator: newVerifyHookCoordinator(st), + firstErr: kv.NewTxnLockedError([]byte("redis-route-lock")), + } + coord.groupForKey = readFenceRouteVersionGroupForKey(routeA, routeB) + coord.beforeDispatch = func(n int) { + if n == 1 { + st.advanceRouteVersionForTest() + } + } + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + onePhaseTxnDedup: true, + } + + results, err := server.runTransactionWithDedup([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, "OK", results[0].str) + require.Equal(t, 2, coord.dispatches) + require.GreaterOrEqual(t, countReadKey(coord.leaseKeys, routeA), 2) + require.GreaterOrEqual(t, countReadKey(coord.leaseKeys, routeB), 2) +} + +func TestRedisTxnCommitRechecksReadFenceRouteVersionAfterPrepare(t *testing.T) { + t.Parallel() + + ctx := context.Background() + key := []byte("txn:prepare-route-version-direct") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + } + require.NoError(t, st.PutAt(ctx, store.HashFieldKey(key, []byte("field")), []byte("old"), redisTxnTestStartTS, 0)) + require.NoError(t, st.PutAt(ctx, store.HashMetaKey(key), store.MarshalHashMeta(store.HashMeta{Len: 1}), redisTxnTestStartTS, 0)) + + var bumpOnce sync.Once + st.onScanAt = func() { + bumpOnce.Do(func() { + st.advanceRouteVersionForTest() + }) + } + + coord := newReadKeyRecordingCoordinator(st) + server := NewRedisServer(nil, "", st, coord, nil, nil) + txn := newRedisTxnTestContext(server) + txn.ctx = ctx + txn.logicalDeletes[string(key)] = key + + err := txn.commit(redisReadFenceRouteVersion{tracked: true, version: 1}) + require.ErrorIs(t, err, store.ErrWriteConflict) + require.Zero(t, coord.dispatches) +} + +func TestRedisExecDedupRetriesWhenReadFenceRouteVersionChangesDuringPrepare(t *testing.T) { + t.Parallel() + + key := []byte("txn:prepare-route-version-dedup") + routeA := []byte("txn-prepare-route-a") + routeB := []byte("txn-prepare-route-b") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + rangeKeysByVersion: map[uint64][][]byte{ + 1: {routeA}, + 2: {routeB}, + }, + } + var bumpOnce sync.Once + st.onScanAt = func() { + bumpOnce.Do(func() { + st.advanceRouteVersionForTest() + }) + } + + coord := newReadKeyRecordingCoordinator(st) + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + onePhaseTxnDedup: true, + } + + results, err := server.runTransactionWithDedup([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, 1, coord.dispatches) + require.Equal(t, uint64(2), coord.lastObservedRouteVersion) +} + +func TestRedisExecRetriesComposedRouteRejectionByRebuilding(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + dedup bool + }{ + {name: "direct", dedup: false}, + {name: "dedup", dedup: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + key := []byte("txn:composed-retry:" + tc.name) + st := store.NewMVCCStore() + coord := &composedRetryCoordinator{ + localAdapterCoordinator: newLocalAdapterCoordinator(st), + } + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + onePhaseTxnDedup: tc.dedup, + } + + results, err := server.runTransaction([]redcon.Command{{ + Args: [][]byte{[]byte(cmdSet), key, []byte("value")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, "OK", results[0].str) + require.Equal(t, 2, coord.dispatches) + require.Equal(t, []uint64{0, 0}, coord.prevCommitTS) + }) + } +} + +func TestRedisExecProxyRouteRetriesWhenReadFenceRouteVersionChanges(t *testing.T) { + t.Parallel() + + key := []byte("txn:route-version-proxy") + routeA := []byte("txn-proxy-route-a") + routeB := []byte("txn-proxy-route-b") + st := &redisReadFenceRouteVersionStore{ + MVCCStore: store.NewMVCCStore(), + routeVersion: 1, + rangeKeysByVersion: map[uint64][][]byte{ + 1: {routeA}, + 2: {routeB}, + }, + } + var bumpOnce sync.Once + st.onReadFenceGroupKeys = func() { + bumpOnce.Do(func() { + st.advanceRouteVersionForTest() + }) + } + coord := &redisTxnFenceRoutingCoordinator{ + localLeader: func([]byte) bool { + return false + }, + raftLeader: func([]byte) string { + return "raft-remote" + }, + groupID: readFenceRouteVersionGroupForKey(routeA, routeB), + } + server := &RedisServer{ + store: st, + coordinator: coord, + leaderRedis: map[string]string{"raft-remote": "redis-remote"}, + } + + route, err := server.retryTransactionProxyRoute(context.Background(), []redcon.Command{{ + Args: [][]byte{[]byte(cmdLRange), key, []byte("0"), []byte("-1")}, + }}) + require.NoError(t, err) + require.Equal(t, redisStrKey(key), route.key) + st.mu.Lock() + lastReadFenceGroupKeys := cloneReadKeys(st.lastReadFenceGroupKeys) + st.mu.Unlock() + require.Equal(t, [][]byte{routeB}, lastReadFenceGroupKeys) + require.Greater(t, st.readFenceGroupKeyCalls(), len(redisListReadFenceRanges(key))) +} + +func TestRedisExecVerifiesLeaderBeforeSnapshot(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + dedup bool + }{ + {name: "direct", dedup: false}, + {name: "dedup", dedup: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + st := store.NewMVCCStore() + key := []byte("list:leader-fence-exec:" + tc.name) + coord := newVerifyHookCoordinator(st) + server := &RedisServer{ + store: st, + coordinator: coord, + scriptCache: map[string]string{}, + onePhaseTxnDedup: tc.dedup, + } + + seeded := false + coord.leaseForKey = func(_ context.Context, _ []byte) (uint64, error) { + if !seeded { + seedRedisListAt(t, st, key, "v1") + seeded = true + } + return 0, nil + } + + results, err := server.runTransaction([]redcon.Command{{ + Args: [][]byte{[]byte(cmdLRange), key, []byte("0"), []byte("-1")}, + }}) + require.NoError(t, err) + require.Len(t, results, 1) + require.Equal(t, resultArray, results[0].typ) + require.Equal(t, []string{"v1"}, results[0].arr) + require.Equal(t, 2, coord.leaseCalls) + }) + } +} + +func TestRedisExecReadFenceUsesRedisStorageKeys(t *testing.T) { + t.Parallel() + + key := []byte("!sqs|user-visible") + keys := redisQueuedCommandReadFenceKeys([]redcon.Command{{ + Args: [][]byte{[]byte(cmdGet), key}, + }}) + keySet := make(map[string]struct{}, len(keys)) + for _, got := range keys { + keySet[string(got)] = struct{}{} + } + + require.Contains(t, keySet, string(redisStrKey(key))) + require.NotContains(t, keySet, string(key)) +} + // TestRedisTxnValidateReadSet_ConcurrentRPushTriggersConflict verifies that a // concurrent RPUSH to a list triggers an OCC read-write conflict for a MULTI // transaction that read the list via LRANGE. Without the boundary key tracking diff --git a/kv/fsm.go b/kv/fsm.go index 6ed81f076..c9820bf9d 100644 --- a/kv/fsm.go +++ b/kv/fsm.go @@ -729,8 +729,8 @@ func (f *kvFSM) verifyComposed1(r *pb.Request) error { if f.routes == nil || f.shardGroupID == 0 { return nil } - observedVer := r.GetObservedRouteVersion() - if observedVer == 0 { + observedVer, pinned := DecodeObservedRouteVersion(r.GetObservedRouteVersion()) + if !pinned { return nil } diff --git a/kv/fsm_composed1_test.go b/kv/fsm_composed1_test.go index c80213f23..05ff2dd67 100644 --- a/kv/fsm_composed1_test.go +++ b/kv/fsm_composed1_test.go @@ -190,6 +190,21 @@ func TestVerifyComposed1_ObservedVersionZeroSkipsGate(t *testing.T) { } } +func TestVerifyComposed1_EncodedObservedVersionZeroRunsGate(t *testing.T) { + t.Parallel() + + e := distribution.NewEngineWithDefaultRoute() + applyComposed1Snapshot(t, e, 1, []distribution.RouteDescriptor{ + {RouteID: 100, Start: []byte(""), End: nil, GroupID: 2, State: distribution.RouteStateActive}, + }) + fsm := newComposed1FSM(t, e, 1) + + err := fsm.verifyComposed1(commitTxnRequest(ObservedRouteVersionZero, "k")) + require.ErrorIs(t, err, ErrComposed1Violation, + "an explicitly pinned version-0 observation must run the Composed-1 gate instead of being mistaken for the unpinned zero sentinel") + require.Contains(t, err.Error(), "current-version") +} + // TestVerifyComposed1_NilRouteHistorySkipsGate documents the // unwired-FSM default: a kvFSM constructed without WithRouteHistory // has routes=nil and the gate short-circuits. Matches the diff --git a/kv/shard_key.go b/kv/shard_key.go index 21afe185d..4ab12297e 100644 --- a/kv/shard_key.go +++ b/kv/shard_key.go @@ -66,6 +66,9 @@ func routeKey(key []byte) []byte { } func normalizeRouteKey(key []byte) []byte { + if user := listRouteKey(key); user != nil { + return user + } if user := redisRouteKey(key); user != nil { return user } @@ -81,10 +84,26 @@ func normalizeRouteKey(key []byte) []byte { if user := fskeys.ExtractRouteKey(key); user != nil { return user } + return key +} + +func listRouteKey(key []byte) []byte { + if store.IsListMetaDeltaKey(key) { + if user := store.ExtractListUserKeyFromDelta(key); user != nil { + return user + } + return store.ExtractListUserKeyFromDeltaScanPrefix(key) + } + if store.IsListClaimKey(key) { + if user := store.ExtractListUserKeyFromClaim(key); user != nil { + return user + } + return store.ExtractListUserKeyFromClaimScanPrefix(key) + } if user := store.ExtractListUserKey(key); user != nil { return user } - return key + return nil } func redisRouteKey(key []byte) []byte { @@ -93,7 +112,7 @@ func redisRouteKey(key []byte) []byte { } rest := key[len(redisInternalRoutePrefix):] sep := bytes.IndexByte(rest, '|') - if sep < 0 || sep+1 >= len(rest) { + if sep <= 0 { return nil } return rest[sep+1:] diff --git a/kv/shard_key_test.go b/kv/shard_key_test.go index 30f27fd94..9f87f9ddc 100644 --- a/kv/shard_key_test.go +++ b/kv/shard_key_test.go @@ -6,6 +6,7 @@ import ( "github.com/bootjp/elastickv/internal/fskeys" "github.com/bootjp/elastickv/internal/s3keys" + "github.com/bootjp/elastickv/store" "github.com/stretchr/testify/require" ) @@ -54,6 +55,40 @@ func TestRouteKey_NormalizesRedisTxnWideFenceKeys(t *testing.T) { } } +func TestRouteKey_NormalizesRedisInternalEmptyUserKey(t *testing.T) { + t.Parallel() + + for _, raw := range [][]byte{ + []byte("!redis|str|"), + []byte("!redis|route|"), + } { + got := routeKey(raw) + require.NotNil(t, got) + require.Empty(t, got) + } +} + +func TestRouteKey_NormalizesRedisListDeltaAndClaimKeys(t *testing.T) { + t.Parallel() + + for _, userKey := range [][]byte{ + []byte("!sqs|foo"), + []byte("!redis|str|foo"), + } { + t.Run(string(userKey), func(t *testing.T) { + t.Parallel() + for _, raw := range [][]byte{ + store.ListMetaDeltaKey(userKey, 12, 0), + store.ListMetaDeltaScanPrefix(userKey), + store.ListClaimKey(userKey, 3), + store.ListClaimScanPrefix(userKey), + } { + require.Equal(t, userKey, routeKey(raw)) + } + }) + } +} + func TestRouteKey_NormalizesDynamoKeysToTable(t *testing.T) { t.Parallel() diff --git a/kv/shard_store.go b/kv/shard_store.go index fc3276de9..0deb2f830 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -404,6 +404,52 @@ func (s *ShardStore) routesForReverseScan(start []byte, end []byte) ([]distribut return s.routesForScan(start, end, true) } +// ReadFenceGroupKeysForRange returns one representative routing key for each +// Raft group that ScanAt can visit for [start, end). It uses the same route +// expansion as ScanAt, so callers that take a snapshot outside ShardStore can +// fence every intersecting group before reading. +func (s *ShardStore) ReadFenceGroupKeysForRange(start []byte, end []byte) [][]byte { + if s == nil || s.engine == nil { + return nil + } + routes, clampToRoutes := s.routesForForwardScan(start, end) + keys := make([][]byte, 0, len(routes)) + seenGroups := make(map[uint64]struct{}, len(routes)) + for _, route := range routes { + if route.GroupID != 0 { + if _, seen := seenGroups[route.GroupID]; seen { + continue + } + seenGroups[route.GroupID] = struct{}{} + } + keys = append(keys, scanReadFenceRouteKey(route, start, clampToRoutes)) + } + return keys +} + +// ReadFenceRouteVersion returns the route catalog version paired with +// ReadFenceGroupKeysForRange so callers can discard reads whose route set +// changed before the scan completed. +func (s *ShardStore) ReadFenceRouteVersion() uint64 { + if s == nil || s.engine == nil { + return 0 + } + return s.engine.Version() +} + +func scanReadFenceRouteKey(route distribution.Route, start []byte, clampToRoutes bool) []byte { + if clampToRoutes { + return bytes.Clone(clampScanStart(start, route.Start)) + } + if listRouteKey(start) != nil { + return bytes.Clone(start) + } + if len(route.Start) > 0 { + return bytes.Clone(route.Start) + } + return bytes.Clone(start) +} + func (s *ShardStore) routesForScan(start []byte, end []byte, useFilesystemChunkRoutes bool) ([]distribution.Route, bool) { if routeStart, routeEnd, ok := s3keys.ManifestScanRouteBounds(start, end); ok { return s.engine.GetIntersectingRoutes(routeStart, routeEnd), false @@ -418,7 +464,7 @@ func (s *ShardStore) routesForScan(start []byte, end []byte, useFilesystemChunkR } // For internal list keys, shard routing is based on the logical user key // rather than the raw key prefix. - if userKey := store.ExtractListUserKey(start); userKey != nil { + if userKey := listRouteKey(start); userKey != nil { route, ok := s.engine.GetRoute(userKey) if !ok { return []distribution.Route{}, false diff --git a/kv/shard_store_test.go b/kv/shard_store_test.go index 83f6ceb63..65688ffef 100644 --- a/kv/shard_store_test.go +++ b/kv/shard_store_test.go @@ -86,6 +86,52 @@ func TestShardStoreScanAt_IncludesListKeysAcrossShards(t *testing.T) { require.Equal(t, itemKey, kvs[0].Key) } +func TestShardStoreReadFenceGroupKeysForRangeIncludesIntersectingRoutes(t *testing.T) { + t.Parallel() + + userKey := []byte("hash:fence-routes") + prefix := store.HashFieldScanPrefix(userKey) + split := append(append([]byte(nil), prefix...), 'm') + + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), split, 1) + engine.UpdateRoute(split, nil, 2) + st := NewShardStore(engine, map[uint64]*ShardGroup{ + 1: {}, + 2: {}, + }) + + got := st.ReadFenceGroupKeysForRange(prefix, store.PrefixScanEnd(prefix)) + + require.Equal(t, [][]byte{prefix, split}, got) +} + +func TestShardStoreReadFenceGroupKeysForListRangeUsesStorageRepresentative(t *testing.T) { + t.Parallel() + + for _, userKey := range [][]byte{ + []byte("!sqs|foo"), + []byte("!redis|str|foo"), + } { + t.Run(string(userKey), func(t *testing.T) { + t.Parallel() + prefix := store.ListMetaDeltaScanPrefix(userKey) + engine := distribution.NewEngine() + engine.UpdateRoute([]byte(""), userKey, 1) + engine.UpdateRoute(userKey, nil, 2) + st := NewShardStore(engine, map[uint64]*ShardGroup{ + 1: {}, + 2: {}, + }) + + got := st.ReadFenceGroupKeysForRange(prefix, store.PrefixScanEnd(prefix)) + + require.Equal(t, [][]byte{prefix}, got) + require.Equal(t, userKey, routeKey(got[0])) + }) + } +} + func TestShardStoreScanAt_RoutesListItemScansByUserKey(t *testing.T) { t.Parallel() diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index 4b4fea8ba..d5f601049 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -858,7 +858,7 @@ func (c *ShardedCoordinator) maybeAutoPinObservedRouteVersion(reqs *OperationGro if c.anyResolverClaimedKey(reqs.Elems) { return } - reqs.ObservedRouteVersion = c.engine.Version() + reqs.ObservedRouteVersion = EncodeObservedRouteVersion(c.engine.Version()) } // anyResolverClaimedKey reports whether any element's key is @@ -936,7 +936,7 @@ func (c *ShardedCoordinator) dispatchTxnWithComposed1Retry(ctx context.Context, // so a key whose owning group changed since the last // attempt naturally lands on the new group's FSM. if c.engine != nil { - reqs.ObservedRouteVersion = c.engine.Version() + reqs.ObservedRouteVersion = EncodeObservedRouteVersion(c.engine.Version()) } // Clear the timestamps so the next attempt allocates a // fresh pair against the post-shift HLC. The OCC diff --git a/kv/transcoder.go b/kv/transcoder.go index 1bb4ac455..4caedea39 100644 --- a/kv/transcoder.go +++ b/kv/transcoder.go @@ -14,6 +14,41 @@ const ( DelPrefix ) +// ObservedRouteVersionZero encodes a transaction pinned to catalog version 0. +// The protobuf field's literal zero remains the legacy/unpinned sentinel, so a +// real version-0 observation needs a distinct value. MaxUint64 is reserved for +// this internal encoding; normal catalog versions are far below this boundary. +const ObservedRouteVersionZero = ^uint64(0) + +// observedRouteVersionZeroWireEncodingEnabled stays disabled until every Raft +// member advertises support for ObservedRouteVersionZero. Mixed-version groups +// must keep literal zero on the wire so old followers do not treat the sentinel +// as an impossibly new catalog version and diverge from a new leader. +var observedRouteVersionZeroWireEncodingEnabled = false + +// EncodeObservedRouteVersion converts a tracked catalog version into the wire +// value carried by OperationGroup. Version zero is left as the legacy unpinned +// zero until the version-zero sentinel is capability-gated across Raft members. +func EncodeObservedRouteVersion(version uint64) uint64 { + if version == 0 && observedRouteVersionZeroWireEncodingEnabled { + return ObservedRouteVersionZero + } + return version +} + +// DecodeObservedRouteVersion converts OperationGroup.ObservedRouteVersion back +// to a catalog version and reports whether it was explicitly pinned. +func DecodeObservedRouteVersion(observed uint64) (uint64, bool) { + switch observed { + case 0: + return 0, false + case ObservedRouteVersionZero: + return 0, true + default: + return observed, true + } +} + // Elem is an element of a transaction. type Elem[T OP] struct { Op T @@ -49,12 +84,11 @@ type OperationGroup[T OP] struct { // ReadKeys carries the transaction's read set so the FSM can validate // read-write conflicts atomically with the commit. ReadKeys [][]byte - // ObservedRouteVersion is the durable catalog version this - // transaction's read set was captured at (typically set on - // BeginTxn from distribution.Engine.Version()). Zero means - // "unpinned" — every existing caller leaves it at zero so this - // is behaviour-neutral on the M1 plumbing PR. M3 of the - // Composed-1 design + // ObservedRouteVersion is the encoded durable catalog version this + // transaction's read set was captured at (typically set on BeginTxn + // from distribution.Engine.Version()). Zero means "unpinned"; the + // version-0 sentinel is decoded for compatibility but is not emitted + // until every Raft member advertises support. M3 of the Composed-1 design // (docs/design/2026_05_29_implemented_composed1_cross_group_commit_guard.md) // will gate the FSM apply path on this version so a route shift // between BeginTxn and Commit is caught before it can produce a diff --git a/kv/transcoder_test.go b/kv/transcoder_test.go new file mode 100644 index 000000000..d8d92aa12 --- /dev/null +++ b/kv/transcoder_test.go @@ -0,0 +1,35 @@ +package kv + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestEncodeObservedRouteVersionZeroGatedForRollingUpgrade(t *testing.T) { + require.Equal(t, uint64(0), EncodeObservedRouteVersion(0)) + + decoded, pinned := DecodeObservedRouteVersion(0) + require.Equal(t, uint64(0), decoded) + require.False(t, pinned) + + decoded, pinned = DecodeObservedRouteVersion(ObservedRouteVersionZero) + require.Equal(t, uint64(0), decoded) + require.True(t, pinned) + + previous := observedRouteVersionZeroWireEncodingEnabled + observedRouteVersionZeroWireEncodingEnabled = true + t.Cleanup(func() { + observedRouteVersionZeroWireEncodingEnabled = previous + }) + require.Equal(t, ObservedRouteVersionZero, EncodeObservedRouteVersion(0)) +} + +func TestEncodeObservedRouteVersionNonZeroPassesThrough(t *testing.T) { + t.Parallel() + + require.Equal(t, uint64(7), EncodeObservedRouteVersion(7)) + decoded, pinned := DecodeObservedRouteVersion(7) + require.Equal(t, uint64(7), decoded) + require.True(t, pinned) +} diff --git a/store/list_helpers.go b/store/list_helpers.go index 0f56f6878..2f5f4b727 100644 --- a/store/list_helpers.go +++ b/store/list_helpers.go @@ -131,28 +131,60 @@ func IsListClaimKey(key []byte) bool { // ExtractListUserKeyFromDelta extracts the logical user key from a list delta key. func ExtractListUserKeyFromDelta(key []byte) []byte { - trimmed := bytes.TrimPrefix(key, []byte(ListMetaDeltaPrefix)) - if len(trimmed) < wideColKeyLenSize+deltaKeyTSSize+deltaKeySeqSize { + trimmed, ok := bytes.CutPrefix(key, []byte(ListMetaDeltaPrefix)) + if !ok { + return nil + } + return extractListUserKeyFromLenPrefixedKey(trimmed, deltaKeyTSSize+deltaKeySeqSize) +} + +// ExtractListUserKeyFromClaim extracts the logical user key from a list claim key. +func ExtractListUserKeyFromClaim(key []byte) []byte { + trimmed, ok := bytes.CutPrefix(key, []byte(ListClaimPrefix)) + if !ok { + return nil + } + return extractListUserKeyFromLenPrefixedKey(trimmed, sortableInt64Bytes) +} + +// ExtractListUserKeyFromDeltaScanPrefix extracts the logical user key from the +// exact prefix produced by ListMetaDeltaScanPrefix. +func ExtractListUserKeyFromDeltaScanPrefix(key []byte) []byte { + return extractListUserKeyFromLenPrefixedScanPrefix(key, []byte(ListMetaDeltaPrefix)) +} + +// ExtractListUserKeyFromClaimScanPrefix extracts the logical user key from the +// exact prefix produced by ListClaimScanPrefix. +func ExtractListUserKeyFromClaimScanPrefix(key []byte) []byte { + return extractListUserKeyFromLenPrefixedScanPrefix(key, []byte(ListClaimPrefix)) +} + +func extractListUserKeyFromLenPrefixedScanPrefix(key []byte, prefix []byte) []byte { + if !bytes.HasPrefix(key, prefix) { + return nil + } + trimmed := key[len(prefix):] + if len(trimmed) < wideColKeyLenSize { return nil } ukLen := binary.BigEndian.Uint32(trimmed[:wideColKeyLenSize]) - if uint32(len(trimmed)) < uint32(wideColKeyLenSize)+ukLen+uint32(deltaKeyTSSize+deltaKeySeqSize) { //nolint:gosec // constants fit in uint32 + if uint64(len(trimmed)) != uint64(wideColKeyLenSize)+uint64(ukLen) { return nil } - return trimmed[wideColKeyLenSize : wideColKeyLenSize+ukLen] + return trimmed[wideColKeyLenSize:] } -// ExtractListUserKeyFromClaim extracts the logical user key from a list claim key. -func ExtractListUserKeyFromClaim(key []byte) []byte { - trimmed := bytes.TrimPrefix(key, []byte(ListClaimPrefix)) - if len(trimmed) < wideColKeyLenSize+sortableInt64Bytes { +func extractListUserKeyFromLenPrefixedKey(trimmed []byte, suffixLen int) []byte { + if len(trimmed) < wideColKeyLenSize+suffixLen { return nil } ukLen := binary.BigEndian.Uint32(trimmed[:wideColKeyLenSize]) - if uint32(len(trimmed)) < uint32(wideColKeyLenSize)+ukLen+uint32(sortableInt64Bytes) { //nolint:gosec // constants fit in uint32 + availableUserKeyBytes := len(trimmed) - wideColKeyLenSize - suffixLen + if int64(ukLen) > int64(availableUserKeyBytes) { return nil } - return trimmed[wideColKeyLenSize : wideColKeyLenSize+ukLen] + userEnd := int64(wideColKeyLenSize) + int64(ukLen) + return trimmed[wideColKeyLenSize:userEnd] } // PrefixScanEnd returns the exclusive end key for a prefix scan. diff --git a/store/list_helpers_test.go b/store/list_helpers_test.go new file mode 100644 index 000000000..f45afeca0 --- /dev/null +++ b/store/list_helpers_test.go @@ -0,0 +1,117 @@ +package store + +import ( + "encoding/binary" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestExtractListUserKeyFromFullWideColumnKeys(t *testing.T) { + t.Parallel() + + userKey := []byte("list:user") + require.Equal(t, userKey, ExtractListUserKeyFromDelta(ListMetaDeltaKey(userKey, 10, 2))) + require.Equal(t, userKey, ExtractListUserKeyFromClaim(ListClaimKey(userKey, -3))) +} + +func TestExtractListUserKeyRejectsMalformedFullKeyLength(t *testing.T) { + t.Parallel() + + for _, tc := range []struct { + name string + prefix string + suffixLen int + extract func([]byte) []byte + }{ + { + name: "delta", + prefix: ListMetaDeltaPrefix, + suffixLen: deltaKeyTSSize + deltaKeySeqSize, + extract: ExtractListUserKeyFromDelta, + }, + { + name: "claim", + prefix: ListClaimPrefix, + suffixLen: sortableInt64Bytes, + extract: ExtractListUserKeyFromClaim, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + key := make([]byte, 0, len(tc.prefix)+wideColKeyLenSize+tc.suffixLen) + key = append(key, tc.prefix...) + lenOffset := len(key) + key = append(key, 0xff, 0xff, 0xff, 0xff) + key = append(key, make([]byte, tc.suffixLen)...) + binary.BigEndian.PutUint32(key[lenOffset:lenOffset+wideColKeyLenSize], ^uint32(0)) + + require.Nil(t, tc.extract(key)) + }) + } +} + +func TestExtractListUserKeyFromScanPrefixes(t *testing.T) { + t.Parallel() + + userKey := []byte("list:user") + deltaPrefix := ListMetaDeltaScanPrefix(userKey) + claimPrefix := ListClaimScanPrefix(userKey) + + for _, tc := range []struct { + name string + key []byte + extract func([]byte) []byte + want []byte + }{ + { + name: "delta valid", + key: deltaPrefix, + extract: ExtractListUserKeyFromDeltaScanPrefix, + want: userKey, + }, + { + name: "delta rejects different prefix", + key: claimPrefix, + extract: ExtractListUserKeyFromDeltaScanPrefix, + }, + { + name: "delta rejects truncated user key", + key: deltaPrefix[:len(deltaPrefix)-1], + extract: ExtractListUserKeyFromDeltaScanPrefix, + }, + { + name: "delta rejects trailing bytes", + key: append(append([]byte{}, deltaPrefix...), 0), + extract: ExtractListUserKeyFromDeltaScanPrefix, + }, + { + name: "claim valid", + key: claimPrefix, + extract: ExtractListUserKeyFromClaimScanPrefix, + want: userKey, + }, + { + name: "claim rejects different prefix", + key: deltaPrefix, + extract: ExtractListUserKeyFromClaimScanPrefix, + }, + { + name: "claim rejects truncated user key", + key: claimPrefix[:len(claimPrefix)-1], + extract: ExtractListUserKeyFromClaimScanPrefix, + }, + { + name: "claim rejects trailing bytes", + key: append(append([]byte{}, claimPrefix...), 0), + extract: ExtractListUserKeyFromClaimScanPrefix, + }, + } { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + + require.Equal(t, tc.want, tc.extract(tc.key)) + }) + } +}