KIKIMR-14503 control requests count to tx-allocator

ref:1469dbafc412933345b8e639f81dd0895ded1a3b

12 files changed, 408 insertions, 13 deletions

diff --git a/CMakeLists.darwin.txt b/CMakeLists.darwin.txt
index 0d3ebf68b3..f5294ac864 100644
--- a/CMakeLists.darwin.txt
+++ b/CMakeLists.darwin.txt
@@ -1135,6 +1135,7 @@ add_subdirectory(ydb/core/tx/sequenceproxy/ut)
 add_subdirectory(ydb/core/tx/sequenceshard/ut)
 add_subdirectory(ydb/core/tx/time_cast/ut)
 add_subdirectory(ydb/core/tx/tx_allocator/ut)
+add_subdirectory(ydb/core/tx/tx_allocator_client/ut)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_base_tenant)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_encrypted_storage)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_ext_tenant)
diff --git a/CMakeLists.linux.txt b/CMakeLists.linux.txt
index 2620574fdb..2eab2e65f5 100644
--- a/CMakeLists.linux.txt
+++ b/CMakeLists.linux.txt
@@ -1231,6 +1231,7 @@ add_subdirectory(ydb/core/tx/sequenceproxy/ut)
 add_subdirectory(ydb/core/tx/sequenceshard/ut)
 add_subdirectory(ydb/core/tx/time_cast/ut)
 add_subdirectory(ydb/core/tx/tx_allocator/ut)
+add_subdirectory(ydb/core/tx/tx_allocator_client/ut)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_base_tenant)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_encrypted_storage)
 add_subdirectory(ydb/core/tx/tx_proxy/ut_ext_tenant)
diff --git a/ydb/core/tx/tx_allocator_client/actor_client.cpp b/ydb/core/tx/tx_allocator_client/actor_client.cpp
index 20909a8401..5ed64e670c 100644
--- a/ydb/core/tx/tx_allocator_client/actor_client.cpp
+++ b/ydb/core/tx/tx_allocator_client/actor_client.cpp
@@ -34,7 +34,13 @@ class TTxAllocatorClientActor: public TActorBootstrapped<TTxAllocatorClientActor
     }
 
     void Handle(TEvTxAllocatorClient::TEvAllocate::TPtr& ev, const TActorContext& ctx) {
-        TVector<ui64> txIds = TxAllocatorClient.AllocateTxIds(ev->Get()->Count, ctx);
+        auto count = ev->Get()->Count;
+        if (0 == count) {
+            Send(ev->Sender, new TEvTxAllocatorClient::TEvAllocateResult(TVector<ui64>{}), 0, ev->Cookie);
+            return;
+        }
+
+        TVector<ui64> txIds = TxAllocatorClient.AllocateTxIds(count, ctx);
 
         if (txIds) {
             Send(ev->Sender, new TEvTxAllocatorClient::TEvAllocateResult(std::move(txIds)), 0, ev->Cookie);
@@ -83,9 +89,9 @@ public:
         return NKikimrServices::TActivity::TX_ALLOCATOR_CLIENT_ACTOR;
     }
 
-    explicit TTxAllocatorClientActor(const TVector<ui64>& txAllocators)
+    explicit TTxAllocatorClientActor(TVector<ui64> txAllocators)
         : PipeClientCache(NTabletPipe::CreateUnboundedClientCache(GetPipeClientConfig()))
-        , TxAllocatorClient(NKikimrServices::TX_ALLOCATOR_CLIENT, PipeClientCache.Get(), txAllocators)
+        , TxAllocatorClient(NKikimrServices::TX_ALLOCATOR_CLIENT, PipeClientCache.Get(), std::move(txAllocators))
     {
     }
 
@@ -115,8 +121,8 @@ private:
 
 }; // TTxAllocatorClientActor
 
-IActor* CreateTxAllocatorClient(const TVector<ui64>& txAllocators) {
-    return new TTxAllocatorClientActor(txAllocators);
+IActor* CreateTxAllocatorClient(TVector<ui64> txAllocators) {
+    return new TTxAllocatorClientActor(std::move(txAllocators));
 }
 
 } // NKikimr
diff --git a/ydb/core/tx/tx_allocator_client/actor_client.h b/ydb/core/tx/tx_allocator_client/actor_client.h
index 93d7caacce..3f1f608928 100644
--- a/ydb/core/tx/tx_allocator_client/actor_client.h
+++ b/ydb/core/tx/tx_allocator_client/actor_client.h
@@ -27,7 +27,6 @@ struct TEvTxAllocatorClient {
         explicit TEvAllocate(ui32 count)
             : Count(count)
         {
-            Y_VERIFY(count > 0);
         }
     };
 
@@ -49,6 +48,6 @@ struct TEvTxAllocatorClient {
 
 }; // TTxAllocatorClientEvents
 
-IActor* CreateTxAllocatorClient(const TVector<ui64>& txAllocators);
+IActor* CreateTxAllocatorClient(TVector<ui64> txAllocators);
 
 } // NKikimr
diff --git a/ydb/core/tx/tx_allocator_client/actor_client_ut.cpp b/ydb/core/tx/tx_allocator_client/actor_client_ut.cpp
new file mode 100644
index 0000000000..751d8a561c
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/actor_client_ut.cpp
@@ -0,0 +1,88 @@
+#include "ut_helpers.h"
+#include "actor_client.h"
+
+#include <ydb/core/tx/tx_allocator/txallocator.h>
+#include <ydb/core/testlib/basics/runtime.h>
+
+#include <util/generic/xrange.h>
+
+using namespace NKikimr;
+using namespace NTxAllocatorUT_Private;
+
+Y_UNIT_TEST_SUITE(TTxAllocatorClientTest) {
+    Y_UNIT_TEST(Boot) {
+        TTestBasicRuntime runtime;
+        TTestEnv env(runtime);
+    }
+
+    Y_UNIT_TEST(InitiatingRequest) {
+        TTestBasicRuntime runtime;
+
+        TMsgCounter initiatingCounter(runtime, TEvTxAllocator::TEvAllocate::EventType);
+
+        TTestEnv env(runtime);
+        env.AllocateAndCheck(1);
+
+        initiatingCounter.Wait(1);
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "one request for first batch");
+    }
+
+    Y_UNIT_TEST(ZeroRange) {
+        TTestBasicRuntime runtime;
+
+        TMsgCounter initiatingCounter(runtime, TEvTxAllocator::TEvAllocate::EventType);
+
+        TTestEnv env(runtime);
+        env.AllocateAndCheck(1);
+
+        initiatingCounter.Wait(1);
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "one request for first batch");
+
+        {
+            TMsgCounter counter(runtime, TEvTxAllocator::TEvAllocate::EventType);
+            for (int i = 0; i < 5500; ++i) {
+                env.AllocateAndCheck(0);
+                UNIT_ASSERT_VALUES_EQUAL_C(counter.Get(), 0, "no msg expected");
+            }
+        }
+    }
+
+    Y_UNIT_TEST(AllocateOverTheEdge) {
+        TTestBasicRuntime runtime;
+
+        TMsgCounter initiatingCounter(runtime, TEvTxAllocator::TEvAllocate::EventType);
+
+        TTestEnv env(runtime);
+        env.AllocateAndCheck(1);
+
+        initiatingCounter.Wait(1);
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "one request for first batch");
+
+        env.AllocateAndCheck(1000);
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "no additional request");
+
+        env.AllocateAndCheck(1000); // -2000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "no additional request");
+
+        env.AllocateAndCheck(1000); // -3000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 1, "no additional request");
+
+        env.AllocateAndCheck(1000); // -4000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 2, "one additional request");
+
+        env.AllocateAndCheck(500); // -4500
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 2, "no additional request");
+
+        env.AllocateAndCheck(1000); // -5000 -500
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 2, "no additional request");
+
+        env.AllocateAndCheck(2500); // -5000 -3000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 2, "no additional request");
+
+        env.AllocateAndCheck(1000); // -5000 -4000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 3, "one additional request");
+
+        env.AllocateAndCheck(3000); // -5000 -5000 -2000
+        UNIT_ASSERT_VALUES_EQUAL_C(initiatingCounter.Get(), 3, "one additional request");
+    }
+}
diff --git a/ydb/core/tx/tx_allocator_client/client.cpp b/ydb/core/tx/tx_allocator_client/client.cpp
index d33144a166..61136fcb50 100644
--- a/ydb/core/tx/tx_allocator_client/client.cpp
+++ b/ydb/core/tx/tx_allocator_client/client.cpp
@@ -5,10 +5,10 @@ void NKikimr::TTxAllocatorClient::AddAllocationRange(const NKikimr::TTxAllocator
     Capacity += ReservedRanges.back().Capacity();
 }
 
-NKikimr::TTxAllocatorClient::TTxAllocatorClient(NKikimrServices::EServiceKikimr service, NKikimr::NTabletPipe::IClientCache *pipeClientCache, const TVector<NKikimr::TTxAllocatorClient::TTabletId> &txAllocators)
+NKikimr::TTxAllocatorClient::TTxAllocatorClient(NKikimrServices::EServiceKikimr service, NKikimr::NTabletPipe::IClientCache *pipeClientCache, TVector<NKikimr::TTxAllocatorClient::TTabletId> txAllocators)
     : Service(service)
     , PipeClientCache(pipeClientCache)
-    , TxAllocators(txAllocators)
+    , TxAllocators(std::move(txAllocators))
     , MaxCapacity(TxAllocators.size() * RequestPerAllocator)
 {
     Y_VERIFY(!TxAllocators.empty());
@@ -52,10 +52,10 @@ TVector<ui64> NKikimr::TTxAllocatorClient::AllocateTxIds(ui64 count, const NActo
             txIds.push_back(head.Allocate());
             --Capacity;
             --count;
-        }
 
-        if (head.Capacity() <= PreRequestThreshhold) {
-            RegisterRequest(head.AllocatedFrom(), ctx);
+            if (head.Capacity() == PreRequestThreshhold) {
+                RegisterRequest(head.AllocatedFrom(), ctx);
+            }
         }
 
         if (head.Empty()) {
diff --git a/ydb/core/tx/tx_allocator_client/client.h b/ydb/core/tx/tx_allocator_client/client.h
index d316e0eac2..f957d8ba46 100644
--- a/ydb/core/tx/tx_allocator_client/client.h
+++ b/ydb/core/tx/tx_allocator_client/client.h
@@ -34,7 +34,7 @@ public:
     explicit TTxAllocatorClient(
         NKikimrServices::EServiceKikimr service,
         NTabletPipe::IClientCache* pipeClientCache,
-        const TVector<TTabletId>& txAllocators);
+        TVector<TTabletId> txAllocators);
 
     void Bootstrap(const TActorContext& ctx);
 
diff --git a/ydb/core/tx/tx_allocator_client/ut/CMakeLists.darwin.txt b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.darwin.txt
new file mode 100644
index 0000000000..f0b90a348d
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.darwin.txt
@@ -0,0 +1,54 @@
+
+# This file was gererated by the build system used internally in the Yandex monorepo.
+# Only simple modifications are allowed (adding source-files to targets, adding simple properties
+# like target_include_directories). These modifications will be ported to original
+# ya.make files by maintainers. Any complex modifications which can't be ported back to the
+# original buildsystem will not be accepted.
+
+
+
+add_executable(ydb-core-tx-tx_allocator_client-ut)
+target_compile_options(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  -DUSE_CURRENT_UDF_ABI_VERSION
+)
+target_include_directories(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client
+)
+target_link_libraries(ydb-core-tx-tx_allocator_client-ut PUBLIC
+  contrib-libs-cxxsupp
+  yutil
+  library-cpp-cpuid_check
+  cpp-testing-unittest_main
+  core-tx-tx_allocator_client
+  cpp-testing-unittest
+  ydb-core-mind
+  ydb-core-testlib
+  ydb-core-tx
+)
+target_sources(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client/actor_client_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client/ut_helpers.cpp
+)
+add_test(
+  NAME
+  ydb-core-tx-tx_allocator_client-ut
+  COMMAND
+  ydb-core-tx-tx_allocator_client-ut
+  --print-before-suite
+  --print-before-test
+  --fork-tests
+  --print-times
+  --show-fails
+)
+target_link_flags(ydb-core-tx-tx_allocator_client-ut
+  PUBLIC
+  -Wl,-no_deduplicate
+  -Wl,-sdk_version,10.15
+  -fPIC
+  -fPIC
+  -framework
+  CoreFoundation
+  -framework
+  CoreFoundation
+)
+vcs_info(ydb-core-tx-tx_allocator_client-ut)
diff --git a/ydb/core/tx/tx_allocator_client/ut/CMakeLists.linux.txt b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.linux.txt
new file mode 100644
index 0000000000..f9197643e3
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.linux.txt
@@ -0,0 +1,55 @@
+
+# This file was gererated by the build system used internally in the Yandex monorepo.
+# Only simple modifications are allowed (adding source-files to targets, adding simple properties
+# like target_include_directories). These modifications will be ported to original
+# ya.make files by maintainers. Any complex modifications which can't be ported back to the
+# original buildsystem will not be accepted.
+
+
+
+add_executable(ydb-core-tx-tx_allocator_client-ut)
+target_compile_options(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  -DUSE_CURRENT_UDF_ABI_VERSION
+)
+target_include_directories(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client
+)
+target_link_libraries(ydb-core-tx-tx_allocator_client-ut PUBLIC
+  contrib-libs-cxxsupp
+  yutil
+  library-cpp-lfalloc
+  library-cpp-cpuid_check
+  cpp-testing-unittest_main
+  core-tx-tx_allocator_client
+  cpp-testing-unittest
+  ydb-core-mind
+  ydb-core-testlib
+  ydb-core-tx
+)
+target_sources(ydb-core-tx-tx_allocator_client-ut PRIVATE
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client/actor_client_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/core/tx/tx_allocator_client/ut_helpers.cpp
+)
+add_test(
+  NAME
+  ydb-core-tx-tx_allocator_client-ut
+  COMMAND
+  ydb-core-tx-tx_allocator_client-ut
+  --print-before-suite
+  --print-before-test
+  --fork-tests
+  --print-times
+  --show-fails
+)
+target_link_flags(ydb-core-tx-tx_allocator_client-ut
+  PUBLIC
+  -ldl
+  -lrt
+  -Wl,--no-as-needed
+  -fPIC
+  -fPIC
+  -lpthread
+  -lrt
+  -ldl
+)
+vcs_info(ydb-core-tx-tx_allocator_client-ut)
diff --git a/ydb/core/tx/tx_allocator_client/ut/CMakeLists.txt b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.txt
new file mode 100644
index 0000000000..a681d385f3
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/ut/CMakeLists.txt
@@ -0,0 +1,13 @@
+
+# This file was gererated by the build system used internally in the Yandex monorepo.
+# Only simple modifications are allowed (adding source-files to targets, adding simple properties
+# like target_include_directories). These modifications will be ported to original
+# ya.make files by maintainers. Any complex modifications which can't be ported back to the
+# original buildsystem will not be accepted.
+
+
+if (APPLE)
+  include(CMakeLists.darwin.txt)
+elseif (UNIX)
+  include(CMakeLists.linux.txt)
+endif()
diff --git a/ydb/core/tx/tx_allocator_client/ut_helpers.cpp b/ydb/core/tx/tx_allocator_client/ut_helpers.cpp
new file mode 100644
index 0000000000..53b25a9bc2
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/ut_helpers.cpp
@@ -0,0 +1,110 @@
+#include "ut_helpers.h"
+
+#include <ydb/core/tx/tx_allocator/txallocator.h>
+
+namespace NTxAllocatorUT_Private {
+
+const TDuration TTestEnv::SimTimeOut = TDuration::Seconds(1u);
+
+ui64 SomeCockie(ui64 size) {
+    return size / 2;
+}
+
+void CheckExpectedCookie(const TVector<ui64>& txIds, ui64 cookie) {
+    const ui64  rangeSize = txIds.size();
+    if (rangeSize > 0) // rangeSize == 0 when status is error
+        UNIT_ASSERT_EQUAL(cookie, SomeCockie(rangeSize));
+}
+
+TAnswerWithCookie GrabAnswer(NActors::TTestActorRuntime &runtime) {
+    TAutoPtr<IEventHandle> handle;
+    TEvTxAllocatorClient::TEvAllocateResult *event = runtime.GrabEdgeEvent<TEvTxAllocatorClient::TEvAllocateResult>(handle);
+    UNIT_ASSERT(event);
+    return TAnswerWithCookie(std::move(event->TxIds), handle->Cookie);
+}
+
+void TTestEnv::AllocateAndCheck(ui64 size) {
+    AsyncAllocate(size);
+    TAnswerWithCookie result = GrabAnswer(Runtime);
+
+    auto& txIds = result.first;
+    UNIT_ASSERT_EQUAL(txIds.size(), size);
+
+    ui64 cookie = result.second;
+    CheckExpectedCookie(txIds, cookie);
+}
+
+void TTestEnv::AsyncAllocate(ui64 size) {
+    Y_VERIFY(TxAllocatorClient);
+    TActorId sender = Runtime.AllocateEdgeActor();
+    TEvTxAllocatorClient::TEvAllocate *ev = new TEvTxAllocatorClient::TEvAllocate(size);
+    Runtime.Send(new IEventHandle(TxAllocatorClient, sender, ev, 0, SomeCockie(size)), 0, true);
+}
+
+void TTestEnv::Boot() {
+    CreateTestBootstrapper(Runtime, CreateTestTabletInfo(TxAllocatorTablet, TTabletTypes::TX_ALLOCATOR), &CreateTxAllocator);
+
+    TDispatchOptions options;
+    options.FinalEvents.push_back(TDispatchOptions::TFinalEventCondition(TEvTablet::EvBoot));
+    Runtime.DispatchEvents(options, SimTimeOut);
+}
+
+void TTestEnv::SetupLogging() {
+    Runtime.SetLogPriority(NKikimrServices::TX_ALLOCATOR, NActors::NLog::PRI_DEBUG);
+    Runtime.SetLogPriority(NKikimrServices::TABLET_MAIN, NActors::NLog::PRI_DEBUG);
+}
+
+void TTestEnv::Setup() {
+    static constexpr ui32 domainId = 0;
+    SetupLogging();
+    TAppPrepare app;
+    auto domain = TDomainsInfo::TDomain::ConstructDomainWithExplicitTabletIds("dc-1", domainId, 0,
+                                                                              domainId, domainId, TVector<ui32>{domainId},
+                                                                              domainId, TVector<ui32>{domainId},
+                                                                              100500,
+                                                                              TVector<ui64>{},
+                                                                              TVector<ui64>{},
+                                                                              TVector<ui64>{},
+                                                                              DefaultPoolKinds(2));
+    app.AddDomain(domain.Release());
+    //app.AddHive(0, 0);
+    SetupChannelProfiles(app);
+    SetupTabletServices(Runtime, &app, true);
+}
+
+void TTestEnv::SetupClient() {
+    TxAllocatorClient = Runtime.Register(CreateTxAllocatorClient({TTestEnv::TxAllocatorTablet}));
+}
+
+void TTestEnv::Reboot() {
+    TActorId sender = Runtime.AllocateEdgeActor();
+    RebootTablet(Runtime, TxAllocatorTablet, sender);
+}
+
+TMsgCounter::TMsgCounter(TTestActorRuntime &runtime, ui32 msgType)
+    : Runtime(runtime)
+    , Counter(0)
+{
+    PrevObserver = Runtime.SetObserverFunc([this, msgType](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& ev) {
+        if (ev->GetTypeRewrite() == msgType) {
+            this->Counter += 1;
+        }
+        return TTestActorRuntime::EEventAction::PROCESS;
+    });
+}
+
+void TMsgCounter::Wait(ui32 count) {
+    if (Counter < count) {
+        TDispatchOptions opts;
+        opts.FinalEvents.emplace_back(TDispatchOptions::TFinalEventCondition([this, count](IEventHandle&) -> bool {
+                                          return Counter >= count;
+                                      }));
+        Runtime.DispatchEvents(opts);
+    }
+}
+
+TMsgCounter::~TMsgCounter() {
+    Runtime.SetObserverFunc(PrevObserver);
+}
+
+}
diff --git a/ydb/core/tx/tx_allocator_client/ut_helpers.h b/ydb/core/tx/tx_allocator_client/ut_helpers.h
new file mode 100644
index 0000000000..ced88078bd
--- /dev/null
+++ b/ydb/core/tx/tx_allocator_client/ut_helpers.h
@@ -0,0 +1,68 @@
+#pragma once
+
+#include <ydb/core/tx/tx_allocator_client/actor_client.h>
+#include <ydb/core/tx/tx.h>
+
+#include <ydb/core/testlib/basics/appdata.h>
+#include <ydb/core/testlib/tablet_helpers.h>
+
+#include <library/cpp/testing/unittest/tests_data.h>
+#include <library/cpp/testing/unittest/registar.h>
+
+namespace NTxAllocatorUT_Private {
+
+using namespace NKikimr;
+
+// Sets up everything needed for the tests (actually we need only boot txallocator tablet)
+class TTestEnv {
+public:
+    static constexpr ui64 TxAllocatorTablet = TTestTxConfig::TxAllocator;
+    static const TDuration SimTimeOut;
+
+private:
+    TTestActorRuntime& Runtime;
+    TActorId TxAllocatorClient;
+
+public:
+    TTestEnv(TTestActorRuntime &runtime)
+        : Runtime(runtime)
+    {
+        Setup();
+        Boot();
+        SetupClient();
+    }
+
+private:
+    void Boot();
+    void SetupLogging();
+    void Setup();
+    void SetupClient();
+
+public:
+    void Reboot();
+
+    void AsyncAllocate(ui64 size);
+    void AllocateAndCheck(ui64 size);
+};
+
+typedef std::pair<TVector<ui64>, ui64> TAnswerWithCookie;
+TAnswerWithCookie GrabAnswer(TTestActorRuntime &runtime);
+void CheckExpectedCookie(const TEvTxAllocatorClient::TEvAllocateResult& result, ui64 cockie);
+
+struct TMsgCounter {
+    TTestActorRuntime& Runtime;
+    ui32 Counter;
+
+    TTestActorRuntimeBase::TEventObserver PrevObserver;
+
+    TMsgCounter(TTestActorRuntime& runtime, ui32 msgType);
+    ~TMsgCounter();
+
+    void Wait(ui32 count);
+
+    ui32 Get() {
+        return Counter;
+    }
+};
+
+}