Description
For some background, I ended up hitting a corner case where a bidirectional gRPC stream (via grpcbox) was closing out on an error before any messages were exchanged. The observed behavior ended up being a timeout on a call to grpcbox_client:recv_data/2. Tracing this down it appears that chatterbox isn't properly following the RST_STREAM protocol which leaves grpcbox in limbo.
I tried to base these new state transitions off the diagram in RFC7540 5.1. Once adding this I appear to be getting the underlying error message that's causing the stream to fail though I'm not even certain on the mechanics of how that's delivered inside grpcbox.
Also given grpcbox, this is also based off @tsloughter's 0.9.1 tag so it won't apply cleanly here. The issues tab is disabled on that fork so I'm posting this here as I haven't got a clue one where better to post it.
diff --git a/src/h2_connection.erl b/src/h2_connection.erl
index 7ee864c..aaa2680 100644
--- a/src/h2_connection.erl
+++ b/src/h2_connection.erl
@@ -718,17 +718,17 @@ route_frame(
stream_id=StreamId,
type=?RST_STREAM
},
- _Payload},
+ Payload},
#connection{} = Conn) ->
- %% TODO: anything with this?
- %% EC = h2_frame_rst_stream:error_code(Payload),
Streams = Conn#connection.streams,
Stream = h2_stream_set:get(StreamId, Streams),
case h2_stream_set:type(Stream) of
idle ->
go_away(?PROTOCOL_ERROR, Conn);
- _Stream ->
- %% TODO: RST_STREAM support
+ active ->
+ recv_rs(Stream, Conn, h2_frame_rst_stream:error_code(Payload)),
+ {next_state, connected, Conn};
+ closed ->
{next_state, connected, Conn}
end;
route_frame({H=#frame_header{}, _P},
@@ -1650,6 +1650,21 @@ recv_es(Stream, Conn) ->
rst_stream(Stream, ?STREAM_CLOSED, Conn)
end.
+-spec recv_rs(Stream :: h2_stream_set:stream(),
+ Conn :: connection(),
+ ErrorCode :: error_code()) ->
+ ok.
+
+recv_rs(Stream, _Conn, ErrorCode) ->
+ case h2_stream_set:type(Stream) of
+ active ->
+ Pid = h2_stream_set:pid(Stream),
+ h2_stream:send_event(Pid, {recv_rs, ErrorCode});
+ _ ->
+ ok
+ end.
+
+
-spec recv_pp(h2_stream_set:stream(),
hpack:headers()) ->
ok.
diff --git a/src/h2_stream.erl b/src/h2_stream.erl
index 7a1bb08..d6aa95d 100644
--- a/src/h2_stream.erl
+++ b/src/h2_stream.erl
@@ -346,6 +346,18 @@ open(cast, recv_es,
Stream}
end;
+open(cast, {recv_rs, _ErrorCode},
+ #stream_state{
+ callback_mod=CB,
+ callback_state=CallbackState
+ }=Stream) ->
+ {ok, NewCBState} = callback(CB, on_end_stream, [], CallbackState),
+ {next_state,
+ closed,
+ Stream#stream_state{
+ callback_state=NewCBState
+ }, 0};
+
open(cast, {recv_data,
{#frame_header{
flags=Flags,
@@ -544,6 +556,8 @@ half_closed_remote(cast,
{next_state, closed, Stream, 0}
end;
+half_closed_remote(cast, {recv_es, _ErrorCode}, Stream) ->
+ {next_state, closed, Stream, 0};
half_closed_remote(cast, _,
#stream_state{}=Stream) ->
@@ -655,6 +669,10 @@ half_closed_local(cast, recv_es,
response_body = Data,
callback_state=NewCBState
}, 0};
+
+half_closed_local(cast, {recv_rs, _ErrorCode}, Stream) ->
+ half_closed_local(cast, recv_es, Stream);
+
half_closed_local(cast, {send_t, _Trailers},
#stream_state{}) ->
keep_state_and_data;