Set WFT Failed cause on grpc message too large incidents (#1229)

Author: Sushisource

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

@@ -695,6 +695,7 @@ async fn legacy_query_too_large_responds_with_failure() {
                 assert!(
                     matches!(&result, LegacyQueryResult::Failed(f) if
                         f.failure.as_ref().unwrap().message == "GRPC Message too large"
+                        && f.force_cause() == WorkflowTaskFailedCause::GrpcMessageTooLarge
                     ),
                     "Expected query failure with message-too-large, got: {:?}",
                     matches!(&result, LegacyQueryResult::Failed(_)),

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

@@ -3109,6 +3109,8 @@ async fn grpc_message_too_large_doesnt_spam_task_fails() {
             Ok(Default::default())
         }
     }));
+    mh.expect_fail_wft_matcher =
+        Box::new(|_, cause, _| *cause == WorkflowTaskFailedCause::GrpcMessageTooLarge);
 
     let mut mock = build_mock_pollers(mh);
     mock.worker_cfg(|wc| wc.max_cached_workflows = 1);

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

@@ -416,22 +416,7 @@ impl Workflows {
                         Err(e)
                             if e.metadata().contains_key(MESSAGE_TOO_LARGE_KEY) && attempt < 2 =>
                         {
-                            let failure = Failure {
-                                failure: Some(
-                                    temporalio_common::protos::temporal::api::failure::v1::Failure {
-                                        message: "GRPC Message too large".to_string(),
-                                        failure_info: Some(FailureInfo::ApplicationFailureInfo(
-                                            ApplicationFailureInfo {
-                                                r#type: "GrpcMessageTooLarge".to_string(),
-                                                non_retryable: true,
-                                                ..Default::default()
-                                            },
-                                        )),
-                                        ..Default::default()
-                                    },
-                                ),
-                                force_cause: 0,
-                            };
+                            let failure = make_grpc_message_too_large_failure();
                             let new_outcome = FailedActivationWFTReport::Report(
                                 task_token,
                                 WorkflowTaskFailedCause::GrpcMessageTooLarge,
@@ -860,22 +845,7 @@ impl Workflows {
             }
             Err(e) if e.metadata().contains_key(MESSAGE_TOO_LARGE_KEY) => {
                 warn!(error=%e, "Query response too large, responding with failure");
-                let failure = Failure {
-                    failure: Some(
-                        temporalio_common::protos::temporal::api::failure::v1::Failure {
-                            message: "GRPC Message too large".to_string(),
-                            failure_info: Some(FailureInfo::ApplicationFailureInfo(
-                                ApplicationFailureInfo {
-                                    r#type: "GrpcMessageTooLarge".to_string(),
-                                    non_retryable: true,
-                                    ..Default::default()
-                                },
-                            )),
-                            ..Default::default()
-                        },
-                    ),
-                    force_cause: 0,
-                };
+                let failure = make_grpc_message_too_large_failure();
                 if let Err(e2) = self
                     .client
                     .respond_legacy_query(tt, LegacyQueryResult::Failed(failure))
@@ -1739,6 +1709,25 @@ fn prepare_to_ship_activation(wfa: &mut WorkflowActivation) {
     });
 }
 
+fn make_grpc_message_too_large_failure() -> Failure {
+    Failure {
+        failure: Some(
+            temporalio_common::protos::temporal::api::failure::v1::Failure {
+                message: "GRPC Message too large".to_string(),
+                failure_info: Some(FailureInfo::ApplicationFailureInfo(
+                    ApplicationFailureInfo {
+                        r#type: "GrpcMessageTooLarge".to_string(),
+                        non_retryable: true,
+                        ..Default::default()
+                    },
+                )),
+                ..Default::default()
+            },
+        ),
+        force_cause: WorkflowTaskFailedCause::GrpcMessageTooLarge as i32,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;