enhance(actions): show the steps the runner reports (#39720)

Gitea derives the steps of a job from the workflow file, so the `pre`
and `post` steps of actions only show up inside "Set up job" and
"Complete job".
Fixes https://github.com/go-gitea/gitea/issues/24604.

A runner can now report the steps it actually runs, and Gitea shows
those instead. Runners that don't report them keep the current view, and
older Gitea versions ignore them.

Related:
- Proto: https://gitea.com/gitea/actions-proto-def/pulls/19
- Runner: https://gitea.com/gitea/runner/pulls/1293

models/actions/task.go

@@ -7,6 +7,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"slices"
 	"strings"
 	"time"
 
@@ -482,6 +483,11 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 	for _, v := range state.Steps {
 		stepStates[v.Id] = v
 	}
+	reportedSteps := state.ReportedSteps
+	if err := validateReportedSteps(reportedSteps); err != nil {
+		log.Warn("Ignore the steps reported for task %d: %v", state.Id, err)
+		reportedSteps = nil
+	}
 
 	// Only one request can update the task because the final state needs to be calculated with all job states.
 	// Otherwise, concurrent requests with transaction will make the SQL read stale job state and result in wrong final state.
@@ -538,22 +544,14 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task
 			return err
 		}
 
+		if len(reportedSteps) > 0 {
+			return updateReportedSteps(ctx, task, reportedSteps, now)
+		}
+		if slices.ContainsFunc(task.Steps, (*ActionTaskStep).IsReported) {
+			return nil // the indexes of the steps of the workflow file no longer match the stored steps
+		}
 		for _, step := range task.Steps {
-			var result runnerv1.Result
-			if v, ok := stepStates[step.Index]; ok {
-				result = v.Result
-				step.LogIndex = v.LogIndex
-				step.LogLength = v.LogLength
-				if step.Started == 0 && v.StartedAt != nil {
-					step.Started = now
-				}
-			}
-			if result != runnerv1.Result_RESULT_UNSPECIFIED {
-				step.Status = StatusFromResult(result)
-				step.Stopped = util.IfZero(step.Stopped, now)
-			} else if step.Started != 0 {
-				step.Status = StatusRunning
-			}
+			step.applyState(stepStates[step.Index], now)
 			if _, err := db.GetEngine(ctx).ID(step.ID).Update(step); err != nil {
 				return err
 			}
@@ -637,13 +635,7 @@ func StopTask(ctx context.Context, taskID int64, status Status) error {
 	}
 
 	for _, step := range task.Steps {
-		if !step.Status.IsDone() {
-			step.Status = status
-			if step.Started == 0 {
-				step.Started = now
-			}
-			step.Stopped = now
-		}
+		step.stop(status, now)
 		if _, err := e.ID(step.ID).Update(step); err != nil {
 			return err
 		}

models/actions/task_step.go

@@ -5,10 +5,14 @@ package actions
 
 import (
 	"context"
+	"fmt"
 	"time"
 
+	runnerv1 "gitea.dev/actionslib/runner/v1"
 	"gitea.dev/models/db"
+	"gitea.dev/modules/container"
 	"gitea.dev/modules/timeutil"
+	"gitea.dev/modules/util"
 )
 
 // ActionTaskStep represents a step of ActionTask
@@ -21,6 +25,8 @@ type ActionTaskStep struct {
 	Status    Status `xorm:"index"`
 	LogIndex  int64
 	LogLength int64
+	Stage     int64 `xorm:"NOT NULL DEFAULT 0"` // a runnerv1.StepStage, unspecified for the steps parsed from the workflow file instead of reported by the runner
+	Number    int64 `xorm:"NOT NULL DEFAULT 0"` // the index in the workflow file of the step a PRE, MAIN or POST stage belongs to
 	Started   timeutil.TimeStamp
 	Stopped   timeutil.TimeStamp
 	Created   timeutil.TimeStamp `xorm:"created"`
@@ -31,6 +37,27 @@ func (step *ActionTaskStep) Duration() time.Duration {
 	return calculateDuration(step.Started, step.Stopped, step.Status, step.Updated)
 }
 
+// StepStage returns the stored stage, which can be one added to the enum after this Gitea version
+func (step *ActionTaskStep) StepStage() runnerv1.StepStage {
+	return runnerv1.StepStage(step.Stage)
+}
+
+// IsReported reports whether the runner reported the step instead of Gitea parsing it from the workflow file.
+func (step *ActionTaskStep) IsReported() bool {
+	return step.StepStage() != runnerv1.StepStage_STEP_STAGE_UNSPECIFIED
+}
+
+// WorkflowStepIndex returns the index of the step in the workflow file, or -1 for a step that is not written there.
+func (step *ActionTaskStep) WorkflowStepIndex() int64 {
+	switch step.StepStage() {
+	case runnerv1.StepStage_STEP_STAGE_UNSPECIFIED:
+		return step.Index
+	case runnerv1.StepStage_STEP_STAGE_MAIN:
+		return step.Number
+	}
+	return -1
+}
+
 func init() {
 	db.RegisterModel(new(ActionTaskStep))
 }
@@ -39,3 +66,120 @@ func GetTaskStepsByTaskID(ctx context.Context, taskID int64) ([]*ActionTaskStep,
 	var steps []*ActionTaskStep
 	return steps, db.GetEngine(ctx).Where("task_id=?", taskID).OrderBy("`index` ASC").Find(&steps)
 }
+
+// applyState applies the state the runner reported for the step, nil meaning it reported none.
+func (step *ActionTaskStep) applyState(state *runnerv1.StepState, now timeutil.TimeStamp) {
+	var result runnerv1.Result
+	if state != nil {
+		result = state.Result
+		step.LogIndex = state.LogIndex
+		step.LogLength = state.LogLength
+		if step.Started == 0 && state.StartedAt != nil {
+			step.Started = now
+		}
+	}
+	if result != runnerv1.Result_RESULT_UNSPECIFIED {
+		step.Status = StatusFromResult(result)
+		step.Stopped = util.IfZero(step.Stopped, now)
+	} else if step.Started != 0 {
+		step.Status = StatusRunning
+	}
+}
+
+// stop finishes a step that had not finished when its task stopped
+func (step *ActionTaskStep) stop(status Status, now timeutil.TimeStamp) {
+	if step.Status.IsDone() {
+		return
+	}
+	step.Status = status
+	step.Started = util.IfZero(step.Started, now)
+	step.Stopped = now
+}
+
+// maxReportedSteps bounds the steps a runner can make Gitea store for one task.
+const maxReportedSteps = 1000
+
+type reportedStepKey struct {
+	stage  runnerv1.StepStage
+	number int64
+}
+
+func validateReportedSteps(steps []*runnerv1.StepState) error {
+	if len(steps) > maxReportedSteps {
+		return fmt.Errorf("%d steps exceed the limit of %d", len(steps), maxReportedSteps)
+	}
+	seen := make(container.Set[reportedStepKey], len(steps))
+	for i, v := range steps {
+		if v.Stage <= runnerv1.StepStage_STEP_STAGE_UNSPECIFIED { // a stage added later is shown by its name
+			return fmt.Errorf("step %d has invalid stage %d", i, v.Stage)
+		}
+		if v.Id != int64(i) || v.Number < 0 || v.LogIndex < 0 || v.LogLength < 0 {
+			return fmt.Errorf("step %d has invalid id %d, number %d or log range %d+%d", i, v.Id, v.Number, v.LogIndex, v.LogLength)
+		}
+		if !seen.Add(reportedStepKey{v.Stage, v.Number}) {
+			return fmt.Errorf("step %d duplicates stage %s of number %d", i, v.Stage, v.Number)
+		}
+	}
+	return nil
+}
+
+// updateReportedSteps replaces the steps of the task with the ones the runner reported.
+// A step keeps its times and status when the runner reports it again at another position.
+func updateReportedSteps(ctx context.Context, task *ActionTask, reported []*runnerv1.StepState, now timeutil.TimeStamp) error {
+	e := db.GetEngine(ctx)
+	sameLayout := len(task.Steps) == len(reported)
+	for i := 0; sameLayout && i < len(reported); i++ {
+		step := task.Steps[i]
+		sameLayout = step.IsReported() && reportedStepKey{step.StepStage(), step.Number} == reportedStepKey{reported[i].Stage, reported[i].Number}
+	}
+
+	if sameLayout {
+		for i, step := range task.Steps {
+			step.Name = util.EllipsisDisplayString(reported[i].Name, 255)
+			step.applyState(reported[i], now)
+			if task.Status.IsDone() {
+				step.stop(task.Status, now)
+			}
+			if _, err := e.ID(step.ID).Update(step); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	known := make(map[reportedStepKey]*ActionTaskStep, len(task.Steps))
+	for _, step := range task.Steps {
+		if step.IsReported() {
+			known[reportedStepKey{step.StepStage(), step.Number}] = step
+		}
+	}
+	steps := make([]*ActionTaskStep, len(reported))
+	for i, v := range reported {
+		key := reportedStepKey{v.Stage, v.Number}
+		step := &ActionTaskStep{
+			Name:   util.EllipsisDisplayString(v.Name, 255),
+			TaskID: task.ID,
+			Index:  int64(i),
+			RepoID: task.RepoID,
+			Stage:  int64(key.stage),
+			Number: key.number,
+			Status: StatusWaiting,
+		}
+		if old, ok := known[key]; ok {
+			step.Status, step.Started, step.Stopped = old.Status, old.Started, old.Stopped
+		}
+		step.applyState(v, now)
+		if task.Status.IsDone() {
+			step.stop(task.Status, now)
+		}
+		steps[i] = step
+	}
+	if _, err := e.Delete(&ActionTaskStep{TaskID: task.ID}); err != nil {
+		return err
+	}
+	if _, err := e.Insert(steps); err != nil {
+		return err
+	}
+	task.Steps = steps
+	return nil
+}

models/actions/task_test.go

@@ -407,6 +407,114 @@ func TestUpdateTaskByStateIsAtomic(t *testing.T) {
 	assert.GreaterOrEqual(t, unittest.AssertExistsAndLoadBean(t, &ActionTaskStep{TaskID: task.ID}).Started, before)
 }
 
+func TestUpdateTaskByStateReportedSteps(t *testing.T) {
+	require.NoError(t, unittest.PrepareTestDatabase())
+	task, _ := newRunningTaskForCancelling(t, "reported-steps-job", true)
+	require.NoError(t, db.Insert(t.Context(), &ActionTaskStep{Name: "Run echo", TaskID: task.ID, RepoID: task.RepoID, Status: StatusWaiting}))
+
+	report := func(t *testing.T, steps []*runnerv1.StepState, reported ...*runnerv1.StepState) []*ActionTaskStep {
+		t.Helper()
+		_, err := UpdateTaskByState(t.Context(), task.RunnerID, &runnerv1.TaskState{Id: task.ID, Steps: steps, ReportedSteps: reported})
+		require.NoError(t, err)
+		got, err := GetTaskStepsByTaskID(t.Context(), task.ID)
+		require.NoError(t, err)
+		return got
+	}
+	started := timestamppb.Now()
+	setup := &runnerv1.StepState{Id: 0, Name: "Set up job", Stage: runnerv1.StepStage_STEP_STAGE_SETUP, StartedAt: started}
+	main := &runnerv1.StepState{Id: 1, Name: "Run echo", Stage: runnerv1.StepStage_STEP_STAGE_MAIN}
+	legacy := []*runnerv1.StepState{{Id: 0, StartedAt: started, LogLength: 3}}
+
+	t.Run("invalid reported steps fall back to the workflow steps", func(t *testing.T) {
+		got := report(t, legacy, setup, &runnerv1.StepState{Id: 1, Stage: runnerv1.StepStage_STEP_STAGE_SETUP})
+		require.Len(t, got, 1)
+		assert.False(t, got[0].IsReported())
+		assert.Equal(t, StatusRunning, got[0].Status)
+		assert.EqualValues(t, 3, got[0].LogLength)
+	})
+
+	t.Run("reported steps replace the workflow steps", func(t *testing.T) {
+		got := report(t, legacy, setup, main)
+		require.Len(t, got, 2)
+		assert.Equal(t, runnerv1.StepStage_STEP_STAGE_SETUP, got[0].StepStage())
+		assert.EqualValues(t, -1, got[0].WorkflowStepIndex())
+		assert.Equal(t, StatusRunning, got[0].Status)
+		assert.NotZero(t, got[0].Started)
+		assert.Equal(t, "Run echo", got[1].Name)
+		assert.Equal(t, StatusWaiting, got[1].Status)
+		assert.EqualValues(t, 0, got[1].WorkflowStepIndex())
+	})
+
+	t.Run("an inserted step keeps the state of the others", func(t *testing.T) {
+		before, err := GetTaskStepsByTaskID(t.Context(), task.ID)
+		require.NoError(t, err)
+		doneSetup := &runnerv1.StepState{Id: 0, Name: "Set up job", Stage: runnerv1.StepStage_STEP_STAGE_SETUP, StartedAt: started, Result: runnerv1.Result_RESULT_SUCCESS, LogLength: 4}
+		pre := &runnerv1.StepState{Id: 1, Name: "Pre Run checkout", Stage: runnerv1.StepStage_STEP_STAGE_PRE, StartedAt: started, LogIndex: 4, LogLength: 2}
+		movedMain := &runnerv1.StepState{Id: 2, Name: "Run echo", Stage: runnerv1.StepStage_STEP_STAGE_MAIN}
+		got := report(t, legacy, doneSetup, pre, movedMain)
+		require.Len(t, got, 3)
+		assert.Equal(t, StatusSuccess, got[0].Status)
+		assert.Equal(t, before[0].Started, got[0].Started)
+		assert.Equal(t, runnerv1.StepStage_STEP_STAGE_PRE, got[1].StepStage())
+		assert.Equal(t, StatusRunning, got[1].Status)
+		assert.EqualValues(t, 4, got[1].LogIndex)
+		assert.EqualValues(t, 2, got[2].Index)
+	})
+
+	t.Run("workflow steps no longer apply once reported", func(t *testing.T) {
+		before, err := GetTaskStepsByTaskID(t.Context(), task.ID)
+		require.NoError(t, err)
+		got := report(t, []*runnerv1.StepState{{Id: 0, Result: runnerv1.Result_RESULT_FAILURE}})
+		assert.Equal(t, before, got)
+	})
+
+	t.Run("unfinished steps stop with the task", func(t *testing.T) {
+		doneSetup := &runnerv1.StepState{Id: 0, Name: "Set up job", Stage: runnerv1.StepStage_STEP_STAGE_SETUP, StartedAt: started, Result: runnerv1.Result_RESULT_SUCCESS}
+		runningMain := &runnerv1.StepState{Id: 1, Name: "Run echo", Stage: runnerv1.StepStage_STEP_STAGE_MAIN, StartedAt: started}
+		cleanup := &runnerv1.StepState{Id: 2, Name: "Complete job", Stage: runnerv1.StepStage_STEP_STAGE_CLEANUP}
+		_, err := UpdateTaskByState(t.Context(), task.RunnerID, &runnerv1.TaskState{
+			Id: task.ID, Result: runnerv1.Result_RESULT_FAILURE,
+			ReportedSteps: []*runnerv1.StepState{doneSetup, runningMain, cleanup},
+		})
+		require.NoError(t, err)
+		got, err := GetTaskStepsByTaskID(t.Context(), task.ID)
+		require.NoError(t, err)
+		require.Len(t, got, 3)
+		assert.Equal(t, StatusSuccess, got[0].Status)
+		for _, step := range got[1:] {
+			assert.Equal(t, StatusFailure, step.Status, step.Name)
+			assert.NotZero(t, step.Started, step.Name)
+			assert.NotZero(t, step.Stopped, step.Name)
+		}
+	})
+}
+
+func TestValidateReportedSteps(t *testing.T) {
+	step := func(id int64, stage runnerv1.StepStage, number int64) *runnerv1.StepState {
+		return &runnerv1.StepState{Id: id, Stage: stage, Number: number}
+	}
+	assert.NoError(t, validateReportedSteps([]*runnerv1.StepState{
+		step(0, runnerv1.StepStage_STEP_STAGE_SETUP, 0),
+		step(1, runnerv1.StepStage_STEP_STAGE_PRE, 0),
+		step(2, runnerv1.StepStage_STEP_STAGE_MAIN, 0),
+		step(3, runnerv1.StepStage_STEP_STAGE_MAIN, 1),
+		step(4, runnerv1.StepStage_STEP_STAGE_CLEANUP, 0),
+		step(5, runnerv1.StepStage(100), 0), // a stage added later, with several steps told apart by number
+		step(6, runnerv1.StepStage(100), 1),
+	}))
+	for name, steps := range map[string][]*runnerv1.StepState{
+		"unspecified stage": {step(0, runnerv1.StepStage_STEP_STAGE_UNSPECIFIED, 0)},
+		"negative stage":    {step(0, runnerv1.StepStage(-1), 0)},
+		"id not position":   {step(1, runnerv1.StepStage_STEP_STAGE_SETUP, 0)},
+		"negative number":   {step(0, runnerv1.StepStage_STEP_STAGE_MAIN, -1)},
+		"duplicate main":    {step(0, runnerv1.StepStage_STEP_STAGE_MAIN, 0), step(1, runnerv1.StepStage_STEP_STAGE_MAIN, 0)},
+		"duplicate setup":   {step(0, runnerv1.StepStage_STEP_STAGE_SETUP, 0), step(1, runnerv1.StepStage_STEP_STAGE_SETUP, 0)},
+		"too many":          make([]*runnerv1.StepState, maxReportedSteps+1),
+	} {
+		assert.Error(t, validateReportedSteps(steps), name)
+	}
+}
+
 // newRunningTaskForCancelling inserts a running run/job/task assigned to a fresh runner,
 // which is the state every cancellation test starts from.
 func newRunningTaskForCancelling(t *testing.T, name string, hasCancellingSupport bool) (*ActionTask, *ActionRunJob) {