Don't retry on grpc message too large errors (#916)

* Don't retry on grpc message too large errors

* Update client/src/retry.rs

Co-authored-by: Chad Retz <chad@temporal.io>

* Add additional error messages, share key, add ApplicationFailureInfo

* Refactor cloud tests to a separate test library and CI job

* Linting

* Allow override of test client, add GRPC test

* Formatting

* Init integ telem in new constructor

* Use same namespace env var as core starter

* Simplify cloud test command, no need for runner

* Simplify cloud test command, no need for runner

---------

Co-authored-by: Chad Retz <chad@temporal.io>

Author: tconley1428

.github/workflows/per-pr.yml

@@ -76,11 +76,6 @@ jobs:
 
   integ-tests:
     name: Integ tests
-    env:
-      TEMPORAL_CLOUD_ADDRESS: https://${{ vars.TEMPORAL_CLIENT_NAMESPACE }}.tmprl.cloud:7233
-      TEMPORAL_CLOUD_NAMESPACE: ${{ vars.TEMPORAL_CLIENT_NAMESPACE }}
-      TEMPORAL_CLIENT_CERT: ${{ secrets.TEMPORAL_CLIENT_CERT }}
-      TEMPORAL_CLIENT_KEY: ${{ secrets.TEMPORAL_CLIENT_KEY }}
     timeout-minutes: 20
     strategy:
       fail-fast: false
@@ -111,6 +106,31 @@ jobs:
       - uses: Swatinem/rust-cache@v2
       - run: cargo integ-test
 
+  cloud-tests:
+    if: github.repository_owner == 'temporalio'
+    name: Cloud tests
+    env:
+      TEMPORAL_CLOUD_ADDRESS: https://${{ vars.TEMPORAL_CLIENT_NAMESPACE }}.tmprl.cloud:7233
+      TEMPORAL_NAMESPACE: ${{ vars.TEMPORAL_CLIENT_NAMESPACE }}
+      TEMPORAL_CLIENT_CERT: ${{ secrets.TEMPORAL_CLIENT_CERT }}
+      TEMPORAL_CLIENT_KEY: ${{ secrets.TEMPORAL_CLIENT_KEY }}
+    timeout-minutes: 20
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: 1.85.0
+      - name: Install protoc
+        uses: arduino/setup-protoc@v3
+        with:
+          # TODO: Upgrade proto once https://github.com/arduino/setup-protoc/issues/99 is fixed
+          version: '23.x'
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo test --test cloud_tests
+
   docker-integ-tests:
     name: Docker integ tests
     env:

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};
@@ -201,7 +201,11 @@ where
 {
     type OutError = tonic::Status;
 
-    fn handle(&mut self, current_attempt: usize, e: tonic::Status) -> RetryPolicy<tonic::Status> {
+    fn handle(
+        &mut self,
+        current_attempt: usize,
+        mut e: tonic::Status,
+    ) -> RetryPolicy<tonic::Status> {
         // 0 max retries means unlimited retries
         if self.max_retries > 0 && current_attempt >= self.max_retries {
             return RetryPolicy::ForwardError(e);
@@ -213,6 +217,24 @@ 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"))
+        {
+            // Leave a marker so we don't have duplicate detection logic in the workflow
+            e.metadata_mut().insert(
+                MESSAGE_TOO_LARGE_KEY,
+                tonic::metadata::MetadataValue::from(0),
+            );
+            return RetryPolicy::ForwardError(e);
+        }
+
         // Task polls are OK with being cancelled or running into the timeout because there's
         // nothing to do but retry anyway
         let long_poll_allowed = self.call_type == CallType::TaskLongPoll
@@ -423,6 +445,47 @@ mod tests {
         assert_matches!(result, RetryPolicy::ForwardError(_))
     }
 
+    #[tokio::test]
+    async fn message_too_large_not_retried() {
+        let mut err_handler = TonicErrorHandler::new_with_clock(
+            CallInfo {
+                call_type: CallType::TaskLongPoll,
+                call_name: POLL_WORKFLOW_METH_NAME,
+                retry_cfg: TEST_RETRY_CONFIG,
+                retry_short_circuit: None,
+            },
+            TEST_RETRY_CONFIG,
+            FixedClock(Instant::now()),
+            FixedClock(Instant::now()),
+        );
+        let result = err_handler.handle(
+            1,
+            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(_));
+    }
+
     #[rstest::rstest]
     #[tokio::test]
     async fn task_poll_retries_forever<R>(

core/Cargo.toml

@@ -120,6 +120,11 @@ name = "global_metric_tests"
 path = "../tests/global_metric_tests.rs"
 test = false
 
+[[test]]
+name = "cloud_tests"
+path = "../tests/cloud_tests.rs"
+test = false
+
 [[bench]]
 name = "workflow_replay"
 harness = false