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

@@ -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),

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