feature: implement file upload retry and configure it in builder

Author: sfauvel

mithril-aggregator/src/dependency_injection/builder.rs

@@ -67,7 +67,8 @@ use crate::{
     entities::AggregatorEpochSettings,
     event_store::{EventMessage, EventStore, TransmitterService},
     file_uploaders::{
-        CloudRemotePath, FileUploader, GcpBackendUploader, GcpUploader, LocalUploader,
+        CloudRemotePath, FileUploadRetryPolicy, FileUploader, GcpBackendUploader, GcpUploader,
+        LocalUploader,
     },
     http_server::{
         routes::router::{self, RouterConfig, RouterState},
@@ -1235,14 +1236,15 @@ impl DependenciesBuilder {
                 DependenciesBuilderError::MissingConfiguration("snapshot_bucket_name".to_string())
             })?;
 
-        Ok(GcpUploader::new(
+        Ok(GcpUploader::with_retry_policy(
             Arc::new(GcpBackendUploader::try_new(
                 bucket,
                 self.configuration.snapshot_use_cdn_domain,
                 logger.clone(),
             )?),
             remote_folder_path,
             allow_overwrite,
+            FileUploadRetryPolicy::default(),
         ))
     }
 
@@ -1279,9 +1281,10 @@ impl DependenciesBuilder {
                         }
                     })?;
 
-                    Ok(vec![Arc::new(LocalUploader::new(
+                    Ok(vec![Arc::new(LocalUploader::with_retry_policy(
                         ancillary_url_prefix,
                         &target_dir,
+                        FileUploadRetryPolicy::default(),
                         logger,
                     )?)])
                 }
@@ -1317,9 +1320,10 @@ impl DependenciesBuilder {
                         .join(CARDANO_DB_ARTIFACTS_DIR)
                         .join("immutable");
 
-                    Ok(vec![Arc::new(LocalUploader::new(
+                    Ok(vec![Arc::new(LocalUploader::with_retry_policy(
                         immutable_url_prefix,
                         &target_dir,
+                        FileUploadRetryPolicy::default(),
                         logger,
                     )?)])
                 }
@@ -1360,9 +1364,10 @@ impl DependenciesBuilder {
                         }
                     })?;
 
-                    Ok(vec![Arc::new(LocalUploader::new(
+                    Ok(vec![Arc::new(LocalUploader::with_retry_policy(
                         digests_url_prefix,
                         &target_dir,
+                        FileUploadRetryPolicy::default(),
                         logger,
                     )?)])
                 }

mithril-aggregator/src/file_uploaders/dumb_uploader.rs

@@ -1,23 +1,35 @@
 use anyhow::anyhow;
 use async_trait::async_trait;
 use mithril_common::{entities::FileUri, StdResult};
-use std::{path::Path, sync::RwLock};
+use std::{path::Path, sync::RwLock, time::Duration};
 
 use crate::file_uploaders::FileUploader;
 
+use super::interface::FileUploadRetryPolicy;
+
 /// Dummy uploader for test purposes.
 ///
 /// It actually does NOT upload any file but remembers the last file it
 /// was asked to upload. This is intended to by used by integration tests.
 pub struct DumbUploader {
     last_uploaded: RwLock<Option<FileUri>>,
+    retry_policy: FileUploadRetryPolicy,
 }
 
 impl DumbUploader {
     /// Create a new instance.
     pub fn new() -> Self {
         Self {
             last_uploaded: RwLock::new(None),
+            retry_policy: FileUploadRetryPolicy::never(),
+        }
+    }
+
+    /// Create a new instance with a custom retry policy.
+    pub fn with_retry_policy(retry_policy: FileUploadRetryPolicy) -> Self {
+        Self {
+            last_uploaded: RwLock::new(None),
+            retry_policy,
         }
     }
 
@@ -52,6 +64,10 @@ impl FileUploader for DumbUploader {
 
         Ok(location)
     }
+
+    fn retry_policy(&self) -> FileUploadRetryPolicy {
+        self.retry_policy.clone()
+    }
 }
 
 #[cfg(test)]
@@ -77,4 +93,17 @@ mod tests {
                 .expect("getting dumb uploader last value after a fake download should not fail")
         );
     }
+
+    #[tokio::test]
+    async fn retry_policy_from_file_uploader_trait_should_be_implemented() {
+        let expected_policy = FileUploadRetryPolicy {
+            attempts: 10,
+            delay_between_attempts: Duration::from_millis(123),
+        };
+
+        let uploader: Box<dyn FileUploader> =
+            Box::new(DumbUploader::with_retry_policy(expected_policy.clone()));
+
+        assert_eq!(expected_policy, uploader.retry_policy());
+    }
 }

mithril-aggregator/src/file_uploaders/gcp_uploader.rs

@@ -17,6 +17,8 @@ use mithril_common::{entities::FileUri, logging::LoggerExtensions, StdResult};
 
 use crate::FileUploader;
 
+use super::FileUploadRetryPolicy;
+
 /// CloudRemotePath represents a cloud remote path
 #[derive(Debug, Clone, PartialEq)]
 pub struct CloudRemotePath(PathBuf);
@@ -198,6 +200,7 @@ pub struct GcpUploader {
     cloud_backend_uploader: Arc<dyn CloudBackendUploader>,
     remote_folder: CloudRemotePath,
     allow_overwrite: bool,
+    retry_policy: FileUploadRetryPolicy,
 }
 
 impl GcpUploader {
@@ -211,6 +214,22 @@ impl GcpUploader {
             cloud_backend_uploader,
             remote_folder,
             allow_overwrite,
+            retry_policy: FileUploadRetryPolicy::never(),
+        }
+    }
+
+    /// Create a new instance with a custom retry policy.
+    pub fn with_retry_policy(
+        cloud_backend_uploader: Arc<dyn CloudBackendUploader>,
+        remote_folder: CloudRemotePath,
+        allow_overwrite: bool,
+        retry_policy: FileUploadRetryPolicy,
+    ) -> Self {
+        Self {
+            cloud_backend_uploader,
+            remote_folder,
+            allow_overwrite,
+            retry_policy,
         }
     }
 }
@@ -242,11 +261,17 @@ impl FileUploader for GcpUploader {
 
         Ok(file_uri)
     }
+
+    fn retry_policy(&self) -> FileUploadRetryPolicy {
+        self.retry_policy.clone()
+    }
 }
 
 #[cfg(test)]
 mod tests {
-    use crate::test_tools::TestLogger;
+    use std::time::Duration;
+
+    use crate::{file_uploaders::FileUploadRetryPolicy, test_tools::TestLogger};
 
     use super::*;
 
@@ -480,4 +505,21 @@ mod tests {
             assert_eq!(FileUri(expected_location), location);
         }
     }
+
+    #[tokio::test]
+    async fn retry_policy_from_file_uploader_trait_should_be_implemented() {
+        let expected_policy = FileUploadRetryPolicy {
+            attempts: 10,
+            delay_between_attempts: Duration::from_millis(123),
+        };
+
+        let file_uploader: Box<dyn FileUploader> = Box::new(GcpUploader::with_retry_policy(
+            Arc::new(MockCloudBackendUploader::new()),
+            CloudRemotePath::new("remote_folder"),
+            true,
+            expected_policy.clone(),
+        ));
+
+        assert_eq!(expected_policy, file_uploader.retry_policy());
+    }
 }

mithril-aggregator/src/file_uploaders/interface.rs

@@ -1,19 +1,19 @@
 use async_trait::async_trait;
 use mithril_common::{entities::FileUri, StdResult};
-use std::{
-    any::{Any, TypeId},
-    path::Path,
-    time::Duration,
-};
+use std::{any::Any, path::Path, time::Duration};
 
 /// Policy for retrying file uploads.
+#[derive(Debug, PartialEq, Clone)]
 pub struct FileUploadRetryPolicy {
-    attempts: usize,
-    delay_between_attempts: Duration,
+    /// Number of attempts to upload a file.
+    pub attempts: usize,
+    /// Delay between two attempts.
+    pub delay_between_attempts: Duration,
 }
 
 impl FileUploadRetryPolicy {
-    fn never() -> Self {
+    /// Create a policy that never retries.
+    pub fn never() -> Self {
         Self {
             attempts: 1,
             delay_between_attempts: Duration::from_secs(0),
@@ -22,6 +22,7 @@ impl FileUploadRetryPolicy {
 }
 
 impl Default for FileUploadRetryPolicy {
+    /// Create a default retry policy.
     fn default() -> Self {
         Self {
             attempts: 3,
@@ -38,11 +39,12 @@ pub trait FileUploader: Any + Sync + Send {
     /// Try to upload once.
     async fn upload_without_retry(&self, filepath: &Path) -> StdResult<FileUri>;
 
+    /// Get the retry policy for this uploader.
     fn retry_policy(&self) -> FileUploadRetryPolicy {
         FileUploadRetryPolicy::never()
     }
 
-    // Upload a file
+    /// Upload a file with retries according to the retry policy.
     async fn upload(&self, filepath: &Path) -> StdResult<FileUri> {
         let retry_policy = self.retry_policy();
 

mithril-aggregator/src/file_uploaders/local_uploader.rs

@@ -6,7 +6,7 @@ use std::path::{Path, PathBuf};
 use mithril_common::StdResult;
 use mithril_common::{entities::FileUri, logging::LoggerExtensions};
 
-use crate::file_uploaders::FileUploader;
+use crate::file_uploaders::{FileUploadRetryPolicy, FileUploader};
 use crate::tools::url_sanitizer::SanitizedUrlWithTrailingSlash;
 
 /// LocalUploader is a file uploader working using local files
@@ -17,6 +17,7 @@ pub struct LocalUploader {
     /// Target folder where to store files archive
     target_location: PathBuf,
 
+    retry_policy: FileUploadRetryPolicy,
     logger: Logger,
 }
 
@@ -26,13 +27,28 @@ impl LocalUploader {
         server_url_prefix: SanitizedUrlWithTrailingSlash,
         target_location: &Path,
         logger: Logger,
+    ) -> StdResult<Self> {
+        Self::with_retry_policy(
+            server_url_prefix,
+            target_location,
+            FileUploadRetryPolicy::never(),
+            logger,
+        )
+    }
+
+    pub(crate) fn with_retry_policy(
+        server_url_prefix: SanitizedUrlWithTrailingSlash,
+        target_location: &Path,
+        retry_policy: FileUploadRetryPolicy,
+        logger: Logger,
     ) -> StdResult<Self> {
         let logger = logger.new_with_component_name::<Self>();
         debug!(logger, "New LocalUploader created"; "server_url_prefix" => &server_url_prefix.as_str());
 
         Ok(Self {
             server_url_prefix,
             target_location: target_location.to_path_buf(),
+            retry_policy,
             logger,
         })
     }
@@ -56,13 +72,18 @@ impl FileUploader for LocalUploader {
         );
         Ok(FileUri(location))
     }
+
+    fn retry_policy(&self) -> FileUploadRetryPolicy {
+        self.retry_policy.clone()
+    }
 }
 
 #[cfg(test)]
 mod tests {
     use std::fs::File;
     use std::io::Write;
     use std::path::{Path, PathBuf};
+    use std::time::Duration;
 
     use mithril_common::test_utils::TempDir;
 
@@ -152,4 +173,25 @@ mod tests {
             .await
             .expect_err("Uploading a directory should fail");
     }
+
+    #[tokio::test]
+    async fn retry_policy_from_file_uploader_trait_should_be_implemented() {
+        let target_dir = TempDir::create("local_uploader", "test_retry_policy");
+        let expected_policy = FileUploadRetryPolicy {
+            attempts: 10,
+            delay_between_attempts: Duration::from_millis(123),
+        };
+
+        let uploader: Box<dyn FileUploader> = Box::new(
+            LocalUploader::with_retry_policy(
+                SanitizedUrlWithTrailingSlash::parse("http://test.com:8080/base-root/").unwrap(),
+                &target_dir,
+                expected_policy.clone(),
+                TestLogger::stdout(),
+            )
+            .unwrap(),
+        );
+
+        assert_eq!(expected_policy, uploader.retry_policy());
+    }
 }

mithril-aggregator/src/file_uploaders/mod.rs

@@ -6,7 +6,7 @@ mod local_uploader;
 
 pub use dumb_uploader::*;
 pub use gcp_uploader::{CloudRemotePath, GcpBackendUploader, GcpUploader};
-pub use interface::FileUploader;
+pub use interface::{FileUploadRetryPolicy, FileUploader};
 pub use local_snapshot_uploader::LocalSnapshotUploader;
 pub use local_uploader::LocalUploader;