diff --git a/.github/workflows/CI.yml b/.github/workflows/CI.yml index e7d27a47..d8f03580 100644 --- a/.github/workflows/CI.yml +++ b/.github/workflows/CI.yml @@ -17,7 +17,8 @@ jobs: - test - msrv - miri - - features + - client-features + - server-features - semver - doc steps: @@ -94,15 +95,27 @@ jobs: MIRIFLAGS: "-Zmiri-disable-isolation" run: cargo miri test --all-features - features: - name: features + client-features: + name: client-features needs: [style] runs-on: ubuntu-latest steps: - uses: actions/checkout@v7 - uses: dtolnay/rust-toolchain@stable - uses: taiki-e/install-action@cargo-hack - - run: cargo hack --no-dev-deps check --feature-powerset --depth 2 + - uses: taiki-e/install-action@just + - run: just check-client-features + + server-features: + name: server-features + needs: [style] + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: dtolnay/rust-toolchain@stable + - uses: taiki-e/install-action@cargo-hack + - uses: taiki-e/install-action@just + - run: just check-server-features semver: name: semver diff --git a/Cargo.toml b/Cargo.toml index d2c2dcb8..39b9c2dc 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -100,9 +100,6 @@ tracing = ["dep:tracing"] # Temporary compatibility opt-in, deliberately excluded from `full`. rt-tracing-exec-force = ["tokio", "tracing"] -# internal features used in CI -__internal_happy_eyeballs_tests = [] - [[example]] name = "client" required-features = ["client-legacy", "http1", "tokio"] diff --git a/justfile b/justfile new file mode 100644 index 00000000..4c84dec7 --- /dev/null +++ b/justfile @@ -0,0 +1,9 @@ +check-client-features: + cargo hack --no-dev-deps check --feature-powerset --depth 2 \ + --features=client \ + --include-features client-legacy,client-pool,client-proxy,client-proxy-system,http1,http2,tokio,tracing + +check-server-features: + cargo hack --no-dev-deps check --feature-powerset --depth 2 \ + --features=server \ + --include-features server-auto,server-graceful,http1,http2,tokio,tracing diff --git a/src/client/legacy/client.rs b/src/client/legacy/client.rs index b11c7d91..9bab238e 100644 --- a/src/client/legacy/client.rs +++ b/src/client/legacy/client.rs @@ -24,9 +24,10 @@ use tracing::{debug, trace, warn}; use super::connect::HttpConnector; use super::connect::capture::CaptureConnectionExtension; use super::connect::{Alpn, Connect, Connected, Connection}; +use super::exec::Exec; +use super::lazy::{Started as Lazy, lazy as hyper_lazy}; use super::pool::{self, Ver}; - -use crate::common::{Exec, Lazy, lazy as hyper_lazy, timer}; +use super::timer; type BoxSendFuture = Pin + Send>>; @@ -64,6 +65,7 @@ pub struct Error { #[derive(Debug)] enum ErrorKind { Canceled, + #[cfg(feature = "http1")] ChannelClosed, Connect, UserUnsupportedRequestMethod, @@ -1549,7 +1551,10 @@ impl Builder { /// details. /// /// [`hyper::client::conn::http2::Builder::timer`]: https://docs.rs/hyper/latest/hyper/client/conn/http2/struct.Builder.html#method.timer - pub fn timer(&mut self, timer: M) -> &mut Self + pub fn timer( + &mut self, + #[cfg_attr(not(feature = "http2"), allow(unused_variables))] timer: M, + ) -> &mut Self where M: Timer + Send + Sync + 'static, { @@ -1707,6 +1712,7 @@ impl Error { e!(SendRequest, src) } + #[cfg(feature = "http1")] fn closed(src: hyper::Error) -> Self { e!(ChannelClosed, src) } diff --git a/src/client/legacy/connect/capture.rs b/src/client/legacy/connect/capture.rs index e57fd972..e6e0a9f8 100644 --- a/src/client/legacy/connect/capture.rs +++ b/src/client/legacy/connect/capture.rs @@ -80,10 +80,12 @@ pub fn capture_connection(request: &mut Request) -> CaptureConnection { /// This is inserted into `Extensions` to allow Hyper to back channel connection info #[derive(Clone)] pub(crate) struct CaptureConnectionExtension { + #[cfg_attr(not(any(feature = "http1", feature = "http2")), allow(dead_code))] tx: Arc>>, } impl CaptureConnectionExtension { + #[cfg(any(feature = "http1", feature = "http2"))] pub(crate) fn set(&self, connected: &Connected) { self.tx.send_replace(Some(connected.clone())); } @@ -119,7 +121,7 @@ impl CaptureConnection { } } -#[cfg(test)] +#[cfg(all(test, any(feature = "http1", feature = "http2")))] mod test { use super::*; diff --git a/src/client/legacy/connect/http.rs b/src/client/legacy/connect/http.rs index cc3e4cee..95af1627 100644 --- a/src/client/legacy/connect/http.rs +++ b/src/client/legacy/connect/http.rs @@ -1020,7 +1020,6 @@ fn set_port(addr: &mut SocketAddr, host_port: u16, explicit: bool) { #[cfg(test)] mod tests { - use std::io; use std::net::SocketAddr; use ::http::Uri; @@ -1028,7 +1027,7 @@ mod tests { use crate::client::legacy::connect::http::TcpKeepaliveConfig; use super::super::sealed::{Connect, ConnectSvc}; - use super::{Config, ConnectError, HttpConnector}; + use super::HttpConnector; use super::set_port; diff --git a/src/client/legacy/connect/mod.rs b/src/client/legacy/connect/mod.rs index fff8262f..47697dd2 100644 --- a/src/client/legacy/connect/mod.rs +++ b/src/client/legacy/connect/mod.rs @@ -132,6 +132,7 @@ impl PoisonPill { self.poisoned.store(true, Ordering::Relaxed) } + #[cfg(any(feature = "http1", feature = "http2"))] pub(crate) fn poisoned(&self) -> bool { self.poisoned.load(Ordering::Relaxed) } @@ -224,6 +225,7 @@ impl Connected { // Don't public expose that `Connected` is `Clone`, unsure if we want to // keep that contract... + #[cfg(any(feature = "http1", feature = "http2"))] pub(super) fn clone(&self) -> Connected { Connected { alpn: self.alpn, diff --git a/src/common/exec.rs b/src/client/legacy/exec.rs similarity index 100% rename from src/common/exec.rs rename to src/client/legacy/exec.rs diff --git a/src/common/lazy.rs b/src/client/legacy/lazy.rs similarity index 100% rename from src/common/lazy.rs rename to src/client/legacy/lazy.rs diff --git a/src/client/legacy/mod.rs b/src/client/legacy/mod.rs index 1649ae7e..5e0d5c46 100644 --- a/src/client/legacy/mod.rs +++ b/src/client/legacy/mod.rs @@ -2,6 +2,11 @@ mod client; #[cfg(any(feature = "http1", feature = "http2"))] pub use client::{Builder, Client, Error, ResponseFuture}; +#[cfg(any(feature = "http1", feature = "http2"))] +mod lazy; + +mod exec; +mod timer; pub mod connect; #[doc(hidden)] diff --git a/src/client/legacy/pool.rs b/src/client/legacy/pool.rs index 02b75604..10ef9095 100644 --- a/src/client/legacy/pool.rs +++ b/src/client/legacy/pool.rs @@ -17,7 +17,8 @@ use tracing::{debug, trace}; use hyper::rt::Timer as _; -use crate::common::{exec, exec::Exec, timer::Timer}; +use super::exec::{self, Exec}; +use super::timer::Timer; // FIXME: allow() required due to `impl Trait` leaking types to this lint #[allow(missing_debug_implementations)] @@ -846,10 +847,9 @@ mod tests { use std::time::Duration; use super::{Connecting, Key, Pool, Poolable, Reservation, WeakOpt}; + use crate::client::legacy::timer; use crate::rt::{TokioExecutor, TokioTimer}; - use crate::common::timer; - #[derive(Clone, Debug, PartialEq, Eq, Hash)] struct KeyImpl(http::uri::Scheme, http::uri::Authority); diff --git a/src/common/timer.rs b/src/client/legacy/timer.rs similarity index 100% rename from src/common/timer.rs rename to src/client/legacy/timer.rs diff --git a/src/common/mod.rs b/src/common/mod.rs deleted file mode 100644 index 7131c0b9..00000000 --- a/src/common/mod.rs +++ /dev/null @@ -1,15 +0,0 @@ -#![allow(missing_docs)] - -pub(crate) mod exec; -#[cfg(feature = "client-legacy")] -mod lazy; -#[cfg(feature = "server")] -// #[cfg(feature = "server-auto")] -pub(crate) mod rewind; -pub(crate) mod timer; - -#[cfg(feature = "client-legacy")] -pub(crate) use exec::Exec; - -#[cfg(feature = "client-legacy")] -pub(crate) use lazy::{Started as Lazy, lazy}; diff --git a/src/lib.rs b/src/lib.rs index 88101741..69535755 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ #![deny(missing_docs)] +#![deny(unused)] #![cfg_attr(docsrs, feature(doc_cfg))] //! Utilities for working with hyper. @@ -60,7 +61,6 @@ #[cfg(feature = "client")] pub mod client; -mod common; pub mod rt; #[cfg(feature = "server")] pub mod server; diff --git a/src/server/conn/auto/mod.rs b/src/server/conn/auto/mod.rs index 30f0db39..bb905c1a 100644 --- a/src/server/conn/auto/mod.rs +++ b/src/server/conn/auto/mod.rs @@ -1,5 +1,6 @@ //! Http1 or Http2 connection. +mod rewind; pub mod upgrade; use hyper::service::HttpService; @@ -29,7 +30,7 @@ use std::marker::PhantomData; use pin_project_lite::pin_project; -use crate::common::rewind::Rewind; +use self::rewind::Rewind; type Error = Box; @@ -1132,22 +1133,27 @@ impl Http2Builder<'_, E> { } } -#[cfg(all(feature = "tokio", test))] +#[cfg(all(feature = "tokio", feature = "http1", feature = "http2", test))] mod tests { - use crate::{ - rt::{TokioExecutor, TokioIo}, - server::conn::auto, - }; - use http::{Request, Response}; - use http_body::Body; - use http_body_util::{BodyExt, Empty, Full}; - use hyper::{body, body::Bytes, client, service::service_fn}; - use std::{convert::Infallible, error::Error as StdError, net::SocketAddr, time::Duration}; - use tokio::{ - net::{TcpListener, TcpStream}, - pin, - }; + use crate::{rt::TokioExecutor, server::conn::auto}; + #[cfg(not(miri))] + use prelude::*; + #[cfg(not(miri))] + mod prelude { + pub use crate::rt::TokioIo; + pub use http::{Request, Response}; + pub use http_body::Body; + pub use http_body_util::{BodyExt, Empty, Full}; + pub use hyper::client; + pub use hyper::{body, body::Bytes, service::service_fn}; + pub use std::{convert::Infallible, time::Duration}; + pub use std::{error::Error as StdError, net::SocketAddr}; + pub use tokio::net::TcpListener; + pub use tokio::net::TcpStream; + } + + #[cfg(not(miri))] const BODY: &[u8] = b"Hello, world!"; #[test] @@ -1299,7 +1305,7 @@ mod tests { let builder = auto::Builder::new(TokioExecutor::new()); let connection = builder.serve_connection(stream, service_fn(hello)); - pin!(connection); + tokio::pin!(connection); connection.as_mut().graceful_shutdown(); @@ -1314,6 +1320,7 @@ mod tests { assert_eq!(connection_error.kind(), std::io::ErrorKind::Interrupted); } + #[cfg(not(miri))] async fn connect_h1(addr: SocketAddr) -> client::conn::http1::SendRequest where B: Body + Send + 'static, @@ -1328,6 +1335,7 @@ mod tests { sender } + #[cfg(not(miri))] async fn connect_h2(addr: SocketAddr) -> client::conn::http2::SendRequest where B: Body + Unpin + Send + 'static, @@ -1345,6 +1353,7 @@ mod tests { sender } + #[cfg(not(miri))] async fn start_server(h1_only: bool, h2_only: bool) -> SocketAddr { let addr: SocketAddr = ([127, 0, 0, 1], 0).into(); let listener = TcpListener::bind(addr).await.unwrap(); @@ -1378,6 +1387,7 @@ mod tests { local_addr } + #[cfg(not(miri))] async fn hello(_req: Request) -> Result>, Infallible> { Ok(Response::new(Full::new(Bytes::from(BODY)))) } diff --git a/src/common/rewind.rs b/src/server/conn/auto/rewind.rs similarity index 100% rename from src/common/rewind.rs rename to src/server/conn/auto/rewind.rs diff --git a/src/server/conn/auto/upgrade.rs b/src/server/conn/auto/upgrade.rs index 8d94c409..291ee8e8 100644 --- a/src/server/conn/auto/upgrade.rs +++ b/src/server/conn/auto/upgrade.rs @@ -6,7 +6,7 @@ use hyper::{ upgrade::Upgraded, }; -use crate::common::rewind::Rewind; +use super::rewind::Rewind; /// Tries to downcast the internal trait object to the type passed. /// diff --git a/tests/legacy_client_connection_regression.rs b/tests/legacy_client_connection_regression.rs new file mode 100644 index 00000000..73ce1267 --- /dev/null +++ b/tests/legacy_client_connection_regression.rs @@ -0,0 +1,436 @@ +#![cfg(all(feature = "client-legacy", feature = "http1"))] + +// ------------------------------------------------------- +// Below is our custom code for testing hyper legacy-client behavior with mock connections for PR #184 +// We use fully qualified paths for all types and identifiers to make this code +// copy/paste-able without relying on external 'use' statements. Detailed inline +// comments explain the purpose and logic of each section. + +//XXX: can manually run like this: +// $ cargo test --features="http1,http2,server,client-legacy" --test legacy_client -- test_connection_error_propagation test_incomplete_message_error --nocapture +// $ cargo test --all-features --test legacy_client -- --nocapture +// $ cargo test --all-features --test legacy_client + +use std::error::Error; // needed for .source() eg. error[E0599]: no method named `source` found for struct `hyper_util::client::legacy::Error` in the current scope + +// Helper function to debug byte slices by attempting to interpret them as UTF-8. +// If the bytes are valid UTF-8, they are printed as a string; otherwise, they are +// printed as a raw byte array. This aids in debugging tokio_test::io::Mock mismatches. +fn debug_bytes(bytes: &[u8], label: &str) { + // Try to convert the byte slice to a UTF-8 string. + // If successful, print it with the provided label for context. + if let Ok(s) = std::str::from_utf8(bytes) { + eprintln!("{}: {}", label, s); + } else { + // If the bytes are not valid UTF-8, print them as a raw byte array. + eprintln!("{}: {:?}", label, bytes); + } +} + +// Struct representing a mock connection for testing hyper client behavior. +// Implements hyper::rt::Read, hyper::rt::Write, and hyper_util::client::legacy::connect::Connection +// traits to simulate I/O operations. Uses tokio_test::io::Mock for controlled I/O behavior. +struct MockConnection { + // The underlying mock I/O object, wrapped in hyper_util::rt::TokioIo for compatibility. + inner: hyper_util::rt::TokioIo, + // Atomic flag to signal a connection failure, controlling poll_read behavior. + failed: std::sync::Arc, + // The error to return when failed=true, simulating an I/O failure. + error: std::sync::Arc, + // Optional channel to signal unexpected writes, used for debugging. + error_tx: Option>, + // Tracks total bytes written, for logging and verification. + bytes_written: usize, +} + +impl MockConnection { + // Constructor for MockConnection, initializing all fields. + // Takes a mock I/O object, failure flag, error, and optional error channel. + fn new( + mock: tokio_test::io::Mock, + failed: std::sync::Arc, + error: std::sync::Arc, + error_tx: Option>, + ) -> Self { + MockConnection { + inner: hyper_util::rt::TokioIo::new(mock), + failed, + error, + error_tx, + bytes_written: 0, + } + } +} + +// Implement hyper::rt::Read trait to handle read operations on the mock connection. +// Controls whether an error or mock I/O data is returned based on the failed flag. +impl hyper::rt::Read for MockConnection { + // Polls the connection for reading, filling the provided buffer. + // If failed=true, returns the stored error; otherwise, delegates to the mock I/O. + fn poll_read( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: hyper::rt::ReadBufCursor<'_>, + ) -> std::task::Poll> { + // Log the current state of the failed flag for debugging. + eprintln!( + "poll_read: failed={}", + self.failed.load(std::sync::atomic::Ordering::SeqCst) + ); + // Check if the connection is marked as failed. + // If true, return the stored error immediately to simulate a connection failure. + if self.failed.load(std::sync::atomic::Ordering::SeqCst) { + // Log the error being returned for traceability. + eprintln!("poll_read: returning error: {}", self.error); + // Create a new io::Error with the same kind and message as the stored error. + return std::task::Poll::Ready(std::result::Result::Err(std::io::Error::new( + self.error.kind(), + self.error.to_string(), + ))); + } + // If not failed, delegate to the mock I/O to simulate normal read behavior. + // This may return EOF (Poll::Ready(Ok(0))) for empty IoBuilder. + let inner = std::pin::Pin::new(&mut self.inner); + inner.poll_read(cx, buf) + } +} + +// Implement hyper::rt::Write trait to handle write operations on the mock connection. +// Logs writes and signals unexpected writes via error_tx. +impl hyper::rt::Write for MockConnection { + // Polls the connection for writing, sending the provided buffer. + // Logs the write operation and tracks total bytes written. + fn poll_write( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + buf: &[u8], + ) -> std::task::Poll> { + // Log the size of the buffer being written for debugging. + eprintln!("poll_write: {} bytes", buf.len()); + // Debug the buffer contents as UTF-8 or raw bytes. + debug_bytes(buf, "poll_write buffer"); + // Delegate the write to the mock I/O object. + let inner = std::pin::Pin::new(&mut self.inner); + match inner.poll_write(cx, buf) { + // If the write succeeds, update the bytes_written counter and log the result. + std::task::Poll::Ready(std::result::Result::Ok(bytes)) => { + // Increment the total bytes written for tracking. + self.bytes_written += bytes; + // Log the number of bytes written and the running total. + eprintln!( + "poll_write: wrote {} bytes, total={}", + bytes, self.bytes_written + ); + // If error_tx is present, signal an unexpected write (used in error tests). + // This helps detect writes when the connection should fail early. + if let Some(tx) = self.error_tx.take() { + // Log that an unexpected write is being signaled. + eprintln!("poll_write: signaling unexpected write"); + // Send a message through the channel, ignoring errors if the receiver is closed. + let _ = tx.try_send(()); + } + // Return the successful write result. + std::task::Poll::Ready(std::result::Result::Ok(bytes)) + } + // For pending or error results, propagate them directly. + other => other, + } + } + + // Polls the connection to flush any buffered data. + // Delegates to the mock I/O object. + fn poll_flush( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + // Log the flush operation for debugging. + eprintln!("poll_flush"); + // Delegate the flush to the mock I/O object. + let inner = std::pin::Pin::new(&mut self.inner); + inner.poll_flush(cx) + } + + // Polls the connection to shut down the write side. + // Delegates to the mock I/O object. + fn poll_shutdown( + mut self: std::pin::Pin<&mut Self>, + cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + // Log the shutdown operation for debugging. + eprintln!("poll_shutdown"); + // Delegate the shutdown to the mock I/O object. + let inner = std::pin::Pin::new(&mut self.inner); + inner.poll_shutdown(cx) + } +} + +// Implement hyper_util::client::legacy::connect::Connection trait to provide connection metadata. +// Required for hyper to use MockConnection as a valid connection. +impl hyper_util::client::legacy::connect::Connection for MockConnection { + // Returns metadata about the connection. + // In this case, a default Connected object indicating a new connection. + fn connected(&self) -> hyper_util::client::legacy::connect::Connected { + hyper_util::client::legacy::connect::Connected::new() + } +} + +// Struct representing a mock connector for creating MockConnection instances. +// Implements tower_service::Service to integrate with hyper’s client. +#[derive(Clone)] +struct MockConnector { + // The IoBuilder used to create mock I/O objects for each connection. + io_builder: tokio_test::io::Builder, + // Optional error to simulate a connection failure, passed to MockConnection. + conn_error: Option>, +} + +impl MockConnector { + // Constructor for MockConnector, initializing the IoBuilder and optional error. + fn new( + io_builder: tokio_test::io::Builder, + conn_error: Option>, + ) -> Self { + MockConnector { + io_builder, + conn_error, + } + } +} + +// Implement tower_service::Service for MockConnector to create MockConnection instances. +// Takes a hyper::Uri and returns a future resolving to a MockConnection. +impl tower_service::Service for MockConnector { + type Response = crate::MockConnection; + type Error = std::io::Error; + type Future = std::pin::Pin< + Box< + dyn std::future::Future> + + Send, + >, + >; + + // Polls the connector to check if it’s ready to handle a request. + // Always ready, as we don’t have resource constraints. + fn poll_ready( + &mut self, + _cx: &mut std::task::Context<'_>, + ) -> std::task::Poll> { + std::task::Poll::Ready(std::result::Result::Ok(())) + } + + // Creates a new MockConnection for the given URI. + // Configures the connection based on io_builder and conn_error. + fn call(&mut self, _req: hyper::Uri) -> Self::Future { + // Clone the IoBuilder to create a fresh mock I/O object. + let mut io_builder = self.io_builder.clone(); + // Clone the optional connection error for this call. + let conn_error = self.conn_error.clone(); + // Return a pinned future that creates the MockConnection. + Box::pin(async move { + // Build the mock I/O object from the IoBuilder. + // This defines the I/O behavior (e.g., EOF for empty builder). + let mock = io_builder.build(); + // Create an atomic flag to track connection failure, initially false. + let failed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); + // Set the default error for non-failure cases. + // Used when conn_error is None, simulating a clean EOF or connection close. + let error = if let Some(ref err) = conn_error { + err.clone() + } else { + std::sync::Arc::new(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "connection closed", + )) + }; + // Create an mpsc channel for signaling unexpected writes, if conn_error is set. + // This helps debug cases where writes occur despite an expected failure. + let error_tx = if conn_error.is_some() { + // Create a channel with a buffer of 1 for signaling writes. + let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1); + // Spawn a task to log unexpected writes when received. + tokio::spawn(async move { + // Wait for a message indicating a write occurred. + if rx.recv().await.is_some() { + // Log the unexpected write for debugging. + eprintln!("Unexpected write occurred"); + } + }); + Some(tx) + } else { + None + }; + // If a connection error is provided, mark the connection as failed. + // This causes poll_read to return the error immediately. + if let Some(err_clone) = conn_error { + // Set the failed flag to true atomically. + failed.store(true, std::sync::atomic::Ordering::SeqCst); + // Log the simulated error for traceability. + eprintln!("Simulated conn task error: {}", err_clone); + } + // Create and return the MockConnection with all configured components. + std::result::Result::Ok(crate::MockConnection::new(mock, failed, error, error_tx)) + }) + } +} + +// Test for connection error propagation with PR #184. +// Simulates a connection failure by setting failed=true and returning a custom io::Error. +// Verifies the error propagates through hyper’s client as a hyper::Error(Io, ...). +#[cfg(feature = "http1")] +#[tokio::test] +async fn test_connection_error_propagation_pr184() { + // Define the error message for the simulated connection failure. + // Reused for creating the error and verifying the result. + let err_str = "mock connection failure"; + // Create an io::Error with Other kind and the custom message. + // Wrapped in Arc for sharing across threads and MockConnection. + let io_error = std::sync::Arc::new(std::io::Error::new(std::io::ErrorKind::Other, err_str)); + // Create an empty IoBuilder, as no I/O is expected. + // The error triggers before any reads or writes occur. + let io_builder = tokio_test::io::Builder::new(); + // Create a MockConnector with the error to simulate a failed connection. + // The error will set failed=true in MockConnection. + let connector = crate::MockConnector::new(io_builder, Some(io_error.clone())); + // Build the hyper client with TokioExecutor and our connector. + // pool_max_idle_per_host(0) disables connection pooling for a fresh connection. + let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .pool_max_idle_per_host(0) + .build::<_, http_body_util::Empty>(connector); + // Build a GET request to a mock URI with custom headers. + // Uses mixed-case headers to match your style, ensuring case-insensitive handling. + let request = hyper::Request::builder() + .uri("http://mocked") + .header("hoSt", "mocked") + .header("conNection", "close") + .body(http_body_util::Empty::::new()) + .expect("failed to build request"); + // Send the request and capture the result. + // Expect it to fail due to the simulated connection error. + let result = client.request(request).await; + // Extract the error, as the request should fail. + let err = result.expect_err("expected request to fail"); + // Log the full error for debugging, including its structure. + // Matches your detailed logging style for traceability. + eprintln!("Actually gotten error is: {:?}", err); + // Downcast the error to a hyper::Error to verify its type. + // Expect a hyper::Error wrapping an io::Error from MockConnection. + let hyper_err = err + .source() + .and_then(|e| e.downcast_ref::()) + .expect("expected hyper::Error"); + // Downcast the hyper::Error’s source to an io::Error. + // Verify it matches the simulated error from MockConnection. + let io_err = hyper_err + .source() + .and_then(|e| e.downcast_ref::()) + .expect(&format!("expected io::Error but got {:?}", hyper_err)); + // Verify the io::Error has the expected kind (Other). + assert_eq!(io_err.kind(), std::io::ErrorKind::Other); + // Verify the io::Error’s message matches err_str. + assert_eq!(io_err.to_string(), err_str); +} + +// Test for consistent IncompleteMessage error with or without PR #184. +// Simulates a connection that returns EOF immediately, causing hyper’s HTTP/1.1 parser +// to fail with IncompleteMessage due to no response data. +// Uses MockConnector with conn_error=None to keep failed=false, ensuring EOF behavior. +#[cfg(feature = "http1")] +#[tokio::test] +async fn test_incomplete_message_error_pr184() { + // Create an empty IoBuilder to simulate a connection with no data. + // No write or read expectations, so poll_read returns EOF (Poll::Ready(Ok(0))). + // This triggers IncompleteMessage in hyper’s parser. + let io_builder = tokio_test::io::Builder::new(); + // Create MockConnector with no error (conn_error=None). + // Keeps failed=false in MockConnection, so poll_read delegates to the mock’s EOF. + let connector = crate::MockConnector::new(io_builder, None); + // Build the hyper client with TokioExecutor and our connector. + // pool_max_idle_per_host(0) disables pooling for a fresh connection. + let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .pool_max_idle_per_host(0) + .build::<_, http_body_util::Empty>(connector); + // Build a GET request to a mock URI with headers. + // Uses mixed-case headers to match test_connection_error_propagation_pr184. + // Empty body ensures focus on response parsing failure. + let request = hyper::Request::builder() + .uri("http://mocked") + .header("hoSt", "mocked") + .header("conNection", "close") + .body(http_body_util::Empty::::new()) + .expect("failed to build request"); + // Send the request and capture the result. + // Expect failure due to EOF causing IncompleteMessage. + let result = client.request(request).await; + // Extract the error, as the request should fail. + // Without PR #184, expect ChannelClosed; with PR #184, expect IncompleteMessage. + let err = result.expect_err("expected request to fail"); + // Log the full error for debugging, matching your style. + eprintln!("Actually gotten error is: {:?}", err); + // Downcast to hyper::Error to verify the error type. + // Expect IncompleteMessage (with PR #184) or ChannelClosed (without). + let hyper_err = err + .source() + .and_then(|e| e.downcast_ref::()) + .expect("expected hyper::Error"); + // Verify the error is IncompleteMessage when PR #184 is applied. + // This checks the parser’s failure due to EOF. + assert!( + hyper_err.is_incomplete_message(), + "expected IncompleteMessage, got {:?}", + hyper_err + ); + // Confirm no io::Error is present, as this is a parsing failure, not I/O. + // Ensures we’re testing the correct error type. + assert!( + hyper_err + .source() + .and_then(|e| e.downcast_ref::()) + .is_none(), + "expected no io::Error, got {:?}", + hyper_err + ); +} + +// Test for a successful HTTP/1.1 connection using a mock connector. +// Simulates a server that accepts a request and responds with a 200 OK. +// Verifies the client correctly sends the request and receives the response. +#[cfg(feature = "http1")] +#[tokio::test] +async fn test_successful_connection() { + // Define the expected server response: a valid HTTP/1.1 200 OK with no body. + let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"; + // Define the expected client request, including headers and CRLF termination. + // This ensures the client sends the correct request format. + let expected_request = b"GET / HTTP/1.1\r\nhost: mocked\r\nconnection: close\r\n\r\n"; + // Create an IoBuilder to simulate the server’s I/O behavior. + // Expect the client to write the request and read the response. + let mut io_builder = tokio_test::io::Builder::new(); + // Configure the IoBuilder to expect the request and provide the response. + io_builder.write(expected_request).read(response); + // Finalize the IoBuilder for use in the connector. + let io_builder = io_builder; + // Create a MockConnector with no error (conn_error=None). + // Ensures failed=false, allowing normal I/O operations. + let connector = crate::MockConnector::new(io_builder, None); + // Build the hyper client with TokioExecutor and our connector. + // pool_max_idle_per_host(0) ensures a fresh connection. + let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) + .pool_max_idle_per_host(0) + .build::<_, http_body_util::Empty>(connector); + // Build a GET request to a mock URI with headers. + // Uses mixed-case headers to match your style and verify case-insensitive handling. + let request = hyper::Request::builder() + .uri("http://mocked") + .header("hOst", "mocked") + .header("coNnection", "close") + .body(http_body_util::Empty::::new()) + .expect("failed to build request"); + // Send the request and capture the response. + // Expect a successful response due to the configured IoBuilder. + let response = client + .request(request) + .await + .expect("request should succeed"); + // Verify the response status is 200 OK. + assert_eq!(response.status(), 200); +} diff --git a/tests/legacy_client.rs b/tests/legacy_client_http1.rs similarity index 52% rename from tests/legacy_client.rs rename to tests/legacy_client_http1.rs index aac04cdd..983a13a3 100644 --- a/tests/legacy_client.rs +++ b/tests/legacy_client_http1.rs @@ -1,9 +1,9 @@ -#![cfg(all(feature = "client-legacy", any(feature = "http1", feature = "http2")))] +#![cfg(all(feature = "client-legacy", feature = "http1"))] mod test_utils; use std::io::{Read, Write}; -use std::net::{SocketAddr, TcpListener}; +use std::net::TcpListener; use std::pin::{Pin, pin}; use std::sync::Arc; use std::sync::atomic::Ordering; @@ -16,7 +16,7 @@ use futures_util::future::{self, FutureExt, TryFutureExt}; use futures_util::stream::StreamExt; use futures_util::{self, Stream}; use http_body_util::BodyExt; -use http_body_util::{Empty, Full, StreamBody}; +use http_body_util::{Empty, StreamBody}; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use hyper::Request; @@ -24,23 +24,15 @@ use hyper::body::Bytes; use hyper::body::Frame; use hyper_util::client::legacy::Client; use hyper_util::client::legacy::connect::{HttpConnector, capture_connection}; -use hyper_util::rt::{TokioExecutor, TokioIo}; +use hyper_util::rt::TokioExecutor; -use test_utils::{DebugConnector, DebugStream}; - -pub fn runtime() -> tokio::runtime::Runtime { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - .expect("new rt") -} +use test_utils::{DebugConnector, DebugStream, runtime}; fn s(buf: &[u8]) -> &str { std::str::from_utf8(buf).expect("from_utf8") } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[test] fn drop_body_before_eof_closes_connection() { // https://github.com/hyperium/hyper/issues/1353 @@ -89,7 +81,6 @@ fn drop_body_before_eof_closes_connection() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn drop_client_closes_idle_connections() { let _ = pretty_env_logger::try_init(); @@ -153,7 +144,6 @@ async fn drop_client_closes_idle_connections() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn drop_response_future_closes_in_progress_connection() { let _ = pretty_env_logger::try_init(); @@ -202,7 +192,6 @@ async fn drop_response_future_closes_in_progress_connection() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn drop_response_body_closes_in_progress_connection() { let _ = pretty_env_logger::try_init(); @@ -258,7 +247,6 @@ async fn drop_response_body_closes_in_progress_connection() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn no_keep_alive_closes_connection() { // https://github.com/hyperium/hyper/issues/1383 @@ -311,7 +299,6 @@ async fn no_keep_alive_closes_connection() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn socket_disconnect_closes_idle_conn() { // notably when keep-alive is enabled @@ -381,7 +368,6 @@ fn connect_call_is_lazy() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[test] fn client_keep_alive_0() { let _ = pretty_env_logger::try_init(); @@ -449,7 +435,6 @@ fn client_keep_alive_0() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[test] fn client_keep_alive_extra_body() { let _ = pretty_env_logger::try_init(); @@ -513,7 +498,6 @@ fn client_keep_alive_extra_body() { } #[cfg(not(miri))] -#[cfg(feature = "http1")] #[tokio::test] async fn client_keep_alive_when_response_before_request_body_ends() { let _ = pretty_env_logger::try_init(); @@ -808,158 +792,6 @@ fn client_upgrade() { assert_eq!(vec, b"bar=foo"); } -#[cfg(not(miri))] -#[cfg(feature = "server")] -#[test] -fn client_http2_upgrade() { - use http::{Method, Response, Version}; - use hyper::service::service_fn; - use tokio::io::{AsyncReadExt, AsyncWriteExt}; - use tokio::net::TcpListener; - - let _ = pretty_env_logger::try_init(); - let rt = runtime(); - let server = rt - .block_on(TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))) - .unwrap(); - let addr = server.local_addr().unwrap(); - let mut connector = DebugConnector::new(); - connector.alpn_h2 = true; - - let client = Client::builder(TokioExecutor::new()).build(connector); - - rt.spawn(async move { - let (stream, _) = server.accept().await.expect("accept"); - let stream = TokioIo::new(stream); - let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new()); - // IMPORTANT: This is required to advertise our support for HTTP/2 websockets to the client. - builder.http2().enable_connect_protocol(); - builder - .serve_connection_with_upgrades( - stream, - service_fn(|req| async move { - assert_eq!(req.headers().get("host"), None); - assert_eq!(req.version(), Version::HTTP_2); - assert_eq!( - req.headers().get(http::header::SEC_WEBSOCKET_VERSION), - Some(&http::header::HeaderValue::from_static("13")) - ); - assert_eq!( - req.extensions().get::(), - Some(&hyper::ext::Protocol::from_static("websocket")) - ); - - let on_upgrade = hyper::upgrade::on(req); - tokio::spawn(async move { - let upgraded = on_upgrade.await.unwrap(); - let mut io = TokioIo::new(upgraded); - - let mut vec = vec![]; - io.read_buf(&mut vec).await.unwrap(); - assert_eq!(vec, b"foo=bar"); - io.write_all(b"bar=foo").await.unwrap(); - }); - - Ok::<_, hyper::Error>(Response::new(Empty::::new())) - }), - ) - .await - .expect("server"); - }); - - let req = Request::builder() - .method(Method::CONNECT) - .uri(&*format!("http://{addr}/up")) - .header(http::header::SEC_WEBSOCKET_VERSION, "13") - .version(Version::HTTP_2) - .extension(hyper::ext::Protocol::from_static("websocket")) - .body(Empty::::new()) - .unwrap(); - - let res = client.request(req); - let res = rt.block_on(res).unwrap(); - - assert_eq!(res.status(), http::StatusCode::OK); - assert_eq!(res.version(), Version::HTTP_2); - - let upgraded = rt.block_on(hyper::upgrade::on(res)).expect("on_upgrade"); - let mut io = TokioIo::new(upgraded); - - rt.block_on(io.write_all(b"foo=bar")).unwrap(); - let mut vec = vec![]; - rt.block_on(io.read_to_end(&mut vec)).unwrap(); - assert_eq!(vec, b"bar=foo"); -} - -#[cfg(not(miri))] -#[cfg(feature = "http2")] -#[test] -fn alpn_h2() { - use http::Response; - use hyper::service::service_fn; - use tokio::net::TcpListener; - - let _ = pretty_env_logger::try_init(); - let rt = runtime(); - let listener = rt - .block_on(TcpListener::bind(SocketAddr::from(([127, 0, 0, 1], 0)))) - .unwrap(); - let addr = listener.local_addr().unwrap(); - let mut connector = DebugConnector::new(); - connector.alpn_h2 = true; - let connects = connector.connects.clone(); - - let client = Client::builder(TokioExecutor::new()).build(connector); - - rt.spawn(async move { - let (stream, _) = listener.accept().await.expect("accept"); - let stream = TokioIo::new(stream); - hyper::server::conn::http2::Builder::new(TokioExecutor::new()) - .serve_connection( - stream, - service_fn(|req| async move { - assert_eq!(req.headers().get("host"), None); - Ok::<_, hyper::Error>(Response::new(Full::::from("Hello, world"))) - }), - ) - .await - .expect("server"); - }); - - assert_eq!(connects.load(Ordering::SeqCst), 0); - - let url = format!("http://{addr}/a").parse::<::hyper::Uri>().unwrap(); - let res1 = client.get(url.clone()); - let res2 = client.get(url.clone()); - let res3 = client.get(url.clone()); - rt.block_on(future::try_join3(res1, res2, res3)).unwrap(); - - // Since the client doesn't know it can ALPN at first, it will have - // started 3 connections. But, the server above will only handle 1, - // so the unwrapped responses futures show it still worked. - assert_eq!(connects.load(Ordering::SeqCst), 3); - - let res4 = client.get(url.clone()); - rt.block_on(res4).unwrap(); - - // HTTP/2 request allowed - let res5 = client.request( - Request::builder() - .uri(url) - .version(hyper::Version::HTTP_2) - .body(Empty::::new()) - .unwrap(), - ); - rt.block_on(res5).unwrap(); - - assert_eq!( - connects.load(Ordering::SeqCst), - 3, - "after ALPN, no more connects" - ); - drop(client); -} - #[cfg(not(miri))] #[cfg(feature = "http1")] #[test] @@ -1065,438 +897,3 @@ fn connection_poisoning() { assert_eq!(num_conns.load(Ordering::SeqCst), 2); assert_eq!(num_requests.load(Ordering::SeqCst), 5); } - -// ------------------------------------------------------- -// Below is our custom code for testing hyper legacy-client behavior with mock connections for PR #184 -// We use fully qualified paths for all types and identifiers to make this code -// copy/paste-able without relying on external 'use' statements. Detailed inline -// comments explain the purpose and logic of each section. - -//XXX: can manually run like this: -// $ cargo test --features="http1,http2,server,client-legacy" --test legacy_client -- test_connection_error_propagation test_incomplete_message_error --nocapture -// $ cargo test --all-features --test legacy_client -- --nocapture -// $ cargo test --all-features --test legacy_client - -use std::error::Error; // needed for .source() eg. error[E0599]: no method named `source` found for struct `hyper_util::client::legacy::Error` in the current scope - -// Helper function to debug byte slices by attempting to interpret them as UTF-8. -// If the bytes are valid UTF-8, they are printed as a string; otherwise, they are -// printed as a raw byte array. This aids in debugging tokio_test::io::Mock mismatches. -fn debug_bytes(bytes: &[u8], label: &str) { - // Try to convert the byte slice to a UTF-8 string. - // If successful, print it with the provided label for context. - if let Ok(s) = std::str::from_utf8(bytes) { - eprintln!("{}: {}", label, s); - } else { - // If the bytes are not valid UTF-8, print them as a raw byte array. - eprintln!("{}: {:?}", label, bytes); - } -} - -// Struct representing a mock connection for testing hyper client behavior. -// Implements hyper::rt::Read, hyper::rt::Write, and hyper_util::client::legacy::connect::Connection -// traits to simulate I/O operations. Uses tokio_test::io::Mock for controlled I/O behavior. -struct MockConnection { - // The underlying mock I/O object, wrapped in hyper_util::rt::TokioIo for compatibility. - inner: hyper_util::rt::TokioIo, - // Atomic flag to signal a connection failure, controlling poll_read behavior. - failed: std::sync::Arc, - // The error to return when failed=true, simulating an I/O failure. - error: std::sync::Arc, - // Optional channel to signal unexpected writes, used for debugging. - error_tx: Option>, - // Tracks total bytes written, for logging and verification. - bytes_written: usize, -} - -impl MockConnection { - // Constructor for MockConnection, initializing all fields. - // Takes a mock I/O object, failure flag, error, and optional error channel. - fn new( - mock: tokio_test::io::Mock, - failed: std::sync::Arc, - error: std::sync::Arc, - error_tx: Option>, - ) -> Self { - MockConnection { - inner: hyper_util::rt::TokioIo::new(mock), - failed, - error, - error_tx, - bytes_written: 0, - } - } -} - -// Implement hyper::rt::Read trait to handle read operations on the mock connection. -// Controls whether an error or mock I/O data is returned based on the failed flag. -impl hyper::rt::Read for MockConnection { - // Polls the connection for reading, filling the provided buffer. - // If failed=true, returns the stored error; otherwise, delegates to the mock I/O. - fn poll_read( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: hyper::rt::ReadBufCursor<'_>, - ) -> std::task::Poll> { - // Log the current state of the failed flag for debugging. - eprintln!( - "poll_read: failed={}", - self.failed.load(std::sync::atomic::Ordering::SeqCst) - ); - // Check if the connection is marked as failed. - // If true, return the stored error immediately to simulate a connection failure. - if self.failed.load(std::sync::atomic::Ordering::SeqCst) { - // Log the error being returned for traceability. - eprintln!("poll_read: returning error: {}", self.error); - // Create a new io::Error with the same kind and message as the stored error. - return std::task::Poll::Ready(std::result::Result::Err(std::io::Error::new( - self.error.kind(), - self.error.to_string(), - ))); - } - // If not failed, delegate to the mock I/O to simulate normal read behavior. - // This may return EOF (Poll::Ready(Ok(0))) for empty IoBuilder. - let inner = std::pin::Pin::new(&mut self.inner); - inner.poll_read(cx, buf) - } -} - -// Implement hyper::rt::Write trait to handle write operations on the mock connection. -// Logs writes and signals unexpected writes via error_tx. -impl hyper::rt::Write for MockConnection { - // Polls the connection for writing, sending the provided buffer. - // Logs the write operation and tracks total bytes written. - fn poll_write( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - buf: &[u8], - ) -> std::task::Poll> { - // Log the size of the buffer being written for debugging. - eprintln!("poll_write: {} bytes", buf.len()); - // Debug the buffer contents as UTF-8 or raw bytes. - debug_bytes(buf, "poll_write buffer"); - // Delegate the write to the mock I/O object. - let inner = std::pin::Pin::new(&mut self.inner); - match inner.poll_write(cx, buf) { - // If the write succeeds, update the bytes_written counter and log the result. - std::task::Poll::Ready(std::result::Result::Ok(bytes)) => { - // Increment the total bytes written for tracking. - self.bytes_written += bytes; - // Log the number of bytes written and the running total. - eprintln!( - "poll_write: wrote {} bytes, total={}", - bytes, self.bytes_written - ); - // If error_tx is present, signal an unexpected write (used in error tests). - // This helps detect writes when the connection should fail early. - if let Some(tx) = self.error_tx.take() { - // Log that an unexpected write is being signaled. - eprintln!("poll_write: signaling unexpected write"); - // Send a message through the channel, ignoring errors if the receiver is closed. - let _ = tx.try_send(()); - } - // Return the successful write result. - std::task::Poll::Ready(std::result::Result::Ok(bytes)) - } - // For pending or error results, propagate them directly. - other => other, - } - } - - // Polls the connection to flush any buffered data. - // Delegates to the mock I/O object. - fn poll_flush( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - // Log the flush operation for debugging. - eprintln!("poll_flush"); - // Delegate the flush to the mock I/O object. - let inner = std::pin::Pin::new(&mut self.inner); - inner.poll_flush(cx) - } - - // Polls the connection to shut down the write side. - // Delegates to the mock I/O object. - fn poll_shutdown( - mut self: std::pin::Pin<&mut Self>, - cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - // Log the shutdown operation for debugging. - eprintln!("poll_shutdown"); - // Delegate the shutdown to the mock I/O object. - let inner = std::pin::Pin::new(&mut self.inner); - inner.poll_shutdown(cx) - } -} - -// Implement hyper_util::client::legacy::connect::Connection trait to provide connection metadata. -// Required for hyper to use MockConnection as a valid connection. -impl hyper_util::client::legacy::connect::Connection for MockConnection { - // Returns metadata about the connection. - // In this case, a default Connected object indicating a new connection. - fn connected(&self) -> hyper_util::client::legacy::connect::Connected { - hyper_util::client::legacy::connect::Connected::new() - } -} - -// Struct representing a mock connector for creating MockConnection instances. -// Implements tower_service::Service to integrate with hyper’s client. -#[derive(Clone)] -struct MockConnector { - // The IoBuilder used to create mock I/O objects for each connection. - io_builder: tokio_test::io::Builder, - // Optional error to simulate a connection failure, passed to MockConnection. - conn_error: Option>, -} - -impl MockConnector { - // Constructor for MockConnector, initializing the IoBuilder and optional error. - fn new( - io_builder: tokio_test::io::Builder, - conn_error: Option>, - ) -> Self { - MockConnector { - io_builder, - conn_error, - } - } -} - -// Implement tower_service::Service for MockConnector to create MockConnection instances. -// Takes a hyper::Uri and returns a future resolving to a MockConnection. -impl tower_service::Service for MockConnector { - type Response = crate::MockConnection; - type Error = std::io::Error; - type Future = std::pin::Pin< - Box< - dyn std::future::Future> - + Send, - >, - >; - - // Polls the connector to check if it’s ready to handle a request. - // Always ready, as we don’t have resource constraints. - fn poll_ready( - &mut self, - _cx: &mut std::task::Context<'_>, - ) -> std::task::Poll> { - std::task::Poll::Ready(std::result::Result::Ok(())) - } - - // Creates a new MockConnection for the given URI. - // Configures the connection based on io_builder and conn_error. - fn call(&mut self, _req: hyper::Uri) -> Self::Future { - // Clone the IoBuilder to create a fresh mock I/O object. - let mut io_builder = self.io_builder.clone(); - // Clone the optional connection error for this call. - let conn_error = self.conn_error.clone(); - // Return a pinned future that creates the MockConnection. - Box::pin(async move { - // Build the mock I/O object from the IoBuilder. - // This defines the I/O behavior (e.g., EOF for empty builder). - let mock = io_builder.build(); - // Create an atomic flag to track connection failure, initially false. - let failed = std::sync::Arc::new(std::sync::atomic::AtomicBool::new(false)); - // Set the default error for non-failure cases. - // Used when conn_error is None, simulating a clean EOF or connection close. - let error = if let Some(ref err) = conn_error { - err.clone() - } else { - std::sync::Arc::new(std::io::Error::new( - std::io::ErrorKind::BrokenPipe, - "connection closed", - )) - }; - // Create an mpsc channel for signaling unexpected writes, if conn_error is set. - // This helps debug cases where writes occur despite an expected failure. - let error_tx = if conn_error.is_some() { - // Create a channel with a buffer of 1 for signaling writes. - let (tx, mut rx) = tokio::sync::mpsc::channel::<()>(1); - // Spawn a task to log unexpected writes when received. - tokio::spawn(async move { - // Wait for a message indicating a write occurred. - if rx.recv().await.is_some() { - // Log the unexpected write for debugging. - eprintln!("Unexpected write occurred"); - } - }); - Some(tx) - } else { - None - }; - // If a connection error is provided, mark the connection as failed. - // This causes poll_read to return the error immediately. - if let Some(err_clone) = conn_error { - // Set the failed flag to true atomically. - failed.store(true, std::sync::atomic::Ordering::SeqCst); - // Log the simulated error for traceability. - eprintln!("Simulated conn task error: {}", err_clone); - } - // Create and return the MockConnection with all configured components. - std::result::Result::Ok(crate::MockConnection::new(mock, failed, error, error_tx)) - }) - } -} - -// Test for connection error propagation with PR #184. -// Simulates a connection failure by setting failed=true and returning a custom io::Error. -// Verifies the error propagates through hyper’s client as a hyper::Error(Io, ...). -#[cfg(feature = "http1")] -#[tokio::test] -async fn test_connection_error_propagation_pr184() { - // Define the error message for the simulated connection failure. - // Reused for creating the error and verifying the result. - let err_str = "mock connection failure"; - // Create an io::Error with Other kind and the custom message. - // Wrapped in Arc for sharing across threads and MockConnection. - let io_error = std::sync::Arc::new(std::io::Error::new(std::io::ErrorKind::Other, err_str)); - // Create an empty IoBuilder, as no I/O is expected. - // The error triggers before any reads or writes occur. - let io_builder = tokio_test::io::Builder::new(); - // Create a MockConnector with the error to simulate a failed connection. - // The error will set failed=true in MockConnection. - let connector = crate::MockConnector::new(io_builder, Some(io_error.clone())); - // Build the hyper client with TokioExecutor and our connector. - // pool_max_idle_per_host(0) disables connection pooling for a fresh connection. - let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) - .pool_max_idle_per_host(0) - .build::<_, http_body_util::Empty>(connector); - // Build a GET request to a mock URI with custom headers. - // Uses mixed-case headers to match your style, ensuring case-insensitive handling. - let request = hyper::Request::builder() - .uri("http://mocked") - .header("hoSt", "mocked") - .header("conNection", "close") - .body(http_body_util::Empty::::new()) - .expect("failed to build request"); - // Send the request and capture the result. - // Expect it to fail due to the simulated connection error. - let result = client.request(request).await; - // Extract the error, as the request should fail. - let err = result.expect_err("expected request to fail"); - // Log the full error for debugging, including its structure. - // Matches your detailed logging style for traceability. - eprintln!("Actually gotten error is: {:?}", err); - // Downcast the error to a hyper::Error to verify its type. - // Expect a hyper::Error wrapping an io::Error from MockConnection. - let hyper_err = err - .source() - .and_then(|e| e.downcast_ref::()) - .expect("expected hyper::Error"); - // Downcast the hyper::Error’s source to an io::Error. - // Verify it matches the simulated error from MockConnection. - let io_err = hyper_err - .source() - .and_then(|e| e.downcast_ref::()) - .expect(&format!("expected io::Error but got {:?}", hyper_err)); - // Verify the io::Error has the expected kind (Other). - assert_eq!(io_err.kind(), std::io::ErrorKind::Other); - // Verify the io::Error’s message matches err_str. - assert_eq!(io_err.to_string(), err_str); -} - -// Test for consistent IncompleteMessage error with or without PR #184. -// Simulates a connection that returns EOF immediately, causing hyper’s HTTP/1.1 parser -// to fail with IncompleteMessage due to no response data. -// Uses MockConnector with conn_error=None to keep failed=false, ensuring EOF behavior. -#[cfg(feature = "http1")] -#[tokio::test] -async fn test_incomplete_message_error_pr184() { - // Create an empty IoBuilder to simulate a connection with no data. - // No write or read expectations, so poll_read returns EOF (Poll::Ready(Ok(0))). - // This triggers IncompleteMessage in hyper’s parser. - let io_builder = tokio_test::io::Builder::new(); - // Create MockConnector with no error (conn_error=None). - // Keeps failed=false in MockConnection, so poll_read delegates to the mock’s EOF. - let connector = crate::MockConnector::new(io_builder, None); - // Build the hyper client with TokioExecutor and our connector. - // pool_max_idle_per_host(0) disables pooling for a fresh connection. - let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) - .pool_max_idle_per_host(0) - .build::<_, http_body_util::Empty>(connector); - // Build a GET request to a mock URI with headers. - // Uses mixed-case headers to match test_connection_error_propagation_pr184. - // Empty body ensures focus on response parsing failure. - let request = hyper::Request::builder() - .uri("http://mocked") - .header("hoSt", "mocked") - .header("conNection", "close") - .body(http_body_util::Empty::::new()) - .expect("failed to build request"); - // Send the request and capture the result. - // Expect failure due to EOF causing IncompleteMessage. - let result = client.request(request).await; - // Extract the error, as the request should fail. - // Without PR #184, expect ChannelClosed; with PR #184, expect IncompleteMessage. - let err = result.expect_err("expected request to fail"); - // Log the full error for debugging, matching your style. - eprintln!("Actually gotten error is: {:?}", err); - // Downcast to hyper::Error to verify the error type. - // Expect IncompleteMessage (with PR #184) or ChannelClosed (without). - let hyper_err = err - .source() - .and_then(|e| e.downcast_ref::()) - .expect("expected hyper::Error"); - // Verify the error is IncompleteMessage when PR #184 is applied. - // This checks the parser’s failure due to EOF. - assert!( - hyper_err.is_incomplete_message(), - "expected IncompleteMessage, got {:?}", - hyper_err - ); - // Confirm no io::Error is present, as this is a parsing failure, not I/O. - // Ensures we’re testing the correct error type. - assert!( - hyper_err - .source() - .and_then(|e| e.downcast_ref::()) - .is_none(), - "expected no io::Error, got {:?}", - hyper_err - ); -} - -// Test for a successful HTTP/1.1 connection using a mock connector. -// Simulates a server that accepts a request and responds with a 200 OK. -// Verifies the client correctly sends the request and receives the response. -#[cfg(feature = "http1")] -#[tokio::test] -async fn test_successful_connection() { - // Define the expected server response: a valid HTTP/1.1 200 OK with no body. - let response = b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"; - // Define the expected client request, including headers and CRLF termination. - // This ensures the client sends the correct request format. - let expected_request = b"GET / HTTP/1.1\r\nhost: mocked\r\nconnection: close\r\n\r\n"; - // Create an IoBuilder to simulate the server’s I/O behavior. - // Expect the client to write the request and read the response. - let mut io_builder = tokio_test::io::Builder::new(); - // Configure the IoBuilder to expect the request and provide the response. - io_builder.write(expected_request).read(response); - // Finalize the IoBuilder for use in the connector. - let io_builder = io_builder; - // Create a MockConnector with no error (conn_error=None). - // Ensures failed=false, allowing normal I/O operations. - let connector = crate::MockConnector::new(io_builder, None); - // Build the hyper client with TokioExecutor and our connector. - // pool_max_idle_per_host(0) ensures a fresh connection. - let client = hyper_util::client::legacy::Client::builder(hyper_util::rt::TokioExecutor::new()) - .pool_max_idle_per_host(0) - .build::<_, http_body_util::Empty>(connector); - // Build a GET request to a mock URI with headers. - // Uses mixed-case headers to match your style and verify case-insensitive handling. - let request = hyper::Request::builder() - .uri("http://mocked") - .header("hOst", "mocked") - .header("coNnection", "close") - .body(http_body_util::Empty::::new()) - .expect("failed to build request"); - // Send the request and capture the response. - // Expect a successful response due to the configured IoBuilder. - let response = client - .request(request) - .await - .expect("request should succeed"); - // Verify the response status is 200 OK. - assert_eq!(response.status(), 200); -} diff --git a/tests/legacy_client_http2.rs b/tests/legacy_client_http2.rs new file mode 100644 index 00000000..9f54970f --- /dev/null +++ b/tests/legacy_client_http2.rs @@ -0,0 +1,169 @@ +#![cfg(all(feature = "client-legacy", feature = "http2"))] + +mod test_utils; + +use std::sync::atomic::Ordering; + +use bytes::Bytes; +use futures_util::future; +use http::{Request, Response}; +use http_body_util::{Empty, Full}; +use tokio::net::TcpListener; + +use hyper::service::service_fn; +use hyper_util::client::legacy::Client; +use hyper_util::rt::TokioExecutor; + +use test_utils::{DebugConnector, runtime}; + +#[cfg(not(miri))] +#[cfg(feature = "server")] +#[test] +fn client_http2_upgrade() { + use http::{Method, Version}; + use hyper_util::rt::TokioIo; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let _ = pretty_env_logger::try_init(); + let rt = runtime(); + let server = rt + .block_on(TcpListener::bind(std::net::SocketAddr::from(( + [127, 0, 0, 1], + 0, + )))) + .unwrap(); + let addr = server.local_addr().unwrap(); + let mut connector = DebugConnector::new(); + connector.alpn_h2 = true; + + let client = Client::builder(TokioExecutor::new()).build(connector); + + rt.spawn(async move { + let (stream, _) = server.accept().await.expect("accept"); + let stream = TokioIo::new(stream); + let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new()); + // IMPORTANT: This is required to advertise our support for HTTP/2 websockets to the client. + builder.http2().enable_connect_protocol(); + builder + .serve_connection_with_upgrades( + stream, + service_fn(|req| async move { + assert_eq!(req.headers().get("host"), None); + assert_eq!(req.version(), Version::HTTP_2); + assert_eq!( + req.headers().get(http::header::SEC_WEBSOCKET_VERSION), + Some(&http::header::HeaderValue::from_static("13")) + ); + assert_eq!( + req.extensions().get::(), + Some(&hyper::ext::Protocol::from_static("websocket")) + ); + + let on_upgrade = hyper::upgrade::on(req); + tokio::spawn(async move { + let upgraded = on_upgrade.await.unwrap(); + let mut io = TokioIo::new(upgraded); + + let mut vec = vec![]; + io.read_buf(&mut vec).await.unwrap(); + assert_eq!(vec, b"foo=bar"); + io.write_all(b"bar=foo").await.unwrap(); + }); + + Ok::<_, hyper::Error>(Response::new(Empty::::new())) + }), + ) + .await + .expect("server"); + }); + + let req = Request::builder() + .method(Method::CONNECT) + .uri(&*format!("http://{addr}/up")) + .header(http::header::SEC_WEBSOCKET_VERSION, "13") + .version(Version::HTTP_2) + .extension(hyper::ext::Protocol::from_static("websocket")) + .body(Empty::::new()) + .unwrap(); + + let res = client.request(req); + let res = rt.block_on(res).unwrap(); + + assert_eq!(res.status(), http::StatusCode::OK); + assert_eq!(res.version(), Version::HTTP_2); + + let upgraded = rt.block_on(hyper::upgrade::on(res)).expect("on_upgrade"); + let mut io = hyper_util::rt::TokioIo::new(upgraded); + + rt.block_on(io.write_all(b"foo=bar")).unwrap(); + let mut vec = vec![]; + rt.block_on(io.read_to_end(&mut vec)).unwrap(); + assert_eq!(vec, b"bar=foo"); +} + +#[cfg(not(miri))] +#[test] +fn alpn_h2() { + let _ = pretty_env_logger::try_init(); + let rt = runtime(); + let listener = rt + .block_on(TcpListener::bind(std::net::SocketAddr::from(( + [127, 0, 0, 1], + 0, + )))) + .unwrap(); + let addr = listener.local_addr().unwrap(); + let mut connector = DebugConnector::new(); + connector.alpn_h2 = true; + let connects = connector.connects.clone(); + + let client = Client::builder(TokioExecutor::new()).build(connector); + + rt.spawn(async move { + let (stream, _) = listener.accept().await.expect("accept"); + let stream = hyper_util::rt::TokioIo::new(stream); + hyper::server::conn::http2::Builder::new(TokioExecutor::new()) + .serve_connection( + stream, + service_fn(|req| async move { + assert_eq!(req.headers().get("host"), None); + Ok::<_, hyper::Error>(Response::new(Full::::from("Hello, world"))) + }), + ) + .await + .expect("server"); + }); + + assert_eq!(connects.load(Ordering::SeqCst), 0); + + let url = format!("http://{addr}/a").parse::<::hyper::Uri>().unwrap(); + let res1 = client.get(url.clone()); + let res2 = client.get(url.clone()); + let res3 = client.get(url.clone()); + rt.block_on(future::try_join3(res1, res2, res3)).unwrap(); + + // Since the client doesn't know it can ALPN at first, it will have + // started 3 connections. But, the server above will only handle 1, + // so the unwrapped responses futures show it still worked. + assert_eq!(connects.load(Ordering::SeqCst), 3); + + let res4 = client.get(url.clone()); + rt.block_on(res4).unwrap(); + + // HTTP/2 request allowed + let res5 = client.request( + Request::builder() + .uri(url) + .version(hyper::Version::HTTP_2) + .body(Empty::::new()) + .unwrap(), + ); + rt.block_on(res5).unwrap(); + + assert_eq!( + connects.load(Ordering::SeqCst), + 3, + "after ALPN, no more connects" + ); + drop(client); +} diff --git a/tests/test_utils/mod.rs b/tests/test_utils/mod.rs index 0af7e90a..3e884559 100644 --- a/tests/test_utils/mod.rs +++ b/tests/test_utils/mod.rs @@ -15,6 +15,13 @@ use hyper_util::client::legacy::connect::HttpConnector; use hyper_util::client::legacy::connect::{Connected, Connection}; use hyper_util::rt::TokioIo; +pub fn runtime() -> tokio::runtime::Runtime { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("new rt") +} + #[derive(Clone)] pub struct DebugConnector { pub http: HttpConnector, @@ -41,6 +48,8 @@ impl DebugConnector { } } + #[cfg(any(feature = "http1", feature = "client-proxy"))] + #[cfg_attr(feature = "http2", allow(dead_code))] pub fn proxy(mut self) -> Self { self.is_proxy = true; self