diff --git a/src/ehttpc.erl b/src/ehttpc.erl index 5773c12..57b7fb4 100644 --- a/src/ehttpc.erl +++ b/src/ehttpc.erl @@ -65,14 +65,15 @@ -type path() :: binary() | string(). -type headers() :: [{binary(), iodata()}]. -type body() :: iodata(). +-type callback_fun() :: {function(), [term()]}. -type callback() :: - {function(), list()} + callback_fun() | #{ %% where to send the final results - final_reply := {function(), list()}, + final_reply := callback_fun(), %% optional; sends the gun stream ref and the worker pid so the caller may cancel %% it. - stream_ref => {function(), list()} + stream_ref => callback_fun() }. -type request() :: path() | {path(), headers()} | {path(), headers(), body()}. @@ -226,9 +227,14 @@ mk_request(delete = Method, Req, ExpireAt) when ?IS_HEADERS_REQ(Req) -> %% response is received. -spec request_async(pid(), method(), request(), timeout(), callback()) -> ok. request_async(Worker, Method, Request, Timeout, ResultCallback) when is_pid(Worker) -> - ExpireAt = fresh_expire_at(Timeout), - _ = erlang:send(Worker, mk_async_request(Method, Request, ExpireAt, ResultCallback)), - ok. + case validate_callback(ResultCallback) of + ok -> + ExpireAt = fresh_expire_at(Timeout), + _ = erlang:send(Worker, mk_async_request(Method, Request, ExpireAt, ResultCallback)), + ok; + {error, Reason} -> + error({invalid_callback, Reason}) + end. mk_async_request(head = Method, Req, ExpireAt, RC) when ?IS_HEADERS_REQ(Req) -> ?ASYNC_REQ(Method, Req, ExpireAt, RC); @@ -718,11 +724,11 @@ drop_expired(#{pending := Pending, pending_count := PC} = Requests, Now) -> end. %% For async-request, we evaluate the result-callback with {error, timeout} -maybe_reply_timeout({F, A}) when is_function(F) -> - _ = erlang:apply(F, A ++ [{error, timeout}]), +maybe_reply_timeout({F, A}) when is_function(F), is_list(A) -> + safe_apply(final_reply, F, A ++ [{error, timeout}]), ok; -maybe_reply_timeout(#{final_reply := {F, A}}) when is_function(F) -> - _ = erlang:apply(F, A ++ [{error, timeout}]), +maybe_reply_timeout(#{final_reply := {F, A}}) when is_function(F), is_list(A) -> + safe_apply(final_reply, F, A ++ [{error, timeout}]), ok; maybe_reply_timeout(_) -> %% This is not a callback, but the gen_server:call's From @@ -1042,21 +1048,75 @@ handle_gun_reply(State, Client, StreamRef, IsFin, StatusCode, Headers, Data) -> end end. -reply({F, A}, Result) when is_function(F) -> - _ = erlang:apply(F, A ++ [Result]), - ok; -reply(#{final_reply := FinalReply}, Result) -> - %% assert - {F, A} = FinalReply, - _ = erlang:apply(F, A ++ [Result]), - ok; -reply(From, Result) -> - gen_server:reply(From, Result). +reply({F, A}, Result) when is_function(F), is_list(A) -> + safe_apply(final_reply, F, A ++ [Result]); +reply(#{final_reply := {F, A}}, Result) when is_function(F), is_list(A) -> + safe_apply(final_reply, F, A ++ [Result]); +reply({Pid, _Tag} = From, Result) when is_pid(Pid) -> + gen_server:reply(From, Result); +reply(_InvalidReplyTarget, _Result) -> + log_invalid_callback_target(), + ok. -maybe_send_stream_ref(#{stream_ref := {F, A}}, StreamRef) when is_function(F) -> - _ = erlang:apply(F, [StreamRef, self() | A]), +maybe_send_stream_ref(#{stream_ref := {F, A}}, StreamRef) when + is_function(F), is_list(A) +-> + safe_apply(stream_ref, F, [StreamRef, self() | A]), ok; -maybe_send_stream_ref(_ReplyTo, _StreamRef) -> +maybe_send_stream_ref(_, _) -> + ok. + +safe_apply(Kind, F, Args) -> + try erlang:apply(F, Args) of + _ -> + ok + catch + Class:Reason:Stacktrace -> + logger:error(#{ + msg => "ehttpc_callback_failed", + callback_kind => Kind, + callback => callback_info(F), + class => Class, + reason_kind => callback_reason_kind(Class, Reason), + stacktrace => callback_stacktrace(Stacktrace) + }), + ok + end. + +callback_info(F) when is_function(F) -> + Info = erlang:fun_info(F), + #{ + module => proplists:get_value(module, Info), + name => proplists:get_value(name, Info), + arity => proplists:get_value(arity, Info) + }. + +callback_reason_kind(error, {case_clause, _}) -> + case_clause; +callback_reason_kind(error, {badmatch, _}) -> + badmatch; +callback_reason_kind(error, function_clause) -> + function_clause; +callback_reason_kind(error, {badarity, _}) -> + badarity; +callback_reason_kind(error, undef) -> + undef; +callback_reason_kind(exit, _) -> + exit; +callback_reason_kind(throw, _) -> + throw; +callback_reason_kind(_, _) -> + other. + +callback_stacktrace([{M, F, A, _Info} | _]) when is_integer(A) -> + [{M, F, A}]; +callback_stacktrace([{M, F, Args, _Info} | _]) when is_list(Args) -> + [{M, F, length(Args)}]; +callback_stacktrace(_) -> + []. + +log_invalid_callback_target() -> + logger:error(#{msg => "ehttpc_invalid_callback_target"}), ok. peek_oldest_fn(#{prioritise_latest := true}) -> @@ -1120,6 +1180,35 @@ take_proplist(Key, Proplist0) -> {ValueFromProplist, Proplist1} end. +validate_callback({F, A}) when is_function(F), is_list(A) -> + validate_callback_arity(final_reply, F, length(A) + 1); +validate_callback(#{final_reply := {F, A}} = Callback) when + is_function(F), is_list(A) +-> + case validate_callback_arity(final_reply, F, length(A) + 1) of + ok -> + validate_stream_ref_callback(Callback); + Error -> + Error + end; +validate_callback(_) -> + {error, invalid_callback}. + +validate_stream_ref_callback(#{stream_ref := {F, A}}) when + is_function(F), is_list(A) +-> + validate_callback_arity(stream_ref, F, length(A) + 2); +validate_stream_ref_callback(_) -> + ok. + +validate_callback_arity(Kind, F, ExpectedArity) -> + case is_function(F, ExpectedArity) of + true -> + ok; + false -> + {error, {invalid_callback_arity, Kind, ExpectedArity}} + end. + log(Level, Data, #state{host = Host, port = Port}) -> logger:log(Level, Data#{host => Host, port => Port}). diff --git a/test/ehttpc_async_tests.erl b/test/ehttpc_async_tests.erl index 4943494..5486387 100644 --- a/test/ehttpc_async_tests.erl +++ b/test/ehttpc_async_tests.erl @@ -45,6 +45,45 @@ send_10_async_test() -> PoolOpts = pool_opts(Port, false), true = ?WITH(ServerOpts, PoolOpts, req_async(10, 1000)). +timeout_callback_exception_does_not_kill_worker_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 3_000, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + Tester = self(), + ok = ehttpc:request_async(Worker, get, req(), 5_000, {fun(_) -> ok end, []}), + %% Ensure the first request occupies the only in-flight slot. + _ = sys:get_state(Worker), + ok = ehttpc:request_async( + Worker, + get, + req(), + 10, + { + fun(Result) -> + Tester ! {timeout_callback, Result}, + error(timeout_callback_boom) + end, + [] + } + ), + timer:sleep(30), + ok = ehttpc:request_async(Worker, get, req(), 1_000, {fun(_) -> ok end, []}), + receive + {timeout_callback, {error, timeout}} -> ok + after 2_000 -> + error(timeout_callback_not_called) + end, + _ = sys:get_state(Worker), + ?assert(erlang:is_process_alive(Worker)), + ?assertEqual(1, length(ehttpc:workers(?POOL))) + end + ). + no_expired_req_send_test() -> Port = ?PORT, % infinity diff --git a/test/ehttpc_tests.erl b/test/ehttpc_tests.erl index 522e77f..c309d4b 100644 --- a/test/ehttpc_tests.erl +++ b/test/ehttpc_tests.erl @@ -104,6 +104,182 @@ send_100_test_() -> {"oneoff=false", fun() -> ?WITH(ServerOpts2, PoolOpts2, req_async(100)) end} ]. +callback_crash_does_not_kill_worker_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 0, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + Tester = self(), + Callback = + { + fun(Marker, _Result) -> + Tester ! {callback_called, Marker}, + error(callback_boom) + end, + [legacy_marker] + }, + ok = ehttpc:request_async(Worker, get, req(), 1_000, Callback), + receive + {callback_called, legacy_marker} -> ok + after 2_000 -> + error(callback_not_called) + end, + _ = sys:get_state(Worker), + ?assert(erlang:is_process_alive(Worker)), + ?assertMatch({ok, 200, _, _}, ehttpc:request(Worker, get, req(), 1_000, 0)) + end + ). + +structured_callback_exception_does_not_kill_worker_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 0, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + Tester = self(), + Callback = #{ + final_reply => + { + fun(Marker, Result) -> + Tester ! {final_reply, Marker, Result}, + error(final_reply_callback_boom) + end, + [final_marker] + }, + stream_ref => + { + fun(StreamRef, WorkerPid, Marker) -> + Tester ! {stream_ref_called, StreamRef, WorkerPid, Marker}, + error(stream_ref_callback_boom) + end, + [stream_marker] + } + }, + ok = ehttpc:request_async(Worker, get, req(), 1_000, Callback), + receive + {stream_ref_called, _, Worker, stream_marker} -> ok + after 2_000 -> + error(stream_ref_callback_not_called) + end, + receive + {final_reply, final_marker, {ok, 200, _, _}} -> ok + after 2_000 -> + error(final_reply_not_called) + end, + _ = sys:get_state(Worker), + ?assertMatch({ok, 200, _, _}, ehttpc:request(Worker, get, req(), 1_000, 0)) + end + ). + +invalid_callback_does_not_kill_worker_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 0, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + ?assertError( + {invalid_callback, invalid_callback}, + ehttpc:request_async(Worker, get, req(), 1_000, #{ + stream_ref => {fun() -> ok end, []} + }) + ), + ?assertError( + {invalid_callback, invalid_callback}, + ehttpc:request_async(Worker, get, req(), 1_000, {self(), make_ref()}) + ), + ?assertError( + {invalid_callback, {invalid_callback_arity, final_reply, 1}}, + ehttpc:request_async(Worker, get, req(), 1_000, {fun() -> ok end, []}) + ), + ?assertError( + {invalid_callback, {invalid_callback_arity, stream_ref, 2}}, + ehttpc:request_async(Worker, get, req(), 1_000, #{ + final_reply => {fun(_) -> ok end, []}, + stream_ref => {fun() -> ok end, []} + }) + ), + _ = sys:get_state(Worker), + ?assertEqual(1, length(ehttpc:workers(?POOL))), + ?assertMatch({ok, 200, _, _}, ehttpc:request(Worker, get, req(), 1_000, 0)) + end + ). + +invalid_optional_stream_ref_is_ignored_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 0, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + Tester = self(), + Callback = #{ + final_reply => + {fun(Result) -> Tester ! {final_reply, Result} end, []}, + stream_ref => undefined + }, + ok = ehttpc:request_async(Worker, get, req(), 1_000, Callback), + receive + {final_reply, {ok, 200, _, _}} -> ok + after 2_000 -> + error(final_reply_not_called) + end, + _ = sys:get_state(Worker), + ?assertMatch({ok, 200, _, _}, ehttpc:request(Worker, get, req(), 1_000, 0)) + end + ). + +repeated_callback_exceptions_do_not_restart_worker_test() -> + Port = ?PORT, + ServerOpts = #{port => Port, name => ?FUNCTION_NAME, delay => 0, oneoff => false}, + PoolOpts = pool_opts(Port, false), + ?WITH( + ServerOpts, + PoolOpts, + begin + Worker = ehttpc_pool:pick_worker(?POOL), + Tester = self(), + lists:foreach( + fun(I) -> + Callback = + { + fun(_Result) -> + Tester ! {callback_called, I}, + error(callback_boom) + end, + [] + }, + ok = ehttpc:request_async(Worker, get, req(), 1_000, Callback) + end, + lists:seq(1, 20) + ), + lists:foreach( + fun(I) -> + receive + {callback_called, I} -> ok + after 2_000 -> + error({callback_not_called, I}) + end + end, + lists:seq(1, 20) + ), + _ = sys:get_state(Worker), + ?assertEqual([{{?POOL, 1}, Worker}], ehttpc:workers(?POOL)), + ?assertEqual(ok, ehttpc:check_pool_integrity(?POOL)) + end + ). + send_1000_async_pipeline_test_() -> TestTimeout = 30, Port = ?PORT,