worflows: activity retry policy

famarting · famarting · commit 4f6c0e1ceca0 · 2024-10-25T13:18:31.000+02:00
Signed-off-by: Fabian Martinez &lt;46371672+famarting@users.noreply.github.com&gt;
diff --git a/examples/go.mod b/examples/go.mod
@@ -26,7 +26,7 @@ require (
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/kr/pretty v0.3.1 // indirect
 	github.com/marusama/semaphore/v2 v2.5.0 // indirect
-	github.com/microsoft/durabletask-go v0.5.0 // indirect
+	github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428 // indirect
 	github.com/xhit/go-str2duration/v2 v2.1.0 // indirect
 	go.opentelemetry.io/otel v1.27.0 // indirect
 	go.opentelemetry.io/otel/metric v1.27.0 // indirect
diff --git a/examples/go.sum b/examples/go.sum
@@ -39,8 +39,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/marusama/semaphore/v2 v2.5.0 h1:o/1QJD9DBYOWRnDhPwDVAXQn6mQYD0gZaS1Tpx6DJGM=
 github.com/marusama/semaphore/v2 v2.5.0/go.mod h1:z9nMiNUekt/LTpTUQdpp+4sJeYqUGpwMHfW0Z8V8fnQ=
-github.com/microsoft/durabletask-go v0.5.0 h1:4DWBgg05wnkV/VwakaiPqZ4cARvATP74ZQJFcXVMC18=
-github.com/microsoft/durabletask-go v0.5.0/go.mod h1:goe2gmMgLptCijMDQ7JsekaR86KjPUG64V9JDXvKBhE=
+github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428 h1:I1yeX4tWqOdBzpRzSbY1TnHU2NI25Pdu6OXUm39emm0=
+github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428/go.mod h1:goe2gmMgLptCijMDQ7JsekaR86KjPUG64V9JDXvKBhE=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
 github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE=
diff --git a/examples/workflow/main.go b/examples/workflow/main.go
@@ -314,7 +314,12 @@ func TestWorkflow(ctx *workflow.WorkflowContext) (any, error) {
 		return nil, err
 	}
 
-	if err := ctx.CallActivity(TestActivity, workflow.ActivityInput(input)).Await(&output); err != nil {
+	if err := ctx.CallActivity(TestActivity, workflow.ActivityInput(input), workflow.RetryPolicy(workflow.ActivityRetryPolicy{
+		MaxAttempts:          3,
+		InitialRetryInterval: 1 * time.Second,
+		BackoffCoefficient:   2,
+		MaxRetryInterval:     3 * time.Second,
+	})).Await(&output); err != nil {
 		return nil, err
 	}
 
diff --git a/go.mod b/go.mod
@@ -7,7 +7,7 @@ require (
 	github.com/go-chi/chi/v5 v5.1.0
 	github.com/golang/mock v1.6.0
 	github.com/google/uuid v1.6.0
-	github.com/microsoft/durabletask-go v0.5.0
+	github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428
 	github.com/stretchr/testify v1.9.0
 	google.golang.org/grpc v1.65.0
 	google.golang.org/protobuf v1.34.2
diff --git a/go.sum b/go.sum
@@ -28,8 +28,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/marusama/semaphore/v2 v2.5.0 h1:o/1QJD9DBYOWRnDhPwDVAXQn6mQYD0gZaS1Tpx6DJGM=
 github.com/marusama/semaphore/v2 v2.5.0/go.mod h1:z9nMiNUekt/LTpTUQdpp+4sJeYqUGpwMHfW0Z8V8fnQ=
-github.com/microsoft/durabletask-go v0.5.0 h1:4DWBgg05wnkV/VwakaiPqZ4cARvATP74ZQJFcXVMC18=
-github.com/microsoft/durabletask-go v0.5.0/go.mod h1:goe2gmMgLptCijMDQ7JsekaR86KjPUG64V9JDXvKBhE=
+github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428 h1:I1yeX4tWqOdBzpRzSbY1TnHU2NI25Pdu6OXUm39emm0=
+github.com/microsoft/durabletask-go v0.5.1-0.20241024170039-0c4afbc95428/go.mod h1:goe2gmMgLptCijMDQ7JsekaR86KjPUG64V9JDXvKBhE=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
diff --git a/workflow/activity_context.go b/workflow/activity_context.go
@@ -17,6 +17,7 @@ package workflow
 import (
 	"context"
 	"encoding/json"
+	"time"
 
 	"google.golang.org/protobuf/types/known/wrapperspb"
 
@@ -38,7 +39,16 @@ func (wfac *ActivityContext) Context() context.Context {
 type callActivityOption func(*callActivityOptions) error
 
 type callActivityOptions struct {
-	rawInput *wrapperspb.StringValue
+	rawInput    *wrapperspb.StringValue
+	retryPolicy *ActivityRetryPolicy
+}
+
+type ActivityRetryPolicy struct {
+	MaxAttempts          int
+	InitialRetryInterval time.Duration
+	BackoffCoefficient   float64
+	MaxRetryInterval     time.Duration
+	RetryTimeout         time.Duration
 }
 
 // ActivityInput is an option to pass a JSON-serializable input
@@ -61,6 +71,26 @@ func ActivityRawInput(input string) callActivityOption {
 	}
 }
 
+func RetryPolicy(policy ActivityRetryPolicy) callActivityOption {
+	return func(opts *callActivityOptions) error {
+		opts.retryPolicy = &policy
+		return nil
+	}
+}
+
+func (opts *callActivityOptions) getRetryPolicy() *task.ActivityRetryPolicy {
+	if opts.retryPolicy == nil {
+		return nil
+	}
+	return &task.ActivityRetryPolicy{
+		MaxAttempts:          opts.retryPolicy.MaxAttempts,
+		InitialRetryInterval: opts.retryPolicy.InitialRetryInterval,
+		BackoffCoefficient:   opts.retryPolicy.BackoffCoefficient,
+		MaxRetryInterval:     opts.retryPolicy.MaxRetryInterval,
+		RetryTimeout:         opts.retryPolicy.RetryTimeout,
+	}
+}
+
 func marshalData(input any) ([]byte, error) {
 	if input == nil {
 		return nil, nil
diff --git a/workflow/context.go b/workflow/context.go
@@ -63,7 +63,7 @@ func (wfc *WorkflowContext) CallActivity(activity interface{}, opts ...callActiv
 		}
 	}
 
-	return wfc.orchestrationContext.CallActivity(activity, task.WithRawActivityInput(options.rawInput.GetValue()))
+	return wfc.orchestrationContext.CallActivity(activity, task.WithRawActivityInput(options.rawInput.GetValue()), task.WithRetryPolicy(options.getRetryPolicy()))
 }
 
 // CallChildWorkflow returns a completable task for a given workflow.

Original file line number	Diff line number	Diff line change
`@@ -314,7 +314,12 @@ func TestWorkflow(ctx *workflow.WorkflowContext) (any, error) {`
`314`	`314`	`return nil, err`
`315`	`315`	`}`
`316`	`316`
`317`		`- if err := ctx.CallActivity(TestActivity, workflow.ActivityInput(input)).Await(&output); err != nil {`
	`317`	`+ if err := ctx.CallActivity(TestActivity, workflow.ActivityInput(input), workflow.RetryPolicy(workflow.ActivityRetryPolicy{`
	`318`	`+ MaxAttempts: 3,`
	`319`	`+ InitialRetryInterval: 1 * time.Second,`
	`320`	`+ BackoffCoefficient: 2,`
	`321`	`+ MaxRetryInterval: 3 * time.Second,`
	`322`	`+ })).Await(&output); err != nil {`
`318`	`323`	`return nil, err`
`319`	`324`	`}`
`320`	`325`
Original file line number	Diff line number	Diff line change
`@@ -63,7 +63,7 @@ func (wfc *WorkflowContext) CallActivity(activity interface{}, opts ...callActiv`
`63`	`63`	`}`
`64`	`64`	`}`
`65`	`65`
`66`		`- return wfc.orchestrationContext.CallActivity(activity, task.WithRawActivityInput(options.rawInput.GetValue()))`
	`66`	`+ return wfc.orchestrationContext.CallActivity(activity, task.WithRawActivityInput(options.rawInput.GetValue()), task.WithRetryPolicy(options.getRetryPolicy()))`
`67`	`67`	`}`
`68`	`68`
`69`	`69`	`// CallChildWorkflow returns a completable task for a given workflow.`