Enforce invariant of not performing reads after stream closure

src/core/surface/call.c

@@ -401,6 +401,7 @@ static int is_op_live(grpc_call *call, grpc_ioreq_op op) {
 static void lock(grpc_call *call) { gpr_mu_lock(&call->mu); }
 
 static int need_more_data(grpc_call *call) {
+  if (call->read_state == READ_STATE_STREAM_CLOSED) return 0;
   return is_op_live(call, GRPC_IOREQ_RECV_INITIAL_METADATA) ||
          (is_op_live(call, GRPC_IOREQ_RECV_MESSAGE) && grpc_bbq_empty(&call->incoming_queue)) ||
          is_op_live(call, GRPC_IOREQ_RECV_TRAILING_METADATA) ||
@@ -408,8 +409,7 @@ static int need_more_data(grpc_call *call) {
          is_op_live(call, GRPC_IOREQ_RECV_STATUS_DETAILS) ||
          (is_op_live(call, GRPC_IOREQ_RECV_CLOSE) &&
           grpc_bbq_empty(&call->incoming_queue)) ||
-         (call->write_state == WRITE_STATE_INITIAL && !call->is_client &&
-          call->read_state < READ_STATE_GOT_INITIAL_METADATA);
+         (call->write_state == WRITE_STATE_INITIAL && !call->is_client);
 }
 
 static void unlock(grpc_call *call) {

src/core/transport/chttp2_transport.c

@@ -1142,6 +1142,7 @@ static void perform_op_locked(transport *t, stream *s, grpc_transport_op *op) {
 
   if (op->recv_ops) {
     GPR_ASSERT(s->incoming_sopb == NULL);
+    GPR_ASSERT(s->published_state != GRPC_STREAM_CLOSED);
     s->recv_done_closure.cb = op->on_done_recv;
     s->recv_done_closure.user_data = op->recv_user_data;
     s->incoming_sopb = op->recv_ops;