Flush buffered events before emitting time-skipping transition

A time-skipping transition must be the last event of the transaction that
produces it: it is a server-generated state change applied synchronously
(it bumps AccumulatedSkippedDuration and drives timer-task regeneration in
the same close-transaction), so it must order after every other event.

WORKFLOW_EXECUTION_OPTIONS_UPDATED is a buffered event type while the
transition is not. When an options update re-enables (or enables) time
skipping on an idle workflow with a pending timer, the close-transaction
emits a transition in the same transaction as the buffered OPTIONS_UPDATED.
The non-buffered transition was allocated an event ID immediately, before
the buffered OPTIONS_UPDATED was flushed in closeTransactionPrepareEvents,
inverting their order and producing non-monotonic event timestamps.

Flush pending buffered events in closeTransactionHandleWorkflowTimeSkipping
before emitting the transition. Safe here: isWorkflowSkippable already
guarantees no in-flight workflow task, which is FlushBufferedEvents' own
precondition. The transition stays non-buffered, so reorderBuffer never
moves it and it remains the last-allocated event.

Add TestTransitionEventIsLastInTransaction to TimeSkippingTestSuite guarding
the invariant.

Author: feiyang3cat

service/history/workflow/timeskipping.go

@@ -404,14 +404,17 @@ func (ms *MutableStateImpl) closeTransactionHandleWorkflowTimeSkipping(
 		if !transition.IsValid() {
 			return false
 		}
-		// 3. state change
+		// 3. flush buffered events so that previous events will have an eventID smaller
+		// than the eventID for time skipping
+		ms.FlushBufferedEvents()
+		// 4. state change
 		_, err := ms.AddWorkflowExecutionTimeSkippingTransitionedEvent(
 			ctx, transition.TargetTime, transition.DisabledAfterFastForward)
 		if err != nil {
 			ms.logger.Error("failed to add workflow execution time skipping transitioned event", tag.Error(err))
 			return false
 		}
-		// 4. task regeneration
+		// 5. task regeneration
 		return true
 	case historyi.TransactionPolicyPassive:
 		return false

tests/timeskipping_test.go

@@ -1246,3 +1246,91 @@ 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))
 }
+
+// TestTransitionEventIsLastInTransaction guarantees that a time-skipping transition is the last
+// event of the transaction that produces it — in particular that it is ordered after a
+// WorkflowExecutionOptionsUpdated event emitted in the same transaction.
+//
+// Repro of the ordering bug this guards against: start without time skipping and schedule a user
+// timer so the workflow goes idle, then enable time skipping via UpdateWorkflowExecutionOptions.
+// That update's close-transaction finds the idle workflow now skippable and emits a transition in
+// the same transaction as the OPTIONS_UPDATED event. OPTIONS_UPDATED is a buffered event type
+// while the transition is not, so without flushing buffered events before emitting the transition,
+// the non-buffered transition is allocated an event ID before the buffered OPTIONS_UPDATED is
+// flushed — inverting their order and producing non-monotonic event timestamps.
+func (s *TimeSkippingTestSuite) TestTransitionEventIsLastInTransaction() {
+	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
+		}
+	}
+	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())
+
+	// Event IDs and timestamps must both be non-decreasing across the whole history.
+	for i := 1; i < len(hist); i++ {
+		prev, cur := hist[i-1], hist[i]
+		s.Greater(cur.GetEventId(), prev.GetEventId(),
+			"event IDs must be strictly increasing: event %d follows event %d",
+			cur.GetEventId(), prev.GetEventId())
+		s.False(cur.GetEventTime().AsTime().Before(prev.GetEventTime().AsTime()),
+			"event %d (%s) time precedes event %d (%s)",
+			cur.GetEventId(), cur.GetEventType(), prev.GetEventId(), prev.GetEventType())
+	}
+
+	_, _ = env.FrontendClient().TerminateWorkflowExecution(ctx, &workflowservice.TerminateWorkflowExecutionRequest{
+		Namespace:         env.Namespace().String(),
+		WorkflowExecution: &commonpb.WorkflowExecution{WorkflowId: tv.WorkflowID(), RunId: runID},
+		Reason:            "test cleanup",
+	})
+}