diff --git a/src/server/command_registry.cc b/src/server/command_registry.cc index 1b8df89ea..76d2d7e6a 100644 --- a/src/server/command_registry.cc +++ b/src/server/command_registry.cc @@ -32,8 +32,7 @@ bool CommandId::IsTransactional() const { if (first_key_ > 0 || (opt_mask_ & CO::GLOBAL_TRANS) || (opt_mask_ & CO::NO_KEY_JOURNAL)) return true; - string_view name{name_}; - if (name == "EVAL" || name == "EVALSHA" || name == "EXEC") + if (name_ == "EVAL" || name_ == "EVALSHA" || name_ == "EXEC") return true; return false; @@ -44,30 +43,41 @@ uint32_t CommandId::OptCount(uint32_t mask) { } CommandRegistry::CommandRegistry() { - CommandId cd("COMMAND", CO::LOADING | CO::NOSCRIPT, -1, 0, 0, 0); + static const char kCMD[] = "COMMAND"; + CommandId cd(kCMD, CO::LOADING | CO::NOSCRIPT, -1, 0, 0, 0); cd.SetHandler([this](const auto& args, auto* cntx) { return Command(args, cntx); }); - const char* nm = cd.name(); - cmd_map_.emplace(nm, std::move(cd)); + cmd_map_.emplace(kCMD, std::move(cd)); } void CommandRegistry::Command(CmdArgList args, ConnectionContext* cntx) { + unsigned cmd_cnt = 0; + for (const auto& val : cmd_map_) { + const CommandId& cd = val.second; + if (cd.opt_mask() & CO::HIDDEN) + continue; + + ++cmd_cnt; + } + if (args.size() > 0) { ToUpper(&args[0]); string_view subcmd = ArgS(args, 0); if (subcmd == "COUNT") { - return (*cntx)->SendLong(cmd_map_.size()); + return (*cntx)->SendLong(cmd_cnt); } else { return (*cntx)->SendError(kSyntaxErr, kSyntaxErrType); } } - size_t len = cmd_map_.size(); - (*cntx)->StartArray(len); + (*cntx)->StartArray(cmd_cnt); for (const auto& val : cmd_map_) { const CommandId& cd = val.second; + if (cd.opt_mask() & CO::HIDDEN) + continue; + (*cntx)->StartArray(6); (*cntx)->SendSimpleString(cd.name()); (*cntx)->SendLong(cd.arity()); @@ -118,6 +128,8 @@ const char* OptName(CO::CommandOpt fl) { return "noscript"; case BLOCKING: return "blocking"; + case HIDDEN: + return "hidden"; case GLOBAL_TRANS: return "global-trans"; case VARIADIC_KEYS: diff --git a/src/server/command_registry.h b/src/server/command_registry.h index f83e4ba2b..007672aa0 100644 --- a/src/server/command_registry.h +++ b/src/server/command_registry.h @@ -31,6 +31,7 @@ enum CommandOpt : uint32_t { ADMIN = 1U << 7, // implies NOSCRIPT, NOSCRIPT = 1U << 8, BLOCKING = 1U << 9, // implies REVERSE_MAPPING + HIDDEN = 1U << 10, // does not show in COMMAND command output GLOBAL_TRANS = 1U << 12, NO_AUTOJOURNAL = 1U << 15, // Skip automatically logging command to journal inside transaction. @@ -43,6 +44,10 @@ constexpr inline bool IsEvalKind(std::string_view name) { return name.compare(0, 4, "EVAL") == 0; } +constexpr inline bool IsTransKind(std::string_view name) { + return (name == "EXEC") || (name == "MULTI") || (name == "DISCARD"); +} + static_assert(IsEvalKind("EVAL") && IsEvalKind("EVALSHA")); static_assert(!IsEvalKind("")); @@ -78,7 +83,7 @@ class CommandId { CommandId(const char* name, uint32_t mask, int8_t arity, int8_t first_key, int8_t last_key, int8_t step); - const char* name() const { + std::string_view name() const { return name_; } @@ -132,7 +137,7 @@ class CommandId { static uint32_t OptCount(uint32_t mask); private: - const char* name_; + std::string_view name_; uint32_t opt_mask_; int8_t arity_; diff --git a/src/server/main_service.cc b/src/server/main_service.cc index a92161f13..076ae2b07 100644 --- a/src/server/main_service.cc +++ b/src/server/main_service.cc @@ -11,6 +11,7 @@ extern "C" { #include #include #include +#include #include #include @@ -192,14 +193,14 @@ void SendMonitor(const std::string& msg) { } } -void DispatchMonitorIfNeeded(bool admin_cmd, ConnectionContext* connection, CmdArgList args) { +void DispatchMonitorIfNeeded(bool admin_cmd, ConnectionContext* cntx, CmdArgList args) { // We are not sending any admin command in the monitor, and we do not want to // do any processing if we don't have any waiting connections with monitor // enabled on them - see https://redis.io/commands/monitor/ const auto& my_monitors = ServerState::tlocal()->Monitors(); if (!(my_monitors.Empty() || admin_cmd)) { // We have connections waiting to get the info on the last command, send it to them - auto monitor_msg = MakeMonitorMessage(connection->conn_state, connection->owner(), args); + auto monitor_msg = MakeMonitorMessage(cntx->conn_state, cntx->owner(), args); VLOG(1) << "sending command '" << monitor_msg << "' to the clients that registered on it"; @@ -583,12 +584,6 @@ void Service::Shutdown() { ThisFiber::SleepFor(10ms); } -static void SetMultiExecErrorFlag(ConnectionContext* cntx) { - if (cntx->conn_state.exec_info.IsActive()) { - cntx->conn_state.exec_info.state = ConnectionState::ExecInfo::EXEC_ERROR; - } -} - // Return OK if all keys are allowed to be accessed: either declared in EVAL or // transaction is running in global or non-atomic mode. OpStatus CheckKeysDeclared(const ConnectionState::ScriptInfo& eval_info, const CommandId* cid, @@ -619,20 +614,19 @@ OpStatus CheckKeysDeclared(const ConnectionState::ScriptInfo& eval_info, const C return OpStatus::OK; } -bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, - facade::ConnectionContext* cntx) { +bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, ConnectionContext* dfly_cntx) { ServerState& etl = *ServerState::tlocal(); string_view cmd_str = ArgS(args, 0); - ConnectionContext* dfly_cntx = static_cast(cntx); - bool is_trans_cmd = (cmd_str == "EXEC" || cmd_str == "MULTI" || cmd_str == "DISCARD"); - bool under_script = bool(dfly_cntx->conn_state.script_info); - - absl::Cleanup multi_error([dfly_cntx] { SetMultiExecErrorFlag(dfly_cntx); }); + absl::Cleanup multi_error([exec_info = &dfly_cntx->conn_state.exec_info] { + if (exec_info->IsActive()) { + exec_info->state = ConnectionState::ExecInfo::EXEC_ERROR; + } + }); if (cid == nullptr) { - (*cntx)->SendError(StrCat("unknown command `", cmd_str, "`"), "unknown_cmd"); + (*dfly_cntx)->SendError(StrCat("unknown command `", cmd_str, "`"), "unknown_cmd"); lock_guard lk(mu_); if (unknown_cmds_.size() < 1024) @@ -640,31 +634,33 @@ bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, return false; } - bool blocked_by_loading = !cntx->journal_emulated && etl.gstate() == GlobalState::LOADING && + bool is_trans_cmd = CO::IsTransKind(cid->name()); + bool under_script = bool(dfly_cntx->conn_state.script_info); + bool blocked_by_loading = !dfly_cntx->journal_emulated && etl.gstate() == GlobalState::LOADING && (cid->opt_mask() & CO::LOADING) == 0; if (blocked_by_loading || etl.gstate() == GlobalState::SHUTTING_DOWN) { string err = StrCat("Can not execute during ", GlobalStateName(etl.gstate())); - (*cntx)->SendError(err); + (*dfly_cntx)->SendError(err); return false; } string_view cmd_name{cid->name()}; - if (cntx->req_auth && !cntx->authenticated) { + if (dfly_cntx->req_auth && !dfly_cntx->authenticated) { if (cmd_name != "AUTH" && cmd_name != "QUIT" && cmd_name != "HELLO") { - (*cntx)->SendError("-NOAUTH Authentication required."); + (*dfly_cntx)->SendError("-NOAUTH Authentication required."); return false; } } // only reset and quit are allow if this connection is used for monitoring if (dfly_cntx->monitor && (cmd_name != "RESET" && cmd_name != "QUIT")) { - (*cntx)->SendError("Replica can't interact with the keyspace"); + (*dfly_cntx)->SendError("Replica can't interact with the keyspace"); return false; } if (under_script && (cid->opt_mask() & CO::NOSCRIPT)) { - (*cntx)->SendError("This Redis command is not allowed from script"); + (*dfly_cntx)->SendError("This Redis command is not allowed from script"); return false; } @@ -673,18 +669,18 @@ bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, bool under_multi = dfly_cntx->conn_state.exec_info.IsActive() && !is_trans_cmd; if (!etl.is_master && is_write_cmd && !dfly_cntx->is_replicating) { - (*cntx)->SendError("-READONLY You can't write against a read only replica."); + (*dfly_cntx)->SendError("-READONLY You can't write against a read only replica."); return false; } if ((cid->arity() > 0 && args.size() != size_t(cid->arity())) || (cid->arity() < 0 && args.size() < size_t(-cid->arity()))) { - (*cntx)->SendError(facade::WrongNumArgsError(cmd_str), kSyntaxErrType); + (*dfly_cntx)->SendError(facade::WrongNumArgsError(cmd_str), kSyntaxErrType); return false; } if (cid->key_arg_step() == 2 && (args.size() % 2) == 0) { - (*cntx)->SendError(facade::WrongNumArgsError(cmd_str), kSyntaxErrType); + (*dfly_cntx)->SendError(facade::WrongNumArgsError(cmd_str), kSyntaxErrType); return false; } @@ -695,12 +691,12 @@ bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, if (under_multi) { if (cmd_name == "SELECT") { - (*cntx)->SendError("Can not call SELECT within a transaction"); + (*dfly_cntx)->SendError("Can not call SELECT within a transaction"); return false; } if (cmd_name == "WATCH" || cmd_name == "FLUSHALL" || cmd_name == "FLUSHDB") { - (*cntx)->SendError(absl::StrCat("'", cmd_name, "' inside MULTI is not allowed")); + (*dfly_cntx)->SendError(absl::StrCat("'", cmd_name, "' inside MULTI is not allowed")); return false; } } @@ -710,12 +706,12 @@ bool Service::VerifyCommand(const CommandId* cid, CmdArgList args, dfly_cntx->transaction); if (status == OpStatus::KEY_NOTFOUND) { - (*cntx)->SendError("script tried accessing undeclared key"); + (*dfly_cntx)->SendError("script tried accessing undeclared key"); return false; } if (status != OpStatus::OK) { - (*cntx)->SendError(status); + (*dfly_cntx)->SendError(status); return false; } } @@ -740,21 +736,20 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx) << " in dbid=" << dfly_cntx->conn_state.db_index; } - string_view cmd_str = ArgS(args, 0); - bool is_trans_cmd = (cmd_str == "EXEC" || cmd_str == "MULTI" || cmd_str == "DISCARD"); - const CommandId* cid = registry_.Find(cmd_str); + const CommandId* cid = FindCmd(args); ServerState& etl = *ServerState::tlocal(); etl.RecordCmd(); - if (!VerifyCommand(cid, args, cntx)) + if (!VerifyCommand(cid, args, dfly_cntx)) return; + bool is_trans_cmd = CO::IsTransKind(cid->name()); etl.connection_stats.cmd_count_map[cid->name()]++; - + auto args_no_cmd = args.subspan(1); if (dfly_cntx->conn_state.exec_info.IsActive() && !is_trans_cmd) { // TODO: protect against aggregating huge transactions. - StoredCmd stored_cmd{cid, args.subspan(1)}; + StoredCmd stored_cmd{cid, args_no_cmd}; dfly_cntx->conn_state.exec_info.body.push_back(std::move(stored_cmd)); return (*cntx)->SendSimpleString("QUEUED"); @@ -771,7 +766,7 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx) if (cid->IsTransactional()) { dfly_cntx->transaction->MultiSwitchCmd(cid); OpStatus status = - dfly_cntx->transaction->InitByArgs(dfly_cntx->conn_state.db_index, args.subspan(1)); + dfly_cntx->transaction->InitByArgs(dfly_cntx->conn_state.db_index, args_no_cmd); if (status != OpStatus::OK) return (*cntx)->SendError(status); @@ -783,7 +778,7 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx) dist_trans.reset(new Transaction{cid, etl.thread_index()}); if (!dist_trans->IsMulti()) { // Multi command initialize themself based on their mode. - if (auto st = dist_trans->InitByArgs(dfly_cntx->conn_state.db_index, args.subspan(1)); + if (auto st = dist_trans->InitByArgs(dfly_cntx->conn_state.db_index, args_no_cmd); st != OpStatus::OK) return (*cntx)->SendError(st); } @@ -807,7 +802,7 @@ void Service::DispatchCommand(CmdArgList args, facade::ConnectionContext* cntx) } end_usec = ProactorBase::GetMonotonicTimeNs(); - request_latency_usec.IncBy(cmd_str, (end_usec - start_usec) / 1000); + request_latency_usec.IncBy(cid->name(), (end_usec - start_usec) / 1000); if (!under_script) { dfly_cntx->transaction = nullptr; @@ -1216,6 +1211,20 @@ bool StartMultiEval(DbIndex dbid, CmdArgList keys, ScriptMgr::ScriptParams param return false; } +const CommandId* Service::FindCmd(CmdArgList args) const { + const CommandId* res = registry_.Find(ArgS(args, 0)); + if (!res) + return nullptr; + + // A workaround for XGROUP HELP that does not fit our static taxonomy of commands. + if (args.size() == 2 && res->name() == "XGROUP") { + if (absl::EqualsIgnoreCase(ArgS(args, 1), "HELP")) { + res = registry_.Find("_XGROUP_HELP"); + } + } + return res; +} + void Service::EvalInternal(const EvalArgs& eval_args, Interpreter* interpreter, ConnectionContext* cntx) { DCHECK(!eval_args.sha.empty()); diff --git a/src/server/main_service.h b/src/server/main_service.h index c73b4eed5..e03ea23ba 100644 --- a/src/server/main_service.h +++ b/src/server/main_service.h @@ -116,7 +116,9 @@ class Service : public facade::ServiceInterface { }; // Return false if command is invalid and reply with error. - bool VerifyCommand(const CommandId* cid, CmdArgList args, facade::ConnectionContext* cntx); + bool VerifyCommand(const CommandId* cid, CmdArgList args, ConnectionContext* cntx); + + const CommandId* FindCmd(CmdArgList args) const; void EvalInternal(const EvalArgs& eval_args, Interpreter* interpreter, ConnectionContext* cntx); diff --git a/src/server/server_family.cc b/src/server/server_family.cc index 9fc147ebe..e9c2931b0 100644 --- a/src/server/server_family.cc +++ b/src/server/server_family.cc @@ -2158,7 +2158,7 @@ void ServerFamily::Register(CommandRegistry* registry) { << CI{"ROLE", CO::LOADING | CO::FAST | CO::NOSCRIPT, 1, 0, 0, 0}.HFUNC(Role) << CI{"SLOWLOG", CO::ADMIN | CO::FAST, -2, 0, 0, 0}.SetHandler(SlowLog) << CI{"SCRIPT", CO::NOSCRIPT | CO::NO_KEY_JOURNAL, -2, 0, 0, 0}.HFUNC(Script) - << CI{"DFLY", CO::ADMIN | CO::GLOBAL_TRANS, -2, 0, 0, 0}.HFUNC(Dfly); + << CI{"DFLY", CO::ADMIN | CO::GLOBAL_TRANS | CO::HIDDEN, -2, 0, 0, 0}.HFUNC(Dfly); } } // namespace dfly diff --git a/src/server/stream_family.cc b/src/server/stream_family.cc index 8dc2162fe..cafb24558 100644 --- a/src/server/stream_family.cc +++ b/src/server/stream_family.cc @@ -563,6 +563,24 @@ void SetId(string_view key, string_view gname, CmdArgList args, ConnectionContex } } +void XGroupHelp(CmdArgList args, ConnectionContext* cntx) { + string_view help_arr[] = { + "CREATE [option]", + " Create a new consumer group. Options are:", + " * MKSTREAM", + " Create the empty stream if it does not exist.", + "CREATECONSUMER ", + " Create a new consumer in the specified group.", + "DELCONSUMER ", + " Remove the specified consumer.", + "DESTROY " + " Remove the specified group.", + "SETID ", + " Set the current group ID.", + }; + return (*cntx)->SendSimpleStrArr(help_arr); +} + } // namespace void StreamFamily::XAdd(CmdArgList args, ConnectionContext* cntx) { @@ -651,23 +669,6 @@ void StreamFamily::XDel(CmdArgList args, ConnectionContext* cntx) { void StreamFamily::XGroup(CmdArgList args, ConnectionContext* cntx) { ToUpper(&args[0]); string_view sub_cmd = ArgS(args, 0); - if (sub_cmd == "HELP") { - string_view help_arr[] = { - "CREATE [option]", - " Create a new consumer group. Options are:", - " * MKSTREAM", - " Create the empty stream if it does not exist.", - "CREATECONSUMER ", - " Create a new consumer in the specified group.", - "DELCONSUMER ", - " Remove the specified consumer.", - "DESTROY " - " Remove the specified group.", - "SETID ", - " Set the current group ID.", - }; - return (*cntx)->SendSimpleStrArr(help_arr); - } if (args.size() >= 2) { string_view key = ArgS(args, 1); @@ -859,12 +860,13 @@ void StreamFamily::Register(CommandRegistry* registry) { *registry << CI{"XADD", CO::WRITE | CO::FAST, -5, 1, 1, 1}.HFUNC(XAdd) << CI{"XDEL", CO::WRITE | CO::FAST, -3, 1, 1, 1}.HFUNC(XDel) - << CI{"XGROUP", CO::WRITE | CO::DENYOOM, -2, 2, 2, 1}.HFUNC(XGroup) + << CI{"XGROUP", CO::WRITE | CO::DENYOOM, -3, 2, 2, 1}.HFUNC(XGroup) << CI{"XINFO", CO::READONLY | CO::NOSCRIPT, -2, 0, 0, 0}.HFUNC(XInfo) << CI{"XLEN", CO::READONLY | CO::FAST, 2, 1, 1, 1}.HFUNC(XLen) << CI{"XRANGE", CO::READONLY, -4, 1, 1, 1}.HFUNC(XRange) << CI{"XREVRANGE", CO::READONLY, -4, 1, 1, 1}.HFUNC(XRevRange) - << CI{"XSETID", CO::WRITE | CO::DENYOOM, 3, 1, 1, 1}.HFUNC(XSetId); + << CI{"XSETID", CO::WRITE | CO::DENYOOM, 3, 1, 1, 1}.HFUNC(XSetId) + << CI{"_XGROUP_HELP", CO::NOSCRIPT | CO::HIDDEN, 2, 0, 0, 0}.SetHandler(XGroupHelp); } } // namespace dfly diff --git a/src/server/stream_family_test.cc b/src/server/stream_family_test.cc index 913413787..25e3b7a4a 100644 --- a/src/server/stream_family_test.cc +++ b/src/server/stream_family_test.cc @@ -84,4 +84,12 @@ TEST_F(StreamFamilyTest, Range) { EXPECT_THAT(sub1, ElementsAre("1-0", ArrLen(2))); } +TEST_F(StreamFamilyTest, Issue854) { + auto resp = Run({"xgroup", "help"}); + EXPECT_THAT(resp, ArgType(RespExpr::ARRAY)); + + resp = Run({"eval", "redis.call('xgroup', 'help')", "0"}); + EXPECT_THAT(resp, ErrArg("is not allowed")); +} + } // namespace dfly diff --git a/src/server/transaction.cc b/src/server/transaction.cc index 45d1e289d..fe5800a0a 100644 --- a/src/server/transaction.cc +++ b/src/server/transaction.cc @@ -920,7 +920,7 @@ void Transaction::UnwatchBlocking(bool should_expire, WaitKeysProvider wcb) { DVLOG(1) << "UnwatchBlocking finished " << DebugId(); } -const char* Transaction::Name() const { +string_view Transaction::Name() const { return cid_->name(); } diff --git a/src/server/transaction.h b/src/server/transaction.h index 68bb43a5d..60094b985 100644 --- a/src/server/transaction.h +++ b/src/server/transaction.h @@ -251,7 +251,7 @@ class Transaction { IntentLock::Mode Mode() const; // Based on command mask - const char* Name() const; // Based on command name + std::string_view Name() const; // Based on command name uint32_t GetUniqueShardCnt() const { return unique_shard_cnt_; diff --git a/src/server/zset_family.cc b/src/server/zset_family.cc index 3d892513f..85fe7fa1d 100644 --- a/src/server/zset_family.cc +++ b/src/server/zset_family.cc @@ -615,7 +615,7 @@ bool ParseLexBound(string_view src, ZSetFamily::LexBound* bound) { } void SendAtLeastOneKeyError(ConnectionContext* cntx) { - string name = cntx->cid->name(); + string name{cntx->cid->name()}; absl::AsciiStrToLower(&name); (*cntx)->SendError(absl::StrCat("at least 1 input key is needed for ", name)); }