HTTPServer: compose and send replies to connected clients

Sockets-touching bits copied and adapted from `CConnman::SocketSendData()`

Testing this requires adding a new feature to the SocketTestingSetup,
returning the DynSock I/O pipes from the mock socket so the received
data can be checked.

Co-authored-by: Vasil Dimov <vd@FreeBSD.org>

src/httpserver.cpp

@@ -14,6 +14,7 @@
 #include <netbase.h>
 #include <node/interface_ui.h>
 #include <rpc/protocol.h>
+#include <span.h>
 #include <sync.h>
 #include <util/check.h>
 #include <util/signalinterrupt.h>
@@ -967,6 +968,101 @@ bool HTTPRequest::LoadBody(LineReader& reader)
     }
 }
 
+void HTTPRequest::WriteReply(HTTPStatusCode status, std::span<const std::byte> reply_body)
+{
+    HTTPResponse res;
+
+    // Some response headers are determined in advance and stored in the request
+    res.m_headers = std::move(m_response_headers);
+
+    // Response version matches request version
+    res.m_version = m_version;
+
+    // Add response code
+    res.m_status = status;
+
+    // See libevent evhttp_response_needs_body()
+    // Response headers are different if no body is needed
+    bool needs_body{status != HTTP_NO_CONTENT && (status < 100 || status >= 200)};
+    bool needs_content_length{false};
+
+    // Keep track of this, will be used in a future commit to disconnect some clients
+    [[maybe_unused]] bool keep_alive{false};
+
+    // See libevent evhttp_make_header_response()
+    // Expected response headers depend on protocol version
+    if (m_version.major == 1) {
+        // HTTP/1.0
+        if (m_version.minor == 0) {
+            auto connection_header{m_headers.FindFirst("Connection")};
+            if (connection_header && ToLower(connection_header.value()) == "keep-alive") {
+                res.m_headers.Write("Connection", "keep-alive");
+                keep_alive = true;
+                // HTTP/1.0 connections are closed by default so EOF is sufficient
+                // to indicate end of the body. Adding Content-Length a special case.
+                if (needs_body) needs_content_length = true;
+            }
+        }
+
+        // HTTP/1.1
+        if (m_version.minor >= 1) {
+            const int64_t now_seconds{TicksSinceEpoch<std::chrono::seconds>(NodeClock::now())};
+            res.m_headers.Write("Date", FormatRFC1123DateTime(now_seconds));
+
+            // HTTP/1.1 connections are kept alive by default and always require Content-Length.
+            if (needs_body) needs_content_length = true;
+
+            // Default for HTTP/1.1
+            keep_alive = true;
+        }
+    }
+
+    if (needs_content_length) {
+        res.m_headers.Write("Content-Length", util::ToString(reply_body.size()));
+    }
+
+    if (needs_body && !res.m_headers.FindFirst("Content-Type")) {
+        // Default type from libevent evhttp_new_object()
+        res.m_headers.Write("Content-Type", "text/html; charset=ISO-8859-1");
+    }
+
+    auto connection_header{m_headers.FindFirst("Connection")};
+    if (connection_header && ToLower(connection_header.value()) == "close") {
+        // Might not exist already but we need to replace it, not append to it
+        res.m_headers.RemoveAll("Connection");
+
+        res.m_headers.Write("Connection", "close");
+        keep_alive = false;
+    }
+
+    // Serialize the response headers
+    const std::string headers{res.StringifyHeaders()};
+    const auto headers_bytes{std::as_bytes(std::span{headers})};
+
+    // Fill the send buffer with the complete serialized response headers + body
+    {
+        LOCK(m_client->m_send_mutex);
+        m_client->m_send_buffer.insert(m_client->m_send_buffer.end(), headers_bytes.begin(), headers_bytes.end());
+
+        // We've been using std::span up until now but it is finally time to copy
+        // data. The original data will go out of scope when WriteReply() returns.
+        // This is analogous to the memcpy() in libevent's evbuffer_add()
+        m_client->m_send_buffer.insert(m_client->m_send_buffer.end(), reply_body.begin(), reply_body.end());
+    }
+
+    // Inform HTTPServer I/O loop that there is data that is ready to be sent to
+    // this client in the next loop iteration.
+    m_client->m_send_ready = true;
+
+    LogDebug(
+        BCLog::HTTP,
+        "HTTPResponse (status code: %d size: %lld) added to send buffer for client %s (id=%llu)",
+        status,
+        headers_bytes.size() + reply_body.size(),
+        m_client->m_origin,
+        m_client->m_id);
+}
+
 util::Expected<void, std::string> HTTPServer::BindAndStartListening(const CService& to)
 {
     // Create socket for listening for incoming connections
@@ -1156,7 +1252,12 @@ void HTTPServer::SocketHandlerConnected(const IOReadiness& io_readiness) const
         bool err_ready = events.occurred & Sock::ERR;
 
         if (send_ready) {
-            // TODO: send data
+            // Try to send as much data as is ready for this client.
+            // If there's an error we can skip the receive phase for this client
+            // because we need to disconnect.
+            if (!client->MaybeSendBytesFromBuffer()) {
+                recv_ready = false;
+            }
         }
 
         if (recv_ready || err_ready) {
@@ -1225,15 +1326,12 @@ HTTPServer::IOReadiness HTTPServer::GenerateWaitSockets() const
     }
 
     for (const auto& http_client : m_connected) {
-        // TODO: Implement HTTPRemoteClient methods for send/receive readiness
-        const bool select_recv{true};
-        const bool select_send{true};
-        if (!select_recv && !select_send) continue;
-
         // Safely copy the shared pointer to the socket
         std::shared_ptr<Sock> sock{WITH_LOCK(http_client->m_sock_mutex, return http_client->m_sock;)};
 
-        Sock::Event event = (select_send ? Sock::SEND : 0) | (select_recv ? Sock::RECV : 0);
+        // Check if client is ready to send data. Don't try to receive again
+        // until the send buffer is cleared (all data sent to client).
+        Sock::Event event = (http_client->m_send_ready ? Sock::SEND : Sock::RECV);
         io_readiness.events_per_sock.emplace(sock, Sock::Events{event});
         io_readiness.httpclients_per_sock.emplace(sock, http_client);
     }
@@ -1318,4 +1416,75 @@ bool HTTPRemoteClient::ReadRequest(HTTPRequest& req)
 
     return true;
 }
+
+bool HTTPRemoteClient::MaybeSendBytesFromBuffer()
+{
+    // Send as much data from this client's buffer as we can
+    LOCK(m_send_mutex);
+    if (!m_send_buffer.empty()) {
+        // Socket flags (See kernel docs for send(2) and tcp(7) for more details).
+        // MSG_NOSIGNAL: If the remote end of the connection is closed,
+        //               fail with EPIPE (an error) as opposed to triggering
+        //               SIGPIPE which terminates the process.
+        // MSG_DONTWAIT: Makes the send operation non-blocking regardless of socket blocking mode.
+        // MSG_MORE:     We do not set this flag here because http responses are usually
+        //               small and we want the kernel to send them right away. Setting MSG_MORE
+        //               would "cork" the socket to prevent sending out partial frames.
+        int flags{MSG_NOSIGNAL | MSG_DONTWAIT};
+
+        // Try to send bytes through socket
+        ssize_t bytes_sent;
+        {
+            LOCK(m_sock_mutex);
+            bytes_sent = m_sock->Send(m_send_buffer.data(),
+                                      m_send_buffer.size(),
+                                      flags);
+        }
+
+        if (bytes_sent < 0) {
+            // Something went wrong
+            const int err{WSAGetLastError()};
+            if (!IOErrorIsPermanent(err)) {
+                // The error can be safely ignored, try the send again on the next I/O loop.
+                m_send_ready = true;
+                return true;
+            } else {
+                // Unrecoverable error, log and disconnect client.
+                LogDebug(
+                    BCLog::HTTP,
+                    "Error sending HTTP response data to client %s (id=%llu): %s",
+                    m_origin,
+                    m_id,
+                    NetworkErrorString(err));
+                // TODO: disconnect
+                // Do not attempt to read from this client.
+                return false;
+            }
+        }
+
+        // Successful send, remove sent bytes from our local buffer.
+        Assume(static_cast<size_t>(bytes_sent) <= m_send_buffer.size());
+        m_send_buffer.erase(m_send_buffer.begin(),
+                            m_send_buffer.begin() + bytes_sent);
+
+        LogDebug(
+            BCLog::HTTP,
+            "Sent %d bytes to client %s (id=%llu)",
+            bytes_sent,
+            m_origin,
+            m_id);
+
+        // This check is inside the if(!empty) block meaning "there was data but now its gone".
+        // We wouldn't want to change the flags if MaybeSendBytesFromBuffer() was called
+        // on an already-empty m_send_buffer because the connection might have just been opened.
+        if (m_send_buffer.empty()) {
+            m_send_ready = false;
+        } else {
+            // The send buffer isn't flushed yet, try to push more on the next loop.
+            m_send_ready = true;
+        }
+    }
+
+    return true;
+}
 } // namespace http_bitcoin

src/httpserver.h

@@ -295,6 +295,9 @@ public:
     //! Pointer to the client that made the request so we know who to respond to.
     std::shared_ptr<HTTPRemoteClient> m_client;
 
+    //! Response headers may be set in advance before response body is known
+    HTTPHeaders m_response_headers;
+
     explicit HTTPRequest(std::shared_ptr<HTTPRemoteClient> client) : m_client{std::move(client)} {}
     //! Construct with a null client for unit tests
     explicit HTTPRequest() : m_client{} {}
@@ -312,6 +315,12 @@ public:
     bool LoadHeaders(LineReader& reader);
     bool LoadBody(LineReader& reader);
     /// @}
+
+    void WriteReply(HTTPStatusCode status, std::span<const std::byte> reply_body = {});
+    void WriteReply(HTTPStatusCode status, std::string_view reply_body_view)
+    {
+        WriteReply(status, std::as_bytes(std::span{reply_body_view}));
+    }
 };
 
 class HTTPServer
@@ -518,6 +527,22 @@ public:
      */
     std::vector<std::byte> m_recv_buffer{};
 
+    /**
+     * Response data destined for this client.
+     * Written to by http worker threads, read and erased by HTTPServer I/O thread
+     */
+    /// @{
+    Mutex m_send_mutex;
+    std::vector<std::byte> m_send_buffer GUARDED_BY(m_send_mutex);
+    /// @}
+
+    /**
+     * Set true by worker threads after writing a response to m_send_buffer.
+     * Set false by the HTTPServer I/O thread after flushing m_send_buffer.
+     * Checked in the HTTPServer I/O loop to avoid locking m_send_mutex if there's nothing to send.
+     */
+    std::atomic_bool m_send_ready{false};
+
     /**
      * Mutex that serializes the Send() and Recv() calls on `m_sock`. Reading
      * from the client occurs in the I/O thread but writing back to a client
@@ -547,6 +572,12 @@ public:
      * @returns true upon reading a complete request, otherwise false (may throw).
      */
     bool ReadRequest(HTTPRequest& req);
+
+    /**
+     * Push data (if there is any) from client's m_send_buffer to the connected socket.
+     * @returns false if we are done with this client and HTTPServer can skip the next read operation from it.
+     */
+    bool MaybeSendBytesFromBuffer() EXCLUSIVE_LOCKS_REQUIRED(!m_send_mutex, !m_sock_mutex);
 };
 } // namespace http_bitcoin
 

src/test/httpserver_tests.cpp

@@ -489,6 +489,10 @@ BOOST_AUTO_TEST_CASE(http_request_tests)
 
 BOOST_AUTO_TEST_CASE(http_server_socket_tests)
 {
+    // Hard code the timestamp for the Date header in the HTTP response
+    // Wed Dec 11 00:47:09 2024 UTC
+    SetMockTime(1733878029);
+
     // Prepare a request handler that just stores received requests so we can examine them.
     // Mutex is required to prevent a race between this test's main thread and the server's I/O loop.
     Mutex requests_mutex;
@@ -524,8 +528,10 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
     // No connections yet
     BOOST_CHECK_EQUAL(server.GetConnectionsCount(), 0);
 
-    // Create a mock client with pre-loaded request data and add it to the local CreateSock queue
-    ConnectClient(std::as_bytes(std::span(full_request)));
+    // Create a mock client with pre-loaded request data and add it to the local CreateSock queue.
+    // Keep a handle for the mock client's send and receive pipes so we can examine
+    // the data it "receives".
+    std::shared_ptr<DynSock::Pipes> mock_client_socket_pipes{ConnectClient(std::as_bytes(std::span(full_request)))};
 
     // Wait up to a minute to find and connect the client in the I/O loop
     int attempts{6000};
@@ -553,6 +559,9 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
                 client = requests.front()->m_client;
                 BOOST_CHECK_EQUAL(client->m_origin, "5.5.5.5:6789");
 
+                // Respond to request
+                requests.front()->WriteReply(HTTP_OK, "874140\n");
+
                 break;
             }
         }
@@ -560,6 +569,31 @@ BOOST_AUTO_TEST_CASE(http_server_socket_tests)
         BOOST_REQUIRE(--attempts > 0);
     }
 
+    // Check the sent response from the mock client at the other end of the mock socket
+    std::string actual;
+    // Wait up to one minute for all the bytes to appear in the "send" pipe.
+    char buf[0x10000] = {};
+    attempts = 6000;
+    while (attempts > 0)
+    {
+        ssize_t bytes_read = mock_client_socket_pipes->send.GetBytes(buf, sizeof(buf), 0);
+        if (bytes_read > 0) {
+            actual.append(buf, bytes_read);
+            if (actual.length() == 146) {
+                break;
+            }
+        }
+        std::this_thread::sleep_for(10ms);
+        --attempts;
+    }
+    BOOST_CHECK(actual.starts_with("HTTP/1.1 200 OK\r\n"));
+    BOOST_CHECK(actual.ends_with("\r\n874140\n"));
+    // Headers can be sorted in any order, and will be, since we use unordered_map
+    BOOST_CHECK(actual.find("Connection: close\r\n") != std::string::npos);
+    BOOST_CHECK(actual.find("Content-Length: 7\r\n") != std::string::npos);
+    BOOST_CHECK(actual.find("Content-Type: text/html; charset=ISO-8859-1\r\n") != std::string::npos);
+    BOOST_CHECK(actual.find("Date: Wed, 11 Dec 2024 00:47:09 GMT\r\n") != std::string::npos);
+
     // Stop the I/O thread before modifying m_connected. CloseConnection() is a
     // temporary test-only API that is not thread-safe against GenerateWaitSockets(),
     // which iterates m_connected on the I/O thread.

src/test/util/setup_common.cpp

@@ -658,7 +658,7 @@ SocketTestingSetup::~SocketTestingSetup()
     CreateSock = m_create_sock_orig;
 }
 
-void SocketTestingSetup::ConnectClient(std::span<const std::byte> data)
+std::shared_ptr<DynSock::Pipes> SocketTestingSetup::ConnectClient(std::span<const std::byte> data)
 {
     // I/O pipes for a mock Connected Socket we can read and write to.
     auto connected_socket_pipes(std::make_shared<DynSock::Pipes>());
@@ -672,6 +672,8 @@ void SocketTestingSetup::ConnectClient(std::span<const std::byte> data)
 
     // Push into the queue of Accepted Sockets returned by the local CreateSock()
     m_accepted_sockets.Push(std::move(connected_socket));
+
+    return connected_socket_pipes;
 }
 
 /**

src/test/util/setup_common.h

@@ -256,8 +256,11 @@ public:
     explicit SocketTestingSetup();
     ~SocketTestingSetup();
 
-    //! Connect to the socket with a mock client (a DynSock) and send pre-loaded data.
-    void ConnectClient(std::span<const std::byte> data);
+    /**
+     * Connect to the socket with a mock client (a DynSock) and send pre-loaded data.
+     * Returns the I/O pipes from the mock client so we can read response data sent to it.
+     */
+    std::shared_ptr<DynSock::Pipes> ConnectClient(std::span<const std::byte> data);
 
 private:
     //! Save the original value of CreateSock here and restore it when the test ends.