fix: send a request a GOAWAY refused again on a fresh connection

guides/http2_guide.md

@@ -375,14 +375,33 @@ The option applies to whole-body requests and to streamed bodies sent with `hack
 
 ## Error Handling
 
+A request the server turns away before processing it is sent again, once.
+That is a stream above the `last_stream_id` of a GOAWAY (RFC 9113 §6.8), or
+one reset with `REFUSED_STREAM` (§8.7): the server guarantees it did nothing
+with the request, so hackney repeats it whatever the method: on a fresh
+connection, or on the same pooled connection when a `REFUSED_STREAM` left it
+serving. The streams a GOAWAY accepted still complete on the draining
+connection.
+
+The refusal still reaches the caller, as `{error, {goaway, ErrorCode}}` or
+`{error, {stream_error, refused_stream}}`, when:
+
+- the second attempt is refused too;
+- the body is a function (`fun/0` or `{fun/1, State}`), which the first
+  attempt may already have consumed;
+- the refusal arrives after `request/5` has returned: a streamed request body
+  (`stream`) gets it from `send_body/2` or `start_response/1`, an `async`
+  request as a `{hackney_response, Ref, {error, _}}` message;
+- the request was made on an explicit connection with `send_request/2`.
+
 HTTP/2 specific errors:
 
 ```erlang
 case hackney:get(URL) of
     {ok, Status, Headers, Body} ->
         ok;
     {error, {goaway, ErrorCode}} ->
-        %% Peer sent GOAWAY
+        %% Refused twice, or refused after request/5 had returned
         io:format("HTTP/2 GOAWAY: ~p~n", [ErrorCode]);
     {error, {stream_error, ErrorCode}} ->
         %% Peer sent RST_STREAM for this request

guides/http3_guide.md

@@ -404,6 +404,15 @@ Both HTTP/2 and HTTP/3 use lowercase header names:
 
 ## Error Handling
 
+A request the server turns away before processing it is sent again, once: a
+stream at or above the identifier of a GOAWAY (RFC 9114 §5.2) goes again on a
+fresh connection, one reset with `H3_REQUEST_REJECTED` (§4.1.1) on the same
+one. The refusal still reaches the caller, as `{error, {goaway, no_error}}`
+or `{error, {stream_reset, 16#10b}}`, in the same cases as for HTTP/2 (see
+the [HTTP/2 guide](http2_guide.md#error-handling)): a second refusal, a body
+function, a refusal after `request/5` has returned, or a request made with
+`send_request/2`.
+
 ```erlang
 case hackney:get(URL, [], <<>>, [{protocols, [http3]}]) of
     {ok, Status, Headers, Body} ->

src/hackney.erl

@@ -79,6 +79,9 @@
 -include("hackney_lib.hrl").
 -include_lib("hackney_internal.hrl").
 
+%% HTTP/3 H3_REQUEST_REJECTED (RFC 9114 8.1): the server did not process the
+%% request.
+-define(H3_REQUEST_REJECTED, 16#10b).
 
 -type url() :: #hackney_url{} | binary().
 -type conn() :: pid().
@@ -720,17 +723,66 @@ request_with_transport(Method, URL, Headers0, Body, Options0,
                 _ -> <<Path/binary, "?", Query/binary>>
               end,
 
-  %% Check for proxy
-  case maybe_proxy(Transport, URL#hackney_url.scheme, Host, Port, Options) of
+  Dial = fun() -> maybe_proxy(Transport, URL#hackney_url.scheme, Host, Port, Options) end,
+  Send = fun(ConnPid, ReqPath, ReqHeaders) ->
+           do_request(ConnPid, Method, ReqPath, ReqHeaders, Body, Options, URL, Host)
+         end,
+  case dial_and_send(Dial, Send, FinalPath, Headers0) of
+    {{error, Reason} = Error, ConnPid} ->
+      %% A request the server turned away unprocessed can go again whatever
+      %% its method, as long as the body can be sent twice. Once only: a
+      %% second refusal is the caller's.
+      case refused(Reason) andalso replayable_body(Body) of
+        true ->
+          maybe_stop_unpooled(ConnPid, Options),
+          {Result, _} = dial_and_send(Dial, Send, FinalPath, Headers0),
+          Result;
+        false ->
+          Error
+      end;
+    {Result, _} ->
+      Result
+  end.
+
+%% @private The server guarantees it did not process the request: a GOAWAY
+%% below its stream (RFC 9113 6.8, RFC 9114 5.2), RST_STREAM(REFUSED_STREAM)
+%% (RFC 9113 8.7) or H3_REQUEST_REJECTED (RFC 9114 4.1.1).
+refused({goaway, _}) -> true;
+refused({stream_error, refused_stream}) -> true;
+refused({stream_reset, ?H3_REQUEST_REJECTED}) -> true;
+refused(_Reason) -> false.
+
+%% @private Dial (possibly through a proxy) and send, returning the connection
+%% alongside the result so a refused request can be retried elsewhere.
+dial_and_send(Dial, Send, FinalPath, Headers0) ->
+  case Dial() of
     {ok, ConnPid} ->
-      do_request(ConnPid, Method, FinalPath, Headers0, Body, Options, URL, Host);
+      {Send(ConnPid, FinalPath, Headers0), ConnPid};
     {ok, ConnPid, {http_proxy, TargetScheme, TargetHost, TargetPort, ProxyAuth}} ->
       %% HTTP proxy mode - use absolute URLs
       AbsolutePath = build_absolute_url(TargetScheme, TargetHost, TargetPort, FinalPath),
       Headers1 = add_proxy_auth_header(Headers0, ProxyAuth),
-      do_request(ConnPid, Method, AbsolutePath, Headers1, Body, Options, URL, Host);
+      {Send(ConnPid, AbsolutePath, Headers1), ConnPid};
     Error ->
-      Error
+      {Error, undefined}
+  end.
+
+%% @private A body produced by a fun may already have been consumed by the
+%% first attempt. Everything else hackney accepts is re-sendable: binaries,
+%% iolists, forms, multipart, files, and `stream', whose chunks only start
+%% after the headers were accepted.
+replayable_body(Body) when is_function(Body) -> false;
+replayable_body({Fun, _State}) when is_function(Fun, 1) -> false;
+replayable_body(_Body) -> true.
+
+%% @private A pooled connection stays the pool's: one a GOAWAY refused has
+%% already left it and closes on its own, one that refused a single stream
+%% still serves the others. An unpooled one belongs to this caller, who is
+%% about to replace it.
+maybe_stop_unpooled(ConnPid, Options) ->
+  case use_pool(Options) of
+    false -> stop_conn(ConnPid);
+    _ -> ok
   end.
 
 %% @doc Send a request on an existing connection.

src/hackney_conn.erl

@@ -4776,8 +4776,9 @@ h3_body_too_large(StreamId, From, Streams, #conn_data{h3_conn = ConnRef} = Data)
 %% @private Handle HTTP/3 stream reset
 handle_h3_stream_reset(StreamId, ErrorCode, Streams, Data) ->
     case maps:get(StreamId, Streams, undefined) of
-        {From, _State} when is_pid(From) ->
-            %% Sync stream - reply with error
+        {From, StreamState} when StreamState =:= waiting_headers;
+                                 element(1, StreamState) =:= waiting_headers_streaming ->
+            %% Sync request awaiting headers - reply via the state machine.
             UpdatedStreams = maps:remove(StreamId, Streams),
             {keep_state, Data#conn_data{h3_streams = UpdatedStreams, request_from = undefined},
              [{reply, From, {error, {stream_reset, ErrorCode}}}]};

src/hackney_pool.erl

@@ -680,7 +680,7 @@ handle_call({checkout_h2, Key}, _From, #state{h2_connections = H2Conns} = State)
         undefined ->
             {reply, none, State};
         Pid ->
-            case h2_conn_usable(Pid) of
+            case shared_conn_usable(Pid) of
                 true ->
                     {reply, {ok, Pid}, State};
                 false ->
@@ -690,17 +690,17 @@ handle_call({checkout_h2, Key}, _From, #state{h2_connections = H2Conns} = State)
     end;
 
 handle_call({checkout_h3, Key}, _From, #state{h3_connections = H3Conns} = State) ->
-    %% HTTP/3 checkout - return existing connection if available
+    %% HTTP/3 checkout - return existing connection if available. The same
+    %% check as HTTP/2: a connection draining after a GOAWAY leaves the pool
+    %% with a cast that can lose the race against this checkout.
     case maps:get(Key, H3Conns, undefined) of
         undefined ->
             {reply, none, State};
         Pid ->
-            %% Verify connection is still alive
-            case erlang:is_process_alive(Pid) of
+            case shared_conn_usable(Pid) of
                 true ->
                     {reply, {ok, Pid}, State};
                 false ->
-                    %% Connection died, remove from pool
                     H3Conns2 = maps:remove(Key, H3Conns),
                     {reply, none, State#state{h3_connections = H3Conns2}}
             end
@@ -1360,9 +1360,10 @@ pool_has_idle_room(#state{available=Available, max_connections=MaxConn}) ->
 idle_count(Available) ->
     maps:fold(fun(_, Pids, Acc) -> Acc + length(Pids) end, 0, Available).
 
-%% @private Check that a pooled HTTP/2 conn is alive and in `connected` state.
-%% Short timeout so a stuck conn doesn't wedge the pool; any failure → unusable.
-h2_conn_usable(Pid) ->
+%% @private Check that a shared HTTP/2 or HTTP/3 conn is alive and in
+%% `connected` state. Short timeout so a stuck conn doesn't wedge the pool;
+%% any failure → unusable.
+shared_conn_usable(Pid) ->
     case erlang:is_process_alive(Pid) of
         false -> false;
         true ->

test/hackney_h2_goaway_server.erl

@@ -6,12 +6,17 @@
 %%% streamed body once that body is complete. Later connections answer every
 %%% request as soon as it is complete. Each response body is the stream id.
 %%%
-%%% The GOAWAY is sent on a trigger:
-%%%   {second_stream, Pick}  when the second stream opens, with
+%%% The refusal is sent on a trigger:
+%%%   {second_stream, Pick}  GOAWAY when the second stream opens, with
 %%%                          Pick(FirstStreamId, SecondStreamId) as last_stream_id
-%%%   first_data             when a stream's first body chunk arrives, with that
-%%%                          stream as last_stream_id
+%%%   first_data             GOAWAY when a stream's first body chunk arrives,
+%%%                          with that stream as last_stream_id
+%%%   first_stream           GOAWAY when the first stream opens, refusing it
+%%%   refused_stream         RST_STREAM(REFUSED_STREAM) on the first stream; the
+%%%                          connection goes on answering the others
 %%% and shaped by options:
+%%%   trigger_on => first | all     only the first connection refuses, or every
+%%%                                 connection does (default first)
 %%%   goaway => direct | two_step   two_step first sends GOAWAY(2^31-1), the
 %%%                                 graceful shutdown of RFC 9113 6.8 (default direct)
 %%%   answer => boolean()           false answers nothing after the GOAWAY, not
@@ -37,19 +42,20 @@ start(Trigger, Opts) ->
          {versions, ['tlsv1.2', 'tlsv1.3']},
          {active, false}, {mode, binary}, {reuseaddr, true}]),
     {ok, {_, Port}} = ssl:sockname(LSock),
-    Options = maps:merge(#{goaway => direct, answer => true}, Opts),
+    Options = maps:merge(#{goaway => direct, answer => true, trigger_on => first}, Opts),
     Pid = spawn(fun() -> accept_loop(LSock, Trigger, Options) end),
     Url = iolist_to_binary([<<"https://localhost:">>, integer_to_list(Port), <<"/">>]),
     {Pid, Url}.
 
 stop(Pid) ->
     exit(Pid, kill).
 
-accept_loop(LSock, Trigger, Opts) ->
+accept_loop(LSock, Trigger, #{trigger_on := TriggerOn} = Opts) ->
     case ssl:transport_accept(LSock, 2000) of
         {ok, TSock} ->
             spawn(fun() -> serve(TSock, Trigger, Opts) end),
-            accept_loop(LSock, none, maps:remove(notify, Opts));
+            Next = case TriggerOn of all -> Trigger; first -> none end,
+            accept_loop(LSock, Next, maps:remove(notify, Opts));
         {error, timeout} -> accept_loop(LSock, Trigger, Opts);
         {error, closed} -> ok
     end.
@@ -131,19 +137,30 @@ on_request_frame(Sock, Kind, Sid, EndStream, #{seen := Seen, complete := Complet
     Seen2 = case lists:member(Sid, Seen) of true -> Seen; false -> Seen ++ [Sid] end,
     Complete2 = case EndStream of true -> [Sid | Complete]; false -> Complete end,
     St1 = St#{seen := Seen2, complete := Complete2},
-    St2 = case goaway_for(Kind, Sid, EndStream, St1) of
+    St2 = case refusal_for(Kind, Sid, EndStream, St1) of
         none -> St1;
-        LastStreamId -> send_goaway(Sock, LastStreamId, St1)
+        {goaway, LastStreamId} -> send_goaway(Sock, LastStreamId, St1);
+        {rst_stream, Sid} -> send_rst_stream(Sock, Sid, St1)
     end,
     answer_ready(Sock, St2).
 
-goaway_for(headers, _Sid, _EndStream, #{trigger := {second_stream, Pick}, seen := [First, Second]}) ->
-    Pick(First, Second);
-goaway_for(data, Sid, false, #{trigger := first_data}) ->
-    Sid;
-goaway_for(_Kind, _Sid, _EndStream, _St) ->
+refusal_for(headers, _Sid, _EndStream, #{trigger := {second_stream, Pick}, seen := [First, Second]}) ->
+    {goaway, Pick(First, Second)};
+refusal_for(data, Sid, false, #{trigger := first_data}) ->
+    {goaway, Sid};
+refusal_for(headers, Sid, _EndStream, #{trigger := first_stream, seen := [Sid]}) ->
+    {goaway, max(Sid - 2, 0)};
+refusal_for(headers, Sid, _EndStream, #{trigger := refused_stream, seen := [Sid]}) ->
+    {rst_stream, Sid};
+refusal_for(_Kind, _Sid, _EndStream, _St) ->
     none.
 
+%% RFC 9113 8.7: the stream was not processed; the connection stays up and
+%% answers every other stream.
+send_rst_stream(Sock, Sid, #{complete := Complete} = St) ->
+    send(Sock, h2_frame:rst_stream(Sid, refused_stream)),
+    St#{trigger := none, complete := Complete -- [Sid]}.
+
 %% One GOAWAY per connection, then a short drain before anything is answered.
 send_goaway(Sock, LastStreamId, #{seen := Seen, goaway := Shape, answer := Answer} = St) ->
     case Shape of

test/hackney_h3_test_server.erl

@@ -23,8 +23,9 @@
 %%%   GET|POST /hold         holds the request: sends {held, Handler, Conn,
 %%%                          StreamId} to the process registered as
 %%%                          hackney_h3_test_hold, then answers 200 with the
-%%%                          stream id as body on `release', or sends
-%%%                          {held_reset, StreamId} there if the client
+%%%                          stream id as body on `release', resets the
+%%%                          stream with H3_REQUEST_REJECTED on `reject', or
+%%%                          sends {held_reset, StreamId} there if the client
 %%%                          resets the stream
 %%%   POST any path          200, echoes the request body
 %%%   anything else          404
@@ -181,6 +182,8 @@ hold(Conn, StreamId) ->
     receive
         release ->
             respond(Conn, StreamId, 200, [], integer_to_binary(StreamId));
+        reject ->
+            quic_h3:cancel(Conn, StreamId, 16#10b);
         {quic_h3, Conn, {stream_reset, StreamId, _Code}} ->
             hackney_h3_test_hold ! {held_reset, StreamId};
         {quic_h3, Conn, _Other} ->