feat(volo-http client): add redirect-following support (#653)

fix(volo-http client): make FollowRedirect compose correctly with HttpProxy

Author: junliurs

examples/Cargo.toml

@@ -116,6 +116,9 @@ path = "src/http/example-http-server.rs"
 name = "example-http-client"
 path = "src/http/example-http-client.rs"
 [[bin]]
+name = "http-client-follow-redirects"
+path = "src/http/client-follow-redirects.rs"
+[[bin]]
 name = "http-tls-server"
 path = "src/http/http-tls-server.rs"
 required-features = ["__tls"]

examples/src/http/client-follow-redirects.rs

@@ -0,0 +1,108 @@
+use std::net::SocketAddr;
+
+use tokio::net::TcpListener;
+use volo::net::DefaultIncoming;
+use volo_http::{
+    body::BodyConversion,
+    client::Client,
+    error::BoxError,
+    server::{
+        Redirect, Server,
+        route::{Router, get, post},
+    },
+    utils::Extension,
+};
+
+#[derive(Clone)]
+struct RedirectTarget {
+    addr: SocketAddr,
+}
+
+async fn redirect_relative() -> Redirect {
+    Redirect::found("/final")
+}
+
+async fn redirect_cross_host(Extension(target): Extension<RedirectTarget>) -> Redirect {
+    Redirect::found(&format!("http://{}/landing", target.addr))
+}
+
+async fn redirect_post_to_get() -> Redirect {
+    Redirect::see_other("/method")
+}
+
+async fn final_relative() -> &'static str {
+    "followed relative redirect"
+}
+
+async fn landing() -> &'static str {
+    "followed cross-host redirect"
+}
+
+async fn method() -> &'static str {
+    "POST became GET after 303"
+}
+
+async fn bind_local() -> Result<(SocketAddr, DefaultIncoming), BoxError> {
+    let listener = TcpListener::bind("127.0.0.1:0").await?;
+    let addr = listener.local_addr()?;
+    Ok((addr, DefaultIncoming::from(listener)))
+}
+
+#[volo::main]
+async fn main() -> Result<(), BoxError> {
+    tracing_subscriber::fmt()
+        .with_max_level(tracing::Level::INFO)
+        .try_init()
+        .ok();
+
+    let (target_addr, target_incoming) = bind_local().await?;
+    let target_app = Router::new()
+        .route("/landing", get(landing))
+        .route("/method", get(method));
+    let target_server = tokio::spawn(Server::new(target_app).run(target_incoming));
+
+    let (entry_addr, entry_incoming) = bind_local().await?;
+    let entry_app = Router::new()
+        .route("/relative", get(redirect_relative))
+        .route("/cross-host", get(redirect_cross_host))
+        .route("/post-to-get", post(redirect_post_to_get))
+        .route("/method", get(method))
+        .route("/final", get(final_relative))
+        .layer(Extension(RedirectTarget { addr: target_addr }));
+    let entry_server = tokio::spawn(Server::new(entry_app).run(entry_incoming));
+
+    let client = Client::builder().follow_redirects(10).build()?;
+
+    let relative = client
+        .get(format!("http://{entry_addr}/relative"))
+        .send()
+        .await?
+        .into_string()
+        .await?;
+    let cross_host = client
+        .get(format!("http://{entry_addr}/cross-host"))
+        .send()
+        .await?
+        .into_string()
+        .await?;
+    let post_to_get = client
+        .post(format!("http://{entry_addr}/post-to-get"))
+        .data("payload that will be dropped")
+        .send()
+        .await?
+        .into_string()
+        .await?;
+
+    assert_eq!(relative, "followed relative redirect");
+    assert_eq!(cross_host, "followed cross-host redirect");
+    assert_eq!(post_to_get, "POST became GET after 303");
+
+    println!("relative: {relative}");
+    println!("cross-host: {cross_host}");
+    println!("post-to-get: {post_to_get}");
+
+    entry_server.abort();
+    target_server.abort();
+
+    Ok(())
+}

volo-http/src/body.rs

@@ -93,6 +93,25 @@ impl Body {
             repr: BodyRepr::Body(BoxBody::new(body.map_err(Into::into))),
         }
     }
+
+    /// Try to clone the body.
+    ///
+    /// Only in-memory bodies (created from bytes, string, [`Body::empty`], etc.) can be cloned,
+    /// and the clone is cheap since the underlying [`Bytes`] is reference-counted (a shallow
+    /// clone, the payload is not copied).
+    ///
+    /// Streaming bodies ([`Body::from_stream`], [`Body::from_incoming`], [`Body::from_body`]) are
+    /// one-shot and cannot be replayed, so this method returns [`None`] for them. This is mainly
+    /// used by redirect-following, where a request may need to be re-sent: a request with a
+    /// non-cloneable body will not be followed across redirects.
+    pub fn try_clone(&self) -> Option<Self> {
+        match &self.repr {
+            BodyRepr::Full(full) => Some(Self {
+                repr: BodyRepr::Full(full.clone()),
+            }),
+            BodyRepr::Hyper(_) | BodyRepr::Stream(_) | BodyRepr::Body(_) => None,
+        }
+    }
 }
 
 impl http_body::Body for Body {
@@ -402,4 +421,26 @@ mod tests {
         let body = Body::from(bytes);
         assert_eq!(body.into_string().await.unwrap(), "Hello, world!");
     }
+
+    #[tokio::test]
+    async fn test_try_clone_in_memory() {
+        // In-memory bodies can be cloned, and the clone carries the same content.
+        let body = Body::from("hello");
+        let cloned = body.try_clone().expect("full body should be cloneable");
+        assert_eq!(cloned.into_string().await.unwrap(), "hello");
+        assert_eq!(body.into_string().await.unwrap(), "hello");
+
+        // Empty body is also cloneable.
+        assert!(Body::empty().try_clone().is_some());
+    }
+
+    #[test]
+    fn test_try_clone_streaming_is_none() {
+        // Streaming bodies are one-shot and cannot be cloned.
+        let stream = futures_util::stream::empty::<
+            Result<http_body::Frame<Bytes>, crate::error::BoxError>,
+        >();
+        let body = Body::from_stream(stream);
+        assert!(body.try_clone().is_none());
+    }
 }

volo-http/src/client/layer/http_proxy.rs

@@ -15,7 +15,10 @@ use volo::{client::Apply, context::Context};
 use crate::{
     client::{Target, target::RemoteHost, utils::is_default_port},
     context::ClientContext,
-    error::{ClientError, client::request_error},
+    error::{
+        ClientError,
+        client::{Result, request_error},
+    },
     request::Request,
 };
 
@@ -132,34 +135,38 @@ pub struct HttpProxyService<S> {
 }
 
 impl<S> HttpProxyService<S> {
-    fn update_req<B>(&self, cx: &mut ClientContext, req: &mut Request<B>) {
+    fn update_req<B>(&self, cx: &mut ClientContext, req: &mut Request<B>) -> Option<Target> {
         let Some(target) = &self.target else {
-            return;
+            return None;
         };
+
+        // A configured proxy is a routing requirement. Unsupported requests must not silently
+        // fall back to a direct connection.
         if req.version() != Version::HTTP_11 {
             tracing::info!("[Volo-HTTP] HttpProxy only works for HTTP/1.1");
-            return;
+            return None;
         }
         if let Some(scheme) = cx.target().scheme() {
             if scheme != &Scheme::HTTP {
                 tracing::info!("[Volo-HTTP] HttpProxy only supports HTTP protocol");
-                return;
+                return None;
             }
         }
 
-        // Generate authority by old target, and then update request
+        // Generate authority from the logical upstream target, then rewrite the HTTP/1.1
+        // request-target to absolute-form for the proxy.
         let Some(authority) = gen_authority(cx.target()) else {
             tracing::warn!(
                 "[Volo-HTTP] HttpProxy: failed to gen authority by {:?}",
                 cx.target()
             );
-            return;
+            return None;
         };
         let authority = match Authority::from_maybe_shared(Bytes::from(authority)) {
             Ok(authority) => authority,
             Err(e) => {
                 tracing::warn!("[Volo-HTTP] HttpProxy: failed to parse authority: {e}");
-                return;
+                return None;
             }
         };
         let mut parts = req.uri().to_owned().into_parts();
@@ -174,19 +181,20 @@ impl<S> HttpProxyService<S> {
             Ok(uri) => uri,
             Err(e) => {
                 tracing::warn!("[Volo-HTTP] HttpProxy: failed to build uri: {e}");
-                return;
+                return None;
             }
         };
         *req.uri_mut() = uri;
 
-        // Clear callee and update proxy target to it
-        // Note: we must apply target after updating request because `target.apply(cx)` will update
-        // self to `cx.target`
+        // Only the transport target becomes the proxy. HttpProxy::call owns restoring the logical
+        // upstream after the inner call returns.
         cx.rpc_info_mut().callee_mut().clear();
-        target
+        let old_target = target
             .to_owned()
-            .apply(cx)
+            .apply_and_replace(cx)
             .expect("infallible: failed to parse target in HttpProxy");
+
+        Some(old_target)
     }
 }
 
@@ -227,6 +235,11 @@ fn gen_authority(target: &Target) -> Option<String> {
     Some(host)
 }
 
+fn restore_target(cx: &mut ClientContext, target: Target) -> Result<()> {
+    cx.rpc_info_mut().callee_mut().clear();
+    target.apply(cx)
+}
+
 impl<B, S> Service<ClientContext, Request<B>> for HttpProxyService<S>
 where
     B: Send,
@@ -240,8 +253,17 @@ where
         cx: &mut ClientContext,
         mut req: Request<B>,
     ) -> Result<Self::Response, Self::Error> {
-        self.update_req(cx, &mut req);
-        match self.inner.call(cx, req).await {
+        let old_target = self.update_req(cx, &mut req);
+        let result = self.inner.call(cx, req).await;
+
+        // HttpProxy only borrows cx.target() as a transport target. Restore the logical upstream
+        // before returning to FollowRedirect or any other outer layer. Do this for both success and
+        // error responses.
+        if let Some(target) = old_target {
+            restore_target(cx, target)?;
+        };
+
+        match result {
             Ok(resp) => Ok(resp),
             Err(e) => {
                 if let Some(target) = &self.target {

volo-http/src/client/layer/mod.rs

@@ -6,11 +6,13 @@ mod fail_on_status;
 pub mod header;
 #[cfg(feature = "http1")]
 pub mod http_proxy;
+mod redirect;
 mod timeout;
-mod utils;
+pub(crate) mod utils;
 
 pub use self::{
     fail_on_status::{FailOnStatus, StatusCodeError},
+    redirect::{FollowRedirect, RedirectPredicate},
     timeout::Timeout,
     utils::TargetLayer,
 };