Commit 3bbf9747 authored by Oran Agra's avatar Oran Agra Committed by antirez
Browse files

slave corrupts replication stream when module blocked client uses large reply (or POSTPONED_ARRAY)

when redis appends the blocked client reply list to the real client, it didn't
bother to check if it is in fact the master client. so a slave executing that
module command will send replies to the master, causing the master to send the
slave error responses, which will mess up the replication offset
(slave will advance it's replication offset, and the master does not)
parent f72f4ea3
...@@ -3706,12 +3706,7 @@ void moduleHandleBlockedClients(void) { ...@@ -3706,12 +3706,7 @@ void moduleHandleBlockedClients(void) {
* We need to glue such replies to the client output buffer and * We need to glue such replies to the client output buffer and
* free the temporary client we just used for the replies. */ * free the temporary client we just used for the replies. */
if (c) { if (c) {
if (bc->reply_client->bufpos) AddReplyFromClient(c, bc->reply_client);
addReplyString(c,bc->reply_client->buf,
bc->reply_client->bufpos);
if (listLength(bc->reply_client->reply))
listJoin(c->reply,bc->reply_client->reply);
c->reply_bytes += bc->reply_client->reply_bytes;
} }
freeClient(bc->reply_client); freeClient(bc->reply_client);
......
...@@ -628,6 +628,19 @@ void addReplySubcommandSyntaxError(client *c) { ...@@ -628,6 +628,19 @@ void addReplySubcommandSyntaxError(client *c) {
sdsfree(cmd); sdsfree(cmd);
} }
/* Append 'src' client output buffers into 'dst' client output buffers.
* This function clears the output buffers of 'src' */
void AddReplyFromClient(client *dst, client *src) {
if (prepareClientToWrite(dst) != C_OK)
return;
addReplyString(dst,src->buf, src->bufpos);
if (listLength(src->reply))
listJoin(dst->reply,src->reply);
dst->reply_bytes += src->reply_bytes;
src->reply_bytes = 0;
src->bufpos = 0;
}
/* Copy 'src' client output buffers into 'dst' client output buffers. /* Copy 'src' client output buffers into 'dst' client output buffers.
* The function takes care of freeing the old output buffers of the * The function takes care of freeing the old output buffers of the
* destination client. */ * destination client. */
......
...@@ -1435,6 +1435,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask); ...@@ -1435,6 +1435,7 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask); void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask); void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
void addReplyString(client *c, const char *s, size_t len); void addReplyString(client *c, const char *s, size_t len);
void AddReplyFromClient(client *c, client *src);
void addReplyBulk(client *c, robj *obj); void addReplyBulk(client *c, robj *obj);
void addReplyBulkCString(client *c, const char *s); void addReplyBulkCString(client *c, const char *s);
void addReplyBulkCBuffer(client *c, const void *p, size_t len); void addReplyBulkCBuffer(client *c, const void *p, size_t len);
......
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