diff --git a/chasm/lib/activity/activity.go b/chasm/lib/activity/activity.go index 48fd5288143..747162bf85d 100644 --- a/chasm/lib/activity/activity.go +++ b/chasm/lib/activity/activity.go @@ -897,6 +897,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 +913,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 +925,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_tasks_test.go b/chasm/lib/activity/activity_tasks_test.go new file mode 100644 index 00000000000..f131ffca9e6 --- /dev/null +++ b/chasm/lib/activity/activity_tasks_test.go @@ -0,0 +1,53 @@ +package activity + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + "go.temporal.io/server/chasm" + "go.temporal.io/server/chasm/lib/activity/gen/activitypb/v1" + "google.golang.org/protobuf/types/known/durationpb" +) + +func TestScheduleToCloseTimeoutTaskValidateStamp(t *testing.T) { + handler := newScheduleToCloseTimeoutTaskHandler() + + newActivity := func(stamp int32) *Activity { + return &Activity{ + ActivityState: &activitypb.ActivityState{ + Status: activitypb.ACTIVITY_EXECUTION_STATUS_SCHEDULED, + ScheduleToCloseTimeout: durationpb.New(30 * time.Second), + ScheduleToCloseStamp: stamp, + }, + } + } + + testCases := []struct { + name string + activityStamp int32 + taskStamp int32 + wantValid bool + }{ + // A legacy activity (created before ScheduleToCloseStamp existed) has stamp 0 on both + // the state and the timer task. Its timer is valid only while the activity stamp is + // still 0; once an options update bumps the stamp, the zero-stamp timer must be rejected. + {"legacy task valid while activity stamp still zero", 0, 0, true}, + {"legacy zero-stamp task invalid after stamp bumped", 5, 0, false}, + {"matching stamp valid", 5, 5, true}, + {"stale non-zero stamp invalid", 5, 3, false}, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + valid, err := handler.Validate( + nil, + newActivity(tc.activityStamp), + chasm.TaskAttributes{}, + &activitypb.ScheduleToCloseTimeoutTask{Stamp: tc.taskStamp}, + ) + require.NoError(t, err) + require.Equal(t, tc.wantValid, valid) + }) + } +} 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..c3dd759e1c5 100644 --- a/tests/activity_standalone_test.go +++ b/tests/activity_standalone_test.go @@ -22,6 +22,7 @@ import ( "go.temporal.io/sdk/temporal" "go.temporal.io/server/chasm/lib/activity" "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" @@ -9722,15 +9723,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 +10863,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 +10874,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 +10930,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 +12184,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) (