diff --git a/.devcontainer/Dockerfile b/.devcontainer/Dockerfile index 3ec2018..805cafe 100644 --- a/.devcontainer/Dockerfile +++ b/.devcontainer/Dockerfile @@ -4,6 +4,8 @@ # FROM docker.io/psedoc/anyhttp:0.28 +# ================================================================================================== + # # install some more interactive utils in the devcontainer # @@ -12,6 +14,8 @@ RUN apt-get update && export DEBIAN_FRONTEND=noninteractive && \ apt-get clean && \ rm -rf /var/lib/apt/lists/* +# ================================================================================================== + # # TEST: claude code # @@ -24,6 +28,8 @@ RUN apt-get update && export DEBIAN_FRONTEND=noninteractive && \ # Install Claude Code # RUN npm install -g @anthropic-ai/claude-code +# ================================================================================================== + # # enable persistent bash history in container # diff --git a/.devcontainer/base/Dockerfile b/.devcontainer/base/Dockerfile index 3b285c8..b0bbc51 100644 --- a/.devcontainer/base/Dockerfile +++ b/.devcontainer/base/Dockerfile @@ -120,7 +120,7 @@ RUN git clone --recursive --depth 1 --branch ${NGTCP2_VERSION} https://github.co # ================================================================================================== -RUN git clone --depth 1 -b v1.6.3 https://github.com/libbpf/libbpf && \ +RUN git clone --depth 1 -b v1.7.0 https://github.com/libbpf/libbpf && \ cd libbpf && \ PREFIX=/usr/local LIBDIR=/usr/local/lib make -j$(nproc) -C src install && \ cd .. && rm -rf libbpf @@ -128,6 +128,8 @@ RUN git clone --depth 1 -b v1.6.3 https://github.com/libbpf/libbpf && \ # # nghttp2 with HTTP/3 support (based on OpenSSL) # +# --enable-debug (very noisy) +# ARG NGHTTP2_VERSION=v1.70.0 RUN git clone --recursive --depth 1 --branch ${NGHTTP2_VERSION} https://github.com/nghttp2/nghttp2.git && \ cd nghttp2 && \ @@ -150,7 +152,7 @@ RUN cd opt && \ git clone --depth 1 https://github.com/curl/curl.git && \ cd curl && autoreconf -fi && \ ./configure --with-openssl --without-libpsl --with-nghttp2 \ - --with-ngtcp2 --with-nghttp3 \ + --with-ngtcp2 --with-nghttp3 --enable-ssls-export \ --prefix /usr/local && \ make -j$(nproc) && make install && \ cd .. && rm -rf curl diff --git a/include/anyhttp/detail/h2_session_details.hpp b/include/anyhttp/detail/h2_session_details.hpp index 0b32fe3..d76a32d 100644 --- a/include/anyhttp/detail/h2_session_details.hpp +++ b/include/anyhttp/detail/h2_session_details.hpp @@ -225,6 +225,34 @@ awaitable ServerSession::do_session(Buffer&& buffer) nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, &ent, 1); #endif + // + // Continue an HTTP/1.1 request upgraded to h2c as stream 1. It has been received completely, + // so nghttp2 opens it half-closed (remote). No HEADERS frame will arrive for it, so do what + // on_begin_headers_callback() and on_frame_recv_callback() would have done. + // + if (m_upgrade) + { + const auto& settings = m_upgrade->settings; + const bool head_request = m_upgrade->method == "HEAD"; + if (auto rv = nghttp2_session_upgrade2(session, + reinterpret_cast(settings.data()), + settings.size(), head_request, nullptr)) + { + mloge("nghttp2_session_upgrade2: {}", nghttp2_strerror(rv)); + nghttp2_session_terminate_session(session, NGHTTP2_PROTOCOL_ERROR); + } + else + { + auto stream = this->create_stream(1); + stream->method = std::move(m_upgrade->method); + stream->url = std::move(m_upgrade->url); + mlogd("upgraded from HTTP/1.1: {} {}", stream->method, stream->url.buffer()); + stream->on_request(); + stream->on_eof(session, 1); + } + m_upgrade.reset(); + } + // // Let NGHTTP2 parse what we have received so far. // This must happen after submitting the server settings. diff --git a/include/anyhttp/h1_session.hpp b/include/anyhttp/h1_session.hpp index 88c993b..3d801a4 100644 --- a/include/anyhttp/h1_session.hpp +++ b/include/anyhttp/h1_session.hpp @@ -86,8 +86,13 @@ class ServerSession : public ServerSessionBase, public BeastSession public: ServerSession(server::Server::Impl& parent, any_io_executor executor, Stream&& stream); + void destroy() noexcept override; void async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers) override; awaitable do_session(Buffer&& data) override; + +private: + /// Takes over the stream after an upgrade to h2c, see do_session(). + std::shared_ptr m_upgraded; }; // ------------------------------------------------------------------------------------------------- diff --git a/include/anyhttp/h2_backend.hpp b/include/anyhttp/h2_backend.hpp index fa092d5..d8fe6c3 100644 --- a/include/anyhttp/h2_backend.hpp +++ b/include/anyhttp/h2_backend.hpp @@ -15,8 +15,10 @@ #include #include #include +#include #include +#include namespace anyhttp::nghttp2 { @@ -25,6 +27,18 @@ namespace anyhttp::nghttp2 using SslStream = boost::asio::ssl::stream; +/** + * A request received as HTTP/1.1 with "Upgrade: h2c" (RFC 7540, section 3.2) that has been answered + * with "101 Switching Protocols". The HTTP/2 session continues it as stream 1. Only requests without + * a body are upgraded, so the stream starts out half-closed (remote). + */ +struct Upgrade +{ + std::string settings; ///< decoded payload of the HTTP2-Settings header + std::string method; + boost::urls::url url; +}; + std::shared_ptr make_server_session(server::Server::Impl& server, boost::asio::any_io_executor executor, boost::asio::ip::tcp::socket&& socket); @@ -37,6 +51,17 @@ std::shared_ptr make_server_session(server::Server::Impl& server, boost::asio::any_io_executor executor, AnyAsyncStream&& stream); +// Cleartext only: there is no upgrade to HTTP/2 over TLS, that is what ALPN is for. + +std::shared_ptr make_server_session(server::Server::Impl& server, + boost::asio::any_io_executor executor, + boost::asio::ip::tcp::socket&& socket, + Upgrade&& upgrade); + +std::shared_ptr make_server_session(server::Server::Impl& server, + boost::asio::any_io_executor executor, + AnyAsyncStream&& stream, Upgrade&& upgrade); + // ------------------------------------------------------------------------------------------------- std::shared_ptr make_client_session(client::Client::Impl& client, diff --git a/include/anyhttp/h2_session.hpp b/include/anyhttp/h2_session.hpp index b5c895a..5901909 100644 --- a/include/anyhttp/h2_session.hpp +++ b/include/anyhttp/h2_session.hpp @@ -2,6 +2,7 @@ #include "anyhttp/common.hpp" #include "client_impl.hpp" +#include "h2_backend.hpp" #include "h2_stream.hpp" #include "server_impl.hpp" #include "session_impl.hpp" @@ -12,6 +13,7 @@ #include #include +#include #include "nghttp2/nghttp2.h" @@ -181,6 +183,9 @@ class ServerSession : public ServerReference, public NGHttp2SessionImpl ServerSession(server::Server::Impl& parent, any_io_executor executor, Stream&& stream); awaitable do_session(Buffer&& data) override; + + /// Set if this session continues an HTTP/1.1 request that has been upgraded to h2c. + std::optional m_upgrade; }; // ================================================================================================= diff --git a/src/h1_session.cpp b/src/h1_session.cpp index 3cd57be..6c0a2db 100644 --- a/src/h1_session.cpp +++ b/src/h1_session.cpp @@ -4,6 +4,7 @@ #include "anyhttp/common.hpp" #include "anyhttp/formatter.hpp" // IWYU pragma: keep #include "anyhttp/h1_backend.hpp" +#include "anyhttp/h2_backend.hpp" #include "anyhttp/server.hpp" #include @@ -17,6 +18,7 @@ #include #include +#include #include #include #include @@ -37,6 +39,9 @@ #include +#include +#include +#include #include using namespace std::chrono_literals; @@ -319,7 +324,7 @@ class WriterBase : public Parent message.body().data = buffer.size() ? const_cast(buffer.data()) : nullptr; #else message.body().data = const_cast(buffer.data()); -#endif +#endif message.body().size = buffer.size(); message.body().more = !eof; @@ -661,6 +666,106 @@ void BeastSession::destroy() noexcept // }); } +template +void ServerSession::destroy() noexcept +{ + if (m_upgraded) + m_upgraded->destroy(); // the stream has been moved there + else + super::destroy(); +} + +// ================================================================================================= + +/** + * Returns what the HTTP/2 session needs to continue \p request as stream 1, if the request asks + * for an upgrade to h2c (RFC 7540, section 3.2) and it can be granted. Otherwise, the upgrade is + * ignored and the request is served as HTTP/1.1, which is always a valid response to it. + * + * Only requests that are complete after their header are upgraded: A request body would have to + * be read in HTTP/1.1 first, before switching protocols. Cleartext only, as h2 over TLS is + * negotiated by ALPN instead. + */ +static std::optional h2c_upgrade(const http::request& request, + const boost::urls::url& url, bool complete) +{ + const auto has_token = [](std::string_view list, std::string_view token) + { + for (auto item : http::token_list(list)) + if (beast::iequals(item, token)) + return true; + return false; + }; + + if (!has_token(request[http::field::upgrade], "h2c") || url.scheme() != "http") + return std::nullopt; + + if (!has_token(request[http::field::connection], "upgrade") || + !has_token(request[http::field::connection], "http2-settings")) + { + logw("upgrade: ignoring h2c upgrade, 'Connection' misses 'Upgrade' or 'HTTP2-Settings'"); + return std::nullopt; + } + + // exactly one HTTP2-Settings header, containing base64url without padding + if (request.count("HTTP2-Settings") != 1) + { + logw("upgrade: ignoring h2c upgrade, need exactly one 'HTTP2-Settings' header"); + return std::nullopt; + } + + if (!complete) + { + logw("upgrade: ignoring h2c upgrade for request with body"); + return std::nullopt; + } + + std::string encoded(request["HTTP2-Settings"]); + std::ranges::replace(encoded, '-', '+'); + std::ranges::replace(encoded, '_', '/'); + + nghttp2::Upgrade upgrade; + upgrade.settings.resize(beast::detail::base64::decoded_size(encoded.size())); + auto [written, read] = + beast::detail::base64::decode(upgrade.settings.data(), encoded.data(), encoded.size()); + upgrade.settings.resize(written); + + // a SETTINGS payload is a sequence of 6-byte entries + if (read != encoded.size() || upgrade.settings.size() % 6 != 0) + { + logw("upgrade: ignoring h2c upgrade, invalid 'HTTP2-Settings' header"); + return std::nullopt; + } + + upgrade.method = request.method_string(); + upgrade.url = url; + return upgrade; +} + +static std::shared_ptr make_h2c_session(server::Server::Impl& server, + any_io_executor executor, + tcp_stream& stream, + nghttp2::Upgrade&& upgrade) +{ + return nghttp2::make_server_session(server, std::move(executor), stream.release_socket(), + std::move(upgrade)); +} + +static std::shared_ptr make_h2c_session(server::Server::Impl& server, + any_io_executor executor, + AnyAsyncStream& stream, + nghttp2::Upgrade&& upgrade) +{ + return nghttp2::make_server_session(server, std::move(executor), std::move(stream), + std::move(upgrade)); +} + +static std::shared_ptr make_h2c_session(server::Server::Impl&, any_io_executor, + ssl::stream&, nghttp2::Upgrade&&) +{ + throw std::logic_error("h2c upgrade over TLS"); // rejected by h2c_upgrade() +} + // ================================================================================================= /** @@ -749,6 +854,32 @@ awaitable ServerSession::do_session(Buffer&& buffer) for (auto& header : request) mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + // + // Upgrade to h2c, if requested: Answer with "101 Switching Protocols" and hand over the + // stream to an HTTP/2 session, which continues this request as stream 1. Anything that + // follows in the buffer (the client preface) is already HTTP/2. + // + if (auto upgrade = h2c_upgrade(request, reader->m_url, parser.is_done())) + { + http::response res{http::status::switching_protocols, request.version()}; + res.set(http::field::connection, "Upgrade"); + res.set(http::field::upgrade, "h2c"); + reader.reset(); // owns the parser and thereby 'request' + + if (auto [ec, n] = co_await http::async_write(m_stream, res, as_tuple); ec) + { + mlogw("upgrade: writing 101 response: {}", ec.message()); + break; + } + + mlogi("upgrading to h2c, {} bytes in buffer", m_buffer.size()); + m_upgraded = make_h2c_session(server(), super::get_executor(), m_stream, + std::move(*upgrade)); + co_await m_upgraded->do_session(std::move(m_buffer)); + mlogi("h2c session done, served {} requests before upgrade", requestCounter - 1); + co_return; + } + // // Prepare response. // diff --git a/src/h2_session.cpp b/src/h2_session.cpp index a2bab2e..b561058 100644 --- a/src/h2_session.cpp +++ b/src/h2_session.cpp @@ -657,6 +657,27 @@ std::shared_ptr make_server_session(server::Server::Impl& server, std::move(socket)); } +std::shared_ptr make_server_session(server::Server::Impl& server, + asio::any_io_executor executor, + asio::ip::tcp::socket&& socket, + Upgrade&& upgrade) +{ + auto session = std::make_shared>( + server, std::move(executor), std::move(socket)); + session->m_upgrade = std::move(upgrade); + return session; +} + +std::shared_ptr make_server_session(server::Server::Impl& server, + asio::any_io_executor executor, + AnyAsyncStream&& stream, Upgrade&& upgrade) +{ + auto session = std::make_shared>(server, std::move(executor), + std::move(stream)); + session->m_upgrade = std::move(upgrade); + return session; +} + std::shared_ptr make_client_session(client::Client::Impl& client, asio::any_io_executor executor, asio::ip::tcp::socket&& socket) diff --git a/src/h3_stream.cpp b/src/h3_stream.cpp index 56a7e4d..96085d7 100644 --- a/src/h3_stream.cpp +++ b/src/h3_stream.cpp @@ -385,6 +385,7 @@ void Http3Stream::bind_write_cancellation(WriteHandler& handler, uint64_t token) }); } +// https://nghttp2.org/nghttp3/types.html#c.nghttp3_read_data_callback nghttp3_ssize Http3Stream::data_reader(nghttp3_vec* vec, size_t veccnt, uint32_t* pflags) { if (veccnt == 0) diff --git a/test/CMakeLists.txt b/test/CMakeLists.txt index 1de2268..bc98d64 100644 --- a/test/CMakeLists.txt +++ b/test/CMakeLists.txt @@ -8,6 +8,9 @@ add_executable(test_all ${SRC_FILES}) target_link_libraries(test_all PRIVATE GTest::GTest GTest::gmock anyhttp) target_link_libraries(test_all PRIVATE spdlog::spdlog_header_only) +# the h2c upgrade tests drive an HTTP/2 client session by hand +target_link_libraries(test_all PRIVATE PkgConfig::NGHTTP2) + # the tests connect over TLS and pass pki/out/root.pem to curl, so the PKI must exist first add_dependencies(test_all pki) diff --git a/test/test_client_async.cpp b/test/test_client_async.cpp new file mode 100644 index 0000000..ab9e0cc --- /dev/null +++ b/test/test_client_async.cpp @@ -0,0 +1,769 @@ +#include "test_fixtures.hpp" + +#include + +#include +#include +#include +#include +#include +#include + +// ================================================================================================= + +INSTANTIATE_TEST_SUITE_P(ClientAsync, ClientAsync, + ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, WHEN_post_data_THEN_receive_echo) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + size_t bytes = 1024; + auto count = co_await (generate(request, bytes) && count_response(request)); + EXPECT_EQ(bytes, count); + }; +} + +TEST_P(ClientAsync, WHEN_post_without_path_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path(""), {}); + co_await generate(request, 1024); + auto [ec, response] = co_await request.async_get_response(as_tuple); + EXPECT_TRUE(ec); + }; +} + +TEST_P(ClientAsync, WHEN_post_to_unknown_path_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("unknown"), {}); + co_await generate(request, 1_m); + auto response = co_await request.async_get_response(); + EXPECT_EQ(response.status_code(), 404); + auto received = co_await drain(response); + }; +} + +TEST_P(ClientAsync, WHEN_server_discards_request_THEN_error_500) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("discard"), {}); + co_await generate(request, 1024); + auto [ec, response] = co_await request.async_get_response(as_tuple); + EXPECT_TRUE(ec); + }; +} + +TEST_P(ClientAsync, WHEN_server_discards_request_delayed_THEN_error_500) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("detach"), {}); + co_await generate(request, 1024); + auto [ec, response] = co_await request.async_get_response(as_tuple); + EXPECT_TRUE(ec); + }; +} + +TEST_P(ClientAsync, WHEN_server_discards_request_with_body_delayed_THEN_error_500) +{ + test = [this](Session session) -> awaitable + { + auto executor = co_await this_coro::executor; + auto request = co_await session.async_submit(url.set_path("detach"), {}); + auto [ep] = co_await co_spawn(executor, send(request, rv::iota(uint8_t{0})), as_tuple); + EXPECT_TRUE(ep); + }; +} + +TEST_P(ClientAsync, WHEN_invalid_port_in_host_header_THEN_reports_error) +{ + test = [this](Session session) -> awaitable + { + Fields fields; + fields.set("Host", "host:12345x"); + auto request = co_await session.async_submit(url.set_path("echo"), fields); + auto response = co_await (send_eof(request) && count_response(request)); + }; +} + +TEST_P(ClientAsync, WHEN_get_response_is_called_twice_THEN_reports_error) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo")); + auto [ec, response] = co_await request.async_get_response(as_tuple); + EXPECT_EQ(ec, boost::system::errc::success); + std::tie(ec, response) = co_await request.async_get_response(as_tuple); + EXPECT_EQ(ec, boost::system::errc::connection_already_in_progress); + EXPECT_EQ(ec, asio::error::basic_errors::already_started); + }; +} + +TEST_P(ClientAsync, WHEN_get_response_is_detached_THEN_does_not_crash) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP(); + + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo")); + request.async_get_response(detached); + }; +} + +TEST_P(ClientAsync, WHEN_server_discards_request_while_writing_THEN_connection_is_reset) +{ + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await sleep(150ms); + request.reset(); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + auto executor = co_await this_coro::executor; + auto [ec] = co_await co_spawn(executor, send(request, rv::iota(uint8_t(0))), as_tuple); + EXPECT_EQ(code(ec), boost::system::errc::connection_reset); + }; +} + +TEST_P(ClientAsync, WHEN_server_discards_request_and_response_THEN_completes_anyway) +{ + // if (GetParam() == anyhttp::Protocol::http11) + // GTEST_SKIP(); // FIXME: timeout + + custom = [this](server::Request request, server::Response response) -> awaitable + { + std::ignore = request; + std::ignore = response; + co_return; + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + auto [ec, _] = co_await request.async_get_response(as_tuple); + EXPECT_EQ(ec, boost::beast::http::error::end_of_stream); + // EXPECT_EQ(ec, std::errc::connection_reset); + }; +} + +TEST_P(ClientAsync, WHEN_client_cancels_write_THEN_can_resume) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP(); // a chunked body cannot be cancelled correctly --> disconnects + + test = [this](Session session) -> awaitable + { + co_await this_coro::throw_if_cancelled(false); + auto executor = co_await this_coro::executor; + auto request = co_await session.async_submit(url.set_path("echo")); + auto response = co_await request.async_get_response(); + + // send as much data as possible within 1s, should run into backpressure + auto [ep] = co_await co_spawn(executor, send(request, rv::iota(uint8_t(0))), + cancel_after(1s, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + + if (GetParam() == anyhttp::Protocol::h3) + { + // + // QUIC: whether the FIN can slip out while the send window is closed depends on flow + // control timing, so don't assert either way here. What matters is that ending the + // upload and draining the response together complete the exchange. + // + auto received = co_await (send_eof(request) && drain(response)); + EXPECT_GT(received, 0); + } + else + { + // now, with a closed window, we cannot even end the upload + std::tie(ep) = co_await co_spawn(executor, send_eof(request), cancel_after(1ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + + // as we have no control over when the send window is re-opened, wait for it in parallel + auto received = co_await (send_eof(request) && drain(response)); + EXPECT_GT(received, 0); + } + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, YieldFuzz) +{ +#if 0 + static std::random_device rd; + static std::mt19937 gen(rd()); +#else + static std::mt19937 gen(42); // fixed seed for reproducibility +#endif + + custom = [this](server::Request request, server::Response response) -> awaitable + { + std::uniform_int_distribution<> dist(0, 10); + constexpr auto msg = "Hello, Client!"sv; + co_await yield(dist(gen)); + Fields fields; + fields.set("Content-Length", std::to_string(msg.size())); + co_await response.async_submit(200, fields); + co_await yield(dist(gen)); + co_await response.async_write(asio::buffer(msg)); + co_await yield(dist(gen)); + co_await response.async_write_eof(); + co_await yield(dist(gen)); + std::array data; + co_await request.async_read_some(asio::buffer(data), as_tuple); + }; + test = [this](Session session) -> awaitable + { + std::uniform_int_distribution<> dist(0, 10); + for (size_t i = 0; i < 100; ++i) + { + std::println( + "=== {} =========================================================================", i); + co_await yield(dist(gen)); + Fields fields; + if (GetParam() == anyhttp::Protocol::http11) + fields.set("Connection", "Keep-Alive"); + fields.set("Content-Length", "0"); + auto request = co_await session.async_submit(url, fields); + co_await yield(dist(gen)); + co_await request.async_write_eof(); + co_await yield(dist(gen)); + co_await count_response(request); + } + }; +} + +// +// The end of an incoming body is an error code, not a zero-sized read -- and it keeps being +// reported for every read issued after it. A zero-length buffer, on the other hand, says nothing +// about the body at all: it completes immediately, at the end of a body just as anywhere else. +// +TEST_P(ClientAsync, WHEN_body_ends_THEN_read_reports_eof) +{ + static const auto hello = "Hello, World!"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); + co_await response.async_write_eof(asio::buffer(hello)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + + std::string body; + std::array buffer; // small on purpose: several reads before the end + for (;;) + { + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + if (ec) + { + EXPECT_EQ(ec, asio::error::eof); + EXPECT_EQ(n, 0u); + break; + } + body.append(buffer.data(), n); + } + EXPECT_EQ(body, hello); + + // + // Reading past the end of a body says the same thing again -- also once the protocol layer + // has torn the underlying stream down in the meantime, which the yield gives it every + // opportunity to do (both sides of the exchange are finished by now). + // + co_await yield(20); + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); + EXPECT_EQ(ec, asio::error::eof); + + // ... but a zero-length read is not a read, and reports nothing + std::array empty; + std::tie(ec, n) = co_await response.async_read_some(asio::buffer(empty), as_tuple); + EXPECT_FALSE(ec); + EXPECT_EQ(n, 0u); + }; +} + +// +// An empty async_write() no longer ends a body -- async_write_eof() does, and nothing else. So a +// message with an empty write in the middle of it still carries everything written after that. +// +TEST_P(ClientAsync, WHEN_empty_buffer_is_written_THEN_body_stays_open) +{ + static const auto tail = "still here"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + EXPECT_EQ(co_await drain(request), 0u); + co_await response.async_submit(200, {}); + co_await response.async_write({}); // writes nothing, leaves the body open + co_await response.async_write_eof(asio::buffer(tail)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write({}); // likewise: the request body stays open + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(co_await read(response), tail); + }; +} + +// +// Ending a body twice is harmless -- the second call has nothing left to do -- and an empty +// write stays a free no-op even then. Data after the end is neither: there is no body left for +// it to belong to, through whichever entry point it tries to sneak in. +// +TEST_P(ClientAsync, WHEN_written_after_eof_THEN_reports_broken_pipe) +{ + static const auto hello = "Hello, World!"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); + co_await response.async_write_eof(asio::buffer(hello)); + + auto [ec] = co_await response.async_write_eof(as_tuple); + EXPECT_FALSE(ec); + + std::tie(ec) = co_await response.async_write({}, as_tuple); + EXPECT_FALSE(ec); + + std::tie(ec) = co_await response.async_write(asio::buffer(hello), as_tuple); + EXPECT_EQ(ec, boost::system::errc::broken_pipe); + + std::tie(ec) = co_await response.async_write_eof(asio::buffer(hello), as_tuple); + EXPECT_EQ(ec, boost::system::errc::broken_pipe); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(co_await read(response), hello); + }; +} + +TEST_P(ClientAsync, HelloWorld) +{ + static const auto hello = "Hello, World!"sv; + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await response.async_submit(200, {}); + co_await response.async_write_eof(asio::buffer(hello)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + auto body = co_await read(response); + EXPECT_EQ(body, hello); + }; +} + +// ------------------------------------------------------------------------------------------------- + +// +// A single async_write() larger than what the transport hands to its peer in one go, i.e. the +// whole body in one call instead of chunk by chunk. HTTP/3 has to keep offering the same buffer +// to nghttp3 across many packets here, and complete the write only once all of it is +// acknowledged. +// +TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) +{ + static const std::vector body = [] + { + std::vector data(256_k); + std::ranges::generate(data, [i = uint8_t(0)]() mutable { return i++; }); + return data; + }(); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + co_await drain(request); + + co_await response.async_submit(200, fields({{"Content-Length", body.size()}})); + co_await response.async_write_eof(asio::buffer(body)); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + EXPECT_EQ(co_await count_response(request), body.size()); + }; +} + +// +// Cancelling an async_write_eof() that carries data. The buffer goes back to the caller the +// moment the handler runs, so the backend must stop referencing it right there -- for HTTP/3's +// zero-copy path that means resetting the stream, exactly as for a cancelled plain write; under +// ASAN this test is what catches a backend that keeps pointing into the freed buffer. +// +TEST_P(ClientAsync, WHEN_server_cancels_write_eof_THEN_client_sees_truncated_body) +{ + static const std::vector body(8_m, 'x'); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await drain(request); + co_await response.async_submit(200, {}); + + // + // Far more than the peer's receive window, and the client below doesn't read a byte until + // this is over, so the write is guaranteed to still be in progress when it is cancelled. + // + auto [ec] = co_await response.async_write_eof(asio::buffer(body), + cancel_after(50ms, as_tuple)); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + + // leave the body untouched until the cancellation above has hit, see the sibling testcase + asio::steady_timer timer(co_await this_coro::executor, 150ms); + co_await timer.async_wait(deferred); + + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + EXPECT_LT(received, body.size()); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + }; +} + +// +// Cancelling an async_write_eof() whose FIN never made it out must not leave the body in limbo: +// the intent to end it is rolled back, and a re-issued async_write_eof() ends the (now shorter) +// body for real -- instead of completing as a no-op while the peer waits forever for the end. +// +TEST_P(ClientAsync, WHEN_client_cancels_write_eof_THEN_can_still_end) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP(); // a chunked body cannot be cancelled correctly --> disconnects + + static const std::vector body(8_m, 'x'); + + test = [this](Session session) -> awaitable + { + co_await this_coro::throw_if_cancelled(false); + auto executor = co_await this_coro::executor; + auto request = co_await session.async_submit(url.set_path("echo")); + auto response = co_await request.async_get_response(); + + // far more than the send window, with nobody reading the echo yet: this cannot complete + auto write_eof = [&]() -> awaitable + { co_await request.async_write_eof(asio::buffer(body)); }; + auto [ep] = co_await co_spawn(executor, write_eof(), cancel_after(100ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + + // the FIN never went out with the cancelled write, so the body can still be ended + auto received = co_await (send_eof(request) && drain(response)); + EXPECT_GT(received, 0u); + EXPECT_LT(received, body.size()); + }; +} + +// +// Cancelling a response write mid-body. HTTP/3 hands the caller's buffer to nghttp3 by reference, +// so bytes already offered and not yet acknowledged cannot simply be abandoned -- the stream is +// reset instead, which is what the peer would see anyway for a body that stops short of its end. +// +TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) +{ + static const std::vector body(8_m, 'x'); + + custom = [this](server::Request request, server::Response response) -> awaitable + { + // drain the request -- HTTP/1.1 closes the connection on an unfinished parser + co_await drain(request); + + co_await response.async_submit(200, {}); + + // + // Far more than the peer's receive window, and the client below doesn't read a byte until + // this is over, so the write is guaranteed to still be in progress when it is cancelled. + // + auto executor = co_await this_coro::executor; + auto [ep] = co_await co_spawn(executor, send(response, std::span(body)), + cancel_after(50ms, as_tuple)); + EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + + // + // Leave the response body untouched for now: whatever the server manages to send fills up + // the receive window and stays there, so its write cannot run to completion before the + // cancellation above hits. + // + asio::steady_timer timer(co_await this_coro::executor, 150ms); + co_await timer.async_wait(deferred); + + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + std::println("received {} of {} bytes ({})", received, body.size(), ec.message()); + EXPECT_LT(received, body.size()); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + }; +} + +// ================================================================================================= + +TEST_P(ClientAsync, ServerYieldFirst) +{ + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await yield(10); + co_await response.async_submit(200, {}); + co_await yield(10); + co_await response.async_write_eof(); + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url); + co_await request.async_write_eof(); + co_await count_response(request); + }; +} + +// ---------------------------------------------------------------------------------------------- + +static std::optional stackRemainingBytes() +{ + pthread_attr_t attr; + if (pthread_getattr_np(pthread_self(), &attr) != 0) + return std::nullopt; + + void* stack_base = nullptr; + size_t stack_size = 0; + if (pthread_attr_getstack(&attr, &stack_base, &stack_size) != 0) + { + pthread_attr_destroy(&attr); + return std::nullopt; + } + + pthread_attr_destroy(&attr); + + if (stack_base == nullptr || stack_size == 0) + return std::nullopt; + + int local = 0; + std::uintptr_t sp = reinterpret_cast(&local); + std::uintptr_t base = reinterpret_cast(stack_base); + + return (sp >= base) ? std::optional(sp - base) : std::nullopt; +} + +TEST_P(ClientAsync, Recursion) +{ +#if __has_feature(address_sanitizer) + GTEST_SKIP() << "skipped under address sanitizer"; +#endif + if (!stackRemainingBytes()) + GTEST_SKIP() << "unable to measure stack on this platform"; + + test = [this](Session session) -> awaitable + { + auto ex = co_await this_coro::executor; + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + + // verify that immediate completion (here, due to an empty buffer) does not cause recursion + std::array empty; + co_await response.async_read_some(asio::buffer(empty)); + auto s0 = stackRemainingBytes().value(); + co_await response.async_read_some(asio::buffer(empty)); + auto s1 = stackRemainingBytes().value(); + EXPECT_EQ(s0, s1); + + // however, ASIO allows us to control this behavior using "immediate executors" + co_await response.async_read_some(asio::buffer(empty), bind_immediate_executor(ex)); + auto s2 = stackRemainingBytes().value(); + EXPECT_GT(s1, s2); + }; +} + +// ---------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, Custom) +{ + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await response.async_submit(200, {}); + std::array buffer; + for (;;) + { + auto [ec, n] = co_await request.async_read_some(asio::buffer(buffer), as_tuple); + if (ec) + { + co_await response.async_write_eof(); + co_return; + } + co_await response.async_write(asio::buffer(buffer, n)); + } + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, {}); + constexpr size_t bytes = 1024; + auto count = co_await (generate(request, bytes) && count_response(request)); + EXPECT_EQ(bytes, count); + }; +} + +TEST_P(ClientAsync, IgnoreRequest) +{ + custom = [this](server::Request request, server::Response response) -> awaitable + { + co_await response.async_submit(200, {}); + co_await response.async_write_eof(); + }; + test = [this](Session session) -> awaitable + { + Fields fields; + fields.set("content-length", "0"); + auto request = co_await session.async_submit(url, fields); + auto count = co_await (generate(request, 0) && count_response(request)); + EXPECT_EQ(count, 0); + }; +} + +TEST_P(ClientAsync, IgnoreRequestAndResponse) +{ + custom = [this](server::Request request, server::Response response) -> awaitable + { + std::ignore = request; + std::ignore = response; + co_return; + }; + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, {}); + auto res = co_await (generate(request, 0) && try_read_response(request)); + EXPECT_FALSE(res.has_value()); + std::println("ERROR: {}", res.error().message()); + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, PostRange) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + // co_await request.async_write(asio::buffer("ping"sv)); // FIXME: + auto response = co_await request.async_get_response(); + // std::string s(10_m, 'a'); + // auto sender = send(request, std::string_view("blah")); + // auto sender = send(request, std::string(10_m, 'a')); + auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); + auto received = co_await (std::move(sender) && drain(response)); + loge("received: {}", received); + EXPECT_EQ(received, 1_m); + }; +} + +TEST_P(ClientAsync, PostRangeImmediate) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); + auto received = co_await (std::move(sender) && count_response(request)); + loge("received: {}", received); + EXPECT_EQ(received, 1_m); + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, WHEN_request_is_sent_THEN_response_is_received_before_body_is_posted) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + constexpr size_t bytes = 1024; + co_await generate(request, bytes); + EXPECT_EQ(co_await drain(response), bytes); + }; +} + +// ------------------------------------------------------------------------------------------------- + +// +// HTTP/1.1 supports pipelining in the sense that multiple, full requests can be made before +// the responses are received. +// +// TODO: Any kind of interleaving is not supported. An attempt to issue another request while the +// previous request is still active should result in an error, immediately. +// +TEST_P(ClientAsync, WHEN_multiple_request_are_made_THEN_responses_are_received_in_order) +{ + test = [this](Session session) -> awaitable + { + auto request1 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request1.async_write_eof(asio::buffer("Hello, Server #1!"sv)); + + auto request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer("Hello, Server #2! XYZ"sv)); + + auto response1 = co_await request1.async_get_response(); + EXPECT_EQ(co_await drain(response1), 17); + + auto response2 = co_await request2.async_get_response(); + EXPECT_EQ(co_await drain(response2), 21); + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, EatRequest) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("eat_request"), {}); + co_await generate(request, 1024); + auto response = co_await request.async_get_response(); + auto received = co_await drain(response); + EXPECT_EQ(received, 0); + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsync, Dump) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit( + url.set_path("dump space").set_params({{"blah", "white space"}, {"x", "y"}}), {}); + co_await send_eof(request); + auto response = co_await request.async_get_response(); + auto dump = co_await read(response); + EXPECT_THAT(dump, testing::HasSubstr("path: /dump space")); + EXPECT_THAT(dump, testing::HasSubstr(" blah=white space")); + }; +} + +// ================================================================================================= diff --git a/test/test_client_async_cancellation.cpp b/test/test_client_async_cancellation.cpp new file mode 100644 index 0000000..d1ccc3e --- /dev/null +++ b/test/test_client_async_cancellation.cpp @@ -0,0 +1,317 @@ +#include "test_fixtures.hpp" + +#include +#include +#include + +// ================================================================================================= + +// +// Backpressure, cancellation and connection loss, on top of the ClientAsync fixture. +// +class ClientAsyncCancellation : public ClientAsync +{ +}; + +INSTANTIATE_TEST_SUITE_P(ClientAsyncCancellation, ClientAsyncCancellation, + ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ClientAsyncCancellation, Backpressure) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + auto sender = send(request, rv::iota(uint8_t(0))); + co_await (std::move(sender) || sleep(2s)); + // FIXME: count bytes sent, just like asio::async_write() does + // FIXME: or even use asio::async_write() on top of a async_write_some() implementation + + // + // Now that the flow control window is 0, we can't even send an EOF any more -- except over + // QUIC, where whether the FIN slips out without credit depends on flow control timing, so + // only assert that for the stream protocols. + // + auto rc = co_await (send_eof(request) || sleep(100ms)); + if (GetParam() != anyhttp::Protocol::h3) + EXPECT_EQ(rc.index(), 1); + + // So instead, we start doing this in background, to be resumed as soon as the window reopens. + co_spawn(co_await this_coro::executor, send_eof(request), detached); // FIXME: join + + std::println("receiving...."); + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + std::println("receiving... done, got {} bytes ({})", received, ec.message()); + EXPECT_GT(received, 0); + // EXPECT_EQ(received, sent); + // FIXME: we should be able to receive the remainders that already have been buffered + // FIXME: in the end, this must be the same as the the bytes sent above + }; +} + +// +// Cancellation of a large buffer with Content-Length. +// +// Any short write of a body with known content length should result in a 'partial message' error. +// +// FIXME: As of nghttp2 version 1.67, the partial message results in a GOAWAY, so that only one +// request can be made. The following request should throw an exception. +// +TEST_P(ClientAsyncCancellation, CancellationContentLength) +{ + test = [this](Session session) -> awaitable + { + const size_t length = 50_m; + const std::vector buffer(length); + for (size_t i = 0; i <= 20; ++i) + { + if (!session) + session = co_await client->async_connect(); + + Fields fields; + fields.set("content-length", std::to_string(length)); + auto request = co_await session.async_submit(url.set_path("echo"), fields); + auto response = co_await request.async_get_response(); + + // + // This is a single large buffer and will be serialized as a single chunk. When writing + // gets cancelled, there is no way to recover gracefully. + // + auto sender = sendAndForceEOF(request, std::string_view(buffer)); + + boost::system::error_code ec; + auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); + std::println("received {} bytes (\x1b[1;31m{}\x1b[0m, yielded {})", std::get<1>(received), + ec.message(), i); + EXPECT_LT(std::get<1>(received), length); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + + session.reset(); + } + }; +} + +// +// Cancellation of sending a single, large buffer without Content-Length. +// +// HTTP/1.1: As always when not providing Content-Length, the data is chunked. When sending data +// as a single, large buffer, this will result in a single, large chunk of same size. +// If sending that chunk is interrupted, there is no way to recover. The sender will +// close the connection in this situation. +// +// HTTP/2: Cancelling a large buffer without Content-Length will look to the server just like a +// short buffer. No error is raised. FIXME: we could try to support cancellation here +// by closing the stream without sending an EOF. But that would also stop the receiving +// direction. +// +TEST_P(ClientAsyncCancellation, Cancellation) +{ + test = [this](Session session) -> awaitable + { + const size_t length = 50_m; + const std::vector buffer(length, 'a'); + for (size_t i = 0; i <= 20; ++i) + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + auto sender = sendAndDrop(std::move(request), std::string_view(buffer)); + + boost::system::error_code ec; + auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); + std::println("received {} bytes ({}, yield {})", std::get<1>(received), ec.message(), i); + EXPECT_LT(std::get<1>(received), length); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + + // HTTP/1.1 needs to reconnect here + // HTTP/2 can handle this without reconnect -- only the stream is cancelled + if (GetParam() == anyhttp::Protocol::http11) + { + session.reset(); + session = co_await client->async_connect(); + } + } + }; +} + +// +// Cancellation of sending a large amount of data that is split into many smaller chunks. +// +// This should work with any protocol, without error. As we don't give a Content-Length in advance, +// cancelling the upload should not be terminal. BUT: cancellation of a parallel group seems to +// do 'terminal' cancellation... +// +// TODO: Aside using operator||, when manually setting up a parallel group, it is possible to +// specify the cancellation type that should be used. +// +// TODO: If an operation supports "partial" as well, it is free to cancel like that even when +// requested to do terminal "cancellation". Cancellation types are backward compatible this +// way. +// +TEST_P(ClientAsyncCancellation, CancellationRange) +{ + test = [this](Session session) -> awaitable + { + for (size_t i = 6; i <= 6; ++i) + { + co_await yield(); + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + // auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0))); + auto sender = sendAndDrop(std::move(request), rv::iota(uint8_t(0))); + + boost::system::error_code ec; + auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); + std::println("received {} bytes ({}, yield {})", std::get<1>(received), ec.message(), i); + EXPECT_EQ(ec, boost::beast::http::error::partial_message); + co_await client->async_connect(); + } + }; +} + +TEST_P(ClientAsyncCancellation, PerOperationCancellation) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + + asio::cancellation_signal cancel; + asio::steady_timer timer(co_await asio::this_coro::executor, 110ms); + timer.async_wait([&cancel](const boost::system::error_code& ec) { // + cancel.emit(asio::cancellation_type::terminal); + }); + + std::array buffer; + auto token = asio::bind_cancellation_slot(cancel.slot(), as_tuple); + auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), std::move(token)); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + }; +} + +TEST_P(ClientAsyncCancellation, CancelAfter) +{ + test = [this](Session session) -> awaitable + { + auto request = + co_await session.async_submit(url.set_path("echo").set_params({{"delay", "1000"}}), {}); + auto [ec, response] = co_await request.async_get_response(cancel_after(250ms, as_tuple)); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + + std::tie(ec, response) = co_await request.async_get_response(cancel_after(0ms, as_tuple)); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + + std::tie(ec, response) = co_await request.async_get_response(as_tuple); + EXPECT_FALSE(ec); + + constexpr auto msg = "Hello, Client!"sv; + co_await request.async_write_eof(asio::buffer(msg)); + EXPECT_EQ(co_await read(response), msg); + }; +} + +TEST_P(ClientAsyncCancellation, WHEN_send_more_than_content_length_THEN_connection_is_reset) +{ + test = [this](Session session) -> awaitable + { + Fields fields; + fields.set("content-length", "1024"); + auto request = co_await session.async_submit(url.set_path("eat_request"), fields); + auto response = co_await request.async_get_response(); + co_await drain(response); + + auto ex = co_await this_coro::executor; + auto [ep] = co_await co_spawn(ex, send(request, rv::iota(uint8_t(0))), as_tuple); + + // + // Which of the two the write reports is a matter of how far the kernel has gotten with the + // peer's RST by the time we get to write again -- the first write after it fails with + // ECONNRESET, any later one with EPIPE. Single-threaded we reliably hit the former, with + // more than one thread the latter; both mean the same thing here. + // + EXPECT_THAT(code(ep), testing::AnyOf(boost::system::errc::connection_reset, + boost::system::errc::broken_pipe)); + }; +} + +// ================================================================================================= + +TEST_P(ClientAsyncCancellation, ClientDropRequest) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + }; +} + +// ================================================================================================= + +TEST_P(ClientAsyncCancellation, ResetServerDuringRequest) +{ + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + + // + // Deliberately NOT use_future(): with more than one thread the client lives on a strand, + // and blocking that strand in future.get() below would keep the very handlers that + // complete this send from ever running. asio::experimental::promise starts the coroutine + // right away, just like use_future, but is awaited instead of waited on. + // + auto promise = co_spawn(request.get_executor(), send(request, rv::iota(uint8_t(0))), + asio::experimental::use_promise); + + std::println("============================================================================="); + for (size_t i = 0; i < 10; ++i) + { + std::println("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -", i); + co_await yield(); + } + + std::println("============================================================================="); + server.reset(); + + for (size_t i = 0; i < 10; ++i) + { + std::println("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -", i); + co_await yield(); + } + + auto exception_ptr = co_await std::move(promise)(as_tuple(use_awaitable)); + + boost::system::error_code ec; + auto received = co_await try_receive(response, ec); + loge("received: {} ({} bytes)", ec.message(), received); + }; +} + +TEST_P(ClientAsyncCancellation, DISABLED_SpawnAndForget) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP(); // FIXME: ASAN errors + + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + auto response = co_await request.async_get_response(); + co_await yield(); + + std::println("- - spawning - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - "); + co_spawn(context, + [request = std::move(request)]() mutable -> awaitable + { // + std::println("- - SPAWNED - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -"); + co_await yield(5); + std::println("- - SPAWNED, sending - - - - - - - - - - - - - - - - - - - - - - - - -"); + co_await send(request, rv::iota(uint8_t(0))); + }, detached); + }; +} + +// ================================================================================================= diff --git a/test/test_client_connect.cpp b/test/test_client_connect.cpp new file mode 100644 index 0000000..19d21de --- /dev/null +++ b/test/test_client_connect.cpp @@ -0,0 +1,68 @@ +#include "test_fixtures.hpp" + +// ================================================================================================= + +class ClientConnect : public testing::Test +{ +public: + void SetUp() override { setupLogging(); } +}; + +// ------------------------------------------------------------------------------------------------- + +TEST_F(ClientConnect, WHEN_unknown_host_THEN_completes_with_host_not_found_eventually) +{ + boost::asio::io_context context; + client::Config config{.url = boost::urls::url("http://this-domain-does-not-exist:12345")}; + client::Client client(context.get_executor(), config); + client.async_connect([this](boost::system::error_code ec, Session session) + { + loge("ERROR: {}", ec.message()); + EXPECT_TRUE(ec == boost::asio::error::netdb_errors::host_not_found || + ec == boost::asio::error::netdb_errors::host_not_found_try_again); + }); + context.run(); +} + +TEST_F(ClientConnect, WHEN_wrong_port_THEN_completes_with_host_not_found_eventually) +{ + boost::asio::io_context context; + auto port = get_unused_port(context); + client::Config config{.url = boost::urls::url("http://localhost").set_port_number(port)}; + client::Client client(context.get_executor(), config); + client.async_connect([this](boost::system::error_code ec, Session session) + { + loge("ERROR: {}", ec.message()); + EXPECT_EQ(ec, boost::system::errc::connection_refused); + }); + context.run(); +} + +TEST_F(ClientConnect, WHEN_async_connect_is_cancelled_THEN_returns_operation_aborted) +{ + boost::asio::io_context context; + client::Config config{.url = boost::urls::url("http://localhost:12345")}; + client::Client client(context.get_executor(), config); + client.async_connect(cancel_after(0ms, [this](boost::system::error_code ec, Session session) + { + loge("ERROR: {}", ec.message()); + EXPECT_EQ(ec, boost::system::errc::operation_canceled); + })); + + context.run(); +} + +TEST_F(ClientConnect, WHEN_connect_to_broadcast_ip_THEN_completes_with_network_unreachable) +{ + boost::asio::io_context context; + client::Config config{.url = boost::urls::url("http://255.255.255.255:12345")}; + client::Client client(context.get_executor(), config); + client.async_connect([this](boost::system::error_code ec, Session session) + { + loge("ERROR: {}", ec.message()); + EXPECT_EQ(ec, boost::system::errc::network_unreachable); + }); + context.run(); +} + +// ================================================================================================= diff --git a/test/test_external.cpp b/test/test_external.cpp new file mode 100644 index 0000000..a3195fe --- /dev/null +++ b/test/test_external.cpp @@ -0,0 +1,442 @@ +#include "test_fixtures.hpp" + +#include + +#include +#include +#include + +#include + +#include +#include +#include +#include +#include + +namespace bp = boost::process::v2; + +// https://github.com/curl/curl/issues/10634 --> use custom built curl +#define CURL_PATH "/usr/local/bin/curl" +#define NGHTTP_PATH "/usr/local/bin/nghttp" +#define H2LOAD_PATH "/usr/local/bin/h2load" + +// ================================================================================================= + +class External : public Server +{ +protected: + auto split_lines(std::string_view lines) + { + if (lines.ends_with('\n')) + lines.remove_suffix(1); + + return lines | std::views::split('\n') | + std::views::transform([](auto range) { return std::string_view(range); }); + } + + awaitable log(std::string prefix, readable_pipe& pipe) + { + std::string buffer; + auto print = [&](std::string_view line) + { + if (line.ends_with('\r')) + line.remove_suffix(1); + + // print trailing '…' if there is more data in the buffer after this line + const auto continuation = (line.size() + 1 == buffer.size()) ? "" : "…"; + std::println("{}: \x1b[32m{}\x1b[0m{}", prefix, line, continuation); + }; + + auto cs = co_await this_coro::cancellation_state; + try + { + for (;;) + { + auto n = co_await async_read_until(pipe, dynamic_buffer(buffer), '\n'); + for (;;) + { + print(std::string_view(buffer).substr(0, n - 1)); + buffer.erase(0, n); + + // try to bundle multiple lines, looks nicer in debug output + auto pos = buffer.find('\n'); + if (pos == std::string::npos) + break; + n = pos + 1; + } + } + } + catch (const boost::system::system_error& ec) + { + std::println("{}: {}", prefix, ec.code().message()); + if (cs.cancelled() != cancellation_type::none) + std::println("{}: CANCELLED ({})", prefix, cs.cancelled()); + + for (auto line : split_lines(buffer)) + print(line); + + if (ec.code() == error::eof) + co_return; + + throw; + } + } + + awaitable read_all(readable_pipe pipe) + { + std::string result; + auto [ec, nread] = co_await asio::async_read(pipe, asio::dynamic_buffer(result), as_tuple); + logi("STDOUT: {} bytes ({})", nread, what(ec)); + if (ec && ec != error::eof) + throw boost::system::system_error(ec); + co_return result; + } + + awaitable spawn_process(std::filesystem::path path, std::vector args) + { + logi("spawn: {} {}", path.generic_string(), boost::algorithm::join(args, " ")); + + auto ex = co_await this_coro::executor; + readable_pipe out(ex), err(ex); + bp::process child(ex, path, args, bp::process_stdio{.out = out, .err = err}); + // bp::process_environment{{"LD_LIBRARY_PATH=/usr/local/lib"}}); + + logi("spawn: starting to communicate..."); +#if 1 + auto result = co_await (log("STDERR", err) && read_all(std::move(out))); +#else + co_await (log("STDERR", err) && log("STDOUT", out)); + auto result = std::string(); +#endif + logi("spawn: starting to communicate... done, read {} bytes", result.size()); + + co_await child.async_wait(); + if (child.exit_code()) + logw("exit_code={}", child.exit_code()); + else + logi("exit_code={}", child.exit_code()); + + if (--numSpawned <= 0) + { + co_await post(server->get_executor()); + logi("all processes exited, stopping server..."); + server.reset(); + logi("all processes exited, stopping server... done"); + } + + co_return result; + } + + std::future spawn(std::filesystem::path path, std::vector args) + { + ++numSpawned; + std::promise promise; + auto future = promise.get_future(); + co_spawn(strand, spawn_process(std::move(path), std::move(args)), + bind_executor(strand, [this, promise = std::move(promise)]( + const std::exception_ptr& ex, std::string str) mutable + { + if (ex) + { + loge("{}", what(ex)); + server.reset(); + } + promise.set_value(std::move(str)); + })); + return std::move(future); + } + + // + // Like spawn(CURL_PATH, args), but for Protocol::h3: QUIC handshakes can hang in ways + // http11/h2 curl invocations don't, so wrap in a hard `timeout 5` safety net. + // + std::future spawn_curl(std::vector args) + { + if (GetParam() == anyhttp::Protocol::h3) + { + args.insert(args.begin(), {"5", CURL_PATH}); + return spawn("/usr/bin/timeout", std::move(args)); + } + return spawn(CURL_PATH, std::move(args)); + } + + any_io_executor strand{make_strand(context.get_executor())}; + std::filesystem::path testFile{"CMakeLists.txt"}; + std::filesystem::path dataFile{"test/data/64kminus1"}; // posted by h2load, one file per request + std::atomic numSpawned = 0; +}; + +using Args = std::vector; + +// ================================================================================================= + +// plain-text only, so no HTTP/3 +INSTANTIATE_TEST_SUITE_P(External, External, + ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 + anyhttp::Protocol::h2), // HTTP/2 + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(External, curl) +{ + auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); + Args args = {"-sS", "-v", "--data-binary", std::format("@{}", testFile.string()), url}; + + if (GetParam() == anyhttp::Protocol::h2) + args.insert(args.begin(), "--http2-prior-knowledge"); + + auto future = spawn(CURL_PATH, std::move(args)); + run(); + + EXPECT_EQ(future.get().size(), file_size(testFile)); +} + +TEST_P(External, curl_multiple) +{ + auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); + Args args = {"-sS", "-v", "--data-binary", std::format("@{}", testFile.string()), url, url}; + + if (GetParam() == anyhttp::Protocol::h2) + args.insert(args.begin(), "--http2-prior-knowledge"); + + auto future = spawn(CURL_PATH, std::move(args)); + run(); + + EXPECT_EQ(future.get().size(), file_size(testFile) * 2); +} + +// ================================================================================================= + +class ExternalTLS : public External +{ +protected: + std::string curlProtocolParam() + { + switch (GetParam()) + { + case anyhttp::Protocol::http11: + return "--http1.1"; + case anyhttp::Protocol::h2: + return "--http2"; + case anyhttp::Protocol::h3: + return "--http3-only"; + } + } + + // + // Run h2load against /echo, posting the contents of 'dataFile' with every request, and check + // that all of it came back. h2load speaks the protocol of the fixture parameter. + // + void h2load(size_t n, size_t clients, size_t streams) + { + auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); + Args args = {"-d", dataFile.string(), "-n", std::to_string(n), // + "-c", std::to_string(clients), "-m", std::to_string(streams), url}; + + switch (GetParam()) + { + case anyhttp::Protocol::http11: + args.insert(args.begin(), "--h1"); + break; + case anyhttp::Protocol::h3: + args.insert(args.begin(), "--h3"); // h2load negotiates h3 itself, http:// URL is fine + break; + default: + break; // h2load defaults to HTTP/2 + } + + auto future = spawn(H2LOAD_PATH, std::move(args)); + run(); + + const std::string output = future.get(); + std::smatch match; + std::regex regex( + R"((\d+) total, \d+ started, (\d+) done, (\d+) succeeded, (\d+) failed, \d+ errored)"); + ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)) << output; + EXPECT_EQ(std::stoul(match[3].str()), n) << match[1]; + EXPECT_EQ(std::stoul(match[4].str()), 0) << match[1]; + + regex = std::regex(R"(\((\d+)\) data)"); + ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)) << output; + EXPECT_EQ(std::stoul(match[1].str()), n * file_size(dataFile)) << match[1]; + } +}; + +INSTANTIATE_TEST_SUITE_P(ExternalTLS, ExternalTLS, + ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 + anyhttp::Protocol::h2, // HTTP/2 + anyhttp::Protocol::h3), // HTTP/3 (QUIC) + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ExternalTLS, curl) +{ + auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); + // clang-format off + Args args = {curlProtocolParam(), "-sS", "-v", + "--cacert", "pki/out/root.pem", + "--data-binary", std::format("@{}", testFile.string()), + url}; + // clang-format on + + auto future = spawn_curl(std::move(args)); + run(); + + EXPECT_EQ(future.get().size(), file_size(testFile)); +} + +TEST_P(ExternalTLS, curl_many) +{ + std::vector> futures; + futures.reserve(10); + + for (size_t i = 0; i < futures.capacity(); ++i) + { + auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); + // clang-format off + Args args = {curlProtocolParam(), "-sS", "-v", + "--cacert", "pki/out/root.pem", + "--data-binary", std::format("@{}", testFile.string()), + url}; + // clang-format on + + futures.emplace_back(spawn_curl(std::move(args))); + } + + run(); + + for (auto& future : futures) + EXPECT_EQ(future.get().size(), file_size(testFile)); +} + +TEST_P(ExternalTLS, curl_multiple) +{ + auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); + // clang-format off + Args args = {curlProtocolParam(), "-sS", "-v", + "--cacert", "pki/out/root.pem", + "--data-binary", std::format("@{}", testFile.string()), + url, url, url, url}; + // clang-format on + + auto future = spawn_curl(std::move(args)); + run(); + + EXPECT_EQ(future.get().size(), file_size(testFile) * 4); +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(ExternalTLS, h2load) { h2load(100, 4, 3); } + +// ================================================================================================= + +// +// Same as ExternalTLS, but with the io_context run on multiple threads, so every connection gets +// its own strand. For HTTP/3 this is the regression test for concurrent access to a single +// ngtcp2_conn, which used to crash right away. +// +class ExternalTLSThreaded : public ExternalTLS +{ +protected: + size_t threads() const override { return 8; } +}; + +INSTANTIATE_TEST_SUITE_P(ExternalTLSThreaded, ExternalTLSThreaded, + ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 + anyhttp::Protocol::h2, // HTTP/2 + anyhttp::Protocol::h3), // HTTP/3 (QUIC) + NameGenerator); + +TEST_P(ExternalTLSThreaded, h2load) { h2load(1000, 8, 5); } + +// ================================================================================================= + +// +// Non-parametrized fixture for external tests that are tied to a specific protocol. +// +class ExternalCustom : public External +{ +}; + +// ------------------------------------------------------------------------------------------------- + +TEST_F(ExternalCustom, netcat_crazy_chunked) +{ + auto cmd = + std::format("nc 127.0.0.2 {} local_endpoint().port()); + auto future = spawn("/usr/bin/bash", {"-c", cmd}); + run(); + + auto out = future.get(); + EXPECT_GT(out.size(), 0); + EXPECT_TRUE(out.contains("Hello, World!\n")); +} + +TEST_F(ExternalCustom, nghttp2) +{ + auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); + auto future = spawn(NGHTTP_PATH, {"-d", testFile.string(), url}); + run(); + + EXPECT_EQ(future.get().size(), file_size(testFile)); +} + +TEST_F(ExternalCustom, h2spec) +{ + auto future = spawn("bin/h2spec", {"--host", server->local_endpoint().address().to_string(), + "--port", std::to_string(server->local_endpoint().port()), + "--path", "/h2spec", "--timeout", "1", "--verbose"}); + run(); + + const std::string output = future.get(); + + std::smatch match; + std::regex regex(R"(((\d+) tests, (\d+) passed, (\d+) skipped, (\d+) failed))"); + ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)); + EXPECT_EQ(std::stoi(match[2].str()), 146) << match[1]; + + // https://github.com/nghttp2/nghttp2/issues/2278 + // https://github.com/nghttp2/nghttp2/issues/2365 + const int expected_ok = std::invoke([] + { + if (NGHTTP2_VERSION_NUM >= 0x004200) // 1.66 + return 138; // 6.9.1 + else if (NGHTTP2_VERSION_NUM == 0x004100) // 1.65 + return 139; + else + return 145; + }); + EXPECT_EQ(std::stoi(match[3].str()), expected_ok) << output; +} + +// +// curl --http2 with an http:// URL asks for an upgrade to h2c. The first request is upgraded, the +// second one is sent as an HTTP/2 stream on the same connection. +// +TEST_F(ExternalCustom, curl_h2c_upgrade) +{ + auto url = std::format("http://127.0.0.2:{}/dump", server->local_endpoint().port()); + // clang-format off + Args args = {"-sS", "-v", "--http2", + "-w", "%{http_code} HTTP/%{http_version}\n", + url + "?first", url + "?second"}; + // clang-format on + auto future = spawn(CURL_PATH, std::move(args)); + run(); + + const std::string output = future.get(); + EXPECT_THAT(output, testing::HasSubstr("query: first")); + EXPECT_THAT(output, testing::HasSubstr("query: second")); + + std::string_view rest = output; + size_t upgraded = 0; + for (size_t pos; (pos = rest.find("200 HTTP/2\n")) != std::string_view::npos; ++upgraded) + rest.remove_prefix(pos + 1); + EXPECT_EQ(upgraded, 2) << output; +} + +// ================================================================================================= diff --git a/test/test_file_handler.cpp b/test/test_file_handler.cpp new file mode 100644 index 0000000..d6925c6 --- /dev/null +++ b/test/test_file_handler.cpp @@ -0,0 +1,223 @@ +#include "test_fixtures.hpp" + +#include "anyhttp/file_handler.hpp" + +#include +#include + +// ================================================================================================= + +// +// serve_file() mounted on "/custom", serving a directory tree created fresh for each testcase. +// +class FileHandler : public ClientAsync +{ +public: + void SetUp() override + { + base = std::filesystem::temp_directory_path() / + std::format("anyhttp-file-handler-{}", ::getpid()); + root = base / "docroot"; + std::filesystem::remove_all(base); + std::filesystem::create_directories(root / "sub"); + + write(root / "hello.txt", "Hello, File!"); + write(root / "empty.txt", ""); + write(root / "sub" / "nested.txt", "Nested!"); + write(root / "large.bin", std::string(256_k, 'x')); + write(root / "secret.txt", "no peeking"); + write(root / "er.txt", "leaked"); // what "/customer.txt" resolves to without a segment check + std::filesystem::permissions(root / "secret.txt", std::filesystem::perms::none); + std::filesystem::create_symlink(base / "outside.txt", root / "escape.txt"); + + // just outside the root, reachable only by escaping it -- so a missing check shows up as + // content served instead of a 404 + write(base / "outside.txt", "outside"); + + ClientAsync::SetUp(); + + custom = [this](server::Request request, server::Response response) -> awaitable { + co_await serve_file(std::move(request), std::move(response), root, "/custom"); + }; + } + + void TearDown() override + { + ClientAsync::TearDown(); + std::filesystem::remove_all(base); + } + + static void write(const std::filesystem::path& path, std::string_view content) + { + std::ofstream(path, std::ios::binary).write(content.data(), content.size()); + } + + // + // Requests \p target and returns status code and body. The request is finished right away -- + // serve_file() ignores the request body, but still has to consume it. + // + awaitable> get(Session& session, boost::urls::url target) + { + auto request = co_await session.async_submit(target, {}); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + auto body = co_await read(response); + co_return std::make_tuple(response.status_code(), std::move(body)); + } + + awaitable> get(Session& session, std::string_view path) + { + co_return co_await get(session, boost::urls::url(url).set_path(path)); + } + + /// Target with a percent-encoded path, passed to the server as-is. + boost::urls::url encoded(std::string_view path) const + { + auto target = boost::urls::url(url); + target.set_encoded_path(path); + return target; + } + +protected: + std::filesystem::path base; ///< holds the docroot and the file just outside of it + std::filesystem::path root; ///< what serve_file() is mounted on +}; + +// ------------------------------------------------------------------------------------------------- + +INSTANTIATE_TEST_SUITE_P(FileHandler, FileHandler, + ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ================================================================================================= + +TEST_P(FileHandler, WHEN_file_exists_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/hello.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Hello, File!"); + }; +} + +TEST_P(FileHandler, WHEN_file_is_in_subdirectory_THEN_serves_content) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/sub/nested.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, "Nested!"); + }; +} + +// +// An empty file cannot be mmap()ed at all, and must not send the empty buffer that already means +// EOF twice. +// +TEST_P(FileHandler, WHEN_file_is_empty_THEN_serves_empty_body) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/empty.txt"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, ""); + }; +} + +// +// Large enough that a single async_write() spans many QUIC packets, so the HTTP/3 write path has +// to keep handing out the caller's buffer across several write_pkt() rounds. +// +TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/large.bin"); + EXPECT_EQ(status, 200); + EXPECT_EQ(body, std::string(256_k, 'x')); + }; +} + +TEST_P(FileHandler, WHEN_file_does_not_exist_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/missing.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + }; +} + +// +// A directory can be open()ed but not mapped, and we do not serve listings. +// +TEST_P(FileHandler, WHEN_path_is_a_directory_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/")), 404); + }; +} + +TEST_P(FileHandler, WHEN_path_escapes_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub/../../outside.txt")), 404); + EXPECT_EQ(std::get<0>(co_await get(session, encoded("/custom/%2e%2e/outside.txt"))), 404); + }; +} + +// +// weakly_canonical() resolves the link, so a link out of the root is caught like any other escape. +// +TEST_P(FileHandler, WHEN_symlink_points_outside_the_root_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<0>(co_await get(session, "/custom/escape.txt")), 404); + }; +} + +// +// The mount prefix must match whole path segments, not just any leading characters: stripping +// "/custom" off "/customer.txt" would otherwise serve "er.txt" out of the docroot. +// +TEST_P(FileHandler, WHEN_prefix_matches_mid_segment_THEN_error_404) +{ + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/customer.txt"); + EXPECT_EQ(status, 404); + EXPECT_EQ(body, ""); + EXPECT_EQ(std::get<0>(co_await get(session, "/customer/hello.txt")), 404); + }; +} + +TEST_P(FileHandler, WHEN_file_is_not_readable_THEN_error_403) +{ + if (::geteuid() == 0) + GTEST_SKIP() << "running as root, permissions do not apply"; + + test = [this](Session session) -> awaitable + { + auto [status, body] = co_await get(session, "/custom/secret.txt"); + EXPECT_EQ(status, 403); + EXPECT_EQ(body, ""); + }; +} + +TEST_P(FileHandler, WHEN_same_file_is_requested_twice_THEN_serves_it_twice) +{ + test = [this](Session session) -> awaitable + { + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); + }; +} + +// ================================================================================================= diff --git a/test/test_fixtures.hpp b/test/test_fixtures.hpp new file mode 100644 index 0000000..8b2e70d --- /dev/null +++ b/test/test_fixtures.hpp @@ -0,0 +1,253 @@ +#pragma once + +// +// Fixtures and helpers shared by the test_*.cpp files. +// +#include "anyhttp/client.hpp" +#include "anyhttp/formatter.hpp" // IWYU pragma: keep +#include "anyhttp/request_handlers.hpp" +#include "anyhttp/server.hpp" +#include "anyhttp/session.hpp" +#include "anyhttp/utils.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +#include +#include +#include + +#include + +#include + +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +using namespace std::string_view_literals; +using namespace std::chrono_literals; + +namespace asio = boost::asio; +using namespace asio; +using namespace asio::experimental::awaitable_operators; +using tcp = ip::tcp; + +namespace rv = std::ranges::views; + +using namespace anyhttp; + +// ================================================================================================= + +/// Returns HTTP11 or HTTP/2 depending on the protocol. +static std::string NameGenerator(const testing::TestParamInfo& info) +{ + return to_string(info.param); +} + +static void setupLogging() +{ +#if defined(GITHUB_ACTIONS) + spdlog::set_level(spdlog::level::warn); +#elif defined(NDEBUG) + spdlog::set_level(spdlog::level::info); +#else + spdlog::set_level(spdlog::level::debug); +#endif +} + +// ================================================================================================= + +// #define MULTITHREADED + +// +// Server fixture with some default request handlers. +// +// Although the server itself supports all protocols at runtime, this is a parametrized fixture +// for use by the clients. +// +class Server : public testing::TestWithParam +{ +protected: + // + // Number of threads run() will run the io_context on. More than one makes the server put + // every connection on its own strand, see below. + // + virtual size_t threads() const + { +#if defined(MULTITHREADED) + return std::max(2u, std::thread::hardware_concurrency()); +#else + return 1; +#endif + } + + void SetUp() override + { + setupLogging(); + + auto config = server::Config{.listen_address = "127.0.0.2", .port = 0}; + config.use_strand = threads() > 1; + + // + // The main server acceptor loop does not need to run on a strand. Instead, a per-connection + // strand is created after accepting a new connection. + // + server.emplace(context.get_executor(), config); + server->setRequestHandler( + [this](server::Request request, server::Response response) -> awaitable + { + logd("{} ({})", request.url().path(), request.url().buffer()); + + if (auto delay = request.get_param_as("delay")) + co_await sleep(std::chrono::milliseconds{*delay}); + + if (request.url().path() == "/echo") + co_await echo(std::move(request), std::move(response)); + else if (request.url().path() == "/eat_request") + co_await eat_request(std::move(request), std::move(response)); + else if (request.url().path() == "/discard") + co_return; + else if (request.url().path() == "/h2spec") + co_await h2spec(std::move(request), std::move(response)); + else if (request.url().path() == "/dump") + co_await dump(std::move(request), std::move(response)); + else if (request.url().path() == "/dump space") + co_await dump(std::move(request), std::move(response)); + else if (request.url().path() == "/detach") + co_await detach(std::move(request), std::move(response)); + else if (request.url().path().starts_with("/custom")) + co_await custom(std::move(request), std::move(response)); + else + co_await not_found(std::move(request), std::move(response)); + }); + } + + void run() + { + const size_t n = threads(); + if (n <= 1) + { + ::run(context); + return; + } + + // + // The extra threads use context.run() directly: the per-operation logging of ::run() is + // meant for single-threaded debugging and would just interleave into noise here. + // + auto pool = rv::iota(size_t{1}, n) | rv::transform([this](size_t) { + return std::jthread([this] { context.run(); }); + }) | std::ranges::to(); + + context.run(); + } + +protected: + boost::asio::io_context context; + std::optional server; + std::function(server::Request request, server::Response response)> custom; +}; + +// ================================================================================================= + +class Client : public Server +{ +protected: + void SetUp() override + { + Server::SetUp(); + url.set_port_number(server->local_endpoint().port()); + client::Config config{.url = url, .protocol = GetParam()}; +#if defined(MULTITHREADED) + client.emplace(make_strand(context.get_executor()), config); +#else + client.emplace(context.get_executor(), config); +#endif + } + +protected: + boost::urls::url url{"http://127.0.0.2/custom"}; + std::optional client; +}; + +// ------------------------------------------------------------------------------------------------- + +class ClientAsync : public Client +{ +public: + auto token() + { + return [this](const std::exception_ptr& ep) + { + auto ec = code(ep); + if (ec) + logw("client completed with \x1b[1;31m{}\x1b[0m", what(ec)); + else + logi("client completed successfully"); + + on_complete(ec); + + logd("stopping server"); + server.reset(); + work.reset(); + }; + } + + MOCK_METHOD(void, on_complete, (boost::system::error_code ec), ()); + static constexpr auto Success = boost::system::error_code{}; + + void SetUp() override + { + Client::SetUp(); + + // + // Spawn the testcase coroutine on the client's executor so that access to it is serialized. + // + co_spawn(client->get_executor(), [this]() -> awaitable + { + if (test) + { + auto session = co_await client->async_connect(); + co_await test(std::move(session)); + } + }, token()); + } + + void TearDown() override + { + EXPECT_CALL(*this, on_complete(boost::system::error_code{})); + run(); + } + +public: + decltype(boost::asio::make_work_guard(context)) work = boost::asio::make_work_guard(context); + std::function(Session session)> test; +}; + +// ================================================================================================= diff --git a/test/test_formatter.cpp b/test/test_formatter.cpp index 82121c8..1140a83 100644 --- a/test/test_formatter.cpp +++ b/test/test_formatter.cpp @@ -1,4 +1,3 @@ -#include #include #include // the nghttp2_nv formatter lives with the rest of the h2 glue @@ -11,6 +10,8 @@ #include #include +#include + // ================================================================================================= // Test thread_id formatter // ================================================================================================= diff --git a/test/test_h2c_upgrade.cpp b/test/test_h2c_upgrade.cpp new file mode 100644 index 0000000..dd4ba3f --- /dev/null +++ b/test/test_h2c_upgrade.cpp @@ -0,0 +1,295 @@ +#include "test_fixtures.hpp" + +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include + +// ================================================================================================= + +// +// Upgrade from HTTP/1.1 to cleartext HTTP/2 (RFC 7540, section 3.2). The HTTP/2 side of the client +// is a bare nghttp2 session driven by hand, so that the test is in control of the handshake. +// +class H2CUpgrade : public Server +{ +protected: + struct Response + { + unsigned status = 0; + std::string body; + bool closed = false; + }; + + using Responses = std::map; + using Request = boost::beast::http::request; + using Http11Response = boost::beast::http::response; + + static std::string base64url(std::span data) + { + namespace base64 = boost::beast::detail::base64; + std::string result(base64::encoded_size(data.size()), '\0'); + result.resize(base64::encode(result.data(), data.data(), data.size())); + std::ranges::replace(result, '+', '-'); + std::ranges::replace(result, '/', '_'); + while (result.ends_with('=')) + result.pop_back(); + return result; + } + + static nghttp2_nv nv(std::string_view name, std::string_view value) + { + return {const_cast(reinterpret_cast(name.data())), + const_cast(reinterpret_cast(value.data())), name.size(), + value.size(), NGHTTP2_NV_FLAG_NONE}; + } + + /// Upgrades a GET for the first target and sends GETs for the others as HTTP/2 streams. + awaitable upgrade(std::vector targets) + { + namespace http = boost::beast::http; + + tcp::socket socket(co_await this_coro::executor); + co_await socket.async_connect(server->local_endpoint()); + const auto authority = std::format("127.0.0.2:{}", server->local_endpoint().port()); + + Responses responses; + auto callbacks = std::invoke([] + { + nghttp2_session_callbacks* cbs; + nghttp2_session_callbacks_new(&cbs); + nghttp2_session_callbacks_set_on_header_callback( + cbs, + [](nghttp2_session*, const nghttp2_frame* frame, const uint8_t* name, size_t namelen, + const uint8_t* value, size_t valuelen, uint8_t, void* user_data) -> int + { + auto& responses = *static_cast(user_data); + if (std::string_view(reinterpret_cast(name), namelen) == ":status") + responses[frame->hd.stream_id].status = + std::stoul(std::string(reinterpret_cast(value), valuelen)); + return 0; + }); + nghttp2_session_callbacks_set_on_data_chunk_recv_callback( + cbs, + [](nghttp2_session*, uint8_t, int32_t stream_id, const uint8_t* data, size_t len, + void* user_data) -> int + { + auto& responses = *static_cast(user_data); + responses[stream_id].body.append(reinterpret_cast(data), len); + return 0; + }); + nghttp2_session_callbacks_set_on_stream_close_callback( + cbs, [](nghttp2_session*, int32_t stream_id, uint32_t, void* user_data) -> int + { + static_cast(user_data)->operator[](stream_id).closed = true; + return 0; + }); + return std::unique_ptr( + cbs, nghttp2_session_callbacks_del); + }); + + nghttp2_session* session; + nghttp2_session_client_new(&session, callbacks.get(), &responses); + boost::scope::scope_exit deleter([&] { nghttp2_session_del(session); }); + + // + // HTTP/1.1 request asking for the upgrade + // + std::array iv{{{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 100}}}; + std::array settings; + auto len = + nghttp2_pack_settings_payload2(settings.data(), settings.size(), iv.data(), iv.size()); + EXPECT_GT(len, 0); + + http::request request{http::verb::get, targets.front(), 11}; + request.set(http::field::host, authority); + request.set(http::field::connection, "Upgrade, HTTP2-Settings"); + request.set(http::field::upgrade, "h2c"); + request.set("HTTP2-Settings", base64url({settings.data(), size_t(len)})); + co_await http::async_write(socket, request); + + boost::beast::flat_buffer buffer; + http::response_parser parser; + co_await http::async_read_header(socket, buffer, parser); + EXPECT_EQ(parser.get().result(), http::status::switching_protocols); + if (parser.get().result() != http::status::switching_protocols) + co_return responses; + + // + // From here on, it's HTTP/2: the upgraded request continues as stream 1. + // + auto result = nghttp2_session_upgrade2(session, settings.data(), len, 0, nullptr); + EXPECT_EQ(result, 0) << nghttp2_strerror(result); + if (result) + co_return responses; + + nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, iv.data(), iv.size()); + for (auto& target : targets | rv::drop(1)) + { + std::array nva{nv(":method", "GET"), nv(":scheme", "http"), nv(":authority", authority), + nv(":path", target)}; + auto id = + nghttp2_submit_request2(session, nullptr, nva.data(), nva.size(), nullptr, nullptr); + EXPECT_GT(id, 0) << nghttp2_strerror(id); + } + + auto recv = [&](const_buffer data) + { + auto n = nghttp2_session_mem_recv2(session, static_cast(data.data()), + data.size()); + EXPECT_EQ(n, data.size()) << nghttp2_strerror(n); + }; + + auto done = [&] + { + return std::ranges::count_if(responses, [](auto& item) { return item.second.closed; }) == + targets.size(); + }; + + std::string out; // nghttp2 starts with the client magic by itself + auto send = [&]() -> awaitable + { + const uint8_t* data; + while (auto n = nghttp2_session_mem_send2(session, &data)) + { + EXPECT_GT(n, 0) << nghttp2_strerror(n); + if (n < 0) + break; + out.append(reinterpret_cast(data), n); + } + if (!out.empty()) + co_await asio::async_write(socket, asio::buffer(out)); + out.clear(); + }; + + recv(buffer.data()); // what came along with the 101 response + std::array data; + for (co_await send(); !done(); co_await send()) + { + auto [ec, n] = co_await socket.async_read_some(asio::buffer(data), as_tuple); + EXPECT_FALSE(ec) << ec.message(); + if (ec) + break; + recv(asio::buffer(data, n)); + } + + nghttp2_session_terminate_session(session, NGHTTP2_NO_ERROR); + co_await send(); + boost::system::error_code ignored; // the server may have closed the connection already + socket.shutdown(tcp::socket::shutdown_send, ignored); + co_return responses; + } + + /// Sends a single HTTP/1.1 request and reads the response. + awaitable http11(Request request) + { + namespace http = boost::beast::http; + + tcp::socket socket(co_await this_coro::executor); + co_await socket.async_connect(server->local_endpoint()); + + request.set(http::field::host, std::format("127.0.0.2:{}", server->local_endpoint().port())); + request.prepare_payload(); + co_await http::async_write(socket, request); + + boost::beast::flat_buffer buffer; + Http11Response response; + co_await http::async_read(socket, buffer, response); + boost::system::error_code ignored; // the server may have closed the connection already + socket.shutdown(tcp::socket::shutdown_send, ignored); + co_return response; + } + + /// Runs \p task to completion, stops the server and returns the result. + template + T run(awaitable task) + { + T result; + co_spawn(context, std::move(task), [&](const std::exception_ptr& ep, T value) + { + if (ep) + ADD_FAILURE() << what(ep); + result = std::move(value); + server.reset(); + }); + Server::run(); + return result; + } + + static Request upgrade_request(boost::beast::http::verb method, std::string target) + { + namespace http = boost::beast::http; + Request request{method, target, 11}; + request.set(http::field::connection, "Upgrade, HTTP2-Settings"); + request.set(http::field::upgrade, "h2c"); + request.set("HTTP2-Settings", "AAMAAABkAAQAAQAAAAIAAAAA"); // as sent by curl + return request; + } +}; + +// ------------------------------------------------------------------------------------------------- + +TEST_F(H2CUpgrade, WHEN_upgrade_is_requested_THEN_request_continues_as_stream_1) +{ + auto responses = run(upgrade({"/dump?first"})); + + ASSERT_EQ(responses.size(), 1); + ASSERT_TRUE(responses.contains(1)); + EXPECT_EQ(responses[1].status, 200); + EXPECT_TRUE(responses[1].closed); + EXPECT_THAT(responses[1].body, testing::HasSubstr("path: /dump")); + EXPECT_THAT(responses[1].body, testing::HasSubstr("query: first")); +} + +TEST_F(H2CUpgrade, WHEN_upgraded_THEN_connection_takes_more_streams) +{ + auto responses = run(upgrade({"/dump?first", "/dump?second", "/unknown"})); + + ASSERT_EQ(responses.size(), 3); + EXPECT_EQ(responses[1].status, 200); + EXPECT_THAT(responses[1].body, testing::HasSubstr("query: first")); + EXPECT_EQ(responses[3].status, 200); + EXPECT_THAT(responses[3].body, testing::HasSubstr("query: second")); + EXPECT_EQ(responses[5].status, 404); +} + +TEST_F(H2CUpgrade, WHEN_request_has_body_THEN_is_served_as_http11) +{ + auto request = upgrade_request(boost::beast::http::verb::post, "/echo"); + request.body() = "Hello, World!"; + auto response = run(http11(std::move(request))); + + EXPECT_EQ(response.result_int(), 200); + EXPECT_EQ(response.body(), "Hello, World!"); +} + +TEST_F(H2CUpgrade, WHEN_http2_settings_are_missing_THEN_is_served_as_http11) +{ + auto request = upgrade_request(boost::beast::http::verb::get, "/dump?no-settings"); + request.erase("HTTP2-Settings"); + auto response = run(http11(std::move(request))); + + EXPECT_EQ(response.result_int(), 200); + EXPECT_THAT(response.body(), testing::HasSubstr("query: no-settings")); +} + +TEST_F(H2CUpgrade, WHEN_http2_settings_are_invalid_THEN_is_served_as_http11) +{ + auto request = upgrade_request(boost::beast::http::verb::get, "/dump?invalid"); + request.set("HTTP2-Settings", "AAMAAABkAA"); // 7 bytes, not a multiple of 6 + auto response = run(http11(std::move(request))); + + EXPECT_EQ(response.result_int(), 200); + EXPECT_THAT(response.body(), testing::HasSubstr("query: invalid")); +} + +// ================================================================================================= diff --git a/test/test_server.cpp b/test/test_server.cpp index 89657d2..6a0828d 100644 --- a/test/test_server.cpp +++ b/test/test_server.cpp @@ -1,258 +1,13 @@ -#include "anyhttp/client.hpp" -#include "anyhttp/file_handler.hpp" -#include "anyhttp/formatter.hpp" // IWYU pragma: keep -#include "anyhttp/literals.hpp" -#include "anyhttp/request_handlers.hpp" -#include "anyhttp/server.hpp" -#include "anyhttp/session.hpp" -#include "anyhttp/utils.hpp" - -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include -#include - -#include -#include - -#include - -#include -#include -#include - -#include -#include -#include -#include - -#include - -#include - -#include - -#include -#include - -#include - -#include -#include -#include +#include "test_fixtures.hpp" +#include #include +#include #include -#include -#include -#include -#include #include -using namespace std::string_view_literals; -using namespace std::chrono_literals; -namespace bp = boost::process::v2; - -namespace asio = boost::asio; -using namespace asio; -using namespace asio::experimental::awaitable_operators; -using tcp = ip::tcp; - -namespace rv = std::ranges::views; - -using namespace anyhttp; - -// https://github.com/curl/curl/issues/10634 --> use custom built curl -#define CURL_PATH "/usr/local/bin/curl" -#define NGHTTP_PATH "/usr/local/bin/nghttp" -#define H2LOAD_PATH "/usr/local/bin/h2load" - -// ================================================================================================= - -/// Returns HTTP11 or HTTP/2 depending on the protocol. -static std::string NameGenerator(const testing::TestParamInfo& info) -{ - return to_string(info.param); -}; - -static void setupLogging() -{ -#if defined(GITHUB_ACTIONS) - spdlog::set_level(spdlog::level::warn); -#elif defined(NDEBUG) - spdlog::set_level(spdlog::level::info); -#else - spdlog::set_level(spdlog::level::debug); -#endif -} - -// ================================================================================================= - -class ClientConnect : public testing::Test -{ -public: - void SetUp() override { setupLogging(); } -}; - -TEST_F(ClientConnect, WHEN_unknown_host_THEN_completes_with_host_not_found_eventually) -{ - boost::asio::io_context context; - client::Config config{.url = boost::urls::url("http://this-domain-does-not-exist:12345")}; - client::Client client(context.get_executor(), config); - client.async_connect([this](boost::system::error_code ec, Session session) - { - loge("ERROR: {}", ec.message()); - EXPECT_TRUE(ec == boost::asio::error::netdb_errors::host_not_found || - ec == boost::asio::error::netdb_errors::host_not_found_try_again); - }); - context.run(); -} - -TEST_F(ClientConnect, WHEN_wrong_port_THEN_completes_with_host_not_found_eventually) -{ - boost::asio::io_context context; - auto port = get_unused_port(context); - client::Config config{.url = boost::urls::url("http://localhost").set_port_number(port)}; - client::Client client(context.get_executor(), config); - client.async_connect([this](boost::system::error_code ec, Session session) - { - loge("ERROR: {}", ec.message()); - EXPECT_EQ(ec, boost::system::errc::connection_refused); - }); - context.run(); -} - -TEST_F(ClientConnect, WHEN_async_connect_is_cancelled_THEN_returns_operation_aborted) -{ - boost::asio::io_context context; - client::Config config{.url = boost::urls::url("http://localhost:12345")}; - client::Client client(context.get_executor(), config); - client.async_connect(cancel_after(0ms, [this](boost::system::error_code ec, Session session) - { - loge("ERROR: {}", ec.message()); - EXPECT_EQ(ec, boost::system::errc::operation_canceled); - })); - - context.run(); -} - -TEST_F(ClientConnect, WHEN_connect_to_broadcast_ip_THEN_completes_with_network_unreachable) -{ - boost::asio::io_context context; - client::Config config{.url = boost::urls::url("http://255.255.255.255:12345")}; - client::Client client(context.get_executor(), config); - client.async_connect([this](boost::system::error_code ec, Session session) - { - loge("ERROR: {}", ec.message()); - EXPECT_EQ(ec, boost::system::errc::network_unreachable); - }); - context.run(); -} - // ================================================================================================= -// #define MULTITHREADED - -// -// Server fixture with some default request handlers. -// -// Although the server itself supports all protocols at runtime, this is a parametrized fixture -// for use by the clients. -// -class Server : public testing::TestWithParam -{ -protected: - // - // Number of threads run() will run the io_context on. More than one makes the server put - // every connection on its own strand, see below. - // - virtual size_t threads() const - { -#if defined(MULTITHREADED) - return std::max(2u, std::thread::hardware_concurrency()); -#else - return 1; -#endif - } - - void SetUp() override - { - setupLogging(); - - auto config = server::Config{.listen_address = "127.0.0.2", .port = 0}; - config.use_strand = threads() > 1; - - // - // The main server acceptor loop does not need to run on a strand. Instead, a per-connection - // strand is created after accepting a new connection. - // - server.emplace(context.get_executor(), config); - server->setRequestHandler( - [this](server::Request request, server::Response response) -> awaitable - { - logd("{} ({})", request.url().path(), request.url().buffer()); - - if (auto delay = request.get_param_as("delay")) - co_await sleep(std::chrono::milliseconds{*delay}); - - if (request.url().path() == "/echo") - co_await echo(std::move(request), std::move(response)); - else if (request.url().path() == "/eat_request") - co_await eat_request(std::move(request), std::move(response)); - else if (request.url().path() == "/discard") - co_return; - else if (request.url().path() == "/h2spec") - co_await h2spec(std::move(request), std::move(response)); - else if (request.url().path() == "/dump") - co_await dump(std::move(request), std::move(response)); - else if (request.url().path() == "/dump space") - co_await dump(std::move(request), std::move(response)); - else if (request.url().path() == "/detach") - co_await detach(std::move(request), std::move(response)); - else if (request.url().path().starts_with("/custom")) - co_await custom(std::move(request), std::move(response)); - else - co_await not_found(std::move(request), std::move(response)); - }); - } - - void run() - { - const size_t n = threads(); - if (n <= 1) - { - ::run(context); - return; - } - - // - // The extra threads use context.run() directly: the per-operation logging of ::run() is - // meant for single-threaded debugging and would just interleave into noise here. - // - auto pool = rv::iota(size_t{1}, n) | rv::transform([this](size_t) { - return std::jthread([this] { context.run(); }); - }) | std::ranges::to(); - - context.run(); - } - -protected: - boost::asio::io_context context; - std::optional server; - std::function(server::Request request, server::Response response)> custom; -}; - INSTANTIATE_TEST_SUITE_P(Server, Server, ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2), NameGenerator); @@ -274,1738 +29,6 @@ TEST_P(Server, Stop) // ================================================================================================= -class External : public Server -{ -protected: - auto split_lines(std::string_view lines) - { - if (lines.ends_with('\n')) - lines.remove_suffix(1); - - return lines | std::views::split('\n') | - std::views::transform([](auto range) { return std::string_view(range); }); - } - - awaitable log(std::string prefix, readable_pipe& pipe) - { - std::string buffer; - auto print = [&](std::string_view line) - { - if (line.ends_with('\r')) - line.remove_suffix(1); - - // print trailing '…' if there is more data in the buffer after this line - const auto continuation = (line.size() + 1 == buffer.size()) ? "" : "…"; - std::println("{}: \x1b[32m{}\x1b[0m{}", prefix, line, continuation); - }; - - auto cs = co_await this_coro::cancellation_state; - try - { - for (;;) - { - auto n = co_await async_read_until(pipe, dynamic_buffer(buffer), '\n'); - for (;;) - { - print(std::string_view(buffer).substr(0, n - 1)); - buffer.erase(0, n); - - // try to bundle multiple lines, looks nicer in debug output - auto pos = buffer.find('\n'); - if (pos == std::string::npos) - break; - n = pos + 1; - } - } - } - catch (const boost::system::system_error& ec) - { - std::println("{}: {}", prefix, ec.code().message()); - if (cs.cancelled() != cancellation_type::none) - std::println("{}: CANCELLED ({})", prefix, cs.cancelled()); - - for (auto line : split_lines(buffer)) - print(line); - - if (ec.code() == error::eof) - co_return; - - throw; - } - } - - awaitable read_all(readable_pipe pipe) - { - std::string result; - auto [ec, nread] = co_await asio::async_read(pipe, asio::dynamic_buffer(result), as_tuple); - logi("STDOUT: {} bytes ({})", nread, what(ec)); - if (ec && ec != error::eof) - throw boost::system::system_error(ec); - co_return result; - } - - awaitable spawn_process(std::filesystem::path path, std::vector args) - { - logi("spawn: {} {}", path.generic_string(), boost::algorithm::join(args, " ")); - - auto ex = co_await this_coro::executor; - readable_pipe out(ex), err(ex); - bp::process child(ex, path, args, bp::process_stdio{.out = out, .err = err}); - // bp::process_environment{{"LD_LIBRARY_PATH=/usr/local/lib"}}); - - logi("spawn: starting to communicate..."); -#if 1 - auto result = co_await (log("STDERR", err) && read_all(std::move(out))); -#else - co_await (log("STDERR", err) && log("STDOUT", out)); - auto result = std::string(); -#endif - logi("spawn: starting to communicate... done, read {} bytes", result.size()); - - co_await child.async_wait(); - if (child.exit_code()) - logw("exit_code={}", child.exit_code()); - else - logi("exit_code={}", child.exit_code()); - - if (--numSpawned <= 0) - { - co_await post(server->get_executor()); - logi("all processes exited, stopping server..."); - server.reset(); - logi("all processes exited, stopping server... done"); - } - - co_return result; - } - - std::future spawn(std::filesystem::path path, std::vector args) - { - ++numSpawned; - std::promise promise; - auto future = promise.get_future(); - co_spawn(strand, spawn_process(std::move(path), std::move(args)), - bind_executor(strand, [this, promise = std::move(promise)]( - const std::exception_ptr& ex, std::string str) mutable - { - if (ex) - { - loge("{}", what(ex)); - server.reset(); - } - promise.set_value(std::move(str)); - })); - return std::move(future); - } - - // - // Like spawn(CURL_PATH, args), but for Protocol::h3: QUIC handshakes can hang in ways - // http11/h2 curl invocations don't, so wrap in a hard `timeout 5` safety net. - // - std::future spawn_curl(std::vector args) - { - if (GetParam() == anyhttp::Protocol::h3) - { - args.insert(args.begin(), {"5", CURL_PATH}); - return spawn("/usr/bin/timeout", std::move(args)); - } - return spawn(CURL_PATH, std::move(args)); - } - - any_io_executor strand{make_strand(context.get_executor())}; - std::filesystem::path testFile{"CMakeLists.txt"}; - std::filesystem::path dataFile{"test/data/64kminus1"}; // posted by h2load, one file per request - std::atomic numSpawned = 0; -}; - -using Args = std::vector; - -// ================================================================================================= - -// plain-text only, so no HTTP/3 -INSTANTIATE_TEST_SUITE_P(External, External, - ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 - anyhttp::Protocol::h2), // HTTP/2 - NameGenerator); - -// ------------------------------------------------------------------------------------------------- - -TEST_P(External, curl) -{ - auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); - Args args = {"-sS", "-v", "--data-binary", std::format("@{}", testFile.string()), url}; - - if (GetParam() == anyhttp::Protocol::h2) - args.insert(args.begin(), "--http2-prior-knowledge"); - - auto future = spawn(CURL_PATH, std::move(args)); - run(); - - EXPECT_EQ(future.get().size(), file_size(testFile)); -} - -TEST_P(External, curl_multiple) -{ - auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); - Args args = {"-sS", "-v", "--data-binary", std::format("@{}", testFile.string()), url, url}; - - if (GetParam() == anyhttp::Protocol::h2) - args.insert(args.begin(), "--http2-prior-knowledge"); - - auto future = spawn(CURL_PATH, std::move(args)); - run(); - - EXPECT_EQ(future.get().size(), file_size(testFile) * 2); -} - -// ================================================================================================= - -class ExternalTLS : public External -{ -protected: - std::string curlProtocolParam() - { - switch (GetParam()) - { - case anyhttp::Protocol::http11: - return "--http1.1"; - case anyhttp::Protocol::h2: - return "--http2"; - case anyhttp::Protocol::h3: - return "--http3-only"; - } - } - - // - // Run h2load against /echo, posting the contents of 'dataFile' with every request, and check - // that all of it came back. h2load speaks the protocol of the fixture parameter. - // - void h2load(size_t n, size_t clients, size_t streams) - { - auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); - Args args = {"-d", dataFile.string(), "-n", std::to_string(n), // - "-c", std::to_string(clients), "-m", std::to_string(streams), url}; - - switch (GetParam()) - { - case anyhttp::Protocol::http11: - args.insert(args.begin(), "--h1"); - break; - case anyhttp::Protocol::h3: - args.insert(args.begin(), "--h3"); // h2load negotiates h3 itself, http:// URL is fine - break; - default: - break; // h2load defaults to HTTP/2 - } - - auto future = spawn(H2LOAD_PATH, std::move(args)); - run(); - - const std::string output = future.get(); - std::smatch match; - std::regex regex( - R"((\d+) total, \d+ started, (\d+) done, (\d+) succeeded, (\d+) failed, \d+ errored)"); - ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)) << output; - EXPECT_EQ(std::stoul(match[3].str()), n) << match[1]; - EXPECT_EQ(std::stoul(match[4].str()), 0) << match[1]; - - regex = std::regex(R"(\((\d+)\) data)"); - ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)) << output; - EXPECT_EQ(std::stoul(match[1].str()), n * file_size(dataFile)) << match[1]; - } -}; - -INSTANTIATE_TEST_SUITE_P(ExternalTLS, ExternalTLS, - ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 - anyhttp::Protocol::h2, // HTTP/2 - anyhttp::Protocol::h3), // HTTP/3 (QUIC) - NameGenerator); - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ExternalTLS, curl) -{ - auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); - // clang-format off - Args args = {curlProtocolParam(), "-sS", "-v", - "--cacert", "pki/out/root.pem", - "--data-binary", std::format("@{}", testFile.string()), - url}; - // clang-format off - - auto future = spawn_curl(std::move(args)); - run(); - - EXPECT_EQ(future.get().size(), file_size(testFile)); -} - -TEST_P(ExternalTLS, curl_many) -{ - std::vector> futures; - futures.reserve(10); - - for (size_t i = 0; i < futures.capacity(); ++i) - { - auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); - // clang-format off - Args args = {curlProtocolParam(), "-sS", "-v", - "--cacert", "pki/out/root.pem", - "--data-binary", std::format("@{}", testFile.string()), - url}; - // clang-format off - - futures.emplace_back(spawn_curl(std::move(args))); - } - - run(); - - for (auto& future : futures) - EXPECT_EQ(future.get().size(), file_size(testFile)); -} - -TEST_P(ExternalTLS, curl_multiple) -{ - auto url = std::format("https://127.0.0.2:{}/echo", server->local_endpoint().port()); - // clang-format off - Args args = {curlProtocolParam(), "-sS", "-v", - "--cacert", "pki/out/root.pem", - "--data-binary", std::format("@{}", testFile.string()), - url, url, url, url}; - // clang-format off - - auto future = spawn_curl(std::move(args)); - run(); - - EXPECT_EQ(future.get().size(), file_size(testFile) * 4); -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ExternalTLS, h2load) { h2load(100, 4, 3); } - -// ================================================================================================= - -// -// Same as ExternalTLS, but with the io_context run on multiple threads, so every connection gets -// its own strand. For HTTP/3 this is the regression test for concurrent access to a single -// ngtcp2_conn, which used to crash right away. -// -class ExternalTLSThreaded : public ExternalTLS -{ -protected: - size_t threads() const override { return 8; } -}; - -INSTANTIATE_TEST_SUITE_P(ExternalTLSThreaded, ExternalTLSThreaded, - ::testing::Values(anyhttp::Protocol::http11, // HTTP/1.1 - anyhttp::Protocol::h2, // HTTP/2 - anyhttp::Protocol::h3), // HTTP/3 (QUIC) - NameGenerator); - -TEST_P(ExternalTLSThreaded, h2load) { h2load(1000, 8, 5); } - -// ================================================================================================= - -// -// Non-parametrized fixture for external tests that are tied to a specific protocol. -// -class ExternalCustom : public External -{ -}; - -// ------------------------------------------------------------------------------------------------- - -TEST_F(ExternalCustom, netcat_crazy_chunked) -{ - auto cmd = - std::format("nc 127.0.0.2 {} local_endpoint().port()); - auto future = spawn("/usr/bin/bash", {"-c", cmd}); - run(); - - auto out = future.get(); - EXPECT_GT(out.size(), 0); - EXPECT_TRUE(out.contains("Hello, World!\n")); -} - -TEST_F(ExternalCustom, nghttp2) -{ - auto url = std::format("http://127.0.0.2:{}/echo", server->local_endpoint().port()); - auto future = spawn(NGHTTP_PATH, {"-d", testFile.string(), url}); - run(); - - EXPECT_EQ(future.get().size(), file_size(testFile)); -} - -TEST_F(ExternalCustom, h2spec) -{ - auto future = spawn("bin/h2spec", {"--host", server->local_endpoint().address().to_string(), - "--port", std::to_string(server->local_endpoint().port()), - "--path", "/h2spec", "--timeout", "1", "--verbose"}); - run(); - - const std::string output = future.get(); - - std::smatch match; - std::regex regex(R"(((\d+) tests, (\d+) passed, (\d+) skipped, (\d+) failed))"); - ASSERT_TRUE(std::regex_search(output.begin(), output.end(), match, regex)); - EXPECT_EQ(std::stoi(match[2].str()), 146) << match[1]; - - // https://github.com/nghttp2/nghttp2/issues/2278 - // https://github.com/nghttp2/nghttp2/issues/2365 - const int expected_ok = std::invoke([] - { - if (NGHTTP2_VERSION_NUM >= 0x004200) // 1.66 - return 138; // 6.9.1 - else if (NGHTTP2_VERSION_NUM == 0x004100) // 1.65 - return 139; - else - return 145; - }); - EXPECT_EQ(std::stoi(match[3].str()), expected_ok) << output; -} - -// ================================================================================================= - -class Client : public Server -{ -protected: - void SetUp() override - { - Server::SetUp(); - url.set_port_number(server->local_endpoint().port()); - client::Config config{.url = url, .protocol = GetParam()}; -#if defined(MULTITHREADED) - client.emplace(make_strand(context.get_executor()), config); -#else - client.emplace(context.get_executor(), config); -#endif - } - -protected: - boost::urls::url url{"http://127.0.0.2/custom"}; - std::optional client; -}; - -// ------------------------------------------------------------------------------------------------- - -class ClientAsync : public Client -{ -public: - auto token() - { - return [this](const std::exception_ptr& ep) - { - auto ec = code(ep); - if (ec) - logw("client completed with \x1b[1;31m{}\x1b[0m", what(ec)); - else - logi("client completed successfully"); - - on_complete(ec); - - logd("stopping server"); - server.reset(); - work.reset(); - }; - } - - MOCK_METHOD(void, on_complete, (boost::system::error_code ec), ()); - static constexpr auto Success = boost::system::error_code{}; - - void SetUp() override - { - Client::SetUp(); - - // - // Spawn the testcase coroutine on the client's executor so that access to it is serialized. - // - co_spawn(client->get_executor(), [this]() -> awaitable - { - if (test) - { - auto session = co_await client->async_connect(); - co_await test(std::move(session)); - } - }, token()); - } - - void TearDown() override - { - EXPECT_CALL(*this, on_complete(boost::system::error_code{})); - run(); - } - -public: - decltype(boost::asio::make_work_guard(context)) work = boost::asio::make_work_guard(context); - std::function(Session session)> test; -}; - -INSTANTIATE_TEST_SUITE_P(ClientAsync, ClientAsync, - ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, - anyhttp::Protocol::h3), - NameGenerator); - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, WHEN_post_data_THEN_receive_echo) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - size_t bytes = 1024; - auto count = co_await (generate(request, bytes) && count_response(request)); - EXPECT_EQ(bytes, count); - }; -} - -TEST_P(ClientAsync, WHEN_post_without_path_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path(""), {}); - co_await generate(request, 1024); - auto [ec, response] = co_await request.async_get_response(as_tuple); - EXPECT_TRUE(ec); - }; -} - -TEST_P(ClientAsync, WHEN_post_to_unknown_path_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("unknown"), {}); - co_await generate(request, 1_m); - auto response = co_await request.async_get_response(); - EXPECT_EQ(response.status_code(), 404); - auto received = co_await drain(response); - }; -} - -TEST_P(ClientAsync, WHEN_server_discards_request_THEN_error_500) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("discard"), {}); - co_await generate(request, 1024); - auto [ec, response] = co_await request.async_get_response(as_tuple); - EXPECT_TRUE(ec); - }; -} - -TEST_P(ClientAsync, WHEN_server_discards_request_delayed_THEN_error_500) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("detach"), {}); - co_await generate(request, 1024); - auto [ec, response] = co_await request.async_get_response(as_tuple); - EXPECT_TRUE(ec); - }; -} - -TEST_P(ClientAsync, WHEN_server_discards_request_with_body_delayed_THEN_error_500) -{ - test = [this](Session session) -> awaitable - { - auto executor = co_await this_coro::executor; - auto request = co_await session.async_submit(url.set_path("detach"), {}); - auto [ep] = co_await co_spawn(executor, send(request, rv::iota(uint8_t{0})), as_tuple); - EXPECT_TRUE(ep); - }; -} - -TEST_P(ClientAsync, WHEN_invalid_port_in_host_header_THEN_reports_error) -{ - test = [this](Session session) -> awaitable - { - Fields fields; - fields.set("Host", "host:12345x"); - auto request = co_await session.async_submit(url.set_path("echo"), fields); - auto response = co_await (send_eof(request) && count_response(request)); - }; -} - -TEST_P(ClientAsync, WHEN_get_response_is_called_twice_THEN_reports_error) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo")); - auto [ec, response] = co_await request.async_get_response(as_tuple); - EXPECT_EQ(ec, boost::system::errc::success); - std::tie(ec, response) = co_await request.async_get_response(as_tuple); - EXPECT_EQ(ec, boost::system::errc::connection_already_in_progress); - EXPECT_EQ(ec, asio::error::basic_errors::already_started); - }; -} - -TEST_P(ClientAsync, WHEN_get_response_is_detached_THEN_does_not_crash) -{ - if (GetParam() == anyhttp::Protocol::http11) - GTEST_SKIP(); - - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo")); - request.async_get_response(detached); - }; -} - -TEST_P(ClientAsync, WHEN_server_discards_request_while_writing_THEN_connection_is_reset) -{ - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await sleep(150ms); - request.reset(); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - auto executor = co_await this_coro::executor; - auto [ec] = co_await co_spawn(executor, send(request, rv::iota(uint8_t(0))), as_tuple); - EXPECT_EQ(code(ec), boost::system::errc::connection_reset); - }; -} - -TEST_P(ClientAsync, WHEN_server_discards_request_and_response_THEN_completes_anyway) -{ - // if (GetParam() == anyhttp::Protocol::http11) - // GTEST_SKIP(); // FIXME: timeout - - custom = [this](server::Request request, server::Response response) -> awaitable - { - std::ignore = request; - std::ignore = response; - co_return; - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - auto [ec, _] = co_await request.async_get_response(as_tuple); - EXPECT_EQ(ec, boost::beast::http::error::end_of_stream); - // EXPECT_EQ(ec, std::errc::connection_reset); - }; -} - -TEST_P(ClientAsync, WHEN_client_cancels_write_THEN_can_resume) -{ - if (GetParam() == anyhttp::Protocol::http11) - GTEST_SKIP(); // a chunked body cannot be cancelled correctly --> disconnects - - test = [this](Session session) -> awaitable - { - co_await this_coro::throw_if_cancelled(false); - auto executor = co_await this_coro::executor; - auto request = co_await session.async_submit(url.set_path("echo")); - auto response = co_await request.async_get_response(); - - // send as much data as possible within 1s, should run into backpressure - auto [ep] = co_await co_spawn(executor, send(request, rv::iota(uint8_t(0))), - cancel_after(1s, as_tuple)); - EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); - - if (GetParam() == anyhttp::Protocol::h3) - { - // - // QUIC: whether the FIN can slip out while the send window is closed depends on flow - // control timing, so don't assert either way here. What matters is that ending the - // upload and draining the response together complete the exchange. - // - auto received = co_await (send_eof(request) && drain(response)); - EXPECT_GT(received, 0); - } - else - { - // now, with a closed window, we cannot even end the upload - std::tie(ep) = co_await co_spawn(executor, send_eof(request), cancel_after(1ms, as_tuple)); - EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); - - // as we have no control over when the send window is re-opened, wait for it in parallel - auto received = co_await (send_eof(request) && drain(response)); - EXPECT_GT(received, 0); - } - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, YieldFuzz) -{ -#if 0 - static std::random_device rd; - static std::mt19937 gen(rd()); -#else - static std::mt19937 gen(42); // fixed seed for reproducibility -#endif - - custom = [this](server::Request request, server::Response response) -> awaitable - { - std::uniform_int_distribution<> dist(0, 10); - constexpr auto msg = "Hello, Client!"sv; - co_await yield(dist(gen)); - Fields fields; - fields.set("Content-Length", std::to_string(msg.size())); - co_await response.async_submit(200, fields); - co_await yield(dist(gen)); - co_await response.async_write(asio::buffer(msg)); - co_await yield(dist(gen)); - co_await response.async_write_eof(); - co_await yield(dist(gen)); - std::array data; - co_await request.async_read_some(asio::buffer(data), as_tuple); - }; - test = [this](Session session) -> awaitable - { - std::uniform_int_distribution<> dist(0, 10); - for (size_t i = 0; i < 100; ++i) - { - std::println( - "=== {} =========================================================================", i); - co_await yield(dist(gen)); - Fields fields; - if (GetParam() == anyhttp::Protocol::http11) - fields.set("Connection", "Keep-Alive"); - fields.set("Content-Length", "0"); - auto request = co_await session.async_submit(url, fields); - co_await yield(dist(gen)); - co_await request.async_write_eof(); - co_await yield(dist(gen)); - co_await count_response(request); - } - }; -} - -// -// The end of an incoming body is an error code, not a zero-sized read -- and it keeps being -// reported for every read issued after it. A zero-length buffer, on the other hand, says nothing -// about the body at all: it completes immediately, at the end of a body just as anywhere else. -// -TEST_P(ClientAsync, WHEN_body_ends_THEN_read_reports_eof) -{ - static const auto hello = "Hello, World!"sv; - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await drain(request); - co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); - co_await response.async_write_eof(asio::buffer(hello)); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - - std::string body; - std::array buffer; // small on purpose: several reads before the end - for (;;) - { - auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); - if (ec) - { - EXPECT_EQ(ec, asio::error::eof); - EXPECT_EQ(n, 0u); - break; - } - body.append(buffer.data(), n); - } - EXPECT_EQ(body, hello); - - // - // Reading past the end of a body says the same thing again -- also once the protocol layer - // has torn the underlying stream down in the meantime, which the yield gives it every - // opportunity to do (both sides of the exchange are finished by now). - // - co_await yield(20); - auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), as_tuple); - EXPECT_EQ(ec, asio::error::eof); - - // ... but a zero-length read is not a read, and reports nothing - std::array empty; - std::tie(ec, n) = co_await response.async_read_some(asio::buffer(empty), as_tuple); - EXPECT_FALSE(ec); - EXPECT_EQ(n, 0u); - }; -} - -// -// An empty async_write() no longer ends a body -- async_write_eof() does, and nothing else. So a -// message with an empty write in the middle of it still carries everything written after that. -// -TEST_P(ClientAsync, WHEN_empty_buffer_is_written_THEN_body_stays_open) -{ - static const auto tail = "still here"sv; - custom = [this](server::Request request, server::Response response) -> awaitable - { - EXPECT_EQ(co_await drain(request), 0u); - co_await response.async_submit(200, {}); - co_await response.async_write({}); // writes nothing, leaves the body open - co_await response.async_write_eof(asio::buffer(tail)); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write({}); // likewise: the request body stays open - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - EXPECT_EQ(co_await read(response), tail); - }; -} - -// -// Ending a body twice is harmless -- the second call has nothing left to do -- and an empty -// write stays a free no-op even then. Data after the end is neither: there is no body left for -// it to belong to, through whichever entry point it tries to sneak in. -// -TEST_P(ClientAsync, WHEN_written_after_eof_THEN_reports_broken_pipe) -{ - static const auto hello = "Hello, World!"sv; - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await drain(request); - co_await response.async_submit(200, fields({{"Content-Length", hello.size()}})); - co_await response.async_write_eof(asio::buffer(hello)); - - auto [ec] = co_await response.async_write_eof(as_tuple); - EXPECT_FALSE(ec); - - std::tie(ec) = co_await response.async_write({}, as_tuple); - EXPECT_FALSE(ec); - - std::tie(ec) = co_await response.async_write(asio::buffer(hello), as_tuple); - EXPECT_EQ(ec, boost::system::errc::broken_pipe); - - std::tie(ec) = co_await response.async_write_eof(asio::buffer(hello), as_tuple); - EXPECT_EQ(ec, boost::system::errc::broken_pipe); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - EXPECT_EQ(co_await read(response), hello); - }; -} - -TEST_P(ClientAsync, HelloWorld) -{ - static const auto hello = "Hello, World!"sv; - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await response.async_submit(200, {}); - co_await response.async_write_eof(asio::buffer(hello)); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - auto body = co_await read(response); - EXPECT_EQ(body, hello); - }; -} - -// ------------------------------------------------------------------------------------------------- - -// -// A single async_write() larger than what the transport hands to its peer in one go, i.e. the -// whole body in one call instead of chunk by chunk. HTTP/3 has to keep offering the same buffer -// to nghttp3 across many packets here, and complete the write only once all of it is -// acknowledged. -// -TEST_P(ClientAsync, WHEN_server_writes_large_buffer_at_once_THEN_receives_all) -{ - static const std::vector body = [] - { - std::vector data(256_k); - std::ranges::generate(data, [i = uint8_t(0)]() mutable { return i++; }); - return data; - }(); - - custom = [this](server::Request request, server::Response response) -> awaitable - { - // drain the request -- HTTP/1.1 closes the connection on an unfinished parser - co_await drain(request); - - co_await response.async_submit(200, fields({{"Content-Length", body.size()}})); - co_await response.async_write_eof(asio::buffer(body)); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - EXPECT_EQ(co_await count_response(request), body.size()); - }; -} - -// -// Cancelling an async_write_eof() that carries data. The buffer goes back to the caller the -// moment the handler runs, so the backend must stop referencing it right there -- for HTTP/3's -// zero-copy path that means resetting the stream, exactly as for a cancelled plain write; under -// ASAN this test is what catches a backend that keeps pointing into the freed buffer. -// -TEST_P(ClientAsync, WHEN_server_cancels_write_eof_THEN_client_sees_truncated_body) -{ - static const std::vector body(8_m, 'x'); - - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await drain(request); - co_await response.async_submit(200, {}); - - // - // Far more than the peer's receive window, and the client below doesn't read a byte until - // this is over, so the write is guaranteed to still be in progress when it is cancelled. - // - auto [ec] = co_await response.async_write_eof(asio::buffer(body), - cancel_after(50ms, as_tuple)); - EXPECT_EQ(ec, boost::system::errc::operation_canceled); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - - // leave the body untouched until the cancellation above has hit, see the sibling testcase - asio::steady_timer timer(co_await this_coro::executor, 150ms); - co_await timer.async_wait(deferred); - - boost::system::error_code ec; - auto received = co_await try_receive(response, ec); - EXPECT_LT(received, body.size()); - EXPECT_EQ(ec, boost::beast::http::error::partial_message); - }; -} - -// -// Cancelling an async_write_eof() whose FIN never made it out must not leave the body in limbo: -// the intent to end it is rolled back, and a re-issued async_write_eof() ends the (now shorter) -// body for real -- instead of completing as a no-op while the peer waits forever for the end. -// -TEST_P(ClientAsync, WHEN_client_cancels_write_eof_THEN_can_still_end) -{ - if (GetParam() == anyhttp::Protocol::http11) - GTEST_SKIP(); // a chunked body cannot be cancelled correctly --> disconnects - - static const std::vector body(8_m, 'x'); - - test = [this](Session session) -> awaitable - { - co_await this_coro::throw_if_cancelled(false); - auto executor = co_await this_coro::executor; - auto request = co_await session.async_submit(url.set_path("echo")); - auto response = co_await request.async_get_response(); - - // far more than the send window, with nobody reading the echo yet: this cannot complete - auto write_eof = [&]() -> awaitable - { co_await request.async_write_eof(asio::buffer(body)); }; - auto [ep] = co_await co_spawn(executor, write_eof(), cancel_after(100ms, as_tuple)); - EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); - - // the FIN never went out with the cancelled write, so the body can still be ended - auto received = co_await (send_eof(request) && drain(response)); - EXPECT_GT(received, 0u); - EXPECT_LT(received, body.size()); - }; -} - -// -// Cancelling a response write mid-body. HTTP/3 hands the caller's buffer to nghttp3 by reference, -// so bytes already offered and not yet acknowledged cannot simply be abandoned -- the stream is -// reset instead, which is what the peer would see anyway for a body that stops short of its end. -// -TEST_P(ClientAsync, WHEN_server_cancels_write_THEN_client_sees_truncated_body) -{ - static const std::vector body(8_m, 'x'); - - custom = [this](server::Request request, server::Response response) -> awaitable - { - // drain the request -- HTTP/1.1 closes the connection on an unfinished parser - co_await drain(request); - - co_await response.async_submit(200, {}); - - // - // Far more than the peer's receive window, and the client below doesn't read a byte until - // this is over, so the write is guaranteed to still be in progress when it is cancelled. - // - auto executor = co_await this_coro::executor; - auto [ep] = co_await co_spawn(executor, send(response, std::span(body)), - cancel_after(50ms, as_tuple)); - EXPECT_EQ(code(ep), boost::system::errc::operation_canceled); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - - // - // Leave the response body untouched for now: whatever the server manages to send fills up - // the receive window and stays there, so its write cannot run to completion before the - // cancellation above hits. - // - asio::steady_timer timer(co_await this_coro::executor, 150ms); - co_await timer.async_wait(deferred); - - boost::system::error_code ec; - auto received = co_await try_receive(response, ec); - std::println("received {} of {} bytes ({})", received, body.size(), ec.message()); - EXPECT_LT(received, body.size()); - EXPECT_EQ(ec, boost::beast::http::error::partial_message); - }; -} - -// ================================================================================================= - -// -// serve_file() mounted on "/custom", serving a directory tree created fresh for each testcase. -// -class FileHandler : public ClientAsync -{ -public: - void SetUp() override - { - base = std::filesystem::temp_directory_path() / - std::format("anyhttp-file-handler-{}", ::getpid()); - root = base / "docroot"; - std::filesystem::remove_all(base); - std::filesystem::create_directories(root / "sub"); - - write(root / "hello.txt", "Hello, File!"); - write(root / "empty.txt", ""); - write(root / "sub" / "nested.txt", "Nested!"); - write(root / "large.bin", std::string(256_k, 'x')); - write(root / "secret.txt", "no peeking"); - write(root / "er.txt", "leaked"); // what "/customer.txt" resolves to without a segment check - std::filesystem::permissions(root / "secret.txt", std::filesystem::perms::none); - std::filesystem::create_symlink(base / "outside.txt", root / "escape.txt"); - - // just outside the root, reachable only by escaping it -- so a missing check shows up as - // content served instead of a 404 - write(base / "outside.txt", "outside"); - - ClientAsync::SetUp(); - - custom = [this](server::Request request, server::Response response) -> awaitable { - co_await serve_file(std::move(request), std::move(response), root, "/custom"); - }; - } - - void TearDown() override - { - ClientAsync::TearDown(); - std::filesystem::remove_all(base); - } - - static void write(const std::filesystem::path& path, std::string_view content) - { - std::ofstream(path, std::ios::binary).write(content.data(), content.size()); - } - - // - // Requests \p target and returns status code and body. The request is finished right away -- - // serve_file() ignores the request body, but still has to consume it. - // - awaitable> get(Session& session, boost::urls::url target) - { - auto request = co_await session.async_submit(target, {}); - co_await request.async_write_eof(); - auto response = co_await request.async_get_response(); - auto body = co_await read(response); - co_return std::make_tuple(response.status_code(), std::move(body)); - } - - awaitable> get(Session& session, std::string_view path) - { - co_return co_await get(session, boost::urls::url(url).set_path(path)); - } - - /// Target with a percent-encoded path, passed to the server as-is. - boost::urls::url encoded(std::string_view path) const - { - auto target = boost::urls::url(url); - target.set_encoded_path(path); - return target; - } - -protected: - std::filesystem::path base; ///< holds the docroot and the file just outside of it - std::filesystem::path root; ///< what serve_file() is mounted on -}; - -INSTANTIATE_TEST_SUITE_P(FileHandler, FileHandler, - ::testing::Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, - anyhttp::Protocol::h3), - NameGenerator); - -// ------------------------------------------------------------------------------------------------- - -TEST_P(FileHandler, WHEN_file_exists_THEN_serves_content) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/hello.txt"); - EXPECT_EQ(status, 200); - EXPECT_EQ(body, "Hello, File!"); - }; -} - -TEST_P(FileHandler, WHEN_file_is_in_subdirectory_THEN_serves_content) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/sub/nested.txt"); - EXPECT_EQ(status, 200); - EXPECT_EQ(body, "Nested!"); - }; -} - -// -// An empty file cannot be mmap()ed at all, and must not send the empty buffer that already means -// EOF twice. -// -TEST_P(FileHandler, WHEN_file_is_empty_THEN_serves_empty_body) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/empty.txt"); - EXPECT_EQ(status, 200); - EXPECT_EQ(body, ""); - }; -} - -// -// Large enough that a single async_write() spans many QUIC packets, so the HTTP/3 write path has -// to keep handing out the caller's buffer across several write_pkt() rounds. -// -TEST_P(FileHandler, WHEN_file_is_large_THEN_serves_all_of_it) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/large.bin"); - EXPECT_EQ(status, 200); - EXPECT_EQ(body, std::string(256_k, 'x')); - }; -} - -TEST_P(FileHandler, WHEN_file_does_not_exist_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/missing.txt"); - EXPECT_EQ(status, 404); - EXPECT_EQ(body, ""); - }; -} - -// -// A directory can be open()ed but not mapped, and we do not serve listings. -// -TEST_P(FileHandler, WHEN_path_is_a_directory_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub")), 404); - EXPECT_EQ(std::get<0>(co_await get(session, "/custom/")), 404); - }; -} - -TEST_P(FileHandler, WHEN_path_escapes_the_root_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - EXPECT_EQ(std::get<0>(co_await get(session, "/custom/../outside.txt")), 404); - EXPECT_EQ(std::get<0>(co_await get(session, "/custom/sub/../../outside.txt")), 404); - EXPECT_EQ(std::get<0>(co_await get(session, encoded("/custom/%2e%2e/outside.txt"))), 404); - }; -} - -// -// weakly_canonical() resolves the link, so a link out of the root is caught like any other escape. -// -TEST_P(FileHandler, WHEN_symlink_points_outside_the_root_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - EXPECT_EQ(std::get<0>(co_await get(session, "/custom/escape.txt")), 404); - }; -} - -// -// The mount prefix must match whole path segments, not just any leading characters: stripping -// "/custom" off "/customer.txt" would otherwise serve "er.txt" out of the docroot. -// -TEST_P(FileHandler, WHEN_prefix_matches_mid_segment_THEN_error_404) -{ - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/customer.txt"); - EXPECT_EQ(status, 404); - EXPECT_EQ(body, ""); - EXPECT_EQ(std::get<0>(co_await get(session, "/customer/hello.txt")), 404); - }; -} - -TEST_P(FileHandler, WHEN_file_is_not_readable_THEN_error_403) -{ - if (::geteuid() == 0) - GTEST_SKIP() << "running as root, permissions do not apply"; - - test = [this](Session session) -> awaitable - { - auto [status, body] = co_await get(session, "/custom/secret.txt"); - EXPECT_EQ(status, 403); - EXPECT_EQ(body, ""); - }; -} - -TEST_P(FileHandler, WHEN_same_file_is_requested_twice_THEN_serves_it_twice) -{ - test = [this](Session session) -> awaitable - { - EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); - EXPECT_EQ(std::get<1>(co_await get(session, "/custom/hello.txt")), "Hello, File!"); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, ServerYieldFirst) -{ - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await yield(10); - co_await response.async_submit(200, {}); - co_await yield(10); - co_await response.async_write_eof(); - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url); - co_await request.async_write_eof(); - co_await count_response(request); - }; -} - -// ---------------------------------------------------------------------------------------------- - -static std::optional stackRemainingBytes() -{ - pthread_attr_t attr; - if (pthread_getattr_np(pthread_self(), &attr) != 0) - return std::nullopt; - - void* stack_base = nullptr; - size_t stack_size = 0; - if (pthread_attr_getstack(&attr, &stack_base, &stack_size) != 0) - { - pthread_attr_destroy(&attr); - return std::nullopt; - } - - pthread_attr_destroy(&attr); - - if (stack_base == nullptr || stack_size == 0) - return std::nullopt; - - int local = 0; - std::uintptr_t sp = reinterpret_cast(&local); - std::uintptr_t base = reinterpret_cast(stack_base); - - return (sp >= base) ? std::optional(sp - base) : std::nullopt; -} - -TEST_P(ClientAsync, Recursion) -{ -#if __has_feature(address_sanitizer) - GTEST_SKIP() << "skipped under address sanitizer"; -#endif - if (!stackRemainingBytes()) - GTEST_SKIP() << "unable to measure stack on this platform"; - - test = [this](Session session) -> awaitable - { - auto ex = co_await this_coro::executor; - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - - // verify that immediate completion (here, due to an empty buffer) does not cause recursion - std::array empty; - co_await response.async_read_some(asio::buffer(empty)); - auto s0 = stackRemainingBytes().value(); - co_await response.async_read_some(asio::buffer(empty)); - auto s1 = stackRemainingBytes().value(); - EXPECT_EQ(s0, s1); - - // however, ASIO allows us to control this behavior using "immediate executors" - co_await response.async_read_some(asio::buffer(empty), bind_immediate_executor(ex)); - auto s2 = stackRemainingBytes().value(); - EXPECT_GT(s1, s2); - }; -} - -// ---------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, Custom) -{ - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await response.async_submit(200, {}); - std::array buffer; - for (;;) - { - auto [ec, n] = co_await request.async_read_some(asio::buffer(buffer), as_tuple); - if (ec) - { - co_await response.async_write_eof(); - co_return; - } - co_await response.async_write(asio::buffer(buffer, n)); - } - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url, {}); - constexpr size_t bytes = 1024; - auto count = co_await (generate(request, bytes) && count_response(request)); - EXPECT_EQ(bytes, count); - }; -} - -TEST_P(ClientAsync, IgnoreRequest) -{ - custom = [this](server::Request request, server::Response response) -> awaitable - { - co_await response.async_submit(200, {}); - co_await response.async_write_eof(); - }; - test = [this](Session session) -> awaitable - { - Fields fields; - fields.set("content-length", "0"); - auto request = co_await session.async_submit(url, fields); - auto count = co_await (generate(request, 0) && count_response(request)); - EXPECT_EQ(count, 0); - }; -} - -TEST_P(ClientAsync, IgnoreRequestAndResponse) -{ - custom = [this](server::Request request, server::Response response) -> awaitable - { - std::ignore = request; - std::ignore = response; - co_return; - }; - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url, {}); - auto res = co_await (generate(request, 0) && try_read_response(request)); - EXPECT_FALSE(res.has_value()); - std::println("ERROR: {}", res.error().message()); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, PostRange) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - // co_await request.async_write(asio::buffer("ping"sv)); // FIXME: - auto response = co_await request.async_get_response(); - // std::string s(10_m, 'a'); - // auto sender = send(request, std::string_view("blah")); - // auto sender = send(request, std::string(10_m, 'a')); - auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); - auto received = co_await (std::move(sender) && drain(response)); - loge("received: {}", received); - EXPECT_EQ(received, 1_m); - }; -} - -TEST_P(ClientAsync, PostRangeImmediate) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0)) | rv::take(1_m)); - auto received = co_await (std::move(sender) && count_response(request)); - loge("received: {}", received); - EXPECT_EQ(received, 1_m); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, WHEN_request_is_sent_THEN_response_is_received_before_body_is_posted) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - constexpr size_t bytes = 1024; - co_await generate(request, bytes); - EXPECT_EQ(co_await drain(response), bytes); - }; -} - -// ------------------------------------------------------------------------------------------------- - -// -// HTTP/1.1 supports pipelining in the sense that multiple, full requests can be made before -// the responses are received. -// -// TODO: Any kind of interleaving is not supported. An attempt to issue another request while the -// previous request is still active should result in an error, immediately. -// -TEST_P(ClientAsync, WHEN_multiple_request_are_made_THEN_responses_are_received_in_order) -{ - test = [this](Session session) -> awaitable - { - auto request1 = co_await session.async_submit(url.set_path("echo"), {}); - co_await request1.async_write_eof(asio::buffer("Hello, Server #1!"sv)); - - auto request2 = co_await session.async_submit(url.set_path("echo"), {}); - co_await request2.async_write_eof(asio::buffer("Hello, Server #2! XYZ"sv)); - - auto response1 = co_await request1.async_get_response(); - EXPECT_EQ(co_await drain(response1), 17); - - auto response2 = co_await request2.async_get_response(); - EXPECT_EQ(co_await drain(response2), 21); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, EatRequest) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("eat_request"), {}); - co_await generate(request, 1024); - auto response = co_await request.async_get_response(); - auto received = co_await drain(response); - EXPECT_EQ(received, 0); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, Dump) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit( - url.set_path("dump space").set_params({{"blah", "white space"}, {"x", "y"}}), {}); - co_await send_eof(request); - auto response = co_await request.async_get_response(); - auto dump = co_await read(response); - EXPECT_THAT(dump, testing::HasSubstr("path: /dump space")); - EXPECT_THAT(dump, testing::HasSubstr(" blah=white space")); - }; -} - -// ------------------------------------------------------------------------------------------------- - -TEST_P(ClientAsync, Backpressure) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - auto sender = send(request, rv::iota(uint8_t(0))); - co_await (std::move(sender) || sleep(2s)); - // FIXME: count bytes sent, just like asio::async_write() does - // FIXME: or even use asio::async_write() on top of a async_write_some() implementation - - // - // Now that the flow control window is 0, we can't even send an EOF any more -- except over - // QUIC, where whether the FIN slips out without credit depends on flow control timing, so - // only assert that for the stream protocols. - // - auto rc = co_await (send_eof(request) || sleep(100ms)); - if (GetParam() != anyhttp::Protocol::h3) - EXPECT_EQ(rc.index(), 1); - - // So instead, we start doing this in background, to be resumed as soon as the window reopens. - co_spawn(co_await this_coro::executor, send_eof(request), detached); // FIXME: join - - std::println("receiving...."); - boost::system::error_code ec; - auto received = co_await try_receive(response, ec); - std::println("receiving... done, got {} bytes ({})", received, ec.message()); - EXPECT_GT(received, 0); - // EXPECT_EQ(received, sent); - // FIXME: we should be able to receive the remainders that already have been buffered - // FIXME: in the end, this must be the same as the the bytes sent above - }; -} - -// -// Cancellation of a large buffer with Content-Length. -// -// Any short write of a body with known content length should result in a 'partial message' error. -// -// FIXME: As of nghttp2 version 1.67, the partial message results in a GOAWAY, so that only one -// request can be made. The following request should throw an exception. -// -TEST_P(ClientAsync, CancellationContentLength) -{ - test = [this](Session session) -> awaitable - { - const size_t length = 50_m; - const std::vector buffer(length); - for (size_t i = 0; i <= 20; ++i) - { - if (!session) - session = co_await client->async_connect(); - - Fields fields; - fields.set("content-length", std::to_string(length)); - auto request = co_await session.async_submit(url.set_path("echo"), fields); - auto response = co_await request.async_get_response(); - - // - // This is a single large buffer and will be serialized as a single chunk. When writing - // gets cancelled, there is no way to recover gracefully. - // - auto sender = sendAndForceEOF(request, std::string_view(buffer)); - - boost::system::error_code ec; - auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); - std::println("received {} bytes (\x1b[1;31m{}\x1b[0m, yielded {})", std::get<1>(received), - ec.message(), i); - EXPECT_LT(std::get<1>(received), length); - EXPECT_EQ(ec, boost::beast::http::error::partial_message); - - session.reset(); - } - }; -} - -// -// Cancellation of sending a single, large buffer without Content-Length. -// -// HTTP/1.1: As always when not providing Content-Length, the data is chunked. When sending data -// as a single, large buffer, this will result in a single, large chunk of same size. -// If sending that chunk is interrupted, there is no way to recover. The sender will -// close the connection in this situation. -// -// HTTP/2: Cancelling a large buffer without Content-Length will look to the server just like a -// short buffer. No error is raised. FIXME: we could try to support cancellation here -// by closing the stream without sending an EOF. But that would also stop the receiving -// direction. -// -TEST_P(ClientAsync, Cancellation) -{ - test = [this](Session session) -> awaitable - { - const size_t length = 50_m; - const std::vector buffer(length, 'a'); - for (size_t i = 0; i <= 20; ++i) - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - auto sender = sendAndDrop(std::move(request), std::string_view(buffer)); - - boost::system::error_code ec; - auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); - std::println("received {} bytes ({}, yield {})", std::get<1>(received), ec.message(), i); - EXPECT_LT(std::get<1>(received), length); - EXPECT_EQ(ec, boost::beast::http::error::partial_message); - - // HTTP/1.1 needs to reconnect here - // HTTP/2 can handle this without reconnect -- only the stream is cancelled - if (GetParam() == anyhttp::Protocol::http11) - { - session.reset(); - session = co_await client->async_connect(); - } - } - }; -} - -// -// Cancellation of sending a large amount of data that is split into many smaller chunks. -// -// This should work with any protocol, without error. As we don't give a Content-Length in advance, -// cancelling the upload should not be terminal. BUT: cancellation of a parallel group seems to -// do 'terminal' cancellation... -// -// TODO: Aside using operator||, when manually setting up a parallel group, it is possible to -// specify the cancellation type that should be used. -// -// TODO: If an operation supports "partial" as well, it is free to cancel like that even when -// requested to do terminal "cancellation". Cancellation types are backward compatible this -// way. -// -TEST_P(ClientAsync, CancellationRange) -{ - test = [this](Session session) -> awaitable - { - for (size_t i = 6; i <= 6; ++i) - { - co_await yield(); - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - // auto sender = sendAndForceEOF(request, rv::iota(uint8_t(0))); - auto sender = sendAndDrop(std::move(request), rv::iota(uint8_t(0))); - - boost::system::error_code ec; - auto received = co_await ((std::move(sender) || yield(i)) && try_receive(response, ec)); - std::println("received {} bytes ({}, yield {})", std::get<1>(received), ec.message(), i); - EXPECT_EQ(ec, boost::beast::http::error::partial_message); - co_await client->async_connect(); - } - }; -} - -TEST_P(ClientAsync, PerOperationCancellation) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - - asio::cancellation_signal cancel; - asio::steady_timer timer(co_await asio::this_coro::executor, 110ms); - timer.async_wait([&cancel](const boost::system::error_code& ec) { // - cancel.emit(asio::cancellation_type::terminal); - }); - - std::array buffer; - auto token = asio::bind_cancellation_slot(cancel.slot(), as_tuple); - auto [ec, n] = co_await response.async_read_some(asio::buffer(buffer), std::move(token)); - EXPECT_EQ(ec, boost::system::errc::operation_canceled); - }; -} - -TEST_P(ClientAsync, CancelAfter) -{ - test = [this](Session session) -> awaitable - { - auto request = - co_await session.async_submit(url.set_path("echo").set_params({{"delay", "1000"}}), {}); - auto [ec, response] = co_await request.async_get_response(cancel_after(250ms, as_tuple)); - EXPECT_EQ(ec, boost::system::errc::operation_canceled); - - std::tie(ec, response) = co_await request.async_get_response(cancel_after(0ms, as_tuple)); - EXPECT_EQ(ec, boost::system::errc::operation_canceled); - - std::tie(ec, response) = co_await request.async_get_response(as_tuple); - EXPECT_FALSE(ec); - - constexpr auto msg = "Hello, Client!"sv; - co_await request.async_write_eof(asio::buffer(msg)); - EXPECT_EQ(co_await read(response), msg); - }; -} - -TEST_P(ClientAsync, WHEN_send_more_than_content_length_THEN_connection_is_reset) -{ - test = [this](Session session) -> awaitable - { - Fields fields; - fields.set("content-length", "1024"); - auto request = co_await session.async_submit(url.set_path("eat_request"), fields); - auto response = co_await request.async_get_response(); - co_await drain(response); - - auto ex = co_await this_coro::executor; - auto [ep] = co_await co_spawn(ex, send(request, rv::iota(uint8_t(0))), as_tuple); - - // - // Which of the two the write reports is a matter of how far the kernel has gotten with the - // peer's RST by the time we get to write again -- the first write after it fails with - // ECONNRESET, any later one with EPIPE. Single-threaded we reliably hit the former, with - // more than one thread the latter; both mean the same thing here. - // - EXPECT_THAT(code(ep), testing::AnyOf(boost::system::errc::connection_reset, - boost::system::errc::broken_pipe)); - }; -} - -// ================================================================================================= - -TEST_P(ClientAsync, ClientDropRequest) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - }; -} - -// ================================================================================================= - -TEST_P(ClientAsync, ResetServerDuringRequest) -{ - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - - // - // Deliberately NOT use_future(): with more than one thread the client lives on a strand, - // and blocking that strand in future.get() below would keep the very handlers that - // complete this send from ever running. asio::experimental::promise starts the coroutine - // right away, just like use_future, but is awaited instead of waited on. - // - auto promise = co_spawn(request.get_executor(), send(request, rv::iota(uint8_t(0))), - asio::experimental::use_promise); - - std::println("============================================================================="); - for (size_t i = 0; i < 10; ++i) - { - std::println("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -", i); - co_await yield(); - } - - std::println("============================================================================="); - server.reset(); - - for (size_t i = 0; i < 10; ++i) - { - std::println("- - {} - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -", i); - co_await yield(); - } - - auto exception_ptr = co_await std::move(promise)(as_tuple(use_awaitable)); - - boost::system::error_code ec; - auto received = co_await try_receive(response, ec); - loge("received: {} ({} bytes)", ec.message(), received); - }; -} - -TEST_P(ClientAsync, DISABLED_SpawnAndForget) -{ - if (GetParam() == anyhttp::Protocol::http11) - GTEST_SKIP(); // FIXME: ASAN errors - - test = [this](Session session) -> awaitable - { - auto request = co_await session.async_submit(url.set_path("echo"), {}); - auto response = co_await request.async_get_response(); - co_await yield(); - - std::println("- - spawning - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - "); - co_spawn(context, - [request = std::move(request)]() mutable -> awaitable - { // - std::println("- - SPAWNED - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -"); - co_await yield(5); - std::println("- - SPAWNED, sending - - - - - - - - - - - - - - - - - - - - - - - - -"); - co_await send(request, rv::iota(uint8_t(0))); - }, detached); - }; -} - -// ================================================================================================= - // // A QUIC peer that goes away without a word -- a killed client, a machine that went to sleep in // the middle of a request -- leaves the server nothing to react to: no CONNECTION_CLOSE arrives,