feat ydb: support retries for interactive tx

Author: Gazizonoki

cmake/SetupYdbCppSDK.cmake

@@ -25,6 +25,10 @@ write_package_stub(jwt-cpp)
 
 set(RAPIDJSON_INCLUDE_DIRS "${USERVER_THIRD_PARTY_DIRS}/rapidjson/include")
 
+if(Protobuf_INCLUDE_DIR)
+    set(Protobuf_INCLUDE_DIR "${Protobuf_INCLUDE_DIR}" CACHE PATH "" FORCE)
+endif()
+
 if(TARGET userver-api-common-protos)
     set(YDB_SDK_GOOGLE_COMMON_PROTOS_TARGET userver-api-common-protos)
 else()
@@ -37,6 +41,7 @@ cpmaddpackage(
     GIT_TAG v3.13.0
     GITHUB_REPOSITORY ydb-platform/ydb-cpp-sdk
     GIT_SHALLOW TRUE
+    PATCHES ydb-cpp-sdk_protobuf_include.patch
     OPTIONS "Brotli_VERSION ${Brotli_VERSION}" "RAPIDJSON_INCLUDE_DIRS ${RAPIDJSON_INCLUDE_DIRS}"
             "YDB_SDK_GOOGLE_COMMON_PROTOS_TARGET ${YDB_SDK_GOOGLE_COMMON_PROTOS_TARGET}" "YDB_SDK_EXAMPLES OFF"
 )

cmake/ydb-cpp-sdk_protobuf_include.patch

@@ -0,0 +1,45 @@
+From 0000000000000000000000000000000000000000 Mon Sep 17 00:00:00 2001
+From: userver <userver@yandex-team.ru>
+Date: Tue, 25 Feb 2026 00:00:00 +0000
+Subject: [PATCH] Fix protobuf 4.24 compatibility
+
+1. cmake/protobuf: Add Protobuf well-known types include path so that
+   protoc can find google/protobuf/struct.proto when Protobuf is fetched
+   via CPM.
+
+2. operation.h: Remove include of google/protobuf/stubs/status.h which
+   was removed in protobuf 4.22+. The include is unnecessary because
+   MessageToJsonString now returns absl::Status.
+
+---
+ cmake/protobuf.cmake                                              | 4 ++++
+ include/ydb-cpp-sdk/client/types/operation/operation.h             | 1 -
+ 2 files changed, 4 insertions(+), 1 deletion(-)
+
+diff --git a/cmake/protobuf.cmake b/cmake/protobuf.cmake
+index 1111111..2222222 100644
+--- a/cmake/protobuf.cmake
++++ b/cmake/protobuf.cmake
+@@ -55,6 +55,10 @@ function(_ydb_sdk_init_proto_library_impl Tgt USE_API_COMMON_PROTOS)
+ 
+   set(proto_incls ${YDB_SDK_SOURCE_DIR})
+ 
++  if(Protobuf_INCLUDE_DIR)
++    list(APPEND proto_incls ${Protobuf_INCLUDE_DIR})
++  endif()
++
+   if (USE_API_COMMON_PROTOS)
+     target_link_libraries(${Tgt} PUBLIC
+       api-common-protos
+diff --git a/include/ydb-cpp-sdk/client/types/operation/operation.h b/include/ydb-cpp-sdk/client/types/operation/operation.h
+index 3333333..4444444 100644
+--- a/include/ydb-cpp-sdk/client/types/operation/operation.h
++++ b/include/ydb-cpp-sdk/client/types/operation/operation.h
+@@ -6,7 +6,6 @@
+ 
+ #include <library/cpp/threading/future/future.h>
+ 
+-#include <google/protobuf/stubs/status.h>
+ #include <google/protobuf/timestamp.pb.h>
+ #include <google/protobuf/util/json_util.h>
+ 

samples/ydb_service/views/upsert-2rows/post/view.cpp

@@ -27,29 +27,31 @@ VALUES ($id_key, $name_key, $service_key, $channel_key, CurrentUtcTimestamp(), $
         ydb::Query::LogMode::kNameOnly,
     };
 
-    auto trx = Ydb().Begin("trx", ydb::TransactionMode::kSerializableRW);
-
-    for (auto i : {1, 2}) {
-        auto response = trx.Execute(
-            kUpsertQuery,  //
-            "$id_key",
-            request["id"].As<std::string>() + std::to_string(i),  //
-            "$name_key",
-            ydb::Utf8{request["name"].As<std::string>() + std::to_string(i)},  //
-            "$service_key",
-            request["service"].As<std::string>(),  //
-            "$channel_key",
-            request["channel"].As<int64_t>(),  //
-            "$state_key",
-            request["state"].As<std::optional<formats::json::Value>>()  //
-        );
-
-        if (response.GetCursorCount() != 0) {
-            throw std::runtime_error("Unexpected response data");
+    Ydb().RetryTx("trx", {.tx_mode = ydb::TransactionMode::kSerializableRW},
+        [&](ydb::TxActor& tx) {
+            for (auto i : {1, 2}) {
+                auto response = tx.Execute(
+                    kUpsertQuery,  //
+                    "$id_key",
+                    request["id"].As<std::string>() + std::to_string(i),  //
+                    "$name_key",
+                    ydb::Utf8{request["name"].As<std::string>() + std::to_string(i)},  //
+                    "$service_key",
+                    request["service"].As<std::string>(),  //
+                    "$channel_key",
+                    request["channel"].As<int64_t>(),  //
+                    "$state_key",
+                    request["state"].As<std::optional<formats::json::Value>>()  //
+                );
+
+                if (response.GetCursorCount() != 0) {
+                    throw std::runtime_error("Unexpected response data");
+                }
+            }
+
+            return ydb::TxAction::kCommit;
         }
-    }
-
-    trx.Commit();
+    );
 
     return formats::json::MakeObject();
 }

ydb/functional_tests/basic/views/upsert-row/post/view.cpp

@@ -34,26 +34,29 @@ UpsertRowHandler::HandleRequestJsonThrow(const server::http::HttpRequest&, const
     const {
     engine::SleepFor(std::chrono::milliseconds(10));
 
-    auto trx = Ydb().Begin("trx", ydb::TransactionMode::kSerializableRW);
-    auto response = trx.Execute(
-        kUpsertQuery,  //
-        "$id_key",
-        request["id"].As<std::string>(),  //
-        "$name_key",
-        ydb::Utf8{request["name"].As<std::string>()},  //
-        "$service_key",
-        request["service"].As<std::string>(),  //
-        "$channel_key",
-        request["channel"].As<int64_t>(),  //
-        "$state_key",
-        request["state"].As<std::optional<formats::json::Value>>()  //
-    );
+    Ydb().RetryTx("trx", {.tx_mode = ydb::TransactionMode::kSerializableRW},
+        [&](ydb::TxActor& tx) {
+            auto response = tx.Execute(
+                kUpsertQuery,  //
+                "$id_key",
+                request["id"].As<std::string>(),  //
+                "$name_key",
+                ydb::Utf8{request["name"].As<std::string>()},  //
+                "$service_key",
+                request["service"].As<std::string>(),  //
+                "$channel_key",
+                request["channel"].As<int64_t>(),  //
+                "$state_key",
+                request["state"].As<std::optional<formats::json::Value>>()  //
+            );
 
-    if (response.GetCursorCount()) {
-        throw std::runtime_error("Unexpected response data");
-    }
+            if (response.GetCursorCount()) {
+                throw std::runtime_error("Unexpected response data");
+            }
 
-    trx.Commit();
+            return ydb::TxAction::kCommit;
+        }
+    );
 
     return formats::json::MakeObject();
 }

ydb/include/userver/ydb/builder.hpp

@@ -37,6 +37,7 @@ class PreparedArgsBuilder final {
 private:
     friend class Transaction;
     friend class TableClient;
+    friend class TxActor;
     struct PreparedArgsWithKey;
 
     NYdb::TParams Build() && { return std::move(builder_).Build(); }

ydb/include/userver/ydb/io/structs.hpp

@@ -13,6 +13,8 @@
 #include <boost/pfr/core.hpp>
 #include <boost/pfr/core_name.hpp>
 
+#include <fmt/ranges.h>
+
 #include <userver/utils/assert.hpp>
 #include <userver/utils/constexpr_indices.hpp>
 #include <userver/utils/enumerate.hpp>

ydb/include/userver/ydb/settings.hpp

@@ -3,6 +3,7 @@
 #include <chrono>
 #include <cstdint>
 #include <optional>
+#include <string>
 #include <string_view>
 
 #include <ydb-cpp-sdk/client/table/query_stats/stats.h>
@@ -33,6 +34,26 @@ struct QuerySettings final {
     std::optional<NYdb::NTable::ECollectQueryStatsMode> collect_query_stats{std::nullopt};
 };
 
+struct RequestSettings final {
+    std::chrono::milliseconds timeout_ms{0};
+
+    std::string trace_id{};
+};
+
+using ExecuteSettings = RequestSettings;
+using CommitSettings = RequestSettings;
+using RollbackSettings = RequestSettings;
+
+struct RetryTxSettings final {
+    TransactionMode tx_mode{TransactionMode::kSerializableRW};
+    std::chrono::milliseconds timeout_ms{0};
+    std::uint32_t retries{10};
+    bool is_idempotent{false};
+
+    CommitSettings commit_settings;
+    RollbackSettings rollback_settings;
+};
+
 }  // namespace ydb
 
 namespace formats::parse {

ydb/include/userver/ydb/table.hpp

@@ -33,6 +33,7 @@ namespace impl {
 struct Stats;
 struct TableSettings;
 class Driver;
+template <typename Settings = OperationSettings>
 struct RequestContext;
 enum class IsStreaming : bool {};
 }  // namespace impl
@@ -48,7 +49,8 @@ using DescribeTableSettings = NYdb::NTable::TDescribeTableSettings;
 using DropTableSettings = NYdb::NTable::TDropTableSettings;
 using ScanQuerySettings = NYdb::NTable::TStreamExecScanQuerySettings;
 
-/// @brief A dynamic transaction name for @see TableClient::Begin.
+/// @brief A dynamic transaction name for @see TableClient::Begin or
+/// @see TableClient::RetryTx.
 ///
 /// @warning Make sure that transaction name has low cardinality.
 /// If transaction name is unique for every call, per-transaction metrics will overflow metrics quota,
@@ -119,19 +121,45 @@ class TableClient final {
     );
     /// @}
 
-    /// @name Transactions
+    /// @name Transactions with retry
+    /// @brief Execute a transactional function with automatic retries.
+    ///
+    /// The user-provided function receives a TxActor for executing queries
+    /// and returns TxAction::kCommit or TxAction::kRollback. On transient
+    /// errors the whole function is retried automatically.
+    ///
+    /// @code
+    /// client.RetryTx("my_tx", {.retries = 3},
+    ///     [](ydb::TxActor& tx) {
+    ///         tx.Execute(query, "$id", 1);
+    ///         return ydb::TxAction::kCommit;
+    ///     });
+    /// @endcode
+    ///
+    /// @{
+    void RetryTx(utils::StringLiteral transaction_name, RetryTxSettings retry_settings, RetryTxFunction fn);
+
+    /// @warning Make sure that `transaction_name` has low cardinality.
+    void RetryTx(DynamicTransactionName transaction_name, RetryTxSettings retry_settings, RetryTxFunction fn);
+    /// @}
+
+    /// @name Transactions (deprecated)
     /// @brief Begin a transaction with the specified name. The settings are used
     /// for the `BEGIN` statement.
+    /// @deprecated Use RetryTx instead for automatic retry support.
     /// @see ydb::Transaction
     ///
     /// @{
+    [[deprecated("Use RetryTx instead")]]
     Transaction Begin(utils::StringLiteral transaction_name, OperationSettings settings = {});
 
     /// @warning Make sure that `transaction_name` has low cardinality.
     /// If `transaction_name` is unique for every call, per-transaction metrics will overflow metrics quota,
     /// and metrics will become unusable.
+    [[deprecated("Use RetryTx instead")]]
     Transaction Begin(DynamicTransactionName transaction_name, OperationSettings settings = {});
 
+    [[deprecated("Use RetryTx instead")]]
     Transaction Begin(utils::StringLiteral transaction_name, TransactionMode tx_mode);
     /// @}
 
@@ -233,6 +261,8 @@ class TableClient final {
 
 private:
     friend class Transaction;
+    friend class TxActor;
+    template <typename Settings>
     friend struct impl::RequestContext;
 
     std::string JoinDbPath(std::string_view path) const;
@@ -248,7 +278,7 @@ class TableClient final {
     //       (TTableClient&, const std::string& full_path, const Settings&)
     //       -> NThreading::TFuture<T>
     // ExecuteSchemeQueryImpl -> T
-    template <typename QuerySettings, typename Func>
+    template <typename FuncResult, typename QuerySettings, typename Func>
     auto ExecuteWithPathImpl(
         std::string_view path,
         std::string_view operation_name,

ydb/include/userver/ydb/transaction.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <functional>
 #include <string>
 
 #include <ydb-cpp-sdk/client/query/client.h>
@@ -18,8 +19,84 @@ USERVER_NAMESPACE_BEGIN
 
 namespace ydb {
 
+/// @brief Transaction actor for use with TableClient::RetryTx.
+///
+/// Provides only query execution within a transaction. Commit and rollback
+/// are controlled by returning TxAction from the retry function.
+/// https://ydb.tech/docs/en/concepts/transactions
+class TxActor {
+public:
+    TxActor(const TxActor&) = delete;
+    TxActor& operator=(const TxActor&) = delete;
+    TxActor(TxActor&&) noexcept = delete;
+    TxActor& operator=(TxActor&&) = delete;
+
+    /// Execute a single data query as a part of the transaction. Query parameters
+    /// are passed in `Args` as "string key - value" pairs:
+    ///
+    /// @code
+    /// tx.Execute(query, "name1", value1, "name2", value2, ...);
+    /// @endcode
+    ///
+    /// Use ydb::PreparedArgsBuilder for storing a generic buffer of query params
+    /// if needed.
+    ///
+    /// @{
+    template <typename... Args>
+    ExecuteResponse Execute(const Query& query, Args&&... args);
+
+    template <typename... Args>
+    ExecuteResponse Execute(ExecuteSettings settings, const Query& query, Args&&... args);
+
+    ExecuteResponse Execute(ExecuteSettings settings, const Query& query, PreparedArgsBuilder&& builder);
+    /// @}
+
+    PreparedArgsBuilder GetBuilder() const;
+
+private:
+    friend class TableClient;
+
+    TxActor(
+        TableClient& table_client,
+        NYdb::NQuery::TTransaction ydb_tx,
+        std::string name
+    ) noexcept;
+
+    TableClient& table_client_;
+    std::string name_;
+    impl::StatsScope stats_scope_;
+    tracing::Span span_;
+    NYdb::NQuery::TTransaction ydb_tx_;
+};
+
+template <typename... Args>
+ExecuteResponse TxActor::Execute(const Query& query, Args&&... args) {
+    auto builder = GetBuilder();
+    builder.AddParams(std::forward<Args>(args)...);
+    return Execute(ExecuteSettings{}, query, std::move(builder));
+}
+
+template <typename... Args>
+ExecuteResponse TxActor::Execute(ExecuteSettings settings, const Query& query, Args&&... args) {
+    auto builder = GetBuilder();
+    builder.AddParams(std::forward<Args>(args)...);
+    return Execute(std::move(settings), query, std::move(builder));
+}
+
+/// Action to take after the retry function completes.
+enum class TxAction {
+    kCommit,
+    kRollback,
+};
+
+/// Signature for the function passed to TableClient::RetryTx.
+using RetryTxFunction = std::function<TxAction(TxActor&)>;
+
 /// @brief YDB Transaction
 ///
+/// @deprecated Use TableClient::RetryTx instead of manually managing
+/// transactions with Begin/Commit/Rollback.
+///
 /// https://ydb.tech/docs/en/concepts/transactions
 class Transaction final {
 public: