feat: global error raw variant

MasterPtato · MasterPtato · commit db7c1e416e2f · 2024-05-29T22:13:58.000Z
diff --git a/lib/chirp-workflow/core/Cargo.toml b/lib/chirp-workflow/core/Cargo.toml
@@ -6,7 +6,6 @@ edition = "2021"
 license = "Apache-2.0"
 
 [dependencies]
-anyhow = "1.0.82"
 async-trait = "0.1.80"
 chirp-client = { path = "../../chirp/client" }
 chirp-workflow-macros = { path = "../macros" }
@@ -32,3 +31,6 @@ tokio = { version = "1.37.0", features = ["full"] }
 tracing = "0.1.40"
 tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
 uuid = { version = "1.8.0", features = ["v4", "serde"] }
+
+[dev-dependencies]
+anyhow = "1.0.82"
diff --git a/lib/chirp-workflow/core/src/activity.rs b/lib/chirp-workflow/core/src/activity.rs
@@ -1,7 +1,7 @@
 use std::{fmt::Debug, hash::Hash};
 
-use anyhow::*;
 use async_trait::async_trait;
+use global_error::GlobalResult;
 use serde::{de::DeserializeOwned, Serialize};
 
 use crate::ActivityCtx;
@@ -13,7 +13,7 @@ pub trait Activity {
 
 	fn name() -> &'static str;
 
-	async fn run(ctx: &mut ActivityCtx, input: &Self::Input) -> Result<Self::Output>;
+	async fn run(ctx: &mut ActivityCtx, input: &Self::Input) -> GlobalResult<Self::Output>;
 }
 
 pub trait ActivityInput: Serialize + DeserializeOwned + Debug + Hash + Send {
diff --git a/lib/chirp-workflow/core/src/ctx/activity.rs b/lib/chirp-workflow/core/src/ctx/activity.rs
@@ -27,9 +27,9 @@ impl ActivityCtx {
 			name.to_string(),
 			std::time::Duration::from_secs(60),
 			conn.clone(),
-			// TODO: req_id
-			Uuid::new_v4(),
 			workflow_id,
+			// TODO: ray_id
+			Uuid::new_v4(),
 			rivet_util::timestamp::now(),
 			// TODO: req_ts
 			rivet_util::timestamp::now(),
diff --git a/lib/chirp-workflow/core/src/ctx/workflow.rs b/lib/chirp-workflow/core/src/ctx/workflow.rs
@@ -1,6 +1,5 @@
 use std::{collections::HashMap, sync::Arc};
 
-use anyhow::*;
 use serde::Serialize;
 use tokio::time::Duration;
 use uuid::Uuid;
@@ -138,7 +137,7 @@ impl WorkflowCtx {
 		}
 	}
 
-	async fn run_workflow_inner(&mut self) -> Result<()> {
+	async fn run_workflow_inner(&mut self) -> WorkflowResult<()> {
 		tracing::info!(id=%self.workflow_id, "running workflow");
 
 		// Lookup workflow
diff --git a/lib/chirp-workflow/core/src/error.rs b/lib/chirp-workflow/core/src/error.rs
@@ -1,4 +1,4 @@
-use anyhow::*;
+use global_error::GlobalError;
 use uuid::Uuid;
 
 pub type WorkflowResult<T> = Result<T, WorkflowError>;
@@ -10,13 +10,13 @@ pub type WorkflowResult<T> = Result<T, WorkflowError>;
 #[derive(thiserror::Error, Debug)]
 pub enum WorkflowError {
 	#[error("workflow failure: {0:?}")]
-	WorkflowFailure(Error),
+	WorkflowFailure(GlobalError),
 
 	#[error("activity failure: {0:?}")]
-	ActivityFailure(Error),
+	ActivityFailure(GlobalError),
 
 	#[error("operation failure: {0:?}")]
-	OperationFailure(Error),
+	OperationFailure(GlobalError),
 
 	#[error("workflow missing from registry: {0}")]
 	WorkflowMissingFromRegistry(String),
diff --git a/lib/chirp-workflow/core/src/operation.rs b/lib/chirp-workflow/core/src/operation.rs
@@ -1,5 +1,5 @@
-use anyhow::*;
 use async_trait::async_trait;
+use global_error::GlobalResult;
 
 use crate::OperationCtx;
 
@@ -10,7 +10,7 @@ pub trait Operation {
 
 	fn name() -> &'static str;
 
-	async fn run(ctx: &mut OperationCtx, input: &Self::Input) -> Result<Self::Output>;
+	async fn run(ctx: &mut OperationCtx, input: &Self::Input) -> GlobalResult<Self::Output>;
 }
 
 pub trait OperationInput: Send {
diff --git a/lib/chirp-workflow/core/src/prelude.rs b/lib/chirp-workflow/core/src/prelude.rs
@@ -28,8 +28,6 @@ pub use chirp_workflow_macros::*;
 
 // External libraries
 #[doc(hidden)]
-pub use anyhow::{self, Result};
-#[doc(hidden)]
 pub use async_trait;
 #[doc(hidden)]
 pub use futures_util;
diff --git a/lib/chirp-workflow/core/src/registry.rs b/lib/chirp-workflow/core/src/registry.rs
@@ -1,6 +1,7 @@
 use std::{collections::HashMap, future::Future, pin::Pin, sync::Arc};
 
 use futures_util::FutureExt;
+use global_error::GlobalError;
 
 use crate::{Workflow, WorkflowCtx, WorkflowError, WorkflowResult};
 
@@ -41,11 +42,20 @@ impl Registry {
 						// Run workflow
 						let output = match W::run(ctx, &input).await {
 							Ok(x) => x,
+							// Differentiate between WorkflowError and user error
 							Err(err) => {
-								// Differentiate between WorkflowError and user error
-								match err.downcast::<WorkflowError>() {
-									Ok(err) => return Err(err),
-									Err(err) => return Err(WorkflowError::WorkflowFailure(err)),
+								match err {
+									GlobalError::Raw(inner_err) => {
+										match inner_err.downcast::<WorkflowError>() {
+											Ok(inner_err) => return Err(*inner_err),
+											Err(err) => {
+												return Err(WorkflowError::WorkflowFailure(
+													GlobalError::Raw(err),
+												))
+											}
+										}
+									}
+									_ => return Err(WorkflowError::WorkflowFailure(err)),
 								}
 							}
 						};
diff --git a/lib/chirp-workflow/core/src/util.rs b/lib/chirp-workflow/core/src/util.rs
@@ -3,7 +3,7 @@ use std::{
 	time::{SystemTime, UNIX_EPOCH},
 };
 
-use anyhow::*;
+use global_error::{macros::*, GlobalResult};
 use rand::Rng;
 use tokio::time::{self, Duration};
 
@@ -16,7 +16,7 @@ const FAULT_RATE: usize = 80;
 
 pub async fn sleep_until_ts(ts: i64) {
 	let target_time = UNIX_EPOCH + Duration::from_millis(ts as u64);
-	if let std::result::Result::Ok(sleep_duration) = target_time.duration_since(SystemTime::now()) {
+	if let Ok(sleep_duration) = target_time.duration_since(SystemTime::now()) {
 		time::sleep(sleep_duration).await;
 	}
 }
@@ -100,12 +100,13 @@ pub fn combine_events(
 		.map(|(k, v)| (k, v.into_iter().map(|(_, v)| v).collect()))
 		.collect();
 
-	WorkflowResult::Ok(event_history)
+	Ok(event_history)
 }
 
-pub fn inject_fault() -> Result<()> {
+pub fn inject_fault() -> GlobalResult<()> {
 	if rand::thread_rng().gen_range(0..100) < FAULT_RATE {
 		bail!("This is a random panic!");
 	}
+
 	Ok(())
 }
diff --git a/lib/chirp-workflow/core/src/workflow.rs b/lib/chirp-workflow/core/src/workflow.rs
@@ -1,5 +1,5 @@
-use anyhow::*;
 use async_trait::async_trait;
+use global_error::GlobalResult;
 use serde::{de::DeserializeOwned, Serialize};
 use std::fmt::Debug;
 
@@ -13,7 +13,7 @@ pub trait Workflow {
 	fn name() -> &'static str;
 
 	// TODO: Is there any reason for input to be a reference?
-	async fn run(ctx: &mut WorkflowCtx, input: &Self::Input) -> Result<Self::Output>;
+	async fn run(ctx: &mut WorkflowCtx, input: &Self::Input) -> GlobalResult<Self::Output>;
 }
 
 pub trait WorkflowInput: Serialize + DeserializeOwned + Debug + Send {
diff --git a/lib/chirp-workflow/macros/src/lib.rs b/lib/chirp-workflow/macros/src/lib.rs
@@ -54,7 +54,7 @@ fn trait_fn(attr: TokenStream, item: TokenStream, opts: TraitFnOpts) -> TokenStr
 		ReturnType::Type(_, ty) => match ty.as_ref() {
 			Type::Path(path) => {
 				let segment = path.path.segments.last().unwrap();
-				if segment.ident == "Result" {
+				if segment.ident == "GlobalResult" {
 					match &segment.arguments {
 						PathArguments::AngleBracketed(args) => {
 							if let Some(GenericArgument::Type(Type::Path(path))) = args.args.first()
@@ -68,7 +68,7 @@ fn trait_fn(attr: TokenStream, item: TokenStream, opts: TraitFnOpts) -> TokenStr
 					}
 				} else {
 					panic!(
-						"{} function must return a Result type",
+						"{} function must return a GlobalResult type",
 						opts.trait_ty.to_token_stream().to_string()
 					);
 				}
@@ -124,7 +124,7 @@ fn trait_fn(attr: TokenStream, item: TokenStream, opts: TraitFnOpts) -> TokenStr
 				#fn_name
 			}
 
-			async fn run(#ctx_ident: #ctx_ty, #input_ident: &Self::Input) -> anyhow::Result<Self::Output> {
+			async fn run(#ctx_ident: #ctx_ty, #input_ident: &Self::Input) -> GlobalResult<Self::Output> {
 				#fn_body
 			}
 		}
diff --git a/lib/global-error/src/error.rs b/lib/global-error/src/error.rs
@@ -1,4 +1,4 @@
-use std::{collections::HashMap, fmt::Display, sync::Arc};
+use std::{collections::HashMap, fmt::Display};
 
 use http::StatusCode;
 use serde::Serialize;
@@ -7,7 +7,7 @@ use types::rivet::chirp;
 
 pub type GlobalResult<T> = Result<T, GlobalError>;
 
-#[derive(Debug, Clone)]
+#[derive(Debug)]
 pub enum GlobalError {
 	/// Errors thrown by any part of the code, such as from sql queries, api calls, etc.
 	Internal {
@@ -35,7 +35,7 @@ pub enum GlobalError {
 	},
 	/// Any kind of error, but stored dynamically. This is used to downcast the error back into its original
 	/// type if needed.
-	Raw(Arc<dyn std::error::Error + Send + Sync>),
+	Raw(Box<dyn std::error::Error + Send + Sync>),
 }
 
 impl Display for GlobalError {
@@ -91,7 +91,7 @@ impl GlobalError {
 	}
 
 	pub fn raw<T: std::error::Error + Send + Sync + 'static>(err: T) -> GlobalError {
-		GlobalError::Raw(Arc::new(err))
+		GlobalError::Raw(Box::new(err))
 	}
 
 	pub fn bad_request_builder(code: &'static str) -> BadRequestBuilder {
diff --git a/svc/Cargo.lock b/svc/Cargo.lock
diff --git a/svc/pkg/foo/worker/src/workflows/test.rs b/svc/pkg/foo/worker/src/workflows/test.rs
@@ -11,7 +11,7 @@ pub struct TestOutput {
 }
 
 #[workflow(Test)]
-async fn test(ctx: &mut WorkflowCtx, input: &TestInput) -> Result<TestOutput> {
+async fn test(ctx: &mut WorkflowCtx, input: &TestInput) -> GlobalResult<TestOutput> {
 	tracing::info!("input {}", input.x);
 
 	let a = ctx.activity(FooInput {}).await?;
@@ -28,29 +28,29 @@ pub struct FooOutput {
 }
 
 #[activity(Foo)]
-pub fn foo(ctx: &mut ActivityCtx, input: &FooInput) -> Result<FooOutput> {
+pub fn foo(ctx: &mut ActivityCtx, input: &FooInput) -> GlobalResult<FooOutput> {
+	chirp_workflow::util::inject_fault()?;
 	let ids = sql_fetch_all!(
 		[ctx, (Uuid,)]
 		"
 		SELECT datacenter_id
 		FROM db_cluster.datacenters
 		",
 	)
-	.await
-	.unwrap()
+	.await?
 	.into_iter()
 	.map(|(id,)| id)
 	.collect();
 
-	let user_id = util::uuid::parse("000b3124-91d9-472e-8104-3dcc41e1a74d").unwrap();
-	let user_get_res = op!([ctx] user_get {
-		user_ids: vec![user_id.into()],
-	})
-	.await
-	.unwrap();
-	let user = user_get_res.users.first().unwrap();
+	// let user_id = util::uuid::parse("000b3124-91d9-472e-8104-3dcc41e1a74d").unwrap();
+	// let user_get_res = op!([ctx] user_get {
+	// 	user_ids: vec![user_id.into()],
+	// })
+	// .await
+	// .unwrap();
+	// let user = user_get_res.users.first().unwrap();
 
-	tracing::info!(?user, "-----------");
+	// tracing::info!(?user, "-----------");
 
 	Ok(FooOutput { ids })
 }

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,5 @@`
`1`	`1`	`use std::{collections::HashMap, sync::Arc};`
`2`	`2`
`3`		`-use anyhow::*;`
`4`	`3`	`use serde::Serialize;`
`5`	`4`	`use tokio::time::Duration;`
`6`	`5`	`use uuid::Uuid;`
`@@ -138,7 +137,7 @@ impl WorkflowCtx {`
`138`	`137`	`}`
`139`	`138`	`}`
`140`	`139`
`141`		`- async fn run_workflow_inner(&mut self) -> Result<()> {`
	`140`	`+ async fn run_workflow_inner(&mut self) -> WorkflowResult<()> {`
`142`	`141`	`tracing::info!(id=%self.workflow_id, "running workflow");`
`143`	`142`
`144`	`143`	`// Lookup workflow`
Original file line number	Diff line number	Diff line change
`@@ -3,7 +3,7 @@ use std::{`
`3`	`3`	`time::{SystemTime, UNIX_EPOCH},`
`4`	`4`	`};`
`5`	`5`
`6`		`-use anyhow::*;`
	`6`	`+use global_error::{macros::*, GlobalResult};`
`7`	`7`	`use rand::Rng;`
`8`	`8`	`use tokio::time::{self, Duration};`
`9`	`9`
`@@ -16,7 +16,7 @@ const FAULT_RATE: usize = 80;`
`16`	`16`
`17`	`17`	`pub async fn sleep_until_ts(ts: i64) {`
`18`	`18`	`let target_time = UNIX_EPOCH + Duration::from_millis(ts as u64);`
`19`		`- if let std::result::Result::Ok(sleep_duration) = target_time.duration_since(SystemTime::now()) {`
	`19`	`+ if let Ok(sleep_duration) = target_time.duration_since(SystemTime::now()) {`
`20`	`20`	`time::sleep(sleep_duration).await;`
`21`	`21`	`}`
`22`	`22`	`}`
`@@ -100,12 +100,13 @@ pub fn combine_events(`
`100`	`100`	`.map(\|(k, v)\| (k, v.into_iter().map(\|(_, v)\| v).collect()))`
`101`	`101`	`.collect();`
`102`	`102`
`103`		`- WorkflowResult::Ok(event_history)`
	`103`	`+ Ok(event_history)`
`104`	`104`	`}`
`105`	`105`
`106`		`-pub fn inject_fault() -> Result<()> {`
	`106`	`+pub fn inject_fault() -> GlobalResult<()> {`
`107`	`107`	`if rand::thread_rng().gen_range(0..100) < FAULT_RATE {`
`108`	`108`	`bail!("This is a random panic!");`
`109`	`109`	`}`
	`110`	`+`
`110`	`111`	`Ok(())`
`111`	`112`	`}`