Ensure the behavior on stream handler crash is consistent

Also corrects the lack of error response when HTTP/1.1 is used.

4 files changed, 219 insertions, 27 deletions

diff --git a/src/cowboy_http.erl b/src/cowboy_http.erl
index a7c9b4d..0be949e 100644
--- a/src/cowboy_http.erl
+++ b/src/cowboy_http.erl
@@ -252,10 +252,9 @@ after_parse({request, Req=#{streamid := StreamID, headers := Headers, version :=
 		cowboy_stream:report_error(init,
 			[StreamID, Req, Opts],
 			Class, Exception, erlang:get_stacktrace()),
-		ok %% @todo send a proper response, etc. note that terminate must NOT be called
-		%% @todo Status code.
-%		stream_reset(State, StreamID, {internal_error, {Class, Reason},
-%			'Exception occurred in StreamHandler:init/10 call.'}) %% @todo Check final arity.
+		early_error(500, State0, {internal_error, {Class, Exception},
+			'Unhandled exception in cowboy_stream:init/3.'}, Req),
+		parse(Buffer, State0)
 	end;
 %% Streams are sequential so the body is always about the last stream created
 %% unless that stream has terminated.
@@ -270,11 +269,8 @@ after_parse({data, StreamID, IsFin, Data, State=#state{
 		cowboy_stream:report_error(data,
 			[StreamID, IsFin, Data, StreamState0],
 			Class, Exception, erlang:get_stacktrace()),
-		%% @todo Bad value returned here. Crashes.
-		ok
-		%% @todo
-%		stream_reset(State, StreamID, {internal_error, {Class, Reason},
-%			'Exception occurred in StreamHandler:data/4 call.'})
+		stream_reset(State, StreamID, {internal_error, {Class, Exception},
+			'Unhandled exception in cowboy_stream:data/4.'})
 	end;
 %% No corresponding stream, skip.
 after_parse({data, _, _, _, State, Buffer}) ->
@@ -733,10 +729,8 @@ info(State=#state{streams=Streams0}, StreamID, Msg) ->
 				cowboy_stream:report_error(info,
 					[StreamID, Msg, StreamState0],
 					Class, Exception, erlang:get_stacktrace()),
-				ok
-%% @todo
-%				stream_reset(State, StreamID, {internal_error, {Class, Reason},
-%					'Exception occurred in StreamHandler:info/3 call.'})
+				stream_reset(State, StreamID, {internal_error, {Class, Exception},
+					'Unhandled exception in cowboy_stream:info/3.'})
 			end;
 		false ->
 			error_logger:error_msg("Received message ~p for unknown stream ~p.~n", [Msg, StreamID]),
@@ -935,6 +929,8 @@ stream_terminate(State0=#state{socket=Socket, transport=Transport,
 	{value, #stream{state=StreamState, version=Version}, Streams}
 		= lists:keytake(StreamID, #stream.id, Streams0),
 	State1 = case OutState of
+		wait when element(1, Reason) =:= internal_error ->
+			info(State0, StreamID, {response, 500, #{<<"content-length">> => <<"0">>}, <<>>});
 		wait ->
 			info(State0, StreamID, {response, 204, #{}, <<>>});
 		chunked when Version =:= 'HTTP/1.1' ->
@@ -1020,8 +1016,7 @@ connection_hd_is_close(Conn) ->
 
 %% This function is only called when an error occurs on a new stream.
 -spec error_terminate(cowboy:http_status(), #state{}, _) -> no_return().
-error_terminate(StatusCode0, State=#state{ref=Ref, socket=Socket, transport=Transport,
-		opts=Opts, peer=Peer, in_streamid=StreamID, in_state=StreamState}, Reason) ->
+error_terminate(StatusCode, State=#state{ref=Ref, peer=Peer, in_state=StreamState}, Reason) ->
 	PartialReq = case StreamState of
 		#ps_request_line{} ->
 			#{};
@@ -1039,7 +1034,12 @@ error_terminate(StatusCode0, State=#state{ref=Ref, socket=Socket, transport=Tran
 			end
 		}
 	end,
-	Resp = {response, StatusCode0, #{<<"content-length">> => <<"0">>}, <<>>},
+	early_error(StatusCode, State, Reason, PartialReq),
+	terminate(State, Reason).
+
+early_error(StatusCode0, #state{socket=Socket, transport=Transport,
+		opts=Opts, in_streamid=StreamID}, Reason, PartialReq) ->
+	Resp = {response, StatusCode0, RespHeaders0=#{<<"content-length">> => <<"0">>}, <<>>},
 	try cowboy_stream:early_error(StreamID, Reason, PartialReq, Resp, Opts) of
 		{response, StatusCode, RespHeaders, RespBody} ->
 			Transport:send(Socket, [
@@ -1049,9 +1049,13 @@ error_terminate(StatusCode0, State=#state{ref=Ref, socket=Socket, transport=Tran
 	catch Class:Exception ->
 		cowboy_stream:report_error(early_error,
 			[StreamID, Reason, PartialReq, Resp, Opts],
-			Class, Exception, erlang:get_stacktrace())
+			Class, Exception, erlang:get_stacktrace()),
+		%% We still need to send an error response, so send what we initially
+		%% wanted to send. It's better than nothing.
+		Transport:send(Socket, cow_http:response(StatusCode0,
+			'HTTP/1.1', maps:to_list(RespHeaders0)))
 	end,
-	terminate(State, Reason).
+	ok.
 
 -spec terminate(_, _) -> no_return().
 terminate(undefined, Reason) ->
diff --git a/src/cowboy_http2.erl b/src/cowboy_http2.erl
index 8c6a0d3..dbd67cb 100644
--- a/src/cowboy_http2.erl
+++ b/src/cowboy_http2.erl
@@ -305,7 +305,7 @@ frame(State0=#state{remote_window=ConnWindow, streams=Streams},
 					[StreamID, IsFin, Data, StreamState0],
 					Class, Exception, erlang:get_stacktrace()),
 				stream_reset(State, StreamID, {internal_error, {Class, Exception},
-					'Exception occurred in cowboy_stream:data/4.'})
+					'Unhandled exception in cowboy_stream:data/4.'})
 			end;
 		#stream{remote=fin} ->
 			stream_reset(State, StreamID, {stream_error, stream_closed,
@@ -444,7 +444,7 @@ info(State=#state{streams=Streams}, StreamID, Msg) ->
 					[StreamID, Msg, StreamState0],
 					Class, Exception, erlang:get_stacktrace()),
 				stream_reset(State, StreamID, {internal_error, {Class, Exception},
-					'Exception occurred in cowboy_stream:info/3.'})
+					'Unhandled exception in cowboy_stream:info/3.'})
 			end;
 		false ->
 			error_logger:error_msg("Received message ~p for unknown stream ~p.", [Msg, StreamID]),
@@ -759,7 +759,7 @@ stream_handler_init(State=#state{opts=Opts,
 			[StreamID, Req, Opts],
 			Class, Exception, erlang:get_stacktrace()),
 		stream_reset(State, StreamID, {internal_error, {Class, Exception},
-			'Exception occurred in cowboy_stream:init/3.'})
+			'Unhandled exception in cowboy_stream:init/3.'})
 	end.
 
 %% @todo We might need to keep track of which stream has been reset so we don't send lots of them.
@@ -801,9 +801,11 @@ stream_terminate(State=#state{socket=Socket, transport=Transport,
 			stream_call_terminate(StreamID, Reason, StreamState),
 			Children = cowboy_children:shutdown(Children0, StreamID),
 			State#state{streams=Streams, children=Children};
+		%% The stream doesn't exist. This can occur for various reasons.
+		%% It can happen before the stream has been created, or because
+		%% the cowboy_stream:init call failed, in which case doing nothing
+		%% is correct.
 		false ->
-			%% @todo Unknown stream. Not sure what to do here. Check again once all
-			%% terminate calls have been written.
 			State
 	end.
 
diff --git a/test/handlers/stream_handler_h.erl b/test/handlers/stream_handler_h.erl
index 3ee4932..74fc478 100644
--- a/test/handlers/stream_handler_h.erl
+++ b/test/handlers/stream_handler_h.erl
@@ -21,6 +21,16 @@ init(StreamID, Req, Opts) ->
 	Pid ! {Pid, self(), init, StreamID, Req, Opts},
 	{init_commands(StreamID, Req, State), State}.
 
+init_commands(_, _, #state{test=crash_in_init}) ->
+	error(crash);
+init_commands(_, _, #state{test=crash_in_data}) ->
+	[];
+init_commands(_, _, #state{test=crash_in_info}) ->
+	[];
+init_commands(_, _, #state{test=crash_in_terminate}) ->
+	[{response, 200, #{<<"content-length">> => <<"12">>}, <<"Hello world!">>}, stop];
+init_commands(_, _, #state{test=crash_in_early_error}) ->
+	error(crash);
 init_commands(_, _, State=#state{test=shutdown_on_stream_stop}) ->
 	Spawn = init_process(false, State),
 	[{headers, 200, #{}}, {spawn, Spawn, 5000}, stop];
@@ -50,20 +60,32 @@ init_process(TrapExit, #state{pid=Pid}) ->
 	receive {Spawn, ready} -> ok after 1000 -> error(timeout) end,
 	Spawn.
 
+data(_, _, _, #state{test=crash_in_data}) ->
+	error(crash);
 data(StreamID, IsFin, Data, State=#state{pid=Pid}) ->
 	Pid ! {Pid, self(), data, StreamID, IsFin, Data, State},
 	{[], State}.
 
+info(_, Resp={response, _, _, _}, State) ->
+	{[Resp], State};
+info(_, crash, #state{test=crash_in_info}) ->
+	error(crash);
 info(StreamID, Info, State=#state{pid=Pid}) ->
 	Pid ! {Pid, self(), info, StreamID, Info, State},
 	{[], State}.
 
+terminate(StreamID, Reason, State=#state{pid=Pid, test=crash_in_terminate}) ->
+	Pid ! {Pid, self(), terminate, StreamID, Reason, State},
+	error(crash);
 terminate(StreamID, Reason, State=#state{pid=Pid}) ->
 	Pid ! {Pid, self(), terminate, StreamID, Reason, State},
 	ok.
 
-%% This clause can only test for early errors that reached the required header.
+%% This clause can only test for early errors that reached the required headers.
 early_error(StreamID, Reason, PartialReq, Resp, Opts) ->
 	Pid = list_to_pid(binary_to_list(cowboy_req:header(<<"x-test-pid">>, PartialReq))),
 	Pid ! {Pid, self(), early_error, StreamID, Reason, PartialReq, Resp, Opts},
-	Resp.
+	case cowboy_req:header(<<"x-test-case">>, PartialReq) of
+		<<"crash_in_early_error",_/bits>> -> error(crash);
+		_ -> Resp
+	end.
diff --git a/test/stream_handler_SUITE.erl b/test/stream_handler_SUITE.erl
index 6a5ff8e..632adff 100644
--- a/test/stream_handler_SUITE.erl
+++ b/test/stream_handler_SUITE.erl
@@ -18,6 +18,7 @@
 -import(ct_helper, [config/2]).
 -import(ct_helper, [doc/1]).
 -import(cowboy_test, [gun_open/1]).
+-import(cowboy_test, [gun_down/1]).
 
 %% ct.
 
@@ -59,6 +60,169 @@ end_per_group(Name, _) ->
 
 %% Tests.
 
+crash_in_init(Config) ->
+	doc("Confirm an error is sent when a stream handler crashes in init/3."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	Ref = gun:get(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_init">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	Pid = receive {Self, P, init, _, _, _} -> P after 1000 -> error(timeout) end,
+	%% Confirm terminate/3 is NOT called. We have no state to give to it.
+	receive {Self, Pid, terminate, _, _, _} -> error(terminate) after 1000 -> ok end,
+	%% Receive a 500 error response.
+	case gun:await(ConnPid, Ref) of
+		{response, fin, 500, _} -> ok;
+		{error, {stream_error, internal_error, _}} -> ok
+	end.
+
+crash_in_data(Config) ->
+	doc("Confirm an error is sent when a stream handler crashes in data/4."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	Ref = gun:post(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"content-length">>, <<"6">>},
+		{<<"x-test-case">>, <<"crash_in_data">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	Pid = receive {Self, P, init, _, _, _} -> P after 1000 -> error(timeout) end,
+	%% Send data to make the stream handler crash.
+	gun:data(ConnPid, Ref, fin, <<"Hello!">>),
+	%% Confirm terminate/3 is called, indicating the stream ended.
+	receive {Self, Pid, terminate, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive a 500 error response.
+	case gun:await(ConnPid, Ref) of
+		{response, fin, 500, _} -> ok;
+		{error, {stream_error, internal_error, _}} -> ok
+	end.
+
+crash_in_info(Config) ->
+	doc("Confirm an error is sent when a stream handler crashes in info/3."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	Ref = gun:get(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_info">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	Pid = receive {Self, P, init, _, _, _} -> P after 1000 -> error(timeout) end,
+	%% Send a message to make the stream handler crash.
+	Pid ! {{Pid, 1}, crash},
+	%% Confirm terminate/3 is called, indicating the stream ended.
+	receive {Self, Pid, terminate, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive a 500 error response.
+	case gun:await(ConnPid, Ref) of
+		{response, fin, 500, _} -> ok;
+		{error, {stream_error, internal_error, _}} -> ok
+	end.
+
+crash_in_terminate(Config) ->
+	doc("Confirm the state is correct when a stream handler crashes in terminate/3."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	%% Do a first request.
+	Ref1 = gun:get(ConnPid, "/hello_world", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_terminate">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	Pid = receive {Self, P, init, _, _, _} -> P after 1000 -> error(timeout) end,
+	%% Confirm terminate/3 is called.
+	receive {Self, Pid, terminate, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive the response.
+	{response, nofin, 200, _} = gun:await(ConnPid, Ref1),
+	{ok, <<"Hello world!">>} = gun:await_body(ConnPid, Ref1),
+	%% Do a second request to make sure the connection state is still good.
+	Ref2 = gun:get(ConnPid, "/hello_world", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_terminate">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called. The pid shouldn't change.
+	receive {Self, Pid, init, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Confirm terminate/3 is called.
+	receive {Self, Pid, terminate, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive the second response.
+	{response, nofin, 200, _} = gun:await(ConnPid, Ref2),
+	{ok, <<"Hello world!">>} = gun:await_body(ConnPid, Ref2),
+	ok.
+
+crash_in_early_error(Config) ->
+	case config(protocol, Config) of
+		http -> do_crash_in_early_error(Config);
+		http2 -> doc("The callback early_error/5 is not currently used for HTTP/2.")
+	end.
+
+do_crash_in_early_error(Config) ->
+	doc("Confirm an error is sent when a stream handler crashes in early_error/5."
+		"The connection is kept open by Cowboy."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	Ref1 = gun:get(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_early_error">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	Pid = receive {Self, P, init, _, _, _} -> P after 1000 -> error(timeout) end,
+	%% Confirm terminate/3 is NOT called. We have no state to give to it.
+	receive {Self, Pid, terminate, _, _, _} -> error(terminate) after 1000 -> ok end,
+	%% Confirm early_error/5 is called.
+	receive {Self, Pid, early_error, _, _, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive a 500 error response.
+	{response, fin, 500, _} = gun:await(ConnPid, Ref1),
+	%% This error is not fatal. We should be able to repeat it on the same connection.
+	Ref2 = gun:get(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"x-test-case">>, <<"crash_in_early_error">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is called.
+	receive {Self, Pid, init, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Confirm terminate/3 is NOT called. We have no state to give to it.
+	receive {Self, Pid, terminate, _, _, _} -> error(terminate) after 1000 -> ok end,
+	%% Confirm early_error/5 is called.
+	receive {Self, Pid, early_error, _, _, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive a 500 error response.
+	{response, fin, 500, _} = gun:await(ConnPid, Ref2),
+	ok.
+
+crash_in_early_error_fatal(Config) ->
+	case config(protocol, Config) of
+		http -> do_crash_in_early_error_fatal(Config);
+		http2 -> doc("The callback early_error/5 is not currently used for HTTP/2.")
+	end.
+
+do_crash_in_early_error_fatal(Config) ->
+	doc("Confirm an error is sent when a stream handler crashes in early_error/5."
+		"The error was fatal and the connection is closed by Cowboy."),
+	Self = self(),
+	ConnPid = gun_open(Config),
+	Ref = gun:get(ConnPid, "/long_polling", [
+		{<<"accept-encoding">>, <<"gzip">>},
+		{<<"host">>, <<"host:port">>},
+		{<<"x-test-case">>, <<"crash_in_early_error_fatal">>},
+		{<<"x-test-pid">>, pid_to_list(Self)}
+	]),
+	%% Confirm init/3 is NOT called. The error occurs before we reach this step.
+	receive {Self, _, init, _, _, _} -> error(init) after 1000 -> ok end,
+	%% Confirm terminate/3 is NOT called. We have no state to give to it.
+	receive {Self, _, terminate, _, _, _} -> error(terminate) after 1000 -> ok end,
+	%% Confirm early_error/5 is called.
+	receive {Self, _, early_error, _, _, _, _, _} -> ok after 1000 -> error(timeout) end,
+	%% Receive a 400 error response. We do not send a 500 when
+	%% early_error/5 crashes, we send the original error.
+	{response, fin, 400, _} = gun:await(ConnPid, Ref),
+	%% Confirm the connection gets closed.
+	gun_down(ConnPid).
+
 shutdown_on_stream_stop(Config) ->
 	doc("Confirm supervised processes are shutdown when stopping the stream."),
 	Self = self(),
@@ -88,7 +252,7 @@ shutdown_on_socket_close(Config) ->
 	doc("Confirm supervised processes are shutdown when the socket closes."),
 	Self = self(),
 	ConnPid = gun_open(Config),
-	Ref = gun:get(ConnPid, "/long_polling", [
+	_ = gun:get(ConnPid, "/long_polling", [
 		{<<"accept-encoding">>, <<"gzip">>},
 		{<<"x-test-case">>, <<"shutdown_on_socket_close">>},
 		{<<"x-test-pid">>, pid_to_list(Self)}
@@ -139,7 +303,7 @@ shutdown_timeout_on_socket_close(Config) ->
 		"when the shutdown timeout triggers after the socket has closed."),
 	Self = self(),
 	ConnPid = gun_open(Config),
-	Ref = gun:get(ConnPid, "/long_polling", [
+	_ = gun:get(ConnPid, "/long_polling", [
 		{<<"accept-encoding">>, <<"gzip">>},
 		{<<"x-test-case">>, <<"shutdown_timeout_on_socket_close">>},
 		{<<"x-test-pid">>, pid_to_list(Self)}