From 444592755428a984983a69ea308af97aed3def70 Mon Sep 17 00:00:00 2001 From: Dennis Wuitz Date: Tue, 20 Oct 2026 22:17:36 +0200 Subject: [PATCH 1/2] trigger: fix: flake update without a second eval --- .../gradient-ci/src/trigger/input_update.rs | 27 ++++-- backend/gradient-ci/src/trigger/mod.rs | 2 +- backend/gradient-ci/src/trigger/tests/mod.rs | 8 +- backend/gradient-fixtures/src/shape/insert.rs | 6 +- backend/gradient-fixtures/src/shape/mod.rs | 6 +- .../gradient-fixtures/src/shape/scheduling.rs | 45 +++++++++- .../gradient-scheduler/src/cases/triggers.rs | 35 ++++++++ .../gradient-scheduler/src/trigger_firing.rs | 11 ++- .../src/endpoints/tasks/triggers.rs | 87 ++++++++++++------- docs/gradient-api.yaml | 2 +- docs/src/guides/flake-updates.md | 3 +- 11 files changed, 177 insertions(+), 55 deletions(-) diff --git a/backend/gradient-ci/src/trigger/input_update.rs b/backend/gradient-ci/src/trigger/input_update.rs index aa46cbb56..ff745319e 100644 --- a/backend/gradient-ci/src/trigger/input_update.rs +++ b/backend/gradient-ci/src/trigger/input_update.rs @@ -11,12 +11,25 @@ use sea_orm::{ ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, IntoActiveModel, QueryFilter, }; +#[derive(Debug, PartialEq, Eq)] +pub enum InputUpdateOutcome { + NotConfigured, + AlreadyRunning, + Started(Vec), +} + +impl InputUpdateOutcome { + pub fn replaces_evaluation(&self) -> bool { + !matches!(self, Self::NotConfigured) + } +} + pub async fn maybe_trigger_input_update( db: &C, task: &MTask, base_commit_hash: Vec, trigger: Option, -) -> Result, TriggerError> { +) -> Result { let Some(action) = ETaskAction::find() .filter(CTaskAction::Task.eq(task.id)) .filter(CTaskAction::Active.eq(true)) @@ -24,13 +37,13 @@ pub async fn maybe_trigger_input_update( .one(db) .await? else { - return Ok(Vec::new()); + return Ok(InputUpdateOutcome::NotConfigured); }; let Ok(ActionConfig::OpenPr { granularity, .. }) = serde_json::from_value::(action.config.clone()) else { - return Ok(Vec::new()); + return Ok(InputUpdateOutcome::NotConfigured); }; let overrides = ETaskFlakeInputOverride::find() @@ -38,12 +51,12 @@ pub async fn maybe_trigger_input_update( .all(db) .await?; if overrides.iter().any(|o| o.url.is_some()) { - return Ok(Vec::new()); + return Ok(InputUpdateOutcome::NotConfigured); } let tracked: Vec = overrides.into_iter().map(|o| o.input_name).collect(); if tracked.is_empty() { - return Ok(Vec::new()); + return Ok(InputUpdateOutcome::NotConfigured); } let active_codes: Vec = EvaluationStatus::ACTIVE @@ -58,7 +71,7 @@ pub async fn maybe_trigger_input_update( .one(db) .await?; if already_running.is_some() { - return Ok(Vec::new()); + return Ok(InputUpdateOutcome::AlreadyRunning); } let base_commit: String = base_commit_hash @@ -105,7 +118,7 @@ pub async fn maybe_trigger_input_update( } } - Ok(created) + Ok(InputUpdateOutcome::Started(created)) } /// The commit is blank because the PR commit does not exist until the branch is force-pushed. diff --git a/backend/gradient-ci/src/trigger/mod.rs b/backend/gradient-ci/src/trigger/mod.rs index f61bdec53..b48382f46 100644 --- a/backend/gradient-ci/src/trigger/mod.rs +++ b/backend/gradient-ci/src/trigger/mod.rs @@ -13,7 +13,7 @@ mod restart; use thiserror::Error; pub use drv_recovery::trigger_drv_recovery; -pub use input_update::{fan_out_expansion, maybe_trigger_input_update}; +pub use input_update::{InputUpdateOutcome, fan_out_expansion, maybe_trigger_input_update}; pub use new_evaluation::trigger_evaluation; pub use restart::trigger_evaluation_retry; diff --git a/backend/gradient-ci/src/trigger/tests/mod.rs b/backend/gradient-ci/src/trigger/tests/mod.rs index c13d3ba4e..db4ffb512 100644 --- a/backend/gradient-ci/src/trigger/tests/mod.rs +++ b/backend/gradient-ci/src/trigger/tests/mod.rs @@ -314,7 +314,7 @@ async fn input_update_noop_without_open_pr_action() { let created = maybe_trigger_input_update(&db, &task, vec![0u8; 20], None) .await .unwrap(); - assert!(created.is_empty()); + assert_eq!(created, InputUpdateOutcome::NotConfigured); } #[tokio::test] @@ -328,7 +328,7 @@ async fn input_update_noop_without_tracked_inputs() { let created = maybe_trigger_input_update(&db, &task, vec![0u8; 20], None) .await .unwrap(); - assert!(created.is_empty()); + assert_eq!(created, InputUpdateOutcome::NotConfigured); } #[tokio::test] @@ -346,7 +346,7 @@ async fn input_update_pinned_override_blocks_run() { let created = maybe_trigger_input_update(&db, &task, vec![0u8; 20], None) .await .unwrap(); - assert!(created.is_empty()); + assert_eq!(created, InputUpdateOutcome::NotConfigured); } #[tokio::test] @@ -374,5 +374,5 @@ async fn input_update_creates_eval_for_tracked_input() { let created = maybe_trigger_input_update(&db, &task, vec![0u8; 20], None) .await .unwrap(); - assert_eq!(created, vec![eval_id]); + assert_eq!(created, InputUpdateOutcome::Started(vec![eval_id])); } diff --git a/backend/gradient-fixtures/src/shape/insert.rs b/backend/gradient-fixtures/src/shape/insert.rs index b1f26461d..99bcad064 100644 --- a/backend/gradient-fixtures/src/shape/insert.rs +++ b/backend/gradient-fixtures/src/shape/insert.rs @@ -9,8 +9,8 @@ use gradient_entity::{ cluster_member, commit, derivation, derivation_build, derivation_dependency, derivation_feature, derivation_metric, derivation_output, dispatched_job, dispatched_job_phase, entry_point, eval_cache_store, evaluation, evaluation_flake_input_override, - evaluation_input_update, evaluation_metric, feature, project, project_cache, task, - task_trigger, user, worker_registration, + evaluation_input_update, evaluation_metric, feature, project, project_cache, task, task_action, + task_flake_input_override, task_trigger, user, worker_registration, }; use sea_orm::sea_query::OnConflict; use sea_orm::{DatabaseConnection, EntityTrait, IntoActiveModel}; @@ -40,6 +40,8 @@ pub(super) async fn all(shape: Shape, db: &DatabaseConnection) { insert!(db, project::Entity, shape.projects); insert!(db, task::Entity, shape.tasks); insert!(db, task_trigger::Entity, shape.triggers); + insert!(db, task_action::Entity, shape.actions); + insert!(db, task_flake_input_override::Entity, shape.tracked_inputs); insert!(db, commit::Entity, shape.commits); insert!(db, evaluation::Entity, shape.evaluations); insert!(db, derivation::Entity, shape.derivations); diff --git a/backend/gradient-fixtures/src/shape/mod.rs b/backend/gradient-fixtures/src/shape/mod.rs index b259d50aa..0ec67a1ab 100644 --- a/backend/gradient-fixtures/src/shape/mod.rs +++ b/backend/gradient-fixtures/src/shape/mod.rs @@ -25,8 +25,8 @@ use gradient_entity::{ cluster_job, cluster_member, commit, derivation, derivation_build, derivation_dependency, derivation_feature, derivation_metric, derivation_output, dispatched_job, dispatched_job_phase, entry_point, eval_cache_store, evaluation, evaluation_flake_input_override, - evaluation_input_update, evaluation_metric, feature, project, project_cache, task, - task_trigger, user, worker_registration, + evaluation_input_update, evaluation_metric, feature, project, project_cache, task, task_action, + task_flake_input_override, task_trigger, user, worker_registration, }; use sea_orm::DatabaseConnection; use uuid::Uuid; @@ -57,6 +57,8 @@ pub struct Shape { pub(super) assignments: Vec, pub(super) attempts: Vec, pub(super) triggers: Vec, + pub(super) actions: Vec, + pub(super) tracked_inputs: Vec, pub(super) clusters: Vec, pub(super) members: Vec, pub(super) cluster_attempts: Vec, diff --git a/backend/gradient-fixtures/src/shape/scheduling.rs b/backend/gradient-fixtures/src/shape/scheduling.rs index b47f8c5c3..f1bd4baa1 100644 --- a/backend/gradient-fixtures/src/shape/scheduling.rs +++ b/backend/gradient-fixtures/src/shape/scheduling.rs @@ -15,7 +15,8 @@ use gradient_entity::ids::{ BuildAttemptId, CacheUpstreamId, ClusterAttemptId, ClusterJobId, ClusterMemberId, CommitId, DerivationFeatureId, DerivationMetricId, DispatchedJobId, DispatchedJobPhaseId, EvalCacheStoreId, EvaluationFlakeInputOverrideId, EvaluationId, EvaluationInputUpdateId, - EvaluationMetricId, FeatureId, ProjectId, TaskId, TaskTriggerId, + EvaluationMetricId, FeatureId, FlakeInputOverrideId, IntegrationId, ProjectId, TaskActionId, + TaskId, TaskTriggerId, }; use gradient_entity::project_cache::CacheSubscriptionMode; use gradient_entity::task_trigger::TriggerType; @@ -23,8 +24,10 @@ use gradient_entity::{ build_attempt, build_job, cache_upstream, cluster_attempt, cluster_job, cluster_member, derivation, derivation_build, derivation_feature, derivation_metric, dispatched_job, dispatched_job_phase, eval_cache_store, evaluation, evaluation_flake_input_override, - evaluation_input_update, evaluation_metric, feature, task_trigger, + evaluation_input_update, evaluation_metric, feature, task_action, task_flake_input_override, + task_trigger, }; +use gradient_types::{ActionConfig, ActionType, PatchGeneratorKind, PrGranularity, VerifyGate}; use super::Shape; use crate::{rows, test_date}; @@ -109,6 +112,44 @@ impl Shape { self } + pub fn flake_update_pull_request(mut self, inputs: &[&str]) -> Self { + let task = self.tasks[self.current_task()].id; + let created_by = self.owner(); + let config = ActionConfig::OpenPr { + integration_id: IntegrationId::now_v7(), + generator: PatchGeneratorKind::FlakeLock, + granularity: PrGranularity::PerRun, + verify_gate: VerifyGate::Build, + branch_pattern: "gradient/flake-lock-update".into(), + title_template: None, + body_template: None, + update_existing: true, + }; + self.actions.push(task_action::Model { + id: TaskActionId::now_v7(), + task, + name: "flake-update".into(), + action_type: ActionType::OpenPr, + config: serde_json::to_value(config).expect("an open_pr config"), + events: serde_json::json!([]), + active: true, + last_fired_at: None, + created_by, + created_at: test_date(), + updated_at: test_date(), + }); + self.tracked_inputs + .extend(inputs.iter().map(|input| task_flake_input_override::Model { + id: FlakeInputOverrideId::now_v7(), + task, + input_name: (*input).into(), + url: None, + created_at: test_date(), + updated_at: test_date(), + })); + self + } + pub fn input_override(mut self, input: &str, url: &str) -> Self { let evaluation = self.evaluations[self.current_evaluation()].id; self.input_overrides diff --git a/backend/gradient-scheduler/src/cases/triggers.rs b/backend/gradient-scheduler/src/cases/triggers.rs index 96cd2e5f9..bc8bc9d9d 100644 --- a/backend/gradient-scheduler/src/cases/triggers.rs +++ b/backend/gradient-scheduler/src/cases/triggers.rs @@ -1183,6 +1183,41 @@ async fn t20_a_poll_of_the_default_branch_and_a_push_to_it_abort_each_other() { case.finish().await; } +#[tokio::test] +async fn t21_a_flake_update_task_fires_only_its_input_update() { + let (e0, t1) = (EvaluationId::now_v7(), TaskTriggerId::now_v7()); + let repo = GitRepo::new(); + let c0 = repo.commit("c0"); + let c1 = repo.commit("c1"); + repo.push("main", &c1); + let case = SchedulerCase::start( + polled( + e0, + t1, + &repo, + &c0, + EvaluationStatus::Completed, + ConcurrencyPolicy::SoftAbort, + ) + .flake_update_pull_request(&["nixpkgs"]), + ) + .await; + + case.fire_once().await; + case.deliver_trigger_fires().await; + + let fresh = evaluations_besides(&case, &[e0]).await; + let kinds: Vec<_> = fresh.iter().map(|e| (e.kind, e.trigger)).collect(); + assert_eq!( + kinds, + vec![(EvaluationKind::InputUpdate, Some(t1))], + "{fresh:?}" + ); + case.check("step 1").await; + + case.finish().await; +} + fn due_polling(e0: EvaluationId, t1: TaskTriggerId, repo: &GitRepo) -> Shape { shape() .evaluation(e0) diff --git a/backend/gradient-scheduler/src/trigger_firing.rs b/backend/gradient-scheduler/src/trigger_firing.rs index 9839a6100..e4efd698a 100644 --- a/backend/gradient-scheduler/src/trigger_firing.rs +++ b/backend/gradient-scheduler/src/trigger_firing.rs @@ -68,7 +68,8 @@ pub(crate) fn cron_due( .unwrap_or(false) } -use gradient_ci::{ApplyInput, ApplyOutcome, apply_trigger, trigger::maybe_trigger_input_update}; +use gradient_ci::trigger::{InputUpdateOutcome, maybe_trigger_input_update}; +use gradient_ci::{ApplyInput, ApplyOutcome, apply_trigger}; use gradient_db::deliveries::trigger_fire::{TriggerFire, enqueue_trigger_fires}; use gradient_db::{DbContext, WorkerDb}; use gradient_entity::task_trigger as ept; @@ -226,15 +227,19 @@ async fn apply_resolved_head( has_update: bool, head: BranchHead, ) { - if let Err(e) = maybe_trigger_input_update( + let input_update = maybe_trigger_input_update( &scheduler.db.worker_db, task, head.commit.clone(), Some(trig.id), ) .await - { + .unwrap_or_else(|e| { warn!(error = %e, trigger_id = %trig.id, "input_update trigger failed"); + InputUpdateOutcome::NotConfigured + }); + if input_update.replaces_evaluation() { + return; } if has_update || matches!(cfg, TriggerConfig::Time { .. }) { diff --git a/backend/gradient-web/src/endpoints/tasks/triggers.rs b/backend/gradient-web/src/endpoints/tasks/triggers.rs index e95b0bac4..ca2bfbd0d 100644 --- a/backend/gradient-web/src/endpoints/tasks/triggers.rs +++ b/backend/gradient-web/src/endpoints/tasks/triggers.rs @@ -15,6 +15,7 @@ use anyhow::Context; use axum::extract::State; use axum::{Extension, Json}; use chrono::Utc; +use gradient_ci::trigger::InputUpdateOutcome; use gradient_ci::{ApplyInput, ApplyOutcome, apply_trigger}; use gradient_core::ServerState; use gradient_db::lookup::by_ids; @@ -303,7 +304,7 @@ pub async fn delete_one( operation_id = "testTaskTrigger", tag = "triggers", summary = "Manually fire trigger", - description = "Resolves the task HEAD and applies the trigger immediately, bypassing commit dedup. Returns the apply outcome.", + description = "Resolves the task HEAD and applies the trigger immediately, bypassing commit dedup. A task with an active `open_pr` action and tracked flake inputs starts only its input update. Returns the outcome.", params(TaskPath, ("id" = TaskTriggerId, Path)), responses((status = OK, description = "Trigger apply outcome", body = serde_json::Value), WebError) )] @@ -340,47 +341,61 @@ pub async fn fire_now( .await .context("resolve the branch head")?; - if let Err(e) = gradient_ci::trigger::maybe_trigger_input_update( + let input_update = gradient_ci::trigger::maybe_trigger_input_update( &state.web_db, &proj, head.commit.clone(), Some(row.id), ) .await - { + .unwrap_or_else(|e| { tracing::warn!(error = %e, trigger_id = %row.id, "manual input_update trigger failed"); - } - - let input = ApplyInput { - trigger_id: row.id, - trigger_type, - branch: head.branch, - commit_hash: head.commit, - commit_message: Some(commit_message), - author_name: Some(author_name), - manual: true, - gate_approval: None, - repository_override: None, - wildcard_override: None, - source_comment: None, - instance_max_storage_gb: state.config.cache.max_storage_gb, + InputUpdateOutcome::NotConfigured + }); + + let body = match input_update { + InputUpdateOutcome::NotConfigured => { + let input = ApplyInput { + trigger_id: row.id, + trigger_type, + branch: head.branch, + commit_hash: head.commit, + commit_message: Some(commit_message), + author_name: Some(author_name), + manual: true, + gate_approval: None, + repository_override: None, + wildcard_override: None, + source_comment: None, + instance_max_storage_gb: state.config.cache.max_storage_gb, + }; + apply_now(&state, &scheduler, &proj, input).await? + } + InputUpdateOutcome::AlreadyRunning => { + serde_json::json!({ "outcome": "InputUpdateRunning" }) + } + InputUpdateOutcome::Started(evaluations) => serde_json::json!({ + "outcome": "InputUpdateStarted", + "evaluation_ids": evaluations, + }), }; - let outcome = apply_trigger(&state.web_db, &proj, input) + stamp_last_fired(&state.web_db, &row).await; + + Ok(ok_json(body)) +} + +async fn apply_now( + state: &ServerState, + scheduler: &Scheduler, + task: &MTask, + input: ApplyInput, +) -> WebResult { + let outcome = apply_trigger(&state.web_db, task, input) .await .context("apply the trigger")?; - { - let now = gradient_types::now(); - let mut active: gradient_entity::task_trigger::ActiveModel = row.clone().into(); - active.last_fired_at = Set(Some(now)); - active.updated_at = Set(now); - if let Err(e) = gradient_db::triggers::update(&state.web_db, active).await { - tracing::warn!(error = %e, trigger_id = %row.id, "failed to stamp trigger last_fired_at"); - } - } - - let body = match outcome { + Ok(match outcome { ApplyOutcome::Created { evaluation: eval, aborted_evaluation, @@ -402,7 +417,15 @@ pub async fn fire_now( ApplyOutcome::SkippedConcurrency => { serde_json::json!({ "outcome": "SkippedConcurrency" }) } - }; + }) +} - Ok(ok_json(body)) +async fn stamp_last_fired(db: &gradient_db::WebDb, row: &MTaskTrigger) { + let now = gradient_types::now(); + let mut active: gradient_entity::task_trigger::ActiveModel = row.clone().into(); + active.last_fired_at = Set(Some(now)); + active.updated_at = Set(now); + if let Err(e) = gradient_db::triggers::update(db, active).await { + tracing::warn!(error = %e, trigger_id = %row.id, "failed to stamp trigger last_fired_at"); + } } diff --git a/docs/gradient-api.yaml b/docs/gradient-api.yaml index f88c11894..0f657ec6a 100644 --- a/docs/gradient-api.yaml +++ b/docs/gradient-api.yaml @@ -9748,7 +9748,7 @@ paths: tags: - triggers summary: Manually fire trigger - description: Resolves the task HEAD and applies the trigger immediately, bypassing commit dedup. Returns the apply outcome. + description: Resolves the task HEAD and applies the trigger immediately, bypassing commit dedup. A task with an active `open_pr` action and tracked flake inputs starts only its input update. Returns the outcome. operationId: testTaskTrigger parameters: - name: project diff --git a/docs/src/guides/flake-updates.md b/docs/src/guides/flake-updates.md index bfd0af2ee..7913889a2 100644 --- a/docs/src/guides/flake-updates.md +++ b/docs/src/guides/flake-updates.md @@ -56,7 +56,8 @@ Supported input types: `github`, `gitlab` and `git`, including `git` over SSH wi ## Update Run -Every trigger fire and every click on **Start Evaluation** launches an update evaluation in addition to the normal one. +- A trigger fire and a **Fire Now** click start the update evaluation alone. +- Clicks on **Start Evaluation** start the update evaluation alongside the normal evaluation. ```mermaid flowchart LR From 0b05332ff8b8f72b43d6c86e3997c63fed61d325 Mon Sep 17 00:00:00 2001 From: Dennis Wuitz Date: Tue, 20 Oct 2026 22:33:08 +0200 Subject: [PATCH 2/2] web: refactor: fire now takes the db context --- backend/gradient-web/src/endpoints/tasks/triggers.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/backend/gradient-web/src/endpoints/tasks/triggers.rs b/backend/gradient-web/src/endpoints/tasks/triggers.rs index ca2bfbd0d..abd39af1e 100644 --- a/backend/gradient-web/src/endpoints/tasks/triggers.rs +++ b/backend/gradient-web/src/endpoints/tasks/triggers.rs @@ -18,6 +18,7 @@ use chrono::Utc; use gradient_ci::trigger::InputUpdateOutcome; use gradient_ci::{ApplyInput, ApplyOutcome, apply_trigger}; use gradient_core::ServerState; +use gradient_db::DbContext; use gradient_db::lookup::by_ids; use gradient_scheduler::Scheduler; use gradient_sources::resolve_head; @@ -369,7 +370,7 @@ pub async fn fire_now( source_comment: None, instance_max_storage_gb: state.config.cache.max_storage_gb, }; - apply_now(&state, &scheduler, &proj, input).await? + apply_now(&db, &scheduler, &proj, input).await? } InputUpdateOutcome::AlreadyRunning => { serde_json::json!({ "outcome": "InputUpdateRunning" }) @@ -386,12 +387,12 @@ pub async fn fire_now( } async fn apply_now( - state: &ServerState, + db: &DbContext, scheduler: &Scheduler, task: &MTask, input: ApplyInput, ) -> WebResult { - let outcome = apply_trigger(&state.web_db, task, input) + let outcome = apply_trigger(&db.web_db, task, input) .await .context("apply the trigger")?; @@ -407,7 +408,7 @@ async fn apply_now( .await; } - state.record_evaluation_created(&eval).await; + db.record_evaluation_created(&eval).await; serde_json::json!({ "outcome": "Created", "evaluation_id": eval.id,