From d32f3c93a992cf53f1199632eae4a0c978c0aee6 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Sat, 3 Oct 2026 01:48:16 +0200 Subject: [PATCH] Drain a GOAWAY with a streamed body in flight An accepted upload or streamed response now finishes like any other accepted stream. A refused upload is marked and its caller gets {error, {goaway, Code}} on its next send or start_response. A failed h2 body send now fails the other streams on the connection instead of leaving them waiting. --- src/hackney_conn.erl | 73 +++++++++--- test/hackney_http2_goaway_drain_tests.erl | 137 ++++++++++++++++++---- 2 files changed, 172 insertions(+), 38 deletions(-) diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 58611889..a9f71a29 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -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, @@ -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) -> @@ -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) -> @@ -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, []). @@ -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 @@ -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; diff --git a/test/hackney_http2_goaway_drain_tests.erl b/test/hackney_http2_goaway_drain_tests.erl index 34a4b3d4..54d48143 100644 --- a/test/hackney_http2_goaway_drain_tests.erl +++ b/test/hackney_http2_goaway_drain_tests.erl @@ -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), @@ -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), @@ -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), @@ -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), @@ -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), @@ -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, <>) + 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. @@ -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}]}], @@ -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")}, @@ -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). @@ -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 @@ -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}.