make get_input_drvs return relation for dynamic inputs

wip

Add dynamic derivation resolution

Do dynamic derivation resolution, still doesn't splice tree

Revert "Do dynamic derivation resolution, still doesn't splice tree"

This reverts commit 389cc05.

Simpler initial dynamic resolution

dynamic dep prototype, needs testing

Remove dynamic_deps, we can store relation in dynamic_rdeps instead

enable dyn-drv test

Let Claude fix dynamic derivation building in queue runner

This was basically unsupervised, except for me running the test and
feeding it the test logs.

Four bugs prevented dynamic derivations from being built:

- `get_dependents` only walked `rdeps`, not `dynamic_rdeps`, so steps
  connected via dynamic dependencies appeared orphaned and were
  "maybe cancelled" instead of dispatched.

- `succeed_step` used `get_direct_builds` to find the owning build for
  dynamic rdeps processing. Intermediate steps (like `hello.drv.drv`)
  have no direct builds, so we now fall back to `get_dependents` to
  walk the dependency chain.

- `make_rdeps_runnable` only processed `rdeps`, never removing the
  finished step from `dynamic_rdeps` dependents' forward `deps`. This
  left the wrapper permanently stuck with an unsatisfied dependency.

- `resolve_dynamic_input` used `1..outputs.len()` which produces an
  empty range for single-element chains (the common case). Changed to
  `1..=outputs.len()` so the chain is actually resolved and the
  produced `.drv` file is discovered and built as a new step.

Test updated from 2 to 5 expected build steps now that Hydra properly
tracks the full dynamic derivation chain.

Combine `rdeps` and `dynamic_rdeps` into a single `rdeps` list

Both were reverse dependency lists differing only in whether they
carried a `relation` (output name chain for dynamic derivations). Having
two separate lists meant every operation that walked one had to remember
to also walk the other — the exact class of bug fixed in the previous
commit.

Now `StepState::rdeps` is `Vec<ReverseDep>` where an empty `relation`
signifies a regular (non-dynamic) reverse dependency. `get_dependents`,
`make_rdeps_runnable`, and `add_referring_data` all operate on the
single unified list. The `dynamic_rdeps_len` atomic counter,
`clone_dyn_rdeps` method, and `DynamicReverseDep` type are removed in
favor of the renamed `ReverseDep`.

Resolve dynamic derivation chains one level at a time

Instead of resolving the entire chain at once via a recursive SQL query
in `resolve_dynamic_input`, resolve one level per step completion: when
a step finishes, `pop_dynamic_rdeps` pops the next output name from the
relation stack, and `succeed_step` looks it up directly in the build
output — no DB round-trip needed.

The relation vector is now stored in reverse order so that `pop()` gives
the next level to resolve in O(1).

`resolve_dynamic_input` is removed from both `State` and
`db::Connection` since it is no longer needed. The `create_step` input
processing no longer attempts eager resolution either; the dynamic rdeps
mechanism in `succeed_step` handles it naturally as each step completes.

TEMP add some asserts, see failure

Fix CA resolution for intermediate steps, extract `DynDrvRelation` type

Three changes:

- **Fix premature build completion**: `realise_drv_on_valid_machine`
  passed `referring_build: Some(build)` when creating resolved steps for
  ALL CA derivations, not just the toplevel. This caused `succeed_step`
  to mark the entire build as finished when an intermediate resolved
  step completed (e.g. resolved `build-b`), orphaning downstream steps
  (d, e, wrapper). Now `referring_build` is only set when the unresolved
  step was the build's toplevel.

- **`DynDrvRelation` newtype**: Extracts the reversed output-name stack
  into its own type, documenting the reversal invariant in the type name
  and providing `is_empty()`/`pop()` methods. Used throughout `ReverseDep`,
  `add_referring_data`, `create_step`, and `pop_dynamic_rdeps`.

- **`input_drvs` free function**: Walks `SingleDerivedPath` directly
  instead of going through `DerivationInputs::from`. Skips `Opaque`
  (source) inputs — only `Built` derivation dependencies are returned.
  The outermost output name (what we consume) is discarded; intermediate
  names form the `DynDrvRelation`. Called inline in `create_step` before
  `set_drv` consumes the derivation.

Non-trivial dyn-drv test now expects 12 build steps covering the full
chain: makeDerivations → a → b,c (resolve+build) → d (resolve+build) →
e (resolve+build) → wrapper (resolve+build).

Extract `state::drv` module with `OutputNameChain` and `SingleDerivedPath` helpers

Move `OutputNameChain` (renamed from `DynDrvRelation`), `input_drvs`,
and `flatten_chain` into a new `state::drv` module. Add `flatten_path`
as the base case for mutual recursion with `flatten_chain`.

`flatten_path`/`flatten_chain` now return `OutputNameChain` directly in
stack order (outermost first), eliminating the separate reverse step.
`input_drvs` calls `flatten_path` on the inner `drv_path` of each
`Built` input, naturally discarding the outermost output name.

`flatten_chain` (previously in `step_info.rs`) and `input_drvs`
(previously in `step.rs`) are unified in one place. The SQL caller in
`step_info.rs` reverses to forward order where needed.

Author: artemist

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

@@ -510,7 +510,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);
                     }
@@ -1377,6 +1377,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
@@ -1922,7 +1990,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>>>>,
@@ -1936,9 +2004,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);
         }
@@ -2034,6 +2106,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 {
@@ -2091,35 +2164,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

@@ -39,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;