feat: enforce Redis maxclients at listener

docs/02-command-processing.md

@@ -47,7 +47,8 @@ service layer only; engine internals are in `data_substrate/docs/` (esp. `02-thr
    listen address, starts the metrics collector thread and namespace GC daemon. In `bootstrap`
    mode it exits the process right here after table creation (`src/redis_service.cpp:704-721`).
 6. brpc `Server::Start()` with `server_options.redis_service = redis_service_impl` (ownership
-   transfers to the server) and optional SSL options; then `RunUntilAskedToQuit()`
+   transfers to the server), the public listener declared Redis-only, `redis_max_connections` set
+   from `maxclients`, and optional force-SSL settings; then `RunUntilAskedToQuit()`
    (`src/redis_server.cpp:502-548`).
 
 ## 2. Request path end-to-end
@@ -99,7 +100,18 @@ key ≤ 32 MB (2 KB for the EloqStore backend) and object ≤ 256 MB
 
 ## 3. Connection state machine
 
-One `RedisConnectionContext` per socket, created by `NewConnectionContext`
+The brpc acceptor enforces `maxclients` only on the Redis-only public listener, immediately after
+the kernel accepts a TCP connection and before it creates a brpc `Socket`. Over-limit plaintext
+connections receive `-ERR max number of clients reached` and are closed; SSL-capable listeners
+close over-limit connections before TLS authentication or handshake work. Other EloqKV brpc/RPC
+servers are not subject to this limit. The acceptor reserves connection slots atomically before
+socket creation, so idle clients count toward the limit and concurrent accepts cannot overshoot it.
+`CONFIG GET maxclients` reports the active limit and `CONFIG SET maxclients <count>` atomically
+changes it for subsequent accepts. Lowering the limit does not disconnect established clients.
+The runtime value is not written back to disk; a restart loads `maxclients` from the `[local]`
+section of `eloqkv.ini` (or its gflag override) again.
+
+One `RedisConnectionContext` per admitted socket, created by `NewConnectionContext`
 (`src/redis_service.cpp:5917`) and owned by brpc's per-socket parsing context. Key fields
 (`include/redis_connection_context.h:61-170`):
 
@@ -304,8 +316,9 @@ the special MOVED/READONLY translations of §6.
 ## 9. Stats, INFO, slow log, metrics
 
 - `RedisStats` (`src/redis_stats.cpp`) exposes brpc bvars when `--enable_redis_stats` (default
-  on): connections received/rejected/closed, blocked clients, read/write/multi-object command
-  counters. Incremented in the connection ctor/dtor and in `ExecuteTxRequest`/
+  on): connections received/closed, blocked clients, and read/write/multi-object command
+  counters. Rejected connections come from the Redis listener's brpc admission counter. The
+  remaining counters are incremented in the connection ctor/dtor and in `ExecuteTxRequest`/
   `ExecuteMultiObjTxRequest` (4486-4496, 4528-4536). `INFO` is a `DirectCommand`
   (`include/redis_command.h:811`) assembled from these counters plus service fields captured at
   Init (OS info, exe path, memory; `src/redis_service.cpp:587-630`).

eloqkv.ini

@@ -24,6 +24,12 @@ ip = 127.0.0.1
 # changed once the server has been launched.
 port = 6379
 
+# Maximum number of simultaneous Redis client connections. This limits only
+# the Redis listener and does not change the process open-file limit. CONFIG
+# SET maxclients updates the running process; after restart this file is read
+# again.
+maxclients = 500000
+
 # EloqKV data directory. By default, the transaction service data, log service data, and local key-value storage data
 # are all stored in the eloq_data_path directory.
 eloq_data_path = eloq_data

include/redis_command.h

@@ -1321,6 +1321,7 @@ struct ConfigCommand : public DirectCommand
     std::vector<std::string_view> keys_;
     std::vector<std::string_view> values_;
     std::vector<std::string> results_;
+    std::string error_message_;
 };
 
 struct TimeCommand : public DirectCommand

include/redis_service.h

@@ -24,8 +24,10 @@
 #include <brpc/redis.h>
 #include <bthread/task_group.h>
 
+#include <atomic>
 #include <cassert>
 #include <cstddef>
+#include <cstdint>
 #include <map>
 #include <memory>  //std::unique_ptr
 #include <string>
@@ -156,6 +158,7 @@ class RedisServiceImpl;
 class NamespaceStorage;
 
 const std::unordered_set<std::string> redis_config_keys = {
+    "maxclients",
     "slowlog-log-slower-than",
     "slowlog-max-len",
 };
@@ -543,6 +546,8 @@ class RedisServiceImpl : public brpc::RedisService
 
     void ResizeSlowLog(uint32_t len);
 
+    // Returns the active Redis listener limit. CONFIG SET may update this
+    // value without changing the startup configuration in DataSubstrate.
     size_t MaxConnectionCount() const;
 
     static bool SendTxRequest(TransactionExecution *txm,
@@ -612,6 +617,15 @@ class RedisServiceImpl : public brpc::RedisService
     std::atomic_bool config_accessing_{false};
     std::unordered_map<std::string, std::string> config_;
 
+    // The Server outlives the service because it owns this RedisServiceImpl
+    // after Start(). The pointer is used only by CONFIG SET to update the
+    // Redis-only public acceptor's atomic admission limit.
+    brpc::Server *brpc_server_{nullptr};
+    // Race-free cache for startup and INFO reads. config_accessing_ serializes
+    // CONFIG mutations; relaxed access here does not synchronize the brpc
+    // acceptor update, which uses its own atomic state.
+    std::atomic<uint32_t> max_connection_count_{0};
+
     bool enable_redis_stats_;
     bool skip_kv_;
     bool skip_wal_;

src/redis_command.cpp

@@ -1788,11 +1788,19 @@ void InfoCommand::Execute(RedisServiceImpl *redis_impl,
     max_connection_count_ = redis_impl->MaxConnectionCount();
     active_extern_txms_ = redis_impl->ActiveExternTxCount();
 
+    conn_rejected_count_ =
+        server_acceptor == nullptr
+            ? 0
+            : server_acceptor->RejectedRedisConnectionCount();
+    // The acceptor owns the maxclients slots, so use the same source for
+    // connected_clients. This includes idle sockets that have not sent a
+    // first Redis command yet.
+    connecting_count_ =
+        server_acceptor == nullptr ? 0 : server_acceptor->ConnectionCount();
+
     if (redis_impl->IsEnableRedisStats())
     {
         conn_received_count_ = RedisStats::GetConnReceivedCount();
-        conn_rejected_count_ = RedisStats::GetConnRejectedCount();
-        connecting_count_ = RedisStats::GetConnectingCount();
         blocked_clients_count_ = RedisStats::GetBlockedClientsCount();
 
         cmd_read_count_ = RedisStats::GetReadCommandsCount();
@@ -2827,6 +2835,11 @@ void ConfigCommand::OutputResult(OutputHandler *reply) const
 {
     if (flag_ == CONFIG_SET)
     {
+        if (!error_message_.empty())
+        {
+            reply->OnError(error_message_);
+            return;
+        }
         reply->OnStatus("OK");
         return;
     }

src/redis_server.cpp

@@ -505,10 +505,17 @@ int main(int argc, char *argv[])
     // Notice: redis_service_impl will be deleted in server's destructor.
     server_options.redis_service = redis_service_impl.release();
     server_options.has_builtin_services = false;
+    // This listener is exclusively RESP. Declaring it as such lets brpc
+    // enforce maxclients immediately after accept, before any optional TLS
+    // handshake, without applying the limit to EloqKV's other RPC servers.
+    server_options.enabled_protocols = "redis";
+    server_options.redis_max_connections =
+        redis_service_ptr->MaxConnectionCount();
 
     // Configure TLS if enabled
     if (redis_service_ptr->IsTlsEnabled())
     {
+        server_options.force_ssl = true;
         brpc::ServerSSLOptions *ssl_options =
             server_options.mutable_ssl_options();
 

src/redis_service.cpp

@@ -42,6 +42,7 @@
 #include <filesystem>
 #include <fstream>
 #include <functional>
+#include <limits>
 #include <memory>
 #include <nlohmann/json.hpp>
 #include <optional>
@@ -232,6 +233,7 @@ RedisServiceImpl::RedisServiceImpl(const std::string &config_file,
 
 bool RedisServiceImpl::Init(brpc::Server &brpc_server)
 {
+    brpc_server_ = &brpc_server;
     INIReader config_reader(config_file_);
 
     if (!config_file_.empty() && config_reader.ParseError() != 0)
@@ -243,6 +245,10 @@ bool RedisServiceImpl::Init(brpc::Server &brpc_server)
     // Engine registration: EloqKv
     auto &ds = DataSubstrate::Instance();
 
+    max_connection_count_.store(ds.GetCoreConfig().maxclients,
+                                std::memory_order_relaxed);
+    config_.try_emplace("maxclients", std::to_string(MaxConnectionCount()));
+
     databases = config_reader.GetInteger("local", "databases", 16);
 
     // Prebuilt Redis tables (same names as today: data_table_0 ..
@@ -4733,13 +4739,58 @@ void RedisServiceImpl::ExecuteSetConfig(ConfigCommand *cmd)
     }
     for (size_t i = 0; i < cmd->keys_.size(); ++i)
     {
-        config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]);
-        if (cmd->keys_[i] == "slowlog-log-slower-than")
+        if (cmd->keys_[i] == "maxclients")
+        {
+            uint64_t value = 0;
+            const std::string_view text = cmd->values_[i];
+            if (!text.empty() && text.front() == '-')
+            {
+                cmd->error_message_ =
+                    "ERR CONFIG SET failed (possibly related to argument "
+                    "'maxclients') - argument must be between 1 and "
+                    "4294967295 inclusive";
+                break;
+            }
+            const auto [end, error] =
+                std::from_chars(text.data(), text.data() + text.size(), value);
+            if (error == std::errc::invalid_argument ||
+                end != text.data() + text.size())
+            {
+                cmd->error_message_ =
+                    "ERR CONFIG SET failed (possibly related to argument "
+                    "'maxclients') - argument couldn't be parsed into an "
+                    "integer";
+                break;
+            }
+            if (error == std::errc::result_out_of_range || value == 0 ||
+                value > std::numeric_limits<uint32_t>::max())
+            {
+                cmd->error_message_ =
+                    "ERR CONFIG SET failed (possibly related to argument "
+                    "'maxclients') - argument must be between 1 and "
+                    "4294967295 inclusive";
+                break;
+            }
+            const uint32_t maxclients = static_cast<uint32_t>(value);
+            if (brpc_server_ == nullptr ||
+                brpc_server_->SetRedisMaxConnections(maxclients) != 0)
+            {
+                cmd->error_message_ =
+                    "ERR CONFIG SET failed (possibly related to argument "
+                    "'maxclients') - unable to update the Redis listener";
+                break;
+            }
+            max_connection_count_.store(maxclients, std::memory_order_relaxed);
+            config_["maxclients"] = std::to_string(maxclients);
+        }
+        else if (cmd->keys_[i] == "slowlog-log-slower-than")
         {
+            config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]);
             slow_log_threshold_ = std::stoul(std::string(cmd->values_[i]));
         }
         else if (cmd->keys_[i] == "slowlog-max-len")
         {
+            config_[std::string(cmd->keys_[i])] = std::string(cmd->values_[i]);
             ResizeSlowLog(std::stoul(std::string(cmd->values_[i])));
         }
     }
@@ -6514,8 +6565,7 @@ metrics::Meter *RedisServiceImpl::GetMeter(std::size_t core_id) const
 
 size_t RedisServiceImpl::MaxConnectionCount() const
 {
-    auto &ds = DataSubstrate::Instance();
-    return ds.GetCoreConfig().maxclients;
+    return max_connection_count_.load(std::memory_order_relaxed);
 }
 
 }  // namespace EloqKV

tests/unit/eloq/maxclients.tcl

@@ -0,0 +1,31 @@
+start_server {tags {"maxclients network"}} {
+    test {CONFIG GET and SET expose the live maxclients limit} {
+        set original [lindex [r config get maxclients] 1]
+
+        assert_equal {OK} [r config set maxclients 1]
+        assert_equal {1} [lindex [r config get maxclients] 1]
+
+        # The connection issuing CONFIG SET remains established when the new
+        # limit is below the current connection count.
+        assert_equal {PONG} [r ping]
+
+        if {$::tls} {
+            set expected_rejection {*I/O error*}
+        } else {
+            set expected_rejection {*ERR max*reached*}
+        }
+        set rejected [catch {redis_deferring_client} rejection]
+        assert_equal {1} $rejected
+        assert_match $expected_rejection $rejection
+
+        assert_error {*argument must be between 1 and 4294967295 inclusive*} {
+            r config set maxclients 0
+        }
+        assert_error {*argument couldn't be parsed into an integer*} {
+            r config set maxclients invalid
+        }
+        assert_equal {1} [lindex [r config get maxclients] 1]
+
+        r config set maxclients $original
+    }
+}