[call-v3] Implement failed_before_recv_message() on {Client,Server}Call (#37810)

Closes #37810

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/37810 from ctiller:CANCELLED 134f49a07b
PiperOrigin-RevId: 681236011

src/core/lib/surface/client_call.h

@@ -98,7 +98,9 @@ class ClientCall final
   char* GetPeer() override;
 
   bool Completed() final { Crash("unimplemented"); }
-  bool failed_before_recv_message() const final { Crash("unimplemented"); }
+  bool failed_before_recv_message() const final {
+    return started_call_initiator_.WasCancelledPushed();
+  }
 
   grpc_compression_algorithm incoming_compression_algorithm() override {
     return message_receiver_.incoming_compression_algorithm();

src/core/lib/surface/server_call.h

@@ -129,7 +129,9 @@ class ServerCall final : public Call, public DualRefCounted<ServerCall> {
   }
 
   bool Completed() final { Crash("unimplemented"); }
-  bool failed_before_recv_message() const final { Crash("unimplemented"); }
+  bool failed_before_recv_message() const final {
+    return call_handler_.WasCancelledPushed();
+  }
 
   uint32_t test_only_message_flags() override {
     return message_receiver_.last_message_flags();

src/core/lib/transport/call_filters.h

@@ -1590,10 +1590,17 @@ class CallFilters {
   GRPC_MUST_USE_RESULT auto WasCancelled() {
     return [this]() { return call_state_.PollWasCancelled(); };
   }
+  // Client & server: returns true if server trailing metadata has been pushed
+  // *and* contained a cancellation, false otherwise.
+  GRPC_MUST_USE_RESULT bool WasCancelledPushed() const {
+    return call_state_.WasCancelledPushed();
+  }
+
   // Returns true if server trailing metadata has been pulled
   bool WasServerTrailingMetadataPulled() const {
     return call_state_.WasServerTrailingMetadataPulled();
   }
+
   // Client & server: fill in final_info with the final status of the call.
   void Finalize(const grpc_call_final_info* final_info);
 

src/core/lib/transport/call_spine.h

@@ -263,6 +263,10 @@ class CallInitiator {
     return spine_->SpawnWaitable(name, std::move(promise_factory));
   }
 
+  bool WasCancelledPushed() const {
+    return spine_->call_filters().WasCancelledPushed();
+  }
+
   Arena* arena() { return spine_->arena(); }
   Party* party() { return spine_.get(); }
 
@@ -304,6 +308,10 @@ class CallHandler {
 
   auto WasCancelled() { return spine_->WasCancelled(); }
 
+  bool WasCancelledPushed() const {
+    return spine_->call_filters().WasCancelledPushed();
+  }
+
   template <typename PromiseFactory>
   void SpawnGuarded(absl::string_view name, PromiseFactory promise_factory,
                     DebugLocation whence = {}) {

src/core/lib/transport/call_state.h

@@ -52,8 +52,14 @@ class CallState {
   Poll<ValueOrFailure<bool>> PollPullServerToClientMessageAvailable();
   void FinishPullServerToClientMessage();
   Poll<Empty> PollServerTrailingMetadataAvailable();
+  void FinishPullServerTrailingMetadata();
   bool WasServerTrailingMetadataPulled() const;
+  // Resolves after server trailing metadata has been pulled, to true if the
+  // call was cancelled, and false otherwise.
   Poll<bool> PollWasCancelled();
+  // Return true if server trailing metadata has been pushed *and* that push was
+  // a cancellation.
+  bool WasCancelledPushed() const;
   // Debug
   std::string DebugString() const;
 
@@ -939,6 +945,23 @@ CallState::PollWasCancelled() {
   Crash("Unreachable");
 }
 
+GPR_ATTRIBUTE_ALWAYS_INLINE_FUNCTION inline bool CallState::WasCancelledPushed()
+    const {
+  GRPC_TRACE_LOG(call_state, INFO)
+      << "[call_state] PollWasCancelledPushed: "
+      << GRPC_DUMP_ARGS(this, server_trailing_metadata_state_);
+  switch (server_trailing_metadata_state_) {
+    case ServerTrailingMetadataState::kNotPushed:
+    case ServerTrailingMetadataState::kPulled:
+    case ServerTrailingMetadataState::kPushed:
+      return false;
+    case ServerTrailingMetadataState::kPushedCancel:
+    case ServerTrailingMetadataState::kPulledCancel:
+      return true;
+  }
+  Crash("Unreachable");
+}
+
 }  // namespace grpc_core
 
 #endif  // GRPC_SRC_CORE_LIB_TRANSPORT_CALL_STATE_H

src/core/util/dump_args.h

@@ -78,20 +78,25 @@ class DumpArgs {
     return 0;
   }
 
-  int AddDumper(void** p) {
+  int AddDumper(void const* const* p) {
     arg_dumpers_.push_back(
         [p](CustomSink& os) { os.Append(absl::StrFormat("%p", *p)); });
     return 0;
   }
 
   template <typename T>
-  int AddDumper(T** p) {
-    return AddDumper(reinterpret_cast<void**>(p));
+  int AddDumper(T const* const* p) {
+    return AddDumper(reinterpret_cast<void const* const*>(p));
   }
 
   template <typename T>
   int AddDumper(T* const* p) {
-    return AddDumper(const_cast<T**>(p));
+    return AddDumper(const_cast<T const* const*>(p));
+  }
+
+  template <typename T>
+  int AddDumper(T const** p) {
+    return AddDumper(const_cast<T const* const*>(p));
   }
 
   void Stringify(CustomSink& sink) const;

test/core/transport/call_state_test.cc

@@ -262,11 +262,14 @@ TEST(CallStateTest, RecallNoCancellation) {
   activity.Activate();
   CallState state;
   state.Start();
+  EXPECT_EQ(state.WasCancelledPushed(), false);
   state.PushServerTrailingMetadata(false);
+  EXPECT_EQ(state.WasCancelledPushed(), false);
   EXPECT_THAT(state.PollPullServerInitialMetadataAvailable(), IsReady(false));
   state.FinishPullServerInitialMetadata();
   EXPECT_THAT(state.PollServerTrailingMetadataAvailable(), IsReady());
   EXPECT_THAT(state.PollWasCancelled(), IsReady(false));
+  EXPECT_EQ(state.WasCancelledPushed(), false);
 }
 
 TEST(CallStateTest, RecallCancellation) {
@@ -274,11 +277,14 @@ TEST(CallStateTest, RecallCancellation) {
   activity.Activate();
   CallState state;
   state.Start();
+  EXPECT_EQ(state.WasCancelledPushed(), false);
   state.PushServerTrailingMetadata(true);
+  EXPECT_EQ(state.WasCancelledPushed(), true);
   EXPECT_THAT(state.PollPullServerInitialMetadataAvailable(), IsReady(false));
   state.FinishPullServerInitialMetadata();
   EXPECT_THAT(state.PollServerTrailingMetadataAvailable(), IsReady());
   EXPECT_THAT(state.PollWasCancelled(), IsReady(true));
+  EXPECT_EQ(state.WasCancelledPushed(), true);
 }
 
 TEST(CallStateTest, ReceiveTrailingMetadataAfterMessageRead) {