timeskipping support replication

Author: feiyang3cat

api/persistence/v1/executions.go-helpers.pb.go

@@ -325,10 +325,22 @@ message TimeSkippingInfo {
   temporal.api.workflow.v1.TimeSkippingConfig config = 1;
 
   // Total skipped duration for the current workflow execution run, including any
-  ./* inherited skipped duration carried over from a preceding execution that started this run. */google.protobuf.Duration accumulated_skipped_duration = 2;
+  // inherited skipped duration carried over from a preceding execution that started this run.
+  google.protobuf.Duration accumulated_skipped_duration = 2;
 
   // The current bound based on elapsed duration for time skipping.
   TimeSkippingBoundInfo current_elapsed_duration_bound = 3;
+
+  repeated TimeSkipTaskRegenEntry task_regen_entries = 4;
+}
+
+message TimeSkipTaskRegenEntry {
+  // The TimeSkippingTransitionedEvent that caused the skipped duration.
+  int64 source_event_id = 1;
+
+  // the mask to have idempotency for task recreation
+  // especially when tasks are regenerated in the passive cluster in a state-based replication
+  int32 task_regen_status = 2;
 }
 
 message TimeSkippingBoundInfo {

api/persistence/v1/executions.pb.go

@@ -110,9 +110,7 @@ func (t *timerQueueStandbyTaskExecutor) Execute(
 	case *tasks.ChasmTask:
 		err = t.executeChasmSideEffectTimerTask(ctx, task)
 	case *tasks.TimeSkippingTimerTask:
-		// todo@time-skipping: replication. The disable-after-bound transition is emitted
-		// on the active side and will replicate; standby drops the local task.
-		err = nil
+		err = t.executeTimeSkippingTimerTask(ctx, task)
 	default:
 		err = queueserrors.NewUnprocessableTaskError("unknown task type")
 	}
@@ -235,17 +233,32 @@ func (t *timerQueueStandbyTaskExecutor) discardChasmTask(
 	)
 }
 
+// executeTimeSkippingTimerTask is a deliberate no-op on the standby side. The
+// disable-after-bound transition is emitted on the active cluster as a regular
+// history event and replicates through the normal path; the bound flag is
+// advisory on standby (it doesn't drive any local progress), so there's nothing
+// for the verifier to gain by checking it.
+func (t *timerQueueStandbyTaskExecutor) executeTimeSkippingTimerTask(
+	_ context.Context,
+	_ *tasks.TimeSkippingTimerTask,
+) error {
+	return nil
+}
+
 func (t *timerQueueStandbyTaskExecutor) executeUserTimerTimeoutTask(
 	ctx context.Context,
 	timerTask *tasks.UserTimerTask,
 ) error {
-	referenceTime := t.Now()
 	actionFn := func(_ context.Context, wfContext historyi.WorkflowContext, mutableState historyi.MutableState, _ historyi.ReleaseWorkflowContextFunc) (any, error) {
 		if !mutableState.IsWorkflowExecutionRunning() {
 			// workflow already finished, no need to process the timer
 			return nil, nil
 		}
 
+		// Use mutableState.Now() as reference time as the mutable state may use virtual time
+		// which can skip duration and be before the wall clock time.
+		referenceTime := mutableState.Now()
+
 		timerSequence := t.getTimerSequence(mutableState)
 		timerSequenceIDs := timerSequence.LoadAndSortUserTimers()
 		if len(timerSequenceIDs) > 0 {
@@ -299,13 +312,14 @@ func (t *timerQueueStandbyTaskExecutor) executeActivityTimeoutTask(
 	//
 	// the overall solution is to attempt to generate a new activity timer task whenever the
 	// task passed in is safe to be throw away.
-	referenceTime := t.Now()
 	actionFn := func(ctx context.Context, wfContext historyi.WorkflowContext, mutableState historyi.MutableState, _ historyi.ReleaseWorkflowContextFunc) (any, error) {
 		if !mutableState.IsWorkflowExecutionRunning() {
 			// workflow already finished, no need to process the timer
 			return nil, nil
 		}
 
+		referenceTime := mutableState.Now()
+
 		timerSequence := t.getTimerSequence(mutableState)
 		updateMutableState := false
 		timerSequenceIDs := timerSequence.LoadAndSortActivityTimers()
@@ -778,9 +792,13 @@ func (t *timerQueueStandbyTaskExecutor) pushActivity(
 	)
 }
 
-// TODO: deprecate this function and always use t.Now()
-// Only test code sets t.clusterName to be non-current cluster name
-// and advance the time by setting calling shardContext.SetCurrentTime.
+// getCurrentTime returns the shard's wall-clock view of "now" for t.clusterName.
+// Must stay wall-clock: it gates standby task-retry timing against VisibilityTime
+// (also wall-clock); mutableState.Now() is virtual time and would force-discard
+// time-skipping workflows. actionFn closures compare against virtual timestamps,
+// so they use mutableState.Now() instead.
+//
+// TODO: deprecate once tests stop using a non-current clusterName and shardContext.SetCurrentTime.
 func (t *timerQueueStandbyTaskExecutor) getCurrentTime() time.Time {
 	return t.shardContext.GetCurrentTime(t.clusterName)
 }

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

@@ -4012,8 +4012,12 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionTimeSkippingTransitionedEvent(
 		tsi.AccumulatedSkippedDuration = durationpb.New(0)
 	}
 	accumulatedSkippedDuration := tsi.GetAccumulatedSkippedDuration().AsDuration()
+
+	// todo@time-skipping: add precision to the entire time-skipping feature
+	var skipDelta time.Duration
 	if !timeNotSet(attr.TargetTime) {
-		accumulatedSkippedDuration += attr.TargetTime.AsTime().Sub(event.GetEventTime().AsTime())
+		skipDelta = attr.TargetTime.AsTime().Sub(event.GetEventTime().AsTime())
+		accumulatedSkippedDuration += skipDelta
 	}
 	tsi.AccumulatedSkippedDuration = durationpb.New(accumulatedSkippedDuration)
 	tsi.Config.Enabled = !attr.GetDisabledAfterBound()
@@ -4022,6 +4026,13 @@ func (ms *MutableStateImpl) ApplyWorkflowExecutionTimeSkippingTransitionedEvent(
 		tsi.CurrentElapsedDurationBound.HasReached = true
 	}
 
+	if skipDelta > 0 {
+		tsi.TaskRegenEntries = append(tsi.TaskRegenEntries, &persistencespb.TimeSkipTaskRegenEntry{
+			SourceEventId:   event.GetEventId(),
+			TaskRegenStatus: TimeSkippingTaskRegenStatusNone,
+		})
+	}
+
 	ms.timeSkippingInfoUpdated = true
 	return nil
 }
@@ -8483,7 +8494,6 @@ func (ms *MutableStateImpl) closeTransactionRegenerateTimerTasksForTimeSkipping(
 ) error {
 	switch transactionPolicy {
 	case historyi.TransactionPolicyActive:
-		// todo@time-skipping: impacts of paused workflow to be considered
 		if !ms.IsWorkflowExecutionRunning() {
 			return nil
 		}
@@ -9011,6 +9021,8 @@ func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowEx
 		}
 	}
 
+	ms.applyIncomingTimeSkippingInfo(current, incoming)
+
 	doNotSync := func(v any) []any {
 		info, ok := v.(*persistencespb.WorkflowExecutionInfo)
 		if !ok || info == nil {
@@ -9044,6 +9056,10 @@ func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowEx
 			&info.StateMachineTimers,
 			&info.TaskGenerationShardClockTimestamp,
 			&info.UpdateInfos,
+			// TimeSkippingInfo is handled by applyIncomingTimeSkippingInfo above
+			// because TimeSkipTaskRegenEntry.TaskRegenStatus is per-cluster local state
+			// and must not be clobbered by the wire value.
+			&info.TimeSkippingInfo,
 		}
 		if !isSnapshot {
 			ignoreFields = append(ignoreFields, &info.SubStateMachineTombstoneBatches)
@@ -9060,6 +9076,40 @@ func (ms *MutableStateImpl) syncExecutionInfo(current *persistencespb.WorkflowEx
 	return nil
 }
 
+// applyIncomingTimeSkippingInfo merges the incoming TimeSkippingInfo (from a
+// replication mutation/snapshot) into current, preserving each local
+// TimeSkipTaskRegenEntry's TaskRegenStatus by SourceEventId. Entries that are
+// new to this cluster get TaskRegenStatus reset to None so the local refresh
+// can regenerate timer tasks. Mirrors the TimerTaskStatus override pattern at
+// applyUpdatesToSubStateMachines.
+func (ms *MutableStateImpl) applyIncomingTimeSkippingInfo(
+	current *persistencespb.WorkflowExecutionInfo,
+	incoming *persistencespb.WorkflowExecutionInfo,
+) {
+	if incoming.GetTimeSkippingInfo() == nil {
+		current.TimeSkippingInfo = nil
+		return
+	}
+
+	localStatusByEventID := make(map[int64]int32)
+	if current.GetTimeSkippingInfo() != nil {
+		for _, d := range current.TimeSkippingInfo.GetTaskRegenEntries() {
+			localStatusByEventID[d.GetSourceEventId()] = d.GetTaskRegenStatus()
+		}
+	}
+
+	newTSI := common.CloneProto(incoming.TimeSkippingInfo)
+	for _, d := range newTSI.GetTaskRegenEntries() {
+		if localStatus, ok := localStatusByEventID[d.GetSourceEventId()]; ok {
+			d.TaskRegenStatus = localStatus
+		} else {
+			d.TaskRegenStatus = TimeSkippingTaskRegenStatusNone
+		}
+	}
+
+	current.TimeSkippingInfo = newTSI
+}
+
 func (ms *MutableStateImpl) syncSubStateMachinesByType(incoming map[string]*persistencespb.StateMachineMap) error {
 	// check if there is node been deleted
 	currentHSM := ms.HSM()
@@ -9541,10 +9591,17 @@ func (ms *MutableStateImpl) initTimeSkippingInfo(
 	initialSkippedDuration *durationpb.Duration,
 	currentEventID int64,
 ) {
-	ms.executionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+	tsi := &persistencespb.TimeSkippingInfo{
 		Config:                     config,
 		AccumulatedSkippedDuration: initialSkippedDuration,
 	}
+	if initialSkippedDuration != nil && initialSkippedDuration.AsDuration() > 0 {
+		tsi.TaskRegenEntries = []*persistencespb.TimeSkipTaskRegenEntry{{
+			SourceEventId:   currentEventID,
+			TaskRegenStatus: TimeSkippingTaskRegenStatusNone,
+		}}
+	}
+	ms.executionInfo.TimeSkippingInfo = tsi
 	ms.wrapTimeSourceWithTimeSkipping()
 	ms.shiftWorkflowTimes(initialSkippedDuration)
 	ms.applyTimeSkippingBound(currentEventID)

service/history/timer_queue_standby_task_executor.go

@@ -8153,3 +8153,115 @@ func (s *mutableStateSuite) TestCloseTransactionTimeSkipping_Bound() {
 		s.InDelta(float64(time.Hour), float64(got), float64(time.Millisecond))
 	})
 }
+
+// TestApplyIncomingTimeSkippingInfo_PreservesLocalRegenStatus verifies the
+// merge contract: TimeSkipTaskRegenEntry.TaskRegenStatus is per-cluster local
+// state. For details the receiver already has (matched by SourceEventId), the
+// local status is preserved; new details are reset to None so the local
+// refresh can regenerate timer tasks.
+func TestApplyIncomingTimeSkippingInfo_PreservesLocalRegenStatus(t *testing.T) {
+	t.Parallel()
+
+	ms := &MutableStateImpl{}
+
+	for _, tc := range []struct {
+		name         string
+		current      *persistencespb.TimeSkippingInfo
+		incoming     *persistencespb.TimeSkippingInfo
+		wantStatuses map[int64]int32 // SourceEventId -> expected TaskRegenStatus
+	}{
+		{
+			name:    "current nil",
+			current: nil,
+			incoming: &persistencespb.TimeSkippingInfo{
+				TaskRegenEntries: []*persistencespb.TimeSkipTaskRegenEntry{
+					{SourceEventId: 5, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+					{SourceEventId: 9, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+				},
+			},
+			wantStatuses: map[int64]int32{
+				5: TimeSkippingTaskRegenStatusNone,
+				9: TimeSkippingTaskRegenStatusNone,
+			},
+		},
+		{
+			name: "matching by SourceEventId preserves local status; new entry resets to None",
+			current: &persistencespb.TimeSkippingInfo{
+				TaskRegenEntries: []*persistencespb.TimeSkipTaskRegenEntry{
+					{SourceEventId: 5, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+					{SourceEventId: 9, TaskRegenStatus: TimeSkippingTaskRegenStatusNone},
+				},
+			},
+			incoming: &persistencespb.TimeSkippingInfo{
+				TaskRegenEntries: []*persistencespb.TimeSkipTaskRegenEntry{
+					// Wire always carries Regenerated (sender marked before sending).
+					{SourceEventId: 5, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+					{SourceEventId: 9, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+					{SourceEventId: 13, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+				},
+			},
+			wantStatuses: map[int64]int32{
+				5:  TimeSkippingTaskRegenStatusRegenerated, // local kept
+				9:  TimeSkippingTaskRegenStatusNone,        // local kept
+				13: TimeSkippingTaskRegenStatusNone,        // new -> None
+			},
+		},
+	} {
+		tc := tc
+		t.Run(tc.name, func(t *testing.T) {
+			t.Parallel()
+
+			current := &persistencespb.WorkflowExecutionInfo{TimeSkippingInfo: tc.current}
+			incoming := &persistencespb.WorkflowExecutionInfo{TimeSkippingInfo: tc.incoming}
+
+			ms.applyIncomingTimeSkippingInfo(current, incoming)
+
+			if tc.wantStatuses == nil {
+				require.Nil(t, current.TimeSkippingInfo)
+				return
+			}
+			require.NotNil(t, current.TimeSkippingInfo)
+			require.Len(t, current.TimeSkippingInfo.TaskRegenEntries, len(tc.wantStatuses))
+			for _, d := range current.TimeSkippingInfo.TaskRegenEntries {
+				want, ok := tc.wantStatuses[d.GetSourceEventId()]
+				require.True(t, ok, "unexpected detail SourceEventId=%d in result", d.GetSourceEventId())
+				require.Equal(t, want, d.GetTaskRegenStatus(),
+					"detail SourceEventId=%d: expected status %d, got %d",
+					d.GetSourceEventId(), want, d.GetTaskRegenStatus())
+			}
+		})
+	}
+}
+
+// TestApplyIncomingTimeSkippingInfo_ClonesToBreakAliasing verifies the merge
+// clones the incoming TimeSkippingInfo so subsequent local mutations
+// (e.g. flipping TaskRegenStatus during PartialRefresh) cannot leak back into
+// the wire message.
+func TestApplyIncomingTimeSkippingInfo_ClonesToBreakAliasing(t *testing.T) {
+	t.Parallel()
+
+	ms := &MutableStateImpl{}
+
+	incoming := &persistencespb.WorkflowExecutionInfo{
+		TimeSkippingInfo: &persistencespb.TimeSkippingInfo{
+			TaskRegenEntries: []*persistencespb.TimeSkipTaskRegenEntry{
+				{SourceEventId: 5, TaskRegenStatus: TimeSkippingTaskRegenStatusRegenerated},
+			},
+		},
+	}
+	current := &persistencespb.WorkflowExecutionInfo{}
+
+	ms.applyIncomingTimeSkippingInfo(current, incoming)
+
+	require.NotSame(t, current.TimeSkippingInfo, incoming.TimeSkippingInfo,
+		"current.TimeSkippingInfo must be a cloned copy, not aliased to incoming")
+	require.NotSame(t, current.TimeSkippingInfo.TaskRegenEntries[0],
+		incoming.TimeSkippingInfo.TaskRegenEntries[0],
+		"detail must also be cloned")
+
+	// Mutate local; incoming must not change.
+	current.TimeSkippingInfo.TaskRegenEntries[0].TaskRegenStatus = TimeSkippingTaskRegenStatusRegenerated
+	require.Equal(t, TimeSkippingTaskRegenStatusRegenerated,
+		incoming.TimeSkippingInfo.TaskRegenEntries[0].GetTaskRegenStatus(),
+		"sanity: incoming was already Regenerated; this assertion is a regression guard if the wire value were ever mutated")
+}