LibDevTools: Handle fragmented protocol packets

Firefox can split a DevTools protocol packet across multiple TCP
reads. The previous reader only checked that some data was available,
then tried to synchronously read the whole length-prefixed packet from
the readiness callback. Once the socket was nonblocking this could make
startup flaky when Firefox opened the inspector.

Buffer incoming bytes instead, and only dispatch messages once a full
length-prefixed JSON payload has arrived. Add a protocol test that sends
a request in two fragments through the real DevTools server.

Libraries/LibDevTools/Connection.cpp

@@ -20,6 +20,8 @@ NonnullRefPtr<Connection> Connection::create(NonnullOwnPtr<Core::BufferedTCPSock
 Connection::Connection(NonnullOwnPtr<Core::BufferedTCPSocket> socket)
     : m_socket(move(socket))
 {
+    (void)m_socket->set_blocking(false);
+
     m_socket->on_ready_to_read = [this]() {
         if (auto result = on_ready_to_read(); result.is_error()) {
             if (on_connection_closed)
@@ -54,47 +56,71 @@ void Connection::send_message(JsonValue const& message)
     }
 }
 
-// https://firefox-source-docs.mozilla.org/devtools/backend/protocol.html#packets
-ErrorOr<JsonValue> Connection::read_message()
+ErrorOr<void> Connection::read_available_data()
 {
-    ByteBuffer length_buffer;
+    auto buffer = TRY(ByteBuffer::create_uninitialized(4096));
 
-    // FIXME: `read_until(':')` would be nicer here, but that seems to return immediately without receiving any data.
-    while (true) {
-        auto byte = TRY(m_socket->read_value<u8>());
-        if (byte == ':') {
+    while (TRY(m_socket->can_read_without_blocking())) {
+        auto bytes = TRY(m_socket->read_some(buffer));
+        if (bytes.is_empty()) {
+            if (m_socket->is_eof())
+                return Error::from_string_literal("DevTools client disconnected");
             break;
         }
 
-        length_buffer.append(byte);
+        TRY(m_incoming_buffer.try_append(bytes));
     }
 
-    auto length = StringView { length_buffer }.to_number<size_t>();
+    return {};
+}
+
+// https://firefox-source-docs.mozilla.org/devtools/backend/protocol.html#packets
+ErrorOr<Optional<JsonValue>> Connection::read_message()
+{
+    auto const packet = StringView { m_incoming_buffer };
+    auto colon_offset = packet.find(':');
+    if (!colon_offset.has_value())
+        return Optional<JsonValue> {};
+
+    auto length = packet.substring_view(0, *colon_offset).to_number<size_t>();
     if (!length.has_value())
         return Error::from_string_literal("Could not read message length from DevTools client");
 
-    ByteBuffer message_buffer;
-    message_buffer.resize(*length);
+    auto const message_offset = *colon_offset + 1;
+    auto const packet_size = message_offset + *length;
+    if (m_incoming_buffer.size() < packet_size)
+        return Optional<JsonValue> {};
 
-    TRY(m_socket->read_until_filled(message_buffer));
+    auto message = TRY(JsonValue::from_string(packet.substring_view(message_offset, *length)));
 
-    auto message = TRY(JsonValue::from_string(message_buffer));
     dbgln_if(DEVTOOLS_DEBUG, "\x1b[1;33m>>\x1b[0m {}", message);
 
+    if (packet_size == m_incoming_buffer.size()) {
+        m_incoming_buffer.clear();
+    } else {
+        m_incoming_buffer = TRY(m_incoming_buffer.slice(packet_size, m_incoming_buffer.size() - packet_size));
+    }
+
     return message;
 }
 
 ErrorOr<void> Connection::on_ready_to_read()
 {
+    TRY(read_available_data());
+
     // https://firefox-source-docs.mozilla.org/devtools/backend/protocol.html#the-request-reply-pattern
     // Note that it is correct for a client to send several requests to a request/reply actor without waiting for a
     // reply to each request before sending the next; requests can be pipelined.
-    while (TRY(m_socket->can_read_without_blocking())) {
+    while (true) {
         auto message = TRY(read_message());
-        if (!message.is_object())
+        if (!message.has_value())
+            break;
+
+        auto value = message.release_value();
+        if (!value.is_object())
             continue;
 
-        Core::deferred_invoke([weak_self = make_weak_ptr<Connection>(), message = move(message)]() mutable {
+        Core::deferred_invoke([weak_self = make_weak_ptr<Connection>(), message = move(value)]() mutable {
             auto self = weak_self.strong_ref();
             if (!self)
                 return;

Libraries/LibDevTools/Connection.h

@@ -6,10 +6,12 @@
 
 #pragma once
 
+#include <AK/ByteBuffer.h>
 #include <AK/Error.h>
 #include <AK/Function.h>
 #include <AK/NonnullOwnPtr.h>
 #include <AK/NonnullRefPtr.h>
+#include <AK/Optional.h>
 #include <AK/RefCounted.h>
 #include <AK/Weakable.h>
 #include <LibCore/Socket.h>
@@ -33,9 +35,11 @@ private:
     explicit Connection(NonnullOwnPtr<Core::BufferedTCPSocket>);
 
     ErrorOr<void> on_ready_to_read();
-    ErrorOr<JsonValue> read_message();
+    ErrorOr<void> read_available_data();
+    ErrorOr<Optional<JsonValue>> read_message();
 
     NonnullOwnPtr<Core::BufferedTCPSocket> m_socket;
+    ByteBuffer m_incoming_buffer;
 };
 
 }

Tests/LibDevTools/CMakeLists.txt

@@ -3,5 +3,5 @@ set(TEST_SOURCES
 )
 
 foreach(source IN LISTS TEST_SOURCES)
-    ladybird_test("${source}" LibDevTools LIBS LibDevTools LibHTTP LibRequests LibWeb LibWebView)
+    ladybird_test("${source}" LibDevTools LIBS LibDevTools LibHTTP LibRequests LibThreading LibWeb LibWebView)
 endforeach()

Tests/LibDevTools/TestDevToolsProtocol.cpp

@@ -4,6 +4,7 @@
  * SPDX-License-Identifier: BSD-2-Clause
  */
 
+#include <AK/Atomic.h>
 #include <AK/ByteBuffer.h>
 #include <AK/JsonArray.h>
 #include <AK/JsonObject.h>
@@ -21,6 +22,7 @@
 #include <LibHTTP/Header.h>
 #include <LibRequests/RequestTimingInfo.h>
 #include <LibTest/TestCase.h>
+#include <LibThreading/Thread.h>
 #include <LibWeb/CSS/StyleSheetIdentifier.h>
 #include <LibWebView/Attribute.h>
 #include <LibWebView/ConsoleOutput.h>
@@ -1287,9 +1289,25 @@ public:
     {
         auto serialized = message.serialized();
         auto packet = MUST(String::formatted("{}:{}", serialized.byte_count(), serialized));
-        MUST(m_socket->set_blocking(true));
-        auto restore_nonblocking = ScopeGuard([&] { MUST(m_socket->set_blocking(false)); });
-        MUST(m_socket->write_until_depleted(packet.bytes()));
+        send_packet_bytes(packet.bytes());
+    }
+
+    NonnullRefPtr<Threading::Thread> send_in_two_fragments(JsonObject message, size_t first_fragment_size, Atomic<bool>& may_send_second_fragment, Atomic<bool>& sent_second_fragment)
+    {
+        auto serialized = message.serialized();
+        auto packet = MUST(String::formatted("{}:{}", serialized.byte_count(), serialized));
+        VERIFY(first_fragment_size < packet.byte_count());
+
+        send_packet_bytes(packet.bytes().slice(0, first_fragment_size));
+
+        return Threading::Thread::construct("DevToolsFragmentSender"sv, [this, packet = move(packet), first_fragment_size, &may_send_second_fragment, &sent_second_fragment]() -> intptr_t {
+            for (auto i = 0; i < 5000 && !may_send_second_fragment; ++i)
+                MUST(Core::System::sleep_ms(1));
+
+            sent_second_fragment = true;
+            send_packet_bytes(packet.bytes().slice(first_fragment_size));
+            return 0;
+        });
     }
 
     JsonObject request(StringView to, StringView type)
@@ -1307,6 +1325,13 @@ public:
     }
 
 private:
+    void send_packet_bytes(ReadonlyBytes bytes)
+    {
+        MUST(m_socket->set_blocking(true));
+        auto restore_nonblocking = ScopeGuard([&] { MUST(m_socket->set_blocking(false)); });
+        MUST(m_socket->write_until_depleted(bytes));
+    }
+
     ProtocolClient(Core::EventLoop& loop, NonnullOwnPtr<Core::TCPSocket> socket)
         : m_loop(loop)
         , m_socket(move(socket))
@@ -1768,6 +1793,34 @@ TEST_CASE(root_actor_and_connection_errors)
     EXPECT(client.read_message().has_array("addons"sv));
 }
 
+TEST_CASE(connection_accepts_fragmented_packets)
+{
+    auto session = create_session();
+    auto& client = *session->client;
+
+    (void)client.read_message();
+    EXPECT_EQ(client.request("root"sv, "connect"sv).get_string("from"sv).value(), "root"sv);
+
+    JsonObject request;
+    request.set("to"sv, "root"sv);
+    request.set("type"sv, "getRoot"sv);
+
+    IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> may_send_second_fragment { false };
+    IGNORE_USE_IN_ESCAPING_LAMBDA Atomic<bool> sent_second_fragment { false };
+    auto thread = client.send_in_two_fragments(move(request), 2, may_send_second_fragment, sent_second_fragment);
+    thread->start();
+
+    pump(session->loop);
+    EXPECT(!sent_second_fragment);
+
+    may_send_second_fragment = true;
+    MUST(thread->join());
+
+    auto root = client.read_message();
+    EXPECT_EQ(root.get_string("from"sv).value(), "root"sv);
+    EXPECT(root.has_string("deviceActor"sv));
+}
+
 TEST_CASE(history_navigation_requests)
 {
     auto session = create_session();