vey-proxy: support h2 informational response

Author: zh-jq

lib/vey-h2/src/client/mod.rs

@@ -0,0 +1,7 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2026 VEY-OSS developers.
+ */
+
+mod response;
+pub use response::H2ResponseHeaderReceiver;

lib/vey-h2/src/client/response.rs

@@ -0,0 +1,47 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright 2026 VEY-OSS developers.
+ */
+
+use std::future::poll_fn;
+use std::pin::Pin;
+use std::task::{Context, Poll, ready};
+
+use h2::client::ResponseFuture;
+use h2::{Error, RecvStream};
+use http::Response;
+
+pub struct H2ResponseHeaderReceiver {
+    rsp_fut: ResponseFuture,
+    rsp_body: Option<RecvStream>,
+}
+
+impl H2ResponseHeaderReceiver {
+    pub fn new(rsp_fut: ResponseFuture) -> Self {
+        Self {
+            rsp_fut,
+            rsp_body: None,
+        }
+    }
+
+    fn poll_recv(&mut self, cx: &mut Context<'_>) -> Poll<Result<Response<()>, Error>> {
+        if let Some(r) = ready!(self.rsp_fut.poll_informational(cx)) {
+            return Poll::Ready(r);
+        }
+
+        let r = ready!(Pin::new(&mut self.rsp_fut).poll(cx))?;
+        let (header, body) = r.into_parts();
+
+        self.rsp_body = Some(body);
+        Poll::Ready(Ok(Response::from_parts(header, ())))
+    }
+
+    pub async fn recv_header(&mut self) -> Result<Response<()>, Error> {
+        poll_fn(move |cx| self.poll_recv(cx)).await
+    }
+
+    #[inline]
+    pub fn take_body(&mut self) -> Option<RecvStream> {
+        self.rsp_body.take()
+    }
+}

lib/vey-h2/src/ext/request.rs

@@ -7,14 +7,15 @@ use std::io::Write;
 
 use bytes::BufMut;
 use http::uri::Authority;
-use http::{HeaderMap, Method, Request, Uri};
+use http::{HeaderMap, Method, Request, Uri, header};
 
 use vey_http::server::HttpAdaptedRequest;
 
 pub trait RequestExt {
     fn serialize_for_adapter(&self) -> Vec<u8>;
     fn adapt_to(self, other: &HttpAdaptedRequest) -> Self;
     fn clone_header(&self) -> Request<()>;
+    fn expect_100_continue(&self) -> bool;
 }
 
 impl<T> RequestExt for Request<T> {
@@ -30,7 +31,7 @@ impl<T> RequestExt for Request<T> {
             let _ = write!(buf, "{method} / HTTP/1.1\r\n");
         }
         for (name, value) in self.headers() {
-            if matches!(name, &http::header::TE) {
+            if matches!(name, &header::TE) {
                 // skip hop-by-hop headers
                 continue;
             }
@@ -39,7 +40,7 @@ impl<T> RequestExt for Request<T> {
             buf.put_slice(value.as_bytes());
             buf.put_slice(b"\r\n");
         }
-        if !self.headers().contains_key(http::header::HOST)
+        if !self.headers().contains_key(header::HOST)
             && let Some(host) = uri.host()
         {
             buf.put_slice(b"Host: ");
@@ -53,19 +54,19 @@ impl<T> RequestExt for Request<T> {
     fn adapt_to(self, other: &HttpAdaptedRequest) -> Self {
         let mut headers = HeaderMap::from(&other.headers);
         // add hop-by-hop headers
-        if let Some(v) = self.headers().get(http::header::TE) {
-            headers.insert(http::header::TE, v.into());
+        if let Some(v) = self.headers().get(header::TE) {
+            headers.insert(header::TE, v.into());
         }
         let (mut parts, body) = self.into_parts();
         parts.method = other.method.clone();
         let mut uri_parts = other.uri.clone().into_parts();
         uri_parts.scheme = parts.uri.scheme().cloned();
         uri_parts.authority = parts.uri.authority().cloned();
-        if let Some(host) = headers.remove(http::header::HOST) {
+        if let Some(host) = headers.remove(header::HOST) {
             // we should always remove the Host header to be compatible with Google,
             // but let's keep the same as client behaviour here
-            if parts.headers.contains_key(http::header::HOST) {
-                headers.insert(http::header::HOST, host.clone());
+            if parts.headers.contains_key(header::HOST) {
+                headers.insert(header::HOST, host.clone());
             }
             if uri_parts.authority.is_none()
                 && let Ok(authority) = Authority::from_maybe_shared(host.clone())
@@ -90,4 +91,14 @@ impl<T> RequestExt for Request<T> {
         parts.headers = self.headers().clone();
         Request::from_parts(parts, ())
     }
+
+    fn expect_100_continue(&self) -> bool {
+        for v in self.headers().get_all(header::EXPECT) {
+            if v.as_bytes() == b"100-continue" {
+                return true;
+            }
+        }
+
+        false
+    }
 }

lib/vey-h2/src/lib.rs

@@ -8,3 +8,6 @@ pub use body::*;
 
 mod ext;
 pub use ext::{RequestExt, ResponseExt};
+
+mod client;
+pub use client::H2ResponseHeaderReceiver;

lib/vey-icap-client/src/reqmod/h2/bidirectional.rs

@@ -8,11 +8,12 @@ use std::time::Duration;
 
 use bytes::Bytes;
 use h2::client::SendRequest;
-use http::Request;
+use h2::server::SendResponse;
+use http::{Request, Response, StatusCode};
 
 use vey_h2::{
-    H2StreamFromChunkedTransfer, H2StreamFromChunkedTransferError, H2StreamToChunkedTransfer,
-    H2StreamToChunkedTransferError, RequestExt,
+    H2ResponseHeaderReceiver, H2StreamFromChunkedTransfer, H2StreamFromChunkedTransferError,
+    H2StreamToChunkedTransfer, H2StreamToChunkedTransferError, RequestExt,
 };
 use vey_http::server::HttpAdaptedRequest;
 use vey_io_ext::{IdleCheck, LimitedBufReadExt, StreamCopyConfig};
@@ -110,7 +111,9 @@ impl<I: IdleCheck> BidirectionalRecvHttpRequest<'_, I> {
         state: &mut ReqmodAdaptationRunState,
         mut clt_body_transfer: &mut H2StreamToChunkedTransfer<'_, IcapClientWriter>,
         orig_http_request: Request<()>,
-        mut ups_send_request: SendRequest<Bytes>,
+        mut ups_send_req: SendRequest<Bytes>,
+        clt_send_rsp: &mut SendResponse<Bytes>,
+        mut allow_continue: bool,
     ) -> Result<ReqmodAdaptationEndState, H2ReqmodAdaptationError> {
         let http_req = HttpAdaptedRequest::parse(
             self.icap_reader,
@@ -120,7 +123,7 @@ impl<I: IdleCheck> BidirectionalRecvHttpRequest<'_, I> {
         .await?;
 
         let final_req = orig_http_request.adapt_to(&http_req);
-        let (mut ups_recv_rsp, mut ups_send_stream) = ups_send_request
+        let (rsp_fut, mut ups_send_stream) = ups_send_req
             .send_request(final_req, false)
             .map_err(H2ReqmodAdaptationError::HttpUpstreamSendHeadFailed)?;
         state.mark_ups_send_header();
@@ -133,22 +136,51 @@ impl<I: IdleCheck> BidirectionalRecvHttpRequest<'_, I> {
             self.http_trailer_max_size,
         );
 
+        let mut ups_recv_rsp = H2ResponseHeaderReceiver::new(rsp_fut);
+
         let mut idle_interval = self.idle_checker.interval_timer();
         let mut idle_count = 0;
 
         loop {
             tokio::select! {
-                r = &mut ups_recv_rsp => {
-                    return match r {
+                r = ups_recv_rsp.recv_header() => {
+                     match r {
                         Ok(ups_rsp) => {
-                            state.mark_ups_recv_header();
-                            if ups_body_transfer.finished() {
-                                self.icap_read_finished = true;
+                            match ups_rsp.status() {
+                                StatusCode::CONTINUE => {
+                                    if allow_continue {
+                                        clt_send_rsp
+                                            .send_informational(ups_rsp)
+                                            .map_err(H2ReqmodAdaptationError::HttpClientSendResponseFailed)?;
+                                        allow_continue = false;
+                                    } else {
+                                        return Err(H2ReqmodAdaptationError::InvalidUpstreamContinueResponse);
+                                    }
+                                }
+                                StatusCode::EARLY_HINTS => {
+                                    clt_send_rsp
+                                        .send_informational(ups_rsp)
+                                        .map_err(H2ReqmodAdaptationError::HttpClientSendResponseFailed)?;
+                                }
+                                status => {
+                                    state.mark_ups_recv_header();
+                                    return if let Some(body) = ups_recv_rsp.take_body() {
+                                        let (headers, _) = ups_rsp.into_parts();
+                                        if ups_body_transfer.finished() {
+                                            self.icap_read_finished = true;
+                                        }
+                                        let ups_rsp = Response::from_parts(headers, body);
+                                        Ok(ReqmodAdaptationEndState::AdaptedTransferred(http_req, ups_rsp))
+                                    } else {
+                                        Err(H2ReqmodAdaptationError::UnsupportedInformationalResponse(
+                                            status,
+                                        ))
+                                    };
+                                }
                             }
-                            Ok(ReqmodAdaptationEndState::AdaptedTransferred(http_req, ups_rsp))
                         }
-                        Err(e) => Err(H2ReqmodAdaptationError::HttpUpstreamRecvResponseFailed(e)),
-                    };
+                        Err(e) => return Err(H2ReqmodAdaptationError::HttpUpstreamRecvResponseFailed(e)),
+                    }
                 }
                 r = &mut clt_body_transfer => {
                     return match r {
@@ -157,7 +189,7 @@ impl<I: IdleCheck> BidirectionalRecvHttpRequest<'_, I> {
                                 Ok(_) => {
                                     state.mark_ups_send_all();
                                     self.icap_read_finished = true;
-                                    let ups_rsp = recv_ups_response_head_after_transfer(ups_recv_rsp, self.http_rsp_head_recv_timeout).await?;
+                                    let ups_rsp = recv_ups_response_head_after_transfer(&mut ups_recv_rsp, clt_send_rsp, allow_continue, self.http_rsp_head_recv_timeout).await?;
                                     Ok(ReqmodAdaptationEndState::AdaptedTransferred(http_req, ups_rsp))
                                 }
                                 Err(H2StreamFromChunkedTransferError::ReadError(e)) => Err(H2ReqmodAdaptationError::IcapServerReadFailed(e)),
@@ -176,7 +208,7 @@ impl<I: IdleCheck> BidirectionalRecvHttpRequest<'_, I> {
                         Ok(_) => {
                             state.mark_ups_send_all();
                             self.icap_read_finished = true;
-                            let ups_rsp = recv_ups_response_head_after_transfer(ups_recv_rsp, self.http_rsp_head_recv_timeout).await?;
+                            let ups_rsp = recv_ups_response_head_after_transfer(&mut ups_recv_rsp, clt_send_rsp, allow_continue, self.http_rsp_head_recv_timeout).await?;
                             Ok(ReqmodAdaptationEndState::AdaptedTransferred(http_req, ups_rsp))
                         }
                         Err(H2StreamFromChunkedTransferError::ReadError(e)) => Err(H2ReqmodAdaptationError::IcapServerReadFailed(e)),

lib/vey-icap-client/src/reqmod/h2/error.rs

@@ -5,6 +5,7 @@
 
 use std::io;
 
+use http::StatusCode;
 use thiserror::Error;
 
 use vey_h2::H2PreviewError;
@@ -47,6 +48,12 @@ pub enum H2ReqmodAdaptationError {
     HttpUpstreamRecvResponseFailed(h2::Error),
     #[error("recv response from http upstream timeout")]
     HttpUpstreamRecvResponseTimeout,
+    #[error("invalid http upstream 100-continue response")]
+    InvalidUpstreamContinueResponse,
+    #[error("unsupported http upstream informational response {0}")]
+    UnsupportedInformationalResponse(StatusCode),
+    #[error("send response to client failed: {0}")]
+    HttpClientSendResponseFailed(h2::Error),
     #[error("internal server error: {0}")]
     InternalServerError(&'static str),
     #[error("force quit from idle checker: {0:?}")]

lib/vey-icap-client/src/reqmod/h2/forward_body.rs

@@ -8,6 +8,7 @@ use std::io::{IoSlice, Write};
 use bytes::{BufMut, Bytes};
 use h2::RecvStream;
 use h2::client::SendRequest;
+use h2::server::SendResponse;
 use http::Request;
 
 use vey_h2::{
@@ -42,7 +43,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
         http_request: Request<()>,
         preview_data: H2PreviewData,
         clt_body: RecvStream,
-        ups_send_request: SendRequest<Bytes>,
+        ups_send_req: SendRequest<Bytes>,
+        clt_send_rsp: &mut SendResponse<Bytes>,
     ) -> Result<ReqmodAdaptationEndState, H2ReqmodAdaptationError> {
         let http_header = http_request.serialize_for_adapter();
         let icap_header = self.build_forward_all_request(http_header.len());
@@ -99,7 +101,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
                     rsp,
                     header_size,
                     http_request,
-                    ups_send_request,
+                    ups_send_req,
+                    clt_send_rsp,
                 )
                 .await
             }
@@ -109,7 +112,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
                     rsp,
                     header_size,
                     http_request,
-                    ups_send_request,
+                    ups_send_req,
+                    clt_send_rsp,
                 )
                 .await
             }
@@ -179,7 +183,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
         state: &mut ReqmodAdaptationRunState,
         http_request: Request<()>,
         mut clt_body: RecvStream,
-        ups_send_request: SendRequest<Bytes>,
+        ups_send_req: SendRequest<Bytes>,
+        clt_send_rsp: &mut SendResponse<Bytes>,
     ) -> Result<ReqmodAdaptationEndState, H2ReqmodAdaptationError> {
         let http_header = http_request.serialize_for_adapter();
         let icap_header = self.build_forward_all_request(http_header.len());
@@ -242,7 +247,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
                     rsp,
                     header_size,
                     http_request,
-                    ups_send_request,
+                    ups_send_req,
+                    clt_send_rsp,
                 )
                 .await
             }
@@ -254,7 +260,8 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
                         rsp,
                         header_size,
                         http_request,
-                        ups_send_request,
+                        ups_send_req,
+                        clt_send_rsp,
                     )
                     .await
                 } else {
@@ -270,7 +277,14 @@ impl<I: IdleCheck> H2RequestAdapter<I> {
                         icap_read_finished: false,
                     };
                     let r = bidirectional_transfer
-                        .transfer(state, &mut body_transfer, http_request, ups_send_request)
+                        .transfer(
+                            state,
+                            &mut body_transfer,
+                            http_request,
+                            ups_send_req,
+                            clt_send_rsp,
+                            self.allow_continue,
+                        )
                         .await?;
 
                     let icap_read_finished = bidirectional_transfer.icap_read_finished;