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`) 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.
 
 ## 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,58 @@
 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_BZPopMinCandidateUsesScoreIndex(t *testing.T) {
+	t.Parallel()
+	ctx := context.Background()
+	key := []byte("zset-score-index-pop")
+	readTS := uint64(10)
+	base := store.NewMVCCStore()
+	require.NoError(t, base.PutAt(ctx, store.ZSetMetaKey(key), store.MarshalZSetMeta(store.ZSetMeta{Len: 2}), readTS, 0))
+	require.NoError(t, base.PutAt(ctx, store.ZSetMemberKey(key, []byte("later")), store.MarshalZSetScore(2), readTS, 0))
+	require.NoError(t, base.PutAt(ctx, store.ZSetScoreKey(key, 2, []byte("later")), []byte{}, readTS, 0))
+	require.NoError(t, base.PutAt(ctx, store.ZSetMemberKey(key, []byte("first")), store.MarshalZSetScore(1), readTS, 0))
+	require.NoError(t, base.PutAt(ctx, store.ZSetScoreKey(key, 1, []byte("first")), []byte{}, readTS, 0))
+
+	rec := &bzpopminRecordingStore{MVCCStore: base}
+	server := &RedisServer{store: rec}
+	candidate, err := server.bzpopminCandidateAt(ctx, key, readTS)
+	require.NoError(t, err)
+	require.NotNil(t, candidate)
+	require.True(t, candidate.isWide)
+	require.False(t, candidate.isLast)
+	require.Equal(t, "first", candidate.entry.Member)
+	require.InDelta(t, 1.0, candidate.entry.Score, 1e-9)
+	require.Len(t, rec.scans, 1)
+	require.Equal(t, store.ZSetScoreScanPrefix(key), rec.scans[0].start)
+	require.Equal(t, 2, rec.scans[0].limit)
+}
+
 // 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

@@ -10,7 +10,7 @@ import (
 
 const (
 	redisPerPeerLimitEnv             = "ELASTICKV_REDIS_PER_PEER_CONNECTIONS"
-	defaultRedisPerPeerConnectionCap = 8
+	defaultRedisPerPeerConnectionCap = 64
 	redisPeerLimitError              = "ERR max connections per client exceeded"
 	unknownRedisPeer                 = "unknown"
 )

adapter/redis_peer_limiter_test.go

@@ -10,6 +10,15 @@ 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, 64, 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 +104,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))

adapter/redis_zset_cmds.go

@@ -27,6 +27,13 @@ type bzpopminResult struct {
 	entry redisZSetEntry
 }
 
+type bzpopminCandidate struct {
+	entry     redisZSetEntry
+	remaining []redisZSetEntry
+	isWide    bool
+	isLast    bool
+}
+
 const zsetOpsPerEntry = 2
 
 // buildZSetLegacyMigrationElems returns ops that atomically migrate a legacy
@@ -1128,44 +1135,115 @@ func (r *RedisServer) tryBZPopMinWithMode(key []byte, fast bool) (*bzpopminResul
 		if typ != redisTypeZSet {
 			return wrongTypeError()
 		}
-		value, _, err := r.loadZSetAt(ctx, key, readTS)
+		candidate, err := r.bzpopminCandidateAt(ctx, key, readTS)
 		if err != nil {
 			return err
 		}
-		if len(value.Entries) == 0 {
+		if candidate == nil {
 			result = nil
 			return nil
 		}
-		popped := value.Entries[0]
-		remaining := append([]redisZSetEntry(nil), value.Entries[1:]...)
-
-		// Detect wide-column storage.
-		memberPrefix := store.ZSetMemberScanPrefix(key)
-		memberEnd := store.PrefixScanEnd(memberPrefix)
-		probeKVs, probeErr := r.store.ScanAt(ctx, memberPrefix, memberEnd, 1, readTS)
-		if probeErr != nil {
-			return cockerrors.WithStack(probeErr)
-		}
-		isWide := len(probeKVs) > 0
-
-		if err := r.persistBZPopMinResult(ctx, key, readTS, popped, remaining, isWide); err != nil {
+		if err := r.persistBZPopMinResult(ctx, key, readTS, candidate); err != nil {
 			return err
 		}
-		result = &bzpopminResult{key: key, entry: popped}
+		result = &bzpopminResult{key: key, entry: candidate.entry}
 		return nil
 	})
 	return result, err
 }
 
-func (r *RedisServer) persistBZPopMinResult(ctx context.Context, key []byte, readTS uint64, popped redisZSetEntry, remaining []redisZSetEntry, isWide bool) error {
-	if len(remaining) == 0 {
+func (r *RedisServer) bzpopminCandidateAt(ctx context.Context, key []byte, readTS uint64) (*bzpopminCandidate, error) {
+	candidate, err := r.bzpopminWideScoreCandidateAt(ctx, key, readTS)
+	if err != nil || candidate != nil {
+		return candidate, err
+	}
+	candidate, err = r.bzpopminMemberOnlyCandidateAt(ctx, key, readTS)
+	if err != nil || candidate != nil {
+		return candidate, err
+	}
+	return r.bzpopminLegacyCandidateAt(ctx, key, readTS)
+}
+
+func (r *RedisServer) bzpopminWideScoreCandidateAt(ctx context.Context, key []byte, readTS uint64) (*bzpopminCandidate, error) {
+	scorePrefix := store.ZSetScoreScanPrefix(key)
+	scoreEnd := store.PrefixScanEnd(scorePrefix)
+	scoreKVs, err := r.store.ScanAt(ctx, scorePrefix, scoreEnd, 2, readTS) //nolint:mnd // first entry plus last-entry sentinel
+	if err != nil {
+		return nil, cockerrors.WithStack(err)
+	}
+	for _, kv := range scoreKVs {
+		score, member, ok := store.ExtractZSetScoreAndMember(kv.Key, key)
+		if !ok {
+			continue
+		}
+		return &bzpopminCandidate{
+			entry:  redisZSetEntry{Member: string(member), Score: score},
+			isWide: true,
+			isLast: len(scoreKVs) == 1,
+		}, nil
+	}
+	return nil, nil
+}
+
+func (r *RedisServer) bzpopminMemberOnlyCandidateAt(ctx context.Context, key []byte, readTS uint64) (*bzpopminCandidate, error) {
+	memberPrefix := store.ZSetMemberScanPrefix(key)
+	memberEnd := store.PrefixScanEnd(memberPrefix)
+	memberKVs, err := r.store.ScanAt(ctx, memberPrefix, memberEnd, 1, readTS)
+	if err != nil {
+		return nil, cockerrors.WithStack(err)
+	}
+	if len(memberKVs) > 0 {
+		value, loadErr := r.loadZSetMembersAt(ctx, key, readTS)
+		if loadErr != nil {
+			return nil, loadErr
+		}
+		if len(value.Entries) == 0 {
+			return nil, nil
+		}
+		return &bzpopminCandidate{
+			entry:     value.Entries[0],
+			remaining: append([]redisZSetEntry(nil), value.Entries[1:]...),
+			isWide:    true,
+			isLast:    len(value.Entries) == 1,
+		}, nil
+	}
+	return nil, nil
+}
+
+func (r *RedisServer) bzpopminLegacyCandidateAt(ctx context.Context, key []byte, readTS uint64) (*bzpopminCandidate, error) {
+	raw, err := r.store.GetAt(ctx, redisZSetKey(key), readTS)
+	if err != nil {
+		if cockerrors.Is(err, store.ErrKeyNotFound) {
+			return nil, nil
+		}
+		return nil, cockerrors.WithStack(err)
+	}
+	value, err := unmarshalZSetValue(raw)
+	if err != nil {
+		return nil, err
+	}
+	if len(value.Entries) == 0 {
+		return nil, nil
+	}
+	return &bzpopminCandidate{
+		entry:     value.Entries[0],
+		remaining: append([]redisZSetEntry(nil), value.Entries[1:]...),
+		isLast:    len(value.Entries) == 1,
+	}, nil
+}
+
+func (r *RedisServer) persistBZPopMinResult(ctx context.Context, key []byte, readTS uint64, candidate *bzpopminCandidate) error {
+	if candidate == nil {
+		return nil
+	}
+	if candidate.isLast {
 		elems, _, err := r.deleteLogicalKeyElems(ctx, key, readTS)
 		if err != nil {
 			return err
 		}
 		return r.dispatchElems(ctx, true, readTS, elems)
 	}
-	if isWide {
+	if candidate.isWide {
 		// Wide-column: delete the popped member key + score index, emit delta -1.
 		startTS := normalizeStartTS(readTS)
 		commitTS, err := r.nextCommitTSAfter(ctx, startTS, "persistBZPopMinResult: allocate commitTS")
@@ -1174,8 +1252,8 @@ func (r *RedisServer) persistBZPopMinResult(ctx context.Context, key []byte, rea
 		}
 		deltaVal := store.MarshalZSetMetaDelta(store.ZSetMetaDelta{LenDelta: -1})
 		elems := []*kv.Elem[kv.OP]{
-			{Op: kv.Del, Key: store.ZSetMemberKey(key, []byte(popped.Member))},
-			{Op: kv.Del, Key: store.ZSetScoreKey(key, popped.Score, []byte(popped.Member))},
+			{Op: kv.Del, Key: store.ZSetMemberKey(key, []byte(candidate.entry.Member))},
+			{Op: kv.Del, Key: store.ZSetScoreKey(key, candidate.entry.Score, []byte(candidate.entry.Member))},
 			{Op: kv.Put, Key: store.ZSetMetaDeltaKey(key, commitTS, 0), Value: deltaVal},
 			redisTxnWideZSetFenceElem(key),
 		}
@@ -1189,7 +1267,7 @@ func (r *RedisServer) persistBZPopMinResult(ctx context.Context, key []byte, rea
 		return cockerrors.WithStack(dispatchErr)
 	}
 	// Legacy blob: write back all remaining entries.
-	payload, err := marshalZSetValue(redisZSetValue{Entries: remaining})
+	payload, err := marshalZSetValue(redisZSetValue{Entries: candidate.remaining})
 	if err != nil {
 		return err
 	}

cmd/redis-proxy/main_test.go

@@ -93,17 +93,17 @@ func TestDeriveSecondaryConcurrency(t *testing.T) {
 			elasticKVPoolSize:       4,
 			wantWriteConcurrency:    64,
 			wantScriptConcurrency:   32,
-			wantBlockingConcurrency: 20,
+			wantBlockingConcurrency: 64,
 		},
 		{
-			name:                    "large remaining pool caps blocking replay",
+			name:                    "large remaining pool caps blocking replay at production default",
 			mode:                    proxy.ModeDualWrite,
 			primaryPoolSize:         128,
 			elasticKVPoolSize:       144,
 			writeConcurrency:        80,
 			wantWriteConcurrency:    80,
 			wantScriptConcurrency:   40,
-			wantBlockingConcurrency: 20,
+			wantBlockingConcurrency: 64,
 		},
 		{
 			name:                    "explicit write keeps derived script",

deploy/redis-proxy/docker-compose.ha.yml

@@ -26,7 +26,7 @@ services:
       - -listen=:6379
       - -primary=${REDIS_PROXY_PRIMARY:-redis:6379}
       - -secondary=${REDIS_PROXY_SECONDARY:-elastickv:6380}
-      - -elastickv-pool-size=${REDIS_PROXY_ELASTICKV_POOL_SIZE:-4}
+      - -elastickv-pool-size=${REDIS_PROXY_ELASTICKV_POOL_SIZE:-64}
       - -mode=${REDIS_PROXY_MODE:-dual-write-shadow}
       - -metrics=:9191
     networks:
@@ -46,7 +46,7 @@ services:
       - -listen=:6379
       - -primary=${REDIS_PROXY_PRIMARY:-redis:6379}
       - -secondary=${REDIS_PROXY_SECONDARY:-elastickv:6380}
-      - -elastickv-pool-size=${REDIS_PROXY_ELASTICKV_POOL_SIZE:-4}
+      - -elastickv-pool-size=${REDIS_PROXY_ELASTICKV_POOL_SIZE:-64}
       - -mode=${REDIS_PROXY_MODE:-dual-write-shadow}
       - -metrics=:9191
     networks:

docs/architecture_overview.md

@@ -170,7 +170,7 @@ sequenceDiagram
   participant H as "HLC (all nodes)"
   participant Tx as "Txn / MVCC read-write path"
 
-  loop "every hlcRenewalInterval (<3s)"
+  loop "every hlcRenewalInterval (<30s)"
     L->>RG: "Propose HLC lease (now + hlcPhysicalWindowMs)"
     RG-->>F: "Apply HLC lease entry"
     F->>H: "SetPhysicalCeiling(ms)"

docs/design/2026_04_24_implemented_workload_isolation.md

@@ -24,7 +24,7 @@ Implementation status:
   gated `EVAL`/`EVALSHA` path.
 - Shipped: Layer 3 per-peer Redis connection admission in
   `adapter/redis_peer_limiter.go`, wired through `RedisServer.Run` accept and
-  close hooks. Default cap is 8 per peer IP and is configurable via
+  close hooks. Default cap is 64 per peer IP and is configurable via
   `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` /
   `WithRedisPerPeerConnectionLimit`. Redis leader-proxy clients use a small
   explicit go-redis pool below that default cap, and Pub/Sub detached sockets
@@ -374,7 +374,7 @@ one check per accept, not per command.
 
 ### Recommended v1 shape
 
-**Per-peer-IP connection cap, default `N=8`, env-configurable,
+**Per-peer-IP connection cap, default `N=64`, env-configurable,
 enforced at accept.** On reject, accept the TCP connection, write a
 `-ERR max connections per client exceeded` RESP error, then close —
 so the client sees a protocol-level message instead of a bare

docs/design/2026_05_28_implemented_tla_safety_spec.md

@@ -199,8 +199,8 @@ cannot check them as state invariants.
     independently reach `ceiling + 1` before a fresh ceiling is
     renewed, the new leader's first `Next()` can tie or undercut the
     old leader's last commit. Bounding inter-node skew to less than
-    one ceiling window (< 3s with the current `hlcPhysicalWindowMs =
-    3000ms`) keeps the window wide enough that the new leader cannot
+    one ceiling window (< 30s with the current `hlcPhysicalWindowMs =
+    30000ms`) keeps the window wide enough that the new leader cannot
     independently reach the overflow value before a renewal applies.
     Should be surfaced in operator docs as a cluster prerequisite.
   - **(ii) Logical-counter handoff.** The 16-bit logical half of the HLC
@@ -270,7 +270,7 @@ cannot check them as state invariants.
     `hlcPhysicalWindowMs` cannot serve any persistence timestamp —
     every client commit is rejected until renewal succeeds. This is
     a CP, not AP, trade-off and operators must size
-    `hlcPhysicalWindowMs` (currently 3s) relative to expected
+    `hlcPhysicalWindowMs` (currently 30s) relative to expected
     partition duration; see §9 risk 7.
 
 ### 5.2 OCC
@@ -676,7 +676,7 @@ does not keep this document in `partial`.
 7. **Fail-closed availability under partition.** HLC-4 precondition
    (iii) makes the ceiling-fence behaviour normative: a leader
    partitioned from the default group's quorum for longer than
-   `hlcPhysicalWindowMs` (currently 3s) cannot serve any persistence
+   `hlcPhysicalWindowMs` (currently 30s) cannot serve any persistence
    timestamp, so client commits are rejected until renewal succeeds.
    This is a CP, not AP, trade-off and is a stricter regime than the
    current implementation (which silently keeps issuing). Mitigation: