The PQ tablet loses its TEvReadSet (#17842) (#17855)

Alek5andr-Kotov · web-flow · commit 32a6a784db88 · 2025-04-29T13:49:57.000Z
diff --git a/ydb/core/persqueue/pq_impl.cpp b/ydb/core/persqueue/pq_impl.cpp
@@ -3133,6 +3133,11 @@ void TPersQueue::SetTxInFlyCounter()
 
 void TPersQueue::Handle(TEvPersQueue::TEvCancelTransactionProposal::TPtr& ev, const TActorContext& ctx)
 {
+    if (!InitCompleted) {
+        AddPendingEvent(ev.Release());
+        return;
+    }
+
     PQ_LOG_D("Handle TEvPersQueue::TEvCancelTransactionProposal");
 
     NKikimrPQ::TEvCancelTransactionProposal& event = ev->Get()->Record;
@@ -3149,6 +3154,11 @@ void TPersQueue::Handle(TEvPersQueue::TEvCancelTransactionProposal::TPtr& ev, co
 
 void TPersQueue::Handle(TEvPersQueue::TEvProposeTransaction::TPtr& ev, const TActorContext& ctx)
 {
+    if (!InitCompleted) {
+        AddPendingEvent(ev.Release());
+        return;
+    }
+
     const NKikimrPQ::TEvProposeTransaction& event = ev->Get()->GetRecord();
     PQ_LOG_D("Handle TEvPersQueue::TEvProposeTransaction " << event.ShortDebugString());
 
@@ -3328,6 +3338,11 @@ void TPersQueue::HandleConfigTransaction(TAutoPtr<TEvPersQueue::TEvProposeTransa
 
 void TPersQueue::Handle(TEvTxProcessing::TEvPlanStep::TPtr& ev, const TActorContext& ctx)
 {
+    if (!InitCompleted) {
+        AddPendingEvent(ev.Release());
+        return;
+    }
+
     PQ_LOG_D("Handle TEvTxProcessing::TEvPlanStep " << ev->Get()->Record.ShortDebugString());
 
     EvPlanStepQueue.emplace_back(ev->Sender, ev->Release().Release());
@@ -3337,6 +3352,11 @@ void TPersQueue::Handle(TEvTxProcessing::TEvPlanStep::TPtr& ev, const TActorCont
 
 void TPersQueue::Handle(TEvTxProcessing::TEvReadSet::TPtr& ev, const TActorContext& ctx)
 {
+    if (!InitCompleted) {
+        AddPendingEvent(ev.Release());
+        return;
+    }
+
     PQ_LOG_D("Handle TEvTxProcessing::TEvReadSet " << ev->Get()->Record.ShortDebugString());
 
     NKikimrTx::TEvReadSet& event = ev->Get()->Record;
@@ -3348,7 +3368,8 @@ void TPersQueue::Handle(TEvTxProcessing::TEvReadSet::TPtr& ev, const TActorConte
     }
 
     if (auto tx = GetTransaction(ctx, event.GetTxId()); tx && tx->PredicatesReceived.contains(event.GetTabletProducer())) {
-        if (tx->State >= NKikimrPQ::TTransaction::EXECUTED) {
+        if ((tx->State > NKikimrPQ::TTransaction::EXECUTED) ||
+            ((tx->State == NKikimrPQ::TTransaction::EXECUTED) && !tx->WriteInProgress)) {
             if (ack) {
                 PQ_LOG_D("send TEvReadSetAck to " << event.GetTabletProducer());
                 ctx.Send(ev->Sender, ack.release());
@@ -3724,15 +3745,17 @@ void TPersQueue::ProcessPlanStepQueue(const TActorContext& ctx)
 void TPersQueue::ProcessWriteTxs(const TActorContext& ctx,
                                  NKikimrClient::TKeyValueRequest& request)
 {
-    Y_ABORT_UNLESS(!WriteTxsInProgress);
+    Y_ABORT_UNLESS(!WriteTxsInProgress, "PQ %" PRIu64, TabletID());
 
     for (auto& [txId, state] : WriteTxs) {
+        // There may be cases when in one iteration of a record we change the state of a transaction and delete it
         auto tx = GetTransaction(ctx, txId);
-        Y_ABORT_UNLESS(tx);
-
-        tx->AddCmdWrite(request, state);
+        if (tx) {
+            PQ_LOG_D("write key for TxId " << txId);
+            tx->AddCmdWrite(request, state);
 
-        ChangedTxs.emplace(tx->Step, txId);
+            ChangedTxs.emplace(tx->Step, txId);
+        }
     }
 
     WriteTxs.clear();
@@ -3741,9 +3764,7 @@ void TPersQueue::ProcessWriteTxs(const TActorContext& ctx,
 void TPersQueue::ProcessDeleteTxs(const TActorContext& ctx,
                                   NKikimrClient::TKeyValueRequest& request)
 {
-    Y_ABORT_UNLESS(!WriteTxsInProgress,
-                   "PQ %" PRIu64,
-                   TabletID());
+    Y_ABORT_UNLESS(!WriteTxsInProgress, "PQ %" PRIu64, TabletID());
 
     for (ui64 txId : DeleteTxs) {
         PQ_LOG_D("delete key for TxId " << txId);
@@ -4840,8 +4861,9 @@ void TPersQueue::DeleteSupportivePartitions(const TActorContext& ctx)
 void TPersQueue::OnInitComplete(const TActorContext& ctx)
 {
     SignalTabletActive(ctx);
-    TryStartTransaction(ctx);
     InitCompleted = true;
+    ProcessPendingEvents();
+    TryStartTransaction(ctx);
 }
 
 ui64 TPersQueue::GetAllowedStep() const
@@ -4877,6 +4899,11 @@ void TPersQueue::Handle(TEvPQ::TEvSubDomainStatus::TPtr& ev, const TActorContext
 
 void TPersQueue::Handle(TEvPersQueue::TEvProposeTransactionAttach::TPtr &ev, const TActorContext &ctx)
 {
+    if (!InitCompleted) {
+        AddPendingEvent(ev.Release());
+        return;
+    }
+
     PQ_LOG_D("Handle TEvPersQueue::TEvProposeTransactionAttach " << ev->Get()->Record.ShortDebugString());
 
     const ui64 txId = ev->Get()->Record.GetTxId();
@@ -5087,6 +5114,21 @@ ui64 TPersQueue::GetGeneration() {
     return *TabletGeneration;
 }
 
+void TPersQueue::AddPendingEvent(IEventHandle* ev)
+{
+    PendingEvents.emplace_back(ev);
+}
+
+void TPersQueue::ProcessPendingEvents()
+{
+    auto events = std::move(PendingEvents);
+    PendingEvents.clear();
+
+    for (auto& ev : events) {
+        HandleHook(ev);
+    }
+}
+
 bool TPersQueue::HandleHook(STFUNC_SIG)
 {
     SetActivityType(NKikimrServices::TActivity::PERSQUEUE_ACTOR);
diff --git a/ydb/core/persqueue/pq_impl.h b/ydb/core/persqueue/pq_impl.h
@@ -555,6 +555,11 @@ class TPersQueue : public NKeyValue::TKeyValueFlat {
     void ResendEvReadSetToReceiversForState(const TActorContext& ctx, NKikimrPQ::TTransaction::EState state);
 
     void DeleteSupportivePartitions(const TActorContext& ctx);
+
+    TDeque<TAutoPtr<IEventHandle>> PendingEvents;
+
+    void AddPendingEvent(IEventHandle* ev);
+    void ProcessPendingEvents();
 };
 
 
diff --git a/ydb/core/persqueue/ut/pqtablet_ut.cpp b/ydb/core/persqueue/ut/pqtablet_ut.cpp
@@ -197,6 +197,7 @@ class TPQTabletFixture : public NUnitTest::TBaseFixture {
 
     void WaitReadSetAck(NHelpers::TPQTabletMock& tablet, const TReadSetAckMatcher& matcher);
     void SendReadSetAck(NHelpers::TPQTabletMock& tablet);
+    void WaitForNoReadSetAck(NHelpers::TPQTabletMock& tablet);
 
     void SendDropTablet(const TDropTabletParams& params);
     void WaitDropTabletReply(const TDropTabletReplyMatcher& matcher);
@@ -206,7 +207,7 @@ class TPQTabletFixture : public NUnitTest::TBaseFixture {
 
     void SendCancelTransactionProposal(const TCancelTransactionProposalParams& params);
 
-    void StartPQWriteTxsObserver();
+    void StartPQWriteTxsObserver(TAutoPtr<IEventHandle>* ev = nullptr);
     void WaitForPQWriteTxs();
 
     template <class T> void WaitForEvent(size_t count);
@@ -215,7 +216,7 @@ class TPQTabletFixture : public NUnitTest::TBaseFixture {
 
     void TestWaitingForTEvReadSet(size_t senders, size_t receivers);
 
-    void StartPQWriteObserver(bool& flag, unsigned cookie);
+    void StartPQWriteObserver(bool& flag, unsigned cookie, TAutoPtr<IEventHandle>* ev = nullptr);
     void WaitForPQWriteComplete(bool& flag);
 
     bool FoundPQWriteState = false;
@@ -241,6 +242,9 @@ class TPQTabletFixture : public NUnitTest::TBaseFixture {
     void WaitForTxState(ui64 txId, NKikimrPQ::TTransaction::EState state);
     void WaitForExecStep(ui64 step);
 
+    void InterceptSaveTxState(TAutoPtr<IEventHandle>& event);
+    void SendSaveTxState(TAutoPtr<IEventHandle>& event);
+
     //
     // TODO(abcdef): для тестирования повторных вызовов нужны примитивы Send+Wait
     //
@@ -522,6 +526,17 @@ void TPQTabletFixture::WaitReadSetAck(NHelpers::TPQTabletMock& tablet, const TRe
     }
 }
 
+void TPQTabletFixture::WaitForNoReadSetAck(NHelpers::TPQTabletMock& tablet)
+{
+    TDispatchOptions options;
+    options.CustomFinalCondition = [&]() {
+        return tablet.ReadSetAck.Defined();
+    };
+    Ctx->Runtime->DispatchEvents(options, TDuration::Seconds(2));
+
+    UNIT_ASSERT(!tablet.ReadSetAck.Defined());
+}
+
 void TPQTabletFixture::SendDropTablet(const TDropTabletParams& params)
 {
     auto event = MakeHolder<TEvPersQueue::TEvDropTablet>();
@@ -745,18 +760,21 @@ void TPQTabletFixture::WaitWriteResponse(const TWriteResponseMatcher& matcher)
     Ctx->Runtime->SetObserverFunc(prev);
 }
 
-void TPQTabletFixture::StartPQWriteObserver(bool& flag, unsigned cookie)
+void TPQTabletFixture::StartPQWriteObserver(bool& flag, unsigned cookie, TAutoPtr<IEventHandle>* ev)
 {
     flag = false;
 
-    auto observer = [&flag, cookie](TAutoPtr<IEventHandle>& event) {
+    auto observer = [&flag, cookie, ev](TAutoPtr<IEventHandle>& event) {
         if (auto* kvResponse = event->CastAsLocal<TEvKeyValue::TEvResponse>()) {
-            if (kvResponse->Record.HasCookie()) {
-            }
             if ((event->Sender == event->Recipient) &&
                 kvResponse->Record.HasCookie() &&
                 (kvResponse->Record.GetCookie() == cookie)) {
                 flag = true;
+
+                if (ev) {
+                    *ev = event;
+                    return TTestActorRuntimeBase::EEventAction::DROP;
+                }
             }
         }
 
@@ -793,9 +811,9 @@ void TPQTabletFixture::SendCancelTransactionProposal(const TCancelTransactionPro
                event.Release());
 }
 
-void TPQTabletFixture::StartPQWriteTxsObserver()
+void TPQTabletFixture::StartPQWriteTxsObserver(TAutoPtr<IEventHandle>* event)
 {
-    StartPQWriteObserver(FoundPQWriteTxs, 5); // TPersQueue::WRITE_TX_COOKIE
+    StartPQWriteObserver(FoundPQWriteTxs, 5, event); // TPersQueue::WRITE_TX_COOKIE
 }
 
 void TPQTabletFixture::WaitForPQWriteTxs()
@@ -1030,6 +1048,40 @@ void TPQTabletFixture::WaitForExecStep(ui64 step)
     UNIT_FAIL("expected execution step " << step);
 }
 
+void TPQTabletFixture::InterceptSaveTxState(TAutoPtr<IEventHandle>& ev)
+{
+    bool found = false;
+
+    TTestActorRuntimeBase::TEventFilter prev;
+    auto filter = [&](TTestActorRuntimeBase&, TAutoPtr<IEventHandle>& event) -> bool {
+        if (auto* msg = event->CastAsLocal<TEvKeyValue::TEvRequest>()) {
+            if (msg->Record.HasCookie() && (msg->Record.GetCookie() == 5)) { // WRITE_TX_COOKIE
+                ev = event;
+                found = true;
+                return true;
+            }
+        }
+
+        return false;
+    };
+    prev = Ctx->Runtime->SetEventFilter(filter);
+
+    TDispatchOptions options;
+    options.CustomFinalCondition = [&found]() {
+        return found;
+    };
+
+    UNIT_ASSERT(Ctx->Runtime->DispatchEvents(options));
+    UNIT_ASSERT(found);
+
+    Ctx->Runtime->SetEventFilter(prev);
+}
+
+void TPQTabletFixture::SendSaveTxState(TAutoPtr<IEventHandle>& event)
+{
+    Ctx->Runtime->Send(event);
+}
+
 Y_UNIT_TEST_F(Parallel_Transactions_1, TPQTabletFixture)
 {
     TestParallelTransactions("consumer", "consumer");
@@ -1926,6 +1978,45 @@ Y_UNIT_TEST_F(After_Restarting_The_Tablet_Sends_A_TEvReadSet_For_Transactions_In
     WaitReadSetEx(*tablet, {.Step=100, .TxId=txId_1, .Decision=NKikimrTx::TReadSetData::DECISION_COMMIT, .Count=2});
 }
 
+Y_UNIT_TEST_F(TEvReadSet_Is_Not_Sent_Ahead_Of_Time, TPQTabletFixture)
+{
+    const ui64 txId = 67890;
+    const ui64 mockTabletId = 22222;
+
+    NHelpers::TPQTabletMock* tablet = CreatePQTabletMock(mockTabletId);
+    PQTabletPrepare({.partitions=1}, {}, *Ctx);
+
+    SendProposeTransactionRequest({.TxId=txId,
+                                  .Senders={mockTabletId}, .Receivers={mockTabletId},
+                                  .TxOps={
+                                  {.Partition=0, .Consumer="user", .Begin=0, .End=0, .Path="/topic"},
+                                  }});
+    WaitProposeTransactionResponse({.TxId=txId,
+                                   .Status=NKikimrPQ::TEvProposeTransactionResult::PREPARED});
+
+    SendPlanStep({.Step=100, .TxIds={txId}});
+
+    WaitForCalcPredicateResult();
+
+    tablet->SendReadSet(*Ctx->Runtime, {.Step=100, .TxId=txId, .Target=Ctx->TabletId, .Decision=NKikimrTx::TReadSetData::DECISION_COMMIT});
+
+    //WaitProposeTransactionResponse({.TxId=txId,
+    //                               .Status=NKikimrPQ::TEvProposeTransactionResult::COMPLETE});
+
+    TAutoPtr<IEventHandle> kvRequest;
+    InterceptSaveTxState(kvRequest);
+
+    tablet->SendReadSet(*Ctx->Runtime, {.Step=100, .TxId=txId, .Target=Ctx->TabletId, .Decision=NKikimrTx::TReadSetData::DECISION_COMMIT});
+
+    WaitForNoReadSetAck(*tablet);
+
+    SendSaveTxState(kvRequest);
+
+    WaitForTxState(txId, NKikimrPQ::TTransaction::EXECUTED);
+
+    WaitReadSetAck(*tablet, {.Step=100, .TxId=txId, .Source=22222, .Target=Ctx->TabletId, .Consumer=Ctx->TabletId});
+}
+
 }
 
 }
