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.

As part of this, add local outputs to resolution table

With the current queue-runner design, all dependency outputs of a
CAFloating derivation must be recorded in the hydra database.

This is true for things built or substituted by hydra, but until now not
by things found on the local nix store. This may occur for outputs that
are part of the system configuration.

Therefore, add all local outputs that are not already in the database to
the resolution table upon creating a step. This makes it possible to
build derivations from `contentAddressedByDefault` nixpkgs.

Author: amaanq

subprojects/crates/db/src/connection.rs

@@ -1,3 +1,5 @@
+use std::collections::BTreeMap;
+
 use sqlx::Acquire;
 
 use harmonia_store_core::store_path::StoreDir;
@@ -338,12 +340,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
@@ -648,6 +652,16 @@ impl Transaction<'_> {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self), err)]
+    pub async fn find_build_step_outputs(
+        &mut self,
+        drv_path: &str,
+    ) -> sqlx::Result<BTreeMap<String, String>> {
+        let items: Vec<(String, String)> = sqlx::query_as("SELECT o.name, o.path FROM buildstepoutputs o JOIN buildsteps s ON s.stepnr = o.stepnr WHERE s.drvpath = ? AND o.path IS NOT NULL").bind(drv_path).fetch_all(&mut *self.tx).await?;
+
+        Ok(items.into_iter().collect())
+    }
+
     #[tracing::instrument(skip(self, res), err)]
     pub async fn update_build_step_in_finish(
         &mut self,
@@ -879,6 +893,79 @@ 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, outputs,),
+        err,
+        ret
+    )]
+    pub async fn create_local_step(
+        &mut self,
+        start_time: i32,
+        stop_time: i32,
+        build_id: crate::models::BuildID,
+        drv_path: &str,
+        outputs: BTreeMap<String, String>,
+    ) -> anyhow::Result<i32> {
+        let step_nr = loop {
+            if let Some(step_nr) = self
+                .insert_build_step(InsertBuildStep {
+                    build_id,
+                    r#type: crate::models::BuildType::Substitution,
+                    drv_path,
+                    status: BuildStatus::Success,
+                    busy: false,
+                    start_time: Some(start_time),
+                    stop_time: Some(stop_time),
+                    platform: None,
+                    propagated_from: None,
+                    error_msg: None,
+                    machine: "",
+                })
+                .await?
+            {
+                break step_nr;
+            }
+        };
+
+        let output_items: Vec<_> = outputs
+            .into_iter()
+            .map(|(name, path)| InsertBuildStepOutput::<String> {
+                build_id,
+                step_nr,
+                name,
+                path: Some(path),
+            })
+            .collect();
+
+        self.insert_build_step_outputs(&output_items).await?;
+
+        Ok(step_nr)
+    }
+
     #[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(),