OSSForks/ladybird
1
0
Fork 0
mirror of https://github.com/zebrajr/ladybird.git synced 2025-12-06 12:20:00 +01:00
Code Issues Actions 3 Packages Projects Releases Wiki Activity

RequestServer: Manage request lifetimes as a simple state machine

Browse Source 
We currently manage request lifetime as both an ActiveRequest structure
and a series of lambda callbacks. In an upcoming patch, we will want to
"pause" a request to de-duplicate equivalent requests, such that only
one request goes over the network and saves its response to the disk
cache.

To make that easier to reason about, this adds a Request class to manage
the lifetime of a request via a state machine. We will now be able to
add a "waiting for disk cache" state to stop the request.
This commit is contained in:
Timothy Flynn 2025-10-23 20:44:55 -04:00 committed by Andreas Kling
parent 6cf22c424e
commit 822fcc39de
6 changed files with 870 additions and 625 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Services/RequestServer/CMakeLists.txt
View File

@ -9,6 +9,7 @@ set(SOURCES
Cache/Utilities.cpp
ConnectionFromClient.cpp
CURL.cpp
Request.cpp
Resolver.cpp
WebSocketImplCurl.cpp
)

655
Services/RequestServer/ConnectionFromClient.cpp
View File

@ -6,20 +6,17 @@
#include <AK/IDAllocator.h>
#include <AK/NonnullOwnPtr.h>
#include <LibCore/ElapsedTimer.h>
#include <LibCore/EventLoop.h>
#include <LibCore/Proxy.h>
#include <LibCore/Socket.h>
#include <LibCore/StandardPaths.h>
#include <LibRequests/NetworkError.h>
#include <LibRequests/RequestTimingInfo.h>
#include <LibRequests/WebSocket.h>
#include <LibTextCodec/Decoder.h>
#include <LibWebSocket/ConnectionInfo.h>
#include <LibWebSocket/Message.h>
#include <RequestServer/CURL.h>
#include <RequestServer/Cache/DiskCache.h>
#include <RequestServer/ConnectionFromClient.h>
#include <RequestServer/Request.h>
#include <RequestServer/Resolver.h>
#include <RequestServer/WebSocketImplCurl.h>
@ -28,287 +25,8 @@ namespace RequestServer {
static HashMap<int, RefPtr<ConnectionFromClient>> s_connections;
static IDAllocator s_client_ids;
static long s_connect_timeout_seconds = 90L;
Optional<DiskCache> g_disk_cache;
struct ConnectionFromClient::ActiveRequest : public Weakable<ActiveRequest> {
CURLM* multi { nullptr };
CURL* easy { nullptr };
Vector<curl_slist*> curl_string_lists;
i32 request_id { 0 };
WeakPtr<ConnectionFromClient> client;
int writer_fd { 0 };
bool is_connect_only { false };
size_t downloaded_so_far { 0 };
URL::URL url;
ByteString method;
Optional<String> reason_phrase;
ByteBuffer body;
AllocatingMemoryStream send_buffer;
NonnullRefPtr<Core::Notifier> write_notifier;
bool done_fetching { false };
Optional<long> http_status_code;
HTTP::HeaderMap headers;
bool got_all_headers { false };
Optional<size_t> start_offset_of_resumed_response;
size_t bytes_transferred_to_client { 0 };
Optional<CacheEntryWriter&> cache_entry;
UnixDateTime request_start_time;
ActiveRequest(ConnectionFromClient& client, CURLM* multi, CURL* easy, i32 request_id, int writer_fd)
: multi(multi)
, easy(easy)
, request_id(request_id)
, client(client)
, writer_fd(writer_fd)
, write_notifier(Core::Notifier::construct(writer_fd, Core::NotificationType::Write))
, request_start_time(UnixDateTime::now())
{
write_notifier->set_enabled(false);
write_notifier->on_activation = [this] {
if (auto maybe_error = write_queued_bytes_without_blocking(); maybe_error.is_error()) {
dbgln("Warning: Failed to write buffered request data (it's likely the client disappeared): {}", maybe_error.error());
}
};
}
void schedule_self_destruction() const
{
Core::deferred_invoke([weak_this = make_weak_ptr()] {
if (!weak_this)
return;
if (weak_this->client)
weak_this->client->m_active_requests.remove(weak_this->request_id);
});
}
ErrorOr<void> write_queued_bytes_without_blocking()
{
auto available_bytes = send_buffer.used_buffer_size();
// If we've received a response to a range request that is not the partial content (206) we requested, we must
// only transfer the subset of data that WebContent now needs. We discard all received bytes up to the expected
// start of the remaining data, and then transfer the remaining bytes.
if (start_offset_of_resumed_response.has_value()) {
if (http_status_code == 206) {
start_offset_of_resumed_response.clear();
} else if (http_status_code == 200) {
// All bytes currently available have already been transferred. Discard them entirely.
if (bytes_transferred_to_client + available_bytes <= *start_offset_of_resumed_response) {
bytes_transferred_to_client += available_bytes;
MUST(send_buffer.discard(available_bytes));
return {};
}
// Some bytes currently available have already been transferred. Discard those bytes and transfer the rest.
if (bytes_transferred_to_client + available_bytes > *start_offset_of_resumed_response) {
auto bytes_to_discard = *start_offset_of_resumed_response - bytes_transferred_to_client;
bytes_transferred_to_client += bytes_to_discard;
available_bytes -= bytes_to_discard;
MUST(send_buffer.discard(bytes_to_discard));
}
start_offset_of_resumed_response.clear();
} else {
return Error::from_string_literal("Unacceptable status code for resumed HTTP request");
}
}
Vector<u8> bytes_to_send;
bytes_to_send.resize(available_bytes);
send_buffer.peek_some(bytes_to_send);
auto result = Core::System::write(this->writer_fd, bytes_to_send);
if (result.is_error()) {
if (result.error().code() != EAGAIN) {
return result.release_error();
}
write_notifier->set_enabled(true);
return {};
}
if (cache_entry.has_value()) {
auto bytes_sent = bytes_to_send.span().slice(0, result.value());
if (cache_entry->write_data(bytes_sent).is_error())
cache_entry.clear();
}
bytes_transferred_to_client += result.value();
MUST(send_buffer.discard(result.value()));
write_notifier->set_enabled(!send_buffer.is_eof());
if (send_buffer.is_eof() && done_fetching)
schedule_self_destruction();
return {};
}
void notify_about_fetching_completion()
{
done_fetching = true;
if (send_buffer.is_eof())
schedule_self_destruction();
}
~ActiveRequest()
{
if (!send_buffer.is_eof()) {
dbgln("Warning: Request destroyed with buffered data (it's likely that the client disappeared or the request was cancelled)");
}
if (writer_fd > 0)
MUST(Core::System::close(writer_fd));
auto result = curl_multi_remove_handle(multi, easy);
VERIFY(result == CURLM_OK);
curl_easy_cleanup(easy);
for (auto* string_list : curl_string_lists)
curl_slist_free_all(string_list);
if (cache_entry.has_value())
(void)cache_entry->flush();
}
void flush_headers_if_needed()
{
if (!http_status_code.has_value())
http_status_code = acquire_http_status_code();
if (got_all_headers)
return;
got_all_headers = true;
client->async_headers_became_available(request_id, headers, *http_status_code, reason_phrase);
if (g_disk_cache.has_value())
cache_entry = g_disk_cache->create_entry(url, method, *http_status_code, reason_phrase, headers, request_start_time);
}
long acquire_http_status_code() const
{
long code = 0;
auto result = curl_easy_getinfo(easy, CURLINFO_RESPONSE_CODE, &code);
VERIFY(result == CURLE_OK);
return code;
}
};
size_t ConnectionFromClient::on_header_received(void* buffer, size_t size, size_t nmemb, void* user_data)
{
auto* request = static_cast<ActiveRequest*>(user_data);
size_t total_size = size * nmemb;
auto header_line = StringView { static_cast<char const*>(buffer), total_size };
// NOTE: We need to extract the HTTP reason phrase since it can be a custom value.
// Fetching infrastructure needs this value for setting the status message.
if (!request->reason_phrase.has_value() && header_line.starts_with("HTTP/"sv)) {
if (auto const space_positions = header_line.find_all(" "sv); space_positions.size() > 1) {
auto const second_space_offset = space_positions.at(1);
auto const reason_phrase_string_view = header_line.substring_view(second_space_offset + 1).trim_whitespace();
if (!reason_phrase_string_view.is_empty()) {
auto decoder = TextCodec::decoder_for_exact_name("ISO-8859-1"sv);
VERIFY(decoder.has_value());
request->reason_phrase = MUST(decoder->to_utf8(reason_phrase_string_view));
return total_size;
}
}
}
if (auto colon_index = header_line.find(':'); colon_index.has_value()) {
auto name = header_line.substring_view(0, colon_index.value()).trim_whitespace();
auto value = header_line.substring_view(colon_index.value() + 1, header_line.length() - colon_index.value() - 1).trim_whitespace();
request->headers.set(name, value);
}
return total_size;
}
size_t ConnectionFromClient::on_data_received(void* buffer, size_t size, size_t nmemb, void* user_data)
{
auto* request = static_cast<ActiveRequest*>(user_data);
request->flush_headers_if_needed();
size_t total_size = size * nmemb;
ReadonlyBytes bytes { static_cast<u8 const*>(buffer), total_size };
auto maybe_write_error = [&] -> ErrorOr<void> {
TRY(request->send_buffer.write_some(bytes));
return request->write_queued_bytes_without_blocking();
}();
if (maybe_write_error.is_error()) {
dbgln("ConnectionFromClient::on_data_received: Aborting request because error occurred whilst writing data to the client: {}", maybe_write_error.error());
return CURL_WRITEFUNC_ERROR;
}
request->downloaded_so_far += total_size;
return total_size;
}
int ConnectionFromClient::on_socket_callback(CURL*, int sockfd, int what, void* user_data, void*)
{
auto* client = static_cast<ConnectionFromClient*>(user_data);
if (what == CURL_POLL_REMOVE) {
client->m_read_notifiers.remove(sockfd);
client->m_write_notifiers.remove(sockfd);
return 0;
}
if (what & CURL_POLL_IN) {
client->m_read_notifiers.ensure(sockfd, [client, sockfd, multi = client->m_curl_multi] {
auto notifier = Core::Notifier::construct(sockfd, Core::NotificationType::Read);
notifier->on_activation = [client, sockfd, multi] {
auto result = curl_multi_socket_action(multi, sockfd, CURL_CSELECT_IN, nullptr);
VERIFY(result == CURLM_OK);
client->check_active_requests();
};
notifier->set_enabled(true);
return notifier;
});
}
if (what & CURL_POLL_OUT) {
client->m_write_notifiers.ensure(sockfd, [client, sockfd, multi = client->m_curl_multi] {
auto notifier = Core::Notifier::construct(sockfd, Core::NotificationType::Write);
notifier->on_activation = [client, sockfd, multi] {
auto result = curl_multi_socket_action(multi, sockfd, CURL_CSELECT_OUT, nullptr);
VERIFY(result == CURLM_OK);
client->check_active_requests();
};
notifier->set_enabled(true);
return notifier;
});
}
return 0;
}
int ConnectionFromClient::on_timeout_callback(void*, long timeout_ms, void* user_data)
{
auto* client = static_cast<ConnectionFromClient*>(user_data);
if (!client->m_timer)
return 0;
if (timeout_ms < 0) {
client->m_timer->stop();
} else {
client->m_timer->restart(timeout_ms);
}
return 0;
}
ConnectionFromClient::ConnectionFromClient(NonnullOwnPtr<IPC::Transport> transport)
: IPC::ConnectionFromClient<RequestClientEndpoint, RequestServerEndpoint>(*this, move(transport), s_client_ids.allocate())
, m_resolver(Resolver::default_resolver())
@ -343,6 +61,14 @@ ConnectionFromClient::~ConnectionFromClient()
m_curl_multi = nullptr;
}
void ConnectionFromClient::request_complete(Badge<Request>, int request_id)
{
Core::deferred_invoke([weak_self = make_weak_ptr<ConnectionFromClient>(), request_id] {
if (auto self = weak_self.strong_ref())
self->m_active_requests.remove(request_id);
});
}
void ConnectionFromClient::die()
{
auto client_id = this->client_id();
@ -455,266 +181,69 @@ void ConnectionFromClient::set_use_system_dns()
m_resolver->dns.reset_connection();
}
#ifdef AK_OS_WINDOWS
void ConnectionFromClient::start_request(i32, ByteString, URL::URL, HTTP::HeaderMap, ByteBuffer, Core::ProxyData)
{
VERIFY(0 && "RequestServer::ConnectionFromClient::start_request is not implemented");
}
void ConnectionFromClient::issue_network_request(i32, ByteString, URL::URL, HTTP::HeaderMap, ByteBuffer, Core::ProxyData, Optional<ResumeRequestForFailedCacheEntry>)
{
VERIFY(0 && "RequestServer::ConnectionFromClient::issue_network_request is not implemented");
}
#else
void ConnectionFromClient::start_request(i32 request_id, ByteString method, URL::URL url, HTTP::HeaderMap request_headers, ByteBuffer request_body, Core::ProxyData proxy_data)
{
dbgln_if(REQUESTSERVER_DEBUG, "RequestServer: start_request({}, {})", request_id, url);
if (g_disk_cache.has_value()) {
if (auto cache_entry = g_disk_cache->open_entry(url, method); cache_entry.has_value()) {
auto fds = MUST(Core::System::pipe2(O_NONBLOCK));
auto writer_fd = fds[1];
auto reader_fd = fds[0];
async_request_started(request_id, IPC::File::adopt_fd(reader_fd));
async_headers_became_available(request_id, cache_entry->headers(), cache_entry->status_code(), cache_entry->reason_phrase());
cache_entry->pipe_to(
writer_fd,
[this, request_id, writer_fd](auto bytes_sent) {
// FIXME: Implement timing info for cache hits.
async_request_finished(request_id, bytes_sent, {}, {});
MUST(Core::System::close(writer_fd));
},
[this, request_id, writer_fd, method = move(method), url = move(url), request_headers = move(request_headers), request_body = move(request_body), proxy_data](auto bytes_sent) mutable {
// FIXME: We should really also have a way to validate the data once CacheEntry is storing its crc.
ResumeRequestForFailedCacheEntry resume_request {
.start_offset = bytes_sent,
.writer_fd = writer_fd,
};
issue_network_request(request_id, move(method), move(url), move(request_headers), move(request_body), proxy_data, resume_request);
});
return;
}
auto request = Request::fetch(request_id, g_disk_cache, *this, m_curl_multi, m_resolver, move(url), move(method), move(request_headers), move(request_body), m_alt_svc_cache_path, proxy_data);
m_active_requests.set(request_id, move(request));
}
issue_network_request(request_id, move(method), move(url), move(request_headers), move(request_body), proxy_data);
}
void ConnectionFromClient::issue_network_request(i32 request_id, ByteString method, URL::URL url, HTTP::HeaderMap request_headers, ByteBuffer request_body, Core::ProxyData proxy_data, Optional<ResumeRequestForFailedCacheEntry> resume_request)
int ConnectionFromClient::on_socket_callback(CURL*, int sockfd, int what, void* user_data, void*)
{
auto host = url.serialized_host().to_byte_string();
auto const& dns_info = DNSInfo::the();
auto* client = static_cast<ConnectionFromClient*>(user_data);
m_resolver->dns.lookup(host, DNS::Messages::Class::IN, { DNS::Messages::ResourceType::A, DNS::Messages::ResourceType::AAAA }, { .validate_dnssec_locally = dns_info.validate_dnssec_locally })
->when_rejected([this, request_id, resume_request](auto const& error) {
dbgln("StartRequest: DNS lookup failed: {}", error);
// FIXME: Implement timing info for DNS lookup failure.
async_request_finished(request_id, 0, {}, Requests::NetworkError::UnableToResolveHost);
if (resume_request.has_value())
MUST(Core::System::close(resume_request->writer_fd));
})
.when_resolved([this, request_id, host = move(host), url = move(url), method = move(method), request_body = move(request_body), request_headers = move(request_headers), proxy_data, resume_request](auto const& dns_result) mutable {
if (dns_result->is_empty() || !dns_result->has_cached_addresses()) {
dbgln("StartRequest: DNS lookup failed for '{}'", host);
// FIXME: Implement timing info for DNS lookup failure.
async_request_finished(request_id, 0, {}, Requests::NetworkError::UnableToResolveHost);
return;
if (what == CURL_POLL_REMOVE) {
client->m_read_notifiers.remove(sockfd);
client->m_write_notifiers.remove(sockfd);
return 0;
}
dbgln_if(REQUESTSERVER_DEBUG, "RequestServer: DNS lookup successful");
auto* easy = curl_easy_init();
if (!easy) {
dbgln("StartRequest: Failed to initialize curl easy handle");
return;
}
int writer_fd = 0;
if (resume_request.has_value()) {
writer_fd = resume_request->writer_fd;
} else {
auto fds_or_error = Core::System::pipe2(O_NONBLOCK);
if (fds_or_error.is_error()) {
dbgln("StartRequest: Failed to create pipe: {}", fds_or_error.error());
return;
}
auto fds = fds_or_error.release_value();
auto reader_fd = fds[0];
writer_fd = fds[1];
async_request_started(request_id, IPC::File::adopt_fd(reader_fd));
}
auto request = make<ActiveRequest>(*this, m_curl_multi, easy, request_id, writer_fd);
request->url = url;
request->method = method;
auto set_option = [easy](auto option, auto value) {
auto result = curl_easy_setopt(easy, option, value);
if (result != CURLE_OK)
dbgln("StartRequest: Failed to set curl option: {}", curl_easy_strerror(result));
};
set_option(CURLOPT_PRIVATE, request.ptr());
if (auto const& path = default_certificate_path(); !path.is_empty())
set_option(CURLOPT_CAINFO, path.characters());
set_option(CURLOPT_ACCEPT_ENCODING, ""); // empty string lets curl define the accepted encodings
set_option(CURLOPT_URL, url.to_string().to_byte_string().characters());
set_option(CURLOPT_PORT, url.port_or_default());
set_option(CURLOPT_CONNECTTIMEOUT, s_connect_timeout_seconds);
set_option(CURLOPT_PIPEWAIT, 1L);
set_option(CURLOPT_ALTSVC, m_alt_svc_cache_path.characters());
set_option(CURLOPT_CUSTOMREQUEST, method.characters());
set_option(CURLOPT_FOLLOWLOCATION, 0);
bool did_set_body = false;
if (method.is_one_of("POST"sv, "PUT"sv, "PATCH"sv, "DELETE"sv)) {
request->body = move(request_body);
set_option(CURLOPT_POSTFIELDSIZE, request->body.size());
set_option(CURLOPT_POSTFIELDS, request->body.data());
did_set_body = true;
} else if (method == "HEAD"sv) {
set_option(CURLOPT_NOBODY, 1L);
}
struct curl_slist* curl_headers = nullptr;
// NOTE: CURLOPT_POSTFIELDS automatically sets the Content-Type header.
// Tell curl to remove it by setting a blank value if the headers passed in don't contain a content type.
if (did_set_body && !request_headers.contains("Content-Type"))
curl_headers = curl_slist_append(curl_headers, "Content-Type:");
for (auto const& header : request_headers.headers()) {
if (header.value.is_empty()) {
// Special case for headers with an empty value. curl will discard the header unless we pass the
// header name followed by a semicolon.
//
// i.e. we need to pass "Content-Type;" instead of "Content-Type: "
//
// See: https://curl.se/libcurl/c/httpcustomheader.html
auto header_string = ByteString::formatted("{};", header.name);
curl_headers = curl_slist_append(curl_headers, header_string.characters());
continue;
}
auto header_string = ByteString::formatted("{}: {}", header.name, header.value);
dbgln_if(REQUESTSERVER_DEBUG, "RequestServer: Request header: {}", header_string);
curl_headers = curl_slist_append(curl_headers, header_string.characters());
}
if (curl_headers) {
set_option(CURLOPT_HTTPHEADER, curl_headers);
request->curl_string_lists.append(curl_headers);
}
if (resume_request.has_value()) {
auto range = ByteString::formatted("{}-", resume_request->start_offset);
set_option(CURLOPT_RANGE, range.characters());
request->got_all_headers = true; // Don't re-send the headers for resumed requests.
request->start_offset_of_resumed_response = resume_request->start_offset;
}
// FIXME: Set up proxy if applicable
(void)proxy_data;
set_option(CURLOPT_WRITEFUNCTION, &on_data_received);
set_option(CURLOPT_WRITEDATA, reinterpret_cast<void*>(request.ptr()));
set_option(CURLOPT_HEADERFUNCTION, &on_header_received);
set_option(CURLOPT_HEADERDATA, reinterpret_cast<void*>(request.ptr()));
auto formatted_address = build_curl_resolve_list(*dns_result, host, url.port_or_default());
if (curl_slist* resolve_list = curl_slist_append(nullptr, formatted_address.characters())) {
set_option(CURLOPT_RESOLVE, resolve_list);
request->curl_string_lists.append(resolve_list);
} else
VERIFY_NOT_REACHED();
auto result = curl_multi_add_handle(m_curl_multi, easy);
if (what & CURL_POLL_IN) {
client->m_read_notifiers.ensure(sockfd, [client, sockfd, multi = client->m_curl_multi] {
auto notifier = Core::Notifier::construct(sockfd, Core::NotificationType::Read);
notifier->on_activation = [client, sockfd, multi] {
auto result = curl_multi_socket_action(multi, sockfd, CURL_CSELECT_IN, nullptr);
VERIFY(result == CURLM_OK);
m_active_requests.set(request_id, move(request));
client->check_active_requests();
};
notifier->set_enabled(true);
return notifier;
});
}
#endif
static Requests::RequestTimingInfo get_timing_info_from_curl_easy_handle(CURL* easy_handle)
{
/*
* curl_easy_perform()
* |
* |--QUEUE
* |--|--NAMELOOKUP
* |--|--|--CONNECT
* |--|--|--|--APPCONNECT
* |--|--|--|--|--PRETRANSFER
* |--|--|--|--|--|--POSTTRANSFER
* |--|--|--|--|--|--|--STARTTRANSFER
* |--|--|--|--|--|--|--|--TOTAL
* |--|--|--|--|--|--|--|--REDIRECT
*/
if (what & CURL_POLL_OUT) {
client->m_write_notifiers.ensure(sockfd, [client, sockfd, multi = client->m_curl_multi] {
auto notifier = Core::Notifier::construct(sockfd, Core::NotificationType::Write);
notifier->on_activation = [client, sockfd, multi] {
auto result = curl_multi_socket_action(multi, sockfd, CURL_CSELECT_OUT, nullptr);
VERIFY(result == CURLM_OK);
auto get_timing_info = [easy_handle](auto option) {
curl_off_t time_value = 0;
auto result = curl_easy_getinfo(easy_handle, option, &time_value);
VERIFY(result == CURLE_OK);
return time_value;
client->check_active_requests();
};
auto queue_time = get_timing_info(CURLINFO_QUEUE_TIME_T);
auto domain_lookup_time = get_timing_info(CURLINFO_NAMELOOKUP_TIME_T);
auto connect_time = get_timing_info(CURLINFO_CONNECT_TIME_T);
auto secure_connect_time = get_timing_info(CURLINFO_APPCONNECT_TIME_T);
auto request_start_time = get_timing_info(CURLINFO_PRETRANSFER_TIME_T);
auto response_start_time = get_timing_info(CURLINFO_STARTTRANSFER_TIME_T);
auto response_end_time = get_timing_info(CURLINFO_TOTAL_TIME_T);
auto encoded_body_size = get_timing_info(CURLINFO_SIZE_DOWNLOAD_T);
long http_version = 0;
auto get_version_result = curl_easy_getinfo(easy_handle, CURLINFO_HTTP_VERSION, &http_version);
VERIFY(get_version_result == CURLE_OK);
auto http_version_alpn = Requests::ALPNHttpVersion::None;
switch (http_version) {
case CURL_HTTP_VERSION_1_0:
http_version_alpn = Requests::ALPNHttpVersion::Http1_0;
break;
case CURL_HTTP_VERSION_1_1:
http_version_alpn = Requests::ALPNHttpVersion::Http1_1;
break;
case CURL_HTTP_VERSION_2_0:
http_version_alpn = Requests::ALPNHttpVersion::Http2_TLS;
break;
case CURL_HTTP_VERSION_3:
http_version_alpn = Requests::ALPNHttpVersion::Http3;
break;
default:
http_version_alpn = Requests::ALPNHttpVersion::None;
break;
notifier->set_enabled(true);
return notifier;
});
}
return Requests::RequestTimingInfo {
.domain_lookup_start_microseconds = queue_time,
.domain_lookup_end_microseconds = queue_time + domain_lookup_time,
.connect_start_microseconds = queue_time + domain_lookup_time,
.connect_end_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time,
.secure_connect_start_microseconds = queue_time + domain_lookup_time + connect_time,
.request_start_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + request_start_time,
.response_start_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + response_start_time,
.response_end_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + response_end_time,
.encoded_body_size = encoded_body_size,
.http_version_alpn_identifier = http_version_alpn,
};
return 0;
}
int ConnectionFromClient::on_timeout_callback(void*, long timeout_ms, void* user_data)
{
auto* client = static_cast<ConnectionFromClient*>(user_data);
if (!client->m_timer)
return 0;
if (timeout_ms < 0)
client->m_timer->stop();
else
client->m_timer->restart(timeout_ms);
return 0;
}
void ConnectionFromClient::check_active_requests()
@ -741,36 +270,8 @@ void ConnectionFromClient::check_active_requests()
continue;
}
auto* request = static_cast<ActiveRequest*>(application_private);
if (!request->is_connect_only) {
auto timing_info = get_timing_info_from_curl_easy_handle(msg->easy_handle);
request->flush_headers_if_needed();
auto result_code = msg->data.result;
// HTTPS servers might terminate their connection without proper notice of shutdown - i.e. they do not send
// a "close notify" alert. OpenSSL version 3.2 began treating this as an error, which curl translates to
// CURLE_RECV_ERROR in the absence of a Content-Length response header. The Python server used by WPT is one
// such server. We ignore this error if we were actually able to download some response data.
if (result_code == CURLE_RECV_ERROR && request->downloaded_so_far != 0 && !request->headers.contains("Content-Length"sv))
result_code = CURLE_OK;
Optional<Requests::NetworkError> network_error;
bool const request_was_successful = result_code == CURLE_OK;
if (!request_was_successful) {
network_error = map_curl_code_to_network_error(result_code);
if (network_error.has_value() && network_error.value() == Requests::NetworkError::Unknown) {
char const* curl_error_message = curl_easy_strerror(result_code);
dbgln("ConnectionFromClient: Unable to map error ({}), message: \"\033[31;1m{}\033[0m\"", static_cast<int>(result_code), curl_error_message);
}
}
async_request_finished(request->request_id, request->downloaded_so_far, timing_info, network_error);
}
request->notify_about_fetching_completion();
auto* request = static_cast<Request*>(application_private);
request->notify_fetch_complete({}, msg->data.result);
}
}
@ -795,56 +296,10 @@ Messages::RequestServer::SetCertificateResponse ConnectionFromClient::set_certif
void ConnectionFromClient::ensure_connection(URL::URL url, ::RequestServer::CacheLevel cache_level)
{
if (cache_level == CacheLevel::CreateConnection) {
auto* easy = curl_easy_init();
if (!easy) {
dbgln("EnsureConnection: Failed to initialize curl easy handle");
return;
}
auto set_option = [easy](auto option, auto value) {
auto result = curl_easy_setopt(easy, option, value);
if (result != CURLE_OK) {
dbgln("EnsureConnection: Failed to set curl option: {}", curl_easy_strerror(result));
return false;
}
return true;
};
auto connect_only_request_id = get_random<i32>();
auto request = make<ActiveRequest>(*this, m_curl_multi, easy, connect_only_request_id, 0);
request->url = url;
request->is_connect_only = true;
set_option(CURLOPT_PRIVATE, request.ptr());
set_option(CURLOPT_URL, url.to_byte_string().characters());
set_option(CURLOPT_PORT, url.port_or_default());
set_option(CURLOPT_CONNECTTIMEOUT, s_connect_timeout_seconds);
set_option(CURLOPT_CONNECT_ONLY, 1L);
auto const result = curl_multi_add_handle(m_curl_multi, easy);
VERIFY(result == CURLM_OK);
auto request = Request::connect(connect_only_request_id, *this, m_curl_multi, m_resolver, move(url), cache_level);
m_active_requests.set(connect_only_request_id, move(request));
return;
}
if (cache_level == CacheLevel::ResolveOnly) {
auto const& dns_info = DNSInfo::the();
[[maybe_unused]] auto promise = m_resolver->dns.lookup(url.serialized_host().to_byte_string(), DNS::Messages::Class::IN, { DNS::Messages::ResourceType::A, DNS::Messages::ResourceType::AAAA }, { .validate_dnssec_locally = dns_info.validate_dnssec_locally });
if constexpr (REQUESTSERVER_DEBUG) {
Core::ElapsedTimer timer;
timer.start();
promise->when_resolved([url, timer](auto const& results) -> ErrorOr<void> {
dbgln("ensure_connection::ResolveOnly({}) OK {} entrie(s) in {}ms", url, results->cached_addresses().size(), timer.elapsed_milliseconds());
return {};
});
promise->when_rejected([url](auto const&) { dbgln("ensure_connection::ResolveOnly({}) rejected", url); });
}
}
}
void ConnectionFromClient::clear_cache()

29
Services/RequestServer/ConnectionFromClient.h
View File

@ -6,6 +6,7 @@
#pragma once
#include <AK/Badge.h>
#include <AK/HashMap.h>
#include <LibIPC/ConnectionFromClient.h>
#include <LibWebSocket/WebSocket.h>
@ -24,6 +25,8 @@ public:
virtual void die() override;
void request_complete(Badge<Request>, int request_id);
private:
explicit ConnectionFromClient(NonnullOwnPtr<IPC::Transport>);
@ -46,31 +49,21 @@ private:
virtual void websocket_close(i64 websocket_id, u16, ByteString) override;
virtual Messages::RequestServer::WebsocketSetCertificateResponse websocket_set_certificate(i64, ByteString, ByteString) override;
struct ResumeRequestForFailedCacheEntry {
size_t start_offset { 0 };
int writer_fd { 0 };
};
void issue_network_request(i32 request_id, ByteString, URL::URL, HTTP::HeaderMap, ByteBuffer, Core::ProxyData, Optional<ResumeRequestForFailedCacheEntry> = {});
HashMap<i32, RefPtr<WebSocket::WebSocket>> m_websockets;
struct ActiveRequest;
friend struct ActiveRequest;
static int on_socket_callback(void*, int sockfd, int what, void* user_data, void*);
static int on_timeout_callback(void*, long timeout_ms, void* user_data);
void check_active_requests();
static ErrorOr<IPC::File> create_client_socket();
static int on_socket_callback(void*, int sockfd, int what, void* user_data, void*);
static int on_timeout_callback(void*, long timeout_ms, void* user_data);
static size_t on_header_received(void* buffer, size_t size, size_t nmemb, void* user_data);
static size_t on_data_received(void* buffer, size_t size, size_t nmemb, void* user_data);
HashMap<i32, NonnullOwnPtr<ActiveRequest>> m_active_requests;
void check_active_requests();
void* m_curl_multi { nullptr };
HashMap<i32, NonnullOwnPtr<Request>> m_active_requests;
HashMap<i32, RefPtr<WebSocket::WebSocket>> m_websockets;
RefPtr<Core::Timer> m_timer;
HashMap<int, NonnullRefPtr<Core::Notifier>> m_read_notifiers;
HashMap<int, NonnullRefPtr<Core::Notifier>> m_write_notifiers;
NonnullRefPtr<Resolver> m_resolver;
ByteString m_alt_svc_cache_path;
};

2
Services/RequestServer/Forward.h
View File

@ -12,7 +12,9 @@ class CacheEntry;
class CacheEntryReader;
class CacheEntryWriter;
class CacheIndex;
class ConnectionFromClient;
class DiskCache;
class Request;
struct DNSInfo;
struct Resolver;

637
Services/RequestServer/Request.cpp Normal file
View File

@ -0,0 +1,637 @@
/*
* Copyright (c) 2024, Andreas Kling <andreas@ladybird.org>
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#include <AK/Enumerate.h>
#include <LibCore/Notifier.h>
#include <LibTextCodec/Decoder.h>
#include <RequestServer/CURL.h>
#include <RequestServer/Cache/DiskCache.h>
#include <RequestServer/ConnectionFromClient.h>
#include <RequestServer/Request.h>
#include <RequestServer/Resolver.h>
namespace RequestServer {
static long s_connect_timeout_seconds = 90L;
NonnullOwnPtr<Request> Request::fetch(
i32 request_id,
Optional<DiskCache&> disk_cache,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
ByteString method,
HTTP::HeaderMap request_headers,
ByteBuffer request_body,
ByteString alt_svc_cache_path,
Core::ProxyData proxy_data)
{
auto request = adopt_own(*new Request { request_id, disk_cache, client, curl_multi, resolver, move(url), move(method), move(request_headers), move(request_body), move(alt_svc_cache_path), proxy_data });
request->process();
return request;
}
NonnullOwnPtr<Request> Request::connect(
i32 request_id,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
CacheLevel cache_level)
{
auto request = adopt_own(*new Request { request_id, client, curl_multi, resolver, move(url) });
switch (cache_level) {
case CacheLevel::ResolveOnly:
request->transition_to_state(State::DNSLookup);
break;
case CacheLevel::CreateConnection:
request->transition_to_state(State::Connect);
break;
}
return request;
}
Request::Request(
i32 request_id,
Optional<DiskCache&> disk_cache,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
ByteString method,
HTTP::HeaderMap request_headers,
ByteBuffer request_body,
ByteString alt_svc_cache_path,
Core::ProxyData proxy_data)
: m_request_id(request_id)
, m_type(Type::Fetch)
, m_disk_cache(disk_cache)
, m_client(client)
, m_curl_multi_handle(curl_multi)
, m_resolver(resolver)
, m_url(move(url))
, m_method(move(method))
, m_request_headers(move(request_headers))
, m_request_body(move(request_body))
, m_alt_svc_cache_path(move(alt_svc_cache_path))
, m_proxy_data(proxy_data)
{
}
Request::Request(
i32 request_id,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url)
: m_request_id(request_id)
, m_type(Type::Connect)
, m_client(client)
, m_curl_multi_handle(curl_multi)
, m_resolver(resolver)
, m_url(move(url))
{
}
Request::~Request()
{
if (!m_response_buffer.is_eof())
dbgln("Warning: Request destroyed with buffered data (it's likely that the client disappeared or the request was cancelled)");
if (m_client_writer_fd != -1)
MUST(Core::System::close(m_client_writer_fd));
if (m_curl_easy_handle) {
auto result = curl_multi_remove_handle(m_curl_multi_handle, m_curl_easy_handle);
VERIFY(result == CURLM_OK);
curl_easy_cleanup(m_curl_easy_handle);
}
for (auto* string_list : m_curl_string_lists)
curl_slist_free_all(string_list);
if (m_cache_entry_writer.has_value())
(void)m_cache_entry_writer->flush();
}
void Request::notify_fetch_complete(Badge<ConnectionFromClient>, int result_code)
{
m_curl_result_code = result_code;
if (m_response_buffer.is_eof())
transition_to_state(State::Complete);
}
void Request::transition_to_state(State state)
{
m_state = state;
process();
}
void Request::process()
{
switch (m_state) {
case State::Init:
handle_initial_state();
break;
case State::ReadCache:
handle_read_cache_state();
break;
case State::DNSLookup:
handle_dns_lookup_state();
break;
case State::Connect:
handle_connect_state();
break;
case State::Fetch:
handle_fetch_state();
break;
case State::Complete:
handle_complete_state();
break;
case State::Error:
handle_error_state();
break;
}
}
void Request::handle_initial_state()
{
if (m_disk_cache.has_value()) {
m_cache_entry_reader = m_disk_cache->open_entry(m_url, m_method);
if (m_cache_entry_reader.has_value()) {
transition_to_state(State::ReadCache);
return;
}
}
transition_to_state(State::DNSLookup);
}
void Request::handle_read_cache_state()
{
#if defined(AK_OS_WINDOWS)
dbgln("FIXME: Request::handle_read_from_cache_state: Not implemented on Windows");
transition_to_state(State::Error);
#else
m_status_code = m_cache_entry_reader->status_code();
m_reason_phrase = m_cache_entry_reader->reason_phrase();
m_response_headers = m_cache_entry_reader->headers();
auto fds = Core::System::pipe2(O_NONBLOCK);
if (fds.is_error()) {
dbgln("Request::handle_read_from_cache_state: Failed to create pipe: {}", fds.error());
transition_to_state(State::Error);
return;
}
m_client.async_request_started(m_request_id, IPC::File::adopt_fd(fds.value().at(0)));
m_client_writer_fd = fds.value().at(1);
m_client.async_headers_became_available(m_request_id, m_response_headers, m_status_code, m_reason_phrase);
m_sent_response_headers_to_client = true;
m_cache_entry_reader->pipe_to(
m_client_writer_fd,
[this](auto bytes_sent) {
m_bytes_transferred_to_client = bytes_sent;
m_curl_result_code = CURLE_OK;
transition_to_state(State::Complete);
},
[this](auto bytes_sent) {
// FIXME: We should also have a way to validate the data once CacheEntry is storing its crc.
m_start_offset_of_response_resumed_from_cache = bytes_sent;
m_disk_cache.clear();
transition_to_state(State::DNSLookup);
});
#endif
}
void Request::handle_dns_lookup_state()
{
auto host = m_url.serialized_host().to_byte_string();
auto const& dns_info = DNSInfo::the();
m_resolver->dns.lookup(host, DNS::Messages::Class::IN, { DNS::Messages::ResourceType::A, DNS::Messages::ResourceType::AAAA }, { .validate_dnssec_locally = dns_info.validate_dnssec_locally })
->when_rejected([this, host](auto const& error) {
dbgln("Request::handle_dns_lookup_state: DNS lookup failed for '{}': {}", host, error);
m_network_error = Requests::NetworkError::UnableToResolveHost;
transition_to_state(State::Error);
})
.when_resolved([this, host](NonnullRefPtr<DNS::LookupResult const> dns_result) mutable {
if (dns_result->is_empty() || !dns_result->has_cached_addresses()) {
dbgln("Request::handle_dns_lookup_state: DNS lookup failed for '{}'", host);
m_network_error = Requests::NetworkError::UnableToResolveHost;
transition_to_state(State::Error);
} else if (m_type == Type::Fetch) {
m_dns_result = move(dns_result);
transition_to_state(State::Fetch);
} else {
transition_to_state(State::Complete);
}
});
}
void Request::handle_connect_state()
{
m_curl_easy_handle = curl_easy_init();
if (!m_curl_easy_handle) {
dbgln("Request::handle_connect_state: Failed to initialize curl easy handle");
return;
}
auto set_option = [&](auto option, auto value) {
if (auto result = curl_easy_setopt(m_curl_easy_handle, option, value); result != CURLE_OK)
dbgln("Request::handle_connect_state: Failed to set curl option: {}", curl_easy_strerror(result));
};
set_option(CURLOPT_PRIVATE, this);
set_option(CURLOPT_URL, m_url.to_byte_string().characters());
set_option(CURLOPT_PORT, m_url.port_or_default());
set_option(CURLOPT_CONNECTTIMEOUT, s_connect_timeout_seconds);
set_option(CURLOPT_CONNECT_ONLY, 1L);
auto result = curl_multi_add_handle(m_curl_multi_handle, m_curl_easy_handle);
VERIFY(result == CURLM_OK);
}
void Request::handle_fetch_state()
{
#if defined(AK_OS_WINDOWS)
dbgln("FIXME: Request::handle_fetch_state: Not implemented on Windows");
transition_to_state(State::Error);
#else
dbgln_if(REQUESTSERVER_DEBUG, "RequestServer: DNS lookup successful");
m_curl_easy_handle = curl_easy_init();
if (!m_curl_easy_handle) {
dbgln("Request::handle_start_fetch_state: Failed to initialize curl easy handle");
transition_to_state(State::Error);
return;
}
if (!m_start_offset_of_response_resumed_from_cache.has_value()) {
auto fds = Core::System::pipe2(O_NONBLOCK);
if (fds.is_error()) {
dbgln("Request::handle_start_fetch_state: Failed to create pipe: {}", fds.error());
transition_to_state(State::Error);
return;
}
m_client.async_request_started(m_request_id, IPC::File::adopt_fd(fds.value().at(0)));
m_client_writer_fd = fds.value().at(1);
}
m_client_writer_notifier = Core::Notifier::construct(m_client_writer_fd, Core::NotificationType::Write);
m_client_writer_notifier->set_enabled(false);
m_client_writer_notifier->on_activation = [this] {
if (auto result = write_queued_bytes_without_blocking(); result.is_error())
dbgln("Warning: Failed to write buffered request data (it's likely the client disappeared): {}", result.error());
};
auto set_option = [&](auto option, auto value) {
if (auto result = curl_easy_setopt(m_curl_easy_handle, option, value); result != CURLE_OK)
dbgln("Request::handle_start_fetch_state: Failed to set curl option: {}", curl_easy_strerror(result));
};
set_option(CURLOPT_PRIVATE, this);
if (auto const& path = default_certificate_path(); !path.is_empty())
set_option(CURLOPT_CAINFO, path.characters());
set_option(CURLOPT_ACCEPT_ENCODING, ""); // Empty string lets curl define the accepted encodings.
set_option(CURLOPT_URL, m_url.to_byte_string().characters());
set_option(CURLOPT_PORT, m_url.port_or_default());
set_option(CURLOPT_CONNECTTIMEOUT, s_connect_timeout_seconds);
set_option(CURLOPT_PIPEWAIT, 1L);
set_option(CURLOPT_ALTSVC, m_alt_svc_cache_path.characters());
set_option(CURLOPT_CUSTOMREQUEST, m_method.characters());
set_option(CURLOPT_FOLLOWLOCATION, 0);
curl_slist* curl_headers = nullptr;
if (m_method.is_one_of("POST"sv, "PUT"sv, "PATCH"sv, "DELETE"sv)) {
set_option(CURLOPT_POSTFIELDSIZE, m_request_body.size());
set_option(CURLOPT_POSTFIELDS, m_request_body.data());
// CURLOPT_POSTFIELDS automatically sets the Content-Type header. Tell curl to remove it by setting a blank
// value if the headers passed in don't contain a content type.
if (!m_request_headers.contains("Content-Type"sv))
curl_headers = curl_slist_append(curl_headers, "Content-Type:");
} else if (m_method == "HEAD"sv) {
set_option(CURLOPT_NOBODY, 1L);
}
for (auto const& header : m_request_headers.headers()) {
if (header.value.is_empty()) {
// curl will discard the header unless we pass the header name followed by a semicolon (i.e. we need to pass
// "Content-Type;" instead of "Content-Type: ").
//
// See: https://curl.se/libcurl/c/httpcustomheader.html
auto header_string = ByteString::formatted("{};", header.name);
curl_headers = curl_slist_append(curl_headers, header_string.characters());
} else {
auto header_string = ByteString::formatted("{}: {}", header.name, header.value);
curl_headers = curl_slist_append(curl_headers, header_string.characters());
}
}
if (curl_headers) {
set_option(CURLOPT_HTTPHEADER, curl_headers);
m_curl_string_lists.append(curl_headers);
}
if (m_start_offset_of_response_resumed_from_cache.has_value()) {
auto range = ByteString::formatted("{}-", *m_start_offset_of_response_resumed_from_cache);
set_option(CURLOPT_RANGE, range.characters());
}
// FIXME: Set up proxy if applicable
(void)m_proxy_data;
set_option(CURLOPT_HEADERFUNCTION, &on_header_received);
set_option(CURLOPT_HEADERDATA, this);
set_option(CURLOPT_WRITEFUNCTION, &on_data_received);
set_option(CURLOPT_WRITEDATA, this);
VERIFY(m_dns_result);
auto formatted_address = build_curl_resolve_list(*m_dns_result, m_url.serialized_host(), m_url.port_or_default());
if (curl_slist* resolve_list = curl_slist_append(nullptr, formatted_address.characters())) {
set_option(CURLOPT_RESOLVE, resolve_list);
m_curl_string_lists.append(resolve_list);
} else {
VERIFY_NOT_REACHED();
}
auto result = curl_multi_add_handle(m_curl_multi_handle, m_curl_easy_handle);
VERIFY(result == CURLM_OK);
#endif
}
void Request::handle_complete_state()
{
if (m_type == Type::Fetch) {
VERIFY(m_curl_result_code.has_value());
auto timing_info = acquire_timing_info();
transfer_headers_to_client_if_needed();
// HTTPS servers might terminate their connection without proper notice of shutdown - i.e. they do not send
// a "close notify" alert. OpenSSL version 3.2 began treating this as an error, which curl translates to
// CURLE_RECV_ERROR in the absence of a Content-Length response header. The Python server used by WPT is one
// such server. We ignore this error if we were actually able to download some response data.
if (m_curl_result_code == CURLE_RECV_ERROR && m_bytes_transferred_to_client != 0 && !m_response_headers.contains("Content-Length"sv))
m_curl_result_code = CURLE_OK;
if (m_curl_result_code != CURLE_OK) {
m_network_error = curl_code_to_network_error(*m_curl_result_code);
if (m_network_error == Requests::NetworkError::Unknown) {
char const* curl_error_message = curl_easy_strerror(static_cast<CURLcode>(*m_curl_result_code));
dbgln("Request::handle_complete_state: Unable to map error ({}): \"\033[31;1m{}\033[0m\"", *m_curl_result_code, curl_error_message);
}
}
m_client.async_request_finished(m_request_id, m_bytes_transferred_to_client, timing_info, m_network_error);
}
m_client.request_complete({}, m_request_id);
}
void Request::handle_error_state()
{
if (m_type == Type::Fetch) {
// FIXME: Implement timing info for failed requests.
m_client.async_request_finished(m_request_id, m_bytes_transferred_to_client, {}, m_network_error.value_or(Requests::NetworkError::Unknown));
}
m_client.request_complete({}, m_request_id);
}
size_t Request::on_header_received(void* buffer, size_t size, size_t nmemb, void* user_data)
{
auto& request = *static_cast<Request*>(user_data);
auto total_size = size * nmemb;
auto header_line = StringView { static_cast<char const*>(buffer), total_size };
// We need to extract the HTTP reason phrase since it can be a custom value. Fetching infrastructure needs this
// value for setting the status message.
if (!request.m_reason_phrase.has_value() && header_line.starts_with("HTTP/"sv)) {
auto space_index = header_line.find(' ');
if (space_index.has_value())
space_index = header_line.find(' ', *space_index + 1);
if (space_index.has_value()) {
if (auto reason_phrase = header_line.substring_view(*space_index + 1).trim_whitespace(); !reason_phrase.is_empty()) {
auto decoder = TextCodec::decoder_for_exact_name("ISO-8859-1"sv);
VERIFY(decoder.has_value());
request.m_reason_phrase = MUST(decoder->to_utf8(reason_phrase));
return total_size;
}
}
}
if (auto colon_index = header_line.find(':'); colon_index.has_value()) {
auto name = header_line.substring_view(0, *colon_index).trim_whitespace();
auto value = header_line.substring_view(*colon_index + 1).trim_whitespace();
request.m_response_headers.set(name, value);
}
return total_size;
}
size_t Request::on_data_received(void* buffer, size_t size, size_t nmemb, void* user_data)
{
auto& request = *static_cast<Request*>(user_data);
request.transfer_headers_to_client_if_needed();
auto total_size = size * nmemb;
ReadonlyBytes bytes { static_cast<u8 const*>(buffer), total_size };
auto result = [&] -> ErrorOr<void> {
TRY(request.m_response_buffer.write_some(bytes));
return request.write_queued_bytes_without_blocking();
}();
if (result.is_error()) {
dbgln("Request::on_data_received: Aborting request because error occurred whilst writing data to the client: {}", result.error());
return CURL_WRITEFUNC_ERROR;
}
return total_size;
}
void Request::transfer_headers_to_client_if_needed()
{
if (exchange(m_sent_response_headers_to_client, true))
return;
m_status_code = acquire_status_code();
m_client.async_headers_became_available(m_request_id, m_response_headers, m_status_code, m_reason_phrase);
if (m_disk_cache.has_value())
m_cache_entry_writer = m_disk_cache->create_entry(m_url, m_method, m_status_code, m_reason_phrase, m_response_headers, m_request_start_time);
}
ErrorOr<void> Request::write_queued_bytes_without_blocking()
{
auto available_bytes = m_response_buffer.used_buffer_size();
// If we've received a response to a range request that is not the partial content (206) we requested, we must
// only transfer the subset of data that WebContent now needs. We discard all received bytes up to the expected
// start of the remaining data, and then transfer the remaining bytes.
if (m_start_offset_of_response_resumed_from_cache.has_value()) {
if (m_status_code == 206) {
m_start_offset_of_response_resumed_from_cache.clear();
} else if (m_status_code == 200) {
// All bytes currently available have already been transferred. Discard them entirely.
if (m_bytes_transferred_to_client + available_bytes <= *m_start_offset_of_response_resumed_from_cache) {
m_bytes_transferred_to_client += available_bytes;
MUST(m_response_buffer.discard(available_bytes));
return {};
}
// Some bytes currently available have already been transferred. Discard those bytes and transfer the rest.
if (m_bytes_transferred_to_client + available_bytes > *m_start_offset_of_response_resumed_from_cache) {
auto bytes_to_discard = *m_start_offset_of_response_resumed_from_cache - m_bytes_transferred_to_client;
m_bytes_transferred_to_client += bytes_to_discard;
available_bytes -= bytes_to_discard;
MUST(m_response_buffer.discard(bytes_to_discard));
}
m_start_offset_of_response_resumed_from_cache.clear();
} else {
return Error::from_string_literal("Unacceptable status code for resumed HTTP request");
}
}
Vector<u8> bytes_to_send;
bytes_to_send.resize(available_bytes);
m_response_buffer.peek_some(bytes_to_send);
auto result = Core::System::write(m_client_writer_fd, bytes_to_send);
if (result.is_error()) {
if (result.error().code() != EAGAIN)
return result.release_error();
m_client_writer_notifier->set_enabled(true);
return {};
}
if (m_cache_entry_writer.has_value()) {
auto bytes_sent = bytes_to_send.span().slice(0, result.value());
if (m_cache_entry_writer->write_data(bytes_sent).is_error())
m_cache_entry_writer.clear();
}
m_bytes_transferred_to_client += result.value();
MUST(m_response_buffer.discard(result.value()));
m_client_writer_notifier->set_enabled(!m_response_buffer.is_eof());
if (m_response_buffer.is_eof() && m_curl_result_code.has_value())
transition_to_state(State::Complete);
return {};
}
u32 Request::acquire_status_code() const
{
long http_status_code = 0;
auto result = curl_easy_getinfo(m_curl_easy_handle, CURLINFO_RESPONSE_CODE, &http_status_code);
VERIFY(result == CURLE_OK);
return static_cast<u32>(http_status_code);
}
Requests::RequestTimingInfo Request::acquire_timing_info() const
{
// curl_easy_perform()
// |
// |--QUEUE
// |--|--NAMELOOKUP
// |--|--|--CONNECT
// |--|--|--|--APPCONNECT
// |--|--|--|--|--PRETRANSFER
// |--|--|--|--|--|--POSTTRANSFER
// |--|--|--|--|--|--|--STARTTRANSFER
// |--|--|--|--|--|--|--|--TOTAL
// |--|--|--|--|--|--|--|--REDIRECT
// FIXME: Implement timing info for cache hits.
if (m_cache_entry_reader.has_value())
return {};
auto get_timing_info = [&](auto option) {
curl_off_t time_value = 0;
auto result = curl_easy_getinfo(m_curl_easy_handle, option, &time_value);
VERIFY(result == CURLE_OK);
return time_value;
};
auto queue_time = get_timing_info(CURLINFO_QUEUE_TIME_T);
auto domain_lookup_time = get_timing_info(CURLINFO_NAMELOOKUP_TIME_T);
auto connect_time = get_timing_info(CURLINFO_CONNECT_TIME_T);
auto secure_connect_time = get_timing_info(CURLINFO_APPCONNECT_TIME_T);
auto request_start_time = get_timing_info(CURLINFO_PRETRANSFER_TIME_T);
auto response_start_time = get_timing_info(CURLINFO_STARTTRANSFER_TIME_T);
auto response_end_time = get_timing_info(CURLINFO_TOTAL_TIME_T);
auto encoded_body_size = get_timing_info(CURLINFO_SIZE_DOWNLOAD_T);
long http_version = 0;
auto get_version_result = curl_easy_getinfo(m_curl_easy_handle, CURLINFO_HTTP_VERSION, &http_version);
VERIFY(get_version_result == CURLE_OK);
auto http_version_alpn = Requests::ALPNHttpVersion::None;
switch (http_version) {
case CURL_HTTP_VERSION_1_0:
http_version_alpn = Requests::ALPNHttpVersion::Http1_0;
break;
case CURL_HTTP_VERSION_1_1:
http_version_alpn = Requests::ALPNHttpVersion::Http1_1;
break;
case CURL_HTTP_VERSION_2_0:
http_version_alpn = Requests::ALPNHttpVersion::Http2_TLS;
break;
case CURL_HTTP_VERSION_3:
http_version_alpn = Requests::ALPNHttpVersion::Http3;
break;
default:
http_version_alpn = Requests::ALPNHttpVersion::None;
break;
}
return Requests::RequestTimingInfo {
.domain_lookup_start_microseconds = queue_time,
.domain_lookup_end_microseconds = queue_time + domain_lookup_time,
.connect_start_microseconds = queue_time + domain_lookup_time,
.connect_end_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time,
.secure_connect_start_microseconds = queue_time + domain_lookup_time + connect_time,
.request_start_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + request_start_time,
.response_start_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + response_start_time,
.response_end_microseconds = queue_time + domain_lookup_time + connect_time + secure_connect_time + response_end_time,
.encoded_body_size = encoded_body_size,
.http_version_alpn_identifier = http_version_alpn,
};
}
}

157
Services/RequestServer/Request.h Normal file
View File

@ -0,0 +1,157 @@
/*
* Copyright (c) 2025, Tim Flynn <trflynn89@ladybird.org>
*
* SPDX-License-Identifier: BSD-2-Clause
*/
#pragma once
#include <AK/ByteBuffer.h>
#include <AK/ByteString.h>
#include <AK/MemoryStream.h>
#include <AK/Optional.h>
#include <AK/Time.h>
#include <AK/Weakable.h>
#include <LibCore/Proxy.h>
#include <LibDNS/Resolver.h>
#include <LibHTTP/HeaderMap.h>
#include <LibRequests/NetworkError.h>
#include <LibRequests/RequestTimingInfo.h>
#include <LibURL/URL.h>
#include <RequestServer/CacheLevel.h>
#include <RequestServer/Forward.h>
struct curl_slist;
namespace RequestServer {
class Request : public Weakable<Request> {
public:
static NonnullOwnPtr<Request> fetch(
i32 request_id,
Optional<DiskCache&> disk_cache,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
ByteString method,
HTTP::HeaderMap request_headers,
ByteBuffer request_body,
ByteString alt_svc_cache_path,
Core::ProxyData proxy_data);
static NonnullOwnPtr<Request> connect(
i32 request_id,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
CacheLevel cache_level);
~Request();
void notify_fetch_complete(Badge<ConnectionFromClient>, int result_code);
private:
enum class Type : u8 {
Fetch,
Connect,
};
enum class State : u8 {
Init, // Decide whether to service this request from cache or the network.
ReadCache, // Read the cached response from disk.
DNSLookup, // Resolve the URL's host.
Connect, // Issue a network request to connect to the URL.
Fetch, // Issue a network request to fetch the URL.
Complete, // Finalize the request with the client.
Error, // Any error occured during the request's lifetime.
};
Request(
i32 request_id,
Optional<DiskCache&> disk_cache,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url,
ByteString method,
HTTP::HeaderMap request_headers,
ByteBuffer request_body,
ByteString alt_svc_cache_path,
Core::ProxyData proxy_data);
Request(
i32 request_id,
ConnectionFromClient& client,
void* curl_multi,
Resolver& resolver,
URL::URL url);
void transition_to_state(State);
void process();
void handle_initial_state();
void handle_read_cache_state();
void handle_dns_lookup_state();
void handle_connect_state();
void handle_fetch_state();
void handle_complete_state();
void handle_error_state();
static size_t on_header_received(void* buffer, size_t size, size_t nmemb, void* user_data);
static size_t on_data_received(void* buffer, size_t size, size_t nmemb, void* user_data);
void transfer_headers_to_client_if_needed();
ErrorOr<void> write_queued_bytes_without_blocking();
u32 acquire_status_code() const;
Requests::RequestTimingInfo acquire_timing_info() const;
ConnectionFromClient& client();
i32 m_request_id { 0 };
Type m_type { Type::Fetch };
State m_state { State::Init };
Optional<DiskCache&> m_disk_cache;
ConnectionFromClient& m_client;
void* m_curl_multi_handle { nullptr };
void* m_curl_easy_handle { nullptr };
Vector<curl_slist*> m_curl_string_lists;
Optional<int> m_curl_result_code;
NonnullRefPtr<Resolver> m_resolver;
RefPtr<DNS::LookupResult const> m_dns_result;
URL::URL m_url;
ByteString m_method;
UnixDateTime m_request_start_time { UnixDateTime::now() };
HTTP::HeaderMap m_request_headers;
ByteBuffer m_request_body;
ByteString m_alt_svc_cache_path;
Core::ProxyData m_proxy_data;
u32 m_status_code { 0 };
Optional<String> m_reason_phrase;
HTTP::HeaderMap m_response_headers;
bool m_sent_response_headers_to_client { false };
AllocatingMemoryStream m_response_buffer;
RefPtr<Core::Notifier> m_client_writer_notifier;
int m_client_writer_fd { -1 };
Optional<size_t> m_start_offset_of_response_resumed_from_cache;
size_t m_bytes_transferred_to_client { 0 };
Optional<CacheEntryReader&> m_cache_entry_reader;
Optional<CacheEntryWriter&> m_cache_entry_writer;
Optional<Requests::NetworkError> m_network_error;
};
}