Merge pull request #732 from Netflix/dev

Merge dev into v0.6

Author: smukil

src/dyn_client.c

@@ -1230,6 +1230,7 @@ static struct msg *all_rspmgrs_get_response(struct context *ctx, struct msg *req
       continue;
     } else {
       ASSERT(rsp->is_error == false);
+
       // If the DCs we've processed so far have not seen errors, we need to
       // make sure that the remaining DCs don't have errors too.
       dc_rsp = rspmgr_get_response(ctx, rspmgr);
@@ -1259,11 +1260,14 @@ static struct msg *all_rspmgrs_get_response(struct context *ctx, struct msg *req
 static rstatus_t msg_each_quorum_rsp_handler(struct context *ctx, struct msg *req,
     struct msg *rsp) {
 
-  if (all_rspmgrs_done(ctx, req->additional_each_rspmgrs)) return swallow_extra_rsp(req, rsp);
+  if (all_rspmgrs_done(ctx, req->additional_each_rspmgrs)) {
+    return swallow_extra_rsp(req, rsp);
+  }
 
   int rspmgr_idx = -1;
   struct conn *rsp_conn = rsp->owner;
   if (rsp_conn == NULL) {
+    // TODO: We should remove this case. Test and confirm.
     rspmgr_idx = 0;
   } else if (rsp_conn->type == CONN_DNODE_PEER_SERVER) {
     struct node *peer_instance = (struct node*) rsp_conn->owner;

src/dyn_dnode_peer.c

@@ -316,6 +316,7 @@ static void dnode_peer_ack_err(struct context *ctx, struct conn *conn,
   rsp->dyn_error_code = req->dyn_error_code = PEER_CONNECTION_REFUSE;
   rsp->dmsg = dmsg_get();
   rsp->dmsg->id = req->id;
+  rsp->owner = conn;
 
   log_info("%s Closing req %u:%u len %" PRIu32 " type %d %c %s",
            print_obj(conn), req->id, req->parent_id, req->mlen, req->type,

src/dyn_message.c

@@ -833,6 +833,11 @@ uint32_t msg_payload_crc32(struct msg *rsp) {
      the beginning of the first mbuf */
   bool start_found = rsp->dmsg ? false : true;
 
+  // If the message is from another DC, the mbufs will have the decrypted
+  // payload without the Dynomite header, so we do have the start.
+  // rsp->dmsg->payload for cross DC msgs will have the encrypted payload.
+  if (rsp->dmsg && !rsp->owner->same_dc) start_found = true;
+
   STAILQ_FOREACH(mbuf, &rsp->mhdr, next) {
     uint8_t *start = mbuf->start;
     uint8_t *end = mbuf->last;

src/dyn_server.c

@@ -197,6 +197,7 @@ static void server_ack_err(struct context *ctx, struct conn *conn,
   rsp->error_code = req->error_code = conn->err;
   rsp->dyn_error_code = req->dyn_error_code = STORAGE_CONNECTION_REFUSE;
   rsp->dmsg = NULL;
+  rsp->owner = conn;
   log_debug(LOG_DEBUG, "%s <-> %s", print_obj(req), print_obj(rsp));
 
   log_info("close %s req %s len %" PRIu32 " from %s %c %s", print_obj(conn),