Revert commit rXXXXXX, YT-18541: Sequoia Response Keeper

commit_hash:8e78a28686a046a10d7766109b2feba4f8ba50da

Author: babenko

yt/yt/core/rpc/response_keeper.cpp

@@ -4,6 +4,7 @@
 #include "helpers.h"
 #include "service.h"
 
+#include <atomic>
 #include <yt/yt/core/concurrency/thread_affinity.h>
 #include <yt/yt/core/concurrency/periodic_executor.h>
 
@@ -240,7 +241,9 @@ class TResponseKeeper
 
                     const auto& responseMessage = responseMessageOrError.Value();
 
-                    auto remember = ValidateHeaderAndParseRememberOption(responseMessage);
+                    NProto::TResponseHeader header;
+                    YT_VERIFY(TryParseResponseHeader(responseMessage, &header));
+                    bool remember = FromProto<NRpc::EErrorCode>(header.error().code()) != NRpc::EErrorCode::Unavailable;
 
                     if (auto setResponseKeeperPromise = EndRequest(mutationId, responseMessage, remember)) {
                         setResponseKeeperPromise();
@@ -310,16 +313,20 @@ class TResponseKeeper
 
         auto pendingIt = PendingResponses_.find(id);
         if (pendingIt != PendingResponses_.end()) {
-            ValidateRetry(id, isRetry);
-
+            if (!isRetry) {
+                THROW_ERROR_EXCEPTION("Duplicate request is not marked as \"retry\"")
+                    << TErrorAttribute("mutation_id", id);
+            }
             YT_LOG_DEBUG("Replying with pending response (MutationId: %v)", id);
             return pendingIt->second;
         }
 
         auto finishedIt = FinishedResponses_.find(id);
         if (finishedIt != FinishedResponses_.end()) {
-            ValidateRetry(id, isRetry);
-
+            if (!isRetry) {
+                THROW_ERROR_EXCEPTION("Duplicate request is not marked as \"retry\"")
+                    << TErrorAttribute("mutation_id", id);
+            }
             YT_LOG_DEBUG("Replying with finished response (MutationId: %v)", id);
             return MakeFuture(finishedIt->second);
         }
@@ -376,23 +383,6 @@ class TResponseKeeper
 
 ////////////////////////////////////////////////////////////////////////////////
 
-bool ValidateHeaderAndParseRememberOption(const TSharedRefArray& responseMessage)
-{
-    NProto::TResponseHeader header;
-    YT_VERIFY(TryParseResponseHeader(responseMessage, &header));
-    return FromProto<EErrorCode>(header.error().code()) != EErrorCode::Unavailable;
-}
-
-void ValidateRetry(TMutationId mutationId, bool isRetry)
-{
-    if (!isRetry) {
-        THROW_ERROR_EXCEPTION("Duplicate request is not marked as \"retry\"")
-             << TErrorAttribute("mutation_id", mutationId);
-    }
-}
-
-////////////////////////////////////////////////////////////////////////////////
-
 IResponseKeeperPtr CreateResponseKeeper(
     TResponseKeeperConfigPtr config,
     IInvokerPtr invoker,

yt/yt/core/rpc/response_keeper.h

@@ -106,12 +106,6 @@ DEFINE_REFCOUNTED_TYPE(IResponseKeeper)
 
 ////////////////////////////////////////////////////////////////////////////////
 
-bool ValidateHeaderAndParseRememberOption(const TSharedRefArray& responseMessage);
-
-void ValidateRetry(TMutationId mutationId, bool isRetry);
-
-////////////////////////////////////////////////////////////////////////////////
-
 IResponseKeeperPtr CreateResponseKeeper(
     TResponseKeeperConfigPtr config,
     IInvokerPtr invoker,

yt/yt/core/rpc/service_detail.h

@@ -295,6 +295,7 @@ class TGenericTypedServiceContext
         return *Response_;
     }
 
+
     using IServiceContext::Reply;
 
     void Reply()
@@ -314,6 +315,7 @@ class TGenericTypedServiceContext
         }
     }
 
+
     const THandlerInvocationOptions& GetOptions() const
     {
         return Options_;