From 87d5f3e1faf16ff48678a6b8572affe224cd5425 Mon Sep 17 00:00:00 2001 From: Aetherance Date: Sat, 29 Aug 2026 18:44:47 +0800 Subject: [PATCH] feat(string): add keyspace notifications for string commands --- kvrocks.conf | 2 +- src/common/keyspace_events.h | 2 +- src/types/redis_string.cc | 30 +++- tests/cppunit/keyspace_events_test.cc | 2 +- .../keyspacenotify/keyspacenotify_test.go | 133 ++++++++++++++++-- 5 files changed, 152 insertions(+), 17 deletions(-) diff --git a/kvrocks.conf b/kvrocks.conf index 0337b6ab103..973e17995ba 100644 --- a/kvrocks.conf +++ b/kvrocks.conf @@ -617,7 +617,7 @@ lua-strict-key-accessing no # K keyspace channels # E keyevent channels # g generic events, currently del -# $ string events, currently set +# $ string events: set, append, setrange, incrby, incrbyfloat # A same as g$, without K or E # # Default namespace uses db 0. Redis database namespaces use db indexes. diff --git a/src/common/keyspace_events.h b/src/common/keyspace_events.h index 5b4744e70da..03bf7156c52 100644 --- a/src/common/keyspace_events.h +++ b/src/common/keyspace_events.h @@ -36,7 +36,7 @@ enum KeyspaceEventChannel { enum KeyspaceEventType { kNotifyNoType = 0, kNotifyGeneric = 1 << 0, // g, emits del - kNotifyString = 1 << 1, // $, emits set + kNotifyString = 1 << 1, // $, string events // A, supported data classes without K or E. kNotifyAll = kNotifyGeneric | kNotifyString, }; diff --git a/src/types/redis_string.cc b/src/types/redis_string.cc index b385d66cee9..5474203b841 100644 --- a/src/types/redis_string.cc +++ b/src/types/redis_string.cc @@ -131,7 +131,11 @@ rocksdb::Status String::Append(engine::Context &ctx, const std::string &user_key } raw_value.append(value); *new_size = raw_value.size() - Metadata::GetOffsetAfterExpire(raw_value[0]); - return updateRawValue(ctx, ns_key, raw_value); + s = updateRawValue(ctx, ns_key, raw_value); + if (!s.ok()) return s; + + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "append", namespace_, user_key); + return rocksdb::Status::OK(); } std::vector String::MGet(engine::Context &ctx, const std::vector &keys, @@ -410,7 +414,13 @@ rocksdb::Status String::SetRange(engine::Context &ctx, const std::string &user_k } } *new_size = raw_value.size() - header_offset; - return updateRawValue(ctx, ns_key, raw_value); + s = updateRawValue(ctx, ns_key, raw_value); + if (!s.ok()) return s; + + if (!value.empty()) { + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "setrange", namespace_, user_key); + } + return rocksdb::Status::OK(); } rocksdb::Status String::IncrBy(engine::Context &ctx, const std::string &user_key, int64_t increment, @@ -451,7 +461,11 @@ rocksdb::Status String::IncrBy(engine::Context &ctx, const std::string &user_key raw_value = raw_value.substr(0, offset); raw_value.append(std::to_string(n)); - return updateRawValue(ctx, ns_key, raw_value); + s = updateRawValue(ctx, ns_key, raw_value); + if (!s.ok()) return s; + + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "incrby", namespace_, user_key); + return rocksdb::Status::OK(); } rocksdb::Status String::IncrByFloat(engine::Context &ctx, const std::string &user_key, double increment, @@ -485,7 +499,11 @@ rocksdb::Status String::IncrByFloat(engine::Context &ctx, const std::string &use raw_value = raw_value.substr(0, offset); raw_value.append(util::Float2String(n)); - return updateRawValue(ctx, ns_key, raw_value); + s = updateRawValue(ctx, ns_key, raw_value); + if (!s.ok()) return s; + + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "incrbyfloat", namespace_, user_key); + return rocksdb::Status::OK(); } rocksdb::Status String::MSet(engine::Context &ctx, const std::vector &pairs, StringMSetArgs args, @@ -535,6 +553,9 @@ rocksdb::Status String::MSet(engine::Context &ctx, const std::vector s = storage_->Write(ctx, storage_->DefaultWriteOptions(), batch->GetWriteBatch()); if (!s.ok()) return s; + for (const auto &pair : pairs) { + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", namespace_, pair.key.ToStringView()); + } if (flag) *flag = true; return rocksdb::Status::OK(); } @@ -586,6 +607,7 @@ rocksdb::Status String::CAS(engine::Context &ctx, const std::string &user_key, c return write_status; } *flag = 1; + ctx.AddKeyspaceEventIfEnabled(kNotifyString, "set", namespace_, user_key); } return rocksdb::Status::OK(); diff --git a/tests/cppunit/keyspace_events_test.cc b/tests/cppunit/keyspace_events_test.cc index 049a8068c0d..4e555c22fd1 100644 --- a/tests/cppunit/keyspace_events_test.cc +++ b/tests/cppunit/keyspace_events_test.cc @@ -109,7 +109,7 @@ TEST(KeyspaceEvents, ParseFlags) { EXPECT_EQ(type_flags->first, kNotifyNoChannel); EXPECT_EQ(type_flags->second, kNotifyGeneric | kNotifyString); - // KEA enables both channels and set or del. + // KEA enables both channels and every supported event class. auto flags = ParseNotifyKeyspaceEventsFlags("KEA"); ASSERT_TRUE(flags.IsOK()); EXPECT_EQ(flags->first, kNotifyKeyspace | kNotifyKeyevent); diff --git a/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go index 16ecda74521..9d66c2cfa25 100644 --- a/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go +++ b/tests/gocase/unit/keyspacenotify/keyspacenotify_test.go @@ -50,6 +50,12 @@ func expectMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub, chan require.Equal(t, payload, m.Payload) } +func expectEvent(t *testing.T, ctx context.Context, pubsub *redis.PubSub, key, event string) { + t.Helper() + expectMessage(t, ctx, pubsub, "__keyspace@0__:"+key, event) + expectMessage(t, ctx, pubsub, "__keyevent@0__:"+event, key) +} + // expectNoMessage checks that no message arrives soon. func expectNoMessage(t *testing.T, ctx context.Context, pubsub *redis.PubSub) { t.Helper() @@ -77,19 +83,126 @@ func TestKeyspaceNotify(t *testing.T) { expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "foo") }) - t.Run("SETEX publishes set from the shared Set API", func(t *testing.T) { - require.NoError(t, rdb.Do(ctx, "SETEX", "setex-key", 60, "value").Err()) - expectMessage(t, ctx, pubsub, "__keyspace@0__:setex-key", "set") - expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "setex-key") + t.Run("SETEX and PSETEX publish set", func(t *testing.T) { + commands := [][]any{ + {"SETEX", "setex-key", 60, "value"}, + {"PSETEX", "psetex-key", 60000, "value"}, + } + for _, command := range commands { + require.NoError(t, rdb.Do(ctx, command...).Err()) + key := command[1].(string) + expectEvent(t, ctx, pubsub, key, "set") + } + expectNoMessage(t, ctx, pubsub) }) - t.Run("SET NX on existing key publishes nothing", func(t *testing.T) { - require.NoError(t, rdb.Set(ctx, "nxkey", "v1", 0).Err()) - expectMessage(t, ctx, pubsub, "__keyspace@0__:nxkey", "set") - expectMessage(t, ctx, pubsub, "__keyevent@0__:set", "nxkey") + t.Run("SETNX and GETSET publish set after a write", func(t *testing.T) { + cmd := rdb.Do(ctx, "SETNX", "setnx-key", "value") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 1, cmd.Val()) + expectEvent(t, ctx, pubsub, "setnx-key", "set") + + cmd = rdb.Do(ctx, "SETNX", "setnx-key", "skipped") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 0, cmd.Val()) + expectNoMessage(t, ctx, pubsub) + + cmd = rdb.Do(ctx, "GETSET", "setnx-key", "new-value") + require.NoError(t, cmd.Err()) + require.Equal(t, "value", cmd.Val()) + expectEvent(t, ctx, pubsub, "setnx-key", "set") + }) + + t.Run("MSET and MSETNX publish set for every written key", func(t *testing.T) { + require.NoError(t, rdb.Do(ctx, "MSET", "mset-1", "v1", "mset-2", "v2").Err()) + expectEvent(t, ctx, pubsub, "mset-1", "set") + expectEvent(t, ctx, pubsub, "mset-2", "set") + + cmd := rdb.Do(ctx, "MSETNX", "msetnx-1", "v1", "msetnx-2", "v2") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 1, cmd.Val()) + expectEvent(t, ctx, pubsub, "msetnx-1", "set") + expectEvent(t, ctx, pubsub, "msetnx-2", "set") + + cmd = rdb.Do(ctx, "MSETNX", "msetnx-1", "new-value", "msetnx-skipped", "value") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 0, cmd.Val()) + expectNoMessage(t, ctx, pubsub) + }) + + t.Run("MSETEX publishes set for every written key", func(t *testing.T) { + cmd := rdb.Do(ctx, "MSETEX", 2, "msetex-1", "v1", "msetex-2", "v2") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 1, cmd.Val()) + expectEvent(t, ctx, pubsub, "msetex-1", "set") + expectEvent(t, ctx, pubsub, "msetex-2", "set") + expectNoMessage(t, ctx, pubsub) - // NX fails, so nothing is published. - require.NoError(t, rdb.SetNX(ctx, "nxkey", "v2", 0).Err()) + cmd = rdb.Do(ctx, "MSETEX", 2, "msetex-nx-1", "v1", "msetex-nx-2", "v2", "NX") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 1, cmd.Val()) + expectEvent(t, ctx, pubsub, "msetex-nx-1", "set") + expectEvent(t, ctx, pubsub, "msetex-nx-2", "set") + + cmd = rdb.Do(ctx, "MSETEX", 2, "msetex-nx-1", "skipped", "msetex-nx-3", "skipped", "NX") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 0, cmd.Val()) + expectNoMessage(t, ctx, pubsub) + }) + + t.Run("APPEND and SETRANGE publish their command events", func(t *testing.T) { + require.NoError(t, rdb.Append(ctx, "append-key", "value").Err()) + expectEvent(t, ctx, pubsub, "append-key", "append") + require.NoError(t, rdb.Append(ctx, "append-key", "").Err()) + expectEvent(t, ctx, pubsub, "append-key", "append") + + require.NoError(t, rdb.SetRange(ctx, "setrange-key", 2, "value").Err()) + expectEvent(t, ctx, pubsub, "setrange-key", "setrange") + require.NoError(t, rdb.SetRange(ctx, "setrange-key", 0, "").Err()) + expectNoMessage(t, ctx, pubsub) + }) + + t.Run("increment commands publish Redis-compatible events", func(t *testing.T) { + commands := [][]any{ + {"INCR", "incr-key"}, + {"DECR", "decr-key"}, + {"INCRBY", "incrby-key", 2}, + {"DECRBY", "decrby-key", 2}, + {"INCRBYFLOAT", "incrbyfloat-key", 1.5}, + } + for _, command := range commands { + require.NoError(t, rdb.Do(ctx, command...).Err()) + key := command[1].(string) + event := "incrby" + if command[0] == "INCRBYFLOAT" { + event = "incrbyfloat" + } + expectEvent(t, ctx, pubsub, key, event) + } + + require.NoError(t, rdb.Set(ctx, "incr-overflow", "9223372036854775807", 0).Err()) + expectEvent(t, ctx, pubsub, "incr-overflow", "set") + require.Error(t, rdb.Incr(ctx, "incr-overflow").Err()) + expectNoMessage(t, ctx, pubsub) + }) + + t.Run("CAS publishes set only after a write", func(t *testing.T) { + cmd := rdb.Do(ctx, "CAS", "cas-key", "old", "new") + require.NoError(t, cmd.Err()) + require.EqualValues(t, -1, cmd.Val()) + expectNoMessage(t, ctx, pubsub) + + require.NoError(t, rdb.Set(ctx, "cas-key", "old", 0).Err()) + expectEvent(t, ctx, pubsub, "cas-key", "set") + cmd = rdb.Do(ctx, "CAS", "cas-key", "other", "new") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 0, cmd.Val()) + expectNoMessage(t, ctx, pubsub) + + cmd = rdb.Do(ctx, "CAS", "cas-key", "old", "new") + require.NoError(t, cmd.Err()) + require.EqualValues(t, 1, cmd.Val()) + expectEvent(t, ctx, pubsub, "cas-key", "set") expectNoMessage(t, ctx, pubsub) })