From 0afd6ecefb965376f866958dcbb6aeca7b506867 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Wed, 11 Nov 2026 12:29:36 +0000 Subject: [PATCH 1/4] Extract the shared parts of streaming sends Move the streaming client, the conversion of a response head to and from the host ABI, the timeout arguments and the redirect decision into helpers of their own, so that another way of sending a request can reuse them. No behaviour changes. Co-Authored-By: Claude Opus 5.5 --- platform/Http.roc | 51 +++++++++++----- src/http.rs | 146 +++++++++++++++++++++++++++------------------- 2 files changed, 120 insertions(+), 77 deletions(-) diff --git a/platform/Http.roc b/platform/Http.roc index 9caa9ea7..0bc786c3 100644 --- a/platform/Http.roc +++ b/platform/Http.roc @@ -171,11 +171,7 @@ Http :: [].{ url = Url.parse(Request.uri(request)) ? InvalidUrl canonical = Url.without_fragment(url) canonical_request = request_with_defaults(request.with_uri(canonical.to_str())) - deadline = match Request.timeout(request) { - NoTimeout | TimeoutMilliseconds(0) => NoTimeout - TimeoutMilliseconds(ms) => At(Host.monotonic_now!().plus_saturated(ms.times_saturated(1_000_000))) - } - send_redirected!(canonical_request, canonical, options, options.max_redirects, deadline) + send_redirected!(canonical_request, canonical, options, options.max_redirects, request_deadline!(request)) } ## The status and headers of a response, and its body still to be read. @@ -310,31 +306,55 @@ transport_error = |error| Other(message) => HttpErr(Other(message)) } +## The request timeout as an instant on the `monotonic_now!` clock. It is +## captured once, so that it covers every hop and the final body. +request_deadline! : Request => [NoTimeout, At(U64)] +request_deadline! = |request| + match Request.timeout(request) { + NoTimeout | TimeoutMilliseconds(0) => NoTimeout + TimeoutMilliseconds(ms) => At(Host.monotonic_now!().plus_saturated(ms.times_saturated(1_000_000))) + } + +## The idle timeout as the host takes it, in nanoseconds. +host_idle_timeout : [NoTimeout, After(Duration)] -> [NoTimeout, After(U64)] +host_idle_timeout = |idle_timeout| + match idle_timeout { + NoTimeout => NoTimeout + After(duration) => After(duration.to_nanos()) + } + ## The same driver serves buffered and streamed responses. The deadline is ## captured once and is passed unchanged to every native hop and final body. send_redirected! : Request, Url.Url, Http.StreamingOptions, U64, [NoTimeout, At(U64)] => Try(Http.StreamingResponse, [Interrupted, HttpErr(Http.TransportErr)]) send_redirected! = |request, url, options, remaining, deadline| { - idle = match options.idle_timeout { - NoTimeout => NoTimeout - After(duration) => After(duration.to_nanos()) - } hop = request.with_headers(sanitize_headers(Request.headers(request), False, False)) - raw = match Host.http_send_streaming!(InternalHttp.to_host_request(hop), idle, deadline) { + raw = match Host.http_send_streaming!(InternalHttp.to_host_request(hop), host_idle_timeout(options.idle_timeout), deadline) { Ok(value) => value Err(error) => { return Err(transport_error(error)) } } + match redirect_step!(raw, request, url, options.max_redirects, remaining)? { + Final(response) => Ok(response) + Follow(next) => send_redirected!(next.request, next.url, options, remaining - 1, deadline) + } +} + +## What the response to one hop means for a request that may follow +## redirects: the final response, or the next hop to send. The body of a +## redirect is closed. +redirect_step! : InternalHttp.StreamingResponseFromHost, Request, Url.Url, U64, U64 => Try([Final(Http.StreamingResponse), Follow({ request : Request, url : Url.Url, status : U16, location : Str })], [HttpErr(Http.TransportErr)]) +redirect_step! = |raw, request, url, max_redirects, remaining| { response = Http.StreamingResponse.{ status: raw.status, headers: InternalHttp.from_host_headers(raw.headers), content_length: raw.content_length, host: raw.body } status = response.status redirect = status == 301 or status == 302 or status == 303 or status == 307 or status == 308 - if options.max_redirects == 0 or !redirect { - return Ok(response) + if max_redirects == 0 or !redirect { + return Ok(Final(response)) } locations = response.headers.keep_if(|header| header.name.with_ascii_lowercased() == "location") location = match locations { [] => { - return Ok(response) + return Ok(Final(response)) } [header] => header.value _ => { @@ -344,7 +364,7 @@ send_redirected! = |request, url, options, remaining, deadline| { } Host.http_close_body!(response.host) if remaining == 0 { - return Err(HttpErr(RedirectLimitExceeded(options.max_redirects))) + return Err(HttpErr(RedirectLimitExceeded(max_redirects))) } match raw.invalid_location { Some(byte) => { @@ -356,8 +376,7 @@ send_redirected! = |request, url, options, remaining, deadline| { if Url.scheme(url) == Https and Url.scheme(next_url) == Http { return Err(HttpErr(InsecureRedirect)) } - next = redirected_request(request, url, next_url, status) - send_redirected!(next, next_url, options, remaining - 1, deadline) + Ok(Follow({ request: redirected_request(request, url, next_url, status), url: next_url, status, location })) } ## HTTP permits network-path references; Url.resolve deliberately does not. diff --git a/src/http.rs b/src/http.rs index 3e783b1a..66844b53 100644 --- a/src/http.rs +++ b/src/http.rs @@ -589,6 +589,67 @@ struct StreamingHead { content_length: Option, } +impl StreamingHead { + fn from_parts(parts: &hyper::http::response::Parts) -> Self { + Self { + status: parts.status.as_u16(), + headers: header_pairs(&parts.headers), + invalid_location: invalid_location(&parts.headers), + content_length: declared_content_length(&parts.headers), + } + } + + /// The head as Roc receives it, with `body` the handle of its body. + fn to_roc(&self, body: *mut u64, roc_host: &RocHost) -> HostHttpSendStreamingOk { + let content_length = match self.content_length { + Some(length) => KnownOrUnknown { + payload: KnownOrUnknownPayload { + known: ManuallyDrop::new(length), + }, + tag: KnownOrUnknownTag::Known, + }, + None => KnownOrUnknown { + payload: KnownOrUnknownPayload { unknown: [] }, + tag: KnownOrUnknownTag::Unknown, + }, + }; + let invalid_location = match self.invalid_location { + Some(byte) => NoneOrSome { + payload: NoneOrSomePayload { + some: ManuallyDrop::new(byte), + }, + tag: NoneOrSomeTag::Some, + }, + None => NoneOrSome { + payload: NoneOrSomePayload { none: [] }, + tag: NoneOrSomeTag::None, + }, + }; + HostHttpSendStreamingOk { + content_length, + invalid_location, + body, + headers: build_roc_headers(&self.headers, roc_host), + status: self.status, + } + } +} + +/// The client of every streaming request. It is the same as +/// `async_send_request`'s, so all entry points put the same request on the +/// wire. Without a pool the connection is not kept for a later request: it +/// closes when its response is finished with. +fn streaming_client() -> hyper_util::client::legacy::Client< + hyper_rustls::HttpsConnector< + hyper_util::client::legacy::connect::HttpConnector, + >, + http_body_util::Full, +> { + hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .pool_max_idle_per_host(0) + .build(https_connector()) +} + /// Send `request` and wait for the status and headers. /// /// `timeout_ms` is the `Request` timeout, the deadline for the whole exchange, @@ -612,8 +673,6 @@ fn send_streaming_at( idle_timeout: Option, deadline: Option, ) -> Result<(StreamingHead, StreamingBody), Failure> { - use hyper_util::client::legacy::Client; - use hyper_util::rt::TokioExecutor; if crate::signals::is_pending() { return Err(Failure::Interrupted); } @@ -624,17 +683,8 @@ fn send_streaming_at( let budget = wait_budget(idle_timeout, deadline, now).map_err(|()| Failure::Timeout)?; let until = budget.and_then(|budget| now.checked_add(budget)); - let response = TOKIO_RUNTIME.block_on(async { - // The same client as `async_send_request`, so both entry points put - // the same request on the wire. Without a pool the connection is not - // kept for a later request: it closes when its response is finished - // with. - let client: Client<_, http_body_util::Full> = - Client::builder(TokioExecutor::new()) - .pool_max_idle_per_host(0) - .build(https_connector()); - await_io(until, client.request(request)).await - }); + let response = TOKIO_RUNTIME + .block_on(async { await_io(until, streaming_client().request(request)).await }); let response = match response { Ok(Ok(response)) => response, Ok(Err(error)) => return Err(Failure::Other(error.to_string())), @@ -660,12 +710,7 @@ fn send_streaming_at( } let (parts, body) = response.into_parts(); - let head = StreamingHead { - status: parts.status.as_u16(), - headers: header_pairs(&parts.headers), - invalid_location: invalid_location(&parts.headers), - content_length: declared_content_length(&parts.headers), - }; + let head = StreamingHead::from_parts(&parts); let body = StreamingBody { state: BodyState::Open(body), pending: bytes::BytesMut::new(), @@ -684,6 +729,24 @@ fn streaming_err(failure: &Failure, roc_host: &RocHost) -> HostHttpSendStreaming } } +/// An instant on the `Host.monotonic_now!` clock. +fn monotonic_instant(at: AtOrNoTimeout) -> Option { + match at.tag { + AtOrNoTimeoutTag::At => { + crate::monotonic_origin().checked_add(Duration::from_nanos(unsafe { *at.payload.at })) + } + AtOrNoTimeoutTag::NoTimeout => None, + } +} + +/// An idle timeout as Roc passes it, in nanoseconds. +fn idle_duration(after: AfterOrNoTimeout) -> Option { + match after.tag { + AfterOrNoTimeoutTag::After => Some(Duration::from_nanos(unsafe { *after.payload.after })), + AfterOrNoTimeoutTag::NoTimeout => None, + } +} + #[no_mangle] pub extern "C" fn hosted_http_send_streaming( request: HostHttpSendStreamingArg0, @@ -691,17 +754,8 @@ pub extern "C" fn hosted_http_send_streaming( deadline: AtOrNoTimeout, ) -> HostHttpSendStreamingResult { let roc_host = roc_host(); - let deadline = match deadline.tag { - AtOrNoTimeoutTag::At => crate::monotonic_origin() - .checked_add(Duration::from_nanos(unsafe { *deadline.payload.at })), - AtOrNoTimeoutTag::NoTimeout => None, - }; - let idle_timeout = match idle_timeout.tag { - AfterOrNoTimeoutTag::After => { - Some(Duration::from_nanos(unsafe { *idle_timeout.payload.after })) - } - AfterOrNoTimeoutTag::NoTimeout => None, - }; + let deadline = monotonic_instant(deadline); + let idle_timeout = idle_duration(idle_timeout); // The hyper request copies what it needs, so the Roc values can go. let hyper_request = build_hyper_request((&request).into()); @@ -718,39 +772,9 @@ pub extern "C" fn hosted_http_send_streaming( Ok(handle) => handle, Err(error) => return streaming_err(&Failure::Other(error.to_string()), roc_host), }; - let content_length = match head.content_length { - Some(length) => KnownOrUnknown { - payload: KnownOrUnknownPayload { - known: ManuallyDrop::new(length), - }, - tag: KnownOrUnknownTag::Known, - }, - None => KnownOrUnknown { - payload: KnownOrUnknownPayload { unknown: [] }, - tag: KnownOrUnknownTag::Unknown, - }, - }; - let invalid_location = match head.invalid_location { - Some(byte) => NoneOrSome { - payload: NoneOrSomePayload { - some: ManuallyDrop::new(byte), - }, - tag: NoneOrSomeTag::Some, - }, - None => NoneOrSome { - payload: NoneOrSomePayload { none: [] }, - tag: NoneOrSomeTag::None, - }, - }; HostHttpSendStreamingResult { payload: HostHttpSendStreamingResultPayload { - ok: ManuallyDrop::new(HostHttpSendStreamingOk { - content_length, - invalid_location, - body: handle, - headers: build_roc_headers(&head.headers, roc_host), - status: head.status, - }), + ok: ManuallyDrop::new(head.to_roc(handle, roc_host)), }, tag: HostHttpSendStreamingResultTag::Ok, } From 53219d0dcd9133b400c4880354004ee94283e773 Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Wed, 11 Nov 2026 12:34:23 +0000 Subject: [PATCH 2/4] Add a host resource for HTTP requests in flight `http_start!` sends a request from a task on the HTTP runtime and returns a `Resource.HttpPending` handle at once. `http_pending_wait!` waits up to a given instant for the status and headers and leaves the request in flight when they have not arrived; once they have, every wait returns the same head and body handle. The task notes when the head arrived, and the idle timeout and deadline are judged by that time, so a head that came in time during another HTTP call is still returned by a later wait. `http_pending_follow!` sends the next redirect hop behind the same handle and records it; each wait that returns a head also returns the redirects followed so far, so that the redirect policy stays in Roc. Closing or releasing the handle aborts the request and closes its connection. Co-Authored-By: Claude Opus 5.5 --- platform/Host.roc | 15 + platform/Resource.roc | 1 + platform/main.roc | 4 + src/http.rs | 419 ++++++++++++++ src/resources.rs | 1 + src/roc_platform_abi.rs | 1203 +++++++++++++++++++++++++++++++++------ 6 files changed, 1463 insertions(+), 180 deletions(-) diff --git a/platform/Host.roc b/platform/Host.roc index e1e86cde..4fefaf09 100644 --- a/platform/Host.roc +++ b/platform/Host.roc @@ -188,4 +188,19 @@ Host :: [].{ unix_socket_listen! : NativePath => Try(Resource.UnixSocketListener, [UnixSocketErr(IOErr)]) unix_socket_accept! : Resource.UnixSocketListener, [NoTimeout, After(U64)] => Try(Resource.UnixSocketConnection, [Closed, UnixSocketErr(IOErr)]) unix_socket_listener_close! : Resource.UnixSocketListener => Try({}, [UnixSocketErr(IOErr)]) + + # `http_start!` sends a request from a task on the HTTP runtime and returns + # at once; the task runs only while a hosted HTTP call or wait is inside the + # runtime. Every failure of the request, including one to build it, is + # reported by `http_pending_wait!`; the error here means that no handle + # could be made. `http_pending_wait!` gives up at the `At` instant (the + # `monotonic_now!` clock) with `NotReady`, which keeps the request. Once the + # head has arrived, every wait returns it with the same body handle, and + # with the status and `Location` of each redirect followed so far, in + # order. `http_pending_follow!` replaces the request behind a handle by the + # next redirect hop and records the status and `Location` that led there. + http_start! : InternalHttp.RequestToAndFromHost, [NoTimeout, After(U64)], [NoTimeout, At(U64)] => Try(Resource.HttpPending, List(U8)) + http_pending_wait! : Resource.HttpPending, [NoTimeout, At(U64)] => Try([Ready({ response : InternalHttp.StreamingResponseFromHost, redirects : List((U16, Str)) }), NotReady], [Closed, HttpErr(InternalHttp.TransportErr)]) + http_pending_follow! : Resource.HttpPending, InternalHttp.RequestToAndFromHost, U16, Str => {} + http_pending_close! : Resource.HttpPending => {} } diff --git a/platform/Resource.roc b/platform/Resource.roc index 5fff217a..9633d8e1 100644 --- a/platform/Resource.roc +++ b/platform/Resource.roc @@ -14,6 +14,7 @@ Resource :: [].{ Child : Handle([ChildKind]) FileReader : Handle([FileReaderKind]) FileWriter : Handle([FileWriterKind]) + HttpPending : Handle([HttpPendingKind]) HttpResponse : Handle([HttpResponseKind]) SqliteDb : Handle([SqliteDbKind]) SqliteRows : Handle([SqliteRowsKind]) diff --git a/platform/main.roc b/platform/main.roc index f198e3dc..fb7c2d68 100644 --- a/platform/main.roc +++ b/platform/main.roc @@ -73,10 +73,14 @@ platform "" "hosted_file_writer_write_bytes": Host.file_writer_write_bytes!, "hosted_file_writer_write_utf8": Host.file_writer_write_utf8!, "hosted_http_close_body": Host.http_close_body!, + "hosted_http_pending_close": Host.http_pending_close!, + "hosted_http_pending_follow": Host.http_pending_follow!, + "hosted_http_pending_wait": Host.http_pending_wait!, "hosted_http_read_body": Host.http_read_body!, # HTTP is likewise kept at the end to avoid renumbering glue types. "hosted_http_send_request": Host.http_send_request!, "hosted_http_send_streaming": Host.http_send_streaming!, + "hosted_http_start": Host.http_start!, "hosted_locale_all": Host.locale_all!, "hosted_locale_get": Host.locale_get!, "hosted_monotonic_now": Host.monotonic_now!, diff --git a/src/http.rs b/src/http.rs index 66844b53..b64d9993 100644 --- a/src/http.rs +++ b/src/http.rs @@ -582,6 +582,7 @@ impl Drop for StreamingBody { } /// The status line and headers of a response whose body has not been read. +#[derive(Clone)] struct StreamingHead { status: u16, headers: Vec<(String, String)>, @@ -849,6 +850,424 @@ pub unsafe extern "C" fn hosted_http_read_body( result } +// ============================================================================ +// Pending responses +// +// `Http.start!` sends a request from a task on `TOKIO_RUNTIME` and returns at +// once, leaving the task behind a `Resource.HttpPending` handle. The runtime +// has no thread of its own (see `TOKIO_RUNTIME`), so the task only runs while +// a hosted call is inside `block_on`: any HTTP call, body read or wait. +// Between those calls the kernel's socket buffers hold what arrives. A wait +// that gives up keeps the task. Closing or dropping the handle aborts it and +// turns the runtime so that its connection closes, as `StreamingBody::close` +// does. +// +// The task notes when the head arrived, and the head is judged by that time, +// not by when a wait comes for it: a head that came in time during another +// HTTP call is not lost to a timeout while the app does other work. +// +// Redirects are decided in Roc. To follow one, Roc replaces the request behind +// the same handle, and the host records the status and `Location` that led +// there. Each wait that returns a head returns that list too, so that Roc can +// rebuild the hop it is on. +// ---------------------------------------------------------------------------- + +type SendResult = Result, hyper_util::client::legacy::Error>; + +enum PendingState { + /// The request is in flight, or its result, with the time it came, waits + /// in the finished task. + Sending(tokio::task::JoinHandle<(Instant, SendResult)>), + /// The head arrived at `at` during a wait that an interruption ended. The + /// next wait returns it. + Arrived { + at: Instant, + response: hyper::Response, + }, + /// The head has been handed to Roc. Every later wait hands over the same + /// head and another reference to the same body handle. + Ready { head: StreamingHead, body: *mut u64 }, + /// The request failed for good. Every later wait reports the same failure. + Failed(Failure), + /// Closed by `Http.PendingResponse.close!`. + Closed, +} + +/// The resource behind a `Resource.HttpPending` handle. +pub struct PendingResponse { + state: PendingState, + idle_timeout: Option, + /// The `Request` timeout, fixed at `Http.start!` for every hop and the body. + deadline: Option, + /// When the status and headers of the current hop must have arrived. + head_until: Option, + /// The status and `Location` of each redirect followed so far, in order. + redirects: Vec<(u16, String)>, + /// Releases the body reference held in `Ready`. + host: RocHost, +} + +/// What a wait for a pending response found. +enum Waited { + NotReady, + /// The head and a new reference to the body handle, owned by the caller. + Ready(StreamingHead, *mut u64), +} + +#[derive(Debug, PartialEq, Eq)] +enum PendingErr { + Closed, + Failed(Failure), +} + +/// Let the caller's own limit of a wait pass, after a few turns of the +/// runtime. The `select!` it is raced in polls the request first. Without the +/// turns, a limit that has passed would end the wait after the first turn: +/// that turn reads what has arrived on the socket and wakes the tasks that +/// handle it, but `block_on` polls this future again before it runs them. The +/// turns let those tasks run, so that even a wait whose limit has passed +/// returns a head that is already on the socket. (`sleep_until` rounds up to +/// the next millisecond, which hides this for a limit that has only just +/// passed, but not for one that passed a millisecond or more ago.) +async fn give_up_at(until: Option) { + for _ in 0..SETTLE_TURNS { + tokio::task::yield_now().await; + } + match until { + Some(until) => tokio::time::sleep_until(tokio::time::Instant::from_std(until)).await, + None => std::future::pending::<()>().await, + } +} + +impl PendingResponse { + fn new(idle_timeout: Option, deadline: Option) -> Self { + Self { + state: PendingState::Closed, + idle_timeout, + deadline, + head_until: None, + redirects: Vec::new(), + host: resources::make_host(), + } + } + + /// Start sending `request` as the current hop, giving up any earlier one. + fn send(&mut self, request: Result) { + self.release(); + self.state = match self.spawn(request) { + Ok(task) => PendingState::Sending(task), + Err(failure) => PendingState::Failed(failure), + }; + } + + fn spawn( + &mut self, + request: Result, + ) -> Result, Failure> { + let request = request.map_err(Failure::Other)?; + // As in `send_streaming_at`: nothing is sent once a timeout has expired. + if self.idle_timeout == Some(Duration::ZERO) { + return Err(Failure::Timeout); + } + let now = Instant::now(); + let budget = + wait_budget(self.idle_timeout, self.deadline, now).map_err(|()| Failure::Timeout)?; + self.head_until = budget.and_then(|budget| now.checked_add(budget)); + let task = TOKIO_RUNTIME.spawn(async move { + let sent = streaming_client().request(request).await; + (Instant::now(), sent) + }); + // Put the request on its way (connect, write what fits) without waiting. + settle(); + Ok(task) + } + + /// Follow a redirect: record it and send `request` as the next hop. + fn follow(&mut self, status: u16, location: String, request: Result) { + if matches!(self.state, PendingState::Closed) { + return; + } + self.redirects.push((status, location)); + self.send(request); + } + + /// Wait until `until` for the status and headers of the current hop. + fn wait(&mut self, until: Option) -> Result { + if matches!(self.state, PendingState::Closed) { + return Err(PendingErr::Closed); + } + if crate::signals::is_pending() { + return Err(PendingErr::Failed(Failure::Interrupted)); + } + let head_until = self.head_until; + if let PendingState::Sending(task) = &mut self.state { + let sent = if task.is_finished() { + // The result came during an earlier hosted call. Take it + // without a look at the clock: it is judged below by the + // time it came. + Ok(Some(TOKIO_RUNTIME.block_on(task))) + } else { + // Only the borrow of the task is given up when the wait ends + // without it, never the task. + TOKIO_RUNTIME.block_on(await_io(head_until, async { + tokio::select! { + biased; + sent = task => Some(sent), + () = give_up_at(until) => None, + } + })) + }; + let arrived = match sent { + Ok(None) => return Ok(Waited::NotReady), + // The task keeps the request; the next wait continues it. + Err(Failure::Interrupted) => { + return Err(PendingErr::Failed(Failure::Interrupted)); + } + Err(failure) => return Err(PendingErr::Failed(self.fail(failure))), + Ok(Some(Err(error))) => { + return Err(PendingErr::Failed( + self.fail(Failure::Other(error.to_string())), + )); + } + Ok(Some(Ok((_, Err(error))))) => { + return Err(PendingErr::Failed( + self.fail(Failure::Other(error.to_string())), + )); + } + Ok(Some(Ok((at, Ok(response))))) => PendingState::Arrived { at, response }, + }; + self.state = arrived; + } + let arrived_at = match &self.state { + PendingState::Failed(failure) => return Err(PendingErr::Failed(failure.clone())), + PendingState::Ready { head, body } => return Ok(Self::hand_over(head.clone(), *body)), + PendingState::Arrived { at, .. } => *at, + PendingState::Sending(_) | PendingState::Closed => { + unreachable!("a wait leaves no request in flight without returning") + } + }; + // As in `send_streaming_at`: an event that came while the head was + // being polled wins over it. An interruption keeps the head for later. + if crate::signals::is_pending() { + return Err(PendingErr::Failed(Failure::Interrupted)); + } + // The head counts from when it arrived, which may be long before this + // wait. + if head_until.is_some_and(|until| until <= arrived_at) { + return Err(PendingErr::Failed(self.fail(Failure::Timeout))); + } + let PendingState::Arrived { response, .. } = + std::mem::replace(&mut self.state, PendingState::Closed) + else { + unreachable!("only a head that has arrived is left"); + }; + let (parts, body) = response.into_parts(); + let head = StreamingHead::from_parts(&parts); + let body = StreamingBody { + state: BodyState::Open(body), + pending: bytes::BytesMut::new(), + idle_timeout: self.idle_timeout, + deadline: self.deadline, + }; + let body = match resources::box_resource(body, &self.host) { + Ok(body) => body, + Err(error) => { + return Err(PendingErr::Failed( + self.fail(Failure::Other(error.to_string())), + )); + } + }; + self.state = PendingState::Ready { + head: head.clone(), + body, + }; + Ok(Self::hand_over(head, body)) + } + + /// The redirects followed so far, as Roc receives them. + fn redirects_to_roc(&self, roc_host: &RocHost) -> RocList { + let list = unsafe { + RocList::::allocate(self.redirects.len(), roc_host) + }; + for (index, (status, location)) in self.redirects.iter().enumerate() { + let redirect = NotReadyOrReadyReadyRedirects { + _0: *status, + _1: RocStr::from_str(location, roc_host), + }; + unsafe { + list.elements.add(index).write(redirect); + } + } + list + } + + /// A new reference to the body for the caller; this resource keeps its own. + fn hand_over(head: StreamingHead, body: *mut u64) -> Waited { + unsafe { incref_box(body.cast(), 1) }; + Waited::Ready(head, body) + } + + /// Give up the request and whatever it holds; the state is `Closed` after. + fn release(&mut self) { + match std::mem::replace(&mut self.state, PendingState::Closed) { + PendingState::Sending(task) => { + task.abort(); + drop(task); + settle(); + } + PendingState::Arrived { response, .. } => { + drop(response); + settle(); + } + PendingState::Ready { body, .. } => resources::release(body, &self.host), + PendingState::Failed(_) | PendingState::Closed => {} + } + } + + fn fail(&mut self, failure: Failure) -> Failure { + self.release(); + self.state = PendingState::Failed(failure.clone()); + failure + } +} + +impl Drop for PendingResponse { + fn drop(&mut self) { + // A request in flight still holds its connection. + self.release(); + } +} + +#[no_mangle] +pub extern "C" fn hosted_http_start( + request: HostHttpStartArg0, + idle_timeout: AfterOrNoTimeout, + deadline: AtOrNoTimeout, +) -> HostHttpStartResult { + let roc_host = roc_host(); + // The hyper request copies what it needs, so the Roc values can go. + let hyper_request = build_hyper_request((&request).into()); + unsafe { request.decref(roc_host) }; + + // Make the handle before anything is sent, so that a failure to make it + // sends nothing. + let pending = PendingResponse::new(idle_duration(idle_timeout), monotonic_instant(deadline)); + let handle = match resources::box_resource(pending, roc_host) { + Ok(handle) => handle, + Err(error) => { + return HostHttpStartResult { + payload: HostHttpStartResultPayload { + err: ManuallyDrop::new(roc_u8_list_from_slice( + error.to_string().as_bytes(), + roc_host, + )), + }, + tag: HostHttpStartResultTag::Err, + }; + } + }; + unsafe { resources::resource_ref::(handle) }.send(hyper_request); + HostHttpStartResult { + payload: HostHttpStartResultPayload { + ok: ManuallyDrop::new(handle), + }, + tag: HostHttpStartResultTag::Ok, + } +} + +fn pending_wait_ok(ok: HostHttpPendingWaitOk) -> HostHttpPendingWaitResult { + HostHttpPendingWaitResult { + payload: HostHttpPendingWaitResultPayload { + ok: ManuallyDrop::new(ok), + }, + tag: HostHttpPendingWaitResultTag::Ok, + } +} + +fn pending_wait_err(err: HostHttpPendingWaitErr) -> HostHttpPendingWaitResult { + HostHttpPendingWaitResult { + payload: HostHttpPendingWaitResultPayload { + err: ManuallyDrop::new(err), + }, + tag: HostHttpPendingWaitResultTag::Err, + } +} + +/// # Safety +/// +/// `handle` must be a live handle from `hosted_http_start`. +#[no_mangle] +pub unsafe extern "C" fn hosted_http_pending_wait( + handle: *mut u64, + until: AtOrNoTimeout, +) -> HostHttpPendingWaitResult { + let roc_host = roc_host(); + let pending = unsafe { resources::resource_ref::(handle) }; + let result = match pending.wait(monotonic_instant(until)) { + Ok(Waited::NotReady) => pending_wait_ok(HostHttpPendingWaitOk { + payload: HostHttpPendingWaitOkPayload { not_ready: [] }, + tag: HostHttpPendingWaitOkTag::NotReady, + }), + Ok(Waited::Ready(head, body)) => pending_wait_ok(HostHttpPendingWaitOk { + payload: HostHttpPendingWaitOkPayload { + ready: ManuallyDrop::new(HostHttpPendingWaitOkReady { + response: head.to_roc(body, roc_host), + redirects: pending.redirects_to_roc(roc_host), + }), + }, + tag: HostHttpPendingWaitOkTag::Ready, + }), + Err(PendingErr::Closed) => pending_wait_err(HostHttpPendingWaitErr { + payload: HostHttpPendingWaitErrPayload { closed: [] }, + tag: HostHttpPendingWaitErrTag::Closed, + }), + Err(PendingErr::Failed(failure)) => pending_wait_err(HostHttpPendingWaitErr { + payload: HostHttpPendingWaitErrPayload { + http_err: ManuallyDrop::new(failure.to_roc(roc_host)), + }, + tag: HostHttpPendingWaitErrTag::HttpErr, + }), + }; + // Only now, outside `block_on`: this may be the last reference. + resources::release(handle, roc_host); + result +} + +/// # Safety +/// +/// `handle` must be a live handle from `hosted_http_start`. +#[no_mangle] +pub unsafe extern "C" fn hosted_http_pending_follow( + handle: *mut u64, + request: HostHttpPendingFollowArg1, + status: u16, + location: RocStr, +) { + let roc_host = roc_host(); + let hyper_request = build_hyper_request((&request).into()); + let followed = location.as_str().to_owned(); + unsafe { + request.decref(roc_host); + location.decref(roc_host); + } + unsafe { resources::resource_ref::(handle) }.follow( + status, + followed, + hyper_request, + ); + resources::release(handle, roc_host); +} + +/// # Safety +/// +/// `handle` must be a live handle from `hosted_http_start`. +#[no_mangle] +pub unsafe extern "C" fn hosted_http_pending_close(handle: *mut u64) { + let roc_host = roc_host(); + unsafe { resources::resource_ref::(handle) }.release(); + resources::release(handle, roc_host); +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/resources.rs b/src/resources.rs index 92c32053..7843d627 100644 --- a/src/resources.rs +++ b/src/resources.rs @@ -93,6 +93,7 @@ resource_kinds! { Child => crate::process_service::Child, FileReader => std::io::BufReader, FileWriter => crate::file_writer::FileWriter, + HttpPending => crate::http::PendingResponse, HttpResponse => crate::http::StreamingBody, SqliteDb => crate::sqlite::SqliteDb, SqliteRows => crate::sqlite::SqliteRows, diff --git a/src/roc_platform_abi.rs b/src/roc_platform_abi.rs index aa9c7518..4d30065f 100644 --- a/src/roc_platform_abi.rs +++ b/src/roc_platform_abi.rs @@ -1521,6 +1521,60 @@ const _: () = assert!(core::mem::size_of::() == 1, "SystemUnix size #[cfg(target_pointer_width = "32")] const _: () = assert!(core::mem::align_of::() == 1, "SystemUnix alignment mismatch"); +/// Element type for __AnonStruct_57b3856c59c615c8 +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct AnonStruct57b3856c59c615c8 { + pub response: AnonStruct60cbed700549ed13, + pub redirects: RocList, +} + +/// Element type for __AnonStruct_57b3856c59c615c8 +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct AnonStruct57b3856c59c615c8 { + pub response: AnonStruct60cbed700549ed13, + pub redirects: RocList, +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 80, "AnonStruct57b3856c59c615c8 size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 8, "AnonStruct57b3856c59c615c8 alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 56, "AnonStruct57b3856c59c615c8 size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 8, "AnonStruct57b3856c59c615c8 alignment mismatch"); + +/// Element type for __AnonStruct_453a8dc75d4311da +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct AnonStruct453a8dc75d4311da { + pub _1: RocStr, + pub _0: u16, +} + +/// Element type for __AnonStruct_453a8dc75d4311da +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct AnonStruct453a8dc75d4311da { + pub _1: RocStr, + pub _0: u16, +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 32, "AnonStruct453a8dc75d4311da size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 8, "AnonStruct453a8dc75d4311da alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 16, "AnonStruct453a8dc75d4311da size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 4, "AnonStruct453a8dc75d4311da alignment mismatch"); + /// Element type for System.Target #[cfg(target_pointer_width = "32")] #[repr(C)] @@ -13940,318 +13994,754 @@ const _: () = assert!(core::mem::align_of::() == 4, "Clos #[cfg(target_pointer_width = "32")] const _: () = assert!(core::mem::offset_of!(ClosedOrUnixSocketErr, tag) == 16, "ClosedOrUnixSocketErr tag offset mismatch"); -/// Tag union: Arm64OrX64 -#[repr(u8)] -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum Arm64OrX64 { - Arm64 = 0, - X64 = 1, -} - -#[cfg(target_pointer_width = "64")] -const _: () = assert!(core::mem::size_of::() == 1, "Arm64OrX64 size mismatch"); -#[cfg(target_pointer_width = "64")] -const _: () = assert!(core::mem::align_of::() == 1, "Arm64OrX64 alignment mismatch"); -#[cfg(target_pointer_width = "32")] -const _: () = assert!(core::mem::size_of::() == 1, "Arm64OrX64 size mismatch"); -#[cfg(target_pointer_width = "32")] -const _: () = assert!(core::mem::align_of::() == 1, "Arm64OrX64 alignment mismatch"); - -/// Tag discriminant for LinuxOrMacOSOrWindows. +/// Tag discriminant for Try. #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum LinuxOrMacOSOrWindowsTag { - Linux = 0, - MacOS = 1, - Windows = 2, +pub enum HostHttpStartResultTag { + Err = 0, + Ok = 1, } #[repr(C)] #[derive(Clone, Copy)] -pub union LinuxOrMacOSOrWindowsPayload { - pub linux: core::mem::ManuallyDrop, - pub mac_os: core::mem::ManuallyDrop, - pub windows: core::mem::ManuallyDrop, +pub union HostHttpStartResultPayload { + pub err: core::mem::ManuallyDrop>, + pub ok: core::mem::ManuallyDrop<*mut u64>, } #[cfg(target_pointer_width = "32")] -#[repr(align(1))] +#[repr(align(4))] #[derive(Clone, Copy)] -pub struct LinuxOrMacOSOrWindowsPayloadAlignment; +pub struct HostHttpStartResultPayloadAlignment; -/// Tag union: LinuxOrMacOSOrWindows +/// Tag union: Try #[cfg(target_pointer_width = "32")] #[repr(C)] #[derive(Clone, Copy)] -pub struct LinuxOrMacOSOrWindows { - pub _payload_alignment: [LinuxOrMacOSOrWindowsPayloadAlignment; 0], - pub payload: [u8; 1], - pub tag: LinuxOrMacOSOrWindowsTag, +pub struct HostHttpStartResult { + pub _payload_alignment: [HostHttpStartResultPayloadAlignment; 0], + pub payload: [u8; 12], + pub tag: HostHttpStartResultTag, } -/// Tag union: LinuxOrMacOSOrWindows +/// Tag union: Try #[cfg(not(target_pointer_width = "32"))] #[repr(C)] #[derive(Clone, Copy)] -pub struct LinuxOrMacOSOrWindows { - pub payload: LinuxOrMacOSOrWindowsPayload, - pub tag: LinuxOrMacOSOrWindowsTag, +pub struct HostHttpStartResult { + pub payload: HostHttpStartResultPayload, + pub tag: HostHttpStartResultTag, } -impl LinuxOrMacOSOrWindows { - /// Borrow the `Linux` payload without creating another owner. - /// - /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux` and the payload must still be initialized. - #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_linux_unchecked(&self) -> &SystemLinux { - unsafe { &*(self.payload.as_ptr() as *const SystemLinux) } - } - - /// Borrow the `Linux` payload without creating another owner. - /// - /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux` and the payload must still be initialized. - #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_linux_unchecked(&self) -> &SystemLinux { - unsafe { &*(&self.payload.linux as *const core::mem::ManuallyDrop as *const SystemLinux) } - } - - /// Move the `Linux` payload out of one owned tag-union shell. - /// - /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux`. After this call, `self` is logically uninitialized and must not be read or destroyed. - #[cfg(target_pointer_width = "32")] - pub unsafe fn take_payload_linux_unchecked(&mut self) -> SystemLinux { - unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemLinux) } - } - - /// Move the `Linux` payload out of one owned tag-union shell. - /// - /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux`. After this call, `self` is logically uninitialized and must not be read or destroyed. - #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn take_payload_linux_unchecked(&mut self) -> SystemLinux { - unsafe { core::mem::ManuallyDrop::take(&mut self.payload.linux) } - } - - /// Borrow the `MacOS` payload without creating another owner. +impl HostHttpStartResult { + /// Borrow the `Err` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS` and the payload must still be initialized. + /// `self.tag` must be `HostHttpStartResultTag::Err` and the payload must still be initialized. #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_mac_os_unchecked(&self) -> &SystemMacOS { - unsafe { &*(self.payload.as_ptr() as *const SystemMacOS) } + pub unsafe fn borrow_payload_err_unchecked(&self) -> &RocListWith { + unsafe { &*(self.payload.as_ptr() as *const RocListWith) } } - /// Borrow the `MacOS` payload without creating another owner. + /// Borrow the `Err` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS` and the payload must still be initialized. + /// `self.tag` must be `HostHttpStartResultTag::Err` and the payload must still be initialized. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_mac_os_unchecked(&self) -> &SystemMacOS { - unsafe { &*(&self.payload.mac_os as *const core::mem::ManuallyDrop as *const SystemMacOS) } + pub unsafe fn borrow_payload_err_unchecked(&self) -> &RocListWith { + unsafe { &*(&self.payload.err as *const core::mem::ManuallyDrop> as *const RocListWith) } } - /// Move the `MacOS` payload out of one owned tag-union shell. + /// Move the `Err` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpStartResultTag::Err`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(target_pointer_width = "32")] - pub unsafe fn take_payload_mac_os_unchecked(&mut self) -> SystemMacOS { - unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemMacOS) } + pub unsafe fn take_payload_err_unchecked(&mut self) -> RocListWith { + unsafe { core::ptr::read(self.payload.as_ptr() as *const RocListWith) } } - /// Move the `MacOS` payload out of one owned tag-union shell. + /// Move the `Err` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpStartResultTag::Err`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn take_payload_mac_os_unchecked(&mut self) -> SystemMacOS { - unsafe { core::mem::ManuallyDrop::take(&mut self.payload.mac_os) } + pub unsafe fn take_payload_err_unchecked(&mut self) -> RocListWith { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.err) } } - /// Borrow the `Windows` payload without creating another owner. + /// Borrow the `Ok` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows` and the payload must still be initialized. + /// `self.tag` must be `HostHttpStartResultTag::Ok` and the payload must still be initialized. #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_windows_unchecked(&self) -> &SystemWindows { - unsafe { &*(self.payload.as_ptr() as *const SystemWindows) } + pub unsafe fn borrow_payload_ok_unchecked(&self) -> &*mut u64 { + unsafe { &*(self.payload.as_ptr() as *const *mut u64) } } - /// Borrow the `Windows` payload without creating another owner. + /// Borrow the `Ok` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows` and the payload must still be initialized. + /// `self.tag` must be `HostHttpStartResultTag::Ok` and the payload must still be initialized. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_windows_unchecked(&self) -> &SystemWindows { - unsafe { &*(&self.payload.windows as *const core::mem::ManuallyDrop as *const SystemWindows) } + pub unsafe fn borrow_payload_ok_unchecked(&self) -> &*mut u64 { + unsafe { &*(&self.payload.ok as *const core::mem::ManuallyDrop<*mut u64> as *const *mut u64) } } - /// Move the `Windows` payload out of one owned tag-union shell. + /// Move the `Ok` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpStartResultTag::Ok`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(target_pointer_width = "32")] - pub unsafe fn take_payload_windows_unchecked(&mut self) -> SystemWindows { - unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemWindows) } + pub unsafe fn take_payload_ok_unchecked(&mut self) -> *mut u64 { + unsafe { core::ptr::read(self.payload.as_ptr() as *const *mut u64) } } - /// Move the `Windows` payload out of one owned tag-union shell. + /// Move the `Ok` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpStartResultTag::Ok`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn take_payload_windows_unchecked(&mut self) -> SystemWindows { - unsafe { core::mem::ManuallyDrop::take(&mut self.payload.windows) } + pub unsafe fn take_payload_ok_unchecked(&mut self) -> *mut u64 { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.ok) } } } #[cfg(target_pointer_width = "64")] -const _: () = assert!(core::mem::size_of::() == 2, "LinuxOrMacOSOrWindows size mismatch"); +const _: () = assert!(core::mem::size_of::() == 32, "HostHttpStartResult size mismatch"); #[cfg(target_pointer_width = "64")] -const _: () = assert!(core::mem::align_of::() == 1, "LinuxOrMacOSOrWindows alignment mismatch"); +const _: () = assert!(core::mem::align_of::() == 8, "HostHttpStartResult alignment mismatch"); #[cfg(target_pointer_width = "64")] -const _: () = assert!(core::mem::offset_of!(LinuxOrMacOSOrWindows, tag) == 1, "LinuxOrMacOSOrWindows tag offset mismatch"); +const _: () = assert!(core::mem::offset_of!(HostHttpStartResult, tag) == 24, "HostHttpStartResult tag offset mismatch"); #[cfg(target_pointer_width = "32")] -const _: () = assert!(core::mem::size_of::() == 2, "LinuxOrMacOSOrWindows size mismatch"); +const _: () = assert!(core::mem::size_of::() == 16, "HostHttpStartResult size mismatch"); #[cfg(target_pointer_width = "32")] -const _: () = assert!(core::mem::align_of::() == 1, "LinuxOrMacOSOrWindows alignment mismatch"); +const _: () = assert!(core::mem::align_of::() == 4, "HostHttpStartResult alignment mismatch"); #[cfg(target_pointer_width = "32")] -const _: () = assert!(core::mem::offset_of!(LinuxOrMacOSOrWindows, tag) == 1, "LinuxOrMacOSOrWindows tag offset mismatch"); +const _: () = assert!(core::mem::offset_of!(HostHttpStartResult, tag) == 12, "HostHttpStartResult tag offset mismatch"); -/// Tag discriminant for OsStr. +/// Tag discriminant for Try. #[repr(u8)] #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum OsStrTag { - UnixBytes = 0, - Utf8 = 1, - WindowsU16s = 2, +pub enum HostHttpPendingWaitResultTag { + Err = 0, + Ok = 1, } #[repr(C)] #[derive(Clone, Copy)] -pub union OsStrPayload { - pub unix_bytes: core::mem::ManuallyDrop>, - pub utf8: core::mem::ManuallyDrop, - pub windows_u16s: core::mem::ManuallyDrop>, +pub union HostHttpPendingWaitResultPayload { + pub err: core::mem::ManuallyDrop, + pub ok: core::mem::ManuallyDrop, } #[cfg(target_pointer_width = "32")] -#[repr(align(4))] +#[repr(align(8))] #[derive(Clone, Copy)] -pub struct OsStrPayloadAlignment; +pub struct HostHttpPendingWaitResultPayloadAlignment; -/// Tag union: OsStr +/// Tag union: Try #[cfg(target_pointer_width = "32")] #[repr(C)] #[derive(Clone, Copy)] -pub struct OsStr { - pub _payload_alignment: [OsStrPayloadAlignment; 0], - pub payload: [u8; 12], - pub tag: OsStrTag, +pub struct HostHttpPendingWaitResult { + pub _payload_alignment: [HostHttpPendingWaitResultPayloadAlignment; 0], + pub payload: [u8; 64], + pub tag: HostHttpPendingWaitResultTag, } -/// Tag union: OsStr +/// Tag union: Try #[cfg(not(target_pointer_width = "32"))] #[repr(C)] #[derive(Clone, Copy)] -pub struct OsStr { - pub payload: OsStrPayload, - pub tag: OsStrTag, +pub struct HostHttpPendingWaitResult { + pub payload: HostHttpPendingWaitResultPayload, + pub tag: HostHttpPendingWaitResultTag, } -impl OsStr { - /// Borrow the `UnixBytes` payload without creating another owner. +impl HostHttpPendingWaitResult { + /// Borrow the `Err` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `OsStrTag::UnixBytes` and the payload must still be initialized. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Err` and the payload must still be initialized. #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_unix_bytes_unchecked(&self) -> &RocListWith { - unsafe { &*(self.payload.as_ptr() as *const RocListWith) } + pub unsafe fn borrow_payload_err_unchecked(&self) -> &ClosedOrHttpErr { + unsafe { &*(self.payload.as_ptr() as *const ClosedOrHttpErr) } } - /// Borrow the `UnixBytes` payload without creating another owner. + /// Borrow the `Err` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `OsStrTag::UnixBytes` and the payload must still be initialized. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Err` and the payload must still be initialized. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_unix_bytes_unchecked(&self) -> &RocListWith { - unsafe { &*(&self.payload.unix_bytes as *const core::mem::ManuallyDrop> as *const RocListWith) } + pub unsafe fn borrow_payload_err_unchecked(&self) -> &ClosedOrHttpErr { + unsafe { &*(&self.payload.err as *const core::mem::ManuallyDrop as *const ClosedOrHttpErr) } } - /// Move the `UnixBytes` payload out of one owned tag-union shell. + /// Move the `Err` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `OsStrTag::UnixBytes`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Err`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(target_pointer_width = "32")] - pub unsafe fn take_payload_unix_bytes_unchecked(&mut self) -> RocListWith { - unsafe { core::ptr::read(self.payload.as_ptr() as *const RocListWith) } + pub unsafe fn take_payload_err_unchecked(&mut self) -> ClosedOrHttpErr { + unsafe { core::ptr::read(self.payload.as_ptr() as *const ClosedOrHttpErr) } } - /// Move the `UnixBytes` payload out of one owned tag-union shell. + /// Move the `Err` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `OsStrTag::UnixBytes`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Err`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn take_payload_unix_bytes_unchecked(&mut self) -> RocListWith { - unsafe { core::mem::ManuallyDrop::take(&mut self.payload.unix_bytes) } + pub unsafe fn take_payload_err_unchecked(&mut self) -> ClosedOrHttpErr { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.err) } } - /// Borrow the `Utf8` payload without creating another owner. + /// Borrow the `Ok` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `OsStrTag::Utf8` and the payload must still be initialized. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Ok` and the payload must still be initialized. #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_utf8_unchecked(&self) -> &RocStr { - unsafe { &*(self.payload.as_ptr() as *const RocStr) } + pub unsafe fn borrow_payload_ok_unchecked(&self) -> &NotReadyOrReady { + unsafe { &*(self.payload.as_ptr() as *const NotReadyOrReady) } } - /// Borrow the `Utf8` payload without creating another owner. + /// Borrow the `Ok` payload without creating another owner. /// /// # Safety - /// `self.tag` must be `OsStrTag::Utf8` and the payload must still be initialized. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Ok` and the payload must still be initialized. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_utf8_unchecked(&self) -> &RocStr { - unsafe { &*(&self.payload.utf8 as *const core::mem::ManuallyDrop as *const RocStr) } + pub unsafe fn borrow_payload_ok_unchecked(&self) -> &NotReadyOrReady { + unsafe { &*(&self.payload.ok as *const core::mem::ManuallyDrop as *const NotReadyOrReady) } } - /// Move the `Utf8` payload out of one owned tag-union shell. + /// Move the `Ok` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `OsStrTag::Utf8`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Ok`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(target_pointer_width = "32")] - pub unsafe fn take_payload_utf8_unchecked(&mut self) -> RocStr { - unsafe { core::ptr::read(self.payload.as_ptr() as *const RocStr) } + pub unsafe fn take_payload_ok_unchecked(&mut self) -> NotReadyOrReady { + unsafe { core::ptr::read(self.payload.as_ptr() as *const NotReadyOrReady) } } - /// Move the `Utf8` payload out of one owned tag-union shell. + /// Move the `Ok` payload out of one owned tag-union shell. /// /// # Safety - /// `self.tag` must be `OsStrTag::Utf8`. After this call, `self` is logically uninitialized and must not be read or destroyed. + /// `self.tag` must be `HostHttpPendingWaitResultTag::Ok`. After this call, `self` is logically uninitialized and must not be read or destroyed. #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn take_payload_utf8_unchecked(&mut self) -> RocStr { - unsafe { core::mem::ManuallyDrop::take(&mut self.payload.utf8) } + pub unsafe fn take_payload_ok_unchecked(&mut self) -> NotReadyOrReady { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.ok) } } - /// Borrow the `WindowsU16s` payload without creating another owner. - /// - /// # Safety - /// `self.tag` must be `OsStrTag::WindowsU16s` and the payload must still be initialized. - #[cfg(target_pointer_width = "32")] - pub unsafe fn borrow_payload_windows_u16s_unchecked(&self) -> &RocListWith { - unsafe { &*(self.payload.as_ptr() as *const RocListWith) } - } +} - /// Borrow the `WindowsU16s` payload without creating another owner. - /// - /// # Safety - /// `self.tag` must be `OsStrTag::WindowsU16s` and the payload must still be initialized. - #[cfg(not(target_pointer_width = "32"))] - pub unsafe fn borrow_payload_windows_u16s_unchecked(&self) -> &RocListWith { - unsafe { &*(&self.payload.windows_u16s as *const core::mem::ManuallyDrop> as *const RocListWith) } +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 96, "HostHttpPendingWaitResult size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 8, "HostHttpPendingWaitResult alignment mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::offset_of!(HostHttpPendingWaitResult, tag) == 88, "HostHttpPendingWaitResult tag offset mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 72, "HostHttpPendingWaitResult size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 8, "HostHttpPendingWaitResult alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::offset_of!(HostHttpPendingWaitResult, tag) == 64, "HostHttpPendingWaitResult tag offset mismatch"); + +/// Tag discriminant for ClosedOrHttpErr. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClosedOrHttpErrTag { + Closed = 0, + HttpErr = 1, +} + +#[repr(C)] +#[derive(Clone, Copy)] +pub union ClosedOrHttpErrPayload { + pub closed: [u8; 0], + pub http_err: core::mem::ManuallyDrop, +} + +#[cfg(target_pointer_width = "32")] +#[repr(align(4))] +#[derive(Clone, Copy)] +pub struct ClosedOrHttpErrPayloadAlignment; + +/// Tag union: ClosedOrHttpErr +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ClosedOrHttpErr { + pub _payload_alignment: [ClosedOrHttpErrPayloadAlignment; 0], + pub payload: [u8; 16], + pub tag: ClosedOrHttpErrTag, +} + +/// Tag union: ClosedOrHttpErr +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct ClosedOrHttpErr { + pub payload: ClosedOrHttpErrPayload, + pub tag: ClosedOrHttpErrTag, +} + +impl ClosedOrHttpErr { + /// Borrow the `HttpErr` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `ClosedOrHttpErrTag::HttpErr` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_http_err_unchecked(&self) -> &BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout { + unsafe { &*(self.payload.as_ptr() as *const BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout) } + } + + /// Borrow the `HttpErr` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `ClosedOrHttpErrTag::HttpErr` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_http_err_unchecked(&self) -> &BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout { + unsafe { &*(&self.payload.http_err as *const core::mem::ManuallyDrop as *const BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout) } + } + + /// Move the `HttpErr` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `ClosedOrHttpErrTag::HttpErr`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_http_err_unchecked(&mut self) -> BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout { + unsafe { core::ptr::read(self.payload.as_ptr() as *const BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout) } + } + + /// Move the `HttpErr` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `ClosedOrHttpErrTag::HttpErr`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_http_err_unchecked(&mut self) -> BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.http_err) } + } + +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 40, "ClosedOrHttpErr size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 8, "ClosedOrHttpErr alignment mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::offset_of!(ClosedOrHttpErr, tag) == 32, "ClosedOrHttpErr tag offset mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 20, "ClosedOrHttpErr size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 4, "ClosedOrHttpErr alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::offset_of!(ClosedOrHttpErr, tag) == 16, "ClosedOrHttpErr tag offset mismatch"); + +/// Tag discriminant for NotReadyOrReady. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NotReadyOrReadyTag { + NotReady = 0, + Ready = 1, +} + +#[repr(C)] +#[derive(Clone, Copy)] +pub union NotReadyOrReadyPayload { + pub not_ready: [u8; 0], + pub ready: core::mem::ManuallyDrop, +} + +#[cfg(target_pointer_width = "32")] +#[repr(align(8))] +#[derive(Clone, Copy)] +pub struct NotReadyOrReadyPayloadAlignment; + +/// Tag union: NotReadyOrReady +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct NotReadyOrReady { + pub _payload_alignment: [NotReadyOrReadyPayloadAlignment; 0], + pub payload: [u8; 56], + pub tag: NotReadyOrReadyTag, +} + +/// Tag union: NotReadyOrReady +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct NotReadyOrReady { + pub payload: NotReadyOrReadyPayload, + pub tag: NotReadyOrReadyTag, +} + +impl NotReadyOrReady { + /// Borrow the `Ready` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `NotReadyOrReadyTag::Ready` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_ready_unchecked(&self) -> &AnonStruct57b3856c59c615c8 { + unsafe { &*(self.payload.as_ptr() as *const AnonStruct57b3856c59c615c8) } + } + + /// Borrow the `Ready` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `NotReadyOrReadyTag::Ready` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_ready_unchecked(&self) -> &AnonStruct57b3856c59c615c8 { + unsafe { &*(&self.payload.ready as *const core::mem::ManuallyDrop as *const AnonStruct57b3856c59c615c8) } + } + + /// Move the `Ready` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `NotReadyOrReadyTag::Ready`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_ready_unchecked(&mut self) -> AnonStruct57b3856c59c615c8 { + unsafe { core::ptr::read(self.payload.as_ptr() as *const AnonStruct57b3856c59c615c8) } + } + + /// Move the `Ready` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `NotReadyOrReadyTag::Ready`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_ready_unchecked(&mut self) -> AnonStruct57b3856c59c615c8 { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.ready) } + } + +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 88, "NotReadyOrReady size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 8, "NotReadyOrReady alignment mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::offset_of!(NotReadyOrReady, tag) == 80, "NotReadyOrReady tag offset mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 64, "NotReadyOrReady size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 8, "NotReadyOrReady alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::offset_of!(NotReadyOrReady, tag) == 56, "NotReadyOrReady tag offset mismatch"); + +/// Tag union: Arm64OrX64 +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Arm64OrX64 { + Arm64 = 0, + X64 = 1, +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 1, "Arm64OrX64 size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 1, "Arm64OrX64 alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 1, "Arm64OrX64 size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 1, "Arm64OrX64 alignment mismatch"); + +/// Tag discriminant for LinuxOrMacOSOrWindows. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LinuxOrMacOSOrWindowsTag { + Linux = 0, + MacOS = 1, + Windows = 2, +} + +#[repr(C)] +#[derive(Clone, Copy)] +pub union LinuxOrMacOSOrWindowsPayload { + pub linux: core::mem::ManuallyDrop, + pub mac_os: core::mem::ManuallyDrop, + pub windows: core::mem::ManuallyDrop, +} + +#[cfg(target_pointer_width = "32")] +#[repr(align(1))] +#[derive(Clone, Copy)] +pub struct LinuxOrMacOSOrWindowsPayloadAlignment; + +/// Tag union: LinuxOrMacOSOrWindows +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct LinuxOrMacOSOrWindows { + pub _payload_alignment: [LinuxOrMacOSOrWindowsPayloadAlignment; 0], + pub payload: [u8; 1], + pub tag: LinuxOrMacOSOrWindowsTag, +} + +/// Tag union: LinuxOrMacOSOrWindows +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct LinuxOrMacOSOrWindows { + pub payload: LinuxOrMacOSOrWindowsPayload, + pub tag: LinuxOrMacOSOrWindowsTag, +} + +impl LinuxOrMacOSOrWindows { + /// Borrow the `Linux` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_linux_unchecked(&self) -> &SystemLinux { + unsafe { &*(self.payload.as_ptr() as *const SystemLinux) } + } + + /// Borrow the `Linux` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_linux_unchecked(&self) -> &SystemLinux { + unsafe { &*(&self.payload.linux as *const core::mem::ManuallyDrop as *const SystemLinux) } + } + + /// Move the `Linux` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_linux_unchecked(&mut self) -> SystemLinux { + unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemLinux) } + } + + /// Move the `Linux` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Linux`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_linux_unchecked(&mut self) -> SystemLinux { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.linux) } + } + + /// Borrow the `MacOS` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_mac_os_unchecked(&self) -> &SystemMacOS { + unsafe { &*(self.payload.as_ptr() as *const SystemMacOS) } + } + + /// Borrow the `MacOS` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_mac_os_unchecked(&self) -> &SystemMacOS { + unsafe { &*(&self.payload.mac_os as *const core::mem::ManuallyDrop as *const SystemMacOS) } + } + + /// Move the `MacOS` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_mac_os_unchecked(&mut self) -> SystemMacOS { + unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemMacOS) } + } + + /// Move the `MacOS` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::MacOS`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_mac_os_unchecked(&mut self) -> SystemMacOS { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.mac_os) } + } + + /// Borrow the `Windows` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_windows_unchecked(&self) -> &SystemWindows { + unsafe { &*(self.payload.as_ptr() as *const SystemWindows) } + } + + /// Borrow the `Windows` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_windows_unchecked(&self) -> &SystemWindows { + unsafe { &*(&self.payload.windows as *const core::mem::ManuallyDrop as *const SystemWindows) } + } + + /// Move the `Windows` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_windows_unchecked(&mut self) -> SystemWindows { + unsafe { core::ptr::read(self.payload.as_ptr() as *const SystemWindows) } + } + + /// Move the `Windows` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `LinuxOrMacOSOrWindowsTag::Windows`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_windows_unchecked(&mut self) -> SystemWindows { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.windows) } + } + +} + +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::size_of::() == 2, "LinuxOrMacOSOrWindows size mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::align_of::() == 1, "LinuxOrMacOSOrWindows alignment mismatch"); +#[cfg(target_pointer_width = "64")] +const _: () = assert!(core::mem::offset_of!(LinuxOrMacOSOrWindows, tag) == 1, "LinuxOrMacOSOrWindows tag offset mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::size_of::() == 2, "LinuxOrMacOSOrWindows size mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::align_of::() == 1, "LinuxOrMacOSOrWindows alignment mismatch"); +#[cfg(target_pointer_width = "32")] +const _: () = assert!(core::mem::offset_of!(LinuxOrMacOSOrWindows, tag) == 1, "LinuxOrMacOSOrWindows tag offset mismatch"); + +/// Tag discriminant for OsStr. +#[repr(u8)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum OsStrTag { + UnixBytes = 0, + Utf8 = 1, + WindowsU16s = 2, +} + +#[repr(C)] +#[derive(Clone, Copy)] +pub union OsStrPayload { + pub unix_bytes: core::mem::ManuallyDrop>, + pub utf8: core::mem::ManuallyDrop, + pub windows_u16s: core::mem::ManuallyDrop>, +} + +#[cfg(target_pointer_width = "32")] +#[repr(align(4))] +#[derive(Clone, Copy)] +pub struct OsStrPayloadAlignment; + +/// Tag union: OsStr +#[cfg(target_pointer_width = "32")] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct OsStr { + pub _payload_alignment: [OsStrPayloadAlignment; 0], + pub payload: [u8; 12], + pub tag: OsStrTag, +} + +/// Tag union: OsStr +#[cfg(not(target_pointer_width = "32"))] +#[repr(C)] +#[derive(Clone, Copy)] +pub struct OsStr { + pub payload: OsStrPayload, + pub tag: OsStrTag, +} + +impl OsStr { + /// Borrow the `UnixBytes` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::UnixBytes` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_unix_bytes_unchecked(&self) -> &RocListWith { + unsafe { &*(self.payload.as_ptr() as *const RocListWith) } + } + + /// Borrow the `UnixBytes` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::UnixBytes` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_unix_bytes_unchecked(&self) -> &RocListWith { + unsafe { &*(&self.payload.unix_bytes as *const core::mem::ManuallyDrop> as *const RocListWith) } + } + + /// Move the `UnixBytes` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::UnixBytes`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_unix_bytes_unchecked(&mut self) -> RocListWith { + unsafe { core::ptr::read(self.payload.as_ptr() as *const RocListWith) } + } + + /// Move the `UnixBytes` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::UnixBytes`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_unix_bytes_unchecked(&mut self) -> RocListWith { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.unix_bytes) } + } + + /// Borrow the `Utf8` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::Utf8` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_utf8_unchecked(&self) -> &RocStr { + unsafe { &*(self.payload.as_ptr() as *const RocStr) } + } + + /// Borrow the `Utf8` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::Utf8` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_utf8_unchecked(&self) -> &RocStr { + unsafe { &*(&self.payload.utf8 as *const core::mem::ManuallyDrop as *const RocStr) } + } + + /// Move the `Utf8` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::Utf8`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(target_pointer_width = "32")] + pub unsafe fn take_payload_utf8_unchecked(&mut self) -> RocStr { + unsafe { core::ptr::read(self.payload.as_ptr() as *const RocStr) } + } + + /// Move the `Utf8` payload out of one owned tag-union shell. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::Utf8`. After this call, `self` is logically uninitialized and must not be read or destroyed. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn take_payload_utf8_unchecked(&mut self) -> RocStr { + unsafe { core::mem::ManuallyDrop::take(&mut self.payload.utf8) } + } + + /// Borrow the `WindowsU16s` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::WindowsU16s` and the payload must still be initialized. + #[cfg(target_pointer_width = "32")] + pub unsafe fn borrow_payload_windows_u16s_unchecked(&self) -> &RocListWith { + unsafe { &*(self.payload.as_ptr() as *const RocListWith) } + } + + /// Borrow the `WindowsU16s` payload without creating another owner. + /// + /// # Safety + /// `self.tag` must be `OsStrTag::WindowsU16s` and the payload must still be initialized. + #[cfg(not(target_pointer_width = "32"))] + pub unsafe fn borrow_payload_windows_u16s_unchecked(&self) -> &RocListWith { + unsafe { &*(&self.payload.windows_u16s as *const core::mem::ManuallyDrop> as *const RocListWith) } } /// Move the `WindowsU16s` payload out of one owned tag-union shell. @@ -14934,6 +15424,37 @@ pub struct HostHttpCloseBodyArgs { pub arg0: *mut u64, } +/// Arguments for Host.http_pending_close! +/// Roc signature: Resource.Handle([HttpPendingKind]) => {} +/// Refcounted fields are owned by the hosted function. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct HostHttpPendingCloseArgs { + pub arg0: *mut u64, +} + +/// Arguments for Host.http_pending_follow! +/// Roc signature: Resource.Handle([HttpPendingKind]), { body : List(U8), headers : List((Str, Str)), method : U8, method_ext : Str, timeout_ms : U64, uri : Str }, U16, Str => {} +/// Refcounted fields are owned by the hosted function. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct HostHttpPendingFollowArgs { + pub arg0: *mut u64, + pub arg1: AnonStruct8bbc5017d7a8cb36, + pub arg2: u16, + pub arg3: RocStr, +} + +/// Arguments for Host.http_pending_wait! +/// Roc signature: Resource.Handle([HttpPendingKind]), [At(U64), NoTimeout] => Try([NotReady, Ready({ redirects : List((U16, Str)), response : { body : Resource.Handle([HttpResponseKind]), content_length : [Known(U64), Unknown], headers : List((Str, Str)), invalid_location : [None, Some(U8)], status : U16 } })], [Closed, HttpErr([BadBody, Interrupted, NetworkError, Other(List(U8)), Timeout])]) +/// Refcounted fields are owned by the hosted function. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct HostHttpPendingWaitArgs { + pub arg0: *mut u64, + pub arg1: AtOrNoTimeout, +} + /// Arguments for Host.http_read_body! /// Roc signature: Resource.Handle([HttpResponseKind]), [Chunk, ToEnd(U64)] => Try(List(U8), [EndOfBody, HttpErr([BadBody, Interrupted, NetworkError, Other(List(U8)), Timeout]), ReadLimitExceeded]) /// Refcounted fields are owned by the hosted function. @@ -15029,6 +15550,17 @@ pub struct HostHttpSendStreamingArgs { pub arg2: AtOrNoTimeout, } +/// Arguments for Host.http_start! +/// Roc signature: { body : List(U8), headers : List((Str, Str)), method : U8, method_ext : Str, timeout_ms : U64, uri : Str }, [After(U64), NoTimeout], [At(U64), NoTimeout] => Try(Resource.Handle([HttpPendingKind]), List(U8)) +/// Refcounted fields are owned by the hosted function. +#[repr(C)] +#[derive(Clone, Copy)] +pub struct HostHttpStartArgs { + pub arg0: AnonStruct8bbc5017d7a8cb36, + pub arg1: AfterOrNoTimeout, + pub arg2: AtOrNoTimeout, +} + /// Arguments for Host.path_absolute! /// Roc signature: [UnixBytes(List(U8)), Utf8(Str), WindowsU16s(List(U16))] => Try([UnixBytes(List(U8)), Utf8(Str), WindowsU16s(List(U16))], IOErr) /// Refcounted fields are owned by the hosted function. @@ -15689,6 +16221,25 @@ pub type HostFileWriterWriteBytesResultTag = HostFileAppendBytesResultTag; pub type HostFileWriterWriteUtf8Result = HostFileAppendBytesResult; pub type HostFileWriterWriteUtf8ResultPayload = HostFileAppendBytesResultPayload; pub type HostFileWriterWriteUtf8ResultTag = HostFileAppendBytesResultTag; +pub type HostHttpPendingFollowArg1 = AnonStruct8bbc5017d7a8cb36; +pub type HostHttpPendingFollowArg1Headers = AnonStruct77eaba63dfee299d; +pub type HostHttpPendingWaitErr = ClosedOrHttpErr; +pub type HostHttpPendingWaitErrPayload = ClosedOrHttpErrPayload; +pub type HostHttpPendingWaitErrTag = ClosedOrHttpErrTag; +pub type HostHttpPendingWaitErrHttpErr = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout; +pub type HostHttpPendingWaitErrHttpErrPayload = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeoutPayload; +pub type HostHttpPendingWaitErrHttpErrTag = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeoutTag; +pub type HostHttpPendingWaitOk = NotReadyOrReady; +pub type HostHttpPendingWaitOkPayload = NotReadyOrReadyPayload; +pub type HostHttpPendingWaitOkTag = NotReadyOrReadyTag; +pub type HostHttpPendingWaitOkReady = AnonStruct57b3856c59c615c8; +pub type ClosedOrHttpErrHttpErr = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeout; +pub type ClosedOrHttpErrHttpErrPayload = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeoutPayload; +pub type ClosedOrHttpErrHttpErrTag = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeoutTag; +pub type NotReadyOrReadyReady = AnonStruct57b3856c59c615c8; +pub type NotReadyOrReadyReadyResponse = AnonStruct60cbed700549ed13; +pub type NotReadyOrReadyReadyResponseHeaders = AnonStruct77eaba63dfee299d; +pub type NotReadyOrReadyReadyRedirects = AnonStruct453a8dc75d4311da; pub type HostHttpReadBodyErr = EndOfBodyOrHttpErrOrReadLimitExceeded; pub type HostHttpReadBodyErrPayload = EndOfBodyOrHttpErrOrReadLimitExceededPayload; pub type HostHttpReadBodyErrTag = EndOfBodyOrHttpErrOrReadLimitExceededTag; @@ -15712,6 +16263,8 @@ pub type HostHttpSendStreamingErrPayload = BadBodyOrInterruptedOrNetworkErrorOrO pub type HostHttpSendStreamingErrTag = BadBodyOrInterruptedOrNetworkErrorOrOtherOrTimeoutTag; pub type HostHttpSendStreamingOk = AnonStruct60cbed700549ed13; pub type HostHttpSendStreamingOkHeaders = AnonStruct77eaba63dfee299d; +pub type HostHttpStartArg0 = AnonStruct8bbc5017d7a8cb36; +pub type HostHttpStartArg0Headers = AnonStruct77eaba63dfee299d; pub type HostPathAbsoluteResult = HostEnvCreateTempDirResult; pub type HostPathAbsoluteResultPayload = HostEnvCreateTempDirResultPayload; pub type HostPathAbsoluteResultTag = HostEnvCreateTempDirResultTag; @@ -21623,6 +22176,250 @@ unsafe impl RocRelease for ClosedOrUnixSocketErrRelease { } } +impl HostHttpStartResult { + /// Recursively decrement Roc-owned payloads. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted payload. + pub unsafe fn decref(self, roc_host: &RocHost) { + let mut value = self; + let _ = roc_host; + match value.tag { + HostHttpStartResultTag::Err => { + let payload = unsafe { value.take_payload_err_unchecked() }; + unsafe { payload.decref(roc_host); } + }, + HostHttpStartResultTag::Ok => { + let payload = unsafe { value.take_payload_ok_unchecked() }; + unsafe { decref_box_with(payload as RocBox, core::mem::align_of::(), false, None, roc_host); } + }, + } + } + + /// Increment Roc-owned payloads. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + let _ = amount; + match value.tag { + HostHttpStartResultTag::Err => { + let payload = unsafe { core::ptr::read(value.borrow_payload_err_unchecked()) }; + unsafe { payload.incref(amount); } + }, + HostHttpStartResultTag::Ok => { + let payload = unsafe { core::ptr::read(value.borrow_payload_ok_unchecked()) }; + unsafe { incref_box(payload as RocBox, amount); } + }, + } + } +} + +pub struct HostHttpStartResultRelease; + +unsafe impl RocRelease for HostHttpStartResultRelease { + unsafe fn release(value: HostHttpStartResult, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + +impl HostHttpPendingWaitResult { + /// Recursively decrement Roc-owned payloads. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted payload. + pub unsafe fn decref(self, roc_host: &RocHost) { + let mut value = self; + let _ = roc_host; + match value.tag { + HostHttpPendingWaitResultTag::Err => { + let payload = unsafe { value.take_payload_err_unchecked() }; + unsafe { payload.decref(roc_host); } + }, + HostHttpPendingWaitResultTag::Ok => { + let payload = unsafe { value.take_payload_ok_unchecked() }; + unsafe { payload.decref(roc_host); } + }, + } + } + + /// Increment Roc-owned payloads. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + let _ = amount; + match value.tag { + HostHttpPendingWaitResultTag::Err => { + let payload = unsafe { core::ptr::read(value.borrow_payload_err_unchecked()) }; + unsafe { payload.incref(amount); } + }, + HostHttpPendingWaitResultTag::Ok => { + let payload = unsafe { core::ptr::read(value.borrow_payload_ok_unchecked()) }; + unsafe { payload.incref(amount); } + }, + } + } +} + +pub struct HostHttpPendingWaitResultRelease; + +unsafe impl RocRelease for HostHttpPendingWaitResultRelease { + unsafe fn release(value: HostHttpPendingWaitResult, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + +impl ClosedOrHttpErr { + /// Recursively decrement Roc-owned payloads. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted payload. + pub unsafe fn decref(self, roc_host: &RocHost) { + let mut value = self; + let _ = roc_host; + match value.tag { + ClosedOrHttpErrTag::Closed => {}, + ClosedOrHttpErrTag::HttpErr => { + let payload = unsafe { value.take_payload_http_err_unchecked() }; + unsafe { payload.decref(roc_host); } + }, + } + } + + /// Increment Roc-owned payloads. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + let _ = amount; + match value.tag { + ClosedOrHttpErrTag::Closed => {}, + ClosedOrHttpErrTag::HttpErr => { + let payload = unsafe { core::ptr::read(value.borrow_payload_http_err_unchecked()) }; + unsafe { payload.incref(amount); } + }, + } + } +} + +pub struct ClosedOrHttpErrRelease; + +unsafe impl RocRelease for ClosedOrHttpErrRelease { + unsafe fn release(value: ClosedOrHttpErr, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + +impl NotReadyOrReady { + /// Recursively decrement Roc-owned payloads. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted payload. + pub unsafe fn decref(self, roc_host: &RocHost) { + let mut value = self; + let _ = roc_host; + match value.tag { + NotReadyOrReadyTag::NotReady => {}, + NotReadyOrReadyTag::Ready => { + let payload = unsafe { value.take_payload_ready_unchecked() }; + unsafe { payload.decref(roc_host); } + }, + } + } + + /// Increment Roc-owned payloads. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + let _ = amount; + match value.tag { + NotReadyOrReadyTag::NotReady => {}, + NotReadyOrReadyTag::Ready => { + let payload = unsafe { core::ptr::read(value.borrow_payload_ready_unchecked()) }; + unsafe { payload.incref(amount); } + }, + } + } +} + +pub struct NotReadyOrReadyRelease; + +unsafe impl RocRelease for NotReadyOrReadyRelease { + unsafe fn release(value: NotReadyOrReady, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + +impl AnonStruct57b3856c59c615c8 { + /// Recursively decrement Roc-owned fields. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted field. + pub unsafe fn decref(self, roc_host: &RocHost) { + let value = self; + unsafe { value.response.decref(roc_host); } + unsafe { decref_list_of_anon_struct453a8dc75d4311da(value.redirects, roc_host); } + } + + /// Increment Roc-owned fields. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + unsafe { value.response.incref(amount); } + unsafe { value.redirects.incref(amount); } + } +} + +pub struct AnonStruct57b3856c59c615c8Release; + +unsafe impl RocRelease for AnonStruct57b3856c59c615c8Release { + unsafe fn release(value: AnonStruct57b3856c59c615c8, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + +impl AnonStruct453a8dc75d4311da { + /// Recursively decrement Roc-owned fields. + /// + /// # Safety + /// `self` must own one live Roc reference for each refcounted field. + pub unsafe fn decref(self, roc_host: &RocHost) { + let value = self; + unsafe { value._1.decref(roc_host); } + } + + /// Increment Roc-owned fields. + /// + /// # Safety + /// `self` must point at live Roc allocations. The retained references must + /// be balanced by later decrefs. + pub unsafe fn incref(self, amount: isize) { + let value = self; + unsafe { value._1.incref(amount); } + } +} + +pub struct AnonStruct453a8dc75d4311daRelease; + +unsafe impl RocRelease for AnonStruct453a8dc75d4311daRelease { + unsafe fn release(value: AnonStruct453a8dc75d4311da, roc_host: &RocHost) { + unsafe { value.decref(roc_host); } + } +} + impl SystemTarget { /// Recursively decrement Roc-owned fields. /// @@ -21956,6 +22753,17 @@ pub unsafe fn decref_list_of_anon_struct69eee2ff6c448fed(value: RocList(roc_host); } } +/// Release one owned reference to a `RocList`. +/// +/// The allocation's final reference is claimed atomically before any element +/// is read, so concurrent owners cannot skip or duplicate element teardown. +/// +/// # Safety +/// `value` must own one live Roc list reference. +pub unsafe fn decref_list_of_anon_struct453a8dc75d4311da(value: RocList, roc_host: &RocHost) { + unsafe { value.release_with::(roc_host); } +} + /// Release one owned reference to a `RocList`. /// /// The allocation's final reference is claimed atomically before any element @@ -22445,6 +23253,31 @@ unsafe extern "C" { /// unsafe { decref_box_with(arg0 as RocBox, core::mem::align_of::(), false, None, roc_host); } pub fn hosted_http_close_body(arg0: *mut u64); + /// Hosted symbol for Host.http_pending_close! + /// Roc signature: Resource.Handle([HttpPendingKind]) => {} + /// Owned arguments. Release each exactly once before returning, unless it is + /// moved into storage or into the result: + /// unsafe { decref_box_with(arg0 as RocBox, core::mem::align_of::(), false, None, roc_host); } + pub fn hosted_http_pending_close(arg0: *mut u64); + + /// Hosted symbol for Host.http_pending_follow! + /// Roc signature: Resource.Handle([HttpPendingKind]), { body : List(U8), headers : List((Str, Str)), method : U8, method_ext : Str, timeout_ms : U64, uri : Str }, U16, Str => {} + /// Owned arguments. Release each exactly once before returning, unless it is + /// moved into storage or into the result: + /// unsafe { decref_box_with(arg0 as RocBox, core::mem::align_of::(), false, None, roc_host); } + /// unsafe { arg1.decref(roc_host); } + /// unsafe { arg3.decref(roc_host); } + pub fn hosted_http_pending_follow(arg0: *mut u64, arg1: AnonStruct8bbc5017d7a8cb36, arg2: u16, arg3: RocStr); + + /// Hosted symbol for Host.http_pending_wait! + /// Roc signature: Resource.Handle([HttpPendingKind]), [At(U64), NoTimeout] => Try([NotReady, Ready({ redirects : List((U16, Str)), response : { body : Resource.Handle([HttpResponseKind]), content_length : [Known(U64), Unknown], headers : List((Str, Str)), invalid_location : [None, Some(U8)], status : U16 } })], [Closed, HttpErr([BadBody, Interrupted, NetworkError, Other(List(U8)), Timeout])]) + /// Owned arguments. Release each exactly once before returning, unless it is + /// moved into storage or into the result: + /// unsafe { decref_box_with(arg0 as RocBox, core::mem::align_of::(), false, None, roc_host); } + /// unsafe { arg1.decref(roc_host); } + /// The result is owned by Roc: return exactly one owned reference. + pub fn hosted_http_pending_wait(arg0: *mut u64, arg1: AtOrNoTimeout) -> HostHttpPendingWaitResult; + /// Hosted symbol for Host.http_read_body! /// Roc signature: Resource.Handle([HttpResponseKind]), [Chunk, ToEnd(U64)] => Try(List(U8), [EndOfBody, HttpErr([BadBody, Interrupted, NetworkError, Other(List(U8)), Timeout]), ReadLimitExceeded]) /// Owned arguments. Release each exactly once before returning, unless it is @@ -22472,6 +23305,16 @@ unsafe extern "C" { /// The result is owned by Roc: return exactly one owned reference. pub fn hosted_http_send_streaming(arg0: AnonStruct8bbc5017d7a8cb36, arg1: AfterOrNoTimeout, arg2: AtOrNoTimeout) -> HostHttpSendStreamingResult; + /// Hosted symbol for Host.http_start! + /// Roc signature: { body : List(U8), headers : List((Str, Str)), method : U8, method_ext : Str, timeout_ms : U64, uri : Str }, [After(U64), NoTimeout], [At(U64), NoTimeout] => Try(Resource.Handle([HttpPendingKind]), List(U8)) + /// Owned arguments. Release each exactly once before returning, unless it is + /// moved into storage or into the result: + /// unsafe { arg0.decref(roc_host); } + /// unsafe { arg1.decref(roc_host); } + /// unsafe { arg2.decref(roc_host); } + /// The result is owned by Roc: return exactly one owned reference. + pub fn hosted_http_start(arg0: AnonStruct8bbc5017d7a8cb36, arg1: AfterOrNoTimeout, arg2: AtOrNoTimeout) -> HostHttpStartResult; + /// Hosted symbol for Host.locale_all! /// Roc signature: {} => List(Str) /// The result is owned by Roc: return exactly one owned reference. From 92d632a85e9b263f197566d723ae7541d581fa0e Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Wed, 11 Nov 2026 12:39:53 +0000 Subject: [PATCH 3/4] Add Http.start! to keep a request in flight `Http.start!` sends a request and returns an `Http.PendingResponse` without waiting for it. `wait_within!` waits up to a given duration for the status and headers and returns `NotReady` without cancelling the request; `wait!` waits as `send_streaming!` would. Both return the same `StreamingResponse` and errors as `send_streaming!`, and follow redirects with the same rules and one deadline across every hop. `close!`, or releasing the last reference, cancels the request. There is no background thread, so a pending request makes progress only during HTTP calls and waits; the documentation of `start!` says so. Co-Authored-By: Claude Opus 5.5 --- README.md | 10 ++- platform/Http.roc | 182 +++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 189 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 436cc1ec..c8e95a3e 100644 --- a/README.md +++ b/README.md @@ -50,7 +50,15 @@ These host behaviors may affect application design: leave the `Request` at `NoTimeout`. After an idle timeout the response can still be read; the `Request` deadline passing is final. Each streaming response has its own connection, closed when the body ends or fails, or when - the response and its streams are released. Request bodies are not streamed. + the response and its streams (and, for `Http.start!`, its pending request) + are released. Request bodies are not streamed. +- `Http.start!` keeps a request in flight while the app does other work, + including other requests; `pending.wait_within!(duration)` waits for its + response for up to `duration`. There is no background thread: requests make + progress only during HTTP calls and waits. Both timeouts count from + `start!`, so a slow server needs a larger `idle_timeout`; a response that + arrived in time is not lost to a later wait. Releasing the last reference to + the pending request, or `pending.close!()`, cancels it. - HTTP redirects are opt-in through trailing options, `{ max_redirects: n }`; the default is zero. The request deadline covers every hop and the final body. Cross-origin hops strip credentials; HTTPS downgrades are refused. diff --git a/platform/Http.roc b/platform/Http.roc index 0bc786c3..f09d228f 100644 --- a/platform/Http.roc +++ b/platform/Http.roc @@ -17,6 +17,8 @@ import http.Response ## memory. [Http.send_streaming!] returns as soon as the status and headers ## have arrived and hands the body over as a `Stream` of chunks, for a ## download larger than memory or one whose progress is shown. +## [Http.start!] sends a request without waiting for it, so that other work, +## including other requests, can go on while it is in flight. ## ## Requests default to `User-Agent: roc-basic-cli/0.26`. An explicit ## `User-Agent` is preserved, regardless of the header name's casing. @@ -131,13 +133,17 @@ Http :: [].{ decode_json_response(response) } - ## Options for [Http.send_streaming!]. `{}` selects the defaults. + ## Options for [Http.send_streaming!] and [Http.start!]. `{}` selects the + ## defaults. ## ## `idle_timeout` bounds each single wait: for the status and headers, and ## then for each part of the body. A server that is silent for that long ## has stalled, however much it has already sent. It defaults to 30 ## seconds; `After(Duration.zero)` expires at once, before anything is - ## sent, and `NoTimeout` waits for as long as the server takes. + ## sent, and `NoTimeout` waits for as long as the server takes. For + ## [Http.start!], the wait for the status and headers counts from + ## `start!`, or from the redirect that led to the hop, rather than from + ## the call that waits; see [Http.start!]. ## ## The timeout of the `Request` keeps its meaning: it is the deadline for ## the whole exchange, body included, and `NoTimeout` (its default) and @@ -288,6 +294,133 @@ Http :: [].{ } } } + + ## Start sending a request and return without waiting for its response. + ## Wait for the response with [PendingResponse.wait_within!] or + ## [PendingResponse.wait!]. Meanwhile the app can do other work, + ## including other requests and other pending requests. + ## + ## The request is validated, built and sent as [Http.send_streaming!] + ## does it, with the same [StreamingOptions], and the waits return the + ## same [StreamingResponse] and the same errors. Besides an invalid URL, + ## `start!` fails only when the host cannot hold another request; every + ## failure of the request itself, including one to build it, is returned + ## by the waits. `start!` does not wait, so a pending signal does not + ## stop it. + ## + ## **When a request makes progress.** basic-cli runs no background + ## thread. A pending request is sent, and its response received, only + ## while the app is inside an HTTP call: `start!` itself, [Http.send!] and + ## the functions built on it, [Http.send_streaming!], a body read, and + ## above all [PendingResponse.wait_within!] and [PendingResponse.wait!]. In + ## between, for example during `Sleep.millis!`, a command or file I/O, + ## the operating system buffers what arrives, up to a limit. A request + ## with a large body, or a server that sends a lot before its status + ## line, can stall until the next such call. Wait with + ## [PendingResponse.wait_within!] instead of sleeping to keep requests + ## moving. + ## + ## Both timeouts count from `start!`, also while the request is not + ## making progress. The timeout of the `Request` stays the deadline for + ## the whole exchange, across redirects and the body. `idle_timeout` + ## bounds the time each hop has for its status and headers, counted from + ## `start!` or from the redirect that led to the hop, and then each wait + ## for part of the body. Status and headers that the host received in + ## time are returned by a later wait, however late it comes; the body + ## then has the timeouts of any [StreamingResponse]. For a server that + ## takes long to answer, set a larger `idle_timeout`, or `NoTimeout`. + ## + ## ```roc + ## commit = Http.start!(commit_request, { idle_timeout: NoTimeout })? + ## while True { + ## match commit.wait_within!(Duration.seconds(1))? { + ## Ready(response) => { + ## report!(response)? + ## break + ## } + ## NotReady => show_progress!(Http.send!(progress_request, {})?)? + ## } + ## } + ## ``` + start! : Request, StreamingOptions => Try(PendingResponse, [InvalidUrl(Url.ParseErr), HttpErr(TransportErr)]) + start! = |request, options| { + url = Url.parse(Request.uri(request)) ? InvalidUrl + canonical = Url.without_fragment(url) + canonical_request = request_with_defaults(request.with_uri(canonical.to_str())) + hop = canonical_request.with_headers(sanitize_headers(Request.headers(canonical_request), False, False)) + match Host.http_start!(InternalHttp.to_host_request(hop), host_idle_timeout(options.idle_timeout), request_deadline!(request)) { + Ok(host) => Ok(PendingResponse.{ host, request: canonical_request, url: canonical, max_redirects: options.max_redirects }) + Err(message) => Err(HttpErr(Other(message))) + } + } + + ## A request started with [Http.start!] whose response may not have + ## arrived yet. + ## + ## Copies of a pending response share one request. Once its response has + ## arrived, every wait returns that same [StreamingResponse], and all of + ## them read its body from one cursor. Until the pending response is + ## released it keeps the request, body included, and the connection of + ## a response that has arrived. + ## + ## The request is cancelled and its connection closed when the last + ## reference to the pending response is released, or earlier by + ## [close!]. A response already returned stays readable until it, too, + ## is released. + PendingResponse :: { host : Resource.HttpPending, request : Request, url : Url.Url, max_redirects : U64 }.{ + + ## Render the pending response without exposing its host handle. + to_inspect : PendingResponse -> Str + to_inspect = |_| "Http.PendingResponse()" + + ## Check on the request, waiting up to `wait` for its status and + ## headers: `Ready(response)` once they have arrived, otherwise + ## `NotReady`, which leaves the request in flight. + ## + ## `Duration.zero` returns at once, with a response whose status and + ## headers the host has already received. A redirect that has arrived + ## then only starts the next hop, so the result is `NotReady` until + ## the response to that hop has arrived too. A wait also moves every + ## other pending request along (see [Http.start!]), so a loop of + ## `wait_within!` calls needs no separate sleep. Redirects whose + ## response arrives during a longer wait are followed within it. + ## + ## Either timeout of the request expiring is `Err(HttpErr(Timeout))`, + ## and every later wait returns it again. A caught pending signal is + ## `Err(Interrupted)` and keeps the request; acknowledge the signal + ## before waiting again. After [close!] this is `Err(Closed)`. + wait_within! : PendingResponse, Duration => Try([Ready(StreamingResponse), NotReady], [Closed, Interrupted, HttpErr(TransportErr)]) + wait_within! = |pending, wait| + match wait_pending!(pending, At(Host.monotonic_now!().plus_saturated(wait.to_nanos()))) { + Ok(Ready(response)) => Ok(Ready(response)) + Ok(NotReady) => Ok(NotReady) + Err(Closed) => Err(Closed) + Err(Interrupted) => Err(Interrupted) + Err(HttpErr(err)) => Err(HttpErr(err)) + } + + ## Wait for the status and headers, as [Http.send_streaming!] would + ## have, bounded only by the timeouts of the request. Errors are those + ## of [wait_within!]. + wait! : PendingResponse => Try(StreamingResponse, [Closed, Interrupted, HttpErr(TransportErr)]) + wait! = |pending| + match wait_pending!(pending, NoTimeout) { + Ok(Ready(response)) => Ok(response) + # Cannot happen: a wait without a limit of its own ends only + # with the response or an error. + Ok(NotReady) => wait!(pending) + Err(Closed) => Err(Closed) + Err(Interrupted) => Err(Interrupted) + Err(HttpErr(err)) => Err(HttpErr(err)) + } + + ## Cancel the request if it is still in flight and close its + ## connection, for every copy of the pending response. A response + ## already returned is not affected. Afterwards [wait_within!] and [wait!] + ## return `Err(Closed)`. Closing again does nothing. + close! : PendingResponse => {} + close! = |pending| Host.http_pending_close!(pending.host) + } } # `{}` asks for the default idle timeout; a field given replaces it. @@ -379,6 +512,51 @@ redirect_step! = |raw, request, url, max_redirects, remaining| { Ok(Follow({ request: redirected_request(request, url, next_url, status), url: next_url, status, location })) } +## The request, URL and remaining redirects of the hop a pending response is +## on: its first request with each redirect the host has followed applied +## again, in order. Each passed [redirect_step!] when it was followed, so +## applying it again gives the same hop. +current_hop : Http.PendingResponse, List((U16, Str)) -> Try({ request : Request, url : Url.Url, remaining : U64 }, [HttpErr(Http.TransportErr)]) +current_hop = |pending, followed| { + var $hop = { request: pending.request, url: pending.url, remaining: pending.max_redirects } + for (status, location) in followed { + next_url = redirect_url($hop.url, location) ? |error| HttpErr(InvalidRedirectUrl(error)) + $hop = { request: redirected_request($hop.request, $hop.url, next_url, status), url: next_url, remaining: $hop.remaining - 1 } + } + Ok($hop) +} + +## Wait until `until` for the final response of a pending request, sending +## each redirect hop as soon as its response arrives. One `until` covers +## every hop. +wait_pending! : Http.PendingResponse, [NoTimeout, At(U64)] => Try([Ready(Http.StreamingResponse), NotReady], [Closed, Interrupted, HttpErr(Http.TransportErr)]) +wait_pending! = |pending, until| { + ready = match Host.http_pending_wait!(pending.host, until) { + Ok(Ready(value)) => value + Ok(NotReady) => { + return Ok(NotReady) + } + Err(Closed) => { + return Err(Closed) + } + Err(HttpErr(error)) => { + return Err(transport_error(error)) + } + } + hop = match ready.redirects { + [] => { request: pending.request, url: pending.url, remaining: pending.max_redirects } + followed => current_hop(pending, followed)? + } + match redirect_step!(ready.response, hop.request, hop.url, pending.max_redirects, hop.remaining)? { + Final(response) => Ok(Ready(response)) + Follow(next) => { + sent = next.request.with_headers(sanitize_headers(Request.headers(next.request), False, False)) + Host.http_pending_follow!(pending.host, InternalHttp.to_host_request(sent), next.status, next.location) + wait_pending!(pending, until) + } + } +} + ## HTTP permits network-path references; Url.resolve deliberately does not. redirect_url : Url.Url, Str -> Try(Url.Url, Url.ParseErr) redirect_url = |base, location| { From c9e7d25ddf9a5c58a9853add2cd9aeb43fa657af Mon Sep 17 00:00:00 2001 From: Karakatiza666 Date: Wed, 11 Nov 2026 13:34:51 +0000 Subject: [PATCH 4/4] Test in-flight HTTP requests Rust tests in src/http.rs cover Http.start! on the host side: not ready and then ready, a zero wait, other HTTP calls while a request is in flight, interruption, drop and close ending the connection, a returned response outliving its pending request, the request timeout counting from the start, a head that arrived in time outliving the timeouts, the idle timeout, transport errors, one deadline across a redirect hop, and the same bytes on the wire as send_streaming. The test server gets a /delay/ route. tests/http-in-flight.roc runs the Roc API against it, and examples/http-pending.roc shows the pattern of waiting on a long request while checking its progress with other requests. Both have entries in scripts/test_spec.json. Co-Authored-By: Claude Opus 5.5 --- ci/rust_http_server/src/main.rs | 11 + examples/http-pending.roc | 64 ++++ scripts/test_spec.json | 69 +++++ src/http.rs | 529 ++++++++++++++++++++++++++++++++ tests/http-in-flight.roc | 228 ++++++++++++++ 5 files changed, 901 insertions(+) create mode 100644 examples/http-pending.roc create mode 100644 tests/http-in-flight.roc diff --git a/ci/rust_http_server/src/main.rs b/ci/rust_http_server/src/main.rs index af727ed3..a92d96c8 100644 --- a/ci/rust_http_server/src/main.rs +++ b/ci/rust_http_server/src/main.rs @@ -86,6 +86,17 @@ async fn handle_request(req: Request) -> Result, Gen if req.uri().path().starts_with("/redirect/") { return handle_redirect(req).await; } + // Answers only after the given number of milliseconds, as a server does + // with a long-running request. Used by tests/http-in-flight.roc and + // examples/http-pending.roc. + if let Some(ms) = req.uri().path().strip_prefix("/delay/") { + let ms: u64 = ms.parse()?; + tokio::time::sleep(Duration::from_millis(ms)).await; + return Ok(Response::builder() + .status(StatusCode::OK) + .header("Content-Type", "text/plain; charset=utf-8") + .body(full(format!("waited {ms} ms")))?); + } let response = match (req.method(), req.uri().path()) { (&Method::GET, "/utf8test") => { // UTF-8 encoded "Hello utf8" diff --git a/examples/http-pending.roc b/examples/http-pending.roc new file mode 100644 index 00000000..cead5883 --- /dev/null +++ b/examples/http-pending.roc @@ -0,0 +1,64 @@ +## Keep a slow request in flight while other requests report on its progress. +app [main!] { + http: "https://github.com/roc-lang/http/releases/download/1.0.0/6ZUwqYhCS8PU9Mo6MF7oV82ET2o7KYb57CLKDq4cq4sS.tar.zst", + pf: platform "https://github.com/roc-lang/basic-cli/releases/download/0.27.0/9yjxqdGTP4xSpSEmDazwTmRsehfRZzA19pwEKkR6fj1g.tar.zst", +} + +import pf.Duration +import pf.Http +import pf.OsStr +import pf.Stdout +import http.Request + +# To run this example, first start the test server in another terminal: +# +# cd ci/rust_http_server && cargo run --release +# +# then: +# +# roc build examples/http-pending.roc +# ./examples/http-pending +main! : List(OsStr) => Try({}, _) +main! = |_args| { + # The server answers only after two seconds, as a server does with a long + # commit. That is longer than a response normally takes, so the request + # must not keep the default 30 second `idle_timeout`, which would end the + # wait for a commit that takes minutes. `NoTimeout` leaves it to the + # `Request` timeout, which is off by default. + commit_request = Request.from_method(POST).with_uri("http://127.0.0.1:9000/delay/2000") + + # `start!` sends the request and returns at once, with a handle to the + # response that is still to come. + commit = Http.start!(commit_request, { idle_timeout: NoTimeout }) ? |err| CommitFailed(err) + + # Each `wait_within!` waits up to half a second for the commit. While it + # waits it also moves the commit along, so the loop needs no `Sleep`. When + # the commit has not finished by then, ask for its progress and wait again. + var $progress_checks = 0.U64 + while True { + match commit.wait_within!(Duration.millis(500)) { + Ok(Ready(response)) => { + # The commit's own response: status and headers are here, the + # body is read as with `Http.send_streaming!`. + body = response.read_to_end!(NoLimit) ? |err| CommitFailed(err) + Stdout.line!("Checked progress while the commit ran: ${Str.inspect($progress_checks >= 2)}")? + Stdout.line!("Commit finished with status ${response.status().to_str()}: ${Str.from_utf8_lossy(body)}")? + break + } + Ok(NotReady) => { + # Any other request can be sent now. The commit stays in flight. + progress = Http.get_utf8!("http://127.0.0.1:9000/utf8test", {}) ? |err| ProgressFailed(err) + $progress_checks = $progress_checks + 1 + Stdout.line!("Progress ${$progress_checks.to_str()}: ${progress}")? + } + Err(err) => { + return Err(CommitFailed(err)) + } + } + } + + # To give up on a commit that is still running, call `commit.close!()`: the + # request is cancelled and its connection closed. To wait for it without + # checking on anything else, call `commit.wait!()`. + Ok({}) +} diff --git a/scripts/test_spec.json b/scripts/test_spec.json index 7003551d..6708365a 100644 --- a/scripts/test_spec.json +++ b/scripts/test_spec.json @@ -362,6 +362,25 @@ } ] }, + { + "path": "examples/http-pending.roc", + "cases": [ + { + "name": "server-available", + "helper": "http", + "timeout": 15, + "live_handles": 0, + "stdout_contains": ["Checked progress while the commit ran: True\nCommit finished with status 200: waited 2000 ms\n"], + "valgrind": false + }, + { + "name": "connection-refused", + "live_handles": 0, + "exit_code": 1, + "contains": ["CommitFailed(HttpErr(Other("] + } + ] + }, { "path": "examples/http-simple.roc", "cases": [ @@ -908,6 +927,56 @@ } ] }, + { + "path": "tests/http-in-flight.roc", + "cases": [ + { + "name": "in-flight-requests", + "args": ["in-flight"], + "helper": "http", + "timeout": 20, + "temp_cwd": true, + "live_handles": 0, + "stdout_contains": ["waiting while other requests are sent: ok\nthe response is returned again: ok\na zero wait returns what has arrived: ok\n"] + }, + { + "name": "redirects-and-one-deadline", + "args": ["redirects"], + "helper": "http", + "timeout": 30, + "temp_cwd": true, + "live_handles": 0, + "stdout_contains": ["following redirects: ok\nthe redirect limit: ok\nsecurity checks: ok\nredirects across waits and one deadline: ok\n"] + }, + { + "name": "start-does-not-wait-and-the-timeout-counts-from-start", + "args": ["timing"], + "helper": "http", + "timeout": 20, + "temp_cwd": true, + "live_handles": 0, + "stdout_contains": ["start! returns at once: ok\nthe timeout counts from start!: ok\na response that arrived in time is kept: ok\n"], + "valgrind": false + }, + { + "name": "close-and-errors", + "args": ["close-and-errors"], + "helper": "http", + "timeout": 20, + "temp_cwd": true, + "live_handles": 0, + "stdout_contains": ["close! cancels: ok\nfailures come back from the waits: ok\nerror rows: ok\n"] + }, + { + "name": "extra-argument-is-the-callers-error", + "args": ["close-and-errors", "unexpected"], + "temp_cwd": true, + "live_handles": 0, + "exit_code": 1, + "contains": ["Program exited with error: UnexpectedArguments"] + } + ] + }, { "path": "tests/http-redirects.roc", "cases": [ diff --git a/src/http.rs b/src/http.rs index b64d9993..a84f1fc4 100644 --- a/src/http.rs +++ b/src/http.rs @@ -2228,6 +2228,535 @@ mod tests { unsafe { args.decref(&host) }; } + // ---- Pending requests --------------------------------------------------- + + fn get_request(uri: &str) -> HyperRequest { + hyper::Request::builder() + .uri(uri) + .body(http_body_util::Full::default()) + .unwrap() + } + + fn start_get( + uri: &str, + idle_timeout: Option, + deadline: Option, + ) -> PendingResponse { + let mut pending = PendingResponse::new(idle_timeout, deadline); + pending.send(Ok(get_request(uri))); + pending + } + + /// The head and body reference of a wait that must have found a response. + fn ready(waited: Result) -> (StreamingHead, *mut u64) { + match waited { + Ok(Waited::Ready(head, body)) => (head, body), + Ok(Waited::NotReady) => panic!("the response was not ready"), + Err(error) => panic!("the wait failed: {error:?}"), + } + } + + /// A wait that is generous enough for a local response on a loaded machine. + fn within_seconds(seconds: u64) -> Option { + Some(Instant::now() + Duration::from_secs(seconds)) + } + + /// Wait without waiting until the head is there. Tests run in parallel and + /// share one runtime, which only the thread inside `block_on` turns, so one + /// attempt may not reach the task. (A single thread, as in a Roc app, does + /// on the first; tests/http-in-flight.roc checks that.) + fn zero_waits_until_ready(pending: &mut PendingResponse) -> (StreamingHead, *mut u64) { + for _ in 0..100 { + match pending.wait(Some(Instant::now())) { + Ok(Waited::NotReady) => thread::sleep(Duration::from_millis(20)), + waited => return ready(waited), + } + } + panic!("a response that had arrived was not returned in 2 s of zero waits"); + } + + fn read_all(body: *mut u64) -> Vec { + unsafe { crate::resources::resource_ref::(body) } + .read_to_end(u64::MAX) + .unwrap() + .to_vec() + } + + const OK_HELLO: &[u8] = b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello"; + + #[test] + fn a_pending_request_is_not_ready_until_its_head_arrives() { + let (release_tx, release_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + release_rx.recv().unwrap(); + stream.write_all(OK_HELLO).unwrap(); + }); + let mut pending = start_get(&uri, None, None); + // Waiting without a response is not an error, and it can be repeated. + for _ in 0..2 { + let started = Instant::now(); + let waited = pending.wait(Some(started + Duration::from_millis(100))); + assert!(matches!(waited, Ok(Waited::NotReady))); + assert!(started.elapsed() >= Duration::from_millis(100)); + assert!(matches!(pending.state, PendingState::Sending(_))); + } + release_tx.send(()).unwrap(); + let (head, body) = ready(pending.wait(within_seconds(5))); + assert_eq!(head.status, 200); + assert_eq!(&read_all(body)[..], b"hello"); + // Another wait hands over the same response: the same body, whose + // cursor is at its end. + let (again_head, again) = ready(pending.wait(Some(Instant::now()))); + assert_eq!(again_head.status, 200); + assert_eq!(again, body); + assert_eq!(&read_all(again)[..], b""); + let host = crate::resources::make_host(); + crate::resources::release(body, &host); + crate::resources::release(again, &host); + drop(pending); + server.join().unwrap(); + } + + #[test] + fn a_zero_wait_returns_a_head_that_has_arrived() { + let (written_tx, written_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + stream + .write_all(b"HTTP/1.1 204 No Content\r\n\r\n") + .unwrap(); + written_tx.send(()).unwrap(); + thread::sleep(Duration::from_millis(300)); + }); + let mut pending = start_get(&uri, None, None); + // The answer is in the socket, though nothing has read it yet. + written_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + // A wait that does not wait still turns the runtime, so this finds it. + let (head, body) = zero_waits_until_ready(&mut pending); + assert_eq!(head.status, 204); + crate::resources::release(body, &crate::resources::make_host()); + drop(pending); + server.join().unwrap(); + } + + #[test] + fn starting_a_request_sends_it_without_a_wait() { + let (sent_tx, sent_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + sent_tx.send(()).unwrap(); + let _ = peer_closed(&mut stream); + }); + // The request reaches the server although the app makes no HTTP call + // after `start`; here, only the test thread could have turned the + // runtime, and it waits for the server instead. + let pending = start_get(&uri, None, None); + assert_eq!(sent_rx.recv_timeout(Duration::from_secs(10)), Ok(())); + drop(pending); + server.join().unwrap(); + } + + #[test] + fn other_http_calls_drive_and_do_not_block_a_pending_request() { + let host = crate::resources::make_host(); + let (release_tx, release_rx) = mpsc::channel::<()>(); + let (written_tx, written_rx) = mpsc::channel::<()>(); + let (uri_a, server_a) = local_server(move |mut stream| { + release_rx.recv().unwrap(); + stream.write_all(OK_HELLO).unwrap(); + written_tx.send(()).unwrap(); + }); + let mut slow = start_get(&uri_a, None, None); + assert!(matches!( + slow.wait(Some(Instant::now() + Duration::from_millis(100))), + Ok(Waited::NotReady) + )); + + // A plain request completes while the first one is in flight ... + let (uri_b, server_b) = local_server(|mut stream| stream.write_all(OK_HELLO).unwrap()); + let result = get(&uri_b, 5_000, &host); + assert!(result.tag == HttpResultTag::Ok); + unsafe { result.decref(&host) }; + server_b.join().unwrap(); + + // ... and so does a second pending request. + let (uri_c, server_c) = local_server(|mut stream| stream.write_all(OK_HELLO).unwrap()); + let mut fast = start_get(&uri_c, None, None); + let (head, body) = ready(fast.wait(within_seconds(5))); + assert_eq!(head.status, 200); + crate::resources::release(body, &host); + server_c.join().unwrap(); + assert!(matches!(slow.state, PendingState::Sending(_))); + + // The held request goes on while the app is in other HTTP calls, so a + // wait that does not wait finds it ready. + release_tx.send(()).unwrap(); + written_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + let (uri_d, server_d) = local_server(|mut stream| stream.write_all(OK_HELLO).unwrap()); + let result = get(&uri_d, 5_000, &host); + assert!(result.tag == HttpResultTag::Ok); + unsafe { result.decref(&host) }; + server_d.join().unwrap(); + let (head, body) = zero_waits_until_ready(&mut slow); + assert_eq!(head.status, 200); + assert_eq!(&read_all(body)[..], b"hello"); + crate::resources::release(body, &host); + server_a.join().unwrap(); + } + + #[test] + fn an_interrupted_wait_keeps_the_request() { + struct Reset; + impl Drop for Reset { + fn drop(&mut self) { + crate::signals::set_test_pending(false); + } + } + let (release_tx, release_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + release_rx.recv().unwrap(); + stream.write_all(OK_HELLO).unwrap(); + }); + let mut pending = start_get(&uri, None, None); + let reset = Reset; + crate::signals::set_test_pending(true); + for _ in 0..2 { + assert_eq!( + pending.wait(within_seconds(5)).err(), + Some(PendingErr::Failed(Failure::Interrupted)) + ); + assert!(matches!(pending.state, PendingState::Sending(_))); + } + drop(reset); + // As after a timeout, the request is still in flight. + release_tx.send(()).unwrap(); + let (head, body) = ready(pending.wait(within_seconds(5))); + assert_eq!(head.status, 200); + assert_eq!(&read_all(body)[..], b"hello"); + crate::resources::release(body, &crate::resources::make_host()); + drop(pending); + server.join().unwrap(); + } + + // Each way of giving a request up while it is in flight must close its + // connection without another hosted call: dropping the resource, releasing + // the last reference to its handle through the allocator, and `close!`. + #[test] + fn dropping_or_closing_a_pending_request_closes_its_connection() { + for mode in ["drop", "handle", "close"] { + let (closed_tx, closed_rx) = mpsc::channel::(); + let (uri, server) = local_server(move |mut stream| { + closed_tx.send(peer_closed(&mut stream)).unwrap(); + }); + let mut pending = start_get(&uri, None, None); + assert!(matches!( + pending.wait(Some(Instant::now() + Duration::from_millis(100))), + Ok(Waited::NotReady) + )); + match mode { + "drop" => drop(pending), + "handle" => { + let host = crate::resources::make_host(); + let handle = crate::resources::box_resource(pending, &host).unwrap(); + crate::resources::release(handle, &host); + } + _ => { + pending.release(); + assert_eq!(pending.wait(None).err(), Some(PendingErr::Closed)); + assert_eq!(pending.wait(None).err(), Some(PendingErr::Closed)); + } + } + assert_eq!( + closed_rx.recv_timeout(Duration::from_secs(10)), + Ok(true), + "the connection stayed open ({mode})" + ); + server.join().unwrap(); + } + } + + #[test] + fn giving_up_a_pending_request_leaves_a_returned_response_readable() { + for close in [false, true] { + let (resume_tx, resume_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + stream + .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhe") + .unwrap(); + resume_rx.recv().unwrap(); + stream.write_all(b"llo").unwrap(); + }); + let mut pending = start_get(&uri, None, None); + let (head, body) = ready(pending.wait(within_seconds(5))); + assert_eq!(head.status, 200); + if close { + pending.release(); + } else { + drop(pending); + } + resume_tx.send(()).unwrap(); + assert_eq!(&read_all(body)[..], b"hello"); + crate::resources::release(body, &crate::resources::make_host()); + server.join().unwrap(); + } + } + + #[test] + fn the_request_timeout_counts_from_start() { + let (uri, server) = local_server(|mut stream| { + let _ = peer_closed(&mut stream); + }); + // The deadline passes while the app is not in any HTTP call. + let mut pending = start_get( + &uri, + None, + Some(Instant::now() + Duration::from_millis(300)), + ); + thread::sleep(Duration::from_millis(500)); + let started = Instant::now(); + assert_eq!( + pending.wait(within_seconds(5)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert!(started.elapsed() < Duration::from_secs(2)); + // The failure is final. + assert_eq!( + pending.wait(within_seconds(5)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert!(matches!( + pending.state, + PendingState::Failed(Failure::Timeout) + )); + drop(pending); + server.join().unwrap(); + + // A wait with a limit of its own longer than the deadline reports the + // deadline's timeout, not "not ready". + let (uri, server) = local_server(|mut stream| { + let _ = peer_closed(&mut stream); + }); + let mut pending = start_get( + &uri, + None, + Some(Instant::now() + Duration::from_millis(300)), + ); + let started = Instant::now(); + assert_eq!( + pending.wait(within_seconds(30)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert!(started.elapsed() < Duration::from_secs(10)); + drop(pending); + server.join().unwrap(); + + // A head that comes only after the deadline is a timeout, although + // the wait finds it there. + let (written_tx, written_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + thread::sleep(Duration::from_millis(500)); + stream.write_all(OK_HELLO).unwrap(); + written_tx.send(()).unwrap(); + let _ = peer_closed(&mut stream); + }); + let mut pending = start_get( + &uri, + None, + Some(Instant::now() + Duration::from_millis(300)), + ); + written_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + assert_eq!( + pending.wait(within_seconds(5)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + drop(pending); + server.join().unwrap(); + } + + #[test] + fn a_head_that_arrived_in_time_outlives_the_timeouts() { + let host = crate::resources::make_host(); + let (written_tx, written_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |mut stream| { + stream.write_all(OK_HELLO).unwrap(); + written_tx.send(()).unwrap(); + let _ = peer_closed(&mut stream); + }); + let mut pending = start_get(&uri, Some(Duration::from_secs(1)), None); + let head_until = pending.head_until.unwrap(); + written_rx.recv_timeout(Duration::from_secs(10)).unwrap(); + // Other HTTP calls turn the runtime, and the task reads the head. + while !matches!(&pending.state, PendingState::Sending(task) if task.is_finished()) { + assert!(Instant::now() < head_until, "the head was not read in time"); + let (other_uri, other) = local_server(|mut stream| stream.write_all(OK_HELLO).unwrap()); + let result = get(&other_uri, 5_000, &host); + assert!(result.tag == HttpResultTag::Ok); + unsafe { result.decref(&host) }; + other.join().unwrap(); + } + // The app does other work until the idle timeout has passed, and + // only then waits. + thread::sleep(head_until.saturating_duration_since(Instant::now())); + thread::sleep(Duration::from_millis(100)); + let (head, body) = ready(pending.wait(Some(Instant::now()))); + assert_eq!(head.status, 200); + assert_eq!(&read_all(body)[..], b"hello"); + crate::resources::release(body, &host); + drop(pending); + server.join().unwrap(); + } + + #[test] + fn the_idle_timeout_bounds_the_head_of_a_pending_request() { + let (release_tx, release_rx) = mpsc::channel::<()>(); + let (uri, server) = local_server(move |_stream| { + let _ = release_rx.recv(); + }); + let mut pending = start_get(&uri, Some(IDLE), None); + let started = Instant::now(); + assert_eq!( + pending.wait(within_seconds(30)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert!(started.elapsed() < Duration::from_secs(10)); + drop(pending); + drop(release_tx); + server.join().unwrap(); + + // A zero idle timeout expires before anything is sent. + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + listener.set_nonblocking(true).unwrap(); + let uri = format!("http://{}/", listener.local_addr().unwrap()); + let mut pending = start_get(&uri, Some(Duration::ZERO), None); + assert!(matches!( + pending.state, + PendingState::Failed(Failure::Timeout) + )); + assert_eq!( + pending.wait(within_seconds(5)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert_eq!( + listener.accept().unwrap_err().kind(), + std::io::ErrorKind::WouldBlock + ); + } + + #[test] + fn transport_errors_come_back_from_the_wait() { + let listener = TcpListener::bind(("127.0.0.1", 0)).unwrap(); + let uri = format!("http://{}/", listener.local_addr().unwrap()); + drop(listener); + let mut pending = start_get(&uri, None, None); + let first = pending.wait(within_seconds(10)); + let Err(PendingErr::Failed(Failure::Other(message))) = first else { + panic!("expected a transport error"); + }; + assert!(!message.is_empty()); + // The same failure, every time. + assert_eq!( + pending.wait(None).err(), + Some(PendingErr::Failed(Failure::Other(message))) + ); + + // A request that could not be built fails the same way. + let mut pending = PendingResponse::new(None, None); + pending.send(Err("invalid HTTP method".into())); + assert_eq!( + pending.wait(None).err(), + Some(PendingErr::Failed(Failure::Other( + "invalid HTTP method".into() + ))) + ); + } + + #[test] + fn a_redirect_hop_keeps_the_deadline() { + let (release_tx, release_rx) = mpsc::channel::<()>(); + let (second_uri, second) = local_server(move |_stream| { + let _ = release_rx.recv(); + }); + let location = format!("{second_uri}next"); + let (first_uri, first) = local_server(move |mut stream| { + stream + .write_all( + format!( + "HTTP/1.1 302 Found\r\nLocation: {location}\r\nContent-Length: 0\r\n\r\n" + ) + .as_bytes(), + ) + .unwrap(); + }); + let host = crate::resources::make_host(); + let started = Instant::now(); + let mut pending = start_get( + &first_uri, + None, + Some(started + Duration::from_millis(1200)), + ); + let (head, body) = ready(pending.wait(within_seconds(5))); + assert_eq!(head.status, 302); + crate::resources::release(body, &host); + // The first hop took almost none of the time; the next hop is not + // given a new allowance. + thread::sleep(Duration::from_millis(600)); + pending.follow( + 302, + "/next".to_string(), + Ok(get_request(&format!("{second_uri}next"))), + ); + assert_eq!(pending.redirects, [(302, "/next".to_string())]); + assert_eq!( + pending.wait(within_seconds(30)).err(), + Some(PendingErr::Failed(Failure::Timeout)) + ); + assert!( + started.elapsed() < Duration::from_millis(1800), + "the deadline restarted with the redirect" + ); + drop(pending); + drop(release_tx); + first.join().unwrap(); + second.join().unwrap(); + } + + #[test] + fn a_pending_request_sends_the_same_request_as_a_streaming_one() { + let host = crate::resources::make_host(); + let build = |uri: &str, headers: &[(&str, &str)]| { + let args = request_args(7, "", uri, headers, &host); + let request = build_hyper_request((&args).into()); + unsafe { args.decref(&host) }; + request + }; + for headers in [ + &[][..], + &[("Content-Type", "application/json"), ("X-A", "b")], + ] { + let streamed = captured_request(|uri| { + let (head, mut stream) = send_streaming( + build(uri, headers).unwrap(), + 5_000, + Some(Duration::from_secs(5)), + ) + .unwrap(); + assert_eq!(head.status, 200); + assert_eq!(stream.read_chunk(), Ok(None)); + }); + let pending = captured_request(|uri| { + let mut pending = PendingResponse::new( + Some(Duration::from_secs(5)), + Some(Instant::now() + Duration::from_secs(5)), + ); + pending.send(build(uri, headers)); + let (head, body) = ready(pending.wait(within_seconds(5))); + assert_eq!(head.status, 200); + crate::resources::release(body, &host); + }); + assert_eq!(pending, streamed); + assert!(pending.starts_with("POST /path?query=1 HTTP/1.1\r\n")); + assert!(pending.ends_with("\r\n\r\npayload")); + } + } + // Guards the shape of `TOKIO_RUNTIME`: a thread that used it must be able to // exit while the blocking pool still holds work. An owned runtime blocks in // `Drop` until every pool thread acknowledges the shutdown, which is the diff --git a/tests/http-in-flight.roc b/tests/http-in-flight.roc new file mode 100644 index 00000000..ae3900d6 --- /dev/null +++ b/tests/http-in-flight.roc @@ -0,0 +1,228 @@ +## Exercise `Http.start!` and `Http.PendingResponse` against +## ci/rust_http_server: a request in flight while others are sent, waits with +## and without a limit, redirects under one deadline, closing, and failures. The deadline +## and timing checks run in their own mode, as valgrind slows them. +## scripts/test.py checks that no handle is left open in any mode. +app [main!] { + http: "https://github.com/roc-lang/http/releases/download/1.0.0/6ZUwqYhCS8PU9Mo6MF7oV82ET2o7KYb57CLKDq4cq4sS.tar.zst", + pf: platform "../platform/main.roc", +} + +import pf.Duration +import pf.Http +import pf.Monotonic +import pf.OsStr +import pf.Sleep +import pf.Stdout +import http.Request + +check : Bool, Str -> Try({}, [CheckFailed(Str)]) +check = |condition, message| if (condition) Ok({}) else Err(CheckFailed(message)) + +main! : List(OsStr) => Try({}, _) +main! = |args| { + mode = match args { + [arg, ..] => OsStr.display(arg) + [] => "" + } + + match mode { + "in-flight" => in_flight!() + "redirects" => redirects!() + "timing" => timing!() + "close-and-errors" => close_and_errors!(args.len()) + _ => Err(UnknownMode(mode)) + } +} + +request : Str -> Request +request = |path| Request.from_method(GET).with_uri("http://127.0.0.1:9000${path}") + +text : List(U8) -> Str +text = |bytes| Str.from_utf8_lossy(bytes) + +## Whether a wait found the response. +is_ready : Try([Ready(Http.StreamingResponse), NotReady], _) -> Bool +is_ready = |result| + match result { + Ok(Ready(_)) => True + _ => False + } + +## Wait for the final response, then read its body. +wait_to_end! : Http.PendingResponse => Try(List(U8), _) +wait_to_end! = |pending| Ok(pending.wait!()?.read_to_end!(NoLimit)?) + +## Wait in small steps until the final response has arrived, then read its body. +wait_in_steps_to_end! : Http.PendingResponse => Try(List(U8), _) +wait_in_steps_to_end! = |pending| { + match pending.wait_within!(Duration.millis(100))? { + Ready(response) => Ok(response.read_to_end!(NoLimit)?) + NotReady => wait_in_steps_to_end!(pending) + } +} + +## What the app can do while a request is in flight, and what it gets back. +in_flight! : () => Try({}, _) +in_flight! = || { + # "Not ready yet" is an answer, not a failure, and the request goes on. + slow = Http.start!(request("/delay/1000"), {})? + var $waits = 0 + var $answer = "" + while $answer == "" { + match slow.wait_within!(Duration.millis(100))? { + Ready(response) => { + $answer = text(response.read_to_end!(NoLimit)?) + } + NotReady => { + $waits = $waits + 1 + # Other requests are served while the first one is in flight. + if $waits == 1 { + greeting = Http.get_utf8!("http://127.0.0.1:9000/utf8test", {})? + check(greeting == "Hello utf8", "a request while another is pending")? + second = Http.start!(request("/utf8test"), {})? + second_response = second.wait!()? + check(second_response.status() == 200 and text(second_response.read_to_end!(NoLimit)?) == "Hello utf8", "a second pending request")? + } + } + } + } + check($waits >= 1, "the response was not ready at first")? + check($answer == "waited 1000 ms", "the response of the pending request")? + Stdout.line!("waiting while other requests are sent: ok")? + + # After the response has arrived, a wait returns it again: the same + # body, read from one cursor. + pending = Http.start!(request("/stream/sized"), {})? + response = pending.wait!()? + first = match Stream.next!(response.chunks()) { + One({ item: Ok(chunk), .. }) => chunk + _ => { + return Err(ExpectedAChunk) + } + } + again = pending.wait!()? + check(again.status() == 200 and again.content_length() == Known(200_000), "the response again")? + rest = again.read_to_end!(NoLimit)? + check(first.len() + rest.len() == 200_000 and rest.len() > 0, "one body cursor")? + check(is_ready(pending.wait_within!(Duration.zero)), "a zero wait finds the response")? + Stdout.line!("the response is returned again: ok")? + + # A wait that does not wait returns a response that has arrived, although + # the app made no HTTP call since the request was started. Several times, + # as the host must turn its runtime more than once to read it. + for _ in [1, 2, 3] { + arrived = Http.start!(request("/utf8test"), {})? + Sleep.millis!(300)? + check(is_ready(arrived.wait_within!(Duration.zero)), "a zero wait finds a response that has arrived")? + } + Stdout.line!("a zero wait returns what has arrived: ok") +} + +## Redirects behave as in `Http.send_streaming!`, hop by hop across waits. +redirects! : () => Try({}, _) +redirects! = || { + chain = Http.start!(request("/redirect/chain/2"), { max_redirects: 2 })? + check(text(chain.wait!()?.read_to_end!(NoLimit)?) == "final", "exact redirect budget")? + unfollowed = Http.start!(request("/redirect/chain/2"), {})? + check(unfollowed.wait!()?.status() == 302, "redirects are opt-in")? + Stdout.line!("following redirects: ok")? + + looped = Http.start!(request("/redirect/loop"), { max_redirects: 2 })? + for _ in [1, 2] { + result = looped.wait!().map_ok(|response| response.status()) + check(result == Err(HttpErr(RedirectLimitExceeded(2))), "redirect budget enforced, every time")? + } + Stdout.line!("the redirect limit: ok")? + + cross = request("/redirect/302/cross").add_header("Authorization", "secret") + report = text(Http.start!(cross, { max_redirects: 1 })?.wait!()?.read_to_end!(NoLimit)?) + check(report.contains("authorization: \n") and report.contains("host: localhost:9000\n"), "credentials are not sent to another origin")? + Stdout.line!("security checks: ok")? + + # One deadline covers every hop and the body, however the waits are cut up. + for path in ["/redirect/delayed/1", "/redirect/delayed/2"] { + timed = request(path).with_timeout(TimeoutMilliseconds(650)) + waited = Http.start!(timed, { max_redirects: 2 })? + check(wait_to_end!(waited) == Err(HttpErr(Timeout)), "one deadline with wait!")? + stepped = Http.start!(timed, { max_redirects: 2 })? + check(wait_in_steps_to_end!(stepped) == Err(HttpErr(Timeout)), "one deadline with wait_within!")? + } + # Without a deadline the hops simply take their time. + patient = Http.start!(request("/redirect/delayed/2"), { max_redirects: 2 })? + check(text(wait_in_steps_to_end!(patient)?) == "final", "redirects followed across waits")? + Stdout.line!("redirects across waits and one deadline: ok") +} + +## Times: that `start!` does not wait, that the `Request` timeout counts +## from `start!`, and that a response that arrived in time is kept. +timing! : () => Try({}, _) +timing! = || { + began = Monotonic.now!() + pending = Http.start!(request("/delay/1000"), {})? + check(began.elapsed!() < Duration.millis(500), "start! does not wait for the response")? + check(is_ready(pending.wait_within!(Duration.millis(100))) == False, "wait_within! returned a response too early")? + Stdout.line!("start! returns at once: ok")? + + # The timeout of the request counts from `start!`, also while the app is + # not waiting for it. The late wait finds that it has expired. + late = Http.start!(request("/delay/1000").with_timeout(TimeoutMilliseconds(300)), {})? + Sleep.millis!(600)? + check(late.wait_within!(Duration.zero).map_ok(|_| {}) == Err(HttpErr(Timeout)), "the timeout counts from start!")? + check(late.wait!().map_ok(|_| {}) == Err(HttpErr(Timeout)), "a timeout is final")? + waited = Monotonic.now!() + early = Http.start!(request("/delay/1000").with_timeout(TimeoutMilliseconds(300)), {})? + check(early.wait!().map_ok(|_| {}) == Err(HttpErr(Timeout)), "the timeout ends an unbounded wait")? + check(waited.elapsed!() < Duration.millis(900), "the timeout ended the wait at 300 ms")? + Stdout.line!("the timeout counts from start!: ok")? + + # A response that arrived in time, here during another request, is not + # lost to a wait that comes only after its idle timeout. + arrived = Http.start!(request("/utf8test"), { idle_timeout: After(Duration.millis(300)) })? + other = Http.get_utf8!("http://127.0.0.1:9000/delay/100", {})? + check(other == "waited 100 ms", "another request")? + Sleep.millis!(600)? + check(arrived.wait!()?.status() == 200, "a response that arrived in time")? + Stdout.line!("a response that arrived in time is kept: ok") +} + +## Closing cancels, and failures come back from the waits. +close_and_errors! : U64 => Try({}, _) +close_and_errors! = |arg_count| { + # A second error beside the platform's, so each call below must widen. + if arg_count > 1 { + return Err(UnexpectedArguments) + } + closed = Http.start!(request("/delay/1000"), {})? + closed.close!() + closed.close!() + check(closed.wait_within!(Duration.zero).map_ok(|_| {}) == Err(Closed), "wait_within! after close!")? + check(closed.wait!().map_ok(|_| {}) == Err(Closed), "wait! after close!")? + Stdout.line!("close! cancels: ok")? + + # Nothing listens on port 9: the failure is a transport error with a message. + refused = Http.start!(Request.from_method(GET).with_uri("http://127.0.0.1:9/"), {})? + for _ in [1, 2] { + match refused.wait!() { + Err(HttpErr(Other(message))) => check(!(message.is_empty()), "the error has a message")? + _ => { + return Err(CheckFailed("a refused connection")) + } + } + } + bad_url = Http.start!(Request.from_method(GET).with_uri("ftp://example.com/"), {}).map_ok(|_| {}) + check(bad_url == Err(InvalidUrl(UnsupportedScheme("ftp"))), "an invalid URL")? + Stdout.line!("failures come back from the waits: ok")? + + # Functions as values beside another error: the rows have to widen. + start! = Http.start! + wait_within! = Http.PendingResponse.wait_within! + wait! = Http.PendingResponse.wait! + close! = Http.PendingResponse.close! + pending = start!(request("/utf8test"), {})? + waited_within = wait_within!(pending, Duration.seconds(10))? + check(is_ready(Ok(waited_within)), "wait_within! as a function value")? + check(wait!(pending)?.status() == 200, "wait! as a function value")? + close!(pending) + Stdout.line!("error rows: ok") +}