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 2aa3efff2 feat(hash): support HFE in WAL-to-RESP conversion (#3634)
2aa3efff2 is described below
commit 2aa3efff250b61f1fe7d6a72fc6fb84368ddcd92
Author: Twice <[email protected]>
AuthorDate: Tue Sep 22 18:57:50 2026 +0800
feat(hash): support HFE in WAL-to-RESP conversion (#3634)
Adds hash encoding context to WAL and preserves field expiration when
converting hash writes and COPY/RENAME to RESP.
Legacy hash LogData remains `2`; HFE hash writes now use `2 13 1`
(`kRedisHash`, `kRedisCmdHSet`, `kFieldExpiration`) to follow the
existing `<type> <operation> <args...>` convention and record the hash's
actual encoding even when a batch contains no metadata update.
Refs #3436; migration batching and kvrocks2redis full export are left
for follow-up PRs.
Assisted-by: Codex / GPT6
---------
Co-authored-by: 纪华裕 <[email protected]>
---
src/storage/batch_extractor.cc | 20 ++-
src/storage/redis_db.cc | 32 +++-
src/storage/redis_db.h | 4 +
src/storage/redis_metadata.h | 1 +
src/types/redis_hash.cc | 33 ++--
tests/cppunit/batch_extractor_test.cc | 101 ++++++++++++
tests/gocase/unit/server/poll_updates_hfe_test.go | 179 ++++++++++++++++++++++
7 files changed, 348 insertions(+), 22 deletions(-)
diff --git a/src/storage/batch_extractor.cc b/src/storage/batch_extractor.cc
index e545b4258..e51fa224c 100644
--- a/src/storage/batch_extractor.cc
+++ b/src/storage/batch_extractor.cc
@@ -133,9 +133,23 @@ rocksdb::Status WriteBatchExtractor::PutCF(uint32_t
column_family_id, const Slic
ns = ikey.GetNamespace().ToString();
switch (log_data_.GetRedisType()) {
- case kRedisHash:
- command_args = {"HSET", user_key, sub_key, value.ToString()};
- break;
+ case kRedisHash: {
+ auto mode = log_data_.GetHashSubkeyEncodingMode();
+ if (!mode) return rocksdb::Status::InvalidArgument(mode.Msg());
+ HashMetadata metadata(false, *mode);
+ Slice field_value = value;
+ uint64_t expire = 0;
+ auto s = metadata.DecodeSubkeyValue(&field_value, &expire);
+ if (!s.ok()) return s;
+
+ resp_commands_[ns].emplace_back(redis::ArrayOfBulkStrings({"HSET",
user_key, sub_key, field_value.ToString()}));
+ // Replay expired writes too: the destination may still contain an
older field value.
+ if (expire > 0) {
+ resp_commands_[ns].emplace_back(
+ redis::ArrayOfBulkStrings({"HPEXPIREAT", user_key,
std::to_string(expire), "FIELDS", "1", sub_key}));
+ }
+ return rocksdb::Status::OK();
+ }
case kRedisList: {
auto args = log_data_.GetArguments();
if (args->empty()) {
diff --git a/src/storage/redis_db.cc b/src/storage/redis_db.cc
index 118468cd9..b8483878d 100644
--- a/src/storage/redis_db.cc
+++ b/src/storage/redis_db.cc
@@ -613,10 +613,30 @@ rocksdb::Status SubKeyScanner::Scan(engine::Context &ctx,
RedisType type, const
return iter->status();
}
+WriteBatchLogData::WriteBatchLogData(HashSubkeyEncodingMode mode) :
type_(kRedisHash) {
+ // Keep legacy logs unchanged. A subkey-only update must describe its own
encoding.
+ if (mode != HashSubkeyEncodingMode::kLegacy) {
+ args_ = {std::to_string(kRedisCmdHSet),
std::to_string(static_cast<uint8_t>(mode))};
+ }
+}
+
RedisType WriteBatchLogData::GetRedisType() const { return type_; }
std::vector<std::string> *WriteBatchLogData::GetArguments() { return &args_; }
+StatusOr<HashSubkeyEncodingMode>
WriteBatchLogData::GetHashSubkeyEncodingMode() const {
+ if (type_ != kRedisHash) return {Status::NotOK, "not a hash log"};
+ if (args_.empty()) return HashSubkeyEncodingMode::kLegacy;
+ if (args_.size() != 2 || args_[0] != std::to_string(kRedisCmdHSet)) {
+ return {Status::NotOK, "invalid hash encoding log"};
+ }
+ auto mode = ParseInt<uint8_t>(args_[1], 10);
+ if (!mode || *mode >
static_cast<uint8_t>(HashSubkeyEncodingMode::kFieldExpiration)) {
+ return {Status::NotOK, "invalid hash subkey encoding mode in log"};
+ }
+ return static_cast<HashSubkeyEncodingMode>(*mode);
+}
+
std::string WriteBatchLogData::Encode() const {
std::string ret = std::to_string(type_);
for (const auto &arg : args_) {
@@ -695,16 +715,22 @@ rocksdb::Status Database::Copy(engine::Context &ctx,
const std::string &key, con
if (key == new_key) return rocksdb::Status::OK();
+ engine::DBIterator iter(ctx, ctx.GetReadOptions());
+ iter.Seek(key);
+
auto batch = storage_->GetWriteBatchBase();
WriteBatchLogData log_data(type);
+ if (type == kRedisHash) {
+ HashMetadata metadata(false);
+ s = metadata.Decode(iter.Value());
+ if (!s.ok()) return s;
+ log_data = WriteBatchLogData(metadata.mode);
+ }
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) {
return s;
}
- engine::DBIterator iter(ctx, ctx.GetReadOptions());
- iter.Seek(key);
-
if (delete_old) {
s = batch->Delete(metadata_cf_handle_, key);
if (!s.ok()) {
diff --git a/src/storage/redis_db.h b/src/storage/redis_db.h
index 506dc250b..d24547bf9 100644
--- a/src/storage/redis_db.h
+++ b/src/storage/redis_db.h
@@ -189,9 +189,13 @@ class WriteBatchLogData {
WriteBatchLogData() = default;
explicit WriteBatchLogData(RedisType type) : type_(type) {}
explicit WriteBatchLogData(RedisType type, std::vector<std::string> &&args)
: type_(type), args_(std::move(args)) {}
+ // Non-legacy hashes record HSET and the encoding mode. Unmarked logs retain
legacy semantics;
+ // HFE WAL written before this marker cannot be replayed without a new full
synchronization.
+ explicit WriteBatchLogData(HashSubkeyEncodingMode mode);
RedisType GetRedisType() const;
std::vector<std::string> *GetArguments();
+ StatusOr<HashSubkeyEncodingMode> GetHashSubkeyEncodingMode() const;
std::string Encode() const;
Status Decode(const rocksdb::Slice &blob);
diff --git a/src/storage/redis_metadata.h b/src/storage/redis_metadata.h
index 5ad5677a1..3f9a793cb 100644
--- a/src/storage/redis_metadata.h
+++ b/src/storage/redis_metadata.h
@@ -99,6 +99,7 @@ enum RedisCommand {
kRedisCmdBitOp,
kRedisCmdBitfield,
kRedisCmdLMove,
+ kRedisCmdHSet,
};
constexpr const char *kErrMsgWrongType = "WRONGTYPE Operation against a key
holding the wrong kind of value";
diff --git a/src/types/redis_hash.cc b/src/types/redis_hash.cc
index 5ba98b1f8..5020ecca7 100644
--- a/src/types/redis_hash.cc
+++ b/src/types/redis_hash.cc
@@ -221,7 +221,7 @@ bool HashMetadataEqual(const HashMetadata &lhs, const
HashMetadata &rhs) {
class HashBatchWriter {
public:
- explicit HashBatchWriter(rocksdb::WriteBatchBase *batch) : batch_(batch) {}
+ HashBatchWriter(rocksdb::WriteBatchBase *batch, HashSubkeyEncodingMode mode)
: batch_(batch), mode_(mode) {}
rocksdb::Status Put(const Slice &key, const Slice &value) {
auto s = ensureLogData();
@@ -260,13 +260,14 @@ class HashBatchWriter {
private:
rocksdb::Status ensureLogData() {
if (has_log_data_) return rocksdb::Status::OK();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(mode_);
auto s = batch_->PutLogData(log_data.Encode());
if (s.ok()) has_log_data_ = true;
return s;
}
rocksdb::WriteBatchBase *batch_;
+ HashSubkeyEncodingMode mode_;
bool has_log_data_ = false;
bool storage_mutated_ = false;
};
@@ -381,7 +382,7 @@ rocksdb::Status Hash::Size(engine::Context &ctx, const
Slice &user_key, uint64_t
}
if (metadata.upper != 0 && now > metadata.upper && metadata.persist == 0) {
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
s = batch->Delete(metadata_cf_handle_, ns_key);
@@ -409,7 +410,7 @@ rocksdb::Status Hash::scanAndRepair(engine::Context &ctx,
const Slice &ns_key, H
repaired_metadata.upper = 0;
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata->mode);
auto s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
@@ -522,7 +523,7 @@ rocksdb::Status Hash::IncrBy(engine::Context &ctx, const
Slice &user_key, const
*new_value = old_value + increment;
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
std::string encoded_value =
metadata.EncodeSubkeyValue(std::to_string(*new_value), keep_expire);
@@ -582,7 +583,7 @@ rocksdb::Status Hash::IncrByFloat(engine::Context &ctx,
const Slice &user_key, c
*new_value = n;
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
std::string encoded_value =
metadata.EncodeSubkeyValue(util::Float2String(*new_value), keep_expire);
@@ -715,7 +716,7 @@ rocksdb::Status Hash::SetFieldsWithExpire(engine::Context
&ctx, const Slice &use
}
auto batch = storage_->GetWriteBatchBase();
- HashBatchWriter writer(batch.Get());
+ HashBatchWriter writer(batch.Get(), metadata.mode);
for (const auto &sub_key : expired_cleanup_keys) {
s = writer.Delete(sub_key);
if (!s.ok()) return s;
@@ -860,7 +861,7 @@ rocksdb::Status Hash::GetFieldsWithExpire(engine::Context
&ctx, const Slice &use
if (!s.ok()) return s;
auto batch = storage_->GetWriteBatchBase();
- HashBatchWriter writer(batch.Get());
+ HashBatchWriter writer(batch.Get(), metadata.mode);
auto make_sub_key = [&](const std::string &field) {
return InternalKey(ns_key, field, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
};
@@ -957,14 +958,14 @@ rocksdb::Status Hash::Delete(engine::Context &ctx, const
Slice &user_key, const
std::string ns_key = AppendNamespacePrefix(user_key);
HashMetadata metadata(false);
+ auto s = getMetadata(ctx, ns_key, &metadata);
+ if (!s.ok()) return s.IsNotFound() ? rocksdb::Status::OK() : s;
+
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
- auto s = batch->PutLogData(log_data.Encode());
+ WriteBatchLogData log_data(metadata.mode);
+ s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
- s = getMetadata(ctx, ns_key, &metadata);
- if (!s.ok()) return s.IsNotFound() ? rocksdb::Status::OK() : s;
-
uint64_t physical_removed = 0;
uint64_t persistent_removed = 0;
std::unordered_set<std::string_view> field_set;
@@ -1032,7 +1033,7 @@ rocksdb::Status Hash::MSet(engine::Context &ctx, const
Slice &user_key, const st
int added = 0;
bool metadata_changed = ttl_updated;
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
std::unordered_set<std::string_view> field_set;
@@ -1371,7 +1372,7 @@ rocksdb::Status Hash::ExpireFields(engine::Context &ctx,
const Slice &user_key,
}
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
@@ -1465,7 +1466,7 @@ rocksdb::Status Hash::PersistFields(engine::Context &ctx,
const Slice &user_key,
}
auto batch = storage_->GetWriteBatchBase();
- WriteBatchLogData log_data(kRedisHash);
+ WriteBatchLogData log_data(metadata.mode);
s = batch->PutLogData(log_data.Encode());
if (!s.ok()) return s;
diff --git a/tests/cppunit/batch_extractor_test.cc
b/tests/cppunit/batch_extractor_test.cc
new file mode 100644
index 000000000..06b121e87
--- /dev/null
+++ b/tests/cppunit/batch_extractor_test.cc
@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+
+#include "storage/batch_extractor.h"
+
+#include <gtest/gtest.h>
+
+#include "server/redis_reply.h"
+
+TEST(WriteBatchLogData, HashEncoding) {
+ redis::WriteBatchLogData legacy(HashSubkeyEncodingMode::kLegacy);
+ redis::WriteBatchLogData hfe(HashSubkeyEncodingMode::kFieldExpiration);
+ EXPECT_EQ(legacy.Encode(), "2");
+ EXPECT_EQ(hfe.Encode(), "2 13 1");
+
+ redis::WriteBatchLogData decoded;
+ for (const auto &log : {hfe, legacy}) {
+ ASSERT_TRUE(decoded.Decode(log.Encode()).IsOK());
+ auto mode = decoded.GetHashSubkeyEncodingMode();
+ ASSERT_TRUE(mode.IsOK());
+ EXPECT_EQ(*mode, *log.GetHashSubkeyEncodingMode());
+ }
+}
+
+class HashBatchExtractorTest : public ::testing::TestWithParam<bool> {};
+
+TEST_P(HashBatchExtractorTest, MixedEncodingsWithoutMetadata) {
+ bool slot_encoded = GetParam();
+ HashMetadata hfe(false, HashSubkeyEncodingMode::kFieldExpiration);
+ auto ns_key = ComposeNamespaceKey("ns", "hash", slot_encoded);
+ auto key = InternalKey(ns_key, "field", 1, slot_encoded).Encode();
+ const std::string binary_value("\0value\xff", 7);
+ const auto legacy_value = hfe.EncodeSubkeyValue(binary_value, 123);
+ rocksdb::WriteBatch batch;
+ std::vector<std::string> expected;
+
+ // Include an already-expired timestamp: replay must remove any older
destination value.
+ for (uint64_t expire : {0ULL, 1ULL, 4000000000000ULL}) {
+
ASSERT_TRUE(batch.PutLogData(redis::WriteBatchLogData(hfe.mode).Encode()).ok());
+ ASSERT_TRUE(batch.Put(key, hfe.EncodeSubkeyValue(binary_value,
expire)).ok());
+ expected.emplace_back(redis::ArrayOfBulkStrings({"HSET", "hash", "field",
binary_value}));
+ if (expire != 0) {
+ expected.emplace_back(
+ redis::ArrayOfBulkStrings({"HPEXPIREAT", "hash",
std::to_string(expire), "FIELDS", "1", "field"}));
+ }
+
+ // An old-format log must reset the mode, even if its value looks like HFE
data.
+
ASSERT_TRUE(batch.PutLogData(redis::WriteBatchLogData(kRedisHash).Encode()).ok());
+ ASSERT_TRUE(batch.Put(key, legacy_value).ok());
+ expected.emplace_back(redis::ArrayOfBulkStrings({"HSET", "hash", "field",
legacy_value}));
+ }
+
ASSERT_TRUE(batch.PutLogData(redis::WriteBatchLogData(hfe.mode).Encode()).ok());
+ ASSERT_TRUE(batch.Put(key, hfe.EncodeSubkeyValue("")).ok());
+ expected.emplace_back(redis::ArrayOfBulkStrings({"HSET", "hash", "field",
""}));
+ ASSERT_TRUE(batch.Delete(key).ok());
+ expected.emplace_back(redis::ArrayOfBulkStrings({"HDEL", "hash", "field"}));
+
+ WriteBatchExtractor extractor(slot_encoded, -1, true);
+ ASSERT_TRUE(batch.Iterate(&extractor).ok());
+ ASSERT_EQ(extractor.GetRESPCommands()->size(), 1);
+ EXPECT_EQ(extractor.GetRESPCommands()->at("ns"), expected);
+}
+
+TEST_P(HashBatchExtractorTest, InvalidHashEncoding) {
+ auto ns_key = ComposeNamespaceKey("ns", "hash", GetParam());
+ auto key = InternalKey(ns_key, "field", 1, GetParam()).Encode();
+ for (const auto &log : {"2 13", "2 13 2", "2 13 -1", "2 13 invalid", "2 13 1
extra"}) {
+ SCOPED_TRACE(log);
+ rocksdb::WriteBatch batch;
+ ASSERT_TRUE(batch.PutLogData(log).ok());
+ ASSERT_TRUE(batch.Put(key, std::string(8, '\0')).ok());
+ WriteBatchExtractor extractor(GetParam(), -1, true);
+ EXPECT_TRUE(batch.Iterate(&extractor).IsInvalidArgument());
+ EXPECT_TRUE(extractor.GetRESPCommands()->empty());
+ }
+
+ rocksdb::WriteBatch batch;
+
ASSERT_TRUE(batch.PutLogData(redis::WriteBatchLogData(HashSubkeyEncodingMode::kFieldExpiration).Encode()).ok());
+ ASSERT_TRUE(batch.Put(key, std::string(7, '\0')).ok());
+ WriteBatchExtractor extractor(GetParam(), -1, true);
+ EXPECT_TRUE(batch.Iterate(&extractor).IsInvalidArgument());
+ EXPECT_TRUE(extractor.GetRESPCommands()->empty());
+}
+
+INSTANTIATE_TEST_SUITE_P(SlotEncoding, HashBatchExtractorTest,
::testing::Bool());
diff --git a/tests/gocase/unit/server/poll_updates_hfe_test.go
b/tests/gocase/unit/server/poll_updates_hfe_test.go
new file mode 100644
index 000000000..79d4c4a2f
--- /dev/null
+++ b/tests/gocase/unit/server/poll_updates_hfe_test.go
@@ -0,0 +1,179 @@
+/*
+ * 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 server
+
+import (
+ "context"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/redis/go-redis/v9"
+ "github.com/stretchr/testify/require"
+
+ "github.com/apache/kvrocks/tests/gocase/util"
+)
+
+func TestPollUpdatesHashFieldExpiration(t *testing.T) {
+ for _, txn := range []string{"no", "yes"} {
+ ctx := context.Background()
+ source := util.StartServer(t,
map[string]string{"txn-context-enabled": txn, "hash-encoding-mode":
"field-expiration"})
+ defer source.Close()
+ destination := util.StartServer(t,
map[string]string{"hash-encoding-mode": "field-expiration"})
+ defer destination.Close()
+ src, dst := source.NewClient(), destination.NewClient()
+ defer func() { require.NoError(t, src.Close()) }()
+ defer func() { require.NoError(t, dst.Close()) }()
+ t.Run("txn-context="+txn, func(t *testing.T) {
+ var sequence int64
+ replay := func(t *testing.T, expected ...[]string) {
+ t.Helper()
+ result, err := src.Do(ctx, "POLLUPDATES",
sequence, "MAX", 1000, "FORMAT", "RESP").Result()
+ require.NoError(t, err)
+ updates := parsePollUpdatesResult(t,
result.(map[any]any), true)
+ require.Equal(t, updates.LatestSeq,
updates.NextSeq)
+ require.Equal(t, []any{RESPFormat{Namespace:
"default", Commands: expected}}, updates.Updates)
+ sequence = updates.NextSeq
+ for _, command := range
updates.Updates[0].(RESPFormat).Commands {
+ args := make([]any, len(command))
+ for i, arg := range command {
+ args[i] = arg
+ }
+ require.NoError(t, dst.Do(ctx,
args...).Err())
+ }
+ }
+ set := func(key, field, value string) []string { return
[]string{"HSET", key, field, value} }
+ expire := func(key, field string, at int64) []string {
+ return []string{"HPEXPIREAT", key,
strconv.FormatInt(at, 10), "FIELDS", "1", field}
+ }
+ checkField := func(t *testing.T, key, field, value
string, at int64) {
+ t.Helper()
+ gotValue, err := dst.HGet(ctx, key,
field).Result()
+ require.NoError(t, err)
+ require.Equal(t, value, gotValue)
+ got, err := dst.Do(ctx, "HPEXPIRETIME", key,
"FIELDS", 1, field).Int64Slice()
+ require.NoError(t, err)
+ require.Equal(t, []int64{at}, got)
+ }
+
+ t.Run("mixed encodings in a transaction", func(t
*testing.T) {
+ binary :=
"\x00\x00\x00\x00\x00\x00\x00\x01payload\x00\xff"
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "legacy").Err())
+ require.NoError(t, src.HSet(ctx, "legacy", "f",
binary).Err())
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "field-expiration").Err())
+ require.NoError(t, src.HSet(ctx, "hfe", "f",
"initial").Err())
+ replay(t, set("legacy", "f", binary),
set("hfe", "f", "initial"))
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "legacy").Err())
+ _, err := src.TxPipelined(ctx, func(pipe
redis.Pipeliner) error {
+ pipe.HSet(ctx, "hfe", "f", "updated")
+ pipe.HSet(ctx, "legacy", "f",
binary+"updated")
+ return nil
+ })
+ require.NoError(t, err)
+ replay(t, set("hfe", "f", "updated"),
set("legacy", "f", binary+"updated"))
+ checkField(t, "hfe", "f", "updated", -1)
+ checkField(t, "legacy", "f", binary+"updated",
-1)
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "field-expiration").Err())
+ })
+
+ t.Run("field TTL transitions and subkey-only updates",
func(t *testing.T) {
+ at := time.Now().Add(time.Minute).UnixMilli()
+ for _, step := range []struct {
+ args []any
+ value string
+ ttl int64
+ }{
+ {[]any{"HSET", "counter", "f", "10"},
"10", -1},
+ {[]any{"HPEXPIREAT", "counter", at,
"FIELDS", 1, "f"}, "10", at},
+ {[]any{"HINCRBY", "counter", "f", 1},
"11", at},
+ {[]any{"HINCRBYFLOAT", "counter", "f",
0.5}, "11.5", at},
+ {[]any{"HSETEX", "counter", "KEEPTTL",
"FIELDS", 1, "f", "12"}, "12", at},
+ {[]any{"HGETEX", "counter", "PXAT", at
+ 1000, "FIELDS", 1, "f"}, "12", at + 1000},
+ {[]any{"HPERSIST", "counter", "FIELDS",
1, "f"}, "12", -1},
+ {[]any{"HSETEX", "counter", "PXAT", at,
"FIELDS", 1, "f", "13"}, "13", at},
+ {[]any{"HGETEX", "counter", "PERSIST",
"FIELDS", 1, "f"}, "13", -1},
+ {[]any{"HPEXPIREAT", "counter", at,
"FIELDS", 1, "f"}, "13", at},
+ {[]any{"HSET", "counter", "f", "14"},
"14", -1},
+ } {
+ require.NoError(t, src.Do(ctx,
step.args...).Err())
+ commands := [][]string{set("counter",
"f", step.value)}
+ if step.ttl > 0 {
+ commands = append(commands,
expire("counter", "f", step.ttl))
+ }
+ replay(t, commands...)
+ checkField(t, "counter", "f",
step.value, step.ttl)
+ }
+ require.NoError(t, src.HDel(ctx, "counter",
"f").Err())
+ replay(t, []string{"HDEL", "counter", "f"})
+ require.ErrorIs(t, dst.HGet(ctx, "counter",
"f").Err(), redis.Nil)
+ })
+
+ t.Run("key TTL and hash log arguments coexist", func(t
*testing.T) {
+ // Legacy metadata stores key expiration with
second precision.
+ at :=
time.Now().Add(time.Minute).Truncate(time.Second).UnixMilli()
+ require.NoError(t, src.HSet(ctx, "key-ttl",
"f", "v").Err())
+ require.NoError(t, src.Do(ctx, "PEXPIREAT",
"key-ttl", at).Err())
+ require.NoError(t, src.HSet(ctx, "key-ttl",
"g", "w").Err())
+ replay(t, set("key-ttl", "f", "v"),
[]string{"PEXPIREAT", "key-ttl", strconv.FormatInt(at, 10)}, set("key-ttl",
"g", "w"))
+ require.Equal(t, at, dst.Do(ctx, "PEXPIRETIME",
"key-ttl").Val())
+ checkField(t, "key-ttl", "g", "w", -1)
+ })
+
+ t.Run("copy and rename preserve field TTL", func(t
*testing.T) {
+ at := time.Now().Add(time.Minute).UnixMilli()
+ require.NoError(t, src.Do(ctx, "HSETEX",
"original", "PXAT", at, "FIELDS", 1, "f", "value").Err())
+ replay(t, set("original", "f", "value"),
expire("original", "f", at))
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "legacy").Err())
+ require.EqualValues(t, 1, src.Copy(ctx,
"original", "copied", 0, false).Val())
+ replay(t, set("copied", "f", "value"),
expire("copied", "f", at))
+ checkField(t, "copied", "f", "value", at)
+ require.NoError(t, src.Rename(ctx, "copied",
"renamed").Err())
+ replay(t, []string{"DEL", "copied"},
set("renamed", "f", "value"), expire("renamed", "f", at))
+ checkField(t, "renamed", "f", "value", at)
+ exists, err := dst.Exists(ctx,
"copied").Result()
+ require.NoError(t, err)
+ require.Zero(t, exists)
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "field-expiration").Err())
+ })
+
+ t.Run("natural expiration and delayed replay", func(t
*testing.T) {
+ at := time.Now().Add(3 *
time.Second).UnixMilli()
+ require.NoError(t, src.Do(ctx, "HSETEX",
"expiry", "PXAT", at, "FIELDS", 2, "live", "v", "persist", "p").Err())
+ replay(t, set("expiry", "live", "v"),
expire("expiry", "live", at), set("expiry", "persist", "p"), expire("expiry",
"persist", at))
+ checkField(t, "expiry", "live", "v", at)
+ checkField(t, "expiry", "persist", "p", at)
+ require.NoError(t, src.Do(ctx, "HPERSIST",
"expiry", "FIELDS", 1, "persist").Err())
+ replay(t, set("expiry", "persist", "p"))
+ checkField(t, "expiry", "persist", "p", -1)
+
+ // Leave this source write unread until its
deadline has passed.
+ require.NoError(t, src.Do(ctx, "HSETEX",
"expiry", "PXAT", at, "FIELDS", 1, "delayed", "new").Err())
+ require.NoError(t, dst.HSet(ctx, "expiry",
"delayed", "old").Err())
+ checkField(t, "expiry", "delayed", "old", -1)
+ require.Eventually(t, func() bool {
+ return time.Now().UnixMilli() > at &&
dst.HGet(ctx, "expiry", "live").Err() == redis.Nil
+ }, 10*time.Second, 10*time.Millisecond)
+ checkField(t, "expiry", "persist", "p", -1)
+ replay(t, set("expiry", "delayed", "new"),
expire("expiry", "delayed", at))
+ require.ErrorIs(t, dst.HGet(ctx, "expiry",
"delayed").Err(), redis.Nil)
+ })
+ })
+ }
+}
