From 7afd1724ddd2e3754cdceb148c419a81ad00fa9f Mon Sep 17 00:00:00 2001 From: Binbin Date: Sun, 12 Mar 2023 23:50:44 +0800 Subject: [PATCH] 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 (cherry picked from commit 416842e6c004dbd951e398a8651df6c56a030a23) (cherry picked from commit f8ae7a414c96b8a0dfff473ddb5f5447599cf5f4) --- src/debug.c | 7 +++ src/networking.c | 7 ++- src/pubsub.c | 18 +++++++ src/server.h | 1 + src/tracking.c | 9 ++++ tests/unit/introspection.tcl | 42 +++++++++++++++++ tests/unit/pubsub.tcl | 28 +++++++++++ tests/unit/tracking.tcl | 91 ++++++++++++++++++++++++++++++++++++ 8 files changed, 201 insertions(+), 2 deletions(-) diff --git a/src/debug.c b/src/debug.c index 1b2bbf2c2..ea6a3f397 100644 --- a/src/debug.c +++ b/src/debug.c @@ -750,9 +750,16 @@ NULL * also have a normal reply type after the attribute. */ addReplyBulkCString(c,"Some real reply following the attribute"); } else if (!strcasecmp(name,"push")) { + if (c->resp < 3) { + addReplyError(c,"RESP2 is not supported by this command"); + return; + } + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; addReplyPushLen(c,2); addReplyBulkCString(c,"server-cpu-usage"); addReplyLongLong(c,42); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; /* Push replies are not synchronous replies, so we emit also a * normal reply in order for blocking clients just discarding the * push reply, to actually consume the reply and continue. */ diff --git a/src/networking.c b/src/networking.c index fe454efe7..9f654063c 100644 --- a/src/networking.c +++ b/src/networking.c @@ -260,8 +260,10 @@ int prepareClientToWrite(client *c) { /* If CLIENT_CLOSE_ASAP flag is set, we need not write anything. */ if (c->flags & CLIENT_CLOSE_ASAP) return C_ERR; - /* CLIENT REPLY OFF / SKIP handling: don't send replies. */ - if (c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) return C_ERR; + /* CLIENT REPLY OFF / SKIP handling: don't send replies. + * CLIENT_PUSHING handling: disables the reply silencing flags. */ + if ((c->flags & (CLIENT_REPLY_OFF|CLIENT_REPLY_SKIP)) && + !(c->flags & CLIENT_PUSHING)) return C_ERR; /* Masters don't receive replies, unless CLIENT_MASTER_FORCE_REPLY flag * is set. */ @@ -819,6 +821,7 @@ void addReplyAttributeLen(client *c, long length) { void addReplyPushLen(client *c, long length) { serverAssert(c->resp >= 3); + serverAssertWithInfo(c, NULL, c->flags & CLIENT_PUSHING); addReplyAggregateLen(c,length,'>'); } diff --git a/src/pubsub.c b/src/pubsub.c index 3409deac2..329a57cae 100644 --- a/src/pubsub.c +++ b/src/pubsub.c @@ -41,6 +41,8 @@ int clientSubscriptionsCount(client *c); * to send a special message (for instance an Array type) by using the * addReply*() API family. */ void addReplyPubsubMessage(client *c, robj *channel, robj *msg) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[3]); else @@ -48,12 +50,15 @@ void addReplyPubsubMessage(client *c, robj *channel, robj *msg) { addReply(c,shared.messagebulk); addReplyBulk(c,channel); if (msg) addReplyBulk(c,msg); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* Send a pubsub message of type "pmessage" to the client. The difference * with the "message" type delivered by addReplyPubsubMessage() is that * this message format also includes the pattern that matched the message. */ void addReplyPubsubPatMessage(client *c, robj *pat, robj *channel, robj *msg) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[4]); else @@ -62,10 +67,13 @@ void addReplyPubsubPatMessage(client *c, robj *pat, robj *channel, robj *msg) { addReplyBulk(c,pat); addReplyBulk(c,channel); addReplyBulk(c,msg); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* Send the pubsub subscription notification to the client. */ void addReplyPubsubSubscribed(client *c, robj *channel) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[3]); else @@ -73,6 +81,7 @@ void addReplyPubsubSubscribed(client *c, robj *channel) { addReply(c,shared.subscribebulk); addReplyBulk(c,channel); addReplyLongLong(c,clientSubscriptionsCount(c)); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* Send the pubsub unsubscription notification to the client. @@ -80,6 +89,8 @@ void addReplyPubsubSubscribed(client *c, robj *channel) { * unsubscribe command but there are no channels to unsubscribe from: we * still send a notification. */ void addReplyPubsubUnsubscribed(client *c, robj *channel) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[3]); else @@ -90,10 +101,13 @@ void addReplyPubsubUnsubscribed(client *c, robj *channel) { else addReplyNull(c); addReplyLongLong(c,clientSubscriptionsCount(c)); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* Send the pubsub pattern subscription notification to the client. */ void addReplyPubsubPatSubscribed(client *c, robj *pattern) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[3]); else @@ -101,6 +115,7 @@ void addReplyPubsubPatSubscribed(client *c, robj *pattern) { addReply(c,shared.psubscribebulk); addReplyBulk(c,pattern); addReplyLongLong(c,clientSubscriptionsCount(c)); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* Send the pubsub pattern unsubscription notification to the client. @@ -108,6 +123,8 @@ void addReplyPubsubPatSubscribed(client *c, robj *pattern) { * punsubscribe command but there are no pattern to unsubscribe from: we * still send a notification. */ void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; if (c->resp == 2) addReply(c,shared.mbulkhdr[3]); else @@ -118,6 +135,7 @@ void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) { else addReplyNull(c); addReplyLongLong(c,clientSubscriptionsCount(c)); + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /*----------------------------------------------------------------------------- diff --git a/src/server.h b/src/server.h index be0c42cbe..7afd6843e 100644 --- a/src/server.h +++ b/src/server.h @@ -286,6 +286,7 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT]; and AOF client */ #define CLIENT_REPL_RDBONLY (1ULL<<42) /* This client is a replica that only wants RDB without replication buffer. */ +#define CLIENT_PUSHING (1ULL<<43) /* This client is pushing notifications. */ /* Client block type (btype field in client structure) * if CLIENT_BLOCKED flag is set. */ diff --git a/src/tracking.c b/src/tracking.c index 5bac6926a..d77886357 100644 --- a/src/tracking.c +++ b/src/tracking.c @@ -260,6 +260,9 @@ void trackingRememberKeys(client *c) { * - Following a flush command, to send a single RESP NULL to indicate * that all keys are now invalid. */ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { + uint64_t old_flags = c->flags; + c->flags |= CLIENT_PUSHING; + int using_redirection = 0; if (c->client_tracking_redirection) { client *redir = lookupClientByID(c->client_tracking_redirection); @@ -273,10 +276,14 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { addReplyBulkCBuffer(c,"tracking-redir-broken",21); addReplyLongLong(c,c->client_tracking_redirection); } + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; return; } + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; c = redir; using_redirection = 1; + old_flags = c->flags; + c->flags |= CLIENT_PUSHING; } /* Only send such info for clients in RESP version 3 or more. However @@ -295,6 +302,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { * redirecting to another client. We can't send anything to * it since RESP2 does not support push messages in the same * connection. */ + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; return; } @@ -305,6 +313,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) { addReplyArrayLen(c,1); addReplyBulkCBuffer(c,keyname,keylen); } + if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING; } /* This function is called when a key is modified in Redis and in the case diff --git a/tests/unit/introspection.tcl b/tests/unit/introspection.tcl index 0c78edf20..b1319211e 100644 --- a/tests/unit/introspection.tcl +++ b/tests/unit/introspection.tcl @@ -13,6 +13,48 @@ start_server {tags {"introspection"}} { r client info } {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* argv-mem=* obl=0 oll=0 omem=0 tot-mem=* events=r cmd=client*} + 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 03c5dfccb..45a91576c 100644 --- a/tests/unit/pubsub.tcl +++ b/tests/unit/pubsub.tcl @@ -140,6 +140,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.*}] @@ -198,6 +222,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@9__:foo set} [$rd1 read] @@ -207,6 +232,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@9__:set foo} [$rd1 read] @@ -216,6 +242,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@9__:foo set} [$rd1 read] diff --git a/tests/unit/tracking.tcl b/tests/unit/tracking.tcl index 35069ba60..23290c9e5 100644 --- a/tests/unit/tracking.tcl +++ b/tests/unit/tracking.tcl @@ -746,6 +746,97 @@ start_server {tags {"tracking network"}} { 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 }