diff --git a/Cargo.lock b/Cargo.lock index 059b2c3..25261f2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1056,7 +1056,7 @@ dependencies = [ "resolv-conf", "rustls", "smallvec", - "system-configuration 0.7.0", + "system-configuration", "thiserror 2.0.18", "tokio", "tokio-rustls", @@ -2138,7 +2138,6 @@ dependencies = [ "base64", "bytes", "futures-channel", - "futures-core", "futures-util", "http", "http-body", @@ -2151,7 +2150,7 @@ dependencies = [ "pretty_env_logger", "primp-hyper", "socket2", - "system-configuration 0.6.1", + "system-configuration", "tokio", "tokio-test", "tower-layer", @@ -3062,17 +3061,6 @@ dependencies = [ "syn", ] -[[package]] -name = "system-configuration" -version = "0.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3c879d448e9d986b661742763247d3693ed13609438cf3d006f51f5368a5ba6b" -dependencies = [ - "bitflags", - "core-foundation 0.9.4", - "system-configuration-sys", -] - [[package]] name = "system-configuration" version = "0.7.0" diff --git a/crates/primp-h2/Cargo.toml b/crates/primp-h2/Cargo.toml index 4476b8e..45f5b91 100644 --- a/crates/primp-h2/Cargo.toml +++ b/crates/primp-h2/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "primp-h2" -version = "0.4.14" # upstream src is 0.4.13 +version = "0.4.14" # based on upstream v0.4.14, cherry-picks through 2026-05-23 license = "MIT" authors = [ "deedy5", diff --git a/crates/primp-h2/src/proto/streams/prioritize.rs b/crates/primp-h2/src/proto/streams/prioritize.rs index 75358d4..a6758d1 100644 --- a/crates/primp-h2/src/proto/streams/prioritize.rs +++ b/crates/primp-h2/src/proto/streams/prioritize.rs @@ -731,6 +731,16 @@ impl Prioritize { let frame = match stream.pending_send.pop_front(buffer) { Some(Frame::Data(mut frame)) => { + if let Some(reason) = stream.state.get_scheduled_reset() { + if reason != Reason::NO_ERROR { + stream.pending_send.push_front(buffer, frame.into()); + self.clear_queue(buffer, &mut stream); + self.reclaim_all_capacity(&mut stream, counts); + self.pending_send.push(&mut stream); + continue; + } + } + // Get the amount of capacity remaining for stream's // window. let stream_capacity = stream.send_flow.available(); diff --git a/crates/primp-h2/src/proto/streams/recv.rs b/crates/primp-h2/src/proto/streams/recv.rs index 1d98caa..c52511f 100644 --- a/crates/primp-h2/src/proto/streams/recv.rs +++ b/crates/primp-h2/src/proto/streams/recv.rs @@ -381,8 +381,15 @@ impl Recv { if let Some(event) = stream.pending_recv.pop_front(&mut self.buffer) { match event { Event::InformationalHeaders(Client(response)) => { - // Found an informational response, return it - return Poll::Ready(Some(Ok(response))); + // Only return if it's actually a 1xx informational response + if response.status().is_informational() { + return Poll::Ready(Some(Ok(response))); + } + // Not actually informational (e.g., a final response), put it back + stream.pending_recv.push_front( + &mut self.buffer, + Event::InformationalHeaders(Client(response)), + ); } other => { // Not an informational response, put it back at the front diff --git a/crates/primp-h2/src/proto/streams/send.rs b/crates/primp-h2/src/proto/streams/send.rs index b52117d..bdcd167 100644 --- a/crates/primp-h2/src/proto/streams/send.rs +++ b/crates/primp-h2/src/proto/streams/send.rs @@ -376,7 +376,14 @@ impl Send { stream.send_capacity_inc = false; - Poll::Ready(Some(Ok(self.capacity(stream)))) + let capacity = self.capacity(stream); + + if capacity == 0 { + stream.wait_send(cx); + return Poll::Pending; + } + + Poll::Ready(Some(Ok(capacity))) } /// Current available stream send capacity diff --git a/crates/primp-h2/src/proto/streams/stream.rs b/crates/primp-h2/src/proto/streams/stream.rs index df93fd6..1ada434 100644 --- a/crates/primp-h2/src/proto/streams/stream.rs +++ b/crates/primp-h2/src/proto/streams/stream.rs @@ -384,6 +384,7 @@ impl Stream { /// Notify the send and receive tasks, if they exist. pub(super) fn set_reset(&mut self, reason: Reason, initiator: Initiator) { self.state.set_reset(self.id, reason, initiator); + self.notify_send(); self.notify_push(); self.notify_recv(); } diff --git a/crates/primp-h2/src/proto/streams/streams.rs b/crates/primp-h2/src/proto/streams/streams.rs index 9074f91..cf9c987 100644 --- a/crates/primp-h2/src/proto/streams/streams.rs +++ b/crates/primp-h2/src/proto/streams/streams.rs @@ -530,6 +530,11 @@ impl Inner { let stream = self.store.resolve(key); + if stream.is_pending_open { + proto_err!(conn: "recv_headers: received frame on idle stream {:?}", id); + return Err(Error::library_go_away(Reason::PROTOCOL_ERROR)); + } + if stream.state.is_local_error() { // Locally reset streams must ignore frames "for some time". // This is because the remote may have sent trailers before @@ -612,6 +617,9 @@ impl Inner { id, self.actions.recv.max_stream_id() ); + let sz = frame.payload().len(); + let sz = sz as WindowSize; + self.actions.recv.ignore_data(sz)?; return Ok(()); } @@ -688,6 +696,11 @@ impl Inner { } }; + if stream.is_pending_open { + proto_err!(conn: "recv_reset: received frame on idle stream {:?}", id); + return Err(Error::library_go_away(Reason::PROTOCOL_ERROR)); + } + let mut send_buffer = send_buffer.inner.lock().unwrap(); let send_buffer = &mut *send_buffer; @@ -720,6 +733,11 @@ impl Inner { // The remote may send window updates for streams that the local now // considers closed. It's ok... if let Some(mut stream) = self.store.find_mut(&id) { + if stream.is_pending_open { + proto_err!(conn: "recv_window_update: received frame on idle stream {:?}", id); + return Err(Error::library_go_away(Reason::PROTOCOL_ERROR)); + } + let res = self .actions .send @@ -783,9 +801,10 @@ impl Inner { actions.send.recv_go_away(last_stream_id)?; let err = Error::remote_go_away(frame.debug_data().clone(), frame.reason()); + let peer = counts.peer(); self.store.for_each(|stream| { - if stream.id > last_stream_id { + if stream.id > last_stream_id && peer.is_local_init(stream.id) { counts.transition(stream, |counts, stream| { actions.recv.handle_error(&err, &mut *stream); actions.send.handle_error(send_buffer, stream, counts); diff --git a/crates/primp-h2/src/share.rs b/crates/primp-h2/src/share.rs index c07402a..1019873 100644 --- a/crates/primp-h2/src/share.rs +++ b/crates/primp-h2/src/share.rs @@ -308,7 +308,7 @@ impl SendStream { /// increased by the connection. Note that `n` here represents the **total** /// amount of assigned capacity at that point in time. It is also possible /// that `n` is lower than the previous call if, since then, the caller has - /// sent data. + /// sent data. `n` will always be greater than zero. pub fn poll_capacity(&mut self, cx: &mut Context) -> Poll>> { self.inner .poll_capacity(cx) diff --git a/crates/primp-hyper-util/Cargo.toml b/crates/primp-hyper-util/Cargo.toml index d87e3ee..33e7029 100644 --- a/crates/primp-hyper-util/Cargo.toml +++ b/crates/primp-hyper-util/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "primp-hyper-util" -version = "0.1.21" # upstream src is 0.1.20 +version = "0.1.21" # based on upstream v0.1.20, cherry-picks through 2026-05-23 license = "MIT" authors = ["deedy5", "Sean McArthur "] edition = "2021" @@ -35,13 +35,13 @@ rustdoc-args = [ __internal_happy_eyeballs_tests = [] client = [ "hyper/client", - "tokio/net", "dep:tracing", "dep:futures-channel", "dep:tower-service", ] client-legacy = [ "client", + "tokio/net", "dep:socket2", "tokio/sync", "dep:libc", @@ -51,6 +51,7 @@ client-pool = [ "client", "dep:futures-util", "dep:tower-layer", + "tokio/sync", ] client-proxy = [ "client", @@ -148,8 +149,6 @@ version = "1.7.1" version = "0.3" optional = true -[dependencies.futures-core] -version = "0.3" [dependencies.futures-util] version = "0.3.16" @@ -163,7 +162,7 @@ version = "1.0" version = "1.0.0" [dependencies.hyper] -version = "1.8" +version = "1.9" path = "../primp-hyper" package = "primp-hyper" @@ -183,7 +182,7 @@ optional = true version = "0.2.4" [dependencies.socket2] -version = ">=0.5.9, <0.7" +version = "0.6" features = ["all"] optional = true @@ -218,7 +217,7 @@ default-features = false version = "0.1.0" [dev-dependencies.hyper] -version = "1.8" +version = "1.9" path = "../primp-hyper" package = "primp-hyper" features = ["full"] @@ -244,7 +243,7 @@ version = "0.4" version = "0.35.0" [target.'cfg(target_os = "macos")'.dependencies.system-configuration] -version = ">=0.5, <0.7" +version = ">=0.7, <0.8" optional = true [target."cfg(windows)".dependencies.windows-registry] diff --git a/crates/primp-hyper-util/src/client/legacy/client.rs b/crates/primp-hyper-util/src/client/legacy/client.rs index e68edce..dd56caf 100644 --- a/crates/primp-hyper-util/src/client/legacy/client.rs +++ b/crates/primp-hyper-util/src/client/legacy/client.rs @@ -25,8 +25,8 @@ use super::connect::HttpConnector; use super::connect::{Alpn, Connect, Connected, Connection}; use super::pool::{self, Ver}; -use crate::common::future::poll_fn; use crate::common::{lazy as hyper_lazy, timer, Exec, Lazy, SyncWrapper}; +use std::future::poll_fn; type BoxSendFuture = Pin + Send>>; @@ -487,7 +487,7 @@ where fn connect_to( &self, pool_key: PoolKey, - ) -> impl Lazy, PoolKey>, Error>> + Send + Unpin + ) -> impl Lazy, PoolKey>, Error>> + Send + Unpin + use { let executor = self.exec.clone(); let pool = self.pool.clone(); @@ -498,7 +498,6 @@ where let ver = self.config.ver; let is_ver_h2 = ver == Ver::Http2; let connector = self.connector.clone(); - let dst = domain_as_uri(pool_key.clone()); hyper_lazy(move || { // Try to take a "connecting lock". // @@ -514,6 +513,7 @@ where return Either::Right(future::err(canceled)); } }; + let dst = domain_as_uri(pool_key); Either::Left( connector .connect(super::connect::sealed::Internal, dst) diff --git a/crates/primp-hyper-util/src/client/legacy/connect/dns.rs b/crates/primp-hyper-util/src/client/legacy/connect/dns.rs index 3da6845..f57179c 100644 --- a/crates/primp-hyper-util/src/client/legacy/connect/dns.rs +++ b/crates/primp-hyper-util/src/client/legacy/connect/dns.rs @@ -297,7 +297,7 @@ pub(super) async fn resolve(resolver: &mut R, name: Name) -> Result { match self.fallback { None => self.preferred.connect(self.config).await, Some(mut fallback) => { - let preferred_fut = self.preferred.connect(self.config); - futures_util::pin_mut!(preferred_fut); - - let fallback_fut = fallback.remote.connect(self.config); - futures_util::pin_mut!(fallback_fut); - - let fallback_delay = fallback.delay; - futures_util::pin_mut!(fallback_delay); + let preferred_fut = pin!(self.preferred.connect(self.config)); + let fallback_fut = pin!(fallback.remote.connect(self.config)); + let fallback_delay = pin!(fallback.delay); let (result, future) = match futures_util::future::select(preferred_fut, fallback_delay).await { diff --git a/crates/primp-hyper-util/src/client/legacy/connect/proxy/tunnel.rs b/crates/primp-hyper-util/src/client/legacy/connect/proxy/tunnel.rs index 34abbd8..32341ff 100644 --- a/crates/primp-hyper-util/src/client/legacy/connect/proxy/tunnel.rs +++ b/crates/primp-hyper-util/src/client/legacy/connect/proxy/tunnel.rs @@ -4,10 +4,10 @@ use std::marker::{PhantomData, Unpin}; use std::pin::Pin; use std::task::{self, Poll}; -use futures_core::ready; use http::{HeaderMap, HeaderValue, Uri}; use hyper::rt::{Read, Write}; use pin_project_lite::pin_project; +use std::task::ready; use tower_service::Service; /// Tunnel Proxy via HTTP CONNECT diff --git a/crates/primp-hyper-util/src/client/legacy/pool.rs b/crates/primp-hyper-util/src/client/legacy/pool.rs index cad2e14..cec4be8 100644 --- a/crates/primp-hyper-util/src/client/legacy/pool.rs +++ b/crates/primp-hyper-util/src/client/legacy/pool.rs @@ -14,7 +14,7 @@ use std::task::{self, Poll}; use std::time::{Duration, Instant}; use futures_channel::oneshot; -use futures_core::ready; +use std::task::ready; use tracing::{debug, trace}; use hyper::rt::Timer as _; diff --git a/crates/primp-hyper-util/src/client/pool/cache.rs b/crates/primp-hyper-util/src/client/pool/cache.rs index 13f1b73..608dd99 100644 --- a/crates/primp-hyper-util/src/client/pool/cache.rs +++ b/crates/primp-hyper-util/src/client/pool/cache.rs @@ -24,8 +24,8 @@ mod internal { use std::sync::{Arc, Mutex, Weak}; use std::task::{self, Poll}; - use futures_core::ready; use futures_util::future; + use std::task::ready; use tokio::sync::oneshot; use tower_service::Service; @@ -448,7 +448,7 @@ mod tests { let mut cache = super::builder().build(mock); handle.allow(1); - crate::common::future::poll_fn(|cx| cache.poll_ready(cx)) + std::future::poll_fn(|cx| cache.poll_ready(cx)) .await .unwrap(); @@ -470,7 +470,7 @@ mod tests { // only 1 connection should ever be made handle.allow(1); - crate::common::future::poll_fn(|cx| cache.poll_ready(cx)) + std::future::poll_fn(|cx| cache.poll_ready(cx)) .await .unwrap(); let f = cache.call(1); @@ -482,7 +482,7 @@ mod tests { .expect("call"); drop(cached); - crate::common::future::poll_fn(|cx| cache.poll_ready(cx)) + std::future::poll_fn(|cx| cache.poll_ready(cx)) .await .unwrap(); let f = cache.call(1); diff --git a/crates/primp-hyper-util/src/client/pool/negotiate.rs b/crates/primp-hyper-util/src/client/pool/negotiate.rs index dd1c678..a91da12 100644 --- a/crates/primp-hyper-util/src/client/pool/negotiate.rs +++ b/crates/primp-hyper-util/src/client/pool/negotiate.rs @@ -51,8 +51,8 @@ mod internal { use std::sync::{Arc, Mutex}; use std::task::{self, Poll}; - use futures_core::ready; use pin_project_lite::pin_project; + use std::task::ready; use tower_layer::Layer; use tower_service::Service; @@ -580,7 +580,7 @@ mod tests { .upgrade(layer_fn(|s| s)) .build(); - crate::common::future::poll_fn(|cx| negotiate.poll_ready(cx)) + std::future::poll_fn(|cx| negotiate.poll_ready(cx)) .await .unwrap(); @@ -605,7 +605,7 @@ mod tests { .upgrade(layer_fn(|s| s)) .build(); - crate::common::future::poll_fn(|cx| negotiate.poll_ready(cx)) + std::future::poll_fn(|cx| negotiate.poll_ready(cx)) .await .unwrap(); diff --git a/crates/primp-hyper-util/src/client/pool/singleton.rs b/crates/primp-hyper-util/src/client/pool/singleton.rs index a45daba..4452329 100644 --- a/crates/primp-hyper-util/src/client/pool/singleton.rs +++ b/crates/primp-hyper-util/src/client/pool/singleton.rs @@ -154,8 +154,8 @@ mod internal { use std::sync::{Mutex, Weak}; use std::task::{self, Poll}; - use futures_core::ready; use pin_project_lite::pin_project; + use std::task::ready; use tokio::sync::oneshot; use tower_service::Service; @@ -351,7 +351,7 @@ mod tests { let mut singleton = Singleton::new(mock_svc); handle.allow(1); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); // First call: should go into Driving @@ -374,7 +374,7 @@ mod tests { let mut singleton = Singleton::new(mock_svc); handle.allow(1); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); // Drive first call to completion @@ -396,7 +396,7 @@ mod tests { // Allow the singleton to be made outer_handle.allow(2); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); @@ -417,7 +417,7 @@ mod tests { )); // Drive poll_ready on cached service - let err = crate::common::future::poll_fn(|cx| cached.poll_ready(cx)) + let err = std::future::poll_fn(|cx| cached.poll_ready(cx)) .await .err() .expect("expected poll_ready error"); @@ -425,7 +425,7 @@ mod tests { // After error, the singleton should be cleared, so a new call drives outer again outer_handle.allow(1); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); let fut2 = singleton.call(()); @@ -436,7 +436,7 @@ mod tests { // The new cached service should still work inner_handle2.allow(1); - crate::common::future::poll_fn(|cx| cached2.poll_ready(cx)) + std::future::poll_fn(|cx| cached2.poll_ready(cx)) .await .expect("expected poll_ready"); let cfut2 = cached2.call(()); @@ -450,7 +450,7 @@ mod tests { let (mock_svc, mut handle) = tower_test::mock::pair::<(), &'static str>(); let mut singleton = Singleton::new(mock_svc); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); let fut1 = singleton.call(()); @@ -469,14 +469,14 @@ mod tests { let (mock_svc, mut handle) = tower_test::mock::pair::<(), &'static str>(); let mut singleton = Singleton::new(mock_svc); - crate::common::future::poll_fn(|cx| singleton.poll_ready(cx)) + std::future::poll_fn(|cx| singleton.poll_ready(cx)) .await .unwrap(); let mut fut1 = singleton.call(()); let fut2 = singleton.call(()); // poll driver just once, and then drop - crate::common::future::poll_fn(move |cx| { + std::future::poll_fn(move |cx| { let _ = Pin::new(&mut fut1).poll(cx); Poll::Ready(()) }) diff --git a/crates/primp-hyper-util/src/client/proxy/matcher.rs b/crates/primp-hyper-util/src/client/proxy/matcher.rs index 4b4548c..4dcf510 100644 --- a/crates/primp-hyper-util/src/client/proxy/matcher.rs +++ b/crates/primp-hyper-util/src/client/proxy/matcher.rs @@ -575,7 +575,7 @@ mod builder { #[cfg(feature = "client-proxy-system")] #[cfg(target_os = "macos")] mod mac { - use system_configuration::core_foundation::base::{CFType, TCFType, TCFTypeRef}; + use system_configuration::core_foundation::base::CFType; use system_configuration::core_foundation::dictionary::CFDictionary; use system_configuration::core_foundation::number::CFNumber; use system_configuration::core_foundation::string::{CFString, CFStringRef}; @@ -586,7 +586,9 @@ mod mac { }; pub(super) fn with_system(builder: &mut super::Builder) { - let store = SCDynamicStoreBuilder::new("").build(); + let Some(store) = SCDynamicStoreBuilder::new("").build() else { + return; + }; let proxies_map = if let Some(proxies_map) = store.get_proxies() { proxies_map diff --git a/crates/primp-hyper-util/src/common/future.rs b/crates/primp-hyper-util/src/common/future.rs deleted file mode 100644 index 47897f2..0000000 --- a/crates/primp-hyper-util/src/common/future.rs +++ /dev/null @@ -1,30 +0,0 @@ -use std::{ - future::Future, - pin::Pin, - task::{Context, Poll}, -}; - -// TODO: replace with `std::future::poll_fn` once MSRV >= 1.64 -pub(crate) fn poll_fn(f: F) -> PollFn -where - F: FnMut(&mut Context<'_>) -> Poll, -{ - PollFn { f } -} - -pub(crate) struct PollFn { - f: F, -} - -impl Unpin for PollFn {} - -impl Future for PollFn -where - F: FnMut(&mut Context<'_>) -> Poll, -{ - type Output = T; - - fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { - (self.f)(cx) - } -} diff --git a/crates/primp-hyper-util/src/common/mod.rs b/crates/primp-hyper-util/src/common/mod.rs index a32c25d..6586aab 100644 --- a/crates/primp-hyper-util/src/common/mod.rs +++ b/crates/primp-hyper-util/src/common/mod.rs @@ -17,5 +17,3 @@ pub(crate) use exec::Exec; pub(crate) use lazy::{lazy, Started as Lazy}; #[cfg(feature = "client-legacy")] pub(crate) use sync::SyncWrapper; - -pub(crate) mod future; diff --git a/crates/primp-hyper-util/src/rt/io.rs b/crates/primp-hyper-util/src/rt/io.rs index 888756f..8dcfca9 100644 --- a/crates/primp-hyper-util/src/rt/io.rs +++ b/crates/primp-hyper-util/src/rt/io.rs @@ -2,10 +2,10 @@ use std::marker::Unpin; use std::pin::Pin; use std::task::Poll; -use futures_core::ready; use hyper::rt::{Read, ReadBuf, Write}; +use std::task::ready; -use crate::common::future::poll_fn; +use std::future::poll_fn; pub(crate) async fn read(io: &mut T, buf: &mut [u8]) -> Result where diff --git a/crates/primp-hyper-util/src/server/conn/auto/mod.rs b/crates/primp-hyper-util/src/server/conn/auto/mod.rs index 275b30c..46032c7 100644 --- a/crates/primp-hyper-util/src/server/conn/auto/mod.rs +++ b/crates/primp-hyper-util/src/server/conn/auto/mod.rs @@ -11,7 +11,6 @@ use std::task::{Context, Poll}; use std::{error::Error as StdError, io, time::Duration}; use bytes::Bytes; -use futures_core::ready; use http::{Request, Response}; use http_body::Body; use hyper::{ @@ -19,6 +18,7 @@ use hyper::{ rt::{Read, ReadBuf, Timer, Write}, service::Service, }; +use std::task::ready; #[cfg(feature = "http1")] use hyper::server::conn::http1; diff --git a/crates/primp-hyper-util/src/service/oneshot.rs b/crates/primp-hyper-util/src/service/oneshot.rs index 2cc3e6e..17892ab 100644 --- a/crates/primp-hyper-util/src/service/oneshot.rs +++ b/crates/primp-hyper-util/src/service/oneshot.rs @@ -1,7 +1,7 @@ -use futures_core::ready; use pin_project_lite::pin_project; use std::future::Future; use std::pin::Pin; +use std::task::ready; use std::task::{Context, Poll}; use tower_service::Service; diff --git a/crates/primp-hyper/Cargo.toml b/crates/primp-hyper/Cargo.toml index acbe779..79fbbb6 100644 --- a/crates/primp-hyper/Cargo.toml +++ b/crates/primp-hyper/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "primp-hyper" -version = "1.9.0" +version = "1.9.0" # based on upstream v1.9.0, cherry-picks through 2026-05-23 description = "Fork of the original hyper HTTP library." readme = "README.md" license = "MIT" diff --git a/crates/primp-hyper/src/body/incoming.rs b/crates/primp-hyper/src/body/incoming.rs index 8c7c6ff..4b0ca60 100644 --- a/crates/primp-hyper/src/body/incoming.rs +++ b/crates/primp-hyper/src/body/incoming.rs @@ -248,14 +248,14 @@ impl Body for Incoming { return Poll::Ready(Some(Ok(Frame::data(bytes)))); } Some(Err(e)) => { - return match e.reason() { - // These reasons should cause the body reading to stop, but not fail it. - // The same logic as for `Read for H2Upgraded` is applied here. - Some(h2::Reason::NO_ERROR) | Some(h2::Reason::CANCEL) => { - Poll::Ready(None) - } - _ => Poll::Ready(Some(Err(crate::Error::new_body(e)))), - }; + if let Some(h2::Reason::NO_ERROR) = e.reason() { + // As mentioned in RFC 7540 Section 8.1, a RST_STREAM with NO_ERROR + // indicates an early response, and should cause the body reading + // to stop, but not fail it: + return Poll::Ready(None); + } else { + return Poll::Ready(Some(Err(crate::Error::new_body(e)))); + } } None => { *data_done = true; @@ -270,7 +270,16 @@ impl Body for Incoming { ping.record_non_data(); Poll::Ready(Ok(t.map(Frame::trailers)).transpose()) } - Err(e) => Poll::Ready(Some(Err(crate::Error::new_h2(e)))), + Err(e) => { + if let Some(h2::Reason::NO_ERROR) = e.reason() { + // Same as above, a RST_STREAM with NO_ERROR indicates an early + // response, and should cause reading the trailers to stop, but + // not fail it: + Poll::Ready(None) + } else { + Poll::Ready(Some(Err(crate::Error::new_h2(e)))) + } + } } } diff --git a/crates/primp-hyper/src/proto/h1/decode.rs b/crates/primp-hyper/src/proto/h1/decode.rs index 102e54e..4c8c9eb 100644 --- a/crates/primp-hyper/src/proto/h1/decode.rs +++ b/crates/primp-hyper/src/proto/h1/decode.rs @@ -152,7 +152,7 @@ impl Decoder { if *remaining == 0 { Poll::Ready(Ok(Frame::data(Bytes::new()))) } else { - let to_read = *remaining as usize; + let to_read = usize::try_from(*remaining).unwrap_or(usize::MAX); let buf = ready!(body.read_mem(cx, to_read))?; let num = buf.as_ref().len() as u64; if num > *remaining { @@ -490,12 +490,7 @@ impl ChunkedState { trace!("Chunked read, remaining={:?}", rem); // cap remaining bytes at the max capacity of usize - let rem_cap = match *rem { - r if r > usize::MAX as u64 => usize::MAX, - r => r as usize, - }; - - let to_read = rem_cap; + let to_read = usize::try_from(*rem).unwrap_or(usize::MAX); let slice = ready!(rdr.read_mem(cx, to_read))?; let count = slice.len(); diff --git a/crates/primp-hyper/src/proto/h1/dispatch.rs b/crates/primp-hyper/src/proto/h1/dispatch.rs index 024cdf1..c2ccac8 100644 --- a/crates/primp-hyper/src/proto/h1/dispatch.rs +++ b/crates/primp-hyper/src/proto/h1/dispatch.rs @@ -22,7 +22,7 @@ use crate::upgrade::OnUpgrade; pub(crate) struct Dispatcher { conn: Conn, dispatch: D, - body_tx: Option, + body_tx: SenderDropGuard, body_rx: Pin>>, is_closing: bool, } @@ -82,7 +82,7 @@ where Dispatcher { conn, dispatch, - body_tx: None, + body_tx: SenderDropGuard::none(), body_rx: Box::pin(None), is_closing: false, } @@ -171,8 +171,14 @@ where // benchmarks often use. Perhaps it should be a config option instead. for _ in 0..16 { let _ = self.poll_read(cx)?; - let _ = self.poll_write(cx)?; - let _ = self.poll_flush(cx)?; + let write_ready = self.poll_write(cx)?.is_ready(); + let flush_ready = self.poll_flush(cx)?.is_ready(); + + // If we can write more body and the connection is ready, we should + // write again. If we return `Ready(Ok(())` here, we will yield + // without a guaranteed wake-up from the write side of the connection. + // This would lead to a deadlock if we also don't expect reads. + let wants_write_again = self.can_write_again() && (write_ready || flush_ready); // This could happen if reading paused before blocking on IO, // such as getting to the end of a framed message, but then @@ -182,10 +188,29 @@ where // // Using this instead of task::current() and notify() inside // the Conn is noticeably faster in pipelined benchmarks. - if !self.conn.wants_read_again() { - //break; + let wants_read_again = self.conn.wants_read_again(); + + // If we cannot write or read again, we yield and rely on the + // wake-up from the connection futures. + if !(wants_write_again || wants_read_again) { return Poll::Ready(Ok(())); } + + // If we are continuing only because "wants_write_again", check if write is ready. + if !wants_read_again && wants_write_again { + // If write was ready, just proceed with the loop + if write_ready { + continue; + } + // Write was previously pending, but may have become ready since polling flush, so + // we need to check it again. If we simply proceeded, the case of an unbuffered + // writer where flush is always ready would cause us to hot loop. + if self.poll_write(cx)?.is_pending() { + // write is pending, so it is safe to yield and rely on wake-up from connection + // futures. + return Poll::Ready(Ok(())); + } + } } trace!("poll_loop yielding (self = {:p})", self); @@ -204,7 +229,7 @@ where match body.poll_ready(cx) { Poll::Ready(Ok(())) => (), Poll::Pending => { - self.body_tx = Some(body); + self.body_tx.set(body); return Poll::Pending; } Poll::Ready(Err(_canceled)) => { @@ -221,7 +246,7 @@ where let chunk = frame.into_data().unwrap_or_else(|_| unreachable!()); match body.try_send_data(chunk) { Ok(()) => { - self.body_tx = Some(body); + self.body_tx.set(body); } Err(_canceled) => { if self.conn.can_read_body() { @@ -235,7 +260,7 @@ where frame.into_trailers().unwrap_or_else(|_| unreachable!()); match body.try_send_trailers(trailers) { Ok(()) => { - self.body_tx = Some(body); + self.body_tx.set(body); } Err(_canceled) => { if self.conn.can_read_body() { @@ -253,7 +278,7 @@ where // just drop, the body will close automatically } Poll::Pending => { - self.body_tx = Some(body); + self.body_tx.set(body); return Poll::Pending; } Poll::Ready(Some(Err(e))) => { @@ -288,7 +313,7 @@ where other => { let (tx, rx) = IncomingBody::new_channel(other, wants.contains(Wants::EXPECT)); - self.body_tx = Some(tx); + self.body_tx.set(tx); rx } }; @@ -434,6 +459,12 @@ where self.conn.close_write(); } + /// If there is pending data in `body_rx`, we can make progress writing if the connection is + /// ready. + fn can_write_again(&mut self) -> bool { + self.body_rx.is_some() + } + fn is_done(&self) -> bool { if self.is_closing { return true; @@ -497,6 +528,38 @@ impl Drop for OptGuard<'_, T> { } } +// ===== impl SenderDropGuard ===== + +/// A drop guard for the body `Sender`. +/// +/// If the `Dispatcher` future is dropped (e.g. the runtime driving the +/// connection is shut down) while it still owns a body `Sender`, the guard +/// sends an incomplete-message error so the receiver sees an error instead +/// of a silent, clean end-of-stream. +struct SenderDropGuard(Option); + +impl SenderDropGuard { + fn none() -> Self { + SenderDropGuard(None) + } + + fn set(&mut self, sender: crate::body::Sender) { + self.0 = Some(sender); + } + + fn take(&mut self) -> Option { + self.0.take() + } +} + +impl Drop for SenderDropGuard { + fn drop(&mut self) { + if let Some(mut sender) = self.0.take() { + sender.send_error(crate::Error::new_incomplete()); + } + } +} + // ===== impl Server ===== cfg_server! { diff --git a/crates/primp-hyper/src/proto/h2/mod.rs b/crates/primp-hyper/src/proto/h2/mod.rs index 22dff28..37fffa9 100644 --- a/crates/primp-hyper/src/proto/h2/mod.rs +++ b/crates/primp-hyper/src/proto/h2/mod.rs @@ -88,11 +88,17 @@ pin_project! { { body_tx: SendStream>, data_done: bool, + buffered_data: Option>, #[pin] stream: S, } } +struct Peeked { + data: D, + is_eos: bool, +} + impl PipeToSendStream where S: Body, @@ -101,6 +107,7 @@ where PipeToSendStream { body_tx: tx, data_done: false, + buffered_data: None, stream, } } @@ -121,55 +128,87 @@ where fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll { let mut me = self.project(); loop { - // we don't have the next chunk of data yet, so just reserve 1 byte to make - // sure there's some capacity available. h2 will handle the capacity management - // for the actual body chunk. - me.body_tx.reserve_capacity(1); + // Register for RST_STREAM notification while we wait for the next + // body chunk or for send capacity, so the task wakes up if the + // peer resets the stream. + if let Poll::Ready(reason) = me + .body_tx + .poll_reset(cx) + .map_err(crate::Error::new_body_write)? + { + debug!("stream received RST_STREAM: {:?}", reason); + return Poll::Ready(Err(crate::Error::new_body_write(::h2::Error::from(reason)))); + } - if me.body_tx.capacity() == 0 { - loop { + // If a previously-polled chunk is still waiting for stream-level + // send capacity, drive that to completion before touching the + // body again. + if me.buffered_data.is_some() { + while me.body_tx.capacity() == 0 { match ready!(me.body_tx.poll_capacity(cx)) { Some(Ok(0)) => {} Some(Ok(_)) => break, Some(Err(e)) => return Poll::Ready(Err(crate::Error::new_body_write(e))), None => { - // None means the stream is no longer in a - // streaming state, we either finished it - // somehow, or the remote reset us. return Poll::Ready(Err(crate::Error::new_body_write( "send stream capacity unexpectedly closed", ))); } } } - } else if let Poll::Ready(reason) = me - .body_tx - .poll_reset(cx) - .map_err(crate::Error::new_body_write)? - { - debug!("stream received RST_STREAM: {:?}", reason); - return Poll::Ready(Err(crate::Error::new_body_write(::h2::Error::from(reason)))); + + let peeked = me.buffered_data.take().expect("checked is_some above"); + let buf = SendBuf::Buf(peeked.data); + me.body_tx + .send_data(buf, peeked.is_eos) + .map_err(crate::Error::new_body_write)?; + + if peeked.is_eos { + return Poll::Ready(Ok(())); + } + continue; } + // Poll for the next body frame *before* reserving any connection + // flow-control capacity. Reserving capacity speculatively (even a + // single byte) pins that capacity on the connection-level window, + // which can deadlock a second stream when talking to peers that + // only emit WINDOW_UPDATE once their receive window is fully + // exhausted. See #4003. match ready!(me.stream.as_mut().poll_frame(cx)) { Some(Ok(frame)) => { if frame.is_data() { let chunk = frame.into_data().unwrap_or_else(|_| unreachable!()); let is_eos = me.stream.is_end_stream(); - trace!( - "send body chunk: {} bytes, eos={}", - chunk.remaining(), - is_eos, - ); + let len = chunk.remaining(); + trace!("send body chunk: {} bytes, eos={}", len, is_eos); - let buf = SendBuf::Buf(chunk); - me.body_tx - .send_data(buf, is_eos) - .map_err(crate::Error::new_body_write)?; + if len == 0 { + // Zero-length data frames need no capacity; send + // them straight through so trailing empty frames + // (e.g. an explicit end-of-stream marker) are + // delivered. + let buf = SendBuf::Buf(chunk); + me.body_tx + .send_data(buf, is_eos) + .map_err(crate::Error::new_body_write)?; - if is_eos { - return Poll::Ready(Ok(())); + if is_eos { + return Poll::Ready(Ok(())); + } + continue; } + + // Reserve exactly the chunk size so we never pin more + // connection-level flow-control window than we are + // about to consume. Stash the chunk in `self` so it + // survives the upcoming `poll_capacity` wait even if + // it returns `Poll::Pending`. + me.body_tx.reserve_capacity(len); + *me.buffered_data = Some(Peeked { + data: chunk, + is_eos, + }); } else if frame.is_trailers() { // no more DATA, so give any capacity back me.body_tx.reserve_capacity(0); diff --git a/crates/primp-python/src/async/client.rs b/crates/primp-python/src/async/client.rs index 30e7287..7506cd3 100644 --- a/crates/primp-python/src/async/client.rs +++ b/crates/primp-python/src/async/client.rs @@ -230,26 +230,19 @@ impl AsyncClient { client_guard.set_cookies(&url_parsed, cookie_values); } - // Handle follow_redirects: set policy before cloning client - if let Some(fr) = follow_redirects { - let mut client_guard = self.client.write().unwrap_or_else(|e| e.into_inner()); - if fr { - client_guard.set_redirect_policy(::primp::redirect::Policy::limited(20)); - } else { - client_guard.set_redirect_policy(::primp::redirect::Policy::none()); - } - } - // Clone the client before entering the async block to avoid holding RwLockGuard across await - let client = { + let mut client = { let client_guard = self.client.read().unwrap_or_else(|e| e.into_inner()); client_guard.clone() }; - // Restore redirect policy immediately after cloning if it was changed - if follow_redirects.is_some() { - let mut client_guard = self.client.write().unwrap_or_else(|e| e.into_inner()); - client_guard.set_redirect_policy(::primp::redirect::Policy::limited(20)); + // Apply follow_redirects on the private clone, not the shared client + if let Some(fr) = follow_redirects { + if fr { + client.set_redirect_policy(::primp::redirect::Policy::limited(20)); + } else { + client.set_redirect_policy(::primp::redirect::Policy::none()); + } } let future = async move { diff --git a/crates/primp-python/src/lib.rs b/crates/primp-python/src/lib.rs index 04c812c..1a90a29 100644 --- a/crates/primp-python/src/lib.rs +++ b/crates/primp-python/src/lib.rs @@ -38,7 +38,6 @@ use utils::extract_encoding; static RUNTIME: PyOnceLock = PyOnceLock::new(); /// Get the global Tokio runtime, initializing it if necessary. -#[inline(always)] pub(crate) fn get_runtime(py: Python<'_>) -> &Runtime { RUNTIME.get_or_init(py, || { runtime::Builder::new_current_thread() @@ -356,25 +355,18 @@ impl Client { client_guard.set_cookies(&url_parsed, cookie_values); } - // Handle follow_redirects: set policy before cloning client + // Clone the inner client to avoid holding the RwLock across await points + let mut client_clone = client.read().unwrap_or_else(|e| e.into_inner()).clone(); + + // Apply follow_redirects on the private clone, not the shared client if let Some(fr) = follow_redirects { - let mut client_guard = client.write().unwrap_or_else(|e| e.into_inner()); if fr { - client_guard.set_redirect_policy(::primp::redirect::Policy::limited(20)); + client_clone.set_redirect_policy(::primp::redirect::Policy::limited(20)); } else { - client_guard.set_redirect_policy(::primp::redirect::Policy::none()); + client_clone.set_redirect_policy(::primp::redirect::Policy::none()); } } - // Clone the inner client to avoid holding the RwLock across await points - let client_clone = client.read().unwrap_or_else(|e| e.into_inner()).clone(); - - // Restore redirect policy immediately after cloning if it was changed - if follow_redirects.is_some() { - let mut client_guard = client.write().unwrap_or_else(|e| e.into_inner()); - client_guard.set_redirect_policy(::primp::redirect::Policy::limited(20)); - } - let future = async move { // Create request builder using the cloned client let mut request_builder = client_clone.request(method, &resolved_url); diff --git a/crates/primp-python/src/utils.rs b/crates/primp-python/src/utils.rs index 07db0ce..6990aac 100644 --- a/crates/primp-python/src/utils.rs +++ b/crates/primp-python/src/utils.rs @@ -93,13 +93,9 @@ pub fn extract_encoding(headers: &::primp::header::HeaderMap) -> &'static encodi mod load_ca_certs_tests { use super::*; use std::fs; - use std::path::Path; + use std::path::{Path, PathBuf}; - #[test] - fn test_load_ca_certs_from_file() { - // Create a temporary file with a CA certificate - let ca_cert_path = Path::new("test_ca_cert.pem"); - let ca_cert = "-----BEGIN CERTIFICATE----- + const TEST_CERT: &str = "-----BEGIN CERTIFICATE----- MIIDdTCCAl2gAwIBAgIVAMIIujU9wQIBADANBgkqhkiG9w0BAQUFADBGMQswCQYD VQQGEwJVUzETMBEGA1UECAwKQ2FsaWZvcm5pYTEWMBQGA1UEBwwNTW91bnRhaW4g Q29sbGVjdGlvbjEgMB4GA1UECgwXUG9zdGdyZXMgQ29uc3VsdGF0aW9uczEhMB8G @@ -111,50 +107,42 @@ Q29sbGVjdGlvbjEgMB4GA1UECgwXUG9zdGdyZXMgQ29uc3VsdGF0aW9uczEhMB8G A1UECwwYUG9zdGdyZXMgQ29uc3VsdGF0aW9uczEhMB8GA1UEAwwYUG9zdGdyZXMg Q29uc3VsdGF0aW9uczEiMCAGCSqGSIb3DQEJARYTcGVyc29uYWwtZW1haWwuY29t -----END CERTIFICATE-----"; - fs::write(ca_cert_path, ca_cert).unwrap(); - // Call the function - let result = load_ca_certs_from_file(ca_cert_path); + struct TempFile { + path: PathBuf, + } - // Check the result - assert!(result.is_some()); + impl TempFile { + fn new(name: &str, content: &str) -> Self { + let path = PathBuf::from(name); + fs::write(&path, content).unwrap(); + TempFile { path } + } + } - // Clean up - fs::remove_file(ca_cert_path).unwrap(); + impl Drop for TempFile { + fn drop(&mut self) { + let _ = fs::remove_file(&self.path); + } } #[test] - fn test_load_ca_certs_with_ca_cert_file_param() { - // Create a temporary file with a CA certificate - let ca_cert_path = Path::new("test_ca_cert2.pem"); - let ca_cert = "-----BEGIN CERTIFICATE----- -MIIDdTCCAl2gAwIBAgIVAMIIujU9wQIBADANBgkqhkiG9w0BAQUFADBGMQswCQYD -VQQGEwJVUzETMBEGA1UECAwKQ2FsaWZvcm5pYTEWMBQGA1UEBwwNTW91bnRhaW4g -Q29sbGVjdGlvbjEgMB4GA1UECgwXUG9zdGdyZXMgQ29uc3VsdGF0aW9uczEhMB8G -A1UECwwYUG9zdGdyZXMgQ29uc3VsdGF0aW9uczEhMB8GA1UEAwwYUG9zdGdyZXMg -Q29uc3VsdGF0aW9uczEiMCAGCSqGSIb3DQEJARYTcGVyc29uYWwtZW1haWwuY29t ------END CERTIFICATE-----"; - fs::write(ca_cert_path, ca_cert).unwrap(); - - // Call the function with Some path - let result = load_ca_certs(&Some(ca_cert_path.to_str().unwrap().to_string())); - - // Check the result + fn test_load_ca_certs_from_file() { + let _file = TempFile::new("test_ca_cert.pem", TEST_CERT); + let result = load_ca_certs_from_file(Path::new("test_ca_cert.pem")); assert!(result.is_some()); + } - // Clean up - fs::remove_file(ca_cert_path).unwrap(); + #[test] + fn test_load_ca_certs_with_ca_cert_file_param() { + let _file = TempFile::new("test_ca_cert2.pem", TEST_CERT); + let result = load_ca_certs(&Some("test_ca_cert2.pem".to_string())); + assert!(result.is_some()); } #[test] fn test_load_ca_certs_with_none() { - // Call the function with None let result = load_ca_certs(&None); - - // The function will return Some(...) if any CA cert env vars are set - // (PRIMP_CA_BUNDLE, SSL_CERT_FILE, CURL_CA_BUNDLE), or None otherwise. - // This is expected behavior - the function falls back to env vars. - // We just verify the function doesn't panic and returns a valid Option. assert!(result.is_some() || result.is_none()); } } diff --git a/crates/primp-reqwest/Cargo.toml b/crates/primp-reqwest/Cargo.toml index 6c67930..994bf9b 100644 --- a/crates/primp-reqwest/Cargo.toml +++ b/crates/primp-reqwest/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "primp-reqwest" -version = "0.13.3" +version = "0.13.3" # based on upstream v0.13.3, cherry-picks through 2026-05-23 description = "Fork of the original reqwest library" keywords = ["http", "request", "client"] categories = ["web-programming::http-client", "wasm"] diff --git a/crates/primp-reqwest/src/async_impl/response.rs b/crates/primp-reqwest/src/async_impl/response.rs index 7c68d63..0678da9 100644 --- a/crates/primp-reqwest/src/async_impl/response.rs +++ b/crates/primp-reqwest/src/async_impl/response.rs @@ -265,9 +265,9 @@ impl Response { #[cfg(feature = "json")] #[cfg_attr(docsrs, doc(cfg(feature = "json")))] pub async fn json(self) -> crate::Result { - let full = self.bytes().await?; + let (full, url) = self.do_bytes().await?; - serde_json::from_slice(&full).map_err(crate::error::decode) + serde_json::from_slice(&full).map_err(|err| crate::error::decode(err).with_url(*url)) } /// Get the full response body as `Bytes`. @@ -286,12 +286,7 @@ impl Response { /// # } /// ``` pub async fn bytes(self) -> crate::Result { - use http_body_util::BodyExt; - - BodyExt::collect(self.res.into_body()) - .await - .map(|buf| buf.to_bytes()) - .map_err(crate::error::decode) + self.do_bytes().await.map(|(bytes, _)| bytes) } /// Stream a chunk of the response body. @@ -430,6 +425,14 @@ impl Response { pub(crate) fn body_mut(&mut self) -> &mut ResponseBody { self.res.body_mut() } + async fn do_bytes(self) -> crate::Result<(Bytes, Box)> { + use http_body_util::BodyExt; + + match BodyExt::collect(self.res.into_body()).await { + Ok(buf) => Ok((buf.to_bytes(), self.url)), + Err(err) => Err(crate::error::decode(err).with_url(*self.url)), + } + } } impl fmt::Debug for Response { diff --git a/crates/primp-reqwest/src/dns/hickory.rs b/crates/primp-reqwest/src/dns/hickory.rs index 59afca8..e258893 100644 --- a/crates/primp-reqwest/src/dns/hickory.rs +++ b/crates/primp-reqwest/src/dns/hickory.rs @@ -1,8 +1,8 @@ //! DNS resolution via the [hickory-resolver](https://github.com/hickory-dns/hickory-dns) crate use hickory_resolver::{ - config::{LookupIpStrategy, ResolverConfig}, - net::runtime::TokioRuntimeProvider, + config::{LookupIpStrategy, ResolverConfig, GOOGLE}, + net::{runtime::TokioRuntimeProvider, NetError}, TokioResolver, }; use once_cell::sync::OnceCell; @@ -29,7 +29,7 @@ impl Resolve for HickoryDnsResolver { fn resolve(&self, name: Name) -> Resolving { let resolver = self.clone(); Box::pin(async move { - let resolver = resolver.state.get_or_init(new_resolver); + let resolver = resolver.state.get_or_try_init(new_resolver)?; let lookup = resolver.lookup_ip(name.as_str()).await?; let ips: Vec = lookup.iter().collect(); @@ -54,17 +54,17 @@ impl Iterator for SocketAddrs { /// it fallbacks to hickory_resolver's default config. /// The options are overridden to look up for both IPv4 and IPv6 addresses /// to work with "happy eyeballs" algorithm. -fn new_resolver() -> TokioResolver { +fn new_resolver() -> Result { let mut builder = TokioResolver::builder_tokio().unwrap_or_else(|err| { log::debug!( - "hickory-dns: failed to load system DNS configuration; falling back to hickory_resolver defaults: {:?}", + "hickory-dns: failed to load system DNS configuration; falling back to Google DNS: {:?}", err ); TokioResolver::builder_with_config( - ResolverConfig::default(), + ResolverConfig::udp_and_tcp(&GOOGLE), TokioRuntimeProvider::default(), ) }); builder.options_mut().ip_strategy = LookupIpStrategy::Ipv4AndIpv6; - builder.build().expect("failed to build hickory resolver") + builder.build() } diff --git a/crates/primp-reqwest/src/redirect.rs b/crates/primp-reqwest/src/redirect.rs index 081d6c2..0377f8a 100644 --- a/crates/primp-reqwest/src/redirect.rs +++ b/crates/primp-reqwest/src/redirect.rs @@ -240,7 +240,8 @@ pub(crate) enum ActionKind { pub(crate) fn remove_sensitive_headers(headers: &mut HeaderMap, next: &Url, previous: &[Url]) { if let Some(previous) = previous.last() { let cross_host = next.host_str() != previous.host_str() - || next.port_or_known_default() != previous.port_or_known_default(); + || next.port_or_known_default() != previous.port_or_known_default() + || next.scheme() != previous.scheme(); if cross_host { headers.remove(AUTHORIZATION); headers.remove(COOKIE);