Merge pull request #327 from cschleiden/make-activity-attempt-available

Make retry # available for activity execution

Author: cschleiden

activity/logger.go

@@ -11,3 +11,8 @@ import (
 func Logger(ctx context.Context) *slog.Logger {
 	return activity.GetActivityState(ctx).Logger
 }
+
+// Attempt returns the current attempt of this activity execution
+func Attempt(ctx context.Context) int {
+	return activity.GetActivityState(ctx).Attempt
+}

activitytester/activitytester.go

@@ -15,5 +15,5 @@ func WithActivityTestState(ctx context.Context, activityID, instanceID string, l
 		logger = slog.Default()
 	}
 
-	return activity.WithActivityState(ctx, activity.NewActivityState(activityID, core.NewWorkflowInstance(instanceID, ""), logger))
+	return activity.WithActivityState(ctx, activity.NewActivityState(activityID, 0, core.NewWorkflowInstance(instanceID, ""), logger))
 }

backend/history/activity_scheduled.go

@@ -8,6 +8,8 @@ import (
 type ActivityScheduledAttributes struct {
 	Name string `json:"name,omitempty"`
 
+	Attempt int `json:"attempt,omitempty"`
+
 	Inputs []payload.Payload `json:"inputs,omitempty"`
 
 	Metadata *metadata.WorkflowMetadata `json:"metadata,omitempty"`

backend/test/e2e_activity.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cschleiden/go-workflows/activity"
 	"github.com/cschleiden/go-workflows/client"
 	"github.com/cschleiden/go-workflows/worker"
 	"github.com/cschleiden/go-workflows/workflow"
@@ -75,6 +76,39 @@ var e2eActivityTests = []backendTest{
 			require.NoError(t, err)
 		},
 	},
+	{
+		name: "Activity_ReceiveAttempt",
+		f: func(t *testing.T, ctx context.Context, c *client.Client, w *worker.Worker, b TestBackend) {
+			var maxAttempt int
+
+			a := func(ctx context.Context) error {
+				attempt := activity.Attempt(ctx)
+				maxAttempt = attempt
+
+				if attempt < 2 {
+					return &CustomError{msg: "custom error"}
+				}
+
+				return nil
+			}
+
+			wf := func(ctx workflow.Context) error {
+				_, err := workflow.ExecuteActivity[int](ctx, workflow.ActivityOptions{
+					RetryOptions: workflow.RetryOptions{
+						MaxAttempts: 4,
+					},
+				}, a).Get(ctx)
+
+				return err
+			}
+			register(t, ctx, w, []interface{}{wf}, []interface{}{a})
+
+			_, err := runWorkflowWithResult[bool](t, ctx, c, wf)
+			require.NoError(t, err)
+
+			require.Equal(t, 2, maxAttempt)
+		},
+	},
 	{
 		name: "Activity_ExtendTask",
 		customWorkerOptions: func(w *worker.Options) {

docs/source/includes/_guide.md

@@ -355,6 +355,17 @@ func Activity1(ctx context.Context, name string) (int, error) {
 
 With the default `DefaultActivityOptions`, Activities are retried up to three times when they return an error. If you want to keep automatic retries, but want to avoid them when hitting certain error types, you can wrap an error with `workflow.NewPermanentError`.
 
+```go
+func Activity1(ctx context.Context, name string) (int, error) {
+	// Current retry attempt
+	attempt := activity.Attempt(ctx)
+
+	return http.Do("POST", "https://example.com", name)
+}
+```
+
+`activity.Attempt` returns the current attempt retry.
+
 ## `ContinueAsNew`
 
 ```go

internal/activity/activitystate.go

@@ -10,18 +10,21 @@ import (
 
 type ActivityState struct {
 	ActivityID string
+	Attempt    int
 	Instance   *workflow.Instance
 	Logger     *slog.Logger
 }
 
-func NewActivityState(activityID string, instance *workflow.Instance, logger *slog.Logger) *ActivityState {
+func NewActivityState(activityID string, attempt int, instance *workflow.Instance, logger *slog.Logger) *ActivityState {
 	return &ActivityState{
 		activityID,
+		attempt,
 		instance,
 		logger.With(
 			log.ActivityIDKey, activityID,
 			log.InstanceIDKey, instance.InstanceID,
 			log.ExecutionIDKey, instance.ExecutionID,
+			log.AttemptKey, attempt,
 		)}
 }
 

internal/activity/executor.go

@@ -49,7 +49,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 
 	activity, err := e.r.GetActivity(a.Name)
 	if err != nil {
-		return nil, err
+		return nil, workflowerrors.NewPermanentError(fmt.Errorf("activity not found: %w", err))
 	}
 
 	activityFn := reflect.ValueOf(activity)
@@ -65,6 +65,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 	// Add activity state to context
 	as := NewActivityState(
 		task.Event.ID,
+		a.Attempt,
 		task.WorkflowInstance,
 		e.logger)
 	activityCtx := WithActivityState(ctx, as)
@@ -80,6 +81,7 @@ func (e *Executor) ExecuteActivity(ctx context.Context, task *backend.ActivityTa
 		attribute.String(log.ActivityNameKey, a.Name),
 		attribute.String(log.InstanceIDKey, task.WorkflowInstance.InstanceID),
 		attribute.String(log.ActivityIDKey, task.ID),
+		attribute.Int(log.AttemptKey, a.Attempt),
 	))
 	defer span.End()
 

internal/activity/executor_test.go

@@ -37,7 +37,7 @@ func TestExecutor_ExecuteActivity(t *testing.T) {
 			result: func(t *testing.T, result payload.Payload, err error) {
 				require.Nil(t, result)
 				require.Error(t, err)
-				require.EqualError(t, err, "activity not found")
+				require.EqualError(t, err, "activity not found: activity not found")
 			},
 		},
 		{

internal/command/schedule_activity.go

@@ -12,19 +12,21 @@ type ScheduleActivityCommand struct {
 
 	Name     string
 	Inputs   []payload.Payload
+	Attempt  int
 	Metadata *metadata.WorkflowMetadata
 }
 
 var _ Command = (*ScheduleActivityCommand)(nil)
 
-func NewScheduleActivityCommand(id int64, name string, inputs []payload.Payload, metadata *metadata.WorkflowMetadata) *ScheduleActivityCommand {
+func NewScheduleActivityCommand(id int64, name string, inputs []payload.Payload, attempt int, metadata *metadata.WorkflowMetadata) *ScheduleActivityCommand {
 	return &ScheduleActivityCommand{
 		command: command{
 			id:    id,
 			name:  "ScheduleActivity",
 			state: CommandState_Pending,
 		},
 		Name:     name,
+		Attempt:  attempt,
 		Inputs:   inputs,
 		Metadata: metadata,
 	}
@@ -41,6 +43,7 @@ func (c *ScheduleActivityCommand) Execute(clock clock.Clock) *CommandResult {
 			&history.ActivityScheduledAttributes{
 				Name:     c.Name,
 				Inputs:   c.Inputs,
+				Attempt:  c.Attempt,
 				Metadata: c.Metadata,
 			},
 			history.ScheduleEventID(c.id))

internal/command/schedule_activity_test.go

@@ -37,7 +37,7 @@ func TestScheduleActivityCommand_StateTransitions(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			clock := clock.NewMock()
-			cmd := NewScheduleActivityCommand(1, "activity", []payload.Payload{}, &metadata.WorkflowMetadata{})
+			cmd := NewScheduleActivityCommand(1, "activity", []payload.Payload{}, 0, &metadata.WorkflowMetadata{})
 
 			tt.f(t, cmd, clock)
 		})