.gitignore

@@ -20,6 +20,7 @@ Makefile.in
 /src/sql/hydra-postgresql.sql
 /src/sql/hydra-sqlite.sql
 /src/sql/tmp.sqlite
+/src/hydra-eval-jobs/hydra-eval-jobs
 /src/root/static/bootstrap
 /src/root/static/js/flot
 /tests

configure.ac

@@ -61,6 +61,7 @@ AC_CONFIG_FILES([
   doc/manual/Makefile
   src/Makefile
   src/hydra-evaluator/Makefile
+  src/hydra-eval-jobs/Makefile
   src/hydra-queue-runner/Makefile
   src/sql/Makefile
   src/ttf/Makefile

flake.lock

@@ -16,28 +16,7 @@
         "type": "github"
       }
     },
-    "flake-parts": {
-      "inputs": {
-        "nixpkgs-lib": [
-          "nix-eval-jobs",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1717285511,
-        "narHash": "sha256-iKzJcpdXih14qYVcZ9QC9XuZYnPc6T8YImb6dX166kw=",
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "rev": "2a55567fcf15b1b1c7ed712a2c6fadaec7412ea8",
-        "type": "github"
-      },
-      "original": {
-        "owner": "hercules-ci",
-        "repo": "flake-parts",
-        "type": "github"
-      }
-    },
-    "lix": {
+    "nix": {
       "inputs": {
         "flake-compat": "flake-compat",
         "nix2container": "nix2container",
@@ -48,64 +27,17 @@
         "pre-commit-hooks": "pre-commit-hooks"
       },
       "locked": {
-        "lastModified": 1721091462,
-        "narHash": "sha256-0cmEeoOiB91BviTJHzIyxkY+Gxv3O8ZnnExVAoXEFGI=",
+        "lastModified": 1720733512,
+        "narHash": "sha256-vq9CLDvqSSvH4L7YhDa0ihTOrAry4jntKiuoNb5n98M=",
         "ref": "refs/heads/main",
-        "rev": "6b4d46e9e0e1dd80e0977684ab20d14bcd1a6bc3",
-        "revCount": 15967,
+        "rev": "4b109ec1a8fc4550150f56f0f46f2f41d844bda8",
+        "revCount": 15950,
         "type": "git",
-        "url": "https://git.lix.systems/lix-project/lix"
+        "url": "https://git@git.lix.systems/lix-project/lix"
       },
       "original": {
         "type": "git",
-        "url": "https://git.lix.systems/lix-project/lix"
-      }
-    },
-    "nix-eval-jobs": {
-      "inputs": {
-        "flake-parts": "flake-parts",
-        "lix": [
-          "lix"
-        ],
-        "nix-github-actions": "nix-github-actions",
-        "nixpkgs": [
-          "nixpkgs"
-        ],
-        "treefmt-nix": "treefmt-nix"
-      },
-      "locked": {
-        "lastModified": 1719352671,
-        "narHash": "sha256-F/RvI9chHywnckEqHO1ggjzCayknhDnnl2kNnnVXpWg=",
-        "ref": "refs/heads/main",
-        "rev": "f8869bdcca7c1d5aaf37de3da3a4176811279a57",
-        "revCount": 606,
-        "type": "git",
-        "url": "https://git.lix.systems/lix-project/nix-eval-jobs"
-      },
-      "original": {
-        "type": "git",
-        "url": "https://git.lix.systems/lix-project/nix-eval-jobs"
-      }
-    },
-    "nix-github-actions": {
-      "inputs": {
-        "nixpkgs": [
-          "nix-eval-jobs",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1703863825,
-        "narHash": "sha256-rXwqjtwiGKJheXB43ybM8NwWB8rO2dSRrEqes0S7F5Y=",
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "rev": "5163432afc817cf8bd1f031418d1869e4c9d5547",
-        "type": "github"
-      },
-      "original": {
-        "owner": "nix-community",
-        "repo": "nix-github-actions",
-        "type": "github"
+        "url": "https://git@git.lix.systems/lix-project/lix"
       }
     },
     "nix2container": {
@@ -174,31 +106,9 @@
     },
     "root": {
       "inputs": {
-        "lix": "lix",
-        "nix-eval-jobs": "nix-eval-jobs",
+        "nix": "nix",
         "nixpkgs": "nixpkgs"
       }
-    },
-    "treefmt-nix": {
-      "inputs": {
-        "nixpkgs": [
-          "nix-eval-jobs",
-          "nixpkgs"
-        ]
-      },
-      "locked": {
-        "lastModified": 1718139168,
-        "narHash": "sha256-1TZQcdETNdJMcfwwoshVeCjwWfrPtkSQ8y8wFX3it7k=",
-        "owner": "numtide",
-        "repo": "treefmt-nix",
-        "rev": "1cb529bffa880746a1d0ec4e0f5076876af931f1",
-        "type": "github"
-      },
-      "original": {
-        "owner": "numtide",
-        "repo": "treefmt-nix",
-        "type": "github"
-      }
     }
   },
   "root": "root",

flake.nix

@@ -2,20 +2,15 @@
   description = "A Nix-based continuous build system";
 
   inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.05";
+  inputs.nix.url = "git+https://git@git.lix.systems/lix-project/lix";
+  inputs.nix.inputs.nixpkgs.follows = "nixpkgs";
 
-  inputs.lix.url = "git+https://git.lix.systems/lix-project/lix";
-  inputs.lix.inputs.nixpkgs.follows = "nixpkgs";
-
-  inputs.nix-eval-jobs.url = "git+https://git.lix.systems/lix-project/nix-eval-jobs";
-  inputs.nix-eval-jobs.inputs.nixpkgs.follows = "nixpkgs";
-  inputs.nix-eval-jobs.inputs.lix.follows = "lix";
-
-  outputs = { self, nix-eval-jobs, nixpkgs, lix }:
+  outputs = { self, nixpkgs, nix }:
     let
       systems = [ "x86_64-linux" "aarch64-linux" ];
       forEachSystem = nixpkgs.lib.genAttrs systems;
 
-      overlayList = [ self.overlays.default lix.overlays.default ];
+      overlayList = [ self.overlays.default nix.overlays.default ];
 
       pkgsBySystem = forEachSystem (system: import nixpkgs {
         inherit system;
@@ -29,7 +24,6 @@
       overlays.default = final: prev: {
         hydra = final.callPackage ./package.nix {
           inherit (final.lib) fileset;
-          nix-eval-jobs = nix-eval-jobs.packages.${final.system}.default;
           rawSrc = self;
         };
       };

package.nix

@@ -48,7 +48,6 @@
 , xz
 , gnutar
 , gnused
-, nix-eval-jobs
 
 , rpm
 , dpkg
@@ -193,7 +192,6 @@ stdenv.mkDerivation (finalAttrs: {
     openldap
     postgresql_13
     pixz
-    nix-eval-jobs
   ];
 
   checkInputs = [
@@ -222,7 +220,6 @@ stdenv.mkDerivation (finalAttrs: {
       darcs
       gnused
       breezy
-      nix-eval-jobs
     ] ++ lib.optionals stdenv.isLinux [ rpm dpkg cdrkit ]
   );
 
@@ -231,7 +228,7 @@ stdenv.mkDerivation (finalAttrs: {
   shellHook = ''
     pushd $(git rev-parse --show-toplevel) >/dev/null
 
-    PATH=$(pwd)/src/hydra-evaluator:$(pwd)/src/script:$(pwd)/src/hydra-queue-runner:$PATH
+    PATH=$(pwd)/src/hydra-evaluator:$(pwd)/src/script:$(pwd)/src/hydra-eval-jobs:$(pwd)/src/hydra-queue-runner:$PATH
     PERL5LIB=$(pwd)/src/lib:$PERL5LIB
     export HYDRA_HOME="$(pwd)/src/"
     mkdir -p .hydra-data

src/Makefile.am

@@ -1,3 +1,3 @@
-SUBDIRS = hydra-evaluator hydra-queue-runner sql script lib root ttf
+SUBDIRS = hydra-evaluator hydra-eval-jobs hydra-queue-runner sql script lib root ttf
 BOOTCLEAN_SUBDIRS = $(SUBDIRS)
 DIST_SUBDIRS      = $(SUBDIRS)

src/hydra-eval-jobs/Makefile.am

@@ -0,0 +1,5 @@
+bin_PROGRAMS = hydra-eval-jobs
+
+hydra_eval_jobs_SOURCES = hydra-eval-jobs.cc
+hydra_eval_jobs_LDADD = $(NIX_LIBS) -llixcmd
+hydra_eval_jobs_CXXFLAGS = $(NIX_CFLAGS) -I ../libhydra

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

@@ -0,0 +1,577 @@
+#include <iostream>
+#include <thread>
+#include <optional>
+#include <unordered_map>
+
+#include "shared.hh"
+#include "store-api.hh"
+#include "eval.hh"
+#include "eval-inline.hh"
+#include "eval-settings.hh"
+#include "signals.hh"
+#include "terminal.hh"
+#include "get-drvs.hh"
+#include "globals.hh"
+#include "lix/libcmd/common-eval-args.hh"
+#include "flake/flakeref.hh"
+#include "flake/flake.hh"
+#include "attr-path.hh"
+#include "derivations.hh"
+#include "local-fs-store.hh"
+
+#include "hydra-config.hh"
+
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/resource.h>
+
+#include <nlohmann/json.hpp>
+
+void check_pid_status_nonblocking(pid_t check_pid)
+{
+    // Only check 'initialized' and known PID's
+    if (check_pid <= 0) { return; }
+
+    int wstatus = 0;
+    pid_t pid = waitpid(check_pid, &wstatus, WNOHANG);
+    // -1 = failure, WNOHANG: 0 = no change
+    if (pid <= 0) { return; }
+
+    std::cerr << "child process (" << pid << ") ";
+
+    if (WIFEXITED(wstatus)) {
+        std::cerr << "exited with status=" << WEXITSTATUS(wstatus) << std::endl;
+    } else if (WIFSIGNALED(wstatus)) {
+        std::cerr << "killed by signal=" << WTERMSIG(wstatus) << std::endl;
+    } else if (WIFSTOPPED(wstatus)) {
+        std::cerr << "stopped by signal=" << WSTOPSIG(wstatus) << std::endl;
+    } else if (WIFCONTINUED(wstatus)) {
+        std::cerr << "continued" << std::endl;
+    }
+}
+
+using namespace nix;
+
+static Path gcRootsDir;
+static size_t maxMemorySize;
+
+struct MyArgs : MixEvalArgs, MixCommonArgs, RootArgs
+{
+    Path releaseExpr;
+    bool flake = false;
+    bool dryRun = false;
+
+    MyArgs() : MixCommonArgs("hydra-eval-jobs")
+    {
+        addFlag({
+            .longName = "gc-roots-dir",
+            .description = "garbage collector roots directory",
+            .labels = {"path"},
+            .handler = {&gcRootsDir}
+        });
+
+        addFlag({
+            .longName = "dry-run",
+            .description = "don't create store derivations",
+            .handler = {&dryRun, true}
+        });
+
+        addFlag({
+            .longName = "flake",
+            .description = "build a flake",
+            .handler = {&flake, true}
+        });
+
+        expectArg("expr", &releaseExpr);
+    }
+};
+
+static MyArgs myArgs;
+
+static std::string queryMetaStrings(EvalState & state, DrvInfo & drv, const std::string & name, const std::string & subAttribute)
+{
+    Strings res;
+    std::function<void(Value & v)> rec;
+
+    rec = [&](Value & v) {
+        state.forceValue(v, noPos);
+        if (v.type() == nString)
+            res.push_back(v.string.s);
+        else if (v.isList())
+            for (unsigned int n = 0; n < v.listSize(); ++n)
+                rec(*v.listElems()[n]);
+        else if (v.type() == nAttrs) {
+            auto a = v.attrs->find(state.symbols.create(subAttribute));
+            if (a != v.attrs->end())
+                res.push_back(std::string(state.forceString(*a->value, a->pos, "while evaluating meta attributes")));
+        }
+    };
+
+    Value * v = drv.queryMeta(name);
+    if (v) rec(*v);
+
+    return concatStringsSep(", ", res);
+}
+
+static void worker(
+    EvalState & state,
+    Bindings & autoArgs,
+    AutoCloseFD & to,
+    AutoCloseFD & from)
+{
+    Value vTop;
+
+    if (myArgs.flake) {
+        using namespace flake;
+
+        auto flakeRef = parseFlakeRef(myArgs.releaseExpr);
+
+        auto vFlake = state.allocValue();
+
+        auto lockedFlake = lockFlake(state, flakeRef,
+            LockFlags {
+                .updateLockFile = false,
+                .useRegistries = false,
+                .allowUnlocked = false,
+            });
+
+        callFlake(state, lockedFlake, *vFlake);
+
+        auto vOutputs = vFlake->attrs->get(state.symbols.create("outputs"))->value;
+        state.forceValue(*vOutputs, noPos);
+
+        auto aHydraJobs = vOutputs->attrs->get(state.symbols.create("hydraJobs"));
+        if (!aHydraJobs)
+            aHydraJobs = vOutputs->attrs->get(state.symbols.create("checks"));
+        if (!aHydraJobs)
+            throw Error("flake '%s' does not provide any Hydra jobs or checks", flakeRef);
+
+        vTop = *aHydraJobs->value;
+
+    } else {
+        state.evalFile(lookupFileArg(state, myArgs.releaseExpr), vTop);
+    }
+
+    auto vRoot = state.allocValue();
+    state.autoCallFunction(autoArgs, vTop, *vRoot);
+
+    while (true) {
+        /* Wait for the master to send us a job name. */
+        writeLine(to.get(), "next");
+
+        auto s = readLine(from.get());
+        if (s == "exit") break;
+        if (!s.starts_with("do ")) abort();
+        std::string attrPath(s, 3);
+
+        debug("worker process %d at '%s'", getpid(), attrPath);
+
+        /* Evaluate it and send info back to the master. */
+        nlohmann::json reply;
+
+        try {
+            auto vTmp = findAlongAttrPath(state, attrPath, autoArgs, *vRoot).first;
+
+            auto v = state.allocValue();
+            state.autoCallFunction(autoArgs, *vTmp, *v);
+
+            if (auto drv = getDerivation(state, *v, false)) {
+
+                // CA derivations do not have static output paths, so we
+                // have to defensively not query output paths in case we
+                // encounter one.
+                DrvInfo::Outputs outputs = drv->queryOutputs(
+                    !experimentalFeatureSettings.isEnabled(Xp::CaDerivations));
+
+                if (drv->querySystem() == "unknown")
+                    state.error<EvalError>("derivation must have a 'system' attribute").debugThrow();
+
+                auto drvPath = state.store->printStorePath(drv->requireDrvPath());
+
+                nlohmann::json job;
+
+                job["nixName"] = drv->queryName();
+                job["system"] =drv->querySystem();
+                job["drvPath"] = drvPath;
+                job["description"] = drv->queryMetaString("description");
+                job["license"] = queryMetaStrings(state, *drv, "license", "shortName");
+                job["homepage"] = drv->queryMetaString("homepage");
+                job["maintainers"] = queryMetaStrings(state, *drv, "maintainers", "email");
+                job["schedulingPriority"] = drv->queryMetaInt("schedulingPriority", 100);
+                job["timeout"] = drv->queryMetaInt("timeout", 36000);
+                job["maxSilent"] = drv->queryMetaInt("maxSilent", 7200);
+                job["isChannel"] = drv->queryMetaBool("isHydraChannel", false);
+
+                /* If this is an aggregate, then get its constituents. */
+                auto a = v->attrs->get(state.symbols.create("_hydraAggregate"));
+                if (a && state.forceBool(*a->value, a->pos, "while evaluating the `_hydraAggregate` attribute")) {
+                    auto a = v->attrs->get(state.symbols.create("constituents"));
+                    if (!a)
+                        state.error<EvalError>("derivation must have a ‘constituents’ attribute").debugThrow();
+
+                    NixStringContext context;
+                    state.coerceToString(a->pos, *a->value, context, "while evaluating the `constituents` attribute", true, false);
+                    for (auto & c : context)
+                        std::visit(overloaded {
+                            [&](const NixStringContextElem::Built & b) {
+                                job["constituents"].push_back(b.drvPath->to_string(*state.store));
+                            },
+                            [&](const NixStringContextElem::Opaque & o) {
+                            },
+                            [&](const NixStringContextElem::DrvDeep & d) {
+                            },
+                        }, c.raw);
+
+                    state.forceList(*a->value, a->pos, "while evaluating the `constituents` attribute");
+                    for (unsigned int n = 0; n < a->value->listSize(); ++n) {
+                        auto v = a->value->listElems()[n];
+                        state.forceValue(*v, noPos);
+                        if (v->type() == nString)
+                            job["namedConstituents"].push_back(v->str());
+                    }
+                }
+
+                /* Register the derivation as a GC root.  !!! This
+                   registers roots for jobs that we may have already
+                   done. */
+                auto localStore = state.store.dynamic_pointer_cast<LocalFSStore>();
+                if (gcRootsDir != "" && localStore) {
+                    Path root = gcRootsDir + "/" + std::string(baseNameOf(drvPath));
+                    if (!pathExists(root))
+                        localStore->addPermRoot(localStore->parseStorePath(drvPath), root);
+                }
+
+                nlohmann::json out;
+                for (auto & [outputName, optOutputPath] : outputs) {
+                    if (optOutputPath) {
+                        out[outputName] = state.store->printStorePath(*optOutputPath);
+                    } else {
+                        // See the `queryOutputs` call above; we should
+                        // not encounter missing output paths otherwise.
+                        assert(experimentalFeatureSettings.isEnabled(Xp::CaDerivations));
+                        out[outputName] = nullptr;
+                    }
+                }
+                job["outputs"] = std::move(out);
+                reply["job"] = std::move(job);
+            }
+
+            else if (v->type() == nAttrs) {
+                auto attrs = nlohmann::json::array();
+                StringSet ss;
+                for (auto & i : v->attrs->lexicographicOrder(state.symbols)) {
+                    std::string name(state.symbols[i->name]);
+                    if (name.find(' ') != std::string::npos) {
+                        printError("skipping job with illegal name '%s'", name);
+                        continue;
+                    }
+                    attrs.push_back(name);
+                }
+                reply["attrs"] = std::move(attrs);
+            }
+
+            else if (v->type() == nNull)
+                ;
+
+            else state.error<TypeError>("attribute '%s' is %s, which is not supported", attrPath, showType(*v)).debugThrow();
+
+        } catch (EvalError & e) {
+            auto msg = e.msg();
+            // Transmits the error we got from the previous evaluation
+            // in the JSON output.
+            reply["error"] = filterANSIEscapes(msg, true);
+            // Don't forget to print it into the STDERR log, this is
+            // what's shown in the Hydra UI.
+            printError(msg);
+        }
+
+        writeLine(to.get(), reply.dump());
+
+        /* If our RSS exceeds the maximum, exit. The master will
+           start a new process. */
+        struct rusage r;
+        getrusage(RUSAGE_SELF, &r);
+        if ((size_t) r.ru_maxrss > maxMemorySize * 1024) break;
+    }
+
+    writeLine(to.get(), "restart");
+}
+
+int main(int argc, char * * argv)
+{
+    /* Prevent undeclared dependencies in the evaluation via
+       $NIX_PATH. */
+    unsetenv("NIX_PATH");
+
+    return handleExceptions(argv[0], [&]() {
+
+        auto config = std::make_unique<HydraConfig>();
+
+        auto nrWorkers = config->getIntOption("evaluator_workers", 1);
+        maxMemorySize = config->getIntOption("evaluator_max_memory_size", 4096);
+
+        initNix();
+        initGC();
+
+        myArgs.parseCmdline(argvToStrings(argc, argv));
+
+        auto pureEval = config->getBoolOption("evaluator_pure_eval", myArgs.flake);
+
+        /* FIXME: The build hook in conjunction with import-from-derivation is causing "unexpected EOF" during eval */
+        settings.builders = "";
+
+        /* Prevent access to paths outside of the Nix search path and
+           to the environment. */
+        evalSettings.restrictEval = true;
+
+        /* When building a flake, use pure evaluation (no access to
+           'getEnv', 'currentSystem' etc. */
+        evalSettings.pureEval = pureEval;
+
+        if (myArgs.dryRun) settings.readOnlyMode = true;
+
+        if (myArgs.releaseExpr == "") throw UsageError("no expression specified");
+
+        if (gcRootsDir == "") printMsg(lvlError, "warning: `--gc-roots-dir' not specified");
+
+        struct State
+        {
+            std::set<std::string> todo{""};
+            std::set<std::string> active;
+            nlohmann::json jobs;
+            std::exception_ptr exc;
+        };
+
+        std::condition_variable wakeup;
+
+        Sync<State> state_;
+
+        /* Start a handler thread per worker process. */
+        auto handler = [&]()
+        {
+            Pid pid;
+            try {
+                AutoCloseFD from, to;
+
+                while (true) {
+
+                    /* Start a new worker process if necessary. */
+                    if (!pid) {
+                        Pipe toPipe, fromPipe;
+                        toPipe.create();
+                        fromPipe.create();
+                        pid = startProcess(
+                            [&,
+                             to{std::make_shared<AutoCloseFD>(std::move(fromPipe.writeSide))},
+                             from{std::make_shared<AutoCloseFD>(std::move(toPipe.readSide))}
+                            ]()
+                            {
+                                try {
+                                    EvalState state(myArgs.searchPath, openStore());
+                                    Bindings & autoArgs = *myArgs.getAutoArgs(state);
+                                    worker(state, autoArgs, *to, *from);
+                                } catch (Error & e) {
+                                    nlohmann::json err;
+                                    auto msg = e.msg();
+                                    err["error"] = filterANSIEscapes(msg, true);
+                                    printError(msg);
+                                    writeLine(to->get(), err.dump());
+                                    // Don't forget to print it into the STDERR log, this is
+                                    // what's shown in the Hydra UI.
+                                    writeLine(to->get(), "restart");
+                                }
+                            });
+                        from = std::move(fromPipe.readSide);
+                        to = std::move(toPipe.writeSide);
+                        debug("created worker process %d", pid.get());
+                    }
+
+                    /* Check whether the existing worker process is still there. */
+                    auto s = readLine(from.get());
+                    if (s == "restart") {
+                        pid.wait();
+                        continue;
+                    } else if (s != "next") {
+                        auto json = nlohmann::json::parse(s);
+                        throw Error("worker error: %s", (std::string) json["error"]);
+                    }
+
+                    /* Wait for a job name to become available. */
+                    std::string attrPath;
+
+                    while (true) {
+                        checkInterrupt();
+                        auto state(state_.lock());
+                        if ((state->todo.empty() && state->active.empty()) || state->exc) {
+                            writeLine(to.get(), "exit");
+                            return;
+                        }
+                        if (!state->todo.empty()) {
+                            attrPath = *state->todo.begin();
+                            state->todo.erase(state->todo.begin());
+                            state->active.insert(attrPath);
+                            break;
+                        } else
+                            state.wait(wakeup);
+                    }
+
+                    /* Tell the worker to evaluate it. */
+                    writeLine(to.get(), "do " + attrPath);
+
+                    /* Wait for the response. */
+                    auto response = nlohmann::json::parse(readLine(from.get()));
+
+                    /* Handle the response. */
+                    StringSet newAttrs;
+
+                    if (response.find("job") != response.end()) {
+                        auto state(state_.lock());
+                        state->jobs[attrPath] = response["job"];
+                    }
+
+                    if (response.find("attrs") != response.end()) {
+                        for (auto & i : response["attrs"]) {
+                            std::string path = i;
+                            if (path.find(".") != std::string::npos){
+                                path = "\"" + path  + "\"";
+                            }
+                            auto s = (attrPath.empty() ? "" : attrPath + ".") + (std::string) path;
+                            newAttrs.insert(s);
+                        }
+                    }
+
+                    if (response.find("error") != response.end()) {
+                        auto state(state_.lock());
+                        state->jobs[attrPath]["error"] = response["error"];
+                    }
+
+                    /* Add newly discovered job names to the queue. */
+                    {
+                        auto state(state_.lock());
+                        state->active.erase(attrPath);
+                        for (auto & s : newAttrs)
+                            state->todo.insert(s);
+                        wakeup.notify_all();
+                    }
+                }
+            } catch (...) {
+                check_pid_status_nonblocking(pid.release());
+                auto state(state_.lock());
+                state->exc = std::current_exception();
+                wakeup.notify_all();
+            }
+        };
+
+        std::vector<std::thread> threads;
+        for (size_t i = 0; i < nrWorkers; i++)
+            threads.emplace_back(std::thread(handler));
+
+        for (auto & thread : threads)
+            thread.join();
+
+        auto state(state_.lock());
+
+        if (state->exc)
+            std::rethrow_exception(state->exc);
+
+        /* For aggregate jobs that have named consistuents
+           (i.e. constituents that are a job name rather than a
+           derivation), look up the referenced job and add it to the
+           dependencies of the aggregate derivation. */
+        auto store = openStore();
+
+        for (auto i = state->jobs.begin(); i != state->jobs.end(); ++i) {
+            auto jobName = i.key();
+            auto & job = i.value();
+
+            auto named = job.find("namedConstituents");
+            if (named == job.end()) continue;
+
+            std::unordered_map<std::string, std::string> brokenJobs;
+            auto getNonBrokenJobOrRecordError = [&brokenJobs, &jobName, &state](
+                    const std::string & childJobName) -> std::optional<nlohmann::json> {
+                auto childJob = state->jobs.find(childJobName);
+                if (childJob == state->jobs.end()) {
+                    printError("aggregate job '%s' references non-existent job '%s'", jobName, childJobName);
+                    brokenJobs[childJobName] = "does not exist";
+                    return std::nullopt;
+                }
+                if (childJob->find("error") != childJob->end()) {
+                    std::string error = (*childJob)["error"];
+                    printError("aggregate job '%s' references broken job '%s': %s", jobName, childJobName, error);
+                    brokenJobs[childJobName] = error;
+                    return std::nullopt;
+                }
+                return *childJob;
+            };
+
+            if (myArgs.dryRun) {
+                for (std::string jobName2 : *named) {
+                    auto job2 = getNonBrokenJobOrRecordError(jobName2);
+                    if (!job2) {
+                        continue;
+                    }
+                    std::string drvPath2 = (*job2)["drvPath"];
+                    job["constituents"].push_back(drvPath2);
+                }
+            } else {
+                auto drvPath = store->parseStorePath((std::string) job["drvPath"]);
+                auto drv = store->readDerivation(drvPath);
+
+                for (std::string jobName2 : *named) {
+                    auto job2 = getNonBrokenJobOrRecordError(jobName2);
+                    if (!job2) {
+                        continue;
+                    }
+                    auto drvPath2 = store->parseStorePath((std::string) (*job2)["drvPath"]);
+                    auto drv2 = store->readDerivation(drvPath2);
+                    job["constituents"].push_back(store->printStorePath(drvPath2));
+                    drv.inputDrvs.map[drvPath2].value = {drv2.outputs.begin()->first};
+                }
+
+                if (brokenJobs.empty()) {
+                    std::string drvName(drvPath.name());
+                    assert(drvName.ends_with(drvExtension));
+                    drvName.resize(drvName.size() - drvExtension.size());
+
+                    auto hashModulo = hashDerivationModulo(*store, drv, true);
+                    if (hashModulo.kind != DrvHash::Kind::Regular) continue;
+                    auto h = hashModulo.hashes.find("out");
+                    if (h == hashModulo.hashes.end()) continue;
+                    auto outPath = store->makeOutputPath("out", h->second, drvName);
+                    drv.env["out"] = store->printStorePath(outPath);
+                    drv.outputs.insert_or_assign("out", DerivationOutput::InputAddressed { .path = outPath });
+                    auto newDrvPath = store->printStorePath(writeDerivation(*store, drv));
+
+                    debug("rewrote aggregate derivation %s -> %s", store->printStorePath(drvPath), newDrvPath);
+
+                    job["drvPath"] = newDrvPath;
+                    job["outputs"]["out"] = store->printStorePath(outPath);
+                }
+            }
+
+            job.erase("namedConstituents");
+
+            /* Register the derivation as a GC root.  !!! This
+                registers roots for jobs that we may have already
+                done. */
+            auto localStore = store.dynamic_pointer_cast<LocalFSStore>();
+            if (gcRootsDir != "" && localStore) {
+                auto drvPath = job["drvPath"].get<std::string>();
+                Path root = gcRootsDir + "/" + std::string(baseNameOf(drvPath));
+                if (!pathExists(root))
+                    localStore->addPermRoot(localStore->parseStorePath(drvPath), root);
+            }
+
+            if (!brokenJobs.empty()) {
+                std::stringstream ss;
+                for (const auto& [jobName, error] : brokenJobs) {
+                    ss << jobName << ": " << error << "\n";
+                }
+                job["error"] = ss.str();
+            }
+        }
+
+        std::cout << state->jobs.dump(2) << "\n";
+    });
+}

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

@@ -121,7 +121,7 @@ static void copyClosureTo(
        the remote host to substitute missing paths. */
     // FIXME: substitute output pollutes our build log
     conn.to << ServeProto::Command::QueryValidPaths << 1 << useSubstitutes;
-    conn.to << ServeProto::write(destStore, conn, closure);
+    ServeProto::write(destStore, conn, closure);
     conn.to.flush();
 
     /* Get back the set of paths that are already valid on the remote
@@ -359,7 +359,7 @@ static std::map<StorePath, ValidPathInfo> queryPathInfos(
     /* Get info about each output path. */
     std::map<StorePath, ValidPathInfo> infos;
     conn.to << ServeProto::Command::QueryPathInfos;
-    conn.to << ServeProto::write(localStore, conn, outputs);
+    ServeProto::write(localStore, conn, outputs);
     conn.to.flush();
     while (true) {
         auto storePathS = readString(conn.from);

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

@@ -36,7 +36,7 @@ BuildOutput getBuildOutput(
             printInfo("fetching NAR contents of '%s'...", outputS);
             auto source = sinkToSource([&](Sink & sink)
             {
-                sink << store->narFromPath(output);
+                store->narFromPath(output, sink);
             });
             extractNarData(*source, outputS, narMembers);
         }

src/hydra-queue-runner/state.hh

@@ -328,6 +328,7 @@ struct Machine : nix::Machine
         operator nix::ServeProto::WriteConn ()
         {
             return {
+                .to = to,
                 .version = remoteVersion,
             };
         }

src/script/hydra-eval-jobset

@@ -357,26 +357,22 @@ sub evalJobs {
     my @cmd;
 
     if (defined $flakeRef) {
-        @cmd = ("nix-eval-jobs",
-                "--flake", $flakeRef . '#hydraJobs',
+        @cmd = ("hydra-eval-jobs",
+                "--flake", $flakeRef,
                 "--gc-roots-dir", getGCRootsDir,
-                "--max-jobs", 1,
-                "--meta",
-                "--force-recurse");
+                "--max-jobs", 1);
     } else {
         my $nixExprInput = $inputInfo->{$nixExprInputName}->[0]
             or die "cannot find the input containing the job expression\n";
 
-        @cmd = ("nix-eval-jobs",
+        @cmd = ("hydra-eval-jobs",
                 "<" . $nixExprInputName . "/" . $nixExprPath . ">",
                 "--gc-roots-dir", getGCRootsDir,
                 "--max-jobs", 1,
-                "--meta",
-                "--force-recurse",
                 inputsToArgs($inputInfo));
     }
 
-    #push @cmd, "--no-allow-import-from-derivation" if $config->{allow_import_from_derivation} // "true" ne "true";
+    push @cmd, "--no-allow-import-from-derivation" if $config->{allow_import_from_derivation} // "true" ne "true";
 
     if (defined $ENV{'HYDRA_DEBUG'}) {
         sub escape {
@@ -388,23 +384,14 @@ sub evalJobs {
         print STDERR "evaluator: @escaped\n";
     }
 
-    (my $res, my $jobsJSONLines, my $stderr) = captureStdoutStderr(21600, @cmd);
-    die "nix-eval-jobs returned " . ($res & 127 ? "signal $res" : "exit code " . ($res >> 8))
+    (my $res, my $jobsJSON, my $stderr) = captureStdoutStderr(21600, @cmd);
+    die "hydra-eval-jobs returned " . ($res & 127 ? "signal $res" : "exit code " . ($res >> 8))
         . ":\n" . ($stderr ? decode("utf-8", $stderr) : "(no output)\n")
         if $res;
 
     print STDERR "$stderr";
 
-    # XXX: take advantage of nix-eval-jobs's streaming instead of parsing everything in one block at
-    # the end.
-    my @jobs;
-    foreach my $line (split(/\n/, $jobsJSONLines)) {
-      last if $line eq "";
-
-      push(@jobs, decode_json($line));
-    };
-
-    return @jobs;
+    return decode_json($jobsJSON);
 }
 
 
@@ -433,7 +420,7 @@ sub checkBuild {
     my $firstOutputName = $outputNames[0];
     my $firstOutputPath = $buildInfo->{outputs}->{$firstOutputName};
 
-    my $jobName = $buildInfo->{attr} or die;
+    my $jobName = $buildInfo->{jobName} or die;
     my $drvPath = $buildInfo->{drvPath} or die;
 
     my $build;
@@ -487,30 +474,9 @@ sub checkBuild {
 
         my $time = time();
 
-        sub getMeta {
-            my ($s, $def) = @_;
-            return ($s || "") eq "" ? $def : $s;
-        }
-
-        sub getMetaStrings {
-            my ($v, $k, $acc) = @_;
-            my $t = ref $v;
-
-            if ($t eq 'HASH') {
-                push @$acc, $v->{$k} if exists $v->{$k};
-            } elsif ($t eq 'ARRAY') {
-                getMetaStrings($_, $k, $acc) foreach @$v;
-            } else {
-                push @$acc, $v;
-            }
-        }
-
-        sub getMetaConcatStrings {
-            my ($v, $k) = @_;
-
-            my @strings;
-            getMetaStrings($v, $k, \@strings);
-            return join(", ", @strings) || undef;
+        sub null {
+            my ($s) = @_;
+            return $s eq "" ? undef : $s;
         }
 
         # Add the build to the database.
@@ -518,19 +484,19 @@ sub checkBuild {
             { timestamp => $time
             , jobset_id => $jobset->id
             , job => $jobName
-            , description => getMeta($buildInfo->{meta}->{description}, undef)
-            , license => getMetaConcatStrings($buildInfo->{meta}->{license}, "shortName")
-            , homepage => getMeta($buildInfo->{meta}->{homepage}, undef)
-            , maintainers => getMetaConcatStrings($buildInfo->{meta}->{maintainers}, "email")
-            , maxsilent => getMeta($buildInfo->{meta}->{maxSilent}, 7200)
-            , timeout => getMeta($buildInfo->{meta}->{timeout}, 36000)
-            , nixname => $buildInfo->{name}
+            , description => null($buildInfo->{description})
+            , license => null($buildInfo->{license})
+            , homepage => null($buildInfo->{homepage})
+            , maintainers => null($buildInfo->{maintainers})
+            , maxsilent => $buildInfo->{maxSilent}
+            , timeout => $buildInfo->{timeout}
+            , nixname => $buildInfo->{nixName}
             , drvpath => $drvPath
             , system => $buildInfo->{system}
-            , priority => getMeta($buildInfo->{meta}->{schedulingPriority}, 100)
+            , priority => $buildInfo->{schedulingPriority}
             , finished => 0
             , iscurrent => 1
-            , ischannel => getMeta($buildInfo->{meta}->{isChannel}, 0)
+            , ischannel => $buildInfo->{isChannel}
             });
 
         $build->buildoutputs->create({ name => $_, path => $buildInfo->{outputs}->{$_} })
@@ -699,7 +665,7 @@ sub checkJobsetWrapped {
         return;
     }
 
-    # Hash the arguments to nix-eval-jobs and check the
+    # Hash the arguments to hydra-eval-jobs and check the
     # JobsetInputHashes to see if the previous evaluation had the same
     # inputs.  If so, bail out.
     my @args = ($jobset->nixexprinput // "", $jobset->nixexprpath // "", inputsToArgs($inputInfo));
@@ -721,18 +687,19 @@ sub checkJobsetWrapped {
 
     # Evaluate the job expression.
     my $evalStart = clock_gettime(CLOCK_MONOTONIC);
-    my @jobs = evalJobs($project->name . ":" . $jobset->name, $inputInfo, $jobset->nixexprinput, $jobset->nixexprpath, $flakeRef);
+    my $jobs = evalJobs($project->name . ":" . $jobset->name, $inputInfo, $jobset->nixexprinput, $jobset->nixexprpath, $flakeRef);
     my $evalStop = clock_gettime(CLOCK_MONOTONIC);
 
     if ($jobsetsJobset) {
+        my @keys = keys %$jobs;
         die "The .jobsets jobset must only have a single job named 'jobsets'"
-            unless (scalar @jobs) == 1 && $jobs[0]->{attr} eq "jobsets";
+            unless (scalar @keys) == 1 && $keys[0] eq "jobsets";
     }
     Net::Statsd::timing("hydra.evaluator.eval_time", int(($evalStop - $evalStart) * 1000));
 
     if ($dryRun) {
-        foreach my $job (@jobs) {
-            my $name = $job->{attr};
+        foreach my $name (keys %{$jobs}) {
+            my $job = $jobs->{$name};
             if (defined $job->{drvPath}) {
                 print STDERR "good job $name: $job->{drvPath}\n";
             } else {
@@ -742,6 +709,11 @@ sub checkJobsetWrapped {
         return;
     }
 
+    die "Jobset contains a job with an empty name. Make sure the jobset evaluates to an attrset of jobs.\n"
+        if defined $jobs->{""};
+
+    $jobs->{$_}->{jobName} = $_ for keys %{$jobs};
+
     my $jobOutPathMap = {};
     my $jobsetChanged = 0;
     my $dbStart = clock_gettime(CLOCK_MONOTONIC);
@@ -750,10 +722,10 @@ sub checkJobsetWrapped {
     # Store the error messages for jobs that failed to evaluate.
     my $evaluationErrorTime = time;
     my $evaluationErrorMsg = "";
-    foreach my $job (@jobs) {
+    foreach my $job (values %{$jobs}) {
         next unless defined $job->{error};
         $evaluationErrorMsg .=
-            ($job->{attr} ne "" ? "in job ‘$job->{attr}’" : "at top-level") .
+            ($job->{jobName} ne "" ? "in job ‘$job->{jobName}’" : "at top-level") .
             ":\n" . $job->{error} . "\n\n";
     }
     setJobsetError($jobset, $evaluationErrorMsg, $evaluationErrorTime);
@@ -788,7 +760,7 @@ sub checkJobsetWrapped {
             });
 
         # Schedule each successfully evaluated job.
-        foreach my $job (permute(@jobs)) {
+        foreach my $job (permute(values %{$jobs})) {
             next if defined $job->{error};
             #print STDERR "considering job " . $project->name, ":", $jobset->name, ":", $job->{jobName} . "\n";
             checkBuild($db, $jobset, $ev, $inputInfo, $job, \%buildMap, $prevEval, $jobOutPathMap, $plugins);
@@ -814,6 +786,40 @@ sub checkJobsetWrapped {
                 $ev->jobsetevalmembers->create({ build => $id, isnew => $x->{new} });
             }
 
+            # Create AggregateConstituents mappings.  Since there can
+            # be jobs that alias each other, if there are multiple
+            # builds for the same derivation, pick the one with the
+            # shortest name.
+            my %drvPathToId;
+            foreach my $id (keys %buildMap) {
+                my $x = $buildMap{$id};
+                my $y = $drvPathToId{$x->{drvPath}};
+                if (defined $y) {
+                    next if length $x->{jobName} > length $y->{jobName};
+                    next if length $x->{jobName} == length $y->{jobName} && $x->{jobName} ge $y->{jobName};
+                }
+                $drvPathToId{$x->{drvPath}} = $x;
+            }
+
+            foreach my $job (values %{$jobs}) {
+                next unless $job->{constituents};
+
+                if (defined $job->{error}) {
+                    die "aggregate job ‘$job->{jobName}’ failed with the error: $job->{error}\n";
+                }
+
+                my $x = $drvPathToId{$job->{drvPath}} or
+                    die "aggregate job ‘$job->{jobName}’ has no corresponding build record.\n";
+                foreach my $drvPath (@{$job->{constituents}}) {
+                    my $constituent = $drvPathToId{$drvPath};
+                    if (defined $constituent) {
+                        $db->resultset('AggregateConstituents')->update_or_create({aggregate => $x->{id}, constituent => $constituent->{id}});
+                    } else {
+                        warn "aggregate job ‘$job->{jobName}’ has a constituent ‘$drvPath’ that doesn't correspond to a Hydra build\n";
+                    }
+                }
+            }
+
             foreach my $name (keys %{$inputInfo}) {
                 for (my $n = 0; $n < scalar(@{$inputInfo->{$name}}); $n++) {
                     my $input = $inputInfo->{$name}->[$n];

t/Hydra/Controller/Build/api.t

@@ -78,4 +78,28 @@ subtest "validating the JSON representation of a build" => sub {
     }, "The build's JSON matches our API.");
 };
 
+subtest "accessing the constituents API" => sub {
+    my $url = $build_url . "/constituents";
+
+    my $constituents = request(GET $url,
+        Accept => 'application/json',
+    );
+
+    ok($constituents->is_success, "Getting the constituent builds");
+
+    my $data;
+    my $valid_json = lives { $data = decode_json($constituents->content); };
+    ok($valid_json, "We get back valid JSON.");
+    if (!$valid_json) {
+        use Data::Dumper;
+        print STDERR Dumper $constituents->content;
+    }
+
+    my ($buildA) = grep { $_->{nixname} eq "empty-dir-a" } @$data;
+    my ($buildB) = grep { $_->{nixname} eq "empty-dir-b" } @$data;
+
+    is($buildA->{job}, "a");
+    is($buildB->{job}, "b");
+};
+
 done_testing;

t/Makefile.am

@@ -13,7 +13,7 @@ TESTS_ENVIRONMENT = \
   PGHOST=/tmp	\
   PERL5LIB="$(srcdir):$(abs_top_srcdir)/src/lib:$$PERL5LIB"	\
   PYTHONPATH= \
-  PATH=$(abs_top_srcdir)/src/hydra-evaluator:$(abs_top_srcdir)/src/script:$(abs_top_srcdir)/src/hydra-queue-runner:$$PATH \
+  PATH=$(abs_top_srcdir)/src/hydra-evaluator:$(abs_top_srcdir)/src/script:$(abs_top_srcdir)/src/hydra-eval-jobs:$(abs_top_srcdir)/src/hydra-queue-runner:$$PATH \
   perl -w
 
 EXTRA_DIST = \

t/evaluator/evaluate-constituents-broken.t

@@ -0,0 +1,32 @@
+use strict;
+use warnings;
+use Setup;
+use Test2::V0;
+use Hydra::Helper::Exec;
+
+my $ctx = test_context();
+
+my $jobsetCtx = $ctx->makeJobset(
+    expression => 'constituents-broken.nix',
+);
+my $jobset = $jobsetCtx->{"jobset"};
+
+my ($res, $stdout, $stderr) = captureStdoutStderr(60,
+    ("hydra-eval-jobset", $jobsetCtx->{"project"}->name, $jobset->name)
+);
+isnt($res, 0, "hydra-eval-jobset exits non-zero");
+ok(utf8::decode($stderr), "Stderr output is UTF8-clean");
+like(
+    $stderr,
+    qr/aggregate job ‘mixed_aggregate’ failed with the error: constituentA: does not exist/,
+    "The stderr record includes a relevant error message"
+);
+
+$jobset->discard_changes({ '+columns' => {'errormsg' => 'errormsg'} });  # refresh from DB
+like(
+    $jobset->errormsg,
+    qr/aggregate job ‘mixed_aggregate’ failed with the error: constituentA: does not exist/,
+    "The jobset records a relevant error message"
+);
+
+done_testing;

t/evaluator/evaluate-constituents-gc.t

@@ -0,0 +1,20 @@
+use strict;
+use warnings;
+use Setup;
+use Test2::V0;
+
+my $ctx = test_context();
+
+my $builds = $ctx->makeAndEvaluateJobset(
+    expression => 'constituents.nix',
+);
+
+my $constituentA = $builds->{"constituentA"};
+my $directAggregate = $builds->{"direct_aggregate"};
+my $indirectAggregate = $builds->{"indirect_aggregate"};
+
+is(system('nix-store', '--delete', $constituentA->drvpath), 256, "Deleting a constituent derivation fails");
+is(system('nix-store', '--delete', $directAggregate->drvpath), 256, "Deleting the direct aggregate derivation fails");
+is(system('nix-store', '--delete', $indirectAggregate->drvpath), 256, "Deleting the indirect aggregate derivation fails");
+
+done_testing;

t/evaluator/evaluate-meta.t

@@ -1,22 +0,0 @@
-use feature 'unicode_strings';
-use strict;
-use warnings;
-use Setup;
-use Test2::V0;
-
-my $ctx = test_context();
-
-my $builds = $ctx->makeAndEvaluateJobset(
-    expression => "meta.nix",
-    build => 1
-);
-
-my $build = $builds->{"full-of-meta"};
-
-is($build->finished, 1, "Build should be finished.");
-is($build->description, "This is the description of the job.", "Wrong description extracted from the build.");
-is($build->license, "MIT, BSD", "Wrong licenses extracted from the build.");
-is($build->homepage, "https://example.com/", "Wrong homepage extracted from the build.");
-is($build->maintainers, 'alice@example.com, bob@not.found', "Wrong maintainers extracted from the build.");
-
-done_testing;

t/evaluator/evaluate-oom-job.t

@@ -0,0 +1,63 @@
+use strict;
+use warnings;
+use Setup;
+use Test2::V0;
+use Hydra::Helper::Exec;
+
+# Ensure that `systemd-run` is
+# - Available in the PATH/envionment
+# - Accessable to the user executing it
+# - Capable of using the command switches we use in our test
+my $sd_res;
+eval {
+  ($sd_res) = captureStdoutStderr(3, (
+    "systemd-run",
+      "--user",
+      "--collect",
+      "--scope",
+      "--property",
+        "MemoryMax=25M",
+    "--",
+    "true"
+  ));
+} or do {
+  # The command failed to execute, likely because `systemd-run` is not present
+  # in `PATH`
+  skip_all("`systemd-run` failed when invoked in this environment");
+};
+if ($sd_res != 0) {
+  # `systemd-run` executed but `sytemd-run` failed to call `true` and return
+  # successfully
+  skip_all("`systemd-run` returned non-zero when executing `true` (expected 0)");
+}
+
+my $ctx = test_context();
+
+# Contain the memory usage to 25 MegaBytes using `systemd-run`
+# Run `hydra-eval-jobs` on test job that will purposefully consume all memory
+# available
+my ($res, $stdout, $stderr) = captureStdoutStderr(60, (
+  "systemd-run",
+    "--user",
+    "--collect",
+    "--scope",
+    "--property",
+      "MemoryMax=25M",
+  "--",
+  "hydra-eval-jobs",
+    "-I", "/dev/zero",
+    "-I", $ctx->jobsdir,
+    ($ctx->jobsdir . "/oom.nix")
+));
+
+isnt($res, 0, "`hydra-eval-jobs` exits non-zero");
+ok(utf8::decode($stderr), "Stderr output is UTF8-clean");
+like(
+  $stderr,
+  # Assert error log contains messages added in PR
+  # https://github.com/NixOS/hydra/pull/1203
+  qr/^child process \(\d+?\) killed by signal=9$/m,
+  "The stderr record includes a relevant error message"
+);
+
+done_testing;

t/jobs/meta.nix

@@ -1,17 +0,0 @@
-with import ./config.nix;
-{
-  full-of-meta =
-    mkDerivation {
-      name = "full-of-meta";
-      builder = ./empty-dir-builder.sh;
-
-      meta = {
-        description = "This is the description of the job.";
-        license = [ { shortName = "MIT"; } "BSD" ];
-        homepage = "https://example.com/";
-        maintainers = [ "alice@example.com" { email = "bob@not.found"; } ];
-
-        outPath = "${placeholder "out"}";
-      };
-    };
-}

t/queue-runner/constituents.t

@@ -0,0 +1,32 @@
+use feature 'unicode_strings';
+use strict;
+use warnings;
+use Setup;
+
+my %ctx = test_init();
+
+require Hydra::Schema;
+require Hydra::Model::DB;
+
+use Test2::V0;
+
+my $db = Hydra::Model::DB->new;
+hydra_setup($db);
+
+my $project = $db->resultset('Projects')->create({name => "tests", displayname => "", owner => "root"});
+
+my $jobset = createBaseJobset("broken-constituent", "broken-constituent.nix", $ctx{jobsdir});
+
+ok(evalSucceeds($jobset),               "Evaluating jobs/broken-constituent.nix should exit with return code 0");
+is(nrQueuedBuildsForJobset($jobset), 0, "Evaluating jobs/broken-constituent.nix should not queue any builds");
+
+like(
+    $jobset->errormsg,
+    qr/^does-not-exist: does not exist$/m,
+    "Evaluating jobs/broken-constituent.nix should log an error for does-not-exist");
+like(
+    $jobset->errormsg,
+    qr/^does-not-evaluate: error: assertion 'false' failed$/m,
+    "Evaluating jobs/broken-constituent.nix should log an error for does-not-evaluate");
+
+done_testing;

t/queue-runner/direct-indirect-constituents.t

@@ -0,0 +1,35 @@
+use strict;
+use warnings;
+use Setup;
+use Test2::V0;
+
+my $ctx = test_context();
+
+my $builds = $ctx->makeAndEvaluateJobset(
+    expression => 'constituents.nix',
+);
+
+my $constituentBuildA = $builds->{"constituentA"};
+my $constituentBuildB = $builds->{"constituentB"};
+
+my $eval = $constituentBuildA->jobsetevals->first();
+is($eval->evaluationerror->has_error, 0);
+
+subtest "Verifying the direct aggregate" => sub {
+    my $aggBuild = $builds->{"direct_aggregate"};
+    is($aggBuild->constituents->first()->id, $constituentBuildA->id, "The ID of the constituent is correct");
+};
+
+subtest "Verifying the indirect aggregate" => sub {
+    my $indirectBuild = $builds->{"indirect_aggregate"};
+    is($indirectBuild->constituents->first()->id, $constituentBuildA->id, "The ID of the constituent is correct");
+};
+
+subtest "Verifying a mix of direct and indirect aggregate references" => sub {
+    my $mixedBuild = $builds->{"mixed_aggregate"};
+    my ($constituentA, $constituentB) = $mixedBuild->constituents()->search({}, {order_by => { -asc => "job"} });
+    is($constituentA->id, $constituentBuildA->id, "The ID of the constituent is correct");
+    is($constituentB->id, $constituentBuildB->id, "The ID of the constituent is correct");
+};
+
+done_testing;