fix: Missed wakeup in poll_trailers

poll_trailers returned Pending without storing a waker when a
non-trailer event was at the front of pending_recv.

Author: ArniDagur

src/proto/streams/recv.rs

@@ -1230,7 +1230,7 @@ impl Recv {
             Some(event) => {
                 // Frame is not trailers.. not ready to poll trailers yet.
                 stream.pending_recv.push_front(&mut self.buffer, event);
-
+                stream.recv_task = Some(cx.waker().clone());
                 Poll::Pending
             }
             None => self.schedule_recv(cx, stream),

tests/h2-tests/tests/trailers.rs

@@ -1,5 +1,8 @@
 use futures::StreamExt;
 use h2_support::prelude::*;
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+use std::task::{Context, Poll, Wake, Waker};
 
 #[tokio::test]
 async fn recv_trailers_only() {
@@ -110,6 +113,87 @@ fn recv_trailers_without_eos() {
     // This should be a protocol error?
 }
 
+#[tokio::test]
+async fn poll_trailers_before_data_is_consumed() {
+    h2_support::trace_init!();
+    let (io, mut srv) = mock::new();
+
+    let srv = async move {
+        let settings = srv.assert_client_handshake().await;
+        assert_default_settings!(settings);
+
+        // 2. Receive the request.
+        srv.recv_frame(
+            frames::headers(1)
+                .request("GET", "https://example.com/")
+                .eos(),
+        )
+        .await;
+
+        // 3. Send response HEADERS followed by DATA and trailers.
+        srv.send_frame(frames::headers(1).response(200)).await;
+        srv.send_frame(frames::data(1, "hello")).await;
+        srv.send_frame(frames::headers(1).field("trailer-key", "trailer-val").eos())
+            .await;
+
+        // 4. Ensure all preceding frames have been processed by the client.
+        srv.ping_pong([1; 8]).await;
+    };
+
+    let client = async move {
+        let (mut client, conn) = client::handshake(io).await.expect("handshake");
+
+        let req = async move {
+            // 1. Send the request and wait for response HEADERS.
+            let resp = client.get("https://example.com/").await.expect("response");
+            assert_eq!(resp.status(), StatusCode::OK);
+
+            let mut body = resp.into_body();
+
+            struct FlagWake(AtomicBool);
+            impl Wake for FlagWake {
+                fn wake(self: Arc<Self>) {
+                    self.0.store(true, Ordering::SeqCst);
+                }
+            }
+
+            let flag = Arc::new(FlagWake(AtomicBool::new(false)));
+            let waker = Waker::from(flag.clone());
+            let mut cx = Context::from_waker(&waker);
+
+            // 5. Poll trailers while DATA is at the front of pending_recv.
+            // This returns Pending and must register the trailers waker.
+            assert!(
+                matches!(body.poll_trailers(&mut cx), Poll::Pending),
+                "poll_trailers should be Pending when DATA is buffered"
+            );
+
+            // 6. Consume the DATA frame.
+            let data = body.data().await.unwrap().unwrap();
+            assert_eq!(data, "hello");
+
+            // 7. poll_data reaches the queued trailers and calls notify_recv().
+            // The waker registered in 5 must be woken.
+            assert!(body.data().await.is_none());
+            assert!(
+                flag.0.load(Ordering::SeqCst),
+                "poll_trailers waker must be notified when DATA is consumed"
+            );
+
+            // 8. Poll trailers again and consume them.
+            let trailers = body.trailers().await.expect("trailers result");
+            let trailers = trailers.expect("should have trailers");
+            assert_eq!(trailers["trailer-key"], "trailer-val");
+        };
+
+        let mut conn = Box::pin(async move { conn.await.expect("client") });
+        conn.drive(req).await;
+        conn.await;
+    };
+
+    join(srv, client).await;
+}
+
 #[tokio::test]
 async fn send_trailers_rejects_connection_specific_headers() {
     // RFC 9113 §8.2.2: endpoints MUST NOT *generate* an HTTP/2 message containing