From acba2fc9b4c8082e5344d2d53e51dc4c1c37942c Mon Sep 17 00:00:00 2001
From: Oran Agra <oran@redislabs.com>
Date: Sun, 24 Mar 2019 13:10:55 +0200
Subject: [PATCH] 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)
---
 src/module.c     |  7 +------
 src/networking.c | 13 +++++++++++++
 src/server.h     |  1 +
 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);