diff --git a/models/actions/task.go b/models/actions/task.go index 9f22e560ae3..a34769bb8ba 100644 --- a/models/actions/task.go +++ b/models/actions/task.go @@ -487,7 +487,7 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task return nil, err } task := &ActionTask{} - err = globallock.LockAndDo(ctx, fmt.Sprintf("UpdateTaskByState-run-%d", runID), func(ctx context.Context) error { + applyState := func(ctx context.Context) error { if has, err := db.GetEngine(ctx).ID(taskID).Get(task); err != nil { return err } else if !has { @@ -552,6 +552,10 @@ func UpdateTaskByState(ctx context.Context, runnerID int64, state *runnerv1.Task } } return nil + } + err = globallock.LockAndDo(ctx, fmt.Sprintf("UpdateTaskByState-run-%d", runID), func(ctx context.Context) error { + // A half-written report leaves the task done with a running job, which no retry repairs. + return db.WithTx(ctx, applyState) }) return task, err } diff --git a/models/actions/task_test.go b/models/actions/task_test.go index 139de0c9ec3..a48dbf0ebe0 100644 --- a/models/actions/task_test.go +++ b/models/actions/task_test.go @@ -4,7 +4,10 @@ package actions import ( + "context" + "errors" "strings" + "sync/atomic" "testing" runnerv1 "gitea.dev/actions-proto-go/runner/v1" @@ -17,6 +20,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" "google.golang.org/protobuf/types/known/timestamppb" + "xorm.io/xorm/contexts" ) func TestActionTask_GetRunJobLink(t *testing.T) { @@ -357,6 +361,35 @@ func TestCreateTaskForRunnerPagination(t *testing.T) { assert.Equal(t, task.ID, claimed.TaskID) } +type failFirstStepWrite struct{ fired atomic.Bool } + +func (h *failFirstStepWrite) BeforeProcess(c *contexts.ContextHook) (context.Context, error) { + if !h.fired.Load() && strings.HasPrefix(c.SQL, "UPDATE") && strings.Contains(c.SQL, "action_task_step") { + h.fired.Store(true) + return nil, errors.New("interrupted") + } + return c.Ctx, nil +} + +func (*failFirstStepWrite) AfterProcess(*contexts.ContextHook) error { return nil } + +// TestUpdateTaskByStateIsAtomic checks that an interrupted report writes nothing: a surviving task or +// job write would hit the "state is final" early return, which no retry or cleanup repairs. +func TestUpdateTaskByStateIsAtomic(t *testing.T) { + require.NoError(t, unittest.PrepareTestDatabase()) + task, job := newRunningTaskForCancelling(t, "atomic-report-job", true) + require.NoError(t, db.Insert(t.Context(), &ActionTaskStep{TaskID: task.ID, RepoID: task.RepoID, Status: StatusRunning})) + unittest.GetXORMEngine().AddHook(&failFirstStepWrite{}) + finalState := &runnerv1.TaskState{Id: task.ID, Result: runnerv1.Result_RESULT_SUCCESS, StoppedAt: timestamppb.Now()} + _, err := UpdateTaskByState(t.Context(), task.RunnerID, finalState) + require.Error(t, err) + assert.Equal(t, StatusRunning, unittest.AssertExistsAndLoadBean(t, &ActionTask{ID: task.ID}).Status) + assert.Equal(t, StatusRunning, unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID}).Status) + _, err = UpdateTaskByState(t.Context(), task.RunnerID, finalState) + require.NoError(t, err) + assert.Equal(t, StatusSuccess, unittest.AssertExistsAndLoadBean(t, &ActionRunJob{ID: job.ID}).Status) +} + // 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) {