linkerd
diff --git a/‎Cargo.lock
Lines changed: 223 additions & 123 deletions b/‎Cargo.lock
Lines changed: 223 additions & 123 deletions
diff --git a/‎Cargo.toml
Lines changed: 19 additions & 15 deletions b/‎Cargo.toml
Lines changed: 19 additions & 15 deletions
diff --git a/‎deny.toml
Lines changed: 11 additions & 8 deletions b/‎deny.toml
Lines changed: 11 additions & 8 deletions
diff --git a/‎hyper-balance/Cargo.toml
Lines changed: 1 addition & 1 deletion b/‎hyper-balance/Cargo.toml
Lines changed: 1 addition & 1 deletion
diff --git a/‎hyper-balance/src/lib.rs
Lines changed: 57 additions & 83 deletions b/‎hyper-balance/src/lib.rs
Lines changed: 57 additions & 83 deletions
diff --git a/‎linkerd/app/Cargo.toml
Lines changed: 1 addition & 0 deletions b/‎linkerd/app/Cargo.toml
Lines changed: 1 addition & 0 deletions
diff --git a/‎linkerd/app/admin/Cargo.toml
Lines changed: 2 additions & 1 deletion b/‎linkerd/app/admin/Cargo.toml
Lines changed: 2 additions & 1 deletion
@@ -24,7 +24,6 @@ members = [
     "linkerd/error-respond",
     "linkerd/exp-backoff",
     "linkerd/http/access-log",
-    "linkerd/http/body-compat",
     "linkerd/http/box",
     "linkerd/http/classify",
     "linkerd/http/detect",
@@ -107,26 +106,31 @@ publish = false
 
 [workspace.dependencies]
 bytes = { version = "1" }
-drain = { version = "0.1", default-features = false }
-h2 = { version = "0.3" }
-http = { version = "0.2" }
-http-body = { version = "0.4" }
-hyper = { version = "0.14.32", default-features = false }
+drain = { version = "0.2", default-features = false }
+h2 = { version = "0.4" }
+http = { version = "1" }
+http-body = { version = "1" }
+hyper = { version = "1", default-features = false }
+hyper-util = { version = "0.1", default-features = false }
 prometheus-client = { version = "0.23" }
-prost = { version = "0.12" }
-prost-build = { version = "0.12", default-features = false }
-prost-types = { version = "0.12" }
+prost = { version = "0.13" }
+prost-build = { version = "0.13", default-features = false }
+prost-types = { version = "0.13" }
 tokio-rustls = { version = "0.26", default-features = false, features = [
     "ring",
     "logging",
 ] }
-tonic = { version = "0.10", default-features = false }
-tonic-build = { version = "0.10", default-features = false }
-tower = { version = "0.4.13", default-features = false }
+tonic = { version = "0.12", default-features = false }
+tonic-build = { version = "0.12", default-features = false }
+tower = { version = "0.4", default-features = false }
 tower-service = { version = "0.3" }
 tower-test = { version = "0.4" }
 
+[workspace.dependencies.http-body-util]
+version = "0.1.3"
+default-features = false
+features = ["channel"]
+
 [workspace.dependencies.linkerd2-proxy-api]
-version = "0.15.0"
-# git = "https://github.com/linkerd/linkerd2-proxy-api.git"
-# branch = "main"
+git = "https://github.com/linkerd/linkerd2-proxy-api"
+branch = "main"
@@ -55,13 +55,11 @@ skip = [
     # `linkerd-trace-context`, `rustls-pemfile` and `tonic` depend on `base64`
     # v0.13.1 while `rcgen` depends on v0.21.5
     { name = "base64" },
-    # https://github.com/hawkw/matchers/pull/4
-    { name = "regex-automata", version = "0.1" },
-    { name = "regex-syntax", version = "0.6" },
-    # Some dependencies still use indexmap v1.
-    { name = "indexmap", version = "1" },
-    { name = "hashbrown", version = "0.12" },
-
+    # itertools still uses itertools v0.13.
+    { name = "itertools", version = "0.13" },
+    # tonic/axum depend on a newer `tower`, which we are still catching up to.
+    # see #3744.
+    { name = "tower", version = "0.5" },
 ]
 skip-tree = [
     # thiserror v2 is still propagating through the ecosystem
@@ -70,9 +68,14 @@ skip-tree = [
     { name = "rand", version = "0.8" },
     # rust v1.0 is still propagating through the ecosystem
     { name = "rustix", version = "0.38" },
+    # `pprof` uses a number of old dependencies. for now, we skip its subtree.
+    { name = "pprof" }
 ]
 
 [sources]
 unknown-registry = "deny"
 unknown-git = "deny"
-allow-registry = ["https://github.com/rust-lang/crates.io-index"]
+allow-registry = [
+    "https://github.com/rust-lang/crates.io-index",
+    "https://github.com/linkerd/linkerd2-proxy-api",
+]
@@ -10,7 +10,7 @@ publish = { workspace = true }
 futures = { version = "0.3", default-features = false }
 http = { workspace = true }
 http-body = { workspace = true }
-hyper = { workspace = true, features = ["deprecated"] }
+hyper = { workspace = true }
 pin-project = "1"
 tower = { workspace = true, default-features = false, features = ["load"] }
 tokio = { version = "1", features = ["macros"] }
 
@@ -102,32 +102,20 @@ where
         self.body.is_end_stream()
     }
 
-    fn poll_data(
+    fn poll_frame(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
         let this = self.project();
-        let ret = futures::ready!(this.body.poll_data(cx));
+        let ret = futures::ready!(this.body.poll_frame(cx));
 
-        // Once a data frame is received, the handle is dropped. On subsequent calls, this
+        // Once a frame is received, the handle is dropped. On subsequent calls, this
         // is a noop.
         drop(this.handle.take());
 
         Poll::Ready(ret)
     }
 
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-        let this = self.project();
-        // If this is being called, the handle definitely should have been dropped
-        // already.
-        drop(this.handle.take());
-
-        this.body.poll_trailers(cx)
-    }
-
     #[inline]
     fn size_hint(&self) -> hyper::body::SizeHint {
         self.body.size_hint()
@@ -157,35 +145,21 @@ impl<T: Send + 'static, B: Body> Body for PendingUntilEosBody<T, B> {
         self.body.is_end_stream()
     }
 
-    fn poll_data(
+    fn poll_frame(
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
-    ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
+    ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
         let mut this = self.project();
         let body = &mut this.body;
         tokio::pin!(body);
-        let ret = futures::ready!(body.poll_data(cx));
+        let frame = futures::ready!(body.poll_frame(cx));
 
         // If this was the last frame, then drop the handle immediately.
         if this.body.is_end_stream() {
             drop(this.handle.take());
         }
 
-        Poll::Ready(ret)
-    }
-
-    fn poll_trailers(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-        let this = self.project();
-        let ret = futures::ready!(this.body.poll_trailers(cx));
-
-        // Once trailers are received, the handle is dropped immediately (in case the body
-        // is retained longer for some reason).
-        drop(this.handle.take());
-
-        Poll::Ready(ret)
+        Poll::Ready(frame)
     }
 
     #[inline]
@@ -198,7 +172,7 @@ impl<T: Send + 'static, B: Body> Body for PendingUntilEosBody<T, B> {
 mod tests {
     use super::{PendingUntilEos, PendingUntilFirstData};
     use futures::future::poll_fn;
-    use http_body::Body;
+    use http_body::{Body, Frame};
     use std::collections::VecDeque;
     use std::io::Cursor;
     use std::pin::Pin;
@@ -225,11 +199,13 @@ mod tests {
         assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll())
-        .expect("data some")
-        .expect("data ok");
+        .expect("frame is some")
+        .expect("frame is ok")
+        .into_data()
+        .expect("frame is data");
         assert!(wk.upgrade().is_none());
     }
 
@@ -282,10 +258,10 @@ mod tests {
         let res = assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll());
-        assert!(res.expect("data is some").is_err());
+        assert!(res.expect("frame is some").is_err());
         assert!(wk.upgrade().is_none());
     }
 
@@ -308,21 +284,21 @@ mod tests {
         assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll())
-        .expect("data some")
-        .expect("data ok");
+        .expect("frame is some")
+        .expect("frame is ok");
         assert!(wk.upgrade().is_some());
 
         assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll())
-        .expect("data some")
-        .expect("data ok");
+        .expect("frame is some")
+        .expect("frame is ok");
         assert!(wk.upgrade().is_none());
     }
 
@@ -355,40 +331,42 @@ mod tests {
         assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll())
-        .expect("data")
-        .expect("data ok");
+        .expect("frame is some")
+        .expect("frame is ok");
         assert!(wk.upgrade().is_some());
 
         assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll())
-        .expect("data")
-        .expect("data ok");
+        .expect("frame is some")
+        .expect("frame is ok");
         assert!(wk.upgrade().is_some());
 
-        let poll = assert_ready!(task::spawn(poll_fn(|cx| {
+        assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
-        .poll());
-        assert!(poll.is_none());
-        assert!(wk.upgrade().is_some());
+        .poll())
+        .expect("frame is some")
+        .expect("frame is ok")
+        .into_trailers()
+        .expect("is trailers");
+        assert!(wk.upgrade().is_none());
 
-        assert_ready!(task::spawn(poll_fn(|cx| {
+        let poll = assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_trailers(cx)
+            body.poll_frame(cx)
         }))
-        .poll())
-        .expect("trailers ok")
-        .expect("trailers");
+        .poll());
+        assert!(poll.is_none());
         assert!(wk.upgrade().is_none());
     }
 
@@ -411,7 +389,7 @@ mod tests {
         let poll = assert_ready!(task::spawn(poll_fn(|cx| {
             let body = &mut body;
             tokio::pin!(body);
-            body.poll_data(cx)
+            body.poll_frame(cx)
         }))
         .poll());
         assert!(poll.expect("some").is_err());
@@ -437,20 +415,21 @@ mod tests {
             self.0.is_empty() & self.1.is_none()
         }
 
-        fn poll_data(
+        fn poll_frame(
             mut self: Pin<&mut Self>,
             _: &mut Context<'_>,
-        ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-            Poll::Ready(self.as_mut().0.pop_front().map(Cursor::new).map(Ok))
-        }
-
-        fn poll_trailers(
-            mut self: Pin<&mut Self>,
-            _: &mut Context<'_>,
-        ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
+        ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
             let mut this = self.as_mut();
-            assert!(this.0.is_empty());
-            Poll::Ready(Ok(this.1.take()))
+
+            // Return the next data frame from the sequence of chunks.
+            if let Some(chunk) = this.0.pop_front() {
+                let frame = Some(Ok(Frame::data(Cursor::new(chunk))));
+                return Poll::Ready(frame);
+            }
+
+            // Yield the trailers once all data frames have been yielded.
+            let trailers = this.1.take().map(Frame::<Self::Data>::trailers).map(Ok);
+            Poll::Ready(trailers)
         }
     }
 
@@ -464,18 +443,13 @@ mod tests {
             self.0.is_none()
         }
 
-        fn poll_data(
+        fn poll_frame(
             mut self: Pin<&mut Self>,
             _: &mut Context<'_>,
-        ) -> Poll<Option<Result<Self::Data, Self::Error>>> {
-            Poll::Ready(Some(Err(self.as_mut().0.take().expect("err"))))
-        }
+        ) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
+            let err = self.as_mut().0.take().expect("err");
 
-        fn poll_trailers(
-            mut self: Pin<&mut Self>,
-            _: &mut Context<'_>,
-        ) -> Poll<Result<Option<http::HeaderMap>, Self::Error>> {
-            Poll::Ready(Err(self.as_mut().0.take().expect("err")))
+            Poll::Ready(Some(Err(err)))
         }
     }
 }
@@ -18,6 +18,7 @@ pprof = ["linkerd-app-admin/pprof"]
 
 [dependencies]
 futures = { version = "0.3", default-features = false }
+hyper-util = { workspace = true }
 linkerd-app-admin = { path = "./admin" }
 linkerd-app-core = { path = "./core" }
 linkerd-app-gateway = { path = "./gateway" }
 
@@ -19,7 +19,8 @@ bytes = { workspace = true }
 deflate = { version = "1", optional = true, features = ["gzip"] }
 http = { workspace = true }
 http-body = { workspace = true }
-hyper = { workspace = true, features = ["deprecated", "http1", "http2"] }
+http-body-util = { workspace = true }
+hyper = { workspace = true, features = ["http1", "http2"] }
 futures = { version = "0.3", default-features = false }
 pprof = { version = "0.14", optional = true, features = ["prost-codec"] }
 serde = "1"