From 062e9281b63a86be564108c0437e0f74b0145299 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Thu, 16 Jul 2026 18:49:02 +0200 Subject: [PATCH 1/5] Block HTTP/2 request-body sends on flow control An HTTP/2 request body larger than the peer's flow control window failed with {error, send_buffer_full}: body sends used the non-blocking h2_connection:send_data/4, which buffers up to a cap instead of waiting for WINDOW_UPDATE. Request-body sends now use send_data/5 with #{block => Timeout}, parking the caller until the window opens. The new send_timeout option bounds the wait (default 30000 ms, infinity allowed, nonblock restores the old fail-fast behavior). Covers the whole-body send, streamed chunks, the end-stream frame and body producer funs. HTTP/1.1 and HTTP/3 paths are unchanged. Tested against a frame-level h2 server with a small initial window and delayed, split or withheld WINDOW_UPDATEs, plus a digest echo server for a body over the old 1 MB cap. --- NEWS.md | 14 ++ README.md | 3 +- guides/http2_guide.md | 18 ++ src/hackney.erl | 17 +- src/hackney_conn.erl | 69 ++++--- src/hackney_pool.erl | 2 + test/hackney_http2_flow_control_tests.erl | 213 ++++++++++++++++++++++ test/repro_h2_raw_server.erl | 28 +++ 8 files changed, 339 insertions(+), 25 deletions(-) create mode 100644 test/hackney_http2_flow_control_tests.erl diff --git a/NEWS.md b/NEWS.md index e91285fb..4660eeb3 100644 --- a/NEWS.md +++ b/NEWS.md @@ -1,5 +1,19 @@ # NEWS +unreleased +---------- + +### Fixed + +- HTTP/2 request bodies larger than the peer's flow control window no longer + fail with `{error, send_buffer_full}`. Body sends now block until the + server opens the window with WINDOW_UPDATE frames, bounded by the new + `send_timeout` request option (default 30000 ms, `infinity` allowed). If + the window never opens the request fails with `{error, timeout}` instead + of hanging. Pass `{send_timeout, nonblock}` to restore the previous + non-blocking behavior. Applies to whole-body and streamed HTTP/2 request + bodies; HTTP/1.1 and HTTP/3 are unchanged. + 4.6.1 - 2026-07-15 ------------------ diff --git a/README.md b/README.md index 0406d157..65d55e1b 100644 --- a/README.md +++ b/README.md @@ -258,7 +258,8 @@ hackney:get(URL, [], <<>>, [{follow_redirect, true}, {max_redirect, 5}]). ```erlang hackney:get(URL, [], <<>>, [ {connect_timeout, 5000}, %% Connection timeout - {recv_timeout, 30000} %% Response timeout + {recv_timeout, 30000}, %% Response timeout + {send_timeout, 30000} %% HTTP/2 body send timeout (flow control) ]). ``` diff --git a/guides/http2_guide.md b/guides/http2_guide.md index 4cef5e52..7dce1453 100644 --- a/guides/http2_guide.md +++ b/guides/http2_guide.md @@ -329,6 +329,24 @@ HTTP/2 has built-in flow control to prevent fast senders from overwhelming slow No configuration is needed for most use cases. +### Sending Large Bodies + +A request body larger than the server's flow control window cannot be sent in one shot: the send waits for the server to open the window with WINDOW_UPDATE frames. Hackney blocks the request until the body is fully handed to the connection, up to `send_timeout` (default 30000 ms). If the server never opens the window, the request fails with `{error, timeout}` instead of hanging. + +```erlang +%% Give a slow server more time to drain a large upload +hackney:post(URL, Headers, LargeBody, [{send_timeout, 120000}]). + +%% Wait forever +hackney:post(URL, Headers, LargeBody, [{send_timeout, infinity}]). + +%% Opt out of blocking: fail fast with {error, send_buffer_full} when the +%% body exceeds the window plus the connection's send buffer +hackney:post(URL, Headers, Body, [{send_timeout, nonblock}]). +``` + +The option applies to whole-body requests and to streamed bodies sent with `hackney:send_body/2`. HTTP/1.1 and HTTP/3 requests ignore it. + ## Error Handling HTTP/2 specific errors: diff --git a/src/hackney.erl b/src/hackney.erl index 6826b301..fe04a096 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -140,6 +140,7 @@ connect_direct(Transport, Host, Port, Options) -> transport => Transport, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), + send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []) }, @@ -288,6 +289,7 @@ try_new_h3_connection(Host, Port, Transport, Options, PoolHandler) -> transport => Transport, connect_timeout => ConnectTimeout, recv_timeout => proplists:get_value(recv_timeout, Options, 5000), + send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => SslOpts, pool_name => PoolName, @@ -499,6 +501,7 @@ start_conn_with_socket_internal(Host, Port, Transport, Socket, Options) -> socket => Socket, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), + send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []), no_reuse => NoReuse @@ -1358,10 +1361,16 @@ sync_request_with_redirect_body(ConnPid, Method, Path, HeadersList, FinalBody, false -> ReqOpts0 end, %% Pass recv_timeout through to the connection so it's applied per-request - ReqOpts = case proplists:get_value(recv_timeout, Options) of + ReqOpts2 = case proplists:get_value(recv_timeout, Options) of undefined -> ReqOpts1; RecvTimeout -> [{recv_timeout, RecvTimeout} | ReqOpts1] end, + %% Pass send_timeout through so HTTP/2 body sends blocked on flow control + %% use the caller's deadline + ReqOpts = case proplists:get_value(send_timeout, Options) of + undefined -> ReqOpts2; + SendTimeout -> [{send_timeout, SendTimeout} | ReqOpts2] + end, case hackney_conn:request(ConnPid, Method, Path, HeadersList, FinalBody, infinity, ReqOpts) of %% HTTP/2 returns body directly - handle 4-tuple first {ok, Status, RespHeaders, RespBody} when Status >= 301, Status =< 303; Status =:= 307; Status =:= 308 -> @@ -1568,10 +1577,14 @@ async_request(ConnPid, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowR {FinalHeaders, FinalBody} = encode_body(Headers, Body, []), HeadersList = hackney_headers:to_list(FinalHeaders), %% Build ReqOpts for recv_timeout (fix for issue #832) - ReqOpts = case proplists:get_value(recv_timeout, Options) of + ReqOpts0 = case proplists:get_value(recv_timeout, Options) of undefined -> []; RecvTimeout -> [{recv_timeout, RecvTimeout}] end, + ReqOpts = case proplists:get_value(send_timeout, Options) of + undefined -> ReqOpts0; + SendTimeout -> [{send_timeout, SendTimeout} | ReqOpts0] + end, %% Note: Issue #646 - ownership transfer to StreamTo (when different from caller) %% is handled atomically inside hackney_conn:do_request_async case hackney_conn:request_async(ConnPid, Method, Path, HeadersList, FinalBody, AsyncMode, StreamTo, FollowRedirect, ReqOpts) of diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 4995e282..d748a4a4 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -115,6 +115,11 @@ -define(CONNECT_TIMEOUT, 8000). -define(IDLE_TIMEOUT, infinity). +%% How long an HTTP/2 request-body send may wait on flow control +%% (WINDOW_UPDATE) before failing with {error, timeout}. The atom nonblock +%% 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). %% 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. @@ -144,6 +149,7 @@ %% Options connect_timeout = ?CONNECT_TIMEOUT :: timeout(), recv_timeout = ?RECV_TIMEOUT :: timeout(), + send_timeout = ?SEND_TIMEOUT :: timeout() | nonblock, idle_timeout = ?IDLE_TIMEOUT :: timeout(), connect_options = [] :: list(), ssl_options = [] :: list(), @@ -635,6 +641,7 @@ init([DefaultOwner, Opts]) -> socket = Socket, connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT), recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT), + send_timeout = maps:get(send_timeout, Opts, ?SEND_TIMEOUT), idle_timeout = maps:get(idle_timeout, Opts, ?IDLE_TIMEOUT), connect_options = maps:get(connect_options, Opts, []), ssl_options = maps:get(ssl_options, Opts, []), @@ -944,7 +951,8 @@ connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_d %% HTTP/2 request - use h2_machine (1xx not applicable for HTTP/2) %% Allow recv_timeout to be overridden per-request (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), - NewData = Data#conn_data{recv_timeout = RecvTimeout}, + SendTimeout = proplists:get_value(send_timeout, ReqOpts, Data#conn_data.send_timeout), + NewData = Data#conn_data{recv_timeout = RecvTimeout, send_timeout = SendTimeout}, do_h2_request(From, Method, Path, Headers, Body, NewData); connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_data{protocol = http3} = Data) -> @@ -1013,7 +1021,8 @@ connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request with ReqOpts (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), - NewData = Data#conn_data{recv_timeout = RecvTimeout}, + SendTimeout = proplists:get_value(send_timeout, ReqOpts, Data#conn_data.send_timeout), + NewData = Data#conn_data{recv_timeout = RecvTimeout, send_timeout = SendTimeout}, do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, NewData); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, _FollowRedirect, ReqOpts}, #conn_data{protocol = http3} = Data) -> @@ -1283,14 +1292,16 @@ streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol = 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} = Data, + #conn_data{h2_conn = H2Conn, h2_stream_id = StreamId, + send_timeout = SendTimeout} = Data, Result = case BodyData of Fun when is_function(Fun, 0) -> - stream_body_fun_h2(H2Conn, StreamId, Fun); + stream_body_fun_h2(H2Conn, StreamId, Fun, SendTimeout); {Fun, State} when is_function(Fun, 1) -> - stream_body_fun_h2(H2Conn, StreamId, {Fun, State}); + stream_body_fun_h2(H2Conn, StreamId, {Fun, State}, SendTimeout); _ -> - h2_send_data(H2Conn, StreamId, iolist_to_binary(BodyData), false) + h2_send_data(H2Conn, StreamId, iolist_to_binary(BodyData), false, + SendTimeout) end, case Result of ok -> @@ -1329,8 +1340,9 @@ streaming_body({call, From}, finish_send_body, #conn_data{protocol = http3} = Da streaming_body({call, From}, finish_send_body, #conn_data{protocol = http2} = Data) -> %% HTTP/2 - close the request stream with an empty END_STREAM DATA frame. - #conn_data{h2_conn = H2Conn, h2_stream_id = StreamId} = Data, - case h2_send_data(H2Conn, StreamId, <<>>, true) of + #conn_data{h2_conn = H2Conn, h2_stream_id = StreamId, + send_timeout = SendTimeout} = Data, + case h2_send_data(H2Conn, StreamId, <<>>, true, SendTimeout) of ok -> {keep_state, Data, [{reply, From, ok}]}; {error, Reason} -> @@ -2962,7 +2974,7 @@ do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, {async, Ref, StreamTo, AsyncMode}, Data). do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> - #conn_data{h2_conn = H2Conn} = Data, + #conn_data{h2_conn = H2Conn, send_timeout = SendTimeout} = Data, {MethodBin, PathBin, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), BodyBin = case Body of @@ -2979,7 +2991,10 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> _ -> case h2_connection:send_request_headers(H2Conn, H2Headers, false) of {ok, SId} -> - case h2_connection:send_data(H2Conn, SId, BodyBin, true) of + %% Block on flow control so bodies larger than the + %% peer's window wait for WINDOW_UPDATE instead of + %% failing with send_buffer_full. + case h2_send_data(H2Conn, SId, BodyBin, true, SendTimeout) of ok -> {ok, SId}; {error, _} = E1 -> E1 end; @@ -3048,25 +3063,34 @@ do_h2_send_headers(From, Method, Path, Headers, Data) -> {keep_state_and_data, [{reply, From, {error, Reason}}]} end. -%% @private Non-blocking h2 send_data, matching the one-shot path. The -%% h2_connection buffers beyond the peer's flow-control window and drains as -%% WINDOW_UPDATEs arrive (returning {error, send_buffer_full} only past its -%% per-stream cap). Normalises a dead h2_connection exit to an error. -h2_send_data(H2Conn, StreamId, Bin, EndStream) -> +%% @private h2 send_data with flow-control blocking. With a timeout (or +%% infinity) the caller parks in h2_connection until the peer's WINDOW_UPDATEs +%% drain the send buffer, so bodies larger than the flow-control window +%% complete instead of failing; {error, timeout} if the window never opens. +%% With nonblock, keeps the historical non-blocking behavior: h2_connection +%% buffers beyond the window and returns {error, send_buffer_full} past its +%% per-stream cap. Normalises a dead h2_connection exit to an error. +h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) -> try - h2_connection:send_data(H2Conn, StreamId, Bin, EndStream) + do_h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) catch exit:{ExitReason, _} -> {error, {closed, ExitReason}}; exit:ExitReason -> {error, {closed, ExitReason}} end. +do_h2_send_data(H2Conn, StreamId, Bin, EndStream, nonblock) -> + h2_connection:send_data(H2Conn, StreamId, Bin, EndStream); +do_h2_send_data(H2Conn, StreamId, Bin, EndStream, Timeout) -> + h2_connection:send_data(H2Conn, StreamId, Bin, EndStream, #{block => Timeout}). + %% @private Drain a body-producer fun, sending each chunk as a non-final h2 DATA %% frame. Mirrors stream_body_fun/3 (HTTP/1.1) and stream_body_fun_h3/3. -stream_body_fun_h2(H2Conn, StreamId, Fun) when is_function(Fun, 0) -> +stream_body_fun_h2(H2Conn, StreamId, Fun, SendTimeout) when is_function(Fun, 0) -> case Fun() of {ok, Data} -> - case h2_send_data(H2Conn, StreamId, iolist_to_binary(Data), false) of - ok -> stream_body_fun_h2(H2Conn, StreamId, Fun); + case h2_send_data(H2Conn, StreamId, iolist_to_binary(Data), false, + SendTimeout) of + ok -> stream_body_fun_h2(H2Conn, StreamId, Fun, SendTimeout); Error -> Error end; eof -> @@ -3074,11 +3098,12 @@ stream_body_fun_h2(H2Conn, StreamId, Fun) when is_function(Fun, 0) -> {error, _} = Error -> Error end; -stream_body_fun_h2(H2Conn, StreamId, {Fun, State}) when is_function(Fun, 1) -> +stream_body_fun_h2(H2Conn, StreamId, {Fun, State}, SendTimeout) when is_function(Fun, 1) -> case Fun(State) of {ok, Data, NewState} -> - case h2_send_data(H2Conn, StreamId, iolist_to_binary(Data), false) of - ok -> stream_body_fun_h2(H2Conn, StreamId, {Fun, NewState}); + case h2_send_data(H2Conn, StreamId, iolist_to_binary(Data), false, + SendTimeout) of + ok -> stream_body_fun_h2(H2Conn, StreamId, {Fun, NewState}, SendTimeout); Error -> Error end; eof -> diff --git a/src/hackney_pool.erl b/src/hackney_pool.erl index d96e07a0..c39fa316 100644 --- a/src/hackney_pool.erl +++ b/src/hackney_pool.erl @@ -994,6 +994,7 @@ start_connection(Key, Owner, Opts, State) -> start_connection(Host, Port, Transport, Owner, Opts, State) -> ConnectTimeout = proplists:get_value(connect_timeout, Opts, 8000), RecvTimeout = proplists:get_value(recv_timeout, Opts, infinity), + SendTimeout = proplists:get_value(send_timeout, Opts, 30000), IdleTimeout = State#state.keepalive_timeout, SslOpts = proplists:get_value(ssl_options, Opts, []), ConnectOpts = proplists:get_value(connect_options, Opts, []), @@ -1004,6 +1005,7 @@ start_connection(Host, Port, Transport, Owner, Opts, State) -> transport => Transport, connect_timeout => ConnectTimeout, recv_timeout => RecvTimeout, + send_timeout => SendTimeout, idle_timeout => IdleTimeout, ssl_options => SslOpts, connect_options => ConnectOpts, diff --git a/test/hackney_http2_flow_control_tests.erl b/test/hackney_http2_flow_control_tests.erl new file mode 100644 index 00000000..f845b71b --- /dev/null +++ b/test/hackney_http2_flow_control_tests.erl @@ -0,0 +1,213 @@ +%%% Tests for HTTP/2 request-body sends under flow control. +%%% +%%% Request bodies larger than the peer's flow-control window used to fail +%%% with {error, send_buffer_full}: hackney called the non-blocking +%%% h2_connection:send_data/4, which caps buffering at 1 MB and never waits +%%% for WINDOW_UPDATE. hackney now uses send_data/5 with #{block => Timeout} +%%% (the send_timeout option), parking the caller until the window opens. +%%% +%%% Uses two servers: +%%% - repro_h2_raw_server: frame-level server with a small initial window and +%%% a window_update knob, so WINDOW_UPDATE emission is delayed, split, or +%%% withheld entirely. +%%% - the h2 dep's high-level echo server, to assert byte-for-byte delivery of +%%% a body larger than the old 1 MB non-blocking buffer cap. +-module(hackney_http2_flow_control_tests). + +-include_lib("eunit/include/eunit.hrl"). + +-define(SMALL_WINDOW, 16384). + +%%==================================================================== +%% Fixtures +%%==================================================================== + +raw_server_test_() -> + [{"large body completes across delayed split WINDOW_UPDATEs", + {timeout, 120, fun t_large_body_completes/0}}, + {"streamed chunks complete across delayed split WINDOW_UPDATEs", + {timeout, 120, fun t_streaming_large_small_window/0}}, + {"window never opens: bounded {error, timeout}, no hang", + {timeout, 60, fun t_window_never_opens_times_out/0}}, + {"send_timeout nonblock keeps the old fail-fast behavior", + {timeout, 60, fun t_nonblock_opt_out/0}}]. + +echo_server_test_() -> + {setup, + fun setup_echo/0, + fun cleanup_echo/1, + fun(Ctx) -> + [{"body over the 1 MB non-blocking cap is delivered intact", + {timeout, 120, fun() -> t_body_over_buffer_cap_echoed(Ctx) end}}] + end}. + +%%==================================================================== +%% Raw-server tests (controlled flow control) +%%==================================================================== + +%% The server advertises a 16 KB stream window and never consumes more than +%% one WINDOW_UPDATE increment at a time, 5 ms apart, so a 300 KB upload is +%% forced through many park/flush cycles on both the stream and connection +%% windows. +t_large_body_completes() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, ?SMALL_WINDOW}], + window_update => {auto, ?SMALL_WINDOW, 5}, + body_size => 128 + }), + Body = binary:copy(<<"x">>, 300 * 1024), + try + Res = hackney:request(post, url(Srv, <<"/">>), [], Body, opts([])), + ?assertMatch({ok, 200, _, _}, Res) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% Same constrained window, driven through the streaming API +%% (send_body/finish_send_body) instead of a one-shot body. +t_streaming_large_small_window() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, ?SMALL_WINDOW}], + window_update => {auto, 8192, 2}, + body_size => 128 + }), + Body = binary:copy(<<"y">>, 300 * 1024), + try + {ok, ConnPid} = hackney:request(post, url(Srv, <<"/">>), [], stream, opts([])), + ok = send_in_chunks(ConnPid, Body, 32768), + ok = hackney:finish_send_body(ConnPid), + {ok, 200, _RespHeaders, ConnPid} = hackney:start_response(ConnPid), + {ok, _RespBody} = hackney:body(ConnPid) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% A server that never sends WINDOW_UPDATE must yield {error, timeout} after +%% send_timeout, not a hang and not send_buffer_full. +t_window_never_opens_times_out() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Body = binary:copy(<<"z">>, 200 * 1024), + try + T0 = erlang:monotonic_time(millisecond), + Res = hackney:request(post, url(Srv, <<"/">>), [], Body, + opts([{send_timeout, 500}])), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assertEqual({error, timeout}, Res), + ?assert(Elapsed < 5000) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% {send_timeout, nonblock} restores the historical non-blocking send: a body +%% past the h2 per-stream buffer cap fails fast with send_buffer_full. +t_nonblock_opt_out() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Body = binary:copy(<<"w">>, 1536 * 1024), + try + Res = hackney:request(post, url(Srv, <<"/">>), [], Body, + opts([{send_timeout, nonblock}])), + ?assertEqual({error, send_buffer_full}, Res) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%%==================================================================== +%% Echo-server test (well-behaved peer, body over the old cap) +%%==================================================================== + +setup_echo() -> + start_apps(), + Certs = cert_dir(), + Handler = fun server_handler/5, + {ok, Server} = h2:start_server(0, #{ + cert => filename:join(Certs, "server.pem"), + key => filename:join(Certs, "server.key"), + handler => Handler, + settings => #{max_concurrent_streams => unlimited} + }), + Port = h2:server_port(Server), + #{server => Server, port => Port}. + +cleanup_echo(#{server := Server}) -> + try h2:stop_server(Server) catch _:_ -> ok end, + ok. + +%% Reply with size + SHA-256 of the request body rather than echoing it: the +%% server-side h2:send_data is itself non-blocking and would hit its own 1 MB +%% cap on a 1.5 MB response. The digest still proves byte-for-byte delivery. +server_handler(Conn, Sid, _Method, _Path, _Headers) -> + ok = h2:set_stream_handler(Conn, Sid, self()), + Body = recv_request_body(Conn, Sid, <<>>), + Digest = binary:encode_hex(crypto:hash(sha256, Body)), + Size = integer_to_binary(byte_size(Body)), + ok = h2:send_response(Conn, Sid, 200, + [{<<"content-type">>, <<"text/plain">>}]), + ok = h2:send_data(Conn, Sid, <>, true). + +recv_request_body(Conn, Sid, Acc) -> + receive + {h2, Conn, {data, Sid, Data, true}} -> + <>; + {h2, Conn, {data, Sid, Data, false}} -> + recv_request_body(Conn, Sid, <>) + after 30000 -> + Acc + end. + +%% 1.5 MB in a single one-shot body: over the h2 non-blocking 1 MB buffer cap, +%% so the old path failed with send_buffer_full even against a well-behaved +%% server. The blocking send must deliver it byte-for-byte. +t_body_over_buffer_cap_echoed(#{port := Port}) -> + Payload = crypto:strong_rand_bytes(1536 * 1024), + Expected = <<(integer_to_binary(byte_size(Payload)))/binary, ":", + (binary:encode_hex(crypto:hash(sha256, Payload)))/binary>>, + Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/echo">>]), + {ok, 200, _RespHeaders, RespBody} = + hackney:request(post, Url, [], Payload, opts([{recv_timeout, 30000}])), + ?assertEqual(Expected, RespBody). + +%%==================================================================== +%% Helpers +%%==================================================================== + +start_apps() -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + ok. + +opts(Extra) -> + Extra ++ + [{pool, false}, + {protocols, [http2]}, + {recv_timeout, 15000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}]. + +url(Srv, Path) -> + Port = repro_h2_raw_server:port(Srv), + iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), Path]). + +send_in_chunks(_ConnPid, <<>>, _Size) -> + ok; +send_in_chunks(ConnPid, Bin, Size) when byte_size(Bin) =< Size -> + hackney:send_body(ConnPid, Bin); +send_in_chunks(ConnPid, Bin, Size) -> + <> = Bin, + ok = hackney:send_body(ConnPid, Chunk), + send_in_chunks(ConnPid, Rest, Size). + +cert_dir() -> + BeamDir = filename:dirname(code:which(?MODULE)), + Root = filename:join([BeamDir, "..", "..", "..", "..", ".."]), + filename:join([filename:absname(Root), "test", "certs"]). diff --git a/test/repro_h2_raw_server.erl b/test/repro_h2_raw_server.erl index 302dc70d..fb4e9116 100644 --- a/test/repro_h2_raw_server.erl +++ b/test/repro_h2_raw_server.erl @@ -14,6 +14,13 @@ %%% quirk :: none | settings | {settings, [{atom(),int()}]} %%% | ping | window_update %%% quirk_when :: after_each | after_first | mid_first | mid_each +%%% window_update :: none | {auto, Increment, DelayMs} +%%% none (default): never open the request-body window; the +%%% client's upload stalls once the initial window is spent. +%%% {auto, Inc, Delay}: on each request DATA frame, release +%%% the consumed bytes back in Inc-sized WINDOW_UPDATE pairs +%%% (connection + stream), sleeping Delay ms before each, so +%%% a large upload drains across several delayed updates. -module(repro_h2_raw_server). -export([start/1, stop/1, port/1]). @@ -111,6 +118,15 @@ handle_frame(_Sock, {ping_ack, _}, St, _Knobs) -> {continue, St}; handle_frame(_Sock, {window_update, _, _}, St, _Knobs) -> {continue, St}; +handle_frame(Sock, {data, StreamId, _Body, _EndStream, PayloadLen}, St, Knobs) -> + %% Request-body DATA: never decoded, only flow-control accounted per the + %% window_update knob. + case maps:get(window_update, Knobs, none) of + none -> ok; + {auto, Inc, DelayMs} -> + release_window(Sock, StreamId, PayloadLen, Inc, DelayMs) + end, + {continue, St}; handle_frame(_Sock, {goaway, _, _, _}, _St, _Knobs) -> stop; handle_frame(_Sock, {rst_stream, _, _}, St, _Knobs) -> @@ -191,6 +207,18 @@ emit_quirk(Sock, {settings, L}) -> send(Sock, h2_frame:settings(L)); emit_quirk(Sock, ping) -> send(Sock, h2_frame:ping(<<0,0,0,0,0,0,0,7>>)); emit_quirk(Sock, window_update) -> send(Sock, h2_frame:window_update(0, 1000)). +%% Release consumed request-body bytes back to the client in Inc-sized +%% WINDOW_UPDATE pairs (connection stream 0 + request stream), with a delay +%% before each, forcing the client through several park/flush cycles. +release_window(_Sock, _StreamId, 0, _Inc, _DelayMs) -> + ok; +release_window(Sock, StreamId, Remaining, Inc, DelayMs) -> + Chunk = min(Inc, Remaining), + case DelayMs of 0 -> ok; _ -> timer:sleep(DelayMs) end, + send(Sock, h2_frame:window_update(0, Chunk)), + send(Sock, h2_frame:window_update(StreamId, Chunk)), + release_window(Sock, StreamId, Remaining - Chunk, Inc, DelayMs). + send(Sock, FrameData) -> ssl:send(Sock, h2_frame:encode(FrameData)). From 66f3e458c864192d3122c8b379f7b21f7d7d09be Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Thu, 16 Jul 2026 19:01:58 +0200 Subject: [PATCH 2/5] Cover the h2 fun-producer and timeout send paths Add flow-control tests for body-producer funs (arity 0 and {fun, State}), a streamed chunk hitting send_timeout, and the async request path with send_timeout set. Every changed line of the blocking-send fix is now exercised. --- test/hackney_http2_flow_control_tests.erl | 79 +++++++++++++++++++++++ 1 file changed, 79 insertions(+) diff --git a/test/hackney_http2_flow_control_tests.erl b/test/hackney_http2_flow_control_tests.erl index f845b71b..9fbdfd39 100644 --- a/test/hackney_http2_flow_control_tests.erl +++ b/test/hackney_http2_flow_control_tests.erl @@ -27,8 +27,14 @@ raw_server_test_() -> {timeout, 120, fun t_large_body_completes/0}}, {"streamed chunks complete across delayed split WINDOW_UPDATEs", {timeout, 120, fun t_streaming_large_small_window/0}}, + {"body-producer funs complete across delayed split WINDOW_UPDATEs", + {timeout, 120, fun t_streaming_fun_bodies/0}}, + {"streamed send hits send_timeout when the window never opens", + {timeout, 60, fun t_streaming_send_times_out/0}}, {"window never opens: bounded {error, timeout}, no hang", {timeout, 60, fun t_window_never_opens_times_out/0}}, + {"async request send hits send_timeout when the window never opens", + {timeout, 60, fun t_async_send_times_out/0}}, {"send_timeout nonblock keeps the old fail-fast behavior", {timeout, 60, fun t_nonblock_opt_out/0}}]. @@ -84,6 +90,61 @@ t_streaming_large_small_window() -> repro_h2_raw_server:stop(element(1, Srv)) end. +%% Same constrained window, driven through the body-producer fun forms of +%% send_body/2: an arity-0 fun and a {fun/1, State} pair, each emitting more +%% than the window in total. +t_streaming_fun_bodies() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, ?SMALL_WINDOW}], + window_update => {auto, ?SMALL_WINDOW, 2}, + body_size => 128 + }), + Chunk = binary:copy(<<"f">>, 32768), + C = counters:new(1, []), + Fun0 = fun() -> + case counters:get(C, 1) of + N when N < 4 -> counters:add(C, 1, 1), {ok, Chunk}; + _ -> eof + end + end, + Fun1 = {fun(N) when N < 4 -> {ok, Chunk, N + 1}; + (_) -> eof + end, 0}, + try + {ok, ConnPid} = hackney:request(post, url(Srv, <<"/">>), [], stream, opts([])), + ok = hackney:send_body(ConnPid, Fun0), + ok = hackney:send_body(ConnPid, Fun1), + ok = hackney:finish_send_body(ConnPid), + {ok, 200, _RespHeaders, ConnPid} = hackney:start_response(ConnPid), + {ok, _RespBody} = hackney:body(ConnPid) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% Streaming picks send_timeout up from the request options at connect time +%% (there is no per-chunk option): a chunk larger than a never-opening window +%% must fail with {error, timeout}, closing the connection. +t_streaming_send_times_out() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Chunk = binary:copy(<<"t">>, 200 * 1024), + try + {ok, ConnPid} = hackney:request(post, url(Srv, <<"/">>), [], stream, + opts([{send_timeout, 500}])), + T0 = erlang:monotonic_time(millisecond), + Res = hackney:send_body(ConnPid, Chunk), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assertEqual({error, timeout}, Res), + ?assert(Elapsed < 5000) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + %% A server that never sends WINDOW_UPDATE must yield {error, timeout} after %% send_timeout, not a hang and not send_buffer_full. t_window_never_opens_times_out() -> @@ -105,6 +166,24 @@ t_window_never_opens_times_out() -> repro_h2_raw_server:stop(element(1, Srv)) end. +%% The async request path sends the body before returning the ref, so a +%% never-opening window surfaces {error, timeout} from the request call. +t_async_send_times_out() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Body = binary:copy(<<"a">>, 200 * 1024), + try + Res = hackney:request(post, url(Srv, <<"/">>), [], Body, + opts([async, {send_timeout, 500}])), + ?assertEqual({error, timeout}, Res) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + %% {send_timeout, nonblock} restores the historical non-blocking send: a body %% past the h2 per-stream buffer cap fails fast with send_buffer_full. t_nonblock_opt_out() -> From dc37ed1ec8c8159966b1d89fca3945fe6013fec3 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Thu, 16 Jul 2026 23:15:13 +0200 Subject: [PATCH 3/5] Address review findings on the h2 blocking-send change A failed body send now RST_STREAM(CANCEL)s the abandoned stream from h2_send_data, so its buffered body no longer lingers on a shared connection. This also covers the nonblock send_buffer_full case. send_timeout no longer leaks between pooled requests: the per-request value is passed along instead of stored, and the connection default is no longer seeded from the first checkout's options. Streamed requests now forward send_timeout through send_request_headers/5 and keep it in a per-stream field set at every stream start, so a reused pooled connection applies the caller's deadline. The new async coverage test exposed a pre-existing bug: the h2 stream entry stored the gen_statem call reference where the async delivery clauses expect the stream_to pid, so every async HTTP/2 response was silently dropped. The entry now stores the stream owner. Regression tests: RST_STREAM observed on the wire after a timed-out send, a 25 ms send_timeout followed by a default request on the same pooled conn, a streamed request with send_timeout on a reused conn, and the direct hackney_conn async API forms. --- NEWS.md | 11 +- src/hackney.erl | 13 +- src/hackney_conn.erl | 88 +++++++++---- src/hackney_pool.erl | 2 - test/hackney_http2_flow_control_tests.erl | 145 +++++++++++++++++++++- test/repro_h2_raw_server.erl | 8 +- 6 files changed, 232 insertions(+), 35 deletions(-) diff --git a/NEWS.md b/NEWS.md index 4660eeb3..1712b699 100644 --- a/NEWS.md +++ b/NEWS.md @@ -10,9 +10,14 @@ unreleased server opens the window with WINDOW_UPDATE frames, bounded by the new `send_timeout` request option (default 30000 ms, `infinity` allowed). If the window never opens the request fails with `{error, timeout}` instead - of hanging. Pass `{send_timeout, nonblock}` to restore the previous - non-blocking behavior. Applies to whole-body and streamed HTTP/2 request - bodies; HTTP/1.1 and HTTP/3 are unchanged. + of hanging, and the abandoned stream is reset (RST_STREAM) so its buffered + body does not linger on a shared connection. Pass `{send_timeout, nonblock}` + to restore the previous non-blocking behavior. Applies to whole-body and + streamed HTTP/2 request bodies; HTTP/1.1 and HTTP/3 are unchanged. +- HTTP/2 async requests now deliver their response messages. The stream + entry stored the internal call reference where the delivery code expected + the `stream_to` pid, so every async HTTP/2 response was silently dropped + and the caller never received `{hackney_response, Ref, ...}` messages. 4.6.1 - 2026-07-15 ------------------ diff --git a/src/hackney.erl b/src/hackney.erl index fe04a096..052e9642 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -140,7 +140,6 @@ connect_direct(Transport, Host, Port, Options) -> transport => Transport, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), - send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []) }, @@ -289,7 +288,6 @@ try_new_h3_connection(Host, Port, Transport, Options, PoolHandler) -> transport => Transport, connect_timeout => ConnectTimeout, recv_timeout => proplists:get_value(recv_timeout, Options, 5000), - send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => SslOpts, pool_name => PoolName, @@ -501,7 +499,6 @@ start_conn_with_socket_internal(Host, Port, Transport, Socket, Options) -> socket => Socket, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), - send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []), no_reuse => NoReuse @@ -1339,8 +1336,14 @@ sync_request_with_redirect(ConnPid, Method, Path, Headers, Body, WithBody, Optio %% Check if this is a streaming body request case FinalBody of stream -> - %% For streaming body, just send headers and return immediately - case hackney_conn:send_request_headers(ConnPid, Method, Path, HeadersList) of + %% For streaming body, send headers and return immediately. Forward + %% send_timeout so the body chunks use the caller's deadline even on a + %% reused pooled connection. + ReqOpts = case proplists:get_value(send_timeout, Options) of + undefined -> []; + SendTimeout -> [{send_timeout, SendTimeout}] + end, + case hackney_conn:send_request_headers(ConnPid, Method, Path, HeadersList, ReqOpts) of ok -> {ok, ConnPid}; {error, Reason} -> {error, Reason} end; diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index d748a4a4..6ad11c8d 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -39,6 +39,7 @@ stream_body/1, %% Streaming body (request body) send_request_headers/4, + send_request_headers/5, send_body_chunk/2, finish_send_body/1, start_response/1, @@ -150,6 +151,11 @@ connect_timeout = ?CONNECT_TIMEOUT :: timeout(), recv_timeout = ?RECV_TIMEOUT :: timeout(), send_timeout = ?SEND_TIMEOUT :: timeout() | nonblock, + %% Effective send_timeout for the streaming-body request in flight, set at + %% every stream start (send_headers) so a per-request override never leaks + %% into later requests. The connection-level default above is never + %% mutated per-request. + req_send_timeout :: timeout() | nonblock | undefined, idle_timeout = ?IDLE_TIMEOUT :: timeout(), connect_options = [] :: list(), ssl_options = [] :: list(), @@ -363,8 +369,14 @@ request_streaming(Pid, Method, Path, Headers, Body) -> %% then start_response/1 to receive the response. -spec send_request_headers(pid(), binary(), binary(), list()) -> ok | {error, term()}. send_request_headers(Pid, Method, Path, Headers) -> + send_request_headers(Pid, Method, Path, Headers, []). + +%% @doc Like send_request_headers/4 with per-request options. Currently only +%% send_timeout is used (HTTP/2 flow-control deadline for the body chunks). +-spec send_request_headers(pid(), binary(), binary(), list(), list()) -> ok | {error, term()}. +send_request_headers(Pid, Method, Path, Headers, ReqOpts) -> case valid_request_target(Path) of - ok -> safe_call(Pid, {send_headers, Method, Path, Headers}, infinity); + ok -> safe_call(Pid, {send_headers, Method, Path, Headers, ReqOpts}, infinity); Err -> Err end. @@ -951,9 +963,11 @@ connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_d %% HTTP/2 request - use h2_machine (1xx not applicable for HTTP/2) %% Allow recv_timeout to be overridden per-request (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), + %% send_timeout is passed along, never stored: a per-request override must + %% not leak into later requests on a pooled connection. SendTimeout = proplists:get_value(send_timeout, ReqOpts, Data#conn_data.send_timeout), - NewData = Data#conn_data{recv_timeout = RecvTimeout, send_timeout = SendTimeout}, - do_h2_request(From, Method, Path, Headers, Body, NewData); + NewData = Data#conn_data{recv_timeout = RecvTimeout}, + do_h2_request(From, Method, Path, Headers, Body, SendTimeout, NewData); connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 request - use hackney_h3 (1xx not applicable for HTTP/3) @@ -964,11 +978,13 @@ connected({call, From}, {request, Method, Path, Headers, Body, ReqOpts}, #conn_d connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request - do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, false, Data); + do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, false, + Data#conn_data.send_timeout, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request with redirect option - do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, Data); + do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, + Data#conn_data.send_timeout, Data); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 async request @@ -1021,9 +1037,10 @@ connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, ReqOpts}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 async request with ReqOpts (fix for issue #832) RecvTimeout = proplists:get_value(recv_timeout, ReqOpts, Data#conn_data.recv_timeout), + %% send_timeout passed along, never stored (no leak across pooled requests) SendTimeout = proplists:get_value(send_timeout, ReqOpts, Data#conn_data.send_timeout), - NewData = Data#conn_data{recv_timeout = RecvTimeout, send_timeout = SendTimeout}, - do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, NewData); + NewData = Data#conn_data{recv_timeout = RecvTimeout}, + do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, SendTimeout, NewData); connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, StreamTo, _FollowRedirect, ReqOpts}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 async request with ReqOpts (fix for issue #832, redirect not yet implemented for H3) @@ -1037,14 +1054,14 @@ connected({call, From}, {request_async, Method, Path, Headers, Body, AsyncMode, NewData = Data#conn_data{recv_timeout = RecvTimeout}, do_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedirect, NewData); -connected({call, From}, {send_headers, Method, Path, Headers}, #conn_data{protocol = http3} = Data) -> +connected({call, From}, {send_headers, Method, Path, Headers, _ReqOpts}, #conn_data{protocol = http3} = Data) -> %% HTTP/3 streaming body - send headers only via QUIC do_h3_send_headers(From, Method, Path, Headers, Data); -connected({call, From}, {send_headers, Method, Path, Headers}, #conn_data{protocol = http2} = Data) -> +connected({call, From}, {send_headers, Method, Path, Headers, ReqOpts}, #conn_data{protocol = http2} = Data) -> %% HTTP/2 streaming body - send headers without END_STREAM, then accept body %% chunks via send_body_chunk/finish_send_body. Mirrors do_h3_send_headers/5. - do_h2_send_headers(From, Method, Path, Headers, Data); + do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data); connected({call, From}, {open_h2_stream, Method, Path, Headers, HandlerPid, Opts}, #conn_data{protocol = http2, h2_conn = H2Conn} = Data) -> @@ -1065,7 +1082,7 @@ connected({call, From}, {open_h2_stream, Method, Path, Headers, HandlerPid, Opts end, {keep_state_and_data, [{reply, From, Reply}]}; -connected({call, From}, {send_headers, Method, Path, Headers}, Data) -> +connected({call, From}, {send_headers, Method, Path, Headers, _ReqOpts}, Data) -> %% Send only headers for streaming body mode (HTTP/1.1) NewData = Data#conn_data{ request_from = From, @@ -1293,7 +1310,7 @@ streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol = 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, - send_timeout = SendTimeout} = Data, + req_send_timeout = SendTimeout} = Data, Result = case BodyData of Fun when is_function(Fun, 0) -> stream_body_fun_h2(H2Conn, StreamId, Fun, SendTimeout); @@ -1341,7 +1358,7 @@ streaming_body({call, From}, finish_send_body, #conn_data{protocol = http3} = Da streaming_body({call, From}, finish_send_body, #conn_data{protocol = http2} = Data) -> %% HTTP/2 - close the request stream with an empty END_STREAM DATA frame. #conn_data{h2_conn = H2Conn, h2_stream_id = StreamId, - send_timeout = SendTimeout} = Data, + req_send_timeout = SendTimeout} = Data, case h2_send_data(H2Conn, StreamId, <<>>, true, SendTimeout) of ok -> {keep_state, Data, [{reply, From, ok}]}; @@ -2961,20 +2978,20 @@ cancel_h2_stream(H2Conn, StreamId) -> try h2_connection:cancel_stream(H2Conn, StreamId) catch _:_ -> ok end. %% @private Send an HTTP/2 request via the h2 library. -do_h2_request(From, Method, Path, Headers, Body, Data) -> +do_h2_request(From, Method, Path, Headers, Body, SendTimeout, Data) -> do_h2_send(From, Method, Path, Headers, Body, - {sync, waiting_headers}, sync, Data). + {sync, waiting_headers}, sync, SendTimeout, Data). %% @private Send an HTTP/2 async request. do_h2_request_async(From, Method, Path, Headers, Body, AsyncMode, StreamTo, - _FollowRedirect, Data) -> + _FollowRedirect, SendTimeout, Data) -> Ref = self(), StreamState = {async, AsyncMode, StreamTo, Ref, waiting_headers}, do_h2_send(From, Method, Path, Headers, Body, StreamState, - {async, Ref, StreamTo, AsyncMode}, Data). + {async, Ref, StreamTo, AsyncMode}, SendTimeout, Data). -do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> - #conn_data{h2_conn = H2Conn, send_timeout = SendTimeout} = Data, +do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Data) -> + #conn_data{h2_conn = H2Conn} = Data, {MethodBin, PathBin, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), BodyBin = case Body of @@ -3007,7 +3024,16 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> end, case SendRes of {ok, StreamId} -> - Streams = maps:put(StreamId, {From, StreamState}, + %% The entry's first element is the stream owner: the parked sync + %% caller, or the async StreamTo pid. The async delivery clauses + %% (h2_on_response/h2_on_data) match it against the StreamTo held + %% in the stream state, so storing the gen_statem From tuple there + %% would silently drop every async response. + Owner = case Mode of + sync -> From; + {async, _Ref0, StreamTo0, _AsyncMode0} -> StreamTo0 + end, + Streams = maps:put(StreamId, {Owner, StreamState}, Data#conn_data.h2_streams), NewData0 = Data#conn_data{ h2_streams = Streams, @@ -3037,10 +3063,14 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, Data) -> %% @private Begin an HTTP/2 streaming-body request: send HEADERS without %% END_STREAM and transition to streaming_body so the caller can push body %% chunks via send_body_chunk/finish_send_body. Mirrors do_h3_send_headers/5. -do_h2_send_headers(From, Method, Path, Headers, Data) -> +do_h2_send_headers(From, Method, Path, Headers, ReqOpts, Data) -> #conn_data{h2_conn = H2Conn, h2_streams = Streams} = Data, {MethodBin, PathBin, H2Headers} = build_h2_request_headers(Method, Path, Headers, Data), + %% Effective send_timeout for this stream's body chunks. Stored in the + %% per-request field, set at every stream start, so an override never + %% leaks into later requests; the connection default stays untouched. + SendTimeout = proplists:get_value(send_timeout, ReqOpts, Data#conn_data.send_timeout), %% h2_connection can die between pool checkout and this call; normalise the %% gen_statem:call exit into an error reply (issue #836). SendRes = try @@ -3054,6 +3084,7 @@ do_h2_send_headers(From, Method, Path, Headers, Data) -> NewData = Data#conn_data{ h2_streams = maps:put(StreamId, {undefined, {stream, sending}}, Streams), h2_stream_id = StreamId, + req_send_timeout = SendTimeout, method = MethodBin, path = PathBin, request_from = undefined @@ -3070,13 +3101,24 @@ do_h2_send_headers(From, Method, Path, Headers, Data) -> %% With nonblock, keeps the historical non-blocking behavior: h2_connection %% buffers beyond the window and returns {error, send_buffer_full} past its %% per-stream cap. Normalises a dead h2_connection exit to an error. +%% +%% Every caller abandons the request on a failed body send, so the half-sent +%% stream is RST_STREAM(CANCEL)ed here: its buffered body must not linger on +%% a shared connection, accumulating memory and stream capacity. h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) -> - try + Res = try do_h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) catch exit:{ExitReason, _} -> {error, {closed, ExitReason}}; exit:ExitReason -> {error, {closed, ExitReason}} - end. + end, + h2_send_result(Res, H2Conn, StreamId). + +h2_send_result(ok, _H2Conn, _StreamId) -> + ok; +h2_send_result({error, _} = Error, H2Conn, StreamId) -> + cancel_h2_stream(H2Conn, StreamId), + Error. do_h2_send_data(H2Conn, StreamId, Bin, EndStream, nonblock) -> h2_connection:send_data(H2Conn, StreamId, Bin, EndStream); diff --git a/src/hackney_pool.erl b/src/hackney_pool.erl index c39fa316..d96e07a0 100644 --- a/src/hackney_pool.erl +++ b/src/hackney_pool.erl @@ -994,7 +994,6 @@ start_connection(Key, Owner, Opts, State) -> start_connection(Host, Port, Transport, Owner, Opts, State) -> ConnectTimeout = proplists:get_value(connect_timeout, Opts, 8000), RecvTimeout = proplists:get_value(recv_timeout, Opts, infinity), - SendTimeout = proplists:get_value(send_timeout, Opts, 30000), IdleTimeout = State#state.keepalive_timeout, SslOpts = proplists:get_value(ssl_options, Opts, []), ConnectOpts = proplists:get_value(connect_options, Opts, []), @@ -1005,7 +1004,6 @@ start_connection(Host, Port, Transport, Owner, Opts, State) -> transport => Transport, connect_timeout => ConnectTimeout, recv_timeout => RecvTimeout, - send_timeout => SendTimeout, idle_timeout => IdleTimeout, ssl_options => SslOpts, connect_options => ConnectOpts, diff --git a/test/hackney_http2_flow_control_tests.erl b/test/hackney_http2_flow_control_tests.erl index 9fbdfd39..eeb0b396 100644 --- a/test/hackney_http2_flow_control_tests.erl +++ b/test/hackney_http2_flow_control_tests.erl @@ -36,7 +36,15 @@ raw_server_test_() -> {"async request send hits send_timeout when the window never opens", {timeout, 60, fun t_async_send_times_out/0}}, {"send_timeout nonblock keeps the old fail-fast behavior", - {timeout, 60, fun t_nonblock_opt_out/0}}]. + {timeout, 60, fun t_nonblock_opt_out/0}}, + {"timed-out send RST_STREAMs the abandoned stream", + {timeout, 60, fun t_timeout_cancels_stream/0}}, + {"per-request send_timeout does not leak into later pooled requests", + {timeout, 60, fun t_send_timeout_not_sticky/0}}, + {"streamed request send_timeout applies on a reused pooled conn", + {timeout, 60, fun t_streamed_send_timeout_reused_conn/0}}, + {"direct hackney_conn async API uses the connection default send_timeout", + {timeout, 60, fun t_async_direct_conn_api/0}}]. echo_server_test_() -> {setup, @@ -202,6 +210,134 @@ t_nonblock_opt_out() -> repro_h2_raw_server:stop(element(1, Srv)) end. +%% A send that times out must not leave the half-sent stream (and its +%% buffered body) behind on the shared connection: the client has to +%% RST_STREAM it. The raw server reports received RST_STREAM frames. +t_timeout_cancels_stream() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + notify => self(), + body_size => 128 + }), + Body = binary:copy(<<"c">>, 200 * 1024), + try + {error, timeout} = hackney:request(post, url(Srv, <<"/">>), [], Body, + opts([{send_timeout, 300}])), + receive + {h2_raw_server, rst_stream, StreamId, _Code} -> + ?assert(StreamId > 0) + after 2000 -> + erlang:error(no_rst_stream_after_send_timeout) + end + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% A {send_timeout, 25} request must not change the timeout of the next +%% request on the same pooled connection: the follow-up uses the default and +%% completes even though the window drains slower than 25 ms. +t_send_timeout_not_sticky() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, ?SMALL_WINDOW}], + window_update => {auto, ?SMALL_WINDOW, 30}, + body_size => 128 + }), + Pool = h2_fc_sticky_pool, + catch hackney_pool:stop_pool(Pool), + ok = hackney_pool:start_pool(Pool, [{max_connections, 5}]), + Body = binary:copy(<<"s">>, 150 * 1024), + try + {error, timeout} = hackney:request(post, url(Srv, <<"/">>), [], Body, + pool_opts(Pool, [{send_timeout, 25}])), + Res = hackney:request(post, url(Srv, <<"/">>), [], Body, pool_opts(Pool, [])), + ?assertMatch({ok, 200, _, _}, Res) + after + catch hackney_pool:stop_pool(Pool), + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% A streamed request must apply its own send_timeout even when it reuses a +%% pooled connection opened by an earlier request without one. +t_streamed_send_timeout_reused_conn() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Pool = h2_fc_stream_reuse_pool, + catch hackney_pool:stop_pool(Pool), + ok = hackney_pool:start_pool(Pool, [{max_connections, 5}]), + Chunk = binary:copy(<<"r">>, 200 * 1024), + try + %% Bodyless request establishes the pooled conn with the default + %% send_timeout. + {ok, 200, _, _} = hackney:request(get, url(Srv, <<"/">>), [], <<>>, + pool_opts(Pool, [])), + {ok, ConnPid} = hackney:request(post, url(Srv, <<"/">>), [], stream, + pool_opts(Pool, [{send_timeout, 25}])), + T0 = erlang:monotonic_time(millisecond), + Res = hackney:send_body(ConnPid, Chunk), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assertEqual({error, timeout}, Res), + ?assert(Elapsed < 1500) + after + catch hackney_pool:stop_pool(Pool), + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% The hackney_conn request_async arities without ReqOpts (and the legacy +%% message form without FollowRedirect) fall back to the connection default: +%% a body larger than the window still completes. +t_async_direct_conn_api() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, ?SMALL_WINDOW}], + window_update => {auto, ?SMALL_WINDOW, 5}, + body_size => 128 + }), + Body = binary:copy(<<"d">>, 150 * 1024), + {ok, ConnPid} = hackney_conn_sup:start_conn(#{ + host => "localhost", + port => repro_h2_raw_server:port(Srv), + transport => hackney_ssl, + connect_options => [{protocols, [http2]}], + ssl_options => [{insecure, true}, {verify, verify_none}] + }), + try + ok = hackney_conn:connect(ConnPid), + {ok, Ref1} = hackney_conn:request_async(ConnPid, <<"POST">>, <<"/">>, [], + Body, true), + ok = await_async_status(Ref1, 200), + %% Legacy message form without the FollowRedirect element. + {ok, Ref2} = gen_statem:call(ConnPid, {request_async, <<"POST">>, <<"/">>, + [], Body, true, self()}), + ok = await_async_status(Ref2, 200) + after + catch hackney_conn:stop(ConnPid), + repro_h2_raw_server:stop(element(1, Srv)) + end. + +await_async_status(Ref, Status) -> + receive + {hackney_response, Ref, {status, Status, _}} -> drain_async(Ref); + {hackney_response, Ref, {error, Reason}} -> {error, Reason} + after 15000 -> + {error, no_async_status} + end. + +drain_async(Ref) -> + receive + {hackney_response, Ref, done} -> ok; + {hackney_response, Ref, {error, Reason}} -> {error, Reason}; + {hackney_response, Ref, _} -> drain_async(Ref) + after 15000 -> + {error, no_async_done} + end. + %%==================================================================== %% Echo-server test (well-behaved peer, body over the old cap) %%==================================================================== @@ -273,6 +409,13 @@ opts(Extra) -> {recv_timeout, 15000}, {ssl_options, [{insecure, true}, {verify, verify_none}]}]. +pool_opts(Pool, Extra) -> + Extra ++ + [{pool, Pool}, + {protocols, [http2]}, + {recv_timeout, 15000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}]. + url(Srv, Path) -> Port = repro_h2_raw_server:port(Srv), iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), Path]). diff --git a/test/repro_h2_raw_server.erl b/test/repro_h2_raw_server.erl index fb4e9116..0b31611f 100644 --- a/test/repro_h2_raw_server.erl +++ b/test/repro_h2_raw_server.erl @@ -14,6 +14,8 @@ %%% quirk :: none | settings | {settings, [{atom(),int()}]} %%% | ping | window_update %%% quirk_when :: after_each | after_first | mid_first | mid_each +%%% notify :: pid() send {h2_raw_server, rst_stream, StreamId, Code} +%%% to this pid when an RST_STREAM frame is received %%% window_update :: none | {auto, Increment, DelayMs} %%% none (default): never open the request-body window; the %%% client's upload stalls once the initial window is spent. @@ -129,7 +131,11 @@ handle_frame(Sock, {data, StreamId, _Body, _EndStream, PayloadLen}, St, Knobs) - {continue, St}; handle_frame(_Sock, {goaway, _, _, _}, _St, _Knobs) -> stop; -handle_frame(_Sock, {rst_stream, _, _}, St, _Knobs) -> +handle_frame(_Sock, {rst_stream, StreamId, ErrorCode}, St, Knobs) -> + case maps:get(notify, Knobs, undefined) of + undefined -> ok; + Pid -> Pid ! {h2_raw_server, rst_stream, StreamId, ErrorCode} + end, {continue, St}; %% After we have sent GOAWAY, ignore new streams (the ALB stops answering %% streams past last_stream_id). The client's read then hangs to recv_timeout. From 45894436ea582aa26001f08f09cdc9156db40427 Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Fri, 17 Jul 2026 00:18:53 +0200 Subject: [PATCH 4/5] Fix review findings: h2 async once, manual send_timeout, dialyzer HTTP/2 {async, once} now honors stream_next/1 with the same contract as HTTP/1.1, backed by h2 manual flow control so an unpulled response keeps the peer's send window closed. Direct connections honor {send_timeout, T} from connect options again (pooled ones keep the constant default). Fix the unmatched return that tripped dialyzer. --- NEWS.md | 10 + guides/http2_guide.md | 17 ++ src/hackney.erl | 7 + src/hackney_conn.erl | 106 ++++++++- test/hackney_http2_async_once_tests.erl | 271 ++++++++++++++++++++++ test/hackney_http2_flow_control_tests.erl | 32 ++- test/repro_h2_raw_server.erl | 20 +- 7 files changed, 454 insertions(+), 9 deletions(-) create mode 100644 test/hackney_http2_async_once_tests.erl diff --git a/NEWS.md b/NEWS.md index 1712b699..a04b23b3 100644 --- a/NEWS.md +++ b/NEWS.md @@ -18,6 +18,16 @@ unreleased entry stored the internal call reference where the delivery code expected the `stream_to` pid, so every async HTTP/2 response was silently dropped and the caller never received `{hackney_response, Ref, ...}` messages. +- HTTP/2 `{async, once}` now honors `stream_next/1` with the same contract + as HTTP/1.1: status and headers are delivered eagerly, then each + `stream_next/1` delivers exactly one message (a body chunk or `done`). + Previously every frame was pushed eagerly, identical to `{async, true}`. + once-mode streams run h2 manual flow control, so a slow consumer keeps the + peer's window closed and in-flight data stays bounded to one window. +- Connections created with `hackney:connect/4` and `{pool, false}` honor a + `{send_timeout, T}` connect option again (`hackney:send_request/2` has no + per-request options channel). Pooled connections keep the constant default + and take the per-request option instead. 4.6.1 - 2026-07-15 ------------------ diff --git a/guides/http2_guide.md b/guides/http2_guide.md index 7dce1453..9d156863 100644 --- a/guides/http2_guide.md +++ b/guides/http2_guide.md @@ -130,6 +130,23 @@ receive end. ``` +With `{async, once}` you pull instead: status and headers arrive eagerly, +then each `hackney:stream_next/1` delivers exactly one message (a body chunk +or `done`). On HTTP/2 the stream uses manual flow control, so data you have +not pulled keeps the server's send window closed and in-flight data stays +bounded to one window. + +```erlang +{ok, Ref} = hackney:get(URL, [], <<>>, [{async, once}]), +%% ... receive status and headers as above, then: +ok = hackney:stream_next(Ref), +receive + {hackney_response, Ref, done} -> ok; + {hackney_response, Ref, Chunk} -> + process(Chunk) %% call stream_next/1 again for the next chunk +end. +``` + ## Connection Multiplexing HTTP/2 allows multiple concurrent requests on a single connection. Unlike HTTP/1.1 where each request needs its own connection, HTTP/2 multiplexes requests as independent "streams" on a shared connection. diff --git a/src/hackney.erl b/src/hackney.erl index 052e9642..e5daf1d8 100644 --- a/src/hackney.erl +++ b/src/hackney.erl @@ -140,6 +140,11 @@ connect_direct(Transport, Host, Port, Options) -> transport => Transport, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), + %% Single-owner connection: seed the conn-level send_timeout so + %% hackney:send_request/2 (which has no options channel) honors it. + %% Pooled connections deliberately skip this (shared conns keep the + %% constant default; per-request ReqOpts override there). + send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []) }, @@ -499,6 +504,8 @@ start_conn_with_socket_internal(Host, Port, Transport, Socket, Options) -> socket => Socket, connect_timeout => proplists:get_value(connect_timeout, Options, 8000), recv_timeout => proplists:get_value(recv_timeout, Options, 5000), + %% Single-owner tunneled connection: same seeding as connect_direct. + send_timeout => proplists:get_value(send_timeout, Options, 30000), connect_options => ConnectOpts, ssl_options => proplists:get_value(ssl_options, Options, []), no_reuse => NoReuse diff --git a/src/hackney_conn.erl b/src/hackney_conn.erl index 6ad11c8d..a38b67cd 100644 --- a/src/hackney_conn.erl +++ b/src/hackney_conn.erl @@ -459,10 +459,11 @@ request_async(Pid, Method, Path, Headers, Body, AsyncMode, StreamTo, FollowRedir Err -> Err end. -%% @doc Request the next message in {async, once} mode. +%% @doc Request the next message in {async, once} mode. The caller pid rides +%% along so a shared HTTP/2 connection can route the pull to the right stream. -spec stream_next(pid()) -> ok | {error, term()}. stream_next(Pid) -> - gen_statem:cast(Pid, stream_next). + gen_statem:cast(Pid, {stream_next, self()}). %% @doc Stop async mode and return to sync mode. -spec stop_async(pid()) -> ok | {error, term()}. @@ -1102,6 +1103,14 @@ connected({call, From}, {send_headers, Method, Path, Headers, _ReqOpts}, Data) - %% Transition to streaming_body state {next_state, streaming_body, NewData, [{next_event, internal, {send_headers_only, Method, Path, Headers}}]}; +%% {async, once} pull for an HTTP/2 stream. The h1 pulls are handled in the +%% streaming_once state; on a connected h2 conn the pull routes to the +%% caller's once-mode stream. The bare atom is the pre-upgrade message form. +connected(cast, {stream_next, Caller}, Data) -> + handle_h2_stream_next(Caller, Data); +connected(cast, stream_next, Data) -> + handle_h2_stream_next(undefined, Data); + %% HTTP/2 owner messages from h2 library connected(info, {h2, H2Conn, Event}, #conn_data{h2_conn = H2Conn} = Data) -> handle_h2_event(Event, Data); @@ -1682,8 +1691,10 @@ streaming_once(enter, _OldState, _Data) -> %% Wait for stream_next keep_state_and_data; +streaming_once(cast, {stream_next, _Caller}, Data) -> + streaming_once(cast, stream_next, Data); streaming_once(cast, stream_next, Data) -> - %% Stream one chunk + %% Stream one chunk (bare-atom form kept for pre-upgrade callers) #conn_data{async_ref = Ref, stream_to = StreamTo} = Data, case stream_body_chunk(Data) of {ok, Chunk, NewData} -> @@ -2998,15 +3009,22 @@ do_h2_send(From, Method, Path, Headers, Body, StreamState, Mode, SendTimeout, Da B when is_binary(B) -> B; L -> iolist_to_binary(L) end, + %% once-mode async pulls response chunks via stream_next/1: open the + %% stream with manual flow control so the peer stalls at one window until + %% the consumer pulls (consume/3 releases bytes as chunks are delivered). + StreamOpts = case Mode of + {async, _, _, once} -> #{flow_control => manual}; + _ -> #{} + end, %% h2_connection can die between pool checkout and this call; gen_statem:call %% on a dead pid raises exit:noproc. Catch that and normalise to an error %% so the caller sees {error, {closed, _}} instead of a gen_statem:call %% blowing up (issue #836). SendRes = try case BodyBin of - <<>> -> h2_connection:send_request_headers(H2Conn, H2Headers, true); + <<>> -> h2_connection:send_request_headers(H2Conn, H2Headers, true, StreamOpts); _ -> - case h2_connection:send_request_headers(H2Conn, H2Headers, false) of + case h2_connection:send_request_headers(H2Conn, H2Headers, false, StreamOpts) of {ok, SId} -> %% Block on flow control so bodies larger than the %% peer's window wait for WINDOW_UPDATE instead of @@ -3117,7 +3135,7 @@ h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) -> h2_send_result(ok, _H2Conn, _StreamId) -> ok; h2_send_result({error, _} = Error, H2Conn, StreamId) -> - cancel_h2_stream(H2Conn, StreamId), + _ = cancel_h2_stream(H2Conn, StreamId), Error. do_h2_send_data(H2Conn, StreamId, Bin, EndStream, nonblock) -> @@ -3289,6 +3307,19 @@ h2_on_response(StreamId, Status, Headers, Data) -> status = Status, response_headers = Headers}), {keep_state, Data2}; + {StreamTo, {async, once, StreamTo, Ref, waiting_headers}} -> + %% once mode, same contract as HTTP/1.1: status and headers are + %% delivered eagerly, then each stream_next/1 pulls exactly one + %% message (a body chunk or done). Queue holds undelivered items; + %% Credit is 1 when a stream_next arrived before any data. + StreamTo ! {hackney_response, Ref, {status, Status, <<>>}}, + StreamTo ! {hackney_response, Ref, {headers, Headers}}, + Streams2 = maps:put(StreamId, + {StreamTo, {async_once, StreamTo, Ref, [], 0}}, + Streams), + {keep_state, Data#conn_data{h2_streams = Streams2, + status = Status, + response_headers = Headers}}; {StreamTo, {async, AsyncMode, StreamTo, Ref, waiting_headers}} -> StreamTo ! {hackney_response, Ref, {status, Status, <<>>}}, StreamTo ! {hackney_response, Ref, {headers, Headers}}, @@ -3321,6 +3352,47 @@ h2_on_response(StreamId, Status, Headers, Data) -> {keep_state, Data} end. +%% @private Deliver exactly one queued item of a once-mode h2 async stream: +%% a body chunk (consume/3 reopens exactly those bytes of the peer's window) +%% or done (stream finished, cleanup). An empty queue parks the pull as a +%% credit; the next DATA frame delivers immediately. +deliver_once_item(StreamId, StreamTo, Ref, [{data, Body} | Rest], Data) -> + StreamTo ! {hackney_response, Ref, Body}, + _ = try h2_connection:consume(Data#conn_data.h2_conn, StreamId, byte_size(Body)) + catch _:_ -> ok end, + Streams2 = maps:put(StreamId, + {StreamTo, {async_once, StreamTo, Ref, Rest, 0}}, + Data#conn_data.h2_streams), + {keep_state, Data#conn_data{h2_streams = Streams2}}; +deliver_once_item(StreamId, StreamTo, Ref, [done], Data) -> + StreamTo ! {hackney_response, Ref, done}, + Streams2 = maps:remove(StreamId, Data#conn_data.h2_streams), + {keep_state, Data#conn_data{h2_streams = Streams2, + async = false, + async_ref = undefined, + stream_to = undefined}}; +deliver_once_item(StreamId, StreamTo, Ref, [], Data) -> + Streams2 = maps:put(StreamId, + {StreamTo, {async_once, StreamTo, Ref, [], 1}}, + Data#conn_data.h2_streams), + {keep_state, Data#conn_data{h2_streams = Streams2}}. + +%% @private Route a stream_next pull to the caller's once-mode h2 stream. +%% With several once-mode streams for the same caller on a shared connection, +%% the oldest stream id wins (the pull API cannot address a specific stream). +handle_h2_stream_next(Caller, #conn_data{h2_streams = Streams} = Data) -> + OnceIds = [SId || {SId, {StreamTo, {async_once, StreamTo, _, _, _}}} + <- maps:to_list(Streams), + StreamTo =:= Caller orelse Caller =:= undefined], + case lists:sort(OnceIds) of + [] -> + keep_state_and_data; + [StreamId | _] -> + {StreamTo, {async_once, StreamTo, Ref, Queue, _Credit}} = + maps:get(StreamId, Streams), + deliver_once_item(StreamId, StreamTo, Ref, Queue, Data) + end. + h2_on_data(StreamId, Body, EndStream, Data) -> #conn_data{h2_streams = Streams} = Data, case maps:get(StreamId, Streams, undefined) of @@ -3342,6 +3414,21 @@ h2_on_data(StreamId, Body, EndStream, Data) -> Data#conn_data{h2_streams = Streams2}), {keep_state, Data2} end; + {StreamTo, {async_once, StreamTo, Ref, Queue, Credit}} -> + %% Queue the frame; deliver only when the consumer has pulled. + %% The stream runs manual flow control, so undelivered bytes keep + %% the peer's window closed (bounded in-flight data). + Queue1 = Queue ++ [{data, Body} || Body =/= <<>>] + ++ [done || EndStream], + case Credit of + 1 -> + deliver_once_item(StreamId, StreamTo, Ref, Queue1, Data); + 0 -> + Streams2 = maps:put(StreamId, + {StreamTo, {async_once, StreamTo, Ref, Queue1, 0}}, + Streams), + {keep_state, Data#conn_data{h2_streams = Streams2}} + end; {StreamTo, {async, AsyncMode, StreamTo, Ref, streaming, Status, Headers}} -> _ = case byte_size(Body) of 0 -> ok; @@ -3426,6 +3513,10 @@ h2_on_stream_reset(StreamId, ErrorCode, Data) -> StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, Streams2 = maps:remove(StreamId, Streams), {keep_state, Data#conn_data{h2_streams = Streams2}}; + {StreamTo, {async_once, StreamTo, Ref, _, _}} -> + StreamTo ! {hackney_response, Ref, {error, {stream_error, ErrorCode}}}, + Streams2 = maps:remove(StreamId, Streams), + {keep_state, Data#conn_data{h2_streams = Streams2}}; {_, Inner} when is_tuple(Inner), element(1, Inner) =:= stream -> %% Streaming-body stream: reply to any parked caller and drop it so a %% later stream_body/start_response sees {error, no_stream}. @@ -3490,6 +3581,9 @@ collect_h2_aborts(Err, #conn_data{h2_streams = Streams} = Data) -> (_SId, {StreamTo, {async, _, StreamTo, Ref, _, _, _}}, Acc) -> StreamTo ! {hackney_response, Ref, {error, Err}}, Acc; + (_SId, {StreamTo, {async_once, StreamTo, Ref, _, _}}, Acc) -> + StreamTo ! {hackney_response, Ref, {error, Err}}, + Acc; (_SId, {_, Inner}, Acc) when is_tuple(Inner), element(1, Inner) =:= stream -> case h2_stream_parked_from(Inner) of undefined -> Acc; diff --git a/test/hackney_http2_async_once_tests.erl b/test/hackney_http2_async_once_tests.erl new file mode 100644 index 00000000..c2211045 --- /dev/null +++ b/test/hackney_http2_async_once_tests.erl @@ -0,0 +1,271 @@ +%%% Tests for HTTP/2 {async, once} pull semantics. +%%% +%%% once mode used to be ignored for HTTP/2: every body frame and done were +%%% pushed eagerly, identical to {async, true}. It now follows the HTTP/1.1 +%%% contract: status and headers are delivered eagerly, then each +%%% stream_next/1 delivers exactly one message (a body chunk or done). The +%%% stream runs h2 manual flow control, so undelivered bytes keep the peer's +%%% window closed: consume/3 releases them only as the consumer pulls, which +%%% the WINDOW_UPDATE assertions below observe on the wire. +-module(hackney_http2_async_once_tests). + +-include_lib("eunit/include/eunit.hrl"). + +-define(FRAME_COUNT, 3). +-define(BODY_SIZE, 30000). +%% repro_h2_raw_server:make_body/1 wraps the fill in {"d":"..."} and yields +%% body_size - 4 actual bytes. +-define(WIRE_SIZE, (?BODY_SIZE - 4)). + +async_once_test_() -> + [{"once: no body frame is pushed until stream_next/1", + {timeout, 60, fun t_once_pull_pacing/0}}, + {"once: no stream WINDOW_UPDATE before the first pull (backpressure)", + {timeout, 60, fun t_once_backpressure_wire/0}}, + {"async true still delivers eagerly", + {timeout, 60, fun t_async_true_still_eager/0}}, + {"once: a pull before data arrives parks as credit", + {timeout, 60, fun t_once_pull_before_data/0}}, + {"once: RST_STREAM surfaces as an error message", + {timeout, 60, fun t_once_stream_reset/0}}, + {"once: connection teardown surfaces as an error message", + {timeout, 60, fun t_once_conn_teardown/0}}, + {"legacy bare stream_next atom still routes", + {timeout, 60, fun t_legacy_bare_stream_next/0}}]. + +%% Status and headers arrive eagerly; body chunks and done arrive strictly +%% one per stream_next/1, never unsolicited. +t_once_pull_pacing() -> + {Srv, Ref} = start_and_request(once), + try + ok = recv_status_headers(Ref), + %% Nothing may be pushed before the first pull. + no_message_within(Ref, 300), + Chunks = pull_all(Ref, []), + ?assertEqual(?WIRE_SIZE, iolist_size(Chunks)), + ?assert(length(Chunks) >= 2) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% Manual flow control on the wire: the client must not open the stream +%% window before the consumer pulls, and must open it (consume/3) after. +t_once_backpressure_wire() -> + {Srv, Ref} = start_and_request(once), + try + ok = recv_status_headers(Ref), + no_message_within(Ref, 300), + ?assertEqual([], drain_stream_window_updates(0)), + _Chunks = pull_all(Ref, []), + %% consume/3 acknowledged the delivered bytes: at least one + %% stream-level WINDOW_UPDATE reaches the server. + WUs = drain_stream_window_updates(2000), + ?assert(length(WUs) >= 1), + ?assertEqual(?WIRE_SIZE, lists:sum([Inc || {_Sid, Inc} <- WUs])) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% {async, true} keeps the eager contract: all frames and done arrive with +%% no stream_next calls. +t_async_true_still_eager() -> + {Srv, Ref} = start_and_request(true), + try + ok = recv_status_headers(Ref), + Chunks = recv_eager(Ref, []), + ?assertEqual(?WIRE_SIZE, iolist_size(Chunks)) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% A stream_next issued while the queue is empty parks as a credit: the next +%% DATA frame is delivered immediately, without another pull. +t_once_pull_before_data() -> + {Srv, Ref} = start_and_request(once, #{ + body_size => ?BODY_SIZE, + frame_count => 2, + inter_frame_ms => 400, + notify => self() + }), + try + ok = recv_status_headers(Ref), + %% First pull gets the first frame. + ok = hackney:stream_next(Ref), + Chunk1 = recv_chunk(Ref), + %% Second pull arrives while the server still sleeps between frames: + %% the pull parks and the late frame is delivered on arrival. + ok = hackney:stream_next(Ref), + Chunk2 = recv_chunk(Ref), + ok = hackney:stream_next(Ref), + done = recv_done_or_chunk(Ref, byte_size(Chunk1) + byte_size(Chunk2)) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% A stream reset by the peer mid-response surfaces as an error message. +t_once_stream_reset() -> + {Srv, Ref} = start_and_request(once, #{rst_after_headers => cancel}), + try + ok = recv_status_headers(Ref), + receive + {hackney_response, Ref, {error, {stream_error, _}}} -> ok + after 5000 -> + erlang:error(no_stream_reset_error) + end + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% Connection-wide teardown (GOAWAY + close) aborts an undrained once stream +%% with an error message. +t_once_conn_teardown() -> + {Srv, Ref} = start_and_request(once, #{ + body_size => ?BODY_SIZE, + frame_count => ?FRAME_COUNT, + goaway_after => 1, + goaway_close => true + }), + try + ok = recv_status_headers(Ref), + %% Do not pull: the queued frames keep the once stream alive until + %% the GOAWAY/close arrives and aborts it. + receive + {hackney_response, Ref, {error, _}} -> ok + after 5000 -> + erlang:error(no_teardown_error) + end + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%% The pre-upgrade bare stream_next atom (no caller pid) still pulls, and is +%% a no-op once no once-mode stream remains. +t_legacy_bare_stream_next() -> + {Srv, Ref} = start_and_request(once, #{ + body_size => ?BODY_SIZE, + frame_count => ?FRAME_COUNT, + notify => self() + }), + try + ok = recv_status_headers(Ref), + ok = gen_statem:cast(Ref, stream_next), + _Chunk = recv_chunk(Ref), + Rest = pull_all(Ref, []), + ?assert(iolist_size(Rest) > 0), + %% Stream is done and removed: a stray legacy pull must not crash. + ok = gen_statem:cast(Ref, stream_next), + no_message_within(Ref, 150), + ?assert(is_process_alive(Ref)) + after + repro_h2_raw_server:stop(element(1, Srv)) + end. + +%%==================================================================== +%% Helpers +%%==================================================================== + +start_and_request(AsyncMode) -> + start_and_request(AsyncMode, #{ + body_size => ?BODY_SIZE, + frame_count => ?FRAME_COUNT, + notify => self() + }). + +start_and_request(AsyncMode, Knobs) -> + _ = application:ensure_all_started(hackney), + _ = application:ensure_all_started(h2), + %% eunit runs sibling tests in the same process: drop notify leftovers + %% from a previous test's server. + flush_server_notifications(), + Srv = repro_h2_raw_server:start(Knobs), + Port = repro_h2_raw_server:port(Srv), + Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]), + Opts = [{async, AsyncMode}, {pool, false}, {protocols, [http2]}, + {recv_timeout, 15000}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}], + {ok, Ref} = hackney:request(get, Url, [], <<>>, Opts), + {Srv, Ref}. + +flush_server_notifications() -> + receive + {h2_raw_server, _, _, _} -> flush_server_notifications() + after 0 -> + ok + end. + +recv_status_headers(Ref) -> + receive + {hackney_response, Ref, {status, 200, _}} -> + receive + {hackney_response, Ref, {headers, _}} -> ok + after 5000 -> erlang:error(no_headers) + end + after 5000 -> + erlang:error(no_status) + end. + +no_message_within(Ref, Ms) -> + receive + {hackney_response, Ref, Msg} -> + erlang:error({unsolicited_message, Msg}) + after Ms -> + ok + end. + +%% One stream_next = exactly one message. After a chunk, nothing else may +%% arrive until the next pull. +pull_all(Ref, Acc) -> + ok = hackney:stream_next(Ref), + receive + {hackney_response, Ref, done} -> + lists:reverse(Acc); + {hackney_response, Ref, Chunk} when is_binary(Chunk) -> + no_message_within(Ref, 150), + pull_all(Ref, [Chunk | Acc]) + after 5000 -> + erlang:error({no_message_after_stream_next, length(Acc)}) + end. + +recv_chunk(Ref) -> + receive + {hackney_response, Ref, Chunk} when is_binary(Chunk) -> Chunk + after 5000 -> + erlang:error(no_chunk) + end. + +%% The tail of a 2-frame body: either done directly (both frames already +%% pulled) or the last small chunk then done on one more pull. +recv_done_or_chunk(Ref, GotSoFar) -> + receive + {hackney_response, Ref, done} -> + ?assertEqual(?WIRE_SIZE, GotSoFar), + done; + {hackney_response, Ref, Chunk} when is_binary(Chunk) -> + ok = hackney:stream_next(Ref), + recv_done_or_chunk(Ref, GotSoFar + byte_size(Chunk)) + after 5000 -> + erlang:error(no_tail_message) + end. + +recv_eager(Ref, Acc) -> + receive + {hackney_response, Ref, done} -> + lists:reverse(Acc); + {hackney_response, Ref, Chunk} when is_binary(Chunk) -> + recv_eager(Ref, [Chunk | Acc]) + after 5000 -> + erlang:error(eager_delivery_stalled) + end. + +%% Collect {h2_raw_server, window_update, ...} for non-zero (stream-level) +%% stream ids. Connection-level (stream 0) updates stay in auto mode and are +%% ignored. +drain_stream_window_updates(TimeoutMs) -> + receive + {h2_raw_server, window_update, 0, _} -> + drain_stream_window_updates(TimeoutMs); + {h2_raw_server, window_update, StreamId, Inc} -> + [{StreamId, Inc} | drain_stream_window_updates(TimeoutMs)] + after TimeoutMs -> + [] + end. diff --git a/test/hackney_http2_flow_control_tests.erl b/test/hackney_http2_flow_control_tests.erl index eeb0b396..a562b0d7 100644 --- a/test/hackney_http2_flow_control_tests.erl +++ b/test/hackney_http2_flow_control_tests.erl @@ -44,7 +44,9 @@ raw_server_test_() -> {"streamed request send_timeout applies on a reused pooled conn", {timeout, 60, fun t_streamed_send_timeout_reused_conn/0}}, {"direct hackney_conn async API uses the connection default send_timeout", - {timeout, 60, fun t_async_direct_conn_api/0}}]. + {timeout, 60, fun t_async_direct_conn_api/0}}, + {"manual hackney:connect honors {send_timeout, T}", + {timeout, 60, fun t_manual_connect_send_timeout/0}}]. echo_server_test_() -> {setup, @@ -338,6 +340,34 @@ drain_async(Ref) -> {error, no_async_done} end. +%% hackney:connect/4 + hackney:send_request/2 has no per-request options +%% channel: the connection-level send_timeout from the connect options must +%% apply (single-owner connection, unlike pooled ones). +t_manual_connect_send_timeout() -> + start_apps(), + Srv = repro_h2_raw_server:start(#{ + server_settings => [{initial_window_size, 1024}], + window_update => none, + body_size => 128 + }), + Body = binary:copy(<<"m">>, 200 * 1024), + %% pool false: hackney:connect defaults to the default pool, but the + %% conn-level send_timeout seeding is for single-owner direct conns. + {ok, ConnPid} = hackney:connect(hackney_ssl, "localhost", + repro_h2_raw_server:port(Srv), + [{send_timeout, 25}, {pool, false}, {protocols, [http2]}, + {ssl_options, [{insecure, true}, {verify, verify_none}]}]), + try + T0 = erlang:monotonic_time(millisecond), + Res = hackney:send_request(ConnPid, {post, <<"/">>, [], Body}), + Elapsed = erlang:monotonic_time(millisecond) - T0, + ?assertEqual({error, timeout}, Res), + ?assert(Elapsed < 1500) + after + catch hackney:close(ConnPid), + repro_h2_raw_server:stop(element(1, Srv)) + end. + %%==================================================================== %% Echo-server test (well-behaved peer, body over the old cap) %%==================================================================== diff --git a/test/repro_h2_raw_server.erl b/test/repro_h2_raw_server.erl index 0b31611f..5b8af810 100644 --- a/test/repro_h2_raw_server.erl +++ b/test/repro_h2_raw_server.erl @@ -15,7 +15,9 @@ %%% | ping | window_update %%% quirk_when :: after_each | after_first | mid_first | mid_each %%% notify :: pid() send {h2_raw_server, rst_stream, StreamId, Code} -%%% to this pid when an RST_STREAM frame is received +%%% when an RST_STREAM frame is received, and +%%% {h2_raw_server, window_update, StreamId, Increment} +%%% when a WINDOW_UPDATE frame is received %%% window_update :: none | {auto, Increment, DelayMs} %%% none (default): never open the request-body window; the %%% client's upload stalls once the initial window is spent. @@ -118,7 +120,11 @@ handle_frame(Sock, {ping, Data}, St, _Knobs) -> {continue, St}; handle_frame(_Sock, {ping_ack, _}, St, _Knobs) -> {continue, St}; -handle_frame(_Sock, {window_update, _, _}, St, _Knobs) -> +handle_frame(_Sock, {window_update, StreamId, Increment}, St, Knobs) -> + case maps:get(notify, Knobs, undefined) of + undefined -> ok; + Pid -> Pid ! {h2_raw_server, window_update, StreamId, Increment} + end, {continue, St}; handle_frame(Sock, {data, StreamId, _Body, _EndStream, PayloadLen}, St, Knobs) -> %% Request-body DATA: never decoded, only flow-control accounted per the @@ -141,6 +147,16 @@ handle_frame(_Sock, {rst_stream, StreamId, ErrorCode}, St, Knobs) -> %% streams past last_stream_id). The client's read then hangs to recv_timeout. handle_frame(_Sock, {headers, _StreamId, _B, _E, _H}, #{goaway := true} = St, _Knobs) -> {continue, St}; +%% rst_after_headers knob: answer with response HEADERS only (no END_STREAM), +%% then reset the stream with the given error code. +handle_frame(Sock, {headers, StreamId, _Block, _EndStream, _EndHeaders}, + #{enc := EncCtx} = St, #{rst_after_headers := Code}) -> + {HBlock, EncCtx2} = h2_hpack:encode( + [{<<":status">>, <<"200">>}, + {<<"content-type">>, <<"application/json">>}], EncCtx), + send(Sock, h2_frame:headers(StreamId, HBlock, false)), + send(Sock, h2_frame:rst_stream(StreamId, Code)), + {continue, St#{enc := EncCtx2}}; handle_frame(Sock, {headers, StreamId, _Block, _EndStream, _EndHeaders}, St, Knobs) -> #{enc := EncCtx, count := Count} = St, Count2 = Count + 1, From 3818b6850c0cc30be6abc86f171f61ca35d552cf Mon Sep 17 00:00:00 2001 From: Benoit Chesneau Date: Fri, 17 Jul 2026 00:28:18 +0200 Subject: [PATCH 5/5] Route raw-server test notifications through a collector notify => self() leaked late {h2_raw_server, ...} messages into the shared eunit process, where later suites' catch-all receives consumed them as unexpected responses. Notifications now go to a per-test collector process that dies with the test, so nothing reaches other tests' mailboxes. --- test/hackney_http2_async_once_tests.erl | 67 ++++++++++++----------- test/hackney_http2_flow_control_tests.erl | 27 ++++++--- test/repro_h2_raw_server.erl | 30 ++++++++++ 3 files changed, 83 insertions(+), 41 deletions(-) diff --git a/test/hackney_http2_async_once_tests.erl b/test/hackney_http2_async_once_tests.erl index c2211045..5533db71 100644 --- a/test/hackney_http2_async_once_tests.erl +++ b/test/hackney_http2_async_once_tests.erl @@ -50,20 +50,28 @@ t_once_pull_pacing() -> %% Manual flow control on the wire: the client must not open the stream %% window before the consumer pulls, and must open it (consume/3) after. +%% Notifications go to a collector process, never to the shared eunit +%% process, so nothing leaks into later tests' mailboxes. t_once_backpressure_wire() -> - {Srv, Ref} = start_and_request(once), + Collector = repro_h2_raw_server:start_collector(), + {Srv, Ref} = start_and_request(once, #{ + body_size => ?BODY_SIZE, + frame_count => ?FRAME_COUNT, + notify => Collector + }), try ok = recv_status_headers(Ref), no_message_within(Ref, 300), - ?assertEqual([], drain_stream_window_updates(0)), + ?assertEqual([], stream_window_updates(Collector)), _Chunks = pull_all(Ref, []), - %% consume/3 acknowledged the delivered bytes: at least one - %% stream-level WINDOW_UPDATE reaches the server. - WUs = drain_stream_window_updates(2000), + %% consume/3 acknowledged the delivered bytes: stream-level + %% WINDOW_UPDATEs totalling the body size reach the server. + WUs = await_stream_window_updates(Collector, ?WIRE_SIZE, 20), ?assert(length(WUs) >= 1), ?assertEqual(?WIRE_SIZE, lists:sum([Inc || {_Sid, Inc} <- WUs])) after - repro_h2_raw_server:stop(element(1, Srv)) + repro_h2_raw_server:stop(element(1, Srv)), + repro_h2_raw_server:stop_collector(Collector) end. %% {async, true} keeps the eager contract: all frames and done arrive with @@ -84,8 +92,7 @@ t_once_pull_before_data() -> {Srv, Ref} = start_and_request(once, #{ body_size => ?BODY_SIZE, frame_count => 2, - inter_frame_ms => 400, - notify => self() + inter_frame_ms => 400 }), try ok = recv_status_headers(Ref), @@ -143,8 +150,7 @@ t_once_conn_teardown() -> t_legacy_bare_stream_next() -> {Srv, Ref} = start_and_request(once, #{ body_size => ?BODY_SIZE, - frame_count => ?FRAME_COUNT, - notify => self() + frame_count => ?FRAME_COUNT }), try ok = recv_status_headers(Ref), @@ -167,16 +173,12 @@ t_legacy_bare_stream_next() -> start_and_request(AsyncMode) -> start_and_request(AsyncMode, #{ body_size => ?BODY_SIZE, - frame_count => ?FRAME_COUNT, - notify => self() + frame_count => ?FRAME_COUNT }). start_and_request(AsyncMode, Knobs) -> _ = application:ensure_all_started(hackney), _ = application:ensure_all_started(h2), - %% eunit runs sibling tests in the same process: drop notify leftovers - %% from a previous test's server. - flush_server_notifications(), Srv = repro_h2_raw_server:start(Knobs), Port = repro_h2_raw_server:port(Srv), Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]), @@ -186,13 +188,6 @@ start_and_request(AsyncMode, Knobs) -> {ok, Ref} = hackney:request(get, Url, [], <<>>, Opts), {Srv, Ref}. -flush_server_notifications() -> - receive - {h2_raw_server, _, _, _} -> flush_server_notifications() - after 0 -> - ok - end. - recv_status_headers(Ref) -> receive {hackney_response, Ref, {status, 200, _}} -> @@ -257,15 +252,21 @@ recv_eager(Ref, Acc) -> erlang:error(eager_delivery_stalled) end. -%% Collect {h2_raw_server, window_update, ...} for non-zero (stream-level) -%% stream ids. Connection-level (stream 0) updates stay in auto mode and are -%% ignored. -drain_stream_window_updates(TimeoutMs) -> - receive - {h2_raw_server, window_update, 0, _} -> - drain_stream_window_updates(TimeoutMs); - {h2_raw_server, window_update, StreamId, Inc} -> - [{StreamId, Inc} | drain_stream_window_updates(TimeoutMs)] - after TimeoutMs -> - [] +%% Stream-level (non-zero stream id) WINDOW_UPDATEs seen by the server so +%% far. Connection-level (stream 0) updates stay in auto mode and are ignored. +stream_window_updates(Collector) -> + [{Sid, Inc} || {h2_raw_server, window_update, Sid, Inc} + <- repro_h2_raw_server:collector_messages(Collector), + Sid =/= 0]. + +%% Poll until the collected stream WINDOW_UPDATEs cover ExpectedSum. +await_stream_window_updates(Collector, ExpectedSum, Retries) -> + WUs = stream_window_updates(Collector), + case lists:sum([Inc || {_, Inc} <- WUs]) >= ExpectedSum of + true -> WUs; + false when Retries > 0 -> + timer:sleep(100), + await_stream_window_updates(Collector, ExpectedSum, Retries - 1); + false -> + WUs end. diff --git a/test/hackney_http2_flow_control_tests.erl b/test/hackney_http2_flow_control_tests.erl index a562b0d7..3d31423f 100644 --- a/test/hackney_http2_flow_control_tests.erl +++ b/test/hackney_http2_flow_control_tests.erl @@ -217,24 +217,35 @@ t_nonblock_opt_out() -> %% RST_STREAM it. The raw server reports received RST_STREAM frames. t_timeout_cancels_stream() -> start_apps(), + %% Notifications go to a collector, never to the shared eunit process + %% (late messages would leak into later tests' mailboxes). + Collector = repro_h2_raw_server:start_collector(), Srv = repro_h2_raw_server:start(#{ server_settings => [{initial_window_size, 1024}], window_update => none, - notify => self(), + notify => Collector, body_size => 128 }), Body = binary:copy(<<"c">>, 200 * 1024), try {error, timeout} = hackney:request(post, url(Srv, <<"/">>), [], Body, opts([{send_timeout, 300}])), - receive - {h2_raw_server, rst_stream, StreamId, _Code} -> - ?assert(StreamId > 0) - after 2000 -> - erlang:error(no_rst_stream_after_send_timeout) - end + Rsts = await_rst_streams(Collector, 20), + ?assertMatch([{StreamId, _Code} | _] when StreamId > 0, Rsts) after - repro_h2_raw_server:stop(element(1, Srv)) + repro_h2_raw_server:stop(element(1, Srv)), + repro_h2_raw_server:stop_collector(Collector) + end. + +await_rst_streams(Collector, Retries) -> + Rsts = [{Sid, Code} || {h2_raw_server, rst_stream, Sid, Code} + <- repro_h2_raw_server:collector_messages(Collector)], + case Rsts of + [] when Retries > 0 -> + timer:sleep(100), + await_rst_streams(Collector, Retries - 1); + _ -> + Rsts end. %% A {send_timeout, 25} request must not change the timeout of the next diff --git a/test/repro_h2_raw_server.erl b/test/repro_h2_raw_server.erl index 5b8af810..1f200626 100644 --- a/test/repro_h2_raw_server.erl +++ b/test/repro_h2_raw_server.erl @@ -28,6 +28,7 @@ -module(repro_h2_raw_server). -export([start/1, stop/1, port/1]). +-export([start_collector/0, collector_messages/1, stop_collector/1]). -define(PREFACE, <<"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n">>). @@ -50,6 +51,35 @@ stop(Pid) -> port({_Pid, Port}) -> Port. +%% Use a collector as the notify target instead of the test process: eunit +%% runs sibling tests (and later suites) in the same process, so notifications +%% arriving after a test ends would otherwise pollute their mailbox. The +%% collector dies with stop_collector/1 and any late message goes nowhere. +start_collector() -> + spawn(fun() -> collector_loop([]) end). + +collector_messages(Collector) -> + Ref = make_ref(), + Collector ! {get, self(), Ref}, + receive + {Ref, Msgs} -> Msgs + after 5000 -> + erlang:error(collector_timeout) + end. + +stop_collector(Collector) -> + exit(Collector, kill), + ok. + +collector_loop(Acc) -> + receive + {get, From, Ref} -> + From ! {Ref, lists:reverse(Acc)}, + collector_loop(Acc); + Msg -> + collector_loop([Msg | Acc]) + end. + accept_loop(LSock, Knobs) -> case ssl:transport_accept(LSock, 5000) of {ok, TSock} ->