The counter value for UserInfo is leaking (#15233)

Author: Alek5andr-Kotov

ydb/core/persqueue/partition.cpp

@@ -2969,6 +2969,7 @@ TPartition::EProcessResult TPartition::PreProcessUserAct(
 void TPartition::CommitUserAct(TEvPQ::TEvSetClientInfo& act) {
     const bool strictCommitOffset = (act.Type == TEvPQ::TEvSetClientInfo::ESCI_OFFSET && act.Strict);
     const TString& user = act.ClientId;
+    RemoveUserAct(user);
     const auto& ctx = ActorContext();
     if (!PendingUsersInfo.contains(user) && AffectedUsers.contains(user)) {
         switch (act.Type) {
@@ -3088,7 +3089,6 @@ void TPartition::CommitUserAct(TEvPQ::TEvSetClientInfo& act) {
         return;
     }
 
-    RemoveUserAct(act.ClientId);
     return EmulatePostProcessUserAct(act, userInfo, ActorContext());
 }
 

ydb/core/persqueue/partition.h

@@ -29,6 +29,7 @@ namespace NKikimr::NPQ {
 
 static const ui32 MAX_BLOB_PART_SIZE = 500_KB;
 static const ui32 DEFAULT_BUCKET_COUNTER_MULTIPLIER = 20;
+static const ui32 MAX_USER_ACTS = 1000;
 
 using TPartitionLabeledCounters = TProtobufTabletLabeledCounters<EPartitionLabeledCounters_descriptor>;
 

ydb/core/persqueue/partition_read.cpp

@@ -28,8 +28,6 @@
 
 namespace NKikimr::NPQ {
 
-static const ui32 MAX_USER_ACTS = 1000;
-
 TMaybe<TInstant> GetReadFrom(ui32 maxTimeLagMs, ui64 readTimestampMs, TInstant consumerReadFromTimestamp, const TActorContext& ctx) {
     if (!(maxTimeLagMs > 0 || readTimestampMs > 0 || consumerReadFromTimestamp > TInstant::MilliSeconds(1))) {
         return {};

ydb/core/persqueue/ut/partition_ut.cpp

@@ -225,7 +225,8 @@ class TPartitionFixture : public NUnitTest::TBaseFixture {
     void SendSetOffset(ui64 cookie,
                        const TString& clientId,
                        ui64 offset,
-                       const TString& sessionId);
+                       const TString& sessionId,
+                       bool strict = false);
     void SendGetOffset(ui64 cookie,
                        const TString& clientId);
     void WaitCmdWrite(const TCmdWriteMatcher& matcher = {});
@@ -488,7 +489,8 @@ void TPartitionFixture::SendCreateSession(ui64 cookie,
 void TPartitionFixture::SendSetOffset(ui64 cookie,
                                       const TString& clientId,
                                       ui64 offset,
-                                      const TString& sessionId)
+                                      const TString& sessionId,
+                                      bool strict)
 {
     auto event = MakeHolder<TEvPQ::TEvSetClientInfo>(cookie,
                                                      clientId,
@@ -498,6 +500,7 @@ void TPartitionFixture::SendSetOffset(ui64 cookie,
                                                      0,
                                                      0,
                                                      TActorId{});
+    event->Strict = strict;
     Ctx->Runtime->SingleSys()->Send(new IEventHandle(ActorId, Ctx->Edge, event.Release()));
 }
 
@@ -712,7 +715,6 @@ void TPartitionFixture::WaitErrorResponse(const TErrorMatcher& matcher)
     }
 
     if (matcher.Error) {
-
         UNIT_ASSERT_VALUES_EQUAL(*matcher.Error, event->Error);
     }
 }
@@ -1796,6 +1798,28 @@ void TPartitionTxTestHelper::WaitTxPredicateReplyImpl(ui64 userActId, bool statu
 #endif
 }
 
+Y_UNIT_TEST_F(UserActCount, TPartitionFixture)
+{
+    // In the test, we check that the reference count for `UserInfo` decreases in case of errors. To do this,
+    // we send a large number of requests to which the server will respond with an error.
+
+    CreatePartition();
+
+    SendCreateSession(1, "client", "session-id", 2, 3);
+    WaitCmdWrite({.Count=2, .UserInfos={{0, {.Session="session-id", .Offset=0, .Generation=2, .Step=3}}}});
+    SendCmdWriteResponse(NMsgBusProxy::MSTATUS_OK);
+    WaitProxyResponse({.Cookie=1});
+
+    for (ui64 k = 0; k <= MAX_USER_ACTS; ++k) {
+        const ui64 cookie = 2 + k;
+        // 1 > EndOffset
+        SendSetOffset(cookie, "client", 1, "session-id", true); // strict = true
+        WaitCmdWrite({.Count=2, .UserInfos={{0, {.Session="session-id", .Offset=0, .Generation=2, .Step=3}}}});
+        SendCmdWriteResponse(NMsgBusProxy::MSTATUS_OK);
+        WaitErrorResponse({.Cookie=cookie, .ErrorCode=NPersQueue::NErrorCode::SET_OFFSET_ERROR_COMMIT_TO_FUTURE});
+    }
+}
+
 Y_UNIT_TEST_F(Batching, TPartitionFixture)
 {
     CreatePartition();