feat(script): support script flags of Eval script and Function (#2446)

This PR is designed to support three script flags within Eval Script and Function: `no-writes`, `no-cluster`, and `allow-cross-slot-keys`.

Before executing the Lua script, `SaveOnRegistry` stores the parsed flags in `REGISTRY_SCRIPT_RUN_CTX_NAME`. During the execution of the Lua script, `GetFromRegistry` retrieves the flags of the currently executing script. After the script execution ends, it should be set to `nil`.

For APIs like `EVAL`, `SCRIPT LOAD`, and `FUNCTION LOAD`, the flags parsed from the Eval Script will be stored in Lua's global variable `f_<sha>_flags_`. The flags parsed by FUNCTION `register_function()` will be stored in the Lua global variable `_registered_flags_<funcname>`.

src/cluster/cluster.cc

@@ -824,7 +824,7 @@ bool Cluster::IsWriteForbiddenSlot(int slot) const {
 }
 
 Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, const std::vector<std::string> &cmd_tokens,
-                                redis::Connection *conn) {
+                                redis::Connection *conn, lua::ScriptRunCtx *script_run_ctx) {
   std::vector<int> keys_indexes;
 
   // No keys
@@ -849,6 +849,21 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
     return {Status::RedisClusterDown, "Hash slot not served"};
   }
 
+  bool cross_slot_ok = false;
+  if (script_run_ctx) {
+    if (script_run_ctx->current_slot != -1 && script_run_ctx->current_slot != slot) {
+      if (getNodeIDBySlot(script_run_ctx->current_slot) != getNodeIDBySlot(slot)) {
+        return {Status::RedisMoved, fmt::format("{} {}:{}", slot, slots_nodes_[slot]->host, slots_nodes_[slot]->port)};
+      }
+      if (!(script_run_ctx->flags & lua::ScriptFlagType::kScriptAllowCrossSlotKeys)) {
+        return {Status::RedisCrossSlot, "Script attempted to access keys that do not hash to the same slot"};
+      }
+    }
+
+    script_run_ctx->current_slot = slot;
+    cross_slot_ok = true;
+  }
+
   if (myself_ && myself_ == slots_nodes_[slot]) {
     // We use central controller to manage the topology of the cluster.
     // Server can't change the topology directly, so we record the migrated slots
@@ -887,7 +902,11 @@ Status Cluster::CanExecByMySelf(const redis::CommandAttributes *attributes, cons
     return Status::OK();  // My master is serving this slot
   }
 
-  return {Status::RedisMoved, fmt::format("{} {}:{}", slot, slots_nodes_[slot]->host, slots_nodes_[slot]->port)};
+  if (!cross_slot_ok) {
+    return {Status::RedisMoved, fmt::format("{} {}:{}", slot, slots_nodes_[slot]->host, slots_nodes_[slot]->port)};
+  }
+
+  return Status::OK();
 }
 
 // Only HARD mode is meaningful to the Kvrocks cluster,

src/cluster/cluster.h

@@ -35,6 +35,7 @@
 #include "redis_slot.h"
 #include "server/redis_connection.h"
 #include "status.h"
+#include "storage/scripting.h"
 
 class ClusterNode {
  public:
@@ -83,7 +84,7 @@ class Cluster {
   bool IsNotMaster();
   bool IsWriteForbiddenSlot(int slot) const;
   Status CanExecByMySelf(const redis::CommandAttributes *attributes, const std::vector<std::string> &cmd_tokens,
-                         redis::Connection *conn);
+                         redis::Connection *conn, lua::ScriptRunCtx *script_run_ctx = nullptr);
   Status SetMasterSlaveRepl();
   Status MigrateSlotRange(const SlotRange &slot_range, const std::string &dst_node_id,
                           SyncMigrateContext *blocking_ctx = nullptr);

src/server/worker.cc

@@ -72,7 +72,7 @@ Worker::Worker(Server *srv, Config *config) : srv(srv), base_(event_base_new())
       LOG(INFO) << "[worker] Listening on: " << bind << ":" << *port;
     }
   }
-  lua_ = lua::CreateState(srv, true);
+  lua_ = lua::CreateState(srv);
 }
 
 Worker::~Worker() {