Method executeTaskRun

backend/runner/taskrun/running_scheduler.go:71–116

executeTaskRun executes a task run that is already in RUNNING status.

(ctx context.Context, projectID string, taskRunUID, taskUID int64)

// executeTaskRun executes a task run that is already in RUNNING status.
func (s *Scheduler) executeTaskRun(ctx context.Context, projectID string, taskRunUID, taskUID int64) error {
	task, err := s.store.GetTaskByID(ctx, projectID, taskUID)
	if err != nil {
		return errors.Wrapf(err, "failed to get task")
	}
	if task == nil {
		return errors.Errorf("task %v not found", taskUID)
	}

	// Validate task freshness before execution.
	if err := s.validateTaskFreshness(ctx, task); err != nil {
		slog.WarnContext(ctx, "task run blocked by drift validation", log.BBError(err))
		taskRunStatusPatch := &store.TaskRunStatusPatch{
			ID:        taskRunUID,
			ProjectID: task.ProjectID,
			Updater:   "",
			Status:    storepb.TaskRun_FAILED,
			AllowedStatuses: []storepb.TaskRun_Status{
				storepb.TaskRun_RUNNING,
			},
			ResultProto: &storepb.TaskRunResult{
				Detail: err.Error(),
			},
		}
		if _, patchErr := s.store.UpdateTaskRunStatus(ctx, taskRunStatusPatch); patchErr != nil {
			if common.ErrorCode(patchErr) == common.Conflict {
				return nil
			}
			return errors.Wrapf(patchErr, "failed to mark task run as failed after drift detection")
		}
		return nil
	}

	executor, ok := s.executorMap[task.Type]
	if !ok {
		return errors.Errorf("executor not found for task type: %v", task.Type)
	}

	// Update started_at
	if err := s.store.UpdateTaskRunStartAt(ctx, task.ProjectID, taskRunUID); err != nil {
		return errors.Wrapf(err, "failed to update task run start at")
	}

	go s.runTaskRunOnce(ctx, taskRunUID, task, executor)
	return nil
}

func (s *Scheduler) runTaskRunOnce(ctx context.Context, taskRunUID int64, task *store.TaskMessage, executor Executor) {
	ctx = taskRunLogContext(ctx, task.ProjectID, taskRunUID)