YT-18541: Sequoia response keeper & cypress dyn config in tests

* Changelog entry
  Type: introduce Sequoia response keeper
  Component: master
commit_hash:cee0869e8223c77582f2345a8b15a5fc7b4eb741

Author: cherepashka

yt/yt/core/rpc/response_keeper.cpp

@@ -4,7 +4,6 @@
 #include "helpers.h"
 #include "service.h"
 
-#include <atomic>
 #include <yt/yt/core/concurrency/thread_affinity.h>
 #include <yt/yt/core/concurrency/periodic_executor.h>
 
@@ -241,9 +240,7 @@ class TResponseKeeper
 
                     const auto& responseMessage = responseMessageOrError.Value();
 
-                    NProto::TResponseHeader header;
-                    YT_VERIFY(TryParseResponseHeader(responseMessage, &header));
-                    bool remember = FromProto<NRpc::EErrorCode>(header.error().code()) != NRpc::EErrorCode::Unavailable;
+                    auto remember = ValidateHeaderAndParseRememberOption(responseMessage);
 
                     if (auto setResponseKeeperPromise = EndRequest(mutationId, responseMessage, remember)) {
                         setResponseKeeperPromise();
@@ -313,20 +310,16 @@ class TResponseKeeper
 
         auto pendingIt = PendingResponses_.find(id);
         if (pendingIt != PendingResponses_.end()) {
-            if (!isRetry) {
-                THROW_ERROR_EXCEPTION("Duplicate request is not marked as \"retry\"")
-                    << TErrorAttribute("mutation_id", id);
-            }
+            ValidateRetry(id, isRetry);
+
             YT_LOG_DEBUG("Replying with pending response (MutationId: %v)", id);
             return pendingIt->second;
         }
 
         auto finishedIt = FinishedResponses_.find(id);
         if (finishedIt != FinishedResponses_.end()) {
-            if (!isRetry) {
-                THROW_ERROR_EXCEPTION("Duplicate request is not marked as \"retry\"")
-                    << TErrorAttribute("mutation_id", id);
-            }
+            ValidateRetry(id, isRetry);
+
             YT_LOG_DEBUG("Replying with finished response (MutationId: %v)", id);
             return MakeFuture(finishedIt->second);
         }
@@ -383,6 +376,23 @@ 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,6 +106,12 @@ 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,7 +295,6 @@ class TGenericTypedServiceContext
         return *Response_;
     }
 
-
     using IServiceContext::Reply;
 
     void Reply()
@@ -315,7 +314,6 @@ class TGenericTypedServiceContext
         }
     }
 
-
     const THandlerInvocationOptions& GetOptions() const
     {
         return Options_;