Merge bitcoin/bitcoin#35735: Add state to HTTPRequest

### src/httpserver.cpp
@@ -30,7 +30,6 @@
 #include <condition_variable>
 #include <cstdio>
 #include <cstdlib>
-#include <deque>
 #include <memory>
 #include <optional>
 #include <span>
@@ -299,17 +298,22 @@ void HTTPHeaders::RemoveAll(std::string_view key)
     m_headers.erase(moved.begin(), moved.end());
 }
 
-bool HTTPHeaders::Read(util::LineReader& reader)
+bool HTTPHeaders::Read(util::LineReader& reader, bool write)
 {
     // Headers https://httpwg.org/specs/rfc9110.html#rfc.section.6.3
     // A sequence of Field Lines https://httpwg.org/specs/rfc9110.html#rfc.section.5.2
+    size_t start{reader.Consumed()};
     while (auto maybe_line = reader.ReadLine()) {
-        if (reader.Consumed() > MAX_HEADERS_SIZE) throw std::runtime_error("HTTP headers exceed size limit");
+        if (reader.Consumed() - start + m_consumed > MAX_HEADERS_SIZE) throw std::runtime_error("HTTP headers exceed size limit");
 
         const std::string_view& line = *maybe_line;
 
         // An empty line indicates end of the headers section https://www.rfc-editor.org/rfc/rfc2616#section-4
-        if (line.empty()) return true;
+        if (line.empty()) {
+            // Ensure all headers are accounted for in case there is a chunked trailer
+            m_consumed += reader.Consumed() - start;
+            return true;
+        }
 
         // "Field values containing CR, LF, or NUL characters are invalid and dangerous"
         // https://httpwg.org/specs/rfc9110.html#rfc.section.5.5
@@ -337,9 +341,16 @@ bool HTTPHeaders::Read(util::LineReader& reader)
         // that can not be empty.
         if (key.empty()) throw std::runtime_error("Empty HTTP header name");
 
-        Write(std::string(key), std::move(value));
+        if (write) {
+            Write(std::string(key), std::move(value));
+        }
     }
 
+    // We have not received all the request headers yet.
+    // Keep track of how much data we have already consumed to enforce
+    // the total limit over multiple read operations.
+    m_consumed += reader.Consumed() - start;
+
     return false;
 }
 
@@ -428,62 +439,68 @@ bool HTTPRequest::LoadBody(LineReader& reader)
         // Chunked Transfer Coding: https://datatracker.ietf.org/doc/html/rfc7230.html#section-4.1
         // see evhttp_handle_chunked_read() in libevent http.c
         while (reader.Remaining() > 0) {
-            auto maybe_chunk_size = reader.ReadLine();
-            if (!maybe_chunk_size) return false;
-
-            // Allow (but ignore) Chunk Extensions
-            // See https://www.rfc-editor.org/rfc/rfc9112.html#name-chunk-extensions
-            std::string_view chunk_size_noext{maybe_chunk_size.value()};
-            const auto semicolon_pos = chunk_size_noext.find(';');
-            if (semicolon_pos != chunk_size_noext.npos) {
-                chunk_size_noext.remove_suffix(chunk_size_noext.size() - semicolon_pos);
-            }
+            if (!m_chunk_size) {
+                auto maybe_chunk_size = reader.ReadLine();
+                if (!maybe_chunk_size) return false;
+
+                // Allow (but ignore) Chunk Extensions
+                // See https://www.rfc-editor.org/rfc/rfc9112.html#name-chunk-extensions
+                std::string_view chunk_size_noext{maybe_chunk_size.value()};
+                const auto semicolon_pos = chunk_size_noext.find(';');
+                if (semicolon_pos != chunk_size_noext.npos) {
+                    chunk_size_noext.remove_suffix(chunk_size_noext.size() - semicolon_pos);
+                }
+
+                m_chunk_size = ToIntegral<uint64_t>(util::TrimStringView(chunk_size_noext), /*base=*/16);
+                if (!m_chunk_size) throw std::runtime_error("Cannot parse chunk length value");
 
-            const auto chunk_size{ToIntegral<uint64_t>(util::TrimStringView(chunk_size_noext), /*base=*/16)};
-            if (!chunk_size) throw std::runtime_error("Cannot parse chunk length value");
+                if ((m_body.size() > MAX_BODY_SIZE) ||
+                    (*m_chunk_size > MAX_BODY_SIZE - m_body.size()))
+                    throw ContentTooLargeError("Chunk will exceed max body size");
+            }
 
-            if ((m_body.size() > MAX_BODY_SIZE) ||
-                (*chunk_size > MAX_BODY_SIZE - m_body.size()))
-                throw ContentTooLargeError("Chunk will exceed max body size");
+            // We either just read the chunk size, or we have it saved
+            // from a prior I/O loop iteration
+            Assume(m_chunk_size);
 
             // Last chunk has size 0
-            if (*chunk_size == 0) {
-                // Allow (but ignore) Chunked Trailer section, by
-                // reading CRLF-terminated lines until we read an empty line,
-                // which indicates the end of this request.
+            if (*m_chunk_size == 0) {
+                // Validate Chunked Trailer section, which is used for
+                // additional headers sent at the end of the message.
+                // Data consumed here is counted towards MAX_HEADERS_SIZE
+                // along with the headers we read in the beginning of the request.
+                // At this time we ignore and drop these data after validating.
                 // See https://httpwg.org/specs/rfc9112.html#rfc.section.7.1.2
-                const size_t trailer_start{reader.Consumed()};
-                while (true) {
-                    auto maybe_trailer = reader.ReadLine();
-                    if (reader.Consumed() - trailer_start > MAX_HEADERS_SIZE) {
-                        throw std::runtime_error("HTTP chunked trailer exceeds size limit");
-                    }
-                    if (!maybe_trailer) return false;
-                    if (maybe_trailer->empty()) break;
-                }
-                // Complete request has been parsed, reader is now pointing
-                // to beginning of next request or end of the buffer.
-                return true;
+                return m_headers.Read(reader, /*write=*/false);
             }
 
-            // We are still expecting more data for this chunk
-            if (reader.Remaining() < *chunk_size) {
-                return false;
-            }
+            // We have not read the entire chunk from the buffer yet
+            if (m_chunk_read < *m_chunk_size) {
+                // Get what we can from the buffer
+                const uint64_t chunk_need{*m_chunk_size - m_chunk_read};
+                const uint64_t buffer_has{std::min(chunk_need, static_cast<uint64_t>(reader.Remaining()))};
 
-            // Pack chunk onto body
-            m_body += reader.ReadLength(*chunk_size);
+                // Pack [partial] chunk onto body and update state
+                m_body += reader.ReadLength(buffer_has);
+                m_chunk_read += buffer_has;
+            }
 
             // Even though every chunk size is explicitly declared,
             // they are still terminated by a CRLF we don't need,
             // just consume it here.
-            auto crlf = reader.ReadLine();
-            if (!crlf) {
-                // CRLF not found before end of buffer: it has not been received by our socket yet.
-                return false;
+            if (m_chunk_read == *m_chunk_size) {
+                auto crlf = reader.ReadLine();
+                if (!crlf) {
+                    // CRLF not found before end of buffer: it has not been received by our socket yet.
+                    return false;
+                }
+                // CRLF was found but there was unexpected data after the chunk_sized chunk
+                if (!crlf.value().empty()) throw std::runtime_error("Improperly terminated chunk");
+
+                // Clear state for next chunk
+                m_chunk_size.reset();
+                m_chunk_read = 0;
             }
-            // CRLF was found but there was unexpected data after the chunk_sized chunk
-            if (!crlf.value().empty()) throw std::runtime_error("Improperly terminated chunk");
         }
 
         // We read all the chunks but never got the last chunk, wait for client to send more
@@ -505,12 +522,15 @@ bool HTTPRequest::LoadBody(LineReader& reader)
 
         if (*content_length > MAX_BODY_SIZE) throw ContentTooLargeError("Max body size exceeded");
 
-        // Not enough data in buffer for expected body
-        if (reader.Remaining() < *content_length) return false;
+        // A large body may arrive over multiple I/O loop iterations. Copy
+        // whatever the buffer has now; m_body's size tracks our progress.
+        const uint64_t body_need{*content_length - m_body.size()};
+        const uint64_t buffer_has{std::min(body_need, static_cast<uint64_t>(reader.Remaining()))};
 
-        m_body = reader.ReadLength(*content_length);
+        // Pack [partial] body on and update state
+        m_body += reader.ReadLength(buffer_has);
 
-        return true;
+        return m_body.size() == *content_length;
     }
 }
 
@@ -996,62 +1016,56 @@ void HTTPServer::ThreadSocketHandler()
 
 void HTTPServer::MaybeDispatchRequestsFromClient(const std::shared_ptr<HTTPRemoteClient>& client) const
 {
-    // Try reading (potentially multiple) HTTP requests from the buffer
-    while (!client->m_recv_buffer.empty()) {
-        // Create a new request object and try to fill it with data from the receive buffer
-        auto req = std::make_unique<HTTPRequest>(client);
-        try {
-            // Stop reading if we need more data from the client to parse a complete request
-            if (!client->ReadRequest(*req)) break;
-        } catch (const ContentTooLargeError& e) {
-            LogDebug(
-                BCLog::HTTP,
-                "HTTP request body too large from client %s (id=%llu): %s",
-                client->m_origin,
-                client->m_id,
-                e.what());
-
-            WriteNoStoreErrorReply(*req, HTTP_CONTENT_TOO_LARGE);
-            client->m_disconnect = true;
-            return;
-        } catch (const std::runtime_error& e) {
-            LogDebug(
-                BCLog::HTTP,
-                "Error reading HTTP request from client %s (id=%llu): %s",
-                client->m_origin,
-                client->m_id,
-                e.what());
-
-            // We failed to read a complete request from the buffer
-            WriteNoStoreErrorReply(*req, HTTP_BAD_REQUEST);
-            client->m_disconnect = true;
-            return;
-        }
+    // If we are already handling a request from
+    // this client, do nothing. We'll check again on the next I/O
+    // loop iteration.
+    if (client->m_req_busy) return;
+
+    if (!client->m_req) {
+        client->m_req = std::make_unique<HTTPRequest>(client);
+    }
 
-        // We read a complete request from the buffer into the queue
+    try {
+        // Read data from the buffer into the current request
+        client->ReadRequest(*client->m_req);
+    } catch (const ContentTooLargeError& e) {
         LogDebug(
             BCLog::HTTP,
-            "Received a %s request for %s from %s (id=%llu)",
-            RequestMethodString(req->m_method),
-            req->m_target,
+            "HTTP request body too large from client %s (id=%llu): %s",
             client->m_origin,
-            client->m_id);
+            client->m_id,
+            e.what());
 
-        // add request to client queue
-        client->m_req_queue.push_back(std::move(req));
+        WriteNoStoreErrorReply(*client->m_req, HTTP_CONTENT_TOO_LARGE);
+        client->m_disconnect = true;
+        return;
+    } catch (const std::runtime_error& e) {
+        LogDebug(
+            BCLog::HTTP,
+            "Error reading HTTP request from client %s (id=%llu): %s",
+            client->m_origin,
+            client->m_id,
+            e.what());
+
+        // We failed to read a complete request from the buffer
+        WriteNoStoreErrorReply(*client->m_req, HTTP_BAD_REQUEST);
+        client->m_disconnect = true;
+        return;
     }
 
-    // If we are already handling a request from
-    // this client, do nothing. We'll check again on the next I/O
-    // loop iteration.
-    if (client->m_req_busy) return;
+    // If the request is ready, hand it to a worker.
+    if (client->m_req->GetState() == HTTPRequest::State::Complete) {
+        LogDebug(
+            BCLog::HTTP,
+            "Received a %s request for %s from %s (id=%llu)",
+            RequestMethodString(client->m_req->m_method),
+            client->m_req->m_target,
+            client->m_origin,
+            client->m_id);
 
-    // Otherwise, if there is a pending request in the queue, handle it.
-    if (!client->m_req_queue.empty()) {
         LOCK(m_request_dispatcher_mutex);
         client->m_req_busy = true;
-        m_request_dispatcher(std::move(client->m_req_queue.front()));
-        client->m_req_queue.pop_front();
+        m_request_dispatcher(std::move(client->m_req));
     }
 }
 
@@ -1102,6 +1116,7 @@ void HTTPServer::DisconnectClients()
                                                  "Disconnecting HTTP client %s (id=%llu)",
                                                  client->m_origin,
                                                  client->m_id);
+                                        client->ReleaseRequest();
                                         return true;
                                     });
     if (erased > 0) {
@@ -1116,25 +1131,53 @@ void HTTPServer::ClearConnectedClients()
     if (m_connected.empty()) return;
     LogWarning("Force-disconnecting %d HTTP client(s) that did not disconnect gracefully", m_connected.size());
     m_connected_size.fetch_sub(m_connected.size(), std::memory_order_relaxed);
+    for (auto& client : m_connected) {
+        client->ReleaseRequest();
+    }
     m_connected.clear();
 }
 
-bool HTTPRemoteClient::ReadRequest(HTTPRequest& req)
+void HTTPRemoteClient::ReadRequest(HTTPRequest& req)
 {
+    if (m_recv_buffer.empty()) return;
+
     LineReader reader(m_recv_buffer, MAX_HEADERS_SIZE);
 
-    if (!req.LoadControlData(reader)) return false;
-    if (!req.LoadHeaders(reader)) return false;
-    if (!req.LoadBody(reader)) return false;
+    try {
+        switch (req.GetState()) {
+        case HTTPRequest::State::Init:
+            if (!req.LoadControlData(reader)) break;
+            req.SetState(HTTPRequest::State::NeedsHeaders);
+            [[fallthrough]];
+
+        case HTTPRequest::State::NeedsHeaders:
+            if (!req.LoadHeaders(reader)) break;
+            req.SetState(HTTPRequest::State::NeedsBody);
+            [[fallthrough]];
+
+        case HTTPRequest::State::NeedsBody:
+            if (!req.LoadBody(reader)) break;
+            req.SetState(HTTPRequest::State::Complete);
+            [[fallthrough]];
+
+        case HTTPRequest::State::Complete:
+            break;
+
+        case HTTPRequest::State::Error:
+            break;
+        }
+    } catch (...) {
+        // Don't try to read any more data for this request
+        req.SetState(HTTPRequest::State::Error);
+        // Clear the memory allocated to this client, caller must disconnect
+        m_recv_buffer.clear();
+        throw;
+    }
 
     // Remove the bytes read out of the buffer.
-    // If one of the above calls throws an error, the caller must
-    // catch it and disconnect the client.
     m_recv_buffer.erase(
         m_recv_buffer.begin(),
         m_recv_buffer.begin() + reader.Consumed());
-
-    return true;
 }
 
 bool HTTPRemoteClient::MaybeSendBytesFromBuffer()

### src/httpserver.h
@@ -6,7 +6,6 @@
 #define BITCOIN_HTTPSERVER_H
 
 #include <atomic>
-#include <deque>
 #include <functional>
 #include <memory>
 #include <optional>
@@ -105,13 +104,15 @@ class HTTPHeaders
      */
     void RemoveAll(std::string_view key);
     /**
+     * @param[in] reader A LineReader instance initialized with the client's receive buffer.
+     * @param[in] write  Whether or not to write the parsed data to the object after validation.
      * @returns false if LineReader hits the end of the buffer before reading an
      *                \n, meaning that we are still waiting on more data from the client.
      *          true  after reading an entire HTTP headers section, terminated
      *                by an empty line and \n.
      * @throws on exceeded read limit and on bad headers syntax (e.g. no ":" in a line)
      */
-    bool Read(util::LineReader& reader);
+    bool Read(util::LineReader& reader, bool write = true);
     std::string Stringify() const;
 
 private:
@@ -120,6 +121,9 @@ class HTTPHeaders
      * https://httpwg.org/specs/rfc9110.html#rfc.section.5.2
      */
     std::vector<std::pair<std::string, std::string>> m_headers;
+
+    //! Track total bytes consumed in Read() for limit checks
+    size_t m_consumed{0};
 };
 
 struct HTTPVersion {
@@ -195,6 +199,27 @@ class HTTPRequest
     std::pair<bool, std::string> GetHeader(std::string_view hdr) const;
     std::string ReadBody() const { return m_body; }
     void WriteHeader(std::string&& hdr, std::string&& value);
+
+    enum class State {
+        Init,
+        NeedsHeaders,
+        NeedsBody,
+        Complete,
+        Error
+    };
+    State GetState() const { return m_state; }
+    void SetState(State state) { m_state = state; }
+
+    // If a large request is sent with "Transfer-encoding: chunked" we may
+    // read the chunk size in a separate I/O loop iteration than the chunk
+    // of data itself. Store the chunk size value here until the chunk is read.
+    std::optional<uint64_t> m_chunk_size;
+    // We may also read a large chunk over multiple loop iterations.
+    // Track the progress of the chunk here.
+    uint64_t m_chunk_read{0};
+
+private:
+    State m_state = State::Init;
 };
 
 class HTTPServer
@@ -479,10 +504,9 @@ class HTTPRemoteClient
     std::string m_recv_buffer{};
 
     //! Requests from a client must be processed in the order in which
-    //! they were received, blocking on a per-client basis. We won't
-    //! process the next request in the queue if we are currently busy
-    //! handling a previous request.
-    std::deque<std::unique_ptr<HTTPRequest>> m_req_queue;
+    //! they were received, blocking on a per-client basis. We read
+    //! one request at a time from the socket buffer then pass it to a worker.
+    std::unique_ptr<HTTPRequest> m_req;
 
     //! Set to true by the I/O thread when a request is popped off
     //! and passed to a worker thread, reset to false by the worker thread.
@@ -555,12 +579,19 @@ class HTTPRemoteClient
     HTTPRemoteClient(const HTTPRemoteClient&) = delete;
     HTTPRemoteClient& operator=(const HTTPRemoteClient&) = delete;
 
+    //! Release any in-progress request. HTTPRequest holds a shared_ptr back to its
+    //! HTTPRemoteClient to keep the client alive from a worker thread. If a request
+    //! hasn't been moved to a worker yet it will prevent the client from destructing
+    //! and never close the socket. Therefore this must be called when disconnecting.
+    void ReleaseRequest() { m_req.reset(); }
+
     /**
      * Try to read an HTTP request from the receive buffer.
+     * Updates HTTPRequest.m_state and drains buffer on error.
      * @param[in]   req     A HTTPRequest to read into
-     * @returns true upon reading a complete request, otherwise false (may throw).
+     * @throws std::runtime_error if request is unreadable or violates protocol
      */
-    bool ReadRequest(HTTPRequest& req);
+    void ReadRequest(HTTPRequest& req);
 
     /**
      * Push data (if there is any) from client's m_send_buffer to the connected socket.

### src/test/httpserver_tests.cpp
@@ -446,8 +446,9 @@ BOOST_AUTO_TEST_CASE(http_request_tests)
         BOOST_CHECK(req.LoadHeaders(reader));
         BOOST_CHECK(req.LoadBody(reader));
         BOOST_CHECK_EQUAL(req.m_body, R"({"method":"getblockcount"})");
-        // Chunk Trailer was cleared
+        // Chunk Trailer was parsed, but ignored
         BOOST_CHECK_EQUAL(reader.Remaining(), 0);
+        BOOST_CHECK(!req.GetHeader("Expires").first);
     }
     {
         // Invalid "chunked" transfer, using roman numerals instead of hex for chunk length
@@ -483,26 +484,342 @@ BOOST_AUTO_TEST_CASE(http_request_tests)
         BOOST_CHECK(req.LoadHeaders(reader));
         BOOST_CHECK_EXCEPTION(req.LoadBody(reader), std::runtime_error, HasReason{"Improperly terminated chunk"});
     }
+}
+
+BOOST_AUTO_TEST_CASE(http_request_state_tests)
+{
+    // For these tests we just need a receive buffer for the requests to read from.
+    class DummyClient : public HTTPRemoteClient
     {
-        // End of buffer reached without chunk termination, caller must wait for more data to arrive
-        HTTPRequest req;
-        std::string delayed_chunked = "GET / HTTP/1.0\n"
-                                      "Transfer-Encoding: chunked\n"
-                                      "\n"
-                                      "10\n"
-                                      R"({"method":"getbl)""\n"
-                                      "a\n"
-                                      R"(ockcount"})";
-        LineReader reader1(delayed_chunked, MAX_HEADERS_SIZE);
-        BOOST_CHECK(req.LoadControlData(reader1));
-        BOOST_CHECK(req.LoadHeaders(reader1));
-        BOOST_CHECK(!req.LoadBody(reader1));
-        // more data arrives!
-        delayed_chunked += "\n0\n\n";
-        LineReader reader2(delayed_chunked, MAX_HEADERS_SIZE);
-        BOOST_CHECK(req.LoadControlData(reader2));
-        BOOST_CHECK(req.LoadHeaders(reader2));
-        BOOST_CHECK(req.LoadBody(reader2));
+    public:
+        DummyClient() : HTTPRemoteClient{/*id=*/0, /*addr=*/CService(), /*socket=*/CreateSock(0, 0, 0)} {}
+
+        void receive(std::string_view s)
+        {
+            m_recv_buffer.insert(
+                m_recv_buffer.end(),
+                s.begin(),
+                s.end());
+        }
+    };
+
+    {
+        // Step through state machine
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST / HTTP/1.0\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsHeaders);
+
+        client->receive("Host: 127.0.0.1\n"
+                        "Content-Length: 10\n\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        client->receive("I miss you\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+
+        // m_req holds a shared_ptr back to the client, so break the cycle
+        // before the client goes out of scope (as the server does on disconnect).
+        client->ReleaseRequest();
+    }
+    {
+        // Read body over multiple data pushes, multiple requests in same push
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST / HTTP/1.0\n"
+                        "Host: 127.0.0.1\n"
+                        "Content-Length: 10\n\n"
+                        "I miss");
+        client->ReadRequest(*client->m_req);
+        // Because of the Content-Length header we know the body is not complete
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Finish sending first request and include second request in the same buffer
+        client->receive(" you"
+                        "GET /endpoint HTTP/1.0\n\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+        BOOST_CHECK_EQUAL(client->m_req->m_body, "I miss you");
+        // Next request sitting in buffer
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 24);
+        // Complete first request hasn't been moved yet, expect no-op
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 24);
+
+        // Reset m_req
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        // Read second request
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+        BOOST_CHECK_EQUAL(client->m_req->m_target, "/endpoint");
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 0);
+        // Buffer is cleared
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 0);
+
+        client->ReleaseRequest();
+    }
+    {
+        // A Content-Length body is drained out of the receive buffer as it
+        // arrives, instead of accumulating there until the request is complete.
+
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST / HTTP/1.0\n"
+                        "Content-Length: 30000\n\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Body arrives in 10kB pieces. Each one is copied onto m_body and
+        // erased from the receive buffer, which never holds more than one piece.
+        for (int i = 1; i <= 3; ++i) {
+            client->receive(std::string(10000, 'x'));
+            BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 10000);
+            client->ReadRequest(*client->m_req);
+            BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 10000 * i);
+            BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 0);
+        }
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+
+        client->ReleaseRequest();
+    }
+    {
+        // A body sent in the same push as the next request is split correctly
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST / HTTP/1.0\n"
+                        "Content-Length: 4\n\n"
+                        "body"
+                        "GET /next HTTP/1.0\n\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+        BOOST_CHECK_EQUAL(client->m_req->m_body, "body");
+        // Only the second request is left over
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 20);
+
+        client->ReleaseRequest();
+    }
+    {
+        // Chunked transfer with state
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+
+        BOOST_CHECK(!client->m_req->m_chunk_size);
+        BOOST_CHECK_EQUAL(client->m_req->m_chunk_read, 0);
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 0);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        // First chunk is incomplete
+        client->receive("GET / HTTP/1.0\n"
+                        "Transfer-Encoding: chunked\n"
+                        "\n"
+                        "10\n"
+                        R"({"method)");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK(client->m_req->m_chunk_size);
+        BOOST_CHECK_EQUAL(*client->m_req->m_chunk_size, 16);
+        BOOST_CHECK_EQUAL(client->m_req->m_chunk_read, 8);
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 8);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // More data arrives, chunk is completed.
+        client->receive(R"(":"getbl)""\n");
+        client->ReadRequest(*client->m_req);
+        // State is reset
+        BOOST_CHECK(!client->m_req->m_chunk_size);
+        BOOST_CHECK_EQUAL(client->m_req->m_chunk_read, 0);
+        // New data is added to body but body is still incomplete
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 16);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Next chunk arrives without terminal CRLF
+        client->receive("a\n"
+                        R"(ockcount"})");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK(client->m_req->m_chunk_size);
+        BOOST_CHECK_EQUAL(*client->m_req->m_chunk_size, 10);
+        BOOST_CHECK_EQUAL(client->m_req->m_chunk_read, 10);
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 26);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Chunk terminal CRLF arrives with final (size 0) chunk
+        client->receive("\n0\n\n");
+        client->ReadRequest(*client->m_req);
+        // Body size hasn't changed
+        BOOST_CHECK_EQUAL(client->m_req->m_body.size(), 26);
+        // We're done
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+        BOOST_CHECK_EQUAL(client->m_req->m_body, R"({"method":"getblockcount"})");
+
+        client->ReleaseRequest();
+    }
+    {
+        // Invalid headers: error state stops reading
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+
+        // Request is in the buffer
+        client->receive("POST / HTTP/1.0\n"
+                        "Host: 127.0.0.1\n"
+                        "Invalid header with no colon\n"
+                        "\n"
+                        "body is not read");
+        BOOST_CHECK(!client->m_recv_buffer.empty());
+
+        // Reading throws an error, sets state
+        BOOST_CHECK_EXCEPTION(client->ReadRequest(*client->m_req),
+                              std::runtime_error,
+                              HasReason{"HTTP header missing colon (:)"});
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Error);
+
+        // We read up to the invalid line
+        BOOST_CHECK_EQUAL(*client->m_req->m_headers.FindFirst("Host"), "127.0.0.1");
+        // Buffer was cleared, client should just be disconnected now
+        BOOST_CHECK(client->m_recv_buffer.empty());
+
+        // Even if more data comes in, trying to read again in error state is a no-op
+        client->receive("Content-Length: 2\n\nok");
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 21);
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_recv_buffer.size(), 21);
+
+        client->ReleaseRequest();
+    }
+    {
+        // Headers sent in batches that are below MAX_HEADERS_SIZE but the total is excessive
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST /huge HTTP/1.0\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsHeaders);
+
+        for (int i = 0; i < 410; ++i) {
+            client->receive("key:value\n");
+        }
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsHeaders);
+
+        for (int i = 0; i < 409; ++i) {
+            client->receive("key:value\n");
+        }
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsHeaders);
+
+        // We're at 819 x 10-byte headers
+        // The limit is 8192, three more bytes should throw.
+        client->receive("k:\n");
+        BOOST_CHECK_EXCEPTION(client->ReadRequest(*client->m_req),
+                              std::runtime_error,
+                              HasReason{"HTTP headers exceed size limit"});
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Error);
+
+        client->ReleaseRequest();
+    }
+    {
+        // Client sends chunks that are below the limit but the total is excessive
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Init);
+
+        client->receive("POST /huge HTTP/1.0\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsHeaders);
+
+        client->receive("Transfer-Encoding: chunked\n\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Send 16-byte chunk
+        client->receive("10\nno auto updates!\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // The next chunk will be of size 32MiB - 16 + 1, below the limit
+        // on its own but not if it were added to the total cumulative body so far.
+        // We don't need to actually send or prepare this amount of data.
+        client->receive("1fffff1\n");
+        BOOST_CHECK_EXCEPTION(client->ReadRequest(*client->m_req),
+                              http_bitcoin::ContentTooLargeError,
+                              HasReason{"Chunk will exceed max body size"});
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Error);
+
+        client->ReleaseRequest();
+    }
+    {
+        // Ensure chunk trailer is parsed over state lines
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+
+        // Send a 1-byte chunk then send the 0-chunk with a trailer but no terminal CRLF
+        client->receive("GET / HTTP/1.0\n"
+                        "Transfer-Encoding: chunked\n"
+                        "\n"
+                        "1\n"
+                        "x\n"
+                        "0\n"
+                        "Digest: sha-4=deadbeef\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Send first part of another trailer line
+        client->receive("Expires:");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Finish the trailer line
+        client->receive("never\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // Terminate
+        client->receive("\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Complete);
+        BOOST_CHECK_EQUAL(client->m_req->m_body, "x");
+
+        client->ReleaseRequest();
+    }
+    {
+        // Ensure chunk trailer counts towards the headers size limit
+        std::shared_ptr<DummyClient> client{std::make_shared<DummyClient>()};
+        client->m_req = std::make_unique<HTTPRequest>(client);
+
+        client->receive("POST /huge HTTP/1.0\n"
+                        "Transfer-Encoding: chunked\n"); //   27 bytes
+        for (int i = 0; i < 816; ++i) {
+            client->receive("key:value\n");              // 8160
+        }
+        client->receive("\n"                             //    1
+                        "1\n"
+                        "x\n"
+                        "0\n");
+        client->ReadRequest(*client->m_req);
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::NeedsBody);
+
+        // We're in the trailer section with a total of 8188 bytes of headers.
+        // The limit is 8192, five more bytes should throw.
+        client->receive("k:vv\n");
+        BOOST_CHECK_EXCEPTION(client->ReadRequest(*client->m_req),
+                              std::runtime_error,
+                              HasReason{"HTTP headers exceed size limit"});
+        BOOST_CHECK_EQUAL(client->m_req->GetState(), HTTPRequest::State::Error);
+
+        client->ReleaseRequest();
     }
 }
 

### test/functional/interface_http.py
@@ -9,6 +9,8 @@
 from test_framework.util import assert_equal, str_to_b64str
 
 import http.client
+import socket
+import threading
 import time
 import urllib.parse
 
@@ -114,7 +116,8 @@ def run_test(self):
         self.check_keepalive_connection()
         self.check_close_connection()
         self.check_excessive_request_size()
-        self.check_pipelining()
+        self.check_pipelining(with_invalid_second_request=False)
+        self.check_pipelining(with_invalid_second_request=True)
         self.check_chunked_transfer()
         self.check_idle_timeout()
         self.check_server_busy_idle_timeout()
@@ -222,39 +225,56 @@ def check_excessive_request_size(self):
         assert_equal(response4.status, http.client.OK)
 
         conn = BitcoinHTTPConnection(self.node)
-        try:
-            # Excessive body size is invalid
-            conn.post_raw('/', f'{{"jsonrpc": "2.0", "id": "0", "method": "submitblock", "params": ["{"0" * bytes_above_limit}"]}}')
-            self.log.info("Client finished sending request before connection was terminated")
-        except NETWORK_ERRORS:
-            self.log.info("Client did not finish sending request before connection was terminated")
 
-        # The server will send a 413 response and disconnect but due to a race
-        # condition, the python client may or may not read the response before
-        # detecting the broken socket (which it may still be trying to write to).
+        # Split off the send into a background thread. When the server detects
+        # the excessive size it will stop reading from the socket, but the client
+        # will continue trying to write until the backpressure eventually
+        # drops the TCP window size to 0. While the send operation is blocking until
+        # it times out, we can still receive the server's response in the foreground.
+
+        def send_excessive_body(self, conn):
+            try:
+                # Excessive body size is invalid
+                conn.post_raw('/', f'{{"jsonrpc": "2.0", "id": "0", "method": "submitblock", "params": ["{"0" * bytes_above_limit}"]}}')
+                # On some platforms (e.g. Windows) the whole request may be
+                # accepted into the OS send buffer before the server disconnects.
+                # It's ok to allow that, the server-side behavior is asserted in
+                # the foreground thread via the 413 response.
+                self.log.info("Client finished sending request before connection was terminated")
+            except NETWORK_ERRORS:
+                self.log.info("Client did not finish sending request before connection was terminated")
+
+        send_thread = threading.Thread(target=send_excessive_body, args=(self, conn))
+        send_thread.start()
+
+        response5 = conn.recv_raw().decode()
+        assert "413 Content too large" in response5
+
         try:
-            response5 = conn.conn.getresponse()
-            assert_equal(response5.status, http.client.REQUEST_ENTITY_TOO_LARGE)
-            self.log.info(f"Client got expected response status {response5.status}")
-            assert conn.sock_closed()
-        except NETWORK_ERRORS:
-            self.log.info("Client did not read response before disconnecting")
+            conn.conn.sock.shutdown(socket.SHUT_RDWR)
+            self.log.info("Send thread force-closed by test framework")
+        except OSError:
+            self.log.info("Send thread was already closed by RST from server")
+        send_thread.join()
 
 
-    def check_pipelining(self):
+    def check_pipelining(self, with_invalid_second_request):
         """
         Requests are responded to in the order in which they were received
         See https://www.rfc-editor.org/rfc/rfc7230#section-6.3.2
         """
-        self.log.info("Check pipelining")
+        self.log.info("Check pipelining" + (" with invalid second request" if with_invalid_second_request else ""))
         tip_height = self.node.getblockcount()
         conn = BitcoinHTTPConnection(self.node)
         conn.set_timeout(5)
 
         # Send two requests in a row.
         # The first request will block the second indefinitely
         conn.post_raw('/', f'{{"method": "waitforblockheight", "params": [{tip_height + 1}]}}')
-        conn.post_raw('/', '{"method": "getblockcount"}')
+        if with_invalid_second_request:
+            conn.post_raw(f'/{"x" * MAX_HEADERS_SIZE * 2}', '{"method": "getblockcount"}')
+        else:
+            conn.post_raw('/', '{"method": "getblockcount"}')
 
         try:
             # The server should not respond to the second request until the first
@@ -268,16 +288,30 @@ def check_pipelining(self):
         # Use a separate http connection to generate a block
         self.generate(self.node, 1, sync_fun=self.no_op)
 
-        # Wait for two responses to be received
+        # Wait for responses to be received
+        if with_invalid_second_request:
+            OK = 1
+            BAD = 1
+        else:
+            OK = 2
+            BAD = 0
         res = b""
-        while res.count(b"result") != 2:
+        while True:
             res += conn.recv_raw()
+            if res.count(b"HTTP/1.1 200") == OK and res.count(b"HTTP/1.1 400") == BAD:
+                break
 
         # waitforblockheight was responded to first, and then getblockcount
         # which includes the block added after the request was made
         chunks = res.split(b'"result":')
         assert chunks[1].startswith(b'{"hash":')
-        assert chunks[2].startswith(bytes(f'{tip_height + 1}', 'utf8'))
+        if with_invalid_second_request:
+            # The response to the in-flight first request is sent before the
+            # error generated by parsing the second one, even though the second
+            # request could have been rejected much earlier.
+            assert res.index(b"HTTP/1.1 200") < res.index(b"HTTP/1.1 400")
+        else:
+            assert chunks[2].startswith(bytes(f'{tip_height + 1}', 'utf8'))
 
 
     def check_chunked_transfer(self):
@@ -314,27 +348,41 @@ def check_chunked_transfer(self):
             b'3' * 10000000,
             b'"]}'
         ]
-        try:
-            conn.conn.request(
-                method='POST',
-                url='/',
-                body=iter(body_chunked),
-                headers=headers_chunked,
-                encode_chunked=True)
-            self.log.info("Client finished sending request before connection was terminated")
-        except NETWORK_ERRORS:
-            self.log.info("Client did not finish sending request before connection was terminated")
 
-        # The server will send a 413 response and disconnect but due to a race
-        # condition, the python client may or may not read the response before
-        # detecting the broken socket (which it may still be trying to write to).
+        # Split off the send into a background thread. When the server detects
+        # the excessive size it will stop reading from the socket, but the client
+        # will continue trying to write until the backpressure eventually
+        # drops the TCP window size to 0. While the send operation is blocking until
+        # it times out, we can still receive the server's response in the foreground.
+
+        def send_excessive_chunked(self, conn):
+            try:
+                conn.conn.request(
+                    method='POST',
+                    url='/',
+                    body=iter(body_chunked),
+                    headers=headers_chunked,
+                    encode_chunked=True)
+                # On some platforms (e.g. Windows) the whole request may be
+                # accepted into the OS send buffer before the server disconnects.
+                # It's ok to allow that, the server-side behavior is asserted in
+                # the foreground thread via the 413 response.
+                self.log.info("Client finished sending request before connection was terminated")
+            except NETWORK_ERRORS:
+                self.log.info("Client did not finish sending request before connection was terminated")
+
+        send_thread = threading.Thread(target=send_excessive_chunked, args=(self, conn))
+        send_thread.start()
+
+        response2 = conn.recv_raw().decode()
+        assert "413 Content too large" in response2
+
         try:
-            response2 = conn.conn.getresponse()
-            assert_equal(response2.status, http.client.REQUEST_ENTITY_TOO_LARGE)
-            self.log.info(f"Client got expected response status {response2.status}")
-            assert conn.sock_closed()
-        except NETWORK_ERRORS:
-            self.log.info("Client did not read response before disconnecting")
+            conn.conn.sock.shutdown(socket.SHUT_RDWR)
+            self.log.info("Send thread force-closed by test framework")
+        except OSError:
+            self.log.info("Send thread was already closed by RST from server")
+        send_thread.join()
 
 
     def check_idle_timeout(self):