Reverted to using stream_write_closed

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

@@ -1207,12 +1207,7 @@ void grpc_chttp2_complete_closure_step(grpc_chttp2_transport* t,
         desc, errstr, write_state_name(t->write_state));
   }
   if (error != GRPC_ERROR_NONE) {
-    // The surface layer treats GRPC_ERROR_EOS specially and does not cancel
-    // calls with batches that failed with GRPC_ERROR_EOS. The transport layer
-    // uses GRPC_ERROR_EOS specifically for send ops that failed due to the
-    // stream being closed for writes.
-    if (closure->error_data.error == GRPC_ERROR_NONE &&
-        error != GRPC_ERROR_EOS) {
+    if (closure->error_data.error == GRPC_ERROR_NONE) {
       closure->error_data.error = GRPC_ERROR_CREATE_FROM_STATIC_STRING(
           "Error in HTTP transport completing operation");
       closure->error_data.error = grpc_error_set_str(
@@ -1489,9 +1484,13 @@ static void perform_stream_op_locked(void* stream_op,
     on_complete->next_data.scratch |= CLOSURE_BARRIER_MAY_COVER_WRITE;
     s->fetching_send_message_finished = add_closure_barrier(op->on_complete);
     if (s->write_closed) {
+      op->payload->send_message.stream_write_closed = true;
+      // We should NOT return an error here, so as to avoid a cancel OP being
+      // started. The surface layer will notice that the stream has been closed
+      // for writes and fail the send message op.
       op->payload->send_message.send_message.reset();
       grpc_chttp2_complete_closure_step(
-          t, s, &s->fetching_send_message_finished, GRPC_ERROR_EOS,
+          t, s, &s->fetching_send_message_finished, GRPC_ERROR_NONE,
           "fetching_send_message_finished");
     } else {
       GPR_ASSERT(s->fetching_send_message == nullptr);

src/core/ext/transport/inproc/inproc_transport.cc

@@ -764,8 +764,9 @@ void op_state_machine_locked(inproc_stream* s, grpc_error* error) {
       // Nothing further will try to receive from this stream, so finish off
       // any outstanding send_message op
       s->send_message_op->payload->send_message.send_message.reset();
+      s->send_message_op->payload->send_message.stream_write_closed = true;
       complete_if_batch_end_locked(
-          s, GRPC_ERROR_EOS, s->send_message_op,
+          s, new_err, s->send_message_op,
           "op_state_machine scheduling send-message-on-complete");
       s->send_message_op = nullptr;
     }

src/core/lib/iomgr/error.cc

@@ -405,9 +405,6 @@ static grpc_error* copy_error_and_unref(grpc_error* in) {
       internal_set_str(&out, GRPC_ERROR_STR_DESCRIPTION,
                        grpc_slice_from_static_string("cancelled"));
       internal_set_int(&out, GRPC_ERROR_INT_GRPC_STATUS, GRPC_STATUS_CANCELLED);
-    } else if (in == GRPC_ERROR_EOS) {
-      internal_set_str(&out, GRPC_ERROR_STR_DESCRIPTION,
-                       grpc_slice_from_static_string("end of stream"));
     }
   } else if (gpr_ref_is_unique(&in->atomics.refs)) {
     out = in;
@@ -462,10 +459,7 @@ const special_error_status_map error_status_map[] = {
      strlen("Out of memory")},              // GRPC_ERROR_OOM
     {GRPC_STATUS_INVALID_ARGUMENT, "", 0},  // GRPC_ERROR_RESERVED_2
     {GRPC_STATUS_CANCELLED, "Cancelled",
-     strlen("Cancelled")},                  // GRPC_ERROR_CANCELLED
-    {GRPC_STATUS_INVALID_ARGUMENT, "", 0},  // GRPC_ERROR_RESERVED_3
-    {GRPC_STATUS_UNAVAILABLE, "End of stream",
-     strlen("End of stream")}  // GRPC_ERROR_EOS
+     strlen("Cancelled")},  // GRPC_ERROR_CANCELLED
 };
 
 bool grpc_error_get_int(grpc_error* err, grpc_error_ints which, intptr_t* p) {
@@ -537,7 +531,6 @@ grpc_error* grpc_error_add_child(grpc_error* src, grpc_error* child) {
 static const char* no_error_string = "\"No Error\"";
 static const char* oom_error_string = "\"Out of memory\"";
 static const char* cancelled_error_string = "\"Cancelled\"";
-static const char* eos_error_string = "\"End of stream\"";
 
 typedef struct {
   char* key;
@@ -752,7 +745,6 @@ const char* grpc_error_string(grpc_error* err) {
   if (err == GRPC_ERROR_NONE) return no_error_string;
   if (err == GRPC_ERROR_OOM) return oom_error_string;
   if (err == GRPC_ERROR_CANCELLED) return cancelled_error_string;
-  if (err == GRPC_ERROR_EOS) return eos_error_string;
 
   void* p = (void*)gpr_atm_acq_load(&err->atomics.error_string);
   if (p != nullptr) {

src/core/lib/iomgr/error.h

@@ -127,11 +127,7 @@ typedef enum {
 #define GRPC_ERROR_OOM ((grpc_error*)2)
 #define GRPC_ERROR_RESERVED_2 ((grpc_error*)3)
 #define GRPC_ERROR_CANCELLED ((grpc_error*)4)
-#define GRPC_ERROR_RESERVED_3 ((grpc_error*)5)
-// GRPC_ERROR_EOS is used only by transports for send ops that failed because
-// the stream was closed for writes.
-#define GRPC_ERROR_EOS ((grpc_error*)6)
-#define GRPC_ERROR_SPECIAL_MAX GRPC_ERROR_EOS
+#define GRPC_ERROR_SPECIAL_MAX GRPC_ERROR_CANCELLED
 
 inline bool grpc_error_is_special(struct grpc_error* err) {
   return err <= GRPC_ERROR_SPECIAL_MAX;

src/core/lib/surface/call.cc

@@ -1178,6 +1178,12 @@ static void post_batch_completion(batch_control* bctl) {
         &call->metadata_batch[0 /* is_receiving */][0 /* is_trailing */]);
   }
   if (bctl->op.send_message) {
+    if (bctl->op.payload->send_message.stream_write_closed &&
+        error == GRPC_ERROR_NONE) {
+      error = grpc_error_add_child(
+          error, GRPC_ERROR_CREATE_FROM_STATIC_STRING(
+                     "Attempt to send message after stream was closed."));
+    }
     call->sending_message = false;
   }
   if (bctl->op.send_trailing_metadata) {
@@ -1532,12 +1538,7 @@ static void finish_batch(void* bctlp, grpc_error* error) {
     gpr_atm_rel_store(&bctl->batch_error,
                       reinterpret_cast<gpr_atm>(GRPC_ERROR_REF(error)));
   }
-  // If the batch had an error, we should normally fail the call. If the batch
-  // ended with GRPC_ERROR_EOS, we should not cancel the call because we
-  // do not want to overwrite the status that will be propagated through the
-  // recv_trailing_metadata callback.
-  if (error != GRPC_ERROR_NONE && error != GRPC_ERROR_EOS) {
-    gpr_log(GPR_ERROR, "got an error %s. cancelling", grpc_error_string(error));
+  if (error != GRPC_ERROR_NONE) {
     cancel_with_error(call, GRPC_ERROR_REF(error));
   }
   finish_batch_step(bctl);

src/core/lib/transport/transport.h

@@ -250,6 +250,18 @@ struct grpc_transport_stream_op_batch_payload {
     // The batch's on_complete will not be called until after the byte
     // stream is orphaned.
     grpc_core::OrphanablePtr<grpc_core::ByteStream> send_message;
+    // Set by the transport if the stream has been closed for writes. If this
+    // is set and send message op is present, we set the operation to be a
+    // failure without sending a cancel OP down the stack. This is so that the
+    // status of the call does not get overwritten by the Cancel OP, which would
+    // be especially problematic if we had received a valid status from the
+    // server.
+    // For send_initial_metadata, it is fine for the status to be overwritten
+    // because at that point, the client will not have received a status.
+    // For send_trailing_metadata, we might overwrite the status if we have
+    // non-zero metadata to send. This is fine because the API does not allow
+    // the client to send trailing metadata.
+    bool stream_write_closed = false;
   } send_message;
 
   struct {