grpc_webnext_client/

tonic_service.rs

//! tonic interop: drive **generated** client stubs over the tunnel.
//!
//! The h2ts path is real HTTP/2, so nothing here translates a protocol — it hands
//! tonic a [`tower_service::Service`] that moves request bytes out and response
//! bytes, headers and trailers back. tonic keeps doing what it already does well
//! (codec, framing, status, compression, interceptors); this crate keeps doing what
//! tonic cannot do in a browser (dial, tunnel, reconnect). A stub generated by
//! `tonic-prost-build` runs unmodified:
//!
//! ```no_run
//! # mod pb { pub mod greeter_client { pub struct GreeterClient<T>(T);
//! #   impl<T> GreeterClient<T> { pub fn new(inner: T) -> Self { Self(inner) } } } }
//! # fn demo(client: grpc_webnext_client::Client) {
//! use pb::greeter_client::GreeterClient;
//!
//! let mut greeter = GreeterClient::new(client.into_tonic());
//! # let _ = greeter;
//! # }
//! ```
//!
//! ## The `Send` bound, honestly — and no, it does not mean threads
//!
//! tonic's generated stubs require `T::ResponseBody: Send + 'static`, and the engine
//! underneath is `!Send` on purpose — that is what a browser is, and asserting
//! otherwise with `unsafe impl` would be a lie the compiler can no longer check.
//! [`SendWrapper`] is the middle path: it carries the value across the bound and
//! records the thread that made it, so a value that really does move threads
//! **panics** at the boundary instead of racing. Natively, keep the client on a
//! `LocalSet`, which is where a `!Send` client belongs anyway.
//!
//! `Send` is a compile-time marker, not a runtime: requiring it links nothing and
//! spawns nothing. This feature brings **no threading** — a release build of the stub
//! path has zero atomic instructions, no shared memory, and no spawn symbols, and
//! `tokio` resolves with `sync` alone, so `tokio::spawn` (behind `rt`) is not even
//! compiled in. Plain `wasm32-unknown-unknown`, no `+atomics`, is the supported
//! configuration. Under wasm threads the rule is one client per worker: moving one
//! across workers panics at the wrapper, deliberately, rather than racing on `Rc`.
//!
//! ## Deadlines
//!
//! `grpc-timeout` on the outgoing request is also **enforced locally**, covering the
//! whole call — opening it and every frame of the response body. This is a
//! deliberate difference from tonic-over-`Channel`, where `Request::set_timeout`
//! sets the header and only an `Endpoint::timeout` layer enforces anything. The
//! header is a request to the far end, and a peer that ignores it leaves a tab
//! waiting forever; see [`CallOptions::timeout`](crate::CallOptions::timeout), which
//! makes the same promise on the native API.

use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use std::time::Duration;

use bytes::Bytes;
use futures::stream::{LocalBoxStream, StreamExt};
use h2ts_client::{RequestBody, Trailers};
use http::{HeaderMap, HeaderName, HeaderValue};
use http_body::{Body, Frame};
use send_wrapper::SendWrapper;

use crate::client::Client;

/// A [`Client`] wearing tonic's transport interface, so generated stubs can drive it.
///
/// Cheap to clone, and clones share one tunnel — the same contract [`Client`] has,
/// which is what makes this a stand-in for `tonic::transport::Channel`. Build one
/// with [`Client::into_tonic`].
#[derive(Clone)]
pub struct TonicService {
    client: Client,
}

impl std::fmt::Debug for TonicService {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("TonicService").finish_non_exhaustive()
    }
}

impl TonicService {
    pub fn new(client: Client) -> TonicService {
        TonicService { client }
    }

    /// The channel underneath, for the things tonic has no vocabulary for —
    /// [`Client::state`], [`Client::state_changes`], [`Client::is_closed`].
    pub fn client(&self) -> &Client {
        &self.client
    }
}

impl Client {
    /// Wrap this channel as a tonic transport, for generated client stubs.
    pub fn into_tonic(self) -> TonicService {
        TonicService::new(self)
    }
}

impl tower_service::Service<http::Request<tonic::body::Body>> for TonicService {
    type Response = http::Response<ResponseBody>;
    /// `tonic::Status` rather than this crate's, because tonic downcasts it back out
    /// of the boxed error — so a deadline arrives at the caller as DEADLINE_EXCEEDED
    /// instead of decaying to UNKNOWN with the text in the message.
    type Error = tonic::Status;
    /// Deliberately **not** `Send`, and deliberately not wrapped. tonic bounds the
    /// response *body* with `Send`, never the future — it awaits this inline inside
    /// `Grpc::streaming`, on whatever executor is polling, which here is one thread.
    /// Wrapping it would assert something no caller asks for and add a second place a
    /// cross-thread move could panic, so `ResponseBody` stays the single seam.
    type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>>>>;

    /// Always ready: there is nothing to reserve. The tunnel is dialed by the call
    /// itself and shared by every other, so readiness here would be reporting on a
    /// resource this service does not own.
    fn poll_ready(&mut self, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        Poll::Ready(Ok(()))
    }

    fn call(&mut self, request: http::Request<tonic::body::Body>) -> Self::Future {
        let client = self.client.clone();
        Box::pin(send(client, request))
    }
}

/// Issue one request and return the response headers plus a body tonic can read.
async fn send(
    client: Client,
    request: http::Request<tonic::body::Body>,
) -> Result<http::Response<ResponseBody>, tonic::Status> {
    let (parts, body) = request.into_parts();
    let path = parts.uri.path().to_string();
    let timeout = grpc_timeout(&parts.headers);
    let headers = to_wire_headers(&parts.headers)?;
    let body = RequestBody::stream(request_chunks(body));

    // The deadline covers the whole call, so the timer that bounds the open is the
    // same one handed to the body afterwards — `select` returns the loser untouched,
    // so it carries its *remaining* time rather than restarting per phase. It is
    // created once and never rechecked against a clock, because
    // `std::time::Instant::now()` panics on `wasm32-unknown-unknown`.
    let (response, deadline) = match timeout {
        None => (client.send(&path, headers, body).await?, None),
        Some(timeout) => {
            use futures::future::{select, Either};
            let mut timer = futures_timer::Delay::new(timeout);
            let opened = {
                let open = client.send(&path, headers, body);
                futures::pin_mut!(open);
                match select(open, &mut timer).await {
                    Either::Left((response, _)) => Some(response),
                    Either::Right(((), _)) => None,
                }
            };
            match opened {
                Some(response) => (response?, Some(timer)),
                None => return Err(tonic::Status::deadline_exceeded("deadline exceeded")),
            }
        }
    };

    let mut builder = http::Response::builder().status(response.status);
    // `raw_headers` rather than the collapsed map: gRPC metadata is multi-valued, and
    // a `HashMap<String, String>` keeps only one of a repeated key.
    for header in &response.raw_headers {
        // Pseudo-headers are HTTP/2's, not HTTP's — `:status` is already the status.
        if header.name.starts_with(':') {
            continue;
        }
        builder = builder.header(&header.name, &header.value);
    }
    let (body, trailers) = response.into_parts();
    builder
        .body(ResponseBody::new(body.boxed_local(), trailers, deadline))
        .map_err(|e| tonic::Status::internal(format!("malformed response headers: {e}")))
}

/// The request body as chunks h2ts can upload.
///
/// tonic has already framed each message (`[compressed flag | u32 len | bytes]`), so
/// this is a byte pass-through and not a re-encode. Request trailers are dropped: a
/// gRPC *client* never sends any, and half-close is the end of the stream.
fn request_chunks(body: tonic::body::Body) -> impl futures::Stream<Item = Vec<u8>> {
    futures::stream::unfold(Box::pin(body), |mut body| async move {
        loop {
            let frame = std::future::poll_fn(|cx| body.as_mut().poll_frame(cx)).await;
            match frame {
                Some(Ok(frame)) => match frame.into_data() {
                    Ok(data) => return Some((data.to_vec(), body)),
                    Err(_trailers) => continue,
                },
                // Ending the body here half-closes the stream mid-message, which the
                // server reports as a truncated request — the honest outcome, since
                // there is no way to signal "this body failed" over HTTP/2 except by
                // resetting, and that would lose the server's own status.
                Some(Err(_)) | None => return None,
            }
        }
    })
}

/// The request's headers, as h2ts wants them.
///
/// tonic has already set `content-type`, `te`, `user-agent` and the compression
/// headers, and sanitized the connection-specific ones HTTP/2 forbids, so this is a
/// conversion and not a policy.
fn to_wire_headers(headers: &HeaderMap) -> Result<Vec<(String, String)>, tonic::Status> {
    headers
        .iter()
        .map(|(name, value)| {
            // gRPC restricts ASCII metadata to printable ASCII and requires `-bin`
            // values to be base64, so anything else is already off-spec — and h2ts
            // header values are `String`. Refusing beats sending mojibake.
            let value = value.to_str().map_err(|_| {
                tonic::Status::internal(format!(
                    "metadata value for `{name}` is not valid ASCII; \
                     binary metadata must use a `-bin` key"
                ))
            })?;
            Ok((name.as_str().to_string(), value.to_string()))
        })
        .collect()
}

/// Parse `grpc-timeout` (a positive integer plus a unit) into a duration. Anything
/// unparseable is `None`: a malformed header means the call is unbounded locally, not
/// that it fails, since the value is the peer's to enforce in the first place.
fn grpc_timeout(headers: &HeaderMap) -> Option<Duration> {
    let raw = headers.get("grpc-timeout")?.to_str().ok()?;
    let (digits, unit) = raw.split_at_checked(raw.len().checked_sub(1)?)?;
    let value: u64 = digits.parse().ok()?;
    Some(match unit {
        "n" => Duration::from_nanos(value),
        "u" => Duration::from_micros(value),
        "m" => Duration::from_millis(value),
        "S" => Duration::from_secs(value),
        "M" => Duration::from_secs(value.checked_mul(60)?),
        "H" => Duration::from_secs(value.checked_mul(3600)?),
        _ => return None,
    })
}

fn to_header_map(headers: std::collections::HashMap<String, String>) -> HeaderMap {
    let mut map = HeaderMap::with_capacity(headers.len());
    for (name, value) in headers {
        // A header the `http` crate rejects cannot be represented; dropping it beats
        // failing the call, since the status itself is what tonic reads out of here
        // and a malformed *other* trailer should not bury it.
        if let (Ok(name), Ok(value)) =
            (HeaderName::try_from(name), HeaderValue::try_from(value))
        {
            map.append(name, value);
        }
    }
    map
}

/// The response body, as an [`http_body::Body`] tonic can decode.
///
/// Data frames pass through untouched; the terminal trailers become the final frame,
/// which is where tonic reads `grpc-status` from. Backpressure is preserved for free:
/// the window is replenished only as this is polled, so tonic's own consumption rate
/// is what throttles the server.
pub struct ResponseBody(SendWrapper<Inner>);

struct Inner {
    body: LocalBoxStream<'static, Result<Vec<u8>, h2ts_client::H2Error>>,
    trailers: Trailers,
    /// The call's deadline, still running from before the response opened.
    deadline: Option<futures_timer::Delay>,
    ended: bool,
}

impl std::fmt::Debug for ResponseBody {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ResponseBody").field("ended", &self.0.ended).finish_non_exhaustive()
    }
}

impl ResponseBody {
    fn new(
        body: LocalBoxStream<'static, Result<Vec<u8>, h2ts_client::H2Error>>,
        trailers: Trailers,
        deadline: Option<futures_timer::Delay>,
    ) -> ResponseBody {
        ResponseBody(SendWrapper::new(Inner { body, trailers, deadline, ended: false }))
    }
}

impl Body for ResponseBody {
    type Data = Bytes;
    type Error = tonic::Status;

    fn poll_frame(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, Self::Error>>> {
        // Deref through the wrapper: this is the thread check, so a body that
        // travelled panics here rather than touching `Rc`s from two threads.
        let inner = &mut *self.get_mut().0;
        if inner.ended {
            return Poll::Ready(None);
        }
        // The deadline is checked before the body, because a server that sends
        // headers promptly and then stalls is exactly the case it exists for.
        if let Some(timer) = inner.deadline.as_mut() {
            if Pin::new(timer).poll(cx).is_ready() {
                inner.ended = true;
                return Poll::Ready(Some(Err(tonic::Status::deadline_exceeded(
                    "deadline exceeded",
                ))));
            }
        }
        match inner.body.poll_next_unpin(cx) {
            Poll::Ready(Some(Ok(chunk))) => Poll::Ready(Some(Ok(Frame::data(Bytes::from(chunk))))),
            Poll::Ready(Some(Err(e))) => {
                inner.ended = true;
                Poll::Ready(Some(Err(tonic::Status::unavailable(format!(
                    "stream failed: {e}"
                )))))
            }
            Poll::Ready(None) => {
                inner.ended = true;
                // The status lives in the trailers. A body that ends without any is a
                // protocol violation, and tonic says so — better it than a guess here.
                match inner.trailers.get() {
                    Some(trailers) => {
                        Poll::Ready(Some(Ok(Frame::trailers(to_header_map(trailers)))))
                    }
                    None => Poll::Ready(None),
                }
            }
            Poll::Pending => Poll::Pending,
        }
    }

    fn is_end_stream(&self) -> bool {
        self.0.ended
    }
}

impl From<crate::Status> for tonic::Status {
    fn from(status: crate::Status) -> tonic::Status {
        let mut headers = HeaderMap::new();
        for (name, value) in status.metadata.to_headers() {
            if let (Ok(name), Ok(value)) =
                (HeaderName::try_from(name), HeaderValue::try_from(value))
            {
                headers.append(name, value);
            }
        }
        tonic::Status::with_metadata(
            tonic::Code::from_i32(status.code as i32),
            status.message,
            tonic::metadata::MetadataMap::from_headers(headers),
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn headers(pairs: &[(&str, &str)]) -> HeaderMap {
        let mut map = HeaderMap::new();
        for (name, value) in pairs {
            map.append(
                HeaderName::try_from(*name).unwrap(),
                HeaderValue::try_from(*value).unwrap(),
            );
        }
        map
    }

    #[test]
    fn every_grpc_timeout_unit_is_understood() {
        // tonic picks the most precise unit that fits in 8 digits, so `n` and `u` are
        // the ones it actually emits — but a peer may send any of them, and a unit we
        // silently misread would be a deadline off by three orders of magnitude.
        for (raw, expected) in [
            ("100n", Duration::from_nanos(100)),
            ("100u", Duration::from_micros(100)),
            ("100m", Duration::from_millis(100)),
            ("2S", Duration::from_secs(2)),
            ("2M", Duration::from_secs(120)),
            ("2H", Duration::from_secs(7200)),
        ] {
            assert_eq!(grpc_timeout(&headers(&[("grpc-timeout", raw)])), Some(expected), "{raw}");
        }
    }

    #[test]
    fn a_missing_or_malformed_timeout_leaves_the_call_unbounded() {
        // Unbounded rather than failed: the header is advisory to begin with, so a
        // value this client cannot read is the peer's problem, not the call's.
        assert_eq!(grpc_timeout(&headers(&[])), None);
        for raw in ["", "m", "100", "100x", "-1S", "abcS"] {
            assert_eq!(grpc_timeout(&headers(&[("grpc-timeout", raw)])), None, "{raw:?}");
        }
    }

    #[test]
    fn tonics_own_encoding_round_trips() {
        // tonic writes `Request::set_timeout` as microseconds up to 8 digits; the pair
        // has to agree or every deadline set through a generated stub is wrong.
        assert_eq!(
            grpc_timeout(&headers(&[("grpc-timeout", "500000u")])),
            Some(Duration::from_millis(500))
        );
    }

    #[test]
    fn headers_convert_and_binary_metadata_survives_as_base64() {
        let mut metadata = tonic::metadata::MetadataMap::new();
        metadata.insert("x-request-id", "abc-123".parse().unwrap());
        metadata.insert_bin(
            "x-trace-bin",
            tonic::metadata::MetadataValue::from_bytes(&[0, 1, 250]),
        );
        let wire = to_wire_headers(&metadata.into_headers()).unwrap();

        assert!(wire.contains(&("x-request-id".to_string(), "abc-123".to_string())));
        // tonic base64s a `-bin` value on the way in, so it is already ASCII here —
        // this crate must not encode it a second time.
        let (_, encoded) = wire.iter().find(|(k, _)| k == "x-trace-bin").expect("the -bin key");
        use base64::Engine as _;
        assert_eq!(
            base64::engine::general_purpose::STANDARD_NO_PAD.decode(encoded).unwrap(),
            vec![0, 1, 250]
        );
    }

    #[test]
    fn a_non_ascii_metadata_value_is_refused_rather_than_mangled() {
        let mut map = HeaderMap::new();
        map.append("x-bad", HeaderValue::from_bytes(&[0xff, 0xfe]).unwrap());
        let error = to_wire_headers(&map).expect_err("not representable on the wire");
        assert_eq!(error.code(), tonic::Code::Internal);
        assert!(error.message().contains("x-bad"), "unhelpful message: {}", error.message());
    }

    #[test]
    fn a_status_carries_its_code_message_and_metadata_into_tonic() {
        let mut metadata = crate::Metadata::new();
        metadata.insert("x-detail", "quota-exhausted");
        metadata.insert_bin("x-detail-bin", vec![0, 1, 250]);
        let status = tonic::Status::from(crate::Status {
            code: crate::Code::FailedPrecondition,
            message: "no".into(),
            metadata,
        });

        assert_eq!(status.code(), tonic::Code::FailedPrecondition);
        assert_eq!(status.message(), "no");
        assert_eq!(status.metadata().get("x-detail").unwrap(), "quota-exhausted");
        assert_eq!(
            status.metadata().get_bin("x-detail-bin").unwrap().to_bytes().unwrap().as_ref(),
            &[0, 1, 250]
        );
    }
}