Add support for NextRetryDelay for local activities (#1456)

Add support for NextRetryDelay for local activities

Author: Quinn-With-Two-Ns

Diff for: internal/error.go

@@ -328,10 +328,11 @@ func NewApplicationError(msg string, errType string, nonRetryable bool, cause er
 
 func NewApplicationErrorWithOptions(msg string, errType string, options ApplicationErrorOptions) error {
 	applicationErr := &ApplicationError{
-		msg:          msg,
-		errType:      errType,
-		cause:        options.Cause,
-		nonRetryable: options.NonRetryable,
+		msg:            msg,
+		errType:        errType,
+		cause:          options.Cause,
+		nonRetryable:   options.NonRetryable,
+		nextRetryDelay: options.NextRetryDelay,
 	}
 	// When return error to user, use EncodedValues as details and data is ready to be decoded by calling Get
 	details := options.Details
@@ -582,6 +583,8 @@ func (e *ApplicationError) Unwrap() error {
 	return e.cause
 }
 
+// NextRetryDelay returns the delay to wait before retrying the activity.
+// a zero value means to use the activities retry policy.
 func (e *ApplicationError) NextRetryDelay() time.Duration { return e.nextRetryDelay }
 
 // Error from error interface

Diff for: internal/internal_task_handlers.go

@@ -1254,7 +1254,7 @@ func (w *workflowExecutionContextImpl) retryLocalActivity(lar *localActivityResu
 		return false
 	}
 
-	retryBackoff := getRetryBackoff(lar, time.Now(), w.wth.dataConverter)
+	retryBackoff := getRetryBackoff(lar, time.Now())
 	if retryBackoff > 0 && retryBackoff <= w.workflowInfo.WorkflowTaskTimeout {
 		// we need a local retry
 		time.AfterFunc(retryBackoff, func() {
@@ -1279,7 +1279,7 @@ func (w *workflowExecutionContextImpl) retryLocalActivity(lar *localActivityResu
 	return false
 }
 
-func getRetryBackoff(lar *localActivityResult, now time.Time, dataConverter converter.DataConverter) time.Duration {
+func getRetryBackoff(lar *localActivityResult, now time.Time) time.Duration {
 	return getRetryBackoffWithNowTime(lar.task.retryPolicy, lar.task.attempt, lar.err, now, lar.task.expireTime)
 }
 
@@ -1291,20 +1291,30 @@ func getRetryBackoffWithNowTime(p *RetryPolicy, attempt int32, err error, now, e
 	if p.MaximumAttempts > 0 && attempt >= p.MaximumAttempts {
 		return noRetryBackoff // max attempt reached
 	}
+
+	var backoffInterval time.Duration
+	// Extract backoff interval from error if it is a retryable error.
+	// Not using errors.As() since we don't want to explore the whole error chain.
+	if applicationErr, ok := err.(*ApplicationError); ok {
+		backoffInterval = applicationErr.nextRetryDelay
+	}
+	// Calculate next backoff interval if the error did not contain the next backoff interval.
 	// attempt starts from 1
-	backoffInterval := time.Duration(float64(p.InitialInterval) * math.Pow(p.BackoffCoefficient, float64(attempt-1)))
-	if backoffInterval <= 0 {
-		// math.Pow() could overflow
-		if p.MaximumInterval > 0 {
+	if backoffInterval == 0 {
+		backoffInterval = time.Duration(float64(p.InitialInterval) * math.Pow(p.BackoffCoefficient, float64(attempt-1)))
+		if backoffInterval <= 0 {
+			// math.Pow() could overflow
+			if p.MaximumInterval > 0 {
+				backoffInterval = p.MaximumInterval
+			}
+		}
+		if p.MaximumInterval > 0 && backoffInterval > p.MaximumInterval {
+			// cap next interval to MaxInterval
 			backoffInterval = p.MaximumInterval
-		} else {
-			return noRetryBackoff
 		}
 	}
-
-	if p.MaximumInterval > 0 && backoffInterval > p.MaximumInterval {
-		// cap next interval to MaxInterval
-		backoffInterval = p.MaximumInterval
+	if backoffInterval <= 0 {
+		return noRetryBackoff
 	}
 
 	nextScheduleTime := now.Add(backoffInterval)

Diff for: internal/internal_workflow_testsuite.go

@@ -1647,7 +1647,7 @@ func (env *testWorkflowEnvironmentImpl) handleLocalActivityResult(result *localA
 		Attempt: 1,
 	}
 	if result.task.retryPolicy != nil && result.err != nil {
-		lar.Backoff = getRetryBackoff(result, env.Now(), env.dataConverter)
+		lar.Backoff = getRetryBackoff(result, env.Now())
 		lar.Attempt = task.attempt
 	}
 	task.callback(lar)

Diff for: internal/internal_workflow_testsuite_test.go

@@ -2412,6 +2412,39 @@ func (s *WorkflowTestSuiteUnitTest) Test_WorkflowWithLocalActivity() {
 	s.Equal("hello local_activity", result)
 }
 
+func (s *WorkflowTestSuiteUnitTest) Test_WorkflowWithLocalActivityNextDelay() {
+	localActivityFn := func(ctx context.Context, delay time.Duration) error {
+		return NewApplicationErrorWithOptions("test delay", "DelayError", ApplicationErrorOptions{
+			NextRetryDelay: delay,
+		})
+	}
+
+	workflowFn := func(ctx Context) (time.Duration, error) {
+		lao := LocalActivityOptions{
+			ScheduleToCloseTimeout: time.Minute,
+			RetryPolicy: &RetryPolicy{
+				MaximumAttempts: 5,
+			},
+		}
+		ctx = WithLocalActivityOptions(ctx, lao)
+		var result string
+		t1 := Now(ctx)
+		f := ExecuteLocalActivity(ctx, localActivityFn, time.Second)
+		_ = f.Get(ctx, &result)
+		t2 := Now(ctx)
+		return t2.Sub(t1), nil
+	}
+
+	env := s.NewTestWorkflowEnvironment()
+	env.ExecuteWorkflow(workflowFn)
+	s.True(env.IsWorkflowCompleted())
+	s.NoError(env.GetWorkflowError())
+	var result time.Duration
+	err := env.GetWorkflowResult(&result)
+	s.NoError(err)
+	s.Equal(4*time.Second, result)
+}
+
 func (s *WorkflowTestSuiteUnitTest) Test_LocalActivity() {
 	localActivityFn := func(ctx context.Context, name string) (string, error) {
 		return "hello " + name, nil

Diff for: test/activity_test.go

@@ -86,6 +86,12 @@ func LocalSleep(_ context.Context, delay time.Duration) error {
 	return nil
 }
 
+func ErrorWithNextDelay(_ context.Context, delay time.Duration) error {
+	return temporal.NewApplicationErrorWithOptions("error with next delay", "NextDelay", temporal.ApplicationErrorOptions{
+		NextRetryDelay: delay,
+	})
+}
+
 func (a *Activities) ActivityToBeCanceled(ctx context.Context) (string, error) {
 	a.append("ActivityToBeCanceled")
 	for {

Diff for: test/integration_test.go

@@ -29,6 +29,7 @@ import (
 	"errors"
 	"flag"
 	"fmt"
+	"math"
 	"math/rand"
 	"os"
 	"strings"
@@ -465,6 +466,16 @@ func (ts *IntegrationTestSuite) TestDeadlockDetectionViaLocalActivity() {
 	ts.True(strings.Contains(applicationErr.Error(), "Potential deadlock detected"))
 }
 
+func (ts *IntegrationTestSuite) TestLocalActivityNextRetryDelay() {
+	var activityExecutionTime time.Duration
+	wfOpts := ts.startWorkflowOptions("test-local-activity-next-retry-delay")
+	wfOpts.WorkflowTaskTimeout = 5 * time.Second
+	err := ts.executeWorkflowWithOption(wfOpts, ts.workflows.LocalActivityNextRetryDelay, &activityExecutionTime)
+	ts.NoError(err)
+	// Check the activity execution time is around 7 seconds
+	ts.LessOrEqual(math.Abs((activityExecutionTime - 7*time.Second).Seconds()), 1.0)
+}
+
 func (ts *IntegrationTestSuite) TestActivityRetryOnError() {
 	var expected []string
 	err := ts.executeWorkflow("test-activity-retry-on-error", ts.workflows.ActivityRetryOnError, &expected)

Diff for: test/workflow_test.go

@@ -101,6 +101,20 @@ func (w *Workflows) DeadlockedWithLocalActivity(ctx workflow.Context) ([]string,
 	return []string{}, nil
 }
 
+func (w *Workflows) LocalActivityNextRetryDelay(ctx workflow.Context) (time.Duration, error) {
+	laCtx := workflow.WithLocalActivityOptions(ctx, workflow.LocalActivityOptions{
+		ScheduleToCloseTimeout: time.Minute,
+		RetryPolicy: &temporal.RetryPolicy{
+			MaximumAttempts: 8,
+		},
+	})
+
+	t1 := workflow.Now(ctx)
+	_ = workflow.ExecuteLocalActivity(laCtx, ErrorWithNextDelay, time.Second).Get(laCtx, nil)
+	t2 := workflow.Now(ctx)
+	return t2.Sub(t1), nil
+}
+
 func (w *Workflows) Panicked(ctx workflow.Context) ([]string, error) {
 	panic("simulated")
 }
@@ -3024,6 +3038,7 @@ func (w *Workflows) register(worker worker.Worker) {
 	worker.RegisterWorkflow(w.UpdateSettingHandlerInHandler)
 	worker.RegisterWorkflow(w.UpdateCancelableWorkflow)
 	worker.RegisterWorkflow(w.UpdateHandlerRegisteredLate)
+	worker.RegisterWorkflow(w.LocalActivityNextRetryDelay)
 
 	worker.RegisterWorkflow(w.child)
 	worker.RegisterWorkflow(w.childWithRetryPolicy)