diff --git a/rebar.config b/rebar.config index 192cb0cf..031a12a4 100644 --- a/rebar.config +++ b/rebar.config @@ -55,7 +55,7 @@ %% Pure Erlang QUIC + HTTP/3 stack {quic, "~>2.1.1"}, %% Pure Erlang HTTP/2 stack - {h2, "~>0.12.4"}, + {h2, "~>0.12.5"}, %% WebTransport client (HTTP/3 and HTTP/2) - powers the wt_* API {webtransport, "~>0.4.8"}, {idna, "~>7.1.0"}, diff --git a/rebar.lock b/rebar.lock index bdcd4110..6bfda5cf 100644 --- a/rebar.lock +++ b/rebar.lock @@ -1,6 +1,6 @@ {"1.2.0", [{<<"certifi">>,{pkg,<<"certifi">>,<<"2.17.0">>},0}, - {<<"h2">>,{pkg,<<"h2">>,<<"0.12.4">>},0}, + {<<"h2">>,{pkg,<<"h2">>,<<"0.12.5">>},0}, {<<"idna">>,{pkg,<<"idna">>,<<"7.1.0">>},0}, {<<"mimerl">>,{pkg,<<"mimerl">>,<<"1.5.0">>},0}, {<<"parse_trans">>,{pkg,<<"parse_trans">>,<<"3.4.2">>},0}, @@ -10,7 +10,7 @@ [ {pkg_hash,[ {<<"certifi">>, <<"835748414307E15E05B17D0E518190228CE648B08D569A5CC93A85A40F3E5C9B">>}, - {<<"h2">>, <<"F8CB769CCC17369D185717135353BEE1D50F2CD193A9E8CD998D88CE6CAFC51D">>}, + {<<"h2">>, <<"03D492B4CCE77466C4C4A2B3FD74B2BB16C1D78F9618BBFEB50EA493C005ACCF">>}, {<<"idna">>, <<"1067A13043538129602D2F2CE6899D8713125C7D19734AA557CE2E3EA55BD4F1">>}, {<<"mimerl">>, <<"F35ACA6F23242339B3666E0AC0702379E362B469D0AEA167F6CC713547E777ED">>}, {<<"parse_trans">>, <<"C352DDC1A0D5E54F9B1654D45F9C432EEF76F9CEA371C55DDFF769EF688FDB74">>}, @@ -19,7 +19,7 @@ {<<"webtransport">>, <<"3C340D7EA381E3F639ECE90378D85FB46B1DA4EECE17BBA3223051E833166501">>}]}, {pkg_hash_ext,[ {<<"certifi">>, <<"8122798A17F0293C80DAADA25D0F81C7F4D708C73FEF782C7C9B1950E26E4D21">>}, - {<<"h2">>, <<"238553579D2599E823F1F7B1CACB763362EE6E2DF8E8F3B0C8FC67DA982C0675">>}, + {<<"h2">>, <<"A4524BF80CAFB86E599710C1D5C9F63ACA0284607CB4D99FB53AE63FD55AA03F">>}, {<<"idna">>, <<"6AE959A025BF36DF61A8CAB8508D9654891B5426A84C44D82DEAFFD6DDF8C71F">>}, {<<"mimerl">>, <<"DB648CE065BAE14EA84CA8B5DD123F42F49417CEF693541110BF6F9E9BE9ECC4">>}, {<<"parse_trans">>, <<"4C25347DE3B7C35732D32E69AB43D1CEEE0BEAE3F3B3ADE1B59CBD3DD224D9CA">>}, diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 034445a8..3cd04213 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -129,6 +129,13 @@ %% opts out of blocking: sends then fail fast with {error, send_buffer_full} %% once the h2 per-stream buffer cap is hit. -define(SEND_TIMEOUT, 30000). +%% HTTP/2 receive windows. The protocol default (65,535 bytes for each stream +%% and for the connection) allows one 64 KB window in flight per round trip, +%% so over a 120 ms link a large download outlasts the server's stream timeout. +%% 8 MB per stream and 32 MB for the connection bring a download over a 250 ms +%% link to HTTP/1.1 speed. +-define(H2_STREAM_WINDOW, 8 * 1024 * 1024). +-define(H2_CONN_WINDOW, 32 * 1024 * 1024). %% Grace window for pooled hackney_conn in `closed` state, during which %% late-arriving calls race the pool DOWN cleanup and still get a proper %% error reply instead of exit:{normal, _}. See issue #836. @@ -826,11 +833,25 @@ terminate(_Reason, _State, #conn_data{socket = Socket, transport = Transport, %% Close socket (only still owned by us for HTTP/1.1; h2 lib closes its own) case {Socket, H2Conn} of {undefined, _} -> ok; - {_, undefined} -> Transport:close(Socket); + {_, undefined} -> close_on_terminate(Transport, Socket); _ -> ok end, ok. +%% ssl:close/1 sends close_notify, then waits up to 5 s for the server to close +%% the TCP connection (ssl:close/2 ignores its timeout on an established +%% connection): at least one round trip. The pool stops a conn with a 100 ms +%% budget and kills it past that, which logged a supervisor report for every +%% pooled TLS conn closed over a slow link. The conn reads nothing more, so +%% shutdown/2 sends close_notify and closes both directions, and close/1 then +%% returns without waiting. +close_on_terminate(hackney_ssl, Socket) -> + _ = hackney_ssl:shutdown(Socket, read_write), + _ = hackney_ssl:close(Socket), + ok; +close_on_terminate(Transport, Socket) -> + Transport:close(Socket). + code_change(_OldVsn, State, Data, _Extra) -> {ok, State, Data}. @@ -3249,7 +3270,9 @@ init_h2_after_upgrade(SslSocket, Data, From) -> start_h2_connection(Socket, Data, From, Origin) -> #conn_data{transport = Transport} = Data, - case h2_connection:start_link(client, Socket, self(), #{}) of + H2Opts = #{settings => #{initial_window_size => ?H2_STREAM_WINDOW}, + connection_window_size => ?H2_CONN_WINDOW}, + case h2_connection:start_link(client, Socket, self(), H2Opts) of {ok, H2Conn} -> %% Transfer socket ownership then activate handshake. _ = Transport:controlling_process(Socket, H2Conn), diff --git a/test/hackney_conn_tls_close_tests.erl b/test/hackney_conn_tls_close_tests.erl new file mode 100644 index 00000000..ab7555a8 --- /dev/null +++ b/test/hackney_conn_tls_close_tests.erl @@ -0,0 +1,83 @@ +%%% A TLS conn stops within the pool's 100 ms stop budget even when the server +%%% does not answer close_notify. ssl:close/1 waited up to 5 s for the server +%%% to close its side, so the pool killed the conn and every pooled TLS conn +%%% closed over a slow link logged a supervisor report (reason killed). +-module(hackney_conn_tls_close_tests). + +-include_lib("eunit/include/eunit.hrl"). + +tls_close_test_() -> + {setup, + fun() -> {ok, _} = application:ensure_all_started(hackney), ok end, + fun(_) -> ok end, + [{"stop/2 on a TLS conn exits normal when the server ignores close_notify", + {timeout, 30, fun t_stop_without_peer_close/0}}]}. + +t_stop_without_peer_close() -> + {Srv, Relay, Port} = start_server(), + try + {ok, Conn} = hackney:connect(hackney_ssl, "localhost", Port, + [{pool, false}, + {ssl_options, [{insecure, true}, + {verify, verify_none}]}]), + Relay ! freeze, + Ref = monitor(process, Conn), + ok = hackney_conn:stop(Conn, 100), + receive + {'DOWN', Ref, process, Conn, Reason} -> ?assertEqual(normal, Reason) + after 5000 -> + error(conn_still_up) + end + after + exit(Relay, kill), + exit(Srv, kill) + end. + +%% A TLS server behind a TCP relay. Once frozen, the relay drops what the +%% client sends: the client's close_notify never reaches the server, so the +%% server never closes its side, as with a peer over a slow link. +start_server() -> + Certs = cert_dir(), + {ok, LSock} = ssl:listen(0, + [{certfile, filename:join(Certs, "server.pem")}, + {keyfile, filename:join(Certs, "server.key")}, + {active, false}, {mode, binary}, {reuseaddr, true}]), + {ok, {_, SrvPort}} = ssl:sockname(LSock), + Srv = spawn(fun() -> + {ok, TSock} = ssl:transport_accept(LSock, 10000), + {ok, _Sock} = ssl:handshake(TSock, 5000), + receive after infinity -> ok end + end), + %% exit_on_close false: the client's half-close must not close the relay + %% socket, which would answer with the FIN the client waits for. + {ok, RLSock} = gen_tcp:listen(0, [binary, {active, false}, {reuseaddr, true}, + {exit_on_close, false}]), + {ok, Port} = inet:port(RLSock), + Relay = spawn(fun() -> + {ok, Client} = gen_tcp:accept(RLSock, 10000), + {ok, Upstream} = gen_tcp:connect("localhost", SrvPort, [binary, {active, once}]), + ok = inet:setopts(Client, [{active, once}]), + relay(Client, Upstream, false) + end), + {Srv, Relay, Port}. + +relay(Client, Upstream, Frozen) -> + receive + freeze -> + relay(Client, Upstream, true); + {tcp, Client, Data} -> + Frozen orelse gen_tcp:send(Upstream, Data), + ok = inet:setopts(Client, [{active, once}]), + relay(Client, Upstream, Frozen); + {tcp, Upstream, Data} -> + ok = gen_tcp:send(Client, Data), + ok = inet:setopts(Upstream, [{active, once}]), + relay(Client, Upstream, Frozen); + {tcp_closed, _} -> + relay(Client, Upstream, Frozen) + end. + +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_recv_window_tests.erl b/test/hackney_http2_recv_window_tests.erl new file mode 100644 index 00000000..587465a3 --- /dev/null +++ b/test/hackney_http2_recv_window_tests.erl @@ -0,0 +1,211 @@ +%%% Tests for HTTP/2 response downloads under receive flow control. +%%% +%%% hackney ran HTTP/2 with the protocol's 65,535 byte initial windows, so +%%% at most 64 KB of a response was in flight per round trip. Over a 120 ms +%%% link a 41 MB body took longer than the server's stream timeout and the +%%% server reset the stream: {error, {stream_error, internal_error}}. +%%% hackney now raises the stream window (SETTINGS_INITIAL_WINDOW_SIZE) and the +%%% connection window (h2 connection_window_size). +%%% +%%% The server below sends DATA only within the windows the client grants, +%%% sleeps ?RTT_MS every time it has to wait for a WINDOW_UPDATE (a round +%%% trip), and resets the stream with INTERNAL_ERROR past ?DEADLINE_MS. +-module(hackney_http2_recv_window_tests). + +-include_lib("eunit/include/eunit.hrl"). + +-define(PREFACE, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>). +-define(INITIAL_WINDOW, 65535). +-define(FRAME, 16384). +-define(BODY_SIZE, 50 * 1024 * 1024). +-define(RTT_MS, 50). +-define(DEADLINE_MS, 10000). +-define(INTERNAL_ERROR, 16#2). + +recv_window_test_() -> + {setup, + fun() -> + {ok, _} = application:ensure_all_started(hackney), + {ok, _} = application:ensure_all_started(h2), + ok + end, + fun(_) -> ok end, + [{"client grants more than 64 KB before the first request", + {timeout, 60, fun t_windows_granted/0}}, + {"50 MB body over a slow link completes before the server deadline", + {timeout, 120, fun t_large_body_slow_link/0}}]}. + +t_windows_granted() -> + {Srv, Port} = start_server(self(), 1024), + try + ?assertMatch({ok, 200, _, _}, + hackney:request(get, url(Port), [], <<>>, opts())), + #{stream_window := StreamWindow, conn_window := ConnWindow} = report(), + ?assert(StreamWindow >= 1024 * 1024), + ?assert(ConnWindow >= 1024 * 1024) + after + stop_server(Srv) + end. + +t_large_body_slow_link() -> + {Srv, Port} = start_server(self(), ?BODY_SIZE), + try + T0 = erlang:monotonic_time(millisecond), + Res = hackney:request(get, url(Port), [], <<>>, opts()), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assertMatch({ok, 200, _, _}, Res), + {ok, 200, _, Body} = Res, + ?assertEqual(?BODY_SIZE, byte_size(Body)), + ?assert(Elapsed < ?DEADLINE_MS) + after + stop_server(Srv) + end. + +opts() -> + [{pool, false}, + {protocols, [http2]}, + {recv_timeout, 60000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}]. + +url(Port) -> + iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]). + +report() -> + receive {recv_window_server, Report} -> Report + after 5000 -> error(no_report) + end. + +%%==================================================================== +%% Server +%%==================================================================== + +-record(srv, {sock, buf = <<>>, test, body_size, + stream_initial = ?INITIAL_WINDOW, conn = ?INITIAL_WINDOW, + conn_granted = ?INITIAL_WINDOW, streams = #{}}). + +start_server(TestPid, BodySize) -> + 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(LSock, TestPid, BodySize) end), + {Pid, Port}. + +stop_server(Pid) -> + exit(Pid, kill), + ok. + +accept(LSock, TestPid, BodySize) -> + {ok, TSock} = ssl:transport_accept(LSock, 10000), + {ok, Sock} = ssl:handshake(TSock, 5000), + Rest = read_preface(Sock, <<>>), + send(Sock, h2_frame:settings([])), + idle(#srv{sock = Sock, buf = Rest, test = TestPid, body_size = BodySize}). + +read_preface(Sock, Buf) when byte_size(Buf) < byte_size(?PREFACE) -> + {ok, Data} = ssl:recv(Sock, 0, 5000), + read_preface(Sock, <>); +read_preface(_Sock, Buf) -> + Size = byte_size(?PREFACE), + <> = Buf, + ?PREFACE = Preface, + Rest. + +%% Process frames until a request arrives, then answer it. +idle(S) -> + case next_frame(S, infinity) of + {ok, {headers, StreamId, _, _, _}, S1} -> + S1#srv.test ! {recv_window_server, + #{stream_window => S1#srv.stream_initial, + conn_window => S1#srv.conn_granted}}, + idle(respond(StreamId, S1)); + {ok, Frame, S1} -> + idle(handle(Frame, S1)) + end. + +respond(StreamId, #srv{sock = Sock, streams = Streams} = S) -> + {Block, _} = h2_hpack:encode([{<<":status">>, <<"200">>}], h2_hpack:new_context()), + send(Sock, h2_frame:headers(StreamId, Block, false)), + Deadline = erlang:monotonic_time(millisecond) + ?DEADLINE_MS, + S1 = S#srv{streams = Streams#{StreamId => S#srv.stream_initial}}, + send_body(StreamId, S#srv.body_size, Deadline, binary:copy(<<"x">>, ?FRAME), S1). + +send_body(_StreamId, 0, _Deadline, _Chunk, S) -> + S; +send_body(StreamId, Left, Deadline, Chunk, #srv{conn = Conn, streams = Streams} = S) -> + Size = lists:min([?FRAME, Left, Conn, maps:get(StreamId, Streams)]), + Now = erlang:monotonic_time(millisecond), + if + Now > Deadline -> + %% What a server does with a stream it cannot make progress on. + send(S#srv.sock, h2_frame:rst_stream(StreamId, ?INTERNAL_ERROR)), + S; + Size > 0 -> + Left1 = Left - Size, + <> = Chunk, + send(S#srv.sock, h2_frame:data(StreamId, Data, Left1 =:= 0)), + send_body(StreamId, Left1, Deadline, Chunk, + S#srv{conn = Conn - Size, + streams = Streams#{StreamId => maps:get(StreamId, Streams) - Size}}); + true -> + %% Window exhausted: one round trip before the update can land. + timer:sleep(?RTT_MS), + send_body(StreamId, Left, Deadline, Chunk, wait_window(S)) + end. + +wait_window(S) -> + case next_frame(S, 5000) of + {ok, {window_update, _, _} = Frame, S1} -> drain(handle(Frame, S1)); + {ok, Frame, S1} -> wait_window(handle(Frame, S1)); + timeout -> error(window_never_opened) + end. + +%% Apply every frame already received, so one round trip lands every update +%% the client sent during it. +drain(#srv{sock = Sock, buf = Buf} = S) -> + case h2_frame:decode(Buf) of + {ok, Frame, Rest} -> + drain(handle(Frame, S#srv{buf = Rest})); + {more, _} -> + case ssl:recv(Sock, 0, 0) of + {ok, Data} -> drain(S#srv{buf = <>}); + {error, timeout} -> S + end + end. + +handle({settings, Settings}, S) -> + send(S#srv.sock, h2_frame:settings_ack()), + S#srv{stream_initial = proplists:get_value(16#4, Settings, S#srv.stream_initial)}; +handle({window_update, 0, Inc}, S) -> + S#srv{conn = S#srv.conn + Inc, conn_granted = S#srv.conn_granted + Inc}; +handle({window_update, StreamId, Inc}, #srv{streams = Streams} = S) -> + S#srv{streams = Streams#{StreamId => maps:get(StreamId, Streams, 0) + Inc}}; +handle({ping, Data}, S) -> + send(S#srv.sock, h2_frame:ping_ack(Data)), + S; +handle(_Frame, S) -> + S. + +next_frame(#srv{sock = Sock, buf = Buf} = S, Timeout) -> + case h2_frame:decode(Buf) of + {ok, Frame, Rest} -> + {ok, Frame, S#srv{buf = Rest}}; + {more, _} -> + case ssl:recv(Sock, 0, Timeout) of + {ok, Data} -> next_frame(S#srv{buf = <>}, Timeout); + {error, timeout} -> timeout; + {error, Reason} -> exit({socket, Reason}) + end + end. + +send(Sock, Frame) -> + ok = ssl:send(Sock, h2_frame:encode(Frame)). + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]).