Fix chunked pipelined requests (#12071)

* Fix pipelined request for chunked bodies

This addresses a variety of bugs concerning pipelined requests. In
particular, the HttpTunnel logic had fundamentally assumed that it could
consume all bytes available in the producer's reader. If a request was
pipelined after a previous request that had a chunked body, this would
result in the second request being unparsed and either sent along to the
origin or dropped on the floor, depening on configuration. This adds an
explicit autest for pipelined requests and addresses these issues.

This patch largely does the following:

1. Updates the copy_partial_post_data data to take the number of bytes
   it consumes rather than consuming all bytes in the reader. It also
   now returns the number of bytes it consumes, which the tunnel needs
   to keep track of the number of bytes it processes.
2. Previous to this patch, the HttpTunnel assumed that it could consume
   all bytes in the reader originally passed to it (all bytes in
   init_bytes_done). This simply will not work for pipelined requests.
   This addresses this issue by adding a new variable to the tunnel:
   bytes_consumed. This way the tunnel can keep track of how many bytes
   it consumed while processing the request body, which allows the
   HttpSM to later process just the right number of bytes from its
   reader rather than eating into any pipelined requests that follow it.
3. The HttpSM must not consume bytes from its client reader that are
   pipelined requests. It now uses the tunnel's processing
   bytes_consumed to process bytes from its reader rather than simply
   consuming all read_available() bytes from it.

* Fix bytes consumed chunk computation

Fix a possible miscalculation of bytes consumed while parsing chunked
content.

* is_read_closed fix

* Verify expected responses are received.

* Updated formatting

---------

Co-authored-by: Brian Neradt <brian.neradt@gmail.com>

Author: bryancall

proxy/ProxyTransaction.cc

@@ -238,6 +238,12 @@ ProxyTransaction::get_version(HTTPHdr &hdr) const
   return hdr.version_get();
 }
 
+bool
+ProxyTransaction::is_read_closed() const
+{
+  return false;
+}
+
 bool
 ProxyTransaction::allow_half_open() const
 {

proxy/ProxyTransaction.h

@@ -49,6 +49,7 @@ class ProxyTransaction : public VConnection
   virtual void set_inactivity_timeout(ink_hrtime timeout_in);
   virtual void cancel_inactivity_timeout();
   virtual void cancel_active_timeout();
+  virtual bool is_read_closed() const;
 
   // Implement VConnection interface.
   VIO *do_io_read(Continuation *c, int64_t nbytes = INT64_MAX, MIOBuffer *buf = nullptr) override;

proxy/http/HttpSM.cc

@@ -973,10 +973,8 @@ HttpSM::wait_for_full_body()
   // Next order of business if copy the remaining data from the
   //  header buffer into new buffer
   int64_t post_bytes = chunked ? INT64_MAX : t_state.hdr_info.request_content_length;
-  client_request_body_bytes =
-    post_buffer->write(ua_txn->get_remote_reader(), chunked ? ua_txn->get_remote_reader()->read_avail() : post_bytes);
+  post_buffer->write(ua_txn->get_remote_reader(), chunked ? ua_txn->get_remote_reader()->read_avail() : post_bytes);
 
-  ua_txn->get_remote_reader()->consume(client_request_body_bytes);
   p = tunnel.add_producer(ua_entry->vc, post_bytes, buf_start, &HttpSM::tunnel_handler_post_ua, HT_BUFFER_READ, "ua post buffer");
   if (chunked) {
     bool const drop_chunked_trailers = t_state.http_config_param->oride.http_drop_chunked_trailers == 1;
@@ -3633,7 +3631,24 @@ int
 HttpSM::tunnel_handler_post_ua(int event, HttpTunnelProducer *p)
 {
   STATE_ENTER(&HttpSM::tunnel_handler_post_ua, event);
-  client_request_body_bytes = p->init_bytes_done + p->bytes_read;
+
+  // Now that the tunnel is done, it can tell us how many bytes were in the
+  // body.
+  if (client_request_body_bytes == 0) {
+    // This is invoked multiple times for a transaction when buffering request
+    // body data, so we only call this the first time when
+    // client_request_body_bytes is 0.
+    client_request_body_bytes     = p->bytes_consumed;
+    IOBufferReader *client_reader = ua_txn->get_remote_reader();
+    // p->bytes_consumed represents the number of body bytes the tunnel parsed
+    // and consumed from the client. However, not all those bytes may have been
+    // written to our _ua client transaction reader. We must not consume past
+    // the number of bytes available.
+    int64_t const bytes_to_consume = std::min(p->bytes_consumed, client_reader->read_avail());
+    SMDebug("http_tunnel", "Consuming %" PRId64 " bytes from client reader with p->bytes_consumed: %" PRId64 " available: %" PRId64,
+            bytes_to_consume, p->bytes_consumed, client_reader->read_avail());
+    client_reader->consume(bytes_to_consume);
+  }
 
   switch (event) {
   case VC_EVENT_INACTIVITY_TIMEOUT:
@@ -6095,8 +6110,8 @@ HttpSM::do_setup_post_tunnel(HttpVC_t to_vc_type)
     IOBufferReader *postdata_producer_reader = postdata_producer_buffer->alloc_reader();
 
     postdata_producer_buffer->write(this->_postbuf.postdata_copy_buffer_start);
-    int64_t post_bytes = postdata_producer_reader->read_avail();
-    transfered_bytes   = post_bytes;
+    int64_t post_bytes = chunked ? INT64_MAX : t_state.hdr_info.request_content_length;
+    transferred_bytes  = post_bytes;
     p = tunnel.add_producer(HTTP_TUNNEL_STATIC_PRODUCER, post_bytes, postdata_producer_reader, (HttpProducerHandler) nullptr,
                             HT_STATIC, "redirect static agent post");
   } else {
@@ -6125,12 +6140,27 @@ HttpSM::do_setup_post_tunnel(HttpVC_t to_vc_type)
 
     // Next order of business if copy the remaining data from the
     //  header buffer into new buffer
-    client_request_body_bytes =
-      post_buffer->write(ua_txn->get_remote_reader(), chunked ? ua_txn->get_remote_reader()->read_avail() : post_bytes);
 
-    ua_txn->get_remote_reader()->consume(client_request_body_bytes);
-    p = tunnel.add_producer(ua_entry->vc, post_bytes - transfered_bytes, buf_start, &HttpSM::tunnel_handler_post_ua, HT_HTTP_CLIENT,
-                            "user agent post");
+    int64_t num_body_bytes = 0;
+    // If is_using_post_buffer has been used, then client_request_body_bytes
+    // will have already been sent in wait_for_full_body and there will be
+    // zero bytes in this user agent buffer. We don't want to clobber
+    // client_request_body_bytes with a zero value here in those cases.
+    if (client_request_body_bytes > 0) {
+      num_body_bytes = client_request_body_bytes;
+    } else {
+      num_body_bytes =
+        post_buffer->write(ua_txn->get_remote_reader(), chunked ? ua_txn->get_remote_reader()->read_avail() : post_bytes);
+    }
+    // Don't consume post_bytes here from ua_txn->get_remote_reader() since
+    // we are not sure how many bytes the tunnel will use yet. Wait until
+    // HttpSM::tunnel_handler_post_ua to consume the bytes.
+    // The user agent has already sent all it has
+    if (ua_txn->is_read_closed()) {
+      post_bytes = num_body_bytes;
+    }
+    p = tunnel.add_producer(ua_entry->vc, post_bytes - transferred_bytes, buf_start, &HttpSM::tunnel_handler_post_ua,
+                            HT_HTTP_CLIENT, "user agent post");
   }
   ua_entry->in_tunnel = true;
 
@@ -6847,6 +6877,8 @@ HttpSM::server_transfer_init(MIOBuffer *buf, int hdr_size)
     //  we'll get is already in the buffer
     nbytes = server_txn->get_remote_reader()->read_avail() + hdr_size;
   } else if (t_state.hdr_info.response_content_length == HTTP_UNDEFINED_CL) {
+    // Chunked or otherwise, no length is defined. Pass -1 to tell the
+    // tunnel that the size is unknown.
     nbytes = -1;
   } else {
     //  Set to copy to the number of bytes we want to write as
@@ -8564,16 +8596,18 @@ HttpSM::rewind_state_machine()
 
 // YTS Team, yamsat Plugin
 // Function to copy the partial Post data while tunnelling
-void
-PostDataBuffers::copy_partial_post_data()
+int64_t
+PostDataBuffers::copy_partial_post_data(int64_t consumed_bytes)
 {
   if (post_data_buffer_done) {
-    return;
+    return 0;
   }
-  Debug("http_redirect", "[PostDataBuffers::copy_partial_post_data] wrote %" PRId64 " bytes to buffers %" PRId64 "",
-        this->ua_buffer_reader->read_avail(), this->postdata_copy_buffer_start->read_avail());
-  this->postdata_copy_buffer->write(this->ua_buffer_reader);
-  this->ua_buffer_reader->consume(this->ua_buffer_reader->read_avail());
+  int64_t const bytes_to_copy = std::min(consumed_bytes, this->ua_buffer_reader->read_avail());
+  Debug("http_redirect", "given %" PRId64 " bytes consumed, copying %" PRId64 " bytes to buffers with %" PRId64 " available bytes",
+        consumed_bytes, bytes_to_copy, this->ua_buffer_reader->read_avail());
+  this->postdata_copy_buffer->write(this->ua_buffer_reader, bytes_to_copy);
+  this->ua_buffer_reader->consume(bytes_to_copy);
+  return bytes_to_copy;
 }
 
 IOBufferReader *

proxy/http/HttpSM.h

@@ -179,7 +179,7 @@ class PostDataBuffers
 
   void clear();
   void init(IOBufferReader *ua_reader);
-  void copy_partial_post_data();
+  int64_t copy_partial_post_data(int64_t consumed_bytes);
   IOBufferReader *get_post_data_buffer_clone_reader();
   void
   set_post_data_buffer_done(bool done)
@@ -313,8 +313,8 @@ class HttpSM : public Continuation, public PluginUserArgs<TS_USER_ARGS_TXN>
   bool debug_on           = false; // Transaction specific debug flag
   char *redirect_url    = nullptr; // url for force redirect (provide users a functionality to redirect to another url when needed)
   int redirect_url_len  = 0;
-  int redirection_tries = 0;    // To monitor number of redirections
-  int64_t transfered_bytes = 0; // Added to calculate POST data
+  int redirection_tries = 0;     // To monitor number of redirections
+  int64_t transferred_bytes = 0; // Added to calculate POST data
 
   // Tunneling request to plugin
   HttpPluginTunnel_t plugin_tunnel_type = HTTP_NO_PLUGIN_TUNNEL;
@@ -331,7 +331,7 @@ class HttpSM : public Continuation, public PluginUserArgs<TS_USER_ARGS_TXN>
   int64_t postbuf_buffer_avail();
   void postbuf_clear();
   void disable_redirect();
-  void postbuf_copy_partial_data();
+  int64_t postbuf_copy_partial_data(int64_t consumed_bytes);
   void postbuf_init(IOBufferReader *ua_reader);
   void set_postbuf_done(bool done);
   IOBufferReader *get_postbuf_clone_reader();
@@ -750,10 +750,10 @@ HttpSM::disable_redirect()
   this->_postbuf.clear();
 }
 
-inline void
-HttpSM::postbuf_copy_partial_data()
+inline int64_t
+HttpSM::postbuf_copy_partial_data(int64_t consumed_bytes)
 {
-  this->_postbuf.copy_partial_post_data();
+  return this->_postbuf.copy_partial_post_data(consumed_bytes);
 }
 
 inline void