fix(server): Implement SCRIPT FLUSH command (#2493)

* fix(server): Implement SCRIPT FLUSH command

src/core/interpreter.cc

@@ -895,4 +895,11 @@ void InterpreterManager::Return(Interpreter* ir) {
   waker_.notify();
 }
 
+void InterpreterManager::Reset() {
+  waker_.await([this]() { return available_.size() == storage_.size(); });
+
+  available_.clear();
+  storage_.clear();
+}
+
 }  // namespace dfly

src/core/interpreter.h

@@ -149,6 +149,8 @@ class InterpreterManager {
 
   void Return(Interpreter*);
 
+  void Reset();
+
  private:
   EventCount waker_;
   std::vector<Interpreter*> available_;

src/server/dragonfly_test.cc

@@ -256,6 +256,29 @@ TEST_F(DflyEngineTest, EvalSha) {
   EXPECT_THAT(resp, "c6459b95a0e81df97af6fdd49b1a9e0287a57363");
 }
 
+TEST_F(DflyEngineTest, ScriptFlush) {
+  auto resp = Run({"script", "load", "return 5"});
+  EXPECT_THAT(resp, ArgType(RespExpr::STRING));
+  string sha{ToSV(resp.GetBuf())};
+  resp = Run({"evalsha", sha, "0"});
+  EXPECT_THAT(5, resp.GetInt());
+  resp = Run({"script", "exists", sha});
+  EXPECT_THAT(1, resp.GetInt());
+
+  resp = Run({"script", "flush"});
+  resp = Run({"script", "exists", sha});
+  EXPECT_THAT(0, resp.GetInt());
+  EXPECT_THAT(Run({"evalsha", sha, "0"}), ErrArg("NOSCRIPT No matching script. Please use EVAL."));
+
+  resp = Run({"script", "load", "return 5"});
+  EXPECT_THAT(resp, ArgType(RespExpr::STRING));
+  sha = string{ToSV(resp.GetBuf())};
+  resp = Run({"evalsha", sha, "0"});
+  EXPECT_THAT(5, resp.GetInt());
+  resp = Run({"script", "exists", sha});
+  EXPECT_THAT(1, resp.GetInt());
+}
+
 TEST_F(DflyEngineTest, Hello) {
   auto resp = Run({"hello"});
   ASSERT_THAT(resp, ArrLen(14));

src/server/script_mgr.cc

@@ -63,6 +63,8 @@ void ScriptMgr::Run(CmdArgList args, ConnectionContext* cntx) {
         "Subcommands are:",
         "EXISTS <sha1> [<sha1> ...]",
         "   Return information about the existence of the scripts in the script cache.",
+        "FLUSH",
+        "   Flush the Lua scripts cache. Very dangerous on replicas.",
         "LOAD <script>",
         "   Load a script into the scripts cache without executing it.",
         "FLAGS <sha> [flags ...]",
@@ -83,6 +85,9 @@ void ScriptMgr::Run(CmdArgList args, ConnectionContext* cntx) {
   if (subcmd == "EXISTS" && args.size() > 1)
     return ExistsCmd(args, cntx);
 
+  if (subcmd == "FLUSH")
+    return FlushCmd(args, cntx);
+
   if (subcmd == "LIST")
     return ListCmd(cntx);
 
@@ -116,6 +121,12 @@ void ScriptMgr::ExistsCmd(CmdArgList args, ConnectionContext* cntx) const {
   return;
 }
 
+void ScriptMgr::FlushCmd(CmdArgList args, ConnectionContext* cntx) {
+  FlushAllScript();
+
+  return cntx->SendOk();
+}
+
 void ScriptMgr::LoadCmd(CmdArgList args, ConnectionContext* cntx) {
   string_view body = ArgS(args, 1);
   auto rb = static_cast<RedisReplyBuilder*>(cntx->reply_builder());
@@ -272,6 +283,16 @@ optional<ScriptMgr::ScriptData> ScriptMgr::Find(std::string_view sha) const {
   return std::nullopt;
 }
 
+void ScriptMgr::FlushAllScript() {
+  lock_guard lk{mu_};
+  db_.clear();
+
+  shard_set->pool()->Await([](auto index, auto* pb) {
+    ServerState* ss = ServerState::tlocal();
+    ss->ResetInterpreter();
+  });
+}
+
 vector<pair<string, ScriptMgr::ScriptData>> ScriptMgr::GetAll() const {
   vector<pair<string, ScriptData>> res;
 

src/server/script_mgr.h

@@ -54,11 +54,14 @@ class ScriptMgr {
   // Returns a list of all scripts in the database with their sha and body.
   std::vector<std::pair<std::string, ScriptData>> GetAll() const;
 
+  void FlushAllScript();
+
   // Returns if scripts run as global transactions by default
   bool AreGlobalByDefault() const;
 
  private:
   void ExistsCmd(CmdArgList args, ConnectionContext* cntx) const;
+  void FlushCmd(CmdArgList args, ConnectionContext* cntx);
   void LoadCmd(CmdArgList args, ConnectionContext* cntx);
   void ConfigCmd(CmdArgList args, ConnectionContext* cntx);
   void ListCmd(ConnectionContext* cntx) const;

src/server/server_state.cc

@@ -181,6 +181,10 @@ void ServerState::ReturnInterpreter(Interpreter* ir) {
   interpreter_mgr_.Return(ir);
 }
 
+void ServerState::ResetInterpreter() {
+  interpreter_mgr_.Reset();
+}
+
 ServerState* ServerState::SafeTLocal() {
   // https://stackoverflow.com/a/75622732
   asm volatile("");

src/server/server_state.h

@@ -181,6 +181,8 @@ class ServerState {  // public struct - to allow initialization.
   // Return interpreter to internal manager to be re-used.
   void ReturnInterpreter(Interpreter*);
 
+  void ResetInterpreter();
+
   // Returns sum of all requests in the last 6 seconds
   // (not including the current one).
   uint32_t MovingSum6() const {