LibWeb: Use unbuffered network requests for all Fetch requests

Previously, unbuffered requests were only available as a special mode
for EventSource. With this change, they are enabled by default, which
means chunks can be read from the stream as soon as they arrive.

This unlocks some interesting possibilities, such as starting to parse
HTML documents before the entire response has been received (that, in
turn, allows us to initiate subresource fetches earlier or begin
executing scripts sooner), or start rendering videos before they are
fully downloaded.

Co-authored-by: Timothy Flynn <trflynn89@pm.me>

Author: kalenikaliaksandr

Libraries/LibWeb/Fetch/Fetching/FetchedDataReceiver.cpp

@@ -1,5 +1,6 @@
 /*
  * Copyright (c) 2024, Tim Flynn <trflynn89@serenityos.org>
+ * Copyright (c) 2025, Aliaksandr Kalenik <kalenik.aliaksandr@gmail.com>
  *
  * SPDX-License-Identifier: BSD-2-Clause
  */
@@ -36,35 +37,82 @@ void FetchedDataReceiver::visit_edges(Visitor& visitor)
 
 void FetchedDataReceiver::set_pending_promise(GC::Ref<WebIDL::Promise> promise)
 {
-    auto had_pending_promise = m_pending_promise != nullptr;
+    VERIFY(!m_pending_promise);
+    VERIFY(!m_has_unfulfilled_promise);
     m_pending_promise = promise;
 
-    if (!had_pending_promise && !m_buffer.is_empty()) {
-        on_data_received(m_buffer);
-        m_buffer.clear();
+    if (!m_buffer.is_empty()) {
+        pull_bytes_into_stream(move(m_buffer));
+    } else if (m_lifecycle_state == LifecycleState::ReadyToClose) {
+        close_stream();
     }
 }
 
 // This implements the parallel steps of the pullAlgorithm in HTTP-network-fetch.
-// https://fetch.spec.whatwg.org/#ref-for-in-parallel④
-void FetchedDataReceiver::on_data_received(ReadonlyBytes bytes)
+// https://fetch.spec.whatwg.org/#ref-for-in-parallel⑤
+void FetchedDataReceiver::handle_network_bytes(ReadonlyBytes bytes, NetworkState state)
 {
-    // FIXME: 1. If the size of buffer is smaller than a lower limit chosen by the user agent and the ongoing fetch
-    //           is suspended, resume the fetch.
-    // FIXME: 2. Wait until buffer is not empty.
+    VERIFY(m_lifecycle_state == LifecycleState::Receiving);
+
+    if (state == NetworkState::Complete) {
+        VERIFY(bytes.is_empty());
+        m_lifecycle_state = LifecycleState::CompletePending;
+    }
 
-    // If the remote end sends data immediately after we receive headers, we will often get that data here before the
-    // stream tasks have all been queued internally. Just hold onto that data.
     if (!m_pending_promise) {
-        m_buffer.append(bytes);
+        if (state == NetworkState::Ongoing)
+            m_buffer.append(bytes);
+        if (m_lifecycle_state == LifecycleState::CompletePending && m_buffer.is_empty() && !m_has_unfulfilled_promise)
+            m_lifecycle_state = LifecycleState::ReadyToClose;
         return;
     }
 
+    // 1. If one or more bytes have been transmitted from response’s message body, then:
+    if (!bytes.is_empty()) {
+        // 1. Let bytes be the transmitted bytes.
+
+        // FIXME: 2. Let codings be the result of extracting header list values given `Content-Encoding` and response’s header list.
+        // FIXME: 3. Increase response’s body info’s encoded size by bytes’s length.
+        // FIXME: 4. Set bytes to the result of handling content codings given codings and bytes.
+        // FIXME: 5. Increase response’s body info’s decoded size by bytes’s length.
+        // FIXME: 6. If bytes is failure, then terminate fetchParams’s controller.
+
+        // 7. Append bytes to buffer.
+        pull_bytes_into_stream(MUST(ByteBuffer::copy(bytes)));
+
+        // FIXME: 8. If the size of buffer is larger than an upper limit chosen by the user agent, ask the user agent
+        //           to suspend the ongoing fetch.
+        return;
+    }
+    // 2. Otherwise, if the bytes transmission for response’s message body is done normally and stream is readable,
+    //    then close stream, and abort these in-parallel steps.
+    if (m_stream->is_readable()) {
+        VERIFY(m_lifecycle_state == LifecycleState::CompletePending);
+        close_stream();
+    }
+}
+
+// This implements the parallel steps of the pullAlgorithm in HTTP-network-fetch.
+// https://fetch.spec.whatwg.org/#ref-for-in-parallel④
+void FetchedDataReceiver::pull_bytes_into_stream(ByteBuffer&& bytes)
+{
+    // FIXME: 1. If the size of buffer is smaller than a lower limit chosen by the user agent and the ongoing fetch
+    //           is suspended, resume the fetch.
+
+    // 2. Wait until buffer is not empty.
+    VERIFY(!bytes.is_empty());
+    VERIFY(m_lifecycle_state == LifecycleState::Receiving || m_lifecycle_state == LifecycleState::CompletePending);
+
     // 3. Queue a fetch task to run the following steps, with fetchParams’s task destination.
+    VERIFY(!m_has_unfulfilled_promise);
+    m_has_unfulfilled_promise = true;
     Infrastructure::queue_fetch_task(
         m_fetch_params->controller(),
         m_fetch_params->task_destination(),
-        GC::create_function(heap(), [this, bytes = MUST(ByteBuffer::copy(bytes))]() mutable {
+        GC::create_function(heap(), [this, bytes = move(bytes), pending_promise = m_pending_promise]() mutable {
+            m_has_unfulfilled_promise = false;
+            VERIFY(m_lifecycle_state == LifecycleState::Receiving || m_lifecycle_state == LifecycleState::CompletePending);
+
             HTML::TemporaryExecutionContext execution_context { m_stream->realm(), HTML::TemporaryExecutionContext::CallbacksEnabled::Yes };
 
             // 1. Pull from bytes buffer into stream.
@@ -82,8 +130,22 @@ void FetchedDataReceiver::on_data_received(ReadonlyBytes bytes)
                 m_fetch_params->controller()->terminate();
 
             // 3. Resolve promise with undefined.
-            WebIDL::resolve_promise(m_stream->realm(), *m_pending_promise, JS::js_undefined());
+            WebIDL::resolve_promise(m_stream->realm(), *pending_promise, JS::js_undefined());
+
+            if (m_lifecycle_state == LifecycleState::CompletePending && m_buffer.is_empty())
+                m_lifecycle_state = LifecycleState::ReadyToClose;
         }));
+
+    m_pending_promise = {};
+}
+
+void FetchedDataReceiver::close_stream()
+{
+    VERIFY(m_has_unfulfilled_promise == 0);
+    WebIDL::resolve_promise(m_stream->realm(), *m_pending_promise, JS::js_undefined());
+    m_pending_promise = {};
+    m_lifecycle_state = LifecycleState::Closed;
+    m_stream->close();
 }
 
 }

Libraries/LibWeb/Fetch/Fetching/FetchedDataReceiver.h

@@ -1,5 +1,6 @@
 /*
  * Copyright (c) 2024, Tim Flynn <trflynn89@serenityos.org>
+ * Copyright (c) 2025, Aliaksandr Kalenik <kalenik.aliaksandr@gmail.com>
  *
  * SPDX-License-Identifier: BSD-2-Clause
  */
@@ -21,17 +22,35 @@ class FetchedDataReceiver final : public JS::Cell {
     virtual ~FetchedDataReceiver() override;
 
     void set_pending_promise(GC::Ref<WebIDL::Promise>);
-    void on_data_received(ReadonlyBytes);
+
+    enum class NetworkState {
+        Ongoing,
+        Complete,
+        Error,
+    };
+    void handle_network_bytes(ReadonlyBytes, NetworkState);
 
 private:
     FetchedDataReceiver(GC::Ref<Infrastructure::FetchParams const>, GC::Ref<Streams::ReadableStream>);
 
     virtual void visit_edges(Visitor& visitor) override;
 
+    void pull_bytes_into_stream(ByteBuffer&&);
+    void close_stream();
+
     GC::Ref<Infrastructure::FetchParams const> m_fetch_params;
     GC::Ref<Streams::ReadableStream> m_stream;
     GC::Ptr<WebIDL::Promise> m_pending_promise;
+
     ByteBuffer m_buffer;
+    enum class LifecycleState {
+        Receiving,
+        CompletePending,
+        ReadyToClose,
+        Closed,
+    };
+    LifecycleState m_lifecycle_state { LifecycleState::Receiving };
+    bool m_has_unfulfilled_promise { false };
 };
 
 }