Fix activity timeout regeneration after unpause (#11666)

## What changed?
- Clear an activity’s timer-task status when it is unpaused so timeout
tasks are regenerated.
- Make ResetActivity with keepPaused=false fully unpause both scheduled
and running activities, including clearing pause metadata.
- Add and strengthen unit and functional coverage for unpause,
reset-unpause, timer regeneration, and keepPaused=true.

## Why?
Timeout tasks can fire while an activity is paused and be discarded.
Previously, the activity’s timer-task status still indicated that those
tasks existed, preventing them from being recreated after unpause and
potentially making the timeout ineffective.
ResetActivity also bypassed normal unpause handling, and running
activities returned early without clearing their paused state.

## How did you test it?
- [X] built
- [ ] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [X] added new functional test(s)

## Potential risks
Unpausing now invalidates existing timeout tasks and recreates the next
applicable timer during transaction close. This is correct, but a
behavioral change. Stale queued tasks may still be processed and
discarded through the existing stamp validation.

service/history/workflow/activity.go

@@ -313,6 +313,8 @@ func ResetActivity(
 	}
 
 	return mutableState.UpdateActivity(ai.ScheduledEventId, func(activityInfo *persistencespb.ActivityInfo, ms historyi.MutableState) error {
+		wasPaused := activityInfo.Paused
+
 		// reset the number of attempts
 		activityInfo.Attempt = 1
 		activityInfo.ActivityReset = true
@@ -339,14 +341,18 @@ func ResetActivity(
 			activityInfo.TimerTaskStatus = TimerTaskStatusNone
 		}
 
-		// if activity is running, or it is paused and we don't want to unpause - we don't need to do anything
-		if GetActivityState(ai) == enumspb.PENDING_ACTIVITY_STATE_STARTED || (ai.Paused && keepPaused) {
+		if wasPaused && !keepPaused {
+			unpauseActivityInfo(activityInfo)
+		}
+
+		// if activity is running, or it remains paused, we don't need to schedule a new run
+		if GetActivityState(activityInfo) == enumspb.PENDING_ACTIVITY_STATE_STARTED || activityInfo.Paused {
 			return nil
 		}
 
-		activityInfo.Stamp++
-		if activityInfo.Paused && !keepPaused {
-			activityInfo.Paused = false
+		// unpauseActivityInfo already incremented the activity stamp.
+		if !wasPaused {
+			activityInfo.Stamp++
 		}
 
 		// if activity is not running - we need to regenerate the retry task as schedule activity immediately
@@ -376,6 +382,10 @@ func unpauseActivityInfo(ai *persistencespb.ActivityInfo) {
 	ai.PauseInfo = nil
 	ai.Stamp++
 
+	// Timer tasks can be dropped while paused without clearing this mask. Clear all
+	// bits so transaction close recreates the earliest timer. This can create duplicate
+	// timer tasks, but after one processes the timeout, the others become no-ops.
+	ai.TimerTaskStatus = TimerTaskStatusNone
 }
 
 func UnpauseActivity(

service/history/workflow/activity_test.go

@@ -370,14 +370,21 @@ func (s *activitySuite) TestResetPausedActivityAcceptance() {
 	s.NotNil(ai.PauseInfo)
 	s.Equal("test_identity", ai.PauseInfo.GetManual().Identity)
 	s.Equal("test_reason", ai.PauseInfo.GetManual().Reason)
+	timerTaskStatus := int32(TimerTaskStatusCreatedScheduleToClose |
+		TimerTaskStatusCreatedScheduleToStart |
+		TimerTaskStatusCreatedStartToClose |
+		TimerTaskStatusCreatedHeartbeat)
+	ai.TimerTaskStatus = timerTaskStatus
 
 	prevStamp = ai.Stamp
 	err = ResetActivity(context.Background(), s.mockShard, s.mutableState, ai.ActivityId,
 		false, true, false, 0)
-	s.NoError(err)
+	s.Require().NoError(err)
 	s.Equal(int32(1), ai.Attempt, "ActivityInfo.Attempt is not reset")
 	s.Equal(prevStamp, ai.Stamp, "ActivityInfo.Stamp should not change")
 	s.True(ai.Paused, "ActivityInfo.Paused shouldn't change by reset")
+	s.Equal(pauseInfo, ai.PauseInfo, "ActivityInfo.PauseInfo shouldn't change by reset")
+	s.Equal(timerTaskStatus, ai.TimerTaskStatus, "ActivityInfo.TimerTaskStatus shouldn't change by reset")
 }
 
 func (s *activitySuite) TestResetAndUnPauseActivityAcceptance() {
@@ -400,14 +407,91 @@ func (s *activitySuite) TestResetAndUnPauseActivityAcceptance() {
 	s.NotNil(ai.PauseInfo)
 	s.Equal("test_identity", ai.PauseInfo.GetManual().Identity)
 	s.Equal("test_reason", ai.PauseInfo.GetManual().Reason)
+	ai.TimerTaskStatus = TimerTaskStatusCreatedScheduleToClose |
+		TimerTaskStatusCreatedScheduleToStart |
+		TimerTaskStatusCreatedStartToClose |
+		TimerTaskStatusCreatedHeartbeat
 
 	prevStamp = ai.Stamp
 	err = ResetActivity(context.Background(), s.mockShard, s.mutableState, ai.ActivityId,
 		false, false, false, 0)
-	s.NoError(err)
+	s.Require().NoError(err)
 	s.Equal(int32(1), ai.Attempt, "ActivityInfo.Attempt is not reset")
 	s.NotEqual(prevStamp, ai.Stamp, "ActivityInfo.Stamp should change")
-	s.False(ai.Paused, "ActivityInfo.Paused shouldn't change by reset")
+	s.False(ai.Paused, "ActivityInfo.Paused should be cleared by reset")
+	s.Nil(ai.PauseInfo, "ActivityInfo.PauseInfo should be cleared by reset")
+	s.Equal(int32(TimerTaskStatusNone), ai.TimerTaskStatus, "ActivityInfo.TimerTaskStatus should be reset to none")
+
+	// Simulate the transaction-close step that recreates activity timers.
+	timerCreated, err := NewTimerSequence(s.mutableState).CreateNextActivityTimer()
+	s.Require().NoError(err)
+	s.True(timerCreated, "activity timeout task should be recreated after reset unpauses the activity")
+}
+
+func (s *activitySuite) TestResetUnpausesRunningActivity() {
+	ai := s.AddActivityInfo()
+	ai.StartedEventId = 2
+	ai.StartedTime = timestamppb.New(s.mockShard.GetTimeSource().Now())
+
+	pauseInfo := &persistencespb.ActivityInfo_PauseInfo{
+		PauseTime: timestamppb.New(s.mockShard.GetTimeSource().Now()),
+		PausedBy: &persistencespb.ActivityInfo_PauseInfo_Manual_{
+			Manual: &persistencespb.ActivityInfo_PauseInfo_Manual{
+				Identity: "test_identity",
+				Reason:   "test_reason",
+			},
+		},
+	}
+	err := PauseActivity(s.mutableState, ai.ActivityId, pauseInfo)
+	s.Require().NoError(err)
+	ai.TimerTaskStatus = TimerTaskStatusCreatedScheduleToClose |
+		TimerTaskStatusCreatedScheduleToStart |
+		TimerTaskStatusCreatedStartToClose |
+		TimerTaskStatusCreatedHeartbeat
+
+	err = ResetActivity(context.Background(), s.mockShard, s.mutableState, ai.ActivityId,
+		false, false, false, 0)
+	s.Require().NoError(err)
+	s.False(ai.Paused, "ActivityInfo.Paused should be cleared by reset")
+	s.Nil(ai.PauseInfo, "ActivityInfo.PauseInfo should be cleared by reset")
+	s.Equal(int32(TimerTaskStatusNone), ai.TimerTaskStatus, "ActivityInfo.TimerTaskStatus should be reset to none")
+
+	// Simulate the transaction-close step that recreates activity timers.
+	timerCreated, err := NewTimerSequence(s.mutableState).CreateNextActivityTimer()
+	s.Require().NoError(err)
+	s.True(timerCreated, "activity timeout task should be recreated after reset unpauses the running activity")
+}
+
+func (s *activitySuite) TestUnpauseActivityAcceptance() {
+	ai := s.AddActivityInfo()
+	pauseInfo := &persistencespb.ActivityInfo_PauseInfo{
+		PauseTime: timestamppb.New(s.mockShard.GetTimeSource().Now()),
+		PausedBy: &persistencespb.ActivityInfo_PauseInfo_Manual_{
+			Manual: &persistencespb.ActivityInfo_PauseInfo_Manual{
+				Identity: "test_identity",
+				Reason:   "test_reason",
+			},
+		},
+	}
+	err := PauseActivity(s.mutableState, ai.ActivityId, pauseInfo)
+	s.Require().NoError(err)
+	ai.TimerTaskStatus = TimerTaskStatusCreatedScheduleToClose |
+		TimerTaskStatusCreatedScheduleToStart |
+		TimerTaskStatusCreatedStartToClose |
+		TimerTaskStatusCreatedHeartbeat
+	prevStamp := ai.Stamp
+
+	err = UnpauseActivity(s.mockShard, s.mutableState, ai, false, false, 0)
+	s.Require().NoError(err)
+	s.False(ai.Paused, "ActivityInfo.Paused should be cleared by unpause")
+	s.Nil(ai.PauseInfo, "ActivityInfo.PauseInfo should be cleared by unpause")
+	s.NotEqual(prevStamp, ai.Stamp, "ActivityInfo.Stamp should change")
+	s.Equal(int32(TimerTaskStatusNone), ai.TimerTaskStatus, "ActivityInfo.TimerTaskStatus should be reset to none")
+
+	// Simulate the transaction-close step that recreates activity timers.
+	timerCreated, err := NewTimerSequence(s.mutableState).CreateNextActivityTimer()
+	s.Require().NoError(err)
+	s.True(timerCreated, "activity timeout task should be recreated after unpause")
 }
 
 func (s *activitySuite) TestUnpauseActivityWithResumeAcceptance() {
@@ -421,6 +505,10 @@ func (s *activitySuite) TestUnpauseActivityWithResumeAcceptance() {
 	s.Equal(int32(1), ai.Attempt, "ActivityInfo.Attempt is shouldn't change")
 	s.NotEqual(prevStamp, ai.Stamp, "ActivityInfo.Stamp should change")
 	s.True(ai.Paused, "ActivityInfo.Paused was not unpaused")
+	ai.TimerTaskStatus = TimerTaskStatusCreatedScheduleToClose |
+		TimerTaskStatusCreatedScheduleToStart |
+		TimerTaskStatusCreatedStartToClose |
+		TimerTaskStatusCreatedHeartbeat
 	prevStamp = ai.Stamp
 	_, err = UnpauseActivityWithResume(s.mockShard, s.mutableState, ai, false, 0)
 	s.NoError(err)
@@ -428,6 +516,12 @@ func (s *activitySuite) TestUnpauseActivityWithResumeAcceptance() {
 	s.Equal(int32(1), ai.Attempt, "ActivityInfo.Attempt is shouldn't change")
 	s.NotEqual(prevStamp, ai.Stamp, "ActivityInfo.Stamp should change")
 	s.False(ai.Paused, "ActivityInfo.Paused was not unpaused")
+	s.Equal(int32(TimerTaskStatusNone), ai.TimerTaskStatus, "ActivityInfo.TimerTaskStatus should be reset to none")
+
+	// Verify invalidating the stale timer mask allows the timeout task to be regenerated.
+	timerCreated, err := NewTimerSequence(s.mutableState).CreateNextActivityTimer()
+	s.Require().NoError(err)
+	s.True(timerCreated, "activity timeout task should be recreated after unpause")
 }
 
 func (s *activitySuite) TestUnpauseActivityWithNewRun() {

tests/activity_api_reset_test.go

@@ -232,6 +232,144 @@ func (s *ActivityApiResetClientTestSuite) TestActivityResetApi_WhileRunning() {
 	s.Equal(int32(1), startedActivityCount.Load())
 }
 
+func (s *ActivityApiResetClientTestSuite) TestActivityResetApi_UnpausesRunningActivity() {
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	activityCompleteCh := make(chan struct{})
+	var startedActivityCount atomic.Int32
+	activityFunction := func() (string, error) {
+		startedActivityCount.Add(1)
+		s.WaitForChannel(ctx, activityCompleteCh) //nolint:staticcheck
+		return "done!", nil
+	}
+
+	workflowFn := s.makeWorkflowFunc(activityFunction)
+	s.SdkWorker().RegisterWorkflow(workflowFn)
+	s.SdkWorker().RegisterActivity(activityFunction)
+
+	workflowRun, err := s.SdkClient().ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
+		ID:        s.tv.WorkflowID(),
+		TaskQueue: s.TaskQueue(),
+	}, workflowFn)
+	s.Require().NoError(err)
+
+	await.Require(ctx, s.T(), func(t *await.T) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, description.PendingActivities[0].State)
+	}, 5*time.Second, 200*time.Millisecond)
+
+	_, err = s.FrontendClient().PauseActivity(ctx, &workflowservice.PauseActivityRequest{
+		Namespace: s.Namespace().String(),
+		Execution: &commonpb.WorkflowExecution{WorkflowId: workflowRun.GetID()},
+		Activity:  &workflowservice.PauseActivityRequest_Id{Id: "activity-id"},
+	})
+	s.Require().NoError(err)
+
+	await.Require(ctx, s.T(), func(t *await.T) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED, description.PendingActivities[0].State)
+		require.True(t, description.PendingActivities[0].Paused)
+		require.NotNil(t, description.PendingActivities[0].PauseInfo)
+	}, 5*time.Second, 200*time.Millisecond)
+
+	s.Require().NoError(s.resetFn(ctx, workflowRun.GetID(), "activity-id", false, false))
+
+	// keepPaused=false must clear both the paused flag and its associated metadata.
+	await.Require(ctx, s.T(), func(t *await.T) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, description.PendingActivities[0].State)
+		require.False(t, description.PendingActivities[0].Paused)
+		require.Nil(t, description.PendingActivities[0].PauseInfo)
+		require.Equal(t, int32(1), description.PendingActivities[0].Attempt)
+	}, 5*time.Second, 200*time.Millisecond)
+
+	activityCompleteCh <- struct{}{}
+
+	s.Require().NoError(workflowRun.Get(ctx, nil))
+	s.Equal(int32(1), startedActivityCount.Load())
+}
+
+func (s *ActivityApiResetClientTestSuite) TestActivityResetApi_TimesOutOnUnpause() {
+	s.startToCloseTimeout = time.Second
+	s.activityRetryPolicy = &temporal.RetryPolicy{MaximumAttempts: 1}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	defer cancel()
+
+	activityCompleteCh := make(chan struct{})
+	defer close(activityCompleteCh)
+	activityFunction := func() (string, error) {
+		<-activityCompleteCh
+		return "done!", nil
+	}
+
+	workflowFn := s.makeWorkflowFunc(activityFunction)
+	s.SdkWorker().RegisterWorkflow(workflowFn)
+	s.SdkWorker().RegisterActivity(activityFunction)
+
+	workflowRun, err := s.SdkClient().ExecuteWorkflow(ctx, sdkclient.StartWorkflowOptions{
+		ID:        s.tv.WorkflowID(),
+		TaskQueue: s.TaskQueue(),
+	}, workflowFn)
+	s.Require().NoError(err)
+
+	var activityStartedAt time.Time
+	await.Require(ctx, s.T(), func(t *await.T) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		pendingActivity := description.PendingActivities[0]
+		require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_STARTED, pendingActivity.State)
+		require.NotNil(t, pendingActivity.LastStartedTime)
+		activityStartedAt = pendingActivity.LastStartedTime.AsTime()
+	}, 5*time.Second, 200*time.Millisecond)
+
+	_, err = s.FrontendClient().PauseActivity(ctx, &workflowservice.PauseActivityRequest{
+		Namespace: s.Namespace().String(),
+		Execution: &commonpb.WorkflowExecution{WorkflowId: workflowRun.GetID()},
+		Activity:  &workflowservice.PauseActivityRequest_Id{Id: "activity-id"},
+	})
+	s.Require().NoError(err)
+
+	await.Require(ctx, s.T(), func(t *await.T) {
+		description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+		require.NoError(t, err)
+		require.Len(t, description.GetPendingActivities(), 1)
+		require.Equal(t, enumspb.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED, description.PendingActivities[0].State)
+		require.True(t, description.PendingActivities[0].Paused)
+	}, 5*time.Second, 200*time.Millisecond)
+
+	// Allow the original timeout task to be processed while the activity is paused.
+	originalDeadline := activityStartedAt.Add(s.startToCloseTimeout)
+	await.RequireTrue(s.T(), func() bool {
+		return time.Now().After(originalDeadline.Add(2 * time.Second))
+	}, 5*time.Second, 100*time.Millisecond)
+
+	description, err := s.SdkClient().DescribeWorkflowExecution(ctx, workflowRun.GetID(), workflowRun.GetRunID())
+	s.Require().NoError(err)
+	s.Require().Len(description.GetPendingActivities(), 1)
+	s.True(description.PendingActivities[0].Paused)
+
+	s.Require().NoError(s.resetFn(ctx, workflowRun.GetID(), "activity-id", false, false))
+
+	workflowResultCtx, workflowResultCancel := context.WithTimeout(ctx, 10*time.Second)
+	defer workflowResultCancel()
+	err = workflowRun.Get(workflowResultCtx, nil)
+	s.Require().Error(err)
+	var activityErr *temporal.ActivityError
+	s.Require().ErrorAs(err, &activityErr)
+	timeoutErr, ok := activityErr.Unwrap().(*temporal.TimeoutError)
+	s.Require().True(ok)
+	s.Equal(enumspb.TIMEOUT_TYPE_START_TO_CLOSE, timeoutErr.TimeoutType())
+}
+
 func (s *ActivityApiResetClientTestSuite) TestActivityResetApi_InRetry() {
 	// reset is called while activity is in retry
 	s.initialRetryInterval = 1 * time.Minute