fix: ignore library resets at connection poll loop level

src/proto/connection.rs

@@ -442,11 +442,17 @@ where
                 Ok(())
             }
             // Attempting to read a frame resulted in a stream level error.
-            // This is handled by resetting the frame then trying to read
-            // another frame.
+            // Locally detected stream errors are reported to the peer with
+            // RST_STREAM. Remotely initiated resets have already been applied
+            // by the streams state machine and must not be echoed back.
             Err(Error::Reset(id, reason, initiator)) => {
+                if initiator == Initiator::Remote {
+                    tracing::trace!(?id, ?reason, ?initiator, "stream reset");
+                    return Ok(());
+                }
+
                 debug_assert_eq!(initiator, Initiator::Library);
-                tracing::trace!(?id, ?reason, "stream error");
+                tracing::trace!(?id, ?reason, ?initiator, "stream error");
                 match self.streams.send_reset(id, reason) {
                     Ok(()) => (),
                     Err(crate::proto::error::GoAway { debug_data, reason }) => {

tests/h2-tests/tests/server.rs

@@ -1750,3 +1750,52 @@ async fn init_window_size_smaller_than_default_should_use_default_before_ack() {
 
     join(client, h2).await;
 }
+
+#[tokio::test]
+async fn remote_reset_does_not_panic_connection_driver() {
+    h2_support::trace_init!();
+
+    const ADVERSARIAL_WIRE: &[u8] = &[
+        // Client connection preface.
+        0x50, 0x52, 0x49, 0x20, 0x2a, 0x20, 0x48, 0x54, 0x54, 0x50, 0x2f, 0x32, 0x2e, 0x30, 0x0d,
+        0x0a, 0x0d, 0x0a, 0x53, 0x4d, 0x0d, 0x0a, 0x0d, 0x0a,
+        // SETTINGS len=0, flags=0, stream=0.
+        0x00, 0x00, 0x00, 0x04, 0x00, 0x00, 0x00, 0x00, 0x00,
+        // Unknown frame type 0x87, len=5, flags=0xc1, stream=257.
+        0x00, 0x00, 0x05, 0x87, 0xc1, 0x00, 0x00, 0x01, 0x01, 0x05, 0x94, 0x05, 0x01, 0x00,
+        // Unknown frame type 0xc1, len=0, flags=0x94, stream=1281.
+        0x00, 0x00, 0x00, 0xc1, 0x94, 0x00, 0x00, 0x05, 0x01,
+        // HEADERS len=4, flags=END_STREAM | END_HEADERS, stream=4353.
+        0x00, 0x00, 0x04, 0x01, 0x05, 0x00, 0x00, 0x11, 0x01, 0x83, 0x87, 0x01, 0x00,
+        // RST_STREAM len=4, flags=0x05, stream=4353.
+        0x00, 0x00, 0x04, 0x03, 0x05, 0x00, 0x00, 0x11, 0x01, 0x83, 0x87, 0x01, 0x00,
+        // HEADERS len=5, flags=0xf6, stream=4353.
+        0x00, 0x00, 0x05, 0x01, 0xf6, 0x00, 0x00, 0x11, 0x01, 0x01, 0x94, 0x00, 0x3d, 0x01,
+        // PUSH_PROMISE len=5, flags=0xf6, stream=4353.
+        0x00, 0x00, 0x05, 0x05, 0xf6, 0x00, 0x00, 0x11, 0x01, 0x3d, 0x94, 0x81, 0x00, 0x95,
+        // HEADERS len=0, flags=END_STREAM | END_HEADERS, stream=4353.
+        0x00, 0x00, 0x00, 0x01, 0x05, 0x00, 0x00, 0x11, 0x01,
+    ];
+
+    let (mut client_io, server_io) = tokio::io::duplex(256 * 1024);
+    let server_task = tokio::spawn(async move {
+        let Ok(mut server) = server::handshake(server_io).await else {
+            return;
+        };
+
+        while let Some(result) = server.next().await {
+            let _ = result;
+        }
+    });
+
+    client_io
+        .write_all(ADVERSARIAL_WIRE)
+        .await
+        .expect("write adversarial wire");
+    drop(client_io);
+
+    tokio::time::timeout(std::time::Duration::from_secs(1), server_task)
+        .await
+        .expect("server task timed out")
+        .expect("server task panicked");
+}