diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 48fd5288143..07e4589c263 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -23,6 +23,7 @@ import ( "fmt" "math/rand" "slices" + "strings" "time" "github.com/nexus-rpc/sdk-go/nexus" @@ -639,13 +640,14 @@ func (a *Activity) UpdateActivityExecutionOptions( } frontendReq := req.GetFrontendRequest() - - // start_delay updates are only valid while the activity is still in its delay window. - var hasStartDelayInMask bool + updateFields := map[string]struct{}{} if mask := frontendReq.GetUpdateMask(); mask != nil { - _, hasStartDelayInMask = util.ParseFieldMask(mask)["startDelay"] + updateFields = util.ParseFieldMask(mask) } - if !frontendReq.GetRestoreOriginal() && hasStartDelayInMask { + + // start_delay updates are only valid while the activity is still in its delay window. + _, hasStartDelayInMask := updateFields["startDelay"] + if hasStartDelayInMask { newDelay := frontendReq.GetActivityOptions().GetStartDelay() if err := validateStartDelay(newDelay); err != nil { return nil, err @@ -663,6 +665,9 @@ func (a *Activity) UpdateActivityExecutionOptions( } } + attempt := a.LastAttempt.Get(ctx) + policyRetryIntervalBeforeUpdate := backoff.CalculateExponentialRetryInterval(a.RetryPolicy, attempt.GetCount()-1) + if frontendReq.GetRestoreOriginal() { ogOptions := a.GetOriginalOptions() a.TaskQueue = common.CloneProto(ogOptions.GetTaskQueue()) @@ -683,15 +688,14 @@ func (a *Activity) UpdateActivityExecutionOptions( } } - attempt := a.LastAttempt.Get(ctx) - - // Recalculate the retry interval based on the (possibly updated) retry policy. - // This ensures a shortened retry interval takes effect immediately on re-dispatch. - status := a.GetStatus() - if (status == activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED || - status == activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED) && - attempt.GetCurrentRetryInterval() != nil { - + // Recalculate policy-derived retry intervals based on the (possibly updated) retry policy. + // Worker-provided NextRetryDelay values are preserved for their already-scheduled retry. + if a.shouldRecalculateCurrentRetryInterval( + attempt, + frontendReq.GetRestoreOriginal(), + updateFields, + policyRetryIntervalBeforeUpdate, + ) { newInterval := backoff.CalculateExponentialRetryInterval(a.RetryPolicy, attempt.GetCount()-1) attempt.CurrentRetryInterval = durationpb.New(newInterval) } @@ -728,6 +732,52 @@ func (a *Activity) UpdateActivityExecutionOptions( }, nil } +// shouldRecalculateCurrentRetryInterval reports whether the pending retry's CurrentRetryInterval +// should be re-derived from the (possibly updated) retry policy. All of the following must hold: +// +// - the activity is waiting to retry, i.e. in SCHEDULED or PAUSED state (a running attempt has no +// pending backoff to recalculate); +// - a retry is actually pending (CurrentRetryInterval is set); +// - the update touches the retry policy: either RestoreOriginal (which replaces the whole policy), +// or the update mask includes "retryPolicy" or one of its subfields; +// - the pending interval is still policy-derived, i.e. it equals the pre-update policy value. +// A mismatch means the interval is a worker-provided NextRetryDelay override, which we preserve. +func (a *Activity) shouldRecalculateCurrentRetryInterval( + attempt *activitypb.ActivityAttemptState, + restoreOriginal bool, + updateFields map[string]struct{}, + policyRetryIntervalBeforeUpdate time.Duration, +) bool { + status := a.GetStatus() + if status != activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED && + status != activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED { + return false + } + + currentRetryInterval := attempt.GetCurrentRetryInterval() + if currentRetryInterval == nil { + return false + } + + if !restoreOriginal { + hasRetryPolicyInMask := false + for field := range updateFields { + if field == "retryPolicy" || strings.HasPrefix(field, "retryPolicy.") { + hasRetryPolicyInMask = true + break + } + } + if !hasRetryPolicyInMask { + return false + } + } + + // CurrentRetryInterval stores either policy-derived backoff or worker-provided NextRetryDelay overrides. + // Only recalculate intervals that match the old policy value; different values are treated as explicit + // worker overrides. + return currentRetryInterval.AsDuration() == policyRetryIntervalBeforeUpdate +} + // mergeActivityOptions applies the field mask from the request to the activity state. // The structure mirrors the field-mask logic in service/history/api/updateactivityoptions/api.go func (a *Activity) mergeActivityOptions( @@ -897,6 +947,8 @@ func (a *Activity) handleUnpauseRequested(ctx chasm.MutableContext, req *activit if err := TransitionUnpausedWhilePauseRequested.Apply(a, ctx, event); err != nil { return nil, err } + case activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED: + a.ResetKeepPaused = false default: return nil, serviceerror.NewFailedPreconditionf("activity is in non-unpausable state %v", a.GetStatus()) } @@ -911,6 +963,8 @@ func (a *Activity) isPaused() bool { case activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED, activitypb.ACTIVITY_EXECUTION_STATUS_PAUSE_REQUESTED: return true + case activitypb.ACTIVITY_EXECUTION_STATUS_RESET_REQUESTED: + return a.GetResetKeepPaused() default: return false } @@ -921,17 +975,16 @@ func (a *Activity) unpause( event unpauseEvent, ) { attempt := a.LastAttempt.Get(ctx) - // Compute the dispatch time while the pending retry state is still intact; it is cleared below. dispatchTime := a.unpauseDispatchTime(ctx, event) if event.req.GetResetAttempts() { attempt.Count = 1 + attempt.CurrentRetryInterval = nil } if event.req.GetResetHeartbeat() { a.LastHeartbeat = chasm.NewDataField(ctx, &activitypb.ActivityHeartbeatState{}) } attempt.Stamp++ - attempt.CurrentRetryInterval = nil if timeout := a.GetScheduleToStartTimeout().AsDuration(); timeout > 0 { ctx.AddTask( a, diff --git a/chasm/lib/activity/activity_tasks.go b/chasm/lib/activity/activity_tasks.go index 70e73d874c2..375e0798960 100644 --- a/chasm/lib/activity/activity_tasks.go +++ b/chasm/lib/activity/activity_tasks.go @@ -137,8 +137,7 @@ func (h *scheduleToCloseTimeoutTaskHandler) Validate( } // Stamp check: discard tasks from before the most recent ScheduleToCloseTimeoutTask was // scheduled (e.g. after a schedule-to-close extension or a disable+re-enable cycle). - // Tasks without a stamp (stamp=0) predate this field and are not validated by stamp. - if task.GetStamp() != 0 && task.GetStamp() != activity.GetScheduleToCloseStamp() { + if task.GetStamp() != activity.GetScheduleToCloseStamp() { return false, nil } return true, nil diff --git a/chasm/lib/activity/activity_test.go b/chasm/lib/activity/activity_test.go index c50cd05827a..135d224c47c 100644 --- a/chasm/lib/activity/activity_test.go +++ b/chasm/lib/activity/activity_test.go @@ -643,3 +643,83 @@ func TestEffectiveUserMetadata_FallsBackToLegacy(t *testing.T) { got := activity.effectiveUserMetadata(ctx) require.Same(t, legacyMD, got) } + +func TestShouldRecalculateCurrentRetryInterval(t *testing.T) { + policyRetryInterval := 2 * time.Second + + testCases := []struct { + name string + status activitypb.ActivityExecutionStatus + restoreOriginal bool + updateFields map[string]struct{} + currentRetryInterval *durationpb.Duration + expectRecalculate bool + }{ + { + name: "retry policy subfield update with policy-derived interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + updateFields: map[string]struct{}{"retryPolicy.initialInterval": {}}, + currentRetryInterval: durationpb.New(policyRetryInterval), + expectRecalculate: true, + }, + { + name: "retry policy replacement with policy-derived interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + updateFields: map[string]struct{}{"retryPolicy": {}}, + currentRetryInterval: durationpb.New(policyRetryInterval), + expectRecalculate: true, + }, + { + name: "restore original with policy-derived interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_PAUSED, + restoreOriginal: true, + currentRetryInterval: durationpb.New(policyRetryInterval), + expectRecalculate: true, + }, + { + name: "unrelated update preserves retry interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + updateFields: map[string]struct{}{"heartbeatTimeout": {}}, + currentRetryInterval: durationpb.New(policyRetryInterval), + }, + { + name: "worker override differs from policy interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + updateFields: map[string]struct{}{"retryPolicy.initialInterval": {}}, + currentRetryInterval: durationpb.New(policyRetryInterval + time.Second), + }, + { + name: "started activity is not in retry backoff", + status: activitypb.ACTIVITY_EXECUTION_STATUS_STARTED, + updateFields: map[string]struct{}{"retryPolicy.initialInterval": {}}, + currentRetryInterval: durationpb.New(policyRetryInterval), + }, + { + name: "missing retry interval", + status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + updateFields: map[string]struct{}{"retryPolicy.initialInterval": {}}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + activity := &Activity{ + ActivityState: &activitypb.ActivityState{ + Status: tc.status, + }, + } + attempt := &activitypb.ActivityAttemptState{ + CurrentRetryInterval: tc.currentRetryInterval, + } + + got := activity.shouldRecalculateCurrentRetryInterval( + attempt, + tc.restoreOriginal, + tc.updateFields, + policyRetryInterval, + ) + + require.Equal(t, tc.expectRecalculate, got) + }) + } +} diff --git a/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go b/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go index 3cb7a1345c3..a725625459e 100644 --- a/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go +++ b/chasm/lib/activity/gen/activitypb/v1/tasks.pb.go @@ -116,7 +116,7 @@ type ScheduleToCloseTimeoutTask struct { state protoimpl.MessageState `protogen:"open.v1"` // The schedule-to-close stamp for this task. Used for task validation. // See also [ActivityState.schedule_to_close_stamp]. - // Tasks without a stamp (stamp=0) predate this field and are not validated by stamp. + // Tasks without a stamp (stamp=0) predate this field and are valid only while the activity stamp is also 0. Stamp int32 `protobuf:"varint,1,opt,name=stamp,proto3" json:"stamp,omitempty"` unknownFields protoimpl.UnknownFields sizeCache protoimpl.SizeCache diff --git a/chasm/lib/activity/proto/v1/tasks.proto b/chasm/lib/activity/proto/v1/tasks.proto index e70753124fd..868309e7063 100644 --- a/chasm/lib/activity/proto/v1/tasks.proto +++ b/chasm/lib/activity/proto/v1/tasks.proto @@ -17,7 +17,7 @@ message ScheduleToStartTimeoutTask { message ScheduleToCloseTimeoutTask { // The schedule-to-close stamp for this task. Used for task validation. // See also [ActivityState.schedule_to_close_stamp]. - // Tasks without a stamp (stamp=0) predate this field and are not validated by stamp. + // Tasks without a stamp (stamp=0) predate this field and are valid only while the activity stamp is also 0. int32 stamp = 1; } diff --git a/tests/activity_standalone_test.go b/tests/activity_standalone_test.go index 66d3b1e6c31..2c348fc6d96 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -20,8 +20,11 @@ import ( taskqueuepb "go.temporal.io/api/taskqueue/v1" "go.temporal.io/api/workflowservice/v1" "go.temporal.io/sdk/temporal" + "go.temporal.io/server/chasm" "go.temporal.io/server/chasm/lib/activity" + activityspb "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" "go.temporal.io/server/chasm/lib/callback" + "go.temporal.io/server/common" "go.temporal.io/server/common/dynamicconfig" "go.temporal.io/server/common/headers" "go.temporal.io/server/common/log" @@ -29,12 +32,16 @@ import ( "go.temporal.io/server/common/nexus/nexusrpc" "go.temporal.io/server/common/payload" "go.temporal.io/server/common/payloads" + "go.temporal.io/server/common/persistence" + "go.temporal.io/server/common/persistence/serialization" + "go.temporal.io/server/common/primitives" "go.temporal.io/server/common/searchattribute/sadefs" "go.temporal.io/server/common/tasktoken" "go.temporal.io/server/common/testing/await" "go.temporal.io/server/common/testing/parallelsuite" "go.temporal.io/server/common/testing/protorequire" "go.temporal.io/server/tests/testcore" + "go.uber.org/fx" "google.golang.org/grpc/codes" "google.golang.org/grpc/metadata" "google.golang.org/protobuf/types/known/durationpb" @@ -7846,6 +7853,156 @@ func (s *standaloneActivityTestSuite) TestUpdateActivityExecutionOptions() { TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, }) require.NoError(t, err) + require.NotEmpty(t, pollResp2.GetTaskToken()) + require.EqualValues(t, 2, pollResp2.Attempt) + }) + + t.Run("ChangeRetryInterval_Consecutive", func(t *testing.T) { + // Consecutive retry_policy updates in backoff: the second update must recalculate the + // interval from the value the first update left behind, not the original interval. + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + // Start with a long retry interval to keep the activity in backoff after failure. + startResp, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + ActivityType: &commonpb.ActivityType{Name: "test-activity"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + StartToCloseTimeout: durationpb.New(30 * time.Minute), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(10 * time.Minute), + MaximumAttempts: 5, + }, + }) + require.NoError(t, err) + + // Poll and fail with a retryable failure — activity enters the 10-minute backoff. + pollResp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }) + require.NoError(t, err) + require.EqualValues(t, 1, pollResp.Attempt) + + _, err = env.FrontendClient().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: env.Namespace().String(), + TaskToken: pollResp.TaskToken, + Failure: &failurepb.Failure{ + Message: "retryable failure", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NonRetryable: false}, + }, + }, + }) + require.NoError(t, err) + + // Update 1: shorten to 5 minutes. Recalculates, but stays in backoff for the rest of the test. + _, err = env.FrontendClient().UpdateActivityExecutionOptions(ctx, &workflowservice.UpdateActivityExecutionOptionsRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.RunId, + ActivityOptions: &activitypb.ActivityOptions{ + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(5 * time.Minute), + }, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"retry_policy.initial_interval"}}, + }) + require.NoError(t, err) + + // Update 2: shorten to 1ms. Must recalculate from update 1's interval so the retry fires now. + updateResp, err := env.FrontendClient().UpdateActivityExecutionOptions(ctx, &workflowservice.UpdateActivityExecutionOptionsRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.RunId, + ActivityOptions: &activitypb.ActivityOptions{ + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(1 * time.Millisecond), + }, + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"retry_policy.initial_interval"}}, + }) + require.NoError(t, err) + require.NotNil(t, updateResp) + + // Activity should now be available to poll for attempt 2 since we shortened it to 1ms. + pollResp2, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }) + require.NoError(t, err) + require.NotEmpty(t, pollResp2.GetTaskToken()) + require.EqualValues(t, 2, pollResp2.Attempt) + }) + + t.Run("UpdateUnrelatedField_PreservesNextRetryDelay", func(t *testing.T) { + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + nextRetryDelay := 2 * time.Second + + startResp, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + ActivityType: &commonpb.ActivityType{Name: "test-activity"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + StartToCloseTimeout: durationpb.New(30 * time.Minute), + RetryPolicy: &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(10 * time.Minute), + MaximumAttempts: 5, + }, + }) + require.NoError(t, err) + + pollResp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }) + require.NoError(t, err) + require.EqualValues(t, 1, pollResp.Attempt) + + _, err = env.FrontendClient().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: env.Namespace().String(), + TaskToken: pollResp.TaskToken, + Failure: &failurepb.Failure{ + Message: "retryable failure", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{ + NonRetryable: false, + NextRetryDelay: durationpb.New(nextRetryDelay), + }, + }, + }, + }) + require.NoError(t, err) + + // Updating an unrelated option must not recalculate the worker-provided next retry interval. + _, err = env.FrontendClient().UpdateActivityExecutionOptions(ctx, &workflowservice.UpdateActivityExecutionOptionsRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.RunId, + ActivityOptions: &activitypb.ActivityOptions{ + HeartbeatTimeout: durationpb.New(5 * time.Second), + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"heartbeat_timeout"}}, + }) + require.NoError(t, err) + + // Backoff is the worker's 2s NextRetryDelay set during RespondActivityTaskFailed, not the policy's 10m, so 8s is enough to poll it. + pollCtx, pollCancel := context.WithTimeout(ctx, 8*time.Second) + defer pollCancel() + pollResp2, err := env.FrontendClient().PollActivityTaskQueue(pollCtx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + }) + require.NoError(t, err) + require.NotEmpty(t, pollResp2.GetTaskToken()) require.EqualValues(t, 2, pollResp2.Attempt) }) @@ -8394,6 +8551,120 @@ func (s *standaloneActivityTestSuite) TestUpdateActivityExecutionOptions() { }) } +// Verifies update-options invalidates a legacy ScheduleToClose timer whose stamp is 0. +// The test seeds persisted activity state as if it was created before ScheduleToCloseStamp existed, +// then extends ScheduleToClose. The original zero-stamp timer must be ignored after the update emits +// a newer stamped ScheduleToClose timer. +func (s *standaloneActivityTestSuite) TestUpdateOptionsInvalidatesLegacyScheduleToClose() { + t := s.T() + env := s.newTestEnv(testcore.WithFxOptions( + primitives.HistoryService, + fx.Decorate(func(executionManager persistence.ExecutionManager) persistence.ExecutionManager { + return &legacyScheduleToCloseStampExecutionManager{ExecutionManager: executionManager} + }), + )) + + ctx, cancel := context.WithTimeout(t.Context(), 30*time.Second) + defer cancel() + + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + originalScheduleToClose := 2 * time.Second + updatedTimeout := 30 * time.Second + + startResp, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + ActivityType: &commonpb.ActivityType{Name: "test-activity"}, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + ScheduleToCloseTimeout: durationpb.New(originalScheduleToClose), + RetryPolicy: &commonpb.RetryPolicy{MaximumAttempts: 1}, + }) + require.NoError(t, err) + + _, err = env.FrontendClient().UpdateActivityExecutionOptions(ctx, &workflowservice.UpdateActivityExecutionOptionsRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + ActivityOptions: &activitypb.ActivityOptions{ + ScheduleToCloseTimeout: durationpb.New(updatedTimeout), + // Extend ScheduleToStart too so this unpolled activity does not time out at the old deadline. + ScheduleToStartTimeout: durationpb.New(updatedTimeout), + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"schedule_to_close_timeout", "schedule_to_start_timeout"}}, + }) + require.NoError(t, err) + + require.Never(t, func() bool { + descResp, err := env.FrontendClient().DescribeActivityExecution(ctx, &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + }) + require.NoError(t, err) + return descResp.GetInfo().GetStatus() == enumspb.ACTIVITY_EXECUTION_STATUS_TIMED_OUT + }, originalScheduleToClose+timerSafetyMargin, 200*time.Millisecond, + "legacy zero-stamp ScheduleToClose task must not remain valid after update-options emits a newer ScheduleToClose stamp") +} + +type legacyScheduleToCloseStampExecutionManager struct { + persistence.ExecutionManager +} + +func (m *legacyScheduleToCloseStampExecutionManager) CreateWorkflowExecution( + ctx context.Context, + request *persistence.CreateWorkflowExecutionRequest, +) (*persistence.CreateWorkflowExecutionResponse, error) { + if request.ArchetypeID == activity.ArchetypeID { + if err := forceLegacyScheduleToCloseStamp(&request.NewWorkflowSnapshot); err != nil { + return nil, err + } + } + return m.ExecutionManager.CreateWorkflowExecution(ctx, request) +} + +func forceLegacyScheduleToCloseStamp(snapshot *persistence.WorkflowSnapshot) error { + rootNode := snapshot.ChasmNodes[""] + if rootNode == nil { + return serviceerror.NewInternal("activity CHASM root node not found") + } + + state := &activityspb.ActivityState{} + if err := serialization.Decode(rootNode.GetData(), state); err != nil { + return err + } + state.ScheduleToCloseStamp = 0 + + encodedState, err := serialization.Encode(state, serialization.WithDeterministicProto3) + if err != nil { + return err + } + rootNode.Data = encodedState + + taskTypeID := chasm.GenerateTypeID(chasm.FullyQualifiedName("activity", "scheduleToCloseTimer")) + taskFound := false + for _, task := range rootNode.GetMetadata().GetComponentAttributes().GetPureTasks() { + if task.GetTypeId() != taskTypeID { + continue + } + scheduleToCloseTask := &activityspb.ScheduleToCloseTimeoutTask{} + if err := serialization.Decode(task.GetData(), scheduleToCloseTask); err != nil { + return err + } + scheduleToCloseTask.Stamp = 0 + encodedTask, err := serialization.Encode(scheduleToCloseTask, serialization.WithDeterministicProto3) + if err != nil { + return err + } + task.Data = encodedTask + taskFound = true + } + if !taskFound { + return serviceerror.NewInternal("activity ScheduleToCloseTimeoutTask not found") + } + return nil +} + func (env *standaloneActivityEnv) pollActivityTaskQueue(ctx context.Context, taskQueue string) (*workflowservice.PollActivityTaskQueueResponse, error) { return env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ Namespace: env.Namespace().String(), @@ -9722,15 +9993,123 @@ func (s *standaloneActivityTestSuite) TestPauseActivityExecution() { }) require.NoError(t, err) - // Honor backoff: the activity must stay SCHEDULED well past the unpause - require.Never(t, func() bool { - dr, dErr := env.FrontendClient().DescribeActivityExecution(ctx, &workflowservice.DescribeActivityExecutionRequest{ + // Honor backoff: the task must not be pollable before the retry deadline. + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + poll2Resp, err := env.FrontendClient().PollActivityTaskQueue(pollCtx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + Identity: env.Tv().WorkerIdentity(), + }) + if err != nil { + require.True(t, common.IsContextDeadlineExceededErr(err), "unexpected poll error: %v", err) + } else { + require.Empty( + t, + poll2Resp.GetActivityId(), + "unpause must honor the remaining retry backoff; activity task should not be pollable", + ) + } + }) + + // After unpause, force dispatch reissue with an options update. The retry backoff should still + // be preserved, so no activity task should be pollable before the original retry deadline. + t.Run("PauseWhileRetryBackoffSurvivesOptionsUpdateAfterUnpause", func(t *testing.T) { + ctx := testcore.NewContext() + activityID := testcore.RandomizeStr(t.Name()) + taskQueue := testcore.RandomizeStr(t.Name()) + + startResp, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + ActivityType: env.Tv().ActivityType(), + Identity: env.Tv().WorkerIdentity(), + Input: defaultInput, + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue}, + StartToCloseTimeout: durationpb.New(defaultStartToCloseTimeout), + RequestId: env.Tv().RequestID(), + RetryPolicy: &commonpb.RetryPolicy{ + MaximumAttempts: 10, + InitialInterval: durationpb.New(30 * time.Second), + BackoffCoefficient: 1.0, + }, + }) + require.NoError(t, err) + + pollResp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + Identity: env.Tv().WorkerIdentity(), + }) + require.NoError(t, err) + require.EqualValues(t, 1, pollResp.Attempt) + + _, err = env.FrontendClient().RespondActivityTaskFailed(ctx, &workflowservice.RespondActivityTaskFailedRequest{ + Namespace: env.Namespace().String(), + TaskToken: pollResp.TaskToken, + Failure: &failurepb.Failure{ + Message: "retryable failure", + FailureInfo: &failurepb.Failure_ApplicationFailureInfo{ + ApplicationFailureInfo: &failurepb.ApplicationFailureInfo{NonRetryable: false}, + }, + }, + Identity: env.Tv().WorkerIdentity(), + }) + require.NoError(t, err) + + await.Require(ctx, t, func(c *await.T) { + dr, dErr := env.FrontendClient().DescribeActivityExecution( + c.Context(), + &workflowservice.DescribeActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + }) + require.NoError(c, dErr) + require.EqualValues(c, 2, dr.GetInfo().GetAttempt()) + }, 10*time.Second, 200*time.Millisecond) + + _, err = env.FrontendClient().PauseActivityExecution(ctx, &workflowservice.PauseActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + Identity: "test-identity", + }) + require.NoError(t, err) + + _, err = env.FrontendClient().UnpauseActivityExecution(ctx, &workflowservice.UnpauseActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + Identity: "test-identity", + }) + require.NoError(t, err) + + _, err = env.FrontendClient().UpdateActivityExecutionOptions( + ctx, + &workflowservice.UpdateActivityExecutionOptionsRequest{ Namespace: env.Namespace().String(), ActivityId: activityID, + RunId: startResp.GetRunId(), + ActivityOptions: &activitypb.ActivityOptions{ + HeartbeatTimeout: durationpb.New(time.Second), + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"heartbeat_timeout"}}, }) - return dErr != nil || dr.GetInfo().GetRunState() != enumspb.PENDING_ACTIVITY_STATE_SCHEDULED - }, 5*time.Second, 100*time.Millisecond, - "unpause must honor the remaining retry backoff; activity should remain SCHEDULED") + require.NoError(t, err) + + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + poll2Resp, err := env.FrontendClient().PollActivityTaskQueue(pollCtx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + Identity: env.Tv().WorkerIdentity(), + }) + if err != nil { + require.True(t, common.IsContextDeadlineExceededErr(err), "unexpected poll error: %v", err) + } else { + require.Empty(t, poll2Resp.GetActivityId(), "options update after unpause must not lose the pending retry backoff") + } }) // PauseWhileCancelRequested: pausing a CANCEL_REQUESTED activity must be rejected with @@ -10754,7 +11133,7 @@ func (s *standaloneActivityTestSuite) TestUnpauseActivityExecution() { activityID := testcore.RandomizeStr(t.Name()) taskQueue := testcore.RandomizeStr(t.Name()) - _, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ + startResp, err := env.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{ Namespace: env.Namespace().String(), ActivityId: activityID, ActivityType: env.Tv().ActivityType(), @@ -10765,7 +11144,7 @@ func (s *standaloneActivityTestSuite) TestUnpauseActivityExecution() { RequestId: env.Tv().RequestID(), RetryPolicy: &commonpb.RetryPolicy{ MaximumAttempts: 10, - InitialInterval: durationpb.New(1 * time.Second), + InitialInterval: durationpb.New(30 * time.Second), BackoffCoefficient: 1.0, }, }) @@ -10821,8 +11200,25 @@ func (s *standaloneActivityTestSuite) TestUnpauseActivityExecution() { }) require.NoError(t, err) - // Poll — attempt count should be reset to 1. - poll2Resp, err := env.FrontendClient().PollActivityTaskQueue(ctx, &workflowservice.PollActivityTaskQueueRequest{ + // Force dispatch reissue before polling. ResetAttempts must clear the old retry backoff, + // otherwise this update would re-delay the reset attempt. + _, err = env.FrontendClient().UpdateActivityExecutionOptions( + ctx, + &workflowservice.UpdateActivityExecutionOptionsRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + ActivityOptions: &activitypb.ActivityOptions{ + HeartbeatTimeout: durationpb.New(time.Second), + }, + UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"heartbeat_timeout"}}, + }) + require.NoError(t, err) + + // Poll: attempt count should be reset to 1 without waiting for the old 30s retry backoff. + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + poll2Resp, err := env.FrontendClient().PollActivityTaskQueue(pollCtx, &workflowservice.PollActivityTaskQueueRequest{ Namespace: env.Namespace().String(), TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, Identity: env.Tv().WorkerIdentity(), @@ -12058,6 +12454,58 @@ func (s *standaloneActivityTestSuite) TestResetActivityExecution() { require.NoError(t, err) }) + // ResetActivityExecution with keep_paused=true can arrive while a worker still owns + // the current attempt. The reset itself must remain deferred until that worker + // reports back, but a later unpause should clear the request to keep the reset + // attempt paused. + t.Run("UnpauseClearsKeepPausedButKeepsDeferredReset", func(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + + activityID := testcore.RandomizeStr(t.Name()) + startResp, pollResp1, taskQueue := startAndPollActivity(ctx, t, activityID, &commonpb.RetryPolicy{ + InitialInterval: durationpb.New(time.Second), + BackoffCoefficient: 1.0, + }) + require.EqualValues(t, 1, pollResp1.Attempt) + + pauseActivity(ctx, t, activityID, startResp.GetRunId()) + waitForState(ctx, t, activityID, startResp.GetRunId(), enumspb.PENDING_ACTIVITY_STATE_PAUSE_REQUESTED) + + _, err := env.FrontendClient().ResetActivityExecution(ctx, &workflowservice.ResetActivityExecutionRequest{ + Namespace: env.Namespace().String(), + ActivityId: activityID, + RunId: startResp.GetRunId(), + KeepPaused: true, + }) + require.NoError(t, err) + + unpauseActivity(ctx, t, activityID, startResp.GetRunId()) + + heartbeatResp, err := env.FrontendClient().RecordActivityTaskHeartbeat(ctx, &workflowservice.RecordActivityTaskHeartbeatRequest{ + Namespace: env.Namespace().String(), + TaskToken: pollResp1.TaskToken, + }) + require.NoError(t, err) + require.True(t, heartbeatResp.GetActivityReset(), "reset should remain pending after unpause") + require.False(t, heartbeatResp.GetActivityPaused(), "unpause should clear the pending keep-paused reset intent") + + failAttemptRetryably(ctx, t, pollResp1.TaskToken, 0) + + pollCtx, pollCancel := context.WithTimeout(ctx, 5*time.Second) + defer pollCancel() + pollResp2, err := env.FrontendClient().PollActivityTaskQueue(pollCtx, &workflowservice.PollActivityTaskQueueRequest{ + Namespace: env.Namespace().String(), + TaskQueue: &taskqueuepb.TaskQueue{Name: taskQueue, Kind: enumspb.TASK_QUEUE_KIND_NORMAL}, + Identity: defaultIdentity, + }) + require.NoError(t, err, "activity should dispatch after the worker failure applies the pending reset") + require.Equal(t, activityID, pollResp2.GetActivityId()) + require.EqualValues(t, 1, pollResp2.Attempt, "attempt should be reset to 1 after deferred reset") + + completeAttempt(ctx, t, pollResp2.TaskToken) + }) + // startAttemptWithTimeouts starts a SAA with the given per-attempt timeouts and retry policy and // polls it to STARTED, returning the start response, the in-flight task token, and the task queue. startAttemptWithTimeouts := func(ctx context.Context, t *testing.T, activityID string, startToClose, heartbeat time.Duration) (