Make client interceptors see notification even if Cancel was done before the RPC was issued. Also add tests

10 files changed, 89 insertions, 16 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 270fc13c11..bafc890121 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -12888,7 +12888,7 @@ if (gRPC_BUILD_TESTS)
 
 add_executable(end2end_test
   test/cpp/end2end/end2end_test.cc
-  test/cpp/end2end/interceptor_util.cc
+  test/cpp/end2end/interceptors_util.cc
   third_party/googletest/googletest/src/gtest-all.cc
   third_party/googletest/googlemock/src/gmock-all.cc
 )
diff --git a/Makefile b/Makefile
index 6594f71e24..9cec533e20 100644
--- a/Makefile
+++ b/Makefile
@@ -17754,7 +17754,7 @@ endif
 
 END2END_TEST_SRC = \
     test/cpp/end2end/end2end_test.cc \
-    test/cpp/end2end/interceptor_util.cc \
+    test/cpp/end2end/interceptors_util.cc \
 
 END2END_TEST_OBJS = $(addprefix $(OBJDIR)/$(CONFIG)/, $(addsuffix .o, $(basename $(END2END_TEST_SRC))))
 ifeq ($(NO_SECURE),true)
@@ -17787,7 +17787,7 @@ endif
 
 $(OBJDIR)/$(CONFIG)/test/cpp/end2end/end2end_test.o:  $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
 
-$(OBJDIR)/$(CONFIG)/test/cpp/end2end/interceptor_util.o:  $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
+$(OBJDIR)/$(CONFIG)/test/cpp/end2end/interceptors_util.o:  $(LIBDIR)/$(CONFIG)/libgrpc++_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc_test_util.a $(LIBDIR)/$(CONFIG)/libgrpc++.a $(LIBDIR)/$(CONFIG)/libgrpc.a $(LIBDIR)/$(CONFIG)/libgpr_test_util.a $(LIBDIR)/$(CONFIG)/libgpr.a
 
 deps_end2end_test: $(END2END_TEST_OBJS:.o=.dep)
 
diff --git a/build.yaml b/build.yaml
index bd30e54258..ba95fcfdd9 100644
--- a/build.yaml
+++ b/build.yaml
@@ -4668,7 +4668,7 @@ targets:
   - test/cpp/end2end/interceptors_util.h
   src:
   - test/cpp/end2end/end2end_test.cc
-  - test/cpp/end2end/interceptor_util.cc
+  - test/cpp/end2end/interceptors_util.cc
   deps:
   - grpc++_test_util
   - grpc_test_util
diff --git a/include/grpcpp/impl/codegen/client_context.h b/include/grpcpp/impl/codegen/client_context.h
index f53b744dcf..75b955e760 100644
--- a/include/grpcpp/impl/codegen/client_context.h
+++ b/include/grpcpp/impl/codegen/client_context.h
@@ -426,6 +426,8 @@ class ClientContext {
 
   grpc::string authority() { return authority_; }
 
+  void SendCancelToInterceptors();
+
   bool initial_metadata_received_;
   bool wait_for_ready_;
   bool wait_for_ready_explicitly_set_;
diff --git a/src/cpp/client/channel_cc.cc b/src/cpp/client/channel_cc.cc
index 15e3ccb3c9..5e7ecf0ebf 100644
--- a/src/cpp/client/channel_cc.cc
+++ b/src/cpp/client/channel_cc.cc
@@ -147,10 +147,14 @@ internal::Call Channel::CreateCallInternal(const internal::RpcMethod& method,
     }
   }
   grpc_census_call_set_context(c_call, context->census_context());
-  context->set_call(c_call, shared_from_this());
 
+  // ClientRpcInfo should be set before call because set_call also checks
+  // whether the call has been cancelled, and if the call was cancelled, we
+  // should notify the interceptors too/
   auto* info = context->set_client_rpc_info(
       method.name(), this, interceptor_creators_, interceptor_pos);
+  context->set_call(c_call, shared_from_this());
+
   return internal::Call(c_call, this, cq, info);
 }
 
diff --git a/src/cpp/client/client_context.cc b/src/cpp/client/client_context.cc
index d5eb029a24..50da75f09c 100644
--- a/src/cpp/client/client_context.cc
+++ b/src/cpp/client/client_context.cc
@@ -87,10 +87,13 @@ void ClientContext::set_call(grpc_call* call,
   call_ = call;
   channel_ = channel;
   if (creds_ && !creds_->ApplyToCall(call_)) {
+    // TODO(yashykt): should interceptors also see this status?
+    SendCancelToInterceptors();
     grpc_call_cancel_with_status(call, GRPC_STATUS_CANCELLED,
                                  "Failed to set credentials to rpc.", nullptr);
   }
   if (call_canceled_) {
+    SendCancelToInterceptors();
     grpc_call_cancel(call_, nullptr);
   }
 }
@@ -111,16 +114,20 @@ void ClientContext::set_compression_algorithm(
 void ClientContext::TryCancel() {
   std::unique_lock<std::mutex> lock(mu_);
   if (call_) {
-    internal::CancelInterceptorBatchMethods cancel_methods;
-    for (size_t i = 0; i < rpc_info_.interceptors_.size(); i++) {
-      rpc_info_.RunInterceptor(&cancel_methods, i);
-    }
+    SendCancelToInterceptors();
     grpc_call_cancel(call_, nullptr);
   } else {
     call_canceled_ = true;
   }
 }
 
+void ClientContext::SendCancelToInterceptors() {
+  internal::CancelInterceptorBatchMethods cancel_methods;
+  for (size_t i = 0; i < rpc_info_.interceptors_.size(); i++) {
+    rpc_info_.RunInterceptor(&cancel_methods, i);
+  }
+}
+
 grpc::string ClientContext::peer() const {
   grpc::string peer;
   if (call_) {
diff --git a/test/cpp/end2end/end2end_test.cc b/test/cpp/end2end/end2end_test.cc
index 6fce4274a1..4558437102 100644
--- a/test/cpp/end2end/end2end_test.cc
+++ b/test/cpp/end2end/end2end_test.cc
@@ -210,8 +210,9 @@ class TestScenario {
 
 static std::ostream& operator<<(std::ostream& out,
                                 const TestScenario& scenario) {
-  return out << "TestScenario{use_proxy="
-             << (scenario.use_proxy ? "true" : "false")
+  return out << "TestScenario{use_interceptors="
+             << (scenario.use_interceptors ? "true" : "false")
+             << ", use_proxy=" << (scenario.use_proxy ? "true" : "false")
              << ", inproc=" << (scenario.inproc ? "true" : "false")
              << ", credentials='" << scenario.credentials_type << "'}";
 }
@@ -275,6 +276,7 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> {
         creators.push_back(std::unique_ptr<DummyInterceptorFactory>(
             new DummyInterceptorFactory()));
       }
+      builder.experimental().SetInterceptorCreators(std::move(creators));
     }
     builder.AddListeningPort(server_address_.str(), server_creds);
     builder.RegisterService(&service_);
@@ -308,10 +310,21 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> {
     args.SetString(GRPC_ARG_SECONDARY_USER_AGENT_STRING, "end2end_test");
 
     if (!GetParam().inproc) {
-      channel_ =
-          CreateCustomChannel(server_address_.str(), channel_creds, args);
+      if (!GetParam().use_interceptors) {
+        channel_ =
+            CreateCustomChannel(server_address_.str(), channel_creds, args);
+      } else {
+        channel_ = CreateCustomChannelWithInterceptors(
+            server_address_.str(), channel_creds, args,
+            CreateDummyClientInterceptors());
+      }
     } else {
-      channel_ = server_->InProcessChannel(args);
+      if (!GetParam().use_interceptors) {
+        channel_ = server_->InProcessChannel(args);
+      } else {
+        channel_ = server_->experimental().InProcessChannelWithInterceptors(
+            args, CreateDummyClientInterceptors());
+      }
     }
   }
 
@@ -336,6 +349,7 @@ class End2endTest : public ::testing::TestWithParam<TestScenario> {
     }
 
     stub_ = grpc::testing::EchoTestService::NewStub(channel_);
+    DummyInterceptor::Reset();
   }
 
   bool is_server_started_;
@@ -392,6 +406,7 @@ class End2endServerTryCancelTest : public End2endTest {
   // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
   void TestRequestStreamServerCancel(
       ServerTryCancelRequestPhase server_try_cancel, int num_msgs_to_send) {
+    RestartServer(std::shared_ptr<AuthMetadataProcessor>());
     ResetStub();
     EchoRequest request;
     EchoResponse response;
@@ -448,6 +463,10 @@ class End2endServerTryCancelTest : public End2endTest {
 
     EXPECT_FALSE(s.ok());
     EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+    // Make sure that the server interceptors were notified
+    if (GetParam().use_interceptors) {
+      EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+    }
   }
 
   // Helper for testing server-streaming RPCs which are cancelled on the server.
@@ -465,6 +484,7 @@ class End2endServerTryCancelTest : public End2endTest {
   // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
   void TestResponseStreamServerCancel(
       ServerTryCancelRequestPhase server_try_cancel) {
+    RestartServer(std::shared_ptr<AuthMetadataProcessor>());
     ResetStub();
     EchoRequest request;
     EchoResponse response;
@@ -524,7 +544,10 @@ class End2endServerTryCancelTest : public End2endTest {
     }
 
     EXPECT_FALSE(s.ok());
-    EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+    // Make sure that the server interceptors were notified
+    if (GetParam().use_interceptors) {
+      EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+    }
   }
 
   // Helper for testing bidirectional-streaming RPCs which are cancelled on the
@@ -542,6 +565,7 @@ class End2endServerTryCancelTest : public End2endTest {
   // NOTE: Do not call this function with server_try_cancel == DO_NOT_CANCEL.
   void TestBidiStreamServerCancel(ServerTryCancelRequestPhase server_try_cancel,
                                   int num_messages) {
+    RestartServer(std::shared_ptr<AuthMetadataProcessor>());
     ResetStub();
     EchoRequest request;
     EchoResponse response;
@@ -608,6 +632,10 @@ class End2endServerTryCancelTest : public End2endTest {
 
     EXPECT_FALSE(s.ok());
     EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+    // Make sure that the server interceptors were notified
+    if (GetParam().use_interceptors) {
+      EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+    }
   }
 };
 
@@ -1005,6 +1033,9 @@ TEST_P(End2endTest, CancelRpcBeforeStart) {
   Status s = stub_->Echo(&context, request, &response);
   EXPECT_EQ("", response.message());
   EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+  if (GetParam().use_interceptors) {
+    EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+  }
 }
 
 // Client cancels request stream after sending two messages
@@ -1025,6 +1056,9 @@ TEST_P(End2endTest, ClientCancelsRequestStream) {
   EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
 
   EXPECT_EQ(response.message(), "");
+  if (GetParam().use_interceptors) {
+    EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+  }
 }
 
 // Client cancels server stream after sending some messages
@@ -1057,6 +1091,9 @@ TEST_P(End2endTest, ClientCancelsResponseStream) {
   // The final status could be either of CANCELLED or OK depending on
   // who won the race.
   EXPECT_GE(grpc::StatusCode::CANCELLED, s.error_code());
+  if (GetParam().use_interceptors) {
+    EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+  }
 }
 
 // Client cancels bidi stream after sending some messages
@@ -1090,6 +1127,9 @@ TEST_P(End2endTest, ClientCancelsBidi) {
 
   Status s = stream->Finish();
   EXPECT_EQ(grpc::StatusCode::CANCELLED, s.error_code());
+  if (GetParam().use_interceptors) {
+    EXPECT_EQ(20, DummyInterceptor::GetNumTimesCancel());
+  }
 }
 
 TEST_P(End2endTest, RpcMaxMessageSize) {
diff --git a/test/cpp/end2end/interceptors_util.cc b/test/cpp/end2end/interceptors_util.cc
index 29fb49d3eb..602d1695a3 100644
--- a/test/cpp/end2end/interceptors_util.cc
+++ b/test/cpp/end2end/interceptors_util.cc
@@ -131,5 +131,21 @@ bool CheckMetadata(const std::multimap<grpc::string_ref, grpc::string_ref>& map,
   }
   return false;
 }
+
+std::unique_ptr<std::vector<
+    std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+CreateDummyClientInterceptors() {
+  auto creators = std::unique_ptr<std::vector<
+      std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>(
+      new std::vector<
+          std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>());
+  // Add 20 dummy interceptors before hijacking interceptor
+  for (auto i = 0; i < 20; i++) {
+    creators->push_back(std::unique_ptr<DummyInterceptorFactory>(
+        new DummyInterceptorFactory()));
+  }
+  return creators;
+}
+
 }  // namespace testing
 }  // namespace grpc
diff --git a/test/cpp/end2end/interceptors_util.h b/test/cpp/end2end/interceptors_util.h
index f5b1d4a110..b4c4791fca 100644
--- a/test/cpp/end2end/interceptors_util.h
+++ b/test/cpp/end2end/interceptors_util.h
@@ -149,6 +149,10 @@ void MakeCallbackCall(const std::shared_ptr<Channel>& channel);
 bool CheckMetadata(const std::multimap<grpc::string_ref, grpc::string_ref>& map,
                    const string& key, const string& value);
 
+std::unique_ptr<std::vector<
+    std::unique_ptr<experimental::ClientInterceptorFactoryInterface>>>
+CreateDummyClientInterceptors();
+
 inline void* tag(int i) { return (void*)static_cast<intptr_t>(i); }
 inline int detag(void* p) {
   return static_cast<int>(reinterpret_cast<intptr_t>(p));
diff --git a/tools/run_tests/generated/sources_and_headers.json b/tools/run_tests/generated/sources_and_headers.json
index 7491bd3e9c..ade814a844 100644
--- a/tools/run_tests/generated/sources_and_headers.json
+++ b/tools/run_tests/generated/sources_and_headers.json
@@ -3609,7 +3609,7 @@
     "name": "end2end_test", 
     "src": [
       "test/cpp/end2end/end2end_test.cc", 
-      "test/cpp/end2end/interceptor_util.cc", 
+      "test/cpp/end2end/interceptors_util.cc", 
       "test/cpp/end2end/interceptors_util.h"
     ], 
     "third_party": false,