Implement TServiceContextWrapper::Cancel

commit_hash:9c450a96bbd9f35d8e984562d75fae5bb870b199

5 files changed, 31 insertions, 1 deletions

diff --git a/yt/yt/core/rpc/server_detail.cpp b/yt/yt/core/rpc/server_detail.cpp
index 9366a95cf1d..56b6cd21838 100644
--- a/yt/yt/core/rpc/server_detail.cpp
+++ b/yt/yt/core/rpc/server_detail.cpp
@@ -702,7 +702,9 @@ bool TServiceContextWrapper::IsCanceled() const
 }
 
 void TServiceContextWrapper::Cancel()
-{ }
+{
+    UnderlyingContext_->Cancel();
+}
 
 TFuture<TSharedRefArray> TServiceContextWrapper::GetAsyncResponseMessage() const
 {
diff --git a/yt/yt/core/rpc/unittests/lib/test_service.cpp b/yt/yt/core/rpc/unittests/lib/test_service.cpp
index 68e42950015..56edc495304 100644
--- a/yt/yt/core/rpc/unittests/lib/test_service.cpp
+++ b/yt/yt/core/rpc/unittests/lib/test_service.cpp
@@ -91,6 +91,7 @@ public:
         RegisterMethod(RPC_SERVICE_METHOD_DESC(GetTraceBaggage));
         RegisterMethod(RPC_SERVICE_METHOD_DESC(CustomMetadata));
         RegisterMethod(RPC_SERVICE_METHOD_DESC(GetChannelFailureError));
+        RegisterMethod(RPC_SERVICE_METHOD_DESC(ManuallyCanceledByServer));
         // NB: NotRegisteredCall is not registered intentionally
 
         DeclareServerFeature(ETestFeature::Great);
@@ -394,6 +395,12 @@ public:
         }
     }
 
+    DECLARE_RPC_SERVICE_METHOD(NTestRpc, ManuallyCanceledByServer)
+    {
+        context->SetRequestInfo();
+        context->Cancel();
+    }
+
     TFuture<void> GetServerStreamsAborted() const override
     {
         return ServerStreamsAborted_.ToFuture();
diff --git a/yt/yt/core/rpc/unittests/lib/test_service.h b/yt/yt/core/rpc/unittests/lib/test_service.h
index a209b1e01d1..27c52f4865a 100644
--- a/yt/yt/core/rpc/unittests/lib/test_service.h
+++ b/yt/yt/core/rpc/unittests/lib/test_service.h
@@ -50,6 +50,7 @@ public:
     DEFINE_RPC_PROXY_METHOD(NTestRpc, GetTraceBaggage);
     DEFINE_RPC_PROXY_METHOD(NTestRpc, CustomMetadata);
     DEFINE_RPC_PROXY_METHOD(NTestRpc, GetChannelFailureError);
+    DEFINE_RPC_PROXY_METHOD(NTestRpc, ManuallyCanceledByServer);
 };
 
 ////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/core/rpc/unittests/lib/test_service.proto b/yt/yt/core/rpc/unittests/lib/test_service.proto
index 70c07a0b873..dd16f98edc6 100644
--- a/yt/yt/core/rpc/unittests/lib/test_service.proto
+++ b/yt/yt/core/rpc/unittests/lib/test_service.proto
@@ -266,3 +266,15 @@ message TReqGetChannelFailureError
 message TRspGetChannelFailureError
 {
 }
+
+////////////////////////////////////////////////////////////////////////////////
+
+message TReqManuallyCanceledByServer
+{
+}
+
+message TRspManuallyCanceledByServer
+{
+}
+
+////////////////////////////////////////////////////////////////////////////////
diff --git a/yt/yt/core/rpc/unittests/rpc_ut.cpp b/yt/yt/core/rpc/unittests/rpc_ut.cpp
index 37a5739d0c8..efb9b0a2de7 100644
--- a/yt/yt/core/rpc/unittests/rpc_ut.cpp
+++ b/yt/yt/core/rpc/unittests/rpc_ut.cpp
@@ -1077,6 +1077,14 @@ TYPED_TEST(TRpcTest, ConnectionLost)
         .ThrowOnError();
 }
 
+TYPED_TEST(TRpcTest, ManuallyCanceledByServer)
+{
+    TTestProxy proxy(this->CreateChannel());
+    auto req = proxy.ManuallyCanceledByServer();
+    auto rspOrError = req->Invoke().Get();
+    EXPECT_EQ(NYT::EErrorCode::Canceled, rspOrError.GetCode());
+}
+
 TYPED_TEST(TNotGrpcTest, ProtocolVersionMismatch)
 {
     TTestIncorrectProtocolVersionProxy proxy(this->CreateChannel());