Revert "Revert "Expose trailers-only response status through C++ callback API"" (#26365)

* Revert "Revert "Expose trailers-only response status through C++ callback API (#26249)" (#26363)"

This reverts commit 05c3b30ea0.

* Move the trailers_only check from ClientContext to ClientReactor

* Remove unneeded change in client_context.cc

include/grpcpp/impl/codegen/client_callback.h

@@ -130,6 +130,15 @@ class ClientReactor {
   /// heavyweight and the cost of the virtual call is not much in comparison.
   /// This function may be removed or devirtualized in the future.
   virtual void InternalScheduleOnDone(::grpc::Status s);
+
+  /// InternalTrailersOnly is not part of the API and is not meant to be
+  /// overridden. It is virtual to allow successful builds for certain bazel
+  /// build users that only want to depend on gRPC codegen headers and not the
+  /// full library (although this is not a generally-supported option). Although
+  /// the virtual call is slower than a direct call, this function is
+  /// heavyweight and the cost of the virtual call is not much in comparison.
+  /// This function may be removed or devirtualized in the future.
+  virtual bool InternalTrailersOnly(const grpc_call* call) const;
 };
 
 }  // namespace internal
@@ -594,7 +603,8 @@ class ClientCallbackReaderWriterImpl
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(
+              ok && !reactor_->InternalTrailersOnly(call_.call()));
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -737,7 +747,8 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(
+              ok && !reactor_->InternalTrailersOnly(call_.call()));
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -995,7 +1006,8 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(
+              ok && !reactor_->InternalTrailersOnly(call_.call()));
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -1121,7 +1133,8 @@ class ClientCallbackUnaryImpl final : public ClientCallbackUnary {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(
+              ok && !reactor_->InternalTrailersOnly(call_.call()));
           MaybeFinish();
         },
         &start_ops_, /*can_inline=*/false);

src/core/lib/surface/call.cc

@@ -170,6 +170,8 @@ struct grpc_call {
   bool destroy_called = false;
   /** flag indicating that cancellation is inherited */
   bool cancellation_is_inherited = false;
+  // Trailers-only response status
+  bool is_trailers_only = false;
   /** which ops are in-flight */
   bool sent_initial_metadata = false;
   bool sending_message = false;
@@ -1825,7 +1827,10 @@ static grpc_call_error call_start_batch(grpc_call* call, const grpc_op* ops,
             &call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */];
         stream_op_payload->recv_initial_metadata.recv_initial_metadata_ready =
             &call->receiving_initial_metadata_ready;
-        if (!call->is_client) {
+        if (call->is_client) {
+          stream_op_payload->recv_initial_metadata.trailing_metadata_available =
+              &call->is_trailers_only;
+        } else {
           stream_op_payload->recv_initial_metadata.peer_string =
               &call->peer_string;
         }
@@ -2017,6 +2022,14 @@ grpc_compression_algorithm grpc_call_compression_for_level(
   return algo;
 }
 
+bool grpc_call_is_trailers_only(const grpc_call* call) {
+  bool result = call->is_trailers_only;
+  GPR_DEBUG_ASSERT(
+      !result || call->metadata_batch[1 /* is_receiving */][0 /* is_trailing */]
+                         .list.count == 0);
+  return result;
+}
+
 bool grpc_call_failed_before_recv_message(const grpc_call* c) {
   return c->call_failed_before_recv_message;
 }

src/core/lib/surface/call.h

@@ -120,6 +120,11 @@ size_t grpc_call_get_initial_size_estimate();
 grpc_compression_algorithm grpc_call_compression_for_level(
     grpc_call* call, grpc_compression_level level);
 
+/* Did this client call receive a trailers-only response */
+/* TODO(markdroth): This is currently available only to the C++ API.
+                    Move to surface API if requested by other languages. */
+bool grpc_call_is_trailers_only(const grpc_call* call);
+
 /* Returns whether or not the call's receive message operation failed because of
  * an error (as opposed to a graceful end-of-stream) */
 /* TODO(markdroth): This is currently available only to the C++ API.

src/cpp/client/client_callback.cc

@@ -20,6 +20,7 @@
 #include "src/core/lib/iomgr/closure.h"
 #include "src/core/lib/iomgr/exec_ctx.h"
 #include "src/core/lib/iomgr/executor.h"
+#include "src/core/lib/surface/call.h"
 
 namespace grpc {
 namespace internal {
@@ -48,5 +49,9 @@ void ClientReactor::InternalScheduleOnDone(grpc::Status s) {
   grpc_core::Executor::Run(&arg->closure, GRPC_ERROR_NONE);
 }
 
+bool ClientReactor::InternalTrailersOnly(const grpc_call* call) const {
+  return grpc_call_is_trailers_only(call);
+}
+
 }  // namespace internal
 }  // namespace grpc

src/proto/grpc/testing/echo.proto

@@ -33,6 +33,7 @@ service EchoTestService {
   rpc ResponseStream(EchoRequest) returns (stream EchoResponse);
   rpc BidiStream(stream EchoRequest) returns (stream EchoResponse);
   rpc Unimplemented(EchoRequest) returns (EchoResponse);
+  rpc UnimplementedBidi(stream EchoRequest) returns (stream EchoResponse);
 }
 
 service EchoTest1Service {

test/cpp/end2end/client_callback_end2end_test.cc

@@ -1403,6 +1403,47 @@ TEST_P(ClientCallbackEnd2endTest, UnimplementedRpc) {
   }
 }
 
+TEST_P(ClientCallbackEnd2endTest, TestTrailersOnlyOnError) {
+  // Note that trailers-only is an HTTP/2 concept so we shouldn't do this test
+  // for any other transport such as inproc.
+  if (GetParam().protocol != Protocol::TCP) {
+    return;
+  }
+
+  ResetStub();
+  class Reactor : public grpc::experimental::ClientBidiReactor<EchoRequest,
+                                                               EchoResponse> {
+   public:
+    explicit Reactor(grpc::testing::EchoTestService::Stub* stub) {
+      stub->experimental_async()->UnimplementedBidi(&context_, this);
+      StartCall();
+    }
+    void Await() {
+      std::unique_lock<std::mutex> l(mu_);
+      while (!done_) {
+        done_cv_.wait(l);
+      }
+    }
+
+   private:
+    void OnReadInitialMetadataDone(bool ok) override { EXPECT_FALSE(ok); }
+    void OnDone(const Status& s) override {
+      EXPECT_EQ(s.error_code(), grpc::StatusCode::UNIMPLEMENTED);
+      EXPECT_EQ(s.error_message(), "");
+      std::unique_lock<std::mutex> l(mu_);
+      done_ = true;
+      done_cv_.notify_one();
+    }
+
+    ClientContext context_;
+    std::mutex mu_;
+    std::condition_variable done_cv_;
+    bool done_ = false;
+  } client(stub_.get());
+
+  client.Await();
+}
+
 TEST_P(ClientCallbackEnd2endTest,
        ResponseStreamExtraReactionFlowReadsUntilDone) {
   ResetStub();

test/cpp/util/grpc_tool_test.cc

@@ -62,7 +62,8 @@ using grpc::testing::EchoResponse;
   "RequestStream\n"               \
   "ResponseStream\n"              \
   "BidiStream\n"                  \
-  "Unimplemented\n"
+  "Unimplemented\n"               \
+  "UnimplementedBidi\n"
 
 #define ECHO_TEST_SERVICE_DESCRIPTION                                          \
   "filename: src/proto/grpc/testing/echo.proto\n"                              \
@@ -88,6 +89,8 @@ using grpc::testing::EchoResponse;
   "grpc.testing.EchoResponse) {}\n"                                            \
   "  rpc Unimplemented(grpc.testing.EchoRequest) returns "                     \
   "(grpc.testing.EchoResponse) {}\n"                                           \
+  "  rpc UnimplementedBidi(stream grpc.testing.EchoRequest) returns (stream "  \
+  "grpc.testing.EchoResponse) {}\n"                                            \
   "}\n"                                                                        \
   "\n"