Change default WFT fail cause to worker unhandled (#1047)

Author: Sushisource

crates/sdk-core/src/core_tests/workflow_tasks.rs

@@ -1286,6 +1286,40 @@ async fn fail_wft_then_recover() {
     core.shutdown().await;
 }
 
+#[tokio::test]
+async fn default_wft_fail_cause_is_worker_unhandled() {
+    let t = canned_histories::single_timer("1");
+    let mut mh = MockPollCfg::from_resp_batches(
+        "fake_wf_id",
+        t,
+        [ResponseType::AllHistory],
+        mock_worker_client(),
+    );
+    mh.num_expected_fails = 1;
+    mh.expect_fail_wft_matcher = Box::new(|_, cause, _| {
+        matches!(
+            cause,
+            WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure
+        )
+    });
+    let mock = build_mock_pollers(mh);
+    let core = mock_worker(mock);
+
+    let act = core.poll_workflow_activation().await.unwrap();
+    core.complete_workflow_activation(WorkflowActivationCompletion::fail(
+        act.run_id,
+        Failure {
+            message: "Simulated workflow failure".to_string(),
+            ..Default::default()
+        },
+        None,
+    ))
+    .await
+    .unwrap();
+
+    core.shutdown().await;
+}
+
 #[tokio::test]
 async fn poll_response_triggers_wf_error() {
     let mut t = TestHistoryBuilder::default();

crates/sdk-core/src/worker/workflow/managed_run.rs

@@ -405,7 +405,7 @@ impl ManagedRun {
             }) {
                 ActivationCompleteOutcome::ReportWFTFail(FailedActivationWFTReport::Report(
                     tt,
-                    WorkflowTaskFailedCause::Unspecified,
+                    WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure,
                     Failure::application_failure(reason, true).into(),
                 ))
             } else {
@@ -888,7 +888,7 @@ impl ManagedRun {
             // We just checked it is some, unwrap OK.
             let c = self.completion_waiting_on_page_fetch.take().unwrap();
             let run_upd = self.failed_completion(
-                WorkflowTaskFailedCause::Unspecified,
+                WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure,
                 info.reason,
                 Failure::application_failure(info.message, false).into(),
                 true,
@@ -1022,7 +1022,7 @@ impl ManagedRun {
                     let fail_cause = if matches!(&fail.source, WFMachinesError::Nondeterminism(_)) {
                         WorkflowTaskFailedCause::NonDeterministicError
                     } else {
-                        WorkflowTaskFailedCause::Unspecified
+                        WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure
                     };
                     self.failed_completion(
                         fail_cause,

crates/sdk-core/src/worker/workflow/workflow_stream.rs

@@ -288,17 +288,24 @@ impl WFStream {
                     failure,
                     is_autocomplete,
                     ..
-                } => rh.failed_completion(
-                    failure.force_cause(),
-                    if is_autocomplete {
-                        EvictionReason::Unspecified
-                    } else {
-                        EvictionReason::LangFail
-                    },
-                    failure,
-                    is_autocomplete,
-                    complete.response_tx,
-                ),
+                } => {
+                    let forced_cause = failure.force_cause();
+                    rh.failed_completion(
+                        if forced_cause == WorkflowTaskFailedCause::Unspecified {
+                            WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure
+                        } else {
+                            forced_cause
+                        },
+                        if is_autocomplete {
+                            EvictionReason::Unspecified
+                        } else {
+                            EvictionReason::LangFail
+                        },
+                        failure,
+                        is_autocomplete,
+                        complete.response_tx,
+                    )
+                }
             },
             NewOrFetchedComplete::Fetched(update, paginator) => {
                 rh.fetched_page_completion(update, *paginator)

crates/sdk-core/tests/integ_tests/workflow_tests/determinism.rs

@@ -121,11 +121,15 @@ async fn test_panic_wf_task_rejected_properly() {
     let t = canned_histories::workflow_fails_with_failure_after_timer("1");
     let mock = mock_worker_client();
     let mut mh = MockPollCfg::from_resp_batches(wf_id, t, [1, 2, 2], mock);
-    // We should see one wft failure which has unspecified cause, since panics don't have a defined
+    // We should see one wft failure which has the default cause, since panics don't have a defined
     // type.
     mh.num_expected_fails = 1;
-    mh.expect_fail_wft_matcher =
-        Box::new(|_, cause, _| matches!(cause, WorkflowTaskFailedCause::Unspecified));
+    mh.expect_fail_wft_matcher = Box::new(|_, cause, _| {
+        matches!(
+            cause,
+            WorkflowTaskFailedCause::WorkflowWorkerUnhandledFailure
+        )
+    });
     let mut worker = mock_sdk(mh);
 
     worker.register_wf(wf_type.to_owned(), timer_wf_fails_once);