Merge pull request #11268

### contrib/epee/include/net/abstract_tcp_server2.h
@@ -173,7 +173,6 @@ namespace net_utils
         struct {
           std::deque<epee::byte_slice> queue;
           std::size_t total_bytes;
-          bool wait_consume;
         } write;
       };
 

### contrib/epee/include/net/abstract_tcp_server2.inl
@@ -54,7 +54,6 @@
 
 #include <algorithm>
 #include <functional>
-#include <random>
 
 #undef MONERO_DEFAULT_LOG_CATEGORY
 #define MONERO_DEFAULT_LOG_CATEGORY "net"
@@ -579,7 +578,6 @@ namespace net_utils
         m_state.data.write.queue.pop_back();
         m_state.data.write.total_bytes -=
           std::min(m_state.data.write.total_bytes, byte_count);
-        m_state.condition.notify_all();
         if (m_state.data.write.queue.empty() && m_state.socket.shutdown_read) {
           // All writes have been sent and reads shutdown already, connection can be closed
           interrupt();
@@ -836,24 +834,9 @@ namespace net_utils
     if (std::numeric_limits<std::size_t>::max() - m_state.data.write.total_bytes < message.size())
       return false;
 
-    // Wait for the write queue to fall below the max. If it doesn't after a
-    // randomized delay, drop the connection. P2P senders fail fast instead of
-    // parking an io_context worker thread here.
-    auto wait_consume = [this] {
-      auto random_delay = []{
-        using engine = std::mt19937;
-        std::random_device dev;
-        std::seed_seq::result_type rand[
-          engine::state_size // Use complete bit space
-        ]{};
-        std::generate_n(rand, engine::state_size, std::ref(dev));
-        std::seed_seq seed(rand, rand + engine::state_size);
-        engine rng(seed);
-        return std::chrono::milliseconds(
-          std::uniform_int_distribution<>(5000, 6000)(rng)
-        );
-      };
-
+    // Do not park an io_context worker waiting for the write queue to drain:
+    // all workers could be sending, leaving none to complete the writes.
+    auto check_send_queue = [this] {
       // The bytes check intentionally does not include incoming message size.
       // This allows for a soft overflow; a single http response will never fail
       // this check, but multiple responses could. Clients can avoid this case
@@ -863,49 +846,10 @@ namespace net_utils
           m_state.data.write.total_bytes <= static_cast<shared_state&>(connection_basic::get_state()).response_soft_limit)
         return true;
 
-      if (m_connection_type == e_connection_type_P2P) {
-        MWARNING("Connection " << get_context().m_connection_id << " tripped write limit, terminating");
-        terminate_async();
-        return false;
-      }
-      m_state.data.write.wait_consume = true;
-      bool success = m_state.condition.wait_for(
-        m_state.lock,
-        random_delay(),
-        [this]{
-          return (
-            m_state.status != status_t::RUNNING ||
-            (
-              m_state.data.write.queue.size() <=
-                ABSTRACT_SERVER_SEND_QUE_MAX_COUNT &&
-              m_state.data.write.total_bytes <=
-                static_cast<shared_state&>(connection_basic::get_state()).response_soft_limit
-            )
-          );
-        }
-      );
-      m_state.data.write.wait_consume = false;
-      if (!success) {
-        terminate_async();
-        return false;
-      }
-      else
-        return m_state.status == status_t::RUNNING;
-    };
-    auto wait_sender = [this] {
-      m_state.condition.wait(
-        m_state.lock,
-        [this] {
-          return (
-            m_state.status != status_t::RUNNING ||
-            !m_state.data.write.wait_consume
-          );
-        }
-      );
-      return m_state.status == status_t::RUNNING;
-    };
-    if (!wait_sender())
+      MWARNING("Connection " << get_context().m_connection_id << " tripped write limit, terminating");
+      terminate_async();
       return false;
+    };
     /* CHUNK_SIZE indirectly caps outgoing to 128 * 1024 * 1000
      (ABSTRACT_SERVER_SEND_QUE_MAX_COUNT). The "soft" limit total is currently
      100 MiB (ABSTRACT_SERVER_SEND_QUE_MAX_BYTES_DEFAULT). These values will
@@ -914,7 +858,7 @@ namespace net_utils
     if (m_connection_type == e_connection_type_RPC ||
       message.size() <= 2 * CHUNK_SIZE
     ) {
-      if (!wait_consume())
+      if (!check_send_queue())
         return false;
       const std::size_t byte_count = message.size();
       m_state.data.write.queue.emplace_front(std::move(message));
@@ -928,7 +872,7 @@ namespace net_utils
       });
 
       while (!message.empty()) {
-        if (!wait_consume())
+        if (!check_send_queue())
           return false;
         m_state.data.write.queue.emplace_front(
           message.take_slice(CHUNK_SIZE)
@@ -937,7 +881,6 @@ namespace net_utils
         start_write();
       }
     }
-    m_state.condition.notify_all();
     return true;
   }
 

### contrib/epee/include/net/http_protocol_handler.inl
@@ -486,12 +486,19 @@ namespace net_utils
 				if(handle_request_and_send_response(m_query_info))
 					set_ready_state();
 				else
+				{
 					m_state = http_state_error;
+					return false;
+				}
 			}
 			m_len_remain = m_len_summary;
 		}else
 		{//current query finished, next will be next query
-			handle_request_and_send_response(m_query_info);
+			if(!handle_request_and_send_response(m_query_info))
+			{
+				m_state = http_state_error;
+				return false;
+			}
 			set_ready_state();
 		}
 
@@ -539,7 +546,10 @@ namespace net_utils
 			if(handle_request_and_send_response(m_query_info))
 				set_ready_state();
 			else
+			{
 				m_state = http_state_error;
+				return false;
+			}
 		}
 		return true;
 	}
@@ -636,7 +646,8 @@ namespace net_utils
 		if ((response.m_body.size() && (query_info.m_http_method != http::http_method_head)) || (query_info.m_http_method == http::http_method_options))
 			response_data += response.m_body;
 
-		m_psnd_hndlr->do_send(byte_slice{std::move(response_data)});
+		if(!m_psnd_hndlr->do_send(byte_slice{std::move(response_data)}))
+			return false;
 		m_psnd_hndlr->send_done();
 		return res;
 	}

### tests/unit_tests/epee_boosted_tcp_server.cpp
@@ -32,7 +32,6 @@
 #include <boost/chrono/chrono.hpp>
 #include <boost/thread/condition_variable.hpp>
 #include <boost/thread/mutex.hpp>
-#include <condition_variable>
 #include <mutex>
 #include <thread>
 
@@ -605,149 +604,6 @@ TEST(test_epee_connection, ssl_handshake)
     workers.back().join();
 }
 
-namespace
-{
-  struct config_t {
-    using condition_t = std::condition_variable_any;
-    using lock_guard_t = std::lock_guard<std::mutex>;
-    void notify_success()
-    {
-      lock_guard_t guard(lock);
-      success = true;
-      condition.notify_all();
-    }
-
-    template<typename T>
-    static bool after_init_connection(const std::shared_ptr<T>& conn)
-    {
-      if (!conn)
-        return false;
-      conn->m_protocol_handler.after_init_connection();
-      return true;
-    }
-
-    std::mutex lock;
-    condition_t condition;
-    bool success;
-  };
-}
-
-TEST(boosted_tcp_server, strand_deadlock)
-{
-  using context_t = epee::net_utils::connection_context_base;
-  using lock_t = std::mutex;
-  using unique_lock_t = std::unique_lock<lock_t>;
-
-  struct handler_t {
-    using config_type = config_t;
-    using connection_context = context_t;
-    using byte_slice_t = epee::byte_slice;
-    using socket_t = epee::net_utils::i_service_endpoint;
-
-    handler_t(socket_t *socket, config_t &config, context_t &context):
-      socket(socket),
-      config(config),
-      context(context)
-    {}
-    void after_init_connection()
-    {
-      unique_lock_t guard(lock);
-      if (!context.m_is_income) {
-        guard.unlock();
-        socket->do_send(byte_slice_t{"."});
-      }
-    }
-    void handle_qued_callback()
-    {
-    }
-    bool handle_recv(const char *data, size_t bytes_transferred)
-    {
-      unique_lock_t guard(lock);
-      if (!context.m_is_income) {
-        if (context.m_recv_cnt == 1024) {
-          guard.unlock();
-          socket->do_send(byte_slice_t{"."});
-        }
-      }
-      else {
-        if (context.m_recv_cnt == 1) {
-          for(size_t i = 0; i < 1024; ++i) {
-            guard.unlock();
-            socket->do_send(byte_slice_t{"."});
-            guard.lock();
-          }
-        }
-        else if(context.m_recv_cnt == 2) {
-          guard.unlock();
-          socket->close(false);
-        }
-      }
-      return true;
-    }
-    void release_protocol()
-    {
-      unique_lock_t guard(lock);
-      if(!context.m_is_income
-        && context.m_recv_cnt == 1024
-        && context.m_send_cnt == 2
-      ) {
-        guard.unlock();
-        config.notify_success();
-      }
-    }
-
-    lock_t lock;
-    socket_t *socket;
-    config_t &config;
-    context_t &context;
-  };
-
-  using server_t = epee::net_utils::boosted_tcp_server<handler_t>;
-  using endpoint_t = boost::asio::ip::tcp::endpoint;
-
-  endpoint_t endpoint(boost::asio::ip::make_address("127.0.0.1"), 5262);
-  server_t server(epee::net_utils::e_connection_type_RPC);
-  server.init_server(
-    endpoint.port(),
-    endpoint.address().to_string(),
-    {},
-    {},
-    {},
-    true,
-    epee::net_utils::ssl_support_t::e_ssl_support_disabled
-  );
-  server.run_server(2, {});
-  server.async_call(
-    [&]{
-      context_t context;
-      ASSERT_TRUE(
-        server.connect(
-          endpoint.address().to_string(),
-          std::to_string(endpoint.port()),
-          5,
-          context,
-          "0.0.0.0",
-          epee::net_utils::ssl_support_t::e_ssl_support_disabled
-        )
-      );
-    }
-  );
-  {
-    unique_lock_t guard(server.get_config_object().lock);
-    EXPECT_TRUE(
-      server.get_config_object().condition.wait_for(
-        guard,
-        std::chrono::seconds(5),
-        [&] { return server.get_config_object().success; }
-      )
-    );
-  }
-
-  server.send_stop_signal();
-  server.timed_wait_server_stop(5 * 1000);
-  server.deinit_server();
-}
-
 namespace
 {
   struct shutdown_handler_t;
@@ -878,110 +734,125 @@ TEST(boosted_tcp_server, shutdown)
   ev.wait();
 }
 
-TEST(boosted_tcp_server, write_failure)
+namespace
 {
-  using context_t = epee::net_utils::connection_context_base;
+  class send_queue_test : public testing::TestWithParam<epee::net_utils::t_connection_type>
+  {
+    struct config_t {
+      static constexpr bool after_init_connection(const std::shared_ptr<epee::net_utils::connection_basic>&) noexcept
+      {
+        return true;
+      }
+    };
 
-  struct config_t {
-    static constexpr bool after_init_connection(const std::shared_ptr<epee::net_utils::connection_basic>&) noexcept
-    {
-      return true;
-    }
-  };
+    struct handler_t {
+      using config_type = config_t;
+      using connection_context = epee::net_utils::connection_context_base;
 
-  struct handler_t {
-    using config_type = config_t;
-    using connection_context = context_t;
-    using socket_t = epee::net_utils::i_service_endpoint;
+      handler_t(epee::net_utils::i_service_endpoint*, config_t&, connection_context&)
+      {}
+      void handle_qued_callback()
+      {}
+      bool handle_recv(const char*, size_t)
+      {
+        ADD_FAILURE() << "Unexpected input";
+        return false;
+      }
+      void release_protocol()
+      {}
+    };
 
-    handler_t(socket_t *socket, config_t &config, context_t &):
-      config(config)
-    {}
-    
-    void handle_qued_callback()
-    {}
+  protected:
+    using connection_t = epee::net_utils::connection<handler_t>;
+    using tcp_t = boost::asio::ip::tcp;
+    using byte_slice_t = epee::byte_slice;
 
-    bool handle_recv(const char *data, size_t bytes_transferred)
+    boost::asio::io_context context;
+    tcp_t::socket peer{context};
+    std::shared_ptr<connection_t::shared_state> shared = std::make_shared<connection_t::shared_state>();
+    std::shared_ptr<connection_t> connection;
+
+    void SetUp() override
     {
-      throw std::runtime_error{"UNEXPECTED!"};
+      tcp_t::acceptor acceptor{context, {boost::asio::ip::make_address("127.0.0.1"), 0}};
+      acceptor.async_accept(peer, [](auto error) { EXPECT_FALSE(error); });
+      tcp_t::socket socket{context};
+      socket.async_connect(acceptor.local_endpoint(), [](auto error) { EXPECT_FALSE(error); });
+      ASSERT_EQ(2u, context.run());
+      connection = std::make_shared<connection_t>(
+        context, std::move(socket), shared, GetParam(),
+        epee::net_utils::ssl_support_t::e_ssl_support_disabled
+      );
+      ASSERT_TRUE(connection->start(false, true));
+      // Leave the io_context stopped so queued writes cannot drain. This makes
+      // queue boundaries independent of socket buffers and worker scheduling.
     }
 
-    void release_protocol()
-    {}
-
-    config_t &config;
-  };
-
-
-  using byte_slice_t = epee::byte_slice;
-  using connection_t = epee::net_utils::connection<handler_t>;
-  using shared_t = connection_t::shared_state;
-  using tcp_t = boost::asio::ip::tcp;
-  using endpoint_t = tcp_t::endpoint;
-  using socket_t = tcp_t::socket;
-  using acceptor_t = tcp_t::acceptor;
-
-  const endpoint_t endpoint{boost::asio::ip::make_address("127.0.0.1"), 5262};
-  boost::asio::io_context context{};
-  acceptor_t acceptor{context};
-  acceptor.open(endpoint.protocol());
-#if !defined(_WIN32)
-  acceptor.set_option(boost::asio::ip::tcp::acceptor::reuse_address(true));
-#endif
-  acceptor.bind(endpoint);
-  acceptor.listen();
+    bool send(byte_slice_t message)
+    {
+      return static_cast<epee::net_utils::i_service_endpoint&>(*connection).do_send(std::move(message));
+    }
 
-  socket_t in_socket{context};
+    void expect_failure(byte_slice_t message)
+    {
+      const auto start = std::chrono::steady_clock::now();
+      EXPECT_FALSE(send(std::move(message)));
+      EXPECT_LT(std::chrono::steady_clock::now() - start, std::chrono::seconds(1));
+      context.restart();
+      EXPECT_LE(1u, context.run_for(std::chrono::seconds(5)));
+      EXPECT_TRUE(context.stopped());
+      EXPECT_EQ(connection_t::WASTED, connection->get_status());
+      EXPECT_FALSE(send(byte_slice_t{"."}));
+    }
 
-  std::shared_ptr<connection_t> out_connection;
-  const auto shared = std::make_shared<shared_t>();
-  const auto make_connection = [&] {
-    in_socket = socket_t{context};
-    acceptor.async_accept(in_socket, [] (auto error) { EXPECT_TRUE(!error); });
-
-    socket_t out_socket{context};
-    out_socket.async_connect(endpoint, [] (auto error) { EXPECT_TRUE(!error); });
-
-    context.restart();
-    ASSERT_EQ(2u, context.run()); // connect and accept
-
-    out_connection = std::make_shared<connection_t>(
-      context,
-      std::move(out_socket),
-      shared,
-      epee::net_utils::e_connection_type_P2P,
-      epee::net_utils::ssl_support_t::e_ssl_support_disabled
-    );
-    EXPECT_TRUE(out_connection->start(false, true));
+    void TearDown() override
+    {
+      if (connection)
+        connection->cancel();
+      context.restart();
+      context.run();
+    }
   };
+}
 
-  make_connection();
-  {
-    const byte_slice_t payload{"."};
-    epee::net_utils::i_service_endpoint& out{*out_connection};
-    static_assert(ABSTRACT_SERVER_SEND_QUE_MAX_COUNT < std::numeric_limits<std::size_t>::max(), "");
-    for (std::size_t i = 0; i <= ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
-      EXPECT_TRUE(out.do_send(payload.clone()));
-    EXPECT_FALSE(out.do_send(payload.clone()));
-  }
-  context.restart();
-  EXPECT_LE(1u, context.run());
-  EXPECT_EQ(connection_t::WASTED, out_connection->get_status());
+TEST_P(send_queue_test, count_limit)
+{
+  const byte_slice_t payload{"."};
+  for (std::size_t i = 0; i <= ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
+    ASSERT_TRUE(send(payload.clone()));
+  expect_failure(payload.clone());
+}
+
+TEST_P(send_queue_test, byte_limit)
+{
+  shared->response_soft_limit = 1024;
+  ASSERT_TRUE(send(byte_slice_t{std::string(shared->response_soft_limit, '.')}));
+  ASSERT_TRUE(send(byte_slice_t{"."}));
+  expect_failure(byte_slice_t{"."});
+}
 
-  make_connection();
+TEST_P(send_queue_test, large_message)
+{
+  const byte_slice_t small{"."};
+  const byte_slice_t large{std::string(std::size_t(3 * 128 * 1024), '.')};
+  for (std::size_t i = 0; i < ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
+    ASSERT_TRUE(send(small.clone()));
+  if (GetParam() == epee::net_utils::e_connection_type_RPC)
   {
-    const byte_slice_t spayload{"."};
-    const byte_slice_t lpayload{std::string(std::size_t(3 * 128 * 1024), '.')};
-    epee::net_utils::i_service_endpoint& out{*out_connection};
-    for (std::size_t i = 0; i < ABSTRACT_SERVER_SEND_QUE_MAX_COUNT; ++i)
-      EXPECT_TRUE(out.do_send(spayload.clone()));
-    EXPECT_FALSE(out.do_send(lpayload.clone()));
+    // RPC responses remain one queue entry, regardless of their size.
+    ASSERT_TRUE(send(large.clone()));
+    expect_failure(small.clone());
   }
-  context.restart();
-  EXPECT_LE(1u, context.run());
-  EXPECT_EQ(connection_t::WASTED, out_connection->get_status());
+  else
+    expect_failure(large.clone()); // Chunking crosses the count limit mid-send.
 }
 
+INSTANTIATE_TEST_SUITE_P(boosted_tcp_server, send_queue_test, testing::Values(
+  epee::net_utils::e_connection_type_P2P,
+  epee::net_utils::e_connection_type_RPC,
+  epee::net_utils::e_connection_type_NET
+));
+
 
 TEST(boosted_tcp_server, slow_reader_is_not_dropped_mid_response)
 {

### tests/unit_tests/epee_http_server.cpp
@@ -119,10 +119,13 @@ TEST(http_server, response_soft_limit)
   req.set(http::field::user_agent, BOOST_BEAST_VERSION_STRING);
   req.body() = make_payload();
   req.prepare_payload();
-  http::write(stream, req, error);
-  EXPECT_FALSE(bool(error));
-
+  // Reading a complete oversized response must allow another on the same
+  // keep-alive connection, even with only one worker available.
+  for (unsigned i = 0; i < 4; ++i)
   {
+    http::write(stream, req, error);
+    ASSERT_FALSE(bool(error));
+
     dummy::response payload{};
     boost::beast::flat_buffer buffer;
     http::response_parser<http::basic_string_body<char>> parser;
@@ -141,6 +144,82 @@ TEST(http_server, response_soft_limit)
   server.send_stop_signal();
 }
 
+TEST(http_server, send_failure_stops_requests)
+{
+  namespace http = epee::net_utils::http;
+
+  struct endpoint_t final : epee::net_utils::i_service_endpoint
+  {
+    bool do_send(epee::byte_slice) override
+    {
+      ++send_count;
+      return allow_send;
+    }
+    bool send_done() override { ++done_count; return true; }
+    bool close(bool) override { return true; }
+    bool call_run_once_service_io() override { return true; }
+    bool request_callback() override { return true; }
+    boost::asio::io_context& get_io_context() override { return context; }
+
+    boost::asio::io_context context;
+    bool allow_send = true;
+    std::size_t send_count = 0;
+    std::size_t done_count = 0;
+  };
+
+  struct handler_t final : http::simple_http_connection_handler<>
+  {
+    using http::simple_http_connection_handler<>::simple_http_connection_handler;
+
+    bool handle_request(const http::http_request_info&, http::http_response_info& response) override
+    {
+      ++request_count;
+      response.m_response_code = 200;
+      response.m_response_comment = "OK";
+      return true;
+    }
+
+    std::size_t request_count = 0;
+  };
+
+  // Exercise all three response paths: absent, empty, and nonempty bodies.
+  const std::string requests[] = {
+    "GET / HTTP/1.1\r\nHost: localhost\r\n\r\n",
+    "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 0\r\n\r\n",
+    "POST / HTTP/1.1\r\nHost: localhost\r\nContent-Length: 1\r\n\r\nx"
+  };
+  for (const auto& request : requests)
+  {
+    for (const auto& input : {request, request + request})
+    {
+      for (const std::size_t successful_requests : {0u, 1u})
+      {
+        SCOPED_TRACE(input);
+        SCOPED_TRACE(successful_requests);
+        endpoint_t endpoint;
+        http::http_server_config config;
+        epee::net_utils::connection_context_base context;
+        handler_t handler{&endpoint, config, context};
+
+        for (std::size_t i = 0; i < successful_requests; ++i)
+          ASSERT_TRUE(handler.handle_recv(request.data(), request.size()));
+
+        endpoint.allow_send = false;
+        // Check both an exhausted input buffer and another pipelined request.
+        EXPECT_FALSE(handler.handle_recv(input.data(), input.size()));
+        EXPECT_EQ(successful_requests + 1, handler.request_count);
+        EXPECT_EQ(successful_requests + 1, endpoint.send_count);
+        EXPECT_EQ(successful_requests, endpoint.done_count);
+
+        // A failed response must leave the parser in its terminal error state.
+        EXPECT_FALSE(handler.handle_recv(request.data(), request.size()));
+        EXPECT_EQ(successful_requests + 1, handler.request_count);
+        EXPECT_EQ(successful_requests + 1, endpoint.send_count);
+      }
+    }
+  }
+}
+
 TEST(http_server, private_ip_limit)
 {
   namespace http = boost::beast::http;