Reject stale legacy SAA schedule-to-close timers

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,

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

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)
+		})
+	}
+}

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;
 }
 

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) (