Reduce Redis proxy drops and Raft snapshot load

adapter/redis_delta_compactor.go

@@ -986,10 +986,87 @@ func (c *DeltaCompactor) buildSimpleCompactElems(
 	return foldSimpleLenDeltas(deltaKVs, 0, baseLen, expireAt, metaKey, unmarshalDelta, marshalBase)
 }
 
-// zsetInlineMetaCompactionThreshold is the number of existing ZSetMetaDeltaKey
-// entries at which an inline compaction is triggered during a Lua ZSet delta commit.
-// Set to MaxDeltaScanLimit so compaction fires just before reads would fail.
-const zsetInlineMetaCompactionThreshold = store.MaxDeltaScanLimit
+const (
+	// Inline compaction fires at the read-side hard limit so a hot key is
+	// folded before the next delta write pushes it into ErrDeltaScanTruncated.
+	inlineMetaCompactionThreshold = store.MaxDeltaScanLimit
+	// zsetInlineMetaCompactionThreshold is kept for existing tests and call sites.
+	zsetInlineMetaCompactionThreshold = inlineMetaCompactionThreshold
+	listInlineMetaCompactionThreshold = inlineMetaCompactionThreshold
+)
+
+func (r *RedisServer) listInlineMetaCompactionElems(
+	ctx context.Context, key []byte, readTS uint64, additionalDelta store.ListMetaDelta, additionalDeltaEntries int,
+) ([]*kv.Elem[kv.OP], bool, error) {
+	prefix := store.ListMetaDeltaScanPrefix(key)
+	end := store.PrefixScanEnd(prefix)
+	deltaKVs, err := r.store.ScanAt(ctx, prefix, end, listInlineMetaCompactionThreshold+1, readTS)
+	if err != nil {
+		return nil, false, errors.WithStack(err)
+	}
+	if len(deltaKVs)+additionalDeltaEntries <= store.MaxDeltaScanLimit {
+		return nil, false, nil
+	}
+
+	baseMeta, raw, err := r.loadListBaseMetaForInlineCompaction(ctx, key, readTS)
+	if err != nil {
+		return nil, false, err
+	}
+	expireAt, err := compactedMetaExpireAt(ctx, r.store, key, readTS, raw, redisWideMetaInlineSizeBytes, baseMeta.ExpireAt, deltaKVs, prefix)
+	if err != nil {
+		return nil, false, err
+	}
+
+	elems, err := buildListInlineMetaCompactionElems(key, baseMeta, expireAt, additionalDelta, deltaKVs)
+	if err != nil {
+		return nil, false, err
+	}
+	return elems, true, nil
+}
+
+func (r *RedisServer) loadListBaseMetaForInlineCompaction(ctx context.Context, key []byte, readTS uint64) (store.ListMeta, []byte, error) {
+	raw, err := r.store.GetAt(ctx, store.ListMetaKey(key), readTS)
+	if err != nil {
+		if errors.Is(err, store.ErrKeyNotFound) {
+			return store.ListMeta{}, nil, nil
+		}
+		return store.ListMeta{}, nil, errors.WithStack(err)
+	}
+	baseMeta, err := store.UnmarshalListMeta(raw)
+	if err != nil {
+		return store.ListMeta{}, nil, errors.WithStack(err)
+	}
+	return baseMeta, raw, nil
+}
+
+func buildListInlineMetaCompactionElems(
+	key []byte, baseMeta store.ListMeta, expireAt uint64, additionalDelta store.ListMetaDelta, deltaKVs []*store.KVPair,
+) ([]*kv.Elem[kv.OP], error) {
+	headDelta, lenDelta, err := sumListMetaDeltas(deltaKVs)
+	if err != nil {
+		return nil, err
+	}
+	newMeta := store.ListMeta{
+		Head:     baseMeta.Head + headDelta + additionalDelta.HeadDelta,
+		Len:      baseMeta.Len + lenDelta + additionalDelta.LenDelta,
+		ExpireAt: expireAt,
+	}
+	if newMeta.Len < 0 {
+		newMeta.Len = 0
+	}
+	newMeta.Tail = newMeta.Head + newMeta.Len
+
+	metaElem, err := listMetaElemForLen(store.ListMetaKey(key), newMeta)
+	if err != nil {
+		return nil, err
+	}
+	elems := make([]*kv.Elem[kv.OP], 0, 1+len(deltaKVs))
+	elems = append(elems, metaElem)
+	for _, d := range deltaKVs {
+		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Del, Key: bytes.Clone(d.Key)})
+	}
+	return elems, nil
+}
 
 // zsetInlineMetaCompactionElems checks whether ZSetMetaDeltaKeys for key have
 // accumulated past the inline threshold. When they have, it returns elems that

adapter/redis_delta_compactor_test.go

@@ -1669,3 +1669,64 @@ func TestZSetInlineMetaCompaction(t *testing.T) {
 	require.NoError(t, err)
 	require.Empty(t, remaining, "all delta keys must be removed after inline compaction")
 }
+
+func TestListInlineMetaCompaction(t *testing.T) {
+	t.Parallel()
+
+	ctx := context.Background()
+	st := store.NewMVCCStore()
+	r := &RedisServer{store: st}
+	userKey := []byte("inline:list")
+
+	baseMeta := store.ListMeta{Head: 100, Len: 10, ExpireAt: 1234}
+	metaBytes, err := store.MarshalListMeta(baseMeta)
+	require.NoError(t, err)
+	require.NoError(t, st.PutAt(ctx, store.ListMetaKey(userKey), metaBytes, 1, 0))
+
+	const threshold = listInlineMetaCompactionThreshold
+	delta := store.MarshalListMetaDelta(store.ListMetaDelta{HeadDelta: 1, LenDelta: 1})
+	for i := range threshold - 1 {
+		ts := uint64(10 + i) //nolint:gosec // i is bounded by threshold.
+		require.NoError(t, st.PutAt(ctx, store.ListMetaDeltaKey(userKey, ts, 0), delta, ts, 0))
+	}
+
+	readTS := st.LastCommitTS()
+	elems, compacted, err := r.listInlineMetaCompactionElems(ctx, userKey, readTS, store.ListMetaDelta{HeadDelta: 1, LenDelta: 1}, 1)
+	require.NoError(t, err)
+	require.False(t, compacted, "below threshold: should not compact")
+	require.Nil(t, elems)
+
+	const lastTS = uint64(10 + threshold - 1)
+	require.NoError(t, st.PutAt(ctx, store.ListMetaDeltaKey(userKey, lastTS, 0), delta, lastTS, 0))
+
+	readTS = st.LastCommitTS()
+	elems, compacted, err = r.listInlineMetaCompactionElems(ctx, userKey, readTS, store.ListMetaDelta{HeadDelta: -2, LenDelta: 3}, 1)
+	require.NoError(t, err)
+	require.True(t, compacted, "at threshold: should compact")
+
+	coord := newLocalAdapterCoordinator(st)
+	commitTS := coord.Clock().Next()
+	_, dispatchErr := coord.Dispatch(ctx, &kv.OperationGroup[kv.OP]{
+		IsTxn:    true,
+		StartTS:  0,
+		CommitTS: commitTS,
+		Elems:    elems,
+	})
+	require.NoError(t, dispatchErr)
+
+	afterTS := st.LastCommitTS()
+	raw, err := st.GetAt(ctx, store.ListMetaKey(userKey), afterTS)
+	require.NoError(t, err)
+	got, err := store.UnmarshalListMeta(raw)
+	require.NoError(t, err)
+	require.Equal(t, int64(100+threshold-2), got.Head)
+	require.Equal(t, int64(10+threshold+3), got.Len)
+	require.Equal(t, uint64(1234), got.ExpireAt)
+	require.Equal(t, got.Head+got.Len, got.Tail)
+
+	prefix := store.ListMetaDeltaScanPrefix(userKey)
+	scanEnd := store.PrefixScanEnd(prefix)
+	remaining, err := st.ScanAt(ctx, prefix, scanEnd, threshold+1, afterTS)
+	require.NoError(t, err)
+	require.Empty(t, remaining, "all list delta keys must be removed after inline compaction")
+}

adapter/redis_lua_context.go

@@ -3773,8 +3773,8 @@ func (c *luaScriptContext) listCommitPlan(ctx context.Context, key string, commi
 		elems, err := c.listCommitElems(ctx, key, commitTS)
 		return luaCommitPlan{elems: elems}, err
 	}
-	elems, err := c.listDeltaCommitElems(key, st, commitTS)
-	return luaCommitPlan{preserveExisting: true, elems: elems}, err
+	elems, err := c.listDeltaCommitElems(ctx, key, st, commitTS)
+	return luaCommitPlan{preserveExisting: true, inlineMetaRewritten: luaListMetaRewritten(elems, []byte(key)), elems: elems}, err
 }
 
 func (c *luaScriptContext) listCommitElems(ctx context.Context, key string, _ uint64) ([]*kv.Elem[kv.OP], error) {
@@ -3812,7 +3812,7 @@ func (c *luaScriptContext) listCommitElems(ctx context.Context, key string, _ ui
 	return elems, nil
 }
 
-func (c *luaScriptContext) listDeltaCommitElems(key string, st *luaListState, commitTS uint64) ([]*kv.Elem[kv.OP], error) {
+func (c *luaScriptContext) listDeltaCommitElems(ctx context.Context, key string, st *luaListState, commitTS uint64) ([]*kv.Elem[kv.OP], error) {
 	if !st.exists || st.currentLen() == 0 {
 		return nil, nil
 	}
@@ -3829,36 +3829,112 @@ func (c *luaScriptContext) listDeltaCommitElems(key string, st *luaListState, co
 
 	// Emit Delta keys for any appended values and trims instead of writing base meta.
 	// Trims are counted separately as negative deltas.
-	var seqInTxn uint32
+	keyBytes := []byte(key)
+	elems, err = c.appendListDeltaMetaElems(ctx, elems, keyBytes, luaListMetaDeltas(st), commitTS)
+	if err != nil {
+		return nil, err
+	}
+	if len(elems) != 0 {
+		elems = append(elems, redisTxnWideListFenceElem(keyBytes))
+	}
+	return elems, nil
+}
+
+func luaListMetaDeltas(st *luaListState) []store.ListMetaDelta {
+	deltas := make([]store.ListMetaDelta, 0, redisPairWidth)
 	if len(st.rightValues) > 0 || st.rightTrim > 0 {
-		rightDelta := store.MarshalListMetaDelta(store.ListMetaDelta{
+		deltas = append(deltas, store.ListMetaDelta{
 			HeadDelta: 0,
 			LenDelta:  int64(len(st.rightValues)) - st.rightTrim,
 		})
-		elems = append(elems, &kv.Elem[kv.OP]{
-			Op:    kv.Put,
-			Key:   store.ListMetaDeltaKey([]byte(key), commitTS, seqInTxn),
-			Value: rightDelta,
-		})
-		seqInTxn++
 	}
 	if len(st.leftValues) > 0 || st.leftTrim > 0 {
-		leftDelta := store.MarshalListMetaDelta(store.ListMetaDelta{
+		deltas = append(deltas, store.ListMetaDelta{
 			HeadDelta: -int64(len(st.leftValues)) + st.leftTrim,
 			LenDelta:  int64(len(st.leftValues)) - st.leftTrim,
 		})
+	}
+	return deltas
+}
+
+func (c *luaScriptContext) appendListDeltaMetaElems(
+	ctx context.Context, elems []*kv.Elem[kv.OP], key []byte, deltas []store.ListMetaDelta, commitTS uint64,
+) ([]*kv.Elem[kv.OP], error) {
+	compactElems, compacted, err := c.listInlineMetaCompactionElems(ctx, key, deltas)
+	if err != nil {
+		return nil, err
+	}
+	if compacted {
+		elems = append(elems, compactElems...)
+		return elems, nil
+	}
+	for seqInTxn, delta := range deltas {
 		elems = append(elems, &kv.Elem[kv.OP]{
 			Op:    kv.Put,
-			Key:   store.ListMetaDeltaKey([]byte(key), commitTS, seqInTxn),
-			Value: leftDelta,
+			Key:   store.ListMetaDeltaKey(key, commitTS, uint32(seqInTxn)), //nolint:gosec // len(deltas) is bounded by redisPairWidth.
+			Value: store.MarshalListMetaDelta(delta),
 		})
 	}
-	if len(elems) != 0 {
-		elems = append(elems, redisTxnWideListFenceElem([]byte(key)))
-	}
 	return elems, nil
 }
 
+func (c *luaScriptContext) listInlineMetaCompactionElems(ctx context.Context, key []byte, deltas []store.ListMetaDelta) ([]*kv.Elem[kv.OP], bool, error) {
+	if len(deltas) == 0 {
+		return nil, false, nil
+	}
+	var additional store.ListMetaDelta
+	for _, delta := range deltas {
+		additional.HeadDelta += delta.HeadDelta
+		additional.LenDelta += delta.LenDelta
+	}
+	elems, compacted, err := c.server.listInlineMetaCompactionElems(ctx, key, c.startTS, additional, len(deltas))
+	if err != nil || !compacted {
+		return elems, compacted, err
+	}
+	return c.listMetaElemsWithFinalTTL(ctx, key, elems)
+}
+
+func (c *luaScriptContext) listMetaElemsWithFinalTTL(ctx context.Context, key []byte, elems []*kv.Elem[kv.OP]) ([]*kv.Elem[kv.OP], bool, error) {
+	ttl, err := c.finalTTL(ctx, key)
+	if err != nil {
+		return nil, false, err
+	}
+	expireAt := ttlMillis(ttl)
+	metaKey := store.ListMetaKey(key)
+	out := make([]*kv.Elem[kv.OP], 0, len(elems))
+	for _, elem := range elems {
+		if elem.Op == kv.Put && string(elem.Key) == string(metaKey) {
+			meta, err := store.UnmarshalListMeta(elem.Value)
+			if err != nil {
+				return nil, false, errors.WithStack(err)
+			}
+			meta.ExpireAt = expireAt
+			metaBytes, err := store.MarshalListMeta(meta)
+			if err != nil {
+				return nil, false, errors.WithStack(err)
+			}
+			out = append(out, &kv.Elem[kv.OP]{
+				Op:    elem.Op,
+				Key:   elem.Key,
+				Value: metaBytes,
+			})
+			continue
+		}
+		out = append(out, elem)
+	}
+	return out, true, nil
+}
+
+func luaListMetaRewritten(elems []*kv.Elem[kv.OP], key []byte) bool {
+	metaKey := store.ListMetaKey(key)
+	for _, elem := range elems {
+		if string(elem.Key) == string(metaKey) {
+			return true
+		}
+	}
+	return false
+}
+
 func validateListDeltaRanges(st *luaListState) (int64, int64, error) {
 	remainingHead := st.meta.Head + st.leftTrim
 	if remainingHead < math.MinInt64+int64(len(st.leftValues)) {

adapter/redis_peer_limiter_test.go

@@ -104,6 +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: "raised cap", limit: defaultRedisPerPeerConnectionCap, wantNormal: 4, wantBlocking: 4},
 	} {
 		t.Run(tc.name, func(t *testing.T) {
 			server := NewRedisServer(nil, "", nil, nil, nil, nil, WithRedisPerPeerConnectionLimit(tc.limit))

docs/redis-proxy-deployment.md

@@ -35,7 +35,7 @@ go build -o redis-proxy ./cmd/redis-proxy/
 | `-secondary-db` | `0` | Secondary Redis DB number |
 | `-secondary-password` | (empty) | Secondary Redis password |
 | `-primary-pool-size` | `128` | Primary Redis backend connection pool size |
-| `-elastickv-pool-size` | `64` | ElasticKV backend connection pool size |
+| `-elastickv-pool-size` | `64` | ElasticKV backend command pool size; keep the server-side `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` above this to leave room for dedicated PubSub connections |
 | `-secondary-write-concurrency` | `0` | Shared maximum for all asynchronous secondary writes, including scripts. `0` derives half of the secondary backend pool size, minimum `1` |
 | `-secondary-script-concurrency` | `0` | Lua-script sublimit within `-secondary-write-concurrency`. `0` derives half of the shared write limit, minimum `1` |
 | `-secondary-write-queue-size` | `0` | Bounded queue for non-script secondary writes. `0` derives `64 * concurrency`, clamped to `64..8192` |

internal/raftengine/etcd/fsm_snapshot_file_test.go

@@ -369,6 +369,30 @@ func TestPrepareFSMSnapshotWriteKeepsNewestRestorablePair(t *testing.T) {
 	require.FileExists(t, filepath.Join(fsmSnapDir, "leftover.fsm.tmp"))
 }
 
+func TestPrepareFSMSnapshotWriteRejectsFooterOnlyRestorablePair(t *testing.T) {
+	snapDir := t.TempDir()
+	fsmSnapDir := t.TempDir()
+
+	crc100, _ := writeFSMFileForTest(t, fsmSnapDir, 100, []byte("valid previous snapshot"))
+	createTokenSnapFileWithTerm(t, snapDir, 1, 100, crc100)
+
+	tokenCRC := crc32.Checksum([]byte("valid newer snapshot"), crc32cTable)
+	f, err := os.Create(fsmSnapPath(fsmSnapDir, 200))
+	require.NoError(t, err)
+	_, err = f.Write([]byte("corrupted newer snapshot"))
+	require.NoError(t, err)
+	require.NoError(t, binary.Write(f, binary.BigEndian, tokenCRC))
+	require.NoError(t, f.Close())
+	createTokenSnapFileWithTerm(t, snapDir, 1, 200, tokenCRC)
+
+	require.NoError(t, prepareFSMSnapshotWrite(snapDir, fsmSnapDir, 300))
+
+	require.FileExists(t, filepath.Join(snapDir, "0000000000000001-0000000000000064.snap"))
+	require.FileExists(t, fsmSnapPath(fsmSnapDir, 100))
+	require.NoFileExists(t, filepath.Join(snapDir, "0000000000000001-00000000000000c8.snap"))
+	require.NoFileExists(t, fsmSnapPath(fsmSnapDir, 200))
+}
+
 func TestPrepareFSMSnapshotWriteKeepsWALValidFallbackPair(t *testing.T) {
 	snapDir := t.TempDir()
 	fsmSnapDir := t.TempDir()

kv/lease_warmup_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/bootjp/elastickv/distribution"
 	"github.com/bootjp/elastickv/internal/monoclock"
 	"github.com/bootjp/elastickv/internal/raftengine"
 	"github.com/stretchr/testify/require"
@@ -427,6 +428,25 @@ func TestShardedCoordinator_RenewHLCLeases_SkipsInFlightGroup(t *testing.T) {
 		"the group must be eligible for renewal after the in-flight proposal finishes")
 }
 
+func TestShardedCoordinator_ProposeHLCLease_UsesDedicatedTimestampGroup(t *testing.T) {
+	t.Parallel()
+	eng0 := newShardedLeaseEngine(300)
+	eng1 := newShardedLeaseEngine(100)
+	distEngine := distribution.NewEngine()
+	distEngine.UpdateRoute([]byte(""), nil, 1)
+	coord := NewShardedCoordinator(distEngine, map[uint64]*ShardGroup{
+		0: {Engine: eng0},
+		1: {Engine: eng1},
+	}, 1, NewHLC(), nil).WithTimestampGroup(0)
+
+	err := coord.ProposeHLCLease(context.Background(), time.Now().UnixMilli()+hlcPhysicalWindowMs)
+	require.NoError(t, err)
+	require.Equal(t, int32(1), eng0.proposeCalls.Load())
+	require.Equal(t, int32(0), eng1.proposeCalls.Load())
+	require.True(t, coord.groups[0].lease.valid(monoclock.Now()),
+		"a synchronous timestamp renewal must warm the timestamp group's lease")
+}
+
 func hlcRenewalInFlight(coord *ShardedCoordinator, gid uint64) bool {
 	coord.hlcRenewalMu.Lock()
 	defer coord.hlcRenewalMu.Unlock()