Merge pull request #5962 from oranagra/module_blocked_reply

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

3 files changed, 15 insertions, 6 deletions

diff --git a/src/module.c b/src/module.c
index ff7f27cdd..0c8197ac7 100644
--- a/src/module.c
+++ b/src/module.c
@@ -3747,12 +3747,7 @@ void moduleHandleBlockedClients(void) {
          * We need to glue such replies to the client output buffer and
          * free the temporary client we just used for the replies. */
         if (c) {
-            if (bc->reply_client->bufpos)
-                addReplyProto(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;
+            AddReplyFromClient(c, bc->reply_client);
         }
         freeClient(bc->reply_client);
 
diff --git a/src/networking.c b/src/networking.c
index 09cbff387..7fdd1984d 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -744,6 +744,19 @@ void addReplySubcommandSyntaxError(client *c) {
     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;
+    addReplyProto(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.
  * The function takes care of freeing the old output buffers of the
  * destination client. */
diff --git a/src/server.h b/src/server.h
index 95e0355a6..dfd9f7698 100644
--- a/src/server.h
+++ b/src/server.h
@@ -1529,6 +1529,7 @@ void addReplyNullArray(client *c);
 void addReplyBool(client *c, int b);
 void addReplyVerbatim(client *c, const char *s, size_t len, const char *ext);
 void addReplyProto(client *c, const char *s, size_t len);
+void AddReplyFromClient(client *c, client *src);
 void addReplyBulk(client *c, robj *obj);
 void addReplyBulkCString(client *c, const char *s);
 void addReplyBulkCBuffer(client *c, const void *p, size_t len);