feat(keyspace): emit set and del notifications (#3541)

Implement initial keyspace notifications for set and del events,
compatible with Redis notify-keyspace-events.

This PR adds the initial notification emitters and configuration
support:
- Support `K`, `E`, `g`, `$`, and `A` flags, where `A` currently expands
to the implemented event classes `g$`.
- Emit `set` notifications only when the shared `Set` path actually
applies a write, including conditional `SET` variants.
- Emit `del` notifications only for keys that are actually deleted
through the shared delete path.
- Deduplicate repeated keys in a single delete operation, so the same
key is deleted and notified once.
- Queue notifications inside `MULTI/EXEC` and publish them after a
successful commit.
  - Map notification DB names correctly:
    - default namespace uses DB `0`
    - `redis-databases` namespaces map back to Redis DB indexes

This PR only adds the initial set and del event paths on primary nodes,
keeping the initial patch small and reviewable. Commands sharing the
same underlying mutation APIs may also produce these events.
Replica-side notifications and additional event types, if supported, can
be added in follow-up PRs.

Unsupported notification classes are rejected for now until their
emitters are implemented.

  Ref Proposal: #3533
  Tracking Issue: #2915
  
  Assisted by Codex/GPT-5.5.

---------

Co-authored-by: hulk <hulk.website@gmail.com>
Co-authored-by: Aleks Lozovyuk <aleks.raiden@gmail.com>

Author: 3 people

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 their original names.
+# 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_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,64 @@
+/*
+ * 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 "config/config.h"
+#include "fmt/format.h"
+
+StatusOr<std::pair<KeyspaceEventChannel, KeyspaceEventType>> ParseNotifyKeyspaceEventsFlags(std::string_view input) {
+  int channel_flags = 0;
+  int type_flags = 0;
+  for (const char c : input) {
+    switch (c) {
+      case 'K':
+        channel_flags |= kNotifyKeyspace;
+        break;
+      case 'E':
+        channel_flags |= kNotifyKeyevent;
+        break;
+      case 'A':
+        type_flags |= kNotifyAll;
+        break;
+      case 'g':
+        type_flags |= kNotifyGeneric;
+        break;
+      case '$':
+        type_flags |= kNotifyString;
+        break;
+      default:
+        return {Status::NotOK, fmt::format("unsupported notify-keyspace-events flag: '{}'", c)};
+    }
+  }
+
+  return std::pair{static_cast<KeyspaceEventChannel>(channel_flags), static_cast<KeyspaceEventType>(type_flags)};
+}
+
+std::string FormatKeyspaceNotificationScope(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;
+}

src/common/keyspace_events.h

@@ -0,0 +1,62 @@
+/*
+ * 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 <utility>
+
+#include "status.h"
+
+enum KeyspaceEventChannel {
+  kNotifyNoChannel = 0,
+  kNotifyKeyspace = 1 << 0,  // K, keyspace channels
+  kNotifyKeyevent = 1 << 1,  // E, keyevent channels
+};
+
+// Event type flags for notify-keyspace-events, separate from RedisType.
+enum KeyspaceEventType {
+  kNotifyNoType = 0,
+  kNotifyGeneric = 1 << 0,  // g, emits del
+  kNotifyString = 1 << 1,   // $, emits set
+  // A, supported data classes without K or E.
+  kNotifyAll = kNotifyGeneric | kNotifyString,
+};
+
+struct KeyspaceEvent {
+  KeyspaceEvent(KeyspaceEventType type_flag, std::string_view event, KeyspaceEventChannel channel_flags,
+                std::string_view ns, std::string_view key)
+      : type_flag(type_flag), channel_flags(channel_flags), event(event), ns(ns), key(key) {}
+
+  KeyspaceEventType type_flag;
+  KeyspaceEventChannel channel_flags;
+  std::string event;
+  std::string ns;
+  std::string key;
+};
+
+// Parses notify-keyspace-events flags into channel flags followed by event type flags.
+StatusOr<std::pair<KeyspaceEventChannel, KeyspaceEventType>> ParseNotifyKeyspaceEventsFlags(std::string_view input);
+
+// Formats the namespace or database scope used in keyspace notification channel names.
+// Default namespace maps to 0; database namespaces map back to db indexes when redis-databases is enabled.
+// Other namespaces use their original names.
+std::string FormatKeyspaceNotificationScope(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,10 @@ void Config::initFieldValidator() {
          }
          return Status::OK();
        }},
+      {"notify-keyspace-events",
+       []([[maybe_unused]] const std::string &k, const std::string &v) -> Status {
+         return ParseNotifyKeyspaceEventsFlags(v).ToStatus();
+       }},
       {"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 +562,13 @@ 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 {
+             const auto flags = GET_OR_RET(ParseNotifyKeyspaceEventsFlags(v));
+             notify_keyspace_event_channels = flags.first;
+             notify_keyspace_event_types = flags.second;
+             return Status::OK();
+           }},
           {"dir",
            [this]([[maybe_unused]] Server *srv, [[maybe_unused]] const std::string &k,
                   [[maybe_unused]] const std::string &v) -> Status {

src/config/config.h

@@ -32,6 +32,7 @@
 
 #include "config_type.h"
 #include "cron.h"
+#include "keyspace_events.h"
 #include "spdlog/common.h"
 #include "status.h"
 #include "storage/redis_metadata.h"
@@ -219,6 +220,10 @@ struct Config {
   // Enable transactional mode in engine::Context
   bool txn_context_enabled = false;
 
+  // Parsed notify-keyspace-events flags.
+  KeyspaceEventChannel notify_keyspace_event_channels = kNotifyNoChannel;
+  KeyspaceEventType notify_keyspace_event_types = kNotifyNoType;
+
   bool skip_block_cache_deallocation_on_close = false;
 
   bool lua_strict_key_accessing = false;
@@ -315,6 +320,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_;

src/server/redis_connection.cc

@@ -23,6 +23,7 @@
 
 #include <mutex>
 #include <nonstd/span.hpp>
+#include <optional>
 #include <shared_mutex>
 
 #include "commands/commander.h"
@@ -634,6 +635,7 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
     }
 
     SetLastCmd(cmd_name);
+    std::vector<KeyspaceEvent> keyspace_events;
     {
       std::optional<MultiLockGuard> guard;
       if (cmd_flags & kCmdWrite) {
@@ -652,6 +654,9 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
         guard.emplace(srv_->storage->GetLockManager(), lock_keys);
       }
       engine::Context ctx(srv_->storage);
+      if (cmd_flags & kCmdWrite) {
+        ctx.EnableKeyspaceEventCollection(config->notify_keyspace_event_channels, config->notify_keyspace_event_types);
+      }
 
       std::vector<GlobalIndexer::RecordResult> index_records;
       if (!srv_->index_mgr.index_map.empty() && IsCmdForIndexing(cmd_flags, attributes->category) &&
@@ -679,6 +684,13 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
           WARN("[connection] index updating failed for key: {}", record.key);
         }
       }
+      if (ctx.HasKeyspaceEvents()) {
+        keyspace_events = ctx.TakeKeyspaceEvents();
+      }
+    }
+    // Nested Lua and function commands reuse their outer context. Publish only after index updates and key unlocking.
+    if (!keyspace_events.empty()) {
+      queueOrPublishKeyspaceEvents(std::move(keyspace_events));
     }
 
     if (!(cmd_flags & redis::kCmdSkipMonitor)) {
@@ -709,10 +721,40 @@ void Connection::ExecuteCommands(std::deque<CommandTokens> *to_process_cmds) {
   }
 }
 
+void Connection::queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events) {
+  if (events.empty()) return;
+
+  if (in_exec_) {
+    // Queue transaction events until commit.
+    for (auto &event : events) {
+      pending_keyspace_events_.emplace_back(std::move(event));
+    }
+    return;
+  }
+
+  for (const auto &event : events) {
+    srv_->NotifyKeyspaceEvent(event);
+  }
+}
+
+void Connection::FlushKeyspaceEvents() {
+  for (const auto &e : pending_keyspace_events_) {
+    srv_->NotifyKeyspaceEvent(e);
+  }
+  pending_keyspace_events_.clear();
+}
+
 void Connection::ResetMultiExec() {
   in_exec_ = false;
   multi_error_ = false;
   multi_cmds_.clear();
+  // Drop events from failed or aborted transactions.
+  pending_keyspace_events_.clear();
+  // Retain capacity for typical transactions, but request releasing unusually large buffers.
+  constexpr std::size_t kMaxRetainedKeyspaceEvents = 1024;
+  if (pending_keyspace_events_.capacity() > kMaxRetainedKeyspaceEvents) {
+    pending_keyspace_events_.shrink_to_fit();
+  }
   DisableFlag(Connection::kMultiExec);
 }
 

src/server/redis_connection.h

@@ -30,6 +30,7 @@
 #include <vector>
 
 #include "commands/commander.h"
+#include "common/keyspace_events.h"
 #include "event_util.h"
 #include "redis_request.h"
 #include "server/redis_reply.h"
@@ -208,6 +209,8 @@ class Connection : public EvbufCallbackBase<Connection> {
   void ResetMultiExec();
   std::deque<redis::CommandTokens> *GetMultiExecCommands() { return &multi_cmds_; }
 
+  void FlushKeyspaceEvents();
+
   std::function<void(int)> close_cb = nullptr;
 
   std::set<std::string> watched_keys;
@@ -218,6 +221,9 @@ class Connection : public EvbufCallbackBase<Connection> {
   ReplyMode GetReplyMode() const { return reply_mode_; }
 
  private:
+  // Queues events while EXEC is running; publishes them otherwise.
+  void queueOrPublishKeyspaceEvents(std::vector<KeyspaceEvent> &&events);
+
   uint64_t id_ = 0;
   std::atomic<int> flags_ = 0;
   std::string ns_;
@@ -248,6 +254,8 @@ class Connection : public EvbufCallbackBase<Connection> {
   bool multi_error_ = false;
   std::atomic<bool> is_running_ = false;
   std::deque<redis::CommandTokens> multi_cmds_;
+
+  std::vector<KeyspaceEvent> pending_keyspace_events_;
   bool in_script_ = false;
 
   bool importing_ = false;

src/server/server.cc

@@ -41,6 +41,7 @@
 
 #include "commands/command_parser.h"
 #include "commands/commander.h"
+#include "common/keyspace_events.h"
 #include "common/string_util.h"
 #include "config/config.h"
 #include "fmt/format.h"
@@ -478,6 +479,17 @@ int Server::PublishMessage(const std::string &channel, const std::string &msg) {
   return cnt;
 }
 
+void Server::NotifyKeyspaceEvent(const KeyspaceEvent &event) {
+  const std::string scope = FormatKeyspaceNotificationScope(event.ns, GetConfig()->redis_databases);
+  // Publish keyspace before keyevent for each key.
+  if (event.channel_flags & kNotifyKeyspace) {
+    PublishMessage("__keyspace@" + scope + "__:" + event.key, event.event);
+  }
+  if (event.channel_flags & kNotifyKeyevent) {
+    PublishMessage("__keyevent@" + scope + "__:" + event.event, event.key);
+  }
+}
+
 void Server::SubscribeChannel(const std::string &channel, redis::Connection *conn) {
   std::lock_guard<std::mutex> guard(pubsub_channels_mu_);
 

src/server/server.h

@@ -218,6 +218,9 @@ class Server {
   int GetFetchFileThreadNum() const { return fetch_file_threads_num_; }
 
   int PublishMessage(const std::string &channel, const std::string &msg);
+
+  // Publishes a keyspace event through the channels selected when it was collected.
+  void NotifyKeyspaceEvent(const KeyspaceEvent &event);
   void SubscribeChannel(const std::string &channel, redis::Connection *conn);
   void UnsubscribeChannel(const std::string &channel, redis::Connection *conn);
   void GetChannelsByPattern(const std::string &pattern, std::vector<std::string> *channels);