Import from S3: check error is retriable KIKIMR-21102 (#2198)

Author: CyberROFL

ydb/core/tx/datashard/export_s3_uploader.cpp

@@ -471,7 +471,7 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
     }
 
     void Retry() {
-        Delay = Min(Delay * ++Attempt, TDuration::Minutes(10));
+        Delay = Min(Delay * ++Attempt, MaxDelay);
         const TDuration random = TDuration::FromValue(TAppData::RandomProvider->GenRand64() % Delay.MicroSeconds());
         this->Schedule(Delay + random, new TEvents::TEvWakeup());
     }
@@ -637,8 +637,10 @@ class TS3Uploader: public TActorBootstrapped<TS3Uploader> {
     const ui32 Retries;
     ui32 Attempt;
 
-    TActorId Client;
     TDuration Delay;
+    static constexpr TDuration MaxDelay = TDuration::Minutes(10);
+
+    TActorId Client;
     bool SchemeUploaded;
     bool MetadataUploaded;
     bool MultiPart;

ydb/core/tx/datashard/import_s3.cpp

@@ -309,10 +309,6 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
 
     }; // TUploadRowsRequestBuilder
 
-    enum class EWakeupTag: ui64 {
-        Restart,
-    };
-
     void AllocateResource() {
         IMPORT_LOG_D("AllocateResource");
 
@@ -386,7 +382,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             default:
                 IMPORT_LOG_E("Error at 'HeadObject'"
                     << ": error# " << result);
-                return RestartOrFinish(result.GetError().GetMessage().c_str());
+                return RetryOrFinish(result.GetError());
             }
 
             CompressionCodec = NBackupRestoreTraits::NextCompressionCodec(CompressionCodec);
@@ -578,7 +574,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             return Finish(false, record.GetErrorDescription());
 
         default:
-            return RestartOrFinish(record.GetErrorDescription());
+            return RetryOrFinish(record.GetErrorDescription());
         }
     }
 
@@ -590,7 +586,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
 
         IMPORT_LOG_E("Error at '" << marker << "'"
             << ": error# " << result);
-        RestartOrFinish(result.GetError().GetMessage().c_str());
+        RetryOrFinish(result.GetError());
 
         return false;
     }
@@ -656,21 +652,35 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
         return true;
     }
 
-    void RestartOrFinish(const TString& error) {
-        if (Attempt++ < Retries) {
-            Delay = Min(Delay * Attempt, MaxDelay);
-            const TDuration random = TDuration::FromValue(TAppData::RandomProvider->GenRand64() % Delay.MicroSeconds());
+    static bool ShouldRetry(const Aws::S3::S3Error& error) {
+        return error.ShouldRetry();
+    }
 
-            Schedule(Delay + random, new TEvents::TEvWakeup(static_cast<ui64>(EWakeupTag::Restart)));
-        } else {
-            Finish(false, error);
-        }
+    static bool ShouldRetry(const TString&) {
+        return true;
     }
 
-    void Handle(TEvents::TEvWakeup::TPtr& ev) {
-        switch (static_cast<EWakeupTag>(ev->Get()->Tag)) {
-        case EWakeupTag::Restart:
-            return Restart();
+    template <typename T>
+    bool CanRetry(const T& error) const {
+        return Attempt < Retries && ShouldRetry(error);
+    }
+
+    void Retry() {
+        Delay = Min(Delay * ++Attempt, MaxDelay);
+        const TDuration random = TDuration::FromValue(TAppData::RandomProvider->GenRand64() % Delay.MicroSeconds());
+        Schedule(Delay + random, new TEvents::TEvWakeup());
+    }
+
+    template <typename T>
+    void RetryOrFinish(const T& error) {
+        if (CanRetry(error)) {
+            Retry();
+        } else {
+            if constexpr (std::is_same_v<T, Aws::S3::S3Error>) {
+                Finish(false, TStringBuilder() << "S3 error: " << error.GetMessage().c_str());
+            } else {
+                Finish(false, error);
+            }
         }
     }
 
@@ -754,7 +764,7 @@ class TS3Downloader: public TActorBootstrapped<TS3Downloader> {
             hFunc(TEvDataShard::TEvS3DownloadInfo, Handle);
             hFunc(TEvDataShard::TEvS3UploadRowsResponse, Handle);
 
-            hFunc(TEvents::TEvWakeup, Handle);
+            sFunc(TEvents::TEvWakeup, Restart);
             sFunc(TEvents::TEvPoisonPill, NotifyDied);
         }
     }

ydb/core/tx/schemeshard/schemeshard_import_scheme_getter.cpp

@@ -94,16 +94,17 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
         LOG_E("Error at '" << marker << "'"
             << ": self# " << SelfId()
             << ", error# " << result);
-        MaybeRetry(result.GetError().GetMessage().c_str());
+        MaybeRetry(result.GetError());
 
         return false;
     }
 
-    void MaybeRetry(const TString& error) {
-        if (Attempt++ < Retries) {
-            Schedule(TDuration::Minutes(1), new TEvents::TEvWakeup());
+    void MaybeRetry(const Aws::S3::S3Error& error) {
+        if (Attempt < Retries && error.ShouldRetry()) {
+            Delay = Min(Delay * ++Attempt, MaxDelay);
+            Schedule(Delay, new TEvents::TEvWakeup());
         } else {
-            Reply(false, error);
+            Reply(false, TStringBuilder() << "S3 error: " << error.GetMessage().c_str());
         }
     }
 
@@ -148,8 +149,8 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
             hFunc(TEvExternalStorage::TEvHeadObjectResponse, Handle);
             hFunc(TEvExternalStorage::TEvGetObjectResponse, Handle);
 
-            cFunc(TEvents::TEvWakeup::EventType, Bootstrap);
-            cFunc(TEvents::TEvPoisonPill::EventType, PassAway);
+            sFunc(TEvents::TEvWakeup, Bootstrap);
+            sFunc(TEvents::TEvPoisonPill, PassAway);
         }
     }
 
@@ -164,6 +165,9 @@ class TSchemeGetter: public TActorBootstrapped<TSchemeGetter> {
     const ui32 Retries;
     ui32 Attempt = 0;
 
+    TDuration Delay = TDuration::Minutes(1);
+    static constexpr TDuration MaxDelay = TDuration::Minutes(10);
+
     TActorId Client;
 
 }; // TSchemeGetter