Raise HTTP/2 receive windows and close TLS conns without waiting

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"},

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">>},

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),

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"]).

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, <<Buf/binary, Data/binary>>);
+read_preface(_Sock, Buf) ->
+    Size = byte_size(?PREFACE),
+    <<Preface:Size/binary, Rest/binary>> = 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,
+            <<Data:Size/binary, _/binary>> = 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 = <<Buf/binary, Data/binary>>});
+                {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 = <<Buf/binary, Data/binary>>}, 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"]).