Merge pull request #1629 from obsidiansystems/ca-resolve-two-phase

queue-runner: resolve CA derivations after dependency outputs land

Author: Ericson2314

subprojects/crates/db/src/connection.rs

@@ -693,6 +693,36 @@ impl Transaction<'_> {
         Ok(())
     }
 
+    #[tracing::instrument(skip(self), err)]
+    pub async fn find_build_step_outputs(
+        &mut self,
+        store_dir: &StoreDir,
+        drv_path: &StorePath,
+    ) -> sqlx::Result<BTreeMap<OutputName, StorePath>> {
+        let drv_path = store_dir.display(drv_path).to_string();
+        let items: Vec<(String, String)> = sqlx::query_as(
+            r"SELECT o.name, o.path
+              FROM buildstepoutputs o
+              JOIN buildsteps s ON s.build = o.build AND s.stepnr = o.stepnr
+              WHERE s.drvpath = $1 AND o.path IS NOT NULL",
+        )
+        .bind(drv_path)
+        .fetch_all(&mut *self.tx)
+        .await?;
+
+        items
+            .into_iter()
+            .map(|(name, path)| -> anyhow::Result<_> {
+                let name: OutputName = name.parse().context("invalid output name from DB")?;
+                let path: StorePath = store_dir
+                    .parse(&path)
+                    .context("invalid store path from DB")?;
+                Ok((name, path))
+            })
+            .collect::<anyhow::Result<_>>()
+            .map_err(|e| sqlx::Error::Decode(e.into_boxed_dyn_error()))
+    }
+
     #[tracing::instrument(skip(self, res), err)]
     pub async fn update_build_step_in_finish(
         &mut self,
@@ -941,6 +971,60 @@ impl Transaction<'_> {
         Ok(step_nr)
     }
 
+    #[tracing::instrument(
+        skip(self, start_time, stop_time, build_id, drv_path, outputs,),
+        err,
+        ret
+    )]
+    pub async fn create_local_step(
+        &mut self,
+        store_dir: &StoreDir,
+        start_time: i32,
+        stop_time: i32,
+        build_id: crate::models::BuildID,
+        drv_path: &StorePath,
+        outputs: BTreeMap<OutputName, StorePath>,
+    ) -> anyhow::Result<i32> {
+        let step_nr = loop {
+            if let Some(step_nr) = self
+                .insert_build_step(
+                    store_dir,
+                    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 {
+                build_id,
+                step_nr,
+                name,
+                path: Some(path),
+            })
+            .collect();
+
+        self.insert_build_step_outputs(store_dir, &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

@@ -22,6 +22,8 @@ pub enum BuildStatus {
     LogLimitExceeded = 10,
     NarSizeLimitExceeded = 11,
     NotDeterministic = 12,
+    /// step was resolved to a CA derivation
+    Resolved = 13,
     /// not stored
     Busy = 100,
 }
@@ -42,6 +44,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,
         )
@@ -536,7 +534,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/drv.rs

@@ -0,0 +1,44 @@
+use nix_utils::SingleDerivedPath;
+
+/// Output names of intermediate derivations for a dynamic derivation
+/// dependency, stored in reverse order so that the next level to resolve
+/// can be cheaply `pop()`-ed.
+///
+/// e.g. for a derivation input that is `aaaa-dyn.drv^foo^bar^out`, the final
+/// derivation would be `aaaa-dyn.drv^foo^bar` (no final `^out`). The output
+/// names are stored as `["bar", "foo"]` (reversed).
+///
+/// In the common case of depending on a static derivation, this is empty.
+#[derive(Debug, Clone, Default, PartialEq, Eq, PartialOrd, Ord)]
+pub struct OutputNameChain(pub Vec<nix_utils::OutputName>);
+
+impl OutputNameChain {
+    pub fn pop(&mut self) -> Option<nix_utils::OutputName> {
+        self.0.pop()
+    }
+}
+
+/// Flatten a [`SingleDerivedPath`] into `(root_drv_path, chain)`.
+///
+/// The output chain is in stack order (outermost first) matching
+/// [`OutputNameChain`]'s convention. For `Built { Opaque(A), "foo" }`,
+/// returns `(A, ["foo"])`. For `Opaque(A)`, returns `(A, [])`.
+pub fn flatten_path(sdp: &SingleDerivedPath) -> (nix_utils::StorePath, OutputNameChain) {
+    match sdp {
+        SingleDerivedPath::Opaque(p) => (p.clone(), OutputNameChain::default()),
+        SingleDerivedPath::Built { drv_path, output } => flatten_chain(drv_path, output),
+    }
+}
+
+/// Like [`flatten_path`] but appends an additional output name.
+///
+/// For `Built { Opaque(A), "foo" }` with output `"bar"`,
+/// returns `(A, ["bar", "foo"])`.
+pub fn flatten_chain(
+    drv_path: &SingleDerivedPath,
+    output_name: &nix_utils::OutputName,
+) -> (nix_utils::StorePath, OutputNameChain) {
+    let (root, mut chain) = flatten_path(drv_path);
+    chain.0.push(output_name.clone());
+    (root, chain)
+}

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,
@@ -696,15 +687,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(),