persistence: regenerate executions.pb.go with to_time api-linter comment

add timeSkippingEvent to workflow rebuilder
fix bugs of taskRefersher, and mutable virtual time

The proto source had the api-linter suppression comment for to_time but
the generated .pb.go was not updated to reflect it.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: feiyang3cat

api/persistence/v1/executions.pb.go

@@ -502,7 +502,7 @@ func NewMutableStateFromDB(
 	mutableState.approximateSize += dbRecord.ExecutionInfo.Size() - mutableState.executionInfo.Size()
 	mutableState.executionInfo = dbRecord.ExecutionInfo
 
-	if mutableState.executionInfo.GetTimeSkippingInfo().GetEnabled() {
+	if mutableState.executionInfo.GetTimeSkippingInfo() != nil {
 		mutableState.timeSource = clock.NewTimeSkippingTimeSource(
 			mutableState.timeSource,
 			mutableState.executionInfo.TimeSkippingInfo.TimeSkippedDetails,

service/history/workflow/mutable_state_impl.go

@@ -6242,3 +6242,26 @@ func (s *mutableStateSuite) TestTimeSkippedDurationRoundtrip() {
 		s.Equal(d, clock.TimeSkippedDurationFromTimestamp(clock.TimeSkippedDurationToTimestamp(d)), "duration: %v", d)
 	}
 }
+
+func (s *mutableStateSuite) TestNewMutableStateFromDB_TimeSkippingDisabled_VirtualTimeUsed() {
+	// Verify that when TimeSkippingInfo exists but Enabled=false, the mutable state still uses
+	// a TimeSkippingTimeSource so that virtual time (real time + skipped offset) is returned.
+	skipDuration := 2 * time.Hour
+	dbState := s.buildWorkflowMutableState()
+	dbState.ExecutionInfo.TimeSkippingInfo = &persistencespb.TimeSkippingInfo{
+		Enabled: false,
+		TimeSkippedDetails: []*persistencespb.TimeSkippedDetails{
+			{
+				DurationToSkip: clock.TimeSkippedDurationToTimestamp(skipDuration),
+			},
+		},
+	}
+
+	ms, err := NewMutableStateFromDB(s.mockShard, s.mockEventsCache, s.logger, s.namespaceEntry, dbState, 123)
+	s.NoError(err)
+
+	// VirtualTimeNow should be ahead of real time by the accumulated skipped duration.
+	realNow := s.mockShard.GetTimeSource().Now()
+	virtualNow := ms.VirtualTimeNow()
+	s.Equal(skipDuration, virtualNow.Sub(realNow))
+}

service/history/workflow/mutable_state_impl_test.go

@@ -672,6 +672,10 @@ func (b *MutableStateRebuilderImpl) applyEvents(
 			if err := b.mutableState.ApplyWorkflowExecutionUnpausedEvent(event); err != nil {
 				return nil, err
 			}
+		case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPED:
+			if err := b.mutableState.ApplyWorkflowExecutionTimeSkippedEvent(ctx, event); err != nil {
+				return nil, err
+			}
 
 		default:
 			def, ok := b.shard.StateMachineRegistry().EventDefinition(event.GetEventType())

service/history/workflow/mutable_state_rebuilder.go

@@ -2159,6 +2159,37 @@ func (s *stateBuilderSuite) TestApplyEvents_HSMRegistry() {
 	s.Equal(enumsspb.NEXUS_OPERATION_STATE_SCHEDULED, sm.State())
 }
 
+func (s *stateBuilderSuite) TestApplyEvents_EventTypeWorkflowExecutionTimeSkipped() {
+	version := int64(1)
+	requestID := uuid.NewString()
+	execution := &commonpb.WorkflowExecution{
+		WorkflowId: "some random workflow ID",
+		RunId:      tests.RunID,
+	}
+
+	now := time.Now().UTC()
+	event := &historypb.HistoryEvent{
+		TaskId:    rand.Int63(),
+		Version:   version,
+		EventId:   130,
+		EventTime: timestamppb.New(now),
+		EventType: enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPED,
+		Attributes: &historypb.HistoryEvent_WorkflowExecutionTimeSkippedEventAttributes{
+			WorkflowExecutionTimeSkippedEventAttributes: &historypb.WorkflowExecutionTimeSkippedEventAttributes{
+				ToTime: timestamppb.New(now.Add(time.Hour)),
+			},
+		},
+	}
+
+	s.mockMutableState.EXPECT().ApplyWorkflowExecutionTimeSkippedEvent(gomock.Any(), protomock.Eq(event)).Return(nil)
+	s.mockUpdateVersion(event)
+	s.mockMutableState.EXPECT().ClearStickyTaskQueue()
+
+	_, err := s.stateRebuilder.ApplyEvents(context.Background(), tests.NamespaceID, requestID, execution, s.toHistory(event), nil, "")
+	s.NoError(err)
+	s.Equal(event.TaskId, s.executionInfo.LastRunningClock)
+}
+
 func (p *testTaskGeneratorProvider) NewTaskGenerator(
 	shardContext historyi.ShardContext,
 	mutableState historyi.MutableState,

service/history/workflow/mutable_state_rebuilder_test.go

@@ -10,6 +10,7 @@ import (
 	enumsspb "go.temporal.io/server/api/enums/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/persistence/transitionhistory"
 	"go.temporal.io/server/common/primitives/timestamp"
 	"go.temporal.io/server/service/history/hsm"
@@ -76,7 +77,7 @@ func (r *TaskRefresherImpl) Refresh(
 		return err
 	}
 
-	r.applyTimeSkippingOffsetToUserTimerTasks(mutableState)
+	r.applyTimeSkippingOffsetToTimerTasks(mutableState)
 
 	if err := mutableState.ChasmTree().RefreshTasks(); err != nil {
 		return err
@@ -692,13 +693,16 @@ func (r *TaskRefresherImpl) refreshTasksForSubStateMachines(
 	return nil
 }
 
-// applyTimeSkippingOffsetToUserTimerTasks shifts the VisibilityTimestamp of all user timer tasks
+// applyTimeSkippingOffsetToTimerTasks shifts the VisibilityTimestamp of all pending timer tasks
 // that were just generated by subtracting the total skipped duration. This ensures the timer queue
 // fires them at the correct real-clock time relative to the advanced virtual clock.
 //
-// skippedDuration = latestSkipToTimePoint - realNow
-// adjustedFireTime = virtualFireTime - skippedDuration
-func (r *TaskRefresherImpl) applyTimeSkippingOffsetToUserTimerTasks(mutableState historyi.MutableState) {
+// skippedDuration = sum of all DurationToSkip in TimeSkippedDetails
+// adjustedFireTime = max(now, virtualFireTime - skippedDuration)
+//
+// DeleteHistoryEventTask and TimeSkippingTimerTask are excluded: the former is a retention-based
+// cleanup task unrelated to virtual time; the latter is the time-skipping mechanism itself.
+func (r *TaskRefresherImpl) applyTimeSkippingOffsetToTimerTasks(mutableState historyi.MutableState) {
 	timeSkippingInfo := mutableState.GetExecutionInfo().GetTimeSkippingInfo()
 	if timeSkippingInfo == nil {
 		return
@@ -708,9 +712,7 @@ func (r *TaskRefresherImpl) applyTimeSkippingOffsetToUserTimerTasks(mutableState
 		return
 	}
 
-	latestDetail := details[len(details)-1]
-	latestTargetVirtualTime := latestDetail.GetToTime().AsTime()
-	skippedDuration := latestTargetVirtualTime.Sub(r.shard.GetTimeSource().Now())
+	skippedDuration := clock.ComputeTotalSkippedOffset(details)
 	if skippedDuration <= 0 {
 		return
 	}
@@ -719,9 +721,18 @@ func (r *TaskRefresherImpl) applyTimeSkippingOffsetToUserTimerTasks(mutableState
 	if !ok {
 		return
 	}
+	now := r.shard.GetTimeSource().Now()
 	for _, task := range ms.InsertTasks[tasks.CategoryTimer] {
-		if userTimerTask, ok := task.(*tasks.UserTimerTask); ok {
-			userTimerTask.VisibilityTimestamp = userTimerTask.VisibilityTimestamp.Add(-skippedDuration)
+		switch task.GetType() {
+		case enumsspb.TASK_TYPE_DELETE_HISTORY_EVENT, enumsspb.TASK_TYPE_TIME_SKIPPING:
+			// not virtual-time based, skip
+			continue
+		default:
+		}
+		adjusted := task.GetVisibilityTime().Add(-skippedDuration)
+		if adjusted.Before(now) {
+			adjusted = now
 		}
+		task.SetVisibilityTime(adjusted)
 	}
 }

service/history/workflow/task_refresher.go

@@ -14,6 +14,7 @@ import (
 	historyspb "go.temporal.io/server/api/history/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/common"
+	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/namespace"
@@ -1434,6 +1435,151 @@ func (s *taskRefresherSuite) TestRefreshSubStateMachineTasks() {
 	s.False(hsmRoot.Dirty())
 }
 
+// buildMutableStateWithTimeSkipping creates a MutableStateImpl with TimeSkippingInfo having
+// the given total skipped duration in a single entry.
+func (s *taskRefresherSuite) buildMutableStateWithTimeSkipping(totalSkip time.Duration, enabled bool) *MutableStateImpl {
+	dbRecord := &persistencespb.WorkflowMutableState{
+		ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
+			NamespaceId: tests.NamespaceID.String(),
+			WorkflowId:  tests.WorkflowID,
+			TimeSkippingInfo: &persistencespb.TimeSkippingInfo{
+				Enabled: enabled,
+				TimeSkippedDetails: []*persistencespb.TimeSkippedDetails{
+					{DurationToSkip: clock.TimeSkippedDurationToTimestamp(totalSkip)},
+				},
+			},
+		},
+		ExecutionState: &persistencespb.WorkflowExecutionState{
+			RunId:  tests.RunID,
+			State:  enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
+			Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
+		},
+		NextEventId: 1,
+	}
+	ms, err := NewMutableStateFromDB(
+		s.mockShard,
+		s.mockShard.GetEventsCache(),
+		log.NewTestLogger(),
+		tests.LocalNamespaceEntry,
+		dbRecord,
+		1,
+	)
+	s.NoError(err)
+	return ms
+}
+
+func (s *taskRefresherSuite) TestApplyTimeSkippingOffset_AdjustsAllTimerTaskTypes() {
+	skipDuration := 2 * time.Hour
+	ms := s.buildMutableStateWithTimeSkipping(skipDuration, true)
+
+	now := s.mockShard.GetTimeSource().Now()
+	fireTime := now.Add(3 * time.Hour)
+	wfKey := ms.GetWorkflowKey()
+
+	ms.InsertTasks[tasks.CategoryTimer] = []tasks.Task{
+		&tasks.UserTimerTask{WorkflowKey: wfKey, VisibilityTimestamp: fireTime},
+		&tasks.ActivityTimeoutTask{WorkflowKey: wfKey, VisibilityTimestamp: fireTime},
+		&tasks.WorkflowRunTimeoutTask{WorkflowKey: wfKey, VisibilityTimestamp: fireTime},
+		&tasks.WorkflowExecutionTimeoutTask{NamespaceID: wfKey.NamespaceID, WorkflowID: wfKey.WorkflowID, VisibilityTimestamp: fireTime},
+	}
+
+	s.taskRefresher.applyTimeSkippingOffsetToTimerTasks(ms)
+
+	expected := fireTime.Add(-skipDuration) // now + 1h
+	for _, task := range ms.InsertTasks[tasks.CategoryTimer] {
+		s.Equal(expected, task.GetVisibilityTime(), "task type %v should be adjusted", task.GetType())
+	}
+}
+
+func (s *taskRefresherSuite) TestApplyTimeSkippingOffset_ExcludesDeleteAndTimeSkippingTasks() {
+	skipDuration := 2 * time.Hour
+	ms := s.buildMutableStateWithTimeSkipping(skipDuration, true)
+
+	now := s.mockShard.GetTimeSource().Now()
+	fireTime := now.Add(3 * time.Hour)
+	wfKey := ms.GetWorkflowKey()
+
+	ms.InsertTasks[tasks.CategoryTimer] = []tasks.Task{
+		&tasks.DeleteHistoryEventTask{WorkflowKey: wfKey, VisibilityTimestamp: fireTime},
+		&tasks.TimeSkippingTimerTask{WorkflowKey: wfKey, VisibilityTimestamp: fireTime},
+	}
+
+	s.taskRefresher.applyTimeSkippingOffsetToTimerTasks(ms)
+
+	for _, task := range ms.InsertTasks[tasks.CategoryTimer] {
+		s.Equal(fireTime, task.GetVisibilityTime(), "task type %v should not be adjusted", task.GetType())
+	}
+}
+
+func (s *taskRefresherSuite) TestApplyTimeSkippingOffset_MultipleSkipEntries_UsesTotalOffset() {
+	// Two separate skip entries: 1h + 1h = 2h total. The old bug used
+	// latestTargetVirtualTime - realNow instead of summing DurationToSkip.
+	dbRecord := &persistencespb.WorkflowMutableState{
+		ExecutionInfo: &persistencespb.WorkflowExecutionInfo{
+			NamespaceId: tests.NamespaceID.String(),
+			WorkflowId:  tests.WorkflowID,
+			TimeSkippingInfo: &persistencespb.TimeSkippingInfo{
+				Enabled: true,
+				TimeSkippedDetails: []*persistencespb.TimeSkippedDetails{
+					{DurationToSkip: clock.TimeSkippedDurationToTimestamp(time.Hour)},
+					{DurationToSkip: clock.TimeSkippedDurationToTimestamp(time.Hour)},
+				},
+			},
+		},
+		ExecutionState: &persistencespb.WorkflowExecutionState{
+			RunId:  tests.RunID,
+			State:  enumsspb.WORKFLOW_EXECUTION_STATE_RUNNING,
+			Status: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
+		},
+		NextEventId: 1,
+	}
+	ms, err := NewMutableStateFromDB(s.mockShard, s.mockShard.GetEventsCache(), log.NewTestLogger(), tests.LocalNamespaceEntry, dbRecord, 1)
+	s.NoError(err)
+
+	now := s.mockShard.GetTimeSource().Now()
+	fireTime := now.Add(3 * time.Hour)
+	ms.InsertTasks[tasks.CategoryTimer] = []tasks.Task{
+		&tasks.UserTimerTask{WorkflowKey: ms.GetWorkflowKey(), VisibilityTimestamp: fireTime},
+	}
+
+	s.taskRefresher.applyTimeSkippingOffsetToTimerTasks(ms)
+
+	// Total offset = 2h, so adjusted = now + 3h - 2h = now + 1h
+	s.Equal(now.Add(time.Hour), ms.InsertTasks[tasks.CategoryTimer][0].GetVisibilityTime())
+}
+
+func (s *taskRefresherSuite) TestApplyTimeSkippingOffset_ClampsToNow() {
+	skipDuration := 5 * time.Hour
+	ms := s.buildMutableStateWithTimeSkipping(skipDuration, true)
+
+	now := s.mockShard.GetTimeSource().Now()
+	fireTime := now.Add(3 * time.Hour) // adjusted = now + 3h - 5h = 2h in the past
+	ms.InsertTasks[tasks.CategoryTimer] = []tasks.Task{
+		&tasks.UserTimerTask{WorkflowKey: ms.GetWorkflowKey(), VisibilityTimestamp: fireTime},
+	}
+
+	s.taskRefresher.applyTimeSkippingOffsetToTimerTasks(ms)
+
+	s.Equal(now, ms.InsertTasks[tasks.CategoryTimer][0].GetVisibilityTime())
+}
+
+func (s *taskRefresherSuite) TestApplyTimeSkippingOffset_DisabledButHasSkips_StillAdjusts() {
+	// Even when Enabled=false, accumulated skips must be applied because virtual time
+	// has deviated from wall clock time and tasks must fire at the correct real time.
+	skipDuration := 2 * time.Hour
+	ms := s.buildMutableStateWithTimeSkipping(skipDuration, false)
+
+	now := s.mockShard.GetTimeSource().Now()
+	fireTime := now.Add(3 * time.Hour)
+	ms.InsertTasks[tasks.CategoryTimer] = []tasks.Task{
+		&tasks.UserTimerTask{WorkflowKey: ms.GetWorkflowKey(), VisibilityTimestamp: fireTime},
+	}
+
+	s.taskRefresher.applyTimeSkippingOffsetToTimerTasks(ms)
+
+	s.Equal(fireTime.Add(-skipDuration), ms.InsertTasks[tasks.CategoryTimer][0].GetVisibilityTime())
+}
+
 type mockTaskGeneratorProvider struct {
 	mockTaskGenerator *MockTaskGenerator
 }