diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 58611889..95593b91 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -234,6 +234,7 @@ %% {stream, headers, Status, Headers, Buffer, Pending} %% {stream, body_full, Status, Headers, Acc, From} %% {stream, done, Status, Headers, Buffer} + %% {stream, refused, ErrorCode} (refused by a GOAWAY, caller not yet told) h2_streams = #{} :: #{pos_integer() => {term(), tuple()}}, %% Shared through the pool (share_h2/1): no owner, each stream monitors %% its caller, and the connection closes itself once idle. @@ -1033,6 +1034,17 @@ connected({call, From}, get_state, #conn_data{h2_goaway = ErrorCode}) connected({call, From}, get_state, _Data) -> {keep_state_and_data, [{reply, From, {ok, connected}}]}; +%% The streaming-body calls only make sense in streaming_body, except for a +%% request a GOAWAY refused while its caller was between them: the connection +%% came back here then (h2_on_goaway/3), and the caller's next call is the +%% first chance to tell it. +connected({call, From}, {send_body_chunk, _}, #conn_data{protocol = http2} = Data) -> + h2_answer_refused_request(From, Data); +connected({call, From}, finish_send_body, #conn_data{protocol = http2} = Data) -> + h2_answer_refused_request(From, Data); +connected({call, From}, start_response, #conn_data{protocol = http2} = Data) -> + h2_answer_refused_request(From, Data); + connected({call, From}, verify_socket, #conn_data{socket = undefined} = Data) -> %% Socket not connected {next_state, closed, Data, [{reply, From, {error, closed}}]}; @@ -3309,9 +3321,16 @@ handle_h2_stream_owner_down(Ref, State, h2_stream_owner_down_result(streaming_body, StreamId, #conn_data{h2_stream_id = StreamId} = Data) -> %% The caller streaming a request body died: its stream is gone, so the - %% connection can take requests again. - {next_state, connected, - Data#conn_data{h2_stream_id = undefined, request_from = undefined}}; + %% connection can take requests again, unless that stream was the last + %% one of a GOAWAY drain. + case h2_stream_result(Data#conn_data{h2_stream_id = undefined, + request_from = undefined}, []) of + {keep_state, Data1, Actions} -> + {next_state, connected, Data1, + [{state_timeout, infinity, idle_timeout} | Actions]}; + Closed -> + Closed + end; h2_stream_owner_down_result(_State, _StreamId, Data) -> h2_stream_result(Data, []). @@ -3991,24 +4010,42 @@ h2_idle_actions(_Data) -> %% here, and close once the accepted streams end. Aborting those too reported %% requests the server went on to complete as failed. %% -%% A streamed request or response body (a `stream' entry) drives the -%% connection through states of its own, so with one in flight the connection -%% still closes at once. +%% A streaming-body request the GOAWAY refuses has no caller parked on it, so +%% it is kept as a `refused' entry and the connection leaves streaming_body: +%% the caller learns of it on its next call (h2_answer_refused_request/2). h2_on_goaway(LastStreamId, ErrorCode, #conn_data{h2_streams = Streams} = Data) -> - {Accepted, Refused} = lists:partition(fun(SId) -> SId =< LastStreamId end, - maps:keys(Streams)), - case Accepted =/= [] andalso not h2_streaming_in_flight(Streams) of + Refused = [SId || SId <- maps:keys(Streams), SId > LastStreamId], + {Replies, Data1} = abort_h2_streams(Refused, ErrorCode, Data), + #conn_data{h2_streams = Streams1, h2_stream_id = StreamId} = Data1, + case map_size(Streams1) =:= 0 of true -> - {Replies, Data1} = abort_h2_streams(Refused, {goaway, ErrorCode}, Data), - ok = leave_h2_pool(Data1), - {keep_state, Data1#conn_data{h2_goaway = ErrorCode}, Replies}; + {next_state, closed, Data2, CloseReplies} = h2_close_on_goaway(ErrorCode, Data1), + {next_state, closed, Data2, Replies ++ CloseReplies}; false -> - h2_close_on_goaway(ErrorCode, Data) + ok = leave_h2_pool(Data1), + Data3 = Data1#conn_data{h2_goaway = ErrorCode}, + case maps:get(StreamId, Streams1, undefined) of + {_, {stream, refused, _}} -> + {next_state, connected, Data3, + [{state_timeout, infinity, idle_timeout} | Replies]}; + _ -> + {keep_state, Data3, Replies} + end end. -h2_streaming_in_flight(Streams) -> - lists:any(fun({_Owner, Inner}) -> element(1, Inner) =:= stream end, - maps:values(Streams)). +%% @private Answer a streaming-body call made after the request's stream was +%% refused by a GOAWAY (see h2_on_goaway/3). Any other such call in +%% `connected' is a misuse, as before. +h2_answer_refused_request({Caller, _} = From, #conn_data{h2_streams = Streams} = Data) -> + case [{Id, Code} || {Id, {Owner, {stream, refused, Code}}} <- maps:to_list(Streams), + h2_stream_owner_pid(Owner) =:= Caller] of + [{StreamId, ErrorCode}] -> + Data1 = Data#conn_data{h2_stream_id = undefined, request_from = undefined}, + h2_stream_result(drop_h2_stream(StreamId, Data1), + [{reply, From, {error, {goaway, ErrorCode}}}]); + [] -> + {keep_state_and_data, [{reply, From, {error, invalid_state}}]} + end. leave_h2_pool(#conn_data{pool_pid = PoolPid}) when is_pid(PoolPid) -> gen_server:cast(PoolPid, {unregister_h2, self()}); @@ -4051,10 +4088,22 @@ collect_h2_aborts(Err, #conn_data{h2_streams = Streams} = Data) -> Data#conn_data{h2_streams = #{}, request_from = undefined}), {Replies, Data1}. -%% @private Fail some streams and keep the rest of the connection running. -abort_h2_streams(StreamIds, Err, #conn_data{h2_streams = Streams} = Data) -> - Replies = h2_abort_replies(Err, maps:with(StreamIds, Streams)), - {Replies, lists:foldl(fun drop_h2_stream/2, Data, StreamIds)}. +%% @private Fail the refused streams and keep the rest of the connection +%% running. A request body still being sent has no caller parked on it, so +%% its stream is kept as a `refused' entry for the caller's next call to find. +abort_h2_streams(StreamIds, ErrorCode, #conn_data{h2_streams = Streams} = Data) -> + Refused = maps:with(StreamIds, Streams), + Replies = h2_abort_replies({goaway, ErrorCode}, Refused), + Data1 = maps:fold(fun + (StreamId, {Owner, {stream, sending}}, D) -> + D#conn_data{h2_streams = maps:put(StreamId, {Owner, {stream, refused, ErrorCode}}, + D#conn_data.h2_streams)}; + (_StreamId, {_Owner, {stream, refused, _}}, D) -> + D; + (StreamId, _Entry, D) -> + drop_h2_stream(StreamId, D) + end, Data, Refused), + {Replies, Data1}. h2_abort_replies(Err, Streams) -> maps:fold(fun diff --git a/test/hackney_h2_goaway_server.erl b/test/hackney_h2_goaway_server.erl new file mode 100644 index 00000000..1d62bb6c --- /dev/null +++ b/test/hackney_h2_goaway_server.erl @@ -0,0 +1,161 @@ +%%% Frame-level HTTP/2 server for the GOAWAY tests, built on the h2 dep's +%%% h2_frame / h2_hpack. +%%% +%%% Its first connection holds every request until it has sent its one GOAWAY, +%%% then answers only the streams at or below last_stream_id, a request with a +%%% streamed body once that body is complete. Later connections answer every +%%% request as soon as it is complete. Each response body is the stream id. +%%% +%%% The GOAWAY is sent on a trigger: +%%% {second_stream, Pick} when the second stream opens, with +%%% Pick(FirstStreamId, SecondStreamId) as last_stream_id +%%% first_data when a stream's first body chunk arrives, with that +%%% stream as last_stream_id +%%% and shaped by options: +%%% goaway => direct | two_step two_step first sends GOAWAY(2^31-1), the +%%% graceful shutdown of RFC 9113 6.8 (default direct) +%%% answer => boolean() false answers nothing after the GOAWAY, not +%%% even the accepted streams (default true) +-module(hackney_h2_goaway_server). + +-export([start/1, start/2, stop/1]). + +-define(PREFACE, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>). + +start(Trigger) -> + start(Trigger, #{}). + +start(Trigger, Opts) -> + Certs = cert_dir(), + {ok, LSock} = ssl:listen(0, + [{certfile, filename:join(Certs, "server.pem")}, + {keyfile, filename:join(Certs, "server.key")}, + {alpn_preferred_protocols, [<<"h2">>]}, + {versions, ['tlsv1.2', 'tlsv1.3']}, + {active, false}, {mode, binary}, {reuseaddr, true}]), + {ok, {_, Port}} = ssl:sockname(LSock), + Options = maps:merge(#{goaway => direct, answer => true}, Opts), + Pid = spawn(fun() -> accept_loop(LSock, Trigger, Options) end), + Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]), + {Pid, Url}. + +stop(Pid) -> + exit(Pid, kill). + +accept_loop(LSock, Trigger, Opts) -> + case ssl:transport_accept(LSock, 2000) of + {ok, TSock} -> + spawn(fun() -> serve(TSock, Trigger, Opts) end), + accept_loop(LSock, none, Opts); + {error, timeout} -> accept_loop(LSock, Trigger, Opts); + {error, closed} -> ok + end. + +serve(TSock, Trigger, Opts) -> + case ssl:handshake(TSock, 5000) of + {ok, Sock} -> + case recv_preface(Sock, <<>>) of + {ok, Rest} -> + send(Sock, h2_frame:settings([])), + %% seen: stream ids in order of arrival. complete: streams + %% whose request has fully arrived and is not answered yet. + %% accepted: after the GOAWAY, the streams it let through. + loop(Sock, Rest, Opts#{enc => h2_hpack:new_context(), trigger => Trigger, + seen => [], complete => [], accepted => all}); + _ -> ok + end; + _ -> ok + end. + +recv_preface(_Sock, Acc) when byte_size(Acc) >= 24 -> + <> = Acc, + case Pre of ?PREFACE -> {ok, Rest}; _ -> {error, bad_preface} end; +recv_preface(Sock, Acc) -> + case ssl:recv(Sock, 0, 5000) of + {ok, Data} -> recv_preface(Sock, <>); + {error, R} -> {error, R} + end. + +loop(Sock, Buf, St) -> + case h2_frame:decode(Buf) of + {ok, Frame, Rest} -> + case handle(Sock, Frame, St) of + {continue, St2} -> loop(Sock, Rest, St2); + stop -> ok + end; + {more, _} -> + case ssl:recv(Sock, 0, 30000) of + {ok, Data} -> loop(Sock, <>, St); + {error, _} -> ok + end; + {error, _, Rest} -> loop(Sock, Rest, St); + {error, _} -> ok + end. + +handle(Sock, {settings, _}, St) -> send(Sock, h2_frame:settings_ack()), {continue, St}; +handle(Sock, {ping, D}, St) -> send(Sock, h2_frame:ping_ack(D)), {continue, St}; +handle(_Sock, {goaway, _, _, _}, _St) -> stop; +handle(Sock, {headers, Sid, _Block, EndStream, _EndHeaders}, St) -> + {continue, on_request_frame(Sock, headers, Sid, EndStream, St)}; +%% decode/1 adds the flow-controlled size as a fifth element. +handle(Sock, {data, Sid, _Bin, EndStream, _FlowControlled}, St) -> + {continue, on_request_frame(Sock, data, Sid, EndStream, St)}; +handle(_Sock, _Other, St) -> {continue, St}. + +on_request_frame(Sock, Kind, Sid, EndStream, #{seen := Seen, complete := Complete} = St) -> + Seen2 = case lists:member(Sid, Seen) of true -> Seen; false -> Seen ++ [Sid] end, + Complete2 = case EndStream of true -> [Sid | Complete]; false -> Complete end, + St1 = St#{seen := Seen2, complete := Complete2}, + St2 = case goaway_for(Kind, Sid, EndStream, St1) of + none -> St1; + LastStreamId -> send_goaway(Sock, LastStreamId, St1) + end, + answer_ready(Sock, St2). + +goaway_for(headers, _Sid, _EndStream, #{trigger := {second_stream, Pick}, seen := [First, Second]}) -> + Pick(First, Second); +goaway_for(data, Sid, false, #{trigger := first_data}) -> + Sid; +goaway_for(_Kind, _Sid, _EndStream, _St) -> + none. + +%% One GOAWAY per connection, then a short drain before anything is answered. +send_goaway(Sock, LastStreamId, #{seen := Seen, goaway := Shape, answer := Answer} = St) -> + case Shape of + two_step -> + send(Sock, h2_frame:goaway(16#7fffffff, no_error, <<>>)), + timer:sleep(50); + direct -> + ok + end, + send(Sock, h2_frame:goaway(LastStreamId, no_error, <<>>)), + timer:sleep(200), + Accepted = case Answer of + true -> [S || S <- Seen, S =< LastStreamId]; + false -> [] + end, + St#{trigger := none, accepted := Accepted}. + +%% Answer every complete request the connection may still answer. Nothing is +%% answered while a GOAWAY is still to come, so the streams it covers are in +%% flight when it does. +answer_ready(_Sock, #{trigger := Trigger} = St) when Trigger =/= none -> + St; +answer_ready(Sock, #{complete := Complete, accepted := Accepted} = St) -> + Ready = [S || S <- lists:reverse(Complete), + Accepted =:= all orelse lists:member(S, Accepted)], + lists:foldl(fun(S, Acc) -> respond(Sock, S, Acc) end, + St#{complete := Complete -- Ready}, Ready). + +respond(Sock, Sid, #{enc := Enc} = St) -> + {HBlock, Enc2} = h2_hpack:encode([{<<":status">>, <<"200">>}], Enc), + send(Sock, h2_frame:headers(Sid, HBlock, false)), + send(Sock, h2_frame:data(Sid, integer_to_binary(Sid), true)), + St#{enc := Enc2}. + +send(Sock, FrameData) -> ssl:send(Sock, h2_frame:encode(FrameData)). + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]). diff --git a/test/hackney_http2_goaway_drain_tests.erl b/test/hackney_http2_goaway_drain_tests.erl index 34a4b3d4..a46b7ef1 100644 --- a/test/hackney_http2_goaway_drain_tests.erl +++ b/test/hackney_http2_goaway_drain_tests.erl @@ -6,14 +6,13 @@ %%% request the server went on to complete came back as an error: for a payment %%% API that is a charge that succeeded and was reported as failed. %%% -%%% The server below holds the first two streams of its first connection, sends -%%% GOAWAY with a chosen last_stream_id, then answers only the streams at or -%%% below it. Later connections answer immediately. +%%% The server (hackney_h2_goaway_server) holds the first two streams of its +%%% first connection, sends GOAWAY with a chosen last_stream_id, then answers +%%% only the streams at or below it. -module(hackney_http2_goaway_drain_tests). -include_lib("eunit/include/eunit.hrl"). --define(PREFACE, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>). -define(POOL, goaway_drain_test_pool). goaway_drain_test_() -> @@ -41,53 +40,55 @@ stop_pool() -> %% complete. A request made while they drain must not land on the draining %% connection, where the peer would ignore it, but dial a fresh one. accepted_streams_complete() -> - {Server, Url} = start_server(fun(_First, Second) -> Second end), + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(_First, Second) -> Second end}), try [R1, R2, R3] = concurrent_requests(Url, 3), ?assertEqual({ok, 200, <<"1">>}, R1), ?assertEqual({ok, 200, <<"3">>}, R2), ?assertEqual({ok, 200, <<"1">>}, R3) after - stop_server(Server) + hackney_h2_goaway_server:stop(Server) end. %% GOAWAY(last_stream_id = 1) after streams 1 and 3: stream 3 was not accepted %% and fails at once with the goaway reason, stream 1 still completes. unaccepted_stream_fails_fast() -> - {Server, Url} = start_server(fun(First, _Second) -> First end), + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}), try [R1, R2] = concurrent_requests(Url, 2), ?assertEqual({ok, 200, <<"1">>}, R1), ?assertEqual({error, {goaway, no_error}}, R2) after - stop_server(Server) + hackney_h2_goaway_server:stop(Server) end. %% The graceful shutdown of RFC 9113 6.8: GOAWAY(2^31-1) stops new streams %% without refusing any, then a second GOAWAY gives the real last_stream_id. %% The first frame alone must not fail anything. two_step_shutdown() -> - {Server, Url} = start_server(fun(First, _Second) -> {two_step, First} end), + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}, + #{goaway => two_step}), try [R1, R2] = concurrent_requests(Url, 2), ?assertEqual({ok, 200, <<"1">>}, R1), ?assertEqual({error, {goaway, no_error}}, R2) after - stop_server(Server) + hackney_h2_goaway_server:stop(Server) end. %% A server that accepts a stream and then never answers it must not keep the %% draining connection around: the stream's own recv_timeout ends it, and with %% it the connection, so the next request gets a fresh one. stalled_drain_ends_with_the_stream() -> - {Server, Url} = start_server(fun(First, _Second) -> {never, First} end), + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}, + #{answer => false}), try [R1, R2] = concurrent_requests(Url, 2, [{recv_timeout, 1000}]), ?assertEqual({error, timeout}, R1), ?assertEqual({error, {goaway, no_error}}, R2), ?assertEqual({ok, 200, <<"1">>}, fetch(Url, [])) after - stop_server(Server) + hackney_h2_goaway_server:stop(Server) end. %% The first request registers the shared connection before the second checks @@ -104,9 +105,6 @@ concurrent_requests(Url, N, Extra) -> end || Delay <- lists:sublist([0, 300, 100], N)], [receive {P, R} -> R after 10000 -> {error, test_timeout} end || P <- Pids]. -fetch(Url) -> - fetch(Url, []). - fetch(Url, Extra) -> Opts = Extra ++ [{pool, ?POOL}, {protocols, [http2]}, {recv_timeout, 5000}, {ssl_options, [{insecure, true}, {verify, verify_none}]}], @@ -114,115 +112,3 @@ fetch(Url, Extra) -> {ok, S, _H, B} when is_binary(B) -> {ok, S, B}; {error, E} -> {error, E} end. - -%%==================================================================== -%% Frame-level h2 server. The body of each response is its stream id. -%%==================================================================== - -start_server(PickLastStreamId) -> - Certs = cert_dir(), - {ok, LSock} = ssl:listen(0, - [{certfile, filename:join(Certs, "server.pem")}, - {keyfile, filename:join(Certs, "server.key")}, - {alpn_preferred_protocols, [<<"h2">>]}, - {versions, ['tlsv1.2', 'tlsv1.3']}, - {active, false}, {mode, binary}, {reuseaddr, true}]), - {ok, {_, Port}} = ssl:sockname(LSock), - Pid = spawn(fun() -> accept_loop(LSock, {hold, PickLastStreamId}) end), - Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]), - {Pid, Url}. - -stop_server(Pid) -> - exit(Pid, kill). - -accept_loop(LSock, Mode) -> - case ssl:transport_accept(LSock, 2000) of - {ok, TSock} -> - spawn(fun() -> serve(TSock, Mode) end), - accept_loop(LSock, immediate); - {error, timeout} -> accept_loop(LSock, Mode); - {error, closed} -> ok - end. - -serve(TSock, Mode) -> - case ssl:handshake(TSock, 5000) of - {ok, Sock} -> - case recv_preface(Sock, <<>>) of - {ok, Rest} -> - send(Sock, h2_frame:settings([])), - loop(Sock, Rest, #{enc => h2_hpack:new_context(), mode => Mode, - held => []}); - _ -> ok - end; - _ -> ok - end. - -recv_preface(_Sock, Acc) when byte_size(Acc) >= 24 -> - <> = Acc, - case Pre of ?PREFACE -> {ok, Rest}; _ -> {error, bad_preface} end; -recv_preface(Sock, Acc) -> - case ssl:recv(Sock, 0, 5000) of - {ok, Data} -> recv_preface(Sock, <>); - {error, R} -> {error, R} - end. - -loop(Sock, Buf, St) -> - case h2_frame:decode(Buf) of - {ok, Frame, Rest} -> - case handle(Sock, Frame, St) of - {continue, St2} -> loop(Sock, Rest, St2); - stop -> ok - end; - {more, _} -> - case ssl:recv(Sock, 0, 30000) of - {ok, Data} -> loop(Sock, <>, St); - {error, _} -> ok - end; - {error, _, Rest} -> loop(Sock, Rest, St); - {error, _} -> ok - end. - -handle(Sock, {settings, _}, St) -> send(Sock, h2_frame:settings_ack()), {continue, St}; -handle(Sock, {ping, D}, St) -> send(Sock, h2_frame:ping_ack(D)), {continue, St}; -handle(_Sock, {goaway, _, _, _}, _St) -> stop; -handle(Sock, {headers, Sid, _B, _E, _H}, #{mode := immediate} = St) -> - {continue, respond(Sock, Sid, St)}; -handle(Sock, {headers, Sid, _B, _E, _H}, #{mode := {hold, Pick}, held := Held} = St) -> - case Held ++ [Sid] of - [First, Second] -> - LastStreamId = send_goaway(Sock, Pick(First, Second)), - %% The drain: the peer is told, then the accepted streams finish. - timer:sleep(200), - St2 = lists:foldl(fun(S, Acc) -> respond(Sock, S, Acc) end, - St#{held := []}, - [S || S <- [First, Second], S =< LastStreamId]), - {continue, St2#{mode := draining}}; - Held2 -> - {continue, St#{held := Held2}} - end; -handle(_Sock, _Other, St) -> {continue, St}. - -send_goaway(Sock, {never, LastStreamId}) -> - send(Sock, h2_frame:goaway(LastStreamId, no_error, <<>>)), - %% Nothing at or below LastStreamId is answered. - 0; -send_goaway(Sock, {two_step, LastStreamId}) -> - send(Sock, h2_frame:goaway(16#7fffffff, no_error, <<>>)), - timer:sleep(50), - send_goaway(Sock, LastStreamId); -send_goaway(Sock, LastStreamId) -> - send(Sock, h2_frame:goaway(LastStreamId, no_error, <<>>)), - LastStreamId. - -respond(Sock, Sid, #{enc := Enc} = St) -> - {HBlock, Enc2} = h2_hpack:encode([{<<":status">>, <<"200">>}], Enc), - send(Sock, h2_frame:headers(Sid, HBlock, false)), - send(Sock, h2_frame:data(Sid, integer_to_binary(Sid), true)), - St#{enc := Enc2}. - -send(Sock, FrameData) -> ssl:send(Sock, h2_frame:encode(FrameData)). - -cert_dir() -> - BeamDir = filename:dirname(code:which(?MODULE)), - Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), - filename:join([filename:absname(Root), "test", "certs"]). diff --git a/test/hackney_http2_goaway_streaming_tests.erl b/test/hackney_http2_goaway_streaming_tests.erl new file mode 100644 index 00000000..25ab73c6 --- /dev/null +++ b/test/hackney_http2_goaway_streaming_tests.erl @@ -0,0 +1,134 @@ +%%% A GOAWAY while a streamed request or response body is in flight. +%%% +%%% Same rule as hackney_http2_goaway_drain_tests: streams up to the GOAWAY's +%%% last_stream_id finish, streams above it fail with {goaway, _}. The streamed +%%% request body adds one wrinkle: when it is refused, nobody is parked on it, +%%% since its caller is between send_body/finish_send_body/start_response +%%% calls, so the refusal has to wait for the caller's next call. +%%% +%%% The server is hackney_h2_goaway_server; each response body is the stream id. +-module(hackney_http2_goaway_streaming_tests). + +-include_lib("eunit/include/eunit.hrl"). + +-define(POOL, goaway_streaming_test_pool). + +goaway_streaming_test_() -> + {foreach, fun setup/0, fun cleanup/1, + [{timeout, 30, fun accepted_upload_completes/0}, + {timeout, 30, fun() -> refused_upload_fails_on(send_body) end}, + {timeout, 30, fun() -> refused_upload_fails_on(finish_send_body) end}, + {timeout, 30, fun() -> refused_upload_fails_on(start_response) end}, + {timeout, 30, fun accepted_streamed_response_completes/0}, + {timeout, 30, fun refused_streamed_response_fails_fast/0}]}. + +setup() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + stop_pool(), + ok = hackney_pool:start_pool(?POOL, [{max_connections, 5}]), + ok. + +cleanup(_) -> + stop_pool(). + +stop_pool() -> + try hackney_pool:stop_pool(?POOL) catch _:_ -> ok end, + ok. + +%% The server sends GOAWAY covering the streamed request as soon as its first +%% body chunk arrives. The rest of the body goes out, the response comes back, +%% and the connection is gone for the request after. +accepted_upload_completes() -> + {Server, Url} = hackney_h2_goaway_server:start(first_data), + try + {ok, Conn} = hackney:request(post, Url, [], stream, opts()), + ok = hackney:send_body(Conn, <<"abc">>), + timer:sleep(300), + ok = hackney:finish_send_body(Conn), + {ok, 200, _Headers, Conn} = hackney:start_response(Conn), + ?assertEqual({ok, <<"1">>}, hackney:body(Conn)), + ?assertEqual({ok, 200, <<"1">>}, fetch(Url)) + after + hackney_h2_goaway_server:stop(Server) + end. + +%% Stream 1 is a plain request the server holds, stream 3 a streamed request, +%% and the GOAWAY covers only stream 1. Whichever call the sender makes next is +%% the one that fails, stream 1 still completes, and the connection is gone for +%% the request after. +refused_upload_fails_on(Call) -> + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}), + try + P1 = spawn_fetch(Url), + timer:sleep(300), + {ok, Conn} = hackney:request(post, Url, [], stream, opts()), + timer:sleep(300), + Result = case Call of + send_body -> hackney:send_body(Conn, <<"abc">>); + finish_send_body -> hackney:finish_send_body(Conn); + start_response -> hackney:start_response(Conn) + end, + ?assertEqual({error, {goaway, no_error}}, Result), + ?assertEqual({ok, 200, <<"1">>}, await(P1)), + ?assertEqual({ok, 200, <<"1">>}, fetch(Url)) + after + hackney_h2_goaway_server:stop(Server) + end. + +%% Stream 1 is a request whose response is read as a stream, stream 3 a plain +%% request, and the GOAWAY covers only stream 1. +accepted_streamed_response_completes() -> + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}), + try + {ok, Conn} = connect(Url), + Self = self(), + P1 = spawn_link(fun() -> + R = case hackney:send_request(Conn, {get, <<"/">>, [], <<>>}) of + {ok, 200, _Headers, Conn} -> hackney:body(Conn); + Other -> Other + end, + Self ! {self(), R} + end), + timer:sleep(300), + ?assertEqual({error, {goaway, no_error}}, fetch(Url)), + ?assertEqual({ok, <<"1">>}, await(P1)) + after + hackney_h2_goaway_server:stop(Server) + end. + +%% Stream 1 is a plain request the server holds, stream 3 a request whose +%% response would be read as a stream, and the GOAWAY covers only stream 1. +refused_streamed_response_fails_fast() -> + {Server, Url} = hackney_h2_goaway_server:start({second_stream, fun(First, _Second) -> First end}), + try + P1 = spawn_fetch(Url), + timer:sleep(300), + {ok, Conn} = connect(Url), + ?assertEqual({error, {goaway, no_error}}, + hackney:send_request(Conn, {get, <<"/">>, [], <<>>})), + ?assertEqual({ok, 200, <<"1">>}, await(P1)) + after + hackney_h2_goaway_server:stop(Server) + end. + +spawn_fetch(Url) -> + Self = self(), + spawn_link(fun() -> Self ! {self(), fetch(Url)} end). + +await(Pid) -> + receive {Pid, R} -> R after 10000 -> {error, test_timeout} end. + +fetch(Url) -> + case hackney:request(get, Url, [], <<>>, opts()) of + {ok, S, _H, B} when is_binary(B) -> {ok, S, B}; + {error, E} -> {error, E} + end. + +connect(Url) -> + #{host := Host, port := Port} = uri_string:parse(Url), + hackney:connect(hackney_ssl, Host, Port, opts()). + +opts() -> + [{pool, ?POOL}, {protocols, [http2]}, {recv_timeout, 5000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}].