fix: backport upstream bug fixes across forked crates

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",

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();

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

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

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();
     }

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);

crates/primp-h2/src/share.rs

@@ -308,7 +308,7 @@ impl<B: Buf> SendStream<B> {
     /// 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<Option<Result<usize, crate::Error>>> {
         self.inner
             .poll_capacity(cx)

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 <sean@seanmonstar.com>"]
 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]

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<Box<dyn Future<Output = ()> + Send>>;
 
@@ -487,7 +487,7 @@ where
     fn connect_to(
         &self,
         pool_key: PoolKey,
-    ) -> impl Lazy<Output = Result<pool::Pooled<PoolClient<B>, PoolKey>, Error>> + Send + Unpin
+    ) -> impl Lazy<Output = Result<pool::Pooled<PoolClient<B>, PoolKey>, Error>> + Send + Unpin + use<C, B>
     {
         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)

crates/primp-hyper-util/src/client/legacy/connect/dns.rs

@@ -297,7 +297,7 @@ pub(super) async fn resolve<R>(resolver: &mut R, name: Name) -> Result<R::Addrs,
 where
     R: Resolve,
 {
-    crate::common::future::poll_fn(|cx| resolver.poll_ready(cx)).await?;
+    std::future::poll_fn(|cx| resolver.poll_ready(cx)).await?;
     resolver.resolve(name).await
 }
 

crates/primp-hyper-util/src/client/legacy/connect/http.rs

@@ -4,16 +4,16 @@ use std::future::Future;
 use std::io;
 use std::marker::PhantomData;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
-use std::pin::Pin;
+use std::pin::{pin, Pin};
 use std::sync::Arc;
 use std::task::{self, Poll};
 use std::time::Duration;
 
-use futures_core::ready;
 use futures_util::future::Either;
 use http::uri::{Scheme, Uri};
 use pin_project_lite::pin_project;
 use socket2::TcpKeepalive;
+use std::task::ready;
 use tokio::net::{TcpSocket, TcpStream};
 use tokio::time::Sleep;
 use tracing::{debug, trace, warn};
@@ -960,14 +960,9 @@ impl ConnectingTcp<'_> {
         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 {

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

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 _;

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);