From 2b92efa64501320955979a92de39c70b6734f835 Mon Sep 17 00:00:00 2001 From: Matt Mastracci Date: Thu, 18 May 2023 20:10:25 -0600 Subject: [PATCH] feat(ext/http): Add support for trailers w/internal API (HTTP/2 only) (#19182) Necessary for #3326. Requested in #10214 as well. --- cli/tests/unit/serve_test.ts | 50 ++++++++++++++++++++++++++++++++++++ ext/http/00_serve.js | 8 ++++++ ext/http/http_next.rs | 16 ++++++++++++ ext/http/lib.rs | 1 + ext/http/response_body.rs | 17 +++++++++++- ext/http/slab.rs | 13 +++++++++- 6 files changed, 103 insertions(+), 2 deletions(-) diff --git a/cli/tests/unit/serve_test.ts b/cli/tests/unit/serve_test.ts index c6cfc45f3325f6..0f97e17b8549dd 100644 --- a/cli/tests/unit/serve_test.ts +++ b/cli/tests/unit/serve_test.ts @@ -15,6 +15,7 @@ import { const { upgradeHttpRaw, + addTrailers, // @ts-expect-error TypeScript (as of 3.7) does not support indexing namespaces by symbol } = Deno[Deno.internal]; @@ -2903,6 +2904,45 @@ Deno.test( }, ); +// TODO(mmastrac): This test should eventually use fetch, when we support trailers there. +// This test is ignored because it's flaky and relies on cURL's verbose output. +Deno.test( + { permissions: { net: true, run: true, read: true }, ignore: true }, + async function httpServerTrailers() { + const ac = new AbortController(); + const listeningPromise = deferred(); + + const server = Deno.serve({ + handler: () => { + const response = new Response("Hello World", { + headers: { + "trailer": "baz", + "transfer-encoding": "chunked", + "foo": "bar", + }, + }); + addTrailers(response, [["baz", "why"]]); + return response; + }, + port: 4501, + signal: ac.signal, + onListen: onListen(listeningPromise), + onError: createOnErrorCb(ac), + }); + + // We don't have a great way to access this right now, so just fetch the trailers with cURL + const [_, stderr] = await curlRequestWithStdErr([ + "http://localhost:4501/path", + "-v", + "--http2", + "--http2-prior-knowledge", + ]); + assertMatch(stderr, /baz: why/); + ac.abort(); + await server; + }, +); + Deno.test( { permissions: { net: true, run: true, read: true } }, async function httpsServeCurlH2C() { @@ -2948,3 +2988,13 @@ async function curlRequest(args: string[]) { assert(success); return new TextDecoder().decode(stdout); } + +async function curlRequestWithStdErr(args: string[]) { + const { success, stdout, stderr } = await new Deno.Command("curl", { + args, + stdout: "piped", + stderr: "piped", + }).output(); + assert(success); + return [new TextDecoder().decode(stdout), new TextDecoder().decode(stderr)]; +} diff --git a/ext/http/00_serve.js b/ext/http/00_serve.js index 9075ae6511557d..7186da1fedb06f 100644 --- a/ext/http/00_serve.js +++ b/ext/http/00_serve.js @@ -60,6 +60,7 @@ const { op_http_set_response_body_text, op_http_set_response_header, op_http_set_response_headers, + op_http_set_response_trailers, op_http_upgrade_raw, op_http_upgrade_websocket_next, op_http_wait, @@ -75,6 +76,7 @@ const { "op_http_set_response_body_text", "op_http_set_response_header", "op_http_set_response_headers", + "op_http_set_response_trailers", "op_http_upgrade_raw", "op_http_upgrade_websocket_next", "op_http_wait", @@ -125,6 +127,11 @@ function upgradeHttpRaw(req, conn) { throw new TypeError("upgradeHttpRaw may only be used with Deno.serve"); } +function addTrailers(resp, headerList) { + const inner = toInnerResponse(resp); + op_http_set_response_trailers(inner.slabId, headerList); +} + class InnerRequest { #slabId; #context; @@ -687,6 +694,7 @@ function serve(arg1, arg2) { return { finished }; } +internals.addTrailers = addTrailers; internals.upgradeHttpRaw = upgradeHttpRaw; export { serve, upgradeHttpRaw }; diff --git a/ext/http/http_next.rs b/ext/http/http_next.rs index 34281ee9210740..30a8c9d51d5ca0 100644 --- a/ext/http/http_next.rs +++ b/ext/http/http_next.rs @@ -318,6 +318,22 @@ pub fn op_http_set_response_headers( } } +#[op] +pub fn op_http_set_response_trailers( + slab_id: SlabId, + trailers: Vec<(ByteString, ByteString)>, +) { + let mut http = slab_get(slab_id); + let mut trailer_map: HeaderMap = HeaderMap::with_capacity(trailers.len()); + for (name, value) in trailers { + // These are valid latin-1 strings + let name = HeaderName::from_bytes(&name).unwrap(); + let value = HeaderValue::from_bytes(&value).unwrap(); + trailer_map.append(name, value); + } + *http.trailers().borrow_mut() = Some(trailer_map); +} + fn is_request_compressible(headers: &HeaderMap) -> Compression { let Some(accept_encoding) = headers.get(ACCEPT_ENCODING) else { return Compression::None; diff --git a/ext/http/lib.rs b/ext/http/lib.rs index 2660f46532859b..7d37c53e1c240e 100644 --- a/ext/http/lib.rs +++ b/ext/http/lib.rs @@ -116,6 +116,7 @@ deno_core::extension!( http_next::op_http_set_response_body_text, http_next::op_http_set_response_header, http_next::op_http_set_response_headers, + http_next::op_http_set_response_trailers, http_next::op_http_track, http_next::op_http_upgrade_websocket_next, http_next::op_http_upgrade_raw, diff --git a/ext/http/response_body.rs b/ext/http/response_body.rs index e30c917c30c487..ea6cc5ab84f2cc 100644 --- a/ext/http/response_body.rs +++ b/ext/http/response_body.rs @@ -158,7 +158,11 @@ impl std::fmt::Debug for ResponseBytesInner { /// required by hyper. As the API requires information about request completion (including a success/fail /// flag), we include a very lightweight [`CompletionHandle`] for interested parties to listen on. #[derive(Debug, Default)] -pub struct ResponseBytes(ResponseBytesInner, CompletionHandle); +pub struct ResponseBytes( + ResponseBytesInner, + CompletionHandle, + Rc>>, +); impl ResponseBytes { pub fn initialize(&mut self, inner: ResponseBytesInner) { @@ -170,6 +174,10 @@ impl ResponseBytes { self.1.clone() } + pub fn trailers(&self) -> Rc>> { + self.2.clone() + } + fn complete(&mut self, success: bool) -> ResponseBytesInner { if matches!(self.0, ResponseBytesInner::Done) { return ResponseBytesInner::Done; @@ -250,6 +258,9 @@ impl Body for ResponseBytes { let res = loop { let res = match &mut self.0 { ResponseBytesInner::Done | ResponseBytesInner::Empty => { + if let Some(trailers) = self.2.borrow_mut().take() { + return std::task::Poll::Ready(Some(Ok(Frame::trailers(trailers)))); + } unreachable!() } ResponseBytesInner::Bytes(..) => { @@ -271,6 +282,9 @@ impl Body for ResponseBytes { }; if matches!(res, ResponseStreamResult::EndOfStream) { + if let Some(trailers) = self.2.borrow_mut().take() { + return std::task::Poll::Ready(Some(Ok(Frame::trailers(trailers)))); + } self.complete(true); } std::task::Poll::Ready(res.into()) @@ -278,6 +292,7 @@ impl Body for ResponseBytes { fn is_end_stream(&self) -> bool { matches!(self.0, ResponseBytesInner::Done | ResponseBytesInner::Empty) + && self.2.borrow_mut().is_none() } fn size_hint(&self) -> SizeHint { diff --git a/ext/http/slab.rs b/ext/http/slab.rs index 24554d68999359..93a56e9ff31b13 100644 --- a/ext/http/slab.rs +++ b/ext/http/slab.rs @@ -4,6 +4,7 @@ use crate::response_body::CompletionHandle; use crate::response_body::ResponseBytes; use deno_core::error::AnyError; use http::request::Parts; +use http::HeaderMap; use hyper1::body::Incoming; use hyper1::upgrade::OnUpgrade; @@ -11,6 +12,7 @@ use slab::Slab; use std::cell::RefCell; use std::cell::RefMut; use std::ptr::NonNull; +use std::rc::Rc; pub type Request = hyper1::Request; pub type Response = hyper1::Response; @@ -23,6 +25,7 @@ pub struct HttpSlabRecord { // The response may get taken before we tear this down response: Option, promise: CompletionHandle, + trailers: Rc>>, been_dropped: bool, #[cfg(feature = "__zombie_http_tracking")] alive: bool, @@ -81,11 +84,14 @@ fn slab_insert_raw( ) -> SlabId { let index = SLAB.with(|slab| { let mut slab = slab.borrow_mut(); + let body = ResponseBytes::default(); + let trailers = body.trailers(); slab.insert(HttpSlabRecord { request_info, request_parts, request_body, - response: Some(Response::new(ResponseBytes::default())), + response: Some(Response::new(body)), + trailers, been_dropped: false, promise: CompletionHandle::default(), #[cfg(feature = "__zombie_http_tracking")] @@ -182,6 +188,11 @@ impl SlabEntry { self.self_mut().response.as_mut().unwrap() } + /// Get a mutable reference to the trailers. + pub fn trailers(&mut self) -> &RefCell> { + &self.self_mut().trailers + } + /// Take the response. pub fn take_response(&mut self) -> Response { self.self_mut().response.take().unwrap()