Expose trailers-only response status through C++ callback API (#26249)

* Expose trailers-only response status through callback API

* Fix change-detector test

* Address reviewer comment about trailers_only headers

* Add TODO requested by markdroth

* Move variable in struct to improve space efficiency

* Don't instantiate ExecCtx in channel_args destructor unless needed

include/grpcpp/impl/codegen/client_callback.h

@@ -594,7 +594,7 @@ class ClientCallbackReaderWriterImpl
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(ok && !context_->trailers_only());
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -737,7 +737,7 @@ class ClientCallbackReaderImpl : public ClientCallbackReader<Response> {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(ok && !context_->trailers_only());
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -995,7 +995,7 @@ class ClientCallbackWriterImpl : public ClientCallbackWriter<Request> {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(ok && !context_->trailers_only());
           MaybeFinish(/*from_reaction=*/true);
         },
         &start_ops_, /*can_inline=*/false);
@@ -1121,7 +1121,7 @@ class ClientCallbackUnaryImpl final : public ClientCallbackUnary {
     start_tag_.Set(
         call_.call(),
         [this](bool ok) {
-          reactor_->OnReadInitialMetadataDone(ok);
+          reactor_->OnReadInitialMetadataDone(ok && !context_->trailers_only());
           MaybeFinish();
         },
         &start_ops_, /*can_inline=*/false);

include/grpcpp/impl/codegen/client_context.h

@@ -493,6 +493,8 @@ class ClientContext {
       const grpc::ServerContextBase& server_context,
       PropagationOptions options);
 
+  bool trailers_only() const;
+
   bool initial_metadata_received_;
   bool wait_for_ready_;
   bool wait_for_ready_explicitly_set_;

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,10 @@ grpc_compression_algorithm grpc_call_compression_for_level(
   return algo;
 }
 
+bool grpc_call_is_trailers_only(const grpc_call* call) {
+  return call->is_trailers_only;
+}
+
 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_context.cc

@@ -31,6 +31,8 @@
 #include <grpcpp/server_context.h>
 #include <grpcpp/support/time.h>
 
+#include "src/core/lib/surface/call.h"
+
 namespace grpc {
 
 class Channel;
@@ -178,4 +180,10 @@ void ClientContext::SetGlobalCallbacks(GlobalCallbacks* client_callbacks) {
   g_client_callbacks = client_callbacks;
 }
 
+bool ClientContext::trailers_only() const {
+  bool result = initial_metadata_received_ && grpc_call_is_trailers_only(call_);
+  GPR_DEBUG_ASSERT(!result || recv_initial_metadata_.arr()->count == 0);
+  return result;
+}
+
 }  // 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"