Allow building content-addressed derivations with hydra - reprise

doc/manual/src/projects.md

@@ -399,3 +399,10 @@ analogous:
   | `String value` | `gitea_status_repo` | *Name of the `Git checkout` input* |
   | `String value` | `gitea_http_url`    | *Public URL of `gitea`*, optional  |
 
+Content-addressed derivations
+-----------------------------
+
+Hydra can to a certain extent use the [`ca-derivations` experimental Nix feature](https://github.com/NixOS/rfcs/pull/62).
+To use it, make sure that the Nix version you use is at least as recent as the one used in hydra's flake.
+
+Be warned that this support is still highly experimental, and anything beyond the basic functionality might be broken at that point.

src/hydra-eval-jobs/hydra-eval-jobs.cc

@@ -174,7 +174,7 @@ static void worker(
 
             if (auto drv = getDerivation(state, *v, false)) {
 
-                DrvInfo::Outputs outputs = drv->queryOutputs();
+                DrvInfo::Outputs outputs = drv->queryOutputs(!settings.isExperimentalFeatureEnabled(Xp::CaDerivations));
 
                 if (drv->querySystem() == "unknown")
                     throw EvalError("derivation must have a 'system' attribute");
@@ -231,12 +231,13 @@ static void worker(
                 }
 
                 nlohmann::json out;
-                for (auto & j : outputs)
-                    // FIXME: handle CA/impure builds.
-                    if (j.second)
-                        out[j.first] = state.store->printStorePath(*j.second);
+                if (settings.isExperimentalFeatureEnabled(Xp::CaDerivations))
+                    for (auto & j : outputs)
+                        out[j.first] = "";
+                else
+                    for (auto & j : outputs)
+                                out[j.first] = state.store->printStorePath(*j.second);
                 job["outputs"] = std::move(out);
-
                 reply["job"] = std::move(job);
             }
 

src/hydra-queue-runner/build-remote.cc

@@ -174,6 +174,71 @@ StorePaths reverseTopoSortPaths(const std::map<StorePath, ValidPathInfo> & paths
     return sorted;
 }
 
+/**
+ * Replace the input derivations by their output paths to send a minimal closure
+ * to the builder.
+ *
+ * If we can afford it, resolve it, so that the newly generated derivation still
+ * has some sensible output paths.
+ */
+BasicDerivation inlineInputDerivations(Store & store, Derivation & drv, const StorePath & drvPath)
+{
+    BasicDerivation ret;
+    auto outputHashes = staticOutputHashes(store, drv);
+    if (!drv.type().hasKnownOutputPaths()) {
+        auto maybeBasicDrv = drv.tryResolve(store);
+        if (!maybeBasicDrv)
+            throw Error(
+                "the derivation '%s' can’t be resolved. It’s probably "
+                "missing some outputs",
+                store.printStorePath(drvPath));
+        ret = *maybeBasicDrv;
+    } else {
+        // If the derivation is a real `InputAddressed` derivation, we must
+        // resolve it manually to keep the original output paths
+        ret = BasicDerivation(drv);
+        for (auto & input : drv.inputDrvs) {
+            auto drv2 = store.readDerivation(input.first);
+            auto drv2Outputs = drv2.outputsAndOptPaths(store);
+            for (auto & name : input.second) {
+                auto inputPath = drv2Outputs.at(name);
+                ret.inputSrcs.insert(*inputPath.second);
+            }
+        }
+    }
+    return ret;
+}
+
+/**
+ * Get the newly built outputs, either from the remote if it supports it, or by
+ * introspecting the derivation if the remote is too old
+ */
+DrvOutputs getBuiltOutputs(Store & store, const int remoteVersion, FdSource & from, Derivation & drv)
+{
+    DrvOutputs builtOutputs;
+    if (GET_PROTOCOL_MINOR(remoteVersion) >= 6) {
+        builtOutputs
+            = worker_proto::read(store, from, Phantom<DrvOutputs> {});
+    } else {
+        // If the remote is too old to handle CA derivations, we can’t get this
+        // far anyways
+        assert(drv.type().hasKnownOutputPaths());
+        DerivationOutputsAndOptPaths drvOutputs
+            = drv.outputsAndOptPaths(store);
+        auto outputHashes = staticOutputHashes(store, drv);
+        for (auto & [outputName, output] : drvOutputs) {
+            auto outputPath = output.second;
+            // We’ve just asserted that the output paths of the derivation
+            // were known
+            assert(outputPath);
+              auto outputHash = outputHashes.at(outputName);
+              auto drvOutput = DrvOutput { outputHash, outputName };
+              builtOutputs.insert(
+                  { drvOutput, Realisation { drvOutput, *outputPath } });
+        }
+    }
+    return builtOutputs;
+}
 
 void State::buildRemote(ref<Store> destStore,
     Machine::ptr machine, Step::ptr step,
@@ -234,7 +299,7 @@ void State::buildRemote(ref<Store> destStore,
         unsigned int remoteVersion;
 
         try {
-            to << SERVE_MAGIC_1 << 0x204;
+            to << SERVE_MAGIC_1 << 0x205;
             to.flush();
 
             unsigned int magic = readInt(from);
@@ -264,22 +329,7 @@ void State::buildRemote(ref<Store> destStore,
            outputs of the input derivations. */
         updateStep(ssSendingInputs);
 
-        StorePathSet inputs;
-        BasicDerivation basicDrv(*step->drv);
-
-        for (auto & p : step->drv->inputSrcs)
-            inputs.insert(p);
-
-        for (auto & input : step->drv->inputDrvs) {
-            auto drv2 = localStore->readDerivation(input.first);
-            for (auto & name : input.second) {
-                if (auto i = get(drv2.outputs, name)) {
-                    auto outPath = i->path(*localStore, drv2.name, name);
-                    inputs.insert(*outPath);
-                    basicDrv.inputSrcs.insert(*outPath);
-                }
-            }
-        }
+        BasicDerivation basicDrv = inlineInputDerivations(*localStore, *step->drv, step->drvPath);
 
         /* Ensure that the inputs exist in the destination store. This is
            a no-op for regular stores, but for the binary cache store,
@@ -304,10 +354,10 @@ void State::buildRemote(ref<Store> destStore,
             /* Copy the input closure. */
             if (machine->isLocalhost()) {
                 StorePathSet closure;
-                destStore->computeFSClosure(inputs, closure);
+                destStore->computeFSClosure(basicDrv.inputSrcs, closure);
                 copyPaths(*destStore, *localStore, closure, NoRepair, NoCheckSigs, NoSubstitute);
             } else {
-                copyClosureTo(machine->state->sendLock, *destStore, from, to, inputs, true);
+                copyClosureTo(machine->state->sendLock, *destStore, from, to, step->drv->inputSrcs, true);
             }
 
             auto now2 = std::chrono::steady_clock::now();
@@ -366,9 +416,6 @@ void State::buildRemote(ref<Store> destStore,
                 result.stopTime = stop;
             }
         }
-        if (GET_PROTOCOL_MINOR(remoteVersion) >= 6) {
-            worker_proto::read(*localStore, from, Phantom<DrvOutputs> {});
-        }
         switch ((BuildResult::Status) res) {
             case BuildResult::Built:
                 result.stepStatus = bsSuccess;
@@ -426,6 +473,11 @@ void State::buildRemote(ref<Store> destStore,
             result.logFile = "";
         }
 
+        auto builtOutputs = getBuiltOutputs(*localStore, remoteVersion, from, *step->drv);
+        StorePathSet outputs;
+        for (auto & [_, realisation] : builtOutputs)
+            outputs.insert(realisation.outPath);
+
         /* Copy the output paths. */
         if (!machine->isLocalhost() || localStore != std::shared_ptr<Store>(destStore)) {
             updateStep(ssReceivingOutputs);
@@ -434,12 +486,6 @@ void State::buildRemote(ref<Store> destStore,
 
             auto now1 = std::chrono::steady_clock::now();
 
-            StorePathSet outputs;
-            for (auto & i : step->drv->outputsAndOptPaths(*localStore)) {
-                if (i.second.second)
-                   outputs.insert(*i.second.second);
-            }
-
             /* Get info about each output path. */
             std::map<StorePath, ValidPathInfo> infos;
             size_t totalNarSize = 0;
@@ -509,6 +555,23 @@ void State::buildRemote(ref<Store> destStore,
             result.overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
         }
 
+        /* Register the outputs of the newly built drv */
+        if (settings.isExperimentalFeatureEnabled(Xp::CaDerivations)) {
+          auto outputHashes = staticOutputHashes(*localStore, *step->drv);
+          for (auto & [outputId, realisation] : builtOutputs) {
+              // Register the resolved drv output
+              localStore->registerDrvOutput(realisation);
+              destStore->registerDrvOutput(realisation);
+
+              // Also register the unresolved one
+              auto unresolvedRealisation = realisation;
+              unresolvedRealisation.signatures.clear();
+              unresolvedRealisation.id.drvHash = outputHashes.at(outputId.outputName);
+              localStore->registerDrvOutput(unresolvedRealisation);
+              destStore->registerDrvOutput(unresolvedRealisation);
+          }
+        }
+
         /* Shut down the connection. */
         child.to = -1;
         child.pid.wait();

src/hydra-queue-runner/build-result.cc

@@ -11,18 +11,18 @@ using namespace nix;
 BuildOutput getBuildOutput(
     nix::ref<Store> store,
     NarMemberDatas & narMembers,
-    const Derivation & drv)
+    const OutputPathMap derivationOutputs)
 {
     BuildOutput res;
 
     /* Compute the closure size. */
     StorePathSet outputs;
     StorePathSet closure;
-    for (auto & i : drv.outputsAndOptPaths(*store))
-        if (i.second.second) {
-            store->computeFSClosure(*i.second.second, closure);
-            outputs.insert(*i.second.second);
-        }
+    for (auto& [outputName, outputPath] : derivationOutputs) {
+        store->computeFSClosure(outputPath, closure);
+        outputs.insert(outputPath);
+        res.outputs.insert({outputName, outputPath});
+    }
     for (auto & path : closure) {
         auto info = store->queryPathInfo(path);
         res.closureSize += info->narSize;
@@ -107,13 +107,12 @@ BuildOutput getBuildOutput(
     /* If no build products were explicitly declared, then add all
        outputs as a product of type "nix-build". */
     if (!explicitProducts) {
-        for (auto & [name, output] : drv.outputs) {
+        for (auto& [name, output] : derivationOutputs) {
             BuildProduct product;
-            auto outPath = output.path(*store, drv.name, name);
-            product.path = store->printStorePath(*outPath);
+            product.path = store->printStorePath(output);
             product.type = "nix-build";
             product.subtype = name == "out" ? "" : name;
-            product.name = outPath->name();
+            product.name = output.name();
 
             auto file = narMembers.find(product.path);
             assert(file != narMembers.end());

src/hydra-queue-runner/builder.cc

@@ -221,7 +221,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         if (result.stepStatus == bsSuccess) {
             updateStep(ssPostProcessing);
-            res = getBuildOutput(destStore, narMembers, *step->drv);
+            res = getBuildOutput(destStore, narMembers, localStore->queryDerivationOutputMap(step->drvPath));
         }
     }
 
@@ -275,9 +275,9 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         assert(stepNr);
 
-        for (auto & i : step->drv->outputsAndOptPaths(*localStore)) {
-            if (i.second.second)
-               addRoot(*i.second.second);
+        for (auto & i : localStore->queryPartialDerivationOutputMap(step->drvPath)) {
+            if (i.second)
+               addRoot(*i.second);
         }
 
         /* Register success in the database for all Build objects that

src/hydra-queue-runner/hydra-build-result.hh

@@ -36,10 +36,12 @@ struct BuildOutput
 
     std::list<BuildProduct> products;
 
+    std::map<std::string, nix::StorePath> outputs;
+
     std::map<std::string, BuildMetric> metrics;
 };
 
 BuildOutput getBuildOutput(
     nix::ref<nix::Store> store,
     NarMemberDatas & narMembers,
-    const nix::Derivation & drv);
+    const nix::OutputPathMap derivationOutputs);

src/hydra-queue-runner/hydra-queue-runner.cc

@@ -318,10 +318,10 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
 
     if (r.affected_rows() == 0) goto restart;
 
-    for (auto & [name, output] : step->drv->outputs)
-        txn.exec_params0
-            ("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
-            buildId, stepNr, name, localStore->printStorePath(*output.path(*localStore, step->drv->name, name)));
+    for (auto& [name, output] : localStore->queryPartialDerivationOutputMap(step->drvPath))
+      txn.exec_params0
+          ("insert into BuildStepOutputs (build, stepnr, name, path, contentAddressed) values ($1, $2, $3, $4, $5)",
+            buildId, stepNr, name, output ? localStore->printStorePath(*output) : "", step->drv->type().isCA());
 
     if (status == bsBusy)
         txn.exec(fmt("notify step_started, '%d\t%d'", buildId, stepNr));
@@ -358,11 +358,23 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
     assert(result.logFile.find('\t') == std::string::npos);
     txn.exec(fmt("notify step_finished, '%d\t%d\t%s'",
             buildId, stepNr, result.logFile));
+
+    if (result.stepStatus == bsSuccess) {
+      // Update the corresponding `BuildStepOutputs` row to add the output path
+      auto res = txn.exec_params1("select drvPath from BuildSteps where build = $1 and stepnr = $2", buildId, stepNr);
+      assert(res.size());
+      StorePath drvPath = localStore->parseStorePath(res[0].as<std::string>());
+      // If we've finished building, all the paths should be known
+      for (auto& [name, output] : localStore->queryDerivationOutputMap(drvPath))
+        txn.exec_params0
+            ("update BuildStepOutputs set path = $4 where build = $1 and stepnr = $2 and name = $3",
+              buildId, stepNr, name, localStore->printStorePath(output));
+    }
 }
 
 
 int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
-    Build::ptr build, const StorePath & drvPath, const std::string & outputName, const StorePath & storePath)
+    Build::ptr build, const StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const StorePath & storePath)
 {
  restart:
     auto stepNr = allocBuildStep(txn, build->id);
@@ -381,9 +393,10 @@ int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t sto
     if (r.affected_rows() == 0) goto restart;
 
     txn.exec_params0
-        ("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
+        ("insert into BuildStepOutputs (build, stepnr, name, path, contentAddressed) values ($1, $2, $3, $4, $5)",
          build->id, stepNr, outputName,
-         localStore->printStorePath(storePath));
+         localStore->printStorePath(storePath),
+         drv.type().isCA());
 
     return stepNr;
 }
@@ -463,6 +476,15 @@ void State::markSucceededBuild(pqxx::work & txn, Build::ptr build,
          res.releaseName != "" ? std::make_optional(res.releaseName) : std::nullopt,
          isCachedBuild ? 1 : 0);
 
+    for (auto & [outputName, outputPath] : res.outputs) {
+      txn.exec_params0
+        ("update BuildOutputs set path = $3 where build = $1 and name = $2",
+         build->id,
+         outputName,
+         localStore->printStorePath(outputPath)
+        );
+    }
+
     txn.exec_params0("delete from BuildProducts where build = $1", build->id);
 
     unsigned int productNr = 1;