gRPC C++: don't wait for initial server metadata during GrpcStream::Start (#1786)

By default, StartCall in gRPC C++ client would wait until the initial server metadata is received, after which the corresponding tag will come back from the completion queue. It can alternatively be configured so that initial server metadata is only received together with the first write operation, effectively making StartCall a synchronous operation. The benefit is that Firestore streams won't have to wait for a network roundtrip before they consider themselves open. It is also consistent with the other platforms.

Author: var-const

Firestore/core/src/firebase/firestore/remote/grpc_stream.cc

@@ -94,9 +94,20 @@ GrpcStream::~GrpcStream() {
 }
 
 void GrpcStream::Start() {
-  GrpcCompletion* completion =
-      NewCompletion([this](const GrpcCompletion*) { OnStart(); });
-  call_->StartCall(completion);
+  // Make starting a quick operation that avoids a roundtrip to the server by
+  // skipping the wait for initial server metadata (instead, it will be
+  // automatically coalesced with the first write operation).
+  context_->set_initial_metadata_corked(true);
+  // It's generally okay to pass a null pointer as a tag; in this case in
+  // particular, the tag will never come back from the completion queue (by
+  // design).
+  call_->StartCall(nullptr);
+
+  if (observer_) {
+    observer_->OnStreamStart();
+    // Start listening for new messages.
+    Read();
+  }
 }
 
 void GrpcStream::Read() {
@@ -203,14 +214,6 @@ GrpcStream::MetadataT GrpcStream::GetResponseHeaders() const {
 
 // Callbacks
 
-void GrpcStream::OnStart() {
-  if (observer_) {
-    observer_->OnStreamStart();
-    // Start listening for new messages.
-    Read();
-  }
-}
-
 void GrpcStream::OnRead(const grpc::ByteBuffer& message) {
   if (observer_) {
     observer_->OnStreamRead(message);

Firestore/core/test/firebase/firestore/remote/grpc_stream_test.cc

@@ -108,7 +108,6 @@ class GrpcStreamTest : public testing::Test {
 
   void StartStream() {
     async_queue().EnqueueBlocking([&] { stream().Start(); });
-    ForceFinish({/*Start*/ Ok});
   }
 
  private:
@@ -221,19 +220,13 @@ TEST_F(GrpcStreamTest, WriteAndFinish) {
   });
 }
 
-TEST_F(GrpcStreamTest, ErrorOnStart) {
-  async_queue().EnqueueBlocking([&] { stream().Start(); });
-  ForceFinish({/*Start*/ Error, /*Finish*/ Ok});
-  EXPECT_EQ(observed_states(), States({"OnStreamError"}));
-}
-
 TEST_F(GrpcStreamTest, ErrorOnWrite) {
   StartStream();
   async_queue().EnqueueBlocking([&] { stream().Write({}); });
 
-  ForceFinish({/*Read*/ Ok, /*Write*/ Error});
+  ForceFinish({/*Write*/ Error, /*Read*/ Error});
   // Give `GrpcStream` a chance to enqueue a finish operation
-  ForceFinish({/*Read*/ Error, /*Finish*/ Ok});
+  ForceFinish({/*Finish*/ Ok});
 
   EXPECT_EQ(observed_states().back(), "OnStreamError");
 }
@@ -245,7 +238,7 @@ TEST_F(GrpcStreamTest, ErrorWithPendingWrites) {
     stream().Write({});
   });
 
-  ForceFinish({/*Read*/ Ok, /*Write*/ Error});
+  ForceFinish({/*Write*/ Ok, /*Write*/ Error});
   // Give `GrpcStream` a chance to enqueue a finish operation
   ForceFinish({/*Read*/ Error, /*Finish*/ Ok});