From 3d1004cb9b8f27bd328d95b4c7524e878c296d61 Mon Sep 17 00:00:00 2001 From: Matthew Zipkin Date: Sun, 30 Aug 2026 16:15:36 -0400 Subject: [PATCH] http: throttle per-connection reads while a request is in flight A client streaming pipelined requests into a busy connection (or any connection whose replies are slower than the sender) could grow server memory without limit, up to remote OOM. Stop selecting RecvEvent for clients whose request is being processed; pipelined data then backs up in the kernel socket buffer, applying TCP backpressure to the sender. One request per connection is in flight at a time. Functional test streams pipelined submitblock requests into a connection blocked on waitforblockheight. Unpatched builds continue draining the socket buffer indefinitely, patched builds will stall. --- src/httpserver.cpp | 15 +++++- src/httpserver.h | 14 ++++-- test/functional/interface_http.py | 80 +++++++++++++++++++++++++++++++ 3 files changed, 104 insertions(+), 5 deletions(-) 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()