Fix unprotected read commit flag check, KIKIMR-13910

ref:46b4d329e084a3d31618a7bc1b85b29eae3cb8ba

2 files changed, 15 insertions, 3 deletions

diff --git a/ydb/core/tx/datashard/datashard_snapshots.cpp b/ydb/core/tx/datashard/datashard_snapshots.cpp
index 39426396f3..80bacf7fe3 100644
--- a/ydb/core/tx/datashard/datashard_snapshots.cpp
+++ b/ydb/core/tx/datashard/datashard_snapshots.cpp
@@ -283,7 +283,7 @@ bool TSnapshotManager::GetPerformedUnprotectedReads() const {
 }
 
 bool TSnapshotManager::IsPerformedUnprotectedReadsCommitted() const {
-    return PerformedUnprotectedReadsUncommitted != 0;
+    return PerformedUnprotectedReadsUncommitted == 0;
 }
 
 void TSnapshotManager::SetPerformedUnprotectedReads(bool performedUnprotectedReads, TTransactionContext& txc) {
diff --git a/ydb/core/tx/datashard/datashard_ut_order.cpp b/ydb/core/tx/datashard/datashard_ut_order.cpp
index 6ad0dc0fbf..2a923845e7 100644
--- a/ydb/core/tx/datashard/datashard_ut_order.cpp
+++ b/ydb/core/tx/datashard/datashard_ut_order.cpp
@@ -4181,6 +4181,9 @@ Y_UNIT_TEST_NEW_ENGINE(TestShardSnapshotReadNoEarlyReply) {
     auto &runtime = *server->GetRuntime();
     auto sender = runtime.AllocateEdgeActor();
 
+    // runtime.SetLogPriority(NKikimrServices::TABLET_MAIN, NLog::PRI_TRACE);
+    // runtime.SetLogPriority(NKikimrServices::TX_DATASHARD, NLog::PRI_TRACE);
+
     InitRoot(server, sender);
 
     CreateShardedTable(server, sender, "/Root", "table-1", 1);
@@ -4202,15 +4205,18 @@ Y_UNIT_TEST_NEW_ENGINE(TestShardSnapshotReadNoEarlyReply) {
     SimulateSleep(server, TDuration::Seconds(1));
 
     auto waitFor = [&](const auto& condition, const TString& description) {
-        if (!condition()) {
+        for (int i = 0; i < 5; ++i) {
+            if (condition()) {
+                return;
+            }
             Cerr << "... waiting for " << description << Endl;
             TDispatchOptions options;
             options.CustomFinalCondition = [&]() {
                 return condition();
             };
             runtime.DispatchEvents(options);
-            UNIT_ASSERT_C(condition(), "... failed to wait for " << description);
         }
+        UNIT_ASSERT_C(condition(), "... failed to wait for " << description);
     };
 
     TVector<THolder<IEventHandle>> blockedCommits;
@@ -4281,6 +4287,12 @@ Y_UNIT_TEST_NEW_ENGINE(TestShardSnapshotReadNoEarlyReply) {
         txId2 = response.GetResponse().GetTxMeta().id();
     }
 
+    // Perform a couple of immediate reads to make sure shards are ready to respond to read-only requests
+    // This may be needed when leases are enabled, otherwise paused leases would block the snapshot read reply
+    ExecSQL(server, sender, Q_("SELECT * FROM `/Root/table-1` WHERE key = 1"));
+    ExecSQL(server, sender, Q_("SELECT * FROM `/Root/table-2` WHERE key = 2"));
+    Cerr << "... shards are ready for read-only immediate transactions" << Endl;
+
     // Start blocking commits again and try performing new writes
     prevObserver = runtime.SetObserverFunc(blockCommits);
     SendRequest(runtime, sender, MakeSimpleRequest(Q_("UPSERT INTO `/Root/table-1` (key, value) VALUES (3, 3)"), "/Root"));