Commit ab05b28b authored by Ran Shidlansik's avatar Ran Shidlansik Committed by Oran Agra
Browse files

fix cluster propagation in case of disconnected cluster node, see #11752

The mentioned PR which was fixed before 7.2 needed these adjustments in
order to fix the problem in redis 7.0.
parent ca0b6cae
......@@ -3093,13 +3093,18 @@ void clusterBroadcastPong(int target) {
* the 'bulk_data', sanitizer generates an out-of-bounds error which is a false
* positive in this context. */
REDIS_NO_SANITIZE("bounds")
void clusterSendPublish(clusterLink *link, robj *channel, robj *message, uint16_t type) {
void clusterSendPublish(clusterLink *link, robj *channel, robj *message, uint16_t type, int bcast) {
unsigned char *payload;
clusterMsg buf[1];
clusterMsg *hdr = (clusterMsg*) buf;
uint32_t totlen;
uint32_t channel_len, message_len;
/* In case we are not going to broadcast we have no point trying to publish on a missing
* clusterbus link. */
if (!bcast && !link)
return;
channel = getDecodedObject(channel);
message = getDecodedObject(message);
channel_len = sdslen(channel->ptr);
......@@ -3125,7 +3130,7 @@ void clusterSendPublish(clusterLink *link, robj *channel, robj *message, uint16_
memcpy(hdr->data.publish.msg.bulk_data+sdslen(channel->ptr),
message->ptr,sdslen(message->ptr));
if (link)
if (!bcast)
clusterSendMessage(link,payload,totlen);
else
clusterBroadcastMessage(payload,totlen);
......@@ -3232,7 +3237,7 @@ int clusterSendModuleMessageToTarget(const char *target, uint64_t module_id, uin
* -------------------------------------------------------------------------- */
void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
if (!sharded) {
clusterSendPublish(NULL, channel, message, CLUSTERMSG_TYPE_PUBLISH);
clusterSendPublish(NULL, channel, message, CLUSTERMSG_TYPE_PUBLISH, 1);
return;
}
......@@ -3245,7 +3250,7 @@ void clusterPropagatePublish(robj *channel, robj *message, int sharded) {
clusterNode *node = listNodeValue(ln);
if (node->flags & (CLUSTER_NODE_MYSELF|CLUSTER_NODE_HANDSHAKE))
continue;
clusterSendPublish(node->link, channel, message, CLUSTERMSG_TYPE_PUBLISHSHARD);
clusterSendPublish(node->link, channel, message, CLUSTERMSG_TYPE_PUBLISHSHARD, 0);
}
}
listRelease(nodes_for_slot);
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment