chore(coprocessor): expose s3 retry policy configs

Author: goshawk-3

coprocessor/fhevm-engine/Cargo.lock

@@ -39,6 +39,7 @@ num-traits = "0.2.19"
 aws-sdk-s3 = "1.78.0"
 bytesize = "2.0.1"
 futures = "0.3.31"
+humantime = "2.2.0"
 
 # local dependencies
 fhevm-engine-common = { path = "../fhevm-engine-common" }

coprocessor/fhevm-engine/sns-executor/Cargo.toml

@@ -1,5 +1,7 @@
 use fhevm_engine_common::telemetry;
-use sns_executor::{compute_128bit_ct, process_s3_uploads, Config, DBConfig, HandleItem, S3Config};
+use sns_executor::{
+    compute_128bit_ct, process_s3_uploads, Config, DBConfig, HandleItem, S3Config, S3RetryPolicy,
+};
 use tokio::{signal::unix, spawn, sync::mpsc};
 use tokio_util::sync::CancellationToken;
 use tracing::{error, Level};
@@ -35,8 +37,14 @@ fn construct_config() -> Config {
         s3: S3Config {
             bucket_ct128: args.bucket_name_ct128,
             bucket_ct64: args.bucket_name_ct64,
-            max_concurrent_uploads: args.max_concurrent_uploads,
-            retry_policy: S3Config::default().retry_policy, // TODO:
+            max_concurrent_uploads: args.s3_max_concurrent_uploads,
+            retry_policy: S3RetryPolicy {
+                max_retries_per_upload: args.s3_max_retries_per_upload,
+                max_backoff: args.s3_max_backoff,
+                max_retries_timeout: args.s3_max_retries_timeout,
+                recheck_duration: args.s3_recheck_duration,
+                regular_recheck_duration: args.s3_recheck_duration,
+            },
         },
     }
 }

coprocessor/fhevm-engine/sns-executor/src/bin/sns_worker.rs

@@ -1,4 +1,7 @@
+use std::time::Duration;
+
 use clap::{command, Parser};
+use humantime::parse_duration;
 
 #[derive(Parser, Debug, Clone)]
 #[command(version, about, long_about = None)]
@@ -52,7 +55,22 @@ pub struct Args {
 
     /// Maximum number of concurrent uploads to S3
     #[arg(long, default_value_t = 100)]
-    pub max_concurrent_uploads: u32,
+    pub s3_max_concurrent_uploads: u32,
+
+    #[arg(long, default_value_t = 100)]
+    pub s3_max_retries_per_upload: u32,
+
+    #[arg(long, default_value = "10s", value_parser = parse_duration)]
+    pub s3_max_backoff: Duration,
+
+    #[arg(long, default_value = "120s", value_parser = parse_duration)]
+    pub s3_max_retries_timeout: Duration,
+
+    #[arg(long, default_value = "2s", value_parser = parse_duration)]
+    pub s3_recheck_duration: Duration,
+
+    #[arg(long, default_value = "120s", value_parser = parse_duration)]
+    pub s3_regular_recheck_duration: Duration,
 }
 
 pub fn parse_args() -> Args {

coprocessor/fhevm-engine/sns-executor/src/bin/utils/daemon_cli.rs

@@ -282,7 +282,7 @@ async fn process_tasks(
         // Enqueue the task for upload in DB
         task.enqueue_upload_task(db_txn).await?;
 
-        // Start uploading the ciphertexts
+        // Start uploading the ciphertexts as soon as the ct128 is computed
         //
         // The service must continue running the squashed noise algorithm,
         // regardless of the availability of the upload worker.

coprocessor/fhevm-engine/sns-executor/src/executor.rs

@@ -43,7 +43,7 @@ pub struct S3Config {
     pub retry_policy: S3RetryPolicy,
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone, Debug, Default)]
 pub struct S3RetryPolicy {
     pub max_retries_per_upload: u32,
     pub max_backoff: Duration,
@@ -52,18 +52,6 @@ pub struct S3RetryPolicy {
     pub regular_recheck_duration: Duration,
 }
 
-impl Default for S3RetryPolicy {
-    fn default() -> Self {
-        Self {
-            max_retries_per_upload: 60,
-            max_backoff: Duration::from_secs(10),
-            max_retries_timeout: Duration::from_secs(2 * 60),
-            recheck_duration: Duration::from_secs(1),
-            regular_recheck_duration: Duration::from_secs(60),
-        }
-    }
-}
-
 #[derive(Clone)]
 pub struct Config {
     pub tenant_api_key: String,