diff --git a/README.md b/README.md index 39f9938..db1fa75 100644 --- a/README.md +++ b/README.md @@ -118,6 +118,40 @@ namespace impl { ``` +## Concurrent Requests + +HTTP/2 and HTTP/3 multiplex requests: each one is a stream of its own, and streams make progress independently. HTTP/1.1 has a single connection instead, which requests are written to and responses read from, one after the other. anyhttp treats HTTP/1.1 as a protocol with **"max concurrent streams = 1"**, and makes that explicit in the client API instead of hiding it. + +### Complete requests and responses + +A request is *complete* when all of it -- header and body -- has been written to the connection: + +* A request **without a body** is complete as soon as `async_submit()` has succeeded. Whether a request has a body is a matter of its framing only ([RFC 9112, section 6.3](https://www.rfc-editor.org/rfc/rfc9112#section-6.3)), never of its method: it has none without `Transfer-Encoding` and with a `Content-Length` of zero or none. As the HTTP/1.1 client sends every request without `Content-Length` chunked, that means `Content-Length: 0`. +* **Any other request** is complete when `async_write_eof()` has succeeded -- even if all of a `Content-Length` has been written before. The body of a request without one is ended implicitly: `async_write_eof()` is an idempotent no-op on it, and writing data fails with `broken_pipe`. + +A response is *complete* when it has been read to its end. + +### Rules for HTTP/1.1 + +1. `async_submit()` fails with `asio::error::would_block` while the previous request is not complete. +2. `async_get_response()` fails with `asio::error::would_block` while the responses to earlier requests have not been read completely -- otherwise, it would read one of those. +3. When a request can not be completed -- a write fails, or it is released before it is complete -- the connection is stuck in the middle of a message: every later `async_submit()` fails with `asio::error::connection_aborted`. +4. When a response can not be read completely -- it is released before its end, or its request is released without asking for it -- every later `async_get_response()` fails with `asio::error::connection_aborted`. + +Pipelining is still possible: complete requests can be sent before any of their responses have been read. + +### Design decisions + +* **Fail instead of waiting.** An operation that has to wait for an earlier request or response does not wait. Very often, the caller waiting is the one who has to finish that earlier request or response, after the operation returns -- waiting would deadlock. Failing immediately with `would_block` turns a hang into an error that can be handled: finish the earlier one, then retry. +* **No queueing of submitted requests.** An earlier version queued the header of a request submitted while the previous one was still incomplete, and sent it as soon as that was complete. That allows code written for HTTP/2 -- submit a couple of requests first, write their bodies later -- to work unchanged. But it needed a queue of pending requests, writes waiting for a header still on its way (and not cancellable while doing so), and failures cascading to queued requests. Refusing the submission keeps it at a single incomplete request per session, which is exactly what the protocol allows. +* **Requests and responses may outlive their session.** The session keeps track of all of its readers and writers, and detaches them when it goes away. Everything a detached request or response is asked to do after that completes with an error (`connection_aborted` for HTTP/1.1 and HTTP/2, `connection_reset` for HTTP/3), without touching the connection that is gone. + +### Outlook: HTTP/2 and HTTP/3 + +HTTP/2 and HTTP/3 have a limit of their own: the peer's `SETTINGS_MAX_CONCURRENT_STREAMS`, or the QUIC stream limit. With that limit reached, they should behave just like HTTP/1.1 -- fail with `would_block` instead of waiting. Currently, anyhttp does not check that limit itself, and leaves it to nghttp2 and ngtcp2; tests for that are still to be added. + +One difference remains to be decided: in HTTP/2 and HTTP/3, a stream counts against the limit until it is closed in *both* directions, that is, until its response has been received as well. Taken strictly, "max concurrent streams = 1" would forbid submitting the next request before the previous response has been read -- which is stricter than HTTP/1.1 pipelining as implemented. + ## Links For now, this section contains just a set of random links collected during development. diff --git a/include/anyhttp/client.hpp b/include/anyhttp/client.hpp index 1a3f20f..c8d0e19 100644 --- a/include/anyhttp/client.hpp +++ b/include/anyhttp/client.hpp @@ -22,6 +22,14 @@ struct Config // FIXME: the client does not connect to an URL, it connects to a host:port or endpoint boost::urls::url url{"localhost:8080"}; Protocol protocol{Protocol::h2}; + + // + // The largest header section of a response the client accepts, in bytes, counted as for + // server::Config::max_header_size. For a response with more, async_get_response() fails with + // boost::beast::http::error::header_limit and the stream is reset -- with HTTP/1.1, which has no + // streams, the connection can not be used any more. + // + size_t max_header_size = default_max_header_size; }; // ================================================================================================= @@ -104,6 +112,16 @@ class Request using GetResponse = void(boost::system::error_code, Response); using GetResponseHandler = asio::any_completion_handler; + /** + * Waits for the response to this request, until its header has been received. + * + * With HTTP/1.1, responses arrive in the order the requests were sent, one after the other. + * Getting the response to a request whose predecessors' responses have not been read to their + * end does not wait for that to happen, but fails immediately with + * \c asio::error::would_block. After a response could not be read -- it was released before + * its end, or its request was released without asking for it -- getting any later response + * fails with \c asio::error::connection_aborted. See README.md, "Concurrent Requests". + */ template auto async_get_response(CompletionToken&& token = CompletionToken()) { diff --git a/include/anyhttp/client_impl.hpp b/include/anyhttp/client_impl.hpp index 9e06014..37fe2d9 100644 --- a/include/anyhttp/client_impl.hpp +++ b/include/anyhttp/client_impl.hpp @@ -48,9 +48,9 @@ class Client::Impl boost::asio::any_io_executor get_executor() const noexcept { return m_executor; } void async_connect(ConnectHandler handler); + const Config& config() const { return m_config; } private: - const Config& config() const { return m_config; } awaitable async_connect(); private: diff --git a/include/anyhttp/common.hpp b/include/anyhttp/common.hpp index ebb5268..f623e23 100644 --- a/include/anyhttp/common.hpp +++ b/include/anyhttp/common.hpp @@ -52,6 +52,19 @@ std::ostream& operator<<(std::ostream& str, Protocol protocol); using Fields = boost::beast::http::fields; static_assert(boost::beast::http::is_fields::value); +/// Default for \c server::Config::max_header_size and \c client::Config::max_header_size. +inline constexpr size_t default_max_header_size = 64 * 1024; + +/** + * What a received header field counts against the limit on the size of a header section: its name + * and value plus 32 bytes of overhead, as for SETTINGS_MAX_HEADER_LIST_SIZE in HTTP/2 (RFC 9113, + * section 6.5.2) and SETTINGS_MAX_FIELD_SECTION_SIZE in HTTP/3 (RFC 9114, section 4.2.2). + */ +constexpr size_t header_field_size(std::string_view name, std::string_view value) noexcept +{ + return name.size() + value.size() + 32; +} + // // A header value as passed to fields() below: either something string-like, or anything // std::format can turn into a string, so sizes and counts need no conversion at the call site. diff --git a/include/anyhttp/detail/h2_session_details.hpp b/include/anyhttp/detail/h2_session_details.hpp index b91a23d..98e2be5 100644 --- a/include/anyhttp/detail/h2_session_details.hpp +++ b/include/anyhttp/detail/h2_session_details.hpp @@ -6,7 +6,9 @@ // #include "anyhttp/any_async_stream.hpp" +#include "anyhttp/h2_common.hpp" #include "anyhttp/h2_session.hpp" +#include "anyhttp/literals.hpp" #include #include @@ -158,7 +160,7 @@ awaitable NGHttp2SessionImpl::send_loop() template awaitable NGHttp2SessionImpl::recv_loop() { - m_buffer.reserve(64 * 1024); + m_buffer.reserve(64_k); unsigned int reason = NGHTTP2_NO_ERROR; while (nghttp2_session_want_read(session) || nghttp2_session_want_write(session)) @@ -190,6 +192,7 @@ ServerSession::ServerSession(server::Server::Impl& parent, any_io_execut Stream&& stream) : ServerReference(parent), super("\x1b[1;31mserver\x1b[0m", executor, std::move(stream)) { + m_max_header_size = parent.config().max_header_size; } // ------------------------------------------------------------------------------------------------- @@ -210,12 +213,19 @@ awaitable ServerSession::do_session(Buffer&& buffer) auto options = nghttp2_option_new(); nghttp2_option_set_no_http_messaging(options.get(), 0); // h2spec: fails ~16 tests if 1 nghttp2_option_set_no_auto_window_update(options.get(), 1); + nghttp2_option_set_max_send_header_block_length(options.get(), 1_m); + nghttp2_option_set_max_continuations(options.get(), max_continuations(m_max_header_size)); if (auto rv = nghttp2_session_server_new2(&session, callbacks.get(), this, options.get())) throw std::runtime_error("nghttp2_session_server_new"); #if 1 - const uint32_t window_size = 1024 * 1024; + const uint32_t window_size = 1_m; + // + // No SETTINGS_MAX_HEADER_LIST_SIZE: it defaults to unlimited and is advisory anyway, as nghttp2 + // enforces it in neither direction. Header sections beyond max_header_size are rejected where + // they arrive, see on_header_callback(). + // std::array iv{{{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 100}, {NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, window_size}}}; nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, iv.data(), iv.size()); @@ -234,9 +244,9 @@ awaitable ServerSession::do_session(Buffer&& buffer) { 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)) + 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); @@ -279,6 +289,7 @@ ClientSession::ClientSession(client::Client::Impl& parent, any_io_execut Stream&& stream) : ClientReference(parent), super("\x1b[1;32mclient\x1b[0m", executor, std::move(stream)) { + m_max_header_size = parent.config().max_header_size; } // ------------------------------------------------------------------------------------------------- @@ -299,12 +310,19 @@ awaitable ClientSession::do_session(Buffer&& buffer) auto options = nghttp2_option_new(); nghttp2_option_set_no_http_messaging(options.get(), 1); nghttp2_option_set_no_auto_window_update(options.get(), 1); + nghttp2_option_set_max_send_header_block_length(options.get(), 1_m); + nghttp2_option_set_max_continuations(options.get(), max_continuations(m_max_header_size)); if (auto rv = nghttp2_session_client_new2(&session, callbacks.get(), this, options.get())) throw std::runtime_error("nghttp2_session_client_new"); #if 1 - const uint32_t window_size = 1024 * 1024; + const uint32_t window_size = 1_m; + // + // No SETTINGS_MAX_HEADER_LIST_SIZE: it defaults to unlimited and is advisory anyway, as nghttp2 + // enforces it in neither direction. Header sections beyond max_header_size are rejected where + // they arrive, see on_header_callback(). + // std::array iv{{{NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS, 100}, {NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE, window_size}}}; nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, iv.data(), iv.size()); diff --git a/include/anyhttp/formatter.hpp b/include/anyhttp/formatter.hpp index 6fc1022..0d0553f 100644 --- a/include/anyhttp/formatter.hpp +++ b/include/anyhttp/formatter.hpp @@ -10,8 +10,10 @@ #include #include -#include +#include #include +#include +#include // ================================================================================================= @@ -80,6 +82,48 @@ struct std::formatter // ================================================================================================= +namespace anyhttp +{ + +/// A string to be logged, cut short if it is longer than \c max_size bytes. See truncated(). +struct Truncated +{ + std::string_view text; + size_t max_size; +}; + +/// Default for truncated(): long enough for any regular header, short enough to keep the log readable. +inline constexpr size_t max_logged_size = 80; + +/** + * Wraps \p text for logging, so that only its first \p max_size bytes are printed, followed by the + * total size. Meant for header names and values, which may be of almost any size: + * \code + * logd("{}: {}", truncated(name), truncated(value)); // x-large: aaaa... (30000 bytes, truncated) + * \endcode + */ +inline Truncated truncated(std::string_view text, size_t max_size = max_logged_size) +{ + return {text, max_size}; +} + +} // namespace anyhttp + +template <> +struct std::formatter : std::formatter +{ + auto format(const anyhttp::Truncated& value, std::format_context& ctx) const + { + if (value.text.size() <= value.max_size) + return std::formatter::format(value.text, ctx); + + return std::format_to(ctx.out(), "{}... ({} bytes, truncated)", + value.text.substr(0, value.max_size), value.text.size()); + } +}; + +// ================================================================================================= + template <> struct std::formatter : std::formatter { diff --git a/include/anyhttp/h1_session.hpp b/include/anyhttp/h1_session.hpp index 3d801a4..c025b22 100644 --- a/include/anyhttp/h1_session.hpp +++ b/include/anyhttp/h1_session.hpp @@ -14,6 +14,9 @@ #include #include +#include +#include + using namespace boost::asio; namespace anyhttp::beast_impl @@ -40,18 +43,47 @@ class BeastSession : public ::anyhttp::Session::Impl // ---------------------------------------------------------------------------------------------- + // + // Readers and writers refer back to their session, and may outlive it. So each of them + // registers here for as long as it exists, to be detach()ed when the session goes away first. + // With pipelining, a client session may have more than one of each at a time. + // + void attach(impl::Reader& reader) { m_readers.push_back(&reader); } + void attach(impl::Writer& writer) { m_writers.push_back(&writer); } + void release(impl::Reader& reader) { std::erase(m_readers, &reader); } + void release(impl::Writer& writer) { std::erase(m_writers, &writer); } + + void detach_readers() + { + for (auto* reader : std::exchange(m_readers, {})) + reader->detach(); + } + + void detach_writers() + { + for (auto* writer : std::exchange(m_writers, {})) + writer->detach(); + } + + /// Called once by each reader when it is done with the stream: either because its message has + /// been read completely (\p complete), or because it is going away before that. + virtual void reader_finished(bool complete) {} + + // ---------------------------------------------------------------------------------------------- + public: std::string m_logPrefix; asio::any_io_executor m_executor; Stream m_stream; Buffer m_buffer; bool m_closed = false; - - /// Non-owning pointer to the currently active reader, detach()ed when this is destroyed. - impl::Reader* rx = nullptr; - /// Non-owning pointer to the currently active writer, detach()ed when this is destroyed. - impl::Writer* wx = nullptr; +private: + /// Non-owning pointers to the attached readers, see attach(). + std::vector m_readers; + + /// Non-owning pointers to the attached writers, see attach(). + std::vector m_writers; }; // ================================================================================================= @@ -80,8 +112,8 @@ class ServerSession : public ServerSessionBase, public BeastSession using super::m_buffer; using super::m_stream; using super::m_closed; - using super::rx; - using super::wx; + using super::detach_readers; + using super::detach_writers; public: ServerSession(server::Server::Impl& parent, any_io_executor executor, Stream&& stream); @@ -111,6 +143,44 @@ class ClientSessionBase client::Client::Impl* m_client = nullptr; }; +template +class RequestWriter; + +/** + * HTTP/1.1 client session. + * + * HTTP/1.1 is treated as a protocol with "max concurrent streams = 1", in the sense that there is + * only a single connection to write requests to and read responses from, without any means to + * interleave them. Pipelining is supported: requests can be written before the responses to the + * earlier ones have been read. But everything that goes onto the connection, or comes off it, has + * to be done one after the other, and completely. See README.md, "Concurrent Requests", for the + * reasoning behind this. + * + * A request is \e complete when all of it has been written to the connection, header and body: + * + * - A request that has no body is complete when async_submit() has succeeded. Whether a request + * has a body is a matter of its framing only (RFC 9112, section 6.3), not of its method: it has + * none without 'Transfer-Encoding' and with a 'Content-Length' of zero or none. As a request + * without 'Content-Length' is always sent chunked, that means "Content-Length: 0". + * - Any other request is complete when async_write_eof() has succeeded -- even if all of a + * 'Content-Length' has been written before. + * + * A response is \e complete when it has been read to its end. + * + * Operations that would have to wait for an earlier request or response don't. Waiting would + * deadlock as soon as the earlier one is taken care of by the same code, after the operation that + * waits for it. Instead, they fail immediately with \c asio::error::would_block, so they can be + * retried later: + * + * - async_submit(), while the previous request is not complete yet, and + * - async_get_response(), while the responses to earlier requests have not been read completely. + * + * When a request can not be completed -- a write fails, or the request goes away without being + * complete -- nothing can be sent after it any more: later calls to async_submit() fail with + * \c asio::error::connection_aborted. Likewise, when a response can not be read completely -- a + * read fails, the response goes away before it has been read, or a request goes away without + * asking for its response -- getting any later response fails. + */ template class ClientSession : public ClientSessionBase, public BeastSession { @@ -120,14 +190,35 @@ class ClientSession : public ClientSessionBase, public BeastSession using super::logPrefix; using super::m_buffer; using super::m_stream; - using super::wx; - using super::rx; public: ClientSession(client::Client::Impl& parent, any_io_executor executor, Stream&& stream); void async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers) override; awaitable do_session(Buffer&& data) override; + + // ---------------------------------------------------------------------------------------------- + + /// Called by the request that is being sent when it is complete. + void request_complete(RequestWriter& request); + + /// Called by the request that is being sent when it can't be completed any more. + void request_failed(RequestWriter& request); + + /// Called by every request that goes away while the session is still there. + void request_released(RequestWriter& request); + + void reader_finished(bool complete) override; + + // ---------------------------------------------------------------------------------------------- + + /// The request that is not complete yet, if any. There can be only one. + RequestWriter* m_sending = nullptr; + + size_t m_requests_sent = 0; ///< number of requests submitted + size_t m_responses_read = 0; ///< number of responses that have been read completely + bool m_send_failed = false; ///< set when a request could not be completed + bool m_receive_failed = false; ///< set when a response could not be read completely }; // ================================================================================================= diff --git a/include/anyhttp/h2_common.hpp b/include/anyhttp/h2_common.hpp index 34a1fb8..976941a 100644 --- a/include/anyhttp/h2_common.hpp +++ b/include/anyhttp/h2_common.hpp @@ -32,6 +32,19 @@ inline nghttp2_nv make_nv_ls(std::string_view key, std::string_view value) return {(uint8_t*)key.data(), (uint8_t*)value.data(), key.size(), value.size(), 0}; } +/** + * The number of CONTINUATION frames to accept after a HEADERS frame, for a header section of up to + * \p max_header_size bytes in frames of the default size. A peer sending more is flooding us and + * loses the connection. nghttp2 accepts 8 by default, which is kept as the minimum. + */ +inline size_t max_continuations(size_t max_header_size) +{ + return std::max(8, max_header_size / 16384 + 1); +} + +inline std::string_view name_of(const nghttp2_nv& nv) { return {(const char*)nv.name, nv.namelen}; } +inline std::string_view value_of(const nghttp2_nv& nv) { return {(const char*)nv.value, nv.valuelen}; } + // ================================================================================================= } // namespace anyhttp diff --git a/include/anyhttp/h2_session.hpp b/include/anyhttp/h2_session.hpp index 5901909..e75dbae 100644 --- a/include/anyhttp/h2_session.hpp +++ b/include/anyhttp/h2_session.hpp @@ -26,6 +26,7 @@ namespace anyhttp::nghttp2 // ================================================================================================= +/// RAII wrapper for NGHTTP2 objects, wrapping them in a std::unique_ptr with a custom deleter. template using nghttp2_unique_ptr = std::unique_ptr; @@ -123,6 +124,9 @@ class NGHttp2Session : public anyhttp::Session::Impl int32_t m_last_id = 0; size_t m_requestCounter = 0; + /// The largest header section accepted from the peer, see Config::max_header_size. + size_t m_max_header_size = default_max_header_size; + Buffer m_buffer; }; @@ -176,6 +180,7 @@ class ServerSession : public ServerReference, public NGHttp2SessionImpl using super::send_loop; using super::m_buffer; + using super::m_max_header_size; using super::m_stream; using super::session; @@ -218,6 +223,7 @@ class ClientSession : public ClientReference, public NGHttp2SessionImpl using super::send_loop; using super::m_buffer; + using super::m_max_header_size; using super::m_stream; using super::session; diff --git a/include/anyhttp/h2_stream.hpp b/include/anyhttp/h2_stream.hpp index 1382ac4..b589bcd 100644 --- a/include/anyhttp/h2_stream.hpp +++ b/include/anyhttp/h2_stream.hpp @@ -168,6 +168,9 @@ class NGHttp2Stream : public std::enable_shared_from_this /// Set to true after responseHandler has been invoked, to make sure that this happens only once. bool response_delivered = false; + /// Set instead of has_response for a response that is not delivered, but fails. + boost::system::error_code response_error; + std::string logPrefix; std::string method; boost::urls::url url; @@ -182,6 +185,10 @@ class NGHttp2Stream : public std::enable_shared_from_this std::optional content_length; Fields fields; // all received headers except the pseudo-headers + /// Size of the received header section so far, see Config::max_header_size. + size_t header_size = 0; + bool header_limit_exceeded = false; + bool closed = false; // set to true after on_stream_close_callback public: diff --git a/include/anyhttp/h3_backend.hpp b/include/anyhttp/h3_backend.hpp index b8b4d65..e2d6e42 100644 --- a/include/anyhttp/h3_backend.hpp +++ b/include/anyhttp/h3_backend.hpp @@ -58,7 +58,8 @@ namespace anyhttp::client /// Connects to `host`:`port` over QUIC and returns the HTTP/3 session running on it. boost::asio::awaitable> -async_connect_http3(boost::asio::any_io_executor executor, std::string host, std::string port); +async_connect_http3(boost::asio::any_io_executor executor, std::string host, std::string port, + const Config& config); // ================================================================================================= diff --git a/include/anyhttp/h3_session.hpp b/include/anyhttp/h3_session.hpp index c702155..66d562a 100644 --- a/include/anyhttp/h3_session.hpp +++ b/include/anyhttp/h3_session.hpp @@ -53,6 +53,9 @@ class Http3Session : public Session::Impl bool closed() const noexcept { return closed_; } const std::string& logPrefix() const noexcept { return log_prefix_; } + /// The largest header section accepted from the peer, see Config::max_header_size. + size_t max_header_size() const noexcept { return max_header_size_; } + // // Returns a shared_ptr, not a raw pointer: callers routinely invoke user handlers on the // stream they looked up, and those can drop the last reference to it (the coroutine they @@ -222,6 +225,7 @@ class Http3Session : public Session::Impl ngtcp2_crypto_conn_ref conn_ref_{}; nghttp3_conn* h3_ = nullptr; + size_t max_header_size_ = default_max_header_size; // set by the derived session's constructor asio::steady_timer timer_; // ngtcp2 expiry (handshake / idle / PTO) ngtcp2_ccerr last_error_{}; diff --git a/include/anyhttp/h3_stream.hpp b/include/anyhttp/h3_stream.hpp index 99e9735..5cb8bab 100644 --- a/include/anyhttp/h3_stream.hpp +++ b/include/anyhttp/h3_stream.hpp @@ -92,6 +92,10 @@ class Http3Stream : public std::enable_shared_from_this std::vector> received_headers; bool headers_received = false; + /// Size of the received header section so far, see Config::max_header_size. + size_t header_size = 0; + bool header_limit_exceeded = false; + // // Outgoing message: the response on the server (set by the user through Http3Writer), the // request on the client (set once by async_submit(), before this stream even exists as far as diff --git a/include/anyhttp/request_handlers.hpp b/include/anyhttp/request_handlers.hpp index c463e2b..f0de405 100644 --- a/include/anyhttp/request_handlers.hpp +++ b/include/anyhttp/request_handlers.hpp @@ -65,6 +65,9 @@ awaitable sleep(T duration) awaitable yield(size_t count = 1); awaitable not_found(server::Response response); awaitable not_found(server::Request request, server::Response response); + +/// Responds with 431 (Request Header Fields Too Large), see server::Config::max_header_size. +awaitable header_fields_too_large(server::Request request, server::Response response); awaitable dump(server::Request request, server::Response response); awaitable echo(server::Request request, server::Response response); awaitable eat_request(server::Request request, server::Response response); diff --git a/include/anyhttp/server.hpp b/include/anyhttp/server.hpp index 93f9f89..ab50e4d 100644 --- a/include/anyhttp/server.hpp +++ b/include/anyhttp/server.hpp @@ -29,6 +29,20 @@ struct Config uint16_t port = 8080; bool use_strand = false; + // + // The largest header section of a request the server accepts, in bytes. A request with more is + // answered with 431 (Request Header Fields Too Large) and never reaches the request handler. + // HTTP/1.1 counts the request line and header lines as received, and closes the connection + // after the 431 response. HTTP/2 and HTTP/3 count each field as its name and value plus 32 + // bytes (pseudo-headers included) and announce the limit to the client in their SETTINGS. + // + // Fields beyond the limit are not stored, so this bounds the memory a request can take up with + // its headers. A single field is also limited by the protocol libraries: 64 KiB for HTTP/2 and + // HTTP/3, where a larger one fails the whole connection. So does an HTTP/2 header block that + // takes up more CONTINUATION frames than a header section of this size needs (but at least 8). + // + size_t max_header_size = default_max_header_size; + // // HTTP/3 only: how long a QUIC connection may go without a packet from its peer before it is // dropped. This is the only way a peer that vanished without a CONNECTION_CLOSE -- a killed @@ -46,6 +60,14 @@ struct Config // double drop_rate_rx = 0.0; double drop_rate_tx = 0.0; + + // + // HTTP/3 only, for benchmarking: turn off the kernel's UDP offloads. Without GRO, every received + // datagram costs a recvmsg() of its own; without GSO, every QUIC packet costs a sendto() of its + // own instead of a whole same-sized run going out in one sendmsg(). + // + bool disable_gro = false; + bool disable_gso = false; }; // ================================================================================================= diff --git a/include/anyhttp/session.hpp b/include/anyhttp/session.hpp index 17a3f01..ac92d0b 100644 --- a/include/anyhttp/session.hpp +++ b/include/anyhttp/session.hpp @@ -38,6 +38,15 @@ class Session * * Use \ref Request::async_get_response() on the request to wait for the response. * + * A session may limit the number of requests in progress. When that limit is reached, this + * does not wait for a request to finish -- the caller might be the one who has to finish it -- + * but fails immediately with \c asio::error::would_block. HTTP/1.1 is such a protocol, with a + * limit of one: a request is in progress until it is \e complete, that is, until all of it, + * header and body, has been written. See README.md, "Concurrent Requests". + * + * After a request could not be completed, an HTTP/1.1 session can't send any more requests, + * and this fails with \c asio::error::connection_aborted. + * * TODO: There is only a single Session interface for both server and client. This even might * make sense for HTTP/2, where the server can also (sort of) submit a push promise to the * client. But in general, it may be better to separate them. diff --git a/src/client_impl.cpp b/src/client_impl.cpp index ef35c77..46b5e72 100644 --- a/src/client_impl.cpp +++ b/src/client_impl.cpp @@ -98,7 +98,7 @@ awaitable Client::Impl::async_connect() // handshake, ...) than the TCP-based http11/h2 paths below. // if (config().protocol == Protocol::h3) - co_return Session{co_await async_connect_http3(m_executor, host, port)}; + co_return Session{co_await async_connect_http3(m_executor, host, port, config())}; std::vector endpoints; logd("Client: resolving {}:{} ...", host, port); diff --git a/src/h1_session.cpp b/src/h1_session.cpp index e5bf31a..f237f73 100644 --- a/src/h1_session.cpp +++ b/src/h1_session.cpp @@ -5,6 +5,7 @@ #include "anyhttp/formatter.hpp" // IWYU pragma: keep #include "anyhttp/h1_backend.hpp" #include "anyhttp/h2_backend.hpp" +#include "anyhttp/literals.hpp" #include "anyhttp/server.hpp" #include @@ -18,8 +19,8 @@ #include #include -#include #include +#include #include #include #include @@ -40,6 +41,7 @@ #include #include +#include #include #include #include @@ -65,6 +67,26 @@ inline auto& get_socket(tcp_stream& stream) { return stream.socket(); } inline auto& get_socket(ssl::stream& stream) { return stream.lowest_layer(); } inline auto& get_socket(AnyAsyncStream& stream) { return stream.get_socket(); } +/** + * Adds the user's header fields to an outgoing message. A field replaces whatever the message + * already has under that name, like a default set before, but repeated fields are all kept. + */ +template +void add_fields(http::message& message, const Fields& headers) +{ + for (auto&& header : headers) + message.erase(header.name_string()); + for (auto&& header : headers) + message.insert(header.name_string(), header.value()); +} + +/// Converts Config::max_header_size into what a Beast parser takes as its header limit. +inline std::uint32_t header_limit(size_t max_header_size) +{ + return static_cast( + std::min(max_header_size, std::numeric_limits::max())); +} + // ================================================================================================= template @@ -75,7 +97,9 @@ class BeastReader : public Interface : session(&session_), stream(stream_), buffer(buffer_), m_executor(session_.get_executor()) // survives detach(), see get_executor() { + // parser.header_limit(std::numeric_limits::max()); parser.body_limit(std::numeric_limits::max()); + session_.attach(*this); } void destroy() noexcept override @@ -95,13 +119,25 @@ class BeastReader : public Interface if (ec) logw("destroy: shutdown: {}", what(ec)); } + finish(); } ~BeastReader() override { assert(!reading); + finish(); if (session) - session->rx = nullptr; + session->release(*this); + } + + /// Tells the session, once, that this reader is done with the stream, see reader_finished(). + void finish() + { + if (session && !finished) + { + finished = true; + session->reader_finished(parser.is_done()); + } } void detach() override { @@ -128,10 +164,6 @@ class BeastReader : public Interface void async_read_some(asio::mutable_buffer body_buffer, ReadSomeHandler&& handler) override { - buffer.reserve(64 * 1024); - mlogd("async_read_some: is_done={} size={} capacity={}", parser.is_done(), buffer.size(), - buffer.capacity()); - assert(!reading); // @@ -155,6 +187,10 @@ class BeastReader : public Interface return; } + buffer.reserve(64_k); // the buffer is the session's, so not before checking for it + mlogd("async_read_some: is_done={} size={} capacity={}", parser.is_done(), buffer.size(), + buffer.capacity()); + reading = true; parser.get().body().data = body_buffer.data(); parser.get().body().size = body_buffer.size(); @@ -173,6 +209,9 @@ class BeastReader : public Interface if (ec == beast::http::error::need_buffer) ec = {}; // FIXME: maybe we should keep 'need_buffer' to avoid extra empty round trip + if (parser.is_done()) + finish(); + // // Nothing of the body came out of this round -- either the parser is now done, in which // case the retry below turns straight into the EOF completion above, or it just needs @@ -207,6 +246,7 @@ class BeastReader : public Interface std::optional m_status_code = 0; boost::url m_url; bool reading = false; + bool finished = false; // see finish() }; // ------------------------------------------------------------------------------------------------- @@ -254,13 +294,14 @@ class WriterBase : public Parent : session(&session_), stream(stream_), m_executor(session_.get_executor()) // survives detach(), see get_executor() { + session_.attach(*this); } ~WriterBase() override { assert(!writing); if (session) - session->wx = nullptr; + session->release(*this); } inline auto logPrefix() const { return session ? session->logPrefix() : "DETACHED"; } @@ -284,6 +325,7 @@ class WriterBase : public Parent // prescribes: a zero-length non-EOF write is a free no-op wherever the body stands (it // would otherwise turn into an empty chunk); after the body has ended, data has no body // left to belong to -- through either entry point -- while a bare re-end is idempotent. + // Only then does it matter that the session, and with it the stream, may be gone. // if (empty && !eof) { @@ -300,6 +342,14 @@ class WriterBase : public Parent return; } + if (!session) + { + mlogw("async_write: session already gone"); + complete_immediately(std::move(handler), get_executor(), + make_error_code(asio::error::connection_aborted)); + return; + } + if (cancelled) { mloge("async_write: already canceled"); @@ -377,8 +427,11 @@ class WriterBase : public Parent // mlogw("async_write: canceled after writing {} of {} bytes", n, expected); cancelled = true; - mlogw("async_write: canceled, closing stream"); - get_socket(stream).shutdown(boost::asio::socket_base::shutdown_send); + if (session) // otherwise, the stream is gone already + { + mlogw("async_write: canceled, closing stream"); + get_socket(stream).shutdown(boost::asio::socket_base::shutdown_send); + } } else if (ec) { @@ -400,6 +453,16 @@ class WriterBase : public Parent if (!ec && eof) eof_submitted = true; + if (session && eof_submitted) + body_ended(); + else if (session && cancelled) + write_failed(); + + // + // The handler may resume the caller right here. If it releases the writer, that has to + // take effect immediately, not only when this callback is gone. + // + self.reset(); std::move(handler)(ec); }; @@ -409,6 +472,12 @@ class WriterBase : public Parent std::move(cb), std::move(ex), std::move(alloc), std::move(cs)}); } + /// Called when a write has ended the body, before its handler is invoked. + virtual void body_ended() {} + + /// Called when a write has failed, leaving the stream unusable for this message. + virtual void write_failed() {} + // ---------------------------------------------------------------------------------------------- /** @@ -418,8 +487,7 @@ class WriterBase : public Parent { message.body().data = nullptr; - for (auto&& header : headers) - message.set(header.name_string(), header.value()); + add_fields(message, headers); if (!message.has_content_length()) message.chunked(true); @@ -454,6 +522,7 @@ class ResponseWriter using super::logPrefix; using super::message; using super::serializer; + using super::session; using super::stream; using super::submit_headers; @@ -473,6 +542,14 @@ class ResponseWriter void async_submit(StatusHandler&& handler, unsigned int status_code, const Fields& headers) override { + if (!session) + { + mlogw("async_submit: session already gone"); + complete_immediately(std::move(handler), super::get_executor(), + make_error_code(asio::error::connection_aborted)); + return; + } + message.result(status_code); if (message.find(http::field::date) == message.end()) @@ -485,7 +562,7 @@ class ResponseWriter mlogd("{} {}", message.result_int(), message.reason()); for (const auto& header : message) - mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + mlogd(" \x1b[1;34m{}\x1b[0m: {}", truncated(header.name_string()), truncated(header.value())); // // TODO: For bundling writing the header and body, we should just post the writing here, @@ -509,6 +586,8 @@ class RequestWriter WriterBase>; public: + using super::cancelled; + using super::eof_submitted; using super::logPrefix; using super::message; using super::response_requested; @@ -518,11 +597,61 @@ class RequestWriter using super::submit_headers; public: - inline RequestWriter(BeastSession& session_, Stream& stream_) : super(session_, stream_) + inline RequestWriter(ClientSession& session_, Stream& stream_) : super(session_, stream_) { } - ~RequestWriter() = default; + ~RequestWriter() override + { + if (session) + client_session().request_released(*this); + } + + ClientSession& client_session() + { + assert(session); + return static_cast&>(*session); + } + + // ---------------------------------------------------------------------------------------------- + + /** + * Whether the request has a body, going by its framing (RFC 9112, section 6.3): a request + * without 'Transfer-Encoding' and without a 'Content-Length' other than zero has none. As + * ClientSession::async_submit() makes every request without 'Content-Length' chunked, that + * leaves "Content-Length: 0". The request method plays no part in this. + */ + bool has_body() const + { + if (message.chunked()) + return true; + + auto value = message[http::field::content_length]; + size_t length = 0; + auto [end, ec] = std::from_chars(value.data(), value.data() + value.size(), length); + return ec != std::errc{} || end != value.data() + value.size() || length != 0; + } + + /// Called when writing the header is done. A request without a body is complete by then. + void header_written(error_code ec) + { + if (ec) + { + mlogw("async_submit: {}", what(ec)); + cancelled = true; + if (session) + client_session().request_failed(*this); + } + else if (!has_body()) + { + eof_submitted = true; + if (session) + client_session().request_complete(*this); + } + } + + void body_ended() override { client_session().request_complete(*this); } + void write_failed() override { client_session().request_failed(*this); } void content_length(std::optional content_length) override { @@ -537,6 +666,14 @@ class RequestWriter void async_submit(StatusHandler&& handler, unsigned int status_code, const Fields& headers) override { + if (!session) + { + mlogw("async_submit: session already gone"); + complete_immediately(std::move(handler), get_executor(), + make_error_code(asio::error::connection_aborted)); + return; + } + submit_headers(headers); message.method(http::verb::post); @@ -565,6 +702,34 @@ class RequestWriter }); return; } + + if (!session) + { + mlogw("async_get_response: session already gone"); + complete_immediately(std::move(handler), get_executor(), + make_error_code(asio::error::connection_aborted), + client::Response{nullptr}); + return; + } + + // + // Responses arrive in the order the requests were sent. Reading the response to this one + // has to wait until the responses to all earlier requests have been read. Instead of + // waiting, this is an error. See ClientSession for details. + // + auto& cs = client_session(); + error_code ec; + if (cs.m_receive_failed) + ec = asio::error::connection_aborted; + else if (sequence != cs.m_responses_read) + ec = asio::error::would_block; + if (ec) + { + mlogw("async_get_response: {} (request #{}, {} responses read)", what(ec), sequence, + cs.m_responses_read); + complete_immediately(std::move(handler), get_executor(), ec, client::Response{nullptr}); + return; + } response_requested = true; auto& buffer = session->m_buffer; @@ -574,8 +739,8 @@ class RequestWriter std::make_unique, decltype(buffer), http::response_parser>>( *session, stream, buffer); - session->rx = reader.get(); http::response_parser& parser = reader->parser; + parser.header_limit(header_limit(cs.client().config().max_header_size)); auto ex = get_associated_executor(handler, get_executor()); auto slot = get_associated_cancellation_slot(handler); @@ -587,7 +752,7 @@ class RequestWriter http::response_parser::value_type& msg = reader->parser.get(); mlogd("{} {}", msg.result_int(), msg.reason()); for (const auto& header : msg) - mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + mlogd(" \x1b[1;34m{}\x1b[0m: {}", truncated(header.name_string()), truncated(header.value())); } else mlogw("async_read_header: {} len={}", ec.message(), len); @@ -596,8 +761,17 @@ class RequestWriter // If reading the headers was cancelled before receiving anything, we can allow another // attempt. TODO: If we move the parser into the session, we can even relax this further. // + // As this reader has not taken anything from the connection, it does not count as having + // failed to read the response, either. + // if (ec == errc::operation_canceled && !reader->parser.got_some()) + { response_requested = false; + reader->finished = true; + } + + if (!ec && reader->parser.is_done()) // a response without body is complete already + reader->finish(); std::move(handler)(ec, client::Response(std::move(reader))); }; @@ -608,6 +782,9 @@ class RequestWriter } client::Request::GetResponseHandler responseHandler; + + /// Position of this request on the connection, which is also the position of its response. + size_t sequence = 0; }; // ================================================================================================= @@ -624,16 +801,12 @@ template BeastSession::~BeastSession() { mlogd("session deleted"); - if (wx) - { - mlogw("dtor: detaching writer"); - wx->detach(); - } - if (rx) - { - mlogw("dtor: detaching reader"); - rx->detach(); - } + if (!m_writers.empty()) + mlogw("dtor: detaching {} writer(s)", m_writers.size()); + detach_writers(); + if (!m_readers.empty()) + mlogw("dtor: detaching {} reader(s)", m_readers.size()); + detach_readers(); } template @@ -764,8 +937,7 @@ static std::optional h2c_upgrade(const http::request make_h2c_session(server::Server::Impl& server, - any_io_executor executor, - tcp_stream& stream, + any_io_executor executor, tcp_stream& stream, nghttp2::Upgrade&& upgrade) { return nghttp2::make_server_session(server, std::move(executor), stream.release_socket(), @@ -820,17 +992,16 @@ awaitable ServerSession::do_session(Buffer&& buffer) size_t requestCounter = 0; while (!m_closed) { + detach_readers(); // the previous request, if still around, is done with the stream auto reader = std::make_unique>>(*this, m_stream, m_buffer); - if (rx) - rx->detach(); - rx = reader.get(); logd(""); mlogd("waiting for request (size={} capacity={})", m_buffer.size(), m_buffer.capacity()); auto& parser = reader->parser; + parser.header_limit(header_limit(server().config().max_header_size)); auto [ec, len] = co_await async_read_header(m_stream, m_buffer, parser, as_tuple); if (!ec) mlogd("async_read_header: len={} size={} capacity={} ec={}", len, m_buffer.size(), @@ -840,6 +1011,20 @@ awaitable ServerSession::do_session(Buffer&& buffer) else mlogw("async_read_header: len={} size={} capacity={} ec=\x1b[1;31m{}\x1b[0m", len, m_buffer.size(), m_buffer.capacity(), ec.message()); + + // + // The rest of the request can not be told apart from whatever follows it on the connection, + // so there is nothing left to do after telling the client why. + // + if (ec == http::error::header_limit) + { + http::response res{http::status::request_header_fields_too_large, 11}; + res.set(http::field::server, "anyhttp"); + res.set(http::field::connection, "close"); + res.content_length(0); + if (auto [ec, n] = co_await http::async_write(m_stream, res, as_tuple); ec) + mlogw("writing 431 response: {}", ec.message()); + } if (ec) break; requestCounter++; @@ -873,7 +1058,7 @@ awaitable ServerSession::do_session(Buffer&& buffer) mlogd("{} {} (need_eof={})", request.method_string(), reader->m_url.buffer(), need_eof); for (auto& header : request) - mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + mlogd(" \x1b[1;34m{}\x1b[0m: {}", truncated(header.name_string()), truncated(header.value())); // // Upgrade to h2c, if requested: Answer with "101 Switching Protocols" and hand over the @@ -894,8 +1079,8 @@ awaitable ServerSession::do_session(Buffer&& buffer) } mlogi("upgrading to h2c, {} bytes in buffer", m_buffer.size()); - m_upgraded = make_h2c_session(server(), super::get_executor(), m_stream, - std::move(*upgrade)); + 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; @@ -904,10 +1089,8 @@ awaitable ServerSession::do_session(Buffer&& buffer) // // Prepare response. // + detach_writers(); // likewise for the previous response auto writer = std::make_shared>(*this, m_stream); - if (wx) - wx->detach(); - wx = writer.get(); http::response& response = writer->message; http::response_serializer& serializer = writer->serializer; @@ -1035,15 +1218,31 @@ template void ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url url, const Fields& headers) { + // + // Only one request can be incomplete at a time, see ClientSession. Instead of waiting for the + // previous one, which might never happen if the caller is the one to complete it, this is an + // error. + // + error_code ec; + if (m_send_failed) + ec = asio::error::connection_aborted; + else if (m_sending) + ec = asio::error::would_block; + if (ec) + { + mlogw("async_submit: {} ({})", what(ec), + m_sending ? "previous request not complete yet" : "an earlier request failed"); + complete_immediately(std::move(handler), super::get_executor(), ec, client::Request{nullptr}); + return; + } + auto writer = std::make_unique>(*this, m_stream); - wx = writer.get(); auto& request = writer->message; request.base().target(url.encoded_target()); request.method(http::verb::post); request.set(http::field::user_agent, "anyhttp"); - for (auto&& header : headers) - request.set(header.name_string(), header.value()); + add_fields(request, headers); if (request.find(http::field::host) == request.end()) request.set(http::field::host, url.encoded_authority()); if (!request.has_content_length()) @@ -1051,24 +1250,72 @@ void ClientSession::async_submit(SubmitHandler&& handler, boost::urls::u mlogd("{} {}", request.method_string(), url.buffer()); for (const auto& header : request) - mlogd(" \x1b[1;34m{}\x1b[0m: {}", header.name_string(), header.value()); + mlogd(" \x1b[1;34m{}\x1b[0m: {}", truncated(header.name_string()), truncated(header.value())); + + writer->sequence = m_requests_sent++; + m_sending = writer.get(); - // - // TODO: make writer shared? put into queue - // auto& serializer = writer->serializer; - auto cs = get_associated_cancellation_slot(handler); auto ex = get_associated_executor(handler, super::get_executor()); - auto alloc = get_associated_allocator(handler); - auto cb = [handler = std::move(handler), writer = std::move(writer), this]( - boost::system::error_code ec, size_t n) mutable { // - std::move(handler)(std::move(ec), - client::Request(std::move(writer))); - }; + auto cb = [handler = std::move(handler), writer = std::move(writer)] // + (error_code ec, size_t) mutable + { + writer->header_written(ec); + std::move(handler)(ec, client::Request(std::move(writer))); + }; async_write_header(m_stream, serializer, bind_executor(ex, std::move(cb))); } +// ------------------------------------------------------------------------------------------------- + +template +void ClientSession::request_complete(RequestWriter& request) +{ + assert(m_sending == &request); + mlogd("request #{} complete", request.sequence); + m_sending = nullptr; +} + +template +void ClientSession::request_failed(RequestWriter& request) +{ + mlogw("request #{} failed, no more requests can be sent", request.sequence); + m_send_failed = true; + m_sending = nullptr; +} + +template +void ClientSession::request_released(RequestWriter& request) +{ + // + // The connection is in the middle of this request, and nothing else can be sent any more. + // + if (m_sending == &request) + request_failed(request); + + // + // Its response will be coming, but nobody is going to read it. + // + if (!request.response_requested) + { + mlogw("request #{} released without getting its response", request.sequence); + m_receive_failed = true; + } +} + +template +void ClientSession::reader_finished(bool complete) +{ + if (complete) + ++m_responses_read; + else + { + mlogw("response #{} not read completely", m_responses_read); + m_receive_failed = true; + } +} + // ================================================================================================= template class ClientSession; diff --git a/src/h2_session.cpp b/src/h2_session.cpp index bb01089..d9d7aca 100644 --- a/src/h2_session.cpp +++ b/src/h2_session.cpp @@ -106,6 +106,24 @@ int on_header_callback(nghttp2_session* session, const nghttp2_frame* frame, con auto stream = handler->find_stream(frame->hd.stream_id); assert(stream); + // + // Beyond the limit, fields are not stored any more, but nghttp2 still has to decode them: HPACK + // state is shared by the whole connection. What happens to the stream is decided once the + // header block is complete, see NGHttp2Stream::on_request() and on_response(). + // + if (stream->header_limit_exceeded) + return 0; + + stream->header_size += header_field_size(name, value); + if (stream->header_size > handler->m_max_header_size) + { + logw("[{}] header section exceeds {} bytes, ignoring the rest", handler->logPrefix(frame), + handler->m_max_header_size); + stream->header_limit_exceeded = true; + stream->received_headers.clear(); + return 0; + } + // // Headers are logged as a block, after the request or status line, see on_frame_recv_callback(). // @@ -166,6 +184,15 @@ int on_frame_not_send_callback(nghttp2_session* session, const nghttp2_frame* fr logw("[{}] on_frame_not_send_callback: {} {}", handler->logPrefix(frame), frameType(frame->hd.type), nghttp2_strerror(lib_error_code)); + // + // nghttp2 closes the stream of a request HEADERS frame that could not be sent, but leaves the + // stream of a response open -- with neither side ever learning that there will be no response. + // Resetting it tells the peer and closes the stream here, too. + // + if (frame->hd.type == NGHTTP2_HEADERS && frame->headers.cat != NGHTTP2_HCAT_REQUEST) + nghttp2_submit_rst_stream(session, NGHTTP2_FLAG_NONE, frame->hd.stream_id, + NGHTTP2_INTERNAL_ERROR); + return 0; } @@ -445,7 +472,8 @@ void NGHttp2Session::async_submit(SubmitHandler&& handler, boost::urls::url url, logd("[{}] {} {}", stream->logPrefix, method, url.buffer()); for (auto nv : nva) - logd("[{0}] \x1b[1;34m{1:n}\x1b[0m: {1:v}", stream->logPrefix, nv); + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", stream->logPrefix, truncated(name_of(nv)), + truncated(value_of(nv))); // // https://nghttp2.org/documentation/types.html#c.nghttp2_data_source_read_callback @@ -565,7 +593,7 @@ void NGHttp2Session::close_stream(int32_t stream_id) // data. This may also happen during normal operation, if the server delivers a response // before the client calls async_get_response(). // - if (stream->has_response) + if (stream->has_response || stream->response_error) { logd("[{}] close_stream: response not delivered yet", logPrefix(stream_id)); it->second->call_read_handler(); // FIXME: this seems to be not needed diff --git a/src/h2_stream.cpp b/src/h2_stream.cpp index 2e790ad..2ef0fdb 100644 --- a/src/h2_stream.cpp +++ b/src/h2_stream.cpp @@ -244,7 +244,8 @@ void NGHttp2Writer::async_submit(StatusHandler&& handler, unsigned int sta } for (auto nv : nva) - logd("[{0}] \x1b[1;34m{1:n}\x1b[0m: {1:v}", stream->logPrefix, nv); + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", stream->logPrefix, truncated(name_of(nv)), + truncated(value_of(nv))); // TODO: If we already know that there is no body, don't set a producer. nghttp2_data_provider2 prd; @@ -805,19 +806,30 @@ ssize_t NGHttp2Stream::producer_callback(uint8_t* buf, size_t length, uint32_t* void NGHttp2Stream::log_received_headers() { for (const auto& [name, value] : received_headers) - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", logPrefix, name, value); + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", logPrefix, truncated(name), truncated(value)); received_headers.clear(); } void NGHttp2Stream::on_response() { - has_response = true; + // + // A response with too large a header section is of no use: fail it, and stop the server from + // sending its body. The stream is kept until the failure has been delivered, see close_stream(). + // + if (header_limit_exceeded) + { + response_error = boost::beast::http::error::header_limit; + nghttp2_submit_rst_stream(parent.session, NGHTTP2_FLAG_NONE, id, NGHTTP2_CANCEL); + } + else + has_response = true; + deliver_response(); } void NGHttp2Stream::deliver_response() { - if (!has_response) + if (!has_response && !response_error) { logd("[{}] deliver_response: no response, yet", logPrefix); } @@ -825,6 +837,16 @@ void NGHttp2Stream::deliver_response() { logw("[{}] deliver_response: not waiting for a response, yet", logPrefix); } + else if (response_error) + { + logw("[{}] deliver_response: {}", logPrefix, what(response_error)); + response_delivered = true; + swap_and_invoke(response_handler, response_error, client::Response{nullptr}); + + // the stream may have been kept around just for this, see close_stream() + if (closed) + parent.close_stream(id); + } else { response_delivered = true; @@ -850,7 +872,10 @@ void NGHttp2Stream::on_request() server::Response response(std::make_unique>(*this)); auto& server = dynamic_cast(parent).server(); - if (auto& handler = server.requestHandler()) + if (header_limit_exceeded) + co_spawn(get_executor(), header_fields_too_large(std::move(request), std::move(response)), + detached); + else if (auto& handler = server.requestHandler()) co_spawn(get_executor(), handler(std::move(request), std::move(response)), detached); else { diff --git a/src/h3_client.cpp b/src/h3_client.cpp index bff6a0a..62a45a7 100644 --- a/src/h3_client.cpp +++ b/src/h3_client.cpp @@ -196,7 +196,7 @@ class Http3ClientWriter : public http3::Http3Writer class Http3ClientSession : public http3::Http3Session { public: - explicit Http3ClientSession(asio::any_io_executor executor); + Http3ClientSession(asio::any_io_executor executor, const Config& config); ~Http3ClientSession() override; // @@ -270,6 +270,19 @@ void Http3ClientStream::on_pseudo_header(std::string_view name, std::string_view void Http3ClientStream::on_headers_complete() { + // + // A response with too large a header section is of no use: fail it, and stop the server from + // sending its body. + // + if (header_limit_exceeded) + { + headers_received = false; // there is no response to deliver, see deliver_failure() + failure_ec = boost::beast::http::error::header_limit; + session.reset_stream(id, NGHTTP3_H3_REQUEST_CANCELLED); + deliver_failure(); + return; + } + using namespace boost::beast::http; logd("[{}] {} {}", log_prefix, status_code, obsolete_reason(int_to_status(status_code))); log_headers(log_prefix, std::exchange(received_headers, {})); @@ -282,7 +295,8 @@ void Http3ClientStream::on_failed(boost::system::error_code ec) // A stream closing gracefully (ec success, e.g. NGHTTP3_H3_NO_ERROR) still means no response // ever arrived if headers were never received -- never report success with a null Response. // - failure_ec = ec ? ec : boost::beast::http::error::end_of_stream; + if (!failure_ec) // the first reason is the one to report, see on_headers_complete() + failure_ec = ec ? ec : boost::beast::http::error::end_of_stream; deliver_failure(); } @@ -391,9 +405,10 @@ void Http3ClientStream::deliver_response() // Http3ClientSession implementation // ================================================================================================= -Http3ClientSession::Http3ClientSession(asio::any_io_executor executor) +Http3ClientSession::Http3ClientSession(asio::any_io_executor executor, const Config& config) : http3::Http3Session(executor), socket_(get_executor()), ready_signal_(get_executor()) { + max_header_size_ = config.max_header_size; // Sentinel timers: expires_at(max) means "not yet"; a wait completes once moved to "min". ready_signal_.expires_at(asio::steady_timer::time_point::max()); logi("Http3ClientSession: ctor"); @@ -674,13 +689,14 @@ void Http3ClientSession::async_submit(SubmitHandler&& handler, boost::urls::url // ================================================================================================= awaitable> async_connect_http3(asio::any_io_executor executor, - std::string host, std::string port) + std::string host, std::string port, + const Config& config) { boost::asio::ip::udp::resolver resolver(executor); auto flags = boost::asio::ip::udp::resolver::numeric_service; auto results = co_await resolver.async_resolve(host, port, flags); // may throw - auto session = std::make_shared(executor); + auto session = std::make_shared(executor, config); if (session->init(results.begin()->endpoint()) != 0) throw boost::system::system_error(errc::make_error_code(errc::connection_refused)); diff --git a/src/h3_common.cpp b/src/h3_common.cpp index c384de1..39bfee0 100644 --- a/src/h3_common.cpp +++ b/src/h3_common.cpp @@ -3,6 +3,7 @@ // #include "anyhttp/h3_common.hpp" #include "anyhttp/common.hpp" // IWYU pragma: keep +#include "anyhttp/formatter.hpp" #include @@ -36,15 +37,15 @@ void log_headers(std::string_view log_prefix, std::span nva) { for (const auto& nv : nva) logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, - std::string_view(reinterpret_cast(nv.name), nv.namelen), - std::string_view(reinterpret_cast(nv.value), nv.valuelen)); + truncated(std::string_view(reinterpret_cast(nv.name), nv.namelen)), + truncated(std::string_view(reinterpret_cast(nv.value), nv.valuelen))); } void log_headers(std::string_view log_prefix, const std::vector>& headers) { for (const auto& [name, value] : headers) - logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, name, value); + logd("[{}] \x1b[1;34m{}\x1b[0m: {}", log_prefix, truncated(name), truncated(value)); } void ngtcp2_log_printf(void* /*user*/, const char* fmt, ...) noexcept diff --git a/src/h3_server.cpp b/src/h3_server.cpp index 8659da3..1db745a 100644 --- a/src/h3_server.cpp +++ b/src/h3_server.cpp @@ -104,10 +104,6 @@ struct Endpoint { Address addr; int fd; - - // Testing aid, see server::Config::drop_rate_rx/tx. - double drop_rate_rx = 0.0; - double drop_rate_tx = 0.0; }; // ================================================================================================= @@ -216,90 +212,14 @@ bool drop_packet(double rate) // ------------------------------------------------------------------------------------------------- -int send_udp(const Endpoint& ep, const sockaddr* sa, socklen_t salen, std::span data) -{ - if (drop_packet(ep.drop_rate_tx)) - { - // logw("*** dropping outgoing packet ({} bytes) ***", data.size()); - return 0; // pretend it went out; ngtcp2 will retransmit - } - - for (;;) - { - auto n = ::sendto(ep.fd, data.data(), data.size(), 0, sa, salen); - if (n == -1) - { - if (errno == EINTR) - continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) - return 0; // best-effort; ngtcp2 will retransmit - loge("sendto: {}", strerror(errno)); - return -1; - } - return 0; - } -} - -// ------------------------------------------------------------------------------------------------- - -// Sends a run of same-sized packets (as produced by ngtcp2_conn_write_aggregate_pkt2(), all but -// the last exactly `gso_size` bytes) with a single sendmsg() using UDP_SEGMENT (GSO), so N QUIC -// packets cost one syscall instead of N. Falls back to one sendto() per segment -- and remembers -// to do so from then on -- if the kernel/NIC doesn't support UDP_SEGMENT here. -int send_udp_gso(const Endpoint& ep, const sockaddr* sa, socklen_t salen, - std::span data, size_t gso_size, bool& no_gso) +std::optional
to_address(const sockaddr_storage& src, socklen_t len) { - // With TX dropping enabled, go packet by packet so each one can be dropped individually. - if (no_gso || data.size() <= gso_size || ep.drop_rate_tx > 0.0) - { - for (; !data.empty();) - { - auto len = std::min(gso_size, data.size()); - if (send_udp(ep, sa, salen, data.first(len)) != 0) - return -1; - data = data.subspan(len); - } - return 0; - } - - iovec msg_iov{const_cast(data.data()), data.size()}; - uint8_t msg_ctrl[CMSG_SPACE(sizeof(uint16_t))]; - msghdr msg{}; - msg.msg_name = const_cast(sa); - msg.msg_namelen = salen; - msg.msg_iov = &msg_iov; - msg.msg_iovlen = 1; - msg.msg_control = msg_ctrl; - msg.msg_controllen = sizeof(msg_ctrl); - - auto* cm = CMSG_FIRSTHDR(&msg); - cm->cmsg_level = SOL_UDP; - cm->cmsg_type = UDP_SEGMENT; - cm->cmsg_len = CMSG_LEN(sizeof(uint16_t)); - auto seg = static_cast(gso_size); - memcpy(CMSG_DATA(cm), &seg, sizeof(seg)); - - for (;;) - { - auto n = ::sendmsg(ep.fd, &msg, 0); - if (n == -1) - { - if (errno == EINTR) - continue; - if (errno == EAGAIN || errno == EWOULDBLOCK) - return 0; // best-effort; ngtcp2 will retransmit - if (errno == EINVAL || errno == EOPNOTSUPP) - { - // GSO unsupported on this socket/NIC: fall back permanently and resend as - // individual datagrams. - no_gso = true; - return send_udp_gso(ep, sa, salen, data, gso_size, no_gso); - } - loge("sendmsg (GSO): {}", strerror(errno)); - return -1; - } - return 0; - } + Address addr{}; + if (len > sizeof(addr.su)) + return std::nullopt; + std::memcpy(&addr.su, &src, len); + addr.len = len; + return addr; } } // namespace @@ -370,6 +290,7 @@ class Http3ServerSession : public http3::Http3Session void on_remove_cid(const ngtcp2_cid& cid) override; private: + int send_udp(const ngtcp2_addr& remote, std::span packet); void signal_done(); void schedule_close_timer(); void do_destroy() noexcept; // the body of destroy(), always run on executor_ @@ -383,22 +304,9 @@ class Http3ServerSession : public http3::Http3Session asio::steady_timer done_signal_; // used to wake do_session() on connection close std::vector conn_closebuf_; // buffered CONNECTION_CLOSE packet - bool no_gso_ = false; + bool no_gso_ = false; // Config::disable_gso, or sendmsg() rejected UDP_SEGMENT }; -namespace -{ -std::optional
to_address(const sockaddr_storage& src, socklen_t len) -{ - Address addr{}; - if (len > sizeof(addr.su)) - return std::nullopt; - std::memcpy(&addr.su, &src, len); - addr.len = len; - return addr; -} -} // namespace - // // What one pass of udp_on_read() hands a session: every datagram of the receive batch that was // addressed to it, copied out of the receive buffer because the session consumes them on its own @@ -520,7 +428,10 @@ void Http3ServerStream::on_headers_complete() server::Response response(std::make_unique>(*this)); auto& sv = static_cast(session).server(); - if (auto& handler = sv.requestHandler()) + if (header_limit_exceeded) + co_spawn(get_executor(), header_fields_too_large(std::move(request), std::move(response)), + detached); + else if (auto& handler = sv.requestHandler()) co_spawn(get_executor(), handler(std::move(request), std::move(response)), detached); else { @@ -575,8 +486,10 @@ Http3ServerSession::Http3ServerSession(Http3ServerImpl& server, Endpoint ep, Add : http3::Http3Session(server.config().use_strand ? asio::any_io_executor{asio::make_strand(server.get_executor())} : server.get_executor()), - server_(server), ep_(ep), remote_(remote), done_signal_(get_executor()) + server_(server), ep_(ep), remote_(remote), done_signal_(get_executor()), + no_gso_(server.config().disable_gso) { + max_header_size_ = server.config().max_header_size; log_prefix_ = std::format("h3:{}", straddr(&remote_.su.sa, remote_.len)); // @@ -674,7 +587,7 @@ void Http3ServerSession::do_destroy() noexcept std::array closebuf; ngtcp2_path_storage ps; if (auto packet = write_connection_close(closebuf, ps); !packet.empty()) - send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, packet); + send_udp(ps.path.remote, packet); } timer_.cancel(); @@ -701,10 +614,94 @@ void Http3ServerSession::on_new_cid(const ngtcp2_cid& cid) void Http3ServerSession::on_remove_cid(const ngtcp2_cid& cid) { server_.dissociate_quic_cid(cid); } +// +// Sends a run of same-sized packets (as produced by ngtcp2_conn_write_aggregate_pkt2(), all but +// the last exactly `gso_size` bytes) with a single sendmsg() using UDP_SEGMENT (GSO), so N QUIC +// packets cost one syscall instead of N. Falls back to one sendto() per segment -- and remembers +// to do so from then on -- if the kernel/NIC doesn't support UDP_SEGMENT here. +// int Http3ServerSession::send_datagrams(const ngtcp2_path& path, std::span data, size_t gso_size) { - return send_udp_gso(ep_, path.remote.addr, path.remote.addrlen, data, gso_size, no_gso_); + // With TX dropping enabled, go packet by packet so each one can be dropped individually. + if (no_gso_ || data.size() <= gso_size || server_.config().drop_rate_tx > 0.0) + { + for (; !data.empty();) + { + auto len = std::min(gso_size, data.size()); + if (send_udp(path.remote, data.first(len)) != 0) + return -1; + data = data.subspan(len); + } + return 0; + } + + iovec msg_iov{const_cast(data.data()), data.size()}; + uint8_t msg_ctrl[CMSG_SPACE(sizeof(uint16_t))]; + msghdr msg{}; + msg.msg_name = path.remote.addr; + msg.msg_namelen = path.remote.addrlen; + msg.msg_iov = &msg_iov; + msg.msg_iovlen = 1; + msg.msg_control = msg_ctrl; + msg.msg_controllen = sizeof(msg_ctrl); + + auto* cm = CMSG_FIRSTHDR(&msg); + cm->cmsg_level = SOL_UDP; + cm->cmsg_type = UDP_SEGMENT; + cm->cmsg_len = CMSG_LEN(sizeof(uint16_t)); + auto seg = static_cast(gso_size); + memcpy(CMSG_DATA(cm), &seg, sizeof(seg)); + + for (;;) + { + auto n = ::sendmsg(ep_.fd, &msg, 0); + if (n == -1) + { + if (errno == EINTR) + continue; + if (errno == EAGAIN || errno == EWOULDBLOCK) + return 0; // best-effort; ngtcp2 will retransmit + if (errno == EINVAL || errno == EOPNOTSUPP) + { + // GSO unsupported on this socket/NIC: fall back permanently and resend as + // individual datagrams. + no_gso_ = true; + return send_datagrams(path, data, gso_size); + } + loge("[{}] sendmsg (GSO): {}", log_prefix_, strerror(errno)); + return -1; + } + return 0; + } +} + +// +// Sends a single QUIC packet. Also where the TX half of the packet-loss testing aid sits, see +// server::Config::drop_rate_tx. +// +int Http3ServerSession::send_udp(const ngtcp2_addr& remote, std::span packet) +{ + if (drop_packet(server_.config().drop_rate_tx)) + { + // logw("*** dropping outgoing packet ({} bytes) ***", packet.size()); + return 0; // pretend it went out; ngtcp2 will retransmit + } + + for (;;) + { + auto n = ::sendto(ep_.fd, packet.data(), packet.size(), 0, remote.addr, remote.addrlen); + if (n == -1) + { + if (errno == EINTR) + continue; + if (errno == EAGAIN || errno == EWOULDBLOCK) + return 0; // best-effort; ngtcp2 will retransmit + loge("[{}] sendto: {}", log_prefix_, strerror(errno)); + return -1; + } + return 0; + } } // ------------------------------------------------------------------------------------------------- @@ -803,7 +800,7 @@ int Http3ServerSession::handle_error(int /*rv*/) { conn_closebuf_.resize(packet.size()); logi("[{}] sending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_, ps.path.remote.addr, ps.path.remote.addrlen, conn_closebuf_); + send_udp(ps.path.remote, conn_closebuf_); } else { @@ -845,7 +842,7 @@ void Http3ServerSession::resend_conn_close() if (!path) return; logd("[{}] resending CONNECTION_CLOSE", log_prefix_); - send_udp(ep_, path->remote.addr, path->remote.addrlen, conn_closebuf_); + send_udp(path->remote, conn_closebuf_); } // ================================================================================================= @@ -876,11 +873,13 @@ Http3ServerImpl::Http3ServerImpl(Server::Impl& parent, const asio::ip::udp::endp socket_->set_option(socket_option::integer(1)); socket_->set_option(socket_option::integer(1)); } - socket_->set_option(socket_option::integer(1)); + if (!config().disable_gro) + socket_->set_option(socket_option::integer(1)); socket_->non_blocking(true); socket_->bind(endpoint); - logi("Server: UDP listening on {}", endpoint); + logi("Server: UDP listening on {} (GRO {}, GSO {})", endpoint, + config().disable_gro ? "off" : "on", config().disable_gso ? "off" : "on"); } // ------------------------------------------------------------------------------------------------- @@ -1017,7 +1016,7 @@ int Http3ServerImpl::udp_on_read(Endpoint& ep) auto data = all_data.subspan(0, std::min(seg_size, all_data.size())); all_data = all_data.subspan(data.size()); - if (drop_packet(ep.drop_rate_rx)) + if (drop_packet(config().drop_rate_rx)) { // logw("*** dropping received packet ({} bytes) ***", data.size()); continue; @@ -1215,8 +1214,6 @@ awaitable Http3ServerImpl::udp_receive_loop() Endpoint ep{}; ep.fd = socket_->native_handle(); - ep.drop_rate_rx = config().drop_rate_rx; - ep.drop_rate_tx = config().drop_rate_tx; auto local = socket_->local_endpoint(); auto data = local.data(); std::memcpy(&ep.addr.su, data, local.size()); diff --git a/src/h3_session.cpp b/src/h3_session.cpp index a19472e..b04504b 100644 --- a/src/h3_session.cpp +++ b/src/h3_session.cpp @@ -499,6 +499,14 @@ int Http3Session::setup_http3() settings.qpack_max_dtable_capacity = 4096; settings.qpack_blocked_streams = 100; + // + // Only announced to the peer: nghttp3 does not enforce it, see Http3Stream::on_header(). The + // setting goes out either way, as nghttp3 always writes it into its SETTINGS frame. Leaving it + // at the default would announce an unlimited header section, and take four more bytes to do so. + // * https://datatracker.ietf.org/doc/html/rfc9114#section-4.2.2 + // + settings.max_field_section_size = max_header_size_; + if (is_server) { if (auto rv = nghttp3_conn_server_new(&h3_, &h3cb, &settings, nullptr, this); rv != 0) diff --git a/src/h3_stream.cpp b/src/h3_stream.cpp index db43f81..cb56dbb 100644 --- a/src/h3_stream.cpp +++ b/src/h3_stream.cpp @@ -616,6 +616,23 @@ void Http3Stream::finish_active_write() void Http3Stream::on_header(std::string_view name, std::string_view value) { + // + // Beyond the limit, fields are not stored any more. What happens to the stream is decided once + // the field section is complete, see on_headers_complete() of the server and client streams. + // + if (header_limit_exceeded) + return; + + header_size += header_field_size(name, value); + if (header_size > session.max_header_size()) + { + logw("[{}] header section exceeds {} bytes, ignoring the rest", log_prefix, + session.max_header_size()); + header_limit_exceeded = true; + received_headers.clear(); + return; + } + if (spdlog::default_logger_raw()->should_log(spdlog::level::debug)) received_headers.emplace_back(name, value); diff --git a/src/request_handlers.cpp b/src/request_handlers.cpp index 6db0e1b..1d42440 100644 --- a/src/request_handlers.cpp +++ b/src/request_handlers.cpp @@ -116,6 +116,12 @@ awaitable not_found(server::Request, server::Response response) co_await response.async_write_eof(); } +awaitable header_fields_too_large(server::Request, server::Response response) +{ + co_await response.async_submit(431, {}); + co_await response.async_write_eof(); +} + awaitable eat_request(server::Request request, server::Response response) { logd("eat_request: going to eat {} bytes", request.content_length().value_or(-1)); diff --git a/src/server_main.cpp b/src/server_main.cpp index f5181af..603897b 100644 --- a/src/server_main.cpp +++ b/src/server_main.cpp @@ -19,6 +19,9 @@ #include #include +#include +#include + namespace rv = std::ranges::views; using namespace std::chrono_literals; @@ -37,8 +40,12 @@ std::expected parseConfig(int argc, char* argv[]) { Config config; - // Define program options - po::options_description desc("Allowed options"); + // Define program options, wrapping the help text at the terminal's width (it goes to stderr) + winsize ws{}; + unsigned columns = ::ioctl(STDERR_FILENO, TIOCGWINSZ, &ws) == 0 && ws.ws_col >= 40 + ? ws.ws_col + : po::options_description::m_default_line_length; + po::options_description desc("Allowed options", columns, columns / 2); auto opts = desc.add_options(); opts("help,h", "produce help message"); opts("verbose,v", po::value>()->zero_tokens()->composing(), @@ -50,6 +57,13 @@ std::expected parseConfig(int argc, char* argv[]) "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a received QUIC packet"); opts("drop-tx", po::value(&config.server.drop_rate_tx)->default_value(0.0), "HTTP/3 testing: probability (0.0 .. 1.0) of dropping a QUIC packet before sending it"); + opts("disable-gro", po::bool_switch(&config.server.disable_gro), + "HTTP/3 benchmarking: don't enable UDP_GRO (receive offload) on the UDP socket"); + opts("disable-gso", po::bool_switch(&config.server.disable_gso), + "HTTP/3 benchmarking: don't use UDP_SEGMENT (send offload), one sendto() per packet"); + opts("max-header-size", + po::value(&config.server.max_header_size)->default_value(config.server.max_header_size), + "largest request header section accepted, in bytes (answered with 431 if exceeded)"); po::variables_map vm; try @@ -138,6 +152,14 @@ int main(int argc, char* argv[]) co_await serve_file(std::move(request), std::move(response), "test", "/test"); else if (path == "/eat_request") co_await eat_request(std::move(request), std::move(response)); + else if (path == "/upload") + { + // Unlike eat_request, respond only after the whole body is in: clients such as h2load + // stop uploading as soon as the response is complete. + co_await drain(request); + co_await response.async_submit(200, {}); + co_await response.async_write_eof(); + } else if (path == "/" || path == "/h2spec") co_await h2spec(std::move(request), std::move(response)); else diff --git a/test/test_client_async.cpp b/test/test_client_async.cpp index aab2c27..0275759 100644 --- a/test/test_client_async.cpp +++ b/test/test_client_async.cpp @@ -123,6 +123,151 @@ TEST_P(ClientAsync, WHEN_get_response_is_detached_THEN_does_not_crash) }; } +// ------------------------------------------------------------------------------------------------- + +// +// Requests and responses may outlive the session they belong to. Once it is gone, there is no +// connection left for them to use: whatever they are asked to do has to complete with an error, +// without touching what used to be the session's stream. +// +// On the client side, only an HTTP/1.1 session is gone as soon as it is reset(). HTTP/2 and HTTP/3 +// sessions shut down asynchronously, so operations may still complete successfully for a little +// while, with data that has already arrived. +// +static bool is_connection_error(const boost::system::error_code& ec) +{ + return ec == boost::system::errc::connection_aborted || + ec == boost::system::errc::connection_reset; +} + +TEST_P(ClientAsync, WHEN_session_is_gone_THEN_request_reports_error) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); + + test = [this](Session session) -> awaitable + { + auto request = co_await session.async_submit(url.set_path("echo"), {}); + session.reset(); + + auto [ec] = co_await request.async_write(asio::buffer("Hello"sv), as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + + std::tie(ec) = co_await request.async_write_eof(as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + + auto [ec2, response] = co_await request.async_get_response(as_tuple); + EXPECT_TRUE(is_connection_error(ec2)) << what(ec2); + }; +} + +TEST_P(ClientAsync, WHEN_server_session_is_gone_THEN_response_reports_error) +{ + auto responded = std::make_shared(false); + custom = [responded](server::Request request, server::Response response) -> awaitable + { + // + // Keep the response around beyond the request handler, until the client has closed the + // connection and the server session has ended. + // + co_spawn(co_await this_coro::executor, + [responded, response = std::move(response)]() mutable -> awaitable + { + co_await sleep(100ms); + + auto [ec] = co_await response.async_submit(200, {}, as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + + std::tie(ec) = co_await response.async_write(asio::buffer("Hello"sv), as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + + *responded = true; + }, detached); + co_return; + }; + test = [this, responded](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, {}); + co_await request.async_write_eof(); + request.reset(); + session.reset(); + + for (int i = 0; i < 100 && !*responded; ++i) + co_await sleep(10ms); + EXPECT_TRUE(*responded); + }; +} + +// +// With HTTP/1.1 pipelining, a session has more than one request at a time. Releasing the earlier +// one must not make the session forget about the later one, which has to learn about the session +// going away all the same. +// +TEST_P(ClientAsync, WHEN_earlier_request_is_released_THEN_later_request_still_learns_session_is_gone) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); + + 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"), {}); + + request1.reset(); + session.reset(); + + auto [ec] = co_await request2.async_write(asio::buffer("Hello"sv), as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + }; +} + +TEST_P(ClientAsync, WHEN_session_is_gone_THEN_earlier_request_reports_error) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); + + 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"), {}); + + session.reset(); + + auto [ec, response] = co_await request1.async_get_response(as_tuple); + EXPECT_TRUE(is_connection_error(ec)) << what(ec); + }; +} + +TEST_P(ClientAsync, WHEN_earlier_response_is_released_THEN_later_response_still_learns_session_is_gone) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); + + 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(asio::buffer("Hello, Server #2!"sv)); + + auto response1 = co_await request1.async_get_response(); + EXPECT_EQ(co_await drain(response1), 17); + auto response2 = co_await request2.async_get_response(); + + response1.reset(); + session.reset(); + + // + // The body of the second response is still open, as its request has not been ended. + // + std::array buffer; + auto [ec, n] = co_await response2.async_read_some(asio::buffer(buffer), as_tuple); + EXPECT_EQ(ec, boost::beast::http::error::partial_message) << what(ec); + }; +} + TEST_P(ClientAsync, WHEN_server_discards_request_while_writing_THEN_connection_is_reset) { custom = [this](server::Request request, server::Response response) -> awaitable @@ -714,11 +859,10 @@ TEST_P(ClientAsync, WHEN_request_is_sent_THEN_response_is_received_before_body_i // ------------------------------------------------------------------------------------------------- // -// 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. +// HTTP/1.1 supports pipelining: multiple requests can be made before the responses are received. +// On the wire, requests and responses can not be interleaved, though, so the HTTP/1.1 client puts +// them in order. See ClientSession in h1_session.hpp for the rules; HTTP/2 and HTTP/3 multiplex +// requests and don't need any of them. // TEST_P(ClientAsync, WHEN_multiple_request_are_made_THEN_responses_are_received_in_order) { @@ -738,6 +882,192 @@ TEST_P(ClientAsync, WHEN_multiple_request_are_made_THEN_responses_are_received_i }; } +static constexpr auto body1 = "Hello, Server #1!"sv; +static constexpr auto body2 = "Hello, Server #2! XYZ"sv; + +// +// HTTP/1.1 behaves like a protocol with "max concurrent streams = 1": submitting has to wait for +// the previous request to be complete. Instead of waiting, which would deadlock here, it reports +// an error, so that it can be retried later. +// +// TODO: HTTP/2 and HTTP/3 should behave the same way when the peer limits concurrent streams. +// +TEST_P(ClientAsync, WHEN_request_is_submitted_before_previous_is_complete_THEN_reports_would_block) +{ + test = [this](Session session) -> awaitable + { + const bool limited = GetParam() == anyhttp::Protocol::http11; + + auto request1 = co_await session.async_submit(url.set_path("echo"), {}); + auto [ec, request2] = co_await session.async_submit(url.set_path("echo"), {}, as_tuple); + if (limited) + { + EXPECT_EQ(ec, asio::error::would_block); + EXPECT_FALSE(request2); + } + else + EXPECT_FALSE(ec) << what(ec); + + co_await request1.async_write_eof(asio::buffer(body1)); + if (limited) + request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer(body2)); + + auto response1 = co_await request1.async_get_response(); + EXPECT_EQ(co_await read(response1), body1); + auto response2 = co_await request2.async_get_response(); + EXPECT_EQ(co_await read(response2), body2); + }; +} + +TEST_P(ClientAsync, WHEN_many_requests_are_made_THEN_all_are_answered_in_order) +{ + test = [this](Session session) -> awaitable + { + std::vector requests; + for (size_t i = 0; i < 10; ++i) + { + requests.push_back(co_await session.async_submit(url.set_path("echo"), {})); + co_await requests.back().async_write_eof(asio::buffer(std::format("request #{}", i))); + } + + for (size_t i = 0; i < requests.size(); ++i) + { + auto response = co_await requests[i].async_get_response(); + EXPECT_EQ(co_await read(response), std::format("request #{}", i)); + } + }; +} + +// +// Likewise, getting a response has to wait until the responses to all earlier requests have been +// read -- otherwise, it would read one of those. +// +TEST_P(ClientAsync, WHEN_getting_response_before_previous_is_read_THEN_reports_would_block) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); // requests are multiplexed, nothing to wait for + + test = [this](Session session) -> awaitable + { + auto request1 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request1.async_write_eof(asio::buffer(body1)); + auto request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer(body2)); + + auto [ec, response2] = co_await request2.async_get_response(as_tuple); + EXPECT_EQ(ec, asio::error::would_block) << "response #1 not requested yet"; + + auto response1 = co_await request1.async_get_response(); + std::tie(ec, response2) = co_await request2.async_get_response(as_tuple); + EXPECT_EQ(ec, asio::error::would_block) << "response #1 not read yet"; + + EXPECT_EQ(co_await read(response1), body1); + response2 = co_await request2.async_get_response(); + EXPECT_EQ(co_await read(response2), body2); + }; +} + +// +// A request without a body -- "Content-Length: 0" -- is complete as soon as it has been submitted, +// so the next one can follow right away. Its body has been ended implicitly: ending it again is a +// no-op, and data has nowhere to go. +// +TEST_P(ClientAsync, WHEN_request_has_no_body_THEN_it_is_complete_after_submit) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); // requests are multiplexed, nothing to wait for + + test = [this](Session session) -> awaitable + { + auto request1 = + co_await session.async_submit(url.set_path("echo"), fields({{"Content-Length", 0}})); + auto request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer(body2)); + + auto [ec] = co_await request1.async_write_eof(as_tuple); + EXPECT_FALSE(ec) << what(ec); + std::tie(ec) = co_await request1.async_write(asio::buffer(body1), as_tuple); + EXPECT_EQ(ec, boost::system::errc::broken_pipe) << what(ec); + + auto response1 = co_await request1.async_get_response(); + EXPECT_EQ(co_await read(response1), ""); + auto response2 = co_await request2.async_get_response(); + EXPECT_EQ(co_await read(response2), body2); + }; +} + +// +// A request with a body is complete only once async_write_eof() has succeeded, even if all of its +// 'Content-Length' has been written already. +// +TEST_P(ClientAsync, WHEN_content_length_is_written_without_eof_THEN_request_is_not_complete) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); // requests are multiplexed, nothing to wait for + + test = [this](Session session) -> awaitable + { + auto request1 = co_await session.async_submit(url.set_path("echo"), + fields({{"Content-Length", body1.size()}})); + co_await request1.async_write(asio::buffer(body1)); + + auto [ec, request2] = co_await session.async_submit(url.set_path("echo"), {}, as_tuple); + EXPECT_EQ(ec, asio::error::would_block); + + co_await request1.async_write_eof(); + request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer(body2)); + + auto response1 = co_await request1.async_get_response(); + EXPECT_EQ(co_await read(response1), body1); + auto response2 = co_await request2.async_get_response(); + EXPECT_EQ(co_await read(response2), body2); + }; +} + +// +// A request that goes away before it is complete leaves the connection in the middle of a message: +// nothing can be sent after it any more. +// +TEST_P(ClientAsync, WHEN_incomplete_request_is_released_THEN_later_requests_report_error) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); // requests are multiplexed, and independent of each other + + test = [this](Session session) -> awaitable + { + auto request1 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request1.async_write(asio::buffer(body1)); + request1.reset(); + + auto [ec, request2] = co_await session.async_submit(url.set_path("echo"), {}, as_tuple); + EXPECT_EQ(ec, asio::error::connection_aborted); + }; +} + +// +// A request that has been sent, but goes away without asking for its response, leaves that +// response unread on the connection -- and with it, all responses after it. +// +TEST_P(ClientAsync, WHEN_request_is_released_without_getting_response_THEN_later_responses_report_error) +{ + if (GetParam() != anyhttp::Protocol::http11) + GTEST_SKIP(); // requests are multiplexed, and independent of each other + + test = [this](Session session) -> awaitable + { + auto request1 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request1.async_write_eof(asio::buffer(body1)); + auto request2 = co_await session.async_submit(url.set_path("echo"), {}); + co_await request2.async_write_eof(asio::buffer(body2)); + request1.reset(); + + auto [ec, response] = co_await request2.async_get_response(as_tuple); + EXPECT_EQ(ec, asio::error::connection_aborted); + }; +} + // ------------------------------------------------------------------------------------------------- TEST_P(ClientAsync, EatRequest) diff --git a/test/test_fixtures.hpp b/test/test_fixtures.hpp index 8b2e70d..60f2d8f 100644 --- a/test/test_fixtures.hpp +++ b/test/test_fixtures.hpp @@ -113,6 +113,7 @@ class Server : public testing::TestWithParam auto config = server::Config{.listen_address = "127.0.0.2", .port = 0}; config.use_strand = threads() > 1; + configure_server(config); // // The main server acceptor loop does not need to run on a strand. Instead, a per-connection @@ -168,6 +169,9 @@ class Server : public testing::TestWithParam context.run(); } + /// Lets a derived fixture adjust the server configuration before the server is created. + virtual void configure_server(server::Config&) {} + protected: boost::asio::io_context context; std::optional server; @@ -184,6 +188,7 @@ class Client : public Server Server::SetUp(); url.set_port_number(server->local_endpoint().port()); client::Config config{.url = url, .protocol = GetParam()}; + configure_client(config); #if defined(MULTITHREADED) client.emplace(make_strand(context.get_executor()), config); #else @@ -191,6 +196,9 @@ class Client : public Server #endif } + /// Lets a derived fixture adjust the client configuration before the client is created. + virtual void configure_client(client::Config&) {} + protected: boost::urls::url url{"http://127.0.0.2/custom"}; std::optional client; diff --git a/test/test_formatter.cpp b/test/test_formatter.cpp index 1140a83..81544bc 100644 --- a/test/test_formatter.cpp +++ b/test/test_formatter.cpp @@ -239,3 +239,25 @@ TEST(FormatterTest, NgHttp2NvEmptyValue) auto formatted = std::format("{}", nv); EXPECT_EQ(formatted, "some-header="); } + +// ================================================================================================= +// Test anyhttp::truncated() +// ================================================================================================= + +TEST(FormatterTest, TruncatedShortStringIsUnchanged) +{ + EXPECT_EQ(std::format("{}", anyhttp::truncated("content-type")), "content-type"); + EXPECT_EQ(std::format("{}", anyhttp::truncated("")), ""); +} + +TEST(FormatterTest, TruncatedAtLimitIsUnchanged) +{ + std::string value(anyhttp::max_logged_size, 'x'); + EXPECT_EQ(std::format("{}", anyhttp::truncated(value)), value); +} + +TEST(FormatterTest, TruncatedLongStringIsCutShort) +{ + std::string value(30000, 'x'); + EXPECT_EQ(std::format("{}", anyhttp::truncated(value, 4)), "xxxx... (30000 bytes, truncated)"); +} diff --git a/test/test_gtest.cpp b/test/test_gtest.cpp deleted file mode 100644 index 2195666..0000000 --- a/test/test_gtest.cpp +++ /dev/null @@ -1,7 +0,0 @@ -#include -#include - -TEST(GTest, HelloWorld) -{ - ASSERT_TRUE(true); -} diff --git a/test/test_headers.cpp b/test/test_headers.cpp new file mode 100644 index 0000000..7848217 --- /dev/null +++ b/test/test_headers.cpp @@ -0,0 +1,358 @@ +#include "test_fixtures.hpp" + +#include + +using namespace testing; + +// ================================================================================================= + +// +// Large and numerous header fields, in both directions. +// +// HTTP/2 splits a header block that does not fit into a single frame (16 KiB by default) into +// HEADERS + CONTINUATION frames, HTTP/3 compresses the whole field section with QPACK, and +// HTTP/1.1 parses it with a Beast parser. All of them are subject to Config::max_header_size. +// +class Headers : public ClientAsync +{ +protected: + void round_trip(Fields sent); +}; + +INSTANTIATE_TEST_SUITE_P(Headers, Headers, + Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +/// Generates \p count fields named x-header-, each with a value of \p size characters. +static Fields make_fields(size_t count, size_t size) +{ + Fields result; + for (size_t i = 0; i < count; ++i) + { + auto value = std::format("value-{}-", i); + value.resize(std::max(size, value.size()), char('a' + i % 26)); + result.insert(std::format("x-header-{}", i), value); + } + return result; +} + +/// All values of the fields named \p name, in the order they appear in. +static std::vector values_of(const Fields& fields, std::string_view name) +{ + auto [begin, end] = fields.equal_range(name); + return std::ranges::subrange(begin, end) | + rv::transform([](auto& field) { return std::string_view(field.value()); }) | + std::ranges::to(); +} + +/// All fields as (name, value) pairs, in the order they appear in. +static std::vector> pairs_of(const Fields& fields) +{ + return fields | rv::transform([](auto& field) { + return std::pair(std::string_view(field.name_string()), + std::string_view(field.value())); + }) | + std::ranges::to(); +} + +/// Expects every field of \p expected to be found in \p actual. +static void expect_contains(const Fields& actual, const Fields& expected) +{ + auto fields = pairs_of(actual); + for (auto&& field : expected) + EXPECT_THAT(fields, Contains(Pair(StrCaseEq(field.name_string()), field.value()))); +} + +/// Number of bytes the fields take up on an HTTP/1.1 wire, roughly. +static size_t wire_size(const Fields& fields) +{ + size_t result = 0; + for (auto&& field : fields) + result += field.name_string().size() + field.value().size() + 4; // ": " and CRLF + return result; +} + +// ------------------------------------------------------------------------------------------------- + +// +// Sends the fields with the request, and has the server send them back with the response. +// +void Headers::round_trip(Fields sent) +{ + custom = [sent](server::Request request, server::Response response) -> awaitable + { + expect_contains(request.fields(), sent); + co_await drain(request); + auto fields = sent; + fields.set("Content-Length", "0"); + co_await response.async_submit(200, fields); + co_await response.async_write_eof(); + }; + test = [this, sent](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, sent); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(response.status_code(), 200); + expect_contains(response.fields(), sent); + EXPECT_EQ(co_await drain(response), 0); + }; +} + +TEST_P(Headers, WHEN_many_small_headers_THEN_all_arrive_in_both_directions) +{ + round_trip(make_fields(200, 8)); // well below the 8 KiB HTTP/1.1 limit +} + +TEST_P(Headers, WHEN_single_header_is_large_THEN_arrives_intact_in_both_directions) +{ + round_trip(make_fields(1, 6000)); +} + +TEST_P(Headers, WHEN_header_name_repeats_THEN_all_values_arrive_in_order) +{ + Fields sent; + std::vector values; + for (size_t i = 0; i < 50; ++i) + sent.insert("x-repeated", values.emplace_back(std::format("value-{}", i))); + + custom = [sent, values](server::Request request, server::Response response) -> awaitable + { + EXPECT_THAT(values_of(request.fields(), "x-repeated"), ElementsAreArray(values)); + co_await drain(request); + auto fields = sent; + fields.set("Content-Length", "0"); + co_await response.async_submit(200, fields); + co_await response.async_write_eof(); + }; + test = [this, sent, values](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, sent); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_THAT(values_of(response.fields(), "x-repeated"), ElementsAreArray(values)); + co_await drain(response); + }; +} + +// +// Larger than a single HTTP/2 frame (16 KiB), so the header block goes out as HEADERS followed by +// CONTINUATION frames. +// +TEST_P(Headers, WHEN_headers_exceed_frame_size_THEN_all_arrive_in_both_directions) +{ + auto sent = make_fields(32, 1100); + ASSERT_GT(wire_size(sent), 32_k); + round_trip(sent); +} + +TEST_P(Headers, WHEN_single_header_exceeds_frame_size_THEN_arrives_intact_in_both_directions) +{ + round_trip(make_fields(1, 40_k)); +} + +TEST_P(Headers, WHEN_request_headers_exceed_default_limit_THEN_server_responds_431) +{ + auto sent = make_fields(3, 30_k); + ASSERT_GT(wire_size(sent), default_max_header_size); + + custom = [](server::Request request, server::Response response) -> awaitable + { + ADD_FAILURE() << "request handler called for oversized request headers"; + co_await response.async_submit(200, {}); + co_await response.async_write_eof(); + }; + test = [this, sent](Session session) -> awaitable + { + auto request = co_await session.async_submit(url, sent); + co_await request.async_write_eof(); + auto response = co_await request.async_get_response(); + EXPECT_EQ(response.status_code(), 431); + }; +} + +// ================================================================================================= + +// +// Header sections beyond Config::max_header_size, with a small limit on both sides. +// +// The receiving side stops storing fields as soon as the limit is exceeded. A server answers such a +// request with 431 without calling the request handler, a client fails async_get_response() with +// http::error::header_limit. Both HTTP/2 and HTTP/3 keep the session usable, as only the stream is +// affected, while an HTTP/1.1 connection is closed. +// +class HeaderLimits : public ClientAsync +{ +protected: + static constexpr size_t limit = 4_k; + + void configure_server(server::Config& config) override { config.max_header_size = limit; } + void configure_client(client::Config& config) override { config.max_header_size = limit; } + + /// Request handler: responds with the headers of size \p response_size given as query parameter. + void respond_with_headers() + { + custom = [this](server::Request request, server::Response response) -> awaitable + { + ++handled; + auto size = request.get_param_as("response_size").value_or(0); + co_await drain(request); + auto fields = make_fields(1, size); + fields.set("Content-Length", "0"); + auto [ec] = co_await response.async_submit(200, fields, as_tuple); + if (!ec) + std::tie(ec) = co_await response.async_write_eof(as_tuple); + logi("server: {}", what(ec)); + }; + } + + /// Sends a request with \p sent headers, returns the response status code or error. + awaitable> request(Session& session, const Fields& sent, + size_t response_size = 0) + { + auto target = url; + if (response_size) + target.params().set("response_size", std::to_string(response_size)); + + auto [ec, request] = co_await session.async_submit(target, sent, as_tuple); + if (!ec) + std::tie(ec) = co_await request.async_write_eof(as_tuple); + if (ec) + co_return std::unexpected(ec); + + auto [ec2, response] = co_await request.async_get_response(as_tuple); + if (ec2) + co_return std::unexpected(ec2); + co_await drain(response); + co_return response.status_code(); + } + + size_t handled = 0; +}; + +INSTANTIATE_TEST_SUITE_P(HeaderLimits, HeaderLimits, + Values(anyhttp::Protocol::http11, anyhttp::Protocol::h2, + anyhttp::Protocol::h3), + NameGenerator); + +// ------------------------------------------------------------------------------------------------- + +TEST_P(HeaderLimits, WHEN_request_headers_are_within_limit_THEN_request_is_handled) +{ + respond_with_headers(); + test = [this](Session session) -> awaitable + { + EXPECT_EQ(co_await request(session, make_fields(1, limit / 2), limit / 2), 200); + EXPECT_EQ(handled, 1); + }; +} + +TEST_P(HeaderLimits, WHEN_request_headers_exceed_limit_THEN_server_responds_431) +{ + respond_with_headers(); + test = [this](Session session) -> awaitable + { + EXPECT_EQ(co_await request(session, make_fields(1, limit)), 431); + EXPECT_EQ(handled, 0); + }; +} + +// +// With HTTP/2 and HTTP/3, each field counts 32 bytes more than its name and value, so many small +// fields exceed the limit early. +// +TEST_P(HeaderLimits, WHEN_many_small_fields_exceed_limit_THEN_server_responds_431) +{ + respond_with_headers(); + test = [this](Session session) -> awaitable + { + auto sent = make_fields(200, 1); + EXPECT_EQ(co_await request(session, sent), 431); + EXPECT_EQ(handled, 0); + }; +} + +// +// A header section 200 times the limit: the server must not store it, but tell the client. +// +// Except for HTTP/2, where it takes up more CONTINUATION frames than a header section within the +// limit ever needs. That is a flood, and nghttp2 closes the connection. +// +TEST_P(HeaderLimits, WHEN_request_headers_far_exceed_limit_THEN_request_is_rejected) +{ + auto sent = make_fields(25, 32_k); + ASSERT_GT(wire_size(sent), limit * 200); + + respond_with_headers(); + test = [this, sent](Session session) -> awaitable + { + auto result = co_await request(session, sent); + if (GetParam() == anyhttp::Protocol::h2) + EXPECT_FALSE(result.has_value()) << "status " << result.value_or(0); + else + EXPECT_EQ(result, 431); + EXPECT_EQ(handled, 0); + }; +} + +TEST_P(HeaderLimits, WHEN_request_is_rejected_THEN_session_serves_next_request) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP() << "HTTP/1.1 closes the connection after 431"; + + respond_with_headers(); + test = [this](Session session) -> awaitable + { + EXPECT_EQ(co_await request(session, make_fields(1, limit)), 431); + EXPECT_EQ(co_await request(session, make_fields(1, 100)), 200); + EXPECT_EQ(handled, 1); + }; +} + +// ------------------------------------------------------------------------------------------------- + +TEST_P(HeaderLimits, WHEN_response_headers_exceed_limit_THEN_get_response_fails) +{ + respond_with_headers(); + test = [this](Session session) -> awaitable + { + auto result = co_await request(session, {}, limit); + EXPECT_EQ(result, std::unexpected(error_code(boost::beast::http::error::header_limit))); + }; +} + +// +// The response has been rejected long before it is asked for: the failure must still be delivered. +// +TEST_P(HeaderLimits, WHEN_response_headers_exceed_limit_before_get_response_THEN_it_fails) +{ + respond_with_headers(); + test = [this](Session session) -> awaitable + { + auto target = url; + target.params().set("response_size", std::to_string(limit)); + auto request = co_await session.async_submit(target, {}); + co_await request.async_write_eof(); + co_await sleep(100ms); + auto [ec, response] = co_await request.async_get_response(as_tuple); + EXPECT_EQ(ec, boost::beast::http::error::header_limit) << what(ec); + }; +} + +TEST_P(HeaderLimits, WHEN_response_is_rejected_THEN_session_serves_next_request) +{ + if (GetParam() == anyhttp::Protocol::http11) + GTEST_SKIP() << "HTTP/1.1 can not skip the rest of a response"; + + respond_with_headers(); + test = [this](Session session) -> awaitable + { + auto result = co_await request(session, {}, limit); + EXPECT_EQ(result, std::unexpected(error_code(boost::beast::http::error::header_limit))); + EXPECT_EQ(co_await request(session, {}, 100), 200); + EXPECT_EQ(handled, 2); + }; +}