Pause activity by rules in executeActivityRetryTimerTask (#7555)

## What changed?
<!-- Describe what has changed in this PR -->
Add code to pause the activity while processing activity retry task.

## Why?
<!-- Tell your future self why have you made these changes -->
Part of the workflow rules efforts. This is another place where we need
to check rules related to the activity.

## How did you test it?
<!-- How have you verified this change? Tested locally? Added a unit
test? Checked in staging env? -->
Add functional test.

## Potential risks
<!-- Assuming the worst case, what can be broken when deploying this
change to production? -->
N/A.
## Documentation
<!-- Have you made sure this change doesn't falsify anything currently
stated in `docs/`? If significant
new behavior is added, have you described that in `docs/`? -->
N/A.

Author: ychebotarev

service/history/timer_queue_active_task_executor.go

@@ -549,6 +549,18 @@ func (t *timerQueueActiveTaskExecutor) executeActivityRetryTimerTask(
 		return consts.ErrWorkflowCompleted
 	}
 
+	err = t.processActivityWorkflowRules(ctx, weContext, mutableState, activityInfo)
+	if err != nil {
+		return err
+	}
+
+	// task can be paused as the result of processing activity workflow rules, so we need to check again
+	if task.Stamp != activityInfo.Stamp || activityInfo.Paused {
+		// if retry task event is from an old stamp of if activity is paused we should ignore the event.
+		release(nil) // release(nil) so mutable state is not unloaded from cache
+		return consts.ErrActivityTaskNotFound
+	}
+
 	taskQueue := &taskqueuepb.TaskQueue{
 		Name: activityInfo.TaskQueue,
 		Kind: enumspb.TASK_QUEUE_KIND_NORMAL,
@@ -880,3 +892,41 @@ func (t *timerQueueActiveTaskExecutor) emitTimeoutMetricScopeWithNamespaceTag(
 		metrics.HeartbeatTimeoutCounter.With(metricsScope).Record(1)
 	}
 }
+
+func (t *timerQueueActiveTaskExecutor) processActivityWorkflowRules(
+	ctx context.Context,
+	weContext historyi.WorkflowContext,
+	ms historyi.MutableState,
+	ai *persistencespb.ActivityInfo,
+) error {
+	if ai.Paused {
+		return nil
+	}
+
+	activityChanged := workflow.ActivityMatchWorkflowRules(ms, t.logger, ai)
+	if !activityChanged {
+		return nil
+	}
+	if ai.Paused {
+		// need to update activity
+		if err := ms.UpdateActivity(ai.ScheduledEventId, func(activityInfo *persistencespb.ActivityInfo, _ historyi.MutableState) error {
+			activityInfo.StartedEventId = common.EmptyEventID
+			activityInfo.StartedTime = nil
+			activityInfo.RequestId = ""
+			return nil
+		}); err != nil {
+			return err
+		}
+
+		// need to update mutable state
+		err := weContext.UpdateWorkflowExecutionAsActive(
+			ctx,
+			t.shardContext,
+		)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}

tests/activity_api_rules_test.go

@@ -40,6 +40,7 @@ import (
 	"go.temporal.io/sdk/temporal"
 	"go.temporal.io/sdk/workflow"
 	"go.temporal.io/server/common/dynamicconfig"
+	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/util"
 	"go.temporal.io/server/tests/testcore"
 )
@@ -72,10 +73,11 @@ type internalRulesTestWorkflow struct {
 	activityFailedCn     chan struct{}
 
 	testSuite *testcore.FunctionalTestBase
+	logger    log.Logger
 	ctx       context.Context
 }
 
-func newInternalRulesTestWorkflow(ctx context.Context, testSuite *testcore.FunctionalTestBase) *internalRulesTestWorkflow {
+func newInternalRulesTestWorkflow(ctx context.Context, testSuite *testcore.FunctionalTestBase, logger log.Logger) *internalRulesTestWorkflow {
 	wf := &internalRulesTestWorkflow{
 		initialRetryInterval:   1 * time.Second,
 		scheduleToCloseTimeout: 30 * time.Minute,
@@ -84,6 +86,7 @@ func newInternalRulesTestWorkflow(ctx context.Context, testSuite *testcore.Funct
 		activityFailedCn:       make(chan struct{}),
 		testSuite:              testSuite,
 		ctx:                    ctx,
+		logger:                 logger,
 	}
 	wf.activityRetryPolicy = &temporal.RetryPolicy{
 		InitialInterval:    wf.initialRetryInterval,
@@ -305,7 +308,7 @@ func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryActivity() {
 	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	defer cancel()
 
-	testWorkflow := newInternalRulesTestWorkflow(ctx, &s.FunctionalTestBase)
+	testWorkflow := newInternalRulesTestWorkflow(ctx, &s.FunctionalTestBase, s.Logger)
 
 	s.Worker().RegisterWorkflow(testWorkflow.WorkflowFuncForRetryActivity)
 	s.Worker().RegisterActivity(testWorkflow.ActivityFuncForRetryActivity)
@@ -339,8 +342,9 @@ func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryActivity() {
 		}
 	}, 5*time.Second, 200*time.Millisecond)
 
-	// Let activity fail
-	err = util.InterruptibleSleep(ctx, 1*time.Second)
+	// Let namespace config propagate.
+	// There is no good way to check if the namespace config has propagated to the history service
+	err = util.InterruptibleSleep(ctx, 4*time.Second)
 	s.NoError(err)
 
 	testWorkflow.activityFailedCn <- struct{}{}
@@ -378,6 +382,11 @@ func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryActivity() {
 		assert.Len(s.T(), nsResp.Rules, 0)
 	}, 5*time.Second, 200*time.Millisecond)
 
+	// Let namespace config propagate.
+	// There is no good way to check if the namespace config has propagated to the history service
+	err = util.InterruptibleSleep(ctx, 4*time.Second)
+	s.NoError(err)
+
 	// unpause the activity
 	_, err = s.FrontendClient().UnpauseActivity(ctx, &workflowservice.UnpauseActivityRequest{
 		Namespace: s.Namespace().String(),
@@ -397,7 +406,7 @@ func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryActivity() {
 			assert.True(t, description.PendingActivities[0].GetActivityType().GetName() == activityType)
 			assert.False(t, description.PendingActivities[0].GetPaused())
 		}
-		assert.Equal(t, int32(1), testWorkflow.startedActivityCount.Load())
+		assert.LessOrEqual(t, int32(1), testWorkflow.startedActivityCount.Load())
 	}, 5*time.Second, 200*time.Millisecond)
 
 	// let activity complete
@@ -409,6 +418,140 @@ func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryActivity() {
 	s.NoError(err)
 }
 
+func (s *ActivityApiRulesClientTestSuite) TestActivityRulesApi_RetryTask() {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	// overall test execution plan:
+	// 1. start workflow
+	// 2. wait for activity to start and fail exactly once
+	// 3. create rule to pause activity
+	// 4. wait for activity to be paused by rule
+	// 5. let activity succeed
+	// 6. Remove the rule so it didn't interfere with the activity
+	// 7. Make sure there is no rules
+	// 6. Unpause the activity. this will also trigger the activity
+	// 7. Wait for activity to be unpaused
+	// 8. Let activity complete
+	// 9. Wait for workflow to finish
+
+	testRetryTaskWorkflow := newInternalRulesTestWorkflow(ctx, &s.FunctionalTestBase, s.Logger)
+
+	// set much longer retry interval to make sure that activity is retried at least once
+	s.initialRetryInterval = 4 * time.Second
+	s.activityRetryPolicy.InitialInterval = s.initialRetryInterval
+
+	s.Worker().RegisterWorkflow(testRetryTaskWorkflow.WorkflowFuncForRetryTask)
+	s.Worker().RegisterActivity(testRetryTaskWorkflow.ActivityFuncForRetryTask)
+
+	// 1. Start workflow
+	workflowRun := s.createWorkflow(ctx, testRetryTaskWorkflow.WorkflowFuncForRetryTask)
+
+	// 2. Wait for activity to start and fail exactly once
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		assert.NoError(t, err)
+		if description.GetPendingActivities() != nil {
+			assert.Len(t, description.PendingActivities, 1)
+		}
+		assert.Equal(t, int32(1), testRetryTaskWorkflow.startedActivityCount.Load())
+	}, 2*time.Second, 200*time.Millisecond)
+
+	// 3. Create rule to pause activity
+	ruleID := "pause-activity"
+	activityType := "ActivityFuncForRetryTask"
+	createRuleRequest := s.createPauseRuleRequest(activityType, ruleID)
+	createRuleResponse, err := s.FrontendClient().CreateWorkflowRule(ctx, createRuleRequest)
+	s.NoError(err)
+	s.NotNil(createRuleResponse)
+
+	// 4. verify that frontend has updated namespaces
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		nsResp, err := s.FrontendClient().ListWorkflowRules(ctx, &workflowservice.ListWorkflowRulesRequest{
+			Namespace: s.Namespace().String(),
+		})
+		assert.NoError(s.T(), err)
+		assert.NotNil(s.T(), nsResp)
+		assert.NotNil(s.T(), nsResp.Rules)
+		if nsResp.GetRules() != nil {
+			assert.Len(s.T(), nsResp.Rules, 1)
+			assert.Equal(s.T(), ruleID, nsResp.Rules[0].Spec.Id)
+		}
+	}, 5*time.Second, 200*time.Millisecond)
+
+	// Let namespace config propagate.
+	// There is no good way to check if the namespace config has propagated to the history service
+	err = util.InterruptibleSleep(ctx, 2*time.Second)
+	s.NoError(err)
+
+	// 5. wait for activity to be paused by rule. This should happen in the activity retry task
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		assert.NoError(t, err)
+		if description.GetPendingActivities() != nil {
+			assert.Len(t, description.PendingActivities, 1)
+			assert.True(t, description.PendingActivities[0].GetActivityType().GetName() == activityType)
+			assert.True(t, description.PendingActivities[0].GetPaused())
+		}
+		assert.Equal(t, int32(1), testRetryTaskWorkflow.startedActivityCount.Load())
+	}, 5*time.Second, 200*time.Millisecond)
+
+	// let activity succeed
+	testRetryTaskWorkflow.letActivitySucceed.Store(true)
+
+	// remove the rule so it didn't interfere with the activity
+	deleteRuleResponse, err := s.FrontendClient().DeleteWorkflowRule(ctx, &workflowservice.DeleteWorkflowRuleRequest{
+		Namespace: s.Namespace().String(),
+		RuleId:    ruleID,
+	})
+	s.NoError(err)
+	s.NotNil(deleteRuleResponse)
+
+	// make sure there is no rules
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		nsResp, err := s.FrontendClient().ListWorkflowRules(ctx, &workflowservice.ListWorkflowRulesRequest{
+			Namespace: s.Namespace().String(),
+		})
+		assert.NoError(s.T(), err)
+		assert.NotNil(s.T(), nsResp)
+		assert.Len(s.T(), nsResp.Rules, 0)
+	}, 5*time.Second, 200*time.Millisecond)
+
+	// Let namespace config propagate.
+	// There is no good way to check if the namespace config has propagated to the history service
+	err = util.InterruptibleSleep(ctx, 2*time.Second)
+	s.NoError(err)
+
+	// unpause the activity. this will also trigger the activity
+	_, err = s.FrontendClient().UnpauseActivity(ctx, &workflowservice.UnpauseActivityRequest{
+		Namespace: s.Namespace().String(),
+		Execution: &commonpb.WorkflowExecution{
+			WorkflowId: workflowRun.GetID(),
+		},
+		Activity: &workflowservice.UnpauseActivityRequest_Type{Type: activityType},
+	})
+	s.NoError(err)
+
+	// wait for activity to be unpaused
+	s.EventuallyWithT(func(t *assert.CollectT) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		assert.NoError(t, err)
+		if description.GetPendingActivities() != nil {
+			assert.Len(t, description.PendingActivities, 1)
+			assert.True(t, description.PendingActivities[0].GetActivityType().GetName() == activityType)
+			assert.False(t, description.PendingActivities[0].GetPaused())
+		}
+		assert.LessOrEqual(t, int32(1), testRetryTaskWorkflow.startedActivityCount.Load())
+	}, 5*time.Second, 200*time.Millisecond)
+
+	// let activity complete
+	testRetryTaskWorkflow.activityCompleteCn <- struct{}{}
+	// wait for workflow to finish
+	var out string
+	err = workflowRun.Get(ctx, &out)
+	s.NoError(err)
+}
+
 func (s *ActivityApiRulesClientTestSuite) createPauseRuleRequest(
 	activityType string, ruleID string,
 ) *workflowservice.CreateWorkflowRuleRequest {