fix: preserve connection task coalescing

.github/workflows/CI.yml

@@ -61,6 +61,95 @@ jobs:
         run: ./ci/h2spec.sh
         if: matrix.rust == 'stable'
 
+  ztunnel-temporary:
+    name: Temporary ztunnel tests
+    needs: [style]
+    runs-on: ubuntu-latest
+    timeout-minutes: 60
+    env:
+      RUST_BACKTRACE: full
+      RUST_TEST_THREADS: 4
+      CARGO_TARGET_X86_64_UNKNOWN_LINUX_GNU_RUNNER: sudo -E
+    steps:
+      - name: Checkout h2
+        uses: actions/checkout@v6
+        with:
+          path: h2
+
+      - name: Checkout ztunnel master
+        uses: actions/checkout@v6
+        with:
+          repository: istio/ztunnel
+          ref: master
+          path: ztunnel
+
+      - name: Install Rust
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Install diagnostics
+        run: sudo apt-get update && sudo apt-get install -y gdb libssl-dev pkg-config protobuf-compiler
+
+      - name: Print runner details
+        run: |
+          nproc
+          lscpu
+          rustc -vV
+          cargo -vV
+
+      - name: Patch ztunnel to use this h2 checkout
+        working-directory: ztunnel
+        run: |
+          cat >> Cargo.toml <<'EOF'
+
+          [patch.crates-io]
+          h2 = { path = "../h2" }
+          EOF
+          cargo update -p h2
+          cargo tree -i h2
+
+      - name: Run ztunnel test suite with diagnostics
+        working-directory: ztunnel
+        run: |
+          set -Eeuo pipefail
+
+          sudo sysctl kernel.yama.ptrace_scope=0 || true
+
+          dump_state() {
+            echo "::group::process table"
+            date -u
+            ps -eT -o pid,ppid,pgid,lwp,stat,pcpu,pmem,wchan:32,comm,args || true
+            echo "::endgroup::"
+
+            echo "::group::ztunnel test process backtraces"
+            pgrep -af 'out/rust/debug/deps|out/rust/debug/ztunnel|target/debug/deps' || true
+            for pid in $(pgrep -f 'out/rust/debug/deps|out/rust/debug/ztunnel|target/debug/deps' || true); do
+              echo "===== gdb thread backtraces for pid ${pid} ====="
+              sudo gdb -q -batch \
+                -ex 'set pagination off' \
+                -ex 'thread apply all bt' \
+                -p "${pid}" || true
+            done
+            echo "::endgroup::"
+          }
+
+          setsid cargo test --benches --tests --bins &
+          cargo_pid=$!
+          deadline=$((SECONDS + 25 * 60))
+
+          while kill -0 "${cargo_pid}" 2>/dev/null; do
+            if (( SECONDS >= deadline )); then
+              echo "::error::ztunnel cargo test timed out after 25 minutes"
+              dump_state
+              kill -TERM "-${cargo_pid}" 2>/dev/null || true
+              sleep 5
+              kill -KILL "-${cargo_pid}" 2>/dev/null || true
+              exit 124
+            fi
+            sleep 10
+          done
+
+          wait "${cargo_pid}"
+
   #clippy_check:
   #  runs-on: ubuntu-latest
   #  steps:
@@ -88,19 +177,6 @@ jobs:
         with:
           toolchain: ${{ steps.msrv.outputs.version }}
 
-      - name: Pin some dependencies for MSRV
-        run: |
-          cargo update --package tokio-util --precise 0.7.11
-          cargo update --package tokio --precise 1.38.1
-          cargo update --package indexmap --precise 2.11.3
-          cargo update --package hashbrown --precise 0.15.0
-          cargo update --package once_cell --precise 1.20.3
-          cargo update --package tracing --precise 0.1.41
-          cargo update --package tracing-subscriber --precise 0.3.19
-          cargo update --package tracing-core --precise 0.1.33
-          cargo update --package itoa --precise 1.0.15
-
-
       - run: cargo check -p h2
 
   minimal-versions:

Cargo.toml

@@ -16,8 +16,8 @@ readme = "README.md"
 keywords = ["http", "async", "non-blocking"]
 categories = ["asynchronous", "web-programming", "network-programming"]
 exclude = ["fixtures/**", "ci/**"]
-edition = "2021"
-rust-version = "1.63"
+edition = "2024"
+rust-version = "1.85"
 
 [features]
 # Enables `futures::Stream` implementations for various types.

benches/main.rs

@@ -1,8 +1,7 @@
 use bytes::Bytes;
 use h2::{
-    client,
+    RecvStream, client,
     server::{self, SendResponse},
-    RecvStream,
 };
 use http::Request;
 

examples/akamai.rs

@@ -3,7 +3,7 @@ use http::{Method, Request};
 use tokio::net::TcpStream;
 use tokio_rustls::TlsConnector;
 
-use tokio_rustls::rustls::{pki_types::ServerName, RootCertStore};
+use tokio_rustls::rustls::{RootCertStore, pki_types::ServerName};
 
 use std::error::Error;
 use std::net::ToSocketAddrs;

examples/server.rs

@@ -1,8 +1,8 @@
 use std::error::Error;
 
 use bytes::Bytes;
-use h2::server::{self, SendResponse};
 use h2::RecvStream;
+use h2::server::{self, SendResponse};
 use http::Request;
 use tokio::net::{TcpListener, TcpStream};
 

src/client.rs

@@ -142,7 +142,7 @@ use crate::proto::{self, Error};
 use crate::{FlowControl, PingPong, RecvStream, SendStream};
 
 use bytes::{Buf, Bytes};
-use http::{uri, HeaderMap, Method, Request, Response, Version};
+use http::{HeaderMap, Method, Request, Response, Version, uri};
 use std::fmt;
 use std::future::Future;
 use std::pin::Pin;
@@ -173,7 +173,7 @@ use tracing::Instrument;
 /// [`Clone`]: https://doc.rust-lang.org/std/clone/trait.Clone.html
 /// [`Error`]: ../struct.Error.html
 pub struct SendRequest<B: Buf> {
-    inner: proto::Streams<B, Peer>,
+    inner: proto::Injector<B>,
     pending: Option<proto::OpaqueStreamRef>,
 }
 
@@ -515,10 +515,10 @@ where
         end_of_stream: bool,
     ) -> Result<(ResponseFuture, SendStream<B>), crate::Error> {
         self.inner
-            .send_request(request, end_of_stream, self.pending.as_ref())
+            .enqueue_request(request, end_of_stream, self.pending.as_ref())
             .map_err(Into::into)
-            .map(|(stream, is_full)| {
-                if stream.is_pending_open() && is_full {
+            .map(|stream| {
+                if stream.is_pending_open() {
                     // Only prevent sending another request when the request queue
                     // is not full.
                     self.pending = Some(stream.clone_to_opaque());
@@ -1220,7 +1220,7 @@ impl Builder {
     pub fn handshake<T, B>(
         &self,
         io: T,
-    ) -> impl Future<Output = Result<(SendRequest<B>, Connection<T, B>), crate::Error>>
+    ) -> impl Future<Output = Result<(SendRequest<B>, Connection<T, B>), crate::Error>> + use<T, B>
     where
         T: AsyncRead + AsyncWrite + Unpin,
         B: Buf,
@@ -1338,7 +1338,7 @@ where
             },
         );
         let send_request = SendRequest {
-            inner: inner.streams().clone(),
+            inner: inner.injector(),
             pending: None,
         };
 

src/error.rs

@@ -163,13 +163,13 @@ impl fmt::Display for Error {
     fn fmt(&self, fmt: &mut fmt::Formatter) -> fmt::Result {
         let debug_data = match self.kind {
             Kind::Reset(_, reason, Initiator::User) => {
-                return write!(fmt, "stream error sent by user: {}", reason)
+                return write!(fmt, "stream error sent by user: {}", reason);
             }
             Kind::Reset(_, reason, Initiator::Library) => {
-                return write!(fmt, "stream error detected: {}", reason)
+                return write!(fmt, "stream error detected: {}", reason);
             }
             Kind::Reset(_, reason, Initiator::Remote) => {
-                return write!(fmt, "stream error received: {}", reason)
+                return write!(fmt, "stream error received: {}", reason);
             }
             Kind::GoAway(ref debug_data, reason, Initiator::User) => {
                 write!(fmt, "connection error sent by user: {}", reason)?;

src/frame/data.rs

@@ -1,4 +1,4 @@
-use crate::frame::{util, Error, Frame, Head, Kind, StreamId};
+use crate::frame::{Error, Frame, Head, Kind, StreamId, util};
 use bytes::{Buf, BufMut, Bytes};
 
 use std::fmt;

src/frame/headers.rs

@@ -1,10 +1,10 @@
-use super::{util, StreamDependency, StreamId};
+use super::{StreamDependency, StreamId, util};
 use crate::ext::Protocol;
 use crate::frame::{Error, Frame, Head, Kind};
 use crate::hpack::{self, BytesStr};
 
 use http::header::{self, HeaderName, HeaderValue};
-use http::{uri, HeaderMap, Method, Request, StatusCode, Uri};
+use http::{HeaderMap, Method, Request, StatusCode, Uri, uri};
 
 use bytes::{Buf, BufMut, Bytes, BytesMut};
 
@@ -1032,7 +1032,7 @@ fn decoded_header_size(name: usize, value: usize) -> usize {
 mod test {
     use super::*;
     use crate::frame;
-    use crate::hpack::{huffman, Encoder};
+    use crate::hpack::{Encoder, huffman};
 
     #[test]
     fn test_nameless_header_at_resume() {
@@ -1072,9 +1072,11 @@ mod test {
 
         dst.clear();
 
-        assert!(continuation
-            .encode(&mut (&mut dst).limit(frame::HEADER_LEN + 16))
-            .is_none());
+        assert!(
+            continuation
+                .encode(&mut (&mut dst).limit(frame::HEADER_LEN + 16))
+                .is_none()
+        );
 
         world.extend_from_slice(&dst[9..12]);
         assert_eq!("world", huff_decode(&world));

src/frame/mod.rs

@@ -52,7 +52,7 @@ pub use self::data::Data;
 pub use self::go_away::GoAway;
 pub use self::head::{Head, Kind};
 pub use self::headers::{
-    parse_u64, Continuation, Headers, Pseudo, PushPromise, PushPromiseHeaderError,
+    Continuation, Headers, Pseudo, PushPromise, PushPromiseHeaderError, parse_u64,
 };
 pub use self::ping::Ping;
 pub use self::priority::{Priority, StreamDependency};

src/frame/settings.rs

@@ -1,6 +1,6 @@
 use std::fmt;
 
-use crate::frame::{util, Error, Frame, FrameSize, Head, Kind, StreamId};
+use crate::frame::{Error, Frame, FrameSize, Head, Kind, StreamId, util};
 use bytes::{BufMut, BytesMut};
 
 #[derive(Clone, Default, Eq, PartialEq)]

src/hpack/decoder.rs

@@ -1,4 +1,4 @@
-use super::{header::BytesStr, huffman, Header};
+use super::{Header, header::BytesStr, huffman};
 use crate::frame;
 
 use bytes::{Buf, Bytes, BytesMut};

src/hpack/encoder.rs

@@ -1,5 +1,5 @@
 use super::table::{Index, Table};
-use super::{huffman, Header};
+use super::{Header, huffman};
 
 use bytes::{BufMut, BytesMut};
 use http::header::{HeaderName, HeaderValue};

src/hpack/test/fixture.rs

@@ -5,8 +5,8 @@ use hex::FromHex;
 use serde_json::Value;
 
 use std::fs::File;
-use std::io::prelude::*;
 use std::io::Cursor;
+use std::io::prelude::*;
 use std::ops::ControlFlow;
 use std::path::Path;
 use std::str;

src/hpack/test/fuzz.rs

@@ -6,7 +6,7 @@ use bytes::BytesMut;
 use quickcheck::{Arbitrary, Gen, QuickCheck, TestResult};
 use rand::distributions::Slice;
 use rand::rngs::StdRng;
-use rand::{thread_rng, Rng, SeedableRng};
+use rand::{Rng, SeedableRng, thread_rng};
 
 use std::io::Cursor;
 use std::ops::ControlFlow;
@@ -182,7 +182,7 @@ impl FuzzHpack {
 
 impl Arbitrary for FuzzHpack {
     fn arbitrary(_: &mut Gen) -> Self {
-        FuzzHpack::new(thread_rng().gen())
+        FuzzHpack::new(thread_rng().r#gen())
     }
 }
 
@@ -234,7 +234,7 @@ fn gen_header(g: &mut StdRng) -> Header<Option<HeaderName>> {
                 Header::Path(to_shared(value))
             }
             4 => {
-                let status = (g.gen::<u16>() % 500) + 100;
+                let status = (g.r#gen::<u16>() % 500) + 100;
 
                 Header::Status(StatusCode::from_u16(status).unwrap())
             }

src/proto/connection.rs

@@ -192,10 +192,11 @@ where
         // The order of these calls don't really matter too much
         ready!(self.inner.ping_pong.send_pending_pong(cx, &mut self.codec))?;
         ready!(self.inner.ping_pong.send_pending_ping(cx, &mut self.codec))?;
-        ready!(self
-            .inner
-            .settings
-            .poll_send(cx, &mut self.codec, &mut self.inner.streams))?;
+        ready!(
+            self.inner
+                .settings
+                .poll_send(cx, &mut self.codec, &mut self.inner.streams)
+        )?;
         ready!(self.inner.streams.send_pending_refusal(cx, &mut self.codec))?;
 
         Poll::Ready(Ok(()))
@@ -589,8 +590,8 @@ where
     T: AsyncRead + AsyncWrite,
     B: Buf,
 {
-    pub(crate) fn streams(&self) -> &Streams<B, client::Peer> {
-        &self.inner.streams
+    pub(crate) fn injector(&self) -> super::Injector<B> {
+        self.inner.streams.injector()
     }
 }
 

src/proto/mod.rs

@@ -10,7 +10,7 @@ pub(crate) use self::connection::{Config, Connection};
 pub use self::error::{Error, Initiator};
 pub(crate) use self::peer::{Dyn as DynPeer, Peer};
 pub(crate) use self::ping_pong::UserPings;
-pub(crate) use self::streams::{DynStreams, OpaqueStreamRef, StreamRef, Streams};
+pub(crate) use self::streams::{DynStreams, Injector, OpaqueStreamRef, StreamRef, Streams};
 pub(crate) use self::streams::{Open, PollReset, Prioritized};
 
 use crate::codec::Codec;

src/proto/ping_pong.rs

@@ -5,8 +5,8 @@ use crate::proto::{self, PingPayload};
 use atomic_waker::AtomicWaker;
 use bytes::Buf;
 use std::io;
-use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use std::task::{Context, Poll};
 use tokio::io::AsyncWrite;