Fixed double closing of the read session (#24717)

Author: nshestakov

ydb/services/persqueue_v1/actors/read_session_actor.cpp

@@ -690,7 +690,9 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPQProxy::TEvCommitDone::
         << ", from# " << msg->StartCookie
         << ", to# " << msg->LastCookie
         << ", offset# " << partition.Offset);
-    WriteToStreamOrDie(ctx, std::move(result));
+    if (!WriteToStreamOrDie(ctx, std::move(result))) {
+        return;
+    }
 
     NotifyChildren(partition, ctx);
 }
@@ -1098,8 +1100,8 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPQProxy::TEvAuthResultOk
 
         if (IsQuotaRequired()) {
             Y_ABORT_UNLESS(MaybeRequestQuota(1, EWakeupTag::RlInit, ctx));
-        } else {
-            InitSession(ctx);
+        } else  if (!InitSession(ctx)) {
+            return;
         }
     } else {
         for (const auto& [name, t] : ev->Get()->TopicAndTablets) {
@@ -1123,17 +1125,17 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPQProxy::TEvAuthResultOk
 }
 
 template <bool UseMigrationProtocol>
-void TReadSessionActor<UseMigrationProtocol>::InitSession(const TActorContext& ctx) {
+bool TReadSessionActor<UseMigrationProtocol>::InitSession(const TActorContext& ctx) {
     TServerMessage result;
     result.set_status(Ydb::StatusIds::SUCCESS);
 
     result.mutable_init_response()->set_session_id(Session);
     if (!WriteToStreamOrDie(ctx, std::move(result))) {
-        return;
+        return false;
     }
 
     if (!ReadFromStreamOrDie(ctx)) {
-        return;
+        return false;
     }
 
     for (auto& [_, holder] : Topics) {
@@ -1153,10 +1155,13 @@ void TReadSessionActor<UseMigrationProtocol>::InitSession(const TActorContext& c
     for (const auto& [topicName, topic] : Topics) {
         if (ReadWithoutConsumer) {
             if (topic->Groups.size() == 0) {
-                return CloseSession(PersQueue::ErrorCode::BAD_REQUEST, "explicitly specify the partitions when reading without a consumer", ctx);
+                CloseSession(PersQueue::ErrorCode::BAD_REQUEST, "explicitly specify the partitions when reading without a consumer", ctx);
+                return false;
             }
             for (auto group : topic->Groups) {
-                SendLockPartitionToSelf(group-1, topicName, topic, ctx);
+                if (!SendLockPartitionToSelf(group-1, topicName, topic, ctx)) {
+                    return false;
+                }
             }
         } else {
             RegisterSession(topic->FullConverter->GetInternalName(), topic->PipeClient, topic->Groups, ctx);
@@ -1166,13 +1171,16 @@ void TReadSessionActor<UseMigrationProtocol>::InitSession(const TActorContext& c
     }
 
     ctx.Schedule(TDuration::Seconds(AppData(ctx)->PQConfig.GetACLRetryTimeoutSec()), new TEvents::TEvWakeup(EWakeupTag::RecheckAcl));
+
+    return true;
 }
 
 template <bool UseMigrationProtocol>
-void TReadSessionActor<UseMigrationProtocol>::SendLockPartitionToSelf(ui32 partitionId, TString topicName, const TTopicHolder::TPtr& topic, const TActorContext& ctx) {
+bool TReadSessionActor<UseMigrationProtocol>::SendLockPartitionToSelf(ui32 partitionId, TString topicName, const TTopicHolder::TPtr& topic, const TActorContext& ctx) {
     auto partitionIt = topic->Partitions.find(partitionId);
     if (partitionIt == topic->Partitions.end()) {
-        return CloseSession(PersQueue::ErrorCode::BAD_REQUEST, TStringBuilder() << "no partition " << partitionId << " in topic " << topicName, ctx);
+        CloseSession(PersQueue::ErrorCode::BAD_REQUEST, TStringBuilder() << "no partition " << partitionId << " in topic " << topicName, ctx);
+        return false;
     }
     THolder<TEvPersQueue::TEvLockPartition> res{new TEvPersQueue::TEvLockPartition};
     res->Record.SetSession(Session);
@@ -1184,6 +1192,8 @@ void TReadSessionActor<UseMigrationProtocol>::SendLockPartitionToSelf(ui32 parti
     res->Record.SetClientId(ClientId);
     res->Record.SetTabletId(partitionIt->second.TabletId);
     ctx.Send(ctx.SelfID, res.Release());
+
+    return true;
 }
 
 template <bool UseMigrationProtocol>
@@ -1783,7 +1793,9 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(NGRpcService::TGRpcRequestP
             TServerMessage result;
             result.set_status(Ydb::StatusIds::SUCCESS);
             result.mutable_update_token_response();
-            WriteToStreamOrDie(ctx, std::move(result));
+            if (!WriteToStreamOrDie(ctx, std::move(result))) {
+                return;
+            }
         }
     } else {
         if (ev->Get()->Retryable) {
@@ -2321,7 +2333,7 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvents::TEvWakeup::TPtr& e
 
     switch (tag) {
         case EWakeupTag::RlInit:
-            return InitSession(ctx);
+            return (void)InitSession(ctx);
 
         case EWakeupTag::RecheckAcl:
             return RecheckACL(ctx);

ydb/services/persqueue_v1/actors/read_session_actor.h

@@ -272,8 +272,8 @@ class TReadSessionActor
         }
     }
 
-    bool ReadFromStreamOrDie(const TActorContext& ctx);
-    bool WriteToStreamOrDie(const TActorContext& ctx, TServerMessage&& response, bool finish = false);
+    [[nodiscard]] bool ReadFromStreamOrDie(const TActorContext& ctx);
+    [[nodiscard]] bool WriteToStreamOrDie(const TActorContext& ctx, TServerMessage&& response, bool finish = false);
     bool SendControlMessage(TPartitionId id, TServerMessage&& message, const TActorContext& ctx);
 
     // grpc events
@@ -324,10 +324,10 @@ class TReadSessionActor
 
     void RunAuthActor(const TActorContext& ctx);
     void RecheckACL(const TActorContext& ctx);
-    void InitSession(const TActorContext& ctx);
+    [[nodiscard]] bool InitSession(const TActorContext& ctx);
     void RegisterSession(const TString& topic, const TActorId& pipe, const TVector<ui32>& groups, const TActorContext& ctx);
     void CloseSession(PersQueue::ErrorCode::ErrorCode code, const TString& reason, const TActorContext& ctx);
-    void SendLockPartitionToSelf(ui32 partitionId, TString topicName, const TTopicHolder::TPtr& topic, const TActorContext& ctx);
+    [[nodiscard]] bool SendLockPartitionToSelf(ui32 partitionId, TString topicName, const TTopicHolder::TPtr& topic, const TActorContext& ctx);
 
     void SetupBytesReadByUserAgentCounter();
     void SetupCounters();