Merge 510f6a7 into 8c72ada

Author: GrigoriyPA

ydb/core/fq/libs/compute/ydb/resources_cleaner_actor.cpp

@@ -7,6 +7,7 @@
 #include <ydb/core/fq/libs/compute/common/run_actor_params.h>
 #include <ydb/core/fq/libs/compute/ydb/events/events.h>
 #include <ydb/core/fq/libs/ydb/ydb.h>
+#include <ydb/core/util/backoff.h>
 #include <ydb/library/services/services.pb.h>
 
 #include <ydb/public/sdk/cpp/client/ydb_query/client.h>
@@ -65,14 +66,19 @@ class TResourcesCleanerActor : public TBaseComputeActor<TResourcesCleanerActor>
         , Connector(connector)
         , OperationId(operationId)
         , Counters(GetStepCountersSubgroup())
+        , BackoffTimer(20, 1000)
     {}
 
     static constexpr char ActorName[] = "FQ_RESOURCES_CLEANER_ACTOR";
 
+    void SendForgetOperation(const TDuration& delay = TDuration::Zero()) {
+        Register(new TRetryActor<TEvYdbCompute::TEvForgetOperationRequest, TEvYdbCompute::TEvForgetOperationResponse, NYdb::TOperation::TOperationId>(Counters.GetCounters(ERequestType::RT_FORGET_OPERATION), delay, SelfId(), Connector, OperationId));
+    }
+
     void Start() {
         LOG_I("Start resources cleaner actor. Compute state: " << FederatedQuery::QueryMeta::ComputeStatus_Name(Params.Status));
         Become(&TResourcesCleanerActor::StateFunc);
-        Register(new TRetryActor<TEvYdbCompute::TEvForgetOperationRequest, TEvYdbCompute::TEvForgetOperationResponse, NYdb::TOperation::TOperationId>(Counters.GetCounters(ERequestType::RT_FORGET_OPERATION), SelfId(), Connector, OperationId));
+        SendForgetOperation();
     }
 
     STRICT_STFUNC(StateFunc,
@@ -81,6 +87,10 @@ class TResourcesCleanerActor : public TBaseComputeActor<TResourcesCleanerActor>
 
     void Handle(const TEvYdbCompute::TEvForgetOperationResponse::TPtr& ev) {
         const auto& response = *ev.Get()->Get();
+        if (response.Status == NYdb::EStatus::TIMEOUT) {
+            SendForgetOperation(TDuration::MilliSeconds(BackoffTimer.NextBackoffMs()));
+            return;
+        }
         if (response.Status != NYdb::EStatus::SUCCESS && response.Status != NYdb::EStatus::NOT_FOUND) {
             LOG_E("Can't forget operation: " << ev->Get()->Issues.ToOneLineString());
             Send(Parent, new TEvYdbCompute::TEvResourcesCleanerResponse(ev->Get()->Issues, ev->Get()->Status));
@@ -98,6 +108,7 @@ class TResourcesCleanerActor : public TBaseComputeActor<TResourcesCleanerActor>
     TActorId Connector;
     NYdb::TOperation::TOperationId OperationId;
     TCounters Counters;
+    NKikimr::TBackoffTimer BackoffTimer;
 };
 
 std::unique_ptr<NActors::IActor> CreateResourcesCleanerActor(const TRunActorParams& params,

ydb/core/kqp/proxy_service/kqp_script_executions.cpp

@@ -478,6 +478,8 @@ class TScriptLeaseUpdater : public TQueryBase {
 
 class TScriptLeaseUpdateActor : public TActorBootstrapped<TScriptLeaseUpdateActor> {
 public:
+    using TLeaseUpdateRetryActor = TQueryRetryActor<TScriptLeaseUpdater, TEvScriptLeaseUpdateResponse, TString, TString, TDuration>;
+
     TScriptLeaseUpdateActor(const TActorId& runScriptActorId, const TString& database, const TString& executionId, TDuration leaseDuration, TIntrusivePtr<TKqpCounters> counters)
         : RunScriptActorId(runScriptActorId)
         , Database(database)
@@ -488,7 +490,11 @@ class TScriptLeaseUpdateActor : public TActorBootstrapped<TScriptLeaseUpdateActo
     {}
 
     void Bootstrap() {
-        Register(new TQueryRetryActor<TScriptLeaseUpdater, TEvScriptLeaseUpdateResponse, TString, TString, TDuration>(SelfId(), Database, ExecutionId, LeaseDuration, LeaseDuration / 2));
+        Register(new TLeaseUpdateRetryActor(
+            SelfId(),
+            TLeaseUpdateRetryActor::IRetryPolicy::GetExponentialBackoffPolicy(TLeaseUpdateRetryActor::Retryable, TDuration::MilliSeconds(10), TDuration::MilliSeconds(200), TDuration::Seconds(1), std::numeric_limits<size_t>::max(), LeaseDuration / 2),
+            Database, ExecutionId, LeaseDuration
+        ));
         Become(&TScriptLeaseUpdateActor::StateFunc);
     }
 
@@ -826,7 +832,6 @@ class TCheckLeaseStatusActor : public TCheckLeaseStatusActorBase {
 
 class TForgetScriptExecutionOperationQueryActor : public TQueryBase {
     static constexpr i64 MAX_NUMBER_ROWS_IN_BATCH = 100000;
-    static constexpr TDuration MINIMAL_DEADLINE_TIME = TDuration::Seconds(1);
 
     struct TResultSetDescription {
         i64 MaxRowId;
@@ -837,7 +842,7 @@ class TForgetScriptExecutionOperationQueryActor : public TQueryBase {
     TForgetScriptExecutionOperationQueryActor(const TString& executionId, const TString& database, TInstant operationDeadline)
         : ExecutionId(executionId)
         , Database(database)
-        , Deadline(operationDeadline - MINIMAL_DEADLINE_TIME)
+        , Deadline(operationDeadline)
     {}
 
     void OnRunQuery() override {
@@ -964,10 +969,14 @@ class TForgetScriptExecutionOperationQueryActor : public TQueryBase {
         Send(Owner, new TEvForgetScriptExecutionOperationResponse(status, std::move(issues)));
     }
 
+    static NYql::TIssues ForgetOperationTimeoutIssues() {
+        return { NYql::TIssue("Forget script execution operation timeout") };
+    }
+
 private:
     bool CheckDeadline() {
         if (TInstant::Now() >= Deadline) {
-            Finish(Ydb::StatusIds::TIMEOUT, "Forget script execution operation timeout");
+            Finish(Ydb::StatusIds::TIMEOUT, ForgetOperationTimeoutIssues());
             return false;
         }
         return true;
@@ -982,6 +991,8 @@ class TForgetScriptExecutionOperationQueryActor : public TQueryBase {
 
 class TForgetScriptExecutionOperationActor : public TActorBootstrapped<TForgetScriptExecutionOperationActor> {
 public:
+    using TForgetOperationRetryActor = TQueryRetryActor<TForgetScriptExecutionOperationQueryActor, TEvForgetScriptExecutionOperationResponse, TString, TString, TInstant>;
+
     explicit TForgetScriptExecutionOperationActor(TEvForgetScriptExecutionOperation::TPtr ev)
         : Request(std::move(ev))
     {}
@@ -1017,7 +1028,18 @@ class TForgetScriptExecutionOperationActor : public TActorBootstrapped<TForgetSc
             }
         }
 
-        Register(new TForgetScriptExecutionOperationQueryActor(ExecutionId, Request->Get()->Database, Request->Get()->Deadline));
+        TDuration minDelay = TDuration::MilliSeconds(10);
+        TDuration maxTime = Request->Get()->Deadline - TInstant::Now() - TDuration::Seconds(1);
+        if (maxTime <= minDelay) {
+            Reply(Ydb::StatusIds::TIMEOUT, TForgetScriptExecutionOperationQueryActor::ForgetOperationTimeoutIssues());
+            return;
+        }
+
+        Register(new TForgetOperationRetryActor(
+            SelfId(),
+            TForgetOperationRetryActor::IRetryPolicy::GetExponentialBackoffPolicy(TForgetOperationRetryActor::Retryable, minDelay, TDuration::MilliSeconds(200), TDuration::Seconds(1), std::numeric_limits<size_t>::max(), maxTime),
+            ExecutionId, Request->Get()->Database, TInstant::Now() + maxTime
+        ));
     }
 
     void Handle(TEvForgetScriptExecutionOperationResponse::TPtr& ev) {

ydb/core/kqp/ut/service/kqp_qs_scripts_ut.cpp

@@ -394,14 +394,13 @@ Y_UNIT_TEST_SUITE(KqpQueryServiceScripts) {
         i32 successCount = 0;
         for (auto& f : forgetFutures) {
             auto forgetStatus = f.ExtractValueSync();
-            UNIT_ASSERT_C(forgetStatus.GetStatus() == NYdb::EStatus::SUCCESS || forgetStatus.GetStatus() == NYdb::EStatus::NOT_FOUND ||
-                          forgetStatus.GetStatus() == NYdb::EStatus::ABORTED, forgetStatus.GetIssues().ToString());
+            UNIT_ASSERT_C(forgetStatus.GetStatus() == NYdb::EStatus::SUCCESS || forgetStatus.GetStatus() == NYdb::EStatus::NOT_FOUND, forgetStatus.GetIssues().ToString());
             if (forgetStatus.GetStatus() == NYdb::EStatus::SUCCESS) {
                 ++successCount;
             }
         }
 
-        UNIT_ASSERT(successCount == 1);
+        UNIT_ASSERT(successCount >= 1);
 
         auto op = opClient.Get<NYdb::NQuery::TScriptExecutionOperation>(scriptExecutionOperation.Id()).ExtractValueSync();
         auto forgetStatus = opClient.Forget(scriptExecutionOperation.Id()).ExtractValueSync();

ydb/library/query_actor/query_actor.h

@@ -175,12 +175,11 @@ class TQueryRetryActor : public NActors::TActorBootstrapped<TQueryRetryActor<TQu
     using TBase = NActors::TActorBootstrapped<TQueryRetryActor<TQueryActor, TResponse, TArgs...>>;
     using IRetryPolicy = IRetryPolicy<Ydb::StatusIds::StatusCode>;
 
-    explicit TQueryRetryActor(const NActors::TActorId& replyActorId, const TArgs&... args, TDuration maxRetryTime = TDuration::Seconds(1))
+    explicit TQueryRetryActor(const NActors::TActorId& replyActorId, const TArgs&... args)
         : ReplyActorId(replyActorId)
         , RetryPolicy(IRetryPolicy::GetExponentialBackoffPolicy(
-            Retryable, TDuration::MilliSeconds(10), 
-            TDuration::MilliSeconds(200), TDuration::Seconds(1),
-            std::numeric_limits<size_t>::max(), maxRetryTime
+            Retryable, TDuration::MilliSeconds(10),
+            TDuration::MilliSeconds(200), TDuration::Seconds(30), 5
         ))
         , CreateQueryActor([=]() {
             return new TQueryActor(args...);
@@ -193,6 +192,7 @@ class TQueryRetryActor : public NActors::TActorBootstrapped<TQueryRetryActor<TQu
         , CreateQueryActor([=]() {
             return new TQueryActor(args...);
         })
+        , RetryState(RetryPolicy->CreateRetryState())
     {}
 
     void StartQueryActor() const {