implement async provider methods with retries and refactor (#47249)

Co-authored-by: Artem Ermoshkin <[email protected]>

25 files changed, 881 insertions, 983 deletions

diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/common/generic_provider.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/common/generic_provider.h
index ea48c937654..f9168caf16c 100644
--- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/common/generic_provider.h
+++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/common/generic_provider.h
@@ -22,6 +22,8 @@
 
 namespace NYdb::inline Dev {
 
+using NCredentials::NDetail::TOwningFacilityCredentialsProvider;
+
 constexpr std::chrono::milliseconds BACKOFF_START{50};
 constexpr std::chrono::milliseconds BACKOFF_MAX{10000};
 constexpr std::chrono::milliseconds PERIODIC_TICK{100};
@@ -77,19 +79,16 @@ private:
               std::weak_ptr<ICoreFacility> responseFacility,
               TCredentialsProviderPtr authTokenProvider)
             : Rpc_(rpc)
-            , Ticket_("")
             , NextTicketUpdate_(SysTimePoint{})
             , IamEndpoint_(iamEndpoint)
             , RequestFiller_(requestFiller)
             , Context_(std::nullopt)
-            , LastRequestError_("")
             , NeedStop_(false)
             , BackoffTimeout_(BACKOFF_START)
             , Lock_()
             , ResponseFacility_(std::move(responseFacility))
             , AuthTokenProvider_(authTokenProvider)
-            , FirstTokenReady_(NThreading::NewPromise<void>())
-            , FirstTokenReadySet_(false)
+            , AuthInfo_(NThreading::NewPromise<std::string>())
         {
             std::shared_ptr<grpc::ChannelCredentials> creds = nullptr;
             if (IamEndpoint_.EnableSsl) {
@@ -112,7 +111,7 @@ private:
         void StartPeriodicTask() {
             auto facility = ResponseFacility_.lock();
             if (!facility) {
-                FailFirstToken("IAM-token provider response facility is not available");
+                Fail("IAM-token provider response facility is not available");
                 return;
             }
 
@@ -125,7 +124,7 @@ private:
                             return false;
                         }
                         if (status != EStatus::SUCCESS) {
-                            self->FailFirstToken(TStringBuilder()
+                            self->Fail(TStringBuilder()
                                 << "IAM-token provider periodic task failed with status "
                                 << static_cast<int>(status));
                             return false;
@@ -135,53 +134,30 @@ private:
                     PERIODIC_TICK
                 );
             } catch (...) {
-                FailFirstToken(TStringBuilder()
+                Fail(TStringBuilder()
                     << "Failed to start IAM-token provider periodic task: "
                     << CurrentExceptionMessage());
             }
         }
 
-        std::string GetTicket() {
+        NThreading::TFuture<std::string> GetAuthInfoAsync() {
             std::lock_guard guard(Lock_);
-            if (Ticket_.empty()) {
-                ythrow yexception() << "IAM-token not ready yet. " << LastRequestError_;
-            }
-            return Ticket_;
-        }
-
-        void WaitForToken() {
-            std::unique_lock guard(Lock_);
-            TokenReady_.wait_for(guard,
-                std::chrono::microseconds(2 * IamEndpoint_.RequestTimeout.MicroSeconds()),
-                [this]() {
-                    return NeedStop_ || !Ticket_.empty(); 
-                }
-            );
-        }
-
-        NThreading::TFuture<void> GetReadyFuture() const {
-            return FirstTokenReady_.GetFuture();
+            return AuthInfo_.GetFuture();
         }
 
         void Stop() {
-            bool setStoppedException = false;
+            NThreading::TPromise<std::string> promise;
             {
                 std::unique_lock guard(Lock_);
-                if (NeedStop_) {
-                    return;
-                }
                 NeedStop_ = true;
-                setStoppedException = MarkFirstTokenReadyLocked();
-                TokenReady_.notify_all();
+                promise = AuthInfo_;
                 if (Context_.has_value()) {
                     Context_->TryCancel();
                 }
                 ContextReady_.wait(guard, [this]() { return !Context_.has_value(); });
             }
-            if (setStoppedException) {
-                FirstTokenReady_.SetException(
-                    std::make_exception_ptr(yexception() << "IAM-token provider stopped before token was ready"));
-            }
+            promise.TrySetException(std::make_exception_ptr(
+                yexception() << "IAM-token provider stopped before token was ready"));
             Stub_.reset();
             Channel_.reset();
         }
@@ -189,21 +165,17 @@ private:
     private:
         using SysDuration = SysClock::duration;
 
-        bool MarkFirstTokenReadyLocked() {
-            return !std::exchange(FirstTokenReadySet_, true);
-        }
-
-        void FailFirstToken(std::string error) {
-            bool setException = false;
+        void Fail(std::string error) {
+            NThreading::TPromise<std::string> promise;
             {
                 std::lock_guard guard(Lock_);
-                if ((setException = MarkFirstTokenReadyLocked())) {
-                    LastRequestError_ = error;
+                NeedStop_ = true;
+                promise = AuthInfo_;
+                if (Context_) {
+                    Context_->TryCancel();
                 }
             }
-            if (setException) {
-                FirstTokenReady_.SetException(std::make_exception_ptr(yexception() << error));
-            }
+            promise.TrySetException(std::make_exception_ptr(yexception() << error));
         }
 
         static SysDuration ToBoundedSysDuration(const TDuration& d) {
@@ -251,16 +223,11 @@ private:
                 }
 
                 if (auto self = weakSelf.lock()) {
-                    bool failFirstToken;
                     {
                         std::lock_guard guard(self->Lock_);
-                        failFirstToken = self->MarkFirstTokenReadyLocked();
                         self->ResetContextImpl();
                     }
-                    if (failFirstToken) {
-                        self->FirstTokenReady_.SetException(std::make_exception_ptr(
-                            yexception() << "IAM-token provider response facility is not available"));
-                    }
+                    self->Fail("IAM-token provider response facility is not available");
                 }
             };
 
@@ -268,42 +235,36 @@ private:
 
             try {
                 RequestFiller_(req);
+                Rpc_(Stub_.get(), &*Context_, &req, response.get(), std::move(cb));
             } catch (...) {
-                std::optional<std::string> firstTokenError;
-                const auto now = SysClock::now();
-                {
-                    std::lock_guard guard(Lock_);
-                    LastRequestError_ = TStringBuilder()
-                        << "Last request error was at " << FormatSysTimeUtcIsoMicros(now)
-                        << ". Failed to prepare IAM request: " << CurrentExceptionMessage();
-                    if (MarkFirstTokenReadyLocked()) {
-                        firstTokenError = LastRequestError_;
-                    }
-                    ResetContextImpl();
-                    RescheduleOnFailure();
-                }
-                if (firstTokenError) {
-                    FirstTokenReady_.SetException(std::make_exception_ptr(yexception() << *firstTokenError));
-                }
-                return;
+                std::lock_guard guard(Lock_);
+                ResetContextImpl();
+                RescheduleOnFailure();
             }
-
-            Rpc_(Stub_.get(), &*Context_, &req, response.get(), std::move(cb));
         }
 
-        void FillContext(std::unique_lock<std::mutex>& guard) {
+        bool FillContext(std::unique_lock<std::mutex>& guard) {
             std::optional<std::string> authToken;
             if (AuthTokenProvider_) {
                 guard.unlock();
                 try {
-                    authToken = AuthTokenProvider_->GetAuthInfo();
+                    if (!AuthTokenInfo_.Initialized()) {
+                        AuthTokenInfo_ = AuthTokenProvider_->GetAuthInfoAsync();
+                    }
+                    if (!AuthTokenInfo_.IsReady()) {
+                        guard.lock();
+                        return false;
+                    }
+                    authToken = AuthTokenInfo_.GetValue();
+                    AuthTokenInfo_ = {};
                 } catch (...) {
+                    AuthTokenInfo_ = {};
                     guard.lock();
                     throw;
                 }
                 guard.lock();
                 if (NeedStop_) {
-                    return;
+                    return false;
                 }
             }
 
@@ -317,6 +278,7 @@ private:
             if (authToken) {
                 context.AddMetadata("authorization", "Bearer " + *authToken);
             }
+            return true;
         }
 
         void ResetContextImpl() {
@@ -332,8 +294,9 @@ private:
         }
 
         bool OnPeriodicTick() {
-            std::optional<std::string> firstTokenError;
+            std::optional<std::string> terminalError;
             bool updateTicket = false;
+            bool authPending = false;
             {
                 std::unique_lock guard(Lock_);
                 if (NeedStop_) {
@@ -342,16 +305,15 @@ private:
                 if (Context_.has_value() || SysClock::now() < NextTicketUpdate_) {
                     return true;
                 }
+                if (AuthInfo_.GetFuture().IsReady()) {
+                    AuthInfo_ = NThreading::NewPromise<std::string>();
+                }
                 try {
-                    FillContext(guard);
+                    authPending = !FillContext(guard);
                 } catch (...) {
-                    const auto now = SysClock::now();
-                    LastRequestError_ = TStringBuilder()
-                        << "Last request error was at " << FormatSysTimeUtcIsoMicros(now)
+                    terminalError = TStringBuilder()
+                        << "Last request error was at " << FormatSysTimeUtcIsoMicros(SysClock::now())
                         << ". Failed to prepare IAM request context: " << CurrentExceptionMessage();
-                    if (MarkFirstTokenReadyLocked()) {
-                        firstTokenError = LastRequestError_;
-                    }
                     ResetContextImpl();
                 }
                 if (NeedStop_) {
@@ -359,13 +321,16 @@ private:
                     return false;
                 }
                 if (!Context_.has_value()) {
-                    RescheduleOnFailure();
+                    if (!authPending && !terminalError) {
+                        RescheduleOnFailure();
+                    }
                 } else {
                     updateTicket = true;
                 }
             }
-            if (firstTokenError) {
-                FirstTokenReady_.SetException(std::make_exception_ptr(yexception() << *firstTokenError));
+            if (terminalError) {
+                Fail(*terminalError);
+                return false;
             }
             if (updateTicket) {
                 UpdateTicket();
@@ -374,38 +339,52 @@ private:
         }
 
         void ProcessIamResponse(grpc::Status&& status, TResponse&& result) {
-            bool setFirstTokenReady = false;
+            std::optional<std::string> token;
+            std::optional<std::string> terminalError;
+            NThreading::TPromise<std::string> promise;
 
             {
                 std::lock_guard guard(Lock_);
 
                 if (!status.ok()) {
-                    LastRequestError_ = TStringBuilder()
+                    const std::string error = TStringBuilder()
                         << "Last request error was at " << FormatSysTimeUtcIsoMicros(SysClock::now())
                         << ". GrpcStatusCode: " << static_cast<int>(status.error_code())
                         << " Message: \"" << status.error_message()
                         << "\" iam-endpoint: \"" << IamEndpoint_.Endpoint << "\"";
 
-                    RescheduleOnFailure();
+                    if (IsRetryable(status.error_code())) {
+                        RescheduleOnFailure();
+                    } else {
+                        terminalError = error;
+                    }
+                } else if (result.iam_token().empty()) {
+                    terminalError = "IAM-token service returned an empty token";
                 } else {
-                    LastRequestError_ = "";
-                    Ticket_ = result.iam_token();
+                    token = result.iam_token();
+                    promise = AuthInfo_;
 
                     const SysTimePoint expiresAt = SysClock::from_time_t(result.expires_at().seconds());
                     RescheduleOnSuccess(expiresAt);
-
-                    setFirstTokenReady = MarkFirstTokenReadyLocked();
-                    TokenReady_.notify_all();
                 }
 
                 ResetContextImpl();
             }
 
-            if (setFirstTokenReady) {
-                FirstTokenReady_.SetValue();
+            if (token) {
+                promise.TrySetValue(std::move(*token));
+            } else if (terminalError) {
+                Fail(*terminalError);
             }
         }
 
+        static bool IsRetryable(grpc::StatusCode code) {
+            return code == grpc::StatusCode::CANCELLED || code == grpc::StatusCode::UNKNOWN ||
+                code == grpc::StatusCode::DEADLINE_EXCEEDED || code == grpc::StatusCode::RESOURCE_EXHAUSTED ||
+                code == grpc::StatusCode::ABORTED || code == grpc::StatusCode::INTERNAL ||
+                code == grpc::StatusCode::UNAVAILABLE;
+        }
+
         void RescheduleOnFailure() { // call with Lock_
             const auto now = SysClock::now();
             const auto retryDelay = std::min(BackoffTimeout_, BACKOFF_MAX);
@@ -431,21 +410,18 @@ private:
         std::shared_ptr<typename TService::Stub> Stub_;
         TAsyncRpc Rpc_;
 
-        std::string Ticket_;
         SysTimePoint NextTicketUpdate_;
         const TIamEndpoint IamEndpoint_;
         const TRequestFiller RequestFiller_;
         std::optional<grpc::ClientContext> Context_;
         std::condition_variable ContextReady_;
-        std::condition_variable TokenReady_;
-        std::string LastRequestError_;
         bool NeedStop_;
         std::chrono::milliseconds BackoffTimeout_;
         std::mutex Lock_;
         std::weak_ptr<ICoreFacility> ResponseFacility_;
         TCredentialsProviderPtr AuthTokenProvider_;
-        NThreading::TPromise<void> FirstTokenReady_;
-        bool FirstTokenReadySet_;
+        NThreading::TFuture<std::string> AuthTokenInfo_;
+        NThreading::TPromise<std::string> AuthInfo_;
     };
 
 public:
@@ -453,14 +429,10 @@ public:
                                 const TRequestFiller& requestFiller,
                                 TAsyncRpc rpc,
                                 std::weak_ptr<ICoreFacility> responseFacility,
-                                TCredentialsProviderPtr authTokenProvider = nullptr,
-                                bool waitForToken = true)
+                                TCredentialsProviderPtr authTokenProvider = nullptr)
         : Impl_(std::make_shared<TImpl>(endpoint, requestFiller, rpc, std::move(responseFacility), authTokenProvider))
     {
         Impl_->StartPeriodicTask();
-        if (waitForToken) {
-            Impl_->WaitForToken();
-        }
     }
 
     ~TGrpcIamCredentialsProvider() {
@@ -468,90 +440,43 @@ public:
     }
 
     std::string GetAuthInfo() const override {
-        return Impl_->GetTicket();
+        return GetAuthInfoAsync().GetValueSync();
     }
 
-    bool IsValid() const override {
-        return true;
-    }
-
-    NThreading::TFuture<void> GetReadyFuture() const {
-        return Impl_->GetReadyFuture();
-    }
-
-private:
-    std::shared_ptr<TImpl> Impl_;
-};
-
-// Adapter that keeps a self-owned ICoreFacility alive for the lifetime of an inner credentials
-// provider. Used by deprecated no-arg ICredentialsProviderFactory::CreateProvider() paths where
-// the caller hasn't supplied a facility.
-class TOwningFacilityCredentialsProvider : public ICredentialsProvider {
-public:
-    TOwningFacilityCredentialsProvider(std::shared_ptr<ICoreFacility> facility,
-                                       TCredentialsProviderPtr inner)
-        : Facility_(std::move(facility))
-        , Inner_(std::move(inner))
-    {}
-
-    std::string GetAuthInfo() const override {
-        return Inner_->GetAuthInfo();
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
+        return Impl_->GetAuthInfoAsync();
     }
 
     bool IsValid() const override {
-        return Inner_->IsValid();
+        return true;
     }
 
 private:
-    // Field declaration order matters: Inner_ is destroyed first so that its Stop() can still
-    // drive the facility's queue (cancel the in-flight gRPC context, drain the response callback),
-    // and only then is Facility_ destroyed.
-    std::shared_ptr<ICoreFacility> Facility_;
-    TCredentialsProviderPtr Inner_;
+    std::shared_ptr<TImpl> Impl_;
 };
 
-namespace NPrivate {
-
-template <typename TProvider, typename TParams>
-NThreading::TFuture<TCredentialsProviderPtr> CreateGrpcIamCredentialsProviderAsync(
-    const TParams& params,
-    std::weak_ptr<ICoreFacility> facility,
-    std::shared_ptr<ICoreFacility> ownedFacility = {})
-{
-    auto inner = std::make_shared<TProvider>(params, std::move(facility), false);
-    auto ready = inner->GetReadyFuture();
-    TCredentialsProviderPtr provider = std::move(inner);
-    if (ownedFacility) {
-        provider = std::make_shared<TOwningFacilityCredentialsProvider>(
-            std::move(ownedFacility), std::move(provider));
-    }
-    return ready.Return(std::move(provider));
-}
-
-} // namespace NPrivate
-
 template<typename TRequest, typename TResponse, typename TService>
 class TIamJwtCredentialsProvider : public TGrpcIamCredentialsProvider<TRequest, TResponse, TService> {
 public:
-    TIamJwtCredentialsProvider(const TIamJwtParams& params, std::weak_ptr<ICoreFacility> responseFacility, bool waitForToken = true)
+    TIamJwtCredentialsProvider(const TIamJwtParams& params, std::weak_ptr<ICoreFacility> responseFacility)
         : TGrpcIamCredentialsProvider<TRequest, TResponse, TService>(params,
             [jwtParams = params.JwtParams](TRequest& req) {
                 req.set_jwt(MakeSignedJwt(jwtParams));
             }, [](typename TService::Stub* stub, grpc::ClientContext* context, const TRequest* request, TResponse* response, std::function<void(grpc::Status)> cb) {
                 stub->async()->Create(context, request, response, std::move(cb));
-            }, std::move(responseFacility), nullptr, waitForToken) {}
+            }, std::move(responseFacility)) {}
 };
 
 template<typename TRequest, typename TResponse, typename TService>
 class TIamOAuthCredentialsProvider : public TGrpcIamCredentialsProvider<TRequest, TResponse, TService> {
 public:
-    TIamOAuthCredentialsProvider(const TIamOAuth& params, std::weak_ptr<ICoreFacility> responseFacility, bool waitForToken = true)
+    TIamOAuthCredentialsProvider(const TIamOAuth& params, std::weak_ptr<ICoreFacility> responseFacility)
         : TGrpcIamCredentialsProvider<TRequest, TResponse, TService>(params,
             [token = params.OAuthToken](TRequest& req) {
                 req.set_yandex_passport_oauth_token(TStringType{token});
             }, [](typename TService::Stub* stub, grpc::ClientContext* context, const TRequest* request, TResponse* response, std::function<void(grpc::Status)> cb) {
                 stub->async()->Create(context, request, response, std::move(cb));
-            }, std::move(responseFacility), nullptr, waitForToken) {}
+            }, std::move(responseFacility)) {}
 };
 
 template<typename TRequest, typename TResponse, typename TService>
@@ -574,12 +499,6 @@ public:
             });
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync() const override {
-        auto facility = CreateSimpleCoreFacility();
-        return NPrivate::CreateGrpcIamCredentialsProviderAsync<
-            TIamJwtCredentialsProvider<TRequest, TResponse, TService>>(Params_, facility, facility);
-    }
-
     TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility> facility) const override {
         return std::make_shared<TIamJwtCredentialsProvider<TRequest, TResponse, TService>>(Params_, std::move(facility));
     }
@@ -595,11 +514,6 @@ public:
             Params_.JwtParams.PrivKey);
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const override {
-        return NPrivate::CreateGrpcIamCredentialsProviderAsync<
-            TIamJwtCredentialsProvider<TRequest, TResponse, TService>>(Params_, std::move(facility));
-    }
-
 private:
     TIamJwtParams Params_;
 };
@@ -622,12 +536,6 @@ public:
             });
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync() const override {
-        auto facility = CreateSimpleCoreFacility();
-        return NPrivate::CreateGrpcIamCredentialsProviderAsync<
-            TIamOAuthCredentialsProvider<TRequest, TResponse, TService>>(Params_, facility, facility);
-    }
-
     TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility> facility) const override {
         return std::make_shared<TIamOAuthCredentialsProvider<TRequest, TResponse, TService>>(Params_, std::move(facility));
     }
@@ -640,11 +548,6 @@ public:
             Params_.OAuthToken);
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const override {
-        return NPrivate::CreateGrpcIamCredentialsProviderAsync<
-            TIamOAuthCredentialsProvider<TRequest, TResponse, TService>>(Params_, std::move(facility));
-    }
-
 private:
     TIamOAuth Params_;
 };
diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h
index e4d428fb02e..398fe89df76 100644
--- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h
+++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h
@@ -19,7 +19,7 @@ public:
     // Add task to execute periodicaly
     // Task should return false to stop execution
     virtual void AddPeriodicTask(TPeriodicCb&& cb, TDeadline::Duration period) = 0;
-    // Post task on SDK response executor.
+    // Post task on SDK response executor, never inline.
     virtual void PostToResponseQueue(TPostTaskCb&& f) = 0;
 };
 
diff --git a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/credentials.h b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/credentials.h
index 67068921f84..8f31135be55 100644
--- a/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/credentials.h
+++ b/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/credentials.h
@@ -5,6 +5,7 @@
 #include <functional>
 #include <library/cpp/threading/future/future.h>
 
+#include <exception>
 #include <memory>
 #include <string>
 #include <utility>
@@ -16,9 +17,17 @@ public:
     virtual ~ICredentialsProvider() = default;
     virtual std::string GetAuthInfo() const = 0;
     virtual bool IsValid() const = 0;
+    virtual NThreading::TFuture<std::string> GetAuthInfoAsync() const {
+        try {
+            return NThreading::MakeFuture(GetAuthInfo());
+        } catch (...) {
+            return NThreading::MakeErrorFuture<std::string>(std::current_exception());
+        }
+    }
 };
 
 using TCredentialsProviderPtr = std::shared_ptr<ICredentialsProvider>;
+class ICoreFacility;
 
 // Implementation detail for SDK credentials factories. Symbols in NCredentials::NDetail are not
 // part of the public YDB C++ SDK API and may change or be removed without notice.
@@ -26,6 +35,35 @@ namespace NCredentials::NDetail {
 
 using TCredentialsProviderCreator = std::function<TCredentialsProviderPtr()>;
 
+class TOwningFacilityCredentialsProvider final : public ICredentialsProvider {
+public:
+    TOwningFacilityCredentialsProvider(std::shared_ptr<ICoreFacility> facility,
+                                       TCredentialsProviderPtr inner,
+                                       bool forwardAsync = false)
+        : Facility_(std::move(facility))
+        , Inner_(std::move(inner))
+        , ForwardAsync_(forwardAsync)
+    {}
+
+    std::string GetAuthInfo() const override {
+        return Inner_->GetAuthInfo();
+    }
+
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
+        return ForwardAsync_ ? Inner_->GetAuthInfoAsync() : ICredentialsProvider::GetAuthInfoAsync();
+    }
+
+    bool IsValid() const override {
+        return Inner_->IsValid();
+    }
+
+private:
+    // Reverse destruction keeps Facility_ alive while Inner_ stops.
+    std::shared_ptr<ICoreFacility> Facility_;
+    TCredentialsProviderPtr Inner_;
+    const bool ForwardAsync_;
+};
+
 // Process-wide weak cache for no-argument factory paths whose providers own their facilities.
 // Facility-bound providers must not use it: their callbacks belong to the supplied facility.
 TCredentialsProviderPtr GetOrCreateCachedProvider(
@@ -34,21 +72,15 @@ TCredentialsProviderPtr GetOrCreateCachedProvider(
 
 } // namespace NCredentials::NDetail
 
-class ICoreFacility;
 class ICredentialsProviderFactory {
 public:
     virtual ~ICredentialsProviderFactory() = default;
     // deprecated, use CreateProvider(std::weak_ptr<ICoreFacility> facility) instead
     virtual TCredentialsProviderPtr CreateProvider() const = 0;
-    virtual NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync() const {
-        return NThreading::MakeFuture(CreateProvider());
-    }
+    // The facility must outlive the returned provider.
     virtual TCredentialsProviderPtr CreateProvider([[maybe_unused]] std::weak_ptr<ICoreFacility> facility) const {
         return CreateProvider();
     }
-    virtual NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const {
-        return NThreading::MakeFuture(CreateProvider(std::move(facility)));
-    }
     virtual std::string GetClientIdentity() const;
 };
 
diff --git a/ydb/public/sdk/cpp/src/client/iam/iam.cpp b/ydb/public/sdk/cpp/src/client/iam/iam.cpp
index 06bbbccc54e..72383cc0bfe 100644
--- a/ydb/public/sdk/cpp/src/client/iam/iam.cpp
+++ b/ydb/public/sdk/cpp/src/client/iam/iam.cpp
@@ -15,120 +15,156 @@ using namespace yandex::cloud::iam::v1;
 
 namespace NYdb::inline Dev {
 
-class TIAMCredentialsProvider : public ICredentialsProvider {
+class TIAMCredentialsProvider : public ICredentialsProvider, public std::enable_shared_from_this<TIAMCredentialsProvider> {
 public:
-    TIAMCredentialsProvider(const TIamHost& params)
+    TIAMCredentialsProvider(const TIamHost& params, std::weak_ptr<ICoreFacility> facility)
         : HttpClient_(TSimpleHttpClient(TString(params.Host), params.Port))
         , Request_("/computeMetadata/v1/instance/service-accounts/default/token")
         , NextTicketUpdate_(TInstant::Zero())
         , RefreshPeriod_(params.RefreshPeriod)
-    {
-        GetTicket();
-    }
+        , Facility_(std::move(facility))
+        , AuthInfo_(NThreading::NewPromise<std::string>())
+    {}
 
-    std::string GetAuthInfo() const override {
-        std::string ticket;
-        TInstant nextTicketUpdate;
-        auto now = TInstant::Now();
-        std::optional<TString> lastErrorMessage;
-        {
-            std::lock_guard lock(Lock_);
-            if (LastErrorMessage_.has_value() && now > ExpiresAt_) {
-                Ticket_.clear();
-            }
-            ticket = Ticket_;
-            nextTicketUpdate = NextTicketUpdate_;
-            lastErrorMessage = LastErrorMessage_;
+    void Start() {
+        auto facility = Facility_.lock();
+        if (!facility) {
+            Fail("IAM-token provider response facility is not available");
+            return;
         }
-        if (now >= nextTicketUpdate) {
-            GetTicket();
-            {
-                std::lock_guard lock(Lock_);
-                if (LastErrorMessage_.has_value() && now > ExpiresAt_) {
-                    Ticket_.clear();
+        try {
+            facility->AddPeriodicTask([weak = weak_from_this()](NYdb::NIssue::TIssues&&, EStatus status) {
+                if (auto self = weak.lock()) {
+                    return self->OnPeriodicTick(status);
                 }
-                ticket = Ticket_;
-                lastErrorMessage = LastErrorMessage_;
-            }
-        }
-        if (ticket.empty() && lastErrorMessage.has_value()) {
-            throw yexception() << *lastErrorMessage;
+                return false;
+            }, PERIODIC_TICK);
+        } catch (...) {
+            Fail(CurrentExceptionMessage());
         }
-        return ticket;
+    }
+
+    std::string GetAuthInfo() const override {
+        return GetAuthInfoAsync().GetValueSync();
+    }
+
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
+        std::lock_guard lock(Lock_);
+        return AuthInfo_.GetFuture();
     }
 
     bool IsValid() const override {
         return true;
     }
 
+    ~TIAMCredentialsProvider() { Fail("IAM-token provider stopped"); }
+
 private:
     TSimpleHttpClient HttpClient_;
     std::string Request_;
     mutable std::mutex Lock_;
-    mutable std::string Ticket_;
     mutable TInstant NextTicketUpdate_;
-    mutable TInstant ExpiresAt_ = TInstant::Zero();
-    mutable std::optional<TString> LastErrorMessage_;
     TDuration RefreshPeriod_;
+    std::weak_ptr<ICoreFacility> Facility_;
+    mutable NThreading::TPromise<std::string> AuthInfo_;
+    mutable bool Stopped_ = false;
 
-    void GetTicket() const {
-        try {
-            TStringStream out;
-            TSimpleHttpClient::THeaders headers;
-            headers["Metadata-Flavor"] = "Google";
-            HttpClient_.DoGet(Request_, &out, headers);
-            NJson::TJsonValue resp;
-            NJson::ReadJsonTree(&out, &resp, true);
-
-            auto respMap = resp.GetMap();
+    void Fail(std::string error) const {
+        NThreading::TPromise<std::string> promise;
+        {
+            std::lock_guard lock(Lock_);
+            Stopped_ = true;
+            promise = AuthInfo_;
+        }
+        promise.TrySetException(std::make_exception_ptr(yexception() << error));
+    }
 
-            std::string ticket;
-            if (auto it = respMap.find("access_token"); it == respMap.end())
-                ythrow yexception() << "Result doesn't contain access_token";
-            else if (ticket = it->second.GetStringSafe(); ticket.empty())
-                ythrow yexception() << "Got empty ticket";
+    bool OnPeriodicTick(EStatus status) const {
+        if (status != EStatus::SUCCESS) {
+            Fail("IAM-token provider periodic task failed");
+            return false;
+        }
 
-            const auto now = TInstant::Now();
-            TInstant nextUpdate;
-            TDuration expiresIn;
-            TInstant expiresAt = TInstant::Max();
-            if (auto it = respMap.find("expires_in"); it != respMap.end()) {
-                auto seconds = it->second.GetUInteger();
-                if (seconds > 0) {
-                    expiresIn = TDuration::Seconds(seconds);
-                    expiresAt = now + expiresIn;
-                }
-            } else if (auto it = respMap.find("expiry"); it != respMap.end()) {
-                try {
-                    TInstant expiry;
-                    if (TInstant::TryParseIso8601(it->second.GetStringSafe(), expiry) && expiry > now) {
-                        expiresIn = expiry - now;
-                        expiresAt = expiry;
-                    }
-                } catch (...) {
-                    expiresAt = now;
-                }
+        NThreading::TPromise<std::string> promise;
+        {
+            std::lock_guard lock(Lock_);
+            if (Stopped_) {
+                return false;
             }
-            if (expiresIn > TDuration::Zero()) {
-                const auto halfLife = expiresIn / 2;
-                const auto interval = std::max(std::min(halfLife, RefreshPeriod_), TDuration::MilliSeconds(100));
-                nextUpdate = now + interval;
-            } else {
-                nextUpdate = now + std::min(RefreshPeriod_, TDuration::Minutes(30));
+            if (TInstant::Now() < NextTicketUpdate_) {
+                return true;
             }
+            if (AuthInfo_.GetFuture().IsReady()) {
+                AuthInfo_ = NThreading::NewPromise<std::string>();
+            }
+            promise = AuthInfo_;
+        }
 
+        try {
+            auto [ticket, nextUpdate] = GetTicket();
             {
                 std::lock_guard lock(Lock_);
-                Ticket_ = std::move(ticket);
                 NextTicketUpdate_ = nextUpdate;
-                ExpiresAt_ = expiresAt;
-                LastErrorMessage_.reset();
             }
+            promise.TrySetValue(std::move(ticket));
         } catch (...) {
+            const auto error = std::current_exception();
+            if (!IsRetryable(error)) {
+                Fail(CurrentExceptionMessage());
+                return false;
+            }
             std::lock_guard lock(Lock_);
             NextTicketUpdate_ = TInstant::Now() + std::min(RefreshPeriod_, TDuration::Seconds(10));
-            LastErrorMessage_ = CurrentExceptionMessage();
         }
+        return true;
+    }
+
+    static bool IsRetryable(const std::exception_ptr& error) {
+        try {
+            std::rethrow_exception(error);
+        } catch (const THttpRequestException& e) {
+            const int code = e.GetStatusCode();
+            return code == 0 || code == HTTP_REQUEST_TIME_OUT || code == HTTP_AUTHENTICATION_TIMEOUT ||
+                code == HTTP_TOO_MANY_REQUESTS || (code >= 500 && code < 600);
+        } catch (const TSystemError&) {
+            return true;
+        } catch (...) {
+            return false;
+        }
+    }
+
+    std::pair<std::string, TInstant> GetTicket() const {
+        TStringStream out;
+        TSimpleHttpClient::THeaders headers;
+        headers["Metadata-Flavor"] = "Google";
+        HttpClient_.DoGet(Request_, &out, headers);
+        NJson::TJsonValue resp;
+        NJson::ReadJsonTree(&out, &resp, true);
+
+        auto respMap = resp.GetMap();
+        std::string ticket;
+        if (auto it = respMap.find("access_token"); it == respMap.end())
+            ythrow yexception() << "Result doesn't contain access_token";
+        else if (ticket = it->second.GetStringSafe(); ticket.empty())
+            ythrow yexception() << "Got empty ticket";
+
+        const auto now = TInstant::Now();
+        TDuration expiresIn;
+        if (auto it = respMap.find("expires_in"); it != respMap.end()) {
+            const auto seconds = it->second.GetUInteger();
+            if (seconds > 0) {
+                expiresIn = TDuration::Seconds(seconds);
+            }
+        } else if (auto it = respMap.find("expiry"); it != respMap.end()) {
+            TInstant expiry;
+            if (TInstant::TryParseIso8601(it->second.GetStringSafe(), expiry) && expiry > now) {
+                expiresIn = expiry - now;
+            }
+        }
+        const auto interval = expiresIn > TDuration::Zero()
+            ? std::max(std::min(expiresIn / 2, RefreshPeriod_), TDuration::MilliSeconds(100))
+            : std::min(RefreshPeriod_, TDuration::Minutes(30));
+        return {std::move(ticket), now + interval};
     }
 };
 
@@ -140,21 +176,17 @@ public:
         return NCredentials::NDetail::GetOrCreateCachedProvider(
             GetClientIdentity(),
             [this] {
-                return std::make_shared<TIAMCredentialsProvider>(Params_);
+                auto facility = CreateSimpleCoreFacility();
+                auto provider = CreateProvider(facility);
+                return std::make_shared<TOwningFacilityCredentialsProvider>(
+                    std::move(facility), std::move(provider));
             });
     }
 
-    // Keep the facility-taking path driver-scoped; only the no-arg path is process-wide cached.
-    TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility>) const final {
-        return std::make_shared<TIAMCredentialsProvider>(Params_);
-    }
-
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync() const final {
-        return CreateProviderAsync(std::weak_ptr<ICoreFacility>{});
-    }
-
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const final {
-        return CreateProviderInBackground(Params_, std::move(facility));
+    TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility> facility) const final {
+        auto provider = std::make_shared<TIAMCredentialsProvider>(Params_, std::move(facility));
+        provider->Start();
+        return provider;
     }
 
     std::string GetClientIdentity() const final {
@@ -164,30 +196,6 @@ public:
     }
 
 private:
-    static NThreading::TFuture<TCredentialsProviderPtr> CreateProviderInBackground(
-        TIamHost params,
-        std::weak_ptr<ICoreFacility> facility)
-    {
-        auto promise = NThreading::NewPromise<TCredentialsProviderPtr>();
-        auto createProvider = [params = std::move(params), promise]() mutable {
-            try {
-                promise.TrySetValue(std::make_shared<TIAMCredentialsProvider>(params));
-            } catch (...) {
-                promise.TrySetException(std::current_exception());
-            }
-        };
-        try {
-            if (auto core = facility.lock()) {
-                core->PostToResponseQueue(std::move(createProvider));
-            } else {
-                createProvider();
-            }
-        } catch (...) {
-            promise.TrySetException(std::current_exception());
-        }
-        return promise.GetFuture();
-    }
-
     TIamHost Params_;
 };
 
diff --git a/ydb/public/sdk/cpp/src/client/iam_private/common/iam.h b/ydb/public/sdk/cpp/src/client/iam_private/common/iam.h
index 676172f40a6..7882ea19c16 100644
--- a/ydb/public/sdk/cpp/src/client/iam_private/common/iam.h
+++ b/ydb/public/sdk/cpp/src/client/iam_private/common/iam.h
@@ -2,8 +2,6 @@
 
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/common/generic_provider.h>
 
-#include <library/cpp/threading/future/core/coroutine_traits.h>
-
 namespace NYdb::inline Dev {
 
 template<typename TRequest, typename TResponse, typename TService>
@@ -27,88 +25,45 @@ private:
 
     class TCredentialsProvider : public TGrpcIamCredentialsProvider<TRequest, TResponse, TService> {
     public:
-        // TDriver path: a shared facility (TGRpcConnectionsImpl) supports multiple periodic tasks,
-        // so we can hand the same weak_ptr to the nested auth provider here.
-        TCredentialsProvider(const TIamServiceParams& params, std::weak_ptr<ICoreFacility> responseFacility)
-            : TGrpcIamCredentialsProvider<TRequest, TResponse, TService>(params,
-                MakeRequestFiller(params),
-                MakeRpc(),
-                responseFacility,
-                params.SystemServiceAccountCredentials->CreateProvider(responseFacility))
-        {}
-
-        // Standalone (no-arg) path: the caller has already built a self-owning auth provider
-        // backed by its OWN facility. We must not share `outerFacility` with the auth provider
-        // because TSimpleCoreFacility allows only one periodic task.
         TCredentialsProvider(const TIamServiceParams& params,
-                             std::weak_ptr<ICoreFacility> outerFacility,
-                             TCredentialsProviderPtr authProvider,
-                             bool waitForToken = true)
+                             std::weak_ptr<ICoreFacility> responseFacility,
+                             TCredentialsProviderPtr authProvider = {})
             : TGrpcIamCredentialsProvider<TRequest, TResponse, TService>(params,
                 MakeRequestFiller(params),
                 MakeRpc(),
-                std::move(outerFacility),
-                std::move(authProvider),
-                waitForToken)
+                responseFacility,
+                authProvider ? std::move(authProvider) :
+                    params.SystemServiceAccountCredentials->CreateProvider(responseFacility))
         {}
     };
 
-    static NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsyncImpl(
-        TIamServiceParams params,
-        NThreading::TFuture<TCredentialsProviderPtr> authProvider,
-        std::weak_ptr<ICoreFacility> facility,
-        std::shared_ptr<ICoreFacility> ownedFacility = {})
-    {
-        auto serviceProvider = std::make_shared<TCredentialsProvider>(
-            params, std::move(facility), co_await authProvider, false);
-        auto ready = serviceProvider->GetReadyFuture();
-        TCredentialsProviderPtr provider = std::move(serviceProvider);
-        if (ownedFacility) {
-            provider = std::make_shared<TOwningFacilityCredentialsProvider>(
-                std::move(ownedFacility), std::move(provider));
-        }
-        co_return co_await ready.Return(std::move(provider));
-    }
-
 public:
     TIamServiceCredentialsProviderFactory(const TIamServiceParams& params)
         : Params_(params)
     {}
 
-    // Deprecated. Kept for backward compatibility — see comment on TIamJwtCredentialsProviderFactory.
-    // The nested auth provider gets its own facility (via a recursive no-arg CreateProvider() that
-    // owns its private facility). Sharing a TSimpleCoreFacility between two gRPC
-    // IAM providers would abort: each one registers a periodic task and the facility allows only one.
     TCredentialsProviderPtr CreateProvider() const override final {
         return NCredentials::NDetail::GetOrCreateCachedProvider(
             GetClientIdentity(),
             [this] {
-                auto authProvider = Params_.SystemServiceAccountCredentials->CreateProvider();
-                auto outerFacility = CreateSimpleCoreFacility();
+                auto authProvider = NCredentials::NDetail::GetOrCreateCachedProvider(
+                    "async:" + Params_.SystemServiceAccountCredentials->GetClientIdentity(), [this] {
+                        auto facility = CreateSimpleCoreFacility();
+                        return std::make_shared<TOwningFacilityCredentialsProvider>(facility,
+                            Params_.SystemServiceAccountCredentials->CreateProvider(facility), true);
+                    });
+                auto facility = CreateSimpleCoreFacility();
                 auto serviceProvider = std::make_shared<TCredentialsProvider>(
-                    Params_, std::weak_ptr<ICoreFacility>(outerFacility), std::move(authProvider));
+                    Params_, facility, std::move(authProvider));
                 return std::make_shared<TOwningFacilityCredentialsProvider>(
-                    std::move(outerFacility), std::move(serviceProvider));
+                    std::move(facility), std::move(serviceProvider));
             });
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync() const override {
-        auto outerFacility = CreateSimpleCoreFacility();
-        return CreateProviderAsyncImpl(
-            Params_, Params_.SystemServiceAccountCredentials->CreateProviderAsync(),
-            outerFacility, outerFacility);
-    }
-
     TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility> facility) const override {
         return std::make_shared<TCredentialsProvider>(Params_, std::move(facility));
     }
 
-    NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const override {
-        return CreateProviderAsyncImpl(
-            Params_, Params_.SystemServiceAccountCredentials->CreateProviderAsync(facility),
-            facility);
-    }
-
     std::string GetClientIdentity() const override final {
         return NIam::NDetail::MakeClientIdentity(
             "TIamServiceCredentialsProviderFactory",
diff --git a/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.cpp b/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.cpp
index a4b031b29af..b44b0b99e8d 100644
--- a/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.cpp
+++ b/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.cpp
@@ -62,29 +62,24 @@ TDbDriverState::TDbDriverState(
 void TDbDriverState::InitCredentials(
     std::shared_ptr<ICredentialsProviderFactory> credentialsProviderFactory
 ) {
-    Credentials = credentialsProviderFactory->CreateProviderAsync(weak_from_this()).Apply(
-        [](const NThreading::TFuture<TCredentialsProviderPtr>& future) {
-            TCredentials result{future.GetValue()};
+    Credentials.Provider = credentialsProviderFactory->CreateProvider(weak_from_this());
 #ifndef YDB_GRPC_UNSECURE_AUTH
-            result.CallCredentials = grpc::MetadataCredentialsFromPlugin(
-                std::unique_ptr<grpc::MetadataCredentialsPlugin>(new TYdbAuthenticator(result.Provider)));
+    Credentials.CallCredentials = grpc::MetadataCredentialsFromPlugin(
+        std::unique_ptr<grpc::MetadataCredentialsPlugin>(new TYdbAuthenticator(Credentials.Provider)));
 #endif
-            return result;
-        });
-    CredentialsReady = Credentials.IgnoreResult();
 }
 
 NThreading::TFuture<void> TDbDriverState::GetCredentialsReady() const {
-    return CredentialsReady;
+    return Credentials.Provider->GetAuthInfoAsync().IgnoreResult();
 }
 
 std::shared_ptr<ICredentialsProvider> TDbDriverState::GetCredentialsProvider() const {
-    return Credentials.HasValue() ? Credentials.GetValue().Provider : nullptr;
+    return Credentials.Provider;
 }
 
 #ifndef YDB_GRPC_UNSECURE_AUTH
 std::shared_ptr<grpc::CallCredentials> TDbDriverState::GetCallCredentials() const {
-    return Credentials.HasValue() ? Credentials.GetValue().CallCredentials : nullptr;
+    return Credentials.CallCredentials;
 }
 #endif
 
diff --git a/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.h b/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.h
index 17cc3596b34..05b8b7dc045 100644
--- a/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.h
+++ b/ydb/public/sdk/cpp/src/client/impl/internal/db_driver_state/state.h
@@ -84,8 +84,7 @@ private:
 #endif
     };
 
-    NThreading::TFuture<void> CredentialsReady;
-    NThreading::TFuture<TCredentials> Credentials;
+    TCredentials Credentials;
     mutable std::once_flag ClientTlsValidationOnceFlag_;
     mutable bool ClientTlsCredentialsValid_ = true;
     mutable std::string ClientTlsValidationDetail_;
diff --git a/ydb/public/sdk/cpp/src/client/impl/internal/grpc_connections/grpc_connections.h b/ydb/public/sdk/cpp/src/client/impl/internal/grpc_connections/grpc_connections.h
index 6f2302303b7..7c56960e22c 100644
--- a/ydb/public/sdk/cpp/src/client/impl/internal/grpc_connections/grpc_connections.h
+++ b/ydb/public/sdk/cpp/src/client/impl/internal/grpc_connections/grpc_connections.h
@@ -335,7 +335,7 @@ public:
         if (auto ready = CredentialsReadyToWaitFor(dbState, requestSettings, context); ready.Initialized()) {
             DeferUntilCredentialsReady(requestSettings, context, std::move(ready),
                 [this, requestWrapper = std::move(requestWrapper), userResponseCb = std::move(userResponseCb),
-                 rpc, dbState, requestSettings, context = std::move(context)]
+                 rpc, dbState, requestSettings, context]
                 (std::optional<TPlainStatus> status) YDB_ASAN_SIZE_ATTRIBUTES mutable {
                     if (status) {
                         userResponseCb(nullptr, std::move(*status));
@@ -592,7 +592,7 @@ public:
 
         if (auto ready = CredentialsReadyToWaitFor(dbState, requestSettings, context); ready.Initialized()) {
             DeferUntilCredentialsReady(requestSettings, context, std::move(ready),
-                [this, request, responseCb = std::move(responseCb), rpc, dbState, requestSettings, context = std::move(context)]
+                [this, request, responseCb = std::move(responseCb), rpc, dbState, requestSettings, context]
                 (std::optional<TPlainStatus> status) YDB_ASAN_SIZE_ATTRIBUTES mutable {
                     if (status) {
                         responseCb(std::move(*status), nullptr);
@@ -699,7 +699,7 @@ public:
 
         if (auto ready = CredentialsReadyToWaitFor(dbState, requestSettings, context); ready.Initialized()) {
             DeferUntilCredentialsReady(requestSettings, context, std::move(ready),
-                [this, connectedCallback = std::move(connectedCallback), rpc, dbState, requestSettings, context = std::move(context)]
+                [this, connectedCallback = std::move(connectedCallback), rpc, dbState, requestSettings, context]
                 (std::optional<TPlainStatus> status) YDB_ASAN_SIZE_ATTRIBUTES mutable {
                     if (status) {
                         connectedCallback(std::move(*status), nullptr);
diff --git a/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.cpp b/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.cpp
index 1da3278a16d..de5adc71936 100644
--- a/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.cpp
+++ b/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.cpp
@@ -29,7 +29,6 @@ TWriteSessionImpl::TWriteSessionImpl(
     , Client(std::move(client))
     , Connections(std::move(connections))
     , DbDriverState(std::move(dbDriverState))
-    , PrevToken(DbDriverState->GetCredentialsProvider() ? DbDriverState->GetCredentialsProvider()->GetAuthInfo() : "")
     , InitSeqNoPromise(NThreading::NewPromise<ui64>())
     , WakeupInterval(
             Settings.BatchFlushInterval_ != TDuration::Zero() ?
@@ -1184,19 +1183,62 @@ void TWriteSessionImpl::UpdateTokenIfNeededImpl() {
     LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session: try to update token");
 
     auto credentialsProvider = DbDriverState->GetCredentialsProvider();
-    if (!credentialsProvider || UpdateTokenInProgress || !SessionEstablished)
+    if (!credentialsProvider || UpdateTokenInProgress || !SessionEstablished || Aborting)
         return;
-    TClientMessage clientMessage;
-    auto* updateRequest = clientMessage.mutable_update_token_request();
-    auto token = credentialsProvider->GetAuthInfo();
-    if (token == PrevToken)
+    auto authInfo = credentialsProvider->GetAuthInfoAsync();
+    if (authInfo.IsReady()) {
+        UpdateTokenImpl(authInfo);
         return;
+    }
+    UpdateTokenInProgress = true;
+    try {
+        authInfo.Subscribe([cbContext = SelfContext](const auto& future) {
+            if (auto self = cbContext->LockShared()) try {
+                self->Connections->ScheduleCallback(TDuration::Zero(), [cbContext, future](bool ok) {
+                    if (auto self = cbContext->LockShared()) {
+                        if (!ok) {
+                            self->UpdateTokenInProgress = false;
+                            return;
+                        }
+                        std::lock_guard guard(self->Lock);
+                        self->UpdateTokenImpl(future);
+                    }
+                });
+            } catch (...) {
+                self->UpdateTokenInProgress = false;
+            }
+        });
+    } catch (...) {
+        UpdateTokenInProgress = false;
+    }
+}
+
+void TWriteSessionImpl::UpdateTokenImpl(const NThreading::TFuture<std::string>& future) {
+    Y_ABORT_UNLESS(Lock.IsLocked());
+
+    UpdateTokenInProgress = false;
+    if (!SessionEstablished || Aborting) {
+        return;
+    }
+
+    std::string token;
+    try {
+        token = future.GetValue();
+    } catch (...) {
+        CloseImpl(EStatus::CLIENT_UNAUTHENTICATED, CurrentExceptionMessage());
+        return;
+    }
+    if (token == PrevToken) {
+        return;
+    }
+
     UpdateTokenInProgress = true;
-    updateRequest->set_token(TStringType{token});
     PrevToken = token;
 
     LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefix() << "Write session: updating token");
 
+    TClientMessage clientMessage;
+    clientMessage.mutable_update_token_request()->set_token(TStringType{token});
     Processor->Write(std::move(clientMessage));
 }
 
diff --git a/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.h b/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.h
index 947095722af..1a56d5dde44 100644
--- a/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.h
+++ b/ydb/public/sdk/cpp/src/client/persqueue_public/impl/write_session_impl.h
@@ -325,6 +325,7 @@ private:
     TStringBuilder LogPrefix() const;
 
     void UpdateTokenIfNeededImpl();
+    void UpdateTokenImpl(const NThreading::TFuture<std::string>& future);
 
     void WriteInternal(TContinuationToken&& continuationToken, std::string_view data, std::optional<ECodec> codec, ui32 originalSize,
                std::optional<ui64> seqNo = std::nullopt, std::optional<TInstant> createTimestamp = std::nullopt);
@@ -387,7 +388,7 @@ private:
     std::shared_ptr<IWriteSessionConnectionProcessorFactory> ConnectionFactory;
     TDbDriverStatePtr DbDriverState;
     std::string PrevToken;
-    bool UpdateTokenInProgress = false;
+    std::atomic_bool UpdateTokenInProgress = false;
     TInstant LastTokenUpdate = TInstant::Zero();
     std::shared_ptr<TWriteSessionEventsQueue> EventsQueue;
     NYdbGrpc::IQueueClientContextPtr ClientContext; // Common client context.
diff --git a/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.cpp b/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.cpp
index df6eb1da819..eea32b545e7 100644
--- a/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.cpp
+++ b/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.cpp
@@ -114,7 +114,6 @@ TWriteSessionImpl::TWriteSessionImpl(
     , Client(std::move(client))
     , Connections(std::move(connections))
     , DbDriverState(std::move(dbDriverState))
-    , PrevToken(DbDriverState->GetCredentialsProvider() ? DbDriverState->GetCredentialsProvider()->GetAuthInfo() : "")
     , MaxBlockMessageCount(Settings.BatchFlushMessageCount_)
     , InitSeqNoPromise(NThreading::NewPromise<uint64_t>())
     , WakeupInterval(
@@ -1587,11 +1586,53 @@ void TWriteSessionImpl::UpdateTokenIfNeededImpl() {
     LOG_LAZY(DbDriverState->Log, TLOG_DEBUG, LogPrefixImpl() << "Write session: try to update token");
 
     auto credentialsProvider = DbDriverState->GetCredentialsProvider();
-    if (!credentialsProvider || UpdateTokenInProgress || !SessionEstablished) {
+    if (!credentialsProvider || UpdateTokenInProgress || !SessionEstablished || Aborting) {
         return;
     }
 
-    auto token = credentialsProvider->GetAuthInfo();
+    auto authInfo = credentialsProvider->GetAuthInfoAsync();
+    if (authInfo.IsReady()) {
+        UpdateTokenImpl(authInfo);
+        return;
+    }
+    UpdateTokenInProgress = true;
+    try {
+        authInfo.Subscribe([cbContext = SelfContext](const auto& future) {
+            if (auto self = cbContext->LockShared()) try {
+                self->Connections->ScheduleCallback(TDuration::Zero(), [cbContext, future](bool ok) {
+                    if (auto self = cbContext->LockShared()) {
+                        if (!ok) {
+                            self->UpdateTokenInProgress = false;
+                            return;
+                        }
+                        std::lock_guard guard(self->Lock);
+                        self->UpdateTokenImpl(future);
+                    }
+                });
+            } catch (...) {
+                self->UpdateTokenInProgress = false;
+            }
+        });
+    } catch (...) {
+        UpdateTokenInProgress = false;
+    }
+}
+
+void TWriteSessionImpl::UpdateTokenImpl(const NThreading::TFuture<std::string>& future) {
+    Y_ABORT_UNLESS(Lock.IsLocked());
+
+    UpdateTokenInProgress = false;
+    if (!SessionEstablished || Aborting) {
+        return;
+    }
+
+    std::string token;
+    try {
+        token = future.GetValue();
+    } catch (...) {
+        CloseImpl(EStatus::CLIENT_UNAUTHENTICATED, CurrentExceptionMessage());
+        return;
+    }
     if (token == PrevToken) {
         return;
     }
diff --git a/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.h b/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.h
index 5f7ec7d92b5..a711a8bf376 100644
--- a/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.h
+++ b/ydb/public/sdk/cpp/src/client/topic/impl/write_session_impl.h
@@ -377,6 +377,7 @@ private:
     TStringBuilder LogPrefixImpl() const;
 
     void UpdateTokenIfNeededImpl();
+    void UpdateTokenImpl(const NThreading::TFuture<std::string>& future);
 
     void WriteInternal(TContinuationToken&& continuationToken, TWriteMessage&& message);
 
@@ -447,7 +448,7 @@ private:
     std::shared_ptr<IWriteSessionConnectionProcessorFactory> ConnectionFactory;
     TDbDriverStatePtr DbDriverState;
     std::string PrevToken;
-    bool UpdateTokenInProgress = false;
+    std::atomic_bool UpdateTokenInProgress = false;
     TInstant LastTokenUpdate = TInstant::Zero();
     std::shared_ptr<TWriteSessionEventsQueue> EventsQueue;
     NYdbGrpc::IQueueClientContextPtr ClientContext; // Common client context.
diff --git a/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.cpp b/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.cpp
index 2047ab4d292..db3c5ace047 100644
--- a/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.cpp
+++ b/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.cpp
@@ -2,7 +2,6 @@
 
 #include <util/generic/yexception.h>
 #include <util/stream/output.h>
-#include <util/system/yassert.h>
 
 namespace NYdb::inline Dev {
 
@@ -30,9 +29,6 @@ TSimpleCoreFacility::~TSimpleCoreFacility() {
 
 void TSimpleCoreFacility::AddPeriodicTask(TPeriodicCb&& cb, TDeadline::Duration period) {
     std::lock_guard lock(Mutex_);
-    Y_ABORT_UNLESS(!PeriodicStarted_);
-    PeriodicStarted_ = true;
-
     auto periodicCb = std::make_shared<TPeriodicCb>(std::move(cb));
     EnqueueTaskNoLock(
         TClock::now(),
@@ -46,15 +42,11 @@ void TSimpleCoreFacility::PostToResponseQueue(TPostTaskCb&& f) {
     if (!f) {
         return;
     }
-    {
-        std::lock_guard lock(Mutex_);
-        if (!Stop_) {
-            EnqueueTaskNoLock(TClock::now(), std::move(f));
-            Cv_.notify_one();
-            return;
-        }
+    std::lock_guard lock(Mutex_);
+    if (!Stop_) {
+        EnqueueTaskNoLock(TClock::now(), std::move(f));
+        Cv_.notify_one();
     }
-    f();
 }
 
 void TSimpleCoreFacility::EnqueueTaskNoLock(TTimePoint executeAt, TPostTaskCb&& task) {
diff --git a/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.h b/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.h
index 7e7fd9ff28f..fe76229a271 100644
--- a/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.h
+++ b/ydb/public/sdk/cpp/src/client/types/core_facility/simple_core_facility.h
@@ -51,7 +51,6 @@ private:
     std::condition_variable Cv_;
     std::priority_queue<TScheduledTask, std::vector<TScheduledTask>, TScheduledTaskLess> Queue_;
     bool Stop_ = false;
-    bool PeriodicStarted_ = false;
     std::uint64_t NextSeqNo_ = 0;
 
     std::thread WorkerThread_;
diff --git a/ydb/public/sdk/cpp/src/client/types/credentials/login/login.cpp b/ydb/public/sdk/cpp/src/client/types/credentials/login/login.cpp
index 221bd92d65c..0a2bf8b010a 100644
--- a/ydb/public/sdk/cpp/src/client/types/credentials/login/login.cpp
+++ b/ydb/public/sdk/cpp/src/client/types/credentials/login/login.cpp
@@ -37,75 +37,99 @@ std::chrono::system_clock::time_point GetTokenExpiresAt(const std::string& token
     }
     return {};
 }
+
+bool IsRetryable(EStatus status) {
+    return status == EStatus::INTERNAL_ERROR || status == EStatus::ABORTED || status == EStatus::UNAVAILABLE ||
+        status == EStatus::OVERLOADED || status == EStatus::GENERIC_ERROR || status == EStatus::TIMEOUT ||
+        status == EStatus::CANCELLED || status == EStatus::UNDETERMINED || status == EStatus::SESSION_BUSY ||
+        status == EStatus::TRANSPORT_UNAVAILABLE || status == EStatus::CLIENT_RESOURCE_EXHAUSTED ||
+        status == EStatus::CLIENT_DEADLINE_EXCEEDED || status == EStatus::CLIENT_INTERNAL_ERROR ||
+        status == EStatus::CLIENT_CANCELLED || status == EStatus::CLIENT_DISCOVERY_FAILED ||
+        status == EStatus::CLIENT_LIMITS_REACHED;
 }
 
-class TLoginCredentialsProvider : public ICredentialsProvider {
+bool IsRetryableOperation(EStatus status) {
+    return status == EStatus::ABORTED || status == EStatus::UNAVAILABLE || status == EStatus::OVERLOADED ||
+        status == EStatus::TIMEOUT || status == EStatus::SESSION_BUSY;
+}
+}
+
+class TLoginCredentialsProvider : public ICredentialsProvider, public std::enable_shared_from_this<TLoginCredentialsProvider> {
 public:
     TLoginCredentialsProvider(std::weak_ptr<ICoreFacility> facility, TLoginCredentialsParams params);
-    virtual std::string GetAuthInfo() const override;
-    virtual bool IsValid() const override;
-    NThreading::TFuture<void> PrepareTokenAsync();
+    std::string GetAuthInfo() const override;
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override;
+    bool IsValid() const override;
+    void Start();
+    ~TLoginCredentialsProvider() { Fail("Login credentials provider stopped"); }
 
 private:
-    void PrepareToken();
+    bool OnPeriodicTick(EStatus status);
     void RequestToken();
     void FinishRequest(Ydb::Auth::LoginResponse* response, TPlainStatus status, bool facilityAvailable);
-    bool IsOk() const;
-    void ParseToken();
-    std::string GetToken() const;
-    std::string GetError() const;
-    std::string GetTokenOrError() const;
-
-    enum class EState {
-        Empty,
-        Requesting,
-        Done,
-    };
+    void Fail(std::string error);
+    static std::string GetError(const TPlainStatus& status, const Ydb::Auth::LoginResponse& response);
 
     std::weak_ptr<ICoreFacility> Facility_;
     TLoginCredentialsParams Params_;
-    EState State_ = EState::Empty;
-    std::mutex Mutex_;
-    std::atomic<ui64> TokenReceived_ = 1;
-    std::atomic<ui64> TokenParsed_ = 0;
-    std::optional<std::string> Token_;
-    std::optional<std::string> Error_;
-    TInstant TokenExpireAt_;
+    mutable std::mutex Mutex_;
     TInstant TokenRequestAt_;
-    TPlainStatus Status_;
-    Ydb::Auth::LoginResponse Response_;
-    NThreading::TPromise<void> TokenReadyPromise_;
+    bool Requesting_ = false;
+    bool HasToken_ = false;
+    bool Stopped_ = false;
+    mutable NThreading::TPromise<std::string> AuthInfo_;
 };
 
 TLoginCredentialsProvider::TLoginCredentialsProvider(std::weak_ptr<ICoreFacility> facility, TLoginCredentialsParams params)
     : Facility_(facility)
     , Params_(std::move(params))
-    , TokenReadyPromise_(NThreading::NewPromise<void>())
-{
-    auto strongFacility = facility.lock();
-    if (strongFacility) {
-        auto periodicTask = [facility, this](NYdb::NIssue::TIssues&&, EStatus status) -> bool {
-            if (status != EStatus::SUCCESS) {
-                return false;
-            }
-
-            auto strongFacility = facility.lock();
-            if (!strongFacility) {
-                return false;
-            }
-
-            if (!TokenRequestAt_) {
-                return true;
-            }
+    , AuthInfo_(NThreading::NewPromise<std::string>())
+{}
 
-            if (TInstant::Now() >= TokenRequestAt_) {
-                RequestToken();
+void TLoginCredentialsProvider::Start() {
+    auto facility = Facility_.lock();
+    if (!facility) {
+        Fail("Login credentials provider response facility is not available");
+        return;
+    }
+    {
+        std::lock_guard lock(Mutex_);
+        Requesting_ = true;
+    }
+    try {
+        facility->AddPeriodicTask([weak = weak_from_this()](NYdb::NIssue::TIssues&&, EStatus status) {
+            if (auto self = weak.lock()) {
+                return self->OnPeriodicTick(status);
             }
+            return false;
+        }, 1s);
+    } catch (...) {
+        Fail(CurrentExceptionMessage());
+        return;
+    }
+    RequestToken();
+}
 
+bool TLoginCredentialsProvider::OnPeriodicTick(EStatus status) {
+    if (status != EStatus::SUCCESS) {
+        Fail("Login credentials provider periodic task failed");
+        return false;
+    }
+    {
+        std::lock_guard lock(Mutex_);
+        if (Stopped_) {
+            return false;
+        }
+        if (Requesting_ || TInstant::Now() < TokenRequestAt_) {
             return true;
-        };
-        strongFacility->AddPeriodicTask(std::move(periodicTask), 1min);
+        }
+        if (AuthInfo_.GetFuture().IsReady()) {
+            AuthInfo_ = NThreading::NewPromise<std::string>();
+        }
+        Requesting_ = true;
     }
+    RequestToken();
+    return true;
 }
 
 bool TLoginCredentialsProvider::IsValid() const {
@@ -113,22 +137,21 @@ bool TLoginCredentialsProvider::IsValid() const {
 }
 
 std::string TLoginCredentialsProvider::GetAuthInfo() const {
-    if (TokenParsed_ == TokenReceived_) {
-        return GetTokenOrError();
-    } else {
-        const_cast<TLoginCredentialsProvider*>(this)->PrepareToken(); // will block here
-        return GetTokenOrError();
-    }
+    return GetAuthInfoAsync().GetValueSync();
+}
+
+NThreading::TFuture<std::string> TLoginCredentialsProvider::GetAuthInfoAsync() const {
+    std::lock_guard lock(Mutex_);
+    return AuthInfo_.GetFuture();
 }
 
 void TLoginCredentialsProvider::RequestToken() {
     auto strongFacility = Facility_.lock();
     if (strongFacility) {
-        TokenRequestAt_ = {};
-
-        auto responseCb = [facility = Facility_, this](Ydb::Auth::LoginResponse* resp, TPlainStatus status) {
-            auto strongFacility = facility.lock();
-            FinishRequest(resp, std::move(status), static_cast<bool>(strongFacility));
+        auto responseCb = [facility = Facility_, weak = weak_from_this()](Ydb::Auth::LoginResponse* resp, TPlainStatus status) {
+            if (auto self = weak.lock()) {
+                self->FinishRequest(resp, std::move(status), !facility.expired());
+            }
         };
 
         Ydb::Auth::LoginRequest request;
@@ -137,9 +160,13 @@ void TLoginCredentialsProvider::RequestToken() {
         TRpcRequestSettings rpcSettings;
         rpcSettings.Deadline = TDeadline::AfterDuration(60s);
 
-        TGRpcConnectionsImpl::RunOnDiscoveryEndpoint<Ydb::Auth::V1::AuthService, Ydb::Auth::LoginRequest, Ydb::Auth::LoginResponse>(
-            strongFacility, std::move(request), std::move(responseCb), &Ydb::Auth::V1::AuthService::Stub::AsyncLogin,
-            rpcSettings);
+        try {
+            TGRpcConnectionsImpl::RunOnDiscoveryEndpoint<Ydb::Auth::V1::AuthService, Ydb::Auth::LoginRequest, Ydb::Auth::LoginResponse>(
+                strongFacility, std::move(request), std::move(responseCb), &Ydb::Auth::V1::AuthService::Stub::AsyncLogin,
+                rpcSettings);
+        } catch (...) {
+            Fail(CurrentExceptionMessage());
+        }
     } else {
         FinishRequest(nullptr, {}, false);
     }
@@ -150,107 +177,78 @@ void TLoginCredentialsProvider::FinishRequest(
     TPlainStatus status,
     bool facilityAvailable)
 {
-    std::optional<std::string> error;
-    {
-        std::lock_guard lock(Mutex_);
-        State_ = EState::Done;
-        ++TokenReceived_;
-        if (facilityAvailable) {
-            Status_ = std::move(status);
-            if (response) {
-                Response_ = std::move(*response);
-            }
-            ParseToken();
-        } else {
-            Token_.reset();
-            Error_ = "Login credentials provider response facility is not available";
-            TokenParsed_ = TokenReceived_.load();
-        }
-        error = Error_;
+    if (!facilityAvailable) {
+        Fail("Login credentials provider response facility is not available");
+        return;
     }
-    if (error) {
-        TokenReadyPromise_.TrySetException(std::make_exception_ptr(yexception() << *error));
-    } else {
-        TokenReadyPromise_.TrySetValue();
-    }
-}
 
-void TLoginCredentialsProvider::PrepareToken() {
-    PrepareTokenAsync().Wait();
-    std::lock_guard lock(Mutex_);
-    ParseToken();
-}
-
-NThreading::TFuture<void> TLoginCredentialsProvider::PrepareTokenAsync() {
-    bool requestToken = false;
-    auto future = TokenReadyPromise_.GetFuture();
-    {
-        std::unique_lock<std::mutex> lock(Mutex_);
-        if (State_ == EState::Empty) {
-            State_ = EState::Requesting;
-            requestToken = true;
+    Ydb::Auth::LoginResponse emptyResponse;
+    const auto& responseValue = response ? *response : emptyResponse;
+    const auto operationStatus = static_cast<EStatus>(responseValue.operation().status());
+    if (!status.Ok() || operationStatus != EStatus::SUCCESS) {
+        bool retry;
+        {
+            std::lock_guard lock(Mutex_);
+            retry = (!status.Ok() && IsRetryable(status.Status)) ||
+                (HasToken_ && status.Ok() && IsRetryableOperation(operationStatus));
+            if (retry) {
+                Requesting_ = false;
+                TokenRequestAt_ = TInstant::Now() + TDuration::Seconds(1);
+            }
         }
-    }
-    if (requestToken) {
-        RequestToken();
-    }
-    return future;
-}
-
-bool TLoginCredentialsProvider::IsOk() const {
-    return State_ == EState::Done
-        && Status_.Ok()
-        && Response_.operation().status() == Ydb::StatusIds::SUCCESS;
-}
-
-void TLoginCredentialsProvider::ParseToken() { // works under mutex
-    if (TokenParsed_ != TokenReceived_) {
-        if (IsOk()) {
-            Token_ = GetToken();
-            Error_.reset();
-            TInstant now = TInstant::Now();
-            TokenExpireAt_ = ToInstant(GetTokenExpiresAt(Token_.value()));
-            TokenRequestAt_ = now + TDuration::Minutes((TokenExpireAt_ - now).Minutes() / 2);
-        } else {
-            Token_.reset();
-            Error_ = GetError();
+        if (retry) {
+            return;
         }
-        TokenParsed_ = TokenReceived_.load();
+        Fail(GetError(status, responseValue));
+        return;
     }
-}
 
-std::string TLoginCredentialsProvider::GetToken() const {
     Ydb::Auth::LoginResult result;
-    Response_.operation().result().UnpackTo(&result);
-    return result.token();
+    if (!responseValue.operation().result().UnpackTo(&result) || result.token().empty()) {
+        Fail("Login service returned an empty token");
+        return;
+    }
+    auto token = result.token();
+    const auto now = TInstant::Now();
+    NThreading::TPromise<std::string> promise;
+    {
+        std::lock_guard lock(Mutex_);
+        Requesting_ = false;
+        HasToken_ = true;
+        TokenRequestAt_ = now + (ToInstant(GetTokenExpiresAt(token)) - now) / 2;
+        promise = AuthInfo_;
+    }
+    promise.TrySetValue(std::move(token));
 }
 
-std::string TLoginCredentialsProvider::GetError() const {
-    if (Status_.Ok()) {
-        if (Response_.operation().issues_size() > 0) {
-            return Response_.operation().issues(0).message();
-        } else {
-            return Ydb::StatusIds_StatusCode_Name(Response_.operation().status());
-        }
-    } else {
-        TStringBuilder str;
-        str << "Couldn't get token for provided credentials from " << Status_.Endpoint
-            << " with status " << Status_.Status << ".";
-        for (const auto& issue : Status_.Issues) {
-            str << Endl << "Issue: " << issue;
-        }
-        return str;
+void TLoginCredentialsProvider::Fail(std::string error) {
+    NThreading::TPromise<std::string> promise;
+    {
+        std::lock_guard lock(Mutex_);
+        Stopped_ = true;
+        Requesting_ = false;
+        promise = AuthInfo_;
     }
+    promise.TrySetException(std::make_exception_ptr(yexception() << error));
 }
 
-std::string TLoginCredentialsProvider::GetTokenOrError() const {
-    if (Token_) {
-        return Token_.value();
+std::string TLoginCredentialsProvider::GetError(
+    const TPlainStatus& status,
+    const Ydb::Auth::LoginResponse& response)
+{
+    if (status.Ok()) {
+        if (response.operation().issues_size() > 0) {
+            return response.operation().issues(0).message();
+        }
+        return Ydb::StatusIds_StatusCode_Name(response.operation().status());
     }
-    if (Error_) {
-        ythrow yexception() << Error_.value();
+    TStringBuilder str;
+    str << "Couldn't get token for provided credentials from " << status.Endpoint
+        << " with status " << status.Status << ".";
+    for (const auto& issue : status.Issues) {
+            str << Endl << "Issue: " << issue;
     }
-    ythrow yexception() << "Wrong state of credentials provider";
+    return str;
 }
 
 class TLoginCredentialsProviderFactory : public ICredentialsProviderFactory {
@@ -258,7 +256,6 @@ public:
     TLoginCredentialsProviderFactory(TLoginCredentialsParams params);
     virtual std::shared_ptr<ICredentialsProvider> CreateProvider() const override;
     virtual std::shared_ptr<ICredentialsProvider> CreateProvider(std::weak_ptr<ICoreFacility> facility) const override;
-    virtual NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const override;
 
 private:
     TLoginCredentialsParams Params_;
@@ -274,13 +271,9 @@ std::shared_ptr<ICredentialsProvider> TLoginCredentialsProviderFactory::CreatePr
 }
 
 std::shared_ptr<ICredentialsProvider> TLoginCredentialsProviderFactory::CreateProvider(std::weak_ptr<ICoreFacility> facility) const {
-    return std::make_shared<TLoginCredentialsProvider>(std::move(facility), Params_);
-}
-
-NThreading::TFuture<TCredentialsProviderPtr> TLoginCredentialsProviderFactory::CreateProviderAsync(std::weak_ptr<ICoreFacility> facility) const {
     auto provider = std::make_shared<TLoginCredentialsProvider>(std::move(facility), Params_);
-    TCredentialsProviderPtr result = provider;
-    return provider->PrepareTokenAsync().Return(std::move(result));
+    provider->Start();
+    return provider;
 }
 
 std::shared_ptr<ICredentialsProviderFactory> CreateLoginCredentialsProviderFactory(TLoginCredentialsParams params) {
diff --git a/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/credentials.cpp b/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/credentials.cpp
index df877f5dce0..367cfbc92dc 100644
--- a/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/credentials.cpp
+++ b/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/credentials.cpp
@@ -1,4 +1,5 @@
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/oauth2_token_exchange/credentials.h>
+#include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h>
 
 #include <library/cpp/cgiparam/cgiparam.h>
 #include <library/cpp/http/misc/httpcodes.h>
@@ -85,18 +86,11 @@ bool IsRetryableError(TKeepAliveHttpClient::THttpCode code) {
         || code == HTTP_GATEWAY_TIME_OUT;
 }
 
-ERetryErrorClass RetryPolicyClass(TKeepAliveHttpClient::THttpCode code) {
-    return IsRetryableError(code) ? ERetryErrorClass::LongRetry : ERetryErrorClass::ShortRetry; // In case when we already have token we know that all params are correct and all errors are temporary
-}
-
-ERetryErrorClass SyncRetryPolicyClass(const std::exception* ex, bool retryAllErrors) {
+ERetryErrorClass RetryPolicyClass(const std::exception* ex) {
     if (const TTokenExchangeError* err = dynamic_cast<const TTokenExchangeError*>(ex)) {
         if (IsRetryableError(err->HttpCode)) {
             return ERetryErrorClass::LongRetry;
         }
-        if (retryAllErrors) {
-            return ERetryErrorClass::ShortRetry;
-        }
     }
     if (dynamic_cast<const TSystemError*>(ex)) {
         return ERetryErrorClass::ShortRetry;
@@ -106,8 +100,7 @@ ERetryErrorClass SyncRetryPolicyClass(const std::exception* ex, bool retryAllErr
     return ERetryErrorClass::NoRetry;
 }
 
-using TRetryPolicy = IRetryPolicy<TKeepAliveHttpClient::THttpCode>;
-using TSyncRetryPolicy = IRetryPolicy<const std::exception*, bool>;
+using TRetryPolicy = IRetryPolicy<const std::exception*>;
 
 struct TPrivateOauth2TokenExchangeParams: public TOauth2TokenExchangeParams {
     TPrivateOauth2TokenExchangeParams(const TOauth2TokenExchangeParams& params)
@@ -172,48 +165,65 @@ private:
     }
 };
 
-class TOauth2TokenExchangeProviderImpl: public std::enable_shared_from_this<TOauth2TokenExchangeProviderImpl>
-{
+class TOauth2TokenExchangeProviderImpl final : public ICredentialsProvider {
     struct TTokenExchangeResult {
         std::string Token;
-        TInstant TokenDeadline;
         TInstant TokenRefreshTime;
     };
 
 public:
-    explicit TOauth2TokenExchangeProviderImpl(const TPrivateOauth2TokenExchangeParams& params)
+    TOauth2TokenExchangeProviderImpl(const TPrivateOauth2TokenExchangeParams& params,
+                                     std::weak_ptr<ICoreFacility> responseFacility)
         : Params(params)
+        , ResponseFacility(std::move(responseFacility))
+        , AuthInfo(NThreading::NewPromise<std::string>())
     {
-        ExchangeTokenSync(TInstant::Now());
+        ResponseFacility.expired() ? Stop() : Start();
+    }
+
+private:
+    void Start() {
+        try {
+            WorkerThread = std::thread([this] { Run(); });
+        } catch (...) {
+            Fail(std::current_exception());
+        }
     }
 
     void Stop() {
+        NThreading::TPromise<std::string> promise;
         {
             std::unique_lock<std::mutex> lock(StopMutex);
             Stopping = true;
             StopVar.notify_all();
         }
+        with_lock (Lock) {
+            promise = AuthInfo;
+        }
+        promise.TrySetException(std::make_exception_ptr(yexception() << PROV_ERR "stopped"));
 
-        if (RefreshTokenThread.joinable()) {
-            RefreshTokenThread.join();
+        if (WorkerThread.joinable()) {
+            WorkerThread.join();
         }
     }
 
-    std::string GetAuthInfo() const {
-        const TInstant now = TInstant::Now();
-        std::string token;
+public:
+    ~TOauth2TokenExchangeProviderImpl() {
+        Stop();
+    }
+
+    std::string GetAuthInfo() const override {
+        return GetAuthInfoAsync().GetValueSync();
+    }
+
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
         with_lock (Lock) {
-            if (Token.empty() || now >= TokenDeadline) { // Update sync. This can be if we have repeating error during token refresh process. In this case we will try for the last time and throw an error
-                ExchangeTokenSync(now, true);
-                token = Token;
-            } else {
-                if (now >= TokenRefreshTime) {
-                    TryRefreshToken();
-                }
-                token = Token; // Still valid
-            }
+            return AuthInfo.GetFuture();
         }
-        return token;
+    }
+
+    bool IsValid() const override {
+        return true;
     }
 
 private:
@@ -344,7 +354,6 @@ private:
 
         const TDuration expireDelta = TDuration::Seconds(expireTime);
         TTokenExchangeResult result;
-        result.TokenDeadline = now + expireDelta;
         result.TokenRefreshTime = now + expireDelta / 2;
         result.Token = TStringBuilder() << "Bearer " << token;
         return result;
@@ -361,31 +370,63 @@ private:
         return ProcessExchangeTokenResponse(now, statusCode, responseStream);
     }
 
-    void ExchangeTokenSync(TInstant now, bool retryAllErrors = false) const { // Is run under lock
-        TSyncRetryPolicy::IRetryState::TPtr retryState;
+    void Run() {
         while (true) {
-            try {
-                TTokenExchangeResult result = ExchangeToken(now);
-                Token = result.Token;
-                TokenDeadline = result.TokenDeadline;
-                TokenRefreshTime = result.TokenRefreshTime;
-                break;
-            } catch (const std::exception& ex) {
-                if (!retryState) {
-                    retryState = TSyncRetryPolicy::GetFixedIntervalPolicy( // retry more aggresively when we are in sync mode
-                        SyncRetryPolicyClass,
-                        TDuration::MilliSeconds(100), // delay // default
-                        TDuration::MilliSeconds(300), // long delay // default
-                        std::numeric_limits<size_t>::max(), // max retries // default
-                        Params.SyncUpdateTimeout_ // max time
-                    )->CreateRetryState();
+            TRetryPolicy::IRetryState::TPtr retryState;
+            TTokenExchangeResult result;
+            while (true) {
+                if (IsStopping()) {
+                    return;
                 }
-                if (auto interval = retryState->GetNextRetryDelay(&ex, retryAllErrors)) {
-                    Sleep(*interval);
-                } else {
-                    throw;
+                try {
+                    result = ExchangeToken(TInstant::Now());
+                    break;
+                } catch (const std::exception& ex) {
+                    if (!retryState) {
+                        retryState = TRetryPolicy::GetExponentialBackoffPolicy(
+                            RetryPolicyClass,
+                            TDuration::MilliSeconds(10),
+                            TDuration::MilliSeconds(200),
+                            TDuration::Seconds(30))->CreateRetryState();
+                    }
+                    auto delay = retryState->GetNextRetryDelay(&ex);
+                    if (!delay) {
+                        Fail(std::current_exception());
+                        return;
+                    }
+                    std::unique_lock<std::mutex> lock(StopMutex);
+                    if (StopVar.wait_for(lock, TChronoDuration(delay->GetValue()), [this] { return Stopping; })) {
+                        return;
+                    }
+                } catch (...) {
+                    Fail(std::current_exception());
+                    return;
                 }
             }
+
+            NThreading::TPromise<std::string> promise;
+            with_lock (Lock) {
+                promise = AuthInfo;
+            }
+            Complete([promise, token = std::move(result.Token)]() mutable {
+                promise.TrySetValue(std::move(token));
+            });
+
+            {
+                std::unique_lock<std::mutex> lock(StopMutex);
+                const auto delay = result.TokenRefreshTime - TInstant::Now();
+                if (delay > TDuration::Zero() &&
+                    StopVar.wait_for(lock, TChronoDuration(delay.GetValue()), [this] { return Stopping; }))
+                {
+                    return;
+                }
+                if (Stopping) {
+                    return;
+                }
+            }
+            with_lock (Lock) {
+                AuthInfo = NThreading::NewPromise<std::string>();
+            }
         }
     }
 
@@ -394,85 +435,32 @@ private:
         return Stopping;
     }
 
-    void RefreshToken() const {
-        TRetryPolicy::IRetryState::TPtr retryState;
-        TInstant deadline;
+    void Fail(std::exception_ptr error) const {
+        NThreading::TPromise<std::string> promise;
         with_lock (Lock) {
-            deadline = TokenDeadline;
+            promise = AuthInfo;
         }
-        while (!IsStopping()) {
-            const TInstant now = TInstant::Now();
-            try {
-                TTokenExchangeResult result = ExchangeToken(now);
-                with_lock (Lock) {
-                    Token = result.Token;
-                    TokenDeadline = result.TokenDeadline;
-                    TokenRefreshTime = result.TokenRefreshTime;
-                    break;
-                }
-            } catch (const std::exception& ex) { // If this error will repeat, we finally will get it syncronously in GetAuthInfo() and pass to client
-                if (!retryState) {
-                    retryState = TRetryPolicy::GetExponentialBackoffPolicy(
-                        RetryPolicyClass,
-                        TDuration::MilliSeconds(10), // min delay // default
-                        TDuration::MilliSeconds(200), // min long delay // default
-                        TDuration::Seconds(30), // max delay // default
-                        std::numeric_limits<size_t>::max(), // max retries // default
-                        deadline - TInstant::Now() // max time
-                    )->CreateRetryState();
-                } else {
-                    TKeepAliveHttpClient::THttpCode code = HTTP_CODE_MAX;
-                    if (const auto* err = dynamic_cast<const TTokenExchangeError*>(&ex)) {
-                        code = err->HttpCode;
-                    }
-                    if (auto delay = retryState->GetNextRetryDelay(code)) {
-                        std::unique_lock<std::mutex> lock(StopMutex);
-                        const bool stopping = StopVar.wait_for(
-                            lock,
-                            TChronoDuration(delay->GetValue()),
-                            [this]() {
-                                return Stopping;
-                            }
-                        );
-                        if (stopping) {
-                            break;
-                        }
-                    } else {
-                        break;
-                    }
-                }
-            }
-        }
-        TokenIsRefreshing.store(false);
+        Complete([promise, error = std::move(error)]() mutable {
+            promise.TrySetException(std::move(error));
+        });
     }
 
-    void TryRefreshToken() const { // Is run under lock
-        if (TokenIsRefreshing.load()) {
-            return;
-        }
-        if (RefreshTokenThread.joinable()) {
-            RefreshTokenThread.join();
-        }
-
-        TokenIsRefreshing.store(true);
-        RefreshTokenThread = std::thread(
-            [w = weak_from_this()]() {
-                if (auto p = w.lock()) {
-                    p->RefreshToken();
-                }
+    void Complete(TPostTaskCb&& callback) const noexcept {
+        try {
+            if (auto facility = ResponseFacility.lock()) {
+                facility->PostToResponseQueue(std::move(callback));
             }
-        );
+        } catch (...) {
+        }
     }
 
 private:
     TPrivateOauth2TokenExchangeParams Params;
+    std::weak_ptr<ICoreFacility> ResponseFacility;
 
-    TAdaptiveLock Lock;
-    mutable std::atomic<bool> TokenIsRefreshing = false;
-    mutable std::thread RefreshTokenThread;
-    mutable std::string Token;
-    mutable TInstant TokenDeadline;
-    mutable TInstant TokenRefreshTime;
+    mutable TAdaptiveLock Lock;
+    mutable NThreading::TPromise<std::string> AuthInfo;
+    std::thread WorkerThread;
 
     // Stop
     bool Stopping = false;
@@ -480,42 +468,31 @@ private:
     mutable std::condition_variable StopVar;
 };
 
-class TOauth2TokenExchangeProvider: public ICredentialsProvider {
-public:
-    explicit TOauth2TokenExchangeProvider(const TPrivateOauth2TokenExchangeParams& params)
-        : Impl(std::make_shared<TOauth2TokenExchangeProviderImpl>(params))
-    {
-    }
-
-    std::string GetAuthInfo() const override {
-        return Impl->GetAuthInfo();
-    }
-
-    bool IsValid() const override {
-        return true;
-    }
-
-    ~TOauth2TokenExchangeProvider() { // The last link tp provider is gone
-        Impl->Stop();
-    }
-
-private:
-    std::shared_ptr<TOauth2TokenExchangeProviderImpl> Impl;
-};
-
 class TOauth2TokenExchangeFactory: public ICredentialsProviderFactory {
 public:
     explicit TOauth2TokenExchangeFactory(const TOauth2TokenExchangeParams& params)
-        : Provider(std::make_shared<TOauth2TokenExchangeProvider>(params))
+        : Params(params)
     {
     }
 
     TCredentialsProviderPtr CreateProvider() const override {
+        std::lock_guard lock(Lock);
+        if (!Provider) {
+            auto facility = CreateSimpleCoreFacility();
+            Provider = std::make_shared<NCredentials::NDetail::TOwningFacilityCredentialsProvider>(
+                facility, std::make_shared<TOauth2TokenExchangeProviderImpl>(Params, facility));
+        }
         return Provider;
     }
 
+    TCredentialsProviderPtr CreateProvider(std::weak_ptr<ICoreFacility> facility) const override {
+        return std::make_shared<TOauth2TokenExchangeProviderImpl>(Params, std::move(facility));
+    }
+
 private:
-    std::shared_ptr<TOauth2TokenExchangeProvider> Provider;
+    TPrivateOauth2TokenExchangeParams Params;
+    mutable std::mutex Lock;
+    mutable TCredentialsProviderPtr Provider;
 };
 
 } // namespace
diff --git a/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/ya.make b/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/ya.make
index 9962d6bb675..a318032d249 100644
--- a/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/ya.make
+++ b/ydb/public/sdk/cpp/src/client/types/credentials/oauth2_token_exchange/ya.make
@@ -16,6 +16,7 @@ PEERDIR(
     library/cpp/string_utils/base64
     library/cpp/uri
     ydb/public/sdk/cpp/src/client/types
+    ydb/public/sdk/cpp/src/client/types/core_facility
     ydb/public/sdk/cpp/src/client/types/credentials
 )
 
diff --git a/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.cpp b/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.cpp
index a07a54fef8c..e6bd55103e2 100644
--- a/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.cpp
+++ b/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.cpp
@@ -10,6 +10,11 @@ void TIamTokenServiceStub::SetResponseToken(const std::string& token, int64_t ex
     ExpiresAtSeconds_ = expiresAtSeconds;
 }
 
+void TIamTokenServiceStub::SetStatus(grpc::Status status) {
+    std::lock_guard lock(Lock_);
+    Status_ = std::move(status);
+}
+
 grpc::Status TIamTokenServiceStub::Create(
     grpc::ServerContext*,
     const yandex::cloud::iam::v1::CreateIamTokenRequest* request,
@@ -19,6 +24,9 @@ grpc::Status TIamTokenServiceStub::Create(
     ++RequestCount_;
     LastRequest_ = *request;
     HasLastRequest_ = true;
+    if (!Status_.ok()) {
+        return Status_;
+    }
     response->set_iam_token(IamToken_);
     response->mutable_expires_at()->set_seconds(ExpiresAtSeconds_);
     response->mutable_expires_at()->set_nanos(0);
diff --git a/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.h b/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.h
index dede93ca1a0..c02001d459b 100644
--- a/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.h
+++ b/ydb/public/sdk/cpp/tests/common/iam_mocks/iam_grpc_mock_server.h
@@ -21,6 +21,7 @@ namespace NYdb::NTest {
 class TIamTokenServiceStub final : public yandex::cloud::iam::v1::IamTokenService::Service {
 public:
     void SetResponseToken(const std::string& token, int64_t expiresAtSeconds = 4102444800);
+    void SetStatus(grpc::Status status);
 
     grpc::Status Create(
         grpc::ServerContext*,
@@ -34,6 +35,7 @@ public:
 private:
     mutable std::mutex Lock_;
     std::string IamToken_;
+    grpc::Status Status_;
     int64_t ExpiresAtSeconds_ = 4102444800;
     int RequestCount_ = 0;
     yandex::cloud::iam::v1::CreateIamTokenRequest LastRequest_;
diff --git a/ydb/public/sdk/cpp/tests/unit/client/driver/driver_ut.cpp b/ydb/public/sdk/cpp/tests/unit/client/driver/driver_ut.cpp
index 330e048cdf1..d69edfdb76d 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/driver/driver_ut.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/driver/driver_ut.cpp
@@ -124,32 +124,54 @@ namespace {
         std::atomic_int& ProviderCount_;
     };
 
+    class TDeferredAuthProvider final : public ICredentialsProvider {
+    public:
+        TDeferredAuthProvider()
+            : AuthInfo_(NThreading::NewPromise<std::string>())
+        {}
+
+        std::string GetAuthInfo() const override {
+            return AuthInfo_.GetFuture().GetValueSync();
+        }
+
+        NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
+            return AuthInfo_.GetFuture();
+        }
+
+        bool IsValid() const override {
+            return true;
+        }
+
+        void SetReady() {
+            AuthInfo_.SetValue("token");
+        }
+
+    private:
+        NThreading::TPromise<std::string> AuthInfo_;
+    };
+
     class TDeferredCredentialsFactory final : public ICredentialsProviderFactory {
     public:
         TDeferredCredentialsFactory()
-            : Provider_(NThreading::NewPromise<TCredentialsProviderPtr>())
+            : Provider_(std::make_shared<TDeferredAuthProvider>())
         {}
 
         TCredentialsProviderPtr CreateProvider() const override {
-            return CreateInsecureCredentialsProviderFactory()->CreateProvider();
-        }
-
-        NThreading::TFuture<TCredentialsProviderPtr> CreateProviderAsync(std::weak_ptr<ICoreFacility>) const override {
-            return Provider_.GetFuture();
+            return Provider_;
         }
 
         void SetReady() {
-            Provider_.SetValue(CreateProvider());
+            Provider_->SetReady();
         }
 
     private:
-        NThreading::TPromise<TCredentialsProviderPtr> Provider_;
+        std::shared_ptr<TDeferredAuthProvider> Provider_;
     };
 
 } // namespace
 
 Y_UNIT_TEST_SUITE(DeferredCredentialsTest) {
-    Y_UNIT_TEST(RequestWaitsForCredentials) {
+    Y_UNIT_TEST(RequestWaitsForAuthInfo) {
         auto factory = std::make_shared<TDeferredCredentialsFactory>();
         auto driver = TDriver(TDriverConfig()
             .SetEndpoint("localhost:100")
diff --git a/ydb/public/sdk/cpp/tests/unit/client/iam/grpc_iam_ut.cpp b/ydb/public/sdk/cpp/tests/unit/client/iam/grpc_iam_ut.cpp
index f0393f26110..fb3346e7f21 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/iam/grpc_iam_ut.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/iam/grpc_iam_ut.cpp
@@ -33,19 +33,19 @@ TEST(GrpcIamCredentialsProvider, TeardownWhileIamCreatePendingCompletes) {
     TIamOAuth params = MakeOAuthParams(server.Endpoint());
     params.RequestTimeout = TDuration::MilliSeconds(400);
 
-    auto work = [&params] {
-        auto facility = std::make_shared<TSimpleCoreFacility>();
-        TIamOAuthCredentialsProvider<CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService> provider(
-            params,
-            facility);
-        (void)provider;
-    };
-
-    std::future<void> done = std::async(std::launch::async, work);
+    auto facility = std::make_shared<TSimpleCoreFacility>();
+    auto provider = std::make_shared<TIamOAuthCredentialsProvider<
+        CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService>>(params, facility);
+    auto authInfo = provider->GetAuthInfoAsync();
+    ASSERT_FALSE(authInfo.IsReady());
 
     ASSERT_TRUE(iamService.WaitUntilRpcEntered(std::chrono::seconds(5)))
         << "server should have accepted the IAM Create call";
 
+    std::future<void> done = std::async(std::launch::async, [provider = std::move(provider)]() mutable {
+        provider.reset();
+    });
+
     ASSERT_EQ(done.wait_for(std::chrono::seconds(20)), std::future_status::ready)
         << "provider destructor must finish while an IAM Create is still blocked on the server "
            "(in-flight RPC vs channel teardown).";
@@ -67,16 +67,15 @@ TEST(GrpcIamCredentialsProvider, TeardownWhileIamCreatePendingCompletesViaFactor
     auto factory = std::make_shared<TIamOAuthCredentialsProviderFactory<
         CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService>>(params);
 
-    auto work = [&factory] {
-        auto provider = factory->CreateProvider();
-        (void)provider;
-    };
-
-    std::future<void> done = std::async(std::launch::async, work);
+    auto provider = factory->CreateProvider();
 
     ASSERT_TRUE(iamService.WaitUntilRpcEntered(std::chrono::seconds(5)))
         << "server should have accepted the IAM Create call";
 
+    std::future<void> done = std::async(std::launch::async, [provider = std::move(provider)]() mutable {
+        provider.reset();
+    });
+
     ASSERT_EQ(done.wait_for(std::chrono::seconds(20)), std::future_status::ready)
         << "factory wrapper teardown must finish while an IAM Create is still blocked on the server";
     done.get();
@@ -107,6 +106,7 @@ TEST(GrpcIamCredentialsProviderFactory, NoArgProvidersAreCachedAcrossFactoryInst
     for (size_t i = 1; i < oauthProviders.size(); ++i) {
         EXPECT_EQ(firstOAuthProvider, oauthProviders[i].get());
     }
+    EXPECT_EQ(firstOAuthProvider->GetAuthInfo(), "unit-test-iam-token");
     EXPECT_EQ(iamStub.GetRequestCount(), 1);
 
     using TJwtFactory = TIamJwtCredentialsProviderFactory<
@@ -116,17 +116,41 @@ TEST(GrpcIamCredentialsProviderFactory, NoArgProvidersAreCachedAcrossFactoryInst
     auto secondJwtProvider = std::make_shared<TJwtFactory>(jwtParams)->CreateProvider();
 
     EXPECT_EQ(firstJwtProvider, secondJwtProvider);
+    EXPECT_EQ(firstJwtProvider->GetAuthInfo(), "unit-test-iam-token");
     EXPECT_EQ(iamStub.GetRequestCount(), 2);
 
     auto differentOAuthParams = oauthParams;
     differentOAuthParams.OAuthToken = "different-oauth-token";
     auto differentOAuthProvider = std::make_shared<TOAuthFactory>(differentOAuthParams)->CreateProvider();
     EXPECT_NE(firstOAuthProvider, differentOAuthProvider);
+    EXPECT_EQ(differentOAuthProvider->GetAuthInfo(), "unit-test-iam-token");
     EXPECT_EQ(iamStub.GetRequestCount(), 3);
 
     server.Stop();
 }
 
+TEST(GrpcIamCredentialsProviderFactory, ReadyCallbackCanReleaseNoArgProvider) {
+    TIamTokenServiceStub iamStub;
+    iamStub.SetResponseToken("unit-test-iam-token");
+    TIamGrpcServer server(&iamStub);
+    ASSERT_TRUE(server.Start());
+
+    auto provider = std::make_shared<TIamOAuthCredentialsProviderFactory<
+        CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService>>(
+            MakeOAuthParams(server.Endpoint()))->CreateProvider();
+    auto released = std::make_shared<std::promise<void>>();
+    auto authInfo = provider->GetAuthInfoAsync();
+    ASSERT_TRUE(authInfo.IsReady());
+    authInfo.Subscribe(
+        [provider = std::move(provider), released](const auto&) mutable {
+            provider.reset();
+            released->set_value();
+        });
+
+    ASSERT_EQ(released->get_future().wait_for(std::chrono::seconds(10)), std::future_status::ready);
+    server.Stop();
+}
+
 namespace {
 
 class TSlowBlockingAuthProvider final : public ICredentialsProvider {
@@ -134,6 +158,7 @@ public:
     std::string GetAuthInfo() const override {
         std::unique_lock lock(Mutex_);
         if (++CallCount_ > 1) {
+            BlockedCv_.notify_all();
             BlockedCv_.wait(lock, [this] { return Released_; });
         }
         return "slow-auth-token";
@@ -152,16 +177,8 @@ public:
     }
 
     bool WaitUntilBlocked(std::chrono::milliseconds timeout) const {
-        const auto deadline = std::chrono::steady_clock::now() + timeout;
-        while (std::chrono::steady_clock::now() < deadline) {
-            std::lock_guard lock(Mutex_);
-            if (CallCount_ > 1 && !Released_) {
-                return true;
-            }
-            std::this_thread::sleep_for(std::chrono::milliseconds(2));
-        }
-        std::lock_guard lock(Mutex_);
-        return CallCount_ > 1 && !Released_;
+        std::unique_lock lock(Mutex_);
+        return BlockedCv_.wait_for(lock, timeout, [this] { return CallCount_ > 1; });
     }
 
 private:
@@ -171,30 +188,36 @@ private:
     bool Released_ = false;
 };
 
-class TFailThenSucceedAuthProvider final : public ICredentialsProvider {
+class TDeferredAuthProvider final : public ICredentialsProvider {
 public:
-    explicit TFailThenSucceedAuthProvider(int failCount)
-        : FailCount_(failCount)
+    TDeferredAuthProvider()
+        : AuthInfo_(NThreading::NewPromise<std::string>())
     {}
 
     std::string GetAuthInfo() const override {
-        if (CallCount_.fetch_add(1) < FailCount_) {
-            ythrow yexception() << "auth failure";
-        }
-        return "auth-token";
+        return AuthInfo_.GetFuture().GetValueSync();
+    }
+
+    NThreading::TFuture<std::string> GetAuthInfoAsync() const override {
+        ++CallCount_;
+        return AuthInfo_.GetFuture();
     }
 
     bool IsValid() const override {
         return true;
     }
 
-    int GetCallCount() const {
+    int CallCount() const {
         return CallCount_.load();
     }
 
+    void SetReady() {
+        AuthInfo_.SetValue("auth-token");
+    }
+
 private:
-    const int FailCount_;
     mutable std::atomic<int> CallCount_{0};
+    NThreading::TPromise<std::string> AuthInfo_;
 };
 
 } // namespace
@@ -250,13 +273,13 @@ TEST(GrpcIamCredentialsProvider, StopDuringFillContextDoesNotHang) {
     server.Stop();
 }
 
-TEST(GrpcIamCredentialsProvider, FillContextAuthExceptionSurvivesAndRecovers) {
+TEST(GrpcIamCredentialsProvider, WaitsForNestedAuthWithoutPollingIt) {
     TIamTokenServiceStub iamStub;
     iamStub.SetResponseToken("unit-test-iam-token");
     TIamGrpcServer server(&iamStub);
     ASSERT_TRUE(server.Start());
 
-    auto authProvider = std::make_shared<TFailThenSucceedAuthProvider>(2);
+    auto authProvider = std::make_shared<TDeferredAuthProvider>();
 
     TIamOAuth params = MakeOAuthParams(server.Endpoint());
     auto facility = std::make_shared<TSimpleCoreFacility>();
@@ -273,9 +296,64 @@ TEST(GrpcIamCredentialsProvider, FillContextAuthExceptionSurvivesAndRecovers) {
         facility,
         authProvider);
 
-    EXPECT_EQ(provider.GetAuthInfo(), "unit-test-iam-token");
+    auto future = provider.GetAuthInfoAsync();
+    for (int i = 0; i < 100 && authProvider->CallCount() == 0; ++i) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    ASSERT_EQ(authProvider->CallCount(), 1);
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+    EXPECT_EQ(authProvider->CallCount(), 1);
+
+    authProvider->SetReady();
+    ASSERT_TRUE(future.Wait(TDuration::Seconds(10)));
+    EXPECT_EQ(future.GetValue(), "unit-test-iam-token");
     EXPECT_EQ(iamStub.GetRequestCount(), 1);
-    EXPECT_GE(authProvider->GetCallCount(), 3);
+
+    server.Stop();
+}
+
+TEST(GrpcIamCredentialsProvider, RetriesTransientFailure) {
+    TIamTokenServiceStub iamStub;
+    iamStub.SetResponseToken("unit-test-iam-token");
+    iamStub.SetStatus(grpc::Status(grpc::StatusCode::UNAVAILABLE, "retry"));
+    TIamGrpcServer server(&iamStub);
+    ASSERT_TRUE(server.Start());
+
+    auto facility = std::make_shared<TSimpleCoreFacility>();
+    auto provider = std::make_shared<TIamOAuthCredentialsProvider<
+        CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService>>(
+            MakeOAuthParams(server.Endpoint()), facility);
+    auto future = provider->GetAuthInfoAsync();
+    for (int i = 0; i < 1000 && iamStub.GetRequestCount() == 0; ++i) {
+        std::this_thread::sleep_for(std::chrono::milliseconds(10));
+    }
+    ASSERT_GT(iamStub.GetRequestCount(), 0);
+    ASSERT_FALSE(future.IsReady());
+
+    iamStub.SetStatus(grpc::Status::OK);
+    ASSERT_TRUE(future.Wait(TDuration::Seconds(10)));
+    EXPECT_EQ(future.GetValue(), "unit-test-iam-token");
+    EXPECT_GT(iamStub.GetRequestCount(), 1);
+
+    server.Stop();
+}
+
+TEST(GrpcIamCredentialsProvider, DoesNotRetryTerminalFailure) {
+    TIamTokenServiceStub iamStub;
+    iamStub.SetStatus(grpc::Status(grpc::StatusCode::PERMISSION_DENIED, "terminal"));
+    TIamGrpcServer server(&iamStub);
+    ASSERT_TRUE(server.Start());
+
+    auto facility = std::make_shared<TSimpleCoreFacility>();
+    auto provider = std::make_shared<TIamOAuthCredentialsProvider<
+        CreateIamTokenRequest, CreateIamTokenResponse, IamTokenService>>(
+            MakeOAuthParams(server.Endpoint()), facility);
+    auto future = provider->GetAuthInfoAsync();
+    ASSERT_TRUE(future.Wait(TDuration::Seconds(10)));
+    EXPECT_THROW(future.GetValue(), yexception);
+    const int requests = iamStub.GetRequestCount();
+    std::this_thread::sleep_for(std::chrono::milliseconds(300));
+    EXPECT_EQ(iamStub.GetRequestCount(), requests);
 
     server.Stop();
 }
diff --git a/ydb/public/sdk/cpp/tests/unit/client/iam/http_iam_ut.cpp b/ydb/public/sdk/cpp/tests/unit/client/iam/http_iam_ut.cpp
index a4a667e983d..baae8eec0d5 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/iam/http_iam_ut.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/iam/http_iam_ut.cpp
@@ -1,4 +1,5 @@
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/iam.h>
+#include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h>
 
 #include <ydb/public/sdk/cpp/tests/common/iam_mocks/iam_http_mock_server.h>
 
@@ -65,7 +66,7 @@ TEST(IamCredentialsProvider, NoExpiryFieldFallback) {
 TEST(IamCredentialsProvider, ServerError) {
     TMetadataServer server;
     server.SetStrictMode(false);
-    server.SetResponse(HTTP_INTERNAL_SERVER_ERROR, "");
+    server.SetResponse(HTTP_BAD_REQUEST, "");
 
     TIamHost params = MakeMetadataParams(server.Port);
 
@@ -75,7 +76,7 @@ TEST(IamCredentialsProvider, ServerError) {
     EXPECT_THROW(provider->GetAuthInfo(), yexception);
 }
 
-TEST(IamCredentialsProvider, GracePeriodOnRefreshError) {
+TEST(IamCredentialsProvider, RetriesTransientError) {
     TMetadataServer server;
     server.SetStrictMode(false);
     server.SetResponse(HTTP_OK, MakeTokenResponse("old-token", 3600));
@@ -83,57 +84,22 @@ TEST(IamCredentialsProvider, GracePeriodOnRefreshError) {
     TIamHost params = MakeMetadataParams(server.Port);
     params.RefreshPeriod = TDuration::MilliSeconds(100);
 
-    auto provider = CreateIamCredentialsProviderFactory(params)->CreateProvider();
+    auto facility = CreateSimpleCoreFacility();
+    auto provider = CreateIamCredentialsProviderFactory(params)->CreateProvider(facility);
     EXPECT_EQ(provider->GetAuthInfo(), "old-token");
 
-    int countBeforeRefresh = server.GetRequestCount();
-    Sleep(TDuration::MilliSeconds(150));
-
+    const int countBeforeRefresh = server.GetRequestCount();
     server.SetResponse(HTTP_INTERNAL_SERVER_ERROR, "");
-    EXPECT_EQ(provider->GetAuthInfo(), "old-token");
-    EXPECT_GT(server.GetRequestCount(), countBeforeRefresh);
-}
-
-TEST(IamCredentialsProvider, ThrowAfterTokenExpiredOnRefreshError) {
-    TMetadataServer server;
-    server.SetStrictMode(false);
-    server.SetResponse(HTTP_OK, MakeTokenResponse("old-token", 1));
-
-    TIamHost params = MakeMetadataParams(server.Port);
-    params.RefreshPeriod = TDuration::MilliSeconds(100);
-
-    auto provider = CreateIamCredentialsProviderFactory(params)->CreateProvider();
-    EXPECT_EQ(provider->GetAuthInfo(), "old-token");
-
-    Sleep(TDuration::MilliSeconds(150));
-    server.SetResponse(HTTP_INTERNAL_SERVER_ERROR, "");
-    EXPECT_EQ(provider->GetAuthInfo(), "old-token");
-
-    Sleep(TDuration::Seconds(1));
-    EXPECT_THROW(provider->GetAuthInfo(), yexception);
-}
-
-TEST(IamCredentialsProvider, RecoveryAfterRefreshError) {
-    TMetadataServer server;
-    server.SetStrictMode(false);
-    server.SetResponse(HTTP_OK, MakeTokenResponse("token-1", 3600));
-
-    TIamHost params = MakeMetadataParams(server.Port);
-    params.RefreshPeriod = TDuration::MilliSeconds(100);
-
-    auto provider = CreateIamCredentialsProviderFactory(params)->CreateProvider();
-    EXPECT_EQ(provider->GetAuthInfo(), "token-1");
-
-    Sleep(TDuration::MilliSeconds(150));
-    server.SetResponse(HTTP_INTERNAL_SERVER_ERROR, "");
-    EXPECT_EQ(provider->GetAuthInfo(), "token-1");
-
-    server.SetResponse(HTTP_OK, MakeTokenResponse("token-2", 3600));
-    int countBeforeRecovery = server.GetRequestCount();
-    Sleep(TDuration::MilliSeconds(150));
+    for (int i = 0; i < 1000 && server.GetRequestCount() == countBeforeRefresh; ++i) {
+        Sleep(TDuration::MilliSeconds(10));
+    }
+    ASSERT_GT(server.GetRequestCount(), countBeforeRefresh);
+    auto future = provider->GetAuthInfoAsync();
+    EXPECT_FALSE(future.IsReady());
 
-    EXPECT_EQ(provider->GetAuthInfo(), "token-2");
-    EXPECT_GT(server.GetRequestCount(), countBeforeRecovery);
+    server.SetResponse(HTTP_OK, MakeTokenResponse("new-token", 3600));
+    ASSERT_TRUE(future.Wait(TDuration::Seconds(10)));
+    EXPECT_EQ(future.GetValue(), "new-token");
 }
 
 TEST(IamCredentialsProvider, ConcurrentAccess) {
diff --git a/ydb/public/sdk/cpp/tests/unit/client/iam_private/grpc_iam_service_ut.cpp b/ydb/public/sdk/cpp/tests/unit/client/iam_private/grpc_iam_service_ut.cpp
index 29b21e145b3..73b1164d347 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/iam_private/grpc_iam_service_ut.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/iam_private/grpc_iam_service_ut.cpp
@@ -107,9 +107,7 @@ public:
         callback(std::move(issues), EStatus::CLIENT_CANCELLED);
     }
 
-    void PostToResponseQueue(TPostTaskCb&& callback) override {
-        callback();
-    }
+    void PostToResponseQueue(TPostTaskCb&&) override {}
 };
 
 using TTestOAuthFactory = TIamOAuthCredentialsProviderFactory<
@@ -183,28 +181,31 @@ TEST(IamServiceCredentialsProvider, NoArgProviderIsCachedAcrossFactoryInstances)
     secondServiceParams.TargetServiceAccountId = "another-target";
     auto differentProvider = CreateIamServiceCredentialsProviderFactory(secondServiceParams)->CreateProvider();
     EXPECT_NE(firstProvider, differentProvider);
+    EXPECT_EQ(differentProvider->GetAuthInfo(), "outer-service-token");
     EXPECT_EQ(stub.GetCreateRequestCount(), 1);
     EXPECT_EQ(stub.GetCreateForServiceRequestCount(), 2);
 
     server.Stop();
 }
 
-TEST(IamCredentialsProvider, AsyncCreationFailsWithExpiredFacility) {
-    auto future = MakeOAuthFactory().CreateProviderAsync(std::weak_ptr<ICoreFacility>{});
+TEST(IamCredentialsProvider, AuthInfoFailsWithExpiredFacility) {
+    auto provider = MakeOAuthFactory().CreateProvider(std::weak_ptr<ICoreFacility>{});
+    auto future = provider->GetAuthInfoAsync();
 
     ASSERT_TRUE(future.IsReady());
     EXPECT_THROW(future.GetValue(), std::exception);
 }
 
-TEST(IamCredentialsProvider, AsyncCreationFailsWhenPeriodicTaskIsRejected) {
+TEST(IamCredentialsProvider, AuthInfoFailsWhenPeriodicTaskIsRejected) {
     auto facility = std::make_shared<TFailingCoreFacility>();
-    auto future = MakeOAuthFactory().CreateProviderAsync(std::weak_ptr<ICoreFacility>(facility));
+    auto provider = MakeOAuthFactory().CreateProvider(std::weak_ptr<ICoreFacility>(facility));
+    auto future = provider->GetAuthInfoAsync();
 
     ASSERT_TRUE(future.IsReady());
     EXPECT_THROW(future.GetValue(), std::exception);
 }
 
-TEST(IamCredentialsProvider, AsyncCreationRetriesTransientIamFailure) {
+TEST(IamCredentialsProvider, AuthInfoRetriesTransientIamFailure) {
     TFlakyIamServiceStub stub;
     TIamGrpcServer server(&stub);
     ASSERT_TRUE(server.Start());
@@ -215,22 +216,17 @@ TEST(IamCredentialsProvider, AsyncCreationRetriesTransientIamFailure) {
     params.OAuthToken = "token";
     params.RequestTimeout = TDuration::Seconds(2);
 
-    auto future = TTestOAuthFactory(params).CreateProviderAsync();
+    auto provider = TTestOAuthFactory(params).CreateProvider();
+    auto future = provider->GetAuthInfoAsync();
     ASSERT_TRUE(future.Wait(TDuration::Seconds(10)));
-    auto provider = future.GetValue();
 
-    EXPECT_EQ(provider->GetAuthInfo(), "oauth-token");
+    EXPECT_EQ(future.GetValue(), "oauth-token");
     EXPECT_GE(stub.Attempts(), 2u);
 
     server.Stop();
 }
 
-// Regression test for the deprecated no-arg CreateProvider() on the IAM service-account
-// factory with a nested gRPC JWT auth provider. Before the fix, both providers shared a single
-// TSimpleCoreFacility, each registered a periodic refresh task, and TSimpleCoreFacility's
-// single-task invariant tripped Y_ABORT_UNLESS and killed the process. The fix gives the
-// nested auth provider its own facility (via a recursive no-arg CreateProvider()), so the two
-// periodic tasks land on separate facilities.
+// The no-arg service provider keeps nested gRPC authentication on a separate facility.
 TEST(IamServiceCredentialsProvider, NoArgCreateProviderWithGrpcInnerCreds) {
     TIamServiceStub stub;
     TIamGrpcServer server(&stub);
diff --git a/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/credentials_ut.cpp b/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/credentials_ut.cpp
index 7b2d0dd973e..724dedcbc33 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/credentials_ut.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/credentials_ut.cpp
@@ -1,5 +1,6 @@
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/oauth2_token_exchange/credentials.h>
 #include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/credentials/oauth2_token_exchange/from_file.h>
+#include <ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/core_facility/core_facility.h>
 #include <ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/helpers/test_token_exchange_server.h>
 
 #include <library/cpp/string_utils/base64/base64.h>
@@ -12,6 +13,8 @@
 #include <util/string/builder.h>
 #include <util/system/tempfile.h>
 
+#include <future>
+
 using namespace NYdb;
 
 extern const std::string TestRSAPrivateKeyContent;
@@ -108,6 +111,19 @@ struct TTestConfigFile : public TJsonFiller<TTestConfigFile> {
 };
 
 Y_UNIT_TEST_SUITE(TestTokenExchange) {
+    bool WaitRequest(TTestTokenExchangeServer& server, TDuration timeout) {
+        const auto deadline = TInstant::Now() + timeout;
+        do {
+            bool received = false;
+            server.WithLock([&] { received = server.Check.InputParams.has_value(); });
+            if (received) {
+                return true;
+            }
+            Sleep(TDuration::MilliSeconds(10));
+        } while (TInstant::Now() < deadline);
+        return false;
+    }
+
     void Exchanges(bool fromConfig) {
         TTestTokenExchangeServer server;
         server.Check.ExpectedInputParams.emplace("grant_type", "urn:ietf:params:oauth:grant-type:token-exchange");
@@ -405,28 +421,6 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
         server.Check.ExpectedInputParams.erase("scope");
 
 
-        server.Check.ExpectedErrorPart = "can not connect to";
-        server.Check.ExpectRequest = false;
-        if (fromConfig) {
-            server.RunFromConfig(
-                TTestConfigFile()
-                    .Field("token-endpoint", "https://localhost:42/aaa")
-                    .SubMap("subject-credentials")
-                        .Field("type", "Fixed")
-                        .Field("token", "test_token")
-                        .Field("token-type", "test_token_type")
-                        .Build()
-                    .Build()
-            );
-        } else {
-            server.Run(
-                TOauth2TokenExchangeParams()
-                    .TokenEndpoint("https://localhost:42/aaa")
-                    .SubjectTokenSource(CreateFixedTokenSource("test_token", "test_token_type"))
-            );
-        }
-        server.Check.ExpectRequest = true;
-
         // parsing response
         server.Check.StatusCode = HTTP_FORBIDDEN;
         server.Check.Response = R"(not json)";
@@ -488,16 +482,13 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
 
         server.WithLock(
             [&]() {
+                server.Check.Reset();
                 server.Check.Response = R"({"access_token": "token_2", "token_type": "bearer", "expires_in": 1})";
             }
         );
 
-        Sleep(TDuration::Seconds(1));
-        server.Run(
-            [&]() {
-                UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_2");
-            }
-        );
+        UNIT_ASSERT(WaitRequest(server, TDuration::Seconds(10)));
+        UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_2");
     }
 
     Y_UNIT_TEST(UpdatesToken) {
@@ -540,138 +531,44 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
         UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer the_only_token");
     }
 
-    Y_UNIT_TEST(UpdatesTokenInBackgroud) {
-        TCredentialsProviderFactoryPtr factory;
-        TInstant startTime;
-
-        TTestTokenExchangeServer server;
-        server.Check.ExpectedInputParams.emplace("grant_type", "urn:ietf:params:oauth:grant-type:token-exchange");
-        server.Check.ExpectedInputParams.emplace("requested_token_type", "urn:ietf:params:oauth:token-type:access_token");
-        server.Check.ExpectedInputParams.emplace("actor_token", "test_token");
-        server.Check.ExpectedInputParams.emplace("actor_token_type", "test_token_type");
-
-        for (int i = 0; i < 2; ++i) {
-            server.WithLock(
-                [&]() {
-                    server.Check.Response = R"({"access_token": "token_1", "token_type": "bearer", "expires_in": 2})";
-                }
-            );
-            if (!factory) {
-                server.Run(
-                    [&]() {
-                        factory = CreateOauth2TokenExchangeCredentialsProviderFactory(
-                            TOauth2TokenExchangeParams()
-                                .TokenEndpoint(server.GetEndpoint())
-                                .ActorTokenSource(CreateFixedTokenSource("test_token", "test_token_type")));
-                        startTime = TInstant::Now();
-                        UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_1");
-                    }
-                );
-            }
-
-            server.WithLock(
-                [&]() {
-                    server.Check.Reset();
-                    if (i == 0) {
-                        server.Check.Response = R"({"access_token": "token_2", "token_type": "bearer", "expires_in": 2})";
-                    } else {
-                        server.Check.Response = R"({"access_token": "token_3", "token_type": "bearer", "expires_in": 2})";
-                    }
-                }
-            );
-
-            SleepUntil(startTime + TDuration::Seconds(1) + TDuration::MilliSeconds(5));
-            const std::string token = factory->CreateProvider()->GetAuthInfo();
-            TInstant halfTimeTokenValid = TInstant::Now();
-            if (halfTimeTokenValid < startTime + TDuration::Seconds(2)) { // valid => got cached token, but async update must be run after half time token is valid
-                if (i == 0) {
-                    UNIT_ASSERT_VALUES_EQUAL(token, "Bearer token_1");
-                } else {
-                    UNIT_ASSERT_VALUES_EQUAL(token, "Bearer token_2");
-                }
-                do {
-                    Sleep(TDuration::MilliSeconds(10));
-                    bool gotRequest = false;
-                    server.WithLock(
-                        [&]() {
-                            if (server.Check.InputParams) { // InputParams are created => got the request
-                                gotRequest = true;
-                            }
-                        }
-                    );
-                    if (gotRequest) {
-                        startTime = TInstant::Now(); // for second iteration
-                        break;
-                    }
-                } while (TInstant::Now() <= startTime + TDuration::Seconds(30));
-                server.CheckExpectations();
-                server.WithLock(
-                    [&]() {
-                        server.Check.Reset();
-                        server.Check.Response = R"(invalid response)"; // update must finish asyncronously
-                    }
-                );
-                Sleep(TDuration::MilliSeconds(500)); // After the request is got, it takes some time to get updated token
-                if (i == 0) { // Finally check that we got updated token
-                    UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_2");
-                } else {
-                    UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_3");
-                }
-                Cerr << "Checked backgroud update on " << i << " iteration" << Endl;
-            }
-        }
-    }
-
     Y_UNIT_TEST(UpdatesTokenAndRetriesErrors) {
         TCredentialsProviderFactoryPtr factory;
+        auto facility = CreateSimpleCoreFacility();
+        TCredentialsProviderPtr provider;
 
         TTestTokenExchangeServer server;
         server.Check.ExpectedInputParams.emplace("grant_type", "urn:ietf:params:oauth:grant-type:token-exchange");
         server.Check.ExpectedInputParams.emplace("requested_token_type", "urn:ietf:params:oauth:token-type:access_token");
         server.Check.ExpectedInputParams.emplace("subject_token", "test_token");
         server.Check.ExpectedInputParams.emplace("subject_token_type", "test_token_type");
-        server.Check.Response = R"({"access_token": "token_1", "token_type": "bearer", "expires_in": 6})";
+        server.Check.Response = R"({"access_token": "token_1", "token_type": "bearer", "expires_in": 2})";
         server.Run(
             [&]() {
                 factory = CreateOauth2TokenExchangeCredentialsProviderFactory(
                     TOauth2TokenExchangeParams()
                         .TokenEndpoint(server.GetEndpoint())
                         .SubjectTokenSource(CreateFixedTokenSource("test_token", "test_token_type")));
-                UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_1");
+                provider = factory->CreateProvider(facility);
+                UNIT_ASSERT_VALUES_EQUAL(provider->GetAuthInfo(), "Bearer token_1");
             }
         );
 
         server.WithLock(
             [&]() {
                 server.Check.Reset();
-                server.Check.StatusCode = HTTP_BAD_REQUEST; // all errors are temporary, because the first attempt is always successful (in constructor)
+                server.Check.StatusCode = HTTP_INTERNAL_SERVER_ERROR;
                 server.Check.Response = R"({"error": "tmp", "error_description": "temporary error"})";
             }
         );
 
-        Sleep(TDuration::Seconds(3) + TDuration::MilliSeconds(5));
-        UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_1");
-
-        auto waitRequest = [&](TDuration howLong) {
-            TInstant startTime = TInstant::Now();
-            bool gotRequest = false;
-            do {
-                Sleep(TDuration::MilliSeconds(10));
-                server.WithLock(
-                    [&]() {
-                        if (server.Check.InputParams) { // InputParams are created => got the request
-                            gotRequest = true;
-                        }
-                    }
-                );
-                if (gotRequest) {
-                    break;
-                }
-            } while (TInstant::Now() <= startTime + howLong);
-            return gotRequest;
-        };
-
-        UNIT_ASSERT(waitRequest(TDuration::Seconds(30)));
+        UNIT_ASSERT(WaitRequest(server, TDuration::Seconds(30)));
+        auto future = provider->GetAuthInfoAsync();
+        UNIT_ASSERT(!future.IsReady());
+        auto released = std::make_shared<std::promise<void>>();
+        future.Subscribe([provider = std::move(provider), released](const auto&) mutable {
+            provider.reset();
+            released->set_value();
+        });
 
         server.WithLock(
             [&]() {
@@ -681,24 +578,15 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
             }
         );
 
-        UNIT_ASSERT(waitRequest(TDuration::Seconds(10)));
-        Sleep(TDuration::MilliSeconds(500)); // After the request is got, it takes some time to get updated token
-        UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_2");
-
-        server.WithLock(
-            [&]() {
-                server.Check.Reset();
-                server.Check.StatusCode = HTTP_INTERNAL_SERVER_ERROR;
-                server.Check.Response = R"({})";
-            }
-        );
-
-        Sleep(TDuration::Seconds(2));
-        UNIT_ASSERT_EXCEPTION(factory->CreateProvider()->GetAuthInfo(), std::runtime_error);
+        UNIT_ASSERT(future.Wait(TDuration::Seconds(10)));
+        UNIT_ASSERT_VALUES_EQUAL(future.GetValue(), "Bearer token_2");
+        UNIT_ASSERT(released->get_future().wait_for(std::chrono::seconds(10)) == std::future_status::ready);
     }
 
     Y_UNIT_TEST(ShutdownWhileRefreshingToken) {
         TCredentialsProviderFactoryPtr factory;
+        auto facility = CreateSimpleCoreFacility();
+        TCredentialsProviderPtr provider;
 
         TTestTokenExchangeServer server;
         server.Check.ExpectedInputParams.emplace("grant_type", "urn:ietf:params:oauth:grant-type:token-exchange");
@@ -713,7 +601,8 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
                     TOauth2TokenExchangeParams()
                         .TokenEndpoint(server.GetEndpoint())
                         .SubjectTokenSource(CreateFixedTokenSource("test_token", "test_token_type")));
-                UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_1");
+                provider = factory->CreateProvider(facility);
+                UNIT_ASSERT_VALUES_EQUAL(provider->GetAuthInfo(), "Bearer token_1");
             }
         );
 
@@ -725,14 +614,16 @@ Y_UNIT_TEST_SUITE(TestTokenExchange) {
             }
         );
 
-        Sleep(TDuration::Seconds(3) + TDuration::MilliSeconds(5));
-
-        UNIT_ASSERT_VALUES_EQUAL(factory->CreateProvider()->GetAuthInfo(), "Bearer token_1");
+        UNIT_ASSERT(WaitRequest(server, TDuration::Seconds(30)));
+        auto future = provider->GetAuthInfoAsync();
+        UNIT_ASSERT(!future.IsReady());
 
         const TInstant shutdownStart = TInstant::Now();
-        factory = nullptr;
+        provider = nullptr;
         const TInstant shutdownStop = TInstant::Now();
-        Cerr << "Shutdown: " << (shutdownStop - shutdownStart) << Endl;
+        UNIT_ASSERT(shutdownStop - shutdownStart < TDuration::Seconds(1));
+        UNIT_ASSERT(future.IsReady());
+        UNIT_ASSERT_EXCEPTION(future.GetValue(), yexception);
     }
 
     Y_UNIT_TEST(ExchangesFromFileConfig) {
diff --git a/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/helpers/test_token_exchange_server.cpp b/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/helpers/test_token_exchange_server.cpp
index 1e745f8e618..26da1c120be 100644
--- a/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/helpers/test_token_exchange_server.cpp
+++ b/ydb/public/sdk/cpp/tests/unit/client/oauth2_token_exchange/helpers/test_token_exchange_server.cpp
@@ -17,9 +17,7 @@ void TTestTokenExchangeServer::Run(const NYdb::TOauth2TokenExchangeParams& param
     std::string token;
     Run([&]() {
         auto factory = CreateOauth2TokenExchangeCredentialsProviderFactory(params);
-        if (!expectedToken.empty()) {
-            token = factory->CreateProvider()->GetAuthInfo();
-        }
+        token = factory->CreateProvider()->GetAuthInfo();
     },
     checkExpectations);
 
@@ -32,9 +30,7 @@ void TTestTokenExchangeServer::RunFromConfig(const std::string& fileName, const
     std::string token;
     Run([&]() {
         auto factory = NYdb::CreateOauth2TokenExchangeFileCredentialsProviderFactory(fileName, explicitTokenEndpoint);
-        if (!expectedToken.empty()) {
-            token = factory->CreateProvider()->GetAuthInfo();
-        }
+        token = factory->CreateProvider()->GetAuthInfo();
     },
     checkExpectations);