Emit GrpcMessageTooLarge as failure_reason for WFT

Author: Quinn-With-Two-Ns

crates/sdk-core/src/telemetry/metrics.rs

@@ -642,6 +642,7 @@ pub(crate) enum FailureReason {
     Timeout,
     NexusOperation(String),
     NexusHandlerError(String),
+    GrpcMessageTooLarge,
 }
 impl Display for FailureReason {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
@@ -651,6 +652,7 @@ impl Display for FailureReason {
             FailureReason::Timeout => "timeout".to_owned(),
             FailureReason::NexusOperation(op) => format!("operation_{op}"),
             FailureReason::NexusHandlerError(op) => format!("handler_error_{op}"),
+            FailureReason::GrpcMessageTooLarge => "GrpcMessageTooLarge".to_owned(),
         };
         write!(f, "{str}")
     }

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

@@ -15,16 +15,10 @@ pub(crate) use driven_workflow::DrivenWorkflow;
 pub(crate) use history_update::HistoryUpdate;
 
 use crate::{
-    MetricsContext,
-    abstractions::{
+    MetricsContext, abstractions::{
         MeteredPermitDealer, TrackedOwnedMeteredSemPermit, UsedMeteredSemPermit, dbg_panic,
         take_cell::TakeCell,
-    },
-    internal_flags::InternalFlags,
-    pollers::TrackedPermittedTqResp,
-    protosext::{ValidPollWFTQResponse, protocol_messages::IncomingProtocolMessage},
-    telemetry::{VecDisplayer, set_trace_subscriber_for_current_thread},
-    worker::{
+    }, internal_flags::InternalFlags, pollers::TrackedPermittedTqResp, protosext::{ValidPollWFTQResponse, protocol_messages::IncomingProtocolMessage}, telemetry::{VecDisplayer, metrics::{self, FailureReason}, set_trace_subscriber_for_current_thread}, worker::{
         LocalActRequest, LocalActivityExecutionResult, LocalActivityResolution,
         PostActivateHookData,
         activities::{ActivitiesFromWFTsHandle, LocalActivityManager},
@@ -37,7 +31,7 @@ use crate::{
             wft_poller::validate_wft,
             workflow_stream::{LocalInput, LocalInputs, WFStream},
         },
-    },
+    }
 };
 use anyhow::anyhow;
 use futures_util::{Stream, StreamExt, future::abortable, stream, stream::BoxStream};
@@ -57,19 +51,15 @@ use std::{
 };
 use temporalio_client::MESSAGE_TOO_LARGE_KEY;
 use temporalio_common::{
-    errors::{CompleteWfError, PollError},
-    protos::{
+    errors::{CompleteWfError, PollError}, protos::{
         TaskToken,
         coresdk::{
             workflow_activation::{
                 QueryWorkflow, WorkflowActivation, WorkflowActivationJob,
                 remove_from_cache::EvictionReason, workflow_activation_job,
             },
             workflow_commands::*,
-            workflow_completion,
-            workflow_completion::{
-                Failure, WorkflowActivationCompletion, workflow_activation_completion,
-            },
+            workflow_completion::{self, Failure, WorkflowActivationCompletion, workflow_activation_completion},
         },
         temporal::api::{
             command::v1::{Command as ProtoCommand, Command, command::Attributes},
@@ -84,8 +74,7 @@ use temporalio_common::{
             taskqueue::v1::StickyExecutionAttributes,
             workflowservice::v1::{PollActivityTaskQueueResponse, get_system_info_response},
         },
-    },
-    worker::{ActivitySlotKind, WorkerConfig, WorkflowSlotKind},
+    }, worker::{ActivitySlotKind, WorkerConfig, WorkflowSlotKind}
 };
 use tokio::{
     sync::{
@@ -134,6 +123,7 @@ pub(crate) struct Workflows {
     local_act_mgr: Option<Arc<LocalActivityManager>>,
     ever_polled: AtomicBool,
     default_versioning_behavior: Option<VersioningBehavior>,
+    metrics: MetricsContext,
 }
 
 pub(crate) struct WorkflowBasics {
@@ -176,6 +166,7 @@ impl Workflows {
         let (fetch_tx, fetch_rx) = unbounded_channel();
         let shutdown_tok = basics.shutdown_token.clone();
         let task_queue = basics.worker_config.task_queue.clone();
+        let metrics = basics.metrics.clone();
         let default_versioning_behavior = basics.default_versioning_behavior;
         let extracted_wft_stream = WFTExtractor::build(
             client.clone(),
@@ -267,6 +258,7 @@ impl Workflows {
             local_act_mgr,
             ever_polled: AtomicBool::new(false),
             default_versioning_behavior,
+            metrics,
         }
     }
 
@@ -431,6 +423,10 @@ impl Workflows {
                             );
                             self.handle_activation_failed(run_id, completion_time, new_outcome)
                                 .await;
+                            self.metrics.with_new_attrs([metrics::failure_reason(
+                                        FailureReason::GrpcMessageTooLarge
+                                    )])
+                                    .wf_task_failed();
                             return Err(e);
                         }
                         e => {

crates/sdk-core/tests/integ_tests/worker_tests.rs

@@ -210,8 +210,12 @@ async fn resource_based_few_pollers_guarantees_non_sticky_poll() {
 
 #[tokio::test]
 async fn oversize_grpc_message() {
+    use crate::common::{prom_metrics, ANY_PORT, NAMESPACE};
     let wf_name = "oversize_grpc_message";
-    let mut starter = CoreWfStarter::new(wf_name);
+    // Enable Prometheus metrics for this test and capture the address
+    let (telemopts, addr, _aborter) = prom_metrics(None);
+    let runtime = CoreRuntime::new_assume_tokio(get_integ_runtime_options(telemopts)).unwrap();
+    let mut starter = CoreWfStarter::new_with_runtime(wf_name, runtime);
     starter
         .worker_config
         .task_types(WorkerTaskTypes::workflow_only());
@@ -238,7 +242,25 @@ async fn oversize_grpc_message() {
             } else {
                 false
             }
-    }))
+    }));
+
+    // Verify the workflow task failure metric includes the GrpcMessageTooLarge reason
+    let tq = starter.get_task_queue();
+    crate::common::eventually(
+        || async {
+            let body = crate::integ_tests::metrics_tests::get_text(format!("http://{addr}/metrics")).await;
+            if body.contains(&format!(
+                "temporal_workflow_task_execution_failed{{failure_reason=\"GrpcMessageTooLarge\",namespace=\"{NAMESPACE}\",service_name=\"temporal-core-sdk\",task_queue=\"{tq}\"}} 1"
+            )) {
+                Ok(())
+            } else {
+                Err(())
+            }
+        },
+        Duration::from_secs(2),
+    )
+    .await
+    .unwrap();
 }
 
 #[tokio::test]