tonic-web 0.14.9

/*
 *
 * Copyright 2025 gRPC authors.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to
 * deal in the Software without restriction, including without limitation the
 * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
 * sell copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
 * IN THE SOFTWARE.
 *
 */

use std::fmt;
use std::pin::Pin;
use std::task::{Context, Poll, ready};

use base64::Engine as _;
use bytes::{Buf, BufMut, Bytes, BytesMut};
use http::{HeaderMap, HeaderName, HeaderValue, header};
use http_body::{Body, Frame, SizeHint};
use pin_project::pin_project;
use tokio_stream::Stream;
use tonic::Status;

use self::content_types::*;

// A grpc header is u8 (flag) + u32 (msg len)
const GRPC_HEADER_SIZE: usize = 1 + 4;

pub(crate) mod content_types {
    use http::{HeaderMap, header::CONTENT_TYPE};

    pub(crate) const GRPC_WEB: &str = "application/grpc-web";
    pub(crate) const GRPC_WEB_PROTO: &str = "application/grpc-web+proto";
    pub(crate) const GRPC_WEB_TEXT: &str = "application/grpc-web-text";
    pub(crate) const GRPC_WEB_TEXT_PROTO: &str = "application/grpc-web-text+proto";

    pub(crate) fn is_grpc_web(headers: &HeaderMap) -> bool {
        matches!(
            content_type(headers),
            Some(GRPC_WEB) | Some(GRPC_WEB_PROTO) | Some(GRPC_WEB_TEXT) | Some(GRPC_WEB_TEXT_PROTO)
        )
    }

    fn content_type(headers: &HeaderMap) -> Option<&str> {
        headers.get(CONTENT_TYPE).and_then(|val| val.to_str().ok())
    }
}

const BUFFER_SIZE: usize = 8 * 1024;

const FRAME_HEADER_SIZE: usize = 5;

// 8th (MSB) bit of the 1st gRPC frame byte
// denotes an uncompressed trailer (as part of the body)
const GRPC_WEB_TRAILERS_BIT: u8 = 0b10000000;

#[derive(Copy, Clone, PartialEq, Debug)]
enum Direction {
    Decode,
    Encode,
    Empty,
}

#[derive(Copy, Clone, PartialEq, Debug)]
pub(crate) enum Encoding {
    Base64,
    None,
}

/// HttpBody adapter for the grpc web based services.
#[derive(Debug)]
#[pin_project]
pub struct GrpcWebCall<B> {
    #[pin]
    inner: B,
    buf: BytesMut,
    decoded: BytesMut,
    direction: Direction,
    encoding: Encoding,
    client: bool,
    trailers: Option<HeaderMap>,
}

impl<B: Default> Default for GrpcWebCall<B> {
    fn default() -> Self {
        Self {
            inner: Default::default(),
            buf: Default::default(),
            decoded: Default::default(),
            direction: Direction::Empty,
            encoding: Encoding::None,
            client: Default::default(),
            trailers: Default::default(),
        }
    }
}

impl<B> GrpcWebCall<B> {
    pub(crate) fn request(inner: B, encoding: Encoding) -> Self {
        Self::new(inner, Direction::Decode, encoding)
    }

    pub(crate) fn response(inner: B, encoding: Encoding) -> Self {
        Self::new(inner, Direction::Encode, encoding)
    }

    pub(crate) fn client_request(inner: B) -> Self {
        Self::new_client(inner, Direction::Encode, Encoding::None)
    }

    pub(crate) fn client_response(inner: B) -> Self {
        Self::new_client(inner, Direction::Decode, Encoding::None)
    }

    fn new_client(inner: B, direction: Direction, encoding: Encoding) -> Self {
        GrpcWebCall {
            inner,
            buf: BytesMut::with_capacity(match (direction, encoding) {
                (Direction::Encode, Encoding::Base64) => BUFFER_SIZE,
                _ => 0,
            }),
            decoded: BytesMut::with_capacity(match direction {
                Direction::Decode => BUFFER_SIZE,
                _ => 0,
            }),
            direction,
            encoding,
            client: true,
            trailers: None,
        }
    }

    fn new(inner: B, direction: Direction, encoding: Encoding) -> Self {
        GrpcWebCall {
            inner,
            buf: BytesMut::with_capacity(match (direction, encoding) {
                (Direction::Encode, Encoding::Base64) => BUFFER_SIZE,
                _ => 0,
            }),
            decoded: BytesMut::with_capacity(0),
            direction,
            encoding,
            client: false,
            trailers: None,
        }
    }

    // This is to avoid passing a slice of bytes with a length that the base64
    // decoder would consider invalid.
    #[inline]
    fn max_decodable(&self) -> usize {
        (self.buf.len() / 4) * 4
    }

    fn decode_chunk(mut self: Pin<&mut Self>) -> Result<Option<Bytes>, Status> {
        // not enough bytes to decode
        if self.buf.is_empty() || self.buf.len() < 4 {
            return Ok(None);
        }

        // Split `buf` at the largest index that is multiple of 4. Decode the
        // returned `Bytes`, keeping the rest for the next attempt to decode.
        let index = self.max_decodable();

        crate::util::base64::STANDARD
            .decode(self.as_mut().project().buf.split_to(index))
            .map(|decoded| Some(Bytes::from(decoded)))
            .map_err(internal_error)
    }
}

impl<B> GrpcWebCall<B>
where
    B: Body,
    B::Data: Buf,
    B::Error: fmt::Display,
{
    // Poll body for data, decoding (e.g. via Base64 if necessary) and returning frames
    // to the caller. If the caller is a client, it should look for trailers before
    // returning these frames.
    fn poll_decode(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, Status>>> {
        match self.encoding {
            Encoding::Base64 => loop {
                if let Some(bytes) = self.as_mut().decode_chunk()? {
                    return Poll::Ready(Some(Ok(Frame::data(bytes))));
                }

                let this = self.as_mut().project();

                match ready!(this.inner.poll_frame(cx)) {
                    Some(Ok(frame)) if frame.is_data() => this
                        .buf
                        .put(frame.into_data().unwrap_or_else(|_| unreachable!())),
                    Some(Ok(frame)) if frame.is_trailers() => {
                        return Poll::Ready(Some(Err(internal_error(
                            "malformed base64 request has unencoded trailers",
                        ))));
                    }
                    Some(Ok(_)) => {
                        return Poll::Ready(Some(Err(internal_error("unexpected frame type"))));
                    }
                    Some(Err(e)) => return Poll::Ready(Some(Err(internal_error(e)))),
                    None => {
                        return if this.buf.has_remaining() {
                            Poll::Ready(Some(Err(internal_error("malformed base64 request"))))
                        } else if let Some(trailers) = this.trailers.take() {
                            Poll::Ready(Some(Ok(Frame::trailers(trailers))))
                        } else {
                            Poll::Ready(None)
                        };
                    }
                }
            },

            Encoding::None => self
                .project()
                .inner
                .poll_frame(cx)
                .map_ok(|f| f.map_data(|mut d| d.copy_to_bytes(d.remaining())))
                .map_err(internal_error),
        }
    }

    fn poll_encode(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Bytes>, Status>>> {
        let this = self.as_mut().project();

        match ready!(this.inner.poll_frame(cx)) {
            Some(Ok(frame)) if frame.is_data() => {
                let mut data = frame.into_data().unwrap_or_else(|_| unreachable!());
                let mut res = data.copy_to_bytes(data.remaining());

                if *this.encoding == Encoding::Base64 {
                    res = crate::util::base64::STANDARD.encode(res).into();
                }

                Poll::Ready(Some(Ok(Frame::data(res))))
            }
            Some(Ok(frame)) if frame.is_trailers() => {
                let trailers = frame.into_trailers().unwrap_or_else(|_| unreachable!());
                let mut res = make_trailers_frame(trailers);

                if *this.encoding == Encoding::Base64 {
                    res = crate::util::base64::STANDARD.encode(res).into();
                }

                Poll::Ready(Some(Ok(Frame::data(res))))
            }
            Some(Ok(_)) => Poll::Ready(Some(Err(internal_error("unexpected frame type")))),
            Some(Err(e)) => Poll::Ready(Some(Err(internal_error(e)))),
            None => Poll::Ready(None),
        }
    }
}

impl<B> Body for GrpcWebCall<B>
where
    B: Body,
    B::Error: fmt::Display,
{
    type Data = Bytes;
    type Error = Status;

    fn poll_frame(
        mut self: Pin<&mut Self>,
        cx: &mut Context<'_>,
    ) -> Poll<Option<Result<Frame<Self::Data>, Self::Error>>> {
        if self.client && self.direction == Direction::Decode {
            let mut me = self.as_mut();

            // If we previously decoded trailers (which can arrive in the same HTTP
            // data frame as the final message), ensure we yield them on a subsequent
            // poll once all message bytes have been emitted.
            let this = me.as_mut().project();
            if this.decoded.is_empty()
                && let Some(trailers) = this.trailers.take()
            {
                return Poll::Ready(Some(Ok(Frame::trailers(trailers))));
            }

            loop {
                // Track whether the underlying HTTP body is complete for this poll.
                // This is used to avoid busy loops when we have an incomplete gRPC
                // frame buffered but can no longer read more bytes.
                let eof = match ready!(me.as_mut().poll_decode(cx)) {
                    Some(Ok(incoming_buf)) if incoming_buf.is_data() => {
                        me.as_mut()
                            .project()
                            .decoded
                            .put(incoming_buf.into_data().unwrap());
                        false
                    }
                    Some(Ok(incoming_buf)) if incoming_buf.is_trailers() => {
                        let trailers = incoming_buf.into_trailers().unwrap();
                        match me.as_mut().project().trailers {
                            Some(current_trailers) => {
                                current_trailers.extend(trailers);
                            }
                            None => {
                                me.as_mut().project().trailers.replace(trailers);
                            }
                        }
                        continue;
                    }
                    Some(Ok(_)) => unreachable!("unexpected frame type"),
                    // No more data to decode, time to look for trailers / remaining messages
                    None => true,
                    Some(Err(e)) => return Poll::Ready(Some(Err(e))),
                };

                // Hold the incoming, decoded data until we have a full message
                // or trailers to return.
                let buf = me.as_mut().project().decoded;

                return match find_trailers(&buf[..])? {
                    FindTrailers::Trailer(len) => {
                        // Extract up to len of where the trailers are at
                        let msg_buf = buf.copy_to_bytes(len);

                        // Only attempt to decode the trailers once the full trailers frame
                        // is buffered. Otherwise, keep the bytes in `decoded` for the next poll.
                        match grpc_web_frame_total_len(&buf[..]) {
                            Some(total_len) if buf.len() >= total_len => {
                                let trailers_bytes = buf.split_to(total_len).freeze();
                                match decode_trailers_frame(trailers_bytes) {
                                    Ok(Some(trailers)) => {
                                        me.as_mut().project().trailers.replace(trailers);
                                    }
                                    Err(e) => return Poll::Ready(Some(Err(e))),
                                    Ok(None) => {}
                                }
                            }
                            // We cannot make forward progress: the trailers frame is only
                            // partially buffered, and the underlying HTTP body has ended.
                            _ if eof && !msg_buf.has_remaining() => {
                                return Poll::Ready(Some(Err(data_loss(
                                    "incomplete grpc-web trailers frame",
                                ))));
                            }
                            _ => {}
                        }

                        if msg_buf.has_remaining() {
                            Poll::Ready(Some(Ok(Frame::data(msg_buf))))
                        } else if let Some(trailers) = me.as_mut().project().trailers.take() {
                            Poll::Ready(Some(Ok(Frame::trailers(trailers))))
                        } else {
                            // If we haven't yet buffered the full trailers frame, request more
                            // bytes from the underlying body.
                            if !eof {
                                continue;
                            }
                            Poll::Ready(None)
                        }
                    }
                    FindTrailers::IncompleteBuf => {
                        if eof {
                            Poll::Ready(Some(Err(data_loss("incomplete grpc-web message frame"))))
                        } else {
                            continue;
                        }
                    }
                    FindTrailers::Done(len) => {
                        // If we have no more message bytes buffered, but we did decode trailers
                        // on a previous poll, yield them now. Besides grpc-web trailer frames,
                        // this covers HTTP trailers stored by the trailers arm above.
                        if len == 0 {
                            if let Some(trailers) = me.as_mut().project().trailers.take() {
                                return Poll::Ready(Some(Ok(Frame::trailers(trailers))));
                            }

                            // Not enough bytes to form a complete gRPC frame yet.
                            // If we can still read from the underlying body, continue polling
                            // for more bytes.
                            if !eof {
                                continue;
                            }

                            Poll::Ready(None)
                        } else {
                            Poll::Ready(Some(Ok(Frame::data(buf.split_to(len).freeze()))))
                        }
                    }
                };
            }
        }

        match self.direction {
            Direction::Decode => self.poll_decode(cx),
            Direction::Encode => self.poll_encode(cx),
            Direction::Empty => Poll::Ready(None),
        }
    }

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

    fn size_hint(&self) -> SizeHint {
        self.inner.size_hint()
    }
}

impl<B> Stream for GrpcWebCall<B>
where
    B: Body,
    B::Error: fmt::Display,
{
    type Item = Result<Frame<Bytes>, Status>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        self.poll_frame(cx)
    }
}

impl Encoding {
    pub(crate) fn from_content_type(headers: &HeaderMap) -> Encoding {
        Self::from_header(headers.get(header::CONTENT_TYPE))
    }

    pub(crate) fn from_accept(headers: &HeaderMap) -> Encoding {
        Self::from_header(headers.get(header::ACCEPT))
    }

    pub(crate) fn to_content_type(self) -> &'static str {
        match self {
            Encoding::Base64 => GRPC_WEB_TEXT_PROTO,
            Encoding::None => GRPC_WEB_PROTO,
        }
    }

    fn from_header(value: Option<&HeaderValue>) -> Encoding {
        match value.and_then(|val| val.to_str().ok()) {
            Some(GRPC_WEB_TEXT_PROTO) | Some(GRPC_WEB_TEXT) => Encoding::Base64,
            _ => Encoding::None,
        }
    }
}

fn internal_error(e: impl std::fmt::Display) -> Status {
    Status::internal(format!("tonic-web: {e}"))
}

fn data_loss(e: impl std::fmt::Display) -> Status {
    Status::data_loss(format!("tonic-web: {e}"))
}

// Key-value pairs encoded as a HTTP/1 headers block (without the terminating newline)
fn encode_trailers(trailers: HeaderMap) -> Vec<u8> {
    trailers.iter().fold(Vec::new(), |mut acc, (key, value)| {
        acc.put_slice(key.as_ref());
        acc.push(b':');
        acc.put_slice(value.as_bytes());
        acc.put_slice(b"\r\n");
        acc
    })
}

fn decode_trailers_frame(mut buf: Bytes) -> Result<Option<HeaderMap>, Status> {
    if buf.remaining() < GRPC_HEADER_SIZE {
        return Ok(None);
    }

    buf.get_u8();
    buf.get_u32();

    let mut map = HeaderMap::new();
    let mut temp_buf = buf.clone();

    let mut trailers = Vec::new();
    let mut cursor_pos = 0;

    for (i, b) in buf.iter().enumerate() {
        // if we are at a trailer delimiter (\r\n)
        if b == &b'\r' && buf.get(i + 1) == Some(&b'\n') {
            // read the bytes of the trailer passed so far
            let trailer = temp_buf.copy_to_bytes(i - cursor_pos);
            // increment cursor beyond the delimiter
            cursor_pos = i + 2;
            trailers.push(trailer);
            if temp_buf.has_remaining() {
                // advance buf beyond the delimiters
                temp_buf.get_u8();
                temp_buf.get_u8();
            }
        }
    }

    for trailer in trailers {
        let Some((key, value)) = trailer
            .iter()
            .position(|b| *b == b':')
            .map(|pos| trailer.split_at(pos))
        else {
            return Err(Status::internal("trailers couldn't parse key/value"));
        };

        // Skip the ':' separator and trim leading OWS (spaces/tabs) from the value
        let value = &value[1..]; // skip ':'
        let value = trim_ascii_start(value);

        let header_key = HeaderName::try_from(key)
            .map_err(|e| Status::internal(format!("Unable to parse HeaderName: {e}")))?;
        let header_value = HeaderValue::try_from(value)
            .map_err(|e| Status::internal(format!("Unable to parse HeaderValue: {e}")))?;
        map.insert(header_key, header_value);
    }

    Ok(Some(map))
}

fn trim_ascii_start(bytes: &[u8]) -> &[u8] {
    let start = bytes
        .iter()
        .position(|b| !b.is_ascii_whitespace())
        .unwrap_or(bytes.len());
    &bytes[start..]
}

fn make_trailers_frame(trailers: HeaderMap) -> Bytes {
    let trailers = encode_trailers(trailers);
    let len = trailers.len();
    assert!(len <= u32::MAX as usize);

    let mut frame = BytesMut::with_capacity(len + FRAME_HEADER_SIZE);
    frame.put_u8(GRPC_WEB_TRAILERS_BIT);
    frame.put_u32(len as u32);
    frame.put_slice(&trailers);

    frame.freeze()
}

/// Return the total length (header + payload) of the first grpc-web frame in `buf`.
///
/// This is used to avoid consuming (and potentially partially parsing) frames when
/// the underlying HTTP body has not yielded enough bytes yet.
#[inline]
fn grpc_web_frame_total_len(buf: &[u8]) -> Option<usize> {
    if buf.len() < FRAME_HEADER_SIZE {
        return None;
    }

    // Bytes 1..=4 are the big-endian payload length.
    let payload_len = u32::from_be_bytes([buf[1], buf[2], buf[3], buf[4]]) as usize;
    // Saturating: on 32-bit targets a payload length near `u32::MAX` would overflow,
    // and a frame that large can never be fully buffered anyway.
    Some(FRAME_HEADER_SIZE.saturating_add(payload_len))
}

/// Search some buffer for grpc-web trailers headers and return
/// its location in the original buf. If `None` is returned we did
/// not find a trailers in this buffer either because its incomplete
/// or the buffer just contained grpc message frames.
fn find_trailers(buf: &[u8]) -> Result<FindTrailers, Status> {
    let mut len = 0;
    let mut temp_buf = buf;

    loop {
        // To check each frame, there must be at least GRPC_HEADER_SIZE
        // amount of bytes available otherwise the buffer is incomplete.
        if temp_buf.is_empty() || temp_buf.len() < GRPC_HEADER_SIZE {
            return Ok(FindTrailers::Done(len));
        }

        let header = temp_buf.get_u8();

        if header == GRPC_WEB_TRAILERS_BIT {
            return Ok(FindTrailers::Trailer(len));
        }

        if !(header == 0 || header == 1) {
            return Err(Status::internal(format!(
                "Invalid header bit {header} expected 0 or 1"
            )));
        }

        let msg_len = temp_buf.get_u32();

        len += msg_len as usize + 4 + 1;

        // If the msg len of a non-grpc-web trailer frame is larger than
        // the overall buffer we know within that buffer there are no trailers.
        if len > buf.len() {
            return Ok(FindTrailers::IncompleteBuf);
        }

        temp_buf = &buf[len..];
    }
}

#[derive(Debug, PartialEq, Eq)]
enum FindTrailers {
    Trailer(usize),
    IncompleteBuf,
    Done(usize),
}

#[cfg(test)]
mod tests {
    use http_body_util::BodyExt;
    use std::collections::VecDeque;
    use std::convert::Infallible;
    use std::pin::Pin;
    use std::task::{Context, Poll};
    use tonic::Code;

    use super::*;

    /// A body that yields each of `chunks` as its own data frame, so tests can
    /// control how grpc-web frames are split across HTTP data frames.
    ///
    /// Once the chunks run out it ends the stream, unless it was built with
    /// [`ChunkedBody::stalled`], in which case it stays pending forever.
    struct ChunkedBody {
        chunks: VecDeque<Bytes>,
        end_of_stream: bool,
    }

    impl ChunkedBody {
        fn new(chunks: impl IntoIterator<Item = Bytes>) -> Self {
            Self {
                chunks: chunks.into_iter().collect(),
                end_of_stream: true,
            }
        }

        /// A body that never signals end-of-stream, modelling a transport that has
        /// delivered every byte but has not yet reported that the body is complete.
        fn stalled(chunks: impl IntoIterator<Item = Bytes>) -> Self {
            Self {
                end_of_stream: false,
                ..Self::new(chunks)
            }
        }
    }

    impl Body for ChunkedBody {
        type Data = Bytes;
        type Error = Infallible;

        fn poll_frame(
            mut self: Pin<&mut Self>,
            _cx: &mut Context<'_>,
        ) -> Poll<Option<Result<Frame<Bytes>, Infallible>>> {
            match self.chunks.pop_front() {
                Some(chunk) => Poll::Ready(Some(Ok(Frame::data(chunk)))),
                None if self.end_of_stream => Poll::Ready(None),
                None => Poll::Pending,
            }
        }
    }

    /// Poll `call` exactly once, so tests can assert that a frame is available
    /// without the poll being retried on their behalf.
    async fn poll_once<B>(call: &mut GrpcWebCall<B>) -> Poll<Option<Result<Frame<Bytes>, Status>>>
    where
        B: Body + Unpin,
        B::Error: fmt::Display,
    {
        let mut polled = None;
        std::future::poll_fn(|cx| {
            polled = Some(Pin::new(&mut *call).poll_frame(cx));
            Poll::Ready(())
        })
        .await;

        polled.unwrap()
    }

    #[test]
    fn encoding_constructors() {
        let cases = &[
            (GRPC_WEB, Encoding::None),
            (GRPC_WEB_PROTO, Encoding::None),
            (GRPC_WEB_TEXT, Encoding::Base64),
            (GRPC_WEB_TEXT_PROTO, Encoding::Base64),
            ("foo", Encoding::None),
        ];

        let mut headers = HeaderMap::new();

        for case in cases {
            headers.insert(header::CONTENT_TYPE, case.0.parse().unwrap());
            headers.insert(header::ACCEPT, case.0.parse().unwrap());

            assert_eq!(Encoding::from_content_type(&headers), case.1, "{}", case.0);
            assert_eq!(Encoding::from_accept(&headers), case.1, "{}", case.0);
        }
    }

    #[test]
    fn decode_trailers() {
        let mut headers = HeaderMap::new();
        headers.insert(Status::GRPC_STATUS, 0.into());
        headers.insert(
            Status::GRPC_MESSAGE,
            "this is a message".try_into().unwrap(),
        );

        let trailers = make_trailers_frame(headers.clone());

        let map = decode_trailers_frame(trailers).unwrap().unwrap();

        assert_eq!(headers, map);
    }

    #[test]
    fn find_trailers_non_buffered() {
        // Byte version of this:
        // b"\x80\0\0\0\x0fgrpc-status:0\r\n"
        let buf = [
            128, 0, 0, 0, 15, 103, 114, 112, 99, 45, 115, 116, 97, 116, 117, 115, 58, 48, 13, 10,
        ];

        let out = find_trailers(&buf[..]).unwrap();

        assert_eq!(out, FindTrailers::Trailer(0));
    }

    #[test]
    fn find_trailers_buffered() {
        // Byte version of this:
        // b"\0\0\0\0L\n$975738af-1a17-4aea-b887-ed0bbced6093\x1a$da609e9b-f470-4cc0-a691-3fd6a005a436\x80\0\0\0\x0fgrpc-status:0\r\n"
        let buf = [
            0, 0, 0, 0, 76, 10, 36, 57, 55, 53, 55, 51, 56, 97, 102, 45, 49, 97, 49, 55, 45, 52,
            97, 101, 97, 45, 98, 56, 56, 55, 45, 101, 100, 48, 98, 98, 99, 101, 100, 54, 48, 57,
            51, 26, 36, 100, 97, 54, 48, 57, 101, 57, 98, 45, 102, 52, 55, 48, 45, 52, 99, 99, 48,
            45, 97, 54, 57, 49, 45, 51, 102, 100, 54, 97, 48, 48, 53, 97, 52, 51, 54, 128, 0, 0, 0,
            15, 103, 114, 112, 99, 45, 115, 116, 97, 116, 117, 115, 58, 48, 13, 10,
        ];

        let out = find_trailers(&buf[..]).unwrap();

        assert_eq!(out, FindTrailers::Trailer(81));

        let trailers = decode_trailers_frame(Bytes::copy_from_slice(&buf[81..]))
            .unwrap()
            .unwrap();
        let status = trailers.get(Status::GRPC_STATUS).unwrap();
        assert_eq!(status.to_str().unwrap(), "0")
    }

    #[test]
    fn find_trailers_buffered_incomplete_message() {
        let buf = vec![
            0, 0, 0, 9, 238, 10, 233, 19, 18, 230, 19, 10, 9, 10, 1, 120, 26, 4, 84, 69, 88, 84,
            18, 60, 10, 58, 10, 56, 3, 0, 0, 0, 44, 0, 0, 0, 0, 0, 0, 0, 116, 104, 105, 115, 32,
            118, 97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114, 105, 116, 116, 101, 110, 32,
            118, 105, 97, 32, 119, 114, 105, 116, 101, 32, 100, 101, 108, 101, 103, 97, 116, 105,
            111, 110, 33, 18, 62, 10, 60, 10, 58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0, 0, 0, 116, 104,
            105, 115, 32, 118, 97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114, 105, 116, 116,
            101, 110, 32, 98, 121, 32, 97, 110, 32, 101, 109, 98, 101, 100, 100, 101, 100, 32, 114,
            101, 112, 108, 105, 99, 97, 33, 18, 62, 10, 60, 10, 58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0,
            0, 0, 116, 104, 105, 115, 32, 118, 97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114,
            105, 116, 116, 101, 110, 32, 98, 121, 32, 97, 110, 32, 101, 109, 98, 101, 100, 100,
            101, 100, 32, 114, 101, 112, 108, 105, 99, 97, 33, 18, 62, 10, 60, 10, 58, 3, 0, 0, 0,
            46, 0, 0, 0, 0, 0, 0, 0, 116, 104, 105, 115, 32, 118, 97, 108, 117, 101, 32, 119, 97,
            115, 32, 119, 114, 105, 116, 116, 101, 110, 32, 98, 121, 32, 97, 110, 32, 101, 109, 98,
            101, 100, 100, 101, 100, 32, 114, 101, 112, 108, 105, 99, 97, 33, 18, 62, 10, 60, 10,
            58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0, 0, 0, 116, 104, 105, 115, 32, 118, 97, 108, 117,
            101, 32, 119, 97, 115, 32, 119, 114, 105, 116, 116, 101, 110, 32, 98, 121, 32, 97, 110,
            32, 101, 109, 98, 101, 100, 100, 101, 100, 32, 114, 101, 112, 108, 105, 99, 97, 33, 18,
            62, 10, 60, 10, 58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0, 0, 0, 116, 104, 105, 115, 32, 118,
            97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114, 105, 116, 116, 101, 110, 32, 98,
            121, 32, 97, 110, 32, 101, 109, 98, 101, 100, 100, 101, 100, 32, 114, 101, 112, 108,
            105, 99, 97, 33, 18, 62, 10, 60, 10, 58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0, 0, 0, 116, 104,
            105, 115, 32, 118, 97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114, 105, 116, 116,
            101, 110, 32, 98, 121, 32, 97, 110, 32, 101, 109, 98, 101, 100, 100, 101, 100, 32, 114,
            101, 112, 108, 105, 99, 97, 33, 18, 62, 10, 60, 10, 58, 3, 0, 0, 0, 46, 0, 0, 0, 0, 0,
            0, 0, 116, 104, 105, 115, 32, 118, 97, 108, 117, 101, 32, 119, 97, 115, 32, 119, 114,
            105, 116, 116, 101, 110, 32, 98, 121, 32,
        ];

        let out = find_trailers(&buf[..]).unwrap();

        assert_eq!(out, FindTrailers::IncompleteBuf);
    }

    #[test]
    #[ignore]
    fn find_trailers_buffered_incomplete_buf_bug() {
        let buf = std::fs::read("tests/incomplete-buf-bug.bin").unwrap();
        let out = find_trailers(&buf[..]).unwrap_err();

        assert_eq!(out.code(), Code::Internal);
    }

    #[test]
    fn decode_multiple_trailers() {
        let buf = b"\x80\0\0\0\x0fgrpc-status:0\r\ngrpc-message:\r\na:1\r\nb:2\r\n";

        let trailers = decode_trailers_frame(Bytes::copy_from_slice(&buf[..]))
            .unwrap()
            .unwrap();

        let mut expected = HeaderMap::new();
        expected.insert(Status::GRPC_STATUS, "0".parse().unwrap());
        expected.insert(Status::GRPC_MESSAGE, "".parse().unwrap());
        expected.insert("a", "1".parse().unwrap());
        expected.insert("b", "2".parse().unwrap());

        assert_eq!(trailers, expected);
    }

    #[test]
    fn decode_trailers_with_space_after_colon() {
        let buf = b"\x80\0\0\0\x0fgrpc-status: 0\r\ngrpc-message: \r\n";

        let trailers = decode_trailers_frame(Bytes::copy_from_slice(&buf[..]))
            .unwrap()
            .unwrap();

        let mut expected = HeaderMap::new();
        expected.insert(Status::GRPC_STATUS, "0".parse().unwrap());
        expected.insert(Status::GRPC_MESSAGE, "".parse().unwrap());

        assert_eq!(trailers, expected);
    }

    #[test]
    fn decode_trailers_space_after_colon() {
        // connect-rpc and standard HTTP use "key: value" (space after colon)
        let trailers_bytes = b"grpc-status: 0\r\ngrpc-message: this is a message\r\n";
        let len = trailers_bytes.len();

        let mut frame = BytesMut::new();
        frame.put_u8(GRPC_WEB_TRAILERS_BIT);
        frame.put_u32(len as u32);
        frame.put_slice(&trailers_bytes[..]);

        let map = decode_trailers_frame(frame.freeze()).unwrap().unwrap();

        let mut expected = HeaderMap::new();
        expected.insert(Status::GRPC_STATUS, HeaderValue::from_static("0"));
        expected.insert(
            Status::GRPC_MESSAGE,
            HeaderValue::from_static("this is a message"),
        );

        assert_eq!(map, expected);
    }

    #[test]
    fn decode_trailers_value_with_colons() {
        let trailers_bytes = b"grpc-status: 0\r\ngrpc-message: error: something: went wrong\r\n";
        let len = trailers_bytes.len();

        let mut frame = BytesMut::new();
        frame.put_u8(GRPC_WEB_TRAILERS_BIT);
        frame.put_u32(len as u32);
        frame.put_slice(&trailers_bytes[..]);

        let map = decode_trailers_frame(frame.freeze()).unwrap().unwrap();

        assert_eq!(map.get("grpc-status").unwrap(), "0");
        assert_eq!(
            map.get("grpc-message").unwrap(),
            "error: something: went wrong"
        );
    }

    const MESSAGE_FRAME: &[u8] = b"\0\0\0\0\x13\n\x11first ok response";
    const TRAILERS_FRAME: &[u8] =
        b"\x80\0\0\0Agrpc-status:10\r\ngrpc-message:error%20after%20partial%20response\r\n";

    async fn assert_message_then_trailers(mut call: GrpcWebCall<ChunkedBody>) {
        let message = call
            .frame()
            .await
            .expect("expected a message frame")
            .unwrap()
            .into_data()
            .expect("expected a data frame");
        assert_eq!(message, Bytes::from_static(MESSAGE_FRAME));

        let trailers = call
            .frame()
            .await
            .expect("expected a trailers frame")
            .unwrap()
            .into_trailers()
            .expect("expected a trailers frame");
        assert_eq!(trailers.get("grpc-status").unwrap(), "10");
        assert_eq!(
            trailers.get("grpc-message").unwrap(),
            "error%20after%20partial%20response"
        );

        assert!(call.frame().await.is_none(), "expected end of stream");
    }

    #[tokio::test]
    async fn client_response_yields_trailers_after_final_message_in_same_chunk() {
        let chunk = Bytes::from([MESSAGE_FRAME, TRAILERS_FRAME].concat());

        assert_message_then_trailers(GrpcWebCall::client_response(ChunkedBody::new([chunk]))).await;
    }

    #[tokio::test]
    async fn client_response_yields_trailers_split_across_chunks() {
        // The trailers frame is only partially buffered when the final message is
        // emitted, so it has to be held back until the rest of it arrives.
        let (head, tail) = TRAILERS_FRAME.split_at(20);

        assert_message_then_trailers(GrpcWebCall::client_response(ChunkedBody::new([
            Bytes::from([MESSAGE_FRAME, head].concat()),
            Bytes::from_static(tail),
        ])))
        .await;
    }

    #[tokio::test]
    async fn client_response_yields_trailers_after_empty_data_frame() {
        // An empty data frame mid-stream is not the end of the body.
        assert_message_then_trailers(GrpcWebCall::client_response(ChunkedBody::new([
            Bytes::from_static(MESSAGE_FRAME),
            Bytes::new(),
            Bytes::from_static(TRAILERS_FRAME),
        ])))
        .await;
    }

    #[tokio::test]
    async fn client_response_yields_trailers_before_trailing_empty_data_frame() {
        // HTTP/2 peers commonly end the stream with an empty DATA frame.
        assert_message_then_trailers(GrpcWebCall::client_response(ChunkedBody::new([
            Bytes::from([MESSAGE_FRAME, TRAILERS_FRAME].concat()),
            Bytes::new(),
        ])))
        .await;
    }

    #[tokio::test]
    async fn client_response_yields_trailers_before_end_of_body() {
        // The trailers frame is the logical end of the RPC, so once it has been
        // decoded it must be yielded without waiting for the transport to report
        // end-of-stream.
        let chunk = Bytes::from([MESSAGE_FRAME, TRAILERS_FRAME].concat());
        let mut call = GrpcWebCall::client_response(ChunkedBody::stalled([chunk]));

        match poll_once(&mut call).await {
            Poll::Ready(Some(Ok(frame))) if frame.is_data() => {}
            other => panic!("expected a message frame, got {other:?}"),
        }

        match poll_once(&mut call).await {
            Poll::Ready(Some(Ok(frame))) if frame.is_trailers() => {}
            other => panic!("expected a trailers frame, got {other:?}"),
        }
    }

    #[tokio::test]
    async fn client_response_errors_on_truncated_trailers_frame() {
        let mut call = GrpcWebCall::client_response(ChunkedBody::new([
            Bytes::from_static(MESSAGE_FRAME),
            Bytes::from_static(&TRAILERS_FRAME[..20]),
        ]));

        assert!(call.frame().await.unwrap().unwrap().is_data());

        let err = call
            .frame()
            .await
            .expect("expected an error frame")
            .unwrap_err();
        assert_eq!(err.code(), Code::DataLoss);
    }

    #[tokio::test]
    async fn client_response_errors_on_truncated_message_frame() {
        // The frame header promises 19 bytes, but the body ends after 5 of them.
        let mut call = GrpcWebCall::client_response(ChunkedBody::new([Bytes::from_static(
            b"\0\0\0\0\x13\n\x11fir",
        )]));

        let err = call
            .frame()
            .await
            .expect("expected an error frame")
            .unwrap_err();
        assert_eq!(err.code(), Code::DataLoss);
    }
}