YT-28163: fix tx pinger tsan race

commit_hash:c55147c95565f8451f2a0ccfcc72e6818bb35959

1 files changed, 6 insertions, 7 deletions

diff --git a/yt/cpp/mapreduce/client/transaction_pinger.cpp b/yt/cpp/mapreduce/client/transaction_pinger.cpp
index 7a659b081ec..855f31d6fd1 100644
--- a/yt/cpp/mapreduce/client/transaction_pinger.cpp
+++ b/yt/cpp/mapreduce/client/transaction_pinger.cpp
@@ -53,12 +53,11 @@ public:
         auto periodic = std::make_shared<NConcurrency::TPeriodicExecutorPtr>(nullptr);
         // Have to use weak_ptr in order to break reference cycle
         // This weak_ptr holds pointer to periodic, which will contain this lambda
-        // Also we consider that lifetime of this lambda is no longer than lifetime of pingableTx
-        // because every pingableTx have to call RemoveTransaction before it is destroyed
-        auto pingRoutine = BIND([this, &pingableTx, periodic = std::weak_ptr{periodic}] {
+        // Don't capture pingableTx by reference: an in-flight ping may outlive it and race with its destruction
+        auto pingRoutine = BIND([this, rawClient = RawClient_, transactionId = pingableTx.GetId(), periodic = std::weak_ptr{periodic}] {
             auto strong_ptr = periodic.lock();
             YT_VERIFY(strong_ptr);
-            DoPingTransaction(pingableTx, *strong_ptr);
+            DoPingTransaction(rawClient, transactionId, *strong_ptr);
         });
         *periodic = New<NConcurrency::TPeriodicExecutor>(ThreadPool_->GetInvoker(), pingRoutine, opts);
         (*periodic)->Start();
@@ -104,10 +103,10 @@ public:
     }
 
 private:
-    void DoPingTransaction(const TPingableTransaction& pingableTx, NConcurrency::TPeriodicExecutorPtr periodic)
+    void DoPingTransaction(const IRawClientPtr& rawClient, const TTransactionId& transactionId, NConcurrency::TPeriodicExecutorPtr periodic)
     {
         try {
-            pingableTx.Ping();
+            rawClient->PingTransaction(transactionId);
         } catch (const TErrorResponse& e) {
             /// NB: No logging here, CheckError() already logged TErrorResponse.
             if (e.GetError().ContainsErrorCode(NYT::NClusterErrorCodes::NTransactionClient::NoSuchTransaction)) {
@@ -117,7 +116,7 @@ private:
             }
         } catch (const std::exception& e) {
             YT_LOG_ERROR("DoPingTransaction has failed (TransactionId: %v, Error: %v)",
-                GetGuidAsString(pingableTx.GetId()),
+                GetGuidAsString(transactionId),
                 e.what());
         }
     }