diff --git a/src/httpserver.cpp b/src/httpserver.cpp index 9bb89863afc..e02c41dbb6b 100644 --- a/src/httpserver.cpp +++ b/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); } diff --git a/src/httpserver.h b/src/httpserver.h index f6810373200..a8236ab61be 100644 --- a/src/httpserver.h +++ b/src/httpserver.h @@ -502,6 +502,7 @@ public: const CService& GetPeer() const { return m_addr; } std::shared_ptr 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 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 @@ public: */ bool MaybeSendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex); - //! Used for tests. - //! @{ - const std::string& GetRecvBuffer() const { return m_recv_buffer; } + /** + * 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; } protected: //! Used for tests. @@ -564,6 +569,7 @@ private: //! 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}; /** diff --git a/test/functional/interface_http.py b/test/functional/interface_http.py index e2f8f621b8e..63c3eb052b4 100755 --- a/test/functional/interface_http.py +++ b/test/functional/interface_http.py @@ -133,6 +133,7 @@ class HTTPBasicsTest (BitcoinTestFramework): 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): @@ -683,5 +684,84 @@ class HTTPBasicsTest (BitcoinTestFramework): 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()