[promises] Implementation of deadline for client based calls (#31902)

* [promises] Implementation of deadline for client based calls

* fix

* Update call.cc

src/core/ext/filters/deadline/deadline_filter.cc

@@ -350,7 +350,10 @@ static void deadline_server_start_transport_stream_op_batch(
 
 const grpc_channel_filter grpc_client_deadline_filter = {
     deadline_client_start_transport_stream_op_batch,
-    nullptr,
+    [](grpc_channel_element*, grpc_core::CallArgs call_args,
+       grpc_core::NextPromiseFactory next_promise_factory) {
+      return next_promise_factory(std::move(call_args));
+    },
     grpc_channel_next_op,
     sizeof(base_call_data),
     deadline_init_call_elem,

src/core/lib/surface/call.cc

@@ -2384,6 +2384,8 @@ void PromiseBasedCall::ResetDeadline() {
 }
 
 void PromiseBasedCall::Run() {
+  ApplicationCallbackExecCtx callback_exec_ctx;
+  ExecCtx exec_ctx;
   CancelWithError(absl::DeadlineExceededError("Deadline exceeded"));
   InternalUnref("deadline");
 }
@@ -2424,6 +2426,9 @@ class ClientPromiseBasedCall final : public PromiseBasedCall {
     if (auto* channelz_channel = channel()->channelz_node()) {
       channelz_channel->RecordCallStarted();
     }
+    if (args->send_deadline != Timestamp::InfFuture()) {
+      UpdateDeadline(args->send_deadline);
+    }
   }
 
   ~ClientPromiseBasedCall() override {
@@ -2880,6 +2885,9 @@ void ClientPromiseBasedCall::PublishStatus(
         gpr_strdup(MakeErrorString(trailing_metadata.get()).c_str());
   }
   PublishMetadataArray(op_args.trailing_metadata, trailing_metadata.get());
+  // Clear state saying we have a RECV_STATUS_ON_CLIENT outstanding
+  // (so we don't call through twice)
+  recv_status_on_client_ = absl::monostate();
   FinishOpOnCompletion(&recv_status_on_client_completion_,
                        PendingOp::kReceiveStatusOnClient);
 }