Fix ScheduleToClose bug

Author: dandavison

chasm/lib/activity/activity_tasks.go

@@ -103,18 +103,12 @@ func newScheduleToCloseTimeoutTaskExecutor() *scheduleToCloseTimeoutTaskExecutor
 }
 
 func (e *scheduleToCloseTimeoutTaskExecutor) Validate(
-	ctx chasm.Context,
+	_ chasm.Context,
 	activity *Activity,
 	_ chasm.TaskAttributes,
-	task *activitypb.ScheduleToCloseTimeoutTask,
+	_ *activitypb.ScheduleToCloseTimeoutTask,
 ) (bool, error) {
-	attempt, err := activity.LastAttempt.Get(ctx)
-	if err != nil {
-		return false, err
-	}
-
-	valid := TransitionTimedOut.Possible(activity) && task.Attempt == attempt.Count
-	return valid, nil
+	return TransitionTimedOut.Possible(activity), nil
 }
 
 func (e *scheduleToCloseTimeoutTaskExecutor) Execute(

chasm/lib/activity/gen/activitypb/v1/tasks.pb.go

@@ -17,9 +17,6 @@ message ScheduleToStartTimeoutTask {
 }
 
 message ScheduleToCloseTimeoutTask {
-  // The current attempt number for this activity execution. Since task validation/exec happen outside of a lock, we
-  // need to guard against any concurrent operations where the originally intended task may be outdated.
-  int32 attempt = 1;
 }
 
 message StartToCloseTimeoutTask {

chasm/lib/activity/proto/v1/tasks.proto

@@ -62,9 +62,7 @@ var TransitionScheduled = chasm.NewTransition(
 				chasm.TaskAttributes{
 					ScheduledTime: currentTime.Add(timeout),
 				},
-				&activitypb.ScheduleToCloseTimeoutTask{
-					Attempt: attempt.GetCount(),
-				})
+				&activitypb.ScheduleToCloseTimeoutTask{})
 		}
 
 		ctx.AddTask(

chasm/lib/activity/statemachine.go

@@ -760,6 +760,67 @@ func (s *standaloneActivityTestSuite) TestCompletedActivity_CannotTerminate() {
 	require.Error(t, err)
 }
 
+func (s *standaloneActivityTestSuite) TestScheduleToCloseTimeout_WithRetry() {
+	t := s.T()
+	ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second)
+	defer cancel()
+	activityID := testcore.RandomizeStr(t.Name())
+	taskQueue := testcore.RandomizeStr(t.Name())
+
+	// Start an activity
+	startResp, err := s.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{
+		Namespace:  s.Namespace().String(),
+		ActivityId: activityID,
+		ActivityType: &commonpb.ActivityType{
+			Name: "test-activity-type",
+		},
+		Options: &activitypb.ActivityOptions{
+			TaskQueue: &taskqueuepb.TaskQueue{
+				Name: taskQueue,
+			},
+			// It's not possible to guarantee (e.g. via NextRetryDelay or RetryPolicy) that a retry
+			// will start with a delay <1s because of the use of TimerProcessorMaxTimeShift in the
+			// timer queue. Therefore we allow 1s for the ActivityDispatchTask to be executed, and
+			// time out the activity 1s into Attempt 2.
+			ScheduleToCloseTimeout: durationpb.New(2 * time.Second),
+		},
+	})
+	require.NoError(t, err)
+
+	// Fail attempt 1, causing the attempt counter to increment.
+	pollTaskResp, err := s.pollActivityTaskQueue(ctx, taskQueue)
+	require.NoError(t, err)
+	_, err = s.FrontendClient().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{
+		Namespace: s.Namespace().String(),
+		TaskToken: pollTaskResp.TaskToken,
+		Failure: &failurepb.Failure{
+			Message: "Retryable failure",
+			FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{
+				NonRetryable:   false,
+				NextRetryDelay: durationpb.New(1 * time.Second),
+			}},
+		},
+	})
+	require.NoError(t, err)
+	pollTaskResp, err = s.pollActivityTaskQueue(ctx, taskQueue)
+	require.NoError(t, err)
+
+	// Wait for schedule-to-close timeout.
+	pollResp, err := s.FrontendClient().PollActivityExecution(ctx, &workflowservice.PollActivityExecutionRequest{
+		Namespace:      s.Namespace().String(),
+		ActivityId:     activityID,
+		RunId:          startResp.RunId,
+		IncludeInfo:    true,
+		IncludeOutcome: true,
+		WaitPolicy: &workflowservice.PollActivityExecutionRequest_WaitCompletion{
+			WaitCompletion: &workflowservice.PollActivityExecutionRequest_CompletionWaitOptions{},
+		},
+	})
+	require.NoError(t, err)
+	require.Equal(t, enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT, pollResp.GetInfo().GetStatus())
+	require.Equal(t, enumspb.TIMEOUT_TYPE_SCHEDULE_TO_CLOSE, pollResp.GetFailure().GetTimeoutFailureInfo().GetTimeoutType())
+}
+
 // TestStartToCloseTimeout tests that a start-to-close timeout is recorded after the activity is
 // started. It also verifies that PollActivityExecution can be used to poll for a TimedOut state
 // change caused by execution of a timer task.
@@ -865,11 +926,6 @@ func (s *standaloneActivityTestSuite) TestStartToCloseTimeout() {
 		"expected StartToCloseTimeout but is %s", pollResp.GetFailure().GetTimeoutFailureInfo().GetTimeoutType())
 }
 
-func (s *standaloneActivityTestSuite) TestScheduleToCloseTimeout() {
-	// TODO implement when we have PollActivityExecution. Make sure we check the attempt vs. outcome failure population.
-	s.T().Skip("Temporarily disabled")
-}
-
 func (s *standaloneActivityTestSuite) TestPollActivityExecution_NoWait() {
 	t := s.T()
 	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)