Only mark the client reprocessing flag when unblocked on keys (#2… (#2231)

NOTE - this is a backport of
#2109

When we refactored the blocking framework we introduced the client
reprocessing infrastructure. In cases the client was blocked on keys, it
will attempt to reprocess the command. One challenge was to keep track
of the command timeout, since we are reprocessing and do not want to
re-register the client with a fresh timeout each time. The solution was
to consider the client reprocessing flag when the client is
blockedOnKeys:

```
if (!c->flag.reprocessing_command) {
        /* If the client is re-processing the command, we do not set the timeout
         * because we need to retain the client's original timeout. */ c->bstate->timeout = timeout; } 
```

However, this introduced a new issue. There are cases where the client
will consecutive blocking of different types for example:
   
```
CLIENT PAUSE 10000 ALL
BZPOPMAX zset 1
```
would have the client blocked on the zset endlessly if nothing will be
written to it.

**Credits to @uriyage for locating this with his fuzzer testing**

The suggested solution is to only flag the client when it is
specifically unblocked on keys.

Signed-off-by: Ran Shidlansik <[email protected]>

Author: ranshid

src/blocked.c

@@ -657,6 +657,7 @@ static void unblockClientOnKey(client *c, robj *key) {
      * we need to re process the command again */
     if (c->flags & CLIENT_PENDING_COMMAND) {
         c->flags &= ~CLIENT_PENDING_COMMAND;
+        c->flags |= CLIENT_REPROCESSING_COMMAND;
         /* We want the command processing and the unblock handler (see RM_Call 'K' option)
          * to run atomically, this is why we must enter the execution unit here before
          * running the command, and exit the execution unit after calling the unblock handler (if exists).
@@ -675,6 +676,8 @@ static void unblockClientOnKey(client *c, robj *key) {
         }
         exitExecutionUnit();
         afterCommand(c);
+        /* Clear the reprocessing command flag after the proc is executed. */
+        c->flags &= ~CLIENT_REPROCESSING_COMMAND;
         server.current_client = old_client;
     }
 }

src/server.c

@@ -3486,7 +3486,7 @@ void call(client *c, int flags) {
      * and a client which is reprocessing command again (after being unblocked).
      * Blocked clients can be blocked in different places and not always it means the call() function has been
      * called. For example this is required for avoiding double logging to monitors.*/
-    int reprocessing_command = flags & CMD_CALL_REPROCESSING;
+    int reprocessing_command = (c->flags & CLIENT_REPROCESSING_COMMAND) ? 1 : 0;
 
     /* Initialization: clear the flags that must be set by the command on
      * demand, and initialize the array for additional commands propagation. */
@@ -3513,20 +3513,12 @@ void call(client *c, int flags) {
      * re-processing and unblock the client.*/
     c->flags |= CLIENT_EXECUTING_COMMAND;
 
-    /* Setting the CLIENT_REPROCESSING_COMMAND flag so that during the actual
-     * processing of the command proc, the client is aware that it is being
-     * re-processed. */
-    if (reprocessing_command) c->flags |= CLIENT_REPROCESSING_COMMAND;
-
     monotime monotonic_start = 0;
     if (monotonicGetType() == MONOTONIC_CLOCK_HW)
         monotonic_start = getMonotonicUs();
 
     c->cmd->proc(c);
 
-    /* Clear the CLIENT_REPROCESSING_COMMAND flag after the proc is executed. */
-    if (reprocessing_command) c->flags &= ~CLIENT_REPROCESSING_COMMAND;
-
     exitExecutionUnit();
 
     /* In case client is blocked after trying to execute the command,
@@ -4166,7 +4158,6 @@ int processCommand(client *c) {
         addReply(c,shared.queued);
     } else {
         int flags = CMD_CALL_FULL;
-        if (client_reprocessing_command) flags |= CMD_CALL_REPROCESSING;
         call(c,flags);
         if (listLength(server.ready_keys) && !isInsideYieldingLongCommand())
             handleClientsBlockedOnKeys();

src/server.h

@@ -593,8 +593,7 @@ typedef enum {
 #define CMD_CALL_NONE 0
 #define CMD_CALL_PROPAGATE_AOF (1<<0)
 #define CMD_CALL_PROPAGATE_REPL (1<<1)
-#define CMD_CALL_REPROCESSING (1<<2)
-#define CMD_CALL_FROM_MODULE (1<<3)  /* From RM_Call */
+#define CMD_CALL_FROM_MODULE (1<<2)  /* From RM_Call */
 #define CMD_CALL_PROPAGATE (CMD_CALL_PROPAGATE_AOF|CMD_CALL_PROPAGATE_REPL)
 #define CMD_CALL_FULL (CMD_CALL_PROPAGATE)
 

tests/unit/type/list.tcl

@@ -2447,4 +2447,27 @@ foreach {pop} {BLPOP BLMPOP_RIGHT} {
         close_replication_stream $repl
     } {} {needs:repl}
 
+    test "Blocking timeout following PAUSE should honor the timeout" {
+        # cleanup first
+        r del mylist
+
+        # create a test client
+        set rd [redis_deferring_client]
+
+        # first PAUSE all writes for a very long time
+        r client pause 10000000000000 write
+
+        # block a client on the list
+        $rd BLPOP mylist 1
+        wait_for_blocked_clients_count 1
+
+        # now unpause the writes
+        r client unpause
+
+        # client should time-out
+        wait_for_blocked_clients_count 0
+
+        $rd close
+    }
+
 } ;# stop servers