fix(workflows): Throw errors for duplicate workflows

Author: MasterPtato

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

@@ -128,6 +128,9 @@ pub enum WorkflowError {
 
 	#[error("operation timed out")]
 	OperationTimeout,
+
+	#[error("duplicate registered workflow: {0}")]
+	DuplicateRegisteredWorkflow(String),
 }
 
 impl WorkflowError {

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

@@ -27,13 +27,25 @@ impl Registry {
 		Arc::new(self)
 	}
 
-	pub fn merge(mut self, registry: Registry) -> Registry {
+	pub fn merge(mut self, registry: Registry) -> WorkflowResult<Registry> {
+		// Check for duplicates
+		for (workflow_name, _) in &registry.workflows {
+			if self.workflows.contains_key(workflow_name.as_str()) {
+				return Err(WorkflowError::DuplicateRegisteredWorkflow(workflow_name.clone()));
+			}
+		}
+
 		self.workflows.extend(registry.workflows.into_iter());
 
-		self
+		Ok(self)
 	}
 
-	pub fn register_workflow<W: Workflow>(&mut self) {
+	pub fn register_workflow<W: Workflow>(&mut self) -> WorkflowResult<()> {
+		// Check for duplicates
+		if self.workflows.contains_key(W::NAME) {
+			return Err(WorkflowError::DuplicateRegisteredWorkflow(W::NAME.to_string()));
+		}
+
 		self.workflows.insert(
 			W::NAME.to_string(),
 			Arc::new(RegistryWorkflow {
@@ -72,6 +84,8 @@ impl Registry {
 				},
 			}),
 		);
+
+		Ok(())
 	}
 
 	pub fn get_workflow(&self, name: &str) -> WorkflowResult<&Arc<RegistryWorkflow>> {

svc/pkg/cluster/src/lib.rs

@@ -5,21 +5,21 @@ pub mod types;
 pub mod util;
 pub mod workflows;
 
-pub fn registry() -> Registry {
+pub fn registry() -> WorkflowResult<Registry> {
 	use workflows::*;
 
 	let mut registry = Registry::new();
-	registry.register_workflow::<cluster::Workflow>();
-	registry.register_workflow::<datacenter::Workflow>();
-	registry.register_workflow::<datacenter::scale::Workflow>();
-	registry.register_workflow::<datacenter::tls_issue::Workflow>();
-	registry.register_workflow::<server::Workflow>();
-	registry.register_workflow::<server::install::Workflow>();
-	registry.register_workflow::<server::dns_create::Workflow>();
-	registry.register_workflow::<server::dns_delete::Workflow>();
-	registry.register_workflow::<server::drain::Workflow>();
-	registry.register_workflow::<server::undrain::Workflow>();
-	registry.register_workflow::<prebake::Workflow>();
+	registry.register_workflow::<cluster::Workflow>()?;
+	registry.register_workflow::<datacenter::Workflow>()?;
+	registry.register_workflow::<datacenter::scale::Workflow>()?;
+	registry.register_workflow::<datacenter::tls_issue::Workflow>()?;
+	registry.register_workflow::<server::Workflow>()?;
+	registry.register_workflow::<server::install::Workflow>()?;
+	registry.register_workflow::<server::dns_create::Workflow>()?;
+	registry.register_workflow::<server::dns_delete::Workflow>()?;
+	registry.register_workflow::<server::drain::Workflow>()?;
+	registry.register_workflow::<server::undrain::Workflow>()?;
+	registry.register_workflow::<prebake::Workflow>()?;
 
-	registry
+	Ok(registry)
 }

svc/pkg/cluster/src/workflows/datacenter/mod.rs

@@ -61,8 +61,11 @@ pub(crate) async fn cluster_datacenter(ctx: &mut WorkflowCtx, input: &Input) ->
 	)
 	.await?;
 
-	// Scale
-	ctx.signal(ctx.workflow_id(), Scale {}).await?;
+	// Scale initially
+	ctx.workflow(scale::Input {
+		datacenter_id: input.datacenter_id,
+	})
+	.await?;
 
 	ctx.repeat(|ctx| {
 		let datacenter_id = input.datacenter_id;

svc/pkg/linode/src/lib.rs

@@ -5,13 +5,13 @@ pub mod types;
 pub mod util;
 pub mod workflows;
 
-pub fn registry() -> Registry {
+pub fn registry() -> WorkflowResult<Registry> {
 	use workflows::*;
 
 	let mut registry = Registry::new();
-	registry.register_workflow::<image::Workflow>();
-	registry.register_workflow::<server::Workflow>();
-	registry.register_workflow::<server::cleanup::Workflow>();
+	registry.register_workflow::<image::Workflow>()?;
+	registry.register_workflow::<server::Workflow>()?;
+	registry.register_workflow::<server::cleanup::Workflow>()?;
 
-	registry
+	Ok(registry)
 }