Add new fields to the pendingActivityInfo. Some refactoring (#6635)

## What changed?
<!-- Describe what has changed in this PR -->
Populate new fields:
* NextAttemptScheduleTime
* LastAttemptCompleteTime
* CurrentRetryInterval
*  
## Why?
<!-- Tell your future self why have you made these changes -->
Feature request https://github.com/temporalio/temporal/issues/6605


## How did you test it?
Add some unit tests. More to come.

## Potential risks
No that I'm aware of.

## Is hotfix candidate?
No

proto/internal/temporal/server/api/persistence/v1/executions.proto

@@ -458,8 +458,11 @@ message ActivityInfo {
         int64 last_redirect_counter = 2;
     }
 
-    // fist time activity was schedulled.
+    // First time the activity was scheduled.
     google.protobuf.Timestamp first_scheduled_time = 39;
+
+    // Last time an activity failure was recorded by the server.
+    google.protobuf.Timestamp last_attempt_complete_time = 40;
 }
 
 // timer_map column

service/history/api/describeworkflow/api.go

@@ -48,7 +48,6 @@ import (
 	"go.temporal.io/server/service/history/shard"
 	"go.temporal.io/server/service/history/workflow"
 	"google.golang.org/protobuf/types/known/durationpb"
-	"google.golang.org/protobuf/types/known/emptypb"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
@@ -168,52 +167,11 @@ func Invoke(
 	}
 
 	for _, ai := range mutableState.GetPendingActivityInfos() {
-		p := &workflowpb.PendingActivityInfo{
-			ActivityId: ai.ActivityId,
-		}
-		if ai.GetUseWorkflowBuildIdInfo() != nil {
-			p.AssignedBuildId = &workflowpb.PendingActivityInfo_UseWorkflowBuildId{UseWorkflowBuildId: &emptypb.Empty{}}
-		} else if ai.GetLastIndependentlyAssignedBuildId() != "" {
-			p.AssignedBuildId = &workflowpb.PendingActivityInfo_LastIndependentlyAssignedBuildId{
-				LastIndependentlyAssignedBuildId: ai.GetLastIndependentlyAssignedBuildId(),
-			}
-		}
-		if ai.CancelRequested {
-			p.State = enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED
-		} else if ai.StartedEventId != common.EmptyEventID {
-			p.State = enumspb.PENDING_ACTIVITY_STATE_STARTED
-		} else {
-			p.State = enumspb.PENDING_ACTIVITY_STATE_SCHEDULED
-		}
-		if ai.LastHeartbeatUpdateTime != nil && !ai.LastHeartbeatUpdateTime.AsTime().IsZero() {
-			p.LastHeartbeatTime = ai.LastHeartbeatUpdateTime
-			p.HeartbeatDetails = ai.LastHeartbeatDetails
-		}
-		p.ActivityType, err = mutableState.GetActivityType(ctx, ai)
+		p, err := workflow.GetPendingActivityInfo(ctx, shard, mutableState, ai)
 		if err != nil {
 			return nil, err
 		}
-		if p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED {
-			p.ScheduledTime = ai.ScheduledTime
-		} else {
-			p.LastStartedTime = ai.StartedTime
-		}
-		p.LastWorkerIdentity = ai.StartedIdentity
-		if ai.HasRetryPolicy {
-			p.Attempt = ai.Attempt
-			p.ExpirationTime = ai.RetryExpirationTime
-			if ai.RetryMaximumAttempts != 0 {
-				p.MaximumAttempts = ai.RetryMaximumAttempts
-			}
-			if ai.RetryLastFailure != nil {
-				p.LastFailure = ai.RetryLastFailure
-			}
-			if p.LastWorkerIdentity == "" && ai.RetryLastWorkerIdentity != "" {
-				p.LastWorkerIdentity = ai.RetryLastWorkerIdentity
-			}
-		} else {
-			p.Attempt = 1
-		}
+
 		result.PendingActivities = append(result.PendingActivities, p)
 	}
 

service/history/api/respondactivitytaskfailed/api.go

@@ -118,6 +118,7 @@ func Invoke(
 
 			postActions := &api.UpdateWorkflowAction{}
 			failure := request.GetFailure()
+			mutableState.RecordLastActivityStarted(ai)
 			retryState, err := mutableState.RetryActivity(ai, failure)
 			if err != nil {
 				return nil, err

service/history/api/respondactivitytaskfailed/api_test.go

@@ -538,6 +538,7 @@ func (s *workflowSuite) setupMutableState(uc UsecaseConfig, ai *persistencepb.Ac
 
 	currentMutableState.EXPECT().GetWorkflowType().Return(uc.wfType).AnyTimes()
 	if uc.expectRetryActivity {
+		currentMutableState.EXPECT().RecordLastActivityStarted(gomock.Any())
 		currentMutableState.EXPECT().RetryActivity(ai, gomock.Any()).Return(uc.retryActivityState, uc.retryActivityError)
 		currentMutableState.EXPECT().HasPendingWorkflowTask().Return(false).AnyTimes()
 	}

service/history/timer_queue_active_task_executor.go

@@ -270,6 +270,7 @@ Loop:
 
 		failureMsg := fmt.Sprintf("activity %v timeout", timerSequenceID.TimerType.String())
 		timeoutFailure := failure.NewTimeoutFailure(failureMsg, timerSequenceID.TimerType)
+		mutableState.RecordLastActivityStarted(activityInfo)
 		var retryState enumspb.RetryState
 		if retryState, err = mutableState.RetryActivity(activityInfo, timeoutFailure); err != nil {
 			return err

service/history/workflow/activity.go

@@ -25,13 +25,19 @@
 package workflow
 
 import (
+	"context"
 	"time"
 
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/emptypb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	enumspb "go.temporal.io/api/enums/v1"
 	failurepb "go.temporal.io/api/failure/v1"
+	workflowpb "go.temporal.io/api/workflow/v1"
 	"go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common"
-	"google.golang.org/protobuf/types/known/durationpb"
-	"google.golang.org/protobuf/types/known/timestamppb"
+	"go.temporal.io/server/service/history/shard"
 )
 
 func makeBackoffAlgorithm(requestedDelay *time.Duration) BackoffCalculatorAlgorithmFunc {
@@ -76,3 +82,81 @@ func updateActivityInfoForRetries(
 
 	return ai
 }
+
+func ActivityState(ai *persistence.ActivityInfo) enumspb.PendingActivityState {
+	if ai.CancelRequested {
+		return enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED
+	}
+	if ai.StartedEventId != common.EmptyEventID {
+		return enumspb.PENDING_ACTIVITY_STATE_STARTED
+	}
+	return enumspb.PENDING_ACTIVITY_STATE_SCHEDULED
+}
+
+func GetPendingActivityInfo(
+	ctx context.Context, // only used as a passthrough to GetActivityType
+	shardContext shard.Context,
+	mutableState MutableState,
+	ai *persistence.ActivityInfo,
+) (*workflowpb.PendingActivityInfo, error) {
+	now := shardContext.GetTimeSource().Now().UTC()
+
+	p := &workflowpb.PendingActivityInfo{
+		ActivityId: ai.ActivityId,
+	}
+	if ai.GetUseWorkflowBuildIdInfo() != nil {
+		p.AssignedBuildId = &workflowpb.PendingActivityInfo_UseWorkflowBuildId{UseWorkflowBuildId: &emptypb.Empty{}}
+	} else if ai.GetLastIndependentlyAssignedBuildId() != "" {
+		p.AssignedBuildId = &workflowpb.PendingActivityInfo_LastIndependentlyAssignedBuildId{
+			LastIndependentlyAssignedBuildId: ai.GetLastIndependentlyAssignedBuildId(),
+		}
+	}
+
+	p.State = ActivityState(ai)
+
+	p.LastAttemptCompleteTime = ai.LastAttemptCompleteTime
+	if !ai.HasRetryPolicy {
+		p.NextAttemptScheduleTime = nil
+	} else {
+		p.Attempt = ai.Attempt
+		if p.State == enumspb.PENDING_ACTIVITY_STATE_SCHEDULED {
+			scheduledTime := ai.ScheduledTime.AsTime()
+			if now.Before(scheduledTime) {
+				// in this case activity is waiting for a retry
+				p.NextAttemptScheduleTime = ai.ScheduledTime
+				currentRetryDuration := p.NextAttemptScheduleTime.AsTime().Sub(p.LastAttemptCompleteTime.AsTime())
+				p.CurrentRetryInterval = durationpb.New(currentRetryDuration)
+			} else {
+				// in this case activity is at least scheduled
+				p.NextAttemptScheduleTime = nil
+				// we rely on the fact that ExponentialBackoffAlgorithm is deterministic, and  there's no random jitter
+				interval := ExponentialBackoffAlgorithm(ai.RetryInitialInterval, ai.RetryBackoffCoefficient, p.Attempt)
+				p.CurrentRetryInterval = durationpb.New(interval)
+			}
+		}
+	}
+	p.Attempt = max(p.Attempt, 1)
+
+	if ai.LastHeartbeatUpdateTime != nil && !ai.LastHeartbeatUpdateTime.AsTime().IsZero() {
+		p.LastHeartbeatTime = ai.LastHeartbeatUpdateTime
+		p.HeartbeatDetails = ai.LastHeartbeatDetails
+	}
+	var err error
+	p.ActivityType, err = mutableState.GetActivityType(ctx, ai)
+	if err != nil {
+		return nil, err
+	}
+	p.ScheduledTime = ai.ScheduledTime
+	p.LastStartedTime = ai.StartedTime
+	p.LastWorkerIdentity = ai.StartedIdentity
+	if ai.HasRetryPolicy {
+		p.ExpirationTime = ai.RetryExpirationTime
+		p.MaximumAttempts = ai.RetryMaximumAttempts
+		p.LastFailure = ai.RetryLastFailure
+		if p.LastWorkerIdentity == "" && ai.RetryLastWorkerIdentity != "" {
+			p.LastWorkerIdentity = ai.RetryLastWorkerIdentity
+		}
+	}
+
+	return p, nil
+}

service/history/workflow/activity_test.go

@@ -0,0 +1,218 @@
+// The MIT License
+//
+// Copyright (c) 2024 Uber Technologies, Inc.
+//
+// Permission is hereby granted, free of charge, to any person obtaining a copy
+// of this software and associated documentation files (the "Software"), to deal
+// in the Software without restriction, including without limitation the rights
+// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+// copies of the Software, and to permit persons to whom the Software is
+// furnished to do so, subject to the following conditions:
+//
+// The above copyright notice and this permission notice shall be included in
+// all copies or substantial portions of the Software.
+//
+// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+// THE SOFTWARE.
+
+package workflow
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+	"go.uber.org/mock/gomock"
+	"google.golang.org/protobuf/types/known/durationpb"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	commonpb "go.temporal.io/api/common/v1"
+	enumspb "go.temporal.io/api/enums/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
+	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/cluster"
+	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/service/history/shard"
+	"go.temporal.io/server/service/history/tests"
+)
+
+type (
+	activitySuite struct {
+		suite.Suite
+		*require.Assertions
+
+		controller            *gomock.Controller
+		mockShard             *shard.ContextTest
+		mockNamespaceRegistry *namespace.MockRegistry
+
+		mutableState *MockMutableState
+	}
+)
+
+func TestActivitySuite(t *testing.T) {
+	s := new(activitySuite)
+	suite.Run(t, s)
+}
+
+func (s *activitySuite) SetupTest() {
+	s.Assertions = require.New(s.T())
+
+	config := tests.NewDynamicConfig()
+	s.controller = gomock.NewController(s.T())
+	s.mockShard = shard.NewTestContext(
+		s.controller,
+		&persistencespb.ShardInfo{ShardId: 1},
+		config,
+	)
+
+	s.mutableState = NewMockMutableState(s.controller)
+
+	s.mockNamespaceRegistry = s.mockShard.Resource.NamespaceCache
+
+	s.mockNamespaceRegistry.EXPECT().GetNamespaceByID(tests.NamespaceID).Return(tests.GlobalNamespaceEntry, nil).AnyTimes()
+	s.mockNamespaceRegistry.EXPECT().GetNamespace(tests.Namespace).Return(tests.GlobalNamespaceEntry, nil).AnyTimes()
+
+	s.mockShard.Resource.ClusterMetadata.EXPECT().IsGlobalNamespaceEnabled().Return(true).AnyTimes()
+	s.mockShard.Resource.ClusterMetadata.EXPECT().GetCurrentClusterName().Return(cluster.TestCurrentClusterName).AnyTimes()
+	s.mockShard.Resource.ClusterMetadata.EXPECT().GetClusterID().Return(int64(1)).AnyTimes()
+	s.mockShard.Resource.ClusterMetadata.EXPECT().ClusterNameForFailoverVersion(true, tests.GlobalNamespaceEntry.FailoverVersion()).Return(cluster.TestCurrentClusterName).AnyTimes()
+}
+
+func (s *activitySuite) TearDownTest() {
+	s.controller.Finish()
+	s.mockShard.StopForTest()
+}
+
+func (s *activitySuite) TestGetActivityState() {
+	testCases := []struct {
+		ai    *persistencespb.ActivityInfo
+		state enumspb.PendingActivityState
+	}{
+		{
+			ai: &persistencespb.ActivityInfo{
+				CancelRequested: true,
+				StartedEventId:  1,
+			},
+			state: enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED,
+		},
+		{
+			ai: &persistencespb.ActivityInfo{
+				CancelRequested: true,
+				StartedEventId:  common.EmptyEventID,
+			},
+			state: enumspb.PENDING_ACTIVITY_STATE_CANCEL_REQUESTED,
+		},
+		{
+			ai: &persistencespb.ActivityInfo{
+				CancelRequested: false,
+				StartedEventId:  common.EmptyEventID,
+			},
+			state: enumspb.PENDING_ACTIVITY_STATE_SCHEDULED,
+		},
+		{
+			ai: &persistencespb.ActivityInfo{
+				CancelRequested: false,
+				StartedEventId:  1,
+			},
+			state: enumspb.PENDING_ACTIVITY_STATE_STARTED,
+		},
+	}
+
+	for _, tc := range testCases {
+		state := ActivityState(tc.ai)
+		s.Equal(tc.state, state)
+	}
+}
+
+func (s *activitySuite) TestGetPendingActivityInfoAcceptance() {
+	now := s.mockShard.GetTimeSource().Now().UTC().Round(time.Hour)
+	activityType := commonpb.ActivityType{
+		Name: "activityType",
+	}
+	ai := &persistencespb.ActivityInfo{
+		ActivityType:            &activityType,
+		ActivityId:              "activityID",
+		CancelRequested:         false,
+		StartedEventId:          1,
+		Attempt:                 2,
+		ScheduledTime:           timestamppb.New(now),
+		LastAttemptCompleteTime: timestamppb.New(now.Add(-1 * time.Hour)),
+		HasRetryPolicy:          false,
+	}
+
+	s.mutableState.EXPECT().GetActivityType(gomock.Any(), gomock.Any()).Return(&activityType, nil).Times(1)
+	pi, err := GetPendingActivityInfo(context.Background(), s.mockShard, s.mutableState, ai)
+	s.NoError(err)
+	s.NotNil(pi)
+}
+
+func (s *activitySuite) TestGetPendingActivityInfoNoRetryPolicy() {
+	now := s.mockShard.GetTimeSource().Now().UTC().Round(time.Hour)
+	activityType := commonpb.ActivityType{
+		Name: "activityType",
+	}
+	ai := &persistencespb.ActivityInfo{
+		ActivityType:            &activityType,
+		ActivityId:              "activityID",
+		CancelRequested:         false,
+		StartedEventId:          1,
+		Attempt:                 2,
+		ScheduledTime:           timestamppb.New(now),
+		LastAttemptCompleteTime: timestamppb.New(now.Add(-1 * time.Hour)),
+		HasRetryPolicy:          false,
+	}
+
+	s.mutableState.EXPECT().GetActivityType(gomock.Any(), gomock.Any()).Return(&activityType, nil).Times(1)
+	pi, err := GetPendingActivityInfo(context.Background(), s.mockShard, s.mutableState, ai)
+	s.NoError(err)
+	s.NotNil(pi)
+	s.Equal(enumspb.PENDING_ACTIVITY_STATE_STARTED, pi.State)
+	s.Equal(int32(1), pi.Attempt)
+	s.Nil(pi.NextAttemptScheduleTime)
+	s.Nil(pi.CurrentRetryInterval)
+}
+
+func (s *activitySuite) TestGetPendingActivityInfoHasRetryPolicy() {
+	now := s.mockShard.GetTimeSource().Now().UTC()
+	activityType := commonpb.ActivityType{
+		Name: "activityType",
+	}
+	ai := &persistencespb.ActivityInfo{
+		ActivityType:            &activityType,
+		ActivityId:              "activityID",
+		CancelRequested:         false,
+		StartedEventId:          common.EmptyEventID,
+		Attempt:                 2,
+		ScheduledTime:           timestamppb.New(now.Add(1 * time.Minute)),
+		LastAttemptCompleteTime: timestamppb.New(now.Add(-1 * time.Minute)),
+		HasRetryPolicy:          true,
+		RetryMaximumInterval:    nil,
+		RetryInitialInterval:    durationpb.New(time.Minute),
+		RetryBackoffCoefficient: 1.0,
+		RetryMaximumAttempts:    10,
+	}
+
+	s.mutableState.EXPECT().GetActivityType(gomock.Any(), gomock.Any()).Return(&activityType, nil).Times(1)
+	pi, err := GetPendingActivityInfo(context.Background(), s.mockShard, s.mutableState, ai)
+	s.NoError(err)
+	s.NotNil(pi)
+	s.Equal(enumspb.PENDING_ACTIVITY_STATE_SCHEDULED, pi.State)
+	s.Equal(int32(2), pi.Attempt)
+	s.Equal(ai.ActivityId, pi.ActivityId)
+	s.Nil(pi.HeartbeatDetails)
+	s.Nil(pi.LastHeartbeatTime)
+	s.Nil(pi.LastStartedTime)
+	s.NotNil(pi.NextAttemptScheduleTime) // activity is waiting for retry
+	s.Equal(ai.RetryMaximumAttempts, pi.MaximumAttempts)
+	s.Nil(pi.AssignedBuildId)
+	s.Equal(durationpb.New(2*time.Minute), pi.CurrentRetryInterval)
+	s.Equal(ai.ScheduledTime, pi.ScheduledTime)
+	s.Equal(ai.LastAttemptCompleteTime, pi.LastAttemptCompleteTime)
+}

service/history/workflow/mutable_state.go

@@ -224,6 +224,7 @@ type (
 		CheckResettable() error
 		CloneToProto() *persistencespb.WorkflowMutableState
 		RetryActivity(ai *persistencespb.ActivityInfo, failure *failurepb.Failure) (enumspb.RetryState, error)
+		RecordLastActivityStarted(ai *persistencespb.ActivityInfo)
 		GetTransientWorkflowTaskInfo(workflowTask *WorkflowTaskInfo, identity string) *historyspb.TransientWorkflowTaskInfo
 		DeleteSignalRequested(requestID string)
 		FlushBufferedEvents()

service/history/workflow/mutable_state_impl.go

@@ -4855,7 +4855,7 @@ func (ms *MutableStateImpl) ApplyChildWorkflowExecutionTimedOutEvent(
 
 func (ms *MutableStateImpl) RetryActivity(
 	ai *persistencespb.ActivityInfo,
-	failure *failurepb.Failure,
+	activityFailure *failurepb.Failure,
 ) (enumspb.RetryState, error) {
 	opTag := tag.WorkflowActionActivityTaskRetry
 	if err := ms.checkMutability(opTag); err != nil {
@@ -4868,13 +4868,13 @@ func (ms *MutableStateImpl) RetryActivity(
 		return enumspb.RETRY_STATE_CANCEL_REQUESTED, nil
 	}
 
-	if !isRetryable(failure, ai.RetryNonRetryableErrorTypes) {
+	if !isRetryable(activityFailure, ai.RetryNonRetryableErrorTypes) {
 		return enumspb.RETRY_STATE_NON_RETRYABLE_FAILURE, nil
 	}
 
 	retryMaxInterval := ai.RetryMaximumInterval
 	// if a delay is specified by the application it should override the maximum interval set by the retry policy.
-	delay := nextRetryDelayFrom(failure)
+	delay := nextRetryDelayFrom(activityFailure)
 	if delay != nil {
 		retryMaxInterval = durationpb.New(*delay)
 	}
@@ -4894,8 +4894,43 @@ func (ms *MutableStateImpl) RetryActivity(
 		return retryState, nil
 	}
 
-	nextScheduledTime := now.Add(retryBackoff)
-	nextAttempt := ai.Attempt + 1
+	ms.updateActivityInfoForRetries(ai,
+		now.Add(retryBackoff),
+		ai.Attempt+1,
+		activityFailure)
+
+	if err := ms.taskGenerator.GenerateActivityRetryTasks(ai); err != nil {
+		return enumspb.RETRY_STATE_INTERNAL_SERVER_ERROR, err
+	}
+	return enumspb.RETRY_STATE_IN_PROGRESS, nil
+}
+
+func (ms *MutableStateImpl) RecordLastActivityStarted(ai *persistencespb.ActivityInfo) {
+	ms.updateActivity(ai, func(info *persistencespb.ActivityInfo, impl *MutableStateImpl) *persistencespb.ActivityInfo {
+		ai.LastAttemptCompleteTime = timestamppb.New(ms.shard.GetTimeSource().Now().UTC())
+		return ai
+	})
+}
+
+func (ms *MutableStateImpl) updateActivityInfoForRetries(
+	ai *persistencespb.ActivityInfo,
+	nextScheduledTime time.Time,
+	nextAttempt int32,
+	activityFailure *failurepb.Failure,
+) {
+	ms.updateActivity(ai, func(info *persistencespb.ActivityInfo, impl *MutableStateImpl) *persistencespb.ActivityInfo {
+		ai = updateActivityInfoForRetries(
+			ai,
+			ms.GetCurrentVersion(),
+			nextAttempt,
+			ms.truncateRetryableActivityFailure(activityFailure),
+			timestamppb.New(nextScheduledTime),
+		)
+		return ai
+	})
+}
+
+func (ms *MutableStateImpl) updateActivity(ai *persistencespb.ActivityInfo, updateCallback func(*persistencespb.ActivityInfo, *MutableStateImpl) *persistencespb.ActivityInfo) {
 	// we need to store activity info size since pendingActivityInfoIDs holds pointers to activity
 	// info and if prev found it points to the same activity info as ai, so updating ai will cause
 	// size of prev change.
@@ -4903,21 +4938,12 @@ func (ms *MutableStateImpl) RetryActivity(
 	if prev, ok := ms.pendingActivityInfoIDs[ai.ScheduledEventId]; ok {
 		originalSize = prev.Size()
 	}
-	ai = updateActivityInfoForRetries(
-		ai,
-		ms.GetCurrentVersion(),
-		nextAttempt,
-		ms.truncateRetryableActivityFailure(failure),
-		timestamppb.New(nextScheduledTime),
-	)
+
+	updateCallback(ai, ms)
+
 	ms.approximateSize += ai.Size() - originalSize
 	ms.updateActivityInfos[ai.ScheduledEventId] = ai
 	ms.syncActivityTasks[ai.ScheduledEventId] = struct{}{}
-
-	if err := ms.taskGenerator.GenerateActivityRetryTasks(ai); err != nil {
-		return enumspb.RETRY_STATE_INTERNAL_SERVER_ERROR, err
-	}
-	return enumspb.RETRY_STATE_IN_PROGRESS, nil
 }
 
 func (ms *MutableStateImpl) truncateRetryableActivityFailure(

service/history/workflow/mutable_state_mock.go

@@ -2798,6 +2798,18 @@ func (mr *MockMutableStateMockRecorder) PopTasks() *gomock.Call {
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "PopTasks", reflect.TypeOf((*MockMutableState)(nil).PopTasks))
 }
 
+// RecordLastActivityStarted mocks base method.
+func (m *MockMutableState) RecordLastActivityStarted(ai *persistence.ActivityInfo) {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "RecordLastActivityStarted", ai)
+}
+
+// RecordLastActivityStarted indicates an expected call of RecordLastActivityStarted.
+func (mr *MockMutableStateMockRecorder) RecordLastActivityStarted(ai any) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordLastActivityStarted", reflect.TypeOf((*MockMutableState)(nil).RecordLastActivityStarted), ai)
+}
+
 // RefreshExpirationTimeoutTask mocks base method.
 func (m *MockMutableState) RefreshExpirationTimeoutTask(ctx context.Context) error {
 	m.ctrl.T.Helper()