feat(http/retry): model PeekTrailersBody<B> with Frame<T> (#3559)

this branch contains a sequence of commits that refactor `PeekTrailersBody<B>`.

this branch is specifically focused on making this body middleware
forward-compatible with the 1.0 interface(s) of `http_body::Body` and
`http_body_util::BodyExt`.

it does this in two main steps: (1) temporarily vendoring `http_body::Frame<T>`
and providing a compatibility shim that provides a `frame()` method for a body,
and (2) modeling `PeekTrailersBody<B>` and its peeking logic in terms of this
`Frame<'a, T>` future.

---

* feat(http/retry): add `Frame<T>` compatibility facilities

this commit introduces a `compat` submodule to `linkerd-http-retry`.

this helps us frontrun the task of replacing all of the finicky control
flow in `PeekTrailersBody<B>` using the antiquated `data()` and
`trailers()` future combinators. instead, we can perform our peeking
in terms of an approximation of `http_body_util::BodyExt::frame()`.

to accomplish this, this commit vendors a copy of the `Frame<T>` type.
we can use this to preemptively model our peek body in terms of this
type, and move to the "real" version of it when we're upgrading in
pr #3504.

additionally, this commit includes a type called
`ForwardCompatibleBody<B>`, and a variant of the `Frame<'a, T>`
combinator. these are a bit boilerplate-y, admittedly, but the pleasant
part of this is that we have, in effect, migrated the trickiest body
middleware in advance of #3504. once we upgrade to http-body 1.0, all of
these types can be removed.

https://docs.rs/http-body-util/latest/http_body_util/trait.BodyExt.html#method.frame
https://docs.rs/http-body-util/0.1.2/src/http_body_util/combinators/frame.rs.html#10

Signed-off-by: katelyn martin <[email protected]>

* refactor(http/retry): `PeekTrailersBody<B>` uses `BodyExt::frame()`

this commit reworks `PeekTrailersBody<B>`.

the most important goal here is replacing the control flow of
`read_body()`, which polls a body using `BodyExt` future combinators
`data()` and `frame()` for up to two frames, with varying levels of
persistence depending on outcomes.

to quote #3556:

> the intent of this type is to only yield the asynchronous task
> responsible for reading the body once. depending on what the inner
> body yields, and when, this can result in combinations of: no data
> frames and no trailers, no data frames with trailers, one data frame
> and no trailers, one data frame with trailers, or two data frames.
> moreover, depending on which of these are yielded, the body will call
> .await some scenarios, and only poll functions once in others.
>
> migrating this to the Frame<T> and poll_frame() style of the 1.0 Body
> interface, away from the 0.4 interface that provides distinct
> poll_data() and poll_trailers() methods, is fundamentally tricky.

this means that `PeekTrailersBody<B>` is notably difficult to port
across the http-body 0.4/1.0 upgrade boundary.

this body middleware must navigate a number of edge conditions, and once
it _has_ obtained a `Frame<T>`, make use of conversion methods to
ascertain whether it is a data or trailers frame, due to the fact that
its internal enum representation is not exposed publicly. one it has
done all of that, it must do the same thing once more to examine the
second frame.

this commit uses the compatibility facilities and backported `Frame<T>`
introduced in the previous commit, and rewrites this control flow using
a form of the `BodyExt::frame()` combinator.

this means that this middleware is forward-compatible with http-body
1.0, which will dramatically simplify the remaining migration work to be
done in #3504.

see linkerd/linkerd2#8733 for more information
and other links related to this ongoing migration work.

Signed-off-by: katelyn martin <[email protected]>

* refactor(http/retry): mock body enforces `poll_trailers()` contract

this commit addresses a `TODO` note, and tightens the enforcement of a
rule defined by the v0.4 signature of the `Body` trait.

this commit changes the mock body type, used in tests, so that it will
panic if the caller improperly polls for a trailers frame before the
final data frame has been yielded.

previously, a comment indicated that we were "fairly sure" this was
okay. while that may have been acceptable in practice, the changes in
the previous commit mean that we now properly respect these rules.

thus, a panic can be placed here, to enforce that "[is] only be called
once `poll_data()` returns `None`", per the documentation.

Signed-off-by: katelyn martin <[email protected]>

* refactor(http/retry): rename `PeekTrailersBody::Buffered`

<#3559 (comment)>

this is a nicer name than `Unknown` for this case. not to mention, we'll
want that name shortly to address the possibility of unknown frame
variants.

Signed-off-by: katelyn martin <[email protected]>

* refactor(http/retry): encapsulate `Inner<B>` enum variants

this commit makes the inner enum variants private.

#3559 (comment)

Signed-off-by: katelyn martin <[email protected]>

* refactor(http/retry): gracefully ignore unknown frames

#3559 (comment)

Signed-off-by: katelyn martin <[email protected]>

---------

Signed-off-by: katelyn martin <[email protected]>

linkerd/http/retry/src/compat.rs

@@ -0,0 +1,115 @@
+//! Compatibility utilities for upgrading to http-body 1.0.
+
+use http_body::Body;
+
+pub(crate) use self::frame::Frame;
+
+mod frame;
+
+#[derive(Debug)]
+pub(crate) struct ForwardCompatibleBody<B> {
+    inner: B,
+    data_finished: bool,
+    trailers_finished: bool,
+}
+
+// === impl ForwardCompatibleBody ===
+
+impl<B: Body> ForwardCompatibleBody<B> {
+    pub(crate) fn new(body: B) -> Self {
+        if body.is_end_stream() {
+            Self {
+                inner: body,
+                data_finished: true,
+                trailers_finished: true,
+            }
+        } else {
+            Self {
+                inner: body,
+                data_finished: false,
+                trailers_finished: false,
+            }
+        }
+    }
+
+    pub(crate) fn into_inner(self) -> B {
+        self.inner
+    }
+
+    /// Returns a future that resolves to the next frame.
+    pub(crate) fn frame(&mut self) -> combinators::Frame<'_, B> {
+        combinators::Frame(self)
+    }
+}
+
+/// Future that resolves to the next frame from a `Body`.
+///
+/// NB: This is a vendored stand-in for [`Frame<'a, T>`][frame], and and can be replaced once
+/// we upgrade from http-body 0.4 to 1.0. This file was vendored, and subsequently adapted to this
+/// project, at commit 86fdf00.
+///
+/// See linkerd/linkerd2#8733 for more information.
+///
+/// [frame]: https://docs.rs/http-body-util/0.1.2/http_body_util/combinators/struct.Frame.html
+mod combinators {
+    use core::future::Future;
+    use core::pin::Pin;
+    use core::task;
+    use http_body::Body;
+    use std::ops::Not;
+    use std::task::ready;
+
+    use super::ForwardCompatibleBody;
+
+    #[must_use = "futures don't do anything unless polled"]
+    #[derive(Debug)]
+    /// Future that resolves to the next frame from a [`Body`].
+    pub struct Frame<'a, T>(pub(super) &'a mut super::ForwardCompatibleBody<T>);
+
+    impl<T: Body + Unpin> Future for Frame<'_, T> {
+        type Output = Option<Result<super::Frame<T::Data>, T::Error>>;
+
+        fn poll(self: Pin<&mut Self>, ctx: &mut task::Context<'_>) -> task::Poll<Self::Output> {
+            let Self(ForwardCompatibleBody {
+                inner,
+                data_finished,
+                trailers_finished,
+            }) = self.get_mut();
+            let mut pinned = Pin::new(inner);
+
+            // We have already yielded the trailers, the body is done.
+            if *trailers_finished {
+                return task::Poll::Ready(None);
+            }
+
+            // We are still yielding data frames.
+            if data_finished.not() {
+                match ready!(pinned.as_mut().poll_data(ctx)) {
+                    Some(Ok(data)) => {
+                        // We yielded a frame.
+                        return task::Poll::Ready(Some(Ok(super::Frame::data(data))));
+                    }
+                    Some(Err(error)) => {
+                        // If we encountered an error, we are finished.
+                        *data_finished = true;
+                        *trailers_finished = true;
+                        return task::Poll::Ready(Some(Err(error)));
+                    }
+                    None => {
+                        // We are done yielding data frames. Mark the corresponding flag, and fall
+                        // through to poll the trailers...
+                        *data_finished = true;
+                    }
+                };
+            }
+
+            // We have yielded all of the data frames but have not yielded the trailers.
+            let trailers = ready!(pinned.poll_trailers(ctx));
+            *trailers_finished = true;
+            let trailers = trailers
+                .transpose()
+                .map(|res| res.map(super::Frame::trailers));
+            task::Poll::Ready(trailers)
+        }
+    }
+}

linkerd/http/retry/src/compat/frame.rs

@@ -0,0 +1,131 @@
+#![allow(unused, reason = "this code is vendored from `http-body v1.0.1")]
+
+//! A frame of any kind related to an HTTP stream (body).
+//!
+//! NB: This is a vendored stand-in for [`Frame<T>`][frame], and and can be replaced once
+//! we upgrade from http-body 0.4 to 1.0. This file was vendored at commit 86fdf00.
+//!
+//! See linkerd/linkerd2#8733 for more information.
+//!
+//! [frame]: https://docs.rs/http-body/1.0.1/http_body/struct.Frame.html>
+
+use http::HeaderMap;
+
+/// A frame of any kind related to an HTTP stream (body).
+#[derive(Debug)]
+pub struct Frame<T> {
+    kind: Kind<T>,
+}
+
+#[derive(Debug)]
+enum Kind<T> {
+    // The first two variants are "inlined" since they are undoubtedly
+    // the most common. This saves us from having to allocate a
+    // boxed trait object for them.
+    Data(T),
+    Trailers(HeaderMap),
+    //Unknown(Box<dyn Frameish>),
+}
+
+impl<T> Frame<T> {
+    /// Create a DATA frame with the provided `Buf`.
+    pub fn data(buf: T) -> Self {
+        Self {
+            kind: Kind::Data(buf),
+        }
+    }
+
+    /// Create a trailers frame.
+    pub fn trailers(map: HeaderMap) -> Self {
+        Self {
+            kind: Kind::Trailers(map),
+        }
+    }
+
+    /// Maps this frame's data to a different type.
+    pub fn map_data<F, D>(self, f: F) -> Frame<D>
+    where
+        F: FnOnce(T) -> D,
+    {
+        match self.kind {
+            Kind::Data(data) => Frame {
+                kind: Kind::Data(f(data)),
+            },
+            Kind::Trailers(trailers) => Frame {
+                kind: Kind::Trailers(trailers),
+            },
+        }
+    }
+
+    /// Returns whether this is a DATA frame.
+    pub fn is_data(&self) -> bool {
+        matches!(self.kind, Kind::Data(..))
+    }
+
+    /// Consumes self into the buf of the DATA frame.
+    ///
+    /// Returns an [`Err`] containing the original [`Frame`] when frame is not a DATA frame.
+    /// `Frame::is_data` can also be used to determine if the frame is a DATA frame.
+    pub fn into_data(self) -> Result<T, Self> {
+        match self.kind {
+            Kind::Data(data) => Ok(data),
+            _ => Err(self),
+        }
+    }
+
+    /// If this is a DATA frame, returns a reference to it.
+    ///
+    /// Returns `None` if not a DATA frame.
+    pub fn data_ref(&self) -> Option<&T> {
+        match self.kind {
+            Kind::Data(ref data) => Some(data),
+            _ => None,
+        }
+    }
+
+    /// If this is a DATA frame, returns a mutable reference to it.
+    ///
+    /// Returns `None` if not a DATA frame.
+    pub fn data_mut(&mut self) -> Option<&mut T> {
+        match self.kind {
+            Kind::Data(ref mut data) => Some(data),
+            _ => None,
+        }
+    }
+
+    /// Returns whether this is a trailers frame.
+    pub fn is_trailers(&self) -> bool {
+        matches!(self.kind, Kind::Trailers(..))
+    }
+
+    /// Consumes self into the buf of the trailers frame.
+    ///
+    /// Returns an [`Err`] containing the original [`Frame`] when frame is not a trailers frame.
+    /// `Frame::is_trailers` can also be used to determine if the frame is a trailers frame.
+    pub fn into_trailers(self) -> Result<HeaderMap, Self> {
+        match self.kind {
+            Kind::Trailers(trailers) => Ok(trailers),
+            _ => Err(self),
+        }
+    }
+
+    /// If this is a trailers frame, returns a reference to it.
+    ///
+    /// Returns `None` if not a trailers frame.
+    pub fn trailers_ref(&self) -> Option<&HeaderMap> {
+        match self.kind {
+            Kind::Trailers(ref trailers) => Some(trailers),
+            _ => None,
+        }
+    }
+
+    /// If this is a trailers frame, returns a mutable reference to it.
+    ///
+    /// Returns `None` if not a trailers frame.
+    pub fn trailers_mut(&mut self) -> Option<&mut HeaderMap> {
+        match self.kind {
+            Kind::Trailers(ref mut trailers) => Some(trailers),
+            _ => None,
+        }
+    }
+}

linkerd/http/retry/src/lib.rs

@@ -4,6 +4,8 @@
 pub mod peek_trailers;
 pub mod replay;
 
+mod compat;
+
 pub use self::{peek_trailers::PeekTrailersBody, replay::ReplayBody};
 pub use tower::retry::budget::Budget;