[call-v3] Updates to CallFilter (#36240)

Updates to `CallFilter` to get it ready for integration to `CallSpine`.

This is a bundle commit grabbed from my working branch so there's a few things rolled in here:

* PipeState picks up methods to close it cleanly and in error, and expands its Pull API to signify that error
* Add a `NextMessage` type that encapsulates one message pulled from a pipe (much like `Pipe::NextResult` - except that we know this is a message, and it's built on the `CallFilters` types)
* Add debug stringifiers to many CallFilters types, and better tracing
* Eliminate unused `CallFilters::PipePromise::Pull` type -- other variants are better for all cases
* Add support for filters having the signature `(Handle, Filter*) -> promise returning StatusOr<Handle>` - will support `ClientAuthFilter` in a future commit asynchronously mutating client initial metadata

Closes #36240

COPYBARA_INTEGRATE_REVIEW=https://github.com/grpc/grpc/pull/36240 from ctiller:cally-1 aa5fe997bf
PiperOrigin-RevId: 623194327

build_autogenerated.yaml

@@ -6411,6 +6411,7 @@ targets:
   - src/core/lib/promise/detail/seq_state.h
   - src/core/lib/promise/detail/status.h
   - src/core/lib/promise/exec_ctx_wakeup_scheduler.h
+  - src/core/lib/promise/latch.h
   - src/core/lib/promise/loop.h
   - src/core/lib/promise/map.h
   - src/core/lib/promise/poll.h

src/core/BUILD

@@ -7252,6 +7252,8 @@ grpc_cc_library(
     ],
     deps = [
         "call_final_info",
+        "latch",
+        "map",
         "message",
         "metadata",
         "ref_counted",

src/core/lib/promise/status_flag.h

@@ -176,6 +176,14 @@ class ValueOrFailure {
     return value_ == other.value_;
   }
 
+  bool operator!=(const ValueOrFailure& other) const {
+    return value_ != other.value_;
+  }
+
+  bool operator==(const T& other) const { return value_ == other; }
+
+  bool operator!=(const T& other) const { return value_ != other; }
+
  private:
   absl::optional<T> value_;
 };

src/core/lib/transport/call_filters.cc

@@ -202,15 +202,53 @@ void CallFilters::Finalize(const grpc_call_final_info* final_info) {
   }
 }
 
-void CallFilters::CancelDueToFailedPipeOperation() {
+void CallFilters::CancelDueToFailedPipeOperation(SourceLocation but_where) {
   // We expect something cancelled before now
   if (server_trailing_metadata_ == nullptr) return;
-  gpr_log(GPR_DEBUG, "Cancelling due to failed pipe operation");
+  if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+    gpr_log(but_where.file(), but_where.line(), GPR_LOG_SEVERITY_DEBUG,
+            "Cancelling due to failed pipe operation: %s",
+            DebugString().c_str());
+  }
   server_trailing_metadata_ =
       ServerMetadataFromStatus(absl::CancelledError("Failed pipe operation"));
   server_trailing_metadata_waiter_.Wake();
 }
 
+void CallFilters::PushServerTrailingMetadata(ServerMetadataHandle md) {
+  if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+    gpr_log(GPR_DEBUG, "%s Push server trailing metadata: %s into %s",
+            GetContext<Activity>()->DebugTag().c_str(),
+            md->DebugString().c_str(), DebugString().c_str());
+  }
+  GPR_ASSERT(md != nullptr);
+  if (server_trailing_metadata_ != nullptr) return;
+  server_trailing_metadata_ = std::move(md);
+  client_initial_metadata_state_.CloseWithError();
+  server_initial_metadata_state_.CloseSending();
+  client_to_server_message_state_.CloseWithError();
+  server_to_client_message_state_.CloseWithError();
+  server_trailing_metadata_waiter_.Wake();
+}
+
+std::string CallFilters::DebugString() const {
+  std::vector<std::string> components = {
+      absl::StrFormat("this:%p", this),
+      absl::StrCat("client_initial_metadata:",
+                   client_initial_metadata_state_.DebugString()),
+      ServerInitialMetadataPromises::DebugString("server_initial_metadata",
+                                                 this),
+      ClientToServerMessagePromises::DebugString("client_to_server_message",
+                                                 this),
+      ServerToClientMessagePromises::DebugString("server_to_client_message",
+                                                 this),
+      absl::StrCat("server_trailing_metadata:",
+                   server_trailing_metadata_ == nullptr
+                       ? "not-set"
+                       : server_trailing_metadata_->DebugString())};
+  return absl::StrCat("CallFilters{", absl::StrJoin(components, ", "), "}");
+};
+
 ///////////////////////////////////////////////////////////////////////////////
 // CallFilters::Stack
 
@@ -251,6 +289,49 @@ void filters_detail::PipeState::Start() {
   wait_recv_.Wake();
 }
 
+void filters_detail::PipeState::CloseWithError() {
+  if (state_ == ValueState::kClosed) return;
+  state_ = ValueState::kError;
+  wait_recv_.Wake();
+  wait_send_.Wake();
+}
+
+Poll<bool> filters_detail::PipeState::PollClosed() {
+  switch (state_) {
+    case ValueState::kIdle:
+    case ValueState::kWaiting:
+    case ValueState::kQueued:
+    case ValueState::kReady:
+    case ValueState::kProcessing:
+      return wait_recv_.pending();
+    case ValueState::kClosed:
+      return false;
+    case ValueState::kError:
+      return true;
+  }
+  GPR_UNREACHABLE_CODE(return Pending{});
+}
+
+void filters_detail::PipeState::CloseSending() {
+  switch (state_) {
+    case ValueState::kIdle:
+      state_ = ValueState::kClosed;
+      break;
+    case ValueState::kWaiting:
+      state_ = ValueState::kClosed;
+      wait_recv_.Wake();
+      break;
+    case ValueState::kClosed:
+    case ValueState::kError:
+      break;
+    case ValueState::kQueued:
+    case ValueState::kReady:
+    case ValueState::kProcessing:
+      Crash("Only one push allowed to be outstanding");
+      break;
+  }
+}
+
 void filters_detail::PipeState::BeginPush() {
   switch (state_) {
     case ValueState::kIdle:
@@ -320,7 +401,7 @@ Poll<StatusFlag> filters_detail::PipeState::PollPush() {
   GPR_UNREACHABLE_CODE(return Pending{});
 }
 
-Poll<StatusFlag> filters_detail::PipeState::PollPull() {
+Poll<ValueOrFailure<bool>> filters_detail::PipeState::PollPull() {
   switch (state_) {
     case ValueState::kWaiting:
       return wait_recv_.pending();
@@ -331,10 +412,11 @@ Poll<StatusFlag> filters_detail::PipeState::PollPull() {
     case ValueState::kQueued:
       if (!started_) return wait_recv_.pending();
       state_ = ValueState::kProcessing;
-      return Success{};
+      return true;
     case ValueState::kProcessing:
       Crash("Only one pull allowed to be outstanding");
     case ValueState::kClosed:
+      return false;
     case ValueState::kError:
       return Failure{};
   }
@@ -358,4 +440,32 @@ void filters_detail::PipeState::AckPull() {
   }
 }
 
+std::string filters_detail::PipeState::DebugString() const {
+  const char* state_str = "<<invalid-value>>";
+  switch (state_) {
+    case ValueState::kIdle:
+      state_str = "Idle";
+      break;
+    case ValueState::kWaiting:
+      state_str = "Waiting";
+      break;
+    case ValueState::kQueued:
+      state_str = "Queued";
+      break;
+    case ValueState::kReady:
+      state_str = "Ready";
+      break;
+    case ValueState::kProcessing:
+      state_str = "Processing";
+      break;
+    case ValueState::kClosed:
+      state_str = "Closed";
+      break;
+    case ValueState::kError:
+      state_str = "Error";
+      break;
+  }
+  return absl::StrCat(state_str, started_ ? "" : " (not started)");
+}
+
 }  // namespace grpc_core

src/core/lib/transport/call_filters.h

@@ -23,6 +23,8 @@
 
 #include "src/core/lib/gprpp/ref_counted.h"
 #include "src/core/lib/gprpp/ref_counted_ptr.h"
+#include "src/core/lib/promise/latch.h"
+#include "src/core/lib/promise/map.h"
 #include "src/core/lib/promise/promise.h"
 #include "src/core/lib/promise/status_flag.h"
 #include "src/core/lib/transport/call_final_info.h"
@@ -771,6 +773,55 @@ struct AddOpImpl<
   }
 };
 
+// PROMISE_RETURNING(absl::StatusOr<$VALUE_HANDLE>)
+// $INTERCEPTOR_NAME($VALUE_HANDLE, FilterType*)
+template <typename FilterType, typename T, typename R,
+          R (FilterType::Call::*impl)(T, FilterType*)>
+struct AddOpImpl<FilterType, T, R (FilterType::Call::*)(T, FilterType*), impl,
+                 absl::enable_if_t<std::is_same<absl::StatusOr<T>,
+                                                PromiseResult<R>>::value>> {
+  static void Add(FilterType* channel_data, size_t call_offset,
+                  Layout<FallibleOperator<T>>& to) {
+    class Promise {
+     public:
+      Promise(T value, typename FilterType::Call* call_data,
+              FilterType* channel_data)
+          : impl_((call_data->*impl)(std::move(value), channel_data)) {}
+
+      Poll<ResultOr<T>> PollOnce() {
+        auto p = impl_();
+        auto* r = p.value_if_ready();
+        if (r == nullptr) return Pending{};
+        this->~Promise();
+        if (r->ok()) return ResultOr<T>{std::move(**r), nullptr};
+        return ResultOr<T>{nullptr, ServerMetadataFromStatus(r->status())};
+      }
+
+     private:
+      GPR_NO_UNIQUE_ADDRESS R impl_;
+    };
+    to.Add(sizeof(Promise), alignof(Promise),
+           FallibleOperator<T>{
+               channel_data,
+               call_offset,
+               [](void* promise_data, void* call_data, void* channel_data,
+                  T value) -> Poll<ResultOr<T>> {
+                 auto* promise = new (promise_data)
+                     Promise(std::move(value),
+                             static_cast<typename FilterType::Call*>(call_data),
+                             static_cast<FilterType*>(channel_data));
+                 return promise->PollOnce();
+               },
+               [](void* promise_data) {
+                 return static_cast<Promise*>(promise_data)->PollOnce();
+               },
+               [](void* promise_data) {
+                 static_cast<Promise*>(promise_data)->~Promise();
+               },
+           });
+  }
+};
+
 struct ChannelDataDestructor {
   void (*destroy)(void* channel_data);
   void* channel_data;
@@ -783,7 +834,7 @@ struct ChannelDataDestructor {
 // in-flight calls.
 struct StackData {
   // Overall size and alignment of the call data for this stack.
-  size_t call_data_alignment = 0;
+  size_t call_data_alignment = 1;
   size_t call_data_size = 0;
   // A complete list of filters for this call, so that we can construct the
   // call data for each filter.
@@ -1104,14 +1155,25 @@ class PipeState {
   void DropPush();
   // Poll for push completion: occurs after the corresponding Pull()
   Poll<StatusFlag> PollPush();
-  Poll<StatusFlag> PollPull();
+  // Poll for pull completion; returns Failure{} if closed with error,
+  // true if a value is available, or false if the pipe was closed without
+  // error.
+  Poll<ValueOrFailure<bool>> PollPull();
   // A pulled value has been consumed: we can unblock the push
   void AckPull();
   // A previously started pull operation has completed
   void DropPull();
+  // Close sending
+  void CloseSending();
+  // Close sending with error
+  void CloseWithError();
+  // Poll for closedness - if true, closed with error
+  Poll<bool> PollClosed();
 
   bool holds_error() const { return state_ == ValueState::kError; }
 
+  std::string DebugString() const;
+
  private:
   enum class ValueState : uint8_t {
     // Nothing sending nor receiving
@@ -1248,6 +1310,44 @@ class CallFilters {
     filters_detail::StackData data_;
   };
 
+  class NextMessage {
+   public:
+    NextMessage() : has_value_(false), cancelled_(false) {}
+    explicit NextMessage(MessageHandle value)
+        : has_value_(true), value_(std::move(value)) {}
+    explicit NextMessage(bool cancelled)
+        : has_value_(false), cancelled_(cancelled) {}
+    NextMessage(const NextMessage&) = delete;
+    NextMessage& operator=(const NextMessage&) = delete;
+    NextMessage(NextMessage&& other) noexcept = default;
+    NextMessage& operator=(NextMessage&& other) = default;
+
+    using value_type = MessageHandle;
+
+    void reset() {
+      has_value_ = false;
+      cancelled_ = false;
+      value_.reset();
+    }
+    bool has_value() const { return has_value_; }
+    const MessageHandle& value() const {
+      GPR_DEBUG_ASSERT(has_value_);
+      return value_;
+    }
+    MessageHandle& value() {
+      GPR_DEBUG_ASSERT(has_value_);
+      return value_;
+    }
+    const MessageHandle& operator*() const { return value(); }
+    MessageHandle& operator*() { return value(); }
+    bool cancelled() const { return !has_value_ && cancelled_; }
+
+   private:
+    bool has_value_;
+    bool cancelled_;
+    MessageHandle value_;
+  };
+
   explicit CallFilters(ClientMetadataHandle client_initial_metadata);
   ~CallFilters();
 
@@ -1258,25 +1358,59 @@ class CallFilters {
 
   void SetStack(RefCountedPtr<Stack> stack);
 
+  // Access client initial metadata before it's processed
   ClientMetadata* unprocessed_client_initial_metadata() {
     return client_initial_metadata_.get();
   }
+
+  // Client: Fetch client initial metadata
+  // Returns a promise that resolves to ValueOrFailure<ClientMetadataHandle>
   GRPC_MUST_USE_RESULT auto PullClientInitialMetadata();
+  // Server: Indicate that no server initial metadata will be sent
+  void NoServerInitialMetadata() {
+    server_initial_metadata_state_.CloseSending();
+  }
+  // Server: Push server initial metadata
+  // Returns a promise that resolves to a StatusFlag indicating success
   GRPC_MUST_USE_RESULT auto PushServerInitialMetadata(ServerMetadataHandle md);
+  // Client: Fetch server initial metadata
+  // Returns a promise that resolves to ValueOrFailure<ServerMetadataHandle>
   GRPC_MUST_USE_RESULT auto PullServerInitialMetadata();
+  // Client: Push client to server message
+  // Returns a promise that resolves to a StatusFlag indicating success
   GRPC_MUST_USE_RESULT auto PushClientToServerMessage(MessageHandle message);
+  // Client: Indicate that no more messages will be sent
+  void FinishClientToServerSends() {
+    client_to_server_message_state_.CloseSending();
+  }
+  // Server: Fetch client to server message
+  // Returns a promise that resolves to ValueOrFailure<MessageHandle>
   GRPC_MUST_USE_RESULT auto PullClientToServerMessage();
+  // Server: Push server to client message
+  // Returns a promise that resolves to a StatusFlag indicating success
   GRPC_MUST_USE_RESULT auto PushServerToClientMessage(MessageHandle message);
+  // Server: Fetch server to client message
+  // Returns a promise that resolves to ValueOrFailure<MessageHandle>
   GRPC_MUST_USE_RESULT auto PullServerToClientMessage();
-  void PushServerTrailingMetadata(ServerMetadataHandle md) {
-    GPR_ASSERT(md != nullptr);
-    if (server_trailing_metadata_ != nullptr) return;
-    server_trailing_metadata_ = std::move(md);
-    server_trailing_metadata_waiter_.Wake();
-  }
+  // Server: Indicate end of response
+  // Closes the request entirely - no messages can be sent/received
+  // If no server initial metadata has been sent, implies
+  // NoServerInitialMetadata() called.
+  void PushServerTrailingMetadata(ServerMetadataHandle md);
+  // Client: Fetch server trailing metadata
+  // Returns a promise that resolves to ServerMetadataHandle
   GRPC_MUST_USE_RESULT auto PullServerTrailingMetadata();
+  // Server: Wait for server trailing metadata to have been sent
+  // Returns a promise that resolves to a StatusFlag indicating whether the
+  // request was cancelled or not -- failure to send trailing metadata is
+  // considered a cancellation, as is actual cancellation -- but not application
+  // errors.
+  GRPC_MUST_USE_RESULT auto WasCancelled();
+  // Client & server: fill in final_info with the final status of the call.
   void Finalize(const grpc_call_final_info* final_info);
 
+  std::string DebugString() const;
+
  private:
   template <filters_detail::PipeState(CallFilters::*state_ptr),
             void*(CallFilters::*push_ptr), typename T,
@@ -1315,6 +1449,10 @@ class CallFilters {
 
       T TakeValue() { return std::move(value_); }
 
+      absl::string_view DebugString() const {
+        return value_ != nullptr ? " (not pulled)" : "";
+      }
+
      private:
       filters_detail::PipeState& state() { return filters_->*state_ptr; }
       void*& push_slot() { return filters_->*push_ptr; }
@@ -1323,24 +1461,36 @@ class CallFilters {
       T value_;
     };
 
-    class Pull {
+    static std::string DebugString(absl::string_view name,
+                                   const CallFilters* filters) {
+      auto* push = static_cast<Push*>(filters->*push_ptr);
+      return absl::StrCat(name, ":", (filters->*state_ptr).DebugString(),
+                          push == nullptr ? "" : push->DebugString());
+    }
+
+    class PullMaybe {
      public:
-      explicit Pull(CallFilters* filters) : filters_(filters) {}
-      ~Pull() {
+      explicit PullMaybe(CallFilters* filters) : filters_(filters) {}
+      ~PullMaybe() {
         if (filters_ != nullptr) {
           state().DropPull();
         }
       }
 
-      Pull(const Pull&) = delete;
-      Pull& operator=(const Pull&) = delete;
-      Pull(Pull&& other) noexcept
+      PullMaybe(const PullMaybe&) = delete;
+      PullMaybe& operator=(const PullMaybe&) = delete;
+      PullMaybe(PullMaybe&& other) noexcept
           : filters_(std::exchange(other.filters_, nullptr)),
             executor_(std::move(other.executor_)) {}
-      Pull& operator=(Pull&&) = delete;
+      PullMaybe& operator=(PullMaybe&&) = delete;
 
-      Poll<ValueOrFailure<T>> operator()() {
+      Poll<ValueOrFailure<absl::optional<T>>> operator()() {
         if (executor_.IsRunning()) {
+          auto c = state().PollClosed();
+          if (c.ready() && c.value()) {
+            filters_->CancelDueToFailedPipeOperation();
+            return Failure{};
+          }
           return FinishOperationExecutor(executor_.Step(filters_->call_data_));
         }
         auto p = state().PollPull();
@@ -1350,6 +1500,7 @@ class CallFilters {
           filters_->CancelDueToFailedPipeOperation();
           return Failure{};
         }
+        if (!**r) return absl::nullopt;
         return FinishOperationExecutor(executor_.Start(
             layout(), push()->TakeValue(), filters_->call_data_));
       }
@@ -1362,7 +1513,7 @@ class CallFilters {
         return &(filters_->stack_->data_.*layout_ptr);
       }
 
-      Poll<ValueOrFailure<T>> FinishOperationExecutor(
+      Poll<ValueOrFailure<absl::optional<T>>> FinishOperationExecutor(
           Poll<filters_detail::ResultOr<T>> p) {
         auto* r = p.value_if_ready();
         if (r == nullptr) return Pending{};
@@ -1376,6 +1527,66 @@ class CallFilters {
       CallFilters* filters_;
       filters_detail::OperationExecutor<T> executor_;
     };
+
+    class PullMessage {
+     public:
+      explicit PullMessage(CallFilters* filters) : filters_(filters) {}
+      ~PullMessage() {
+        if (filters_ != nullptr) {
+          state().DropPull();
+        }
+      }
+
+      PullMessage(const PullMessage&) = delete;
+      PullMessage& operator=(const PullMessage&) = delete;
+      PullMessage(PullMessage&& other) noexcept
+          : filters_(std::exchange(other.filters_, nullptr)),
+            executor_(std::move(other.executor_)) {}
+      PullMessage& operator=(PullMessage&&) = delete;
+
+      Poll<NextMessage> operator()() {
+        if (executor_.IsRunning()) {
+          auto c = state().PollClosed();
+          if (c.ready() && c.value()) {
+            filters_->CancelDueToFailedPipeOperation();
+            return NextMessage(true);
+          }
+          return FinishOperationExecutor(executor_.Step(filters_->call_data_));
+        }
+        auto p = state().PollPull();
+        auto* r = p.value_if_ready();
+        if (r == nullptr) return Pending{};
+        if (!r->ok()) {
+          filters_->CancelDueToFailedPipeOperation();
+          return NextMessage(true);
+        }
+        if (!**r) return NextMessage(false);
+        return FinishOperationExecutor(executor_.Start(
+            layout(), push()->TakeValue(), filters_->call_data_));
+      }
+
+     private:
+      filters_detail::PipeState& state() { return filters_->*state_ptr; }
+      Push* push() { return static_cast<Push*>(filters_->*push_ptr); }
+      const filters_detail::Layout<filters_detail::FallibleOperator<T>>*
+      layout() {
+        return &(filters_->stack_->data_.*layout_ptr);
+      }
+
+      Poll<NextMessage> FinishOperationExecutor(
+          Poll<filters_detail::ResultOr<T>> p) {
+        auto* r = p.value_if_ready();
+        if (r == nullptr) return Pending{};
+        GPR_DEBUG_ASSERT(!executor_.IsRunning());
+        state().AckPull();
+        if (r->ok != nullptr) return NextMessage(std::move(r->ok));
+        filters_->PushServerTrailingMetadata(std::move(r->error));
+        return NextMessage(true);
+      }
+
+      CallFilters* filters_;
+      filters_detail::OperationExecutor<T> executor_;
+    };
   };
 
   class PullClientInitialMetadataPromise {
@@ -1400,7 +1611,12 @@ class CallFilters {
       }
       auto p = state().PollPull();
       auto* r = p.value_if_ready();
-      gpr_log(GPR_INFO, "%s", r == nullptr ? "PENDING" : r->ToString().c_str());
+      if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+        gpr_log(GPR_INFO, "%s",
+                r == nullptr
+                    ? "PENDING"
+                    : (r->ok() ? (r->value() ? "TRUE" : "FALSE") : "FAILURE"));
+      }
       if (r == nullptr) return Pending{};
       if (!r->ok()) {
         filters_->CancelDueToFailedPipeOperation();
@@ -1450,11 +1666,39 @@ class CallFilters {
 
     Poll<ServerMetadataHandle> operator()() {
       if (executor_.IsRunning()) {
-        return executor_.Step(filters_->call_data_);
+        auto r = executor_.Step(filters_->call_data_);
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+          if (r.pending()) {
+            gpr_log(GPR_INFO,
+                    "%s PullServerTrailingMetadata[%p]: Pending(but executing)",
+                    GetContext<Activity>()->DebugTag().c_str(), filters_);
+          } else {
+            gpr_log(GPR_INFO, "%s PullServerTrailingMetadata[%p]: Ready: %s",
+                    GetContext<Activity>()->DebugTag().c_str(), filters_,
+                    r.value()->DebugString().c_str());
+          }
+        }
+        return r;
       }
       if (filters_->server_trailing_metadata_ == nullptr) {
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+          gpr_log(GPR_INFO,
+                  "%s PullServerTrailingMetadata[%p]: Pending(not pushed)",
+                  GetContext<Activity>()->DebugTag().c_str(), filters_);
+        }
         return filters_->server_trailing_metadata_waiter_.pending();
       }
+      // If no stack has been set, we can just return the result of the call
+      if (filters_->stack_ == nullptr) {
+        if (GRPC_TRACE_FLAG_ENABLED(grpc_trace_promise_primitives)) {
+          gpr_log(GPR_INFO,
+                  "%s PullServerTrailingMetadata[%p]: Ready(no-stack): %s",
+                  GetContext<Activity>()->DebugTag().c_str(), filters_,
+                  filters_->server_trailing_metadata_->DebugString().c_str());
+        }
+        return std::move(filters_->server_trailing_metadata_);
+      }
+      // Otherwise we need to process it through all the filters.
       return executor_.Start(&filters_->stack_->data_.server_trailing_metadata,
                              std::move(filters_->server_trailing_metadata_),
                              filters_->call_data_);
@@ -1465,7 +1709,7 @@ class CallFilters {
     filters_detail::InfallibleOperationExecutor<ServerMetadataHandle> executor_;
   };
 
-  void CancelDueToFailedPipeOperation();
+  void CancelDueToFailedPipeOperation(SourceLocation but_where = {});
 
   RefCountedPtr<Stack> stack_;
 
@@ -1475,6 +1719,7 @@ class CallFilters {
   filters_detail::PipeState client_to_server_message_state_;
   filters_detail::PipeState server_to_client_message_state_;
   IntraActivityWaiter server_trailing_metadata_waiter_;
+  Latch<bool> cancelled_;
 
   void* call_data_;
   ClientMetadataHandle client_initial_metadata_;
@@ -1516,7 +1761,7 @@ inline auto CallFilters::PushServerInitialMetadata(ServerMetadataHandle md) {
 }
 
 inline auto CallFilters::PullServerInitialMetadata() {
-  return ServerInitialMetadataPromises::Pull{this};
+  return ServerInitialMetadataPromises::PullMaybe{this};
 }
 
 inline auto CallFilters::PushClientToServerMessage(MessageHandle message) {
@@ -1526,7 +1771,7 @@ inline auto CallFilters::PushClientToServerMessage(MessageHandle message) {
 }
 
 inline auto CallFilters::PullClientToServerMessage() {
-  return ClientToServerMessagePromises::Pull{this};
+  return ClientToServerMessagePromises::PullMessage{this};
 }
 
 inline auto CallFilters::PushServerToClientMessage(MessageHandle message) {
@@ -1536,13 +1781,19 @@ inline auto CallFilters::PushServerToClientMessage(MessageHandle message) {
 }
 
 inline auto CallFilters::PullServerToClientMessage() {
-  return ServerToClientMessagePromises::Pull{this};
+  return ServerToClientMessagePromises::PullMessage{this};
 }
 
 inline auto CallFilters::PullServerTrailingMetadata() {
-  return PullServerTrailingMetadataPromise(this);
+  return Map(PullServerTrailingMetadataPromise(this),
+             [this](ServerMetadataHandle h) {
+               cancelled_.Set(h->get(GrpcCallWasCancelled()).value_or(false));
+               return h;
+             });
 }
 
+inline auto CallFilters::WasCancelled() { return cancelled_.Wait(); }
+
 }  // namespace grpc_core
 
 #endif  // GRPC_SRC_CORE_LIB_TRANSPORT_CALL_FILTERS_H

test/core/transport/call_filters_test.cc

@@ -119,7 +119,7 @@ namespace filters_detail {
 
 TEST(StackDataTest, Empty) {
   StackData d;
-  EXPECT_EQ(d.call_data_alignment, 0u);
+  EXPECT_EQ(d.call_data_alignment, 1u);
   EXPECT_EQ(d.call_data_size, 0u);
 }
 
@@ -1243,7 +1243,7 @@ TEST(PipeStateTest, OnePull) {
   ps.BeginPush();
   Mock::VerifyAndClearExpectations(&activity);
   // now we should see a value on the pull poll
-  EXPECT_THAT(ps.PollPull(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPull(), IsReady(true));
   // push should be pending though!
   EXPECT_THAT(ps.PollPush(), IsPending());
   // ack the pull, should see a wakeup
@@ -1251,7 +1251,7 @@ TEST(PipeStateTest, OnePull) {
   ps.AckPull();
   Mock::VerifyAndClearExpectations(&activity);
   // now the push is complete
-  EXPECT_THAT(ps.PollPush(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPush(), IsReady(Success()));
   ps.DropPush();
   ps.DropPull();
   EXPECT_FALSE(ps.holds_error());
@@ -1270,7 +1270,7 @@ TEST(PipeStateTest, StartThenPull) {
   ps.BeginPush();
   Mock::VerifyAndClearExpectations(&activity);
   // now we should see a value on the pull poll
-  EXPECT_THAT(ps.PollPull(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPull(), IsReady(true));
   // push should be pending though!
   EXPECT_THAT(ps.PollPush(), IsPending());
   // ack the pull, should see a wakeup
@@ -1278,7 +1278,7 @@ TEST(PipeStateTest, StartThenPull) {
   ps.AckPull();
   Mock::VerifyAndClearExpectations(&activity);
   // now the push is complete
-  EXPECT_THAT(ps.PollPush(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPush(), IsReady(Success()));
   ps.DropPush();
   ps.DropPull();
   EXPECT_FALSE(ps.holds_error());
@@ -1294,7 +1294,7 @@ TEST(PipeStateTest, PushFirst) {
   // push should be pending
   EXPECT_THAT(ps.PollPush(), IsPending());
   // pull should immediately see a value
-  EXPECT_THAT(ps.PollPull(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPull(), IsReady(true));
   // push should still be pending though!
   EXPECT_THAT(ps.PollPush(), IsPending());
   // ack the pull, should see a wakeup
@@ -1302,7 +1302,7 @@ TEST(PipeStateTest, PushFirst) {
   ps.AckPull();
   Mock::VerifyAndClearExpectations(&activity);
   // now the push is complete
-  EXPECT_THAT(ps.PollPush(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPush(), IsReady(Success()));
   ps.DropPush();
   ps.DropPull();
   EXPECT_FALSE(ps.holds_error());
@@ -1339,7 +1339,7 @@ TEST(PipeStateTest, DropProcessing) {
   activity.Activate();
   ps.Start();
   ps.BeginPush();
-  EXPECT_THAT(ps.PollPull(), IsReady(Success{}));
+  EXPECT_THAT(ps.PollPull(), IsReady(true));
   ps.DropPull();
   EXPECT_TRUE(ps.holds_error());
   EXPECT_THAT(ps.PollPull(), IsReady(Failure()));