paths to the actors library

- исправил пути к библиотеке акторов
- вернул тесты для `TPartitionWriterCacheActor`

ydb/services/persqueue_v1/actors/partition_writer_cache_actor.cpp

@@ -215,7 +215,7 @@ auto TPartitionWriterCacheActor::GetPartitionWriter(const TString& sessionId, co
         return p->second.get();
     }
 
-    if (Writers.size() >= MAX_TRANSACTIONS_COUNT) {
+    if (Writers.size() >= (1 + MAX_TRANSACTIONS_COUNT)) {
         if (!TryDeleteOldestWriter(ctx)) {
             return nullptr;
         }
@@ -237,8 +237,13 @@ bool TPartitionWriterCacheActor::TryDeleteOldestWriter(const TActorContext& ctx)
     auto oldest = Writers.end();
 
     for (auto p = Writers.begin(); p != Writers.end(); ++p) {
+        auto& tx = p->first;
         auto& writer = *p->second;
 
+        if ((tx.first == "") && (tx.second == "")) {
+            continue;
+        }
+
         if ((writer.LastActivity < minLastActivity) && !writer.HasPendingRequests()) {
             minLastActivity = writer.LastActivity;
             oldest = p;

ydb/services/persqueue_v1/ut/CMakeLists.darwin-arm64.txt

@@ -52,6 +52,10 @@ target_sources(ydb-services-persqueue_v1-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/functions_executor_wrapper.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/topic_service_ut.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/demo_tx.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/pqtablet_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/kqp_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp
 )
 set_property(
   TARGET

ydb/services/persqueue_v1/ut/CMakeLists.darwin-x86_64.txt

@@ -53,6 +53,10 @@ target_sources(ydb-services-persqueue_v1-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/functions_executor_wrapper.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/topic_service_ut.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/demo_tx.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/pqtablet_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/kqp_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp
 )
 set_property(
   TARGET

ydb/services/persqueue_v1/ut/CMakeLists.linux-aarch64.txt

@@ -56,6 +56,10 @@ target_sources(ydb-services-persqueue_v1-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/functions_executor_wrapper.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/topic_service_ut.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/demo_tx.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/pqtablet_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/kqp_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp
 )
 set_property(
   TARGET

ydb/services/persqueue_v1/ut/CMakeLists.linux-x86_64.txt

@@ -57,6 +57,10 @@ target_sources(ydb-services-persqueue_v1-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/functions_executor_wrapper.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/topic_service_ut.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/demo_tx.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/pqtablet_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/kqp_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp
 )
 set_property(
   TARGET

ydb/services/persqueue_v1/ut/CMakeLists.windows-x86_64.txt

@@ -46,6 +46,10 @@ target_sources(ydb-services-persqueue_v1-ut PRIVATE
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/functions_executor_wrapper.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/topic_service_ut.cpp
   ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/demo_tx.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_ut.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/pqtablet_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/kqp_mock.cpp
+  ${CMAKE_SOURCE_DIR}/ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp
 )
 set_property(
   TARGET

ydb/services/persqueue_v1/ut/kqp_mock.cpp

@@ -0,0 +1,32 @@
+#include "kqp_mock.h"
+
+namespace NKikimr::NPersQueueTests {
+
+void TKqpProxyServiceMock::Bootstrap()
+{
+    Become(&TKqpProxyServiceMock::StateWork);
+}
+
+STFUNC(TKqpProxyServiceMock::StateWork)
+{
+    switch (ev->GetTypeRewrite()) {
+        HFunc(NKqp::TEvKqp::TEvQueryRequest, Handle);
+    }
+}
+
+void TKqpProxyServiceMock::Handle(NKqp::TEvKqp::TEvQueryRequest::TPtr& ev, const TActorContext& ctx)
+{
+    auto& event = *ev->Get();
+
+    Y_ABORT_UNLESS(event.HasAction());
+    Y_ABORT_UNLESS(event.GetAction() == NKikimrKqp::QUERY_ACTION_TOPIC);
+
+    auto queryResponse = std::make_unique<NKqp::TEvKqp::TEvQueryResponse>();
+    auto* response = queryResponse->Record.GetRef().MutableResponse();
+
+    response->MutableTopicOperations()->SetWriteId(NextWriteId++);
+
+    ctx.Send(ev->Sender, std::move(queryResponse));
+}
+
+}

ydb/services/persqueue_v1/ut/kqp_mock.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include <ydb/core/kqp/common/events/events.h>
+#include <ydb/library/actors/core/actor_bootstrapped.h>
+
+namespace NKikimr::NPersQueueTests {
+
+class TKqpProxyServiceMock : public TActorBootstrapped<TKqpProxyServiceMock> {
+public:
+    void Bootstrap();
+
+    STFUNC(StateWork);
+
+    void Handle(NKqp::TEvKqp::TEvQueryRequest::TPtr& ev, const TActorContext& ctx);
+
+    ui64 NextWriteId = 1;
+};
+
+}

ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp

@@ -0,0 +1,184 @@
+#include "partition_writer_cache_actor_fixture.h"
+#include "kqp_mock.h"
+
+#include <ydb/core/kqp/common/simple/services.h>
+#include <ydb/services/persqueue_v1/actors/partition_writer_cache_actor.h>
+
+namespace NKikimr::NPersQueueTests {
+
+void TPartitionWriterCacheActorFixture::SetUp(NUnitTest::TTestContext&)
+{
+    SetupContext();
+    SetupKqpMock();
+    SetupPQTabletMock(PQTabletId);
+    SetupEventObserver();
+}
+
+void TPartitionWriterCacheActorFixture::TearDown(NUnitTest::TTestContext&)
+{
+    CleanupContext();
+}
+
+void TPartitionWriterCacheActorFixture::SetupContext()
+{
+    Ctx.ConstructInPlace();
+    Finalizer.ConstructInPlace(*Ctx);
+
+    Ctx->Prepare();
+}
+
+void TPartitionWriterCacheActorFixture::SetupKqpMock()
+{
+    for (ui32 node = 0; node < Ctx->Runtime->GetNodeCount(); ++node) {
+        Ctx->Runtime->RegisterService(NKqp::MakeKqpProxyID(Ctx->Runtime->GetNodeId(node)),
+                                      Ctx->Runtime->Register(new TKqpProxyServiceMock(), node),
+                                      node);
+    }
+}
+
+void TPartitionWriterCacheActorFixture::SetupPQTabletMock(ui64 tabletId)
+{
+    auto createPQTabletMock = [&](const NActors::TActorId& tablet, NKikimr::TTabletStorageInfo* info) -> IActor* {
+        PQTablet = new TPQTabletMock(tablet, info);
+        return PQTablet;
+    };
+
+    CreateTestBootstrapper(*Ctx->Runtime,
+                           CreateTestTabletInfo(tabletId, NKikimrTabletBase::TTabletTypes::Dummy, TErasureType::ErasureNone),
+                           createPQTabletMock);
+
+    TDispatchOptions options;
+    options.FinalEvents.push_back(TDispatchOptions::TFinalEventCondition(TEvTablet::EvBoot));
+
+    Ctx->Runtime->DispatchEvents(options);
+}
+
+void TPartitionWriterCacheActorFixture::CleanupContext()
+{
+    Finalizer.Clear();
+    Ctx.Clear();
+}
+
+TActorId TPartitionWriterCacheActorFixture::CreatePartitionWriterCacheActor(const TCreatePartitionWriterCacheActorParams& params)
+{
+    using TPartitionWriterCacheActor = NKikimr::NGRpcProxy::V1::TPartitionWriterCacheActor;
+
+    NPQ::TPartitionWriterOpts options;
+    options.WithDeduplication(params.WithDeduplication);
+    options.WithDatabase(params.Database);
+
+    auto actor = std::make_unique<TPartitionWriterCacheActor>(Ctx->Edge,
+                                                              params.Partition,
+                                                              PQTabletId,
+                                                              params.Generation,
+                                                              params.SourceId,
+                                                              options);
+    TActorId actorId = Ctx->Runtime->Register(actor.release());
+
+    auto event = Ctx->Runtime->GrabEdgeEvent<NPQ::TEvPartitionWriter::TEvInitResult>();
+    UNIT_ASSERT(event != nullptr);
+
+    return actorId;
+}
+
+auto TPartitionWriterCacheActorFixture::MakeTxId(const TString& sessionId, const TString& txId) -> TTxId
+{
+    return {sessionId, txId};
+}
+
+void TPartitionWriterCacheActorFixture::SetupEventObserver()
+{
+    auto observer = [this](TAutoPtr<IEventHandle>& ev) {
+        if (auto event = ev->CastAsLocal<NPQ::TEvPartitionWriter::TEvTxWriteRequest>(); event) {
+            CookieToTxId[event->Request->GetCookie()] = MakeTxId(event->SessionId, event->TxId);
+        } else if (auto event = ev->CastAsLocal<NPQ::TEvPartitionWriter::TEvWriteRequest>(); event) {
+            auto p = CookieToTxId.find(event->GetCookie());
+            UNIT_ASSERT(p != CookieToTxId.end());
+
+            if (!TxIdToPartitionWriter.contains(p->second)) {
+                TxIdToPartitionWriter[p->second] = ev->Recipient;
+                PartitionWriterToTxId[ev->Recipient] = p->second;
+
+                ++CreatePartitionWriterCount;
+            }
+        } else if (auto event = ev->CastAsLocal<TEvents::TEvPoisonPill>(); event) {
+            auto p = PartitionWriterToTxId.find(ev->Recipient);
+            if (p != PartitionWriterToTxId.end()) {
+                TxIdToPartitionWriter.erase(p->second);
+                PartitionWriterToTxId.erase(p);
+
+                ++DeletePartitionWriterCount;
+            }
+        }
+
+        return TTestActorRuntime::EEventAction::PROCESS;
+    };
+
+    Ctx->Runtime->SetObserverFunc(std::move(observer));
+}
+
+void TPartitionWriterCacheActorFixture::SendTxWriteRequest(const TActorId& recipient,
+                                                           const TSendTxWriteRequestParams& params)
+{
+    auto write =
+        std::make_unique<NPQ::TEvPartitionWriter::TEvTxWriteRequest>(params.SessionId, params.TxId,
+                                                                     MakeHolder<NPQ::TEvPartitionWriter::TEvWriteRequest>(params.Cookie));
+    auto* w = write->Request->Record.MutablePartitionRequest()->AddCmdWrite();
+    Y_UNUSED(w);
+
+    Ctx->Runtime->Send(recipient, Ctx->Edge, write.release(), 0, true);
+}
+
+void TPartitionWriterCacheActorFixture::EnsurePartitionWriterExist(const TEnsurePartitionWriterExistParams& params)
+{
+    auto p = TxIdToPartitionWriter.find(MakeTxId(params.SessionId, params.TxId));
+    UNIT_ASSERT(p != TxIdToPartitionWriter.end());
+}
+
+void TPartitionWriterCacheActorFixture::EnsurePartitionWriterNotExist(const TEnsurePartitionWriterExistParams& params)
+{
+    auto p = TxIdToPartitionWriter.find(MakeTxId(params.SessionId, params.TxId));
+    UNIT_ASSERT(p == TxIdToPartitionWriter.end());
+}
+
+void TPartitionWriterCacheActorFixture::EnsureWriteSessionClosed(EErrorCode errorCode)
+{
+    while (true) {
+        auto event = Ctx->Runtime->GrabEdgeEvent<NPQ::TEvPartitionWriter::TEvWriteResponse>();
+        UNIT_ASSERT(event != nullptr);
+        if (!event->IsSuccess()) {
+            UNIT_ASSERT_VALUES_EQUAL((int)event->GetError().Code, (int)errorCode);
+            break;
+        }
+    }
+}
+
+void TPartitionWriterCacheActorFixture::WaitForPartitionWriterOps(const TWaitForPartitionWriterOpsParams& params)
+{
+    CreatePartitionWriterCount = 0;
+    DeletePartitionWriterCount = 0;
+
+    TDispatchOptions options;
+    options.CustomFinalCondition = [this, &params]() -> bool {
+        if (params.CreateCount && params.DeleteCount) {
+            return
+                (DeletePartitionWriterCount == *params.DeleteCount) &&
+                (CreatePartitionWriterCount == *params.CreateCount);
+        } else if (params.DeleteCount) {
+            return DeletePartitionWriterCount == *params.DeleteCount;
+        } else if (params.CreateCount) {
+            return CreatePartitionWriterCount == *params.CreateCount;
+        } else {
+            return false;
+        }
+    };
+
+    Ctx->Runtime->DispatchEvents(options);
+}
+
+void TPartitionWriterCacheActorFixture::AdvanceCurrentTime(TDuration d)
+{
+    Ctx->Runtime->AdvanceCurrentTime(d);
+}
+
+}

ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.h

@@ -0,0 +1,80 @@
+#pragma once
+
+#include "pqtablet_mock.h"
+
+#include <ydb/core/persqueue/ut/common/pq_ut_common.h>
+#include <ydb/core/persqueue/writer/writer.h>
+
+#include <library/cpp/testing/unittest/registar.h>
+
+#include <optional>
+
+namespace NKikimr::NPersQueueTests {
+
+struct TCreatePartitionWriterCacheActorParams {
+    ui32 Partition = 0;
+    std::optional<ui32> Generation;
+    TString SourceId = "source_id";
+    bool WithDeduplication = true;
+    TString Database = "database";
+};
+
+struct TSendTxWriteRequestParams {
+    TString SessionId;
+    TString TxId;
+    ui64 Cookie;
+};
+
+struct TEnsurePartitionWriterExistParams {
+    TString SessionId;
+    TString TxId;
+};
+
+struct TWaitForPartitionWriterOpsParams {
+    TMaybe<size_t> CreateCount;
+    TMaybe<size_t> DeleteCount;
+};
+
+class TPartitionWriterCacheActorFixture : public NUnitTest::TBaseFixture {
+protected:
+    using EErrorCode = NPQ::TEvPartitionWriter::TEvWriteResponse::EErrorCode;
+    using TTxId = std::pair<TString, TString>;
+
+    void SetUp(NUnitTest::TTestContext&) override;
+    void TearDown(NUnitTest::TTestContext&) override;
+
+    void SetupContext();
+    void SetupKqpMock();
+    void SetupPQTabletMock(ui64 tabletId);
+    void SetupEventObserver();
+
+    void CleanupContext();
+
+    TActorId CreatePartitionWriterCacheActor(const TCreatePartitionWriterCacheActorParams& params = {});
+    void SendTxWriteRequest(const TActorId& recipient, const TSendTxWriteRequestParams& params);
+
+    void EnsurePartitionWriterExist(const TEnsurePartitionWriterExistParams& params);
+    void EnsurePartitionWriterNotExist(const TEnsurePartitionWriterExistParams& params);
+    void EnsureWriteSessionClosed(EErrorCode errorCode);
+
+    void WaitForPartitionWriterOps(const TWaitForPartitionWriterOpsParams& params);
+
+    void AdvanceCurrentTime(TDuration d);
+
+    static TTxId MakeTxId(const TString& sessionId, const TString& txId);
+
+    TMaybe<NPQ::TTestContext> Ctx;
+    TMaybe<NPQ::TFinalizer> Finalizer;
+
+    const ui64 PQTabletId = 12345;
+    TPQTabletMock* PQTablet = nullptr;
+
+    THashMap<ui64, TTxId> CookieToTxId;
+    THashMap<TTxId, TActorId> TxIdToPartitionWriter;
+    THashMap<TActorId, TTxId> PartitionWriterToTxId;
+
+    size_t CreatePartitionWriterCount = 0;
+    size_t DeletePartitionWriterCount = 0;
+};
+
+}