Fix RM_Yield bug processing future commands of the current client. (#10573)

RM_Yield was missing a call to protectClient to prevent redis from
processing future commands of the yielding client.

Adding tests that fail without this fix.

This would be complicated to solve since nested calls to RM_Call used to
replace the current_client variable with the module temp client.

It looks like it's no longer necessary to do that, since it was added
back in #9890 to solve two issues, both already gone:
1. call to CONFIG SET maxmemory could trigger a module hook calling
   RM_Call. although this specific issue is gone, arguably other hooks
   like keyspace notification, can do the same.
2. an assertion in lookupKey that checks the current command of the
   current client, introduced in #9572 and removed in #10248

src/module.c

@@ -717,6 +717,8 @@ void moduleFreeContext(RedisModuleCtx *ctx) {
             if (server.busy_module_yield_flags) {
                 blockingOperationEnds();
                 server.busy_module_yield_flags = BUSY_MODULE_YIELD_NONE;
+                if (server.current_client)
+                    unprotectClient(server.current_client);
                 unblockPostponedClients();
             }
         }
@@ -2108,6 +2110,8 @@ void RM_Yield(RedisModuleCtx *ctx, int flags, const char *busy_reply) {
             if (!server.busy_module_yield_flags) {
                 server.busy_module_yield_flags = BUSY_MODULE_YIELD_EVENTS;
                 blockingOperationStarts();
+                if (server.current_client)
+                    protectClient(server.current_client);
             }
             if (flags & REDISMODULE_YIELD_FLAG_CLIENTS)
                 server.busy_module_yield_flags |= BUSY_MODULE_YIELD_CLIENTS;
@@ -2125,7 +2129,7 @@ void RM_Yield(RedisModuleCtx *ctx, int flags, const char *busy_reply) {
         /* decide when the next event should fire. */
         ctx->next_yield_time = now + 1000000 / server.hz;
     }
-    yield_nesting --;
+    yield_nesting--;
 }
 
 /* Set flags defining capabilities or behavior bit flags.
@@ -5909,11 +5913,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
         if (!(flags & REDISMODULE_ARGV_NO_REPLICAS))
             call_flags |= CMD_CALL_PROPAGATE_REPL;
     }
-    /* Set server.current_client */
-    client *old_client = server.current_client;
-    server.current_client = c;
     call(c,call_flags);
-    server.current_client = old_client;
     server.replication_allowed = prev_replication_allowed;
 
     serverAssert((c->flags & CLIENT_BLOCKED) == 0);
@@ -7650,6 +7650,8 @@ void moduleGILBeforeUnlock() {
     if (server.busy_module_yield_flags) {
         blockingOperationEnds();
         server.busy_module_yield_flags = BUSY_MODULE_YIELD_NONE;
+        if (server.current_client)
+            unprotectClient(server.current_client);
         unblockPostponedClients();
     }
 }

tests/unit/moduleapi/blockedclient.tcl

@@ -106,6 +106,10 @@ foreach call_type {nested normal} {
         }
         $rd flush
 
+        # send another command after the blocked one, to make sure we don't attempt to process it
+        $rd ping
+        $rd flush
+
         # make sure we get BUSY error, and that we didn't get it too early
         assert_error {*BUSY Slow module operation*} {r ping}
         assert_morethan_equal [expr [clock clicks -milliseconds]-$start] $busy_time_limit
@@ -117,7 +121,8 @@ foreach call_type {nested normal} {
         } else {
             fail "Failed waiting for busy command to end"
         }
-        $rd read
+        assert_equal [$rd read] "1"
+        assert_equal [$rd read] "PONG"
 
         # run command that blocks for 200ms
         set start [clock clicks -milliseconds]
@@ -151,6 +156,10 @@ foreach call_type {nested normal} {
         set start [clock clicks -milliseconds]
         $rd do_bg_rm_call hgetall hash
 
+        # send another command after the blocked one, to make sure we don't attempt to process it
+        $rd ping
+        $rd flush
+
         # wait till we know we're blocked inside the module
         wait_for_condition 50 100 {
             [r is_in_slow_bg_operation] eq 1
@@ -172,10 +181,10 @@ foreach call_type {nested normal} {
         assert_equal [r ping] {PONG}
 
         r config set busy-reply-threshold $old_time_limit
-        set res [$rd read]
+        assert_equal [$rd read] {foo bar}
+        assert_equal [$rd read] {PONG}
         $rd close
-        set _ $res
-    } {foo bar}
+    }
 
     test {blocked client reaches client output buffer limit} {
         r hset hash big [string repeat x 50000]