Skip to content

feat(keyspace): add keyspace notifications for string commands #3600

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
Aetherance wants to merge 1 commit into
base: unstable
Choose a base branch
from
Draft

feat(keyspace): add keyspace notifications for string commands #3600

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion kvrocks.conf
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
2 changes: 1 addition & 1 deletion src/common/keyspace_events.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
};
Expand Down
30 changes: 26 additions & 4 deletions src/types/redis_string.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<rocksdb::Status> String::MGet(engine::Context &ctx, const std::vector<Slice> &keys,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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<StringPair> &pairs, StringMSetArgs args,
Expand Down Expand Up @@ -535,6 +553,9 @@ rocksdb::Status String::MSet(engine::Context &ctx, const std::vector<StringPair>
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();
}
Expand Down Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion tests/cppunit/keyspace_events_test.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down
133 changes: 123 additions & 10 deletions tests/gocase/unit/keyspacenotify/keyspacenotify_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand Down Expand Up @@ -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)
})

Expand Down