Regression test

7 files changed, 115 insertions, 2 deletions

diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
index 1b744e906c..7519ee7ba9 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.darwin-x86_64.txt
@@ -38,6 +38,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/sticking_ut.cpp
 )
 set_property(
   TARGET
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
index b77f4f4e6c..1d488c4550 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.linux-aarch64.txt
@@ -41,6 +41,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/sticking_ut.cpp
 )
 set_property(
   TARGET
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
index 96489d5e6b..3ee5b5f656 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.linux-x86_64.txt
@@ -42,6 +42,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/sticking_ut.cpp
 )
 set_property(
   TARGET
diff --git a/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt b/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
index ddae9008cf..b928771974 100644
--- a/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
+++ b/library/cpp/actors/interconnect/ut/CMakeLists.windows-x86_64.txt
@@ -31,6 +31,7 @@ target_sources(library-cpp-actors-interconnect-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/outgoing_stream_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/poller_actor_ut.cpp
   ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/dynamic_proxy_ut.cpp
+  ${CMAKE_SOURCE_DIR}/library/cpp/actors/interconnect/ut/sticking_ut.cpp
 )
 set_property(
   TARGET
diff --git a/library/cpp/actors/interconnect/ut/lib/node.h b/library/cpp/actors/interconnect/ut/lib/node.h
index e15baf639e..3a82ba07e7 100644
--- a/library/cpp/actors/interconnect/ut/lib/node.h
+++ b/library/cpp/actors/interconnect/ut/lib/node.h
@@ -22,7 +22,7 @@ public:
           NMonitoring::TDynamicCounterPtr counters, TDuration deadPeerTimeout,
           TChannelsConfig channelsSettings = TChannelsConfig(),
           ui32 numDynamicNodes = 0, ui32 numThreads = 1,
-          TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr) {
+          TIntrusivePtr<NLog::TSettings> loggerSettings = nullptr, ui32 inflight = 512 * 1024) {
         TActorSystemSetup setup;
         setup.NodeId = nodeId;
         setup.ExecutorsCount = 2;
@@ -43,7 +43,7 @@ public:
         common->Settings.DeadPeer = deadPeerTimeout;
         common->Settings.CloseOnIdle = TDuration::Minutes(1);
         common->Settings.SendBufferDieLimitInMB = 512;
-        common->Settings.TotalInflightAmountOfData = 512 * 1024;
+        common->Settings.TotalInflightAmountOfData = inflight;
         common->Settings.TCPSocketBufferSize = 2048 * 1024;
         common->OutgoingHandshakeInflightLimit = 3;
 
diff --git a/library/cpp/actors/interconnect/ut/sticking_ut.cpp b/library/cpp/actors/interconnect/ut/sticking_ut.cpp
new file mode 100644
index 0000000000..510baa3a1f
--- /dev/null
+++ b/library/cpp/actors/interconnect/ut/sticking_ut.cpp
@@ -0,0 +1,108 @@
+#include <library/cpp/actors/interconnect/ut/lib/node.h>
+#include <library/cpp/actors/interconnect/ut/lib/ic_test_cluster.h>
+#include <library/cpp/testing/unittest/registar.h>
+
+using namespace NActors;
+
+struct TEvPing : TEventBase<TEvPing, TEvents::THelloWorld::Ping> {
+    TString Data;
+
+    TEvPing(TString data)
+        : Data(data)
+    {}
+
+    TEvPing() = default;
+
+    ui32 CalculateSerializedSize() const override { return Data.size(); }
+    bool IsSerializable() const override { return true; }
+    bool SerializeToArcadiaStream(TChunkSerializer *serializer) const override { serializer->WriteAliasedRaw(Data.data(), Data.size()); return true; }
+    TString ToStringHeader() const override { return {}; }
+};
+
+class TPonger : public TActor<TPonger> {
+public:
+    TPonger()
+        : TActor(&TThis::StateFunc)
+    {}
+
+    void Handle(TEvPing::TPtr ev) {
+        Send(ev->Sender, new TEvents::TEvPong(), 0, ev->Cookie);
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvPing, Handle);
+    )
+};
+
+class TPinger : public TActorBootstrapped<TPinger> {
+    ui32 PingInFlight = 0;
+    TActorId PongerId;
+    TDuration MaxRTT;
+
+public:
+    TPinger(TActorId pongerId)
+        : PongerId(pongerId)
+    {}
+
+    void Bootstrap() {
+        Become(&TThis::StateFunc);
+        Action();
+    }
+
+    void Action() {
+        if (PingInFlight) {
+            return;
+        }
+        const ui32 max = 1 + RandomNumber(10u);
+        while (PingInFlight < max) {
+            IssuePing();
+        }
+    }
+
+    void IssuePing() {
+        TString data = TString::Uninitialized(RandomNumber<size_t>(256 * 1024) + 1);
+        memset(data.Detach(), 0, data.size());
+        Send(PongerId, new TEvPing(data), 0, GetCycleCountFast());
+        ++PingInFlight;
+    }
+
+    void Handle(TEvents::TEvPong::TPtr ev) {
+        const TDuration rtt = CyclesToDuration(GetCycleCountFast() - ev->Cookie);
+        if (MaxRTT < rtt) {
+            MaxRTT = rtt;
+            Cerr << "Updated MaxRTT# " << MaxRTT << Endl;
+            Y_VERIFY(MaxRTT <= TDuration::MilliSeconds(500));
+        }
+        --PingInFlight;
+        Action();
+    }
+
+    STRICT_STFUNC(StateFunc,
+        hFunc(TEvents::TEvPong, Handle);
+    )
+};
+
+Y_UNIT_TEST_SUITE(Sticking) {
+    Y_UNIT_TEST(Check) {
+        TPortManager portman;
+        THashMap<ui32, ui16> nodeToPort;
+        nodeToPort.emplace(1, portman.GetPort());
+        nodeToPort.emplace(2, portman.GetPort());
+
+        NMonitoring::TDynamicCounterPtr counters = new NMonitoring::TDynamicCounters;
+        std::list<TNode> nodes;
+        for (auto [nodeId, _] : nodeToPort) {
+            nodes.emplace_back(nodeId, nodeToPort.size(), nodeToPort, "127.1.0.0",
+                counters->GetSubgroup("nodeId", TStringBuilder() << nodeId), TDuration::Seconds(10),
+                TChannelsConfig(), 0, 1, nullptr, 40 << 20);
+        }
+
+        auto& node1 = *nodes.begin();
+        auto& node2 = *++nodes.begin();
+
+        const TActorId ponger = node2.RegisterActor(new TPonger());
+        node1.RegisterActor(new TPinger(ponger));
+
+        Sleep(TDuration::Seconds(10));
+    }
+}
diff --git a/library/cpp/actors/interconnect/ut/ya.make b/library/cpp/actors/interconnect/ut/ya.make
index d524fc69f0..e5b838635f 100644
--- a/library/cpp/actors/interconnect/ut/ya.make
+++ b/library/cpp/actors/interconnect/ut/ya.make
@@ -17,6 +17,7 @@ SRCS(
     outgoing_stream_ut.cpp
     poller_actor_ut.cpp
     dynamic_proxy_ut.cpp
+    sticking_ut.cpp
 )
 
 PEERDIR(