Fix the bug that CLIENT REPLY OFF|SKIP cannot receive push notifications (#11875)

This bug seems to be there forever, CLIENT REPLY OFF|SKIP will
mark the client with CLIENT_REPLY_OFF or CLIENT_REPLY_SKIP flags.
With these flags, prepareClientToWrite called by addReply* will
return C_ERR directly. So the client can't receive the Pub/Sub
messages and any other push notifications, e.g client side tracking.

In this PR, we adding a CLIENT_PUSHING flag, disables the reply
silencing flags. When adding push replies, set the flag, after the reply,
clear the flag. Then add the flag check in prepareClientToWrite.

Fixes #11874

Note, the SUBSCRIBE command response is a bit awkward,
see https://github.com/redis/redis-doc/pull/2327

Co-authored-by: Oran Agra <oran@redislabs.com>

4 files changed, 191 insertions, 3 deletions

diff --git a/tests/unit/introspection.tcl b/tests/unit/introspection.tcl
index 097074047..10d3a15e9 100644
--- a/tests/unit/introspection.tcl
+++ b/tests/unit/introspection.tcl
@@ -94,6 +94,48 @@ start_server {tags {"introspection"}} {
         }
     } {} {needs:save}
 
+    test "CLIENT REPLY OFF/ON: disable all commands reply" {
+        set rd [redis_deferring_client]
+
+        # These replies were silenced.
+        $rd client reply off
+        $rd ping pong
+        $rd ping pong2
+
+        $rd client reply on
+        assert_equal {OK} [$rd read]
+        $rd ping pong3
+        assert_equal {pong3} [$rd read]
+
+        $rd close
+    }
+
+    test "CLIENT REPLY SKIP: skip the next command reply" {
+        set rd [redis_deferring_client]
+
+        # The first pong reply was silenced.
+        $rd client reply skip
+        $rd ping pong
+
+        $rd ping pong2
+        assert_equal {pong2} [$rd read]
+
+        $rd close
+    }
+
+    test "CLIENT REPLY ON: unset SKIP flag" {
+        set rd [redis_deferring_client]
+
+        $rd client reply skip
+        $rd client reply on
+        assert_equal {OK} [$rd read] ;# OK from CLIENT REPLY ON command
+
+        $rd ping
+        assert_equal {PONG} [$rd read]
+
+        $rd close
+    }
+
     test {MONITOR can log executed commands} {
         set rd [redis_deferring_client]
         $rd monitor
diff --git a/tests/unit/pubsub.tcl b/tests/unit/pubsub.tcl
index fe486edf3..0dd32c397 100644
--- a/tests/unit/pubsub.tcl
+++ b/tests/unit/pubsub.tcl
@@ -166,6 +166,30 @@ start_server {tags {"pubsub network"}} {
         $rd1 close
     }
 
+    test "PubSub messages with CLIENT REPLY OFF" {
+        set rd [redis_deferring_client]
+        $rd hello 3
+        $rd read ;# Discard the hello reply
+
+        # Test that the subscribe/psubscribe notification is ok
+        $rd client reply off
+        assert_equal {1} [subscribe $rd channel]
+        assert_equal {2} [psubscribe $rd ch*]
+
+        # Test that the publish notification is ok
+        $rd client reply off
+        assert_equal 2 [r publish channel hello]
+        assert_equal {message channel hello} [$rd read]
+        assert_equal {pmessage ch* channel hello} [$rd read]
+
+        # Test that the unsubscribe/punsubscribe notification is ok
+        $rd client reply off
+        assert_equal {1} [unsubscribe $rd channel]
+        assert_equal {0} [punsubscribe $rd ch*]
+
+        $rd close
+    }
+
     test "PUNSUBSCRIBE from non-subscribed channels" {
         set rd1 [redis_deferring_client]
         assert_equal {0 0 0} [punsubscribe $rd1 {foo.* bar.* quux.*}]
@@ -226,6 +250,7 @@ start_server {tags {"pubsub network"}} {
     test "Keyspace notifications: we receive keyspace notifications" {
         r config set notify-keyspace-events KA
         set rd1 [redis_deferring_client]
+        $rd1 CLIENT REPLY OFF ;# Make sure it works even if replies are silenced
         assert_equal {1} [psubscribe $rd1 *]
         r set foo bar
         assert_equal "pmessage * __keyspace@${db}__:foo set" [$rd1 read]
@@ -235,6 +260,7 @@ start_server {tags {"pubsub network"}} {
     test "Keyspace notifications: we receive keyevent notifications" {
         r config set notify-keyspace-events EA
         set rd1 [redis_deferring_client]
+        $rd1 CLIENT REPLY SKIP ;# Make sure it works even if replies are silenced
         assert_equal {1} [psubscribe $rd1 *]
         r set foo bar
         assert_equal "pmessage * __keyevent@${db}__:set foo" [$rd1 read]
@@ -244,6 +270,8 @@ start_server {tags {"pubsub network"}} {
     test "Keyspace notifications: we can receive both kind of events" {
         r config set notify-keyspace-events KEA
         set rd1 [redis_deferring_client]
+        $rd1 CLIENT REPLY ON ;# Just coverage
+        assert_equal {OK} [$rd1 read]
         assert_equal {1} [psubscribe $rd1 *]
         r set foo bar
         assert_equal "pmessage * __keyspace@${db}__:foo set" [$rd1 read]
diff --git a/tests/unit/pubsubshard.tcl b/tests/unit/pubsubshard.tcl
index 8cccdcff6..6e3fb61c1 100644
--- a/tests/unit/pubsubshard.tcl
+++ b/tests/unit/pubsubshard.tcl
@@ -40,7 +40,7 @@ start_server {tags {"pubsubshard external:skip"}} {
         $rd2 close
     }
 
-    test "PUBLISH/SUBSCRIBE after UNSUBSCRIBE without arguments" {
+    test "SPUBLISH/SSUBSCRIBE after UNSUBSCRIBE without arguments" {
         set rd1 [redis_deferring_client]
         assert_equal {1} [ssubscribe $rd1 {chan1}]
         assert_equal {2} [ssubscribe $rd1 {chan2}]
@@ -54,7 +54,7 @@ start_server {tags {"pubsubshard external:skip"}} {
         $rd1 close
     }
 
-    test "SUBSCRIBE to one channel more than once" {
+    test "SSUBSCRIBE to one channel more than once" {
         set rd1 [redis_deferring_client]
         assert_equal {1 1 1} [ssubscribe $rd1 {chan1 chan1 chan1}]
         assert_equal 1 [r SPUBLISH chan1 hello]
@@ -64,7 +64,7 @@ start_server {tags {"pubsubshard external:skip"}} {
         $rd1 close
     }
 
-    test "UNSUBSCRIBE from non-subscribed channels" {
+    test "SUNSUBSCRIBE from non-subscribed channels" {
         set rd1 [redis_deferring_client]
         assert_equal {0} [sunsubscribe $rd1 {foo}]
         assert_equal {0} [sunsubscribe $rd1 {bar}]
@@ -105,6 +105,33 @@ start_server {tags {"pubsubshard external:skip"}} {
         assert_equal "chan1 1" [r pubsub numsub chan1]
         assert_equal "chan1" [r pubsub shardchannels]
         assert_equal "chan1" [r pubsub channels]
+
+        $rd1 close
+        $rd2 close
+    }
+
+    test "PubSubShard with CLIENT REPLY OFF" {
+        set rd [redis_deferring_client]
+        $rd hello 3
+        $rd read ;# Discard the hello reply
+
+        # Test that the ssubscribe notification is ok
+        $rd client reply off
+        $rd ping
+        assert_equal {1} [ssubscribe $rd channel]
+
+        # Test that the spublish notification is ok
+        $rd client reply off
+        $rd ping
+        assert_equal 1 [r spublish channel hello]
+        assert_equal {smessage channel hello} [$rd read]
+
+        # Test that sunsubscribe notification is ok
+        $rd client reply off
+        $rd ping
+        assert_equal {0} [sunsubscribe $rd channel]
+
+        $rd close
     }
 }
 
diff --git a/tests/unit/tracking.tcl b/tests/unit/tracking.tcl
index 21036352f..bea8508b1 100644
--- a/tests/unit/tracking.tcl
+++ b/tests/unit/tracking.tcl
@@ -781,7 +781,98 @@ start_server {tags {"tracking network logreqres:skip"}} {
         r debug pause-cron 0
     } {OK} {needs:debug}
 
+    foreach resp {3 2} {
+        test "RESP$resp based basic invalidation with client reply off" {
+            # This entire test is mostly irrelevant for RESP2, but we run it anyway just for some extra coverage.
+            clean_all
+
+            $rd hello $resp
+            $rd read
+            $rd client tracking on
+            $rd read
+
+            $rd_sg set foo bar
+            $rd get foo
+            $rd read
+
+            $rd client reply off
+
+            $rd_sg set foo bar2
+
+            if {$resp == 3} {
+                assert_equal {invalidate foo} [$rd read]
+            } elseif {$resp == 2} { } ;# Just coverage
+
+            # Verify things didn't get messed up and no unexpected reply was pushed to the client.
+            $rd client reply on
+            assert_equal {OK} [$rd read]
+            $rd ping
+            assert_equal {PONG} [$rd read]
+        }
+    }
+
+    test {RESP3 based basic redirect invalidation with client reply off} {
+        clean_all
+
+        set rd_redir [redis_deferring_client]
+        $rd_redir hello 3
+        $rd_redir read
+
+        $rd_redir client id
+        set rd_redir_id [$rd_redir read]
+
+        $rd client tracking on redirect $rd_redir_id
+        $rd read
+
+        $rd_sg set foo bar
+        $rd get foo
+        $rd read
+
+        $rd_redir client reply off
+
+        $rd_sg set foo bar2
+        assert_equal {invalidate foo} [$rd_redir read]
+
+        # Verify things didn't get messed up and no unexpected reply was pushed to the client.
+        $rd_redir client reply on
+        assert_equal {OK} [$rd_redir read]
+        $rd_redir ping
+        assert_equal {PONG} [$rd_redir read]
+
+        $rd_redir close
+    }
+
+    test {RESP3 based basic tracking-redir-broken with client reply off} {
+        clean_all
+
+        $rd hello 3
+        $rd read
+        $rd client tracking on redirect $redir_id
+        $rd read
+
+        $rd_sg set foo bar
+        $rd get foo
+        $rd read
+
+        $rd client reply off
+
+        $rd_redirection quit
+        $rd_redirection read
+
+        $rd_sg set foo bar2
+
+        set res [lsearch -exact [$rd read] "tracking-redir-broken"]
+        assert_morethan_equal $res 0
+
+        # Verify things didn't get messed up and no unexpected reply was pushed to the client.
+        $rd client reply on
+        assert_equal {OK} [$rd read]
+        $rd ping
+        assert_equal {PONG} [$rd read]
+    }
+
     $rd_redirection close
+    $rd_sg close
     $rd close
 }