Fix bug where an abandoned child which was cancelled externally errors (#524)

when the event is processed after cancelling inside the workflow.

Author: Sushisource

core/src/core_tests/child_workflows.rs

@@ -88,11 +88,12 @@ async fn parent_cancels_child_wf(ctx: WfContext) -> WorkflowResult<()> {
         .await
         .into_started()
         .expect("Child should get started");
-    let cancel_fut = start_res.cancel(&ctx);
-    let resfut = start_res.result();
-    let (cancel_res, res) = join!(cancel_fut, resfut);
-    cancel_res.expect("cancel result is ok");
-    let stat = res.status.expect("child wf result is ok");
+    start_res.cancel(&ctx);
+    let stat = start_res
+        .result()
+        .await
+        .status
+        .expect("child wf result is ok");
     assert_matches!(stat, child_workflow_result::Status::Cancelled(_));
     Ok(().into())
 }

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

@@ -74,6 +74,8 @@ fsm! {
 
     // Ignore any spurious cancellations after resolution
     Cancelled --(Cancel) --> Cancelled;
+    Cancelled --(ChildWorkflowExecutionCancelled,
+        on_child_workflow_execution_cancelled) --> Cancelled;
     Failed --(Cancel) --> Failed;
     StartFailed --(Cancel) --> StartFailed;
     TimedOut --(Cancel) --> TimedOut;
@@ -111,7 +113,30 @@ pub(super) struct ChildWorkflowInitiatedData {
 }
 
 #[derive(Default, Clone)]
-pub(super) struct Cancelled {}
+pub(super) struct Cancelled {
+    seen_cancelled_event: bool,
+}
+
+impl Cancelled {
+    pub(super) fn on_child_workflow_execution_cancelled(
+        self,
+    ) -> ChildWorkflowMachineTransition<Cancelled> {
+        if self.seen_cancelled_event {
+            ChildWorkflowMachineTransition::Err(WFMachinesError::Fatal(
+                "Child workflow has already seen a ChildWorkflowExecutionCanceledEvent, and now \
+                 another is being applied! This is a bug, please report."
+                    .to_string(),
+            ))
+        } else {
+            ChildWorkflowMachineTransition::ok(
+                [],
+                Cancelled {
+                    seen_cancelled_event: true,
+                },
+            )
+        }
+    }
+}
 
 #[derive(Default, Clone)]
 pub(super) struct Completed {}
@@ -263,7 +288,12 @@ impl Started {
         )
     }
     fn on_child_workflow_execution_cancelled(self) -> ChildWorkflowMachineTransition<Cancelled> {
-        ChildWorkflowMachineTransition::ok(vec![ChildWorkflowCommand::Cancel], Cancelled::default())
+        ChildWorkflowMachineTransition::ok(
+            vec![ChildWorkflowCommand::Cancel],
+            Cancelled {
+                seen_cancelled_event: true,
+            },
+        )
     }
     fn on_child_workflow_execution_terminated(
         self,
@@ -870,7 +900,9 @@ mod test {
     #[test]
     fn cancels_ignored_terminal() {
         for state in [
-            ChildWorkflowMachineState::Cancelled(Cancelled {}),
+            ChildWorkflowMachineState::Cancelled(Cancelled {
+                seen_cancelled_event: false,
+            }),
             Failed {}.into(),
             StartFailed {}.into(),
             TimedOut {}.into(),

sdk/src/lib.rs

@@ -549,11 +549,13 @@ pub enum TimerResult {
 }
 
 /// Successful result of sending a signal to an external workflow
+#[derive(Debug)]
 pub struct SignalExternalOk;
 /// Result of awaiting on sending a signal to an external workflow
 pub type SignalExternalWfResult = Result<SignalExternalOk, Failure>;
 
 /// Successful result of sending a cancel request to an external workflow
+#[derive(Debug)]
 pub struct CancelExternalOk;
 /// Result of awaiting on sending a cancel request to an external workflow
 pub type CancelExternalWfResult = Result<CancelExternalOk, Failure>;
@@ -656,6 +658,20 @@ pub enum CancellableID {
     },
 }
 
+impl CancellableID {
+    /// Returns the type-specific sequence number used for this command
+    pub fn seq_num(&self) -> u32 {
+        match self {
+            CancellableID::Timer(seq) => *seq,
+            CancellableID::Activity(seq) => *seq,
+            CancellableID::LocalActivity(seq) => *seq,
+            CancellableID::ChildWorkflow(seq) => *seq,
+            CancellableID::SignalExternalWorkflow(seq) => *seq,
+            CancellableID::ExternalWorkflow { seqnum, .. } => *seqnum,
+        }
+    }
+}
+
 #[derive(derive_more::From)]
 #[allow(clippy::large_enum_variant)]
 enum RustWfCmd {

sdk/src/workflow_context.rs

@@ -34,7 +34,8 @@ use temporal_sdk_core_protos::{
         workflow_commands::{
             request_cancel_external_workflow_execution as cancel_we,
             signal_external_workflow_execution as sig_we, workflow_command,
-            ModifyWorkflowProperties, RequestCancelExternalWorkflowExecution, SetPatchMarker,
+            CancelChildWorkflowExecution, ModifyWorkflowProperties,
+            RequestCancelExternalWorkflowExecution, SetPatchMarker,
             SignalExternalWorkflowExecution, StartTimer, UpsertWorkflowSearchAttributes,
         },
     },
@@ -679,13 +680,13 @@ impl StartedChildWorkflow {
     }
 
     /// Cancel the child workflow
-    pub fn cancel(&self, cx: &WfContext) -> impl Future<Output = CancelExternalWfResult> {
-        let target = NamespacedWorkflowExecution {
-            namespace: cx.namespace().to_string(),
-            workflow_id: self.common.workflow_id.clone(),
-            ..Default::default()
-        };
-        cx.cancel_external(target)
+    pub fn cancel(&self, cx: &WfContext) {
+        cx.send(RustWfCmd::NewNonblockingCmd(
+            CancelChildWorkflowExecution {
+                child_workflow_seq: self.common.result_future.cancellable_id.seq_num(),
+            }
+            .into(),
+        ));
     }
 
     /// Signal the child workflow

sdk/src/workflow_context/options.rs

@@ -9,7 +9,10 @@ use temporal_sdk_core_protos::{
             StartChildWorkflowExecution,
         },
     },
-    temporal::api::common::v1::{Payload, RetryPolicy},
+    temporal::api::{
+        common::v1::{Payload, RetryPolicy},
+        enums::v1::ParentClosePolicy,
+    },
 };
 
 // TODO: Before release, probably best to avoid using proto types entirely here. They're awkward.
@@ -180,6 +183,8 @@ pub struct ChildWorkflowOptions {
     pub cancel_type: ChildWorkflowCancellationType,
     /// Common options
     pub options: WorkflowOptions,
+    /// How to respond to parent workflow ending
+    pub parent_close_policy: ParentClosePolicy,
 }
 
 impl IntoWorkflowCommand for ChildWorkflowOptions {
@@ -203,6 +208,7 @@ impl IntoWorkflowCommand for ChildWorkflowOptions {
             workflow_task_timeout: self.options.task_timeout.and_then(|d| d.try_into().ok()),
             search_attributes: self.options.search_attributes.unwrap_or_default(),
             cron_schedule: self.options.cron_schedule.unwrap_or_default(),
+            parent_close_policy: self.parent_close_policy as i32,
             ..Default::default()
         }
     }

tests/integ_tests/workflow_tests/child_workflows.rs

@@ -1,8 +1,13 @@
 use anyhow::anyhow;
-use temporal_client::WorkflowOptions;
-use temporal_sdk::{ChildWorkflowOptions, WfContext, WorkflowResult};
-use temporal_sdk_core_protos::coresdk::child_workflow::{child_workflow_result, Success};
+use std::time::Duration;
+use temporal_client::{WorkflowClientTrait, WorkflowOptions};
+use temporal_sdk::{ChildWorkflowOptions, WfContext, WfExitValue, WorkflowResult};
+use temporal_sdk_core_protos::{
+    coresdk::child_workflow::{child_workflow_result, ChildWorkflowCancellationType, Success},
+    temporal::api::enums::v1::ParentClosePolicy,
+};
 use temporal_sdk_core_test_utils::CoreWfStarter;
+use tokio::sync::Barrier;
 
 static PARENT_WF_TYPE: &str = "parent_wf";
 static CHILD_WF_TYPE: &str = "child_wf";
@@ -49,3 +54,74 @@ async fn child_workflow_happy_path() {
         .unwrap();
     worker.run_until_done().await.unwrap();
 }
+
+#[tokio::test]
+async fn abandoned_child_bug_repro() {
+    let mut starter = CoreWfStarter::new("child-workflow-abandon-bug");
+    starter.no_remote_activities();
+    let mut worker = starter.worker().await;
+    let barr: &'static Barrier = Box::leak(Box::new(Barrier::new(2)));
+
+    worker.register_wf(
+        PARENT_WF_TYPE.to_string(),
+        move |mut ctx: WfContext| async move {
+            let child = ctx.child_workflow(ChildWorkflowOptions {
+                workflow_id: "abandoned-child".to_owned(),
+                workflow_type: CHILD_WF_TYPE.to_owned(),
+                parent_close_policy: ParentClosePolicy::Abandon,
+                cancel_type: ChildWorkflowCancellationType::Abandon,
+                ..Default::default()
+            });
+
+            let started = child
+                .start(&ctx)
+                .await
+                .into_started()
+                .expect("Child chould start OK");
+            barr.wait().await;
+            // Wait for cancel signal
+            ctx.cancelled().await;
+            // Cancel the child immediately
+            started.cancel(&ctx);
+            // Need to do something else, so we'll see the ChildWorkflowExecutionCanceled event
+            ctx.timer(Duration::from_secs(1)).await;
+            started.result().await;
+            Ok(().into())
+        },
+    );
+    worker.register_wf(CHILD_WF_TYPE.to_string(), |mut ctx: WfContext| async move {
+        ctx.cancelled().await;
+        Ok(WfExitValue::Cancelled)
+    });
+
+    worker
+        .submit_wf(
+            "parent-abandoner".to_string(),
+            PARENT_WF_TYPE.to_owned(),
+            vec![],
+            WorkflowOptions::default(),
+        )
+        .await
+        .unwrap();
+    let client = starter.get_client().await;
+    let canceller = async {
+        barr.wait().await;
+        client
+            .cancel_workflow_execution(
+                "parent-abandoner".to_string(),
+                None,
+                "die".to_string(),
+                None,
+            )
+            .await
+            .unwrap();
+        client
+            .cancel_workflow_execution("abandoned-child".to_string(), None, "die".to_string(), None)
+            .await
+            .unwrap();
+    };
+    let runner = async move {
+        worker.run_until_done().await.unwrap();
+    };
+    tokio::join!(canceller, runner);
+}