feat(keyspace): emit set and del notifications

kvrocks.conf

@@ -610,6 +610,23 @@ lua-strict-key-accessing no
 #
 # tls-replication yes
 
+############################# KEYSPACE NOTIFICATIONS ##########################
+
+# Keyspace notifications publish key changes to SUBSCRIBE and PSUBSCRIBE clients.
+# Supported flags:
+#   K  keyspace channels
+#   E  keyevent channels
+#   g  generic events, currently del
+#   $  string events, currently set
+#   A  same as g$, without K or E
+#
+# Default namespace uses db 0. Redis database namespaces use db indexes.
+# Other namespaces use ns:encoded-name.
+# Notifications are emitted only when at least one channel flag (K or E) and one event class are enabled.
+#
+# Default: "" disabled
+notify-keyspace-events ""
+
 ################################## SLOW LOG ###################################
 
 # The Kvrocks Slow Log is a mechanism to log queries that exceeded a specified

src/commands/cmd_key.cc

@@ -19,9 +19,11 @@
  */
 
 #include <cstdint>
+#include <string_view>
 
 #include "commander.h"
 #include "commands/ttl_util.h"
+#include "common/keyspace_events.h"
 #include "error_constants.h"
 #include "server/redis_reply.h"
 #include "server/server.h"
@@ -372,9 +374,17 @@ class CommandDel : public Commander {
     uint64_t cnt = 0;
     redis::Database redis(srv->storage, conn->GetNamespace());
 
-    auto s = redis.MDel(ctx, keys, &cnt);
+    const bool notify_del = GetAttributes()->name == "del" && conn->IsKeyspaceEventEnabled(kNotifyGeneric);
+    std::vector<rocksdb::Slice> deleted_keys;
+    if (notify_del) deleted_keys.reserve(keys.size());
+
+    auto s = redis.MDel(ctx, keys, &cnt, notify_del ? &deleted_keys : nullptr);
     if (!s.ok()) return {Status::RedisExecErr, s.ToString()};
 
+    for (const auto &key : deleted_keys) {
+      conn->AddKeyspaceEvent(kNotifyGeneric, "del", std::string_view(key.data(), key.size()));
+    }
+
     *output = redis::Integer(cnt);
     return Status::OK();
   }

src/commands/cmd_string.cc

@@ -24,6 +24,8 @@
 
 #include "commander.h"
 #include "commands/command_parser.h"
+#include "common/keyspace_events.h"
+#include "common/string_util.h"
 #include "error_constants.h"
 #include "server/redis_reply.h"
 #include "server/redis_request.h"
@@ -383,13 +385,17 @@ class CommandSet : public Commander {
   Status Execute(engine::Context &ctx, Server *srv, Connection *conn, std::string *output) override {
     std::optional<std::string> ret;
     redis::String string_db(srv->storage, conn->GetNamespace());
+    const StringSetArgs set_args{expire_, set_flag_, get_, keep_ttl_, cmp_value_};
+    bool set_applied = false;
 
-    rocksdb::Status s = string_db.Set(ctx, args_[1], args_[2], {expire_, set_flag_, get_, keep_ttl_, cmp_value_}, ret);
+    rocksdb::Status s = string_db.Set(ctx, args_[1], args_[2], set_args, ret, &set_applied);
 
     if (!s.ok()) {
       return {Status::RedisExecErr, s.ToString()};
     }
 
+    if (set_applied) conn->AddKeyspaceEvent(kNotifyString, "set", args_[1]);
+
     if (get_) {
       if (ret.has_value()) {
         *output = redis::BulkString(ret.value());

src/commands/cmd_txn.cc

@@ -90,6 +90,11 @@ class CommandExec : public Commander {
       s = storage->CommitTxn();
     }
 
+    // Publish queued notifications after a successful commit.
+    if (s.IsOK()) {
+      conn->FlushKeyspaceEvents();
+    }
+
     conn->ResetMultiExec();
     reset_multiexec.Disable();
 

src/common/keyspace_events.cc

@@ -0,0 +1,96 @@
+/*
+ * 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 "keyspace_events.h"
+
+#include <cstring>
+#include <utility>
+
+#include "config/config.h"
+#include "fmt/format.h"
+
+bool ShouldNotifyKeyspaceEvent(int notify_flags, int type_flag) {
+  return (notify_flags & type_flag) != 0 && (notify_flags & (kNotifyKeyspace | kNotifyKeyevent)) != 0;
+}
+
+KeyspaceEventCollector::KeyspaceEventCollector(std::string ns, int notify_flags)
+    : notify_flags_(notify_flags), ns_(std::move(ns)) {}
+
+void KeyspaceEventCollector::Add(int type_flag, std::string_view event, std::string_view key) {
+  if (!ShouldNotifyKeyspaceEvent(notify_flags_, type_flag)) return;
+  const int channel_flags = notify_flags_ & (kNotifyKeyspace | kNotifyKeyevent);
+  events_.emplace_back(KeyspaceEvent{channel_flags, std::string(event), ns_, std::string(key)});
+}
+
+std::vector<KeyspaceEvent> KeyspaceEventCollector::Take() { return std::move(events_); }
+
+Status ParseNotifyKeyspaceEventsFlags(const std::string &input, int *flags) {
+  int result = 0;
+  for (const char c : input) {
+    switch (c) {
+      case 'K':
+        result |= kNotifyKeyspace;
+        break;
+      case 'E':
+        result |= kNotifyKeyevent;
+        break;
+      case 'A':
+        result |= kNotifyAll;
+        break;
+      case 'g':
+        result |= kNotifyGeneric;
+        break;
+      case '$':
+        result |= kNotifyString;
+        break;
+      default:
+        return {Status::NotOK, fmt::format("unsupported notify-keyspace-events flag: '{}'", c)};
+    }
+  }
+
+  *flags = result;
+  return Status::OK();
+}
+
+namespace {
+std::string PercentEncode(const std::string &input) {
+  std::string output;
+  output.reserve(input.size());
+  for (const unsigned char c : input) {
+    if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '_' || c == '.' ||
+        c == '-') {
+      output += static_cast<char>(c);
+    } else {
+      output += fmt::format("%{:02X}", c);
+    }
+  }
+  return output;
+}
+}  // namespace
+
+std::string MapNamespaceToKeyspaceDB(const std::string &ns, int redis_databases) {
+  if (ns == kDefaultNamespace) {
+    return "0";
+  }
+  if (redis_databases > 0 && ns.rfind(kDatabaseNamespacePrefix, 0) == 0) {
+    return ns.substr(strlen(kDatabaseNamespacePrefix));
+  }
+  return "ns:" + PercentEncode(ns);
+}

src/common/keyspace_events.h

@@ -0,0 +1,68 @@
+/*
+ * 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.
+ *
+ */
+
+#pragma once
+
+#include <string>
+#include <string_view>
+#include <vector>
+
+#include "status.h"
+
+// Flags for notify-keyspace-events, separate from RedisType.
+enum NotifyKeyspaceEventFlag {
+  kNotifyKeyspace = 1 << 0,  // K, keyspace channels
+  kNotifyKeyevent = 1 << 1,  // E, keyevent channels
+  kNotifyGeneric = 1 << 2,   // g, emits del
+  kNotifyString = 1 << 3,    // $, emits set
+  // A, supported data classes without K or E.
+  kNotifyAll = kNotifyGeneric | kNotifyString,
+};
+
+bool ShouldNotifyKeyspaceEvent(int notify_flags, int type_flag);
+
+struct KeyspaceEvent {
+  int channel_flags;
+  std::string event;
+  std::string ns;
+  std::string key;
+};
+
+// Collects semantic keyspace events for one command; Connection owns publish timing.
+class KeyspaceEventCollector {
+ public:
+  KeyspaceEventCollector(std::string ns, int notify_flags);
+  void Add(int type_flag, std::string_view event, std::string_view key);
+  // Moves out events collected during Execute.
+  std::vector<KeyspaceEvent> Take();
+
+ private:
+  int notify_flags_;
+  std::string ns_;
+  std::vector<KeyspaceEvent> events_;
+};
+
+// Parses notify-keyspace-events flags.
+Status ParseNotifyKeyspaceEventsFlags(const std::string &input, int *flags);
+
+// Maps namespaces to notification db names.
+// Default namespace maps to 0; database namespaces map back to db indexes when redis-databases is enabled.
+// Other namespaces map to ns:encoded-name.
+std::string MapNamespaceToKeyspaceDB(const std::string &ns, int redis_databases);

src/config/config.cc

@@ -35,6 +35,7 @@
 #include <utility>
 #include <vector>
 
+#include "common/keyspace_events.h"
 #include "common/string_util.h"
 #include "config_type.h"
 #include "config_util.h"
@@ -192,6 +193,7 @@ Config::Config() {
       {"compact-cron", false, new StringField(&compact_cron_str_, "")},
       {"bgsave-cron", false, new StringField(&bgsave_cron_str_, "")},
       {"dbsize-scan-cron", false, new StringField(&dbsize_scan_cron_str_, "")},
+      {"notify-keyspace-events", false, new StringField(&notify_keyspace_events_str_, "")},
       {"replica-announce-ip", false, new StringField(&replica_announce_ip, "")},
       {"replica-announce-port", false, new UInt32Field(&replica_announce_port, 0, 0, PORT_LIMIT)},
       {"compaction-checker-range", false, new StringField(&compaction_checker_range_str_, "")},
@@ -379,6 +381,11 @@ void Config::initFieldValidator() {
          }
          return Status::OK();
        }},
+      {"notify-keyspace-events",
+       []([[maybe_unused]] const std::string &k, const std::string &v) -> Status {
+         int flags = 0;
+         return ParseNotifyKeyspaceEventsFlags(v, &flags);
+       }},
       {"compact-cron",
        [this]([[maybe_unused]] const std::string &k, const std::string &v) -> Status {
          std::vector<std::string> args = util::Split(v, " \t");
@@ -556,6 +563,10 @@ void Config::initFieldCallback() {
              srv->AdjustWorkerThreads();
              return Status::OK();
            }},
+          {"notify-keyspace-events",
+           [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const std::string &k, const std::string &v) -> Status {
+             return ParseNotifyKeyspaceEventsFlags(v, &notify_keyspace_events);
+           }},
           {"dir",
            [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const std::string &k,
                   [[maybe_unused]] const std::string &v) -> Status {

src/config/config.h

@@ -219,6 +219,9 @@ struct Config {
   // Enable transactional mode in engine::Context
   bool txn_context_enabled = false;
 
+  // Parsed notify-keyspace-events flags.
+  int notify_keyspace_events = 0;
+
   bool skip_block_cache_deallocation_on_close = false;
 
   bool lua_strict_key_accessing = false;
@@ -315,6 +318,7 @@ struct Config {
   std::string compaction_checker_cron_str_;
   std::string profiling_sample_commands_str_;
   std::string client_output_buffer_limit_str_;
+  std::string notify_keyspace_events_str_;
   std::map<std::string, std::unique_ptr<ConfigField>> fields_;
   std::vector<std::string> rename_command_;
   std::string histogram_bucket_boundaries_str_;