fix errors in tests, change version and stamp to VersionedTransition

Author: feiyang3cat

api/persistence/v1/executions.pb.go

@@ -144,14 +144,13 @@ func serializeTimerTask(
 
 func timeSkippingTimerTaskToProto(task *tasks.TimeSkippingTimerTask) *persistencespb.TimerTaskInfo {
 	return &persistencespb.TimerTaskInfo{
-		NamespaceId:    task.NamespaceID,
-		WorkflowId:     task.WorkflowID,
-		RunId:          task.RunID,
-		TaskId:         task.TaskID,
-		VisibilityTime: timestamppb.New(task.VisibilityTimestamp),
-		TaskType:       enumsspb.TASK_TYPE_TIMESKIPPING_TIMER,
-		Version:        task.Version,
-		Stamp:          task.Stamp,
+		NamespaceId:         task.NamespaceID,
+		WorkflowId:          task.WorkflowID,
+		RunId:               task.RunID,
+		TaskId:              task.TaskID,
+		VisibilityTime:      timestamppb.New(task.VisibilityTimestamp),
+		TaskType:            enumsspb.TASK_TYPE_TIMESKIPPING_TIMER,
+		VersionedTransition: task.VersionedTransition,
 	}
 }
 
@@ -164,8 +163,7 @@ func timeSkippingTimerTaskFromProto(info *persistencespb.TimerTaskInfo) *tasks.T
 		),
 		VisibilityTimestamp: info.VisibilityTime.AsTime(),
 		TaskID:              info.TaskId,
-		Version:             info.Version,
-		Stamp:               info.Stamp,
+		VersionedTransition: info.VersionedTransition,
 	}
 }
 

common/persistence/serialization/task_serializers.go

@@ -352,14 +352,7 @@ message FastForwardInfo {
   // Indicates whether this fast-forward has already been reached, used for idempotency checks.
   bool has_reached = 2;
 
-  // This is a monotonically increasing counter that is used to identify the version of the fast-forward.
-  // It is bumped every time the fast-forward is set.
-  // It is used to validate the time skipping fast-forward timer task.
-  int32 stamp = 4;
-
-  // This is the NamespaceFailoverVersion at which this fast-forward was configured.
-  // It is used to validate the time skipping fast-forward timer task.
-  int64 version = 5;
+  VersionedTransition last_update_versioned_transition = 4;
 }
 
 // Internal wrapper message to distinguish "never notified" (nil wrapper) from
@@ -529,6 +522,11 @@ message TimerTaskInfo {
     // If the task addresses a CHASM component, this field will be set.
     ChasmTaskInfo chasm_task_info = 17;
   }
+
+  // VersionedTransition provides a unified solution for both intra-cluster and inter-cluster
+  // task replication. It is intended to replace previous validation mechanisms, such as
+  // version + stamp.
+  VersionedTransition versioned_transition = 18;
 }
 
 message ArchivalTaskInfo {

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

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	enumsspb "go.temporal.io/server/api/enums/v1"
+	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common/definition"
 )
 
@@ -13,41 +14,18 @@ var _ Task = (*TimeSkippingTimerTask)(nil)
 type (
 	// TimeSkippingTimerTask wakes a workflow when the fast-forward configured
 	// on its TimeSkippingConfig should take effect.
-	//
-	// Stamp identifies the fast-forward this task targets. It is matched against
-	// TimeSkippingInfo.FastForwardInfo.Stamp at firing time to detect superseded tasks:
-	// re-applying the fast-forward emits a new task with a bumped Stamp, and the old task is
-	// silently dropped on mismatch. Version is the namespace failover version, validated via
-	// CheckTaskVersion. Both are event-free, so the task works for workflow and CHASM executions.
 	TimeSkippingTimerTask struct {
 		definition.WorkflowKey
 		VisibilityTimestamp time.Time
+		VersionedTransition *persistencespb.VersionedTransition
 		TaskID              int64
-		Version             int64
-		Stamp               int32
 	}
 )
 
 func (t *TimeSkippingTimerTask) GetKey() Key {
 	return NewKey(t.VisibilityTimestamp, t.TaskID)
 }
 
-func (t *TimeSkippingTimerTask) GetVersion() int64 {
-	return t.Version
-}
-
-func (t *TimeSkippingTimerTask) SetVersion(version int64) {
-	t.Version = version
-}
-
-func (t *TimeSkippingTimerTask) GetStamp() int32 {
-	return t.Stamp
-}
-
-func (t *TimeSkippingTimerTask) SetStamp(stamp int32) {
-	t.Stamp = stamp
-}
-
 func (t *TimeSkippingTimerTask) GetTaskID() int64 {
 	return t.TaskID
 }
@@ -73,11 +51,12 @@ func (t *TimeSkippingTimerTask) GetType() enumsspb.TaskType {
 }
 
 func (t *TimeSkippingTimerTask) String() string {
-	return fmt.Sprintf("TimeSkippingTimerTask{WorkflowKey: %s, VisibilityTimestamp: %v, TaskID: %v, Version: %v, Stamp: %v}",
+	vt := t.VersionedTransition
+	return fmt.Sprintf("TimeSkippingTimerTask{WorkflowKey: %s, VisibilityTimestamp: %v, TaskID: %v, FailoverVersion: %v, TransitionCount: %v}",
 		t.WorkflowKey.String(),
 		t.VisibilityTimestamp,
 		t.TaskID,
-		t.Version,
-		t.Stamp,
+		vt.GetNamespaceFailoverVersion(),
+		vt.GetTransitionCount(),
 	)
 }

service/history/tasks/timeskipping_timer_task.go

@@ -116,6 +116,7 @@ func (t *timerQueueActiveTaskExecutor) Execute(
 		err = t.executeActivityRetryTimerTask(ctx, task)
 	case *tasks.WorkflowBackoffTimerTask:
 		err = t.executeWorkflowBackoffTimerTask(ctx, task)
+		// retention timer
 	case *tasks.DeleteHistoryEventTask:
 		err = t.executeDeleteHistoryEventTask(ctx, task)
 	case *tasks.StateMachineTimerTask:
@@ -939,21 +940,31 @@ func (t *timerQueueActiveTaskExecutor) executeTimeSkippingTimerTask(
 	}
 
 	// task staleness check
+	// 1) time skipping is disabled
 	tsi := mutableState.GetExecutionInfo().GetTimeSkippingInfo()
 	if tsi == nil || !tsi.GetConfig().GetEnabled() {
 		release(nil)
 		return errNoTimerFired
 	}
+	// 2) fast-forward is disabled/reached, or this task was superseded by a newer fast-forward.
+	// The versioned transition's TransitionCount identifies the specific fast-forward instance
+	// (a re-applied fast-forward records a fresh transition), so a mismatch means this task is
+	// stale and is dropped silently.
 	ff := tsi.GetFastForwardInfo()
-	if ff.GetTargetTime() == nil || ff.GetHasReached() || ff.GetStamp() != task.Stamp {
+	ffVT := ff.GetLastUpdateVersionedTransition()
+	if ff.GetTargetTime() == nil || ff.GetHasReached() ||
+		ffVT.GetTransitionCount() != task.VersionedTransition.GetTransitionCount() {
 		release(nil)
 		return errNoTimerFired
 	}
-	ffVersion := ff.GetVersion()
-	if err := CheckTaskVersion(t.shardContext, t.logger, mutableState.GetNamespaceEntry(), ffVersion, task.Version, task); err != nil {
+	// The failover version is validated separately so a task from a stale failover version is
+	// rejected with the dedicated task-version-mismatch error.
+	if err := CheckTaskVersion(t.shardContext, t.logger, mutableState.GetNamespaceEntry(),
+		ffVT.GetNamespaceFailoverVersion(), task.VersionedTransition.GetNamespaceFailoverVersion(), task); err != nil {
 		return err
 	}
-
+	// 3) firing fast-forward timer only turns off the time-skipping enabled flag and it doesn't regnerate tasks.
+	// TODO@time-skipping: chasm execution path is not implemented yet.
 	_, err = mutableState.AddWorkflowExecutionTimeSkippingTransitionedEvent(
 		ctx, time.Time{}, true)
 	if err != nil {

service/history/timer_queue_active_task_executor.go

@@ -2407,101 +2407,96 @@ func (s *timerQueueActiveTaskExecutorSuite) makeTimeSkippingMS() (*persistencesp
 	return pms, workflowKey
 }
 
-// TestExecuteTimeSkippingTimerTask covers the full validity-check ladder of
-// executeTimeSkippingTimerTask plus the happy-path disable transition.
 func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask() {
 	target := timestamppb.New(s.now.Add(time.Hour))
-	enabledFF := func(stamp int32, hasReached bool) *persistencespb.TimeSkippingInfo {
+	// The fast-forward's versioned transition carries the failover version (NamespaceFailoverVersion)
+	// and the fast-forward instance identity (TransitionCount, the successor of the old Stamp).
+	enabledFF := func(version int64, transitionCount int64, hasReached bool) *persistencespb.TimeSkippingInfo {
 		return &persistencespb.TimeSkippingInfo{
 			Config: &commonpb.TimeSkippingConfig{Enabled: true},
 			FastForwardInfo: &persistencespb.FastForwardInfo{
 				TargetTime: target,
-				Stamp:      stamp,
 				HasReached: hasReached,
-				Version:    s.version,
+				LastUpdateVersionedTransition: &persistencespb.VersionedTransition{
+					NamespaceFailoverVersion: version,
+					TransitionCount:          transitionCount,
+				},
 			},
 		}
 	}
-
 	for _, tc := range []struct {
-		name        string
-		completed   bool // mark workflow execution completed
-		tsi         *persistencespb.TimeSkippingInfo
-		taskVersion int64
-		taskStamp   int32
-		wantErr     error // nil => happy path: disable transition is applied
+		name                string
+		completed           bool // mark workflow execution completed
+		tsi                 *persistencespb.TimeSkippingInfo
+		taskVersion         int64
+		taskTransitionCount int64
+		wantErr             error // nil => happy path: disable transition is applied
 	}{
 		{
-			name:        "WorkflowNotRunning",
-			completed:   true,
-			tsi:         enabledFF(1, false),
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     consts.ErrWorkflowCompleted,
+			name:                "WorkflowNotRunning",
+			completed:           true,
+			tsi:                 enabledFF(s.version, 1, false),
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             consts.ErrWorkflowCompleted,
 		},
 		{
-			name:        "TSINil",
-			tsi:         nil,
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     errNoTimerFired,
+			name:                "TSINil",
+			tsi:                 nil,
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             errNoTimerFired,
 		},
 		{
 			name: "ConfigDisabled",
 			tsi: &persistencespb.TimeSkippingInfo{
-				Config:          &commonpb.TimeSkippingConfig{Enabled: false},
-				FastForwardInfo: &persistencespb.FastForwardInfo{TargetTime: target, Stamp: 1},
+				Config: &commonpb.TimeSkippingConfig{Enabled: false},
+				FastForwardInfo: &persistencespb.FastForwardInfo{
+					TargetTime:                    target,
+					LastUpdateVersionedTransition: &persistencespb.VersionedTransition{NamespaceFailoverVersion: s.version, TransitionCount: 1},
+				},
 			},
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     errNoTimerFired,
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             errNoTimerFired,
 		},
 		{
-			name:        "NoFastForwardInfo",
-			tsi:         &persistencespb.TimeSkippingInfo{Config: &commonpb.TimeSkippingConfig{Enabled: true}},
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     errNoTimerFired,
+			name:                "FastForwardInfoDeleted",
+			tsi:                 &persistencespb.TimeSkippingInfo{Config: &commonpb.TimeSkippingConfig{Enabled: true}},
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             errNoTimerFired,
 		},
 		{
-			// Stamp 0 is a degenerate fast-forward (a real one always has Stamp >= 1); even a
-			// matching zero-stamp task must be treated as invalid and dropped.
-			name:        "StampZero",
-			tsi:         enabledFF(0, false),
-			taskVersion: s.version,
-			taskStamp:   0,
-			wantErr:     errNoTimerFired,
+			name:                "HasReached",
+			tsi:                 enabledFF(s.version, 1, true),
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             errNoTimerFired,
 		},
 		{
-			name:        "HasReached",
-			tsi:         enabledFF(1, true),
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     errNoTimerFired,
+			// fast-forward was reconfigured (transition count 2); the old task carries 1.
+			name:                "TransitionCountMismatch",
+			tsi:                 enabledFF(s.version, 2, false),
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             errNoTimerFired,
 		},
 		{
-			// fast-forward was reconfigured (stamp 2); the old task carries stamp 1.
-			name:        "StampMismatch",
-			tsi:         enabledFF(2, false),
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     errNoTimerFired,
+			// Transition count matches the live fast-forward, but the task carries a stale failover
+			// version, so CheckTaskVersion rejects it before the disable transition is applied.
+			name:                "VersionMismatch",
+			tsi:                 enabledFF(s.version, 1, false),
+			taskVersion:         s.version + 1,
+			taskTransitionCount: 1,
+			wantErr:             consts.ErrTaskVersionMismatch,
 		},
 		{
-			// Stamp matches the live fast-forward, but the task carries a stale failover version,
-			// so CheckTaskVersion rejects it before the disable transition is applied.
-			name:        "VersionMismatch",
-			tsi:         enabledFF(1, false),
-			taskVersion: s.version + 1,
-			taskStamp:   1,
-			wantErr:     consts.ErrTaskVersionMismatch,
-		},
-		{
-			name:        "HappyPath",
-			tsi:         enabledFF(1, false),
-			taskVersion: s.version,
-			taskStamp:   1,
-			wantErr:     nil,
+			name:                "HappyPath",
+			tsi:                 enabledFF(s.version, 1, false),
+			taskVersion:         s.version,
+			taskTransitionCount: 1,
+			wantErr:             nil,
 		},
 	} {
 		s.Run(tc.name, func() {
@@ -2515,8 +2510,10 @@ func (s *timerQueueActiveTaskExecutorSuite) TestExecuteTimeSkippingTimerTask() {
 				WorkflowKey:         workflowKey,
 				TaskID:              s.mustGenerateTaskID(),
 				VisibilityTimestamp: s.now.Add(time.Hour),
-				Version:             tc.taskVersion,
-				Stamp:               tc.taskStamp,
+				VersionedTransition: &persistencespb.VersionedTransition{
+					NamespaceFailoverVersion: tc.taskVersion,
+					TransitionCount:          tc.taskTransitionCount,
+				},
 			}
 			s.mockExecutionMgr.EXPECT().GetWorkflowExecution(gomock.Any(), gomock.Any()).
 				Return(&persistence.GetWorkflowExecutionResponse{State: pms}, nil)

service/history/timer_queue_active_task_executor_test.go

@@ -18,6 +18,7 @@ import (
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/common/namespace"
+	"go.temporal.io/server/common/persistence/transitionhistory"
 	"go.temporal.io/server/common/resource"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/consts"
@@ -245,7 +246,9 @@ func (t *timerQueueStandbyTaskExecutor) executeTimeSkippingTimerTask(
 		ffi := tsi.GetFastForwardInfo()
 
 		// the fast-forward this timer task is associated with is still valid and has not been reached so keep waiting
-		if ffi != nil && ffi.GetStamp() == timerTask.Stamp && !ffi.GetHasReached() {
+		if ffi != nil &&
+			transitionhistory.Compare(ffi.GetLastUpdateVersionedTransition(), timerTask.VersionedTransition) == 0 &&
+			!ffi.GetHasReached() {
 			return &struct{}{}, nil
 		}
 		return nil, nil