Handle NexusOperationCancelRequest{Completed, Failed}

These events are written by the caller server on receiving a response
from the nexus cancel handler. The only cancellation type that is
sensitive to them is WaitRequested: when that is in effect, these
events cause an activation job to be emitted that resolves the nexus
operation future.

The fsm macro is modified to allow the same union type to be
constructed twice, since Started -> Started|Cancelled is now caused by
two different inputs.

Author: dandavison

core/src/worker/workflow/machines/nexus_operation_state_machine.rs

@@ -21,6 +21,8 @@ use temporal_sdk_core_protos::{
         enums::v1::{CommandType, EventType},
         failure::v1::{self as failure, Failure, failure::FailureInfo},
         history::v1::{
+            NexusOperationCancelRequestCompletedEventAttributes,
+            NexusOperationCancelRequestFailedEventAttributes,
             NexusOperationCanceledEventAttributes, NexusOperationCompletedEventAttributes,
             NexusOperationFailedEventAttributes, NexusOperationStartedEventAttributes,
             NexusOperationTimedOutEventAttributes, history_event,
@@ -63,6 +65,12 @@ fsm! {
       --(NexusOperationFailed(NexusOperationFailedEventAttributes), on_failed)--> Failed;
     Started
       --(NexusOperationCanceled(NexusOperationCanceledEventAttributes), on_canceled)--> Cancelled;
+
+
+    Started --(NexusOperationCancelRequestCompleted(NexusOperationCancelRequestCompletedEventAttributes), shared on_cancel_request_completed)--> Started;
+    Started --(NexusOperationCancelRequestCompleted(NexusOperationCancelRequestCompletedEventAttributes), shared on_cancel_request_completed)--> Cancelled;
+    Started --(NexusOperationCancelRequestFailed(NexusOperationCancelRequestFailedEventAttributes), shared on_cancel_request_failed)--> Started;
+
     Started
       --(NexusOperationTimedOut(NexusOperationTimedOutEventAttributes), on_timed_out)--> TimedOut;
 
@@ -72,6 +80,8 @@ fsm! {
     Cancelled --(NexusOperationCompleted(NexusOperationCompletedEventAttributes), shared on_completed)--> Cancelled;
     Cancelled --(NexusOperationFailed(NexusOperationFailedEventAttributes), shared on_failed)--> Cancelled;
     Cancelled --(NexusOperationTimedOut(NexusOperationTimedOutEventAttributes), shared on_timed_out)--> Cancelled;
+    Cancelled --(NexusOperationCancelRequestCompleted(NexusOperationCancelRequestCompletedEventAttributes), on_cancel_request_completed)--> Cancelled;
+    Cancelled --(NexusOperationCancelRequestFailed(NexusOperationCancelRequestFailedEventAttributes))--> Cancelled;
     Cancelled --(NexusOperationCanceled(NexusOperationCanceledEventAttributes))--> Cancelled;
 
     // Ignore cancels in all terminal states
@@ -311,6 +321,45 @@ impl Started {
         )])
     }
 
+    pub(super) fn on_cancel_request_completed(
+        self,
+        ss: &SharedState,
+        _: NexusOperationCancelRequestCompletedEventAttributes,
+    ) -> NexusOperationMachineTransition<StartedOrCancelled> {
+        if ss.cancel_type == NexusOperationCancellationType::WaitCancellationRequested {
+            TransitionResult::ok(
+                [NexusOperationCommand::Cancel(ss.cancelled_failure(
+                    "Nexus operation cancellation request completed".to_owned(),
+                ))],
+                StartedOrCancelled::Cancelled(Default::default()),
+            )
+        } else {
+            TransitionResult::ok([], StartedOrCancelled::Started(Default::default()))
+        }
+    }
+
+    pub(super) fn on_cancel_request_failed(
+        self,
+        ss: &SharedState,
+        fa: NexusOperationCancelRequestFailedEventAttributes,
+    ) -> NexusOperationMachineTransition<Started> {
+        if ss.cancel_type == NexusOperationCancellationType::WaitCancellationRequested {
+            let message = "Nexus operation cancellation request failed".to_string();
+            TransitionResult::ok(
+                [NexusOperationCommand::Fail(ss.failure(
+                    message.clone(),
+                    fa.failure.unwrap_or_else(|| Failure {
+                        message,
+                        ..Default::default()
+                    }),
+                ))],
+                self,
+            )
+        } else {
+            TransitionResult::ok([], self)
+        }
+    }
+
     pub(super) fn on_timed_out(
         self,
         toa: NexusOperationTimedOutEventAttributes,
@@ -356,6 +405,13 @@ impl Cancelled {
         NexusOperationMachineTransition::ok([], self)
     }
 
+    pub(super) fn on_cancel_request_completed(
+        self,
+        _: NexusOperationCancelRequestCompletedEventAttributes,
+    ) -> NexusOperationMachineTransition<Cancelled> {
+        NexusOperationMachineTransition::ok([], self)
+    }
+
     pub(super) fn on_failed(
         self,
         ss: &mut SharedState,
@@ -454,6 +510,36 @@ impl TryFrom<HistEventData> for NexusOperationMachineEvents {
                 }
             }
             Ok(EventType::NexusOperationCancelRequested) => Self::NexusOperationCancelRequested,
+            Ok(EventType::NexusOperationCancelRequestCompleted) => {
+                if let Some(
+                    history_event::Attributes::NexusOperationCancelRequestCompletedEventAttributes(
+                        attrs,
+                    ),
+                ) = e.attributes
+                {
+                    Self::NexusOperationCancelRequestCompleted(attrs)
+                } else {
+                    return Err(WFMachinesError::Nondeterminism(
+                        "NexusOperationCancelRequestCompleted attributes were unset or malformed"
+                            .to_string(),
+                    ));
+                }
+            }
+            Ok(EventType::NexusOperationCancelRequestFailed) => {
+                if let Some(
+                    history_event::Attributes::NexusOperationCancelRequestFailedEventAttributes(
+                        attrs,
+                    ),
+                ) = e.attributes
+                {
+                    Self::NexusOperationCancelRequestFailed(attrs)
+                } else {
+                    return Err(WFMachinesError::Nondeterminism(
+                        "NexusOperationCancelRequestFailed attributes were unset or malformed"
+                            .to_string(),
+                    ));
+                }
+            }
             _ => {
                 return Err(WFMachinesError::Nondeterminism(format!(
                     "Nexus operation machine does not handle this event: {e:?}"
@@ -599,12 +685,19 @@ impl TryFrom<CommandType> for NexusOperationMachineEvents {
 
 impl SharedState {
     fn cancelled_failure(&self, message: String) -> Failure {
-        Failure {
+        self.failure(
             message,
-            cause: Some(Box::new(Failure {
+            Failure {
                 failure_info: Some(FailureInfo::CanceledFailureInfo(Default::default())),
                 ..Default::default()
-            })),
+            },
+        )
+    }
+
+    fn failure(&self, message: String, cause: Failure) -> Failure {
+        Failure {
+            message,
+            cause: Some(Box::new(cause)),
             failure_info: Some(FailureInfo::NexusOperationExecutionFailureInfo(
                 failure::NexusOperationFailureInfo {
                     scheduled_event_id: self.scheduled_event_id,

fsm/rustfsm_procmacro/src/lib.rs

@@ -434,6 +434,7 @@ impl StateMachineDefinition {
             }
         };
         let mut multi_dest_enums = vec![];
+        let mut multi_dest_enum_names = HashSet::new();
         let state_branches: Vec<_> = statemap.into_iter().map(|(from, transitions)| {
             let occupied_current_state = quote! { Some(#state_enum_name::#from(state_data)) };
             // Merge transition dest states with the same handler
@@ -468,7 +469,11 @@ impl StateMachineDefinition {
                                         }
                                     }
                                 };
-                                multi_dest_enums.push(multi_dest_enum);
+                                // Deduplicate; two different events may each result in a transition
+                                // set with the same set of dest states
+                                if multi_dest_enum_names.insert(enum_ident.clone()) {
+                                    multi_dest_enums.push(multi_dest_enum);
+                                }
                                 quote! {
                                     #transition_result_name<#enum_ident>
                                 }

tests/integ_tests/workflow_tests/nexus.rs

@@ -765,8 +765,14 @@ async fn nexus_cancellation_types(
             );
             assert_eq!(f.message, "Nexus operation cancelled after starting");
         }
-        NexusOperationCancellationType::WaitCancellationRequested
-        | NexusOperationCancellationType::WaitCancellationCompleted => {
+        NexusOperationCancellationType::WaitCancellationRequested => {
+            let f = assert_matches!(
+                res.status,
+                Some(nexus_operation_result::Status::Cancelled(f)) => f
+            );
+            assert_eq!(f.message, "Nexus operation cancellation request completed");
+        }
+        NexusOperationCancellationType::WaitCancellationCompleted => {
             let f = assert_matches!(
                 res.status,
                 Some(nexus_operation_result::Status::Cancelled(f)) => f

tests/runner.rs

@@ -107,6 +107,8 @@ async fn main() -> Result<(), anyhow::Error> {
                     "frontend.workerVersioningDataAPIs=true".to_owned(),
                     "--dynamic-config-value".to_owned(),
                     "system.enableDeploymentVersions=true".to_owned(),
+                    "--dynamic-config-value".to_owned(),
+                    "component.nexusoperations.recordCancelRequestCompletionEvents=true".to_owned(),
                     "--http-port".to_string(),
                     "7243".to_string(),
                     "--search-attribute".to_string(),