Merge pull request #120 from cschleiden/update-sub-workflow-retries

Adjust sub-workflow retry behavior

Author: cschleiden

internal/workflow/executor_test.go

@@ -317,8 +317,8 @@ func Test_ExecuteWorkflowWithSelector(t *testing.T) {
 	require.Equal(t, 1, workflowWithSelectorHits)
 	require.Len(t, e.workflowState.Commands(), 2)
 
-	require.IsType(t, &command.ScheduleTimerCommand{}, e.workflowState.Commands()[0])
-	require.IsType(t, &command.ScheduleActivityCommand{}, e.workflowState.Commands()[1])
+	require.IsType(t, &command.ScheduleActivityCommand{}, e.workflowState.Commands()[0])
+	require.IsType(t, &command.ScheduleTimerCommand{}, e.workflowState.Commands()[1])
 }
 
 func Test_ExecuteNewEvents(t *testing.T) {

workflow/retries.go

@@ -30,16 +30,20 @@ var DefaultRetryOptions = RetryOptions{
 }
 
 func withRetries[T any](ctx sync.Context, retryOptions RetryOptions, fn func(ctx sync.Context, attempt int) Future[T]) Future[T] {
+	attempt := 0
+	firstAttempt := Now(ctx)
+
+	f := fn(ctx, attempt)
+
 	if retryOptions.MaxAttempts <= 1 {
 		// Short-circuit if we don't need to retry
-		return fn(ctx, 0)
+		return f
 	}
 
+	// Start a separate co-routine for retries
 	r := sync.NewFuture[T]()
 
 	sync.Go(ctx, func(ctx sync.Context) {
-		firstAttempt := Now(ctx)
-
 		var result T
 		var err error
 
@@ -48,32 +52,38 @@ func withRetries[T any](ctx sync.Context, retryOptions RetryOptions, fn func(ctx
 			retryExpiration = firstAttempt.Add(retryOptions.RetryTimeout)
 		}
 
-		for attempt := 0; attempt < retryOptions.MaxAttempts; attempt++ {
-			if !retryExpiration.IsZero() && Now(ctx).After(retryExpiration) {
-				// Reached maximum retry time, abort retries
+		for {
+			// Wait for active operation to finish
+			result, err = f.Get(ctx)
+			if err == nil {
 				break
 			}
 
-			result, err = fn(ctx, attempt).Get(ctx)
-			if err != nil {
-				if err == sync.Canceled {
-					break
-				}
+			if err == sync.Canceled {
+				break
+			}
 
-				backoffDuration := time.Duration(float64(retryOptions.FirstRetryInterval) * math.Pow(retryOptions.BackoffCoefficient, float64(attempt)))
-				if retryOptions.MaxRetryInterval > 0 {
-					backoffDuration = time.Duration(math.Min(float64(backoffDuration), float64(retryOptions.MaxRetryInterval)))
-				}
+			backoffDuration := time.Duration(float64(retryOptions.FirstRetryInterval) * math.Pow(retryOptions.BackoffCoefficient, float64(attempt)))
+			if retryOptions.MaxRetryInterval > 0 {
+				backoffDuration = time.Duration(math.Min(float64(backoffDuration), float64(retryOptions.MaxRetryInterval)))
+			}
 
-				if err := Sleep(ctx, backoffDuration); err != nil {
-					r.Set(*new(T), err)
-					return
-				}
+			if err := Sleep(ctx, backoffDuration); err != nil {
+				r.Set(*new(T), err)
+				return
+			}
 
-				continue
+			if !retryExpiration.IsZero() && Now(ctx).After(retryExpiration) {
+				// Reached maximum retry time, abort retries
+				break
+			}
+
+			attempt++
+			if attempt >= retryOptions.MaxAttempts {
+				break
 			}
 
-			break
+			f = fn(ctx, attempt+1)
 		}
 
 		r.Set(result, err)

workflow/subworkflow.go

@@ -22,9 +22,16 @@ type SubWorkflowOptions struct {
 	RetryOptions RetryOptions
 }
 
-var DefaultSubWorkflowOptions = SubWorkflowOptions{
-	RetryOptions: DefaultRetryOptions,
-}
+var (
+	DefaultSubWorkflowRetryOptions = RetryOptions{
+		// Disable retries by default for sub-workflows
+		MaxAttempts: 1,
+	}
+
+	DefaultSubWorkflowOptions = SubWorkflowOptions{
+		RetryOptions: DefaultSubWorkflowRetryOptions,
+	}
+)
 
 func CreateSubWorkflowInstance[TResult any](ctx sync.Context, options SubWorkflowOptions, workflow interface{}, args ...interface{}) Future[TResult] {
 	return withRetries(ctx, options.RetryOptions, func(ctx sync.Context, attempt int) Future[TResult] {