From 11ae407eed92002339fc6cde8acd767e7be953c1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Lo=C3=AFc=20Hoguin?= Date: Wed, 27 Sep 2017 14:17:27 +0200 Subject: Ensure the behavior on stream handler crash is consistent Also corrects the lack of error response when HTTP/1.1 is used. --- src/cowboy_http.erl | 40 +++++---- src/cowboy_http2.erl | 12 +-- test/handlers/stream_handler_h.erl | 26 +++++- test/stream_handler_SUITE.erl | 168 ++++++++++++++++++++++++++++++++++++- 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)} -- cgit v1.2.3