pipeline

func AnnotateAutoscalerProvisioning ¶ added in v5.6.0

func AnnotateAutoscalerProvisioning(s store.Store, pipeline *model.Pipeline)

AnnotateAutoscalerProvisioning sets AutoscalerProvisioning on pending workflows when a healthy autoscaler with pending agents can match the workflow's task labels.

func Approve ¶

func Approve(ctx context.Context, store store.Store, currentPipeline *model.Pipeline, user *model.User, repo *model.Repo) (*model.Pipeline, error)

Approve update the status to pending for a blocked pipeline so it can be executed.

func ApproveWorkflow ¶ added in v5.13.0

func ApproveWorkflow(
	ctx context.Context,
	s store.Store,
	q queue.Queue,
	user *model.User,
	workflowID int64,
	reason string,
) error

ApproveWorkflow transitions a pending approval gate to Approved, records an audit row, and notifies the queue so the workflow can be dispatched.

func BlockOnApproval ¶ added in v5.13.0

func BlockOnApproval(
	ctx context.Context,
	s store.Store,
	q queue.Queue,
	workflowID int64,
	defaultTimeout time.Duration,
) error

BlockOnApproval transitions a workflow into "blocked awaiting approval". Called by the queue when a gated workflow's upstream deps have all cleared. The deadline is fixed here (now + effective timeout), not at queue time, so long-running upstream workflows do not consume the gate's clock.

func Cancel ¶

func Cancel(ctx context.Context, _forge forge.Forge, store store.Store, repo *model.Repo, user *model.User, pipeline *model.Pipeline) error

Cancel the pipeline and returns the status.

func CancelWorkflow ¶

func CancelWorkflow(ctx context.Context, _forge forge.Forge, store store.Store, repo *model.Repo, user *model.User, pipeline *model.Pipeline, workflowID int64) error

CancelWorkflow cancels a specific workflow and returns the status.

Note: We intentionally do NOT gate on the parent pipeline's status here. A workflow can legitimately be in pending/running while the pipeline is in a terminal state — the typical case is an autoscaler-provisioned agent that dies before acking a queue.Error from a previous Cancel(), leaving the workflow as a "ghost" in the DB. The UI keeps showing the cancel button (it's gated on workflow state, not pipeline state), and the user must be able to clean up that ghost. The actual cancel-eligibility check happens against the target workflow's state below.

func CascadeSkipBlockedWorkflows ¶ added in v5.10.4

func CascadeSkipBlockedWorkflows(ctx context.Context, _store store.Store, q queue.Queue, pipeline *model.Pipeline) error

CascadeSkipBlockedWorkflows synchronously marks pending workflows as skipped when their dependencies have completed in a way that prevents them from ever running (failure / killed / error / skipped, given the workflow's run_on semantics).

Without this, blocked pending workflows would linger in the DB as Pending until an agent eventually polls them and the queue's lazy ShouldRun() evaluation discards them. That delay leaves the parent pipeline stuck in "running" state — sometimes indefinitely on idle agents — even though no further work can ever execute.

The cascade is iterative so transitive chains (A fails → B skipped → C skipped) collapse in a single call. Each iteration:

1. Re-reads the workflow tree from the store.
2. For every pending workflow, looks up its task to honor run_on rules.
3. Marks the workflow + its pending children as skipped and evicts the queue entry.

Errors are logged per workflow but do not abort the cascade: it's better to skip as many blocked workflows as possible than to leave the pipeline stuck because one update failed.

func Create ¶

func Create(ctx context.Context, _store store.Store, repo *model.Repo, pipeline *model.Pipeline, opts ...CreateOptions) (*model.Pipeline, error)

Create a new pipeline and start it.

func Decline ¶

func Decline(ctx context.Context, store store.Store, pipeline *model.Pipeline, user *model.User, repo *model.Repo) (*model.Pipeline, error)

Decline updates the status to declined for blocked pipelines.

func DeclineWorkflow ¶ added in v5.13.0

func DeclineWorkflow(
	ctx context.Context,
	s store.Store,
	q queue.Queue,
	user *model.User,
	workflowID int64,
	reason string,
) error

DeclineWorkflow transitions a pending approval gate to Declined, records an audit row, and notifies the queue. Downstream workflows that depend on this one are cascade-skipped (see cascadeSkipFromDeclined, Task 23).

func ExpireWorkflow ¶ added in v5.13.0

func ExpireWorkflow(
	ctx context.Context,
	s store.Store,
	q queue.Queue,
	workflowID int64,
) error

ExpireWorkflow is invoked by the timeout sweeper when an approval deadline passes without a decision. UserID is 0 to mark the action as system-driven.

func RerunStep ¶ added in v5.9.0

func RerunStep(ctx context.Context, _forge forge.Forge, _store store.Store, currentPipeline *model.Pipeline, workflowID, stepID int64, user *model.User, repo *model.Repo) error

RerunStep reruns a specific step (and its dependents) within an existing pipeline. It creates a new workflow attempt with a filtered config containing only the target step, its dependent steps, and any clone/service steps needed for the workspace.

func RerunWorkflow ¶ added in v5.5.0

func RerunWorkflow(ctx context.Context, _forge forge.Forge, _store store.Store, currentPipeline *model.Pipeline, workflowID int64, user *model.User, repo *model.Repo) error

RerunWorkflow reruns a specific workflow within an existing pipeline. It creates new workflow+step records (preserving the original) and queues them for execution. If the rerun'd workflow has dependents that failed/were skipped, those are also rerun.

func Restart ¶

func Restart(ctx context.Context, store store.Store, lastPipeline *model.Pipeline, user *model.User, repo *model.Repo, envs map[string]string) (*model.Pipeline, error)

Restart a pipeline by creating a new one out of the old and start it.

func UpdateStatusToDone ¶

func UpdateStatusToDone(store store.Store, pipeline model.Pipeline, status model.StatusValue, stopped int64) (*model.Pipeline, error)

func UpdateStepStatus ¶

func UpdateStepStatus(store store.Store, step *model.Step, state rpc.StepState) error

func UpdateStepStatusToDone ¶

func UpdateStepStatusToDone(store store.Store, step model.Step, state rpc.StepState) (*model.Step, error)

func UpdateStepToStatusKilled ¶

func UpdateStepToStatusKilled(store store.Store, step model.Step) (*model.Step, error)

func UpdateStepToStatusSkipped ¶

func UpdateStepToStatusSkipped(store store.Store, step model.Step, finished int64) (*model.Step, error)

func UpdateStepToStatusStarted ¶

func UpdateStepToStatusStarted(store store.Store, step model.Step, state rpc.StepState) (*model.Step, error)

func UpdateToStatusDeclined ¶

func UpdateToStatusDeclined(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error)

func UpdateToStatusError ¶

func UpdateToStatusError(store store.Store, pipeline model.Pipeline, err error) (*model.Pipeline, error)

func UpdateToStatusKilled ¶

func UpdateToStatusKilled(store store.Store, pipeline model.Pipeline) (*model.Pipeline, error)

func UpdateToStatusPending ¶

func UpdateToStatusPending(store store.Store, pipeline model.Pipeline, reviewer string) (*model.Pipeline, error)

func UpdateToStatusRunning ¶

func UpdateToStatusRunning(store store.Store, pipeline model.Pipeline, started int64) (*model.Pipeline, error)

func UpdateWorkflowStatusToDone ¶

func UpdateWorkflowStatusToDone(store store.Store, workflow model.Workflow, state rpc.WorkflowState) (*model.Workflow, error)

func UpdateWorkflowStatusToRunning ¶

func UpdateWorkflowStatusToRunning(store store.Store, workflow model.Workflow, state rpc.WorkflowState) (*model.Workflow, error)

func UpdateWorkflowToStatusKilled ¶

func UpdateWorkflowToStatusKilled(store store.Store, workflow model.Workflow) (*model.Workflow, error)

func UpdateWorkflowToStatusSkipped ¶

func UpdateWorkflowToStatusSkipped(store store.Store, workflow model.Workflow) (*model.Workflow, error)