CCBC-156 Custom handling for OBSERVE during packet errors

The ep-engine renders meaningful body for observe responses only if
status code is 0 (PROTOCOL_BINARY_RESPONSE_SUCCESS). We shouldn't
interpret response body in other cases, just decode & failout request
instead.

Change-Id: Ic1e0ea5d8ac5e7b44f20e0ab79c6ddc58b5f758b
Reviewed-on: http://review.couchbase.org/23570
Reviewed-by: Sergey Avseyev <sergey.avseyev@gmail.com>
Tested-by: Sergey Avseyev <sergey.avseyev@gmail.com>

RELEASE_NOTES.markdown

@@ -27,6 +27,13 @@ bugfixes. Do not forget to update this doc in every important patch.
 * [critical] CCBC-155 Observe malfunctions in the case of multiple
   keys and server failure
 
+* [major] CCBC-156 The ep-engine renders meaningful body for observe
+  responses only if status code is 0 (PROTOCOL_BINARY_RESPONSE_SUCCESS).
+  We shouldn't interpret response body in other cases, just decode &
+  failout request instead. Also we shouldn't retry observe commands on
+  PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET, because it can cause the
+  client to loop infinitely
+
 ## 2.0.2 (2013-01-04)
 
 * [major] CCBC-150 commands sent to multiple servers fail to detect

src/event.c

@@ -152,7 +152,8 @@ static int parse_single(lcb_server_t *c, hrtime_t stop)
         }
 
         if (ntohs(header.response.status) != PROTOCOL_BINARY_RESPONSE_NOT_MY_VBUCKET
-                || header.response.opcode == CMD_GET_REPLICA) {
+                || header.response.opcode == CMD_GET_REPLICA
+                || header.response.opcode == CMD_OBSERVE) {
             if (lcb_dispatch_response(c, &ct, (void*)packet) == -1) {
                 /*
                  * Internal error.. we received an unsupported response

src/handler.c

@@ -460,6 +460,46 @@ static void observe_response_handler(lcb_server_t *server,
     VBUCKET_CONFIG_HANDLE config;
     const char *end, *ptr = (const char *)&res->response.cas;
 
+    /**
+     * If we have an error we must decode the request instead
+     */
+    if (rc != LCB_SUCCESS) {
+        protocol_binary_request_header req;
+        lcb_size_t nr;
+
+        nr = ringbuffer_peek(&server->cmd_log, req.bytes, sizeof(req.bytes));
+        if (nr != sizeof(req.bytes)) {
+            lcb_error_handler(server->instance, LCB_EINTERNAL, NULL);
+            abort();
+        }
+        if (req.request.bodylen) {
+            lcb_size_t npacket = sizeof(req.bytes) + ntohl(req.request.bodylen);
+            char *packet = server->cmd_log.read_head;
+            int allocated = 0;
+
+            if (!ringbuffer_is_continous(&server->cmd_log, RINGBUFFER_READ, npacket)) {
+                packet = malloc(npacket);
+                if (packet == NULL) {
+                    lcb_error_handler(root, LCB_CLIENT_ENOMEM, NULL);
+                    abort();
+                }
+                nr = ringbuffer_peek(&server->cmd_log, packet, npacket);
+                if (nr != npacket) {
+                    lcb_error_handler(root, LCB_EINTERNAL, NULL);
+                    free(packet);
+                    abort();
+                }
+                allocated = 1;
+            }
+            lcb_failout_observe_request(server, command_data, packet, npacket, rc);
+            if (allocated) {
+                free(packet);
+            }
+        }
+        return;
+    }
+
+
     memcpy(&ttp, ptr, sizeof(ttp));
     ttp = ntohl(ttp);
     memcpy(&ttr, ptr + sizeof(ttp), sizeof(ttr));