[fuzzing] Fix api-fuzzer found bug in ClientPromiseBasedCall (#34655)

A little debatable if it's a bug in the fuzzer or the call, but erring
on the side of compatibility (and we'll fix when we drop the batch api).

src/core/lib/promise/pipe.h

@@ -601,7 +601,10 @@ class PipeReceiver {
   // Checks closed from the receivers perspective: that is, if there is a value
   // in the pipe but the pipe is closed, reports open until that value is read.
   auto AwaitClosed() {
-    return [center = center_]() { return center->PollClosedForReceiver(); };
+    return [center = center_]() -> Poll<bool> {
+      if (center == nullptr) return false;
+      return center->PollClosedForReceiver();
+    };
   }
 
   auto AwaitEmpty() {

src/core/lib/surface/call.cc

@@ -2772,7 +2772,9 @@ class ClientPromiseBasedCall final : public PromiseBasedCall {
           "cancel_before_initial_metadata",
           [error = std::move(error), this]() {
             server_to_client_messages_.sender.Close();
-            Finish(ServerMetadataFromStatus(error));
+            auto md = ServerMetadataFromStatus(error);
+            md->Set(GrpcCallWasCancelled(), true);
+            Finish(std::move(md));
             return Empty{};
           },
           [](Empty) {});
@@ -3008,9 +3010,12 @@ void ClientPromiseBasedCall::StartRecvInitialMetadata(
     Party::BulkSpawner& spawner) {
   spawner.Spawn(
       "recv_initial_metadata",
-      Race(server_initial_metadata_.receiver.Next(),
-           Map(finished(),
-               [](Empty) { return NextResult<ServerMetadataHandle>(true); })),
+      [this]() {
+        return Race(server_initial_metadata_.receiver.Next(),
+                    Map(finished(), [](Empty) {
+                      return NextResult<ServerMetadataHandle>(true);
+                    }));
+      },
       [this, array,
        completion =
            AddOpToCompletion(completion, PendingOp::kReceiveInitialMetadata)](
@@ -3049,6 +3054,7 @@ void ClientPromiseBasedCall::Finish(ServerMetadataHandle trailing_metadata) {
   client_to_server_messages_.receiver.CloseWithError();
   if (trailing_metadata->get(GrpcCallWasCancelled()).value_or(false)) {
     server_to_client_messages_.receiver.CloseWithError();
+    server_initial_metadata_.receiver.CloseWithError();
   }
   if (auto* channelz_channel = channel()->channelz_node()) {
     if (trailing_metadata->get(GrpcStatusMetadata())

test/core/end2end/fuzzers/api_fuzzer_corpus/5782679267115008

@@ -0,0 +1,27 @@
+actions {
+  create_channel {
+    target: "unix:"
+    channel_creds {
+    }
+  }
+}
+actions {
+  create_call {
+    propagation_mask: 9869440
+    method {
+      value: "v"
+    }
+    timeout: 1000000000
+  }
+}
+actions {
+  queue_batch {
+    operations {
+      receive_message {
+      }
+    }
+  }
+}
+config_vars {
+  experiments: 390842023935
+}