Merge pull request grpc#6969 from ctiller/stream_finish

Fix errored streams prematurely terminating, add a test

Makefile

@@ -6389,6 +6389,7 @@ LIBEND2END_TESTS_SRC = \
     test/core/end2end/tests/simple_delayed_request.c \
     test/core/end2end/tests/simple_metadata.c \
     test/core/end2end/tests/simple_request.c \
+    test/core/end2end/tests/streaming_error_response.c \
     test/core/end2end/tests/trailing_metadata.c \
 
 PUBLIC_HEADERS_C += \
@@ -6465,6 +6466,7 @@ LIBEND2END_NOSEC_TESTS_SRC = \
     test/core/end2end/tests/simple_delayed_request.c \
     test/core/end2end/tests/simple_metadata.c \
     test/core/end2end/tests/simple_request.c \
+    test/core/end2end/tests/streaming_error_response.c \
     test/core/end2end/tests/trailing_metadata.c \
 
 PUBLIC_HEADERS_C += \

src/core/ext/transport/chttp2/transport/chttp2_transport.c

@@ -1093,6 +1093,7 @@ static void perform_stream_op_locked(grpc_exec_ctx *exec_ctx,
     stream_global->recv_trailing_metadata_finished =
         add_closure_barrier(on_complete);
     stream_global->recv_trailing_metadata = op->recv_trailing_metadata;
+    stream_global->final_metadata_requested = true;
     grpc_chttp2_list_add_check_read_ops(transport_global, stream_global);
   }
 
@@ -1246,7 +1247,8 @@ static void check_read_ops(grpc_exec_ctx *exec_ctx,
       stream_global->recv_initial_metadata_ready = NULL;
     }
     if (stream_global->recv_message_ready != NULL) {
-      while (stream_global->seen_error &&
+      while (stream_global->final_metadata_requested &&
+             stream_global->seen_error &&
              (bs = grpc_chttp2_incoming_frame_queue_pop(
                   &stream_global->incoming_frames)) != NULL) {
         incoming_byte_stream_destroy_locked(exec_ctx, NULL, NULL, bs);

src/core/ext/transport/chttp2/transport/internal.h

@@ -441,6 +441,7 @@ typedef struct {
 
   bool published_initial_metadata;
   bool published_trailing_metadata;
+  bool final_metadata_requested;
 
   grpc_chttp2_incoming_metadata_buffer received_initial_metadata;
   grpc_chttp2_incoming_metadata_buffer received_trailing_metadata;

test/core/end2end/end2end_nosec_tests.c

@@ -115,6 +115,8 @@ extern void simple_metadata(grpc_end2end_test_config config);
 extern void simple_metadata_pre_init(void);
 extern void simple_request(grpc_end2end_test_config config);
 extern void simple_request_pre_init(void);
+extern void streaming_error_response(grpc_end2end_test_config config);
+extern void streaming_error_response_pre_init(void);
 extern void trailing_metadata(grpc_end2end_test_config config);
 extern void trailing_metadata_pre_init(void);
 
@@ -157,6 +159,7 @@ void grpc_end2end_tests_pre_init(void) {
   simple_delayed_request_pre_init();
   simple_metadata_pre_init();
   simple_request_pre_init();
+  streaming_error_response_pre_init();
   trailing_metadata_pre_init();
 }
 
@@ -203,6 +206,7 @@ void grpc_end2end_tests(int argc, char **argv,
     simple_delayed_request(config);
     simple_metadata(config);
     simple_request(config);
+    streaming_error_response(config);
     trailing_metadata(config);
     return;
   }
@@ -352,6 +356,10 @@ void grpc_end2end_tests(int argc, char **argv,
       simple_request(config);
       continue;
     }
+    if (0 == strcmp("streaming_error_response", argv[i])) {
+      streaming_error_response(config);
+      continue;
+    }
     if (0 == strcmp("trailing_metadata", argv[i])) {
       trailing_metadata(config);
       continue;

test/core/end2end/end2end_tests.c

@@ -117,6 +117,8 @@ extern void simple_metadata(grpc_end2end_test_config config);
 extern void simple_metadata_pre_init(void);
 extern void simple_request(grpc_end2end_test_config config);
 extern void simple_request_pre_init(void);
+extern void streaming_error_response(grpc_end2end_test_config config);
+extern void streaming_error_response_pre_init(void);
 extern void trailing_metadata(grpc_end2end_test_config config);
 extern void trailing_metadata_pre_init(void);
 
@@ -160,6 +162,7 @@ void grpc_end2end_tests_pre_init(void) {
   simple_delayed_request_pre_init();
   simple_metadata_pre_init();
   simple_request_pre_init();
+  streaming_error_response_pre_init();
   trailing_metadata_pre_init();
 }
 
@@ -207,6 +210,7 @@ void grpc_end2end_tests(int argc, char **argv,
     simple_delayed_request(config);
     simple_metadata(config);
     simple_request(config);
+    streaming_error_response(config);
     trailing_metadata(config);
     return;
   }
@@ -360,6 +364,10 @@ void grpc_end2end_tests(int argc, char **argv,
       simple_request(config);
       continue;
     }
+    if (0 == strcmp("streaming_error_response", argv[i])) {
+      streaming_error_response(config);
+      continue;
+    }
     if (0 == strcmp("trailing_metadata", argv[i])) {
       trailing_metadata(config);
       continue;

test/core/end2end/gen_build_yaml.py

@@ -126,6 +126,7 @@
     'simple_delayed_request': connectivity_test_options,
     'simple_metadata': default_test_options,
     'simple_request': default_test_options,
+    'streaming_error_response': default_test_options,
     'trailing_metadata': default_test_options,
 }