Reduce Redis proxy replay backpressure

README.md

@@ -14,7 +14,7 @@ Elastickv is an experimental project undertaking the challenge of creating a dis
 - **DynamoDB Compatibility Scope**: `CreateTable`/`DeleteTable`/`DescribeTable`/`ListTables`/`PutItem`/`GetItem`/`DeleteItem`/`UpdateItem`/`Query`/`Scan`/`BatchWriteItem`/`TransactWriteItems` are implemented.
 - **S3 Compatibility Scope**: `ListBuckets`, `CreateBucket`, `HeadBucket`, `DeleteBucket`, `PutObject`, `GetObject`, `HeadObject`, `DeleteObject`, and `ListObjectsV2` (path-style) are implemented. AWS Signature Version 4 authentication with static credentials is supported. The server exposes an S3-compatible HTTP endpoint via `--s3Address`.
 - **Basic Consistency Behaviors**: Write-after-read checks, leader redirection/forwarding paths, and OCC conflict detection for transactional writes are covered by tests.
-- **Hybrid Logical Clock (HLC)**: Transactions are ordered by a 64-bit HLC split into an upper 48-bit physical component (Unix milliseconds) and a lower 16-bit logical counter. The logical half advances in memory with atomic CAS on every `Next()` call — no Raft round-trip per timestamp — so timestamp issuance stays in the nanosecond range. The physical half is bounded by a leader-lease style ceiling: the leader periodically commits a lease entry (`hlcRenewalInterval ≈ 1s`, window `hlcPhysicalWindowMs = 3s`) so that a newly elected leader inherits a safe lower bound and never issues timestamps overlapping the previous leader's window, without blocking per-request on consensus. See `docs/architecture_overview.md` §4 for details.
+- **Hybrid Logical Clock (HLC)**: Transactions are ordered by a 64-bit HLC split into an upper 48-bit physical component (Unix milliseconds) and a lower 16-bit logical counter. The logical half advances in memory with atomic CAS on every `Next()` call — no Raft round-trip per timestamp — so timestamp issuance stays in the nanosecond range. The physical half is bounded by a leader-lease style ceiling: the leader periodically commits a lease entry (`hlcRenewalInterval ≈ 2s`, window `hlcPhysicalWindowMs = 30s`). Timestamp safety across leadership changes relies on the new leader observing committed HLC values and on `NextFenced` refusing persistence timestamps after the ceiling expires, without blocking per-request on consensus. See `docs/architecture_overview.md` §4 for details.
 
 ## Planned Features
 - **Dynamic Node Scaling**: Automatic node/range scaling based on load is not yet implemented (current sharding operations are configuration/manual driven).
@@ -202,9 +202,10 @@ docker run --rm \
   -listen :6479 \
   -primary redis.internal:6379 \
   -secondary elastickv.internal:6380 \
-  -elastickv-pool-size 4 \
-  -secondary-write-concurrency 2 \
-  -secondary-script-concurrency 1 \
+  -elastickv-pool-size 64 \
+  -secondary-write-concurrency 32 \
+  -secondary-script-concurrency 16 \
+  -secondary-blocking-replay-concurrency 32 \
   -mode dual-write
 ```
 

adapter/redis_bzpopmin_wake_test.go

@@ -1,15 +1,167 @@
 package adapter
 
 import (
+	"bytes"
 	"context"
 	"runtime"
 	"testing"
 	"time"
 
+	"github.com/bootjp/elastickv/store"
 	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 )
 
+type bzpopminScanRecord struct {
+	start []byte
+	limit int
+}
+
+type bzpopminRecordingStore struct {
+	store.MVCCStore
+	scans []bzpopminScanRecord
+}
+
+func (s *bzpopminRecordingStore) ScanAt(ctx context.Context, start []byte, end []byte, limit int, ts uint64) ([]*store.KVPair, error) {
+	s.scans = append(s.scans, bzpopminScanRecord{start: bytes.Clone(start), limit: limit})
+	return s.MVCCStore.ScanAt(ctx, start, end, limit, ts)
+}
+
+func TestRedis_BZPopMinCandidateSelection(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	readTS := uint64(10)
+
+	for _, tc := range []struct {
+		name       string
+		seed       func(t *testing.T, st store.MVCCStore, key []byte)
+		wantMember string
+		wantScore  float64
+		wantWide   bool
+		wantLast   bool
+		wantScans  []int
+	}{
+		{
+			name: "wide score index",
+			seed: func(t *testing.T, st store.MVCCStore, key []byte) {
+				seedZSetScoreRowsForTest(t, st, key, readTS, []redisZSetEntry{
+					{Member: "later", Score: 2},
+					{Member: "first", Score: 1},
+				})
+			},
+			wantMember: "first",
+			wantScore:  1,
+			wantWide:   true,
+			wantLast:   false,
+			wantScans:  []int{bzpopminScoreScanLimit},
+		},
+		{
+			name: "single score entry",
+			seed: func(t *testing.T, st store.MVCCStore, key []byte) {
+				seedZSetScoreRowsForTest(t, st, key, readTS, []redisZSetEntry{
+					{Member: "only", Score: 3},
+				})
+			},
+			wantMember: "only",
+			wantScore:  3,
+			wantWide:   true,
+			wantLast:   true,
+			wantScans:  []int{bzpopminScoreScanLimit},
+		},
+		{
+			name: "member only fallback",
+			seed: func(t *testing.T, st store.MVCCStore, key []byte) {
+				seedZSetMemberRowsForBZPopMinTest(t, st, key, readTS, []redisZSetEntry{
+					{Member: "later", Score: 2},
+					{Member: "first", Score: 1},
+				})
+			},
+			wantMember: "first",
+			wantScore:  1,
+			wantWide:   true,
+			wantLast:   false,
+			wantScans:  []int{bzpopminScoreScanLimit, 1, maxWideScanLimit},
+		},
+		{
+			name: "legacy blob fallback",
+			seed: func(t *testing.T, st store.MVCCStore, key []byte) {
+				seedLegacyZSetForBZPopMinTest(t, st, key, readTS, []redisZSetEntry{
+					{Member: "later", Score: 2},
+					{Member: "first", Score: 1},
+				})
+			},
+			wantMember: "first",
+			wantScore:  1,
+			wantWide:   false,
+			wantLast:   false,
+			wantScans:  []int{bzpopminScoreScanLimit, 1},
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+			key := []byte("zset-bzpopmin-" + tc.name)
+			base := store.NewMVCCStore()
+			tc.seed(t, base, key)
+			rec := &bzpopminRecordingStore{MVCCStore: base}
+			server := &RedisServer{store: rec}
+
+			candidate, err := server.bzpopminCandidateAt(ctx, key, readTS)
+			require.NoError(t, err)
+			require.NotNil(t, candidate)
+			require.Equal(t, tc.wantWide, candidate.isWide)
+			require.Equal(t, tc.wantLast, candidate.isLast)
+			require.Equal(t, tc.wantMember, candidate.entry.Member)
+			require.InDelta(t, tc.wantScore, candidate.entry.Score, 1e-9)
+			require.Len(t, rec.scans, len(tc.wantScans))
+			for i, wantLimit := range tc.wantScans {
+				require.Equal(t, wantLimit, rec.scans[i].limit)
+			}
+			require.Equal(t, store.ZSetScoreScanPrefix(key), rec.scans[0].start)
+		})
+	}
+}
+
+func seedZSetMemberRowsForBZPopMinTest(
+	t *testing.T,
+	st store.MVCCStore,
+	key []byte,
+	commitTS uint64,
+	entries []redisZSetEntry,
+) {
+	t.Helper()
+	ctx := context.Background()
+	for _, entry := range entries {
+		require.NoError(t, st.PutAt(
+			ctx,
+			store.ZSetMemberKey(key, []byte(entry.Member)),
+			store.MarshalZSetScore(entry.Score),
+			commitTS,
+			0,
+		))
+	}
+	require.NoError(t, st.PutAt(
+		ctx,
+		store.ZSetMetaKey(key),
+		store.MarshalZSetMeta(store.ZSetMeta{Len: int64(len(entries))}),
+		commitTS,
+		0,
+	))
+}
+
+func seedLegacyZSetForBZPopMinTest(
+	t *testing.T,
+	st store.MVCCStore,
+	key []byte,
+	commitTS uint64,
+	entries []redisZSetEntry,
+) {
+	t.Helper()
+	ctx := context.Background()
+	raw, err := marshalZSetValue(redisZSetValue{Entries: entries})
+	require.NoError(t, err)
+	require.NoError(t, st.PutAt(ctx, redisZSetKey(key), raw, commitTS, 0))
+}
+
 // TestRedis_BZPopMinWakesOnZAdd verifies the event-driven wake path:
 // an in-process ZADD on the leader's redis adapter must wake a
 // BZPOPMIN waiter on the same node so the reader returns the new

adapter/redis_peer_limiter.go

@@ -6,11 +6,13 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+
+	"github.com/bootjp/elastickv/internal/redislimits"
 )
 
 const (
 	redisPerPeerLimitEnv             = "ELASTICKV_REDIS_PER_PEER_CONNECTIONS"
-	defaultRedisPerPeerConnectionCap = 8
+	defaultRedisPerPeerConnectionCap = redislimits.DefaultElasticKVRedisConnections
 	redisPeerLimitError              = "ERR max connections per client exceeded"
 	unknownRedisPeer                 = "unknown"
 )

adapter/redis_peer_limiter_test.go

@@ -3,13 +3,23 @@ package adapter
 import (
 	"testing"
 
+	"github.com/bootjp/elastickv/internal/redislimits"
 	"github.com/stretchr/testify/require"
 )
 
 const (
 	testPeerLimit = 2
 )
 
+func TestRedisPeerLimiterDefaultMatchesProxyPool(t *testing.T) {
+	t.Setenv(redisPerPeerLimitEnv, "")
+	server := NewRedisServer(nil, "", nil, nil, nil, nil)
+
+	require.NotNil(t, server.peerLimiter)
+	require.Equal(t, defaultRedisPerPeerConnectionCap, server.peerLimiter.limit)
+	require.Equal(t, redislimits.DefaultElasticKVRedisConnections, server.peerLimiter.limit)
+}
+
 func TestRedisPeerLimiterRejectsAndReleases(t *testing.T) {
 	server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(testPeerLimit))
 	c1 := &remoteCommandRecorder{remote: "192.168.0.64:10001"}
@@ -95,7 +105,7 @@ func TestRedisLeaderClientPoolsSharePeerBudget(t *testing.T) {
 	}{
 		{name: "low cap", limit: 2, wantNormal: 1, wantBlocking: 1},
 		{name: "four cap", limit: 4, wantNormal: 2, wantBlocking: 2},
-		{name: "default cap", limit: 8, wantNormal: 4, wantBlocking: 4},
+		{name: "small cap", limit: 8, wantNormal: 4, wantBlocking: 4},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(tc.limit))