feat(ds): add server create failed message

errors/servers/servers_server_failed_to_create.md

@@ -0,0 +1,9 @@
+---
+name = "SERVERS_SERVER_FAILED_TO_CREATE"
+description = "Server failed to create."
+http_status = 400
+---
+
+# Server Failed To Create
+
+Server failed to create.

svc/api/servers/src/route/servers.rs

@@ -72,11 +72,16 @@ pub async fn create(
 
 	let server_id = Uuid::new_v4();
 
-	let mut sub = ctx
+	let mut create_sub = ctx
 		.subscribe::<ds::workflows::server::CreateComplete>(&json!({
 			"server_id": server_id,
 		}))
 		.await?;
+	let mut fail_sub = ctx
+		.subscribe::<ds::workflows::server::CreateFailed>(&json!({
+			"server_id": server_id,
+		}))
+		.await?;
 
 	ctx.dispatch_tagged_workflow(
 		&json!({
@@ -136,7 +141,13 @@ pub async fn create(
 	)
 	.await?;
 
-	sub.next().await?;
+	tokio::select! {
+		res = create_sub.next() => { res?; },
+		res = fail_sub.next() => {
+			res?;
+			bail_with!(SERVERS_SERVER_FAILED_TO_CREATE);
+		}
+	}
 
 	let servers_res = ctx
 		.op(ds::ops::server::get::Input {

svc/pkg/ds/src/workflows/server/mod.rs

@@ -77,6 +77,48 @@ pub struct Port {
 
 #[workflow]
 pub async fn ds_server(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResult<()> {
+	let res = setup(ctx, input).await;
+	match ctx.catch_unrecoverable(res)? {
+		Ok(_) => {}
+		// If we cannot recover a setup error, send a failed signal
+		Err(err) => {
+			tracing::warn!("unrecoverable setup");
+
+			// TODO: Cleanup
+
+			ctx.msg(
+				json!({
+					"server_id": input.server_id,
+				}),
+				CreateFailed {},
+			)
+			.await?;
+
+			// Throw the original error from the setup activities
+			return Err(err);
+		}
+	};
+
+	ctx.msg(
+		json!({
+			"server_id": input.server_id
+		}),
+		CreateComplete {},
+	)
+	.await?;
+
+	let destroy_sig = ctx.listen::<Destroy>().await?;
+
+	ctx.workflow(destroy::Input {
+		server_id: input.server_id,
+		override_kill_timeout_ms: destroy_sig.override_kill_timeout_ms,
+	})
+	.await?;
+
+	Ok(())
+}
+
+async fn setup(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResult<()> {
 	let (_, prereq) = ctx
 		.join((
 			InsertDbInput {
@@ -140,22 +182,6 @@ pub async fn ds_server(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResult<()>
 	})
 	.await?;
 
-	ctx.msg(
-		json!({
-			"server_id": input.server_id
-		}),
-		CreateComplete {},
-	)
-	.await?;
-
-	let destroy_sig = ctx.listen::<Destroy>().await?;
-
-	ctx.workflow(destroy::Input {
-		server_id: input.server_id,
-		override_kill_timeout_ms: destroy_sig.override_kill_timeout_ms,
-	})
-	.await?;
-
 	Ok(())
 }
 
@@ -1072,6 +1098,9 @@ async fn update_db(ctx: &ActivityCtx, input: &UpdateDbInput) -> GlobalResult<()>
 #[message("ds_server_create_complete")]
 pub struct CreateComplete {}
 
+#[message("ds_server_create_failed")]
+pub struct CreateFailed {}
+
 #[signal("ds_server_destroy")]
 pub struct Destroy {
 	pub override_kill_timeout_ms: i64,

svc/pkg/linode/src/workflows/server/mod.rs

@@ -33,7 +33,7 @@ pub async fn linode_server(ctx: &mut WorkflowCtx, input: &Input) -> GlobalResult
 		Ok(x) => x,
 		// If we cannot recover a provisioning error, send a failed signal and clean up resources
 		Err(err) => {
-			tracing::warn!(?err, "unrecoverable provision, cleaning up");
+			tracing::warn!("unrecoverable provision, cleaning up");
 
 			ctx.dispatch_workflow(cleanup::Input {
 				api_token: input.api_token.clone(),