stable-25-3: add backoff on retry when s3 is unavailable (#26406)

Author: ydbot

ydb/core/kqp/ut/olap/tiering_ut.cpp

@@ -457,11 +457,9 @@ Y_UNIT_TEST_SUITE(KqpOlapTiering) {
         putController->WaitActualization(TDuration::Seconds(5));
         Sleep(TDuration::Seconds(5));
 
-        UNIT_ASSERT_C(putController->GetAbortedWrites() > 20,
-            "Expected load spike, but only " << putController->GetAbortedWrites() << " PutObject requests recorded");   // comment after fix
-        // UNIT_ASSERT_C(putController->GetAbortedWrites() < 10,
-        //               "Expected load spike, but was "
-        //               << putController->GetAbortedWrites() << " PutObject requests recorded"); // uncomment after fix
+        UNIT_ASSERT_C(putController->GetAbortedWrites() < 10,
+                      "Expected reduced load, but was "
+                      << putController->GetAbortedWrites() << " PutObject requests recorded");
     }
 
     Y_UNIT_TEST(ReconfigureTier) {

ydb/core/wrappers/abstract.h

@@ -1,6 +1,10 @@
 #pragma once
 
 #include <ydb/library/actors/core/actorsystem.h>
+#include <ydb/library/actors/core/log.h>
+#include <ydb/core/util/backoff.h>
+#include <ydb/core/wrappers/retry_policy.h>
+#include <util/system/mutex.h>
 
 #include <ydb/core/protos/s3_settings.pb.h>
 #include <ydb/core/wrappers/events/abstract.h>
@@ -41,6 +45,25 @@ struct TEvExternalStorage {
 
 namespace NExternalStorage {
 
+class TThreadSafeBackoff {
+private:
+    mutable TMutex Mutex;
+    NKikimr::TBackoff Policy;
+public:
+    TThreadSafeBackoff(size_t maxRetries = 100, TDuration initial = TDuration::Seconds(3), TDuration max = TDuration::Seconds(10))
+        : Policy(maxRetries, initial, max) {}
+    void Reset() {
+        with_lock(Mutex) {
+            Policy.Reset();
+        }
+    }
+    TDuration Next() {
+        with_lock(Mutex) {
+            return Policy.Next();
+        }
+    }
+};
+
 class IReplyAdapter {
 private:
     std::optional<NActors::TActorId> CustomRecipient;
@@ -69,10 +92,12 @@ class IReplyAdapter {
 class TReplyAdapterContainer {
 private:
     IReplyAdapter::TPtr Adapter;
+    std::shared_ptr<TThreadSafeBackoff> Backoff;
 public:
     TReplyAdapterContainer() = default;
-    TReplyAdapterContainer(IReplyAdapter::TPtr adapter)
-        : Adapter(adapter) {
+    TReplyAdapterContainer(IReplyAdapter::TPtr adapter, std::shared_ptr<TThreadSafeBackoff> backoff = {})
+        : Adapter(adapter)
+        , Backoff(std::move(backoff)) {
 
     }
 
@@ -95,10 +120,31 @@ class TReplyAdapterContainer {
 
     template <class TBaseEventObject>
     void Reply(const NActors::TActorId& recipientId, std::unique_ptr<TBaseEventObject>&& ev) const {
+        bool doBackoff = false;
+        TDuration delay = TDuration::Zero();
+
+        if (ev->IsSuccess()) {
+            
+            Backoff->Reset();
+        } else if (NWrappers::ShouldBackoff(ev->GetError())) {
+            AFL_VERIFY(Backoff);
+            delay = Backoff->Next();
+            doBackoff = delay > TDuration::Zero();
+        }
+
+        std::unique_ptr<NActors::IEventBase> finalEvent;
         if (Adapter) {
-            TlsActivationContext->ActorSystem()->Send(Adapter->GetRecipient(recipientId), Adapter->RebuildReplyEvent(std::move(ev)).release());
+            finalEvent = Adapter->RebuildReplyEvent(std::move(ev));
+        } else {
+            finalEvent.reset(ev.release());
+        }
+
+        const NActors::TActorId recipient = Adapter ? Adapter->GetRecipient(recipientId) : recipientId;
+        if (doBackoff) {
+            auto* handle = new NActors::IEventHandle(recipient, NActors::TActorId(), finalEvent.release());
+            TlsActivationContext->ActorSystem()->Schedule(delay, handle);
         } else {
-            TlsActivationContext->ActorSystem()->Send(recipientId, ev.release());
+            TlsActivationContext->ActorSystem()->Send(recipient, finalEvent.release());
         }
 
     }
@@ -108,14 +154,15 @@ class TReplyAdapterContainer {
 class IExternalStorageOperator {
 protected:
     TReplyAdapterContainer ReplyAdapter;
+    std::shared_ptr<TThreadSafeBackoff> BackoffPolicy = std::make_shared<TThreadSafeBackoff>();
     virtual TString DoDebugString() const {
         return "";
     }
 public:
     using TPtr = std::shared_ptr<IExternalStorageOperator>;
     void InitReplyAdapter(IReplyAdapter::TPtr adapter) {
         Y_ABORT_UNLESS(!ReplyAdapter);
-        ReplyAdapter = TReplyAdapterContainer(adapter);
+        ReplyAdapter = TReplyAdapterContainer(adapter, BackoffPolicy);
     }
 
 

ydb/core/wrappers/retry_policy.cpp

@@ -14,13 +14,30 @@ bool ShouldRetry(const Aws::S3::S3Error& error) {
     }
 
     const auto& exceptionName = error.GetExceptionName();
-    if ("TooManyRequests" == exceptionName) {
-        return true;
-    } else if ("OperationAborted" == exceptionName) {
+    if (exceptionName == "TooManyRequests" ||
+        exceptionName == "OperationAborted") {
         return true;
     }
 
     return false;
 }
 
+bool ShouldBackoff(const Aws::S3::S3Error& error) {
+    if (ShouldRetry(error)) {
+        return true;
+    }
+
+    const auto& exceptionName = error.GetExceptionName();
+    if (exceptionName == "AccessDenied" ||
+        exceptionName == "InvalidAccessKeyId" ||
+        exceptionName == "InvalidToken" ||
+        exceptionName == "ExpiredToken" ||
+        exceptionName == "AuthFailure" ||
+        exceptionName == "ServiceUnavailable")
+    {
+        return true;
+    }
+
+    return false;
+}
 }

ydb/core/wrappers/retry_policy.h

@@ -7,5 +7,6 @@ namespace Aws::S3 {
 namespace NKikimr::NWrappers {
 
 bool ShouldRetry(const Aws::S3::S3Error& error);
+bool ShouldBackoff(const Aws::S3::S3Error& error);
 
 }

ydb/core/wrappers/unavailable_storage.h

@@ -13,7 +13,7 @@ class TUnavailableExternalStorageOperator: public IExternalStorageOperator {
     template <class TResponse, class TRequestPtr>
     void ExecuteImpl(TRequestPtr& ev) const {
         const Aws::S3::S3Error error = Aws::S3::S3Error(
-            Aws::Client::AWSError<Aws::Client::CoreErrors>(Aws::Client::CoreErrors::SERVICE_UNAVAILABLE, Exception, Reason, false));
+            Aws::Client::AWSError<Aws::Client::CoreErrors>(Aws::Client::CoreErrors::SERVICE_UNAVAILABLE, Exception, Reason, true));
         std::unique_ptr<TResponse> response;
         constexpr bool hasKey = requires(const TRequestPtr& r) { r->Get()->GetRequest().GetKey(); };
         constexpr bool hasRange = std::is_same_v<TResponse, TEvGetObjectResponse>;