Cleanup around script_caller, fix tracking of scripts and ACL logging for RM_Call (#11770)

* Make it clear that current_client is the root client that was called by
  external connection
* add executing_client which is the client that runs the current command
  (can be a module or a script)
* Remove script_caller that was used for commands that have CLIENT_SCRIPT
  to get the client that called the script. in most cases, that's the current_client,
  and in others (when being called from a module), it could be an intermediate
  client when we actually want the original one used by the external connection.

bugfixes:
* RM_Call with C flag should log ACL errors with the requested user rather than
  the one used by the original client, this also solves a crash when RM_Call is used
  with C flag from a detached thread safe context.
* addACLLogEntry would have logged info about the script_caller, but in case the
  script was issued by a module command we actually want the current_client. the
  exception is when RM_Call is called from a timer event, in which case we don't
  have a current_client.

behavior changes:
* client side tracking for scripts now tracks the keys that are read by the script
  instead of the keys that are declared by the caller for EVAL

other changes:
* Log both current_client and executing_client in the crash log.
* remove prepareLuaClient and resetLuaClient, being dead code that was forgotten.
* remove scriptTimeSnapshot and snapshot_time and instead add cmd_time_snapshot
  that serves all commands and is reset only when execution nesting starts.
* remove code to propagate CLIENT_FORCE_REPL from the executed command
  to the script caller since scripts aren't propagated anyway these days and anyway
  this flag wouldn't have had an effect since CLIENT_PREVENT_PROP is added by scriptResetRun.
* fix a module GIL violation issue in afterSleep that was introduced in #10300 (unreleased)

1 files changed, 99 insertions, 0 deletions

diff --git a/tests/modules/usercall.c b/tests/modules/usercall.c
index 9eb84ef39..6b23974d4 100644
--- a/tests/modules/usercall.c
+++ b/tests/modules/usercall.c
@@ -1,4 +1,8 @@
 #include "redismodule.h"
+#include <pthread.h>
+#include <assert.h>
+
+#define UNUSED(V) ((void) V)
 
 RedisModuleUser *user = NULL;
 
@@ -103,6 +107,98 @@ int reset_user(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     return REDISMODULE_OK;
 }
 
+typedef struct {
+    RedisModuleString **argv;
+    int argc;
+    RedisModuleBlockedClient *bc;
+} bg_call_data;
+
+void *bg_call_worker(void *arg) {
+    bg_call_data *bg = arg;
+
+    // Get Redis module context
+    RedisModuleCtx *ctx = RedisModule_GetThreadSafeContext(bg->bc);
+
+    // Acquire GIL
+    RedisModule_ThreadSafeContextLock(ctx);
+
+    // Set user
+    RedisModule_SetContextUser(ctx, user);
+
+    // Call the command
+    size_t format_len;
+    RedisModuleString *format_redis_str = RedisModule_CreateString(NULL, "v", 1);
+    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);
+    format = RedisModule_StringPtrLen(format_redis_str, NULL);
+    const char *cmd = RedisModule_StringPtrLen(bg->argv[2], NULL);
+    RedisModuleCallReply *rep = RedisModule_Call(ctx, cmd, format, bg->argv + 3, bg->argc - 3);
+    RedisModule_FreeString(NULL, format_redis_str);
+
+    // Release GIL
+    RedisModule_ThreadSafeContextUnlock(ctx);
+
+    // Reply to client
+    if (!rep) {
+        RedisModule_ReplyWithError(ctx, "NULL reply returned");
+    } else {
+        RedisModule_ReplyWithCallReply(ctx, rep);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    // Unblock client
+    RedisModule_UnblockClient(bg->bc, NULL);
+
+    /* Free the arguments */
+    for (int i=0; i<bg->argc; i++)
+        RedisModule_FreeString(ctx, bg->argv[i]);
+    RedisModule_Free(bg->argv);
+    RedisModule_Free(bg);
+
+    // Free the Redis module context
+    RedisModule_FreeThreadSafeContext(ctx);
+
+    return NULL;
+}
+
+int call_with_user_bg(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
+{
+    UNUSED(argv);
+    UNUSED(argc);
+
+    /* Make sure we're not trying to block a client when we shouldn't */
+    int flags = RedisModule_GetContextFlags(ctx);
+    int allFlags = RedisModule_GetContextFlagsAll();
+    if ((allFlags & REDISMODULE_CTX_FLAGS_MULTI) &&
+        (flags & REDISMODULE_CTX_FLAGS_MULTI)) {
+        RedisModule_ReplyWithSimpleString(ctx, "Blocked client is not supported inside multi");
+        return REDISMODULE_OK;
+    }
+    if ((allFlags & REDISMODULE_CTX_FLAGS_DENY_BLOCKING) &&
+        (flags & REDISMODULE_CTX_FLAGS_DENY_BLOCKING)) {
+        RedisModule_ReplyWithSimpleString(ctx, "Blocked client is not allowed");
+        return REDISMODULE_OK;
+    }
+
+    /* Make a copy of the arguments and pass them to the thread. */
+    bg_call_data *bg = RedisModule_Alloc(sizeof(bg_call_data));
+    bg->argv = RedisModule_Alloc(sizeof(RedisModuleString*)*argc);
+    bg->argc = argc;
+    for (int i=0; i<argc; i++)
+        bg->argv[i] = RedisModule_HoldString(ctx, argv[i]);
+
+    /* Block the client */
+    bg->bc = RedisModule_BlockClient(ctx, NULL, NULL, NULL, 0);
+
+    /* Start a thread to handle the request */
+    pthread_t tid;
+    int res = pthread_create(&tid, NULL, bg_call_worker, bg);
+    assert(res == 0);
+
+    return REDISMODULE_OK;
+}
+
 int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     REDISMODULE_NOT_USED(argv);
     REDISMODULE_NOT_USED(argc);
@@ -116,6 +212,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
     if (RedisModule_CreateCommand(ctx,"usercall.call_with_user_flag", call_with_user_flag,"write",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx, "usercall.call_with_user_bg", call_with_user_bg, "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     if (RedisModule_CreateCommand(ctx, "usercall.add_to_acl", add_to_acl, "write",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;