Propagate deadline in SDK components (#27497)

Author: Gazizonoki

.github/last_commit.txt

@@ -1 +1 @@
-b9b3351e7e628df62ee9335ff1b38d831e7fbe66
+eaa66fe4cc75727a60d89a9ec9d99970e03ce0e0

include/ydb-cpp-sdk/client/types/request_settings.h

@@ -20,7 +20,7 @@ struct TRequestSettings {
     FLUENT_SETTING(std::string, TraceId);
     FLUENT_SETTING(std::string, RequestType);
     FLUENT_SETTING(THeader, Header);
-    FLUENT_SETTING(TDuration, ClientTimeout);
+    FLUENT_SETTING_DEFAULT(TDuration, ClientTimeout, TDuration::Max());
     FLUENT_SETTING(std::string, TraceParent);
 
     TRequestSettings() = default;

src/client/common_client/impl/client.h

@@ -53,7 +53,7 @@ class TClientImplCommon
         return DbDriverState_->DiscoveryCompleted();
     }
 
-    void ScheduleTask(const std::function<void()>& fn, TDuration timeout) override {
+    void ScheduleTask(const std::function<void()>& fn, TDeadline::Duration timeout) override {
         std::weak_ptr<IClientImplCommon> weak = this->shared_from_this();
         auto cbGuard = [weak, fn]() {
             auto strongClient = weak.lock();

src/client/common_client/impl/iface.h

@@ -1,14 +1,15 @@
 #pragma once
 
+#include <src/library/time/time.h>
+
 #include <functional>
-#include <util/datetime/base.h>
 
 namespace NYdb::inline V3 {
 
 class IClientImplCommon {
 public:
     virtual ~IClientImplCommon() = default;
-    virtual void ScheduleTask(const std::function<void()>& fn, TDuration timeout) = 0;
+    virtual void ScheduleTask(const std::function<void()>& fn, TDeadline::Duration timeout) = 0;
 };
 
 }

src/client/driver/driver.cpp

@@ -265,9 +265,9 @@ TDriverConfig TDriver::GetConfig() const {
     );
     config.SetDrainOnDtors(Impl_->DrainOnDtors_);
     config.SetBalancingPolicy(std::make_unique<TBalancingPolicy::TImpl>(Impl_->BalancingSettings_));
-    config.SetGRpcKeepAliveTimeout(Impl_->GRpcKeepAliveTimeout_);
+    config.SetGRpcKeepAliveTimeout(std::chrono::duration_cast<std::chrono::microseconds>(Impl_->GRpcKeepAliveTimeout_));
     config.SetGRpcKeepAlivePermitWithoutCalls(Impl_->GRpcKeepAlivePermitWithoutCalls_);
-    config.SetSocketIdleTimeout(Impl_->SocketIdleTimeout_);
+    config.SetSocketIdleTimeout(std::chrono::duration_cast<std::chrono::microseconds>(Impl_->SocketIdleTimeout_));
     config.SetMaxInboundMessageSize(Impl_->MaxInboundMessageSize_);
     config.SetMaxOutboundMessageSize(Impl_->MaxOutboundMessageSize_);
     config.SetMaxMessageSize(Impl_->MaxMessageSize_);

src/client/federated_topic/impl/federation_observer.cpp

@@ -13,9 +13,9 @@ TFederatedDbObserverImpl::TFederatedDbObserverImpl(std::shared_ptr<TGRpcConnecti
     , PromiseToInitState(NThreading::NewPromise())
     , FederationDiscoveryRetryPolicy(settings.RetryPolicy_)
 {
-    RpcSettings.ClientTimeout = settings.ConnectionTimeout_;
     RpcSettings.EndpointPolicy = TRpcRequestSettings::TEndpointPolicy::UseDiscoveryEndpoint;
     RpcSettings.UseAuth = true;
+    ConnectionTimeout = TDeadline::SafeDurationCast(settings.ConnectionTimeout_);
 }
 
 TFederatedDbObserverImpl::~TFederatedDbObserverImpl() {
@@ -82,6 +82,9 @@ void TFederatedDbObserverImpl::RunFederationDiscoveryImpl() {
         }
     };
 
+    TRpcRequestSettings settings = RpcSettings;
+    settings.Deadline = TDeadline::AfterDuration(ConnectionTimeout);
+
     Connections_->RunDeferred<Ydb::FederationDiscovery::V1::FederationDiscoveryService,
                              Ydb::FederationDiscovery::ListFederationDatabasesRequest,
                              Ydb::FederationDiscovery::ListFederationDatabasesResponse>(
@@ -90,7 +93,7 @@ void TFederatedDbObserverImpl::RunFederationDiscoveryImpl() {
         &Ydb::FederationDiscovery::V1::FederationDiscoveryService::Stub::AsyncListFederationDatabases,
         DbDriverState_,
         {},  // no polling unready operations, so no need in delay parameter
-        RpcSettings,
+        settings,
         FederationDiscoveryDelayContext);
 }
 

src/client/federated_topic/impl/federation_observer.h

@@ -84,6 +84,7 @@ class TFederatedDbObserverImpl : public TClientImplCommon<TFederatedDbObserverIm
     std::shared_ptr<TFederatedDbState> FederatedDbState;
     NThreading::TPromise<void> PromiseToInitState;
     TRpcRequestSettings RpcSettings;
+    TDeadline::Duration ConnectionTimeout;
     TSpinLock Lock;
 
     NTopic::IRetryPolicy::TPtr FederationDiscoveryRetryPolicy;

src/client/impl/internal/db_driver_state/state.cpp

@@ -20,9 +20,11 @@ namespace {
 
 namespace NYdb::inline V3 {
 
+using namespace std::chrono_literals;
+
 constexpr int PESSIMIZATION_DISCOVERY_THRESHOLD = 50; // percent of endpoints pessimized by transport error to start recheck
 constexpr TDuration ENDPOINT_UPDATE_PERIOD = TDuration::Minutes(1); // period to perform endpoints update in "normal" case
-constexpr TDuration DISCOVERY_RECHECK_PERIOD = TDuration::Seconds(5); // period to run periodic discovery task
+constexpr TDeadline::Duration DISCOVERY_RECHECK_PERIOD = 5s; // period to run periodic discovery task
 
 TDbDriverState::TDbDriverState(
     const std::string& database,
@@ -237,7 +239,7 @@ TDbDriverStatePtr TDbDriverStateTracker::GetDriverState(
     return strongState;
 }
 
-void TDbDriverState::AddPeriodicTask(TPeriodicCb&& cb, TDuration period) {
+void TDbDriverState::AddPeriodicTask(TPeriodicCb&& cb, TDeadline::Duration period) {
     Client->AddPeriodicTask(std::move(cb), period);
 }
 

src/client/impl/internal/db_driver_state/state.h

@@ -39,7 +39,7 @@ class TDbDriverState
 
     void SignalDiscoveryCompleted();
 
-    void AddPeriodicTask(TPeriodicCb&& cb, TDuration period) override;
+    void AddPeriodicTask(TPeriodicCb&& cb, TDeadline::Duration period) override;
 
     void AddCb(TCb&& cb, ENotifyType type);
     void ForEachEndpoint(const TEndpointElectorSafe::THandleCb& cb, const void* tag) const;

src/client/impl/internal/grpc_connections/actions.cpp

@@ -4,9 +4,11 @@
 
 #include <src/api/grpc/ydb_operation_v1.grpc.pb.h>
 
+using namespace std::chrono_literals;
+
 namespace NYdb::inline V3 {
 
-constexpr TDuration MAX_DEFERRED_CALL_DELAY = TDuration::Seconds(10); // The max delay between GetOperation calls for one operation
+constexpr TDeadline::Duration MAX_DEFERRED_CALL_DELAY = 10s; // The max delay between GetOperation calls for one operation
 
 TSimpleCbResult::TSimpleCbResult(
         TSimpleCb&& cb,
@@ -24,18 +26,18 @@ TDeferredAction::TDeferredAction(const std::string& operationId,
         TDeferredOperationCb&& userCb,
         TGRpcConnectionsImpl* connection,
         std::shared_ptr<IQueueClientContext> context,
-        TDuration delay,
+        TDeadline::Duration delay,
+        TDeadline globalDeadline,
         TDbDriverStatePtr dbState,
         const std::string& endpoint)
     : TAlarmActionBase(std::move(userCb), connection, std::move(context))
-    , NextDelay_(Min(delay * 2, MAX_DEFERRED_CALL_DELAY))
+    , NextDelay_(std::min(delay * 2, MAX_DEFERRED_CALL_DELAY))
+    , GlobalDeadline_(globalDeadline)
     , DbDriverState_(dbState)
     , OperationId_(operationId)
     , Endpoint_(endpoint)
 {
-    Deadline_ = gpr_time_add(
-        gpr_now(GPR_CLOCK_MONOTONIC),
-        gpr_time_from_micros(delay.MicroSeconds(), GPR_TIMESPAN));
+    Deadline_ = std::min(GlobalDeadline_, TDeadline::AfterDuration(delay));
 }
 
 void TDeferredAction::OnAlarm() {
@@ -46,7 +48,8 @@ void TDeferredAction::OnAlarm() {
 
     TRpcRequestSettings settings;
     settings.PreferredEndpoint = TEndpointKey(Endpoint_, 0);
-    
+    settings.Deadline = GlobalDeadline_;
+
     Connection_->RunDeferred<Ydb::Operation::V1::OperationService, Ydb::Operations::GetOperationRequest, Ydb::Operations::GetOperationResponse>(
         std::move(getOperationRequest),
         std::move(UserResponseCb_),
@@ -56,7 +59,7 @@ void TDeferredAction::OnAlarm() {
         settings,
         true,
         std::move(Context_));
-    }
+}
 
 void TDeferredAction::OnError() {
     Y_ABORT_UNLESS(Connection_);
@@ -76,13 +79,11 @@ TPeriodicAction::TPeriodicAction(
     TPeriodicCb&& userCb,
     TGRpcConnectionsImpl* connection,
     std::shared_ptr<NYdbGrpc::IQueueClientContext> context,
-    TDuration period)
+    TDeadline::Duration period)
     : TAlarmActionBase(std::move(userCb), connection, std::move(context))
     , Period_(period)
 {
-    Deadline_ = gpr_time_add(
-        gpr_now(GPR_CLOCK_MONOTONIC),
-        gpr_time_from_micros(Period_.MicroSeconds(), GPR_TIMESPAN));
+    Deadline_ = TDeadline::AfterDuration(period);
 }
 
 void TPeriodicAction::OnAlarm() {