fix: HTTP/2 sends now wait up to 5 seconds for flow-control credit instead of returning send_buffer_full

src/hackney.erl

@@ -137,6 +137,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),
+    h2_send_timeout => proplists:get_value(h2_send_timeout, Options, undefined),
     connect_options => ConnectOpts,
     ssl_options => proplists:get_value(ssl_options, Options, [])
   },
@@ -496,6 +497,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),
+    h2_send_timeout => proplists:get_value(h2_send_timeout, Options, undefined),
     connect_options => ConnectOpts,
     ssl_options => proplists:get_value(ssl_options, Options, []),
     no_reuse => NoReuse
@@ -1043,7 +1045,8 @@ shutdown_wt(WtPid) ->
 %% one stream on it. Returns a pid driven with h2_send/h2_recv etc. The method
 %% defaults to POST.
 %%
-%% Options: connect_timeout, recv_timeout, connect_options, ssl_options,
+%% Options: connect_timeout, recv_timeout, h2_send_timeout (optional, ms),
+%% connect_options, ssl_options,
 %% {flow_control, auto | manual}, {active, true | false | once},
 %% {max_recv_buffer, bytes | infinity}.
 -spec h2_open(binary() | string(), list()) -> {ok, pid()} | {error, term()}.
@@ -1080,6 +1083,7 @@ h2_open(Method, URL, Headers, Opts) ->
         headers => Headers,
         connect_timeout => proplists:get_value(connect_timeout, Opts, 8000),
         recv_timeout => proplists:get_value(recv_timeout, Opts, infinity),
+        h2_send_timeout => proplists:get_value(h2_send_timeout, Opts, undefined),
         connect_options => proplists:get_value(connect_options, Opts, []),
         ssl_options => proplists:get_value(ssl_options, Opts, []),
         flow_control => proplists:get_value(flow_control, Opts, auto),

src/hackney_conn.erl

@@ -144,6 +144,9 @@
     %% Options
     connect_timeout = ?CONNECT_TIMEOUT :: timeout(),
     recv_timeout = ?RECV_TIMEOUT :: timeout(),
+    %% Optional maximum time to wait for HTTP/2 peer flow-control credit while
+    %% sending. When unset, preserve h2's non-blocking send behavior.
+    h2_send_timeout = undefined :: timeout() | undefined,
     idle_timeout = ?IDLE_TIMEOUT :: timeout(),
     connect_options = [] :: list(),
     ssl_options = [] :: list(),
@@ -635,6 +638,7 @@ init([DefaultOwner, Opts]) ->
         socket = Socket,
         connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT),
         recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT),
+        h2_send_timeout = maps:get(h2_send_timeout, Opts, undefined),
         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 +948,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},
+    H2SendTimeout = proplists:get_value(h2_send_timeout, ReqOpts, Data#conn_data.h2_send_timeout),
+    NewData = Data#conn_data{recv_timeout = RecvTimeout, h2_send_timeout = H2SendTimeout},
     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 +1018,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},
+    H2SendTimeout = proplists:get_value(h2_send_timeout, ReqOpts, Data#conn_data.h2_send_timeout),
+    NewData = Data#conn_data{recv_timeout = RecvTimeout, h2_send_timeout = H2SendTimeout},
     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) ->
@@ -1281,16 +1287,17 @@ streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol =
             {next_state, closed, Data, [{reply, From, {error, Reason}}]}
     end;
 
-streaming_body({call, From}, {send_body_chunk, BodyData}, #conn_data{protocol = http2} = Data) ->
+streaming_body({call, From}, {send_body_chunk, BodyData},
+               #conn_data{protocol = http2, h2_send_timeout = SendTimeout} = Data) ->
     %% HTTP/2 - send a DATA frame without END_STREAM.
     #conn_data{h2_conn = H2Conn, h2_stream_id = StreamId} = 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 ->
@@ -1327,10 +1334,11 @@ streaming_body({call, From}, finish_send_body, #conn_data{protocol = http3} = Da
             {next_state, closed, Data, [{reply, From, {error, Reason}}]}
     end;
 
-streaming_body({call, From}, finish_send_body, #conn_data{protocol = http2} = Data) ->
+streaming_body({call, From}, finish_send_body,
+               #conn_data{protocol = http2, h2_send_timeout = SendTimeout} = 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
+    case h2_send_data(H2Conn, StreamId, <<>>, true, SendTimeout) of
         ok ->
             {keep_state, Data, [{reply, From, ok}]};
         {error, Reason} ->
@@ -2918,7 +2926,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, h2_send_timeout = SendTimeout} = Data,
     {MethodBin, PathBin, H2Headers} =
         build_h2_request_headers(Method, Path, Headers, Data),
     BodyBin = case Body of
@@ -2935,8 +2943,15 @@ 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
+                        case h2_send_data(H2Conn, SId, BodyBin, true, SendTimeout) of
                             ok -> {ok, SId};
+                            {error, timeout} = E1 ->
+                                %% h2 may retain the timed-out payload in its
+                                %% send buffer. The request is failing, so
+                                %% reset the stream rather than let it be sent
+                                %% after the caller has received an error.
+                                _ = cancel_h2_stream(H2Conn, SId),
+                                E1;
                             {error, _} = E1 -> E1
                         end;
                     Err -> Err
@@ -3004,37 +3019,43 @@ 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 Preserve non-blocking h2 sends unless the caller explicitly opts
+%% into a blocking deadline with h2_send_timeout.
+h2_send_data(H2Conn, StreamId, Bin, EndStream, undefined) ->
     try
         h2_connection:send_data(H2Conn, StreamId, Bin, EndStream)
     catch
         exit:{ExitReason, _} -> {error, {closed, ExitReason}};
         exit:ExitReason      -> {error, {closed, ExitReason}}
+    end;
+h2_send_data(H2Conn, StreamId, Bin, EndStream, SendTimeout) ->
+    try
+        h2_connection:send_data(H2Conn, StreamId, Bin, EndStream,
+                                #{block => SendTimeout})
+    catch
+        exit:{ExitReason, _} -> {error, {closed, ExitReason}};
+        exit:ExitReason      -> {error, {closed, ExitReason}}
     end.
 
 %% @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 ->
             ok;
         {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 ->

src/hackney_h2_stream.erl

@@ -91,6 +91,7 @@
     ssl_options = [] :: list(),
     connect_timeout = ?CONNECT_TIMEOUT :: timeout(),
     recv_timeout = ?RECV_TIMEOUT :: timeout(),
+    h2_send_timeout = undefined :: timeout() | undefined,
     flow_control = auto :: auto | manual,
     active = false :: false | true | once,
 
@@ -196,6 +197,7 @@ init([Owner, Opts]) ->
         ssl_options = maps:get(ssl_options, Opts, []),
         connect_timeout = maps:get(connect_timeout, Opts, ?CONNECT_TIMEOUT),
         recv_timeout = maps:get(recv_timeout, Opts, ?RECV_TIMEOUT),
+        h2_send_timeout = maps:get(h2_send_timeout, Opts, undefined),
         flow_control = maps:get(flow_control, Opts, auto),
         active = maps:get(active, Opts, false),
         max_recv_buffer = maps:get(max_recv_buffer, Opts, ?DEFAULT_MAX_RECV_BUFFER)
@@ -246,11 +248,20 @@ connected(enter, _OldState, _Data) ->
 
 %% --- send ------------------------------------------------------------
 connected({call, From}, {send, SData, Fin},
-          #h2s_data{h2_conn = H2Conn, stream_id = Sid}) ->
+          #h2s_data{h2_conn = H2Conn, stream_id = Sid, h2_send_timeout = SendTimeout}) ->
     EndStream = (Fin =:= fin),
-    Reply = h2_call(fun() ->
-        h2_connection:send_data(H2Conn, Sid, iolist_to_binary(SData), EndStream)
+    Reply0 = h2_call(fun() ->
+        h2_send_data(H2Conn, Sid, iolist_to_binary(SData), EndStream, SendTimeout)
     end),
+    %% h2 can retain data after its blocking-send deadline expires. Reset the
+    %% stream so a caller that receives timeout never has that payload sent
+    %% later in the background.
+    Reply = case Reply0 of
+        {error, timeout} = Error ->
+            _ = h2_call(fun() -> h2_connection:cancel_stream(H2Conn, Sid) end),
+            Error;
+        _ -> Reply0
+    end,
     {keep_state_and_data, [{reply, From, Reply}]};
 
 connected({call, From}, {send_trailers, Trailers},
@@ -579,6 +590,14 @@ h2_call(Fun) ->
         exit:ExitReason      -> {error, {closed, ExitReason}}
     end.
 
+%% @private Use h2's existing non-blocking behavior unless the caller opted
+%% into waiting for peer flow-control credit.
+h2_send_data(H2Conn, Sid, Data, EndStream, undefined) ->
+    h2_connection:send_data(H2Conn, Sid, Data, EndStream);
+h2_send_data(H2Conn, Sid, Data, EndStream, SendTimeout) ->
+    h2_connection:send_data(H2Conn, Sid, Data, EndStream,
+                            #{block => SendTimeout}).
+
 cancel_stream_safe(#h2s_data{h2_conn = undefined}) ->
     ok;
 cancel_stream_safe(#h2s_data{h2_conn = H2Conn, stream_id = Sid}) ->

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),
+    H2SendTimeout = proplists:get_value(h2_send_timeout, Opts, undefined),
     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,
+        h2_send_timeout => H2SendTimeout,
         idle_timeout => IdleTimeout,
         ssl_options => SslOpts,
         connect_options => ConnectOpts,