Processing inconsistent proxied response length.

Keepalive connection is disabled if upstream response length
differs from specified in the "Content-Length" field value.

src/nxt_buf.c

@@ -183,7 +183,10 @@ nxt_buf_chain_length(nxt_buf_t *b)
     length = 0;
 
     while (b != NULL) {
-        length += b->mem.free - b->mem.pos;
+        if (!nxt_buf_is_sync(b)) {
+            length += b->mem.free - b->mem.pos;
+        }
+
         b = b->next;
     }
 

src/nxt_h1proto.c

@@ -1549,6 +1549,7 @@ nxt_h1p_request_close(nxt_task_t *task, nxt_http_proto_t proto,
     nxt_debug(task, "h1p request close");
 
     h1p = proto.h1;
+    h1p->keepalive &= !h1p->request->inconsistent;
     h1p->request = NULL;
 
     nxt_router_conf_release(task, joint);

src/nxt_http.h

@@ -112,6 +112,7 @@ typedef struct {
     nxt_sockaddr_t                  *sockaddr;
     nxt_list_t                      *fields;
     nxt_buf_t                       *body;
+    nxt_off_t                       remainder;
 
     nxt_http_status_t               status:16;
     nxt_http_protocol_t             protocol:8;       /* 2 bits */
@@ -170,6 +171,7 @@ struct nxt_http_request_s {
     nxt_http_protocol_t             protocol:8;   /* 2 bits */
     uint8_t                         logged;       /* 1 bit  */
     uint8_t                         header_sent;  /* 1 bit  */
+    uint8_t                         inconsistent; /* 1 bit  */
     uint8_t                         error;        /* 1 bit  */
     uint8_t                         websocket_handshake;  /* 1 bit */
 };

src/nxt_http_proxy.c

@@ -29,6 +29,8 @@ static void nxt_http_proxy_header_send(nxt_task_t *task, void *obj, void *data);
 static void nxt_http_proxy_header_sent(nxt_task_t *task, void *obj, void *data);
 static void nxt_http_proxy_header_read(nxt_task_t *task, void *obj, void *data);
 static void nxt_http_proxy_send_body(nxt_task_t *task, void *obj, void *data);
+static void nxt_http_proxy_request_send(nxt_task_t *task,
+    nxt_http_request_t *r, nxt_buf_t *out);
 static void nxt_http_proxy_read(nxt_task_t *task, void *obj, void *data);
 static void nxt_http_proxy_buf_mem_completion(nxt_task_t *task, void *obj,
     void *data);
@@ -185,6 +187,10 @@ nxt_http_proxy_header_read(nxt_task_t *task, void *obj, void *data)
 
     nxt_debug(task, "http proxy status: %d", peer->status);
 
+    if (r->resp.content_length_n > 0) {
+        peer->remainder = r->resp.content_length_n;
+    }
+
     nxt_list_each(field, peer->fields) {
 
         nxt_debug(task, "http proxy header: \"%*s: %*s\"",
@@ -220,7 +226,7 @@ nxt_http_proxy_send_body(nxt_task_t *task, void *obj, void *data)
 
     if (out != NULL) {
         peer->body = NULL;
-        nxt_http_request_send(task, r, out);
+        nxt_http_proxy_request_send(task, r, out);
     }
 
     r->state = &nxt_http_proxy_read_state;
@@ -229,6 +235,21 @@ nxt_http_proxy_send_body(nxt_task_t *task, void *obj, void *data)
 }
 
 
+static void
+nxt_http_proxy_request_send(nxt_task_t *task, nxt_http_request_t *r,
+    nxt_buf_t *out)
+{
+    size_t  length;
+
+    if (r->peer->remainder > 0) {
+        length = nxt_buf_chain_length(out);
+        r->peer->remainder -= length;
+    }
+
+    nxt_http_request_send(task, r, out);
+}
+
+
 static const nxt_http_request_state_t  nxt_http_proxy_read_state
     nxt_aligned(64) =
 {
@@ -251,12 +272,14 @@ nxt_http_proxy_read(nxt_task_t *task, void *obj, void *data)
     peer->body = NULL;
     last = nxt_buf_is_last(out);
 
-    nxt_http_request_send(task, r, out);
+    nxt_http_proxy_request_send(task, r, out);
 
     if (!last) {
         nxt_http_proto[peer->protocol].peer_read(task, peer);
 
     } else {
+        r->inconsistent = (peer->remainder != 0);
+
         nxt_http_proto[peer->protocol].peer_close(task, peer);
 
         nxt_mp_release(r->mem_pool);