Ydb stable 23-2-1123.2.11

x-stable-origin-commit: 758ace972646c843c5e0785d75c8f4fe044580a1

3 files changed, 43 insertions, 22 deletions

diff --git a/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h b/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h
index 2b6d27cd3f3..dd2557e25ee 100644
--- a/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h
+++ b/library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h
@@ -26,13 +26,15 @@ private:
     TList<TTrafficInterrupter> interrupters;
     NActors::TChannelsConfig ChannelsConfig;
     TPortManager PortManager;
+    TIntrusivePtr<NLog::TSettings> LoggerSettings;
 
 public:
     TTestICCluster(ui32 numNodes = 1, NActors::TChannelsConfig channelsConfig = NActors::TChannelsConfig(),
-                   TTrafficInterrupterSettings* tiSettings = nullptr)
+                   TTrafficInterrupterSettings* tiSettings = nullptr, TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr)
         : NumNodes(numNodes)
         , Counters(new NMonitoring::TDynamicCounters)
         , ChannelsConfig(channelsConfig)
+        , LoggerSettings(loggerSettings)
     {
         THashMap<ui32, ui16> nodeToPortMap;
         THashMap<ui32, THashMap<ui32, ui16>> specificNodePortMap;
@@ -59,7 +61,8 @@ public:
 
         for (ui32 i = 1; i <= NumNodes; ++i) {
             auto& portMap = tiSettings ? specificNodePortMap[i] : nodeToPortMap;
-            Nodes.emplace(i, MakeHolder<TNode>(i, NumNodes, portMap, Address, Counters, DeadPeerTimeout, ChannelsConfig));
+            Nodes.emplace(i, MakeHolder<TNode>(i, NumNodes, portMap, Address, Counters, DeadPeerTimeout, ChannelsConfig,
+                /*numDynamicNodes=*/0, /*numThreads=*/1, LoggerSettings));
         }
     }
 
diff --git a/library/cpp/actors/interconnect/ut/lib/node.h b/library/cpp/actors/interconnect/ut/lib/node.h
index ff30b1445e8..0b538cdb1cb 100644
--- a/library/cpp/actors/interconnect/ut/lib/node.h
+++ b/library/cpp/actors/interconnect/ut/lib/node.h
@@ -6,6 +6,7 @@
 #include <library/cpp/actors/core/mailbox.h>
 #include <library/cpp/actors/dnsresolver/dnsresolver.h>
 
+#include <library/cpp/actors/interconnect/handshake_broker.h>
 #include <library/cpp/actors/interconnect/interconnect_tcp_server.h>
 #include <library/cpp/actors/interconnect/interconnect_tcp_proxy.h>
 #include <library/cpp/actors/interconnect/interconnect_proxy_wrapper.h>
@@ -19,7 +20,8 @@ public:
     TNode(ui32 nodeId, ui32 numNodes, const THashMap<ui32, ui16>& nodeToPort, const TString& address,
           NMonitoring::TDynamicCounterPtr counters, TDuration deadPeerTimeout,
           TChannelsConfig channelsSettings = TChannelsConfig(),
-          ui32 numDynamicNodes = 0, ui32 numThreads = 1) {
+          ui32 numDynamicNodes = 0, ui32 numThreads = 1,
+          TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) {
         TActorSystemSetup setup;
         setup.NodeId = nodeId;
         setup.ExecutorsCount = 1;
@@ -43,6 +45,7 @@ public:
         common->Settings.SendBufferDieLimitInMB = 512;
         common->Settings.TotalInflightAmountOfData = 512 * 1024;
         common->Settings.TCPSocketBufferSize = 2048 * 1024;
+        common->OutgoingHandshakeInflightLimit = 3;
 
         setup.Interconnect.ProxyActors.resize(numNodes + 1 - numDynamicNodes);
         setup.Interconnect.ProxyWrapperFactory = CreateProxyWrapperFactory(common, interconnectPoolId);
@@ -62,29 +65,31 @@ public:
         setup.LocalServices.emplace_back(MakePollerActorId(), TActorSetupCmd(CreatePollerActor(),
             TMailboxType::ReadAsFilled, 0));
 
-        const TActorId loggerActorId(0, "logger");
-        constexpr ui32 LoggerComponentId = 410; // NKikimrServices::LOGGER
+        const TActorId loggerActorId = loggerSettings ? loggerSettings->LoggerActorId : TActorId(0, "logger");
 
-        auto loggerSettings = MakeIntrusive<NLog::TSettings>(
-            loggerActorId,
-            (NLog::EComponent)LoggerComponentId,
-            NLog::PRI_INFO,
-            NLog::PRI_DEBUG,
-            0U);
+        if (!loggerSettings) {
+            constexpr ui32 LoggerComponentId = 410; // NKikimrServices::LOGGER
+            loggerSettings = MakeIntrusive<NLog::TSettings>(
+                loggerActorId,
+                (NLog::EComponent)LoggerComponentId,
+                NLog::PRI_INFO,
+                NLog::PRI_DEBUG,
+                0U);
 
-        loggerSettings->Append(
-            NActorsServices::EServiceCommon_MIN,
-            NActorsServices::EServiceCommon_MAX,
-            NActorsServices::EServiceCommon_Name
-        );
+            loggerSettings->Append(
+                NActorsServices::EServiceCommon_MIN,
+                NActorsServices::EServiceCommon_MAX,
+                NActorsServices::EServiceCommon_Name
+            );
 
-        constexpr ui32 WilsonComponentId = 430; // NKikimrServices::WILSON
-        static const TString WilsonComponentName = "WILSON";
+            constexpr ui32 WilsonComponentId = 430; // NKikimrServices::WILSON
+            static const TString WilsonComponentName = "WILSON";
 
-        loggerSettings->Append(
-            (NLog::EComponent)WilsonComponentId,
-            (NLog::EComponent)WilsonComponentId + 1,
-            [](NLog::EComponent) -> const TString & { return WilsonComponentName; });
+            loggerSettings->Append(
+                (NLog::EComponent)WilsonComponentId,
+                (NLog::EComponent)WilsonComponentId + 1,
+                [](NLog::EComponent) -> const TString & { return WilsonComponentName; });
+        }
 
         // register nameserver table
         auto names = MakeIntrusive<TTableNameserverSetup>();
@@ -105,6 +110,14 @@ public:
             CreateStderrBackend(), counters->GetSubgroup("subsystem", "logger")),
             TMailboxType::ReadAsFilled, interconnectPoolId));
 
+
+        if (common->OutgoingHandshakeInflightLimit) {
+            // create handshake broker actor
+            setup.LocalServices.emplace_back(MakeHandshakeBrokerOutId(), TActorSetupCmd(
+                    CreateHandshakeBroker(*common->OutgoingHandshakeInflightLimit),
+                    TMailboxType::ReadAsFilled, interconnectPoolId));
+        }
+
         auto sp = MakeHolder<TActorSystemSetup>(std::move(setup));
         ActorSystem.Reset(new TActorSystem(sp, nullptr, loggerSettings));
         ActorSystem->Start();
diff --git a/library/cpp/actors/interconnect/ut/lib/test_events.h b/library/cpp/actors/interconnect/ut/lib/test_events.h
index cd0d9e01520..1bb5eb7d383 100644
--- a/library/cpp/actors/interconnect/ut/lib/test_events.h
+++ b/library/cpp/actors/interconnect/ut/lib/test_events.h
@@ -9,6 +9,7 @@ namespace NActors {
         EvTestSmall,
         EvTestLarge,
         EvTestResponse,
+        EvTestStartPolling,
     };
 
     struct TEvTest : TEventPB<TEvTest, NInterconnectTest::TEvTest, EvTest> {
@@ -46,4 +47,8 @@ namespace NActors {
         }
     };
 
+    struct TEvTestStartPolling : TEventPB<TEvTestStartPolling, NInterconnectTest::TEvTestStartPolling, EvTestStartPolling> {
+        TEvTestStartPolling() = default;
+    };
+
 }