LibHTTP+LibCore+RequestServer: Use async streams for HTTP

We no longer use blocking reads and writes, yielding a nice performance
boost and a significantly cleaner (and more readable) implementation :^)

AK/AsyncStreamHelpers.h

@@ -13,14 +13,17 @@
 namespace AK {
 
 struct AsyncStreamHelpers {
-    static Coroutine<ErrorOr<ReadonlyBytes>> consume_until(AsyncInputStream& stream, StringView delimiter)
+    static Coroutine<ErrorOr<ReadonlyBytes>> consume_until(AsyncInputStream& stream, StringView delimiter, Optional<size_t> max_size = {})
     {
         size_t start_position = 0;
         while (true) {
             auto buffer = CO_TRY(co_await stream.peek());
             auto position = StringUtils::find(StringView { buffer }, delimiter, start_position);
-            if (position.has_value())
-                co_return must_sync(stream.read(*position + delimiter.length()));
+            auto end_position = position.map([&](auto value) { return value + delimiter.length(); });
+            if (max_size.has_value() && !end_position.has_value() && buffer.size() >= max_size.value())
+                end_position = buffer.size();
+            if (end_position.has_value())
+                co_return must_sync(stream.read(*end_position));
             start_position = max(buffer.size(), delimiter.length() - 1) - delimiter.length() + 1;
         }
     }

AK/Forward.h

@@ -38,6 +38,7 @@ class DeprecatedStringCodePointIterator;
 class Duration;
 class Error;
 class FlyString;
+class GenericAwaiter;
 class GenericLexer;
 class IPv4Address;
 class JsonArray;

AK/GenericAwaiter.h

@@ -0,0 +1,48 @@
+/*
+ * Copyright (c) 2024, Ali Mohammad Pur <mpfard@serenityos.org>
+ *
+ * SPDX-License-Identifier: BSD-2-Clause
+ */
+
+#pragma once
+
+#include <AK/Coroutine.h>
+#include <AK/Error.h>
+#include <AK/Function.h>
+
+namespace AK {
+class GenericAwaiter {
+    AK_MAKE_NONMOVABLE(GenericAwaiter);
+    AK_MAKE_NONCOPYABLE(GenericAwaiter);
+
+public:
+    GenericAwaiter(Function<void(Function<void()>)>&& fn)
+    {
+        fn([this] { ready(); });
+    }
+
+    bool await_ready() const { return false; }
+    void await_suspend(std::coroutine_handle<> handle)
+    {
+        m_handle = handle;
+    }
+    ErrorOr<void> await_resume()
+    {
+        m_handle = {};
+        return {};
+    }
+
+private:
+    void ready()
+    {
+        if (m_handle)
+            m_handle.resume();
+    }
+
+    std::coroutine_handle<> m_handle;
+};
+}
+
+#ifdef USING_AK_GLOBALLY
+using AK::GenericAwaiter;
+#endif

Userland/Libraries/LibCore/File.h

@@ -9,10 +9,13 @@
 
 #include <AK/Badge.h>
 #include <AK/BufferedStream.h>
+#include <AK/Coroutine.h>
+#include <AK/GenericAwaiter.h>
 #include <AK/Noncopyable.h>
 #include <AK/NonnullOwnPtr.h>
 #include <AK/Stream.h>
 #include <LibCore/Forward.h>
+#include <LibCore/Notifier.h>
 #include <LibIPC/Forward.h>
 
 namespace Core {
@@ -77,6 +80,12 @@ public:
     // See also Socket::set_blocking.
     ErrorOr<void> set_blocking(bool enabled);
 
+    Coroutine<ErrorOr<void>> wait_for_state(Core::Notifier::Type state)
+    {
+        auto notifier = CO_TRY(Core::Notifier::try_create(m_fd, state));
+        co_return co_await GenericAwaiter([&](auto ready) { notifier->on_activation = move(ready); });
+    }
+
     template<OneOf<::IPC::File, ::Core::MappedFile> VIP>
     int leak_fd(Badge<VIP>)
     {

Userland/Libraries/LibCore/NetworkJob.cpp

@@ -11,7 +11,7 @@
 
 namespace Core {
 
-NetworkJob::NetworkJob(Stream& output_stream)
+NetworkJob::NetworkJob(Core::File& output_stream)
     : m_output_stream(output_stream)
 {
 }

Userland/Libraries/LibCore/NetworkJob.h

@@ -7,6 +7,7 @@
 
 #pragma once
 
+#include <AK/Coroutine.h>
 #include <AK/Function.h>
 #include <AK/Stream.h>
 #include <LibCore/EventReceiver.h>
@@ -53,16 +54,20 @@ public:
     }
 
 protected:
-    NetworkJob(Stream&);
+    NetworkJob(Core::File&);
     void did_finish(NonnullRefPtr<NetworkResponse>&&);
     void did_fail(Error);
     void did_progress(Optional<u64> total_size, u64 downloaded);
 
-    ErrorOr<size_t> do_write(ReadonlyBytes bytes) { return m_output_stream.write_some(bytes); }
+    Coroutine<ErrorOr<size_t>> do_write(ReadonlyBytes bytes)
+    {
+        CO_TRY(co_await m_output_stream.wait_for_state(Core::Notifier::Type::Write));
+        co_return m_output_stream.write_some(bytes);
+    }
 
 private:
     RefPtr<NetworkResponse> m_response;
-    Stream& m_output_stream;
+    Core::File& m_output_stream;
     Error m_error { Error::None };
 };
 

Userland/Libraries/LibGemini/Job.cpp

@@ -8,13 +8,14 @@
 #include <AK/Error.h>
 #include <AK/String.h>
 #include <AK/Utf8View.h>
+#include <LibCore/EventLoop.h>
 #include <LibGemini/GeminiResponse.h>
 #include <LibGemini/Job.h>
 #include <unistd.h>
 
 namespace Gemini {
 
-Job::Job(GeminiRequest const& request, Stream& output_stream)
+Job::Job(GeminiRequest const& request, Core::File& output_stream)
     : Core::NetworkJob(output_stream)
     , m_request(request)
 {
@@ -83,7 +84,7 @@ void Job::flush_received_buffers()
 {
     for (size_t i = 0; i < m_received_buffers.size(); ++i) {
         auto& payload = m_received_buffers[i];
-        auto result = do_write(payload);
+        auto result = Core::run_async_in_new_event_loop([&] { return do_write(payload); });
         if (result.is_error()) {
             if (!result.error().is_errno()) {
                 dbgln("Job: Failed to flush received buffers: {}", result.error());

Userland/Libraries/LibGemini/Job.h

@@ -18,7 +18,7 @@ class Job : public Core::NetworkJob {
     C_OBJECT(Job);
 
 public:
-    explicit Job(GeminiRequest const&, Stream&);
+    explicit Job(GeminiRequest const&, Core::File&);
     virtual ~Job() override = default;
 
     virtual void start(Core::BufferedSocketBase&) override;

Userland/Libraries/LibHTTP/HttpsJob.h

@@ -30,7 +30,7 @@ public:
     Function<Vector<TLS::Certificate>()> on_certificate_requested;
 
 private:
-    explicit HttpsJob(HttpRequest&& request, Stream& output_stream)
+    explicit HttpsJob(HttpRequest&& request, Core::File& output_stream)
         : Job(move(request), output_stream)
     {
     }

Userland/Libraries/LibHTTP/Job.cpp

@@ -5,14 +5,20 @@
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
+#include <AK/AsyncStreamBuffer.h>
+#include <AK/AsyncStreamHelpers.h>
+#include <AK/AsyncStreamTransform.h>
 #include <AK/CharacterTypes.h>
 #include <AK/Debug.h>
+#include <AK/GenericAwaiter.h>
+#include <AK/MemMem.h>
 #include <AK/MemoryStream.h>
 #include <AK/Try.h>
 #include <LibCompress/Brotli.h>
 #include <LibCompress/Gzip.h>
 #include <LibCompress/Zlib.h>
 #include <LibCore/Event.h>
+#include <LibCore/EventLoop.h>
 #include <LibHTTP/HttpResponse.h>
 #include <LibHTTP/Job.h>
 #include <stdio.h>
@@ -20,6 +26,62 @@
 
 namespace HTTP {
 
+template<typename Stream>
+struct SyncStreamAsyncWrapper final : public AsyncInputStream {
+    explicit SyncStreamAsyncWrapper(MaybeOwned<Stream>&& stream)
+        : m_stream(move(stream))
+        , m_awaiter(GenericAwaiter { [&](auto cb) { m_stream->on_ready_to_read = move(cb); } })
+    {
+    }
+
+    virtual void reset() override
+    {
+    }
+
+    virtual Coroutine<ErrorOr<void>> close() override
+    {
+        if (m_stream->is_open())
+            m_stream->close();
+        co_return {};
+    }
+
+    virtual bool is_open() const override
+    {
+        return !m_stream->is_eof() || !m_buffer.is_empty();
+    }
+
+    virtual Coroutine<ErrorOr<bool>> enqueue_some(Badge<AsyncInputStream>) override
+    {
+        CO_TRY(co_await m_awaiter);
+        auto buffer = AK::Detail::ByteBuffer<16 * KiB> {};
+        auto bytes = CO_TRY(m_stream->read_some(buffer.must_get_bytes_for_writing(16 * KiB)));
+        if (bytes.is_empty())
+            co_return !m_stream->is_eof();
+        m_buffer.append(bytes);
+        co_return true;
+    }
+
+    virtual ReadonlyBytes buffered_data_unchecked(Badge<AsyncInputStream>) const override
+    {
+        return m_buffer.data();
+    }
+
+    virtual void dequeue(Badge<AsyncInputStream>, size_t bytes) override
+    {
+        m_buffer.dequeue(bytes);
+    }
+
+    Coroutine<ErrorOr<StringView>> read_line(size_t max_size)
+    {
+        co_return StringView { CO_TRY(co_await AsyncStreamHelpers::consume_until(*this, "\r\n"sv, max_size)) }.trim("\r\n"sv, TrimMode::Right);
+    }
+
+private:
+    MaybeOwned<Stream> m_stream;
+    GenericAwaiter m_awaiter;
+    AsyncStreamBuffer m_buffer;
+};
+
 static ErrorOr<ByteBuffer> handle_content_encoding(ByteBuffer const& buf, ByteString const& content_encoding)
 {
     dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf has content_encoding={}", content_encoding);
@@ -89,7 +151,7 @@ static ErrorOr<ByteBuffer> handle_content_encoding(ByteBuffer const& buf, ByteSt
     return buf;
 }
 
-Job::Job(HttpRequest&& request, Stream& output_stream)
+Job::Job(HttpRequest&& request, Core::File& output_stream)
     : Core::NetworkJob(output_stream)
     , m_request(move(request))
 {
@@ -119,81 +181,335 @@ void Job::shutdown(ShutdownMode mode)
     }
 }
 
-void Job::flush_received_buffers()
+Coroutine<void> Job::flush_received_buffers()
 {
-    if (!m_can_stream_response || m_buffered_size == 0)
-        return;
-    dbgln_if(JOB_DEBUG, "Job: Flushing received buffers: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
-    for (size_t i = 0; i < m_received_buffers.size(); ++i) {
-        auto& payload = m_received_buffers[i]->pending_flush;
-        auto result = do_write(payload);
-        if (result.is_error()) {
-            if (!result.error().is_errno()) {
-                dbgln_if(JOB_DEBUG, "Job: Failed to flush received buffers: {}", result.error());
-                continue;
-            }
-            if (result.error().code() == EINTR) {
-                i--;
-                continue;
+    if (!m_can_stream_response || m_has_scheduled_flush)
+        co_return;
+
+    TemporaryChange change(m_has_scheduled_flush, true);
+
+    while (m_buffered_size > 0) {
+        dbgln_if(JOB_DEBUG, "Job: Flushing received buffers: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
+        for (size_t i = 0; i < m_received_buffers.size(); ++i) {
+            auto& payload = m_received_buffers[i]->pending_flush;
+            auto result = co_await do_write(payload);
+            if (result.is_error())
+                break;
+            auto written = result.release_value();
+            m_buffered_size -= written;
+            if (written == payload.size()) {
+                // FIXME: Make this a take-first-friendly object?
+                (void)m_received_buffers.take_first();
+                break;
             }
+            VERIFY(written < payload.size());
+            payload = payload.slice(written, payload.size() - written);
             break;
         }
-        auto written = result.release_value();
-        m_buffered_size -= written;
-        if (written == payload.size()) {
-            // FIXME: Make this a take-first-friendly object?
-            (void)m_received_buffers.take_first();
-            --i;
-            continue;
-        }
-        VERIFY(written < payload.size());
-        payload = payload.slice(written, payload.size() - written);
-        break;
+        dbgln_if(JOB_DEBUG, "Job: Flushing received buffers done: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
+    }
+
+    if (m_request_done) {
+        m_has_scheduled_finish = true;
+        auto response = HttpResponse::create(m_code, move(m_headers), m_received_size);
+        // If the server responded with "Connection: close", close the connection
+        // as the server may or may not want to close the socket. Also, if this is
+        // a legacy HTTP server (1.0 or older), assume close is the default value.
+        if (auto result = response->headers().get("Connection"sv); result.has_value() ? result->equals_ignoring_ascii_case("close"sv) : m_legacy_connection)
+            shutdown(ShutdownMode::CloseSocket);
+        did_finish(response);
     }
-    dbgln_if(JOB_DEBUG, "Job: Flushing received buffers done: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
 }
 
-void Job::register_on_ready_to_read(Function<void()> callback)
+auto Job::parse_status(auto& stream) -> Coroutine<ErrorOr<void>>
 {
-    m_socket->on_ready_to_read = [this, callback = move(callback)] {
-        callback();
+    auto line = CO_TRY(co_await stream.read_line(PAGE_SIZE));
 
-        // As `m_socket` is a buffered object, we might not get notifications for data in the buffer
-        // so exhaust the buffer to ensure we don't end up waiting forever.
-        auto can_read_without_blocking = m_socket->can_read_without_blocking();
-        if (can_read_without_blocking.is_error())
-            return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-        if (can_read_without_blocking.value() && m_state != State::Finished && !has_error()) {
-            deferred_invoke([this] {
-                if (m_socket && m_socket->on_ready_to_read)
-                    m_socket->on_ready_to_read();
-            });
+    dbgln_if(JOB_DEBUG, "Job {} read line of length {}", m_request.url(), line.length());
+    auto parts = line.split_view(' ');
+    if (parts.size() < 2) {
+        dbgln("Job: Expected 2-part or 3-part HTTP status line, got '{}'", line);
+        deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+        co_return {};
+    }
+
+    if (!parts[0].matches("HTTP/?.?"sv, CaseSensitivity::CaseSensitive) || !is_ascii_digit(parts[0][5]) || !is_ascii_digit(parts[0][7])) {
+        dbgln("Job: Expected HTTP-Version to be of the form 'HTTP/X.Y', got '{}'", parts[0]);
+        deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+        co_return {};
+    }
+    auto http_major_version = parse_ascii_digit(parts[0][5]);
+    auto http_minor_version = parse_ascii_digit(parts[0][7]);
+    m_legacy_connection = http_major_version < 1 || (http_major_version == 1 && http_minor_version == 0);
+
+    auto code = parts[1].template to_number<unsigned>();
+    if (!code.has_value()) {
+        dbgln("Job: Expected numeric HTTP status");
+        deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+        co_return {};
+    }
+    m_code = code.value();
+    co_return {};
+}
+
+auto Job::parse_headers(auto& stream, bool in_trailers) -> Coroutine<ErrorOr<void>>
+{
+    while (true) {
+        // There's no max limit defined on headers, but for our sanity, let's limit it to 32K.
+        auto line = StringView { CO_TRY(co_await stream.read_line(32 * KiB)) };
+
+        if (line.is_empty()) {
+            if (in_trailers) {
+                co_await finish_up();
+                co_return {};
+            }
+            if (on_headers_received)
+                on_headers_received(m_headers, m_code > 0 ? m_code : Optional<u32> {});
+
+            // We've reached the end of the headers, there's a possibility that the server
+            // responds with nothing (content-length = 0 with normal encoding); if that's the case,
+            // quit early as we won't be reading anything anyway.
+            if (auto result = m_headers.get("Content-Length"sv).value_or(""sv).to_number<unsigned>(); result.has_value()) {
+                if (result.value() == 0) {
+                    auto transfer_encoding = m_headers.get("Transfer-Encoding"sv);
+                    if (!transfer_encoding.has_value() || !transfer_encoding->view().trim_whitespace().equals_ignoring_ascii_case("chunked"sv)) {
+                        co_await finish_up();
+                        co_return {};
+                    }
+                }
+            }
+            // There's also the possibility that the server responds with 204 (No Content),
+            // and manages to set a Content-Length anyway, in such cases ignore Content-Length and quit early;
+            // As the HTTP spec explicitly prohibits presence of Content-Length when the response code is 204.
+            if (m_code == 204) {
+                co_await finish_up();
+                co_return {};
+            }
+
+            break;
         }
-    };
+        auto parts = line.split_view(':');
+        if (parts.is_empty()) {
+            if (in_trailers) {
+                // Some servers like to send two ending chunks
+                // use this fact as an excuse to ignore anything after the last chunk
+                // that is not a valid trailing header.
+                co_await finish_up();
+                co_return {};
+            }
+            dbgln("Job: Expected HTTP header with key/value");
+            deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+            co_return {};
+        }
+        auto name = parts[0];
+        if (line.length() < name.length() + 2) {
+            if (in_trailers) {
+                // Some servers like to send two ending chunks
+                // use this fact as an excuse to ignore anything after the last chunk
+                // that is not a valid trailing header.
+                co_await finish_up();
+                co_return {};
+            }
+            dbgln("Job: Malformed HTTP header: '{}' ({})", line, line.length());
+            deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+            co_return {};
+        }
+        auto value = line.substring_view(name.length() + 2, line.length() - name.length() - 2);
+        m_headers.set(name, value);
+        if (name.equals_ignoring_ascii_case("Content-Encoding"sv)) {
+            // Assume that any content-encoding means that we can't decode it as a stream :(
+            dbgln_if(JOB_DEBUG, "Content-Encoding {} detected, cannot stream output :(", value);
+            m_can_stream_response = false;
+        } else if (name.equals_ignoring_ascii_case("Content-Length"sv)) {
+            auto length = value.to_number<u64>();
+            if (length.has_value())
+                m_content_length = length.value();
+        }
+        dbgln_if(JOB_DEBUG, "Job: [{}] = '{}'", name, value);
+    }
+
+    co_return {};
 }
 
-ErrorOr<ByteString> Job::read_line(size_t size)
+auto Job::parse_body(auto& stream) -> Coroutine<ErrorOr<bool>>
 {
-    auto buffer = TRY(ByteBuffer::create_uninitialized(size));
-    auto bytes_read = TRY(m_socket->read_until(buffer, "\r\n"sv));
-    return ByteString::copy(bytes_read);
+    auto should_read_trailers = false;
+    while (true) {
+        auto read_size = 64 * KiB;
+        if (m_current_chunk_remaining_size.has_value()) {
+        read_chunk_size:;
+            auto remaining = m_current_chunk_remaining_size.value();
+            if (remaining == -1) {
+                // read size
+                auto size_data = CO_TRY(co_await stream.read_line(PAGE_SIZE));
+                if (m_should_read_chunk_ending_line) {
+                    // NOTE: Some servers seem to send an extra \r\n here despite there being no size.
+                    //       This makes us tolerate that.
+                    size_data = size_data.trim("\r\n"sv, TrimMode::Right);
+                    VERIFY(size_data.is_empty());
+                    m_should_read_chunk_ending_line = false;
+                    continue;
+                }
+                auto size_lines = size_data.trim_whitespace().lines();
+                dbgln_if(JOB_DEBUG, "Job: Received a chunk with size '{}'", size_data);
+                if (size_lines.size() == 0) {
+                    if (!m_socket->is_eof())
+                        break;
+                    dbgln("Job: Reached end of stream");
+                    co_await finish_up();
+                    break;
+                }
+                auto chunk = size_lines[0].split_view(';', SplitBehavior::KeepEmpty);
+                ByteString size_string = chunk[0];
+                char* endptr;
+                auto size = strtoul(size_string.characters(), &endptr, 16);
+                if (*endptr) {
+                    // invalid number
+                    deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
+                    break;
+                }
+                if (size == 0) {
+                    // This is the last chunk
+                    // '0' *[; chunk-ext-name = chunk-ext-value]
+                    // We're going to ignore _all_ chunk extensions
+                    read_size = 0;
+                    m_current_chunk_total_size = 0;
+                    m_current_chunk_remaining_size = 0;
+
+                    dbgln_if(JOB_DEBUG, "Job: Received the last chunk with extensions '{}'", size_string.substring_view(1, size_string.length() - 1));
+                } else {
+                    m_current_chunk_total_size = size;
+                    m_current_chunk_remaining_size = size;
+                    read_size = size;
+
+                    dbgln_if(JOB_DEBUG, "Job: Chunk of size '{}' started", size);
+                }
+            } else {
+                read_size = remaining;
+
+                dbgln_if(JOB_DEBUG, "Job: Resuming chunk with '{}' bytes left over", remaining);
+            }
+        } else {
+            auto transfer_encoding = m_headers.get("Transfer-Encoding");
+            if (transfer_encoding.has_value()) {
+                // HTTP/1.1 3.3.3.3:
+                // If a message is received with both a Transfer-Encoding and a Content-Length header field, the Transfer-Encoding overrides the Content-Length. [...]
+                // https://httpwg.org/specs/rfc7230.html#message.body.length
+                m_content_length = {};
+
+                // Note: Some servers add extra spaces around 'chunked', see #6302.
+                auto encoding = transfer_encoding.value().trim_whitespace();
+
+                dbgln_if(JOB_DEBUG, "Job: This content has transfer encoding '{}'", encoding);
+                if (encoding.equals_ignoring_ascii_case("chunked"sv)) {
+                    m_current_chunk_remaining_size = -1;
+                    goto read_chunk_size;
+                } else {
+                    dbgln("Job: Unknown transfer encoding '{}', the result will likely be wrong!", encoding);
+                }
+            } else if (m_content_length.has_value()) {
+                read_size = m_content_length.value() - m_received_size;
+                if (read_size == 0) {
+                    co_await finish_up();
+                    break;
+                }
+            }
+        }
+
+        auto payload = CO_TRY(ByteBuffer::copy(CO_TRY(co_await stream.read(read_size))));
+
+        if (payload.is_empty() && !stream.is_open()) {
+            co_await finish_up();
+            break;
+        }
+
+        bool read_everything = false;
+        if (m_content_length.has_value()) {
+            auto length = m_content_length.value();
+            if (m_received_size + payload.size() >= length) {
+                payload.resize(length - m_received_size);
+                read_everything = true;
+            }
+        }
+
+        m_received_buffers.append(make<ReceivedBuffer>(payload));
+        m_buffered_size += payload.size();
+        m_received_size += payload.size();
+        Core::EventLoop::current().adopt_coroutine(flush_received_buffers());
+
+        deferred_invoke([this] { did_progress(m_content_length, m_received_size); });
+
+        if (read_everything) {
+            VERIFY(m_received_size <= m_content_length.value());
+            co_await finish_up();
+            break;
+        }
+
+        // Check after reading all the buffered data if we have reached the end of stream
+        // for cases where the server didn't send a content length, chunked encoding but is
+        // directly closing the connection.
+        if (!m_content_length.has_value() && !m_current_chunk_remaining_size.has_value() && m_socket->is_eof()) {
+            co_await finish_up();
+            break;
+        }
+
+        if (m_current_chunk_remaining_size.has_value()) {
+            auto size = m_current_chunk_remaining_size.value() - payload.size();
+
+            dbgln_if(JOB_DEBUG, "Job: We have {} bytes left over in this chunk", size);
+            if (size == 0) {
+                dbgln_if(JOB_DEBUG, "Job: Finished a chunk of {} bytes", m_current_chunk_total_size.value());
+
+                if (m_current_chunk_total_size.value() == 0) {
+                    should_read_trailers = true;
+                    break;
+                }
+
+                // we've read everything, now let's get the next chunk
+                size = -1;
+
+                auto line = CO_TRY(co_await stream.read_line(PAGE_SIZE));
+                VERIFY(line.is_empty());
+            }
+            m_current_chunk_remaining_size = size;
+        }
+    }
+    co_return should_read_trailers;
 }
 
-ErrorOr<ByteBuffer> Job::receive(size_t size)
+auto Job::read_response() -> Coroutine<ErrorOr<void>>
 {
-    if (size == 0)
-        return ByteBuffer {};
+    auto stream = SyncStreamAsyncWrapper(MaybeOwned { *m_socket });
 
-    auto buffer = TRY(ByteBuffer::create_uninitialized(size));
-    size_t nread;
-    do {
-        auto result = m_socket->read_some(buffer);
-        if (result.is_error() && result.error().is_errno() && result.error().code() == EINTR)
-            continue;
-        nread = TRY(result).size();
-        break;
-    } while (true);
-    return buffer.slice(0, nread);
+    if (is_cancelled())
+        co_return {};
+
+    if (m_socket->is_eof()) {
+        // Some servers really like terminating connections by simply closing them (even TLS ones)
+        // to signal end-of-data, if there's no:
+        // - connection
+        // - content-size
+        // - transfer-encoding: chunked
+        // header, simply treat EOF as a termination signal.
+        if (m_headers.contains("connection"sv) || m_content_length.has_value() || m_current_chunk_total_size.has_value()) {
+            dbgln_if(JOB_DEBUG, "Read failure: Actually EOF!");
+            deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
+            co_return {};
+        }
+
+        co_await finish_up();
+        co_return {};
+    }
+
+    CO_TRY(co_await parse_status(stream));
+    CO_TRY(co_await parse_headers(stream, false));
+    auto should_read_trailers = CO_TRY(co_await parse_body(stream));
+    if (should_read_trailers)
+        CO_TRY(co_await parse_headers(stream, true));
+
+    co_await finish_up();
+
+    co_return {};
 }
 
 void Job::on_socket_connected()
@@ -205,381 +521,27 @@ void Job::on_socket_connected()
         dbgln("{}", ByteString::copy(raw_request));
     }
 
-    bool success = !m_socket->write_until_depleted(raw_request).is_error();
-    if (!success)
-        deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
+    if (m_socket->write_until_depleted(raw_request).is_error())
+        return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
 
-    register_on_ready_to_read([&] {
-        dbgln_if(JOB_DEBUG, "Ready to read for {}, state = {}, cancelled = {}", m_request.url(), to_underlying(m_state), is_cancelled());
-        if (is_cancelled())
-            return;
-
-        if (m_state == State::Finished) {
-            // We have everything we want, at this point, we can either get an EOF, or a bunch of extra newlines
-            // (unless "Connection: close" isn't specified)
-            // So just ignore everything after this.
-            return;
+    Core::EventLoop::current().adopt_coroutine([this] mutable -> Coroutine<void> {
+        auto result = co_await read_response();
+        if (result.is_error()) {
+            dbgln("Job: Failed to read response: {}", result.error());
+            did_fail(Core::NetworkJob::Error::TransmissionFailed);
         }
-
-        if (m_socket->is_eof()) {
-            // Some servers really like terminating connections by simply closing them (even TLS ones)
-            // to signal end-of-data, if there's no:
-            // - connection
-            // - content-size
-            // - transfer-encoding: chunked
-            // header, simply treat EOF as a termination signal.
-            if (m_headers.contains("connection"sv) || m_content_length.has_value() || m_current_chunk_total_size.has_value()) {
-                dbgln_if(JOB_DEBUG, "Read failure: Actually EOF!");
-                deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-                return;
-            }
-
-            finish_up();
-            return;
-        }
-
-        while (m_state == State::InStatus) {
-            auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes());
-            if (can_read_line.is_error()) {
-                dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-
-            if (!can_read_line.value()) {
-                dbgln_if(JOB_DEBUG, "Job {} cannot read a full line", m_request.url());
-                // TODO: Should we retry here instead of failing instantly?
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-
-            auto maybe_line = read_line(PAGE_SIZE);
-            if (maybe_line.is_error()) {
-                dbgln_if(JOB_DEBUG, "Job {} could not read line: {}", m_request.url(), maybe_line.error());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-
-            auto line = maybe_line.release_value();
-
-            dbgln_if(JOB_DEBUG, "Job {} read line of length {}", m_request.url(), line.length());
-            auto parts = line.split_view(' ');
-            if (parts.size() < 2) {
-                dbgln("Job: Expected 2-part or 3-part HTTP status line, got '{}'", line);
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-            }
-
-            if (!parts[0].matches("HTTP/?.?"sv, CaseSensitivity::CaseSensitive) || !is_ascii_digit(parts[0][5]) || !is_ascii_digit(parts[0][7])) {
-                dbgln("Job: Expected HTTP-Version to be of the form 'HTTP/X.Y', got '{}'", parts[0]);
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-            }
-            auto http_major_version = parse_ascii_digit(parts[0][5]);
-            auto http_minor_version = parse_ascii_digit(parts[0][7]);
-            m_legacy_connection = http_major_version < 1 || (http_major_version == 1 && http_minor_version == 0);
-
-            auto code = parts[1].to_number<unsigned>();
-            if (!code.has_value()) {
-                dbgln("Job: Expected numeric HTTP status");
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-            }
-            m_code = code.value();
-            m_state = State::InHeaders;
-
-            auto can_read_without_blocking = m_socket->can_read_without_blocking();
-            if (can_read_without_blocking.is_error())
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-
-            if (!can_read_without_blocking.value())
-                return;
-        }
-        while (m_state == State::InHeaders || m_state == State::Trailers) {
-            auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes());
-            if (can_read_line.is_error()) {
-                dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-
-            if (!can_read_line.value()) {
-                dbgln_if(JOB_DEBUG, "Can't read lines anymore :(");
-                return;
-            }
-
-            // There's no max limit defined on headers, but for our sanity, let's limit it to 32K.
-            auto maybe_line = read_line(32 * KiB);
-            if (maybe_line.is_error()) {
-                dbgln_if(JOB_DEBUG, "Job {} could not read a header line: {}", m_request.url(), maybe_line.error());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-            auto line = maybe_line.release_value();
-
-            if (line.is_empty()) {
-                if (m_state == State::Trailers) {
-                    return finish_up();
-                }
-                if (on_headers_received) {
-                    on_headers_received(m_headers, m_code > 0 ? m_code : Optional<u32> {});
-                }
-                m_state = State::InBody;
-
-                // We've reached the end of the headers, there's a possibility that the server
-                // responds with nothing (content-length = 0 with normal encoding); if that's the case,
-                // quit early as we won't be reading anything anyway.
-                if (auto result = m_headers.get("Content-Length"sv).value_or(""sv).to_number<unsigned>(); result.has_value()) {
-                    if (result.value() == 0) {
-                        auto transfer_encoding = m_headers.get("Transfer-Encoding"sv);
-                        if (!transfer_encoding.has_value() || !transfer_encoding->view().trim_whitespace().equals_ignoring_ascii_case("chunked"sv))
-                            return finish_up();
-                    }
-                }
-                // There's also the possibility that the server responds with 204 (No Content),
-                // and manages to set a Content-Length anyway, in such cases ignore Content-Length and quit early;
-                // As the HTTP spec explicitly prohibits presence of Content-Length when the response code is 204.
-                if (m_code == 204)
-                    return finish_up();
-
-                break;
-            }
-            auto parts = line.split_view(':');
-            if (parts.is_empty()) {
-                if (m_state == State::Trailers) {
-                    // Some servers like to send two ending chunks
-                    // use this fact as an excuse to ignore anything after the last chunk
-                    // that is not a valid trailing header.
-                    return finish_up();
-                }
-                dbgln("Job: Expected HTTP header with key/value");
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-            }
-            auto name = parts[0];
-            if (line.length() < name.length() + 2) {
-                if (m_state == State::Trailers) {
-                    // Some servers like to send two ending chunks
-                    // use this fact as an excuse to ignore anything after the last chunk
-                    // that is not a valid trailing header.
-                    return finish_up();
-                }
-                dbgln("Job: Malformed HTTP header: '{}' ({})", line, line.length());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
-            }
-            auto value = line.substring(name.length() + 2, line.length() - name.length() - 2);
-            m_headers.set(name, value);
-            if (name.equals_ignoring_ascii_case("Content-Encoding"sv)) {
-                // Assume that any content-encoding means that we can't decode it as a stream :(
-                dbgln_if(JOB_DEBUG, "Content-Encoding {} detected, cannot stream output :(", value);
-                m_can_stream_response = false;
-            } else if (name.equals_ignoring_ascii_case("Content-Length"sv)) {
-                auto length = value.to_number<u64>();
-                if (length.has_value())
-                    m_content_length = length.value();
-            }
-            dbgln_if(JOB_DEBUG, "Job: [{}] = '{}'", name, value);
-
-            auto can_read_without_blocking = m_socket->can_read_without_blocking();
-            if (can_read_without_blocking.is_error())
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            if (!can_read_without_blocking.value()) {
-                dbgln_if(JOB_DEBUG, "Can't read headers anymore, byebye :(");
-                return;
-            }
-        }
-        VERIFY(m_state == State::InBody);
-
-        while (true) {
-            auto can_read_without_blocking = m_socket->can_read_without_blocking();
-            if (can_read_without_blocking.is_error())
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            if (!can_read_without_blocking.value())
-                break;
-
-            auto read_size = 64 * KiB;
-            if (m_current_chunk_remaining_size.has_value()) {
-            read_chunk_size:;
-                auto remaining = m_current_chunk_remaining_size.value();
-                if (remaining == -1) {
-                    // read size
-                    auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes());
-                    if (can_read_line.is_error())
-                        return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-                    if (!can_read_line.value()) // We'll try later.
-                        return;
-                    auto maybe_size_data = read_line(PAGE_SIZE);
-                    if (maybe_size_data.is_error()) {
-                        dbgln_if(JOB_DEBUG, "Job: Could not receive chunk: {}", maybe_size_data.error());
-                    }
-                    auto size_data = maybe_size_data.release_value();
-
-                    if (m_should_read_chunk_ending_line) {
-                        // NOTE: Some servers seem to send an extra \r\n here despite there being no size.
-                        //       This makes us tolerate that.
-                        size_data = size_data.trim("\r\n"sv, TrimMode::Right);
-                        VERIFY(size_data.is_empty());
-                        m_should_read_chunk_ending_line = false;
-                        continue;
-                    }
-                    auto size_lines = size_data.view().trim_whitespace().lines();
-                    dbgln_if(JOB_DEBUG, "Job: Received a chunk with size '{}'", size_data);
-                    if (size_lines.size() == 0) {
-                        if (!m_socket->is_eof())
-                            break;
-                        dbgln("Job: Reached end of stream");
-                        finish_up();
-                        break;
-                    } else {
-                        auto chunk = size_lines[0].split_view(';', SplitBehavior::KeepEmpty);
-                        ByteString size_string = chunk[0];
-                        char* endptr;
-                        auto size = strtoul(size_string.characters(), &endptr, 16);
-                        if (*endptr) {
-                            // invalid number
-                            deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-                            break;
-                        }
-                        if (size == 0) {
-                            // This is the last chunk
-                            // '0' *[; chunk-ext-name = chunk-ext-value]
-                            // We're going to ignore _all_ chunk extensions
-                            read_size = 0;
-                            m_current_chunk_total_size = 0;
-                            m_current_chunk_remaining_size = 0;
-
-                            dbgln_if(JOB_DEBUG, "Job: Received the last chunk with extensions '{}'", size_string.substring_view(1, size_string.length() - 1));
-                        } else {
-                            m_current_chunk_total_size = size;
-                            m_current_chunk_remaining_size = size;
-                            read_size = size;
-
-                            dbgln_if(JOB_DEBUG, "Job: Chunk of size '{}' started", size);
-                        }
-                    }
-                } else {
-                    read_size = remaining;
-
-                    dbgln_if(JOB_DEBUG, "Job: Resuming chunk with '{}' bytes left over", remaining);
-                }
-            } else {
-                auto transfer_encoding = m_headers.get("Transfer-Encoding");
-                if (transfer_encoding.has_value()) {
-                    // HTTP/1.1 3.3.3.3:
-                    // If a message is received with both a Transfer-Encoding and a Content-Length header field, the Transfer-Encoding overrides the Content-Length. [...]
-                    // https://httpwg.org/specs/rfc7230.html#message.body.length
-                    m_content_length = {};
-
-                    // Note: Some servers add extra spaces around 'chunked', see #6302.
-                    auto encoding = transfer_encoding.value().trim_whitespace();
-
-                    dbgln_if(JOB_DEBUG, "Job: This content has transfer encoding '{}'", encoding);
-                    if (encoding.equals_ignoring_ascii_case("chunked"sv)) {
-                        m_current_chunk_remaining_size = -1;
-                        goto read_chunk_size;
-                    } else {
-                        dbgln("Job: Unknown transfer encoding '{}', the result will likely be wrong!", encoding);
-                    }
-                }
-            }
-
-            can_read_without_blocking = m_socket->can_read_without_blocking();
-            if (can_read_without_blocking.is_error())
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            if (!can_read_without_blocking.value())
-                break;
-
-            dbgln_if(JOB_DEBUG, "Waiting for payload for {}", m_request.url());
-            auto maybe_payload = receive(read_size);
-            if (maybe_payload.is_error()) {
-                dbgln_if(JOB_DEBUG, "Could not read the payload: {}", maybe_payload.error());
-                return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-            }
-
-            auto payload = maybe_payload.release_value();
-
-            if (payload.is_empty() && m_socket->is_eof()) {
-                finish_up();
-                break;
-            }
-
-            bool read_everything = false;
-            if (m_content_length.has_value()) {
-                auto length = m_content_length.value();
-                if (m_received_size + payload.size() >= length) {
-                    payload.resize(length - m_received_size);
-                    read_everything = true;
-                }
-            }
-
-            m_received_buffers.append(make<ReceivedBuffer>(payload));
-            m_buffered_size += payload.size();
-            m_received_size += payload.size();
-            flush_received_buffers();
-
-            deferred_invoke([this] { did_progress(m_content_length, m_received_size); });
-
-            if (read_everything) {
-                VERIFY(m_received_size <= m_content_length.value());
-                finish_up();
-                break;
-            }
-
-            // Check after reading all the buffered data if we have reached the end of stream
-            // for cases where the server didn't send a content length, chunked encoding but is
-            // directly closing the connection.
-            if (!m_content_length.has_value() && !m_current_chunk_remaining_size.has_value() && m_socket->is_eof()) {
-                finish_up();
-                break;
-            }
-
-            if (m_current_chunk_remaining_size.has_value()) {
-                auto size = m_current_chunk_remaining_size.value() - payload.size();
-
-                dbgln_if(JOB_DEBUG, "Job: We have {} bytes left over in this chunk", size);
-                if (size == 0) {
-                    dbgln_if(JOB_DEBUG, "Job: Finished a chunk of {} bytes", m_current_chunk_total_size.value());
-
-                    if (m_current_chunk_total_size.value() == 0) {
-                        m_state = State::Trailers;
-                        break;
-                    }
-
-                    // we've read everything, now let's get the next chunk
-                    size = -1;
-
-                    auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes());
-                    if (can_read_line.is_error())
-                        return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-                    if (can_read_line.value()) {
-                        auto maybe_line = read_line(PAGE_SIZE);
-                        if (maybe_line.is_error()) {
-                            return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
-                        }
-
-                        VERIFY(maybe_line.value().is_empty());
-                    } else {
-                        m_should_read_chunk_ending_line = true;
-                    }
-                }
-                m_current_chunk_remaining_size = size;
-            }
-        }
-
-        if (!m_socket->is_open()) {
-            dbgln_if(JOB_DEBUG, "Connection appears to have closed, finishing up");
-            finish_up();
-        }
-    });
+    }());
 }
 
-void Job::timer_event(Core::TimerEvent& event)
+Coroutine<void> Job::finish_up()
 {
-    event.accept();
-    finish_up();
-    if (m_buffered_size == 0)
-        stop_timer();
-}
+    if (m_has_scheduled_finish)
+        co_return;
 
-void Job::finish_up()
-{
-    VERIFY(!m_has_scheduled_finish);
-    m_state = State::Finished;
     if (!m_can_stream_response) {
         auto maybe_flattened_buffer = ByteBuffer::create_uninitialized(m_buffered_size);
         if (maybe_flattened_buffer.is_error())
-            return did_fail(Core::NetworkJob::Error::TransmissionFailed);
+            co_return did_fail(Core::NetworkJob::Error::TransmissionFailed);
         auto flattened_buffer = maybe_flattened_buffer.release_value();
 
         u8* flat_ptr = flattened_buffer.data();
@@ -596,7 +558,7 @@ void Job::finish_up()
             if (auto result = handle_content_encoding(flattened_buffer, content_encoding.value()); !result.is_error())
                 flattened_buffer = result.release_value();
             else
-                return did_fail(Core::NetworkJob::Error::TransmissionFailed);
+                co_return did_fail(Core::NetworkJob::Error::TransmissionFailed);
         }
 
         m_buffered_size = flattened_buffer.size();
@@ -604,28 +566,7 @@ void Job::finish_up()
         m_can_stream_response = true;
     }
 
-    flush_received_buffers();
-    if (m_buffered_size != 0) {
-        // We have to wait for the client to consume all the downloaded data
-        // before we can actually call `did_finish`. in a normal flow, this should
-        // never be hit since the client is reading as we are writing, unless there
-        // are too many concurrent downloads going on.
-        dbgln_if(JOB_DEBUG, "Flush finished with {} bytes remaining, will try again later", m_buffered_size);
-        if (!has_timer())
-            start_timer(50);
-        return;
-    }
-
-    stop_timer();
-    m_has_scheduled_finish = true;
-    auto response = HttpResponse::create(m_code, move(m_headers), m_received_size);
-    deferred_invoke([this, response = move(response)] {
-        // If the server responded with "Connection: close", close the connection
-        // as the server may or may not want to close the socket. Also, if this is
-        // a legacy HTTP server (1.0 or older), assume close is the default value.
-        if (auto result = response->headers().get("Connection"sv); result.has_value() ? result->equals_ignoring_ascii_case("close"sv) : m_legacy_connection)
-            shutdown(ShutdownMode::CloseSocket);
-        did_finish(response);
-    });
+    m_request_done = true;
+    Core::EventLoop::current().adopt_coroutine(flush_received_buffers());
 }
 }

Userland/Libraries/LibHTTP/Job.h

@@ -19,7 +19,7 @@ class Job : public Core::NetworkJob {
     C_OBJECT(Job);
 
 public:
-    explicit Job(HttpRequest&&, Stream&);
+    explicit Job(HttpRequest&&, Core::File&);
     virtual ~Job() override = default;
 
     virtual void start(Core::BufferedSocketBase&) override;
@@ -31,25 +31,18 @@ public:
     HttpResponse* response() { return static_cast<HttpResponse*>(Core::NetworkJob::response()); }
     HttpResponse const* response() const { return static_cast<HttpResponse const*>(Core::NetworkJob::response()); }
 
-protected:
-    void finish_up();
-    void on_socket_connected();
-    void flush_received_buffers();
-    void register_on_ready_to_read(Function<void()>);
-    ErrorOr<ByteString> read_line(size_t);
-    ErrorOr<ByteBuffer> receive(size_t);
-    void timer_event(Core::TimerEvent&) override;
+private:
+    auto parse_status(auto& stream) -> Coroutine<ErrorOr<void>>;
+    auto parse_headers(auto& stream, bool in_trailers) -> Coroutine<ErrorOr<void>>;
+    auto parse_body(auto& stream) -> Coroutine<ErrorOr<bool>>;
+    auto read_response() -> Coroutine<ErrorOr<void>>;
 
-    enum class State {
-        InStatus,
-        InHeaders,
-        InBody,
-        Trailers,
-        Finished,
-    };
+protected:
+    Coroutine<void> finish_up();
+    void on_socket_connected();
+    Coroutine<void> flush_received_buffers();
 
     HttpRequest m_request;
-    State m_state { State::InStatus };
     Core::BufferedSocketBase* m_socket { nullptr };
     bool m_legacy_connection { false };
     int m_code { -1 };
@@ -79,6 +72,8 @@ protected:
     bool m_can_stream_response { true };
     bool m_should_read_chunk_ending_line { false };
     bool m_has_scheduled_finish { false };
+    bool m_has_scheduled_flush { false };
+    bool m_request_done { false };
 };
 
 }

Userland/Services/RequestServer/ConnectionCache.h

@@ -20,6 +20,9 @@
 #include <LibThreading/RWLockProtected.h>
 #include <LibURL/URL.h>
 
+#undef REQUESTSERVER_DEBUG
+#define REQUESTSERVER_DEBUG 0
+
 namespace RequestServer {
 
 enum class CacheLevel {