fix: Fix bugs in new webserver (#1780)

Fixes #919. Fixes bugs for new webserver: - Unhandled exception when closing already closed websocket - No pings for plain websocket connection - Server drops websocket connection when client responds to pings but doesn't send anything Also changing API of ng connections. Now timeout is set by a separate method instead of providing it for each call.
2025-12-06 17:27:58 +00:00 · 2024-12-19 15:14:04 +00:00
parent c4b87d2a0a
commit 7d4e3619b0
13 changed files with 238 additions and 204 deletions
--- a/src/web/ng/Connection.hpp
+++ b/src/web/ng/Connection.hpp
@@ -104,8 +104,9 @@ protected:
 public:
    /**
     * @brief The default timeout for send, receive, and close operations.
+     * @note This value should be higher than forwarding timeout to not disconnect clients if rippled is slow.
     */
-    static constexpr std::chrono::steady_clock::duration DEFAULT_TIMEOUT = std::chrono::seconds{30};
+    static constexpr std::chrono::steady_clock::duration DEFAULT_TIMEOUT = std::chrono::seconds{11};

    /**
     * @brief Construct a new Connection object
@@ -116,39 +117,41 @@ public:
     */
    Connection(std::string ip, boost::beast::flat_buffer buffer, util::TagDecoratorFactory const& tagDecoratorFactory);

+    /**
+     * @brief Get the timeout for send, receive, and close operations. For WebSocket connections, this is the ping
+     * interval.
+     *
+     * @param newTimeout The new timeout to set.
+     */
+    virtual void
+    setTimeout(std::chrono::steady_clock::duration newTimeout) = 0;
+
    /**
     * @brief Send a response to the client.
     *
     * @param response The response to send.
     * @param yield The yield context.
-     * @param timeout The timeout for the operation.
     * @return An error if the operation failed or nullopt if it succeeded.
     */
    virtual std::optional<Error>
-    send(
-        Response response,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
-    ) = 0;
+    send(Response response, boost::asio::yield_context yield) = 0;

    /**
     * @brief Receive a request from the client.
     *
     * @param yield The yield context.
-     * @param timeout The timeout for the operation.
     * @return The request if it was received or an error if the operation failed.
     */
    virtual std::expected<Request, Error>
-    receive(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) = 0;
+    receive(boost::asio::yield_context yield) = 0;

    /**
     * @brief Gracefully close the connection.
     *
     * @param yield The yield context.
-     * @param timeout The timeout for the operation.
     */
    virtual void
-    close(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) = 0;
+    close(boost::asio::yield_context yield) = 0;
 };

 /**
--- a/src/web/ng/Server.cpp
+++ b/src/web/ng/Server.cpp
@@ -223,6 +223,7 @@ Server::onWs(MessageHandler handler)
 std::optional<std::string>
 Server::run()
 {
+    LOG(log_.info()) << "Starting ng::Server";
    auto acceptor = makeAcceptor(ctx_.get(), endpoint_);
    if (not acceptor.has_value())
        return std::move(acceptor).error();
@@ -236,6 +237,7 @@ Server::run()
                boost::asio::ip::tcp::socket socket{ctx_.get().get_executor()};

                acceptor.async_accept(socket, yield[errorCode]);
+                LOG(log_.trace()) << "Accepted a new connection";
                if (errorCode) {
                    LOG(log_.debug()) << "Error accepting a connection: " << errorCode.what();
                    continue;
@@ -290,6 +292,7 @@ Server::handleConnection(boost::asio::ip::tcp::socket socket, boost::asio::yield
        }
        return;
    }
+    LOG(log_.trace()) << connectionExpected.value()->tag() << "Connection created";

    boost::asio::spawn(
        ctx_.get(),
--- a/src/web/ng/impl/ConnectionHandler.cpp
+++ b/src/web/ng/impl/ConnectionHandler.cpp
@@ -154,6 +154,7 @@ ConnectionHandler::processConnection(ConnectionPtr connectionPtr, boost::asio::y
            yield,
            [this](Error const& e, Connection const& c) { return handleError(e, c); }
        );
+        LOG(log_.trace()) << connectionRef.tag() << "Created SubscriptionContext for the connection";
    }
    SubscriptionContextPtr subscriptionContextInterfacePtr = subscriptionContext;

@@ -166,14 +167,21 @@ ConnectionHandler::processConnection(ConnectionPtr connectionPtr, boost::asio::y
            break;
    }

-    if (subscriptionContext != nullptr)
+    if (subscriptionContext != nullptr) {
        subscriptionContext->disconnect(yield);
+        LOG(log_.trace()) << connectionRef.tag() << "SubscriptionContext disconnected";
+    }

-    if (shouldCloseGracefully)
+    if (shouldCloseGracefully) {
        connectionRef.close(yield);
+        LOG(log_.trace()) << connectionRef.tag() << "Closed gracefully";
+    }

    signalConnection.disconnect();
+    LOG(log_.trace()) << connectionRef.tag() << "Signal disconnected";
+
    onDisconnectHook_(connectionRef);
+    LOG(log_.trace()) << connectionRef.tag() << "Processing finished";
 }

 void
@@ -185,6 +193,7 @@ ConnectionHandler::stop()
 bool
 ConnectionHandler::handleError(Error const& error, Connection const& connection) const
 {
+    LOG(log_.trace()) << connection.tag() << "Got error: " << error << " " << error.message();
    // ssl::error::stream_truncated, also known as an SSL "short read",
    // indicates the peer closed the connection without performing the
    // required closing handshake (for example, Google does this to
@@ -201,7 +210,8 @@ ConnectionHandler::handleError(Error const& error, Connection const& connection)
    // Beast returns the error boost::beast::http::error::partial_message.
    // Therefore, if we see a short read here, it has occurred
    // after the message has been completed, so it is safe to ignore it.
-    if (error == boost::beast::http::error::end_of_stream || error == boost::asio::ssl::error::stream_truncated)
+    if (error == boost::beast::http::error::end_of_stream || error == boost::asio::ssl::error::stream_truncated ||
+        error == boost::asio::error::eof)
        return false;

    // WebSocket connection was gracefully closed
@@ -229,6 +239,7 @@ ConnectionHandler::sequentRequestResponseLoop(
    //   an error appears.
    // - When server is shutting down it will cancel all operations on the connection so an error appears.

+    LOG(log_.trace()) << connection.tag() << "Processing sequentially";
    while (true) {
        auto expectedRequest = connection.receive(yield);
        if (not expectedRequest)
@@ -250,12 +261,14 @@ ConnectionHandler::parallelRequestResponseLoop(
    boost::asio::yield_context yield
 )
 {
+    LOG(log_.trace()) << connection.tag() << "Processing in parallel";
    // atomic_bool is not needed here because everything happening on coroutine's strand
    bool stop = false;
    bool closeConnectionGracefully = true;
    util::CoroutineGroup tasksGroup{yield, maxParallelRequests_};

    while (not stop) {
+        LOG(log_.trace()) << connection.tag() << "Receiving request";
        auto expectedRequest = connection.receive(yield);
        if (not expectedRequest) {
            auto const closeGracefully = handleError(expectedRequest.error(), connection);
@@ -282,7 +295,9 @@ ConnectionHandler::parallelRequestResponseLoop(
                }
            );
            ASSERT(spawnSuccess, "The coroutine was expected to be spawned");
+            LOG(log_.trace()) << connection.tag() << "Spawned a coroutine to process request";
        } else {
+            LOG(log_.trace()) << connection.tag() << "Too many requests from one connection, rejecting the request";
            connection.send(
                Response{
                    boost::beast::http::status::too_many_requests,
@@ -305,8 +320,10 @@ ConnectionHandler::processRequest(
    boost::asio::yield_context yield
 )
 {
+    LOG(log_.trace()) << connection.tag() << "Processing request: " << request.message();
    auto response = handleRequest(connection, subscriptionContext, request, yield);

+    LOG(log_.trace()) << connection.tag() << "Sending response: " << response.message();
    auto const maybeError = connection.send(std::move(response), yield);
    if (maybeError.has_value()) {
        return handleError(maybeError.value(), connection);
--- a/src/web/ng/impl/HttpConnection.hpp
+++ b/src/web/ng/impl/HttpConnection.hpp
@@ -54,10 +54,7 @@ public:
    using Connection::Connection;

    virtual std::expected<bool, Error>
-    isUpgradeRequested(
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
-    ) = 0;
+    isUpgradeRequested(boost::asio::yield_context yield) = 0;

    virtual std::expected<ConnectionPtr, Error>
    upgrade(
@@ -69,8 +66,7 @@ public:
    virtual std::optional<Error>
    sendRaw(
        boost::beast::http::response<boost::beast::http::string_body> response,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
+        boost::asio::yield_context yield
    ) = 0;
 };

@@ -80,6 +76,7 @@ template <typename StreamType>
 class HttpConnection : public UpgradableConnection {
    StreamType stream_;
    std::optional<boost::beast::http::request<boost::beast::http::string_body>> request_;
+    std::chrono::steady_clock::duration timeout_{DEFAULT_TIMEOUT};

 public:
    HttpConnection(
@@ -113,40 +110,39 @@ public:
    }

    std::optional<Error>
-    sendRaw(
-        boost::beast::http::response<boost::beast::http::string_body> response,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
-    ) override
+    sendRaw(boost::beast::http::response<boost::beast::http::string_body> response, boost::asio::yield_context yield)
+        override
    {
        boost::system::error_code error;
-        boost::beast::get_lowest_layer(stream_).expires_after(timeout);
+        boost::beast::get_lowest_layer(stream_).expires_after(timeout_);
        boost::beast::http::async_write(stream_, response, yield[error]);
        if (error)
            return error;
        return std::nullopt;
    }

+    void
+    setTimeout(std::chrono::steady_clock::duration newTimeout) override
+    {
+        timeout_ = newTimeout;
+    }
+
    std::optional<Error>
-    send(
-        Response response,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
-    ) override
+    send(Response response, boost::asio::yield_context yield) override
    {
        auto httpResponse = std::move(response).intoHttpResponse();
-        return sendRaw(std::move(httpResponse), yield, timeout);
+        return sendRaw(std::move(httpResponse), yield);
    }

    std::expected<Request, Error>
-    receive(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) override
+    receive(boost::asio::yield_context yield) override
    {
        if (request_.has_value()) {
            Request result{std::move(request_).value()};
            request_.reset();
            return result;
        }
-        auto expectedRequest = fetch(yield, timeout);
+        auto expectedRequest = fetch(yield);
        if (expectedRequest.has_value())
            return Request{std::move(expectedRequest).value()};

@@ -154,27 +150,22 @@ public:
    }

    void
-    close(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) override
+    close(boost::asio::yield_context yield) override
    {
        [[maybe_unused]] boost::system::error_code error;
        if constexpr (IsSslTcpStream<StreamType>) {
-            boost::beast::get_lowest_layer(stream_).expires_after(timeout);
-            stream_.async_shutdown(yield[error]);
-        }
-        if constexpr (IsTcpStream<StreamType>) {
-            stream_.socket().shutdown(boost::asio::ip::tcp::socket::shutdown_type::shutdown_both, error);
-        } else {
-            boost::beast::get_lowest_layer(stream_).socket().shutdown(
-                boost::asio::ip::tcp::socket::shutdown_type::shutdown_both, error
-            );
+            boost::beast::get_lowest_layer(stream_).expires_after(timeout_);
+            stream_.async_shutdown(yield[error]);  // Close the SSL connection gracefully
        }
+        boost::beast::get_lowest_layer(stream_).socket().shutdown(
+            boost::asio::ip::tcp::socket::shutdown_type::shutdown_both, error
+        );
    }

    std::expected<bool, Error>
-    isUpgradeRequested(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT)
-        override
+    isUpgradeRequested(boost::asio::yield_context yield) override
    {
-        auto expectedRequest = fetch(yield, timeout);
+        auto expectedRequest = fetch(yield);
        if (not expectedRequest.has_value())
            return std::unexpected{std::move(expectedRequest).error()};

@@ -217,11 +208,11 @@ public:

 private:
    std::expected<boost::beast::http::request<boost::beast::http::string_body>, Error>
-    fetch(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout)
+    fetch(boost::asio::yield_context yield)
    {
        boost::beast::http::request<boost::beast::http::string_body> request{};
        boost::system::error_code error;
-        boost::beast::get_lowest_layer(stream_).expires_after(timeout);
+        boost::beast::get_lowest_layer(stream_).expires_after(timeout_);
        boost::beast::http::async_read(stream_, buffer_, request, yield[error]);
        if (error)
            return std::unexpected{error};
--- a/src/web/ng/impl/WsConnection.hpp
+++ b/src/web/ng/impl/WsConnection.hpp
@@ -20,7 +20,6 @@
 #pragma once

 #include "util/Taggable.hpp"
-#include "util/WithTimeout.hpp"
 #include "util/build/Build.hpp"
 #include "web/ng/Connection.hpp"
 #include "web/ng/Error.hpp"
@@ -58,11 +57,7 @@ public:
    using Connection::Connection;

    virtual std::optional<Error>
-    sendBuffer(
-        boost::asio::const_buffer buffer,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = Connection::DEFAULT_TIMEOUT
-    ) = 0;
+    sendBuffer(boost::asio::const_buffer buffer, boost::asio::yield_context yield) = 0;
 };

 template <typename StreamType>
@@ -83,6 +78,7 @@ public:
        , stream_(std::move(socket))
        , initialRequest_(std::move(initialRequest))
    {
+        setupWsStream();
    }

    WsConnection(
@@ -98,14 +94,7 @@ public:
        , stream_(std::move(socket), sslContext)
        , initialRequest_(std::move(initialRequest))
    {
-        // Disable the timeout. The websocket::stream uses its own timeout settings.
-        boost::beast::get_lowest_layer(stream_).expires_never();
-        stream_.set_option(boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server));
-        stream_.set_option(
-            boost::beast::websocket::stream_base::decorator([](boost::beast::websocket::response_type& res) {
-                res.set(boost::beast::http::field::server, util::build::getClioFullVersionString());
-            })
-        );
+        setupWsStream();
    }

    std::optional<Error>
@@ -125,33 +114,39 @@ public:
    }

    std::optional<Error>
-    sendBuffer(
-        boost::asio::const_buffer buffer,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = Connection::DEFAULT_TIMEOUT
-    ) override
+    sendBuffer(boost::asio::const_buffer buffer, boost::asio::yield_context yield) override
    {
-        auto error =
-            util::withTimeout([this, buffer](auto&& yield) { stream_.async_write(buffer, yield); }, yield, timeout);
+        boost::beast::websocket::stream_base::timeout timeoutOption{};
+        stream_.get_option(timeoutOption);
+
+        boost::system::error_code error;
+        stream_.async_write(buffer, yield[error]);
        if (error)
            return error;
        return std::nullopt;
    }

-    std::optional<Error>
-    send(
-        Response response,
-        boost::asio::yield_context yield,
-        std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT
-    ) override
+    void
+    setTimeout(std::chrono::steady_clock::duration newTimeout) override
    {
-        return sendBuffer(response.asWsResponse(), yield, timeout);
+        boost::beast::websocket::stream_base::timeout wsTimeout =
+            boost::beast::websocket::stream_base::timeout::suggested(boost::beast::role_type::server);
+        wsTimeout.idle_timeout = newTimeout;
+        wsTimeout.handshake_timeout = newTimeout;
+        stream_.set_option(wsTimeout);
+    }
+
+    std::optional<Error>
+    send(Response response, boost::asio::yield_context yield) override
+    {
+        return sendBuffer(response.asWsResponse(), yield);
    }

    std::expected<Request, Error>
-    receive(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) override
+    receive(boost::asio::yield_context yield) override
    {
-        auto error = util::withTimeout([this](auto&& yield) { stream_.async_read(buffer_, yield); }, yield, timeout);
+        Error error;
+        stream_.async_read(buffer_, yield[error]);
        if (error)
            return std::unexpected{error};

@@ -162,14 +157,24 @@ public:
    }

    void
-    close(boost::asio::yield_context yield, std::chrono::steady_clock::duration timeout = DEFAULT_TIMEOUT) override
+    close(boost::asio::yield_context yield) override
    {
-        boost::beast::websocket::stream_base::timeout wsTimeout{};
-        stream_.get_option(wsTimeout);
-        wsTimeout.handshake_timeout = timeout;
-        stream_.set_option(wsTimeout);
+        boost::system::error_code error;  // unused
+        stream_.async_close(boost::beast::websocket::close_code::normal, yield[error]);
+    }

-        stream_.async_close(boost::beast::websocket::close_code::normal, yield);
+private:
+    void
+    setupWsStream()
+    {
+        // Disable the timeout. The websocket::stream uses its own timeout settings.
+        boost::beast::get_lowest_layer(stream_).expires_never();
+        setTimeout(DEFAULT_TIMEOUT);
+        stream_.set_option(
+            boost::beast::websocket::stream_base::decorator([](boost::beast::websocket::response_type& res) {
+                res.set(boost::beast::http::field::server, util::build::getClioFullVersionString());
+            })
+        );
    }
 };