mirrors
/
redis
mirror of https://mirror.osredm.com/root/redis.git
2
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

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>
(cherry picked from commit 416842e6c0)
(cherry picked from commit f8ae7a414c)
(cherry picked from commit 96814a32da61e5ed523864e00609a4aa6be065b3)
This commit is contained in:
Binbin 2023-03-12 23:50:44 +08:00 committed by Oran Agra
parent c924ac3fdf
commit 61cf011388
9 changed files with 272 additions and 22 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/debug.c
View File

@ -681,9 +681,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. */

7
src/networking.c
View File

@ -248,8 +248,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. */
@ -722,6 +724,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,'>');
}

18
src/pubsub.c
View File

@ -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;
}
/*-----------------------------------------------------------------------------

1
src/server.h
View File

@ -272,6 +272,7 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
#define CLIENT_PROTOCOL_ERROR (1ULL<<39) /* Protocol error chatting with it. */
#define CLIENT_CLOSE_AFTER_COMMAND (1ULL<<40) /* Close after executing commands
* and writing entire reply. */
#define CLIENT_PUSHING (1ULL<<41) /* This client is pushing notifications. */
/* Client block type (btype field in client structure)
* if CLIENT_BLOCKED flag is set. */

9
src/tracking.c
View File

@ -209,6 +209,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);
@ -222,10 +225,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
@ -244,6 +251,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;
}
@ -254,6 +262,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

48
tests/support/test.tcl
View File

@ -31,6 +31,15 @@ proc assert_match {pattern value} {
}
}
proc assert_failed {expected_err detail} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -2])"
}
error "assertion:$expected_err $detail"
}
proc assert_not_equal {value expected {detail ""}} {
if {!($expected ne $value)} {
assert_failed "Expected '$value' not equal to '$expected'" $detail
@ -39,34 +48,37 @@ proc assert_not_equal {value expected {detail ""}} {
proc assert_equal {value expected {detail ""}} {
if {$expected ne $value} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
}
error "assertion:Expected '$value' to be equal to '$expected' $detail"
assert_failed "Expected '$value' to be equal to '$expected'" $detail
}
}
proc assert_lessthan {value expected {detail ""}} {
if {!($value < $expected)} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
}
error "assertion:Expected '$value' to be lessthan to '$expected' $detail"
assert_failed "Expected '$value' to be less than '$expected'" $detail
}
}
proc assert_lessthan_equal {value expected {detail ""}} {
if {!($value <= $expected)} {
assert_failed "Expected '$value' to be less than or equal to '$expected'" $detail
}
}
proc assert_morethan {value expected {detail ""}} {
if {!($value > $expected)} {
assert_failed "Expected '$value' to be more than '$expected'" $detail
}
}
proc assert_morethan_equal {value expected {detail ""}} {
if {!($value >= $expected)} {
assert_failed "Expected '$value' to be more than or equal to '$expected'" $detail
}
}
proc assert_range {value min max {detail ""}} {
if {!($value <= $max && $value >= $min)} {
if {$detail ne ""} {
set detail "(detail: $detail)"
} else {
set detail "(context: [info frame -1])"
}
error "assertion:Expected '$value' to be between to '$min' and '$max' $detail"
assert_failed "Expected '$value' to be between to '$min' and '$max'" $detail
}
}

42
tests/unit/introspection.tcl
View File

@ -3,6 +3,48 @@ start_server {tags {"introspection"}} {
r client list
} {*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

28
tests/unit/pubsub.tcl
View File

@ -188,6 +188,30 @@ start_server {tags {"pubsub"}} {
$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.*}]
@ -228,6 +252,7 @@ start_server {tags {"pubsub"}} {
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]
@ -237,6 +262,7 @@ start_server {tags {"pubsub"}} {
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]
@ -246,6 +272,8 @@ start_server {tags {"pubsub"}} {
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]

134
tests/unit/tracking.tcl
View File

@ -145,6 +145,136 @@ start_server {tags {"tracking"}} {
# the two keys (only one must remain)
assert {$keys eq {key1} || $keys eq {key2}}
}
$rd1 close
}
start_server {tags {"tracking network"}} {
# Create a deferred client we'll use to redirect invalidation
# messages to.
set rd_redirection [redis_deferring_client]
$rd_redirection client id
set redir_id [$rd_redirection read]
$rd_redirection subscribe __redis__:invalidate
$rd_redirection read ; # Consume the SUBSCRIBE reply.
# Create another client that's not used as a redirection client
# We should always keep this client's buffer clean
set rd [redis_deferring_client]
# Client to be used for SET and GET commands
# We don't read this client's buffer
set rd_sg [redis_client]
proc clean_all {} {
uplevel {
# We should make r TRACKING off first. If r is in RESP3,
# r FLUSH ALL will send us tracking-redir-broken or other
# info which will not be consumed.
r CLIENT TRACKING off
$rd QUIT
$rd_redirection QUIT
set rd [redis_deferring_client]
set rd_redirection [redis_deferring_client]
$rd_redirection client id
set redir_id [$rd_redirection read]
$rd_redirection subscribe __redis__:invalidate
$rd_redirection read ; # Consume the SUBSCRIBE reply.
r FLUSHALL
r HELLO 2
r config set tracking-table-max-keys 1000000
}
}
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
}