fix: loading state error type to be compatible with redis (#2629)

* add -LOADING prefix for loading errors
* replace -ERR with -LOADING for loading errors

Author: kostasrim

src/facade/error.h

@@ -31,6 +31,7 @@ extern const char kIndexOutOfRange[];
 extern const char kOutOfMemory[];
 extern const char kInvalidNumericResult[];
 extern const char kClusterNotConfigured[];
+extern const char kLoadingErr[];
 
 extern const char kSyntaxErrType[];
 extern const char kScriptErrType[];

src/facade/facade.cc

@@ -91,6 +91,7 @@ const char kIndexOutOfRange[] = "index out of range";
 const char kOutOfMemory[] = "Out of memory";
 const char kInvalidNumericResult[] = "result is not a number";
 const char kClusterNotConfigured[] = "Cluster is not yet configured";
+const char kLoadingErr[] = "-LOADING Dragonfly is loading the dataset in memory";
 
 const char kSyntaxErrType[] = "syntax_error";
 const char kScriptErrType[] = "script_error";

src/facade/reply_builder.cc

@@ -308,10 +308,11 @@ void RedisReplyBuilder::SendError(string_view str, string_view err_type) {
   if (str[0] == '-') {
     iovec v[] = {IoVec(str), IoVec(kCRLF)};
     Send(v, ABSL_ARRAYSIZE(v));
-  } else {
-    iovec v[] = {IoVec(kErrPref), IoVec(str), IoVec(kCRLF)};
-    Send(v, ABSL_ARRAYSIZE(v));
+    return;
   }
+
+  iovec v[] = {IoVec(kErrPref), IoVec(str), IoVec(kCRLF)};
+  Send(v, ABSL_ARRAYSIZE(v));
 }
 
 void RedisReplyBuilder::SendProtocolError(std::string_view str) {

src/server/main_service.cc

@@ -983,7 +983,8 @@ std::optional<ErrorReply> Service::VerifyCommandState(const CommandId* cid, CmdA
 
   // Check if the command is allowed to execute under this global state
   bool allowed_by_state = true;
-  switch (etl.gstate()) {
+  const GlobalState gstate = etl.gstate();
+  switch (gstate) {
     case GlobalState::LOADING:
       allowed_by_state = dfly_cntx.journal_emulated || (cid->opt_mask() & CO::LOADING);
       break;
@@ -999,8 +1000,13 @@ std::optional<ErrorReply> Service::VerifyCommandState(const CommandId* cid, CmdA
 
   if (!allowed_by_state) {
     VLOG(1) << "Command " << cid->name() << " not executed because global state is "
-            << GlobalStateName(etl.gstate());
-    return ErrorReply{StrCat("Can not execute during ", GlobalStateName(etl.gstate()))};
+            << GlobalStateName(gstate);
+
+    if (gstate == GlobalState::LOADING) {
+      return ErrorReply(kLoadingErr);
+    }
+
+    return ErrorReply{StrCat("Can not execute during ", GlobalStateName(gstate))};
   }
 
   string_view cmd_name{cid->name()};

src/server/server_family.cc

@@ -2196,7 +2196,7 @@ void ServerFamily::ReplicaOfInternal(string_view host, string_view port_sv, Conn
 
   // We should not execute replica of command while loading from snapshot.
   if (ServerState::tlocal()->is_master && service_.GetGlobalState() == GlobalState::LOADING) {
-    cntx->SendError("Can not execute during LOADING");
+    cntx->SendError(kLoadingErr);
     return;
   }
 

tests/dragonfly/replication_test.py

@@ -1859,8 +1859,8 @@ async def test_replicaof_reject_on_load(df_local_factory, df_seeder_factory):
     try:
         await c_replica.execute_command(f"REPLICAOF localhost {master.port}")
         assert False
-    except aioredis.ResponseError as e:
-        assert "Can not execute during LOADING" in str(e)
+    except aioredis.BusyLoadingError as e:
+        assert "Dragonfly is loading the dataset in memory" in str(e)
     # Check one we finish loading snapshot replicaof success
     await wait_available_async(c_replica)
     await c_replica.execute_command(f"REPLICAOF localhost {master.port}")

tests/dragonfly/utility.py

@@ -52,7 +52,8 @@ async def wait_available_async(client: aioredis.Redis, timeout=10):
             if "MOVED" in str(e):
                 # MOVED means we *can* serve traffic, but 'key' does not belong to an owned slot
                 return
-            assert "Can not execute during LOADING" in str(e)
+        except aioredis.BusyLoadingError as e:
+            assert "Dragonfly is loading the dataset in memory" in str(e)
 
         # Print W to indicate test is waiting for replica
         print("W", end="", flush=True)