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

Author: Ericson2314

subprojects/crates/db/src/connection.rs

@@ -355,14 +355,20 @@ impl Connection {
                 CROSS JOIN LATERAL (
                     SELECT o.path
                     FROM buildsteps s
+                    -- If this step was resolved, look up outputs from
+                    -- the resolved drv's successful buildstep instead.
+                    -- resolvedDrvPath is a bare basename; drvPath is a
+                    -- full path, so strip the directory prefix to compare.
                     LEFT JOIN buildsteps sr
-                        ON s.build = sr.build AND s.resolvedToStep = sr.stepnr
+                        ON sr.drvPath = $2 || '/' || s.resolvedDrvPath
+                        AND sr.status = 0
                     JOIN buildstepoutputs o
-                        ON s.build = o.build AND (s.stepnr = o.stepnr OR sr.stepnr = o.stepnr)
+                        ON o.build = COALESCE(sr.build, s.build)
+                        AND o.stepnr = COALESCE(sr.stepnr, s.stepnr)
                     WHERE s.drvPath = r.drv_path
                       AND o.name = i.chain[r.step]
                       AND o.path IS NOT NULL
-                      AND (s.status = 0 OR (s.status = 13 AND sr.status = 0))
+                      AND (s.status = 0 OR s.status = 13)
                     ORDER BY s.build DESC
                     LIMIT 1
                 ) sub
@@ -378,6 +384,7 @@ impl Connection {
             ",
         )
         .bind(&json_input)
+        .bind(store_dir.to_str())
         .fetch_all(&mut *self.conn)
         .await?;
 
@@ -698,11 +705,31 @@ impl Transaction<'_> {
     #[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?;
+        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?;
 
-        Ok(items.into_iter().collect())
+        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)]
@@ -953,25 +980,24 @@ 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.
+    /// Record which resolved drv path this step was resolved to.
     #[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,
+        resolved_drv_path: &StorePath,
     ) -> sqlx::Result<()> {
         sqlx::query(
             r"
               UPDATE buildsteps
-              SET resolvedToStep = $3
+              SET resolvedDrvPath = $3
               WHERE build = $1 AND stepnr = $2
             ",
         )
         .bind(origin_build_id)
         .bind(origin_step_nr)
-        .bind(resolved_step_nr)
+        .bind(resolved_drv_path.to_string())
         .execute(&mut *self.tx)
         .await?;
         Ok(())
@@ -1281,11 +1307,24 @@ mod tests {
     }
 
     async fn insert_step(conn: &mut Connection, build: i32, stepnr: i32, drv_path: &StorePath) {
+        insert_step_with_status(conn, build, stepnr, drv_path, 0, None).await;
+    }
+
+    async fn insert_step_with_status(
+        conn: &mut Connection,
+        build: i32,
+        stepnr: i32,
+        drv_path: &StorePath,
+        status: i32,
+        resolved_drv_path: Option<&StorePath>,
+    ) {
         let sd = test_store_dir();
-        sqlx::query("INSERT INTO BuildSteps (build, stepnr, type, busy, drvPath, status) VALUES ($1, $2, 0, 0, $3, 0)")
+        sqlx::query("INSERT INTO BuildSteps (build, stepnr, type, busy, drvPath, status, resolvedDrvPath) VALUES ($1, $2, 0, 0, $3, $4, $5)")
             .bind(build)
             .bind(stepnr)
             .bind(sd.display(drv_path).to_string())
+            .bind(status)
+            .bind(resolved_drv_path.map(|p| p.to_string()))
             .execute(&mut *conn.conn)
             .await
             .unwrap();
@@ -1423,6 +1462,70 @@ mod tests {
         );
     }
 
+    /// A step that was resolved (status=13) with `resolvedDrvPath` pointing
+    /// to a different drv whose successful buildstep has the outputs.
+    #[tokio::test]
+    async fn resolve_through_resolved_step() {
+        let (_pg, mut conn) = setup().await;
+
+        // Step 1: unresolved ca-depending-on-ca.drv, status=Resolved(13),
+        // resolvedDrvPath points to the resolved drv
+        insert_step_with_status(
+            &mut conn,
+            1,
+            1,
+            &sp("unresolved.drv"),
+            13,
+            Some(&sp("resolved.drv")),
+        )
+        .await;
+        // A successful buildstep for the resolved drv (could be any build)
+        insert_step(&mut conn, 2, 1, &sp("resolved.drv")).await;
+        insert_output(&mut conn, 2, 1, "out", &sp("result")).await;
+
+        // Looking up via the unresolved drv path should find the output
+        // through the resolvedDrvPath chain.
+        let results = conn
+            .resolve_drv_output_chains(&test_store_dir(), &[(&sp("unresolved.drv"), &[&on("out")])])
+            .await
+            .unwrap();
+        assert_eq!(results, vec![Some(sp("result"))]);
+    }
+
+    /// A depth-2 chain where the first step was resolved:
+    /// unresolved.drv (status=13, resolvedDrvPath→resolved.drv) →
+    /// resolved.drv outputs a .drv path → that .drv has the final output.
+    #[tokio::test]
+    async fn resolve_depth_2_through_resolved_step() {
+        let (_pg, mut conn) = setup().await;
+
+        // Build 1: unresolved step, resolved to bbb-resolved.drv
+        insert_step_with_status(
+            &mut conn,
+            1,
+            1,
+            &sp("unresolved.drv"),
+            13,
+            Some(&sp("resolved.drv")),
+        )
+        .await;
+        insert_step(&mut conn, 2, 1, &sp("resolved.drv")).await;
+        insert_output(&mut conn, 2, 1, "out", &sp("intermediate.drv")).await;
+
+        // Build 3: the intermediate drv
+        insert_step(&mut conn, 3, 1, &sp("intermediate.drv")).await;
+        insert_output(&mut conn, 3, 1, "out", &sp("final")).await;
+
+        let results = conn
+            .resolve_drv_output_chains(
+                &test_store_dir(),
+                &[(&sp("unresolved.drv"), &[&on("out"), &on("out")])],
+            )
+            .await
+            .unwrap();
+        assert_eq!(results, vec![Some(sp("final"))]);
+    }
+
     /// Batch with ragged depths: one depth-1 (Opaque), one depth-2 (Built),
     /// one depth-3 (Built(Built(...))).
     #[tokio::test]
@@ -1467,4 +1570,41 @@ mod tests {
             ]
         );
     }
+
+    /// Depth-1 lookup where the only buildstep for the drv has
+    /// status=Resolved(13) with `resolvedDrvPath` pointing to
+    /// a different drv whose successful buildstep has the outputs.
+    /// This matches the production scenario: ca-depending-on-ca.drv
+    /// was resolved to a different drv, and ca-depending-on-ca-
+    /// depending-on-ca needs to look up its output.
+    #[tokio::test]
+    async fn resolve_depth_1_via_resolved_step() {
+        let (_pg, mut conn) = setup().await;
+
+        // Build 1, step 1: unresolved ca-depending-on-ca.drv
+        //   status=13 (Resolved), resolvedDrvPath points to the resolved drv
+        insert_step_with_status(
+            &mut conn,
+            1,
+            1,
+            &sp("unresolved-ca-dep.drv"),
+            13,
+            Some(&sp("resolved-ca-dep.drv")),
+        )
+        .await;
+        // Build 2: the resolved drv was built successfully
+        insert_step(&mut conn, 2, 1, &sp("resolved-ca-dep.drv")).await;
+        insert_output(&mut conn, 2, 1, "out", &sp("ca-dep-output")).await;
+
+        // Depth-1 chain: look up "out" of the unresolved drv path.
+        // The query should follow resolvedDrvPath to find the output.
+        let results = conn
+            .resolve_drv_output_chains(
+                &test_store_dir(),
+                &[(&sp("unresolved-ca-dep.drv"), &[&on("out")])],
+            )
+            .await
+            .unwrap();
+        assert_eq!(results, vec![Some(sp("ca-dep-output"))]);
+    }
 }