Skip to content

Drain a GOAWAY with a streamed body in flight #963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
benoitc wants to merge 1 commit into from
Closed

Drain a GOAWAY with a streamed body in flight #963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
73 changes: 55 additions & 18 deletions src/hackney_conn.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1521,6 +1521,18 @@ streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol =
{next_state, closed, Data, [{reply, From, {error, Reason}}]}
end;

streaming_body({call, From}, Req, #conn_data{protocol = http2, h2_stream_id = StreamId,
h2_streams = Streams,
h2_goaway = ErrorCode} = Data)
when (Req =:= finish_send_body orelse Req =:= start_response orelse
element(1, Req) =:= send_body_chunk),
is_map_key(StreamId, Streams),
element(2, map_get(StreamId, Streams)) =:= {stream, refused} ->
%% A GOAWAY refused this upload while other streams drain: nothing more
%% goes out on it, and its caller learns it the next time it asks.
h2_upload_done(drop_h2_stream(StreamId, Data),
[{reply, From, {error, {goaway, ErrorCode}}}]);

streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol = http2} = Data) ->
%% HTTP/2 - send a DATA frame without END_STREAM.
#conn_data{h2_conn = H2Conn, h2_stream_id = StreamId,
Expand All @@ -1538,7 +1550,7 @@ streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol =
ok ->
{keep_state_and_data, [{reply, From, ok}]};
{error, Reason} ->
{next_state, closed, Data, [{reply, From, {error, Reason}}]}
h2_upload_failed(From, Reason, Data)
end;

streaming_body({call, From}, {send_body_chunk, BodyData}, Data) ->
Expand Down Expand Up @@ -1577,7 +1589,7 @@ streaming_body({call, From}, finish_send_body, #conn_data{protocol = http2} = Da
ok ->
{keep_state, Data, [{reply, From, ok}]};
{error, Reason} ->
{next_state, closed, Data, [{reply, From, {error, Reason}}]}
h2_upload_failed(From, Reason, Data)
end;

streaming_body({call, From}, finish_send_body, Data) ->
Expand Down Expand Up @@ -3310,8 +3322,7 @@ 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}};
h2_upload_done(Data, []);
h2_stream_owner_down_result(_State, _StreamId, Data) ->
h2_stream_result(Data, []).

Expand Down Expand Up @@ -3991,31 +4002,54 @@ 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 refused upload has no caller parked to tell, so it stays marked until its
%% caller next sends or asks for the response.
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
true ->
{Replies, Data1} = abort_h2_streams(Refused, {goaway, ErrorCode}, Data),
case Accepted of
[] ->
h2_close_on_goaway(ErrorCode, Data);
_ ->
{Uploads, Others} =
lists:partition(fun(SId) ->
element(2, maps:get(SId, Streams)) =:= {stream, sending}
end, Refused),
{Replies, Data1} = abort_h2_streams(Others, {goaway, ErrorCode}, Data),
Streams1 = lists:foldl(fun(SId, Acc) ->
{Owner, _} = maps:get(SId, Acc),
maps:put(SId, {Owner, {stream, refused}}, Acc)
end, Data1#conn_data.h2_streams, Uploads),
ok = leave_h2_pool(Data1),
{keep_state, Data1#conn_data{h2_goaway = ErrorCode}, Replies};
false ->
h2_close_on_goaway(ErrorCode, Data)
{keep_state, Data1#conn_data{h2_streams = Streams1, h2_goaway = ErrorCode},
Replies}
end.

h2_streaming_in_flight(Streams) ->
lists:any(fun({_Owner, Inner}) -> element(1, Inner) =:= stream end,
maps:values(Streams)).
%% @private Leave streaming_body once the upload stream is gone: back to
%% connected, or closed when a GOAWAY drain has nothing left.
h2_upload_done(Data, Replies) ->
Data1 = Data#conn_data{h2_stream_id = undefined, request_from = undefined},
case Data1 of
#conn_data{h2_goaway = ErrorCode, h2_streams = Streams}
when ErrorCode =/= undefined, map_size(Streams) =:= 0 ->
{next_state, closed, Data2, CloseReplies} = h2_close_on_goaway(ErrorCode, Data1),
{next_state, closed, Data2, Replies ++ CloseReplies};
_ ->
{next_state, connected, Data1, Replies}
end.

%% @private A body send failed: the connection is unusable, so fail every
%% other stream on it too instead of leaving their callers waiting.
h2_upload_failed(From, Reason, Data) ->
{next_state, closed, Data1, Replies} = h2_teardown({closed, Reason}, Data),
{next_state, closed, Data1, [{reply, From, {error, Reason}} | Replies]}.

leave_h2_pool(#conn_data{pool_pid = PoolPid}) when is_pid(PoolPid) ->
gen_server:cast(PoolPid, {unregister_h2, self()});
leave_h2_pool(_Data) ->
ok.

h2_close_on_goaway(ErrorCode, #conn_data{h2_conn = H2Conn, h2_mon = H2Mon} = Data) ->
h2_close_on_goaway(ErrorCode, Data) ->
%% A GOAWAY means the peer will not service new streams on this connection.
%% AWS ALBs recycle connections this way, sending GOAWAY but keeping the
%% socket open for a drain window. Leaving the conn `connected` and pooled
Expand All @@ -4024,7 +4058,10 @@ h2_close_on_goaway(ErrorCode, #conn_data{h2_conn = H2Conn, h2_mon = H2Mon} = Dat
%% to recv_timeout. Tear the connection down and transition to `closed` (like
%% h2_on_closed/2): the pool then stops reusing it (h2_conn_usable requires
%% `connected`) and new requests dial a fresh connection.
{Replies, Data1} = collect_h2_aborts({goaway, ErrorCode}, Data),
h2_teardown({goaway, ErrorCode}, Data).

h2_teardown(Err, #conn_data{h2_conn = H2Conn, h2_mon = H2Mon} = Data) ->
{Replies, Data1} = collect_h2_aborts(Err, Data),
Data2 = cancel_all_h2_timers(Data1),
_ = case H2Mon of
undefined -> ok;
Expand Down
137 changes: 117 additions & 20 deletions test/hackney_http2_goaway_drain_tests.erl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@ goaway_drain_test_() ->
[{timeout, 30, fun accepted_streams_complete/0},
{timeout, 30, fun unaccepted_stream_fails_fast/0},
{timeout, 30, fun two_step_shutdown/0},
{timeout, 30, fun stalled_drain_ends_with_the_stream/0}]}.
{timeout, 30, fun stalled_drain_ends_with_the_stream/0},
{timeout, 30, fun accepted_upload_completes/0},
{timeout, 30, fun refused_upload_fails_on_next_send/0},
{timeout, 30, fun accepted_streamed_response_completes/0}]}.

setup() ->
_ = application:ensure_all_started(hackney),
Expand All @@ -41,7 +44,7 @@ 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} = start_server(fun([_First, Second]) -> Second end),
try
[R1, R2, R3] = concurrent_requests(Url, 3),
?assertEqual({ok, 200, <<"1">>}, R1),
Expand All @@ -54,7 +57,7 @@ accepted_streams_complete() ->
%% 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} = start_server(fun([First, _Second]) -> First end),
try
[R1, R2] = concurrent_requests(Url, 2),
?assertEqual({ok, 200, <<"1">>}, R1),
Expand All @@ -67,7 +70,7 @@ unaccepted_stream_fails_fast() ->
%% 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} = start_server(fun([First, _Second]) -> {two_step, First} end),
try
[R1, R2] = concurrent_requests(Url, 2),
?assertEqual({ok, 200, <<"1">>}, R1),
Expand All @@ -80,7 +83,7 @@ two_step_shutdown() ->
%% 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} = start_server(fun([First, _Second]) -> {never, First} end),
try
[R1, R2] = concurrent_requests(Url, 2, [{recv_timeout, 1000}]),
?assertEqual({error, timeout}, R1),
Expand All @@ -90,6 +93,86 @@ stalled_drain_ends_with_the_stream() ->
stop_server(Server)
end.

%% GOAWAY(last_stream_id = 1) while stream 1 still uploads its body: the
%% stream was accepted, so the rest of the body goes out and the response
%% arrives. The drained connection then closes.
accepted_upload_completes() ->
{Server, Port} = start_server(1, fun([First]) -> First end),
Conn = direct_conn(Port),
try
ok = hackney_conn:send_request_headers(Conn, <<"POST">>, <<"/">>, []),
ok = hackney_conn:send_body_chunk(Conn, <<"part">>),
ok = hackney_conn:finish_send_body(Conn),
{ok, 200, _, _} = hackney_conn:start_response(Conn),
?assertEqual({ok, <<"1">>}, hackney_conn:body(Conn)),
?assertEqual({ok, closed}, hackney_conn:get_state(Conn))
after
stop_conn(Conn),
stop_server(Server)
end.

%% A GET on stream 1 and an upload on stream 3, then GOAWAY(last_stream_id = 1):
%% the GET completes, the upload was refused and its next send says so. With
%% nothing left to drain the connection closes.
refused_upload_fails_on_next_send() ->
{Server, Port} = start_server(2, fun([First, _Second]) -> First end),
Conn = direct_conn(Port),
try
{ok, Ref} = hackney_conn:request_async(Conn, <<"GET">>, <<"/">>, [], <<>>,
false),
ok = hackney_conn:send_request_headers(Conn, <<"POST">>, <<"/">>, []),
?assertEqual({ok, 200, <<"1">>}, await_async(Ref, <<>>)),
?assertEqual({error, {goaway, no_error}},
hackney_conn:send_body_chunk(Conn, <<"part">>)),
?assertEqual({ok, closed}, hackney_conn:get_state(Conn))
after
stop_conn(Conn),
stop_server(Server)
end.

%% GOAWAY(last_stream_id = 1) while a streamed response waits on stream 1: the
%% stream was accepted, so its headers and body still arrive.
accepted_streamed_response_completes() ->
{Server, Port} = start_server(1, fun([First]) -> First end),
Conn = direct_conn(Port),
try
{ok, 200, _, _} = hackney_conn:request_streaming(Conn, <<"GET">>, <<"/">>,
[], <<>>),
?assertEqual({ok, <<"1">>}, hackney_conn:stream_body(Conn)),
?assertEqual(done, hackney_conn:stream_body(Conn)),
?assertEqual({ok, closed}, hackney_conn:get_state(Conn))
after
stop_conn(Conn),
stop_server(Server)
end.

stop_conn(Conn) ->
try hackney_conn:stop(Conn) catch _:_ -> ok end.

direct_conn(Port) ->
{ok, Conn} = hackney_conn_sup:start_conn(#{
host => "localhost",
port => Port,
transport => hackney_ssl,
connect_options => [{protocols, [http2]}],
ssl_options => [{insecure, true}, {verify, verify_none}]
}),
ok = hackney_conn:connect(Conn),
Conn.

await_async(Ref, Acc) ->
receive
{hackney_response, Ref, {status, Status, _}} ->
put(async_status, Status),
await_async(Ref, Acc);
{hackney_response, Ref, {headers, _}} -> await_async(Ref, Acc);
{hackney_response, Ref, done} -> {ok, get(async_status), Acc};
{hackney_response, Ref, {error, E}} -> {error, E};
{hackney_response, Ref, Bin} when is_binary(Bin) ->
await_async(Ref, <<Acc/binary, Bin/binary>>)
after 10000 -> {error, test_timeout}
end.

%% The first request registers the shared connection before the second checks
%% one out, so both streams share it. The server sends GOAWAY on the second and
%% waits 200ms before answering, so a third lands inside that drain.
Expand All @@ -104,9 +187,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}]}],
Expand All @@ -120,6 +200,13 @@ fetch(Url, Extra) ->
%%====================================================================

start_server(PickLastStreamId) ->
{Pid, Port} = start_server(2, PickLastStreamId),
Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]),
{Pid, Url}.

%% Hold the first Count streams of the first connection, then send GOAWAY with
%% the last_stream_id PickLastStreamId(HeldIds) returns.
start_server(Count, PickLastStreamId) ->
Certs = cert_dir(),
{ok, LSock} = ssl:listen(0,
[{certfile, filename:join(Certs, "server.pem")},
Expand All @@ -128,9 +215,8 @@ start_server(PickLastStreamId) ->
{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}.
Pid = spawn(fun() -> accept_loop(LSock, {hold, Count, PickLastStreamId}) end),
{Pid, Port}.

stop_server(Pid) ->
exit(Pid, kill).
Expand All @@ -151,7 +237,7 @@ serve(TSock, Mode) ->
{ok, Rest} ->
send(Sock, h2_frame:settings([])),
loop(Sock, Rest, #{enc => h2_hpack:new_context(), mode => Mode,
held => []});
held => [], ended => [], waiting => []});
_ -> ok
end;
_ -> ok
Expand Down Expand Up @@ -185,20 +271,31 @@ loop(Sock, Buf, St) ->
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) ->
handle(Sock, {headers, Sid, _B, true, _H}, #{mode := immediate} = St) ->
{continue, respond(Sock, Sid, St)};
handle(Sock, {headers, Sid, _B, _E, _H}, #{mode := {hold, Pick}, held := Held} = St) ->
handle(Sock, {headers, Sid, _B, EndStream, _H},
#{mode := {hold, Count, Pick}, held := Held, ended := Ended} = St) ->
Ended2 = case EndStream of true -> [Sid | Ended]; false -> Ended end,
case Held ++ [Sid] of
[First, Second] ->
LastStreamId = send_goaway(Sock, Pick(First, Second)),
%% The drain: the peer is told, then the accepted streams finish.
Held2 when length(Held2) =:= Count ->
LastStreamId = send_goaway(Sock, Pick(Held2)),
%% The drain: the peer is told, then the accepted streams finish,
%% an upload once its body has ended.
timer:sleep(200),
Accepted = [S || S <- Held2, S =< LastStreamId],
St2 = lists:foldl(fun(S, Acc) -> respond(Sock, S, Acc) end,
St#{held := []},
[S || S <- [First, Second], S =< LastStreamId]),
{continue, St2#{mode := draining}};
[S || S <- Accepted, lists:member(S, Ended2)]),
{continue, St2#{mode := draining, waiting := Accepted -- Ended2}};
Held2 ->
{continue, St#{held := Held2}}
{continue, St#{held := Held2, ended := Ended2}}
end;
handle(Sock, Data, #{waiting := Waiting} = St)
when element(1, Data) =:= data, element(4, Data) =:= true ->
Sid = element(2, Data),
case lists:member(Sid, Waiting) of
true -> {continue, respond(Sock, Sid, St#{waiting := Waiting -- [Sid]})};
false -> {continue, St#{ended := [Sid | maps:get(ended, St)]}}
end;
handle(_Sock, _Other, St) -> {continue, St}.

Expand Down
Loading