Merge bitcoin/bitcoin#36123: http: throttle per-connection reads while a request is in flight

### src/httpserver.cpp
@@ -1003,7 +1003,20 @@ HTTPServer::IOReadiness HTTPServer::GenerateWaitSockets() const
         // never hold m_sock_mutex and m_send_mutex at the same time here.
         // MaybeSendBytesFromBuffer() locks m_send_mutex then m_sock_mutex, so nesting
         // them in the opposite order here would risk a lock-order inversion deadlock.
-        Sock::Event event = (http_client->ReadyToSend() ? Sock::SendEvent : Sock::RecvEvent);
+        Sock::Event event{0};
+        if (http_client->ReadyToSend()) {
+            event = Sock::SendEvent;
+        } else if (http_client->GetRequest() != nullptr || http_client->ReceiveBufferEmpty()) {
+            // Read from the socket when the parser has an incomplete request in
+            // progress (needs more bytes) or when the buffer is empty. If the
+            // buffer is non-empty but no parse is in progress, leave event=0:
+            // the client stays in the I/O map so TryReadRequest() runs first to
+            // consume buffered bytes before admitting more socket data. Excess
+            // pipelined data then backs up in the kernel socket buffer, applying
+            // TCP backpressure instead of accumulating without bound in m_recv_buffer.
+            event = Sock::RecvEvent;
+        }
+
         io_readiness.events_per_sock.emplace(sock, Sock::Events{event});
         io_readiness.httpclients_per_sock.emplace(sock, http_client);
     }

### src/httpserver.h
@@ -502,6 +502,7 @@ class HTTPRemoteClient
     const CService& GetPeer() const { return m_addr; }
     std::shared_ptr<Sock> GetSock() EXCLUSIVE_LOCKS_REQUIRED(!m_sock_mutex) { return WITH_LOCK(m_sock_mutex, return m_sock;); }
     bool ReadyToSend() const EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex) { return WITH_LOCK(m_send_mutex, return m_send_ready;); }
+    bool ReceiveBufferEmpty() const { return m_recv_buffer.empty(); }
 
     void Send(const HTTPResponse& res, std::span<const std::byte> reply_body, bool keep_alive) EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex);
     void Receive() EXCLUSIVE_LOCKS_REQUIRED(!m_sock_mutex);
@@ -522,11 +523,15 @@ class HTTPRemoteClient
      */
     bool MaybeSendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex);
 
+    /**
+     * Used to determine if an incomplete request is in progress.
+     * @returns nullptr after a complete request is moved to a worker thread,
+     *          but before reading any new data from m_recv_buffer.
+     */
+    const HTTPRequest* GetRequest() const { return m_req.get(); }
+
     //! Used for tests.
-    //! @{
     const std::string& GetRecvBuffer() const { return m_recv_buffer; }
-    const HTTPRequest* GetRequest() const { return m_req.get(); }
-    //! @}
 
 protected:
     //! Used for tests.
@@ -564,6 +569,7 @@ class HTTPRemoteClient
 
     //! 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.
+    //! Only one request per connection is ever in flight.
     std::atomic_bool m_req_busy{false};
 
     /**

### test/functional/interface_http.py
@@ -153,6 +153,7 @@ def run_test(self):
         self.check_invalid_http_version()
         self.check_whitespace_in_headers()
         self.check_connection_limit()
+        self.check_pipelined_data_is_throttled()
 
 
     def check_default_connection(self):
@@ -699,5 +700,84 @@ def wait_for_send(conn):
                 client.close_sock()
 
 
+    def check_pipelined_data_is_throttled(self):
+        self.log.info("Check that pipelined data is throttled while a request is in flight")
+        self.restart_node(0, extra_args=["-rpcservertimeout=0"])
+
+        conn = BitcoinHTTPConnection(self.node)
+
+        # A blocking RPC request: the server reads it fully and it enters the
+        # worker pool as "in flight" (m_req_busy) until a new block arrives.
+        tip_height = self.node.getblockcount()
+        conn.post_raw('/', f'{{"method": "waitforblockheight", "params": [{tip_height + 1}]}}')
+
+        # Flood the same connection with big pipelined requests:
+        # Large garbage submitblock (just under MAX_BODY_SIZE each, including HTTP/jsonrpc overhead)
+        garbage_block = "0" * (MAX_BODY_SIZE - 100)
+        body = f'{{"method": "submitblock", "params": ["{garbage_block}"]}}'
+        flood = (
+            f'POST / HTTP/1.1\r\nAuthorization: Basic {str_to_b64str(conn.authpair)}\r\n'
+            f'Content-Length: {len(body)}\r\n\r\n' +
+            body
+        ).encode("ascii")
+
+        # Non-blocking send: When the server stops reading from the buffer
+        # due to TCP backpressure, Python will raise an error. If the socket
+        # was set to blocking, we would have to wait for an ambiguous timeout.
+        conn.conn.sock.setblocking(False)
+
+        # Kernel socket buffer sizes vary widely across platforms,
+        # so we can't rely on counting sent() bytes to determine if the
+        # server is actually draining its end of the socket.
+        # When the server is busy, a continuous flood from the client SHOULD,
+        # at some point, stall indefinitely. An unpatched server will continue
+        # to accept data from the socket, at some rate, indefinitely.
+
+        # If send() is blocked for this many seconds, we assume the server
+        # is behaving correctly.
+        STALL_TIMEOUT = 5
+        # If send() continues to progress for this many seconds, we assume
+        # the server is vulnerable to memory exhaustion.
+        PROGRESS_TIMEOUT = 10
+
+        sent = 0
+        stuck_since = None
+        start = time.monotonic()
+        while True:
+            try:
+                sent += conn.conn.sock.send(flood[sent % len(flood):])
+                # Progress: the server is still reading
+                stuck_since = None
+                self.log.debug(f"sent: {sent}")
+                assert sent <= len(flood) * 10, (
+                    f"Server accepted {sent} bytes of pipelined data while a "
+                    "request was still in flight: the receive buffer is not throttled")
+            except BlockingIOError:
+                # The kernel send buffer is full (EAGAIN).
+                # That's good, but we still need to determine if we are
+                # feeling backpressure from the server or the client-side buffer.
+                if stuck_since is None:
+                    stuck_since = time.monotonic()
+                elif time.monotonic() - stuck_since > STALL_TIMEOUT:
+                    # No progress: the server has stopped reading.
+                    break
+            if stuck_since is None and time.monotonic() - start > PROGRESS_TIMEOUT:
+                # Continuous progress: the server is still draining the
+                # receive buffer while a request is in flight.
+                raise AssertionError(
+                    f"Server kept reading pipelined data ({sent} bytes) while a "
+                    f"request was still in flight for {PROGRESS_TIMEOUT}s.")
+            time.sleep(0.05)
+
+        self.log.info(f"Pipelined flood stalled after {sent} bytes; no progress for {STALL_TIMEOUT}s.")
+
+        # Unblock the client request queue.
+        conn.conn.sock.settimeout(10)
+        generated_block = self.generate(self.node, 1, sync_fun=self.no_op)[0]
+        # First reply is for the blocking request.
+        response = conn.recv_raw().decode()
+        assert generated_block in response
+
+
 if __name__ == '__main__':
     HTTPBasicsTest(__file__).main()