Fix cancel-before-start abandon activities

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

@@ -407,7 +407,7 @@ impl ScheduleCommandCreated {
 
     pub(super) fn on_abandoned(self, dat: &mut SharedState) -> ActivityMachineTransition<Canceled> {
         dat.cancelled_before_sent = true;
-        ActivityMachineTransition::default()
+        notify_lang_activity_cancelled(None)
     }
 }
 
@@ -806,7 +806,7 @@ mod test {
         internal_flags::InternalFlags,
         replay::TestHistoryBuilder,
         test_help::{build_fake_sdk, MockPollCfg, ResponseType},
-        worker::workflow::machines::Machines,
+        worker::workflow::{machines::Machines, OutgoingJob},
     };
     use std::{cell::RefCell, mem::discriminant, rc::Rc};
     use temporal_sdk::{ActivityOptions, CancellableFuture, WfContext, WorkflowFunction};
@@ -905,7 +905,20 @@ mod test {
             panic!("Wrong machine type");
         };
         let cmds = s.cancel().unwrap();
-        assert_eq!(cmds.len(), 0);
+        // We should always be notifying lang that the activity got cancelled, even if it's
+        // abandoned and we aren't telling server
+        assert_matches!(
+            cmds.as_slice(),
+            [MachineResponse::PushWFJob(OutgoingJob {
+                variant: workflow_activation_job::Variant::ResolveActivity(ResolveActivity {
+                    result: Some(ActivityResolution {
+                        status: Some(activity_resolution::Status::Cancelled(_))
+                    }),
+                    ..
+                }),
+                ..
+            })]
+        );
         let curstate = s.state();
         assert!(matches!(curstate, &ActivityMachineState::Canceled(_)));
     }

tests/integ_tests/workflow_tests/activities.rs

@@ -787,6 +787,50 @@ async fn activity_cancelled_after_heartbeat_times_out() {
         .unwrap();
 }
 
+#[tokio::test]
+async fn one_activity_abandon_cancelled_before_started() {
+    let wf_name = "one_activity_abandon_cancelled_before_started";
+    let mut starter = CoreWfStarter::new(wf_name);
+    let mut worker = starter.worker().await;
+    let client = starter.get_client().await;
+    worker.register_wf(wf_name.to_owned(), |ctx: WfContext| async move {
+        let act_fut = ctx.activity(ActivityOptions {
+            activity_type: "echo_activity".to_string(),
+            start_to_close_timeout: Some(Duration::from_secs(5)),
+            input: "hi!".as_json_payload().expect("serializes fine"),
+            cancellation_type: ActivityCancellationType::Abandon,
+            ..Default::default()
+        });
+        act_fut.cancel(&ctx);
+        act_fut.await;
+        Ok(().into())
+    });
+    worker.register_activity(
+        "echo_activity",
+        |_ctx: ActContext, echo_me: String| async move {
+            sleep(Duration::from_secs(2)).await;
+            Ok(echo_me)
+        },
+    );
+
+    let run_id = worker
+        .submit_wf(
+            wf_name.to_owned(),
+            wf_name.to_owned(),
+            vec![],
+            WorkflowOptions::default(),
+        )
+        .await
+        .unwrap();
+    worker.run_until_done().await.unwrap();
+    let handle = client.get_untyped_workflow_handle(wf_name, run_id);
+    let res = handle
+        .get_workflow_result(Default::default())
+        .await
+        .unwrap();
+    assert_matches!(res, WorkflowExecutionResult::Succeeded(_));
+}
+
 #[tokio::test]
 async fn one_activity_abandon_cancelled_after_complete() {
     let wf_name = "one_activity_abandon_cancelled_after_complete";