From fefd809e68c8d5913bb0c24bcf2023a2d86186e5 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 14:55:03 +0900 Subject: [PATCH 1/9] Fix sparse Lua list pop fallback --- adapter/redis_lua_context.go | 76 +++++++++++++--- adapter/redis_lua_list_holes_test.go | 126 ++++++++++++++++++++++++++- kv/leader_routed_store.go | 14 +++ kv/shard_store.go | 30 +++++++ 4 files changed, 235 insertions(+), 11 deletions(-) diff --git a/adapter/redis_lua_context.go b/adapter/redis_lua_context.go index 4a57051b5..afb9340de 100644 --- a/adapter/redis_lua_context.go +++ b/adapter/redis_lua_context.go @@ -182,6 +182,10 @@ type luaPhysicalLimitedScanStore interface { ReverseScanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64) ([]*store.KVPair, bool, error) } +type luaExactScanFallbackDecider interface { + AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64, reverse bool) bool +} + type luaCommandHandler func(*luaScriptContext, []string) (luaReply, error) type luaRenameHandler func(*luaScriptContext, []byte, []byte) error @@ -2189,6 +2193,66 @@ func (c *luaScriptContext) scanListItemWindow(key []byte, meta store.ListMeta, s if err != nil { return luaLazyListBoundaryItem{}, false, errors.WithStack(err) } + if item, ok := luaListBoundaryItemFromKVs(key, meta, kvs); ok { + return item, true, nil + } + if physicalLimitReached { + if !c.allowExactListScanFallback(startKey, endKey, scanLimit, !left) { + return luaLazyListBoundaryItem{}, false, errors.Wrapf(ErrCollectionTooLarge, + "list %q sparse pop scanned %d physical item rows", string(key), scanLimit) + } + return c.scanListItemWindowExact(key, meta, startKey, endKey, left) + } + if len(kvs) >= scanLimit { + return luaLazyListBoundaryItem{}, false, errors.Wrapf(ErrCollectionTooLarge, + "list %q sparse pop scanned %d non-matching item rows", string(key), scanLimit) + } + return luaLazyListBoundaryItem{}, false, nil +} + +func (c *luaScriptContext) allowExactListScanFallback(startKey, endKey []byte, scanLimit int, reverse bool) bool { + decider, ok := c.server.store.(luaExactScanFallbackDecider) + if !ok { + return true + } + return decider.AllowExactScanFallbackAfterPhysicalLimit(c.ctx, startKey, endKey, scanLimit, scanLimit, c.startTS, reverse) +} + +func (c *luaScriptContext) scanListItemWindowExact(key []byte, meta store.ListMeta, startKey, endKey []byte, left bool) (luaLazyListBoundaryItem, bool, error) { + for { + var ( + kvs []*store.KVPair + err error + ) + if left { + kvs, err = c.server.store.ScanAt(c.ctx, startKey, endKey, 1, c.startTS) + } else { + kvs, err = c.server.store.ReverseScanAt(c.ctx, startKey, endKey, 1, c.startTS) + } + if err != nil { + return luaLazyListBoundaryItem{}, false, errors.WithStack(err) + } + if len(kvs) == 0 { + return luaLazyListBoundaryItem{}, false, nil + } + if item, ok := luaListBoundaryItemFromKVs(key, meta, kvs); ok { + return item, true, nil + } + if left { + startKey = scanStartAfterKey(kvs[0].Key) + } else { + endKey = append([]byte(nil), kvs[0].Key...) + } + } +} + +func scanStartAfterKey(key []byte) []byte { + next := make([]byte, len(key)+1) + copy(next, key) + return next +} + +func luaListBoundaryItemFromKVs(key []byte, meta store.ListMeta, kvs []*store.KVPair) (luaLazyListBoundaryItem, bool) { for _, kvp := range kvs { seq, ok := store.ExtractListItemSeq(kvp.Key, key) if !ok { @@ -2197,17 +2261,9 @@ func (c *luaScriptContext) scanListItemWindow(key []byte, meta store.ListMeta, s return luaLazyListBoundaryItem{ value: string(kvp.Value), index: seq - meta.Head, - }, true, nil - } - if physicalLimitReached { - return luaLazyListBoundaryItem{}, false, errors.Wrapf(ErrCollectionTooLarge, - "list %q sparse pop scanned %d physical item rows", string(key), scanLimit) + }, true } - if len(kvs) >= scanLimit { - return luaLazyListBoundaryItem{}, false, errors.Wrapf(ErrCollectionTooLarge, - "list %q sparse pop scanned %d non-matching item rows", string(key), scanLimit) - } - return luaLazyListBoundaryItem{}, false, nil + return luaLazyListBoundaryItem{}, false } func (c *luaScriptContext) scanAtPhysicalLimit(startKey, endKey []byte, scanLimit int) ([]*store.KVPair, bool, error) { diff --git a/adapter/redis_lua_list_holes_test.go b/adapter/redis_lua_list_holes_test.go index f3211385e..80ac1cd7c 100644 --- a/adapter/redis_lua_list_holes_test.go +++ b/adapter/redis_lua_list_holes_test.go @@ -217,7 +217,7 @@ func TestRedisLua_RPopLPushKeepsRemainingSparseHeadItem(t *testing.T) { require.Equal(t, []string{"job-2"}, dstValues) } -func TestRedisLua_RPopLPushFailsOnTooManyPhysicalTailTombstones(t *testing.T) { +func TestRedisLua_RPopLPushScansPastPhysicalTailTombstones(t *testing.T) { t.Parallel() ctx := context.Background() @@ -226,6 +226,8 @@ func TestRedisLua_RPopLPushFailsOnTooManyPhysicalTailTombstones(t *testing.T) { dst := []byte("bull:test:active:tombstone-tail") largeLen := int64(luaSparseListPopScanLimit) + 2 seedListMeta(t, r, src, store.ListMeta{Head: 0, Len: largeLen, Tail: largeLen}) + require.NoError(t, r.store.PutAt(ctx, listItemKey(src, 0), []byte("job-1"), 2, 0)) + seedListPrefixCollider(t, r, ctx, src, 0) commitTS := uint64(3) for seq := int64(1); seq <= int64(luaSparseListPopScanLimit)+1; seq++ { require.NoError(t, r.store.DeleteAt(ctx, listItemKey(src, seq), commitTS)) @@ -236,6 +238,103 @@ func TestRedisLua_RPopLPushFailsOnTooManyPhysicalTailTombstones(t *testing.T) { require.NoError(t, err) defer scriptCtx.Close() + reply, err := scriptCtx.cmdRPopLPush([]string{string(src), string(dst)}) + require.NoError(t, err) + require.Equal(t, luaReplyString, reply.kind) + require.Equal(t, "job-1", reply.text) + require.NoError(t, scriptCtx.commit()) + + readTS := r.readTS() + typ, err := r.keyTypeAt(ctx, src, readTS) + require.NoError(t, err) + require.Equal(t, redisTypeNone, typ) + dstValues, err := r.listValuesAt(ctx, dst, readTS) + require.NoError(t, err) + require.Equal(t, []string{"job-1"}, dstValues) +} + +func TestRedisLua_LPopScansPastPhysicalHeadTombstones(t *testing.T) { + t.Parallel() + + ctx := context.Background() + r := newListPopTestServer(t) + src := []byte("bull:test:wait:tombstone-head") + largeLen := int64(luaSparseListPopScanLimit) + 2 + seedListMeta(t, r, src, store.ListMeta{Head: 0, Len: largeLen, Tail: largeLen}) + require.NoError(t, r.store.PutAt(ctx, listItemKey(src, largeLen-1), []byte("job-1"), 2, 0)) + seedListPrefixCollider(t, r, ctx, src, 0) + commitTS := uint64(3) + for seq := int64(0); seq <= int64(luaSparseListPopScanLimit); seq++ { + require.NoError(t, r.store.DeleteAt(ctx, listItemKey(src, seq), commitTS)) + commitTS++ + } + + scriptCtx, err := newLuaScriptContext(ctx, r) + require.NoError(t, err) + defer scriptCtx.Close() + + reply, err := scriptCtx.cmdLPop([]string{string(src)}) + require.NoError(t, err) + require.Equal(t, luaReplyString, reply.kind) + require.Equal(t, "job-1", reply.text) + require.NoError(t, scriptCtx.commit()) + + readTS := r.readTS() + typ, err := r.keyTypeAt(ctx, src, readTS) + require.NoError(t, err) + require.Equal(t, redisTypeNone, typ) +} + +func TestRedisLua_RPopLPushDeletesPhysicalTombstoneOnlyList(t *testing.T) { + t.Parallel() + + ctx := context.Background() + r := newListPopTestServer(t) + src := []byte("bull:test:wait:tombstone-only") + dst := []byte("bull:test:active:tombstone-only") + largeLen := int64(luaSparseListPopScanLimit) + 2 + seedListMeta(t, r, src, store.ListMeta{Head: 0, Len: largeLen, Tail: largeLen}) + commitTS := uint64(2) + for seq := int64(0); seq <= int64(luaSparseListPopScanLimit)+1; seq++ { + require.NoError(t, r.store.DeleteAt(ctx, listItemKey(src, seq), commitTS)) + commitTS++ + } + + scriptCtx, err := newLuaScriptContext(ctx, r) + require.NoError(t, err) + defer scriptCtx.Close() + + reply, err := scriptCtx.cmdRPopLPush([]string{string(src), string(dst)}) + require.NoError(t, err) + require.Equal(t, luaReplyNil, reply.kind) + require.NoError(t, scriptCtx.commit()) + + readTS := r.readTS() + typ, err := r.keyTypeAt(ctx, src, readTS) + require.NoError(t, err) + require.Equal(t, redisTypeNone, typ) + dstValues, err := r.listValuesAt(ctx, dst, readTS) + require.NoError(t, err) + require.Empty(t, dstValues) +} + +func TestRedisLua_RPopLPushSyntheticPhysicalLimitFailsClosed(t *testing.T) { + t.Parallel() + + ctx := context.Background() + base := store.NewMVCCStore() + st := noExactFallbackPhysicalLimitStore{MVCCStore: base} + coord := newLocalAdapterCoordinator(st) + r := NewRedisServer(nil, "", st, coord, nil, nil) + src := []byte("bull:test:wait:synthetic-limit") + dst := []byte("bull:test:active:synthetic-limit") + seedListMeta(t, r, src, store.ListMeta{Head: 0, Len: 1, Tail: 1}) + require.NoError(t, r.store.PutAt(ctx, listItemKey(src, 0), []byte("job-1"), 2, 0)) + + scriptCtx, err := newLuaScriptContext(ctx, r) + require.NoError(t, err) + defer scriptCtx.Close() + _, err = scriptCtx.cmdRPopLPush([]string{string(src), string(dst)}) require.ErrorIs(t, err, ErrCollectionTooLarge) } @@ -275,3 +374,28 @@ func seedListMeta(t *testing.T, r *RedisServer, key []byte, meta store.ListMeta) require.NoError(t, err) require.NoError(t, r.store.PutAt(context.Background(), store.ListMetaKey(key), raw, 1, 0)) } + +func seedListPrefixCollider(t *testing.T, r *RedisServer, ctx context.Context, key []byte, seq int64) { + t.Helper() + + collider := append([]byte{}, key...) + collider = append(collider, listItemKey(key, seq)[len(store.ListItemPrefix)+len(key):]...) + collider = append(collider, 'x') + require.NoError(t, r.store.PutAt(ctx, listItemKey(collider, 0), []byte("other-list-job"), 2, 0)) +} + +type noExactFallbackPhysicalLimitStore struct { + store.MVCCStore +} + +func (s noExactFallbackPhysicalLimitStore) ScanAtPhysicalLimit(context.Context, []byte, []byte, int, int, uint64) ([]*store.KVPair, bool, error) { + return nil, true, nil +} + +func (s noExactFallbackPhysicalLimitStore) ReverseScanAtPhysicalLimit(context.Context, []byte, []byte, int, int, uint64) ([]*store.KVPair, bool, error) { + return nil, true, nil +} + +func (s noExactFallbackPhysicalLimitStore) AllowExactScanFallbackAfterPhysicalLimit(context.Context, []byte, []byte, int, int, uint64, bool) bool { + return false +} diff --git a/kv/leader_routed_store.go b/kv/leader_routed_store.go index f92cc39f5..8e169ac81 100644 --- a/kv/leader_routed_store.go +++ b/kv/leader_routed_store.go @@ -392,6 +392,20 @@ func (s *LeaderRoutedStore) ReverseScanAtPhysicalLimit(ctx context.Context, star return s.scanAtPhysicalLimit(ctx, start, end, visibleLimit, physicalLimit, ts, true) } +func (s *LeaderRoutedStore) AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, _ []byte, visibleLimit, physicalLimit int, _ uint64, _ bool) bool { + if s == nil || s.local == nil { + return false + } + if visibleLimit <= 0 || physicalLimit <= 0 { + return false + } + if ok, _ := s.leaderFenceTS(ctx, start); !ok { + return false + } + _, ok := s.local.(physicalLimitedStore) + return ok +} + func (s *LeaderRoutedStore) scanAtPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64, reverse bool) ([]*store.KVPair, bool, error) { if s == nil || s.local == nil { return []*store.KVPair{}, false, nil diff --git a/kv/shard_store.go b/kv/shard_store.go index 81b21e28d..2ef4dd868 100644 --- a/kv/shard_store.go +++ b/kv/shard_store.go @@ -302,6 +302,36 @@ func (s *ShardStore) ReverseScanAtPhysicalLimit(ctx context.Context, start []byt return s.scanRouteAtDirectionPhysicalLimit(ctx, routes[0], start, end, visibleLimit, physicalLimit, ts, true) } +func (s *ShardStore) AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, _ uint64, _ bool) bool { + if visibleLimit <= 0 || physicalLimit <= 0 { + return false + } + g := s.exactFallbackPhysicalLimitGroup(start, end) + if g == nil { + return false + } + if _, ok := g.Store.(physicalLimitedStore); !ok { + return false + } + engine := engineForGroup(g) + return engine == nil || isLinearizableRaftLeader(ctx, engine) +} + +func (s *ShardStore) exactFallbackPhysicalLimitGroup(start []byte, end []byte) *ShardGroup { + if s == nil || s.engine == nil { + return nil + } + routes, clampToRoutes := s.routesForScan(start, end) + if len(routes) != 1 || clampToRoutes { + return nil + } + g, ok := s.groupForID(routes[0].GroupID) + if !ok || g == nil || g.Store == nil { + return nil + } + return g +} + func (s *ShardStore) routesForScan(start []byte, end []byte) ([]distribution.Route, bool) { if routeStart, routeEnd, ok := s3keys.ManifestScanRouteBounds(start, end); ok { return s.engine.GetIntersectingRoutes(routeStart, routeEnd), false From e1b32b5704a2acc875f8dff4a3b9d98d3e208d1c Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 15:48:25 +0900 Subject: [PATCH 2/9] Align ElasticKV proxy socket timeouts --- cmd/redis-proxy/main.go | 20 ++++++++++++++++++++ cmd/redis-proxy/main_test.go | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 52 insertions(+) diff --git a/cmd/redis-proxy/main.go b/cmd/redis-proxy/main.go index 4ad067987..b9483ec1e 100644 --- a/cmd/redis-proxy/main.go +++ b/cmd/redis-proxy/main.go @@ -23,6 +23,8 @@ const ( sentryFlushTimeout = 2 * time.Second metricsShutdownTimeout = 5 * time.Second secondaryConcurrencyDivisor = 2 + elasticKVDispatchTimeout = 10 * time.Second + backendTimeoutGrace = time.Second ) func main() { @@ -121,6 +123,7 @@ func newBackends(cfg proxy.ProxyConfig, primaryPoolSize, elasticKVPoolSize int, secondaryOpts.DB = cfg.SecondaryDB secondaryOpts.Password = cfg.SecondaryPassword secondaryOpts.PoolSize = elasticKVPoolSize + alignElasticKVBackendTimeouts(&secondaryOpts, cfg.SecondaryTimeout) secondarySeeds := parseAddrList(cfg.SecondaryAddr) if len(secondarySeeds) == 0 { @@ -145,6 +148,23 @@ func newBackends(cfg proxy.ProxyConfig, primaryPoolSize, elasticKVPoolSize int, } } +func alignElasticKVBackendTimeouts(opts *proxy.BackendOptions, operationTimeout time.Duration) { + if opts == nil { + return + } + floor := elasticKVDispatchTimeout + if operationTimeout > floor { + floor = operationTimeout + } + floor += backendTimeoutGrace + if opts.ReadTimeout > 0 && opts.ReadTimeout < floor { + opts.ReadTimeout = floor + } + if opts.WriteTimeout > 0 && opts.WriteTimeout < floor { + opts.WriteTimeout = floor + } +} + func serveMetrics(ctx context.Context, addr string, reg *prometheus.Registry, logger *slog.Logger) { mux := http.NewServeMux() mux.Handle("/metrics", promhttp.HandlerFor(reg, promhttp.HandlerOpts{})) diff --git a/cmd/redis-proxy/main_test.go b/cmd/redis-proxy/main_test.go index 29e9a47b7..8690f483c 100644 --- a/cmd/redis-proxy/main_test.go +++ b/cmd/redis-proxy/main_test.go @@ -2,6 +2,7 @@ package main import ( "testing" + "time" "github.com/bootjp/elastickv/proxy" "github.com/stretchr/testify/assert" @@ -96,3 +97,34 @@ func TestDeriveSecondaryConcurrency(t *testing.T) { }) } } + +func TestAlignElasticKVBackendTimeouts(t *testing.T) { + t.Run("uses ElasticKV dispatch floor by default", func(t *testing.T) { + opts := proxy.DefaultElasticKVBackendOptions() + + alignElasticKVBackendTimeouts(&opts, 5*time.Second) + + assert.Equal(t, 11*time.Second, opts.ReadTimeout) + assert.Equal(t, 11*time.Second, opts.WriteTimeout) + }) + + t.Run("follows longer secondary timeout", func(t *testing.T) { + opts := proxy.DefaultElasticKVBackendOptions() + + alignElasticKVBackendTimeouts(&opts, 15*time.Second) + + assert.Equal(t, 16*time.Second, opts.ReadTimeout) + assert.Equal(t, 16*time.Second, opts.WriteTimeout) + }) + + t.Run("keeps explicit larger timeout", func(t *testing.T) { + opts := proxy.DefaultElasticKVBackendOptions() + opts.ReadTimeout = 30 * time.Second + opts.WriteTimeout = 31 * time.Second + + alignElasticKVBackendTimeouts(&opts, 15*time.Second) + + assert.Equal(t, 30*time.Second, opts.ReadTimeout) + assert.Equal(t, 31*time.Second, opts.WriteTimeout) + }) +} From cf478580778795fee500f1a48de7461160a41828 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 15:54:20 +0900 Subject: [PATCH 3/9] Replay blocking zset pops deterministically --- proxy/blocking.go | 53 +++++++++++++++++++++++++++++++++++++++++ proxy/dualwrite.go | 9 +++---- proxy/proxy_test.go | 57 +++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 113 insertions(+), 6 deletions(-) diff --git a/proxy/blocking.go b/proxy/blocking.go index 18a42a4ec..8a1bf64be 100644 --- a/proxy/blocking.go +++ b/proxy/blocking.go @@ -2,6 +2,7 @@ package proxy import ( "context" + "fmt" "strconv" "strings" "time" @@ -52,3 +53,55 @@ func parseBlockingMillisecondsArg(raw []byte) time.Duration { } return time.Duration(millis) * time.Millisecond } + +func blockingReplayCommand(cmd string, _ [][]byte, resp any) (string, []any, bool) { + switch strings.ToUpper(cmd) { + case "BZPOPMIN", "BZPOPMAX": + parts, ok := redisArray(resp) + if !ok || len(parts) < 2 { + return "", nil, false + } + key, keyOK := redisArg(parts[0]) + member, memberOK := redisArg(parts[1]) + if !keyOK || !memberOK { + return "", nil, false + } + return "ZREM", []any{[]byte("ZREM"), key, member}, true + default: + return "", nil, false + } +} + +func redisArray(v any) ([]any, bool) { + switch x := v.(type) { + case []any: + return x, true + case []string: + out := make([]any, len(x)) + for i := range x { + out[i] = x[i] + } + return out, true + case [][]byte: + out := make([]any, len(x)) + for i := range x { + out[i] = x[i] + } + return out, true + default: + return nil, false + } +} + +func redisArg(v any) (any, bool) { + switch x := v.(type) { + case nil: + return nil, false + case []byte: + return append([]byte(nil), x...), true + case string: + return []byte(x), true + default: + return []byte(fmt.Sprint(x)), true + } +} diff --git a/proxy/dualwrite.go b/proxy/dualwrite.go index 1c940e557..fc69b6773 100644 --- a/proxy/dualwrite.go +++ b/proxy/dualwrite.go @@ -241,13 +241,10 @@ func (d *DualWriter) Blocking(ctx context.Context, cmd string, args [][]byte) (a } d.metrics.CommandTotal.WithLabelValues(cmd, d.primary.Name(), "ok").Inc() - // Warmup: send to secondary with short timeout (fire-and-forget, bounded) if d.hasSecondaryWrite() { - d.goWrite(func() { - sCtx, cancel := context.WithTimeout(context.Background(), time.Second) - defer cancel() - d.secondary.Do(sCtx, iArgs...) - }) + if replayCmd, replayArgs, ok := blockingReplayCommand(cmd, args, resp); ok { + d.runSecondaryWrite(func() { d.writeSecondary(replayCmd, replayArgs) }) + } } return resp, err //nolint:wrapcheck // redis.Nil must pass through unwrapped for callers to detect nil replies diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index 4398aca72..8561534f2 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -62,6 +62,16 @@ func (b *mockBackend) CallCount() int { return len(b.calls) } +func (b *mockBackend) Calls() [][]any { + b.mu.Lock() + defer b.mu.Unlock() + out := make([][]any, len(b.calls)) + for i := range b.calls { + out[i] = append([]any(nil), b.calls[i]...) + } + return out +} + // Helper to create a doFunc that returns a specific value. func makeCmd(val any, err error) func(ctx context.Context, args ...any) *redis.Cmd { return func(ctx context.Context, args ...any) *redis.Cmd { @@ -623,6 +633,53 @@ func TestDualWriter_Blocking_UsesTimeoutAwareBackend(t *testing.T) { assert.Equal(t, []any{[]byte("BZPOPMIN"), []byte("queue"), []byte("5")}, primary.args) } +func TestDualWriter_Blocking_ReplaysBZPopMinAsZRem(t *testing.T) { + primary := &timeoutCapturingBackend{ + name: "primary", + returnValue: []any{[]byte("queue"), []byte("job-1"), []byte("42")}, + } + secondary := newMockBackend("secondary") + + metrics := newTestMetrics() + cfg := ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: 10 * time.Second} + d := NewDualWriter(primary, secondary, cfg, metrics, newTestSentry(), testLogger) + + resp, err := d.Blocking(context.Background(), "BZPOPMIN", [][]byte{[]byte("BZPOPMIN"), []byte("queue"), []byte("5")}) + assert.NoError(t, err) + assert.Equal(t, []any{[]byte("queue"), []byte("job-1"), []byte("42")}, resp) + d.Close() + + assert.Equal(t, [][]any{{[]byte("ZREM"), []byte("queue"), []byte("job-1")}}, secondary.Calls()) + assert.InDelta(t, 0, testutil.ToFloat64(metrics.AsyncDrops), 0.001) + assert.InDelta(t, 1, testutil.ToFloat64(metrics.CommandTotal.WithLabelValues("ZREM", "secondary", "ok")), 0.001) +} + +func TestDualWriter_Blocking_XReadDoesNotUseWriteSemaphore(t *testing.T) { + primary := &timeoutCapturingBackend{name: "primary", returnValue: []any{}} + secondary := newMockBackend("secondary") + + metrics := newTestMetrics() + cfg := ProxyConfig{Mode: ModeDualWrite, SecondaryWriteConcurrency: 1, SecondaryTimeout: 10 * time.Second} + d := NewDualWriter(primary, secondary, cfg, metrics, newTestSentry(), testLogger) + + blocker := make(chan struct{}) + d.goWrite(func() { + <-blocker + }) + + resp, err := d.Blocking(context.Background(), "XREAD", [][]byte{ + []byte("XREAD"), []byte("BLOCK"), []byte("1"), []byte("STREAMS"), []byte("jobs"), []byte("0"), + }) + assert.NoError(t, err) + assert.Equal(t, []any{}, resp) + assert.Empty(t, secondary.Calls()) + assert.InDelta(t, 0, testutil.ToFloat64(metrics.AsyncDrops), 0.001) + assert.InDelta(t, 0, testutil.ToFloat64(metrics.AsyncBackpressure), 0.001) + + close(blocker) + d.Close() +} + func TestDualWriter_GoAsync_Bounded(t *testing.T) { primary := newMockBackend("primary") primary.doFunc = makeCmd("OK", nil) From f69764063c661d71876867fe190b31d1a65d8a82 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 16:10:00 +0900 Subject: [PATCH 4/9] Cap ElasticKV secondary script concurrency --- cmd/redis-proxy/main.go | 4 ++++ cmd/redis-proxy/main_test.go | 4 ++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/cmd/redis-proxy/main.go b/cmd/redis-proxy/main.go index b9483ec1e..16117c4cd 100644 --- a/cmd/redis-proxy/main.go +++ b/cmd/redis-proxy/main.go @@ -23,6 +23,7 @@ const ( sentryFlushTimeout = 2 * time.Second metricsShutdownTimeout = 5 * time.Second secondaryConcurrencyDivisor = 2 + elasticKVScriptConcurrency = 1 elasticKVDispatchTimeout = 10 * time.Second backendTimeoutGrace = time.Second ) @@ -229,6 +230,9 @@ func deriveSecondaryConcurrency(mode proxy.ProxyMode, primaryPoolSize, elasticKV } if scriptConcurrency == 0 { scriptConcurrency = defaultSecondaryScriptConcurrency(writeConcurrency) + if mode != proxy.ModeElasticKVPrimary && scriptConcurrency > elasticKVScriptConcurrency { + scriptConcurrency = elasticKVScriptConcurrency + } } return writeConcurrency, scriptConcurrency } diff --git a/cmd/redis-proxy/main_test.go b/cmd/redis-proxy/main_test.go index 8690f483c..0e51ff630 100644 --- a/cmd/redis-proxy/main_test.go +++ b/cmd/redis-proxy/main_test.go @@ -44,7 +44,7 @@ func TestDeriveSecondaryConcurrency(t *testing.T) { primaryPoolSize: 128, elasticKVPoolSize: 8, wantWriteConcurrency: 4, - wantScriptConcurrency: 2, + wantScriptConcurrency: 1, }, { name: "ElasticKV primary derives from Redis secondary pool", @@ -61,7 +61,7 @@ func TestDeriveSecondaryConcurrency(t *testing.T) { elasticKVPoolSize: 4, writeConcurrency: 5, wantWriteConcurrency: 5, - wantScriptConcurrency: 2, + wantScriptConcurrency: 1, }, { name: "explicit values win", From 7e409795c6724fe949318a8a4dc712dbdf210f5a Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 16:13:52 +0900 Subject: [PATCH 5/9] Refresh ElasticKV leader after not-leader replies --- proxy/dualwrite.go | 23 +++++++++++++-- proxy/leader_aware_backend.go | 30 ++++++++++++++++++- proxy/leader_aware_backend_test.go | 34 +++++++++++++++++++++ proxy/proxy_test.go | 47 ++++++++++++++++++++++++++++++ 4 files changed, 131 insertions(+), 3 deletions(-) diff --git a/proxy/dualwrite.go b/proxy/dualwrite.go index fc69b6773..c1e284d48 100644 --- a/proxy/dualwrite.go +++ b/proxy/dualwrite.go @@ -52,6 +52,10 @@ const ( asyncDropLogInterval = 5 * time.Second ) +type leaderRefreshingBackend interface { + RefreshLeaderNow(context.Context) +} + // readTSCompactedMarker is the substring produced by // store.ErrReadTSCompacted as it flows through gRPC (wrapped as // FailedPrecondition) and Lua PCall. Matching on substring is necessary @@ -73,13 +77,26 @@ func isRetryableSecondaryWriteError(err error) bool { return true } switch classifySecondaryWriteError(err) { - case "retry_limit", "write_conflict", "txn_locked": + case "retry_limit", "write_conflict", "txn_locked", "not_leader": return true default: return false } } +func isNotLeaderError(err error) bool { + return classifySecondaryWriteError(err) == "not_leader" +} + +func refreshSecondaryLeader(ctx context.Context, backend Backend, err error) { + if !isNotLeaderError(err) { + return + } + if refresher, ok := backend.(leaderRefreshingBackend); ok { + refresher.RefreshLeaderNow(ctx) + } +} + // DualWriter routes commands to primary and secondary backends based on mode. type DualWriter struct { primary Backend @@ -334,8 +351,10 @@ func (d *DualWriter) writeSecondary(cmd string, iArgs []any) { if attempt >= maxSecondaryTransientRetries { break } + reason := classifySecondaryWriteError(sErr) + refreshSecondaryLeader(sCtx, d.secondary, sErr) d.logger.Debug("retrying secondary write after transient error", - "cmd", cmd, "attempt", attempt+1, "backoff", backoff, "reason", classifySecondaryWriteError(sErr), "err", sErr) + "cmd", cmd, "attempt", attempt+1, "backoff", backoff, "reason", reason, "err", sErr) if !waitCompactedRetryBackoff(sCtx, backoff) { break } diff --git a/proxy/leader_aware_backend.go b/proxy/leader_aware_backend.go index 878d51ebc..3932d4a9d 100644 --- a/proxy/leader_aware_backend.go +++ b/proxy/leader_aware_backend.go @@ -179,6 +179,13 @@ func (b *LeaderAwareRedisBackend) TriggerRefresh() { } } +// RefreshLeaderNow re-probes the cluster before returning. It is used by +// callers that already observed a not-leader response and need the next retry +// to use a fresh target instead of waiting for the background loop. +func (b *LeaderAwareRedisBackend) RefreshLeaderNow(ctx context.Context) { + b.refreshLeader(ctx) +} + // refreshLeader probes INFO replication on the current leader first, then on // each seed, and adopts the first advertised leader address. The current // leader's Redis address is returned by the leader node itself when it's @@ -353,8 +360,19 @@ func (b *LeaderAwareRedisBackend) currentClient() *redis.Client { return b.clients[b.leader] } -// Do forwards a single command to the current leader. +// Do forwards a single command to the current leader. A not-leader Redis reply +// means the command was rejected before applying, so it is safe to refresh the +// leader and retry once. func (b *LeaderAwareRedisBackend) Do(ctx context.Context, args ...any) *redis.Cmd { + cmd := b.doOnce(ctx, args...) + if !isNotLeaderError(cmd.Err()) { + return cmd + } + b.RefreshLeaderNow(ctx) + return b.doOnce(ctx, args...) +} + +func (b *LeaderAwareRedisBackend) doOnce(ctx context.Context, args ...any) *redis.Cmd { cli := b.currentClient() if cli == nil { cmd := redis.NewCmd(ctx, args...) @@ -365,7 +383,17 @@ func (b *LeaderAwareRedisBackend) Do(ctx context.Context, args ...any) *redis.Cm } // DoWithTimeout forwards a blocking command with a per-call socket timeout. +// Like Do, a not-leader rejection is refreshed and retried once. func (b *LeaderAwareRedisBackend) DoWithTimeout(ctx context.Context, timeout time.Duration, args ...any) *redis.Cmd { + cmd := b.doWithTimeoutOnce(ctx, timeout, args...) + if !isNotLeaderError(cmd.Err()) { + return cmd + } + b.RefreshLeaderNow(ctx) + return b.doWithTimeoutOnce(ctx, timeout, args...) +} + +func (b *LeaderAwareRedisBackend) doWithTimeoutOnce(ctx context.Context, timeout time.Duration, args ...any) *redis.Cmd { cli := b.currentClient() if cli == nil { cmd := redis.NewCmd(ctx, args...) diff --git a/proxy/leader_aware_backend_test.go b/proxy/leader_aware_backend_test.go index ea4204272..034822d59 100644 --- a/proxy/leader_aware_backend_test.go +++ b/proxy/leader_aware_backend_test.go @@ -127,6 +127,10 @@ func (n *fakeElasticKVNode) handleConn(conn net.Conn) { _, _ = fmt.Fprintf(conn, "$%d\r\n%s\r\n", len(body), body) default: n.commands.Add(1) + if leader := n.Leader(); leader != "" && leader != n.addr { + _, _ = conn.Write([]byte("-ERR etcd raft engine is not leader\r\n")) + continue + } _, _ = conn.Write([]byte("+OK\r\n")) } } @@ -214,6 +218,36 @@ func TestLeaderAwareRedisBackend_FollowsLeaderChange(t *testing.T) { require.Equal(t, beforeB+1, nodeB.commands.Load(), "command must reach new leader B") } +func TestLeaderAwareRedisBackend_RetryRefreshesOnNotLeader(t *testing.T) { + nodeA := newFakeElasticKVNode(t) + nodeB := newFakeElasticKVNode(t) + + nodeA.SetLeader(nodeA.addr) + nodeB.SetLeader(nodeA.addr) + + backend := NewLeaderAwareRedisBackendWithInterval( + []string{nodeA.addr, nodeB.addr}, + "elastickv", + DefaultBackendOptions(), + time.Hour, 500*time.Millisecond, + testLogger, + ) + t.Cleanup(func() { _ = backend.Close() }) + + require.Eventually(t, func() bool { + return backend.CurrentLeader() == nodeA.addr + }, 2*time.Second, 10*time.Millisecond, "initial leader must be A") + + nodeA.SetLeader(nodeB.addr) + nodeB.SetLeader(nodeB.addr) + + res := backend.Do(context.Background(), "SET", "k", "v") + require.NoError(t, res.Err()) + require.Equal(t, nodeB.addr, backend.CurrentLeader(), "not-leader retry must synchronously adopt the advertised leader") + require.Equal(t, int64(1), nodeA.commands.Load(), "first attempt should hit the former leader and be rejected") + require.Equal(t, int64(1), nodeB.commands.Load(), "retry should land on the refreshed leader") +} + func TestLeaderAwareRedisBackend_ConcurrentCloseIsRaceFree(t *testing.T) { // Regression guard: Close() must not race concurrent Do() callers — // currentClient and ensureClientLocked hold the lock consistently with diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index 8561534f2..dcfa37ffc 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -72,6 +72,24 @@ func (b *mockBackend) Calls() [][]any { return out } +type refreshableMockBackend struct { + *mockBackend + refreshMu sync.Mutex + refreshes int +} + +func (b *refreshableMockBackend) RefreshLeaderNow(ctx context.Context) { + b.refreshMu.Lock() + defer b.refreshMu.Unlock() + b.refreshes++ +} + +func (b *refreshableMockBackend) RefreshCount() int { + b.refreshMu.Lock() + defer b.refreshMu.Unlock() + return b.refreshes +} + // Helper to create a doFunc that returns a specific value. func makeCmd(val any, err error) func(ctx context.Context, args ...any) *redis.Cmd { return func(ctx context.Context, args ...any) *redis.Cmd { @@ -1185,6 +1203,35 @@ func TestDualWriter_writeSecondary_RetriesRetryLimitWriteConflict(t *testing.T) "a retried success must not count as a secondary write error") } +func TestDualWriter_writeSecondary_RetriesNotLeaderAfterRefresh(t *testing.T) { + primary := newMockBackend("primary") + primary.doFunc = makeCmd("OK", nil) + + secondary := &refreshableMockBackend{mockBackend: newMockBackend("secondary")} + notLeaderErr := testRedisErr("ERR etcd raft engine is not leader") + var calls int + secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd { + calls++ + cmd := redis.NewCmd(ctx, args...) + if calls == 1 { + cmd.SetErr(notLeaderErr) + return cmd + } + cmd.SetVal("OK") + return cmd + } + + metrics := newTestMetrics() + d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger) + + d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte("deadbeef"), []byte("0")}) + + assert.Equal(t, 2, calls, "secondary must retry after a not-leader rejection") + assert.Equal(t, 1, secondary.RefreshCount(), "not-leader retries must force leader rediscovery before retrying") + assert.InDelta(t, 0, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001, + "a refreshed retry success must not count as a secondary write error") +} + func TestDualWriter_writeSecondary_RetriesDoNotRepeatNoScriptProbe(t *testing.T) { // After the EVAL fallback resolves a NOSCRIPT, a compacted-retry must // re-send the resolved EVAL form directly — never the known-missing From 3df9239e81fe618f2a8c21a7c38c072940ab275c Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 16:36:56 +0900 Subject: [PATCH 6/9] Avoid full stream rewrites for Lua XADD --- adapter/redis_lua_compat_test.go | 69 +++++++++ adapter/redis_lua_context.go | 241 ++++++++++++++++++++++++++++++- 2 files changed, 304 insertions(+), 6 deletions(-) diff --git a/adapter/redis_lua_compat_test.go b/adapter/redis_lua_compat_test.go index a77bc631f..169bc6650 100644 --- a/adapter/redis_lua_compat_test.go +++ b/adapter/redis_lua_compat_test.go @@ -2,6 +2,7 @@ package adapter import ( "context" + "strconv" "testing" "time" @@ -97,6 +98,74 @@ return {eventId, redis.call("HGET", KEYS[1], "name"), cjson.encode(event), redis require.Equal(t, map[string]any{"event": "waiting", "jobId": "job-1"}, events[0].Values) } +func TestRedis_LuaXAddMaxLenExistingStream(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const stream = "bull:test:events-existing" + for i := range 128 { + _, err := rdb.XAdd(ctx, &redis.XAddArgs{ + Stream: stream, + ID: "*", + Values: []string{"i", strconv.Itoa(i)}, + }).Result() + require.NoError(t, err) + } + + id, err := rdb.Eval(ctx, ` +return redis.call("XADD", KEYS[1], "MAXLEN", "~", 10, "*", "event", "waiting", "jobId", "job-1") +`, []string{stream}).Text() + require.NoError(t, err) + require.NotEmpty(t, id) + + xlen, err := rdb.XLen(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, int64(10), xlen) + + events, err := rdb.XRange(ctx, stream, "-", "+").Result() + require.NoError(t, err) + require.Len(t, events, 10) + require.Equal(t, id, events[len(events)-1].ID) + require.Equal(t, map[string]any{"event": "waiting", "jobId": "job-1"}, events[len(events)-1].Values) +} + +func TestRedis_LuaXAddMaxLenZero(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const stream = "bull:test:events-maxlen0" + for i := range 3 { + _, err := rdb.XAdd(ctx, &redis.XAddArgs{ + Stream: stream, + ID: "*", + Values: []string{"i", strconv.Itoa(i)}, + }).Result() + require.NoError(t, err) + } + + id, err := rdb.Eval(ctx, ` +return redis.call("XADD", KEYS[1], "MAXLEN", "0", "*", "event", "trimmed") +`, []string{stream}).Text() + require.NoError(t, err) + require.NotEmpty(t, id) + + xlen, err := rdb.XLen(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, int64(0), xlen) + + events, err := rdb.XRange(ctx, stream, "-", "+").Result() + require.NoError(t, err) + require.Empty(t, events) +} + func TestRedis_LuaReplyHelpers(t *testing.T) { nodes, _, _ := createNode(t, 3) defer shutdown(nodes) diff --git a/adapter/redis_lua_context.go b/adapter/redis_lua_context.go index afb9340de..b145d6e25 100644 --- a/adapter/redis_lua_context.go +++ b/adapter/redis_lua_context.go @@ -127,6 +127,22 @@ type luaStreamState struct { exists bool dirty bool value redisStreamValue + delta *luaStreamDeltaState +} + +type luaStreamDeltaState struct { + legacyCleanup []*kv.Elem[kv.OP] + meta store.StreamMeta + metaFound bool + appends []luaStreamDeltaAppend + trimCount int + selfDeletes []redisStreamID + forceEmpty bool +} + +type luaStreamDeltaAppend struct { + id redisStreamID + entry redisStreamEntry } type luaTTLState struct { @@ -399,6 +415,7 @@ func (c *luaScriptContext) deleteLogical(key []byte) { st.exists = false st.dirty = true st.value = redisStreamValue{} + st.delta = nil } } @@ -459,7 +476,7 @@ func hasLoadedZSetValue(st *luaZSetState) bool { } func hasLoadedStreamValue(st *luaStreamState) bool { - return st != nil && st.loaded && st.exists + return st != nil && st.exists && (st.loaded || st.delta != nil) } // maxNegativeTypeCacheEntries caps the size of luaScriptContext.negativeType @@ -943,7 +960,10 @@ func (c *luaScriptContext) zsetStateForRead(key []byte) (*luaZSetState, bool, er func (c *luaScriptContext) streamState(key []byte) (*luaStreamState, error) { k := string(key) if st, ok := c.streams[k]; ok { - return st, nil + if st.loaded { + return st, nil + } + return c.materializeStreamState(key, st) } st := &luaStreamState{} c.streams[k] = st @@ -974,6 +994,49 @@ func (c *luaScriptContext) streamState(key []byte) (*luaStreamState, error) { return st, nil } +func (c *luaScriptContext) materializeStreamState(key []byte, st *luaStreamState) (*luaStreamState, error) { + value, err := c.server.loadStreamAt(context.Background(), key, c.startTS) + if err != nil { + return nil, err + } + if st.delta != nil { + value = applyLuaStreamDelta(value, st.delta) + } + st.loaded = true + st.exists = st.exists || len(value.Entries) > 0 || (st.delta != nil && st.delta.metaFound) + st.value = cloneStreamValue(value) + st.delta = nil + return st, nil +} + +func applyLuaStreamDelta(value redisStreamValue, delta *luaStreamDeltaState) redisStreamValue { + if delta == nil { + return value + } + entries := append([]redisStreamEntry(nil), value.Entries...) + if delta.trimCount > 0 { + if delta.trimCount >= len(entries) { + entries = entries[:0] + } else { + entries = entries[delta.trimCount:] + } + } + selfDeleted := make(map[redisStreamID]struct{}, len(delta.selfDeletes)) + for _, id := range delta.selfDeletes { + selfDeleted[id] = struct{}{} + } + for _, appendOp := range delta.appends { + if _, deleted := selfDeleted[appendOp.id]; deleted { + continue + } + entries = append(entries, appendOp.entry) + } + if delta.forceEmpty { + entries = entries[:0] + } + return redisStreamValue{Entries: entries} +} + func (c *luaScriptContext) markStringValue(key []byte, value []byte) { st := c.ttlState(key) _ = st @@ -1067,6 +1130,7 @@ func (c *luaScriptContext) markStreamValue(key []byte, value redisStreamValue) e st.exists = true st.dirty = true st.value = cloneStreamValue(value) + st.delta = nil c.markTouched(key) c.deleted[string(key)] = false return nil @@ -3206,6 +3270,17 @@ func (c *luaScriptContext) cmdXAdd(args []string) (luaReply, error) { if err != nil { return luaReply{}, err } + deltaID, handled, err := c.cmdXAddDelta(parsed) + if err != nil { + return luaReply{}, err + } + if handled { + return luaStringReply(deltaID), nil + } + return c.cmdXAddMaterialized(parsed) +} + +func (c *luaScriptContext) cmdXAddMaterialized(parsed luaXAddArgs) (luaReply, error) { st, err := c.streamState(parsed.key) if err != nil { return luaReply{}, err @@ -3231,6 +3306,96 @@ func (c *luaScriptContext) cmdXAdd(args []string) (luaReply, error) { return luaStringReply(id), nil } +func (c *luaScriptContext) cmdXAddDelta(parsed luaXAddArgs) (string, bool, error) { + st, handled, err := c.streamDeltaStateForXAdd(parsed.key) + if err != nil || !handled { + return "", handled, err + } + + id, parsedID, err := resolveXAddID(st.delta.meta, st.delta.metaFound, parsed.id) + if err != nil { + return "", true, err + } + if err := xaddEnforceMaxWideColumn(parsed.key, st.delta.meta.Length, parsed.maxLen); err != nil { + return "", true, err + } + st.delta.appends = append(st.delta.appends, luaStreamDeltaAppend{ + id: parsedID, + entry: newRedisStreamEntry(id, append([]string(nil), parsed.fields...)), + }) + c.applyLuaXAddDeltaTrim(st.delta, parsed.maxLen, parsedID) + st.delta.meta.LastMs = parsedID.ms + st.delta.meta.LastSeq = parsedID.seq + st.delta.metaFound = true + st.exists = true + st.dirty = true + c.markTouched(parsed.key) + c.deleted[string(parsed.key)] = false + return id, true, nil +} + +func (c *luaScriptContext) streamDeltaStateForXAdd(key []byte) (*luaStreamState, bool, error) { + keyString := string(key) + st := c.streams[keyString] + if st != nil && st.loaded { + return nil, false, nil + } + if st == nil { + st = &luaStreamState{} + c.streams[keyString] = st + } + if st.delta != nil { + return st, true, nil + } + delta, err := c.newLuaStreamDelta(key) + if err != nil { + return nil, true, err + } + st.delta = delta + return st, true, nil +} + +func (c *luaScriptContext) newLuaStreamDelta(key []byte) (*luaStreamDeltaState, error) { + typ, err := c.server.keyTypeAtExpect(c.scriptCtx(), key, c.startTS, redisTypeStream) + if err != nil { + return nil, err + } + if typ != redisTypeNone && typ != redisTypeStream { + return nil, wrongTypeError() + } + legacyCleanup, meta, metaFound, err := c.server.streamWriteBase(c.scriptCtx(), key, c.startTS) + if err != nil { + return nil, err + } + return &luaStreamDeltaState{ + legacyCleanup: legacyCleanup, + meta: meta, + metaFound: metaFound, + }, nil +} + +func (c *luaScriptContext) applyLuaXAddDeltaTrim(delta *luaStreamDeltaState, maxLen int, appendedID redisStreamID) { + candidateLen := delta.meta.Length + 1 + if maxLen < 0 || candidateLen <= int64(maxLen) { + delta.meta.Length = candidateLen + return + } + diff := candidateLen - int64(maxLen) + if diff > int64(maxWideColumnItems) { + diff = int64(maxWideColumnItems) + } + if diff > 0 { + delta.trimCount += int(diff) + } + if maxLen == 0 { + delta.meta.Length = 0 + delta.selfDeletes = append(delta.selfDeletes, appendedID) + delta.forceEmpty = true + return + } + delta.meta.Length = candidateLen - diff +} + func parseLuaXAddArgs(args []string) (luaXAddArgs, error) { parsed := luaXAddArgs{ key: []byte(args[0]), @@ -3429,7 +3594,7 @@ func (c *luaScriptContext) commitPlanForKey(ctx context.Context, key string, com return luaKeyPlan{}, err } - valuePlan, err := c.valueCommitPlan(key, finalType, commitTS) + valuePlan, err := c.valueCommitPlan(ctx, key, finalType, commitTS) if err != nil { return luaKeyPlan{}, err } @@ -3478,7 +3643,7 @@ func luaWideFenceReadKeysForPlan(key []byte, finalType, startType redisValueType return nil } -func (c *luaScriptContext) valueCommitPlan(key string, finalType redisValueType, commitTS uint64) (luaCommitPlan, error) { +func (c *luaScriptContext) valueCommitPlan(ctx context.Context, key string, finalType redisValueType, commitTS uint64) (luaCommitPlan, error) { switch finalType { case redisTypeNone: return luaCommitPlan{}, nil @@ -3496,8 +3661,7 @@ func (c *luaScriptContext) valueCommitPlan(key string, finalType redisValueType, case redisTypeZSet: return c.zsetCommitPlan(key, commitTS) case redisTypeStream: - elems, err := c.streamCommitElems(key) - return luaCommitPlan{elems: elems}, err + return c.streamCommitPlan(ctx, key) default: return luaCommitPlan{}, errors.New("ERR unsupported final redis type") } @@ -3843,6 +4007,71 @@ func (c *luaScriptContext) zsetDeltaMetaElems(key []byte, st *luaZSetState, comm }} } +func (c *luaScriptContext) streamCommitPlan(ctx context.Context, key string) (luaCommitPlan, error) { + st := c.streams[key] + if st == nil || !st.dirty { + return luaCommitPlan{preserveExisting: true}, nil + } + if st.delta != nil && !st.loaded { + elems, err := c.streamDeltaCommitElems(ctx, key, st.delta) + return luaCommitPlan{preserveExisting: true, elems: elems}, err + } + elems, err := c.streamCommitElems(key) + return luaCommitPlan{elems: elems}, err +} + +func (c *luaScriptContext) streamDeltaCommitElems(ctx context.Context, key string, delta *luaStreamDeltaState) ([]*kv.Elem[kv.OP], error) { + keyBytes := []byte(key) + trim, err := c.server.buildXTrimHeadElems(ctx, keyBytes, c.startTS, delta.trimCount) + if err != nil { + return nil, err + } + + elems := make([]*kv.Elem[kv.OP], 0, len(delta.legacyCleanup)+len(delta.appends)+len(trim)+len(delta.selfDeletes)+1) + elems = append(elems, delta.legacyCleanup...) + for _, appendOp := range delta.appends { + entryValue, err := marshalStreamEntry(appendOp.entry) + if err != nil { + return nil, err + } + elems = append(elems, &kv.Elem[kv.OP]{ + Op: kv.Put, + Key: store.StreamEntryKey(keyBytes, appendOp.id.ms, appendOp.id.seq), + Value: entryValue, + }) + } + elems = append(elems, trim...) + for _, id := range delta.selfDeletes { + elems = append(elems, &kv.Elem[kv.OP]{ + Op: kv.Del, + Key: store.StreamEntryKey(keyBytes, id.ms, id.seq), + }) + } + + meta := delta.meta + if delta.forceEmpty { + meta.Length = 0 + } else { + meta.Length = int64(len(delta.appends)) + deltaBaseLength(delta) - int64(len(trim)) - int64(len(delta.selfDeletes)) + if meta.Length < 0 { + meta.Length = 0 + } + } + metaBytes, err := store.MarshalStreamMeta(meta) + if err != nil { + return nil, errors.WithStack(err) + } + elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: store.StreamMetaKey(keyBytes), Value: metaBytes}) + return elems, nil +} + +func deltaBaseLength(delta *luaStreamDeltaState) int64 { + if delta == nil { + return 0 + } + return delta.meta.Length + int64(delta.trimCount) - int64(len(delta.appends)) +} + // streamCommitElems writes the script's final stream state in the // wide-column layout — one StreamEntryKey Put per entry plus a StreamMetaKey // Put for the aggregate meta. The legacy single-blob path is no longer From d19b2a472394450c37bc2d01d3a2898ef27761d4 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 16 Jul 2026 19:55:19 +0900 Subject: [PATCH 7/9] Handle Redis proxy replay edge cases --- adapter/redis_lua_compat_test.go | 74 +++++++++++++++++ adapter/redis_lua_context.go | 41 +++++++--- proxy/blocking.go | 17 ++++ proxy/dualwrite.go | 44 ++++++++++- proxy/leader_aware_backend.go | 22 +++++- proxy/leader_aware_backend_test.go | 123 +++++++++++++++++++++++------ proxy/proxy_test.go | 88 +++++++++++++++++++++ 7 files changed, 373 insertions(+), 36 deletions(-) diff --git a/adapter/redis_lua_compat_test.go b/adapter/redis_lua_compat_test.go index 169bc6650..f9d27c222 100644 --- a/adapter/redis_lua_compat_test.go +++ b/adapter/redis_lua_compat_test.go @@ -166,6 +166,80 @@ return redis.call("XADD", KEYS[1], "MAXLEN", "0", "*", "event", "trimmed") require.Empty(t, events) } +func TestRedis_LuaXAddMultipleMaxLenTrimsInScript(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const stream = "bull:test:events-multi-xadd" + result, err := rdb.Eval(ctx, ` +local first = redis.call("XADD", KEYS[1], "MAXLEN", "1", "*", "event", "first") +local second = redis.call("XADD", KEYS[1], "MAXLEN", "1", "*", "event", "second") +return {first, second} +`, []string{stream}).Result() + require.NoError(t, err) + ids, ok := result.([]any) + require.True(t, ok) + require.Len(t, ids, 2) + + xlen, err := rdb.XLen(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, int64(1), xlen) + + events, err := rdb.XRange(ctx, stream, "-", "+").Result() + require.NoError(t, err) + require.Len(t, events, 1) + require.Equal(t, ids[1], events[0].ID) + require.Equal(t, map[string]any{"event": "second"}, events[0].Values) +} + +func TestRedis_LuaXAddRecreatesTTLExpiredStream(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const ( + stream = "bull:test:events-expired" + ttl = 80 * time.Millisecond + ) + _, err := rdb.XAdd(ctx, &redis.XAddArgs{ + Stream: stream, + ID: "*", + Values: []string{"event", "old"}, + }).Result() + require.NoError(t, err) + require.NoError(t, rdb.PExpire(ctx, stream, ttl).Err()) + eventuallyExpired(t, ttl, func() bool { + xlen, err := rdb.XLen(ctx, stream).Result() + return err == nil && xlen == 0 + }, "stream must be logically expired before Lua XADD recreates it") + + id, err := rdb.Eval(ctx, ` +return redis.call("XADD", KEYS[1], "MAXLEN", "~", 10, "*", "event", "new") +`, []string{stream}).Text() + require.NoError(t, err) + require.NotEmpty(t, id) + + xlen, err := rdb.XLen(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, int64(1), xlen) + ttlAfter, err := rdb.TTL(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, time.Duration(-1), ttlAfter) + + events, err := rdb.XRange(ctx, stream, "-", "+").Result() + require.NoError(t, err) + require.Len(t, events, 1) + require.Equal(t, id, events[0].ID) + require.Equal(t, map[string]any{"event": "new"}, events[0].Values) +} + func TestRedis_LuaReplyHelpers(t *testing.T) { nodes, _, _ := createNode(t, 3) defer shutdown(nodes) diff --git a/adapter/redis_lua_context.go b/adapter/redis_lua_context.go index b145d6e25..358718c94 100644 --- a/adapter/redis_lua_context.go +++ b/adapter/redis_lua_context.go @@ -3340,38 +3340,61 @@ func (c *luaScriptContext) streamDeltaStateForXAdd(key []byte) (*luaStreamState, if st != nil && st.loaded { return nil, false, nil } + if st != nil && st.delta != nil { + return nil, false, nil + } if st == nil { st = &luaStreamState{} c.streams[keyString] = st } - if st.delta != nil { - return st, true, nil - } - delta, err := c.newLuaStreamDelta(key) + delta, useDelta, err := c.newLuaStreamDelta(key) if err != nil { return nil, true, err } + if !useDelta { + delete(c.streams, keyString) + return nil, false, nil + } st.delta = delta return st, true, nil } -func (c *luaScriptContext) newLuaStreamDelta(key []byte) (*luaStreamDeltaState, error) { +func (c *luaScriptContext) newLuaStreamDelta(key []byte) (*luaStreamDeltaState, bool, error) { typ, err := c.server.keyTypeAtExpect(c.scriptCtx(), key, c.startTS, redisTypeStream) if err != nil { - return nil, err + return nil, false, err } if typ != redisTypeNone && typ != redisTypeStream { - return nil, wrongTypeError() + return nil, false, wrongTypeError() } legacyCleanup, meta, metaFound, err := c.server.streamWriteBase(c.scriptCtx(), key, c.startTS) if err != nil { - return nil, err + return nil, false, err + } + useDelta, err := c.canUseLuaStreamDelta(key, typ, metaFound, legacyCleanup) + if err != nil || !useDelta { + return nil, false, err } return &luaStreamDeltaState{ legacyCleanup: legacyCleanup, meta: meta, metaFound: metaFound, - }, nil + }, true, nil +} + +func (c *luaScriptContext) canUseLuaStreamDelta(key []byte, typ redisValueType, metaFound bool, legacyCleanup []*kv.Elem[kv.OP]) (bool, error) { + hasPhysicalStream := metaFound || len(legacyCleanup) != 0 + if !hasPhysicalStream { + return true, nil + } + expired, err := c.server.hasExpired(c.scriptCtx(), key, c.startTS, true) + if err != nil { + return false, err + } + if expired || typ == redisTypeNone { + return false, nil + } + return true, nil } func (c *luaScriptContext) applyLuaXAddDeltaTrim(delta *luaStreamDeltaState, maxLen int, appendedID redisStreamID) { diff --git a/proxy/blocking.go b/proxy/blocking.go index 8a1bf64be..772c011f2 100644 --- a/proxy/blocking.go +++ b/proxy/blocking.go @@ -56,6 +56,10 @@ func parseBlockingMillisecondsArg(raw []byte) time.Duration { func blockingReplayCommand(cmd string, _ [][]byte, resp any) (string, []any, bool) { switch strings.ToUpper(cmd) { + case "BLPOP": + return blockingListPopReplay(1, resp) + case "BRPOP": + return blockingListPopReplay(-1, resp) case "BZPOPMIN", "BZPOPMAX": parts, ok := redisArray(resp) if !ok || len(parts) < 2 { @@ -72,6 +76,19 @@ func blockingReplayCommand(cmd string, _ [][]byte, resp any) (string, []any, boo } } +func blockingListPopReplay(count int64, resp any) (string, []any, bool) { + parts, ok := redisArray(resp) + if !ok || len(parts) < 2 { + return "", nil, false + } + key, keyOK := redisArg(parts[0]) + value, valueOK := redisArg(parts[1]) + if !keyOK || !valueOK { + return "", nil, false + } + return "LREM", []any{[]byte("LREM"), key, count, value}, true +} + func redisArray(v any) ([]any, bool) { switch x := v.(type) { case []any: diff --git a/proxy/dualwrite.go b/proxy/dualwrite.go index c1e284d48..0963c9ebb 100644 --- a/proxy/dualwrite.go +++ b/proxy/dualwrite.go @@ -76,8 +76,11 @@ func isRetryableSecondaryWriteError(err error) bool { if isReadTSCompactedError(err) { return true } + if isElasticKVNotLeaderError(err) { + return true + } switch classifySecondaryWriteError(err) { - case "retry_limit", "write_conflict", "txn_locked", "not_leader": + case "retry_limit", "write_conflict", "txn_locked": return true default: return false @@ -85,7 +88,23 @@ func isRetryableSecondaryWriteError(err error) bool { } func isNotLeaderError(err error) bool { - return classifySecondaryWriteError(err) == "not_leader" + return isElasticKVNotLeaderError(err) +} + +func isElasticKVNotLeaderError(err error) bool { + if err == nil { + return false + } + msg := strings.TrimSpace(err.Error()) + if strings.HasPrefix(strings.ToUpper(msg), "NOTLEADER ") { + return true + } + if strings.Contains(msg, "etcd raft engine is not leader") || + strings.Contains(msg, "raft engine: not leader") { + return true + } + return msg == "leader not found" || + strings.HasSuffix(msg, "desc = leader not found") } func refreshSecondaryLeader(ctx context.Context, backend Backend, err error) { @@ -401,15 +420,34 @@ func (d *DualWriter) replaySecondaryPipeline(cmds [][]any) { d.runSecondaryWrite(func() { sCtx, cancel := context.WithTimeout(context.Background(), d.cfg.SecondaryTimeout) defer cancel() - _, pErr := d.secondary.Pipeline(sCtx, cmds) + results, pErr := d.secondary.Pipeline(sCtx, cmds) if pErr != nil { d.logger.Warn("secondary txn replay failed", "err", pErr) d.metrics.SecondaryWriteErrors.Inc() d.metrics.SecondaryWriteErrorsByReason.WithLabelValues("PIPELINE", classifySecondaryWriteError(pErr)).Inc() + return + } + if rErr := firstPipelineResultError(results); rErr != nil { + d.logger.Warn("secondary txn replay failed", "err", rErr) + d.metrics.SecondaryWriteErrors.Inc() + d.metrics.SecondaryWriteErrorsByReason.WithLabelValues("PIPELINE", classifySecondaryWriteError(rErr)).Inc() } }) } +func firstPipelineResultError(results []*redis.Cmd) error { + for _, result := range results { + if result == nil { + continue + } + err := result.Err() + if err != nil && !errors.Is(err, redis.Nil) { + return fmt.Errorf("pipeline result: %w", err) + } + } + return nil +} + // waitCompactedRetryBackoff sleeps for a jittered interval or returns early // when the context is cancelled. Returns false if the caller should abort // the retry loop (context done). diff --git a/proxy/leader_aware_backend.go b/proxy/leader_aware_backend.go index 3932d4a9d..a9e372699 100644 --- a/proxy/leader_aware_backend.go +++ b/proxy/leader_aware_backend.go @@ -403,8 +403,19 @@ func (b *LeaderAwareRedisBackend) doWithTimeoutOnce(ctx context.Context, timeout return cli.WithTimeout(effectiveBlockingReadTimeout(timeout)).Do(ctx, args...) } -// Pipeline forwards a batch to the current leader. +// Pipeline forwards a batch to the current leader. A not-leader Redis reply in +// any result means the leader rejected the batch before applying it, so retrying +// once after synchronous discovery is safe. func (b *LeaderAwareRedisBackend) Pipeline(ctx context.Context, cmds [][]any) ([]*redis.Cmd, error) { + results, err := b.pipelineOnce(ctx, cmds) + if err != nil || !pipelineHasElasticKVNotLeader(results) { + return results, err + } + b.RefreshLeaderNow(ctx) + return b.pipelineOnce(ctx, cmds) +} + +func (b *LeaderAwareRedisBackend) pipelineOnce(ctx context.Context, cmds [][]any) ([]*redis.Cmd, error) { cli := b.currentClient() if cli == nil { return nil, ErrNoLeaderBackend @@ -425,6 +436,15 @@ func (b *LeaderAwareRedisBackend) Pipeline(ctx context.Context, cmds [][]any) ([ return results, nil } +func pipelineHasElasticKVNotLeader(results []*redis.Cmd) bool { + for _, result := range results { + if result != nil && isElasticKVNotLeaderError(result.Err()) { + return true + } + } + return false +} + // NewPubSub opens a subscribe connection on the current leader. func (b *LeaderAwareRedisBackend) NewPubSub(ctx context.Context) *redis.PubSub { cli := b.currentClient() diff --git a/proxy/leader_aware_backend_test.go b/proxy/leader_aware_backend_test.go index 034822d59..ce2f32a6b 100644 --- a/proxy/leader_aware_backend_test.go +++ b/proxy/leader_aware_backend_test.go @@ -62,6 +62,7 @@ type fakeElasticKVNode struct { addr string ln net.Listener leaderAddr atomic.Pointer[string] + commandErr atomic.Pointer[string] commands atomic.Int64 infoCalls atomic.Int64 } @@ -89,6 +90,10 @@ func (n *fakeElasticKVNode) Leader() string { return "" } +func (n *fakeElasticKVNode) SetCommandError(err string) { + n.commandErr.Store(&err) +} + func (n *fakeElasticKVNode) serve() { for { conn, err := n.ln.Accept() @@ -110,32 +115,43 @@ func (n *fakeElasticKVNode) handleConn(conn net.Conn) { if len(args) == 0 { return } - cmd := strings.ToUpper(args[0]) - switch cmd { - case "HELLO": - // Force go-redis to fall back to RESP2 without HELLO: return an - // error the client treats as "HELLO not supported". - _, _ = conn.Write([]byte("-ERR unknown command 'HELLO'\r\n")) - case "CLIENT", "AUTH", "SELECT", "PING": - _, _ = conn.Write([]byte("+OK\r\n")) - case "INFO": - n.infoCalls.Add(1) - body := fmt.Sprintf( - "# Replication\r\nrole:slave\r\nraft_leader_redis:%s\r\n", - n.Leader(), - ) - _, _ = fmt.Fprintf(conn, "$%d\r\n%s\r\n", len(body), body) - default: - n.commands.Add(1) - if leader := n.Leader(); leader != "" && leader != n.addr { - _, _ = conn.Write([]byte("-ERR etcd raft engine is not leader\r\n")) - continue - } - _, _ = conn.Write([]byte("+OK\r\n")) - } + n.handleCommand(conn, args) + } +} + +func (n *fakeElasticKVNode) handleCommand(conn net.Conn, args []string) { + switch strings.ToUpper(args[0]) { + case "HELLO": + // Force go-redis to fall back to RESP2 without HELLO: return an + // error the client treats as "HELLO not supported". + _, _ = conn.Write([]byte("-ERR unknown command 'HELLO'\r\n")) + case "CLIENT", "AUTH", "SELECT", "PING": + _, _ = conn.Write([]byte("+OK\r\n")) + case "INFO": + n.infoCalls.Add(1) + body := fmt.Sprintf( + "# Replication\r\nrole:slave\r\nraft_leader_redis:%s\r\n", + n.Leader(), + ) + _, _ = fmt.Fprintf(conn, "$%d\r\n%s\r\n", len(body), body) + default: + n.writeCommandReply(conn) } } +func (n *fakeElasticKVNode) writeCommandReply(conn net.Conn) { + n.commands.Add(1) + if p := n.commandErr.Load(); p != nil { + _, _ = fmt.Fprintf(conn, "-%s\r\n", *p) + return + } + if leader := n.Leader(); leader != "" && leader != n.addr { + _, _ = conn.Write([]byte("-ERR etcd raft engine is not leader\r\n")) + return + } + _, _ = conn.Write([]byte("+OK\r\n")) +} + // readRESPArray reads a single RESP array of bulk strings from rd. func readRESPArray(rd *bufio.Reader) ([]string, error) { header, err := rd.ReadString('\n') @@ -248,6 +264,67 @@ func TestLeaderAwareRedisBackend_RetryRefreshesOnNotLeader(t *testing.T) { require.Equal(t, int64(1), nodeB.commands.Load(), "retry should land on the refreshed leader") } +func TestLeaderAwareRedisBackend_DoesNotRetryUserNotLeaderError(t *testing.T) { + node := newFakeElasticKVNode(t) + node.SetLeader(node.addr) + node.SetCommandError("ERR not leader") + + backend := NewLeaderAwareRedisBackendWithInterval( + []string{node.addr}, + "elastickv", + DefaultBackendOptions(), + time.Hour, 500*time.Millisecond, + testLogger, + ) + t.Cleanup(func() { _ = backend.Close() }) + + require.Eventually(t, func() bool { + return backend.CurrentLeader() == node.addr + }, 2*time.Second, 10*time.Millisecond, "initial leader must be set") + + res := backend.Do(context.Background(), "EVAL", "return redis.error_reply('not leader')", 0) + require.Error(t, res.Err()) + require.Contains(t, res.Err().Error(), "not leader") + require.Equal(t, int64(1), node.commands.Load(), "user Redis error must not be retried") +} + +func TestLeaderAwareRedisBackend_PipelineRetryRefreshesOnNotLeader(t *testing.T) { + nodeA := newFakeElasticKVNode(t) + nodeB := newFakeElasticKVNode(t) + + nodeA.SetLeader(nodeA.addr) + nodeB.SetLeader(nodeA.addr) + + backend := NewLeaderAwareRedisBackendWithInterval( + []string{nodeA.addr, nodeB.addr}, + "elastickv", + DefaultBackendOptions(), + time.Hour, 500*time.Millisecond, + testLogger, + ) + t.Cleanup(func() { _ = backend.Close() }) + + require.Eventually(t, func() bool { + return backend.CurrentLeader() == nodeA.addr + }, 2*time.Second, 10*time.Millisecond, "initial leader must be A") + + nodeA.SetLeader(nodeB.addr) + nodeB.SetLeader(nodeB.addr) + + results, err := backend.Pipeline(context.Background(), [][]any{ + {"MULTI"}, + {"SET", "k", "v"}, + {"EXEC"}, + }) + require.NoError(t, err) + for _, result := range results { + require.NoError(t, result.Err()) + } + require.Equal(t, nodeB.addr, backend.CurrentLeader(), "pipeline retry must synchronously adopt the advertised leader") + require.Equal(t, int64(3), nodeA.commands.Load(), "first pipeline attempt should hit the former leader") + require.Equal(t, int64(3), nodeB.commands.Load(), "retry should replay the whole pipeline on the refreshed leader") +} + func TestLeaderAwareRedisBackend_ConcurrentCloseIsRaceFree(t *testing.T) { // Regression guard: Close() must not race concurrent Do() callers — // currentClient and ensureClientLocked hold the lock consistently with diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index dcfa37ffc..f092a492d 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -672,6 +672,51 @@ func TestDualWriter_Blocking_ReplaysBZPopMinAsZRem(t *testing.T) { assert.InDelta(t, 1, testutil.ToFloat64(metrics.CommandTotal.WithLabelValues("ZREM", "secondary", "ok")), 0.001) } +func TestDualWriter_Blocking_ReplaysListPopAsLRem(t *testing.T) { + tests := []struct { + name string + cmd string + resp []any + want [][]any + }{ + { + name: "BLPOP removes first matching element", + cmd: "BLPOP", + resp: []any{[]byte("queue"), []byte("job-1")}, + want: [][]any{{[]byte("LREM"), []byte("queue"), int64(1), []byte("job-1")}}, + }, + { + name: "BRPOP removes last matching element", + cmd: "BRPOP", + resp: []any{[]byte("queue"), []byte("job-2")}, + want: [][]any{{[]byte("LREM"), []byte("queue"), int64(-1), []byte("job-2")}}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + primary := &timeoutCapturingBackend{ + name: "primary", + returnValue: tc.resp, + } + secondary := newMockBackend("secondary") + + metrics := newTestMetrics() + cfg := ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: 10 * time.Second} + d := NewDualWriter(primary, secondary, cfg, metrics, newTestSentry(), testLogger) + + resp, err := d.Blocking(context.Background(), tc.cmd, [][]byte{[]byte(tc.cmd), []byte("queue"), []byte("5")}) + assert.NoError(t, err) + assert.Equal(t, tc.resp, resp) + d.Close() + + assert.Equal(t, tc.want, secondary.Calls()) + assert.InDelta(t, 0, testutil.ToFloat64(metrics.AsyncDrops), 0.001) + assert.InDelta(t, 1, testutil.ToFloat64(metrics.CommandTotal.WithLabelValues("LREM", "secondary", "ok")), 0.001) + }) + } +} + func TestDualWriter_Blocking_XReadDoesNotUseWriteSemaphore(t *testing.T) { primary := &timeoutCapturingBackend{name: "primary", returnValue: []any{}} secondary := newMockBackend("secondary") @@ -1232,6 +1277,49 @@ func TestDualWriter_writeSecondary_RetriesNotLeaderAfterRefresh(t *testing.T) { "a refreshed retry success must not count as a secondary write error") } +func TestDualWriter_writeSecondary_DoesNotRetryUserNotLeaderError(t *testing.T) { + primary := newMockBackend("primary") + primary.doFunc = makeCmd("OK", nil) + + secondary := &refreshableMockBackend{mockBackend: newMockBackend("secondary")} + userErr := testRedisErr("ERR not leader") + var calls int + secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd { + calls++ + cmd := redis.NewCmd(ctx, args...) + cmd.SetErr(userErr) + return cmd + } + + metrics := newTestMetrics() + d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger) + + d.writeSecondary("EVALSHA", []any{[]byte("EVALSHA"), []byte("deadbeef"), []byte("0")}) + + assert.Equal(t, 1, calls, "user Redis not-leader text must not be retried") + assert.Equal(t, 0, secondary.RefreshCount(), "user Redis not-leader text must not force rediscovery") + assert.InDelta(t, 1, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001) +} + +func TestDualWriter_ReplaySecondaryPipeline_RecordsReplyErrors(t *testing.T) { + primary := newMockBackend("primary") + secondary := newMockBackend("secondary") + secondary.doFunc = func(ctx context.Context, args ...any) *redis.Cmd { + cmd := redis.NewCmd(ctx, args...) + cmd.SetErr(testRedisErr("ERR etcd raft engine is not leader")) + return cmd + } + + metrics := newTestMetrics() + d := NewDualWriter(primary, secondary, ProxyConfig{Mode: ModeDualWrite, SecondaryTimeout: time.Second}, metrics, newTestSentry(), testLogger) + + d.replaySecondaryPipeline([][]any{{[]byte("MULTI")}, {[]byte("SET"), []byte("k"), []byte("v")}, {[]byte("EXEC")}}) + d.Close() + + assert.InDelta(t, 1, testutil.ToFloat64(metrics.SecondaryWriteErrors), 0.001) + assert.InDelta(t, 1, testutil.ToFloat64(metrics.SecondaryWriteErrorsByReason.WithLabelValues("PIPELINE", "not_leader")), 0.001) +} + func TestDualWriter_writeSecondary_RetriesDoNotRepeatNoScriptProbe(t *testing.T) { // After the EVAL fallback resolves a NOSCRIPT, a compacted-retry must // re-send the resolved EVAL form directly — never the known-missing From b8bb6ad605532bd8872583e9b1372e8e628fc7c3 Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 17 Jul 2026 02:18:31 +0900 Subject: [PATCH 8/9] Honor Lua XADD cached state --- adapter/redis_lua_compat_test.go | 18 ++++++++ adapter/redis_lua_context.go | 50 ++++++++++++++++---- adapter/redis_lua_list_holes_test.go | 68 ++++++++++++++++++++++++++++ 3 files changed, 126 insertions(+), 10 deletions(-) diff --git a/adapter/redis_lua_compat_test.go b/adapter/redis_lua_compat_test.go index f9d27c222..8f56dd31b 100644 --- a/adapter/redis_lua_compat_test.go +++ b/adapter/redis_lua_compat_test.go @@ -3,6 +3,7 @@ package adapter import ( "context" "strconv" + "strings" "testing" "time" @@ -240,6 +241,23 @@ return redis.call("XADD", KEYS[1], "MAXLEN", "~", 10, "*", "event", "new") require.Equal(t, map[string]any{"event": "new"}, events[0].Values) } +func TestRedis_LuaXAddHonorsScriptLocalStringType(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const key = "bull:test:events-local-string" + _, err := rdb.Eval(ctx, ` +redis.call("SET", KEYS[1], "string-value") +return redis.call("XADD", KEYS[1], "*", "event", "bad") +`, []string{key}).Result() + require.Error(t, err) + require.Contains(t, strings.ToUpper(err.Error()), "WRONGTYPE") +} + func TestRedis_LuaReplyHelpers(t *testing.T) { nodes, _, _ := createNode(t, 3) defer shutdown(nodes) diff --git a/adapter/redis_lua_context.go b/adapter/redis_lua_context.go index 358718c94..9eda2ad57 100644 --- a/adapter/redis_lua_context.go +++ b/adapter/redis_lua_context.go @@ -202,6 +202,10 @@ type luaExactScanFallbackDecider interface { AllowExactScanFallbackAfterPhysicalLimit(ctx context.Context, start []byte, end []byte, visibleLimit, physicalLimit int, ts uint64, reverse bool) bool } +type luaRoutePinnedScanStore interface { + ScanAtWithReadFence(ctx context.Context, start []byte, end []byte, limit int, ts uint64, reverse bool, groupID uint64, readRouteVersion uint64, routeStart []byte, routeEnd []byte) ([]*store.KVPair, error) +} + type luaCommandHandler func(*luaScriptContext, []string) (luaReply, error) type luaRenameHandler func(*luaScriptContext, []byte, []byte) error @@ -2283,18 +2287,12 @@ func (c *luaScriptContext) allowExactListScanFallback(startKey, endKey []byte, s } func (c *luaScriptContext) scanListItemWindowExact(key []byte, meta store.ListMeta, startKey, endKey []byte, left bool) (luaLazyListBoundaryItem, bool, error) { + routeStart := append([]byte(nil), startKey...) + routeEnd := append([]byte(nil), endKey...) for { - var ( - kvs []*store.KVPair - err error - ) - if left { - kvs, err = c.server.store.ScanAt(c.ctx, startKey, endKey, 1, c.startTS) - } else { - kvs, err = c.server.store.ReverseScanAt(c.ctx, startKey, endKey, 1, c.startTS) - } + kvs, err := c.scanExactListPage(startKey, endKey, routeStart, routeEnd, left) if err != nil { - return luaLazyListBoundaryItem{}, false, errors.WithStack(err) + return luaLazyListBoundaryItem{}, false, err } if len(kvs) == 0 { return luaLazyListBoundaryItem{}, false, nil @@ -2310,6 +2308,24 @@ func (c *luaScriptContext) scanListItemWindowExact(key []byte, meta store.ListMe } } +func (c *luaScriptContext) scanExactListPage(startKey, endKey, routeStart, routeEnd []byte, left bool) ([]*store.KVPair, error) { + var ( + kvs []*store.KVPair + err error + ) + if scanner, ok := c.server.store.(luaRoutePinnedScanStore); ok { + kvs, err = scanner.ScanAtWithReadFence(c.ctx, startKey, endKey, 1, c.startTS, !left, 0, 0, routeStart, routeEnd) + } else if left { + kvs, err = c.server.store.ScanAt(c.ctx, startKey, endKey, 1, c.startTS) + } else { + kvs, err = c.server.store.ReverseScanAt(c.ctx, startKey, endKey, 1, c.startTS) + } + if err != nil { + return nil, errors.WithStack(err) + } + return kvs, nil +} + func scanStartAfterKey(key []byte) []byte { next := make([]byte, len(key)+1) copy(next, key) @@ -3343,6 +3359,9 @@ func (c *luaScriptContext) streamDeltaStateForXAdd(key []byte) (*luaStreamState, if st != nil && st.delta != nil { return nil, false, nil } + if cached, handled, err := c.streamDeltaDecisionFromCachedType(key); cached { + return nil, handled, err + } if st == nil { st = &luaStreamState{} c.streams[keyString] = st @@ -3359,6 +3378,17 @@ func (c *luaScriptContext) streamDeltaStateForXAdd(key []byte) (*luaStreamState, return st, true, nil } +func (c *luaScriptContext) streamDeltaDecisionFromCachedType(key []byte) (bool, bool, error) { + typ, cached := c.cachedType(key) + if !cached { + return false, false, nil + } + if typ != redisTypeNone && typ != redisTypeStream { + return true, true, wrongTypeError() + } + return true, false, nil +} + func (c *luaScriptContext) newLuaStreamDelta(key []byte) (*luaStreamDeltaState, bool, error) { typ, err := c.server.keyTypeAtExpect(c.scriptCtx(), key, c.startTS, redisTypeStream) if err != nil { diff --git a/adapter/redis_lua_list_holes_test.go b/adapter/redis_lua_list_holes_test.go index 80ac1cd7c..4483540d4 100644 --- a/adapter/redis_lua_list_holes_test.go +++ b/adapter/redis_lua_list_holes_test.go @@ -339,6 +339,41 @@ func TestRedisLua_RPopLPushSyntheticPhysicalLimitFailsClosed(t *testing.T) { require.ErrorIs(t, err, ErrCollectionTooLarge) } +func TestRedisLua_ListExactFallbackPinsRouteBounds(t *testing.T) { + t.Parallel() + + ctx := context.Background() + src := []byte("bull:test:wait:route-pin") + meta := store.ListMeta{Head: 0, Len: 2, Tail: 2} + startKey := listItemKey(src, 0) + endKey := listItemKey(src, 2) + + collider := append([]byte{}, src...) + collider = append(collider, listItemKey(src, 0)[len(store.ListItemPrefix)+len(src):]...) + collider = append(collider, 'x') + st := &routePinnedExactScanStore{ + MVCCStore: store.NewMVCCStore(), + t: t, + wantRouteStart: startKey, + wantRouteEnd: endKey, + pages: [][]*store.KVPair{ + {{Key: listItemKey(collider, 0), Value: []byte("other-list-job")}}, + {{Key: listItemKey(src, 1), Value: []byte("job-1")}}, + }, + } + r := NewRedisServer(nil, "", st, newLocalAdapterCoordinator(st), nil, nil) + scriptCtx, err := newLuaScriptContext(ctx, r) + require.NoError(t, err) + defer scriptCtx.Close() + + item, ok, err := scriptCtx.scanListItemWindowExact(src, meta, startKey, endKey, true) + require.NoError(t, err) + require.True(t, ok) + require.Equal(t, int64(1), item.index) + require.Equal(t, "job-1", item.value) + require.Equal(t, 2, st.calls) +} + func TestRedisLua_RPopLPushDeletesLargeSparseListWithoutItems(t *testing.T) { t.Parallel() @@ -399,3 +434,36 @@ func (s noExactFallbackPhysicalLimitStore) ReverseScanAtPhysicalLimit(context.Co func (s noExactFallbackPhysicalLimitStore) AllowExactScanFallbackAfterPhysicalLimit(context.Context, []byte, []byte, int, int, uint64, bool) bool { return false } + +type routePinnedExactScanStore struct { + store.MVCCStore + t *testing.T + wantRouteStart []byte + wantRouteEnd []byte + pages [][]*store.KVPair + calls int +} + +func (s *routePinnedExactScanStore) ScanAtWithReadFence( + _ context.Context, + _ []byte, + _ []byte, + _ int, + _ uint64, + _ bool, + _ uint64, + _ uint64, + routeStart []byte, + routeEnd []byte, +) ([]*store.KVPair, error) { + s.t.Helper() + require.Equal(s.t, s.wantRouteStart, routeStart) + require.Equal(s.t, s.wantRouteEnd, routeEnd) + if s.calls >= len(s.pages) { + s.calls++ + return nil, nil + } + page := s.pages[s.calls] + s.calls++ + return page, nil +} From adf5cdb508928334d651e951aae5860401e8688b Mon Sep 17 00:00:00 2001 From: bootjp Date: Fri, 17 Jul 2026 02:25:35 +0900 Subject: [PATCH 9/9] Cover Lua XADD maxlen zero append --- adapter/redis_lua_compat_test.go | 30 ++++++++++++++++++++++++++++++ 1 file changed, 30 insertions(+) diff --git a/adapter/redis_lua_compat_test.go b/adapter/redis_lua_compat_test.go index 8f56dd31b..e22234762 100644 --- a/adapter/redis_lua_compat_test.go +++ b/adapter/redis_lua_compat_test.go @@ -167,6 +167,36 @@ return redis.call("XADD", KEYS[1], "MAXLEN", "0", "*", "event", "trimmed") require.Empty(t, events) } +func TestRedis_LuaXAddMaxLenZeroThenAppendInScript(t *testing.T) { + nodes, _, _ := createNode(t, 3) + defer shutdown(nodes) + + ctx := context.Background() + rdb := redis.NewClient(&redis.Options{Addr: nodes[0].redisAddress}) + defer func() { _ = rdb.Close() }() + + const stream = "bull:test:events-maxlen0-then-append" + result, err := rdb.Eval(ctx, ` +local trimmed = redis.call("XADD", KEYS[1], "MAXLEN", "0", "*", "event", "trimmed") +local kept = redis.call("XADD", KEYS[1], "MAXLEN", "1", "*", "event", "kept") +return {trimmed, kept} +`, []string{stream}).Result() + require.NoError(t, err) + ids, ok := result.([]any) + require.True(t, ok) + require.Len(t, ids, 2) + + xlen, err := rdb.XLen(ctx, stream).Result() + require.NoError(t, err) + require.Equal(t, int64(1), xlen) + + events, err := rdb.XRange(ctx, stream, "-", "+").Result() + require.NoError(t, err) + require.Len(t, events, 1) + require.Equal(t, ids[1], events[0].ID) + require.Equal(t, map[string]any{"event": "kept"}, events[0].Values) +} + func TestRedis_LuaXAddMultipleMaxLenTrimsInScript(t *testing.T) { nodes, _, _ := createNode(t, 3) defer shutdown(nodes)