fix: make epoxy activities retry forever

Author: MasterPtato

packages/common/gasoline/core/src/activity.rs

@@ -12,6 +12,7 @@ pub trait Activity {
 	type Output: Serialize + DeserializeOwned + Debug + Send;
 
 	const NAME: &'static str;
+	/// Seconds.
 	const MAX_RETRIES: usize;
 	const TIMEOUT: std::time::Duration;
 

packages/common/gasoline/core/src/ctx/workflow.rs

@@ -612,23 +612,23 @@ impl WorkflowCtx {
 						// errors
 						let err = match err {
 							WorkflowError::ActivityFailure(err, _) => {
-								if error_count + 1 >= I::Activity::MAX_RETRIES {
+								if error_count.saturating_add(1) >= I::Activity::MAX_RETRIES {
 									WorkflowError::ActivityMaxFailuresReached(err)
 								} else {
 									// Add error count to the error for backoff calculation
 									WorkflowError::ActivityFailure(err, error_count)
 								}
 							}
 							WorkflowError::ActivityTimeout(_) => {
-								if error_count + 1 >= I::Activity::MAX_RETRIES {
+								if error_count.saturating_add(1) >= I::Activity::MAX_RETRIES {
 									WorkflowError::ActivityMaxFailuresReached(err.into())
 								} else {
 									// Add error count to the error for backoff calculation
 									WorkflowError::ActivityTimeout(error_count)
 								}
 							}
 							WorkflowError::OperationTimeout(_) => {
-								if error_count + 1 >= I::Activity::MAX_RETRIES {
+								if error_count.saturating_add(1) >= I::Activity::MAX_RETRIES {
 									WorkflowError::ActivityMaxFailuresReached(err.into())
 								} else {
 									// Add error count to the error for backoff calculation

packages/common/gasoline/macros/src/lib.rs

@@ -7,17 +7,25 @@ use syn::{
 	spanned::Spanned,
 };
 
+#[derive(Default)]
 struct Config {
-	max_retries: usize,
-	timeout: u64,
+	max_retries: Option<syn::Expr>,
+	timeout: Option<syn::Expr>,
 }
 
-impl Default for Config {
-	fn default() -> Self {
-		Config {
-			max_retries: 5,
-			timeout: 30,
-		}
+impl Config {
+	fn max_retries(&self) -> proc_macro2::TokenStream {
+		self.max_retries
+			.as_ref()
+			.map(|e| e.to_token_stream())
+			.unwrap_or_else(|| quote! { 5 })
+	}
+
+	fn timeout(&self) -> proc_macro2::TokenStream {
+		self.timeout
+			.as_ref()
+			.map(|e| e.to_token_stream())
+			.unwrap_or_else(|| quote! { 30 })
 	}
 }
 
@@ -130,8 +138,8 @@ pub fn activity(attr: TokenStream, item: TokenStream) -> TokenStream {
 	let fn_body = item_fn.block;
 	let vis = item_fn.vis;
 
-	let max_retries = config.max_retries;
-	let timeout = config.timeout;
+	let max_retries = config.max_retries();
+	let timeout = config.timeout();
 
 	let expanded = quote! {
 		#vis struct #struct_ident;
@@ -201,7 +209,7 @@ pub fn operation(attr: TokenStream, item: TokenStream) -> TokenStream {
 	let fn_body = item_fn.block;
 	let vis = item_fn.vis;
 
-	let timeout = config.timeout;
+	let timeout = config.timeout();
 
 	let expanded = quote! {
 		#vis struct #struct_ident;
@@ -441,12 +449,13 @@ fn parse_config(attrs: &[syn::Attribute]) -> syn::Result<Config> {
 
 		// Verify config property
 		if ident == "max_retries" {
-			config.max_retries =
-				syn::parse::<syn::LitInt>(name_value.value.to_token_stream().into())?
-					.base10_parse()?;
+			config.max_retries = Some(syn::parse::<syn::Expr>(
+				name_value.value.to_token_stream().into(),
+			)?);
 		} else if ident == "timeout" {
-			config.timeout = syn::parse::<syn::LitInt>(name_value.value.to_token_stream().into())?
-				.base10_parse()?;
+			config.timeout = Some(syn::parse::<syn::Expr>(
+				name_value.value.to_token_stream().into(),
+			)?);
 		} else if ident != "doc" {
 			return Err(syn::Error::new(
 				name_value.span(),

packages/services/epoxy/src/workflows/coordinator/reconfigure.rs

@@ -135,6 +135,7 @@ pub struct HealthCheckNewReplicasInput {
 }
 
 #[activity(HealthCheckNewReplicas)]
+#[max_retries = usize::MAX]
 pub async fn health_check_new_replicas(
 	ctx: &ActivityCtx,
 	input: &HealthCheckNewReplicasInput,

packages/services/epoxy/src/workflows/replica/setup.rs

@@ -740,6 +740,7 @@ pub struct NotifyActiveInput {
 }
 
 #[activity(NotifyActive)]
+#[max_retries = usize::MAX]
 pub async fn notify_active(ctx: &ActivityCtx, input: &NotifyActiveInput) -> Result<()> {
 	let config = &input.learning_config;
 	let proto_config: protocol::ClusterConfig = config.clone().into();