Panic with "poll_response called after response returned"

[nox]

We have a panic deep in h2, I looked into it for a bit and AFAIK the panic message is wrong and the future did not get polled twice and this is instead a protocol bug where h2 popped an event which it expected to be a headers event but it got data or trailers instead.

h2/src/proto/streams/recv.rs

Lines 302 to 313 in bcaaaf6

	// If the buffer is not empty, then the first frame must be a HEADERS
	// frame or the user violated the contract.
	match stream.pending_recv.pop_front(&mut self.buffer) {
	Some(Event::Headers(Client(response))) => Poll::Ready(Ok(response)),
	Some(_) => panic!("poll_response called after response returned"),
	None => {
	stream.state.ensure_recv_open()?;
	stream.recv_task = Some(cx.waker().clone());
	Poll::Pending
	}
	}

I concluded that the panic didn't actually happen because the future was polled twice because the future is only ever polled by Hyper from behind a futures_util::future::future::Map<T>, as shown in the attached stacktrace (see below):

4   h2              h2::client::ResponseFuture::poll (client.rs:1300)
5   futures_util    [inlined] futures_util::future::future::map::Map<T>::poll (map.rs:55)
6   futures_util    futures_util::future::future::Map<T>::poll (lib.rs:102)

That mapping is done in Hyper itself:

https://github.com/hyperium/hyper/blob/237b2ce08341266b62a8d1cfa974779c511a0710/src/proto/h2/client.rs#L269-L288

In this snippet, fut in fut.map is the ResponseFuture whose poll method panicked, as you can see there is no opportunity to poll it twice, as <Map<T>>::poll would have panicked already:

https://github.com/rust-lang/futures-rs/blob/2fa7babdf92e4e6b1e874054a2773c5c330a3299/futures-util/src/future/future/map.rs#L53-L64

---

Click to expand stacktrace

Application Specific Information:
poll_response called after response returned

Thread 0 Crashed:
0   std             std::sys_common::backtrace::__rust_end_short_backtrace (backtrace.rs:141)
1   std             std::panicking::begin_panic (panicking.rs:519)
2   h2              h2::proto::streams::recv::Recv::poll_response (recv.rs:306)
3   h2              h2::proto::streams::streams::OpaqueStreamRef::poll_response (streams.rs:1237)
4   h2              h2::client::ResponseFuture::poll (client.rs:1300)
5   futures_util    [inlined] futures_util::future::future::map::Map<T>::poll (map.rs:55)
6   futures_util    futures_util::future::future::Map<T>::poll (lib.rs:102)
7   hyper           [inlined] hyper::client::dispatch::Callback<T>::send_when::{{closure}} (dispatch.rs:243)
8   futures_util    futures_util::future::poll_fn::PollFn<T>::poll (poll_fn.rs:55)
9   tokio           [inlined] tokio::runtime::task::core::CoreStage<T>::poll::{{closure}} (core.rs:235)
10  tokio           [inlined] tokio::loom::std::unsafe_cell::UnsafeCell<T>::with_mut (unsafe_cell.rs:14)
11  tokio           [inlined] tokio::runtime::task::core::CoreStage<T>::poll (core.rs:225)
12  tokio           [inlined] tokio::runtime::task::harness::poll_future::{{closure}} (harness.rs:422)
13  std             std::panic::AssertUnwindSafe<T>::call_once (panic.rs:322)
14  std             [inlined] std::panicking::try::do_call (panicking.rs:379)
15  std             [inlined] std::panicking::try (panicking.rs:343)
16  std             [inlined] std::panic::catch_unwind (panic.rs:396)
17  tokio           tokio::runtime::task::harness::poll_future (harness.rs:409)
18  tokio           [inlined] tokio::runtime::task::harness::Harness<T>::poll_inner (harness.rs:89)
19  tokio           tokio::runtime::task::harness::Harness<T>::poll (harness.rs:59)
20  tokio           [inlined] tokio::runtime::task::raw::RawTask::poll (raw.rs:66)
21  tokio           [inlined] tokio::runtime::task::Notified<T>::run (mod.rs:171)
22  tokio           [inlined] tokio::runtime::thread_pool::worker::Context::run_task::{{closure}} (worker.rs:370)
23  tokio           [inlined] tokio::coop::with_budget::{{closure}} (coop.rs:121)
24  std             [inlined] std::thread::local::LocalKey<T>::try_with (local.rs:272)
25  std             std::thread::local::LocalKey<T>::with (local.rs:248)
26  tokio           [inlined] tokio::coop::with_budget (coop.rs:114)
27  tokio           [inlined] tokio::coop::budget (coop.rs:98)
28  tokio           tokio::runtime::thread_pool::worker::Context::run_task (worker.rs:348)
29  tokio           tokio::runtime::thread_pool::worker::Context::run (worker.rs:318)
30  tokio           [inlined] tokio::runtime::thread_pool::worker::run::{{closure}} (worker.rs:303)
31  tokio           tokio::macros::scoped_tls::ScopedKey<T>::set (scoped_tls.rs:61)
32  tokio           tokio::runtime::thread_pool::worker::run (worker.rs:300)
33  tokio           [inlined] tokio::runtime::thread_pool::worker::Launch::launch::{{closure}} (worker.rs:279)
34  tokio           [inlined] tokio::runtime::blocking::task::BlockingTask<T>::poll (task.rs:42)
35  tokio           [inlined] tokio::runtime::task::core::CoreStage<T>::poll::{{closure}} (core.rs:235)
36  tokio           tokio::loom::std::unsafe_cell::UnsafeCell<T>::with_mut (unsafe_cell.rs:14)
37  tokio           [inlined] tokio::runtime::task::core::CoreStage<T>::poll (core.rs:225)
38  tokio           [inlined] tokio::runtime::task::harness::poll_future::{{closure}} (harness.rs:422)
39  std             [inlined] std::panic::AssertUnwindSafe<T>::call_once (panic.rs:322)
40  std             [inlined] std::panicking::try::do_call (panicking.rs:379)
41  std             std::panicking::try (panicking.rs:343)
42  std             [inlined] std::panic::catch_unwind (panic.rs:396)
43  tokio           [inlined] tokio::runtime::task::harness::poll_future (harness.rs:409)
44  tokio           [inlined] tokio::runtime::task::harness::Harness<T>::poll_inner (harness.rs:89)
45  tokio           tokio::runtime::task::harness::Harness<T>::poll (harness.rs:59)
46  tokio           [inlined] tokio::runtime::task::raw::RawTask::poll (raw.rs:66)
47  tokio           [inlined] tokio::runtime::task::Notified<T>::run (mod.rs:171)
48  tokio           [inlined] tokio::runtime::blocking::pool::Inner::run (pool.rs:278)
49  tokio           [inlined] tokio::runtime::blocking::pool::Spawner::spawn_thread::{{closure}} (pool.rs:258)
50  std             std::sys_common::backtrace::__rust_begin_short_backtrace (backtrace.rs:125)
51  std             [inlined] std::thread::Builder::spawn_unchecked::{{closure}}::{{closure}} (mod.rs:474)
52  std             [inlined] std::panic::AssertUnwindSafe<T>::call_once (panic.rs:322)
53  std             [inlined] std::panicking::try::do_call (panicking.rs:379)
54  std             [inlined] std::panicking::try (panicking.rs:343)
55  std             [inlined] std::panic::catch_unwind (panic.rs:396)
56  std             [inlined] std::thread::Builder::spawn_unchecked::{{closure}} (mod.rs:473)
57  core            core::ops::function::FnOnce::call_once{{vtable.shim}} (function.rs:227)
58  alloc           [inlined] alloc::boxed::Box<T>::call_once (boxed.rs:1328)
59  alloc           [inlined] alloc::boxed::Box<T>::call_once (boxed.rs:1328)
60  std             std::sys::unix::thread::Thread::new::thread_start (thread.rs:71)

[nox]

It is a bit unfortunate that there is no way to see if the event that caused the panic was an Event::Headers(Server(_)) or something completely different.

[nox]

In #521 h2 started ignoring informational frames.

h2/src/proto/streams/recv.rs

Lines 229 to 239 in bcaaaf6

	if !pseudo.is_informational() {
	let message = counts
	.peer()
	.convert_poll_message(pseudo, fields, stream_id)?;
	// Push the frame onto the stream's recv buffer
	stream
	.pending_recv
	.push_back(&mut self.buffer, Event::Headers(message));
	stream.notify_recv();
	}

AFAIK this code can be reached from a stream whose state was Idle, which means that we can move a stream's state from Idle to Open { local: AwaitingHeaders, remote: Streaming } without pushing the Event::Headers that caused the state change.

h2/src/proto/streams/recv.rs

Line 164 in bcaaaf6

let is_initial = stream.state.recv_open(&frame)?;

h2/src/proto/streams/state.rs

Lines 144 to 155 in 89d91b0

	Idle => {
	initial = true;
	if eos {
	HalfClosedRemote(AwaitingHeaders)
	} else {
	Open {
	local: AwaitingHeaders,
	remote,
	}
	}
	}

I'm wondering if the stream can be then passed to recv_data with its state as Open { remote: Streaming, .. } and push an Event::Data even though we still did not push Event::Headers.

This wouldn't trigger:

h2/src/proto/streams/recv.rs

Lines 542 to 552 in bcaaaf6

	let is_ignoring_frame = stream.state.is_local_reset();
	if !is_ignoring_frame && !stream.state.is_recv_streaming() {
	// TODO: There are cases where this can be a stream error of
	// STREAM_CLOSED instead...
	// Receiving a DATA frame when not expecting one is a protocol
	// error.
	proto_err!(conn: "unexpected DATA frame; stream={:?}", stream.id);
	return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
	}

Because is_recv_streaming returns true:

h2/src/proto/streams/state.rs

Lines 385 to 393 in 89d91b0

	pub fn is_recv_streaming(&self) -> bool {
	match self.inner {
	Open {
	remote: Streaming, ..
	} => true,
	HalfClosedLocal(Streaming) => true,
	_ => false,
	}
	}

So we end up pushing an Event::Data, which AFAIU would be the first event pushed ever for that stream, making ResponseFuture::poll panic:

h2/src/proto/streams/recv.rs

Lines 627 to 629 in bcaaaf6

	// Push the frame onto the recv buffer
	stream.pending_recv.push_back(&mut self.buffer, event);
	stream.notify_recv();

[nox]

Fixed in #527.