gRPC: in Datastore, delay serializing domain object to a grpc::ByteBuffer right until the call is started (#1949)

(see [here](#1935 (comment)) and [here](grpc/grpc#16875) for context)

The problem with serializing a domain object immediately is that the resulting `ByteBuffer` is stored in a `std::function` within Auth. `ByteBuffer`s become invalid once gRPC core shuts down, so if Auth happens to outlive Firestore, once the `ByteBuffer`'s destructor is invoked, the app will crash.

Author: var-const

Firestore/core/src/firebase/firestore/remote/datastore.h

@@ -120,14 +120,14 @@ class Datastore : public std::enable_shared_from_this<Datastore> {
   void PollGrpcQueue();
 
   void CommitMutationsWithCredentials(const auth::Token& token,
-                                      const grpc::ByteBuffer& message,
+                                      NSArray<FSTMutation*>* mutations,
                                       FSTVoidErrorBlock completion);
   void OnCommitMutationsResponse(const util::StatusOr<grpc::ByteBuffer>& result,
                                  FSTVoidErrorBlock completion);
 
   void LookupDocumentsWithCredentials(
       const auth::Token& token,
-      const grpc::ByteBuffer& message,
+      const std::vector<model::DocumentKey>& keys,
       FSTVoidMaybeDocumentArrayErrorBlock completion);
   void OnLookupDocumentsResponse(
       const util::StatusOr<std::vector<grpc::ByteBuffer>>& result,

Firestore/core/src/firebase/firestore/remote/datastore.mm

@@ -151,23 +151,23 @@ void LogGrpcCallFinished(absl::string_view rpc_name,
 
 void Datastore::CommitMutations(NSArray<FSTMutation*>* mutations,
                                 FSTVoidErrorBlock completion) {
-  grpc::ByteBuffer message = serializer_bridge_.ToByteBuffer(
-      serializer_bridge_.CreateCommitRequest(mutations));
-
   ResumeRpcWithCredentials(
-      [this, message, completion](const StatusOr<Token>& maybe_credentials) {
+      [this, mutations, completion](const StatusOr<Token>& maybe_credentials) {
         if (!maybe_credentials.ok()) {
           completion(util::MakeNSError(maybe_credentials.status()));
           return;
         }
-        CommitMutationsWithCredentials(maybe_credentials.ValueOrDie(), message,
-                                       completion);
+        CommitMutationsWithCredentials(maybe_credentials.ValueOrDie(),
+                                       mutations, completion);
       });
 }
 
 void Datastore::CommitMutationsWithCredentials(const Token& token,
-                                               const grpc::ByteBuffer& message,
+                                               NSArray<FSTMutation*>* mutations,
                                                FSTVoidErrorBlock completion) {
+  grpc::ByteBuffer message = serializer_bridge_.ToByteBuffer(
+      serializer_bridge_.CreateCommitRequest(mutations));
+
   std::unique_ptr<GrpcUnaryCall> call_owning = grpc_connection_.CreateUnaryCall(
       kRpcNameCommit, token, std::move(message));
   GrpcUnaryCall* call = call_owning.get();
@@ -196,24 +196,24 @@ void LogGrpcCallFinished(absl::string_view rpc_name,
 void Datastore::LookupDocuments(
     const std::vector<DocumentKey>& keys,
     FSTVoidMaybeDocumentArrayErrorBlock completion) {
-  grpc::ByteBuffer message = serializer_bridge_.ToByteBuffer(
-      serializer_bridge_.CreateLookupRequest(keys));
-
   ResumeRpcWithCredentials(
-      [this, message, completion](const StatusOr<Token>& maybe_credentials) {
+      [this, keys, completion](const StatusOr<Token>& maybe_credentials) {
         if (!maybe_credentials.ok()) {
           completion(nil, util::MakeNSError(maybe_credentials.status()));
           return;
         }
-        LookupDocumentsWithCredentials(maybe_credentials.ValueOrDie(), message,
+        LookupDocumentsWithCredentials(maybe_credentials.ValueOrDie(), keys,
                                        completion);
       });
 }
 
 void Datastore::LookupDocumentsWithCredentials(
     const Token& token,
-    const grpc::ByteBuffer& message,
+    const std::vector<DocumentKey>& keys,
     FSTVoidMaybeDocumentArrayErrorBlock completion) {
+  grpc::ByteBuffer message = serializer_bridge_.ToByteBuffer(
+      serializer_bridge_.CreateLookupRequest(keys));
+
   std::unique_ptr<GrpcStreamingReader> call_owning =
       grpc_connection_.CreateStreamingReader(kRpcNameLookup, token,
                                              std::move(message));

Firestore/core/test/firebase/firestore/remote/datastore_test.mm

@@ -340,10 +340,7 @@ void ForceFinishAnyTypeOrder(
   EXPECT_NO_THROW(credentials.InvokeGetToken());
 }
 
-// TODO(varconst): this test currently fails due to a gRPC issue, see here
-// https://github.com/firebase/firebase-ios-sdk/pull/1935#discussion_r224900667
-// for details. Reenable when/if possible.
-TEST_F(DatastoreTest, DISABLED_AuthOutlivesDatastore) {
+TEST_F(DatastoreTest, AuthOutlivesDatastore) {
   credentials.DelayGetToken();
 
   worker_queue.EnqueueBlocking([&] {