From 356e0b671804f6299fe9c97920a937e41932c295 Mon Sep 17 00:00:00 2001
From: Chetan Gowda <gow@users.noreply.github.com>
Date: Wed, 5 Jun 2024 10:33:17 -0700
Subject: [PATCH] Next retry delay should not be bounded by retry policy
 MaximumInterval (#6063)

When an activity specifies `next_retry_delay` in the application
failure, we should be ignoring the max interval set in the retry policy.
This PR allows it.

The intention of the application here is to override the retry policy
and customize the next attempt time.

https://github.com/temporalio/api/blob/5b356b506e0b86ba26dc6f795bfb011eeb4dfa6e/temporal/api/failure/v1/message.proto#L49

Added unit test.

N/A

N/A

No
---
 service/history/workflow/activity.go          |  8 +++-
 ...utable_state_impl_restart_activity_test.go | 38 +++++++++++++++++++
 2 files changed, 45 insertions(+), 1 deletion(-)

diff --git a/service/history/workflow/activity.go b/service/history/workflow/activity.go
index 40cbfd03189..2d504411ab8 100644
--- a/service/history/workflow/activity.go
+++ b/service/history/workflow/activity.go
@@ -75,14 +75,20 @@ func newActivityVisitor(
 	}
 
 	now := timesource.Now().In(time.UTC)
+	retryMaxInterval := ai.RetryMaximumInterval
 	delay := nextRetryDelayFrom(failure)
 
+	// if a delay is specified by the application it should override the maximum interval set by the retry policy.
+	if delay != nil {
+		retryMaxInterval = durationpb.New(*delay)
+	}
+
 	backoff, retryState := nextBackoffInterval(
 		now,
 		ai.Attempt,
 		ai.RetryMaximumAttempts,
 		ai.RetryInitialInterval,
-		ai.RetryMaximumInterval,
+		retryMaxInterval,
 		ai.RetryExpirationTime,
 		ai.RetryBackoffCoefficient,
 		makeBackoffAlgorithm(delay),
diff --git a/service/history/workflow/mutable_state_impl_restart_activity_test.go b/service/history/workflow/mutable_state_impl_restart_activity_test.go
index 413f063b57f..2164b55f5e7 100644
--- a/service/history/workflow/mutable_state_impl_restart_activity_test.go
+++ b/service/history/workflow/mutable_state_impl_restart_activity_test.go
@@ -188,6 +188,44 @@ func (s *retryActivitySuite) TestRetryActivity_should_be_scheduled_when_next_bac
 	s.assertTruncateFailureCalled()
 }
 
+// TestRetryActivity_should_be_scheduled_when_next_retry_delay_is_set asserts that the activity is retried after NextRetryDelay period specified in the application failure.
+func (s *retryActivitySuite) TestRetryActivity_should_be_scheduled_when_next_retry_delay_is_set() {
+	s.mutableState.timeSource = s.timeSource
+	taskGeneratorMock := NewMockTaskGenerator(s.controller)
+	nextAttempt := s.activity.Attempt + 1
+	expectedScheduledTime := s.timeSource.Now().Add(time.Minute).UTC()
+	taskGeneratorMock.EXPECT().GenerateActivityRetryTasks(s.activity.ScheduledEventId, expectedScheduledTime, nextAttempt)
+	s.mutableState.taskGenerator = taskGeneratorMock
+
+	s.failure.GetApplicationFailureInfo().NextRetryDelay = durationpb.New(time.Minute)
+	_, err := s.mutableState.RetryActivity(s.activity, s.failure)
+	s.NoError(err)
+	s.Equal(s.onActivityCreate.mutableStateApproximateSize-s.onActivityCreate.activitySize+s.activity.Size(), s.mutableState.approximateSize)
+	s.Equal(s.activity.Version, s.mutableState.currentVersion)
+	s.Equal(s.activity.Attempt, nextAttempt)
+
+	s.Equal(expectedScheduledTime, s.activity.ScheduledTime.AsTime(), "Activity scheduled time is incorrect")
+	s.assertTruncateFailureCalled()
+}
+
+func (s *retryActivitySuite) TestRetryActivity_next_retry_delay_should_override_max_interval() {
+	s.mutableState.timeSource = s.timeSource
+	taskGeneratorMock := NewMockTaskGenerator(s.controller)
+	nextAttempt := s.activity.Attempt + 1
+	expectedScheduledTime := s.timeSource.Now().Add(3 * time.Minute).UTC()
+	taskGeneratorMock.EXPECT().GenerateActivityRetryTasks(s.activity.ScheduledEventId, expectedScheduledTime, nextAttempt)
+	s.mutableState.taskGenerator = taskGeneratorMock
+
+	s.failure.GetApplicationFailureInfo().NextRetryDelay = durationpb.New(3 * time.Minute)
+	s.activity.RetryMaximumInterval = durationpb.New(2 * time.Minute) // set retry max interval to be less than next retry delay duration.
+	_, err := s.mutableState.RetryActivity(s.activity, s.failure)
+	s.NoError(err)
+	s.Equal(s.activity.Attempt, nextAttempt)
+
+	s.Equal(expectedScheduledTime, s.activity.ScheduledTime.AsTime(), "Activity scheduled time is incorrect")
+	s.assertTruncateFailureCalled()
+}
+
 func (s *retryActivitySuite) TestRetryActivity_when_no_next_backoff_interval_should_fail() {
 	taskGeneratorMock := NewMockTaskGenerator(s.controller)
 	s.mutableState.taskGenerator = taskGeneratorMock