Merge pull request #4 from Netflix/fix/large-payload-crash

Fix/large payload crash

.travis.yml

@@ -1,8 +1,14 @@
 language: c
 script: bash ./travis.sh
 
+before_install:
+  - sudo apt remove --purge python3-pip
+  - curl -O https://bootstrap.pypa.io/pip/3.5/get-pip.py
+  - sudo -E python3 get-pip.py
+  - sudo -E python3 -m pip install --upgrade "pip < 22.3"
+
 addons:
   apt:
+   update: true
    packages:
-   - python3
-   - python3-pip
+   - python3.9

src/dyn_client.c

@@ -1109,6 +1109,12 @@ void req_recv_done(struct context *ctx, struct conn *conn, struct msg *req,
   if (status != DN_OK) goto error;
 
   status = fragment_query_if_necessary(req, conn, &frag_msgq);
+  if (status == DYNOMITE_PAYLOAD_TOO_LARGE) {
+    dictAdd(conn->outstanding_msgs_dict, &req->id, req);
+    req->nfrag = 0; // Since we failed to fragment the payload, we don't have any fragments for this request
+    goto error;
+  }
+
   if (status != DN_OK) goto error;
 
   status = rewrite_query_with_timestamp_md(&req, ctx);

src/dyn_message.h

@@ -301,6 +301,7 @@ typedef enum dyn_error {
   BAD_FORMAT,
   DYNOMITE_NO_QUORUM_ACHIEVED,
   DYNOMITE_SCRIPT_SPANS_NODES,
+  DYNOMITE_PAYLOAD_TOO_LARGE,
 } dyn_error_t;
 
 static inline char *dn_strerror(dyn_error_t err) {
@@ -325,6 +326,8 @@ static inline char *dn_strerror(dyn_error_t err) {
       return "Failed to achieve Quorum";
     case DYNOMITE_SCRIPT_SPANS_NODES:
       return "Keys in the script cannot span multiple nodes";
+    case DYNOMITE_PAYLOAD_TOO_LARGE:
+      return "MSET/MGET/SCAN payload too large";
     default:
       return strerror(err);
   }
@@ -336,6 +339,7 @@ static inline char *dyn_error_source(dyn_error_t err) {
     case DYNOMITE_INVALID_STATE:
     case DYNOMITE_NO_QUORUM_ACHIEVED:
     case DYNOMITE_SCRIPT_SPANS_NODES:
+    case DYNOMITE_PAYLOAD_TOO_LARGE:
       return "Dynomite:";
     case PEER_CONNECTION_REFUSE:
     case PEER_HOST_DOWN:

src/proto/dyn_redis.c

@@ -3037,7 +3037,9 @@ static rstatus_t redis_copy_bulk(struct msg *dst, struct msg *src, bool log) {
   }
 
   p = mbuf->pos;
-  ASSERT(*p == '$');
+  if (*p != '$') {
+    return DYNOMITE_PAYLOAD_TOO_LARGE;
+  }
   p++;
 
   if (p[0] == '-' && p[1] == '1') {
@@ -3051,7 +3053,7 @@ static rstatus_t redis_copy_bulk(struct msg *dst, struct msg *src, bool log) {
     }
     len += CRLF_LEN * 2;
     len += (p - mbuf->pos);
-  }
+  } 
   bytes = len;
 
   /* copy len bytes to dst */
@@ -3385,7 +3387,7 @@ static rstatus_t redis_append_key(struct msg *r, struct keypos *kpos_src) {
  *   +--------------------+     +---------------------+ +----+----------------+
  *   |   frag_id = 10     |     |   frag_id = 10      |     |   frag_id = 10 |
  *   |     nfrag = 3      |     |      nfrag = 0      |     |      nfrag = 0 |
- *   | frag_seq = x x x   |     |     key1, key3      |     |         key2 |
+ *   | frag_seq = x x x   |     |     key1, key3      |     |         key2   |
  *   +------------|-|-|---+     +---------------------+ +---------------------+
  *                | | |          ^    ^                          ^
  *                | \ \          |    |                          |
@@ -3513,21 +3515,21 @@ static rstatus_t redis_fragment_argx(struct msg *r, struct server_pool *pool,
       TAILQ_INSERT_TAIL(frag_msgq, sub_msg, m_tqe);
     }
 
-    status = redis_append_key(sub_msg, kpos);
+    status = redis_append_key(sub_msg, kpos); // Adds key to the sub_msg
     if (status != DN_OK) {
       dn_free(sub_msgs);
       return status;
     }
     if (key_step == 1) { // mget,del
       continue;
     } else {                                    // mset
-      status = redis_copy_bulk(NULL, r, false); // eat key
+      status = redis_copy_bulk(NULL, r, false); // Consumes key portion of the payload
       if (status != DN_OK) {
         dn_free(sub_msgs);
         return status;
       }
 
-      status = redis_copy_bulk(sub_msg, r, false);
+      status = redis_copy_bulk(sub_msg, r, false); // Consumes and copies value to the sub_msg fragment
       if (status != DN_OK) {
         dn_free(sub_msgs);
         return status;