use core::ptr::NonNull;
use core::sync::atomic::{AtomicU32, AtomicUsize, Ordering};
use bun_ast::{Loc, Log};
use bun_core::FeatureFlags;
use bun_core::{MutableString, ZigStringSlice};
use bun_threading::IntrusiveWorkTask as _;
use bun_threading::thread_pool::{self, Batch, Task};
use bun_url::{PercentEncoding, URL};
use bun_dotenv::Loader as DotEnvLoader;
use bun_http_types::Encoding::Encoding;
use bun_picohttp as picohttp;
use crate::headers::{self, Headers};
use crate::{
FetchRedirect, Flags, HTTPClient, HTTPRequestBody, HTTPVerboseLevel, InternalState, Method,
Signals, ThreadlocalAsyncHTTP,
};
use crate::{HTTPClientResult, HTTPClientResultCallback};
use crate::ssl_config::SharedPtr as SSLConfigSharedPtr;
bun_core::declare_scope!(AsyncHTTP, visible);
pub struct AsyncHTTP<'a> {
pub request: Option<picohttp::Request<'static>>,
pub response: Option<picohttp::Response<'static>>,
pub request_headers: headers::EntryList,
pub response_headers: headers::EntryList,
pub response_buffer: *mut MutableString,
pub request_body: HTTPRequestBody<'a>,
pub request_header_buf: &'a [u8],
pub method: Method,
pub url: URL<'a>,
pub http_proxy: Option<URL<'a>>,
pub real: Option<NonNull<AsyncHTTP<'a>>>,
pub next: bun_threading::Link<AsyncHTTP<'static>>,
pub task: thread_pool::Task,
pub result_callback: HTTPClientResultCallback,
pub redirected: bool,
pub response_encoding: Encoding,
pub verbose: HTTPVerboseLevel,
pub client: HTTPClient<'a>,
pub waiting_deffered: bool,
pub finalized: bool,
pub err: Option<bun_core::Error>,
pub async_http_id: u32,
pub state: AtomicState,
pub elapsed: u64,
pub gzip_elapsed: u64,
pub signals: Signals,
}
bun_threading::intrusive_work_task!(['a] AsyncHTTP<'a>, task);
unsafe impl bun_threading::Linked for AsyncHTTP<'static> {
#[inline]
unsafe fn link(item: *mut Self) -> *const bun_threading::Link<Self> {
unsafe { core::ptr::addr_of!((*item).next) }
}
}
pub static ACTIVE_REQUESTS_COUNT: AtomicUsize = AtomicUsize::new(0);
pub static MAX_SIMULTANEOUS_REQUESTS: AtomicUsize = AtomicUsize::new(256);
fn noop_result_callback(_: *mut (), _: *mut AsyncHTTP<'static>, _: HTTPClientResult<'_>) {}
#[inline(always)]
const fn noop_callback() -> HTTPClientResultCallback {
HTTPClientResultCallback {
ctx: core::ptr::null_mut(),
function: noop_result_callback,
release_at_shutdown: None,
}
}
#[inline]
unsafe fn free_owned_href(href: &'static [u8]) {
if !href.is_empty() {
unsafe { bun_core::heap::destroy(core::ptr::from_ref(href).cast_mut()) };
}
}
#[inline]
fn http_thread_timer_read() -> u64 {
crate::http_thread().timer.elapsed().as_nanos() as u64
}
fn build_proxy_authorization(proxy: &URL<'_>) -> Option<Vec<u8>> {
if proxy.username.is_empty() {
return None;
}
let username = match PercentEncoding::decode_alloc(proxy.username) {
Ok(u) => u,
Err(err) => {
bun_core::scoped_log!(AsyncHTTP, "failed to decode proxy username: {:?}", err);
return None;
}
};
let auth: Vec<u8> = if !proxy.password.is_empty() {
let password = match PercentEncoding::decode_alloc(proxy.password) {
Ok(p) => p,
Err(err) => {
bun_core::scoped_log!(AsyncHTTP, "failed to decode proxy password: {:?}", err);
return None;
}
};
let mut auth: Vec<u8> = Vec::with_capacity(username.len() + 1 + password.len());
auth.extend_from_slice(&username);
auth.push(b':');
auth.extend_from_slice(&password);
auth
} else {
username.into_vec()
};
let size = bun_base64::encode_len_from_size(auth.len());
let mut buf = vec![0u8; size + b"Basic ".len()];
let encoded_len = bun_base64::encode_url_safe(&mut buf[b"Basic ".len()..], &auth);
buf[..b"Basic ".len()].copy_from_slice(b"Basic ");
buf.truncate(b"Basic ".len() + encoded_len);
Some(buf)
}
fn make_client<'a>(
method: Method,
url: URL<'a>,
header_entries: headers::EntryList,
header_buf: &'a [u8],
hostname: Option<&'a [u8]>,
signals: Signals,
async_http_id: u32,
http_proxy: Option<URL<'a>>,
proxy_headers: Option<Headers>,
redirect_type: FetchRedirect,
) -> HTTPClient<'a> {
HTTPClient {
method,
extension_method: None,
header_entries,
header_buf,
url,
connected_url: URL::default(),
verbose: HTTPVerboseLevel::None,
remaining_redirect_count: 127,
allow_retry: false,
h2_retries: 0,
redirect_type,
redirect: Vec::new(),
prev_redirect: Vec::new(),
progress_node: None,
flags: Flags::default(),
state: InternalState::default(),
tls_props: None,
custom_ssl_ctx: None,
result_callback: noop_callback(),
if_modified_since: b"",
request_content_len_buf: [0u8; b"-4294967295".len()],
http_proxy,
proxy_headers,
proxy_authorization: None,
proxy_tunnel: None,
h2: None,
h3: None,
pending_h2: None,
signals,
async_http_id,
hostname,
unix_socket_path: ZigStringSlice::EMPTY,
tls_info: None,
}
}
pub fn load_env(logger: &mut Log, env: &DotEnvLoader) {
if let Some(max_http_requests) = env.get(b"BUN_CONFIG_MAX_HTTP_REQUESTS") {
let max: u16 = match bun_core::parse_int::<u16>(max_http_requests, 10) {
Ok(v) => v,
Err(_) => {
logger
.add_error_fmt(
None,
Loc::EMPTY,
format_args!(
"BUN_CONFIG_MAX_HTTP_REQUESTS value \"{}\" is not a valid integer between 1 and 65535",
bstr::BStr::new(max_http_requests),
),
);
return;
}
};
if max == 0 {
logger.add_warning_fmt(
None,
Loc::EMPTY,
format_args!(
"BUN_CONFIG_MAX_HTTP_REQUESTS value must be a number between 1 and 65535"
),
);
return;
}
MAX_SIMULTANEOUS_REQUESTS.store(usize::from(max), Ordering::Relaxed);
}
}
#[derive(Default)]
pub struct Options<'a> {
pub http_proxy: Option<URL<'a>>,
pub proxy_headers: Option<Headers>,
pub hostname: Option<&'a [u8]>,
pub signals: Option<Signals>,
pub unix_socket_path: Option<ZigStringSlice>,
pub disable_timeout: Option<bool>,
pub verbose: Option<HTTPVerboseLevel>,
pub disable_keepalive: Option<bool>,
pub disable_decompression: Option<bool>,
pub max_redirects: Option<u8>,
pub reject_unauthorized: Option<bool>,
pub tls_props: Option<SSLConfigSharedPtr>,
pub is_page_egress: Option<bool>,
pub omit_connection_header: Option<bool>,
pub extension_method: Option<&'static [u8]>,
}
impl<'a> AsyncHTTP<'a> {
#[inline(always)]
pub fn as_erased_ptr(&self) -> *mut AsyncHTTP<'static> {
std::ptr::from_ref::<Self>(self)
.cast_mut()
.cast::<AsyncHTTP<'static>>()
}
#[inline]
pub fn max_simultaneous_requests() -> &'static core::sync::atomic::AtomicUsize {
&MAX_SIMULTANEOUS_REQUESTS
}
pub fn signal_header_progress(&mut self) {
self.signals.store(
crate::signals::Field::HeaderProgress,
true,
Ordering::Release,
);
}
pub fn enable_response_body_streaming(&mut self) {
self.signals.store(
crate::signals::Field::ResponseBodyStreaming,
true,
Ordering::Release,
);
}
pub fn sync_progress_from(&mut self, src: &AsyncHTTP<'a>) {
self.url = src.url.clone();
self.redirected = src.redirected;
self.elapsed = src.elapsed;
self.gzip_elapsed = src.gzip_elapsed;
self.err = src.err;
self.response = src.response;
self.response_encoding = src.response_encoding;
self.response_buffer = src.response_buffer;
self.state
.store(src.state.load(Ordering::Relaxed), Ordering::Relaxed);
self.client.url = src.client.url.clone();
self.client.flags = src.client.flags;
self.client.remaining_redirect_count = src.client.remaining_redirect_count;
}
pub fn clear_data(&mut self) {
self.response_headers = headers::EntryList::default();
self.request = None;
self.response = None;
self.client.unix_socket_path = ZigStringSlice::EMPTY;
}
}
struct Preconnect {
async_http: Option<AsyncHTTP<'static>>,
response_buffer: MutableString,
url: URL<'static>,
is_url_owned: bool,
}
impl Preconnect {
fn on_result(this: *mut Preconnect, _: *mut AsyncHTTP<'static>, _: HTTPClientResult<'_>) {
unsafe {
(*this).response_buffer = MutableString::default();
(*this)
.async_http
.as_mut()
.expect("Preconnect.async_http set in preconnect()")
.clear_data();
if (*this).is_url_owned {
free_owned_href((*this).url.href);
}
drop(bun_core::heap::take(this));
}
}
}
pub fn preconnect(url: URL<'static>, is_url_owned: bool) {
if !FeatureFlags::IS_FETCH_PRECONNECT_SUPPORTED {
if is_url_owned {
unsafe { free_owned_href(url.href) };
}
return;
}
crate::http_thread::init(&Default::default());
let this: *mut Preconnect = bun_core::heap::into_raw(Box::new(Preconnect {
async_http: None,
response_buffer: MutableString::default(),
url,
is_url_owned,
}));
unsafe {
let response_buffer: *mut MutableString = core::ptr::addr_of_mut!((*this).response_buffer);
let url = (*this).url.clone();
let async_http = (*this).async_http.insert(AsyncHTTP::init(
Method::GET,
url,
headers::EntryList::default(),
b"",
response_buffer,
b"",
HTTPClientResultCallback::new::<Preconnect>(this, Preconnect::on_result),
FetchRedirect::Manual,
Options::default(),
));
async_http.client.flags.is_preconnect_only = true;
crate::HTTPThread::schedule(Batch::from(core::ptr::addr_of_mut!(async_http.task)));
}
}
impl<'a> AsyncHTTP<'a> {
pub fn init(
method: Method,
url: URL<'a>,
headers: headers::EntryList,
headers_buf: &'a [u8],
response_buffer: *mut MutableString,
request_body: &'a [u8],
callback: HTTPClientResultCallback,
redirect_type: FetchRedirect,
options: Options<'a>,
) -> AsyncHTTP<'a> {
let async_http_id = if options
.signals
.as_ref()
.map(|s| s.aborted.is_some())
.unwrap_or(false)
{
crate::ASYNC_HTTP_ID_MONOTONIC.fetch_add(1, Ordering::Relaxed)
} else {
0
};
let signals = options.signals.unwrap_or_default();
let http_proxy = options.http_proxy.clone();
let client = make_client(
method,
url.clone(),
headers.clone().expect("OOM"),
headers_buf,
options.hostname,
signals,
async_http_id,
http_proxy.clone(),
options.proxy_headers,
redirect_type,
);
let mut this = AsyncHTTP {
request: None,
response: None,
request_headers: headers,
response_headers: headers::EntryList::default(),
response_buffer,
request_body: HTTPRequestBody::Bytes(request_body),
request_header_buf: headers_buf,
method,
url,
http_proxy,
real: None,
next: bun_threading::Link::new(),
task: thread_pool::Task {
node: thread_pool::Node::default(),
callback: start_async_http,
},
result_callback: callback,
redirected: false,
response_encoding: Encoding::Identity,
verbose: HTTPVerboseLevel::None,
client,
waiting_deffered: false,
finalized: false,
err: None,
async_http_id,
state: AtomicState::new(State::Pending),
elapsed: 0,
gzip_elapsed: 0,
signals,
};
if let Some(val) = options.unix_socket_path {
debug_assert!(this.client.unix_socket_path.slice().is_empty());
this.client.unix_socket_path = val;
}
if let Some(val) = options.disable_timeout {
this.client.flags.disable_timeout = val;
}
if let Some(val) = options.verbose {
this.client.verbose = val;
}
if let Some(val) = options.disable_decompression {
this.client.flags.disable_decompression = val;
}
if let Some(val) = options.max_redirects {
this.client.remaining_redirect_count = (val.min(126) + 1) as i8;
}
if let Some(val) = options.disable_keepalive {
this.client.flags.disable_keepalive = val;
}
if let Some(val) = options.reject_unauthorized {
this.client.flags.reject_unauthorized = val;
}
if let Some(val) = options.is_page_egress {
this.client.flags.is_page_egress = val;
}
if let Some(val) = options.omit_connection_header {
this.client.flags.omit_connection_header = val;
}
if let Some(val) = options.extension_method {
this.client.method = Method::EXTENSION;
this.client.extension_method = Some(val);
}
if let Some(val) = options.tls_props {
this.client.tls_props = Some(val);
}
if let Some(proxy) = &this.http_proxy {
if let Some(auth) = build_proxy_authorization(proxy) {
this.client.proxy_authorization = Some(auth);
}
}
this
}
pub fn init_sync(
method: Method,
url: URL<'a>,
headers: headers::EntryList,
headers_buf: &'a [u8],
response_buffer: *mut MutableString,
request_body: &'a [u8],
http_proxy: Option<URL<'a>>,
hostname: Option<&'a [u8]>,
redirect_type: FetchRedirect,
) -> AsyncHTTP<'a> {
Self::init(
method,
url,
headers,
headers_buf,
response_buffer,
request_body,
noop_callback(),
redirect_type,
Options {
http_proxy,
hostname,
..Options::default()
},
)
}
pub fn schedule(&mut self, batch: &mut Batch) {
self.state.store(State::Scheduled, Ordering::Relaxed);
batch.push(Batch::from(core::ptr::addr_of_mut!(self.task)));
}
}
pub struct SingleHTTPChannel {
slot: bun_threading::Guarded<Option<HTTPClientResult<'static>>>,
cv: bun_threading::Condvar,
}
impl SingleHTTPChannel {
pub fn init() -> SingleHTTPChannel {
SingleHTTPChannel {
slot: bun_threading::Guarded::new(None),
cv: bun_threading::Condvar::new(),
}
}
pub fn reset(&mut self) {
*self.slot.lock() = None;
}
fn write_item(&self, item: HTTPClientResult<'static>) {
let mut g = self.slot.lock();
*g = Some(item);
self.cv.notify_one();
}
fn read_item(&self) -> HTTPClientResult<'static> {
let mut g = self.slot.lock();
loop {
if let Some(item) = g.take() {
return item;
}
self.cv.wait_guarded(&mut g);
}
}
}
fn send_sync_callback(
this: *mut SingleHTTPChannel,
async_http: *mut AsyncHTTP<'static>,
result: HTTPClientResult<'_>,
) {
let async_http = unsafe { &mut *async_http };
if let Some(mut real) = async_http.real {
let real = unsafe { real.as_mut() };
real.response = async_http.response;
real.request = async_http.request.take();
real.response_headers = core::mem::take(&mut async_http.response_headers);
real.response_encoding = async_http.response_encoding;
real.err = async_http.err;
real.redirected = async_http.redirected;
real.elapsed = async_http.elapsed;
real.gzip_elapsed = async_http.gzip_elapsed;
real.state
.store(async_http.state.load(Ordering::Relaxed), Ordering::Relaxed);
real.response_buffer = async_http.response_buffer;
}
unsafe {
(*this).write_item(result.detach_lifetime());
}
}
impl<'a> AsyncHTTP<'a> {
pub fn send_sync(&mut self) -> Result<picohttp::Response<'static>, bun_core::Error> {
crate::http_thread::init(&Default::default());
let ctx = bun_core::heap::into_raw_nn(Box::new(SingleHTTPChannel::init()));
self.result_callback =
HTTPClientResultCallback::new::<SingleHTTPChannel>(ctx.as_ptr(), send_sync_callback);
let mut batch = Batch::default();
self.schedule(&mut batch);
crate::HTTPThread::schedule(batch);
let result = bun_ptr::ParentRef::from(ctx).read_item();
drop(unsafe { bun_core::heap::take(ctx.as_ptr()) });
if let Some(err) = result.fail {
return Err(err);
}
debug_assert!(result.metadata.is_some());
let metadata = core::mem::ManuallyDrop::new(result.metadata.unwrap());
Ok(metadata.response)
}
fn on_async_http_callback_raw(
this: *mut AsyncHTTP<'static>,
async_http: *mut AsyncHTTP<'static>,
result: HTTPClientResult<'_>,
) {
unsafe {
debug_assert!((*this).real.is_some());
let callback = (*this).result_callback;
(*this).elapsed = http_thread_timer_read().saturating_sub((*this).elapsed);
(*this).redirected = (*this).client.flags.redirected;
if result.is_success() {
(*this).err = None;
if let Some(metadata) = &result.metadata {
(*this).response = Some(metadata.response);
}
(*this).state.store(State::Success, Ordering::Relaxed);
} else {
(*this).err = result.fail;
(*this).response = None;
(*this).state.store(State::Fail, Ordering::Relaxed);
}
let has_more = result.has_more;
if has_more {
callback.run(async_http, result);
} else {
{
let client = &mut (*this).client;
drop(core::mem::take(&mut client.redirect));
drop(core::mem::take(&mut client.prev_redirect));
if let Some(tunnel) = client.proxy_tunnel.take() {
(*tunnel.as_ptr()).detach_socket();
tunnel.deref();
}
debug_assert!(client.h2.is_none());
if let Some(ctx) = client.custom_ssl_ctx.take() {
ctx.deref();
}
drop(core::mem::take(&mut client.state));
}
let elapsed = (*this).elapsed;
bun_core::scoped_log!(AsyncHTTP, "onAsyncHTTPCallback: {:?}", elapsed);
callback.run(async_http, result);
let threadlocal_http: *mut ThreadlocalAsyncHTTP =
bun_core::from_field_ptr!(ThreadlocalAsyncHTTP, async_http, async_http);
{
let in_flight = &mut crate::http_thread().in_flight;
if let Some(i) = in_flight
.iter()
.position(|n| n.as_ptr() == threadlocal_http)
{
in_flight.swap_remove(i);
}
}
std::alloc::dealloc(
threadlocal_http.cast::<u8>(),
std::alloc::Layout::new::<ThreadlocalAsyncHTTP>(),
);
let active_requests = ACTIVE_REQUESTS_COUNT.fetch_sub(1, Ordering::Relaxed);
debug_assert!(active_requests > 0);
}
}
let thread = crate::http_thread();
if (!thread.queued_tasks.is_empty() || !thread.deferred_tasks.is_empty())
&& ACTIVE_REQUESTS_COUNT.load(Ordering::Relaxed)
< MAX_SIMULTANEOUS_REQUESTS.load(Ordering::Relaxed)
{
thread.wakeup();
}
}
pub fn on_async_http_callback(
&mut self,
async_http: *mut AsyncHTTP<'static>,
result: HTTPClientResult<'_>,
) {
Self::on_async_http_callback_raw(self.as_erased_ptr(), async_http, result);
}
}
pub unsafe fn start_async_http(task: *mut Task) {
let this = unsafe { &mut *AsyncHTTP::<'static>::from_task_ptr(task) };
this.on_start();
}
impl<'a> AsyncHTTP<'a> {
pub fn on_start(&mut self) {
let _ = ACTIVE_REQUESTS_COUNT.fetch_add(1, Ordering::Relaxed);
self.err = None;
self.state.store(State::Sending, Ordering::Relaxed);
self.client.result_callback = HTTPClientResultCallback::new::<AsyncHTTP<'static>>(
self.as_erased_ptr(),
AsyncHTTP::on_async_http_callback_raw,
);
self.elapsed = http_thread_timer_read();
let response_buffer = crate::body_out::as_mut(
NonNull::new(self.response_buffer).expect("response_buffer set in init"),
);
let body = core::mem::replace(&mut self.request_body, HTTPRequestBody::Bytes(b""));
self.client.start(body, response_buffer);
}
}
pub type HTTPCallbackPair = (*mut AsyncHTTP<'static>, HTTPClientResult<'static>);
pub type HTTPChannel = bun_threading::Channel<
*mut HTTPCallbackPair,
bun_collections::linear_fifo::StaticBuffer<*mut HTTPCallbackPair, 1000>,
>;
pub struct HTTPChannelContext<'a> {
pub http: AsyncHTTP<'a>,
pub channel: Option<bun_ptr::BackRef<HTTPChannel>>,
}
impl HTTPChannelContext<'_> {
pub fn callback(data: HTTPCallbackPair) {
let this: &mut HTTPChannelContext =
unsafe { &mut *(bun_core::from_field_ptr!(HTTPChannelContext, http, data.0)) };
let boxed = bun_core::heap::into_raw(Box::new(data));
this.channel
.expect("HTTPChannelContext.channel set before scheduling")
.write_item(boxed)
.expect("HTTPChannel full");
}
}
#[repr(u32)]
#[derive(Copy, Clone, PartialEq, Eq)]
pub enum State {
Pending = 0,
Scheduled = 1,
Sending = 2,
Success = 3,
Fail = 4,
}
#[repr(transparent)]
pub struct AtomicState(AtomicU32);
impl AtomicState {
pub const fn new(s: State) -> Self {
Self(AtomicU32::new(s as u32))
}
pub fn store(&self, s: State, order: Ordering) {
self.0.store(s as u32, order);
}
pub fn load(&self, order: Ordering) -> State {
match self.0.load(order) {
0 => State::Pending,
1 => State::Scheduled,
2 => State::Sending,
3 => State::Success,
4 => State::Fail,
_ => unreachable!("invalid AsyncHTTP::State discriminant"),
}
}
}