queue-runner: resolve CA derivations at dispatch time

Instead of resolving at StepInfo construction and carrying two drv identities
through the gRPC layer, resolve in realise_drv_on_valid_machine once all deps
are built. If resolution yields a different drv, the original step is marked
Resolved and a new DB step is created for the resolved drv with a resolvedTo FK
linking them. The builder only ever sees one drv.

Author: amaanq

subprojects/crates/db/src/connection.rs

@@ -866,6 +866,32 @@ impl Transaction<'_> {
         Ok(step_nr)
     }
 
+    /// Set resolvedToBuild/resolvedToStep on a dependency step after the
+    /// resolved step has been created, linking the dependency to its resolution.
+    #[tracing::instrument(skip(self), err)]
+    pub async fn set_resolved_to(
+        &mut self,
+        origin_build_id: crate::models::BuildID,
+        origin_step_nr: i32,
+        resolved_build_id: crate::models::BuildID,
+        resolved_step_nr: i32,
+    ) -> sqlx::Result<()> {
+        sqlx::query(
+            r"
+              UPDATE buildsteps
+              SET resolvedToBuild = $3, resolvedToStep = $4
+              WHERE build = $1 AND stepnr = $2
+            ",
+        )
+        .bind(origin_build_id)
+        .bind(origin_step_nr)
+        .bind(resolved_build_id)
+        .bind(resolved_step_nr)
+        .execute(&mut *self.tx)
+        .await?;
+        Ok(())
+    }
+
     #[tracing::instrument(
         skip(self, start_time, stop_time, build_id, drv_path, output,),
         err,

subprojects/crates/db/src/models.rs

@@ -17,7 +17,8 @@ pub enum BuildStatus {
     LogLimitExceeded = 10,
     NarSizeLimitExceeded = 11,
     NotDeterministic = 12,
-    Busy = 100, // not stored
+    Resolved = 13, // step was resolved to a CA derivation, see resolvedTo FK
+    Busy = 100,    // not stored
 }
 
 impl BuildStatus {
@@ -36,6 +37,7 @@ impl BuildStatus {
             10 => Some(Self::LogLimitExceeded),
             11 => Some(Self::NarSizeLimitExceeded),
             12 => Some(Self::NotDeterministic),
+            13 => Some(Self::Resolved),
             100 => Some(Self::Busy),
             _ => None,
         }

subprojects/hydra-builder/src/state.rs

@@ -400,11 +400,6 @@ impl State {
 
         let machine_id = self.id;
         let drv = nix_utils::parse_store_path(&m.drv);
-        let resolved_drv = m
-            .resolved_drv
-            .as_ref()
-            .map(|v| nix_utils::parse_store_path(v));
-        let maybe_resolved_drv = resolved_drv.as_ref().unwrap_or(&drv);
 
         let before_import = Instant::now();
         let gcroot_prefix = uuid::Uuid::new_v4().to_string();
@@ -422,7 +417,7 @@ impl State {
             .await;
         let requisites = client
             .fetch_drv_requisites(FetchRequisitesRequest {
-                path: maybe_resolved_drv.to_string().to_owned(),
+                path: drv.to_string().to_owned(),
                 include_outputs: false,
             })
             .await
@@ -435,7 +430,7 @@ impl State {
             store.clone(),
             self.metrics.clone(),
             &gcroot,
-            maybe_resolved_drv,
+            &drv,
             requisites
                 .into_iter()
                 .map(|s| nix_utils::parse_store_path(&s)),
@@ -456,7 +451,7 @@ impl State {
         let before_build = Instant::now();
         let (mut child, stdout, mut stderr) = nix_utils::realise_drv(
             &store,
-            maybe_resolved_drv,
+            &drv,
             &nix_utils::BuildOptions::complete(m.max_log_size, m.max_silent_time, m.build_timeout),
             true,
         )
@@ -516,7 +511,7 @@ impl State {
             .store_dir()
             .parse(&output_raw[0].drv_path)
             .map_err(|e: nix_utils::ParseStorePathError| JobFailure::PostProcessing(e.into()))?;
-        if &actual_out_drv != maybe_resolved_drv {
+        if actual_out_drv != drv {
             return Err(JobFailure::PostProcessing(anyhow::anyhow!(
                 "Nix returned outputs for {actual_out_drv} when we expected {drv}"
             )));

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

@@ -491,22 +491,16 @@ impl Machines {
 pub struct Job {
     pub internal_build_id: uuid::Uuid,
     pub path: nix_utils::StorePath,
-    pub resolved_drv: Option<nix_utils::StorePath>,
     pub build_id: BuildID,
     pub step_nr: i32,
     pub result: RemoteBuild,
 }
 
 impl Job {
-    pub fn new(
-        build_id: BuildID,
-        path: nix_utils::StorePath,
-        resolved_drv: Option<nix_utils::StorePath>,
-    ) -> Self {
+    pub fn new(build_id: BuildID, path: nix_utils::StorePath) -> Self {
         Self {
             internal_build_id: uuid::Uuid::new_v4(),
             path,
-            resolved_drv,
             build_id,
             step_nr: 0,
             result: RemoteBuild::new(),
@@ -538,7 +532,6 @@ pub enum Message {
     BuildMessage {
         build_id: uuid::Uuid,
         drv: nix_utils::StorePath,
-        resolved_drv: Option<nix_utils::StorePath>,
         max_log_size: u64,
         max_silent_time: i32,
         build_timeout: i32,
@@ -560,15 +553,13 @@ impl Message {
             Self::BuildMessage {
                 build_id,
                 drv,
-                resolved_drv,
                 max_log_size,
                 max_silent_time,
                 build_timeout,
                 presigned_url_opts,
             } => runner_request::Message::Build(BuildMessage {
                 build_id: build_id.to_string(),
                 drv: drv.to_string(),
-                resolved_drv: resolved_drv.map(|p| p.to_string()),
                 max_log_size,
                 max_silent_time,
                 build_timeout,
@@ -694,15 +685,15 @@ impl Machine {
     pub async fn build_drv(
         &self,
         job: Job,
+        effective_drv: nix_utils::StorePath,
         opts: &nix_utils::BuildOptions,
         presigned_url_opts: Option<PresignedUrlOpts>,
     ) -> anyhow::Result<()> {
-        let drv = job.path.clone();
+        let drv = effective_drv;
         self.msg_queue
             .send(Message::BuildMessage {
                 build_id: job.internal_build_id,
                 drv,
-                resolved_drv: job.resolved_drv.clone(),
                 max_log_size: opts.get_max_log_size(),
                 max_silent_time: opts.get_max_silent_time(),
                 build_timeout: opts.get_build_timeout(),

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

@@ -317,11 +317,7 @@ impl State {
             build.id
         };
 
-        let mut job = machine::Job::new(
-            build_id,
-            drv.to_owned(),
-            step_info.resolved_drv_path.clone(),
-        );
+        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 {
             job.result.step_status = BuildStatus::CachedFailure;
@@ -363,33 +359,100 @@ impl State {
                         .collect(),
                 )
                 .await?;
+
             tx.commit().await?;
             step_nr
         };
         job.step_nr = step_nr;
 
+        // Try to resolve CA derivation inputs. If resolution yields a
+        // different drv, mark this step as Resolved in the DB and create a new
+        // step for the resolved drv that the builder will actually build.
+        let resolved = if let Some(guard) = step_info.step.get_drv() {
+            let drv_ref = guard.as_ref().unwrap();
+            if let Some(basic_drv) =
+                StepInfo::try_resolve(self.store.store_dir(), &self.db, drv_ref).await
+            {
+                let resolved_path = self.store.write_derivation(&basic_drv).await?;
+                (&resolved_path != drv).then_some(resolved_path)
+            } else {
+                None
+            }
+        } else {
+            None
+        };
+
+        let effective_drv = if let Some(resolved_path) = resolved {
+            tracing::info!("resolved CA derivation {drv} -> {resolved_path}");
+
+            // Finish original step as "resolved".
+            let mut resolved_result = RemoteBuild::new();
+            resolved_result.step_status = BuildStatus::Resolved;
+            resolved_result.set_start_time_now();
+            resolved_result.set_stop_time_now();
+            resolved_result.log_file.clone_from(&job.result.log_file);
+            finish_build_step(
+                &self.db,
+                &self.store,
+                build_id,
+                step_nr,
+                &resolved_result,
+                Some(&machine.hostname),
+                None,
+            )
+            .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?;
+
+                tx.set_resolved_to(build_id, step_nr, build_id, nr).await?;
+                tx.commit().await?;
+                nr
+            };
+            job.step_nr = resolved_step_nr;
+            resolved_path
+        } else {
+            drv.to_owned()
+        };
+
         {
             let mut tx = db.begin_transaction().await?;
             tx.notify_build_started(build_id).await?;
             tx.commit().await?;
         }
         tracing::info!(
-            "Submitting build drv={drv} on machine={} hostname={} build_id={build_id} step_nr={step_nr}",
+            "Submitting build drv={effective_drv} on machine={} hostname={} build_id={build_id} step_nr={}",
             machine.id,
-            machine.hostname
+            machine.hostname,
+            job.step_nr,
         );
         self.db
             .get()
             .await?
             .update_build_step(db::models::UpdateBuildStep {
                 build_id,
-                step_nr,
+                step_nr: job.step_nr,
                 status: db::models::StepStatus::Connecting,
             })
             .await?;
         machine
             .build_drv(
                 job,
+                effective_drv,
                 &build_options,
                 // TODO: cleanup
                 if self.config.use_presigned_uploads() {
@@ -962,7 +1025,7 @@ impl State {
             if r.atomic_state.tries.load(Ordering::Relaxed) > 0 {
                 continue;
             }
-            let step_info = StepInfo::new(&self.store, &self.db, r.clone()).await;
+            let step_info = StepInfo::new(r.clone());
 
             new_queues
                 .entry(system)
@@ -2090,7 +2153,7 @@ impl State {
                 continue;
             };
 
-            let mut job = machine::Job::new(build.id, drv.to_owned(), None);
+            let mut job = machine::Job::new(build.id, drv.to_owned());
             job.result.set_start_and_stop(now);
             job.result.step_status = BuildStatus::Unsupported;
             job.result.error_msg = Some(format!(

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

@@ -525,3 +525,33 @@ impl Steps {
         steps.remove(drv_path);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn drv(name: &str) -> nix_utils::StorePath {
+        nix_utils::parse_store_path(&format!("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa-{name}.drv"))
+    }
+
+    #[test]
+    fn steps_create_and_remove() {
+        let steps = Steps::new();
+        let (step, is_new) = steps.create(&drv("test"), None, None);
+        assert!(is_new);
+        assert_eq!(steps.len(), 1);
+
+        steps.remove(step.get_drv_path());
+        assert_eq!(steps.len(), 0);
+    }
+
+    #[test]
+    fn steps_weak_ref_dies_without_strong_ref() {
+        let steps = Steps::new();
+        let (step, _) = steps.create(&drv("ephemeral"), None, None);
+        assert_eq!(steps.len(), 1);
+
+        drop(step);
+        assert_eq!(steps.len(), 0);
+    }
+}

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

@@ -2,7 +2,6 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
 use db::models::BuildID;
-use nix_utils::BaseStore as _;
 use nix_utils::SingleDerivedPath;
 
 use super::Step;
@@ -37,27 +36,15 @@ fn flatten_chain(
 #[derive(Debug)]
 pub struct StepInfo {
     pub step: Arc<Step>,
-    pub resolved_drv_path: Option<nix_utils::StorePath>,
     already_scheduled: AtomicBool,
     cancelled: AtomicBool,
     pub runnable_since: jiff::Timestamp,
     lowest_share_used: atomic_float::AtomicF64,
 }
 
 impl StepInfo {
-    pub async fn new(store: &nix_utils::LocalStore, db: &db::Database, step: Arc<Step>) -> Self {
+    pub fn new(step: Arc<Step>) -> Self {
         Self {
-            resolved_drv_path: match step.get_drv() {
-                Some(guard) => {
-                    let resolved =
-                        Self::try_resolve(store.store_dir(), db, guard.as_ref().unwrap()).await;
-                    match resolved {
-                        Some(ref basic_drv) => store.write_derivation(basic_drv).await.ok(),
-                        None => None,
-                    }
-                }
-                None => None,
-            },
             already_scheduled: false.into(),
             cancelled: false.into(),
             runnable_since: step.get_runnable_since(),
@@ -77,7 +64,7 @@ impl StepInfo {
     ///
     /// We only need a store dir, not a store, because all the info we need comes from the Hydra
     /// database.
-    async fn try_resolve(
+    pub(super) async fn try_resolve(
         store_dir: &nix_utils::StoreDir,
         db: &db::Database,
         drv: &nix_utils::Derivation,
@@ -297,7 +284,6 @@ mod tests {
 
         StepInfo {
             step,
-            resolved_drv_path: None,
             already_scheduled: false.into(),
             cancelled: false.into(),
             runnable_since: jiff::Timestamp::now(),