fix many scheduled events (#7098)

Author: nshestakov

ydb/services/persqueue_v1/actors/partition_actor.cpp

@@ -149,12 +149,11 @@ void TPartitionActor::MakeCommit(const TActorContext& ctx) {
 TPartitionActor::~TPartitionActor() = default;
 
 
-void TPartitionActor::Bootstrap(const TActorContext&) {
-
+void TPartitionActor::Bootstrap(const TActorContext& ctx) {
     Become(&TThis::StateFunc);
+    ctx.Schedule(PREWAIT_DATA, new TEvents::TEvWakeup());
 }
 
-
 void TPartitionActor::SendCommit(const ui64 readId, const ui64 offset, const TActorContext& ctx) {
     NKikimrClient::TPersQueueRequest request;
     request.MutablePartitionRequest()->SetTopic(Topic->GetPrimaryPath());
@@ -1014,7 +1013,6 @@ void TPartitionActor::WaitDataInPartition(const TActorContext& ctx) {
 
     NTabletPipe::SendData(ctx, PipeClient, event.Release());
 
-    ctx.Schedule(PREWAIT_DATA, new TEvents::TEvWakeup());
     ctx.Schedule(WAIT_DATA, new TEvPQProxy::TEvDeadlineExceeded(WaitDataCookie));
 
     WaitDataInfly.insert(WaitDataCookie);
@@ -1213,13 +1211,18 @@ void TPartitionActor::HandlePoison(TEvents::TEvPoisonPill::TPtr&, const TActorCo
 }
 
 void TPartitionActor::Handle(TEvPQProxy::TEvDeadlineExceeded::TPtr& ev, const TActorContext& ctx) {
-    WaitDataInfly.erase(ev->Get()->Cookie);
-    HandleWakeup(ctx);
+    if (WaitDataInfly.erase(ev->Get()->Cookie)) {
+        DoWakeup(ctx);
+    }
 }
 
 void TPartitionActor::HandleWakeup(const TActorContext& ctx) {
-    if (ReadOffset >= EndOffset && WaitDataInfly.size() <= 1 && PipeClient) { //send one more
-        Y_ABORT_UNLESS(WaitForData);
+    DoWakeup(ctx);
+    ctx.Schedule(PREWAIT_DATA, new TEvents::TEvWakeup());
+}
+
+void TPartitionActor::DoWakeup(const TActorContext& ctx) {
+    if (WaitForData && ReadOffset >= EndOffset && WaitDataInfly.size() <= 1 && PipeClient) { //send one more
         WaitDataInPartition(ctx);
     }
 }

ydb/services/persqueue_v1/actors/partition_actor.h

@@ -125,6 +125,7 @@ class TPartitionActor : public NActors::TActorBootstrapped<TPartitionActor> {
 
     void HandlePoison(NActors::TEvents::TEvPoisonPill::TPtr& ev, const NActors::TActorContext& ctx);
     void HandleWakeup(const NActors::TActorContext& ctx);
+    void DoWakeup(const NActors::TActorContext& ctx);
 
     void InitLockPartition(const NActors::TActorContext& ctx);
     void InitStartReading(const NActors::TActorContext& ctx);