Fix handling of RST_STREAM frames

[davisp]

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;

[tsloughter]

Thanks, this it the place to post it, I maintain this as well. I haven't merged in my client changes from my fork yet because I found them to be a bit hacky and haven't had time to clean them.

I can try to get some time this week to finally do that so we have a single base to work off of. It is simple shit like a callback I added doesn't check for existence and should probably be able to return stop.

Though I remember some other client patches from others that need cleaning to, some messages get sent in too adhoc of a way to be good.

[davisp]

Ah, cool. Glad I found the right place on the first try.

Also yeah, I didn't bother forking and opening a PR precisely cause it was hard to tell if I was anywhere near correct in my approach. For instance, I'm fairly certain my recv_rs callbacks end up triggering multiple eos messages back to the client process which seems Not Very Good™.