From 657681946ab3e48f101807d0ae17f4cb7e56e005 Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 08:35:30 +0900 Subject: [PATCH 1/3] redis: reduce proxy drops and raft snapshot load --- adapter/redis_delta_compactor.go | 85 +++++++++++++- adapter/redis_delta_compactor_test.go | 61 ++++++++++ adapter/redis_lua_context.go | 110 +++++++++++++++--- adapter/redis_peer_limiter.go | 2 +- adapter/redis_peer_limiter_test.go | 1 + deploy/redis-proxy/docker-compose.ha.yml | 4 +- docs/redis-proxy-deployment.md | 18 +-- internal/raftengine/etcd/fsm_snapshot_file.go | 34 +++++- proxy/backend.go | 7 +- proxy/dualwrite.go | 6 +- proxy/proxy_test.go | 31 ++++- 11 files changed, 319 insertions(+), 40 deletions(-) diff --git a/adapter/redis_delta_compactor.go b/adapter/redis_delta_compactor.go index 424b71a7c..b90ed2fa0 100644 --- a/adapter/redis_delta_compactor.go +++ b/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 diff --git a/adapter/redis_delta_compactor_test.go b/adapter/redis_delta_compactor_test.go index b63e23451..517a806ab 100644 --- a/adapter/redis_delta_compactor_test.go +++ b/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") +} diff --git a/adapter/redis_lua_context.go b/adapter/redis_lua_context.go index 94cd49627..35326e762 100644 --- a/adapter/redis_lua_context.go +++ b/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)) { diff --git a/adapter/redis_peer_limiter.go b/adapter/redis_peer_limiter.go index d1bace22c..c9d44329a 100644 --- a/adapter/redis_peer_limiter.go +++ b/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" ) diff --git a/adapter/redis_peer_limiter_test.go b/adapter/redis_peer_limiter_test.go index ea631375f..84d50d8b0 100644 --- a/adapter/redis_peer_limiter_test.go +++ b/adapter/redis_peer_limiter_test.go @@ -96,6 +96,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)) diff --git a/deploy/redis-proxy/docker-compose.ha.yml b/deploy/redis-proxy/docker-compose.ha.yml index c150fd9d1..1376c0a84 100644 --- a/deploy/redis-proxy/docker-compose.ha.yml +++ b/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: diff --git a/docs/redis-proxy-deployment.md b/docs/redis-proxy-deployment.md index 003597f2d..26030d19c 100644 --- a/docs/redis-proxy-deployment.md +++ b/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` | `4` | ElasticKV backend connection pool size | +| `-elastickv-pool-size` | `64` | ElasticKV backend connection pool size. Keep this at or below `ELASTICKV_REDIS_PER_PEER_CONNECTIONS` on the ElasticKV servers | | `-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` | @@ -94,9 +94,9 @@ docker run --rm \ -primary redis.internal:6379 \ -primary-password "${REDIS_PASSWORD}" \ -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 \ -mode dual-write-shadow \ -secondary-timeout 5s \ -shadow-timeout 3s \ @@ -118,9 +118,9 @@ services: - -listen=:6479 - -primary=redis:6379 - -secondary=elastickv: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 - -mode=dual-write-shadow - -metrics=:9191 depends_on: @@ -213,7 +213,7 @@ Override backend wiring via env vars before `docker compose up`: ```bash REDIS_PROXY_PRIMARY=redis.prod.internal:6379 \ REDIS_PROXY_SECONDARY=elastickv-1.prod.internal:6380,elastickv-2.prod.internal:6380,elastickv-3.prod.internal:6380 \ -REDIS_PROXY_ELASTICKV_POOL_SIZE=4 \ +REDIS_PROXY_ELASTICKV_POOL_SIZE=64 \ REDIS_PROXY_MODE=dual-write-shadow \ docker compose -f docker-compose.ha.yml up -d ``` @@ -414,7 +414,7 @@ groups: | Parameter | Value | Description | |-----------|-------|-------------| | Redis connection pool size | 128 | Default go-redis pool size for Redis | -| ElasticKV connection pool size | 4 | Default per-leader pool; keep within the server per-peer connection limit | +| ElasticKV connection pool size | 64 | Default per-leader pool; keep within the server per-peer connection limit | | Dial timeout | 5s | Backend connection timeout | | Read timeout | 3s | Backend read timeout | | Write timeout | 3s | Backend write timeout | diff --git a/internal/raftengine/etcd/fsm_snapshot_file.go b/internal/raftengine/etcd/fsm_snapshot_file.go index 574724cae..595966e55 100644 --- a/internal/raftengine/etcd/fsm_snapshot_file.go +++ b/internal/raftengine/etcd/fsm_snapshot_file.go @@ -831,7 +831,39 @@ func fsmSnapshotPairRestorable(snapDir, fsmSnapDir, snapName string, term, index if !ok { return false } - return verifyFSMSnapshotFileWithToken(fsmSnapPath(fsmSnapDir, index), tok.CRC32C, true) == nil + // Prewrite cleanup runs on the snapshot receive path. Avoid a full payload + // CRC here; startup and actual restore still do full verification before + // trusting the snapshot bytes. + return fsmSnapshotFooterMatchesToken(fsmSnapPath(fsmSnapDir, index), tok.CRC32C) == nil +} + +func fsmSnapshotFooterMatchesToken(path string, tokenCRC uint32) error { + f, err := os.Open(path) + if err != nil { + return statFSMFileError(err) + } + defer f.Close() + + info, err := f.Stat() + if err != nil { + return errors.WithStack(err) + } + if info.Size() < fsmMinFileSize { + return errors.Wrapf(ErrFSMSnapshotTooSmall, + "file too small: %d bytes (minimum %d)", info.Size(), fsmMinFileSize) + } + if _, err := f.Seek(-fsmFooterSize, io.SeekEnd); err != nil { + return errors.WithStack(err) + } + var footer uint32 + if err := binary.Read(f, binary.BigEndian, &footer); err != nil { + return errors.WithStack(err) + } + if footer != tokenCRC { + return errors.Wrapf(ErrFSMSnapshotFileCRC, + "path=%s footer=%08x token=%08x", path, footer, tokenCRC) + } + return nil } func snapshotTokenFromSnapFile(snapDir, snapName string, term, index uint64) (snapshotToken, bool) { diff --git a/proxy/backend.go b/proxy/backend.go index 694155dba..3c93e71b0 100644 --- a/proxy/backend.go +++ b/proxy/backend.go @@ -11,7 +11,7 @@ import ( const ( defaultPoolSize = 128 - defaultElasticKVPoolSize = 4 + defaultElasticKVPoolSize = 64 defaultDialTimeout = 5 * time.Second defaultReadTimeout = 3 * time.Second defaultWriteTimeout = 3 * time.Second @@ -72,9 +72,8 @@ func DefaultBackendOptions() BackendOptions { } // DefaultElasticKVBackendOptions returns defaults for proxy backends that -// connect to ElasticKV's Redis adapter. ElasticKV limits concurrent Redis -// connections per peer by default, so keep the pool below that cap unless the -// operator also raises ELASTICKV_REDIS_PER_PEER_CONNECTIONS on the cluster. +// connect to ElasticKV's Redis adapter. Keep this below the server-side +// ELASTICKV_REDIS_PER_PEER_CONNECTIONS cap for the proxy host. func DefaultElasticKVBackendOptions() BackendOptions { opts := DefaultBackendOptions() opts.PoolSize = defaultElasticKVPoolSize diff --git a/proxy/dualwrite.go b/proxy/dualwrite.go index dc1e16bf7..34c1c3259 100644 --- a/proxy/dualwrite.go +++ b/proxy/dualwrite.go @@ -369,7 +369,7 @@ func (d *DualWriter) Blocking(ctx context.Context, cmd string, args [][]byte) (a deadlineAsMiss: true, }) }) - } else if shouldReplayBlockingToSecondary(cmd) { + } else if shouldReplayBlockingToSecondary(cmd) && blockingResultMayHaveMutated(resp, err) { d.goBlockingReplay(func(ctx context.Context) { sCtx, cancel := context.WithTimeout(ctx, time.Second) defer cancel() @@ -381,6 +381,10 @@ func (d *DualWriter) Blocking(ctx context.Context, cmd string, args [][]byte) (a return resp, err //nolint:wrapcheck // redis.Nil must pass through unwrapped for callers to detect nil replies } +func blockingResultMayHaveMutated(resp any, err error) bool { + return err == nil && resp != nil +} + // Admin forwards an admin command to the primary only. // cmd must be the pre-uppercased command name. func (d *DualWriter) Admin(ctx context.Context, cmd string, args [][]byte) (any, error) { diff --git a/proxy/proxy_test.go b/proxy/proxy_test.go index 2a5cd808f..01389e6d1 100644 --- a/proxy/proxy_test.go +++ b/proxy/proxy_test.go @@ -679,6 +679,35 @@ func TestDualWriter_Blocking_ReplaysBZPopAsZRem(t *testing.T) { } } +func TestDualWriter_Blocking_DoesNotReplayNilResult(t *testing.T) { + primary := &timeoutCapturingBackend{ + name: "primary", + returnErr: redis.Nil, + } + secondary := newMockBackend("secondary") + + metrics := newTestMetrics() + d := NewDualWriter( + primary, + secondary, + ProxyConfig{ + Mode: ModeDualWrite, + SecondaryTimeout: time.Second, + SecondaryBlockingReplayConcurrency: 1, + }, + metrics, + newTestSentry(), + testLogger, + ) + + resp, err := d.Blocking(context.Background(), "BZPOPMIN", [][]byte{[]byte("BZPOPMIN"), []byte("queue"), []byte("5")}) + assert.ErrorIs(t, err, redis.Nil) + assert.Nil(t, resp) + d.Close() + + assert.Equal(t, 0, secondary.CallCount()) +} + func TestDualWriter_Blocking_RetriesBZPopReplayUntilRemoved(t *testing.T) { primary := &timeoutCapturingBackend{ name: "primary", @@ -1334,7 +1363,7 @@ func TestDefaultBackendOptions(t *testing.T) { func TestDefaultElasticKVBackendOptions(t *testing.T) { opts := DefaultElasticKVBackendOptions() - assert.Equal(t, 4, opts.PoolSize) + assert.Equal(t, 64, opts.PoolSize) assert.Equal(t, 5*time.Second, opts.DialTimeout) assert.Equal(t, 3*time.Second, opts.ReadTimeout) assert.Equal(t, 3*time.Second, opts.WriteTimeout) From a30bf25010abe874294022144503b30f476a835a Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 08:44:19 +0900 Subject: [PATCH 2/3] kv: renew expired HLC lease before TSO retry --- kv/lease_warmup_test.go | 20 +++++++++++++++ kv/sharded_coordinator.go | 52 ++++++++++++++++++++++++++++++++++----- kv/tso.go | 27 ++++++++++++++++++++ kv/tso_test.go | 51 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 144 insertions(+), 6 deletions(-) diff --git a/kv/lease_warmup_test.go b/kv/lease_warmup_test.go index 5558a913c..0617ed5fb 100644 --- a/kv/lease_warmup_test.go +++ b/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" @@ -369,6 +370,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() diff --git a/kv/sharded_coordinator.go b/kv/sharded_coordinator.go index da532dcb2..22c6c727b 100644 --- a/kv/sharded_coordinator.go +++ b/kv/sharded_coordinator.go @@ -2338,6 +2338,40 @@ func (c *ShardedCoordinator) RunHLCLeaseRenewal(ctx context.Context) { } } +func (c *ShardedCoordinator) ProposeHLCLease(ctx context.Context, ceilingMs int64) error { + if ctx == nil { + ctx = context.Background() + } + var firstErr error + for _, gid := range c.timestampLeaseRenewalGroupIDs() { + group := c.groups[gid] + if group == nil || group.Engine == nil || group.Engine.State() != raftengine.StateLeader { + continue + } + if err := c.proposeHLCLeaseToGroup(ctx, gid, group, ceilingMs); err != nil { + if firstErr == nil { + firstErr = err + } + continue + } + return nil + } + if firstErr != nil { + return firstErr + } + return errors.WithStack(ErrLeaderNotFound) +} + +func (c *ShardedCoordinator) timestampLeaseRenewalGroupIDs() []uint64 { + if c == nil { + return nil + } + if c.timestampGroupConfigured { + return []uint64{c.timestampGroup} + } + return c.timestampBridgeCandidateGroupIDs() +} + // renewHLCLeases starts one renewal proposal for every shard group this node // currently leads. It does not wait for those proposals before returning; the // returned channel closes when the launched proposals finish and exists for @@ -2418,6 +2452,16 @@ func (c *ShardedCoordinator) finishHLCLeaseRenewal(gid uint64) { // lease -- doing so would force every read onto the slow path. func (c *ShardedCoordinator) renewHLCLease(ctx context.Context, gid uint64, group *ShardGroup) { ceilingMs := time.Now().UnixMilli() + hlcPhysicalWindowMs + if err := c.proposeHLCLeaseToGroup(ctx, gid, group, ceilingMs); err != nil { + c.logger().WarnContext(ctx, "hlc lease renewal failed", + slog.Uint64("group_id", gid), + slog.Int64("ceiling_ms", ceilingMs), + slog.Any("err", err), + ) + } +} + +func (c *ShardedCoordinator) proposeHLCLeaseToGroup(ctx context.Context, gid uint64, group *ShardGroup, ceilingMs int64) error { start := monoclock.Now() expectedGen := group.lease.generation() // Route through the ShardGroup's wrap-aware proposer chain — NOT a @@ -2430,16 +2474,12 @@ func (c *ShardedCoordinator) renewHLCLease(ctx context.Context, gid uint64, grou if isLeadershipLossError(err) { group.lease.invalidate() } - c.logger().WarnContext(ctx, "hlc lease renewal failed", - slog.Uint64("group_id", gid), - slog.Int64("ceiling_ms", ceilingMs), - slog.Any("err", err), - ) - return + return errors.Wrapf(err, "hlc lease renewal group %d", gid) } if lp, ok := group.Engine.(raftengine.LeaseProvider); ok { group.lease.extend(start.Add(lp.LeaseDuration()), expectedGen) } + return nil } func keyMutations(muts []*pb.Mutation) []*pb.Mutation { diff --git a/kv/tso.go b/kv/tso.go index e07cb34f9..aa968e790 100644 --- a/kv/tso.go +++ b/kv/tso.go @@ -178,6 +178,10 @@ type tsoLeaseRenewer interface { RunHLCLeaseRenewal(context.Context) } +type tsoLeaseProposer interface { + ProposeHLCLease(context.Context, int64) error +} + type LocalTSOAllocator struct { coord tsoCoordinator pollInterval time.Duration @@ -241,9 +245,32 @@ func (a *LocalTSOAllocator) nextBatchAfter(ctx context.Context, n int, min uint6 clock.Observe(min) } base, err := clock.NextBatchFenced(n) + if errors.Is(err, ErrCeilingExpired) { + if renewErr := a.renewExpiredCeiling(ctx); renewErr != nil { + return 0, errors.Wrap(renewErr, "tso renew expired HLC lease") + } + if min > 0 { + clock.Observe(min) + } + base, err = clock.NextBatchFenced(n) + } return base, errors.Wrap(err, "tso next batch") } +func (a *LocalTSOAllocator) renewExpiredCeiling(ctx context.Context) error { + proposer, ok := a.coord.(tsoLeaseProposer) + if !ok { + return errors.WithStack(ErrCeilingExpired) + } + if ctx == nil { + ctx = context.Background() + } + pctx, cancel := context.WithTimeout(ctx, hlcRenewalInterval) + defer cancel() + ceilingMs := time.Now().UnixMilli() + hlcPhysicalWindowMs + return errors.Wrap(proposer.ProposeHLCLease(pctx, ceilingMs), "propose HLC lease") +} + func (a *LocalTSOAllocator) IsLeader() bool { if a == nil || a.coord == nil { return false diff --git a/kv/tso_test.go b/kv/tso_test.go index ef4e58510..84f943963 100644 --- a/kv/tso_test.go +++ b/kv/tso_test.go @@ -193,6 +193,33 @@ func TestLocalTSOAllocatorAcceptsTimestampLeader(t *testing.T) { require.False(t, coord.IsLeader()) } +func TestLocalTSOAllocatorRenewsExpiredCeilingBeforeIssuing(t *testing.T) { + coord := &renewingTSOCoordinator{clock: NewHLC()} + coord.leader.Store(true) + coord.clock.SetPhysicalCeiling(time.Now().Add(-time.Millisecond).UnixMilli()) + alloc, err := NewLocalTSOAllocator(coord, WithTSOLeaderPollInterval(testTSOPollInterval)) + require.NoError(t, err) + + got, err := alloc.Next(context.Background()) + require.NoError(t, err) + require.NotZero(t, got) + require.EqualValues(t, 1, coord.proposeCalls.Load()) + require.Greater(t, coord.clock.PhysicalCeiling(), time.Now().UnixMilli()) +} + +func TestLocalTSOAllocatorKeepsFailClosedWhenRenewalFails(t *testing.T) { + sentinel := errors.New("renewal rejected") + coord := &renewingTSOCoordinator{clock: NewHLC(), proposeErr: sentinel} + coord.leader.Store(true) + coord.clock.SetPhysicalCeiling(time.Now().Add(-time.Millisecond).UnixMilli()) + alloc, err := NewLocalTSOAllocator(coord, WithTSOLeaderPollInterval(testTSOPollInterval)) + require.NoError(t, err) + + _, err = alloc.Next(context.Background()) + require.ErrorIs(t, err, sentinel) + require.EqualValues(t, 1, coord.proposeCalls.Load()) +} + func TestShardedCoordinatorReportsAnyShardAsTimestampLeader(t *testing.T) { engine := distribution.NewEngine() engine.UpdateRoute([]byte("a"), []byte("m"), 1) @@ -452,3 +479,27 @@ func (f *fakeTimestampLeaderCoordinator) IsTimestampLeader() bool { func (f *fakeTimestampLeaderCoordinator) Clock() *HLC { return f.clock } + +type renewingTSOCoordinator struct { + leader atomic.Bool + clock *HLC + proposeErr error + proposeCalls atomic.Uint64 +} + +func (f *renewingTSOCoordinator) IsLeader() bool { + return f.leader.Load() +} + +func (f *renewingTSOCoordinator) Clock() *HLC { + return f.clock +} + +func (f *renewingTSOCoordinator) ProposeHLCLease(_ context.Context, ceilingMs int64) error { + f.proposeCalls.Add(1) + if f.proposeErr != nil { + return f.proposeErr + } + f.clock.SetPhysicalCeiling(ceilingMs) + return nil +} From 1e6ed43a67fed4f732a2ee45634ce2f9f525f1df Mon Sep 17 00:00:00 2001 From: bootjp Date: Thu, 23 Jul 2026 20:10:36 +0900 Subject: [PATCH 3/3] snapshot: verify prewrite retention candidates --- internal/raftengine/etcd/fsm_snapshot_file.go | 34 +------------------ .../raftengine/etcd/fsm_snapshot_file_test.go | 24 +++++++++++++ 2 files changed, 25 insertions(+), 33 deletions(-) diff --git a/internal/raftengine/etcd/fsm_snapshot_file.go b/internal/raftengine/etcd/fsm_snapshot_file.go index 595966e55..574724cae 100644 --- a/internal/raftengine/etcd/fsm_snapshot_file.go +++ b/internal/raftengine/etcd/fsm_snapshot_file.go @@ -831,39 +831,7 @@ func fsmSnapshotPairRestorable(snapDir, fsmSnapDir, snapName string, term, index if !ok { return false } - // Prewrite cleanup runs on the snapshot receive path. Avoid a full payload - // CRC here; startup and actual restore still do full verification before - // trusting the snapshot bytes. - return fsmSnapshotFooterMatchesToken(fsmSnapPath(fsmSnapDir, index), tok.CRC32C) == nil -} - -func fsmSnapshotFooterMatchesToken(path string, tokenCRC uint32) error { - f, err := os.Open(path) - if err != nil { - return statFSMFileError(err) - } - defer f.Close() - - info, err := f.Stat() - if err != nil { - return errors.WithStack(err) - } - if info.Size() < fsmMinFileSize { - return errors.Wrapf(ErrFSMSnapshotTooSmall, - "file too small: %d bytes (minimum %d)", info.Size(), fsmMinFileSize) - } - if _, err := f.Seek(-fsmFooterSize, io.SeekEnd); err != nil { - return errors.WithStack(err) - } - var footer uint32 - if err := binary.Read(f, binary.BigEndian, &footer); err != nil { - return errors.WithStack(err) - } - if footer != tokenCRC { - return errors.Wrapf(ErrFSMSnapshotFileCRC, - "path=%s footer=%08x token=%08x", path, footer, tokenCRC) - } - return nil + return verifyFSMSnapshotFileWithToken(fsmSnapPath(fsmSnapDir, index), tok.CRC32C, true) == nil } func snapshotTokenFromSnapFile(snapDir, snapName string, term, index uint64) (snapshotToken, bool) { diff --git a/internal/raftengine/etcd/fsm_snapshot_file_test.go b/internal/raftengine/etcd/fsm_snapshot_file_test.go index a7d509b44..1c9108f78 100644 --- a/internal/raftengine/etcd/fsm_snapshot_file_test.go +++ b/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()