Stop test runtime as soon as stop condition is satisfied (#7887)

Author: snaury

ydb/core/quoter/kesus_quoter_ut.cpp

@@ -237,6 +237,9 @@ Y_UNIT_TEST_SUITE(KesusProxyTest) {
             return setup.GetPipeFactory().GetPipesCreatedCount() >= 2;
         };
         setup.GetRuntime().DispatchEvents(reconnected);
+
+        // Dispatch some events to let poison pill reach the mock
+        setup.GetRuntime().SimulateSleep(TDuration::Zero());
     }
 
     Y_UNIT_TEST(ReconnectsWithKesusWhenPipeDestroyed) {
@@ -249,6 +252,9 @@ Y_UNIT_TEST_SUITE(KesusProxyTest) {
         setup.SendDestroyed(pipeMock);
 
         setup.WaitPipesCreated(2);
+
+        // Dispatch some events to let poison pill reach the mock
+        setup.GetRuntime().SimulateSleep(TDuration::Zero());
     }
 
     Y_UNIT_TEST(RejectsNotCanonizedResourceName) {
@@ -391,6 +397,9 @@ Y_UNIT_TEST_SUITE(KesusProxyTest) {
         setup.WaitEvent<NKesus::TEvKesus::TEvUpdateConsumptionState>();
         setup.SendCloseSession("res", 42);
         setup.WaitEvent<TEvQuota::TEvProxyCloseSession>();
+
+        // Dispatch some events to let pending events reach their destinations
+        setup.GetRuntime().SimulateSleep(TDuration::Zero());
     }
 
     void SendsProxySessionOnce(bool onSuccess) {

ydb/core/tx/datashard/datashard_ut_volatile.cpp

@@ -1224,8 +1224,12 @@ Y_UNIT_TEST_SUITE(DataShardVolatile) {
                 "value:   [\n"
                 "    2\n"
                 "  ]\n");
+        }
+
+        SimulateSleep(runtime, TDuration::MilliSeconds(0));
 
-            msg = readResults.back()->Get<TEvDataShard::TEvReadResult>();
+        {
+            auto* msg = readResults.back()->Get<TEvDataShard::TEvReadResult>();
             UNIT_ASSERT_VALUES_EQUAL(msg->Record.GetStatus().GetCode(), Ydb::StatusIds::SUCCESS);
             UNIT_ASSERT_VALUES_EQUAL(msg->Record.GetFinished(), true);
         }

ydb/library/actors/testlib/test_runtime.cpp

@@ -1166,6 +1166,26 @@ namespace NActors {
             tempEdgeEventsCaptor.Reset(new TTempEdgeEventsCaptor(*this));
         }
 
+        auto checkStopConditions = [&](bool perMessage = false) -> bool {
+            // Note: too many tests expect unrelated messages to be
+            // processed before simulation is stopped.
+            if (!perMessage) {
+                if (localContext.FinalEventFound) {
+                    return true;
+                }
+
+                if (!localContext.FoundNonEmptyMailboxes.empty()) {
+                    return true;
+                }
+            }
+
+            if (options.CustomFinalCondition && options.CustomFinalCondition()) {
+                return true;
+            }
+
+            return false;
+        };
+
         TEventMailBoxList& currentMailboxes = useRestrictedMailboxes ? restrictedMailboxes : Mailboxes;
         while (!currentMailboxes.empty()) {
             bool hasProgress = true;
@@ -1195,7 +1215,8 @@ namespace NActors {
                     isEmpty = true;
                     auto mboxIt = startWithMboxIt;
                     TDeque<TEventMailboxId> suspectedBoxes;
-                    while (true) {
+                    bool stopCondition = false;
+                    while (!stopCondition) {
                         auto& mbox = *mboxIt;
                         bool isIgnored = true;
                         if (!mbox.second->IsEmpty()) {
@@ -1264,6 +1285,9 @@ namespace NActors {
                                         case EEventAction::PROCESS:
                                             UpdateFinalEventsStatsForEachContext(*ev);
                                             SendInternal(ev.Release(), mbox.first.NodeId - FirstNodeId, false);
+                                            if (checkStopConditions(/* perMessage */ true)) {
+                                                stopCondition = true;
+                                            }
                                             break;
                                         case EEventAction::DROP:
                                             // do nothing
@@ -1305,19 +1329,17 @@ namespace NActors {
                             currentMailboxes.erase(it);
                         }
                     }
+
+                    if (stopCondition) {
+                        return true;
+                    }
                 }
             }
 
-            if (localContext.FinalEventFound) {
+            if (checkStopConditions()) {
                 return true;
             }
 
-            if (!localContext.FoundNonEmptyMailboxes.empty())
-                return true;
-
-            if (options.CustomFinalCondition && options.CustomFinalCondition())
-                return true;
-
             if (options.FinalEvents.empty()) {
                 for (auto& mbox : currentMailboxes) {
                     if (!mbox.second->IsActive(TInstant::MicroSeconds(CurrentTimestamp)))

ydb/services/persqueue_v1/ut/partition_writer_cache_actor_fixture.cpp

@@ -188,6 +188,9 @@ void TPartitionWriterCacheActorFixture::WaitForPartitionWriterOps(const TWaitFor
     };
 
     Ctx->Runtime->DispatchEvents(options);
+
+    // Tests rely on unrelated events processed after the condition is satisfied
+    Ctx->Runtime->DispatchEvents({}, TInstant::Zero());
 }
 
 void TPartitionWriterCacheActorFixture::AdvanceCurrentTime(TDuration d)