This is an automated email from the ASF dual-hosted git repository.
jihuayu pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new b29d8c4e1 test(cluster): cover HFE replication and slot migration
(#3632)
b29d8c4e1 is described below
commit b29d8c4e161b0449d579ccad973aed2519834997
Author: Twice <[email protected]>
AuthorDate: Tue Sep 22 16:01:41 2026 +0800
test(cluster): cover HFE replication and slot migration (#3632)
Adds HFE coverage for full/incremental replication and snapshot/WAL slot
migration, including mixed hash encodings, metadata preservation, and
field expiration after transfer.
Refs #3436.
Assisted-by: Codex
---
.../hash_field_expiration_commands_test.go | 111 +++++++++++
.../slotmigrate/hash_field_expiration_test.go | 207 +++++++++++++++++++++
2 files changed, 318 insertions(+)
diff --git
a/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
b/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
index db3138268..46bf76dde 100644
---
a/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
+++
b/tests/gocase/integration/replication/hash_field_expiration_commands_test.go
@@ -47,6 +47,117 @@ func waitForHashFieldToExpire(t *testing.T, rdb
*redis.Client, ctx context.Conte
}, 5*time.Second, 50*time.Millisecond)
}
+func TestHashFieldExpirationReplication(t *testing.T) {
+ ctx := context.Background()
+ master := util.StartServer(t, util.KvrocksServerConfigs{
+ "hash-encoding-mode": "legacy",
+ "rocksdb.disable_auto_compactions": "yes",
+ "use-rsid-psync": "yes",
+ })
+ defer master.Close()
+ masterClient := master.NewClient()
+ defer func() { require.NoError(t, masterClient.Close()) }()
+
+ replica := util.StartServer(t, util.KvrocksServerConfigs{
+ "hash-encoding-mode": "legacy",
+ "rocksdb.disable_auto_compactions": "yes",
+ "use-rsid-psync": "yes",
+ })
+ defer replica.Close()
+ replicaClient := replica.NewClient()
+ defer func() { require.NoError(t, replicaClient.Close()) }()
+
+ legacyKey := "legacy-hash"
+ binaryValue := "\x00\x00\x00\x00\x00\x00\x00\x00value\xff"
+ require.NoError(t, masterClient.HSet(ctx, legacyKey, "field",
binaryValue).Err())
+ require.NoError(t, masterClient.ConfigSet(ctx, "hash-encoding-mode",
"field-expiration").Err())
+ key := "hfe-hash"
+ require.NoError(t, masterClient.HSet(ctx, key, "persistent",
binaryValue, "live", "10", "expired", "gone").Err())
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+ require.NoError(t, masterClient.Do(ctx, "hpexpireat", key, expireAt,
"FIELDS", 1, "live").Err())
+ expiredValue, err := masterClient.HGet(ctx, key, "expired").Result()
+ require.NoError(t, err)
+ require.Equal(t, "gone", expiredValue)
+ expiredResult, err := masterClient.Do(ctx, "hpexpire", key, 1000,
"FIELDS", 1, "expired").Result()
+ require.NoError(t, err)
+ requireHFEArray(t, expiredResult, int64(1))
+ require.NoError(t, masterClient.PExpireAt(ctx, key,
time.Now().Add(20*time.Minute)).Err())
+ // This field must already be expired on the source before the
checkpoint is taken.
+ waitForHashFieldToExpire(t, masterClient, ctx, key, "expired")
+
+ metadata := util.GetKMetadata(t, masterClient, ctx, key)
+ require.Equal(t, "field-expiration", metadata.Mode)
+ require.Equal(t, int64(3), metadata.Size)
+ require.Equal(t, int64(1), metadata.Persist)
+
+ requireReplicatedHash := func(t *testing.T, values map[string]string,
expires ...interface{}) {
+ t.Helper()
+ util.WaitForOffsetSync(t, masterClient, replicaClient,
5*time.Second)
+ got, err := replicaClient.HGetAll(ctx, key).Result()
+ require.NoError(t, err)
+ require.Equal(t, values, got)
+ fieldExpires, err := replicaClient.Do(ctx, "hpexpiretime", key,
"FIELDS", 3, "persistent", "live", "expired").Result()
+ require.NoError(t, err)
+ requireHFEArray(t, fieldExpires, expires...)
+ require.Equal(t, util.GetKMetadata(t, masterClient, ctx, key),
util.GetKMetadata(t, replicaClient, ctx, key))
+ }
+
+ expiringKey := "hfe-full-sync-expiry"
+ require.NoError(t, masterClient.HSet(ctx, expiringKey, "field",
"value", "keeper", "persistent").Err())
+ fieldExpireAt := time.Now().Add(3 * time.Second).UnixMilli()
+ require.NoError(t, masterClient.Do(ctx, "hpexpireat", expiringKey,
fieldExpireAt, "FIELDS", 1, "field").Err())
+
+ // Distinct replication histories force a checkpoint transfer even
while the source WAL is retained.
+ util.SlaveOf(t, replicaClient, master)
+ util.WaitForSync(t, replicaClient)
+ require.Equal(t, "1", util.FindInfoEntry(masterClient, "sync_full"))
+ requireReplicatedHash(t, map[string]string{"persistent": binaryValue,
"live": "10"}, int64(-1), expireAt, int64(-2))
+ require.Equal(t, metadata, util.GetKMetadata(t, replicaClient, ctx,
key))
+ got, err := replicaClient.HGet(ctx, legacyKey, "field").Result()
+ require.NoError(t, err)
+ require.Equal(t, binaryValue, got)
+ require.Equal(t, "legacy", util.GetKMetadata(t, replicaClient, ctx,
legacyKey).Mode)
+
+ got, err = replicaClient.HGet(ctx, expiringKey, "field").Result()
+ require.NoError(t, err)
+ require.Equal(t, "value", got)
+ fieldExpires, err := replicaClient.Do(ctx, "hpexpiretime", expiringKey,
"FIELDS", 1, "field").Result()
+ require.NoError(t, err)
+ requireHFEArray(t, fieldExpires, fieldExpireAt)
+ waitForHashFieldToExpire(t, replicaClient, ctx, expiringKey, "field")
+ require.GreaterOrEqual(t, time.Now().UnixMilli(), fieldExpireAt)
+ fields, err := replicaClient.HGetAll(ctx, expiringKey).Result()
+ require.NoError(t, err)
+ require.Equal(t, map[string]string{"keeper": "persistent"}, fields)
+
+ // Subsequent updates must use WAL replication without another full
sync.
+ require.NoError(t, masterClient.HIncrBy(ctx, key, "live", 5).Err())
+ requireReplicatedHash(t, map[string]string{"persistent": binaryValue,
"live": "15"}, int64(-1), expireAt, int64(-2))
+ require.Equal(t, metadata, util.GetKMetadata(t, replicaClient, ctx,
key))
+
+ require.NoError(t, masterClient.Do(ctx, "hpexpireat", key, expireAt,
"FIELDS", 1, "persistent").Err())
+ requireReplicatedHash(t, map[string]string{"persistent": binaryValue,
"live": "15"}, expireAt, expireAt, int64(-2))
+ require.NoError(t, masterClient.Do(ctx, "hpersist", key, "FIELDS", 1,
"live").Err())
+ requireReplicatedHash(t, map[string]string{"persistent": binaryValue,
"live": "15"}, expireAt, int64(-1), int64(-2))
+ require.NoError(t, masterClient.HSet(ctx, key, "persistent",
"updated").Err())
+ requireReplicatedHash(t, map[string]string{"persistent": "updated",
"live": "15"}, int64(-1), int64(-1), int64(-2))
+
+ require.NoError(t, masterClient.HSet(ctx, legacyKey, "field",
binaryValue+"updated").Err())
+ util.WaitForOffsetSync(t, masterClient, replicaClient, 5*time.Second)
+ got, err = replicaClient.HGet(ctx, legacyKey, "field").Result()
+ require.NoError(t, err)
+ require.Equal(t, binaryValue+"updated", got)
+ require.Equal(t, "legacy", util.GetKMetadata(t, replicaClient, ctx,
legacyKey).Mode)
+
+ incrementalExpireAt := time.Now().Add(3 * time.Second).UnixMilli()
+ require.NoError(t, masterClient.Do(ctx, "hpexpireat", key,
incrementalExpireAt, "FIELDS", 1, "live").Err())
+ requireReplicatedHash(t, map[string]string{"persistent": "updated",
"live": "15"}, int64(-1), incrementalExpireAt, int64(-2))
+ waitForHashFieldToExpire(t, replicaClient, ctx, key, "live")
+ require.GreaterOrEqual(t, time.Now().UnixMilli(), incrementalExpireAt)
+ requireReplicatedHash(t, map[string]string{"persistent": "updated"},
int64(-1), int64(-2), int64(-2))
+ require.Equal(t, "1", util.FindInfoEntry(masterClient, "sync_full"))
+}
+
func TestHashFieldExpirationHSetExHGetExReplication(t *testing.T) {
configs := util.KvrocksServerConfigs{
"hash-encoding-mode": "field-expiration",
diff --git a/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
b/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
new file mode 100644
index 000000000..1cf489ae3
--- /dev/null
+++ b/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package slotmigrate
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/apache/kvrocks/tests/gocase/util"
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+)
+
+func TestSlotMigrateHashFieldExpiration(t *testing.T) {
+ ctx := context.Background()
+ source := util.StartServer(t, util.KvrocksServerConfigs{
+ "cluster-enabled": "yes",
+ "hash-encoding-mode": "legacy",
+ "rocksdb.disable_auto_compactions": "yes",
+ "migrate-batch-size-kb": "1",
+ })
+ defer source.Close()
+ sourceClient := source.NewClient()
+ defer func() { require.NoError(t, sourceClient.Close()) }()
+ sourceID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx00"
+ require.NoError(t, sourceClient.Do(ctx, "clusterx", "SETNODEID",
sourceID).Err())
+
+ destination := util.StartServer(t, util.KvrocksServerConfigs{
+ "cluster-enabled": "yes",
+ "hash-encoding-mode": "legacy",
+ "rocksdb.disable_auto_compactions": "yes",
+ })
+ defer destination.Close()
+ destinationClient := destination.NewClient()
+ defer func() { require.NoError(t, destinationClient.Close()) }()
+ destinationID := "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx01"
+ require.NoError(t, destinationClient.Do(ctx, "clusterx", "SETNODEID",
destinationID).Err())
+ nodes := fmt.Sprintf("%s %s %d master - 0-10000\n%s %s %d master -
10001-16383",
+ sourceID, source.Host(), source.Port(), destinationID,
destination.Host(), destination.Port())
+ require.NoError(t, sourceClient.Do(ctx, "clusterx", "SETNODES", nodes,
"1").Err())
+ require.NoError(t, destinationClient.Do(ctx, "clusterx", "SETNODES",
nodes, "1").Err())
+
+ for slot, phase := range []string{"snapshot", "wal"} {
+ t.Run(phase, func(t *testing.T) {
+ key := fmt.Sprintf("hfe_{%s}", util.SlotTable[slot])
+ legacyKey := fmt.Sprintf("legacy_{%s}",
util.SlotTable[slot])
+ binaryValue := strings.Repeat("\x00", 8) +
strings.Repeat("value\xff", 512)
+ require.NoError(t, sourceClient.ConfigSet(ctx,
"hash-encoding-mode", "legacy").Err())
+ require.NoError(t, sourceClient.HSet(ctx, legacyKey,
"field", binaryValue).Err())
+ require.NoError(t, sourceClient.ConfigSet(ctx,
"hash-encoding-mode", "field-expiration").Err())
+ require.NoError(t, sourceClient.HSet(ctx, key,
"persistent", binaryValue, "live", "10",
+ "persist", "20", "overwrite", "old", "deleted",
"gone", "expired", "gone").Err())
+ expireAt := time.Now().Add(10 * time.Minute).UnixMilli()
+ require.NoError(t, sourceClient.Do(ctx, "hpexpireat",
key, expireAt, "FIELDS", 4,
+ "live", "persist", "overwrite",
"deleted").Err())
+ expiredValue, err := sourceClient.HGet(ctx, key,
"expired").Result()
+ require.NoError(t, err)
+ require.Equal(t, "gone", expiredValue)
+ expiredResult, err := sourceClient.Do(ctx, "hpexpire",
key, 1000, "FIELDS", 1, "expired").Result()
+ require.NoError(t, err)
+ require.Equal(t, []interface{}{int64(1)}, expiredResult)
+ require.NoError(t, sourceClient.PExpireAt(ctx, key,
time.Now().Add(20*time.Minute)).Err())
+ // This field must already be expired on the source
before migration starts.
+ require.Eventually(t, func() bool {
+ return errors.Is(sourceClient.HGet(ctx, key,
"expired").Err(), redis.Nil)
+ }, 5*time.Second, 10*time.Millisecond)
+
+ values := map[string]string{
+ "persistent": binaryValue, "live": "10",
"persist": "20", "overwrite": "old", "deleted": "gone",
+ }
+ expires := []interface{}{int64(-1), expireAt, expireAt,
expireAt, expireAt, int64(-2)}
+ var fieldExpireAt int64
+ expiringKey, expiringField, expiringValue := key,
"live", "11"
+ if phase == "wal" {
+ // Keep the snapshot transfer running while
subsequent hash changes enter the WAL.
+ require.NoError(t, sourceClient.ConfigSet(ctx,
"migrate-batch-rate-limit-mb", "1").Err())
+ fillerKey := fmt.Sprintf("filler_{%s}",
util.SlotTable[slot])
+ pipe := sourceClient.Pipeline()
+ for i := 0; i < 1024; i++ {
+ pipe.RPush(ctx, fillerKey,
strings.Repeat("x", 4096))
+ }
+ _, err := pipe.Exec(ctx)
+ require.NoError(t, err)
+ require.NoError(t, sourceClient.Do(ctx,
"clusterx", "migrate", slot, destinationID).Err())
+ // Import starts only after the source has
acquired its snapshot.
+ waitForImportState(t, destinationClient, slot,
"start")
+ requireMigrateState(t, sourceClient, slot,
SlotMigrationStateStarted)
+
+ require.NoError(t, sourceClient.HIncrBy(ctx,
key, "live", 5).Err())
+ require.NoError(t, sourceClient.Do(ctx,
"hpexpireat", key, expireAt+60000, "FIELDS", 1, "persistent").Err())
+ require.NoError(t, sourceClient.Do(ctx,
"hpersist", key, "FIELDS", 1, "persist").Err())
+ require.NoError(t, sourceClient.HSet(ctx, key,
"overwrite", "new").Err())
+ require.NoError(t, sourceClient.HDel(ctx, key,
"deleted").Err())
+ require.NoError(t, sourceClient.HSet(ctx,
legacyKey, "field", binaryValue+"updated").Err())
+ require.NoError(t, sourceClient.Do(ctx,
"hsetex", key+"_new", "PXAT", expireAt,
+ "FIELDS", 2, "field",
"created-during-migration", "short-lived", "temporary").Err())
+ // The new key can only arrive through WAL;
leave time for the rate-limited snapshot transfer.
+ fieldExpireAt = time.Now().Add(8 *
time.Second).UnixMilli()
+ require.NoError(t, sourceClient.Do(ctx,
"hpexpireat", key+"_new", fieldExpireAt, "FIELDS", 1, "short-lived").Err())
+ expiringKey, expiringField, expiringValue =
key+"_new", "short-lived", "temporary"
+ values["live"] = "15"
+ values["overwrite"] = "new"
+ delete(values, "deleted")
+ expires = []interface{}{expireAt + 60000,
expireAt, int64(-1), int64(-1), int64(-2), int64(-2)}
+ } else {
+ fieldExpireAt = time.Now().Add(3 *
time.Second).UnixMilli()
+ require.NoError(t, sourceClient.Do(ctx,
"hpexpireat", key, fieldExpireAt, "FIELDS", 1, "live").Err())
+ expires[1] = fieldExpireAt
+ }
+
+ metadata := util.GetKMetadata(t, sourceClient, ctx, key)
+ require.Equal(t, "field-expiration", metadata.Mode)
+ legacyMetadata := util.GetKMetadata(t, sourceClient,
ctx, legacyKey)
+ require.Equal(t, "legacy", legacyMetadata.Mode)
+ if phase == "snapshot" {
+ require.Equal(t, int64(6), metadata.Size)
+ require.Equal(t, int64(1), metadata.Persist)
+ require.NoError(t, sourceClient.Do(ctx,
"clusterx", "migrate", slot, destinationID).Err())
+ }
+ waitForMigrateStateInDuration(t, sourceClient, slot,
SlotMigrationStateSuccess, time.Minute)
+ waitForImportState(t, destinationClient, slot,
SlotImportStateSuccess)
+
+ got, err := destinationClient.HGetAll(ctx, key).Result()
+ require.NoError(t, err)
+ require.Equal(t, values, got)
+ fieldExpires, err := destinationClient.Do(ctx,
"hpexpiretime", key, "FIELDS", 6,
+ "persistent", "live", "persist", "overwrite",
"deleted", "expired").Result()
+ require.NoError(t, err)
+ require.Equal(t, expires, fieldExpires)
+ require.Equal(t, metadata, util.GetKMetadata(t,
destinationClient, ctx, key))
+ require.Equal(t, legacyMetadata, util.GetKMetadata(t,
destinationClient, ctx, legacyKey))
+ legacyValue, err := destinationClient.HGet(ctx,
legacyKey, "field").Result()
+ require.NoError(t, err)
+ if phase == "wal" {
+ require.Equal(t, binaryValue+"updated",
legacyValue)
+ newValue, err := destinationClient.HGetAll(ctx,
key+"_new").Result()
+ require.NoError(t, err)
+ require.Equal(t, map[string]string{"field":
"created-during-migration", "short-lived": "temporary"}, newValue)
+ newExpiry, err := destinationClient.Do(ctx,
"hpexpiretime", key+"_new", "FIELDS", 2, "field", "short-lived").Result()
+ require.NoError(t, err)
+ require.Equal(t, []interface{}{expireAt,
fieldExpireAt}, newExpiry)
+ } else {
+ require.Equal(t, binaryValue, legacyValue)
+ }
+
+ // Writes must use the migrated mode even though the
destination defaults to legacy encoding.
+ wantLive := int64(11)
+ if phase == "wal" {
+ wantLive = 16
+ }
+ newLive, err := destinationClient.HIncrBy(ctx, key,
"live", 1).Result()
+ require.NoError(t, err)
+ require.Equal(t, wantLive, newLive)
+ values["live"] = fmt.Sprint(wantLive)
+ liveValue, err := destinationClient.HGet(ctx, key,
"live").Result()
+ require.NoError(t, err)
+ require.Equal(t, values["live"], liveValue)
+ fieldExpires, err = destinationClient.Do(ctx,
"hpexpiretime", key, "FIELDS", 1, "live").Result()
+ require.NoError(t, err)
+ require.Equal(t, []interface{}{expires[1]},
fieldExpires)
+
+ expiringValueOnDestination, err :=
destinationClient.HGet(ctx, expiringKey, expiringField).Result()
+ require.NoError(t, err)
+ require.Equal(t, expiringValue,
expiringValueOnDestination)
+ fieldExpires, err = destinationClient.Do(ctx,
"hpexpiretime", expiringKey, "FIELDS", 1, expiringField).Result()
+ require.NoError(t, err)
+ require.Equal(t, []interface{}{fieldExpireAt},
fieldExpires)
+ require.Eventually(t, func() bool {
+ return errors.Is(destinationClient.HGet(ctx,
expiringKey, expiringField).Err(), redis.Nil)
+ }, 10*time.Second, 10*time.Millisecond)
+ require.GreaterOrEqual(t, time.Now().UnixMilli(),
fieldExpireAt)
+ if phase == "snapshot" {
+ delete(values, "live")
+ }
+ got, err = destinationClient.HGetAll(ctx, key).Result()
+ require.NoError(t, err)
+ require.Equal(t, values, got)
+ if phase == "wal" {
+ require.ErrorIs(t, destinationClient.HGet(ctx,
key+"_new", "short-lived").Err(), redis.Nil)
+ newValue, err := destinationClient.HGetAll(ctx,
key+"_new").Result()
+ require.NoError(t, err)
+ require.Equal(t, map[string]string{"field":
"created-during-migration"}, newValue)
+ }
+ })
+ }
+}
