Propagate timeouts to the right layer in HTTP/2 tunnels

This temporarily depends on Cowlib master.

5 files changed, 60 insertions, 37 deletions

diff --git a/Makefile b/Makefile
index a09039d..a988624 100644
--- a/Makefile
+++ b/Makefile
@@ -14,7 +14,7 @@ CT_OPTS += -ct_hooks gun_ct_hook [] # -boot start_sasl
 LOCAL_DEPS = ssl
 
 DEPS = cowlib
-dep_cowlib = git https://github.com/ninenines/cowlib 2.9.0
+dep_cowlib = git https://github.com/ninenines/cowlib master
 
 DOC_DEPS = asciideck
 
diff --git a/src/gun_http2.erl b/src/gun_http2.erl
index e2aeb4a..1072aca 100644
--- a/src/gun_http2.erl
+++ b/src/gun_http2.erl
@@ -173,10 +173,10 @@ init(_ReplyTo, Socket, Transport, Opts0) ->
 		initial_connection_window_size => maps:get(initial_connection_window_size, Opts0, 8000000),
 		initial_stream_window_size => maps:get(initial_stream_window_size, Opts0, 8000000)
 	},
-	{ok, Preface, HTTP2Machine} = cow_http2_machine:init(client, Opts),
 	Handlers = maps:get(content_handlers, Opts, [gun_data_h]),
 	BaseStreamRef = maps:get(stream_ref, Opts, undefined),
 	TunnelTransport = maps:get(tunnel_transport, Opts, undefined),
+	{ok, Preface, HTTP2Machine} = cow_http2_machine:init(client, Opts#{message_tag => BaseStreamRef}),
 	%% @todo Better validate the preface being received.
 	State = #http2_state{socket=Socket, transport=Transport, opts=Opts,
 		base_stream_ref=BaseStreamRef, tunnel_transport=TunnelTransport,
@@ -1046,13 +1046,31 @@ cancel(State=#http2_state{socket=Socket, transport=Transport, http2_machine=HTTP
 				EvHandlerState0}
 	end.
 
-%% @todo What about tunnels?
-timeout(State=#http2_state{http2_machine=HTTP2Machine0}, {cow_http2_machine, Name}, TRef) ->
+timeout(State=#http2_state{http2_machine=HTTP2Machine0}, {cow_http2_machine, undefined, Name}, TRef) ->
 	case cow_http2_machine:timeout(Name, TRef, HTTP2Machine0) of
 		{ok, HTTP2Machine} ->
 			{state, State#http2_state{http2_machine=HTTP2Machine}};
 		{error, Error={connection_error, _, _}, _HTTP2Machine} ->
 			connection_error(State, Error)
+	end;
+%% Timeouts occurring in tunnels.
+timeout(State, {cow_http2_machine, RealStreamRef, Name}, TRef) ->
+	{StreamRef, SubStreamRef} = if
+		is_reference(RealStreamRef) -> {RealStreamRef, undefined};
+		true -> {hd(RealStreamRef), tl(RealStreamRef)}
+	end,
+	case get_stream_by_ref(State, StreamRef) of
+		Stream=#stream{id=StreamID, tunnel=Tunnel=#tunnel{protocol=Proto, protocol_state=ProtoState0}} ->
+			case Proto:timeout(ProtoState0, {cow_http2_machine, SubStreamRef, Name}, TRef) of
+				{state, ProtoState} ->
+					{state, store_stream(State, Stream#stream{
+						tunnel=Tunnel#tunnel{protocol_state=ProtoState}})};
+				{error, {connection_error, Reason, Human}} ->
+					{state, reset_stream(State, StreamID, {stream_error, Reason, Human})}
+			end;
+		%% We ignore timeout events for streams that no longer exist.
+		error ->
+			{state, State}
 	end.
 
 stream_info(State, StreamRef) when is_reference(StreamRef) ->
diff --git a/src/gun_tunnel.erl b/src/gun_tunnel.erl
index f7d0232..ca9d3aa 100644
--- a/src/gun_tunnel.erl
+++ b/src/gun_tunnel.erl
@@ -324,10 +324,13 @@ cancel(State=#tunnel_state{protocol=Proto, protocol_state=ProtoState},
 	{Commands, EvHandlerState} = Proto:cancel(ProtoState, StreamRef, ReplyTo, EvHandler, EvHandlerState0),
 	{{state, commands(Commands, State)}, EvHandlerState}.
 
-timeout(_State, {cow_http2_machine, _Name}, _TRef) ->
-	%% @todo We currently have no way of routing timeout events to the right layer.
-	%% We will need to update Cowlib to include routing information in the timeout message.
-	[].
+timeout(State=#tunnel_state{protocol=Proto, protocol_state=ProtoState0}, Msg, TRef) ->
+	case Proto:timeout(ProtoState0, Msg, TRef) of
+		{state, ProtoState} ->
+			{state, State#tunnel_state{protocol_state=ProtoState}};
+		Other ->
+			Other
+	end.
 
 stream_info(#tunnel_state{transport=Transport0, stream_ref=TunnelStreamRef, reply_to=ReplyTo,
 		tunnel_protocol=TunnelProtocol,
diff --git a/test/rfc7540_SUITE.erl b/test/rfc7540_SUITE.erl
index 9cb4ca2..a66c5ae 100644
--- a/test/rfc7540_SUITE.erl
+++ b/test/rfc7540_SUITE.erl
@@ -139,6 +139,9 @@ do_proxy_parse(<<Len:24, 1:8, _:8, StreamID:32, ReqHeadersBlock:Len/binary, Rest
 		Stream#proxy_stream{origin_socket=OriginSocket}),
 	do_proxy_parse(Rest, Proxy#proxy{streams=Streams,
 		decode_state=DecodeState, encode_state=EncodeState});
+%% An RST_STREAM was received. Stop the proxy.
+do_proxy_parse(<<_:24, 3:8, _/bits>>, _) ->
+	ok;
 do_proxy_parse(<<Len:24, Header:6/binary, Payload:Len/binary, Rest/bits>>, Proxy) ->
 	ct:pal("Ignoring packet header ~0p~npayload ~p", [Header, Payload]),
 	do_proxy_parse(Rest, Proxy);
@@ -557,15 +560,6 @@ do_connect_http(OriginScheme, OriginTransport, OriginProtocol, ProxyScheme, Prox
 	gun:close(ConnPid).
 
 connect_cowboy_http_via_h2c(_) ->
-
-%dbg:tracer(),
-%dbg:tpl(gun_http, []),
-%dbg:tpl(gun_http2, []),
-%dbg:tpl(gun_tunnel, []),
-%dbg:tpl(gun_tcp_proxy, []),
-%dbg:tpl(gun, []),
-%dbg:p(all, c),
-
 	doc("CONNECT can be used to establish a TCP connection "
 		"to an HTTP/1.1 server via a TCP HTTP/2 proxy. (RFC7540 8.3)"),
 	do_connect_cowboy(<<"http">>, tcp, http, <<"http">>, tcp).
@@ -659,6 +653,34 @@ do_cowboy_origin(OriginTransport, OriginProtocol) ->
 	end,
 	{ok, Ref, Port}.
 
+connect_handshake_timeout(_) ->
+	doc("HTTP/2 timeouts are properly routed to the appropriate "
+		"tunnel layer. (RFC7540 3.5, RFC7540 8.3)"),
+	{ok, _, OriginPort} = init_origin(tcp, raw, fun(_, _, _) ->
+		timer:sleep(5000)
+	end),
+	{ok, ProxyPid, ProxyPort} = do_proxy_start(tcp, [
+		#proxy_stream{id=1, status=200}
+	]),
+	{ok, ConnPid} = gun:open("localhost", ProxyPort, #{
+		protocols => [http2]
+	}),
+	{ok, http2} = gun:await_up(ConnPid),
+	handshake_completed = receive_from(ProxyPid),
+	StreamRef = gun:connect(ConnPid, #{
+		host => "localhost",
+		port => OriginPort,
+		protocols => [{http2, #{preface_timeout => 500}}]
+	}),
+	{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
+	{up, http2} = gun:await(ConnPid, StreamRef),
+	%% @todo The error should be normalized.
+	%% @todo Do we want to indicate that a connection_error occurred within the tunnel stream?
+	{error, {stream_error, {stream_error, protocol_error,
+		'The preface was not received in a reasonable amount of time.'}}}
+		= gun:await(ConnPid, StreamRef),
+	gun:close(ConnPid).
+
 connect_http_via_http_via_h2c(_) ->
 	doc("CONNECT can be used to establish a TCP connection "
 		"to an HTTP/1.1 server via a tunnel going through both "
@@ -666,14 +688,6 @@ connect_http_via_http_via_h2c(_) ->
 	do_connect_via_multiple_proxies(tcp, http, tcp, http, tcp).
 
 connect_https_via_https_via_h2(_) ->
-
-%dbg:tracer(),
-%dbg:tpl(?MODULE, []),
-%dbg:tpl(gun, []),
-%dbg:tpl(gun_http, []),
-%dbg:tpl(gun_http2, []),
-%dbg:p(all, c),
-
 	doc("CONNECT can be used to establish a TLS connection "
 		"to an HTTP/1.1 server via a tunnel going through both "
 		"a TLS HTTP/2 and a TLS HTTP/1.1 proxy. (RFC7540 8.3)"),
diff --git a/test/tunnel_SUITE.erl b/test/tunnel_SUITE.erl
index 190eb17..6cc50f8 100644
--- a/test/tunnel_SUITE.erl
+++ b/test/tunnel_SUITE.erl
@@ -774,18 +774,6 @@ socks5tls_socks5tls_rawtls(_) ->
 }).
 
 do_tunnel(FunctionName) ->
-
-%dbg:tracer(),
-%dbg:tpl(gun, []),
-%dbg:tpl(gun_http, []),
-%dbg:tpl(gun_http2, []),
-%dbg:tpl(gun_tunnel, []),
-%dbg:p(all, c),
-
-%dbg:tracer(),
-%dbg:tpl(?MODULE, []),
-%dbg:p(all, c),
-
 	[Proxy1, Proxy2, Origin] = [list_to_atom(Lex) || Lex <- string:lexemes(atom_to_list(FunctionName), "_")],
 	do_doc(Proxy1, Proxy2, Origin),
 	{ok, OriginPid, OriginPort} = do_origin_start(Origin),