Support for RM_Call on blocking commands (#11568)

Allow running blocking commands from within a module using `RM_Call`.

Today, when `RM_Call` is used, the fake client that is used to run command
is marked with `CLIENT_DENY_BLOCKING` flag. This flag tells the command
that it is not allowed to block the client and in case it needs to block, it must
fallback to some alternative (either return error or perform some default behavior).
For example, `BLPOP` fallback to simple `LPOP` if it is not allowed to block.

All the commands must respect the `CLIENT_DENY_BLOCKING` flag (including
module commands). When the command invocation finished, Redis asserts that
the client was not blocked.

This PR introduces the ability to call blocking command using `RM_Call` by
passing a callback that will be called when the client will get unblocked.
In order to do that, the user must explicitly say that he allow to perform blocking
command by passing a new format specifier argument, `K`, to the `RM_Call`
function. This new flag will tell Redis that it is allow to run blocking command
and block the client. In case the command got blocked, Redis will return a new
type of call reply (`REDISMODULE_REPLY_PROMISE`). This call reply indicates
that the command got blocked and the user can set the on_unblocked handler using
`RM_CallReplyPromiseSetUnblockHandler`.

When clients gets unblocked, it eventually reaches `processUnblockedClients` function.
This is where we check if the client is a fake module client and if it is, we call the unblock
callback instead of performing the usual unblock operations.

**Notice**: `RM_CallReplyPromiseSetUnblockHandler` must be called atomically
along side the command invocation (without releasing the Redis lock in between).
In addition, unlike other CallReply types, the promise call reply must be released
by the module when the Redis GIL is acquired.

The module can abort the execution on the blocking command (if it was not yet
executed) using `RM_CallReplyPromiseAbort`. the API will return `REDISMODULE_OK`
on success and `REDISMODULE_ERR` if the operation is already executed.
**Notice** that in case of misbehave module, Abort might finished successfully but the
operation will not really be aborted. This can only happened if the module do not respect
the disconnect callback of the blocked client. 
For pure Redis commands this can not happened.

### Atomicity Guarantees

The API promise that the unblock handler will run atomically as an execution unit.
This means that all the operation performed on the unblock handler will be wrapped
with a multi exec transaction when replicated to the replica and AOF.
The API **do not** grantee any other atomicity properties such as when the unblock
handler will be called. This gives us the flexibility to strengthen the grantees (or not)
in the future if we will decide that we need a better guarantees.

That said, the implementation **does** provide a better guarantees when performing
pure Redis blocking command like `BLPOP`. In this case the unblock handler will run
atomically with the operation that got unblocked (for example, in case of `BLPOP`, the
unblock handler will run atomically with the `LPOP` operation that run when the command
got unblocked). This is an implementation detail that might be change in the future and the
module writer should not count on that.

### Calling blocking commands while running on script mode (`S`)

`RM_Call` script mode (`S`) was introduced on #0372. It is used for usecases where the
command that was invoked on `RM_Call` comes from a user input and we want to make
sure the user will not run dangerous commands like `shutdown`. Some command, such
as `BLPOP`, are marked with `NO_SCRIPT` flag, which means they will not be allowed on
script mode. Those commands are marked with  `NO_SCRIPT` just because they are
blocking commands and not because they are dangerous. Now that we can run blocking
commands on RM_Call, there is no real reason not to allow such commands on script mode.

The underline problem is that the `NO_SCRIPT` flag is abused to also mark some of the
blocking commands (notice that those commands know not to block the client if it is not
allowed to do so, and have a fallback logic to such cases. So even if those commands
were not marked with `NO_SCRIPT` flag, it would not harm Redis, and today we can
already run those commands within multi exec).

In addition, not all blocking commands are marked with `NO_SCRIPT` flag, for example
`blmpop` are not marked and can run from within a script.

Those facts shows that there are some ambiguity about the meaning of the `NO_SCRIPT`
flag, and its not fully clear where it should be use.

The PR suggest that blocking commands should not be marked with `NO_SCRIPT` flag,
those commands should handle `CLIENT_DENY_BLOCKING` flag and only block when
it's safe (like they already does today). To achieve that, the PR removes the `NO_SCRIPT`
flag from the following commands:
* `blmove`
* `blpop`
* `brpop`
* `brpoplpush`
* `bzpopmax`
* `bzpopmin`
* `wait`

This might be considered a breaking change as now, on scripts, instead of getting
`command is not allowed from script` error, the user will get some fallback behavior
base on the command implementation. That said, the change matches the behavior
of scripts and multi exec with respect to those commands and allow running them on
`RM_Call` even when script mode is used.

### Additional RedisModule API and changes

* `RM_BlockClientSetPrivateData` - Set private data on the blocked client without the
  need to unblock the client. This allows up to set the promise CallReply as the private
  data of the blocked client and abort it if the client gets disconnected.
* `RM_BlockClientGetPrivateData` - Return the current private data set on a blocked client.
  We need it so we will have access to this private data on the disconnect callback.
* On RM_Call, the returned reply will be added to the auto memory context only if auto
  memory is enabled, this allows us to keep the call reply for longer time then the context
  lifetime and does not force an unneeded borrow relationship between the CallReply and
  the RedisModuleContext.

3 files changed, 689 insertions, 30 deletions

diff --git a/tests/modules/blockedclient.c b/tests/modules/blockedclient.c
index f4234b0d3..db3e80b29 100644
--- a/tests/modules/blockedclient.c
+++ b/tests/modules/blockedclient.c
@@ -219,6 +219,257 @@ int do_rm_call(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
     return REDISMODULE_OK;
 }
 
+static void rm_call_async_send_reply(RedisModuleCtx *ctx, RedisModuleCallReply *reply) {
+    RedisModule_ReplyWithCallReply(ctx, reply);
+    RedisModule_FreeCallReply(reply);
+}
+
+/* Called when the command that was blocked on 'RM_Call' gets unblocked
+ * and send the reply to the blocked client. */
+static void rm_call_async_on_unblocked(RedisModuleCtx *ctx, RedisModuleCallReply *reply, void *private_data) {
+    UNUSED(ctx);
+    RedisModuleBlockedClient *bc = private_data;
+    RedisModuleCtx *bctx = RedisModule_GetThreadSafeContext(bc);
+    rm_call_async_send_reply(bctx, reply);
+    RedisModule_FreeThreadSafeContext(bctx);
+    RedisModule_UnblockClient(bc, RedisModule_BlockClientGetPrivateData(bc));
+}
+
+int do_rm_call_async_fire_and_forget(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
+    UNUSED(argv);
+    UNUSED(argc);
+
+    if(argc < 2){
+        return RedisModule_WrongArity(ctx);
+    }
+    const char* cmd = RedisModule_StringPtrLen(argv[1], NULL);
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, cmd, "!KEv", argv + 2, argc - 2);
+
+    if(RedisModule_CallReplyType(rep) != REDISMODULE_REPLY_PROMISE) {
+        RedisModule_ReplyWithCallReply(ctx, rep);
+    } else {
+        RedisModule_ReplyWithSimpleString(ctx, "Blocked");
+    }
+    RedisModule_FreeCallReply(rep);
+
+    return REDISMODULE_OK;
+}
+
+static void do_rm_call_async_free_pd(RedisModuleCtx * ctx, void *pd) {
+    UNUSED(ctx);
+    RedisModule_FreeCallReply(pd);
+}
+
+static void do_rm_call_async_disconnect(RedisModuleCtx *ctx, struct RedisModuleBlockedClient *bc) {
+    UNUSED(ctx);
+    RedisModuleCallReply* rep = RedisModule_BlockClientGetPrivateData(bc);
+    RedisModule_CallReplyPromiseAbort(rep, NULL);
+    RedisModule_FreeCallReply(rep);
+    RedisModule_AbortBlock(bc);
+}
+
+/*
+ * Callback for do_rm_call_async / do_rm_call_async_script_mode
+ * Gets the command to invoke as the first argument to the command and runs it,
+ * passing the rest of the arguments to the command invocation.
+ * If the command got blocked, blocks the client and unblock it when the command gets unblocked,
+ * this allows check the K (allow blocking) argument to RM_Call.
+ */
+int do_rm_call_async(RedisModuleCtx *ctx, RedisModuleString **argv, int argc){
+    UNUSED(argv);
+    UNUSED(argc);
+
+    if(argc < 2){
+        return RedisModule_WrongArity(ctx);
+    }
+
+    size_t format_len = 0;
+    char format[6] = {0};
+
+    if (!(RedisModule_GetContextFlags(ctx) & REDISMODULE_CTX_FLAGS_DENY_BLOCKING)) {
+        /* We are allowed to block the client so we can allow RM_Call to also block us */
+        format[format_len++] = 'K';
+    }
+
+    const char* invoked_cmd = RedisModule_StringPtrLen(argv[0], NULL);
+    if (strcasecmp(invoked_cmd, "do_rm_call_async_script_mode") == 0) {
+        format[format_len++] = 'S';
+    }
+
+    format[format_len++] = 'E';
+    format[format_len++] = 'v';
+    if (strcasecmp(invoked_cmd, "do_rm_call_async_no_replicate") != 0) {
+        /* Notice, without the '!' flag we will have inconsistency between master and replica.
+         * This is used only to check '!' flag correctness on blocked commands. */
+        format[format_len++] = '!';
+    }
+
+    const char* cmd = RedisModule_StringPtrLen(argv[1], NULL);
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, cmd, format, argv + 2, argc - 2);
+
+    if(RedisModule_CallReplyType(rep) != REDISMODULE_REPLY_PROMISE) {
+        rm_call_async_send_reply(ctx, rep);
+    } else {
+        RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, NULL, NULL, do_rm_call_async_free_pd, 0);
+        RedisModule_SetDisconnectCallback(bc, do_rm_call_async_disconnect);
+        RedisModule_BlockClientSetPrivateData(bc, rep);
+        RedisModule_CallReplyPromiseSetUnblockHandler(rep, rm_call_async_on_unblocked, bc);
+    }
+
+    return REDISMODULE_OK;
+}
+
+/* Private data for wait_and_do_rm_call_async that holds information about:
+ * 1. the block client, to unblock when done.
+ * 2. the arguments, contains the command to run using RM_Call */
+typedef struct WaitAndDoRMCallCtx {
+    RedisModuleBlockedClient *bc;
+    RedisModuleString **argv;
+    int argc;
+} WaitAndDoRMCallCtx;
+
+/*
+ * This callback will be called when the 'wait' command invoke on 'wait_and_do_rm_call_async' will finish.
+ * This callback will continue the execution flow just like 'do_rm_call_async' command.
+ */
+static void wait_and_do_rm_call_async_on_unblocked(RedisModuleCtx *ctx, RedisModuleCallReply *reply, void *private_data) {
+    WaitAndDoRMCallCtx *wctx = private_data;
+    if (RedisModule_CallReplyType(reply) != REDISMODULE_REPLY_INTEGER) {
+        goto done;
+    }
+
+    if (RedisModule_CallReplyInteger(reply) != 1) {
+        goto done;
+    }
+
+    RedisModule_FreeCallReply(reply);
+    reply = NULL;
+
+    const char* cmd = RedisModule_StringPtrLen(wctx->argv[0], NULL);
+    reply = RedisModule_Call(ctx, cmd, "!EKv", wctx->argv + 1, wctx->argc - 1);
+
+done:
+    if(RedisModule_CallReplyType(reply) != REDISMODULE_REPLY_PROMISE) {
+        RedisModuleCtx *bctx = RedisModule_GetThreadSafeContext(wctx->bc);
+        rm_call_async_send_reply(bctx, reply);
+        RedisModule_FreeThreadSafeContext(bctx);
+        RedisModule_UnblockClient(wctx->bc, NULL);
+    } else {
+        RedisModule_CallReplyPromiseSetUnblockHandler(reply, rm_call_async_on_unblocked, wctx->bc);
+        RedisModule_FreeCallReply(reply);
+    }
+    for (int i = 0 ; i < wctx->argc ; ++i) {
+        RedisModule_FreeString(NULL, wctx->argv[i]);
+    }
+    RedisModule_Free(wctx->argv);
+    RedisModule_Free(wctx);
+}
+
+/*
+ * Callback for wait_and_do_rm_call
+ * Gets the command to invoke as the first argument, runs 'wait'
+ * command (using the K flag to RM_Call). Once the wait finished, runs the
+ * command that was given (just like 'do_rm_call_async').
+ */
+int wait_and_do_rm_call_async(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    UNUSED(argv);
+    UNUSED(argc);
+
+    if(argc < 2){
+        return RedisModule_WrongArity(ctx);
+    }
+
+    int flags = RedisModule_GetContextFlags(ctx);
+    if (flags & REDISMODULE_CTX_FLAGS_DENY_BLOCKING) {
+        return RedisModule_ReplyWithError(ctx, "Err can not run wait, blocking is not allowed.");
+    }
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, "wait", "!EKcc", "1", "0");
+    if(RedisModule_CallReplyType(rep) != REDISMODULE_REPLY_PROMISE) {
+        rm_call_async_send_reply(ctx, rep);
+    } else {
+        RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, NULL, NULL, NULL, 0);
+        WaitAndDoRMCallCtx *wctx = RedisModule_Alloc(sizeof(*wctx));
+        *wctx = (WaitAndDoRMCallCtx){
+                .bc = bc,
+                .argv = RedisModule_Alloc((argc - 1) * sizeof(RedisModuleString*)),
+                .argc = argc - 1,
+        };
+
+        for (int i = 1 ; i < argc ; ++i) {
+            wctx->argv[i - 1] = RedisModule_HoldString(NULL, argv[i]);
+        }
+        RedisModule_CallReplyPromiseSetUnblockHandler(rep, wait_and_do_rm_call_async_on_unblocked, wctx);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    return REDISMODULE_OK;
+}
+
+static void blpop_and_set_multiple_keys_on_unblocked(RedisModuleCtx *ctx, RedisModuleCallReply *reply, void *private_data) {
+    /* ignore the reply */
+    RedisModule_FreeCallReply(reply);
+    WaitAndDoRMCallCtx *wctx = private_data;
+    for (int i = 0 ; i < wctx->argc ; i += 2) {
+        RedisModuleCallReply* rep = RedisModule_Call(ctx, "set", "!ss", wctx->argv[i], wctx->argv[i + 1]);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    RedisModuleCtx *bctx = RedisModule_GetThreadSafeContext(wctx->bc);
+    RedisModule_ReplyWithSimpleString(bctx, "OK");
+    RedisModule_FreeThreadSafeContext(bctx);
+    RedisModule_UnblockClient(wctx->bc, NULL);
+
+    for (int i = 0 ; i < wctx->argc ; ++i) {
+        RedisModule_FreeString(NULL, wctx->argv[i]);
+    }
+    RedisModule_Free(wctx->argv);
+    RedisModule_Free(wctx);
+
+}
+
+/*
+ * Performs a blpop command on a given list and when unblocked set multiple string keys.
+ * This command allows checking that the unblock callback is performed as a unit
+ * and its effect are replicated to the replica and AOF wrapped with multi exec.
+ */
+int blpop_and_set_multiple_keys(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    UNUSED(argv);
+    UNUSED(argc);
+
+    if(argc < 2 || argc % 2 != 0){
+        return RedisModule_WrongArity(ctx);
+    }
+
+    int flags = RedisModule_GetContextFlags(ctx);
+    if (flags & REDISMODULE_CTX_FLAGS_DENY_BLOCKING) {
+        return RedisModule_ReplyWithError(ctx, "Err can not run wait, blocking is not allowed.");
+    }
+
+    RedisModuleCallReply* rep = RedisModule_Call(ctx, "blpop", "!EKsc", argv[1], "0");
+    if(RedisModule_CallReplyType(rep) != REDISMODULE_REPLY_PROMISE) {
+        rm_call_async_send_reply(ctx, rep);
+    } else {
+        RedisModuleBlockedClient *bc = RedisModule_BlockClient(ctx, NULL, NULL, NULL, 0);
+        WaitAndDoRMCallCtx *wctx = RedisModule_Alloc(sizeof(*wctx));
+        *wctx = (WaitAndDoRMCallCtx){
+                .bc = bc,
+                .argv = RedisModule_Alloc((argc - 2) * sizeof(RedisModuleString*)),
+                .argc = argc - 2,
+        };
+
+        for (int i = 0 ; i < argc - 2 ; ++i) {
+            wctx->argv[i] = RedisModule_HoldString(NULL, argv[i + 2]);
+        }
+        RedisModule_CallReplyPromiseSetUnblockHandler(rep, blpop_and_set_multiple_keys_on_unblocked, wctx);
+        RedisModule_FreeCallReply(rep);
+    }
+
+    return REDISMODULE_OK;
+}
+
 /* simulate a blocked client replying to a thread safe context without creating a thread */
 int do_fake_bg_true(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     UNUSED(argv);
@@ -316,6 +567,30 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc)
                                   "write", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx, "do_rm_call_async", do_rm_call_async,
+                                  "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "do_rm_call_async_script_mode", do_rm_call_async,
+                                  "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "do_rm_call_async_no_replicate", do_rm_call_async,
+                                  "write", 0, 0, 0) == REDISMODULE_ERR)
+            return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "do_rm_call_fire_and_forget", do_rm_call_async_fire_and_forget,
+                                  "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "wait_and_do_rm_call", wait_and_do_rm_call_async,
+                                  "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx, "blpop_and_set_multiple_keys", blpop_and_set_multiple_keys,
+                                      "write", 0, 0, 0) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     if (RedisModule_CreateCommand(ctx, "do_bg_rm_call", do_bg_rm_call, "", 0, 0, 0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
diff --git a/tests/unit/moduleapi/async_rm_call.tcl b/tests/unit/moduleapi/async_rm_call.tcl
new file mode 100644
index 000000000..142b098f0
--- /dev/null
+++ b/tests/unit/moduleapi/async_rm_call.tcl
@@ -0,0 +1,380 @@
+set testmodule [file normalize tests/modules/blockedclient.so]
+set testmodule2 [file normalize tests/modules/postnotifications.so]
+set testmodule3 [file normalize tests/modules/blockonkeys.so]
+
+start_server {tags {"modules"}} {
+    r module load $testmodule
+
+    test {Locked GIL acquisition from async RM_Call} {
+        assert_equal {OK} [r do_rm_call_async acquire_gil]
+    }
+
+    test "Blpop on async RM_Call fire and forget" {
+        assert_equal {Blocked} [r do_rm_call_fire_and_forget blpop l 0]
+        r lpush l a
+        assert_equal {0} [r llen l]
+    }
+
+    foreach cmd {do_rm_call_async do_rm_call_async_script_mode} {
+
+        test "Blpop on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd blpop l 0
+            wait_for_blocked_clients_count 1
+            r lpush l a
+            assert_equal [$rd read] {l a}
+            wait_for_blocked_clients_count 0
+        }
+
+        test "Brpop on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd brpop l 0
+            wait_for_blocked_clients_count 1
+            r lpush l a
+            assert_equal [$rd read] {l a}
+            wait_for_blocked_clients_count 0
+        }
+
+        test "Brpoplpush on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd brpoplpush l1 l2 0
+            wait_for_blocked_clients_count 1
+            r lpush l1 a
+            assert_equal [$rd read] {a}
+            wait_for_blocked_clients_count 0
+            r lpop l2
+        } {a}
+
+        test "Blmove on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd blmove l1 l2 LEFT LEFT 0
+            wait_for_blocked_clients_count 1
+            r lpush l1 a
+            assert_equal [$rd read] {a}
+            wait_for_blocked_clients_count 0
+            r lpop l2
+        } {a}
+
+        test "Bzpopmin on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd bzpopmin s 0
+            wait_for_blocked_clients_count 1
+            r zadd s 10 foo
+            assert_equal [$rd read] {s foo 10}
+            wait_for_blocked_clients_count 0
+        }
+
+        test "Bzpopmax on async RM_Call using $cmd" {
+            set rd [redis_deferring_client]
+
+            $rd $cmd bzpopmax s 0
+            wait_for_blocked_clients_count 1
+            r zadd s 10 foo
+            assert_equal [$rd read] {s foo 10}
+            wait_for_blocked_clients_count 0
+        }
+    }
+
+    test {Nested async RM_Call} {
+        set rd [redis_deferring_client]
+
+        $rd do_rm_call_async do_rm_call_async do_rm_call_async do_rm_call_async blpop l 0
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {l a}
+        wait_for_blocked_clients_count 0
+    }
+
+    test {Test multiple async RM_Call waiting on the same event} {
+        set rd1 [redis_deferring_client]
+        set rd2 [redis_deferring_client]
+
+        $rd1 do_rm_call_async do_rm_call_async do_rm_call_async do_rm_call_async blpop l 0
+        $rd2 do_rm_call_async do_rm_call_async do_rm_call_async do_rm_call_async blpop l 0
+        wait_for_blocked_clients_count 2
+        r lpush l element element
+        assert_equal [$rd1 read] {l element}
+        assert_equal [$rd2 read] {l element}
+        wait_for_blocked_clients_count 0
+    }
+
+    test {async RM_Call calls RM_Call} {
+        assert_equal {PONG} [r do_rm_call_async do_rm_call ping]
+    }
+
+    test {async RM_Call calls background RM_Call calls RM_Call} {
+        assert_equal {PONG} [r do_rm_call_async do_bg_rm_call do_rm_call ping]
+    }
+
+    test {async RM_Call calls background RM_Call calls RM_Call calls async RM_Call} {
+        assert_equal {PONG} [r do_rm_call_async do_bg_rm_call do_rm_call do_rm_call_async ping]
+    }
+
+    test {async RM_Call inside async RM_Call callback} {
+        set rd [redis_deferring_client]
+        $rd wait_and_do_rm_call blpop l 0
+        wait_for_blocked_clients_count 1
+
+        start_server {} {
+            test "Connect a replica to the master instance" {
+                r slaveof [srv -1 host] [srv -1 port]
+                wait_for_condition 50 100 {
+                    [s role] eq {slave} &&
+                    [string match {*master_link_status:up*} [r info replication]]
+                } else {
+                    fail "Can't turn the instance into a replica"
+                }
+            }
+
+            assert_equal {1} [r -1 lpush l a]
+            assert_equal [$rd read] {l a}
+        }
+
+        wait_for_blocked_clients_count 0
+    }
+
+    test {Become replica while having async RM_Call running} {
+        r flushall
+        set rd [redis_deferring_client]
+        $rd do_rm_call_async blpop l 0
+        wait_for_blocked_clients_count 1
+
+        #become a replica of a not existing redis
+        r replicaof localhost 30000
+
+        catch {[$rd read]} e
+        assert_match {UNBLOCKED force unblock from blocking operation*} $e
+        wait_for_blocked_clients_count 0
+
+        r replicaof no one
+
+        r lpush l 1
+        # make sure the async rm_call was aborted
+        assert_equal [r llen l] {1}
+    }
+
+    test {Pipeline with blocking RM_Call} {
+        r flushall
+        set rd [redis_deferring_client]
+        set buf ""
+        append buf "do_rm_call_async blpop l 0\r\n"
+        append buf "ping\r\n"
+        $rd write $buf
+        $rd flush
+        wait_for_blocked_clients_count 1
+
+        # release the blocked client
+        r lpush l 1
+
+        assert_equal [$rd read] {l 1}
+        assert_equal [$rd read] {PONG}
+
+        wait_for_blocked_clients_count 0
+    }
+
+    test {blocking RM_Call abort} {
+        r flushall
+        set rd [redis_deferring_client]
+        
+        $rd client id
+        set client_id [$rd read]
+
+        $rd do_rm_call_async blpop l 0
+        wait_for_blocked_clients_count 1
+
+        r client kill ID $client_id
+        assert_error {*error reading reply*} {$rd read}
+
+        wait_for_blocked_clients_count 0
+
+        r lpush l 1
+        # make sure the async rm_call was aborted
+        assert_equal [r llen l] {1}
+    }
+}
+
+start_server {tags {"modules"}} {
+    r module load $testmodule
+
+    test {Test basic replication stream on unblock handler} {
+        r flushall
+        set repl [attach_to_replication_stream]
+
+        set rd [redis_deferring_client]
+
+        $rd do_rm_call_async blpop l 0
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {l a}
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush l a}
+            {lpop l}
+        }
+        close_replication_stream $repl
+
+        wait_for_blocked_clients_count 0
+    }
+
+    test {Test unblock handler are executed as a unit} {
+        r flushall
+        set repl [attach_to_replication_stream]
+
+        set rd [redis_deferring_client]
+
+        $rd blpop_and_set_multiple_keys l x 1 y 2
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {OK}
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush l a}
+            {multi}
+            {lpop l}
+            {set x 1}
+            {set y 2}
+            {exec}
+        }
+        close_replication_stream $repl
+
+        wait_for_blocked_clients_count 0
+    }
+
+    test {Test no propagation of blocking command} {
+        r flushall
+        set repl [attach_to_replication_stream]
+
+        set rd [redis_deferring_client]
+
+        $rd do_rm_call_async_no_replicate blpop l 0
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {l a}
+
+        # make sure the lpop are not replicated
+        r set x 1
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush l a}
+            {set x 1}
+        }
+        close_replication_stream $repl
+
+        wait_for_blocked_clients_count 0
+    }
+}
+
+start_server {tags {"modules"}} {
+    r module load $testmodule
+    r module load $testmodule2
+
+    test {Test unblock handler are executed as a unit with key space notifications} {
+        r flushall
+        set repl [attach_to_replication_stream]
+
+        set rd [redis_deferring_client]
+
+        $rd blpop_and_set_multiple_keys l string_foo 1 string_bar 2
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {OK}
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush l a}
+            {multi}
+            {lpop l}
+            {set string_foo 1}
+            {set string_bar 2}
+            {incr string_changed{string_foo}}
+            {incr string_changed{string_bar}}
+            {incr string_total}
+            {incr string_total}
+            {exec}
+        }
+        close_replication_stream $repl
+
+        wait_for_blocked_clients_count 0
+    }
+
+    test {Test unblock handler are executed as a unit with lazy expire} {
+        r flushall
+        r DEBUG SET-ACTIVE-EXPIRE 0
+        set repl [attach_to_replication_stream]
+
+        set rd [redis_deferring_client]
+
+        $rd blpop_and_set_multiple_keys l string_foo 1 string_bar 2
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {OK}
+
+        # set expiration on string_foo
+        r pexpire string_foo 1
+        after 10
+
+        # now the key should have been expired
+        $rd blpop_and_set_multiple_keys l string_foo 1 string_bar 2
+        wait_for_blocked_clients_count 1
+        r lpush l a
+        assert_equal [$rd read] {OK}
+
+        assert_replication_stream $repl {
+            {select *}
+            {lpush l a}
+            {multi}
+            {lpop l}
+            {set string_foo 1}
+            {set string_bar 2}
+            {incr string_changed{string_foo}}
+            {incr string_changed{string_bar}}
+            {incr string_total}
+            {incr string_total}
+            {exec}
+            {pexpireat string_foo *}
+            {lpush l a}
+            {multi}
+            {lpop l}
+            {del string_foo}
+            {set string_foo 1}
+            {set string_bar 2}
+            {incr expired}
+            {incr string_changed{string_foo}}
+            {incr string_changed{string_bar}}
+            {incr string_total}
+            {incr string_total}
+            {exec}
+        }
+        close_replication_stream $repl
+        r DEBUG SET-ACTIVE-EXPIRE 1
+    }
+
+    wait_for_blocked_clients_count 0
+}
+
+start_server {tags {"modules"}} {
+    r module load $testmodule
+    r module load $testmodule3
+
+    test {Test unblock handler on module blocked on keys} {
+        set rd [redis_deferring_client]
+
+        r fsl.push l 1
+        $rd do_rm_call_async FSL.BPOPGT l 3 0
+        wait_for_blocked_clients_count 1
+        r fsl.push l 2
+        r fsl.push l 3
+        r fsl.push l 4
+        assert_equal [$rd read] {4}
+
+        wait_for_blocked_clients_count 0
+    }
+}
\ No newline at end of file
diff --git a/tests/unit/scripting.tcl b/tests/unit/scripting.tcl
index 02459354a..29193f642 100644
--- a/tests/unit/scripting.tcl
+++ b/tests/unit/scripting.tcl
@@ -217,41 +217,45 @@ start_server {tags {"scripting"}} {
         } {*execution time*}
     }
 
-    test {EVAL - Scripts can't run blpop command} {
-        set e {}
-        catch {run_script {return redis.pcall('blpop','x',0)} 1 x} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on blpop command} {
+        r lpush l 1
+        r lpop l
+        run_script {return redis.pcall('blpop','l',0)} 1 l
+    } {}
 
-    test {EVAL - Scripts can't run brpop command} {
-        set e {}
-        catch {run_script {return redis.pcall('brpop','empty_list',0)} 1 empty_list} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on brpop command} {
+        r lpush l 1
+        r lpop l
+        run_script {return redis.pcall('brpop','l',0)} 1 l
+    } {}
 
-    test {EVAL - Scripts can't run brpoplpush command} {
-        set e {}
-        catch {run_script {return redis.pcall('brpoplpush','empty_list1{t}', 'empty_list2{t}',0)} 2 empty_list1{t} empty_list2{t}} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on brpoplpush command} {
+        r lpush empty_list1{t} 1
+        r lpop empty_list1{t}
+        run_script {return redis.pcall('brpoplpush','empty_list1{t}', 'empty_list2{t}',0)} 2 empty_list1{t} empty_list2{t}
+    } {}
 
-    test {EVAL - Scripts can't run blmove command} {
-        set e {}
-        catch {run_script {return redis.pcall('blmove','empty_list1{t}', 'empty_list2{t}', 'LEFT', 'LEFT', 0)} 2 empty_list1{t} empty_list2{t}} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on blmove command} {
+        r lpush empty_list1{t} 1
+        r lpop empty_list1{t}
+        run_script {return redis.pcall('blmove','empty_list1{t}', 'empty_list2{t}', 'LEFT', 'LEFT', 0)} 2 empty_list1{t} empty_list2{t}
+    } {}
 
-    test {EVAL - Scripts can't run bzpopmin command} {
-        set e {}
-        catch {run_script {return redis.pcall('bzpopmin','empty_zset', 0)} 1 empty_zset} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on bzpopmin command} {
+        r zadd empty_zset 10 foo
+        r zmpop 1 empty_zset MIN
+        run_script {return redis.pcall('bzpopmin','empty_zset', 0)} 1 empty_zset
+    } {}
 
-    test {EVAL - Scripts can't run bzpopmax command} {
-        set e {}
-        catch {run_script {return redis.pcall('bzpopmax','empty_zset', 0)} 1 empty_zset} e
-        set e
-    } {*not allowed*}
+    test {EVAL - Scripts do not block on bzpopmax command} {
+        r zadd empty_zset 10 foo
+        r zmpop 1 empty_zset MIN
+        run_script {return redis.pcall('bzpopmax','empty_zset', 0)} 1 empty_zset
+    } {}
+
+    test {EVAL - Scripts do not block on wait} {
+        run_script {return redis.pcall('wait','1','0')} 0
+    } {0}
 
     test {EVAL - Scripts can't run XREAD and XREADGROUP with BLOCK option} {
         r del s