queue-runner: resolve dynamic derivation dependencies

Reverse dependencies now carry an `OutputNameChain` (empty for regular
deps, non-empty for dynamic ones), stored in a unified `rdeps` list
via `ReverseDep`. This replaces the previous `get_input_drvs` method
with `input_drvs`, which walks `SingleDerivedPath` directly, skipping
`Opaque` (source) inputs and returning only `Built` derivation
dependencies with their relation chain.

When a step completes and has dynamic reverse dependencies (`rdeps` with a
non-empty `OutputNameChain`), pop one level from the relation, look up
the produced output, and create a new step for the resulting derivation.
This resolves dynamic derivation chains one level at a time as each step
finishes, with no extra DB round-trip needed.

The dyn-drv tests are re-enabled and updated to verify the full build
step chains: 4 steps for simple dynamic derivations, 12 for the
non-trivial DAG case.

Co-Authored-By: Artemis Tosini <artemis.tosini@obsidian.systems>
Co-Authored-By: John Ericson <John.Ericson@Obsidian.Systems>
Co-Authored-By: Amaan Qureshi <git@amaanq.com>

Author: 3 people

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

@@ -1,3 +1,5 @@
+use std::collections::BTreeSet;
+
 use nix_utils::SingleDerivedPath;
 
 /// Output names of intermediate derivations for a dynamic derivation
@@ -42,3 +44,21 @@ pub fn flatten_chain(
     chain.0.push(output_name.clone());
     (root, chain)
 }
+
+/// Extract `Built` input dependencies from a derivation.
+///
+/// Returns `(root_drv_path, relation)` pairs. `Opaque` (source) inputs are
+/// skipped — only derivation build dependencies are returned. For each
+/// `Built` input, the outermost output name (what we consume) is discarded;
+/// intermediate output names form the [`OutputNameChain`].
+pub fn input_drvs(
+    drv: &nix_utils::Derivation,
+) -> BTreeSet<(nix_utils::StorePath, OutputNameChain)> {
+    drv.inputs
+        .iter()
+        .filter_map(|sdp| match sdp {
+            SingleDerivedPath::Opaque(_) => None,
+            SingleDerivedPath::Built { drv_path, .. } => Some(flatten_path(drv_path)),
+        })
+        .collect()
+}

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

@@ -521,7 +521,7 @@ impl State {
                 // original step's drv path differs from the resolved one, so
                 // completing the resolved step wouldn't clear the dep).
                 for rdep in step_info.step.clone_rdeps() {
-                    if let Some(rdep) = rdep.upgrade() {
+                    if let Some(rdep) = rdep.step.upgrade() {
                         rdep.remove_dep(&step_info.step);
                         resolved_step.make_rdep(&rdep);
                     }
@@ -1388,6 +1388,74 @@ impl State {
             tx.commit().await?;
         }
 
+        // Process dynamic rdeps first, as we must add new step dependencies for dynamically
+        // generated derivations
+        {
+            for (dep_step, output_name, relation) in item.step_info.step.pop_dynamic_rdeps() {
+                let Some(dependent_step) = dep_step.upgrade() else {
+                    continue;
+                };
+
+                let resolved_drv = output.outputs.get(&output_name).cloned().ok_or_else(|| {
+                    anyhow::anyhow!(
+                        "Dynamic rdep references output `{output_name}` not produced by {drv_path}"
+                    )
+                })?;
+
+                // Find a build associated with this step. For intermediate steps
+                // (not top-level), `direct` is empty, so we walk the dependency
+                // chain via `get_dependents` to find the owning build.
+                let build = if let Some(b) = direct.get(0) {
+                    b.clone()
+                } else {
+                    let mut dependents = HashSet::new();
+                    let mut visited_steps = HashSet::new();
+                    item.step_info
+                        .step
+                        .get_dependents(&mut dependents, &mut visited_steps);
+                    let Some(b) = dependents.into_iter().next() else {
+                        tracing::warn!("Finished step does not have associated build");
+                        continue;
+                    };
+                    b
+                };
+
+                // Create the actual step for the new derivation.
+                // finished_drvs is not necessary as it is only a memoization table to reduce
+                // checks if a dependency is finished from the database.
+                // new_steps is not necessary either as
+                let new_runnable: Arc<parking_lot::RwLock<HashSet<Arc<Step>>>> = Default::default();
+                let new_step = match self
+                    .create_step(
+                        build.clone(),
+                        resolved_drv.clone(),
+                        None,
+                        Some((dependent_step.clone(), relation)),
+                        Default::default(),
+                        Default::default(),
+                        new_runnable.clone(),
+                    )
+                    .await
+                {
+                    CreateStepResult::None => continue,
+                    CreateStepResult::Valid(step) => step,
+                    CreateStepResult::PreviousFailure(step) => {
+                        if let Err(e) = self.handle_previous_failure(build.clone(), step).await {
+                            tracing::error!("Failed to handle previous failure: {e}");
+                        }
+                        // TODO: figure out what to do here
+                        continue;
+                    }
+                };
+
+                for r in new_runnable.read().iter() {
+                    r.make_runnable();
+                }
+
+                // create_step already added rdeps, but we need to add a forward dep as well
+                dependent_step.add_dep(new_step);
+            }
+        }
         item.step_info.step.make_rdeps_runnable();
 
         // always trigger dispatch, as we now might have a free machine again
@@ -1933,7 +2001,7 @@ impl State {
         build: Arc<Build>,
         drv_path: nix_utils::StorePath,
         referring_build: Option<Arc<Build>>,
-        referring_step: Option<Arc<Step>>,
+        referring_step: Option<(Arc<Step>, drv::OutputNameChain)>,
         finished_drvs: Arc<parking_lot::RwLock<HashSet<nix_utils::StorePath>>>,
         new_steps: Arc<parking_lot::RwLock<HashSet<Arc<Step>>>>,
         new_runnable: Arc<parking_lot::RwLock<HashSet<Arc<Step>>>>,
@@ -1947,9 +2015,13 @@ impl State {
             }
         }
 
-        let (step, is_new) =
-            self.steps
-                .create(&drv_path, referring_build.as_ref(), referring_step.as_ref());
+        let (step, is_new) = self.steps.create(
+            &drv_path,
+            referring_build.as_ref(),
+            referring_step
+                .as_ref()
+                .map(|(step, relation)| (step, relation.clone())),
+        );
         if !is_new {
             return CreateStepResult::Valid(step);
         }
@@ -2045,6 +2117,7 @@ impl State {
             self.store.query_missing_outputs(output_paths).await
         };
 
+        let input_drvs = drv::input_drvs(&drv);
         step.set_drv(drv);
 
         if self.check_cached_failure(step.clone()).await {
@@ -2102,35 +2175,32 @@ impl State {
         }
 
         tracing::debug!("creating build step '{drv_path}");
-        let Some(input_drvs) = step.get_input_drvs() else {
-            // this should never happen because we always a a drv set at this point in time
-            return CreateStepResult::None;
-        };
 
         let step2 = step.clone();
-        let mut stream = futures::StreamExt::map(tokio_stream::iter(input_drvs), |i| {
-            let build = build.clone();
-            let step = step2.clone();
-            let finished_drvs = finished_drvs.clone();
-            let new_steps = new_steps.clone();
-            let new_runnable = new_runnable.clone();
-            async move {
-                Box::pin(self.create_step(
-                    // conn,
-                    build,
-                    i,
-                    None,
-                    Some(step),
-                    finished_drvs,
-                    new_steps,
-                    new_runnable,
-                ))
-                .await
-            }
-        })
-        .buffered(25);
-        while let Some(v) = tokio_stream::StreamExt::next(&mut stream).await {
-            match v {
+        let mut stream =
+            futures::StreamExt::map(tokio_stream::iter(input_drvs), |(input_path, relation)| {
+                let build = build.clone();
+                let step = step2.clone();
+                let finished_drvs = finished_drvs.clone();
+                let new_steps = new_steps.clone();
+                let new_runnable = new_runnable.clone();
+
+                async move {
+                    Box::pin(self.create_step(
+                        build,
+                        input_path,
+                        None,
+                        Some((step, relation)),
+                        finished_drvs,
+                        new_steps,
+                        new_runnable,
+                    ))
+                    .await
+                }
+            })
+            .buffered(25);
+        while let Some(result) = tokio_stream::StreamExt::next(&mut stream).await {
+            match result {
                 CreateStepResult::None => (),
                 CreateStepResult::Valid(dep) => {
                     if !dep.get_finished() && !dep.get_previous_failure() {

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

@@ -8,6 +8,15 @@ use hashbrown::{HashMap, HashSet};
 use super::{Build, Jobset};
 use db::models::BuildID;
 
+use super::drv::OutputNameChain;
+
+#[derive(Debug, Clone)]
+pub struct ReverseDep {
+    /// The step that depends on us
+    pub step: Weak<Step>,
+    pub relation: OutputNameChain,
+}
+
 #[derive(Debug)]
 pub struct StepAtomicState {
     /// Whether the step has finished initialisation.
@@ -66,10 +75,11 @@ impl StepAtomicState {
 
 #[derive(Debug)]
 pub(super) struct StepState {
-    /// The build steps on which this step depends.
+    /// The resolved build steps on which this step depends
     deps: HashSet<Arc<Step>>,
     /// The build steps that depend on this step.
-    rdeps: Vec<Weak<Step>>,
+    /// An empty `relation` signifies a regular (non-dynamic) reverse dependency.
+    rdeps: Vec<ReverseDep>,
     /// Builds that have this step as the top-level derivation.
     builds: Vec<Weak<Build>>,
     /// Jobsets to which this step belongs. Used for determining scheduling priority.
@@ -186,16 +196,6 @@ impl Step {
         })
     }
 
-    pub fn get_input_drvs(&self) -> Option<Vec<nix_utils::StorePath>> {
-        let drv = self.drv.load_full();
-        drv.as_ref().map(|drv| {
-            harmonia_store_core::derivation::DerivationInputs::from(&drv.inputs)
-                .drvs
-                .into_keys()
-                .collect::<Vec<_>>()
-        })
-    }
-
     pub fn get_after(&self) -> jiff::Timestamp {
         self.atomic_state.after.load()
     }
@@ -270,7 +270,9 @@ impl Step {
         };
 
         for rdep in rdeps {
-            let Some(rdep) = rdep.upgrade() else { continue };
+            let Some(rdep) = rdep.step.upgrade() else {
+                continue;
+            };
             rdep.get_dependents(builds, steps);
         }
     }
@@ -286,7 +288,7 @@ impl Step {
 
         let mut state = self.state.write();
         state.rdeps.retain(|rdep| {
-            let Some(rdep) = rdep.upgrade() else {
+            let Some(rdep) = rdep.step.upgrade() else {
                 return false;
             };
 
@@ -374,21 +376,45 @@ impl Step {
     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));
+        state.rdeps.push(ReverseDep {
+            step: Arc::downgrade(dep),
+            relation: OutputNameChain::default(),
+        });
         self.atomic_state
             .rdeps_len
             .store(state.rdeps.len() as u64, Ordering::Relaxed);
     }
 
-    pub fn clone_rdeps(&self) -> Vec<Weak<Step>> {
+    pub fn clone_rdeps(&self) -> Vec<ReverseDep> {
         let state = self.state.read();
         state.rdeps.clone()
     }
 
+    /// Pop one level of dynamic indirection from each dynamic rdep,
+    /// returning `(dependent_step, popped_output_name, remaining_relation)` triples.
+    ///
+    /// The rdep entries remain in the list (with shortened relations) so that
+    /// `make_rdeps_runnable` can still clean up forward deps.
+    ///
+    /// We collect into a `Vec` rather than returning an iterator because the
+    /// write lock on the step's state must be released before the caller can
+    /// do async work (e.g. `create_step`) with the results.
+    pub fn pop_dynamic_rdeps(&self) -> Vec<(Weak<Step>, nix_utils::OutputName, OutputNameChain)> {
+        let mut state = self.state.write();
+        state
+            .rdeps
+            .iter_mut()
+            .filter_map(|rdep| {
+                let output_name = rdep.relation.pop()?;
+                Some((rdep.step.clone(), output_name, rdep.relation.clone()))
+            })
+            .collect()
+    }
+
     pub fn add_referring_data(
         &self,
         referring_build: Option<&Arc<Build>>,
-        referring_step: Option<&Arc<Self>>,
+        referring_step: Option<(&Arc<Self>, OutputNameChain)>,
     ) {
         if referring_build.is_none() && referring_step.is_none() {
             return;
@@ -398,8 +424,11 @@ impl Step {
         if let Some(referring_build) = referring_build {
             state.builds.push(Arc::downgrade(referring_build));
         }
-        if let Some(referring_step) = referring_step {
-            state.rdeps.push(Arc::downgrade(referring_step));
+        if let Some((referring_step, relation)) = referring_step {
+            state.rdeps.push(ReverseDep {
+                step: Arc::downgrade(referring_step),
+                relation,
+            });
             self.atomic_state
                 .rdeps_len
                 .store(state.rdeps.len() as u64, Ordering::Relaxed);
@@ -531,7 +560,7 @@ impl Steps {
         &self,
         drv_path: &nix_utils::StorePath,
         referring_build: Option<&Arc<Build>>,
-        referring_step: Option<&Arc<Step>>,
+        referring_step: Option<(&Arc<Step>, OutputNameChain)>,
     ) -> (Arc<Step>, bool) {
         let mut is_new = false;
         let mut steps = self.inner.write();

subprojects/hydra-tests/content-addressed/dyn-drv-non-trivial.t

@@ -4,10 +4,6 @@ use warnings;
 use Setup;
 use Test2::V0;
 
-# FIXME now that we're properly resolving things in Hydra rather than Nix,
-# dynamic derivations stopped-fake working
-plan skip_all => 'dynamic derivation resolution not yet implemented';
-
 # Adapted from https://github.com/NixOS/nix/blob/master/tests/functional/dyn-drv/non-trivial.nix
 #
 # A single derivation uses recursive-nix to dynamically create a DAG of
@@ -43,6 +39,38 @@ if ($wrapper) {
     $wrapper->discard_changes;
     is($wrapper->finished, 1, "wrapper should be finished");
     is($wrapper->buildstatus, 0, "wrapper should succeed");
+
+    # Full dynamic derivation chain: 12 steps total
+    # 1.  make-derivations.drv.drv  (status=0,  build makeDerivations)
+    # 2.  build-a.drv               (status=0,  build a)
+    # 3.  build-c.drv               (status=13, resolve c)
+    # 4.  build-b.drv               (status=13, resolve b)
+    # 5.  build-b.drv               (status=0,  build resolved b)
+    # 6.  build-c.drv               (status=0,  build resolved c)
+    # 7.  build-d.drv               (status=13, resolve d)
+    # 8.  build-d.drv               (status=0,  build resolved d)
+    # 9.  make-derivations.drv      (status=13, resolve e — named after makeDerivations output)
+    # 10. make-derivations.drv      (status=0,  build resolved e)
+    # 11. wrapper.drv               (status=13, resolve wrapper)
+    # 12. wrapper.drv               (status=0,  build resolved wrapper)
+    my @steps = $wrapper->buildsteps->search({}, { order_by => 'stepnr' })->all;
+    is(scalar @steps, 12, "wrapper should have 12 build steps");
+
+    # Check that derivations a-d each got a successful (status=0) build step.
+    # build-e is named make-derivations.drv (the output of makeDerivations),
+    # so we check for it separately.
+    my @built = sort map {
+        my $drv = $_->drvpath // "";
+        (defined $_->status && $_->status == 0 && $drv =~ m{-build-([a-d])\.drv$}) ? $1 : ()
+    } @steps;
+    is(\@built, [qw(a b c d)], "derivations a-d should each have a successful build step");
+
+    # build-e is the make-derivations.drv step (status=0, not the .drv.drv)
+    my @build_e = grep {
+        my $drv = $_->drvpath // "";
+        defined $_->status && $_->status == 0 && $drv =~ m{-make-derivations\.drv$}
+    } @steps;
+    is(scalar @build_e, 1, "build-e (make-derivations.drv) should have a successful build step");
 }
 
 done_testing;