d3d11f4
use std::collections::HashMap;
use ci_protocol::runner::{JobFinishState, StepReportState};
use models::ci::workflow_job::{self, WorkflowJobState};
use models::ci::workflow_run::{self, WorkflowRunState};
use models::ci::workflow_step::{self, WorkflowStepState};
use sea_orm::sea_query::Expr;
use sea_orm::{
ActiveModelTrait, ColumnTrait, DatabaseBackend, EntityTrait, FromQueryResult, PaginatorTrait,
QueryFilter, QueryOrder, QuerySelect, Set, Statement,
};
use time::OffsetDateTime;
use uuid::Uuid;
use crate::errors::{CiError, CiResult};
use crate::service::CiService;
#[derive(Debug)]
pub struct GroupRow {
pub repo_id: Uuid,
pub concurrency_group: String,
}
impl FromQueryResult for GroupRow {
fn from_query_result(res: &sea_orm::QueryResult, _pre: &str) -> Result<Self, sea_orm::DbErr> {
Ok(Self {
repo_id: res.try_get("", "repo_id")?,
concurrency_group: res.try_get("", "concurrency_group")?,
})
}
}
trait LabelsExt {
fn contains_all(&self, wanted: &[String]) -> bool;
}
impl LabelsExt for Vec<String> {
fn contains_all(&self, wanted: &[String]) -> bool {
wanted
.iter()
.all(|label| self.iter().any(|have| have == label))
}
}
impl CiService {
pub async fn advance_jobs(&self) -> CiResult<u64> {
let runs = workflow_run::Entity::find()
.filter(
workflow_run::Column::State
.is_in([WorkflowRunState::Queued, WorkflowRunState::Running]),
)
.all(self.db_read())
.await?;
let mut moved = 0u64;
for run in runs {
moved += if ci_protocol::actions::from_snapshot(&run.definition).is_some() {
self.advance_actions_run(&run).await?
} else {
self.advance_run(run.id).await?
};
}
Ok(moved)
}
async fn advance_run(&self, run_id: Uuid) -> CiResult<u64> {
let jobs = workflow_job::Entity::find()
.filter(workflow_job::Column::RunId.eq(run_id))
.all(self.db_read())
.await?;
let by_key: HashMap<String, WorkflowJobState> = jobs
.iter()
.map(|job| (job.job_key.clone(), job.state))
.collect();
let mut moved = 0u64;
let now = OffsetDateTime::now_utc();
for job in &jobs {
if job.state != WorkflowJobState::Pending {
continue;
}
let needs: Vec<String> = serde_json::from_value(job.needs.clone()).unwrap_or_default();
if needs.is_empty() {
continue;
}
let mut ready = true;
let mut blocked = false;
for need in &needs {
match by_key.get(need) {
Some(WorkflowJobState::Success) => {}
Some(
WorkflowJobState::Failure
| WorkflowJobState::TimedOut
| WorkflowJobState::Cancelled
| WorkflowJobState::Skipped,
) => blocked = true,
_ => ready = false,
}
}
if blocked {
let updated = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Skipped),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job.id))
.filter(workflow_job::Column::State.eq(WorkflowJobState::Pending))
.exec(self.db_write())
.await?;
if updated.rows_affected > 0 {
moved += 1;
self.skip_steps(job.id).await?;
}
continue;
}
if !ready {
continue;
}
let updated = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Queued),
)
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job.id))
.filter(workflow_job::Column::State.eq(WorkflowJobState::Pending))
.exec(self.db_write())
.await?;
if updated.rows_affected > 0 {
moved += 1;
}
}
Ok(moved)
}
async fn skip_steps(&self, job_id: Uuid) -> CiResult<()> {
workflow_step::Entity::update_many()
.col_expr(
workflow_step::Column::State,
Expr::value(WorkflowStepState::Skipped),
)
.filter(workflow_step::Column::JobId.eq(job_id))
.filter(workflow_step::Column::State.eq(WorkflowStepState::Pending))
.exec(self.db_write())
.await?;
Ok(())
}
pub async fn enforce_concurrency(&self) -> CiResult<u64> {
let actions_changed = self.enforce_actions_concurrency().await?;
let groups: Vec<GroupRow> = GroupRow::find_by_statement(Statement::from_string(
DatabaseBackend::Postgres,
"SELECT DISTINCT repo_id, concurrency_group FROM workflow_run \
WHERE concurrency_group IS NOT NULL \
AND state IN (0, 1, 4)"
.to_string(),
))
.all(self.db_read())
.await?;
let mut affected = actions_changed;
for group in groups {
let runs = workflow_run::Entity::find()
.filter(workflow_run::Column::RepoId.eq(group.repo_id))
.filter(workflow_run::Column::ConcurrencyGroup.eq(group.concurrency_group.clone()))
.filter(workflow_run::Column::State.is_in([
WorkflowRunState::Queued,
WorkflowRunState::Running,
WorkflowRunState::CancelRequested,
]))
.order_by_asc(workflow_run::Column::CreatedAt)
.all(self.db_read())
.await?;
if runs.len() < 2 {
continue;
}
if runs
.iter()
.any(|run| ci_protocol::actions::from_snapshot(&run.definition).is_some())
{
continue;
}
let oldest = &runs[0];
for run in runs.iter().skip(1) {
let cancel_in_progress = run
.definition
.get("concurrency")
.and_then(|c| c.get("cancel_in_progress"))
.and_then(|v| v.as_bool())
.unwrap_or(false);
if !cancel_in_progress {
continue;
}
if matches!(
oldest.state,
WorkflowRunState::Running | WorkflowRunState::CancelRequested
) {
let now = OffsetDateTime::now_utc();
let updated = workflow_run::Entity::update_many()
.col_expr(
workflow_run::Column::State,
Expr::value(WorkflowRunState::CancelRequested),
)
.col_expr(workflow_run::Column::UpdatedAt, Expr::value(now))
.filter(workflow_run::Column::Id.eq(oldest.id))
.filter(workflow_run::Column::State.eq(WorkflowRunState::Running))
.exec(self.db_write())
.await?;
if updated.rows_affected > 0 {
affected += 1;
self.record_event(
oldest.id,
"workflow_run",
serde_json::json!({
"action": "cancel_requested",
"reason": "concurrency",
}),
)
.await;
}
}
break;
}
}
Ok(affected)
}
pub async fn claim_job(&self, runner_id: Uuid, labels: &[String]) -> CiResult<Option<Uuid>> {
let runner = models::ci::runner::Entity::find_by_id(runner_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if runner.state != models::ci::runner::RunnerState::Active {
return Err(CiError::Forbidden);
}
let active = workflow_job::Entity::find()
.filter(workflow_job::Column::RunnerId.eq(runner_id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.count(self.db_read())
.await? as i32;
if active >= runner.capacity.max(1) {
return Ok(None);
}
let jobs = workflow_job::Entity::find()
.filter(workflow_job::Column::State.eq(WorkflowJobState::Queued))
.order_by_asc(workflow_job::Column::CreatedAt)
.limit(200)
.all(self.db_read())
.await?;
let now = OffsetDateTime::now_utc();
for job in jobs {
if job.is_template {
continue;
}
if job.actions_context.is_some()
&& (runner.protocol_version.unwrap_or(1) < 2
|| !runner
.capabilities
.as_ref()
.and_then(|v| v.as_array())
.is_some_and(|v| v.iter().any(|c| c.as_str() == Some("actions-v2"))))
{
continue;
}
let job_labels: Vec<String> =
serde_json::from_value(job.runs_on.clone()).unwrap_or_default();
if !labels.to_vec().contains_all(&job_labels) {
continue;
}
if let Some(space_id) = runner.space_id
&& job.space_id != space_id
{
continue;
}
if job.actions_context.is_some() {
if self.claim_actions_job(&runner, &job, labels).await? {
self.mark_run_running(job.run_id, runner_id).await?;
self.start_deployment(&job).await?;
return Ok(Some(job.id));
}
continue;
}
if job.actions_context.is_none()
&& let Some(environment) = &job.environment
&& !self
.runner_matches_environment(job.space_id, environment, labels)
.await?
{
continue;
}
if job.actions_context.is_none()
&& !self
.environment_available(job.space_id, job.environment.as_ref())
.await?
{
continue;
}
let lease_token = crate::secrets::random_secret(24);
let lease_expires = now + time::Duration::seconds(self.config().ci.lease_secs as i64);
let updated = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Claimed),
)
.col_expr(workflow_job::Column::RunnerId, Expr::value(runner_id))
.col_expr(
workflow_job::Column::LeaseToken,
Expr::value(lease_token.clone()),
)
.col_expr(
workflow_job::Column::LeaseExpiresAt,
Expr::value(Some(lease_expires)),
)
.col_expr(workflow_job::Column::Attempt, Expr::value(job.attempt + 1))
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job.id))
.filter(workflow_job::Column::State.eq(WorkflowJobState::Queued))
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
continue;
}
self.mark_run_running(job.run_id, runner_id).await?;
self.start_deployment(&job).await?;
tracing::info!(
job_id = %job.id,
run_id = %job.run_id,
runner_id = %runner_id,
attempt = job.attempt + 1,
"job claimed"
);
return Ok(Some(job.id));
}
Ok(None)
}
async fn runner_matches_environment(
&self,
space_id: Uuid,
environment: &str,
labels: &[String],
) -> CiResult<bool> {
let row = models::ci::ci_environment::Entity::find()
.filter(models::ci::ci_environment::Column::SpaceId.eq(space_id))
.filter(models::ci::ci_environment::Column::Name.eq(environment.to_string()))
.one(self.db_read())
.await?;
let Some(row) = row else {
return Ok(true);
};
let required: Vec<String> =
serde_json::from_value(row.required_labels.clone()).unwrap_or_default();
Ok(labels.to_vec().contains_all(&required))
}
async fn environment_available(
&self,
space_id: Uuid,
environment: Option<&String>,
) -> CiResult<bool> {
let Some(environment) = environment else {
return Ok(true);
};
let row = models::ci::ci_environment::Entity::find()
.filter(models::ci::ci_environment::Column::SpaceId.eq(space_id))
.filter(models::ci::ci_environment::Column::Name.eq(environment.to_string()))
.one(self.db_read())
.await?;
let Some(row) = row else {
return Ok(true);
};
let active = models::ci::ci_deployment::Entity::find()
.filter(models::ci::ci_deployment::Column::EnvironmentId.eq(row.id))
.filter(models::ci::ci_deployment::Column::State.is_in([
models::ci::ci_deployment::DeploymentState::Pending,
models::ci::ci_deployment::DeploymentState::Running,
]))
.count(self.db_read())
.await?;
Ok(active == 0)
}
async fn start_deployment(&self, job: &workflow_job::Model) -> CiResult<()> {
let Some(environment) = &job.environment else {
return Ok(());
};
let env_row = models::ci::ci_environment::Entity::find()
.filter(models::ci::ci_environment::Column::SpaceId.eq(job.space_id))
.filter(models::ci::ci_environment::Column::Name.eq(environment.clone()))
.one(self.db_read())
.await?;
let Some(env_row) = env_row else {
return Ok(());
};
let run = workflow_run::Entity::find_by_id(job.run_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let now = OffsetDateTime::now_utc();
let existing = models::ci::ci_deployment::Entity::find()
.filter(models::ci::ci_deployment::Column::JobId.eq(job.id))
.one(self.db_read())
.await?;
if existing.is_some() {
return Ok(());
}
models::ci::ci_deployment::ActiveModel {
id: Set(Uuid::now_v7()),
environment_id: Set(env_row.id),
run_id: Set(job.run_id),
job_id: Set(job.id),
repo_id: Set(job.repo_id),
commit_sha: Set(run.commit_sha.clone()),
git_ref: Set(run.git_ref.clone()),
state: Set(models::ci::ci_deployment::DeploymentState::Running),
actor_id: Set(run.trigger_user_id),
created_at: Set(now),
updated_at: Set(now),
}
.insert(self.db_write())
.await?;
Ok(())
}
pub async fn finish_deployment(
&self,
job_id: Uuid,
state: models::ci::ci_deployment::DeploymentState,
) -> CiResult<()> {
let now = OffsetDateTime::now_utc();
models::ci::ci_deployment::Entity::update_many()
.col_expr(models::ci::ci_deployment::Column::State, Expr::value(state))
.col_expr(
models::ci::ci_deployment::Column::UpdatedAt,
Expr::value(now),
)
.filter(models::ci::ci_deployment::Column::JobId.eq(job_id))
.exec(self.db_write())
.await?;
Ok(())
}
async fn mark_run_running(&self, run_id: Uuid, runner_id: Uuid) -> CiResult<()> {
let now = OffsetDateTime::now_utc();
let updated = workflow_run::Entity::update_many()
.col_expr(
workflow_run::Column::State,
Expr::value(WorkflowRunState::Running),
)
.col_expr(workflow_run::Column::RunnerId, Expr::value(runner_id))
.col_expr(workflow_run::Column::StartedAt, Expr::value(Some(now)))
.col_expr(workflow_run::Column::UpdatedAt, Expr::value(now))
.filter(workflow_run::Column::Id.eq(run_id))
.filter(workflow_run::Column::State.eq(WorkflowRunState::Queued))
.exec(self.db_write())
.await?;
if updated.rows_affected > 0
&& let Some(run) = workflow_run::Entity::find_by_id(run_id)
.one(self.db_write())
.await?
{
self.emit_actions_workflow_run(&run, "in_progress").await?;
}
Ok(())
}
pub async fn renew_lease(&self, job_id: Uuid, runner_id: Uuid) -> CiResult<()> {
let now = OffsetDateTime::now_utc();
let expires = now + time::Duration::seconds(self.config().ci.lease_secs as i64);
let updated = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::LeaseExpiresAt,
Expr::value(Some(expires)),
)
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job_id))
.filter(workflow_job::Column::RunnerId.eq(runner_id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
return Err(CiError::NotFound);
}
Ok(())
}
pub async fn expire_leases(&self) -> CiResult<u64> {
let now = OffsetDateTime::now_utc();
let max_attempts = self.config().ci.max_attempts;
let jobs = workflow_job::Entity::find()
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.filter(workflow_job::Column::LeaseExpiresAt.lt(now))
.all(self.db_read())
.await?;
let mut affected = 0u64;
for job in jobs {
if job.attempt >= max_attempts {
workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Failure),
)
.col_expr(
workflow_job::Column::Error,
Expr::value(Some("runner lost".to_string())),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(workflow_job::Column::RunnerId, Expr::value(None::<Uuid>))
.col_expr(
workflow_job::Column::LeaseToken,
Expr::value(None::<String>),
)
.col_expr(
workflow_job::Column::LeaseExpiresAt,
Expr::value(None::<OffsetDateTime>),
)
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job.id))
.exec(self.db_write())
.await?;
self.finish_deployment(job.id, models::ci::ci_deployment::DeploymentState::Failure)
.await?;
affected += 1;
self.record_event(
job.run_id,
"workflow_job",
serde_json::json!({
"action": "failed",
"job": job.name,
"reason": "runner lost",
}),
)
.await;
} else {
workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Queued),
)
.col_expr(workflow_job::Column::RunnerId, Expr::value(None::<Uuid>))
.col_expr(
workflow_job::Column::LeaseToken,
Expr::value(None::<String>),
)
.col_expr(
workflow_job::Column::LeaseExpiresAt,
Expr::value(None::<OffsetDateTime>),
)
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job.id))
.exec(self.db_write())
.await?;
self.finish_deployment(
job.id,
models::ci::ci_deployment::DeploymentState::Cancelled,
)
.await?;
affected += 1;
tracing::warn!(
job_id = %job.id,
run_id = %job.run_id,
attempt = job.attempt,
"job lease expired; requeued"
);
}
}
Ok(affected)
}
pub async fn apply_cancellations(&self) -> CiResult<u64> {
let now = OffsetDateTime::now_utc();
let grace = self.config().ci.cancel_grace_secs as i64;
let runs = workflow_run::Entity::find()
.filter(workflow_run::Column::State.eq(WorkflowRunState::CancelRequested))
.all(self.db_read())
.await?;
let mut affected = 0u64;
for run in runs {
workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Cancelled),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::RunId.eq(run.id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Pending, WorkflowJobState::Queued]),
)
.exec(self.db_write())
.await?;
workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Cancelled),
)
.col_expr(
workflow_job::Column::Outcome,
Expr::value(Some("cancelled".to_owned())),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.filter(workflow_job::Column::RunId.eq(run.id))
.filter(workflow_job::Column::IsTemplate.eq(true))
.filter(workflow_job::Column::State.eq(WorkflowJobState::Running))
.exec(self.db_write())
.await?;
let stale = now - time::Duration::seconds(grace.max(1));
let forced = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Cancelled),
)
.col_expr(
workflow_job::Column::Error,
Expr::value(Some("cancelled by janitor".to_string())),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::RunId.eq(run.id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.filter(workflow_job::Column::UpdatedAt.lt(stale))
.exec(self.db_write())
.await?;
affected += forced.rows_affected;
let remaining = workflow_job::Entity::find()
.filter(workflow_job::Column::RunId.eq(run.id))
.all(self.db_read())
.await?;
if remaining.iter().all(|j| j.state.is_terminal()) {
self.finish_run(run.id, WorkflowRunState::Cancelled, None)
.await?;
affected += 1;
}
}
Ok(affected)
}
pub async fn finalize_runs(&self) -> CiResult<u64> {
let runs = workflow_run::Entity::find()
.filter(
workflow_run::Column::State
.is_in([WorkflowRunState::Running, WorkflowRunState::Queued]),
)
.all(self.db_read())
.await?;
let mut affected = 0u64;
for run in runs {
let jobs = workflow_job::Entity::find()
.filter(workflow_job::Column::RunId.eq(run.id))
.all(self.db_read())
.await?;
if jobs.is_empty() || !jobs.iter().all(|j| j.state.is_terminal()) {
continue;
}
let failed = jobs
.iter()
.filter(|job| {
ci_protocol::actions::from_snapshot(&run.definition).is_none()
|| (job.is_template && job.logical_job_id.is_none())
})
.any(|j| {
matches!(
j.state,
WorkflowJobState::Failure
| WorkflowJobState::TimedOut
| WorkflowJobState::Cancelled
)
});
let state = if failed {
WorkflowRunState::Failure
} else {
WorkflowRunState::Success
};
self.finish_run(run.id, state, None).await?;
affected += 1;
}
Ok(affected)
}
pub async fn finish_run(
&self,
run_id: Uuid,
state: WorkflowRunState,
error: Option<String>,
) -> CiResult<()> {
let now = OffsetDateTime::now_utc();
let updated = workflow_run::Entity::update_many()
.col_expr(workflow_run::Column::State, Expr::value(state))
.col_expr(workflow_run::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(workflow_run::Column::UpdatedAt, Expr::value(now))
.filter(workflow_run::Column::Id.eq(run_id))
.filter(workflow_run::Column::State.is_in([
WorkflowRunState::Queued,
WorkflowRunState::Running,
WorkflowRunState::CancelRequested,
]))
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
return Ok(());
}
let run = workflow_run::Entity::find_by_id(run_id)
.one(self.db_write())
.await?
.ok_or(CiError::NotFound)?;
self.publish_commit_status(&run, state).await?;
self.record_event(
run_id,
"workflow_run",
serde_json::json!({
"action": "finished",
"state": crate::share::states::run_state_str(state),
"error": error,
}),
)
.await;
tracing::info!(run_id = %run_id, state = ?state, "workflow run finished");
self.emit_actions_workflow_run(&run, "completed").await?;
Ok(())
}
pub async fn mark_run_cancel_requested(&self, run_id: Uuid) -> CiResult<()> {
let now = OffsetDateTime::now_utc();
let updated = workflow_run::Entity::update_many()
.col_expr(
workflow_run::Column::State,
Expr::value(WorkflowRunState::CancelRequested),
)
.col_expr(workflow_run::Column::UpdatedAt, Expr::value(now))
.filter(workflow_run::Column::Id.eq(run_id))
.filter(
workflow_run::Column::State
.is_in([WorkflowRunState::Queued, WorkflowRunState::Running]),
)
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
return Err(CiError::Conflict(
"only queued or running runs can be cancelled".to_string(),
));
}
self.record_event(
run_id,
"workflow_run",
serde_json::json!({ "action": "cancel_requested" }),
)
.await;
Ok(())
}
pub async fn cancel_list_for_runner(&self, runner_id: Uuid) -> CiResult<Vec<Uuid>> {
let jobs = workflow_job::Entity::find()
.filter(workflow_job::Column::RunnerId.eq(runner_id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.all(self.db_read())
.await?;
let mut out = Vec::new();
for job in jobs {
let run = workflow_run::Entity::find_by_id(job.run_id)
.one(self.db_read())
.await?;
if let Some(run) = run
&& run.state == WorkflowRunState::CancelRequested
{
out.push(job.id);
} else if job
.actions_context
.as_ref()
.is_some_and(|context| context["cancel_requested"].as_bool().unwrap_or(false))
{
out.push(job.id);
}
}
Ok(out)
}
pub async fn start_job(&self, job_id: Uuid, runner_id: Uuid) -> CiResult<Uuid> {
let now = OffsetDateTime::now_utc();
let updated = workflow_job::Entity::update_many()
.col_expr(
workflow_job::Column::State,
Expr::value(WorkflowJobState::Running),
)
.col_expr(workflow_job::Column::StartedAt, Expr::value(Some(now)))
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job_id))
.filter(workflow_job::Column::RunnerId.eq(runner_id))
.filter(
workflow_job::Column::State
.is_in([WorkflowJobState::Claimed, WorkflowJobState::Running]),
)
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
return Err(CiError::NotFound);
}
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
Ok(job.run_id)
}
pub async fn report_step(
&self,
job_id: Uuid,
runner_id: Uuid,
index: i32,
report: &ci_protocol::runner::StepReport,
) -> CiResult<()> {
self.ensure_job_owned(job_id, runner_id).await?;
let now = OffsetDateTime::now_utc();
let (state, exit_code, started_at, finished_at) = match report.state {
StepReportState::Started => (
WorkflowStepState::Running,
None,
report.started_at.or(Some(now)),
None,
),
StepReportState::Finished => {
let exit = report.exit_code.unwrap_or(0);
let state = if exit == 0 {
WorkflowStepState::Success
} else {
WorkflowStepState::Failure
};
(
state,
Some(exit),
report.started_at,
report.finished_at.or(Some(now)),
)
}
};
let step = workflow_step::Entity::find()
.filter(workflow_step::Column::JobId.eq(job_id))
.filter(workflow_step::Column::StepIndex.eq(index))
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
let mut am: workflow_step::ActiveModel = step.into();
am.state = Set(state);
am.exit_code = Set(exit_code);
if let Some(name) = &report.name {
am.name = Set(name.clone());
}
if let Some(shell) = &report.shell {
am.shell = Set(Some(shell.clone()));
}
if let Some(workdir) = &report.working_directory {
am.working_directory = Set(Some(workdir.clone()));
}
if let Some(script) = &report.script {
am.script = Set(script.clone());
}
if let Some(started) = started_at {
am.started_at = Set(Some(started));
}
if let Some(finished) = finished_at {
am.finished_at = Set(Some(finished));
}
am.updated_at = Set(now);
am.update(self.db_write()).await?;
Ok(())
}
pub async fn ensure_job_owned(&self, job_id: Uuid, runner_id: Uuid) -> CiResult<()> {
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if job.runner_id != Some(runner_id) {
return Err(CiError::Forbidden);
}
Ok(())
}
pub async fn finish_job(
&self,
job_id: Uuid,
runner_id: Uuid,
finish: &ci_protocol::runner::JobFinishRequest,
) -> CiResult<Uuid> {
let job = workflow_job::Entity::find_by_id(job_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)?;
if job.runner_id != Some(runner_id) {
return Err(CiError::Forbidden);
}
if job.state.is_terminal() {
return Ok(job.run_id);
}
if finish.lease_token.is_some() && finish.lease_token != job.lease_token {
return Err(CiError::Conflict("job lease was superseded".into()));
}
let now = OffsetDateTime::now_utc();
let state = match finish.state {
JobFinishState::Success => WorkflowJobState::Success,
JobFinishState::Failure => WorkflowJobState::Failure,
JobFinishState::TimedOut => WorkflowJobState::TimedOut,
JobFinishState::Cancelled => WorkflowJobState::Cancelled,
};
let updated = workflow_job::Entity::update_many()
.col_expr(workflow_job::Column::State, Expr::value(state))
.col_expr(
workflow_job::Column::Outputs,
Expr::value(Some(serde_json::to_value(&finish.outputs)?)),
)
.col_expr(
workflow_job::Column::Outcome,
Expr::value(Some(state.as_str().to_owned())),
)
.col_expr(
workflow_job::Column::ExitCode,
Expr::value(finish.exit_code),
)
.col_expr(
workflow_job::Column::Error,
Expr::value(finish.error.clone()),
)
.col_expr(workflow_job::Column::FinishedAt, Expr::value(Some(now)))
.col_expr(
workflow_job::Column::LeaseToken,
Expr::value(None::<String>),
)
.col_expr(
workflow_job::Column::LeaseExpiresAt,
Expr::value(None::<OffsetDateTime>),
)
.col_expr(workflow_job::Column::UpdatedAt, Expr::value(now))
.filter(workflow_job::Column::Id.eq(job_id))
.filter(workflow_job::Column::RunnerId.eq(runner_id))
.filter(
sea_orm::Condition::all().add_option(
finish
.lease_token
.as_ref()
.map(|token| workflow_job::Column::LeaseToken.eq(token.clone())),
),
)
.exec(self.db_write())
.await?;
if updated.rows_affected == 0 {
return Ok(job.run_id);
}
let duration = job
.started_at
.and_then(|started| (now - started).try_into().ok());
export::metrics::ci_job_finished(state.as_str(), duration);
if state != WorkflowJobState::Success {
workflow_step::Entity::update_many()
.col_expr(
workflow_step::Column::State,
Expr::value(WorkflowStepState::Cancelled),
)
.filter(workflow_step::Column::JobId.eq(job_id))
.filter(
workflow_step::Column::State
.is_in([WorkflowStepState::Pending, WorkflowStepState::Running]),
)
.exec(self.db_write())
.await?;
}
let deployment_state = match state {
WorkflowJobState::Success => models::ci::ci_deployment::DeploymentState::Success,
WorkflowJobState::Cancelled => models::ci::ci_deployment::DeploymentState::Cancelled,
_ => models::ci::ci_deployment::DeploymentState::Failure,
};
self.finish_deployment(job.id, deployment_state).await?;
self.record_event(
job.run_id,
"workflow_job",
serde_json::json!({
"action": "finished",
"job": job.name,
"state": state.as_str(),
"exit_code": finish.exit_code,
}),
)
.await;
Ok(job.run_id)
}
pub async fn record_event(&self, run_id: Uuid, event_type: &str, payload: serde_json::Value) {
let run = match workflow_run::Entity::find_by_id(run_id)
.one(self.db_read())
.await
{
Ok(Some(run)) => run,
Ok(None) => return,
Err(error) => {
tracing::warn!(%error, "ci event run lookup failed");
return;
}
};
if let Err(error) =
crate::outbound::enqueue_for_repo(self, run.repo_id, event_type, &payload).await
{
tracing::warn!(%error, "outbound webhook enqueue failed");
}
}
pub async fn repo_by_id(&self, repo_id: Uuid) -> CiResult<models::repo::repos::Model> {
models::repo::repos::Entity::find_by_id(repo_id)
.one(self.db_read())
.await?
.ok_or(CiError::NotFound)
}
async fn publish_commit_status(
&self,
run: &workflow_run::Model,
state: WorkflowRunState,
) -> CiResult<()> {
let Some(creator) = run.trigger_user_id else {
tracing::warn!(run_id = %run.id, "run has no trigger user; skipping commit status");
return Ok(());
};
let repo = self.repo_by_id(run.repo_id).await?;
let space = self.space_of_repo(&repo).await?;
let git = git::service::GitService::new(self.state.clone());
let dto_state = match state {
WorkflowRunState::Success => git::service::commits::CommitStatusDtoState::Success,
WorkflowRunState::Failure => git::service::commits::CommitStatusDtoState::Failure,
WorkflowRunState::Cancelled => git::service::commits::CommitStatusDtoState::Error,
_ => git::service::commits::CommitStatusDtoState::Pending,
};
let description = match state {
WorkflowRunState::Success => "workflow succeeded",
WorkflowRunState::Failure => "workflow failed",
WorkflowRunState::Cancelled => "workflow cancelled",
_ => "workflow running",
};
let base_url = self.config().git.base_url.trim_end_matches('/').to_string();
let target_url = format!(
"{base_url}/app/{}/repo/{}?tab=pipelines&run={}",
space.lower_name, repo.name, run.id
);
if let Err(error) = git
.set_commit_status(
space.lower_name.clone(),
repo.name.clone(),
creator,
git::service::commits::SetCommitStatusCmd {
sha: run.commit_sha.clone(),
state: dto_state,
context: format!("ci/{}", run.workflow_name),
description: Some(description.to_string()),
target_url: Some(target_url),
},
)
.await
{
tracing::warn!(run_id = %run.id, error = %error, "commit status write failed");
}
Ok(())
}
}