trigger: fix: flake update without a second eval

backend/gradient-ci/src/trigger/input_update.rs

@@ -11,39 +11,52 @@ use sea_orm::{
     ActiveModelTrait, ColumnTrait, ConnectionTrait, EntityTrait, IntoActiveModel, QueryFilter,
 };
 
+#[derive(Debug, PartialEq, Eq)]
+pub enum InputUpdateOutcome {
+    NotConfigured,
+    AlreadyRunning,
+    Started(Vec<EvaluationId>),
+}
+
+impl InputUpdateOutcome {
+    pub fn replaces_evaluation(&self) -> bool {
+        !matches!(self, Self::NotConfigured)
+    }
+}
+
 pub async fn maybe_trigger_input_update<C: ConnectionTrait>(
     db: &C,
     task: &MTask,
     base_commit_hash: Vec<u8>,
     trigger: Option<TaskTriggerId>,
-) -> Result<Vec<EvaluationId>, TriggerError> {
+) -> Result<InputUpdateOutcome, TriggerError> {
     let Some(action) = ETaskAction::find()
         .filter(CTaskAction::Task.eq(task.id))
         .filter(CTaskAction::Active.eq(true))
         .filter(CTaskAction::ActionType.eq(ActionType::OpenPr))
         .one(db)
         .await?
     else {
-        return Ok(Vec::new());
+        return Ok(InputUpdateOutcome::NotConfigured);
     };
 
     let Ok(ActionConfig::OpenPr { granularity, .. }) =
         serde_json::from_value::<ActionConfig>(action.config.clone())
     else {
-        return Ok(Vec::new());
+        return Ok(InputUpdateOutcome::NotConfigured);
     };
 
     let overrides = ETaskFlakeInputOverride::find()
         .filter(CTaskFlakeInputOverride::Task.eq(task.id))
         .all(db)
         .await?;
     if overrides.iter().any(|o| o.url.is_some()) {
-        return Ok(Vec::new());
+        return Ok(InputUpdateOutcome::NotConfigured);
     }
 
     let tracked: Vec<String> = 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<i32> = EvaluationStatus::ACTIVE
@@ -58,7 +71,7 @@ pub async fn maybe_trigger_input_update<C: ConnectionTrait>(
         .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<C: ConnectionTrait>(
         }
     }
 
-    Ok(created)
+    Ok(InputUpdateOutcome::Started(created))
 }
 
 /// The commit is blank because the PR commit does not exist until the branch is force-pushed.

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;
 

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]));
 }

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);

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<dispatched_job::Model>,
     pub(super) attempts: Vec<build_attempt::Model>,
     pub(super) triggers: Vec<task_trigger::Model>,
+    pub(super) actions: Vec<task_action::Model>,
+    pub(super) tracked_inputs: Vec<task_flake_input_override::Model>,
     pub(super) clusters: Vec<cluster_job::Model>,
     pub(super) members: Vec<cluster_member::Model>,
     pub(super) cluster_attempts: Vec<cluster_attempt::Model>,

backend/gradient-fixtures/src/shape/scheduling.rs

@@ -15,16 +15,19 @@ 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;
 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

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)

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 { .. }) {