keep a reasonable history events order when time skipping happens

service/history/historybuilder/event_store.go

@@ -340,9 +340,13 @@ func (b *EventStore) bufferEvent(
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_UPDATE_COMPLETED:
 		return false
 
-	case // time skipping related events should not be buffered
+	case // Buffer this event to ensure:
+		// 1) The transition is emitted after WORKFLOW_EXECUTION_OPTIONS_UPDATED in the
+		//    same transaction, so history reflects when time skipping actually becomes enabled.
+		// 2) The transition is not inserted between workflow task events when the
+		//    fast-forward timer disables time skipping.
 		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED:
-		return false
+		return true
 
 	// A paused workflow event *should be* allowed to be buffered since we want to accept any inflight workflow task completion.
 	// Since we buffer the paused event, we need to buffer unpaused event as well so that they don't go out of order.

service/history/historybuilder/history_builder_test.go

@@ -2229,14 +2229,13 @@ func (s *historyBuilderSuite) TestHasBufferEvent() {
 func (s *historyBuilderSuite) TestBufferEvent() {
 	// workflow status events will be assign event ID immediately
 	workflowEvents := map[enumspb.EventType]bool{
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED:                    true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:                  true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:                     true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:                  true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED:                 true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW:           true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED:                   true,
-		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED: true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED:          true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_COMPLETED:        true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_FAILED:           true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT:        true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TERMINATED:       true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CONTINUED_AS_NEW: true,
+		enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_CANCELED:         true,
 	}
 
 	// workflow task events will be assign event ID immediately

service/history/workflow/timeskipping.go

@@ -404,7 +404,7 @@ func (ms *MutableStateImpl) closeTransactionHandleWorkflowTimeSkipping(
 		if !transition.IsValid() {
 			return false
 		}
-		// 3. state change
+		// 3. state change.
 		_, err := ms.AddWorkflowExecutionTimeSkippingTransitionedEvent(
 			ctx, transition.TargetTime, transition.DisabledAfterFastForward)
 		if err != nil {

tests/timeskipping_fast_forward_test.go

@@ -442,3 +442,95 @@ func (s *TimeSkippingFastForwardFunctionalSuite) TestFastForward_EqualsRunTimeou
 	s.GreaterOrEqual(timedOutIdx, 0, "expected the workflow to time out")
 	s.Less(transitionIdx, timedOutIdx, "the time-skipping transition must be recorded before the timeout")
 }
+
+func (s *TimeSkippingFastForwardFunctionalSuite) TestEventsOrderOfFastForwardTimerFiredDuringStartedWorkflowTask() {
+	env := testcore.NewEnv(s.T())
+	env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
+	tv := testvars.New(s.T())
+	ctx := testcore.NewContext()
+
+	const (
+		fastForward = time.Minute
+		timer1Dur   = 45 * time.Second
+	)
+
+	cfg := &commonpb.TimeSkippingConfig{Enabled: true, FastForward: durationpb.New(fastForward)}
+	startResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           env.Namespace().String(),
+		WorkflowId:          tv.WorkflowID(),
+		WorkflowType:        tv.WorkflowType(),
+		TaskQueue:           tv.TaskQueue(),
+		WorkflowRunTimeout:  durationpb.New(24 * time.Hour),
+		WorkflowTaskTimeout: durationpb.New(90 * time.Second),
+		TimeSkippingConfig:  cfg,
+	})
+	s.NoError(err)
+	runID := startResp.RunId
+
+	// WT1: start timer1. Close-tx skips to timer1; timer1 fires; WT2 is scheduled.
+	_, err = env.TaskPoller().PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+		return &workflowservice.RespondWorkflowTaskCompletedRequest{
+			Commands: []*commandpb.Command{startTimerCmd("t1", timer1Dur)},
+		}, nil
+	})
+	s.NoError(err)
+
+	// WT2: poll (marks it STARTED), then hold it open until the fast-forward disable fires, so the
+	// transition is emitted while this workflow task is in flight. The eventual complete response is
+	// expected to be rejected (UnhandledCommand) because the buffered transition invalidated the WT.
+	_, _ = env.TaskPoller().PollWorkflowTask(&workflowservice.PollWorkflowTaskQueueRequest{}).
+		HandleTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+			s.AwaitTruef(func() bool {
+				ms := s.getMutableState(env, tv.WorkflowID(), runID)
+				return ms.State.ExecutionInfo.GetTimeSkippingInfo().GetFastForwardInfo().GetHasReached()
+			}, 75*time.Second, 200*time.Millisecond, "fast-forward disable did not fire while the workflow task was started")
+
+			// The disable transition mutates state synchronously in the close-transaction,
+			// even though its history event is still buffered while this WT is in flight:
+			// time skipping is already disabled here, before the event is flushed to history.
+			ms := s.getMutableState(env, tv.WorkflowID(), runID)
+			s.False(ms.State.ExecutionInfo.GetTimeSkippingInfo().GetConfig().GetEnabled(),
+				"time skipping must be disabled in mutable state once the fast-forward timer fired")
+			return &workflowservice.RespondWorkflowTaskCompletedRequest{
+				Commands: []*commandpb.Command{completeWorkflowCmd()},
+			}, nil
+		}, taskpoller.WithTimeout(90*time.Second))
+
+	hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
+
+	// A fast-forward disable transition (DisabledAfterFastForward) must be present.
+	sawDisableTransition := false
+	for _, e := range s.findTransitionedEvents(hist) {
+		if e.GetWorkflowExecutionTimeSkippingTransitionedEventAttributes().GetDisabledAfterFastForward() {
+			sawDisableTransition = true
+		}
+	}
+	s.True(sawDisableTransition, "expected a fast-forward disable transition")
+
+	// No transition may sit inside an open WorkflowTaskStarted->Completed/Failed/TimedOut window.
+	insideStartedWT := false
+	for _, e := range hist {
+		switch e.GetEventType() {
+		case enumspb.EVENT_TYPE_WORKFLOW_TASK_STARTED:
+			insideStartedWT = true
+		case enumspb.EVENT_TYPE_WORKFLOW_TASK_COMPLETED,
+			enumspb.EVENT_TYPE_WORKFLOW_TASK_FAILED,
+			enumspb.EVENT_TYPE_WORKFLOW_TASK_TIMED_OUT:
+			insideStartedWT = false
+		case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED:
+			s.False(insideStartedWT,
+				"time-skipping transition (event %d) must not be wedged inside an open workflow-task window",
+				e.GetEventId())
+		default:
+			// other event types are irrelevant to this ordering check
+		}
+	}
+
+	// Event IDs must be strictly increasing across the whole history.
+	for i := 1; i < len(hist); i++ {
+		s.Greater(hist[i].GetEventId(), hist[i-1].GetEventId(),
+			"event IDs must be strictly increasing: event %d follows event %d",
+			hist[i].GetEventId(), hist[i-1].GetEventId())
+	}
+}

tests/timeskipping_test.go

@@ -1246,3 +1246,70 @@ func (s *TimeSkippingFastForwardFunctionalSuite) TestTimeSkipping_ExecutionTimeo
 		"timing out at the execution timeout is not a fast-forward disable")
 	s.True(hasEventType(hist, enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIMED_OUT))
 }
+
+func (s *TimeSkippingTestSuite) TestTimeSkippingTransitionEventOrdersAfterOptionsUpdated() {
+	env := testcore.NewEnv(s.T())
+	env.OverrideDynamicConfig(dynamicconfig.TimeSkippingEnabled, true)
+	tv := testvars.New(s.T())
+	ctx := testcore.NewContext()
+
+	// Start WITHOUT time skipping.
+	startResp, err := env.FrontendClient().StartWorkflowExecution(ctx, &workflowservice.StartWorkflowExecutionRequest{
+		RequestId:           uuid.NewString(),
+		Namespace:           env.Namespace().String(),
+		WorkflowId:          tv.WorkflowID(),
+		WorkflowType:        tv.WorkflowType(),
+		TaskQueue:           tv.TaskQueue(),
+		WorkflowRunTimeout:  durationpb.New(24 * time.Hour),
+		WorkflowTaskTimeout: durationpb.New(10 * time.Second),
+	})
+	s.NoError(err)
+	runID := startResp.RunId
+
+	// WT1: schedule a user timer. Time skipping is off, so the workflow just goes idle with a
+	// pending timer (no skip yet).
+	_, err = env.TaskPoller().PollAndHandleWorkflowTask(tv, func(_ *workflowservice.PollWorkflowTaskQueueResponse) (*workflowservice.RespondWorkflowTaskCompletedRequest, error) {
+		return &workflowservice.RespondWorkflowTaskCompletedRequest{
+			Commands: []*commandpb.Command{startTimerCmd("t1", time.Hour)},
+		}, nil
+	})
+	s.NoError(err)
+
+	// Enable time skipping. The workflow is idle with a pending timer, so this update's close-tx
+	// emits a transition skipping to the timer — in the same transaction as OPTIONS_UPDATED.
+	_, err = env.FrontendClient().UpdateWorkflowExecutionOptions(ctx, &workflowservice.UpdateWorkflowExecutionOptionsRequest{
+		Namespace:         env.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
+		WorkflowExecutionOptions: &workflowpb.WorkflowExecutionOptions{
+			TimeSkippingConfig: &commonpb.TimeSkippingConfig{Enabled: true},
+		},
+		UpdateMask: &fieldmaskpb.FieldMask{Paths: []string{"time_skipping_config"}},
+	})
+	s.NoError(err)
+
+	hist := env.GetHistory(env.Namespace().String(), &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID})
+
+	// The transition and the OPTIONS_UPDATED that triggered it must both be present, and the
+	// OPTIONS_UPDATED must come first (the transition is the last event of the transaction).
+	optionsUpdatedIdx, transitionIdx := -1, -1
+	for i, e := range hist {
+		switch e.GetEventType() {
+		case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_OPTIONS_UPDATED:
+			optionsUpdatedIdx = i
+		case enumspb.EVENT_TYPE_WORKFLOW_EXECUTION_TIME_SKIPPING_TRANSITIONED:
+			transitionIdx = i
+		default:
+			// other event types are irrelevant to this ordering check
+		}
+	}
+	s.NotEqual(-1, optionsUpdatedIdx, "expected an OPTIONS_UPDATED event")
+	s.NotEqual(-1, transitionIdx, "expected a TIME_SKIPPING_TRANSITIONED event")
+	s.Less(optionsUpdatedIdx, transitionIdx,
+		"OPTIONS_UPDATED (event %d) must precede the transition it triggered (event %d)",
+		hist[optionsUpdatedIdx].GetEventId(), hist[transitionIdx].GetEventId())
+	_, _ = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
+		Namespace:         env.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
+		Reason:            "test cleanup",
+	})
+}