Import YDB C++ SDK 43

.github/import_generation.txt

@@ -1 +1 @@
-43
+44

.github/last_commit.txt

@@ -1 +1 @@
-b3b57fcb5915100b21bf9a507bf0a8c0627b7fee
+fb25fb6453264260a02c6ec49492382fcaf801ce

.github/scripts/run_iam_integration_tests.sh

@@ -0,0 +1,47 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+IAM_REGEX='^(DriverAuth|TMetadataFixture|TJwtIamFixture|TOAuthIamFixture|OAuth_WithFacility)\.'
+IAM_CONTAINER_NAME="${IAM_CONTAINER_NAME:-ydb-iam}"
+IAM_CTEST_JOBS="${IAM_CTEST_JOBS:-2}"
+IAM_READY_ATTEMPTS="${IAM_READY_ATTEMPTS:-60}"
+IAM_READY_SLEEP_SECONDS="${IAM_READY_SLEEP_SECONDS:-2}"
+IAM_TOKEN="${IAM_TOKEN:-root@builtin}"
+
+cleanup_iam() {
+  docker rm -f "${IAM_CONTAINER_NAME}" >/dev/null 2>&1 || true
+}
+
+wait_for_iam_ydb() {
+  for _ in $(seq 1 "${IAM_READY_ATTEMPTS}"); do
+    if docker exec -e "YDB_TOKEN=${IAM_TOKEN}" "${IAM_CONTAINER_NAME}" /ydb \
+        --endpoint grpc://localhost:2136 \
+        --database /local \
+        sql -s 'select 1' >/dev/null 2>&1; then
+      return 0
+    fi
+    sleep "${IAM_READY_SLEEP_SECONDS}"
+  done
+
+  echo "ydb-iam failed to become ready"
+  docker logs "${IAM_CONTAINER_NAME}" 2>&1 | tail -50 || true
+  return 1
+}
+
+trap cleanup_iam EXIT
+cleanup_iam
+
+docker run -d --name "${IAM_CONTAINER_NAME}" --hostname localhost \
+  -p 2235:2135 -p 2236:2136 -p 28765:8765 \
+  -v /tmp/ydb_iam_certs:/ydb_certs \
+  -e YDB_USE_IN_MEMORY_PDISKS=true \
+  -e YDB_TABLE_ENABLE_PREPARED_DDL=true \
+  -e YDB_ENFORCE_USER_TOKEN_REQUIREMENT=true \
+  -e YDB_DEFAULT_CLUSTERADMIN=root@builtin \
+  ghcr.io/ydb-platform/local-ydb:trunk
+
+wait_for_iam_ydb
+
+YDB_ENDPOINT=localhost:2236 YDB_DATABASE=/local \
+  ctest -j"${IAM_CTEST_JOBS}" --test-dir build -R "${IAM_REGEX}" --output-on-failure

.github/workflows/coverage.yml

@@ -61,7 +61,15 @@ jobs:
       - name: Test
         shell: bash
         run: |
-          ctest -j1 --preset coverage-all --output-on-failure
+          set -euo pipefail
+
+          IAM_REGEX='^(DriverAuth|TMetadataFixture|TJwtIamFixture|TOAuthIamFixture|OAuth_WithFacility)\.'
+          FLAKY_REGEX='(ManyMessages|DiscoveryHang|DescribeHang)'
+          EXCLUDE_REGEX="${IAM_REGEX}|${FLAKY_REGEX}"
+
+          ctest -j1 --preset coverage-all -E "${EXCLUDE_REGEX}" --output-on-failure
+
+          IAM_CTEST_JOBS=1 ./.github/scripts/run_iam_integration_tests.sh
 
       - name: Generate coverage report
         uses: ./.github/actions/gcovr

.github/workflows/tests.yaml

@@ -117,7 +117,16 @@ jobs:
     - name: Test
       shell: bash
       run: |
-        YDB_VERSION=${{ matrix.ydb-version }} ctest -j2 --preset integration
+        IAM_REGEX='^(DriverAuth|TMetadataFixture|TJwtIamFixture|TOAuthIamFixture|OAuth_WithFacility)\.'
+
+        YDB_VERSION=${{ matrix.ydb-version }} ctest -j2 --preset integration \
+          -E "${IAM_REGEX}" --output-on-failure
+
+        case '${{ matrix.ydb-version }}' in
+          25.1|trunk)
+            ./.github/scripts/run_iam_integration_tests.sh
+            ;;
+        esac
 
   test-install:
     name: "Test CMake Install"
@@ -267,4 +276,4 @@ jobs:
       - name: Smoke-test generated .deb packages
         shell: bash
         run: |
-          ./scripts/test_deb_packages.sh build-deb
+          ./scripts/test_deb_packages.sh build-deb

CHANGELOG.md

@@ -1,3 +1,9 @@
+## v3.20.0
+
+* Added automatic retries for unary methods of table and query clients(ExecuteQuery, ExecuteScript, BulkUpsert, ReadRows).
+
+* Implemented native ranges(TRowRange) and iterators over both streaming query results and TResultSet.
+
 ## v3.19.0
 
 * Added metric buffer for batched metric writes to reduce OpenTelemetry hot-path overhead.

Testing/Temporary/LastTest.log

@@ -0,0 +1,3 @@
+Start testing: Jul 01 00:28 MSK
+----------------------------------------------------------
+End testing: Jul 01 00:28 MSK

examples/basic_example/basic_example.cpp

@@ -1,14 +1,14 @@
 #include "basic_example.h"
 
 #include <ydb-cpp-sdk/client/retry/retry.h>
-
+#include <ydb-cpp-sdk/client/result/rows.h>
 #include <util/string/cast.h>
 
-#include <format>
 
 using namespace NYdb;
 using namespace NYdb::NQuery;
 using namespace NYdb::NStatusHelpers;
+using NYdb::NRetry::TRetryOperationSettings;
 
 template <class T>
 std::string OptionalToString(const std::optional<T>& opt) {
@@ -338,14 +338,18 @@ void MultiStep(TQueryClient client) {
         return result2;
     })); // The end of the retried lambda
 
-    TResultSetParser parser(*resultSet);
     std::cout << "> MultiStep:" << std::endl;
-    while (parser.TryNextRow()) {
-        auto airDate = TInstant::Days(*parser.ColumnParser("air_date").GetOptionalUint64());
-
-        std::cout << "Episode " << OptionalToString(parser.ColumnParser("episode_id").GetOptionalUint64())
-            << ", Season: " << OptionalToString(parser.ColumnParser("season_id").GetOptionalUint64())
-            << ", Title: " << OptionalToString(parser.ColumnParser("title").GetOptionalUtf8())
+    TRowRange range(std::move(*resultSet));
+    for (auto [episodeId, seasonId, title, airDateDays] : range.Get<
+            std::optional<uint64_t>,
+            std::optional<uint64_t>,
+            std::optional<std::string>,
+            std::optional<uint64_t>>({"episode_id", "season_id", "title", "air_date"})) {
+        auto airDate = TInstant::Days(*airDateDays);
+
+        std::cout << "Episode " << OptionalToString(episodeId)
+            << ", Season: " << OptionalToString(seasonId)
+            << ", Title: " << OptionalToString(title)
             << ", Air date: " << airDate.FormatLocalTime("%a %b %d, %Y")
             << std::endl;
     }
@@ -395,7 +399,7 @@ void ExplicitTcl(TQueryClient client) {
 void StreamQuerySelect(TQueryClient client) {
     std::cout << "> StreamQuery:" << std::endl;
 
-    ThrowOnError(client.RetryQuerySync([](TQueryClient client) -> TStatus {
+    ThrowOnError(client.RetryQuerySync([](TSession session) -> TStatus {
         auto query = R"(
             DECLARE $series AS List<UInt64>;
 
@@ -419,44 +423,16 @@ void StreamQuerySelect(TQueryClient client) {
                                 .Build()
                                 .Build();
 
-        // Executes stream query
-        auto resultStreamQuery = client.StreamExecuteQuery(query, TTxControl::NoTx(), parameters).GetValueSync();
-
-        if (!resultStreamQuery.IsSuccess()) {
-            return resultStreamQuery;
-        }
-
-        // Iterates over results
-        bool eos = false;
-
-        while (!eos) {
-            auto streamPart = resultStreamQuery.ReadNext().ExtractValueSync();
-
-            if (!streamPart.IsSuccess()) {
-                eos = true;
-                if (!streamPart.EOS()) {
-                    return streamPart;
-                }
-                continue;
-            }
-
-            // It is possible to duplicate lines in the output stream due to an external retryer.
-            if (streamPart.HasResultSet()) {
-                auto rs = streamPart.ExtractResultSet();
-                TResultSetParser parser(rs);
-                while (parser.TryNextRow()) {
-                    std::cout << "Season"
-                            << ", SeriesId: " << OptionalToString(parser.ColumnParser("series_id").GetOptionalUint64())
-                            << ", SeasonId: " << OptionalToString(parser.ColumnParser("season_id").GetOptionalUint64())
-                            << ", Title: " << OptionalToString(parser.ColumnParser("title").GetOptionalUtf8())
-                            << ", Air date: " << parser.ColumnParser("first_aired").GetOptionalDate()->FormatLocalTime("%Y-%m-%d")
-                            << std::endl;
-                }
-            }
+        for (auto& parser : TRowRange(
+                session.StreamExecuteQuery(query, TTxControl::NoTx(), parameters).ExtractValueSync())) {
+            std::cout << "Season" << ", SeriesId: " << OptionalToString(parser.ColumnParser("series_id").GetOptionalUint64())
+                    << ", SeasonId: " << OptionalToString(parser.ColumnParser("season_id").GetOptionalUint64())
+                    << ", Title: " << OptionalToString(parser.ColumnParser("title").GetOptionalUtf8())
+                    << ", Air date: " << parser.ColumnParser("first_aired").GetOptionalDate()->FormatLocalTime("%Y-%m-%d")
+                    << std::endl;
         }
         return TStatus(EStatus::SUCCESS, NYdb::NIssue::TIssues());
     }));
-
 }
 
 

include/ydb-cpp-sdk/client/iam/common/generic_provider.h

@@ -168,7 +168,6 @@ class TGrpcIamCredentialsProvider : public ICredentialsProvider {
                     }
                 };
                 auto facility = weakFacility.lock();
-                auto self = weakSelf.lock();
 
                 try {
                     if (facility) {
@@ -178,7 +177,7 @@ class TGrpcIamCredentialsProvider : public ICredentialsProvider {
                 } catch (...) {
                 }
 
-                if (self) {
+                if (auto self = weakSelf.lock()) {
                     std::lock_guard guard(self->Lock_);
                     self->ResetContextImpl();
                 }
@@ -189,9 +188,13 @@ class TGrpcIamCredentialsProvider : public ICredentialsProvider {
             try {
                 RequestFiller_(req);
             } catch (...) {
+                const auto now = SysClock::now();
                 std::lock_guard guard(Lock_);
-                LastRequestError_ = TStringBuilder() << "Request failed: " << CurrentExceptionMessage();
+                LastRequestError_ = TStringBuilder()
+                    << "Last request error was at " << FormatSysTimeUtcIsoMicros(now)
+                    << ". Failed to prepare IAM request: " << CurrentExceptionMessage();
                 ResetContextImpl();
+                RescheduleOnFailure();
                 return;
             }
 
@@ -247,11 +250,23 @@ class TGrpcIamCredentialsProvider : public ICredentialsProvider {
                 if (Context_.has_value() || SysClock::now() < NextTicketUpdate_) {
                     return true;
                 }
-                FillContext(guard);
+                try {
+                    FillContext(guard);
+                } catch (...) {
+                    const auto now = SysClock::now();
+                    LastRequestError_ = TStringBuilder()
+                        << "Last request error was at " << FormatSysTimeUtcIsoMicros(now)
+                        << ". Failed to prepare IAM request context: " << CurrentExceptionMessage();
+                    ResetContextImpl();
+                }
                 if (NeedStop_) {
                     ResetContextImpl();
                     return false;
                 }
+                if (!Context_.has_value()) {
+                    RescheduleOnFailure();
+                    return true;
+                }
             }
             UpdateTicket();
             return true;
@@ -267,31 +282,40 @@ class TGrpcIamCredentialsProvider : public ICredentialsProvider {
                     << " Message: \"" << status.error_message()
                     << "\" iam-endpoint: \"" << IamEndpoint_.Endpoint << "\"";
 
-                const auto now = SysClock::now();
-                const auto retryDelay = std::min(BackoffTimeout_, BACKOFF_MAX);
-                NextTicketUpdate_ = SafeAddSystemTime(now, ToBoundedSysDuration(retryDelay));
-                BackoffTimeout_ = std::min(BackoffTimeout_ * 2, BACKOFF_MAX);
+                RescheduleOnFailure();
             } else {
                 LastRequestError_ = "";
                 Ticket_ = result.iam_token();
-                BackoffTimeout_ = BACKOFF_START;
 
-                const auto now = SysClock::now();
-                const SysTimePoint refreshAt = SafeAddSystemTime(now, ToBoundedSysDuration(IamEndpoint_.RefreshPeriod));
                 const SysTimePoint expiresAt = SysClock::from_time_t(result.expires_at().seconds());
-                const SysDuration requestMargin = ToBoundedSysDuration(IamEndpoint_.RequestTimeout);
-
-                SysTimePoint nextUpdate = std::min(refreshAt, expiresAt);
-                nextUpdate = SafeAddSystemTime(nextUpdate, -requestMargin);
-                nextUpdate = std::max(nextUpdate, SafeAddSystemTime(now, ToBoundedSysDuration(MINIMUM_REFRESH_INTERVAL)));
-                NextTicketUpdate_ = nextUpdate;
+                RescheduleOnSuccess(expiresAt);
 
                 TokenReady_.notify_all();
             }
 
             ResetContextImpl();
         }
 
+        void RescheduleOnFailure() { // call with Lock_
+            const auto now = SysClock::now();
+            const auto retryDelay = std::min(BackoffTimeout_, BACKOFF_MAX);
+            NextTicketUpdate_ = SafeAddSystemTime(now, ToBoundedSysDuration(retryDelay));
+            BackoffTimeout_ = std::min(BackoffTimeout_ * 2, BACKOFF_MAX);
+        }
+
+        void RescheduleOnSuccess(const SysTimePoint expiresAt) { // call with Lock_
+            BackoffTimeout_ = BACKOFF_START;
+
+            const auto now = SysClock::now();
+            const SysTimePoint refreshAt = SafeAddSystemTime(now, ToBoundedSysDuration(IamEndpoint_.RefreshPeriod));
+            const SysDuration requestMargin = ToBoundedSysDuration(IamEndpoint_.RequestTimeout);
+
+            SysTimePoint nextUpdate = std::min(refreshAt, expiresAt);
+            nextUpdate = SafeAddSystemTime(nextUpdate, -requestMargin);
+            nextUpdate = std::max(nextUpdate, SafeAddSystemTime(now, ToBoundedSysDuration(MINIMUM_REFRESH_INTERVAL)));
+            NextTicketUpdate_ = nextUpdate;
+        }
+
     private:
         std::shared_ptr<grpc::Channel> Channel_;
         std::shared_ptr<typename TService::Stub> Stub_;