diff --git a/src/flow/mod.rs b/src/flow/mod.rs index 8ebf69d..89c8cac 100644 --- a/src/flow/mod.rs +++ b/src/flow/mod.rs @@ -19,6 +19,12 @@ pub fn flow_belongs_to_action(flow: &ValidationFlow, action_identifier: &str) -> flow.definition_source.as_deref() == Some(source.as_str()) } +/// Whether `flow` has been disabled, and if so, why. Callers must reject +/// execution requests against a disabled flow rather than dispatching them. +pub fn flow_disable_reason(flow: &ValidationFlow) -> Option<&str> { + flow.disable_reason.as_deref() +} + /// Projects a `ValidationFlow` down to the fields an action needs to /// execute/validate against it - an "action flow". pub fn to_action_flow(flow: &ValidationFlow) -> ActionFlow { @@ -135,6 +141,21 @@ mod tests { assert!(!flow_belongs_to_action(&flow(None), "send-email")); } + #[test] + fn flow_disable_reason_returns_reason_when_disabled() { + let disabled = ValidationFlow { + disable_reason: Some("maintenance".to_string()), + ..Default::default() + }; + let enabled = ValidationFlow { + disable_reason: None, + ..Default::default() + }; + + assert_eq!(flow_disable_reason(&disabled), Some("maintenance")); + assert_eq!(flow_disable_reason(&enabled), None); + } + #[test] fn to_action_flow_projects_matching_fields() { let flow = ValidationFlow { diff --git a/src/sagittarius/test_execution_client_impl/mod.rs b/src/sagittarius/test_execution_client_impl/mod.rs index f4da066..7dd4273 100644 --- a/src/sagittarius/test_execution_client_impl/mod.rs +++ b/src/sagittarius/test_execution_client_impl/mod.rs @@ -124,6 +124,39 @@ impl SagittariusTestExecutionServiceClient { } }; + if let Some(reason) = flow::flow_disable_reason(&validation_flow) { + log::warn!( + "Rejecting Sagittarius execution request for a disabled flow requested_execution_id={} flow_id={} reason={}", + request.execution_identifier, + request.flow_id, + reason + ); + + let execution_id = if request.execution_identifier.is_empty() { + uuid::Uuid::new_v4().to_string() + } else { + request.execution_identifier.clone() + }; + + let rejection = validation::disabled_flow_rejection_result( + execution_id, + request.flow_id, + reason, + ); + + if let Err(status) = + self.response_sender.send_execution_result(rejection).await + { + log::error!( + "Failed to send disabled flow rejection result flow_id={} error={:?}", + request.flow_id, + status + ); + } + + continue; + } + if validation::is_rest_flow(&validation_flow) { let input_schema = validation::extract_input_schema(&validation_flow); if let Err(err) = validation::validate_body_against_schema( diff --git a/src/server/action_transfer/nats_bridge.rs b/src/server/action_transfer/nats_bridge.rs index 921b368..03e1201 100644 --- a/src/server/action_transfer/nats_bridge.rs +++ b/src/server/action_transfer/nats_bridge.rs @@ -113,6 +113,25 @@ pub(super) async fn handle_flow_execution( return; } + if let Some(reason) = flow::flow_disable_reason(&validation_flow) { + log::warn!( + "Rejected action flow execution request for a disabled flow action={} flow_id={} reason={}", + action_identifier, + flow_id, + reason + ); + send_flow_execution_failure( + &tx, + execution_id, + format!( + "flow {} has been disabled for the reason: {}", + flow_id, reason + ), + ) + .await; + return; + } + if validation::is_rest_flow(&validation_flow) { let input_schema = validation::extract_input_schema(&validation_flow); if let Err(err) = diff --git a/src/validation.rs b/src/validation.rs index 5fe054f..51db94b 100644 --- a/src/validation.rs +++ b/src/validation.rs @@ -187,6 +187,31 @@ pub fn rejection_result( } } +/// Synthesizes the [`ExecutionResult`] sent back in place of dispatching an +/// execution request against a disabled flow. +pub fn disabled_flow_rejection_result( + execution_identifier: String, + flow_id: i64, + reason: &str, +) -> ExecutionResult { + let now = epoch_millis_now(); + ExecutionResult { + execution_identifier, + flow_id, + started_at: now, + finished_at: now, + result: Some(execution_result::Result::Error(Error { + code: "A-VALIDATION-000002".to_string(), + category: "InvalidArgument".to_string(), + message: format!("flow {} has been disabled for the reason: {}", flow_id, reason), + timestamp: now, + version: crate::version::runtime_version().to_string(), + ..Default::default() + })), + ..Default::default() + } +} + pub(crate) fn epoch_millis_now() -> i64 { match SystemTime::now().duration_since(UNIX_EPOCH) { Ok(duration) => duration.as_millis() as i64, @@ -321,4 +346,18 @@ mod tests { other => panic!("expected error result, got {:?}", other), } } + + #[test] + fn disabled_flow_rejection_result_carries_reason() { + let result = disabled_flow_rejection_result("exec-1".to_string(), 42, "maintenance"); + + assert_eq!(result.execution_identifier, "exec-1"); + assert_eq!(result.flow_id, 42); + match result.result { + Some(execution_result::Result::Error(err)) => { + assert!(err.message.contains("maintenance")); + } + other => panic!("expected error result, got {:?}", other), + } + } }