feat(workflows): add retries and timeouts

docs/libraries/workflow/ERRORS.md

@@ -1,3 +1,13 @@
 # Errors
 
-Only errors from inside of activities will be retried. Errors thrown in the workflow body will not be retried because they will never succeed (the state is consistent up the point of error).
+## Retries
+
+Only errors from inside of activities will be retried. Errors thrown in the workflow body will not be retried
+because they will never succeed (the state is consistent up the point of error).
+
+## Workflow errors
+
+Sub workflow errors cannot be caught because it's up to the workflow to handle its own errors gracefully.
+
+We return OK responses from workflows for failure cases we explicitly handle (e.g. linode server provision
+cleaning itself up)

lib/chirp-workflow/core/src/activity.rs

@@ -11,7 +11,9 @@ pub trait Activity {
 	type Input: ActivityInput;
 	type Output: Serialize + DeserializeOwned + Debug + Send;
 
-	fn name() -> &'static str;
+	const NAME: &'static str;
+	const MAX_RETRIES: u32;
+	const TIMEOUT: std::time::Duration;
 
 	async fn run(ctx: &mut ActivityCtx, input: &Self::Input) -> GlobalResult<Self::Output>;
 }

lib/chirp-workflow/core/src/compat.rs

@@ -24,7 +24,7 @@ where
 		bail!("cannot dispatch a workflow from an operation within a workflow execution. trigger it from the workflow's body.");
 	}
 
-	let name = I::Workflow::name();
+	let name = I::Workflow::NAME;
 
 	tracing::debug!(%name, ?input, "dispatching workflow");
 
@@ -50,7 +50,7 @@ pub async fn wait_for_workflow<W: Workflow, B: Debug + Clone>(
 	ctx: &rivet_operation::OperationContext<B>,
 	workflow_id: Uuid,
 ) -> GlobalResult<W::Output> {
-	tracing::info!(name=W::name(), id=?workflow_id, "waiting for workflow");
+	tracing::info!(name=W::NAME, id=?workflow_id, "waiting for workflow");
 
 	let period = Duration::from_millis(50);
 	let mut interval = tokio::time::interval(period);
@@ -94,7 +94,7 @@ pub async fn signal<I: Signal + Serialize, B: Debug + Clone>(
 		bail!("cannot dispatch a signal from an operation within a workflow execution. trigger it from the workflow's body.");
 	}
 
-	tracing::debug!(name=%I::name(), %workflow_id, "dispatching signal");
+	tracing::debug!(name=%I::NAME, %workflow_id, "dispatching signal");
 
 	let signal_id = Uuid::new_v4();
 
@@ -105,7 +105,7 @@ pub async fn signal<I: Signal + Serialize, B: Debug + Clone>(
 
 	db(ctx)
 		.await?
-		.publish_signal(ctx.ray_id(), workflow_id, signal_id, I::name(), input_val)
+		.publish_signal(ctx.ray_id(), workflow_id, signal_id, I::NAME, input_val)
 		.await
 		.map_err(GlobalError::raw)?;
 
@@ -127,7 +127,7 @@ where
 		ctx.ray_id(),
 		ctx.req_ts(),
 		ctx.from_workflow(),
-		I::Operation::name(),
+		I::Operation::NAME,
 	);
 
 	I::Operation::run(&mut ctx, &input)

lib/chirp-workflow/core/src/ctx/activity.rs

@@ -66,11 +66,12 @@ impl ActivityCtx {
 			self.ray_id,
 			self.op_ctx.req_ts(),
 			true,
-			I::Operation::name(),
+			I::Operation::NAME,
 		);
 
-		I::Operation::run(&mut ctx, &input)
+		tokio::time::timeout(I::Operation::TIMEOUT, I::Operation::run(&mut ctx, &input))
 			.await
+			.map_err(|_| WorkflowError::OperationTimeout)?
 			.map_err(WorkflowError::OperationFailure)
 			.map_err(GlobalError::raw)
 	}

lib/chirp-workflow/core/src/ctx/operation.rs

@@ -67,7 +67,7 @@ impl OperationCtx {
 			self.ray_id,
 			self.op_ctx.req_ts(),
 			self.op_ctx.from_workflow(),
-			I::Operation::name(),
+			I::Operation::NAME,
 		);
 
 		I::Operation::run(&mut ctx, &input)

lib/chirp-workflow/core/src/ctx/test.rs

@@ -67,7 +67,7 @@ impl TestCtx {
 		I: WorkflowInput,
 		<I as WorkflowInput>::Workflow: Workflow<Input = I>,
 	{
-		let name = I::Workflow::name();
+		let name = I::Workflow::NAME;
 
 		tracing::debug!(%name, ?input, "dispatching workflow");
 
@@ -92,7 +92,7 @@ impl TestCtx {
 		&self,
 		workflow_id: Uuid,
 	) -> GlobalResult<W::Output> {
-		tracing::info!(name=W::name(), id=?workflow_id, "waiting for workflow");
+		tracing::info!(name=W::NAME, id=?workflow_id, "waiting for workflow");
 
 		let period = Duration::from_millis(50);
 		let mut interval = tokio::time::interval(period);
@@ -131,7 +131,7 @@ impl TestCtx {
 		workflow_id: Uuid,
 		input: I,
 	) -> GlobalResult<Uuid> {
-		tracing::debug!(name=%I::name(), %workflow_id, "dispatching signal");
+		tracing::debug!(name=%I::NAME, %workflow_id, "dispatching signal");
 
 		let signal_id = Uuid::new_v4();
 
@@ -141,7 +141,7 @@ impl TestCtx {
 			.map_err(GlobalError::raw)?;
 
 		self.db
-			.publish_signal(self.ray_id, workflow_id, signal_id, I::name(), input_val)
+			.publish_signal(self.ray_id, workflow_id, signal_id, I::NAME, input_val)
 			.await
 			.map_err(GlobalError::raw)?;
 
@@ -164,7 +164,7 @@ impl TestCtx {
 			self.ray_id,
 			self.ts,
 			false,
-			I::Operation::name(),
+			I::Operation::NAME,
 		);
 
 		I::Operation::run(&mut ctx, &input)