[v2] Event update time bug

executor/pkg/controller/taskaction_controller.go

@@ -383,7 +383,7 @@ func (r *TaskActionReconciler) buildActionEvent(
 	}
 
 	info := phaseInfo.Info()
-	updatedTime := updatedTimestamp(info, taskAction.Status.PhaseHistory)
+	updatedTime := updatedTimestamp(taskAction.Status.PhaseHistory)
 	reportedTime := reportedTimestamp(info)
 
 	event := &workflow.ActionEvent{
@@ -423,14 +423,11 @@ func observedCacheStatus(info *pluginsCore.TaskInfo) core.CatalogCacheStatus {
 	return cacheStatusFromExternalResources(info.ExternalResources)
 }
 
-func updatedTimestamp(info *pluginsCore.TaskInfo, history []flyteorgv1.PhaseTransition) *timestamppb.Timestamp {
-	if info != nil && info.OccurredAt != nil {
-		return timestamppb.New(*info.OccurredAt)
-	}
+func updatedTimestamp(history []flyteorgv1.PhaseTransition) *timestamppb.Timestamp {
 	if n := len(history); n > 0 {
 		return timestamppb.New(history[n-1].OccurredAt.Time)
 	}
-	return nil
+	return timestamppb.Now()
 }
 
 func reportedTimestamp(info *pluginsCore.TaskInfo) *timestamppb.Timestamp {

runs/service/get_action_details_test.go

@@ -363,6 +363,36 @@ func TestMergeEvents_PhaseTransitions(t *testing.T) {
 	assert.Equal(t, common.ActionPhase_ACTION_PHASE_SUCCEEDED, result.PhaseTransitions[2].Phase)
 }
 
+func TestMergeEvents_PrefersUpdatedTimeOverReportedTime(t *testing.T) {
+	now := time.Now()
+	events := []*workflow.ActionEvent{
+		{
+			Phase:        common.ActionPhase_ACTION_PHASE_QUEUED,
+			UpdatedTime:  timestamppb.New(now),
+			ReportedTime: timestamppb.New(now.Add(2 * time.Second)),
+		},
+		{
+			Phase:        common.ActionPhase_ACTION_PHASE_RUNNING,
+			UpdatedTime:  timestamppb.New(now.Add(1 * time.Second)),
+			ReportedTime: timestamppb.New(now),
+		},
+		{
+			Phase:        common.ActionPhase_ACTION_PHASE_SUCCEEDED,
+			UpdatedTime:  timestamppb.New(now.Add(2 * time.Second)),
+			ReportedTime: timestamppb.New(now.Add(3 * time.Second)),
+		},
+	}
+
+	result := mergeEvents(1, events)
+	assert.Equal(t, common.ActionPhase_ACTION_PHASE_SUCCEEDED, result.Phase)
+	assert.Equal(t, 3, len(result.PhaseTransitions))
+	assert.Equal(t, common.ActionPhase_ACTION_PHASE_QUEUED, result.PhaseTransitions[0].Phase)
+	assert.Equal(t, common.ActionPhase_ACTION_PHASE_RUNNING, result.PhaseTransitions[1].Phase)
+	assert.Equal(t, common.ActionPhase_ACTION_PHASE_SUCCEEDED, result.PhaseTransitions[2].Phase)
+	assert.True(t, result.PhaseTransitions[0].GetStartTime().AsTime().Equal(now))
+	assert.True(t, result.PhaseTransitions[1].GetStartTime().AsTime().Equal(now.Add(1*time.Second)))
+}
+
 func TestMergeEvents_MergesLogs(t *testing.T) {
 	now := time.Now()
 	events := []*workflow.ActionEvent{

runs/service/run_service.go

@@ -469,14 +469,23 @@ func (s *RunService) getAttempts(ctx context.Context, actionId *common.ActionIde
 
 // mergeEvents merges a set of events for the same attempt into a single ActionAttempt.
 func mergeEvents(attempt uint32, events []*workflow.ActionEvent) *workflow.ActionAttempt {
-	// Order events by reported time, falling back to updated time.
+	// Order events by the controller-observed phase transition time.
+	// ReportedTime reflects when an observation was emitted and can arrive out of order,
+	// so it is only used as a tie-breaker for otherwise-identical UpdatedTime values.
 	sort.SliceStable(events, func(i, j int) bool {
+		updatedTimeI := events[i].GetUpdatedTime()
+		updatedTimeJ := events[j].GetUpdatedTime()
+		if updatedTimeI != nil && updatedTimeJ != nil && !updatedTimeI.AsTime().Equal(updatedTimeJ.AsTime()) {
+			return updatedTimeI.AsTime().Before(updatedTimeJ.AsTime())
+		}
+
 		reportedTimeI := events[i].GetReportedTime()
 		reportedTimeJ := events[j].GetReportedTime()
-		if reportedTimeI != nil && reportedTimeJ != nil {
+		if reportedTimeI != nil && reportedTimeJ != nil && !reportedTimeI.AsTime().Equal(reportedTimeJ.AsTime()) {
 			return reportedTimeI.AsTime().Before(reportedTimeJ.AsTime())
 		}
-		return events[i].GetUpdatedTime().AsTime().Before(events[j].GetUpdatedTime().AsTime())
+
+		return phaseOrder(events[i].GetPhase()) < phaseOrder(events[j].GetPhase())
 	})
 
 	if len(events) == 0 {
@@ -578,6 +587,26 @@ func mergeEvents(attempt uint32, events []*workflow.ActionEvent) *workflow.Actio
 	}
 }
 
+func phaseOrder(phase common.ActionPhase) int {
+	switch phase {
+	case common.ActionPhase_ACTION_PHASE_QUEUED:
+		return 0
+	case common.ActionPhase_ACTION_PHASE_WAITING_FOR_RESOURCES:
+		return 1
+	case common.ActionPhase_ACTION_PHASE_INITIALIZING:
+		return 2
+	case common.ActionPhase_ACTION_PHASE_RUNNING:
+		return 3
+	case common.ActionPhase_ACTION_PHASE_SUCCEEDED,
+		common.ActionPhase_ACTION_PHASE_FAILED,
+		common.ActionPhase_ACTION_PHASE_ABORTED,
+		common.ActionPhase_ACTION_PHASE_TIMED_OUT:
+		return 4
+	default:
+		return 5
+	}
+}
+
 func IsTerminalPhase(phase common.ActionPhase) bool {
 	return phase == common.ActionPhase_ACTION_PHASE_FAILED ||
 		phase == common.ActionPhase_ACTION_PHASE_SUCCEEDED ||