Fix crash on RM_Call with script mode. (#10886)

The PR fixes 2 issues:

### RM_Call crash on script mode

`RM_Call` can potentially be called from a background thread where `server.current_client`
are not set. In such case we get a crash on `NULL` dereference.
The fix is to check first if `server.current_client` is `NULL`, if it does we should
verify disc errors and readonly replica as we do to any normal clients (no masters nor AOF).

### RM_Call block OOM commands when not needed

Again `RM_Call` can be executed on a background thread using a `ThreadSafeCtx`.
In such case `server.pre_command_oom_state` can be irrelevant and should not be
considered when check OOM state. This cause OOM commands to be blocked when
not necessarily needed.

In such case, check the actual used memory (and not the cached value). Notice that in
order to know if the cached value can be used, we check that the ctx that was used on
the `RM_Call` is a ThreadSafeCtx. Module writer can potentially abuse the API and use
ThreadSafeCtx on the main thread. We consider this as a API miss used.

src/module.c

@@ -5783,7 +5783,16 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
 
     if (flags & REDISMODULE_ARGV_RESPECT_DENY_OOM) {
         if (cmd->flags & CMD_DENYOOM) {
-            if (server.pre_command_oom_state) {
+            int oom_state;
+            if (ctx->flags & REDISMODULE_CTX_THREAD_SAFE) {
+                /* On background thread we can not count on server.pre_command_oom_state.
+                 * Because it is only set on the main thread, in such case we will check
+                 * the actual memory usage. */
+                oom_state = (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_ERR);
+            } else {
+                oom_state = server.pre_command_oom_state;
+            }
+            if (oom_state) {
                 errno = ENOSPC;
                 if (error_as_call_replies) {
                     sds msg = sdsdup(shared.oomerr->ptr);
@@ -5823,7 +5832,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
             }
 
             int deny_write_type = writeCommandsDeniedByDiskError();
-            int obey_client = mustObeyClient(server.current_client);
+            int obey_client = (server.current_client && mustObeyClient(server.current_client));
 
             if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) {
                 errno = ESPIPE;

tests/modules/blockedclient.c

@@ -7,6 +7,7 @@
 #include <assert.h>
 #include <stdio.h>
 #include <pthread.h>
+#include <strings.h>
 
 #define UNUSED(V) ((void) V)
 
@@ -119,8 +120,20 @@ void *bg_call_worker(void *arg) {
     }
 
     // Call the command
-    const char* cmd = RedisModule_StringPtrLen(bg->argv[1], NULL);
-    RedisModuleCallReply* rep = RedisModule_Call(ctx, cmd, "v", bg->argv + 2, bg->argc - 2);
+    const char *module_cmd = RedisModule_StringPtrLen(bg->argv[0], NULL);
+    int cmd_pos = 1;
+    RedisModuleString *format_redis_str = RedisModule_CreateString(NULL, "v", 1);
+    if (!strcasecmp(module_cmd, "do_bg_rm_call_format")) {
+        cmd_pos = 2;
+        size_t format_len;
+        const char *format = RedisModule_StringPtrLen(bg->argv[1], &format_len);
+        RedisModule_StringAppendBuffer(NULL, format_redis_str, format, format_len);
+        RedisModule_StringAppendBuffer(NULL, format_redis_str, "E", 1);
+    }
+    const char *format = RedisModule_StringPtrLen(format_redis_str, NULL);
+    const char *cmd = RedisModule_StringPtrLen(bg->argv[cmd_pos], NULL);
+    RedisModuleCallReply *rep = RedisModule_Call(ctx, cmd, format, bg->argv + cmd_pos + 1, bg->argc - cmd_pos - 1);
+    RedisModule_FreeString(NULL, format_redis_str);
 
     // Release GIL
     RedisModule_ThreadSafeContextUnlock(ctx);
@@ -306,6 +319,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     if (RedisModule_CreateCommand(ctx, "do_bg_rm_call", do_bg_rm_call, "", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx, "do_bg_rm_call_format", do_bg_rm_call, "", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     if (RedisModule_CreateCommand(ctx, "do_fake_bg_true", do_fake_bg_true, "", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 

tests/unit/moduleapi/blockedclient.tcl

@@ -76,6 +76,19 @@ start_server {tags {"modules"}} {
         r do_bg_rm_call hgetall hash
     } {foo bar}
 
+    test {RM_Call from blocked client with script mode} {
+        r do_bg_rm_call_format S hset k foo bar
+    } {1}
+
+    test {RM_Call from blocked client with oom mode} {
+        r config set maxmemory 1
+        # will set server.pre_command_oom_state to 1
+        assert_error {OOM command not allowed*} {r hset hash foo bar}
+        r config set maxmemory 0
+        # now its should be OK to call OOM commands
+        r do_bg_rm_call_format M hset k1 foo bar
+    } {1} {needs:config-maxmemory}
+
     test {RESP version carries through to blocked client} {
         for {set client_proto 2} {$client_proto <= 3} {incr client_proto} {
             r hello $client_proto