Fix two-phase CA derivation resolution and StorePath type safety

Fix several issues with the `ca-resolve-two-phase` branch:

- Fix `find_build_step_outputs` SQL: `?` → `$1` (PostgreSQL), add
  `s.build = o.build` join condition

- Fix dep graph propagation: when resolving a CA derivation, remove
  the original step from each dependent's deps and replace with the
  resolved step, so downstream steps become runnable

- Skip two-phase resolution for CA floating drvs with only `Opaque`
  inputs (no `Built` deps) — these are already resolved

- Replace `resolvedToStep` (integer FK) with `resolvedDrvPath` (text
  basename) so output lookups work across builds by drv path

  So it turns out that only scheduled or finishes steps go in the DB
  with the current desing. Steps which totally pending (blocked on other
  steps, for example) just live in memory. So we won't get our step
  number at resolution time to use as a foreign key.

  Instead, we can just write down the drv path, which is enough for the
  in-memory step, and only when that step is processsed will something
  end up in the data base.

- Remove eager `create_build_step` for resolved steps — the DB entry
  is created when the step is actually dispatched

  very much corresponds to the above

- Skip dyn-drv integration test (dynamic derivation resolution not
  yet implemented in Hydra)

  Our passing it was very much a lie before. Our failing it now is more
  honest.

- Add unit tests for `resolve_drv_output_chains` with resolved steps

Also, did this general cleanup which should be pulled out of this commit
and landed on master independently, first.

- Refactor `db` crate to use `StorePath` and `OutputName` types instead
  of raw `&str` for drv paths, output paths, and output names.

  We want to make the core logic all using nice types, and get away from
  not-nice types as soon as possible after getting out the database.
  This does that.

- Add `store_dir: &StoreDir` parameter to db functions that need to
  render `StorePath` to full paths for legacy DB columns

Author: Ericson2314

subprojects/crates/db/src/connection.rs

@@ -1,5 +1,5 @@
 use harmonia_store_core::derived_path::OutputName;
-use harmonia_store_core::store_path::{ParseStorePathError, StoreDir};
+use harmonia_store_core::store_path::{ParseStorePathError, StoreDir, StorePath};
 use hashbrown::HashMap;
 
 pub type BuildID = i32;
@@ -132,7 +132,7 @@ pub struct UpdateBuild<'a> {
 pub struct InsertBuildStep<'a> {
     pub build_id: BuildID,
     pub r#type: BuildType,
-    pub drv_path: &'a str,
+    pub drv_path: &'a StorePath,
     pub status: BuildStatus,
     pub busy: bool,
     pub start_time: Option<i32>,
@@ -147,7 +147,7 @@ pub struct InsertBuildStep<'a> {
 pub struct InsertBuildStepOutput<StorePath = harmonia_store_core::store_path::StorePath> {
     pub build_id: BuildID,
     pub step_nr: i32,
-    pub name: String,
+    pub name: OutputName,
     pub path: Option<StorePath>,
 }
 

subprojects/crates/db/src/models.rs

@@ -20,7 +20,6 @@ pub use step::{Step, Steps};
 pub use step_info::StepInfo;
 
 use std::collections::BTreeMap;
-use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicI64, Ordering};
 use std::time::Instant;
@@ -338,9 +337,10 @@ impl State {
 
             let step_nr = tx
                 .create_build_step(
+                    self.store.store_dir(),
                     Some(job.result.get_start_time_as_i32()?),
                     build_id,
-                    &self.store.print_store_path(step_info.step.get_drv_path()),
+                    step_info.step.get_drv_path(),
                     step_info.step.get_system().as_deref(),
                     machine.hostname.clone(),
                     BuildStatus::Busy,
@@ -351,12 +351,6 @@ impl State {
                         .get_output_paths(self.store.store_dir())
                         .unwrap_or_default()
                         .into_iter()
-                        .map(|(name, path)| {
-                            (
-                                name.to_string(),
-                                path.map(|p| self.store.print_store_path(&p)),
-                            )
-                        })
                         .collect(),
                 )
                 .await?;
@@ -376,59 +370,58 @@ impl State {
             };
             let drv_ref = guard.as_ref().unwrap();
 
-            let is_ca_floating = drv_ref
-                .outputs
-                .iter()
-                .all(|output| matches!(output.1, nix_utils::DerivationOutput::CAFloating(_)));
-
-            let needs_resolution = is_ca_floating
-                && drv_ref.inputs.iter().any(|input| {
-                    matches!(
-                        input,
-                        nix_utils::SingleDerivedPath::Built {
-                            drv_path: _,
-                            output: _
-                        }
-                    )
-                });
-
-            if needs_resolution {
-                let resolved_path = if let Some(basic_drv) =
-                    StepInfo::try_resolve(self.store.store_dir(), &self.db, drv_ref).await
-                {
-                    let resolved_path = self.store.write_derivation(&basic_drv).await?;
-                    (&resolved_path != drv).then_some(resolved_path)
-                } else {
-                    None
-                };
+            // `Some(path)` if this CA derivation was resolved to a
+            // different drv; `None` if resolution is not needed or is
+            // a no-op (same drv path).
+            let resolved_path = (|| async {
+                // Only CA floating derivations need resolution, and only
+                // when they have `Built` inputs (dynamic derivation deps).
+                // `Opaque`-only inputs are already resolved.
+                let is_ca_with_built_inputs =
+                    drv_ref.outputs.iter().all(|output| {
+                        matches!(output.1, nix_utils::DerivationOutput::CAFloating(_))
+                    }) && drv_ref
+                        .inputs
+                        .iter()
+                        .any(|input| matches!(input, nix_utils::SingleDerivedPath::Built { .. }));
+                tracing::info!(
+                    is_ca_with_built_inputs,
+                    ca_floating = drv_ref
+                        .outputs
+                        .iter()
+                        .all(|o| matches!(o.1, nix_utils::DerivationOutput::CAFloating(_))),
+                    has_built_inputs = drv_ref
+                        .inputs
+                        .iter()
+                        .any(|i| matches!(i, nix_utils::SingleDerivedPath::Built { .. })),
+                    "resolution check for {drv}"
+                );
+                if !is_ca_with_built_inputs {
+                    return Ok::<_, anyhow::Error>(None);
+                }
 
-                // If we try to build an unresolved CA derivation it might work, but it might not.
-                // Make sure it always fails for consistency.
-                let Some(resolved_path) = resolved_path else {
-                    return Err(anyhow::anyhow!(
-                        "Failed to resolve CAFloating derivation {drv}"
-                    ));
-                };
+                // Resolve `Built` input placeholders to concrete store
+                // paths using outputs recorded in the DB.
+                let basic_drv = StepInfo::try_resolve(self.store.store_dir(), &self.db, drv_ref)
+                    .await
+                    .ok_or_else(|| {
+                        anyhow::anyhow!("Failed to resolve CAFloating derivation {drv}")
+                    })?;
+                let resolved_path = self.store.write_derivation(&basic_drv).await?;
+                // If resolution changed the drv, we need a two-phase
+                // build; otherwise just build the original directly.
+                Ok((&resolved_path != drv).then_some(resolved_path))
+            })()
+            .await?;
 
+            if let Some(resolved_path) = resolved_path {
                 tracing::info!("resolved CA derivation {drv} -> {resolved_path}");
 
-                // Create DB step for the resolved drv under the same build.
+                // Record the resolved drv path on the original step so
+                // future output lookups can follow the chain.
                 let mut tx = db.begin_transaction().await?;
-                let nr = tx
-                    .create_build_step(
-                        Some(job.result.get_start_time_as_i32()?),
-                        build_id,
-                        &self.store.print_store_path(&resolved_path),
-                        step_info.step.get_system().as_deref(),
-                        machine.hostname.clone(),
-                        BuildStatus::Busy,
-                        None,
-                        None,
-                        vec![],
-                    )
+                tx.set_resolved_to(build_id, step_nr, &resolved_path)
                     .await?;
-
-                tx.set_resolved_to(build_id, step_nr, nr).await?;
                 tx.commit().await?;
 
                 // Finish original step as "resolved" in the DB and in-memory
@@ -499,11 +492,15 @@ impl State {
                 // dependencies and, if it was the root derivation, as the new root
                 // derivation of the build.
 
-                // Copy rdeps so the resolved step is kept around.
-                // This also prevents reverse dependencies from being kept
-                // around for too long.
+                // Replace the original step with the resolved step in the
+                // dependency graph.  Each step that depended on the original
+                // (unresolved) step must now depend on the resolved step
+                // instead, otherwise it will never become runnable (the
+                // 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() {
+                        rdep.remove_dep(&step_info.step);
                         resolved_step.make_rdep(&rdep);
                     }
                 }
@@ -717,8 +714,9 @@ impl State {
             .await?
             .ok_or_else(|| anyhow::anyhow!("drv not found"))?;
         db.insert_debug_build(
+            self.store.store_dir(),
             jobset_id,
-            &self.store.print_store_path(drv_path),
+            drv_path,
             std::str::from_utf8(&drv.platform).expect("platform must be valid UTF-8"),
         )
         .await?;
@@ -1560,9 +1558,10 @@ impl State {
                     }
 
                     tx.create_build_step(
+                        self.store.store_dir(),
                         None,
                         b.id,
-                        &self.store.print_store_path(step.get_drv_path()),
+                        step.get_drv_path(),
                         step.get_system().as_deref(),
                         machine
                             .as_deref()
@@ -1578,12 +1577,6 @@ impl State {
                         step.get_output_paths(self.store.store_dir())
                             .unwrap_or_default()
                             .into_iter()
-                            .map(|(name, path)| {
-                                (
-                                    name.to_string(),
-                                    path.map(|p| self.store.print_store_path(&p)),
-                                )
-                            })
                             .collect(),
                     )
                     .await?;
@@ -1622,7 +1615,7 @@ impl State {
                         .unwrap_or_default()
                     {
                         if let Some(path) = path {
-                            tx.insert_failed_paths(&self.store.print_store_path(&path))
+                            tx.insert_failed_paths(self.store.store_dir(), &path)
                                 .await?;
                         }
                     }
@@ -1698,7 +1691,7 @@ impl State {
         // Find the previous build step record, first by derivation path, then by output
         // path.
         let mut propagated_from = tx
-            .get_last_build_step_id(&self.store.print_store_path(step.get_drv_path()))
+            .get_last_build_step_id(self.store.store_dir(), step.get_drv_path())
             .await?
             .unwrap_or_default();
 
@@ -1711,11 +1704,12 @@ impl State {
                 .unwrap_or_default();
             for (name, path) in &outputs {
                 let res = if let Some(path) = path {
-                    tx.get_last_build_step_id_for_output_path(&self.store.print_store_path(path))
+                    tx.get_last_build_step_id_for_output_path(self.store.store_dir(), path)
                         .await
                 } else {
                     tx.get_last_build_step_id_for_output_with_drv(
-                        &self.store.print_store_path(step.get_drv_path()),
+                        self.store.store_dir(),
+                        step.get_drv_path(),
                         name.as_ref(),
                     )
                     .await
@@ -1728,9 +1722,10 @@ impl State {
         }
 
         tx.create_build_step(
+            self.store.store_dir(),
             None,
             build.id,
-            &self.store.print_store_path(step.get_drv_path()),
+            step.get_drv_path(),
             step.get_system().as_deref(),
             String::new(),
             BuildStatus::CachedFailure,
@@ -1739,12 +1734,6 @@ impl State {
             step.get_output_paths(self.store.store_dir())
                 .unwrap_or_default()
                 .into_iter()
-                .map(|(name, path)| {
-                    (
-                        name.to_string(),
-                        path.map(|p| self.store.print_store_path(&p)),
-                    )
-                })
                 .collect(),
         )
         .await?;
@@ -2155,17 +2144,9 @@ impl State {
     ) -> anyhow::Result<BTreeMap<nix_utils::OutputName, nix_utils::StorePath>> {
         let mut db = self.db.get().await?;
         let mut tx = db.begin_transaction().await?;
-        tx.find_build_step_outputs(&self.store.print_store_path(drv_path))
-            .await?
-            .iter()
-            .map(|(name, path)| -> anyhow::Result<_> {
-                Ok((
-                    nix_utils::OutputName::from_str(name)?,
-                    self.store.get_store_dir().parse(path)?,
-                ))
-            })
-            .collect::<anyhow::Result<_>>()
-            .into()
+        Ok(tx
+            .find_build_step_outputs(self.store.store_dir(), drv_path)
+            .await?)
     }
 
     #[tracing::instrument(skip(self, step), ret, level = "debug")]
@@ -2179,9 +2160,10 @@ impl State {
         };
 
         conn.check_if_paths_failed(
+            self.store.store_dir(),
             &drv_outputs
                 .iter()
-                .filter_map(|(_, path)| path.as_ref().map(|p| self.store.print_store_path(p)))
+                .filter_map(|(_, path)| path.as_ref().cloned())
                 .collect::<Vec<_>>(),
         )
         .await
@@ -2233,7 +2215,7 @@ impl State {
                     continue;
                 };
                 let Some(db_build_output) = db
-                    .get_build_output_for_path(&self.store.print_store_path(out_path))
+                    .get_build_output_for_path(self.store.store_dir(), out_path)
                     .await?
                 else {
                     continue;

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

@@ -351,6 +351,14 @@ impl Step {
             .store(state.deps.len() as u64, Ordering::Relaxed);
     }
 
+    pub fn remove_dep(&self, dep: &Arc<Self>) {
+        let mut state = self.state.write();
+        state.deps.remove(dep);
+        self.atomic_state
+            .deps_len
+            .store(state.deps.len() as u64, Ordering::Relaxed);
+    }
+
     pub fn make_rdep(self: &Arc<Self>, dep: &Arc<Self>) {
         dep.add_dep(self.clone());
         let mut state = self.state.write();