Add additional error messages, share key, add ApplicationFailureInfo

Author: tconley1428

client/src/lib.rs

@@ -86,6 +86,10 @@ use uuid::Uuid;
 static CLIENT_NAME_HEADER_KEY: &str = "client-name";
 static CLIENT_VERSION_HEADER_KEY: &str = "client-version";
 static TEMPORAL_NAMESPACE_HEADER_KEY: &str = "temporal-namespace";
+
+/// Key used to communicate when a GRPC message is too large
+pub static MESSAGE_TOO_LARGE_KEY: &str = "message-too-large";
+
 /// The server times out polls after 60 seconds. Set our timeout to be slightly beyond that.
 const LONG_POLL_TIMEOUT: Duration = Duration::from_secs(70);
 const OTHER_CALL_TIMEOUT: Duration = Duration::from_secs(30);

client/src/retry.rs

@@ -1,6 +1,6 @@
 use crate::{
-    Client, IsWorkerTaskLongPoll, NamespacedClient, NoRetryOnMatching, Result, RetryConfig,
-    raw::IsUserLongPoll,
+    Client, IsWorkerTaskLongPoll, MESSAGE_TOO_LARGE_KEY, NamespacedClient, NoRetryOnMatching,
+    Result, RetryConfig, raw::IsUserLongPoll,
 };
 use backoff::{Clock, SystemClock, backoff::Backoff, exponential::ExponentialBackoff};
 use futures_retry::{ErrorHandler, FutureRetry, RetryPolicy};
@@ -219,14 +219,19 @@ where
 
         // Short circuit if message is too large - this is not retryable
         if e.code() == Code::ResourceExhausted
-            && (e.message()
-                .starts_with("grpc: received message larger than max") ||
-                e.message().starts_with("grpc: message after decompression larger than max") ||
-                e.message().starts_with("grpc: received message after decompression larger than max")
+            && (e
+                .message()
+                .starts_with("grpc: received message larger than max")
+                || e.message()
+                    .starts_with("grpc: message after decompression larger than max")
+                || e.message()
+                    .starts_with("grpc: received message after decompression larger than max"))
         {
             // Leave a marker so we don't have duplicate detection logic in the workflow
-            e.metadata_mut()
-                .insert("message-too-large", tonic::metadata::MetadataValue::from(0));
+            e.metadata_mut().insert(
+                MESSAGE_TOO_LARGE_KEY,
+                tonic::metadata::MetadataValue::from(0),
+            );
             return RetryPolicy::ForwardError(e);
         }
 
@@ -455,7 +460,28 @@ mod tests {
         );
         let result = err_handler.handle(
             1,
-            Status::new(Code::ResourceExhausted, "received message larger than max"),
+            Status::new(
+                Code::ResourceExhausted,
+                "grpc: received message larger than max",
+            ),
+        );
+        assert_matches!(result, RetryPolicy::ForwardError(_));
+
+        let result = err_handler.handle(
+            1,
+            Status::new(
+                Code::ResourceExhausted,
+                "grpc: message after decompression larger than max",
+            ),
+        );
+        assert_matches!(result, RetryPolicy::ForwardError(_));
+
+        let result = err_handler.handle(
+            1,
+            Status::new(
+                Code::ResourceExhausted,
+                "grpc: received message after decompression larger than max",
+            ),
         );
         assert_matches!(result, RetryPolicy::ForwardError(_));
     }

core/src/worker/workflow/mod.rs

@@ -57,6 +57,7 @@ use std::{
     thread,
     time::{Duration, Instant},
 };
+use temporal_client::MESSAGE_TOO_LARGE_KEY;
 use temporal_sdk_core_api::{
     errors::{CompleteWfError, PollError},
     worker::{ActivitySlotKind, WorkerConfig, WorkflowSlotKind},
@@ -78,6 +79,7 @@ use temporal_sdk_core_protos::{
         command::v1::{Command as ProtoCommand, Command, command::Attributes},
         common::v1::{Memo, MeteringMetadata, RetryPolicy, SearchAttributes, WorkflowExecution},
         enums::v1::{VersioningBehavior, WorkflowTaskFailedCause},
+        failure::v1::{ApplicationFailureInfo, failure::FailureInfo},
         protocol::v1::Message as ProtocolMessage,
         query::v1::WorkflowQuery,
         sdk::v1::{UserMetadata, WorkflowTaskCompletedMetadata},
@@ -388,11 +390,18 @@ impl Workflows {
                                 response.activity_tasks,
                             );
                         }
-                        Err(e) if e.metadata().contains_key("message-too-large") => {
+                        Err(e) if e.metadata().contains_key(MESSAGE_TOO_LARGE_KEY) => {
                             let failure = Failure {
                                 failure: Some(
                                     temporal_sdk_core_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()
                                     },
                                 ),

tests/integ_tests/worker_tests.rs

@@ -159,14 +159,14 @@ async fn resource_based_few_pollers_guarantees_non_sticky_poll() {
     worker.run_until_done().await.unwrap();
 }
 
-static OVERSIZE_GRPC_MESSAGE_RUN: AtomicBool = AtomicBool::new(false);
 #[tokio::test]
 async fn oversize_grpc_message() {
     let wf_name = "oversize_grpc_message";
     let mut starter = CoreWfStarter::new(wf_name);
     starter.worker_config.no_remote_activities(true);
     let mut core = starter.worker().await;
 
+    static OVERSIZE_GRPC_MESSAGE_RUN: AtomicBool = AtomicBool::new(false);
     core.register_wf(wf_name.to_owned(), |_ctx: WfContext| async move {
         if OVERSIZE_GRPC_MESSAGE_RUN.load(Relaxed) {
             Ok(vec![].into())