Skip to content

Commit

Permalink
Fix the bug that CLIENT REPLY OFF|SKIP cannot receive push notificati…
Browse files Browse the repository at this point in the history 
…ons (#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 redis/redis-doc#2327

Co-authored-by: Oran Agra <oran@redislabs.com>
(cherry picked from commit 416842e)
  • Loading branch information
@enjoy-binbin @oranagra
enjoy-binbin authored and oranagra committed Mar 20, 2023
1 parent 1718151 commit f8ae7a4
Show file tree
Hide file tree
Showing 9 changed files with 227 additions and 5 deletions.
3 changes: 3 additions & 0 deletions src/debug.c
View file
Expand Up @@ -804,9 +804,12 @@ NULL
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. */
Expand Down
7 changes: 5 additions & 2 deletions src/networking.c
View file
Expand Up @@ -290,8 +290,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. */
Expand Down Expand Up @@ -973,6 +975,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,'>');
}

Expand Down
18 changes: 18 additions & 0 deletions src/pubsub.c
View file
Expand Up @@ -105,19 +105,24 @@ pubsubtype pubSubShardType = {
* to send a special message (for instance an Array type) by using the
* addReply*() API family. */
void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk) {
uint64_t old_flags = c->flags;
c->flags |= CLIENT_PUSHING;
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,message_bulk);
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
Expand All @@ -126,24 +131,30 @@ 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, pubsubtype type) {
uint64_t old_flags = c->flags;
c->flags |= CLIENT_PUSHING;
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,*type.subscribeMsg);
addReplyBulk(c,channel);
addReplyLongLong(c,type.subscriptionCount(c));
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
}

/* Send the pubsub unsubscription notification to the client.
* Channel can be NULL: this is useful when the client sends a mass
* unsubscribe command but there are no channels to unsubscribe from: we
* still send a notification. */
void addReplyPubsubUnsubscribed(client *c, robj *channel, pubsubtype type) {
uint64_t old_flags = c->flags;
c->flags |= CLIENT_PUSHING;
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
Expand All @@ -154,24 +165,30 @@ void addReplyPubsubUnsubscribed(client *c, robj *channel, pubsubtype type) {
else
addReplyNull(c);
addReplyLongLong(c,type.subscriptionCount(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
addReplyPushLen(c,3);
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.
* Pattern can be NULL: this is useful when the client sends a mass
* 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
Expand All @@ -182,6 +199,7 @@ void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) {
else
addReplyNull(c);
addReplyLongLong(c,clientSubscriptionsCount(c));
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
}

/*-----------------------------------------------------------------------------
Expand Down
1 change: 1 addition & 0 deletions src/server.h
View file
Expand Up @@ -364,6 +364,7 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
RDB without replication buffer. */
#define CLIENT_NO_EVICT (1ULL<<43) /* This client is protected against client
memory eviction. */
#define CLIENT_PUSHING (1ULL<<44) /* This client is pushing notifications. */

/* Client block type (btype field in client structure)
* if CLIENT_BLOCKED flag is set. */
Expand Down
9 changes: 9 additions & 0 deletions src/tracking.c
View file
Expand Up @@ -266,6 +266,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);
Expand All @@ -279,10 +282,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
Expand All @@ -301,6 +308,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;
}

Expand All @@ -312,6 +320,7 @@ void sendTrackingMessage(client *c, char *keyname, size_t keylen, int proto) {
addReplyBulkCBuffer(c,keyname,keylen);
}
updateClientMemUsageAndBucket(c);
if (!(old_flags & CLIENT_PUSHING)) c->flags &= ~CLIENT_PUSHING;
}

/* This function is called when a key is modified in Redis and in the case
Expand Down
42 changes: 42 additions & 0 deletions tests/unit/introspection.tcl
View file
Expand Up @@ -56,6 +56,48 @@ start_server {tags {"introspection"}} {
assert_equal "PONG" [r ping]
}

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
Expand Down
28 changes: 28 additions & 0 deletions tests/unit/pubsub.tcl
View file
Expand Up @@ -146,6 +146,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.*}]
Expand Down Expand Up @@ -206,6 +230,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]
Expand All @@ -215,6 +240,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]
Expand All @@ -224,6 +250,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]
Expand Down
33 changes: 30 additions & 3 deletions tests/unit/pubsubshard.tcl
View file
Expand Up @@ -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}]
Expand All @@ -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]
Expand All @@ -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}]
Expand Down Expand Up @@ -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
}
}

Expand Down

0 comments on commit f8ae7a4

Please sign in to comment.