Create new Steps for resolved derivations

Instead of replacing the current step with a new resolved one
when something is resolved, punt it back to the scheduler.
This will grant us more flexibility in execution and the
method can be used in the future for dynamic derivations,
which won't map 1:1 with the original derivations.

Author: artemist

subprojects/hydra-queue-runner/src/state/build.rs

@@ -25,7 +25,7 @@ pub struct Build {
     pub local_priority: i32,
     pub global_priority: AtomicI32,
 
-    toplevel: arc_swap::ArcSwapOption<Step>,
+    pub toplevel: arc_swap::ArcSwapOption<Step>,
     pub jobset: Arc<Jobset>,
 
     finished_in_db: AtomicBool,

subprojects/hydra-queue-runner/src/state/mod.rs

@@ -10,6 +10,7 @@ mod step;
 mod step_info;
 mod uploader;
 
+use anyhow::Context as _;
 pub use atomic::AtomicDateTime;
 pub use build::{Build, BuildOutput, BuildResultState, BuildTimings, Builds, RemoteBuild};
 pub use jobset::{Jobset, JobsetID, Jobsets};
@@ -46,6 +47,8 @@ enum CreateStepResult {
 
 enum RealiseStepResult {
     None,
+    /// Created a new resolved BuildStep
+    Resolved,
     Valid(Arc<Machine>),
     MaybeCancelled,
     CachedFailure,
@@ -275,7 +278,7 @@ impl State {
         let drv = step_info.step.get_drv_path();
         let mut build_options = nix_utils::BuildOptions::new(None);
 
-        let build_id = {
+        let build = {
             let mut dependents = HashSet::new();
             let mut steps = HashSet::new();
             step_info.step.get_dependents(&mut dependents, &mut steps);
@@ -308,9 +311,11 @@ impl State {
             build_options
                 .set_max_silent_time(biggest_max_silent_time.unwrap_or(build.max_silent_time));
             build_options.set_build_timeout(biggest_build_timeout.unwrap_or(build.timeout));
-            build.id
+            build.clone()
         };
 
+        let build_id = build.id;
+
         let mut job = machine::Job::new(build_id, drv.to_owned());
         job.result.set_start_time_now();
         if self.check_cached_failure(step_info.step.clone()).await {
@@ -403,10 +408,11 @@ impl State {
             }
         };
 
-        let effective_drv = if let Some(resolved_path) = resolved {
+        if let Some(resolved_path) = resolved {
             tracing::info!("resolved CA derivation {drv} -> {resolved_path}");
 
             // Finish original step as "resolved".
+            step_info.step.set_finished(true);
             let mut resolved_result = RemoteBuild::new();
             resolved_result.step_status = BuildStatus::Resolved;
             resolved_result.set_start_time_now();
@@ -424,30 +430,77 @@ impl State {
             .await?;
 
             // Create DB step for the resolved drv under the same build.
-            let resolved_step_nr = {
-                let mut tx = db.begin_transaction().await?;
-                let nr = tx
-                    .create_build_step(
-                        Some(job.result.get_start_time_as_i32()?),
-                        build_id,
-                        &self.store.print_store_path(&resolved_path),
-                        step_info.step.get_system().as_deref(),
-                        machine.hostname.clone(),
-                        BuildStatus::Busy,
-                        None,
-                        None,
-                        vec![],
-                    )
-                    .await?;
+            let mut tx = db.begin_transaction().await?;
+            let nr = tx
+                .create_build_step(
+                    Some(job.result.get_start_time_as_i32()?),
+                    build_id,
+                    &self.store.print_store_path(&resolved_path),
+                    step_info.step.get_system().as_deref(),
+                    machine.hostname.clone(),
+                    BuildStatus::Busy,
+                    None,
+                    None,
+                    vec![],
+                )
+                .await?;
 
-                tx.set_resolved_to(build_id, step_nr, nr).await?;
-                tx.commit().await?;
-                nr
+            tx.set_resolved_to(build_id, step_nr, nr).await?;
+            tx.commit().await?;
+
+            // Actually schedule the step
+            let resolved_step = match self
+                .create_step(
+                    build.clone(),
+                    resolved_path,
+                    Some(build.clone()),
+                    None,
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                    Arc::new(parking_lot::RwLock::new(HashSet::new())),
+                )
+                .await
+            {
+                CreateStepResult::None => {
+                    return Err(anyhow::anyhow!("Could not create resolved build step"));
+                }
+                CreateStepResult::Valid(step) => step,
+                CreateStepResult::PreviousFailure(step) => {
+                    self.handle_previous_failure(build.clone(), step.clone())
+                        .await
+                        .with_context(|| {
+                            format!(
+                                "Failed to handle previous failure in resolved version of {drv}"
+                            )
+                        })?;
+                    return Ok(RealiseStepResult::CachedFailure);
+                }
             };
-            job.step_nr = resolved_step_nr;
-            resolved_path
-        } else {
-            drv.to_owned()
+
+            resolved_step.make_runnable();
+
+            // Copy rdeps so dependents don't try to start too early
+            for rdep in step_info.step.clone_rdeps() {
+                if let Some(rdep) = rdep.upgrade() {
+                    resolved_step.make_rdep(&rdep);
+                }
+            }
+
+            // If we're the root of a build then we need to become the new root,
+            // lest the Step get garbage collected.
+            if *build
+                .toplevel
+                .compare_and_swap(&step_info.step, Some(resolved_step))
+                == Some(step_info.step.clone())
+            {
+                build.propagate_priorities();
+            }
+
+            // New steps runnable
+            self.trigger_dispatch();
+
+            // No more work to do, build will happen in another step
+            return Ok(RealiseStepResult::Resolved);
         };
 
         {
@@ -456,7 +509,7 @@ impl State {
             tx.commit().await?;
         }
         tracing::info!(
-            "Submitting build drv={effective_drv} on machine={} hostname={} build_id={build_id} step_nr={}",
+            "Submitting build drv={drv} on machine={} hostname={} build_id={build_id} step_nr={}",
             machine.id,
             machine.hostname,
             job.step_nr,
@@ -473,7 +526,7 @@ impl State {
         machine
             .build_drv(
                 job,
-                effective_drv,
+                drv.clone(),
                 &build_options,
                 // TODO: cleanup
                 if self.config.use_presigned_uploads() {

subprojects/hydra-queue-runner/src/state/queue.rs

@@ -504,6 +504,7 @@ impl Queues {
                         nr_waiting += 1;
                         nr_steps_waiting_all_queues += 1;
                     }
+                    Ok(crate::state::RealiseStepResult::Resolved) => {}
                     Ok(
                         crate::state::RealiseStepResult::MaybeCancelled
                         | crate::state::RealiseStepResult::CachedFailure,

subprojects/hydra-queue-runner/src/state/step.rs

@@ -351,6 +351,20 @@ impl Step {
             .store(state.deps.len() as u64, Ordering::Relaxed);
     }
 
+    pub fn make_rdep(self: &Arc<Self>, dep: &Arc<Self>) {
+        dep.add_dep(self.clone());
+        let mut state = self.state.write();
+        state.rdeps.push(Arc::downgrade(dep));
+        self.atomic_state
+            .rdeps_len
+            .store(state.rdeps.len() as u64, Ordering::Relaxed);
+    }
+
+    pub fn clone_rdeps(&self) -> Vec<Weak<Step>> {
+        let state = self.state.read();
+        state.rdeps.clone()
+    }
+
     pub fn add_referring_data(
         &self,
         referring_build: Option<&Arc<Build>>,