Fix wrong HTTP/1 timeout being used in some cases

Added many tests to ensure the right timeout is picked in
the appropriate situation. Should there be any issues
remaining we can add more tests.

2 files changed, 187 insertions, 8 deletions

diff --git a/src/cowboy_http.erl b/src/cowboy_http.erl
index 33ad7af..7aa799f 100644
--- a/src/cowboy_http.erl
+++ b/src/cowboy_http.erl
@@ -266,9 +266,24 @@ loop(State=#state{parent=Parent, socket=Socket, transport=Transport, opts=Opts,
 		terminate(State, {internal_error, timeout, 'No message or data received before timeout.'})
 	end.
 
-%% We do not set request_timeout if there are active streams.
-set_timeout(State=#state{streams=[_|_]}, request_timeout) ->
-	State;
+%% For HTTP/1.1 we have two types of timeouts: the request_timeout
+%% is used when there is no currently ongoing request. This means
+%% that we are not currently sending or receiving data and that
+%% the next data to be received will be a new request. The
+%% request_timeout is set once when we no longer have ongoing
+%% requests, and runs until the full set of request headers
+%% is received. It is not reset.
+%%
+%% After that point we use the idle_timeout. We continue using
+%% the idle_timeout if pipelined requests come in: we are doing
+%% work and just want to ensure the socket is not half-closed.
+%% We continue using the idle_timeout up until there is no
+%% ongoing request. This includes requests that were processed
+%% and for which we only want to skip the body. Once the body
+%% has been read fully we can go back to request_timeout. The
+%% idle_timeout is reset every time we receive data and,
+%% optionally, every time we send data.
+
 %% We do not set request_timeout if we are skipping a body.
 set_timeout(State=#state{in_state=#ps_body{}}, request_timeout) ->
 	State;
@@ -392,10 +407,7 @@ after_parse({data, StreamID, IsFin, Data, State0=#state{opts=Opts, buffer=Buffer
 		{Commands, StreamState} ->
 			Streams = lists:keyreplace(StreamID, #stream.id, Streams0,
 				Stream#stream{state=StreamState}),
-			State1 = set_timeout(State0, case IsFin of
-				fin -> request_timeout;
-				nofin -> idle_timeout
-			end),
+			State1 = set_timeout(State0, idle_timeout),
 			State = update_flow(IsFin, Data, State1#state{streams=Streams}),
 			parse(Buffer, commands(State, StreamID, Commands))
 	catch Class:Exception:Stacktrace ->
@@ -1357,7 +1369,10 @@ stream_next(State0=#state{opts=Opts, active=Active, out_streamid=OutStreamID, st
 	NextOutStreamID = OutStreamID + 1,
 	case lists:keyfind(NextOutStreamID, #stream.id, Streams) of
 		false ->
-			State0#state{out_streamid=NextOutStreamID, out_state=wait};
+			State = State0#state{out_streamid=NextOutStreamID, out_state=wait},
+			%% There are no streams remaining. We therefore can
+			%% and want to switch back to the request_timeout.
+			set_timeout(State, request_timeout);
 		#stream{queue=Commands} ->
 			State = case Active of
 				true -> State0;
diff --git a/test/http_SUITE.erl b/test/http_SUITE.erl
index 8b8473f..3088383 100644
--- a/test/http_SUITE.erl
+++ b/test/http_SUITE.erl
@@ -227,6 +227,68 @@ http10_keepalive_false(Config) ->
 		cowboy:stop_listener(?FUNCTION_NAME)
 	end.
 
+idle_timeout_read_body(Config) ->
+	doc("Ensure the idle_timeout drops connections when the "
+		"connection is idle too long reading the request body."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 60000,
+		idle_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
+		{ok, http} = gun:await_up(ConnPid),
+		_StreamRef = gun:post(ConnPid, "/echo/read_body",
+			#{<<"content-length">> => <<"12">>}),
+		{error, {down, {shutdown, closed}}} = gun:await(ConnPid, undefined, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
+idle_timeout_read_body_pipeline(Config) ->
+	doc("Ensure the idle_timeout drops connections when the "
+		"connection is idle too long reading the request body."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 60000,
+		idle_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
+		{ok, http} = gun:await_up(ConnPid),
+		StreamRef1 = gun:get(ConnPid, "/"),
+		StreamRef2 = gun:get(ConnPid, "/"),
+		_StreamRef3 = gun:post(ConnPid, "/echo/read_body",
+			#{<<"content-length">> => <<"12">>}),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef1),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef2),
+		{error, {down, {shutdown, closed}}} = gun:await(ConnPid, undefined, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
+idle_timeout_skip_body(Config) ->
+	doc("Ensure the idle_timeout drops connections when the "
+		"connection is idle too long skipping the request body."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 60000,
+		idle_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
+		{ok, http} = gun:await_up(ConnPid),
+		StreamRef = gun:post(ConnPid, "/",
+			#{<<"content-length">> => <<"12">>}),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef),
+		{error, {down, {shutdown, closed}}} = gun:await(ConnPid, undefined, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
 idle_timeout_infinity(Config) ->
 	doc("Ensure the idle_timeout option accepts the infinity value."),
 	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
@@ -352,6 +414,108 @@ do_persistent_term_router(Config) ->
 		cowboy:stop_listener(?FUNCTION_NAME)
 	end.
 
+request_timeout(Config) ->
+	doc("Ensure the request_timeout drops connections when requests "
+		"fail to come in fast enough."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
+		{ok, http} = gun:await_up(ConnPid),
+		{error, {down, {shutdown, closed}}} = gun:await(ConnPid, undefined, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
+request_timeout_pipeline(Config) ->
+	doc("Ensure the request_timeout drops connections when requests "
+		"fail to come in fast enough after pipelined requests went through."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		ConnPid = gun_open([{type, tcp}, {protocol, http}, {port, Port}|Config]),
+		{ok, http} = gun:await_up(ConnPid),
+		StreamRef1 = gun:get(ConnPid, "/"),
+		StreamRef2 = gun:get(ConnPid, "/"),
+		StreamRef3 = gun:get(ConnPid, "/"),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef1),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef2),
+		{response, nofin, 200, _} = gun:await(ConnPid, StreamRef3),
+		{error, {down, {shutdown, closed}}} = gun:await(ConnPid, undefined, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
+request_timeout_skip_body(Config) ->
+	doc("Ensure the request_timeout drops connections when requests "
+		"fail to come in fast enough after skipping a request body."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		Client = raw_open([{type, tcp}, {port, Port}, {opts, []}|Config]),
+		ok = raw_send(Client, <<
+			"POST / HTTP/1.1\r\n"
+			"host: localhost\r\n"
+			"content-length: 12\r\n\r\n"
+		>>),
+		Data = raw_recv_head(Client),
+		{'HTTP/1.1', 200, _, Rest0} = cow_http:parse_status_line(Data),
+		{Headers, Rest} = cow_http:parse_headers(Rest0),
+		{_, Len} = lists:keyfind(<<"content-length">>, 1, Headers),
+		<<"Hello world!">> = raw_recv_rest(Client, binary_to_integer(Len), Rest),
+		%% We then send the request data that should be skipped by Cowboy.
+		timer:sleep(100),
+		raw_send(Client, <<"Hello world!">>),
+		%% Connection should be closed by the request_timeout after that.
+		{error, closed} = raw_recv(Client, 1, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
+request_timeout_skip_body_more(Config) ->
+	doc("Ensure the request_timeout drops connections when requests "
+		"fail to come in fast enough after skipping a request body."),
+	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{
+		env => #{dispatch => init_dispatch(Config)},
+		request_timeout => 500
+	}),
+	Port = ranch:get_port(?FUNCTION_NAME),
+	try
+		Client = raw_open([{type, tcp}, {port, Port}, {opts, []}|Config]),
+		ok = raw_send(Client, <<
+			"POST / HTTP/1.1\r\n"
+			"host: localhost\r\n"
+			"content-length: 12\r\n\r\n"
+		>>),
+		Data = raw_recv_head(Client),
+		{'HTTP/1.1', 200, _, Rest0} = cow_http:parse_status_line(Data),
+		{Headers, Rest} = cow_http:parse_headers(Rest0),
+		{_, Len} = lists:keyfind(<<"content-length">>, 1, Headers),
+		<<"Hello world!">> = raw_recv_rest(Client, binary_to_integer(Len), Rest),
+		%% We then send the request data that should be skipped by Cowboy.
+		timer:sleep(100),
+		raw_send(Client, <<"Hello world!">>),
+		%% Send the start of another request.
+		ok = raw_send(Client, <<
+			"GET / HTTP/1.1\r\n"
+			"host: localhost\r\n"
+			%% Missing final \r\n on purpose.
+		>>),
+		%% Connection should be closed by the request_timeout after that.
+		{error, closed} = raw_recv(Client, 1, 1000)
+	after
+		cowboy:stop_listener(?FUNCTION_NAME)
+	end.
+
 request_timeout_infinity(Config) ->
 	doc("Ensure the request_timeout option accepts the infinity value."),
 	{ok, _} = cowboy:start_clear(?FUNCTION_NAME, [{port, 0}], #{