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.

We create a new Step for that resolution and bunt it back to the
scheduler. This grants 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.

In order to make tests more consistent, CA derivations will fail if they
cannot be fully resolved. Otherwise, there could be inconsistent
successes depending on which builder a step was performed on.

Author: amaanq

subprojects/crates/db/src/connection.rs

@@ -338,12 +338,14 @@ impl Connection {
                 CROSS JOIN LATERAL (
                     SELECT o.path
                     FROM buildsteps s
+                    LEFT JOIN buildsteps sr
+                        ON s.build = sr.build AND s.resolvedToStep = sr.stepnr
                     JOIN buildstepoutputs o
-                        ON s.build = o.build AND s.stepnr = o.stepnr
+                        ON s.build = o.build AND (s.stepnr = o.stepnr OR sr.stepnr = o.stepnr)
                     WHERE s.drvPath = r.drv_path
                       AND o.name = i.chain[r.step]
                       AND o.path IS NOT NULL
-                      AND s.status = 0
+                      AND (s.status = 0 OR (s.status = 13 AND sr.status = 0))
                     ORDER BY s.build DESC
                     LIMIT 1
                 ) sub
@@ -879,6 +881,30 @@ 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_step_nr: i32,
+    ) -> sqlx::Result<()> {
+        sqlx::query(
+            r"
+              UPDATE buildsteps
+              SET resolvedToStep = $3
+              WHERE build = $1 AND stepnr = $2
+            ",
+        )
+        .bind(origin_build_id)
+        .bind(origin_step_nr)
+        .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

@@ -19,7 +19,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 {
@@ -38,6 +39,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

@@ -408,8 +408,6 @@ impl State {
             .drv
             .ok_or(JobFailure::Preparing(anyhow::anyhow!("missing drv")))?
             .0;
-        let resolved_drv = m.resolved_drv.map(|v| v.0);
-        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();
@@ -427,7 +425,7 @@ impl State {
             .await;
         let requisites = client
             .fetch_drv_requisites(FetchRequisitesRequest {
-                path: Some(ProtoStorePath::from(maybe_resolved_drv.clone())),
+                path: Some(ProtoStorePath::from(drv.clone())),
                 include_outputs: false,
             })
             .await
@@ -440,7 +438,7 @@ impl State {
             store.clone(),
             self.metrics.clone(),
             &gcroot,
-            maybe_resolved_drv,
+            &drv,
             requisites.into_iter().map(|s| s.0),
             usize::try_from(self.max_concurrent_downloads.load(Ordering::Relaxed)).unwrap_or(5),
             self.config.use_substitutes,
@@ -459,7 +457,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,
         )
@@ -519,7 +517,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/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/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: Some(shared::proto::ProtoStorePath::from(drv)),
-                resolved_drv: resolved_drv.map(shared::proto::ProtoStorePath::from),
                 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(),