Merge pull request #1460 from NixOS/nix-2.27

Nix 2.27

.gitignore

@@ -3,6 +3,7 @@
 /src/sql/hydra-postgresql.sql
 /src/sql/hydra-sqlite.sql
 /src/sql/tmp.sqlite
+.hydra-data
 result
 result-*
 outputs

README.md

@@ -72,17 +72,16 @@ Make sure **State** at the top of the page is set to "_Enabled_" and click on "_
 You can build Hydra via `nix-build` using the provided [default.nix](./default.nix):
 
 ```
-$ nix-build
+$ nix build
 ```
 
 ### Development Environment
 
 You can use the provided shell.nix to get a working development environment:
 ```
-$ nix-shell
-$ autoreconfPhase
-$ configurePhase # NOTE: not ./configure
-$ make
+$ nix develop
+$ mesonConfigurePhase
+$ ninja
 ```
 
 ### Executing Hydra During Development
@@ -91,9 +90,9 @@ When working on new features or bug fixes you need to be able to run Hydra from
 can be done using [foreman](https://github.com/ddollar/foreman):
 
 ```
-$ nix-shell
+$ nix develop
 $ # hack hack
-$ make
+$ ninja -C build
 $ foreman start
 ```
 
@@ -115,22 +114,24 @@ Start by following the steps in [Development Environment](#development-environme
 Then, you can run the tests and the perlcritic linter together with:
 
 ```console
-$ nix-shell
-$ make check
+$ nix develop
+$ ninja -C build test
 ```
 
 You can run a single test with:
 
 ```
-$ nix-shell
-$ yath test ./t/foo/bar.t
+$ nix develop
+$ cd build
+$ meson test --test-args=../t/Hydra/Event.t testsuite
 ```
 
 And you can run just perlcritic with:
 
 ```
-$ nix-shell
-$ make perlcritic
+$ nix develop
+$ cd build
+$ meson test perlcritic
 ```
 
 ### JSON API

doc/manual/src/hacking.md

@@ -11,12 +11,6 @@ $ cd hydra
 To enter a shell in which all environment variables (such as `PERL5LIB`)
 and dependencies can be found:
 
-```console
-$ nix-shell
-```
-
-of when flakes are enabled:
-
 ```console
 $ nix develop
 ```
@@ -24,15 +18,15 @@ $ nix develop
 To build Hydra, you should then do:
 
 ```console
-[nix-shell]$ autoreconfPhase
-[nix-shell]$ configurePhase
-[nix-shell]$ make -j$(nproc)
+$ mesonConfigurePhase
+$ ninja
 ```
 
 You start a local database, the webserver, and other components with
 foreman:
 
 ```console
+$ ninja -C build
 $ foreman start
 ```
 
@@ -47,18 +41,11 @@ $ ./src/script/hydra-server
 You can run Hydra's test suite with the following:
 
 ```console
-[nix-shell]$ make check
-[nix-shell]$ # to run as many tests as you have cores:
-[nix-shell]$ make check YATH_JOB_COUNT=$NIX_BUILD_CORES
-[nix-shell]$ # or run yath directly:
-[nix-shell]$ yath test
-[nix-shell]$ # to run as many tests as you have cores:
-[nix-shell]$ yath test -j $NIX_BUILD_CORES
+$ meson test
+# to run as many tests as you have cores:
+$ YATH_JOB_COUNT=$NIX_BUILD_CORES meson test
 ```
 
-When using `yath` instead of `make check`, ensure you have run `make`
-in the root of the repository at least once.
-
 **Warning**: Currently, the tests can fail
 if run with high parallelism [due to an issue in
 `Test::PostgreSQL`](https://github.com/TJC/Test-postgresql/issues/40)
@@ -75,7 +62,7 @@ will reload the page every time you save.
 To build Hydra and its dependencies:
 
 ```console
-$ nix-build release.nix -A build.x86_64-linux
+$ nix build .#packages.x86_64-linux.default
 ```
 
 ## Development Tasks

flake.lock

@@ -12,16 +12,16 @@
         "nixpkgs-regression": []
       },
       "locked": {
-        "lastModified": 1739899400,
-        "narHash": "sha256-q/RgA4bB7zWai4oPySq9mch7qH14IEeom2P64SXdqHs=",
+        "lastModified": 1744029599,
+        "narHash": "sha256-u6RhBWQ1XohTZ4Ub5ml1PTcaxQgtqFNng6Sohy1rojw=",
         "owner": "NixOS",
         "repo": "nix",
-        "rev": "e310c19a1aeb1ce1ed4d41d5ab2d02db596e0918",
+        "rev": "d0f98c76f962147610489e84c10033ca92e9c532",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "2.26-maintenance",
+        "ref": "2.27-maintenance",
         "repo": "nix",
         "type": "github"
       }
@@ -29,15 +29,16 @@
     "nix-eval-jobs": {
       "flake": false,
       "locked": {
-        "lastModified": 1739500569,
-        "narHash": "sha256-3wIReAqdTALv39gkWXLMZQvHyBOc3yPkWT2ZsItxedY=",
+        "lastModified": 1744009874,
+        "narHash": "sha256-PypQspB7h7EENe4RQQUQj2Ay8J1+O49AKNO9JbAU4Ek=",
         "owner": "nix-community",
         "repo": "nix-eval-jobs",
-        "rev": "4b392b284877d203ae262e16af269f702df036bc",
+        "rev": "62f9c9e8d00d2ff6ab27a6197ab459a8e0808e59",
         "type": "github"
       },
       "original": {
         "owner": "nix-community",
+        "ref": "v2.27.0",
         "repo": "nix-eval-jobs",
         "type": "github"
       }

flake.nix

@@ -4,7 +4,7 @@
   inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-24.11-small";
 
   inputs.nix = {
-    url = "github:NixOS/nix/2.26-maintenance";
+    url = "github:NixOS/nix/2.27-maintenance";
     inputs.nixpkgs.follows = "nixpkgs";
 
     # hide nix dev tooling from our lock file
@@ -16,7 +16,7 @@
   };
 
   inputs.nix-eval-jobs = {
-    url = "github:nix-community/nix-eval-jobs";
+    url = "github:nix-community/nix-eval-jobs/v2.27.0";
     # We want to control the deps precisely
     flake = false;
   };
@@ -85,7 +85,6 @@
             nix-util
             nix-store
             nix-main
-            nix-cmd
             nix-cli
             ;
           nix-perl-bindings = nix.hydraJobs.perlBindings.${system};

foreman/start-manual.sh

@@ -1,6 +1,6 @@
 #!/bin/sh
 
-exec mdbook serve \
+mdbook serve \
   --port 63332 \
   --dest-dir ./.hydra-data/manual \
   ./doc/manual/

nixos-modules/default.nix

@@ -15,7 +15,6 @@
     systemd.services.hydra-send-stats.enable = false;
 
     services.postgresql.enable = true;
-    services.postgresql.package = pkgs.postgresql_12;
 
     # The following is to work around the following error from hydra-server:
     #   [error] Caught exception in engine "Cannot determine local time zone"

nixos-modules/hydra.nix

@@ -468,7 +468,7 @@ in
             elif [[ $compression == zstd ]]; then
               compression="zstd --rm"
             fi
-            find ${baseDir}/build-logs -type f -name "*.drv" -mtime +3 -size +0c | xargs -r "$compression" --force --quiet
+            find ${baseDir}/build-logs -ignore_readdir_race -type f -name "*.drv" -mtime +3 -size +0c | xargs -r "$compression" --force --quiet
           '';
         startAt = "Sun 01:45";
       };

package.nix

@@ -11,7 +11,6 @@
 , nix-util
 , nix-store
 , nix-main
-, nix-cmd
 , nix-cli
 , nix-perl-bindings
 , git
@@ -179,7 +178,6 @@ stdenv.mkDerivation (finalAttrs: {
     nix-util
     nix-store
     nix-main
-    nix-cmd
     perlDeps
     perl
     boost
@@ -243,7 +241,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)/build/src/hydra-evaluator:$(pwd)/build/src/script:$(pwd)/build/src/hydra-queue-runner:$PATH
     PERL5LIB=$(pwd)/src/lib:$PERL5LIB
     export HYDRA_HOME="$(pwd)/src/"
     mkdir -p .hydra-data

src/hydra-build-step/hydra-build-step.cc

@@ -1,213 +0,0 @@
-/* This is a helper program that performs a build step, i.e. a single
-   derivation. In addition to a derivation path, it takes three store
-   URLs as arguments:
-
-   * --store: The store that will hold the resulting store paths
-       (typically a binary cache).
-
-   * --eval-store: The store that holds the .drv files, as produced by
-       hydra-evaluator.
-
-   * --build-store: The store that performs the build (often a
-       SSHStore for remote builds).
-
-   The build log is written to the path indicated by --log-file.
-*/
-
-#include "util.hh"
-#include "shared.hh"
-#include "common-eval-args.hh"
-#include "store-api.hh"
-#include "build-result.hh"
-#include "derivations.hh"
-#include "worker-protocol.hh"
-
-#include <chrono>
-
-using namespace nix;
-
-// FIXME: cut&paste
-static std::string_view getS(const std::vector<Logger::Field> & fields, size_t n)
-{
-    assert(n < fields.size());
-    assert(fields[n].type == Logger::Field::tString);
-    return fields[n].s;
-}
-
-void mainWrapped(std::list<std::string> args)
-{
-    verbosity = lvlError;
-
-    struct MyArgs : MixEvalArgs, MixCommonArgs, RootArgs
-    {
-        Path drvPath;
-        std::optional<std::string> buildStoreUrl;
-        std::optional<Path> logPath;
-        std::optional<uint64_t> maxOutputSize;
-
-        MyArgs() : MixCommonArgs("hydra-build-step")
-        {
-            expectArg("drv-path", &drvPath);
-
-            addFlag({
-                .longName = "build-store",
-                .description = "The Nix store to use for building the derivation.",
-                //.category = category,
-                .labels = {"store-url"},
-                .handler = {&buildStoreUrl},
-            });
-
-            addFlag({
-                .longName = "log-file",
-                .description = "The path to the build log.",
-                .labels = {"path"},
-                .handler = {&logPath},
-            });
-
-            addFlag({
-                .longName = "max-output-size",
-                .description = "Maximum size of the outputs.",
-                .labels = {"bytes"},
-                .handler = {&maxOutputSize},
-            });
-        }
-    };
-
-    /* A logger that intercepts all build log lines and writes them to
-       the log file. */
-    MyArgs myArgs;
-    myArgs.parseCmdline(args);
-
-    struct MyLogger : public Logger
-    {
-        Logger & prev;
-        AutoCloseFD logFile;
-
-        MyLogger(Logger & prev, Path logPath) : prev(prev)
-        {
-            logFile = open(logPath.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0666);
-            if (!logFile)
-                throw SysError("creating log file '%s'", logPath);
-        }
-
-        void log(Verbosity lvl, std::string_view s) override
-        { prev.log(lvl, s); }
-
-        void logEI(const ErrorInfo & ei) override
-        { prev.logEI(ei); }
-
-        void writeToStdout(std::string_view s) override
-        { prev.writeToStdout(s); }
-
-        void result(ActivityId act, ResultType type, const Fields & fields) override
-        {
-            if (type == resBuildLogLine)
-                writeLine(logFile.get(), std::string(getS(fields, 0)));
-            else
-                prev.result(act, type, fields);
-        }
-    };
-
-    auto destStore = openStore();
-    auto evalStore = myArgs.evalStoreUrl ? openStore(*myArgs.evalStoreUrl) : destStore;
-    auto buildStore = myArgs.buildStoreUrl ? openStore(*myArgs.buildStoreUrl) : destStore;
-
-    auto drvPath = evalStore->parseStorePath(myArgs.drvPath);
-
-    auto drv = evalStore->readDerivation(drvPath);
-    BasicDerivation basicDrv(drv);
-
-    uint64_t overhead = 0;
-
-    /* Gather the inputs. */
-    StorePathSet inputs;
-
-    for (auto & p : drv.inputSrcs)
-        inputs.insert(p);
-
-    for (auto & [drvPath, node] : drv.inputDrvs.map) {
-        auto drv2 = evalStore->readDerivation(drvPath);
-        for (auto & name : node.value) {
-            if (auto i = get(drv2.outputs, name)) {
-                auto outPath = i->path(*evalStore, drv2.name, name);
-                inputs.insert(*outPath);
-                basicDrv.inputSrcs.insert(*outPath);
-            }
-        }
-    }
-
-    /* Ensure that the inputs exist in the destination store (so that
-       the builder can substitute them from the destination
-       store). This is a no-op for regular stores, but for the binary
-       cache store, this will copy the inputs to the binary cache from
-       the local store. */
-    {
-        auto now1 = std::chrono::steady_clock::now();
-
-        debug("sending closure of '%s' to '%s'",
-            evalStore->printStorePath(drvPath), destStore->getUri());
-
-        if (evalStore != destStore)
-            copyClosure(*evalStore, *destStore, drv.inputSrcs, NoRepair, NoCheckSigs);
-
-        copyClosure(*destStore, *buildStore, inputs, NoRepair, NoCheckSigs, Substitute);
-
-        auto now2 = std::chrono::steady_clock::now();
-
-        overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
-    }
-
-    /* Perform the build. */
-    if (myArgs.logPath)
-        logger = new MyLogger(*logger, *myArgs.logPath);
-
-    auto buildResult = buildStore->buildDerivation(drvPath, basicDrv);
-
-    /* Copy the output paths from the build store to the destination
-       store. */
-    size_t totalNarSize = 0;
-
-    if (buildResult.success()) {
-
-        std::map<StorePath, ValidPathInfo> infos;
-        StorePathSet outputs;
-        for (auto & [output, realisation] : buildResult.builtOutputs) {
-            auto info = buildStore->queryPathInfo(realisation.outPath);
-            totalNarSize += info->narSize;
-            infos.insert_or_assign(info->path, *info);
-            outputs.insert(info->path);
-        }
-
-        if ((!myArgs.maxOutputSize || totalNarSize <= *myArgs.maxOutputSize)
-            && buildStore != destStore)
-        {
-            debug("copying outputs of '%s' from '%s' (%d bytes)",
-                buildStore->printStorePath(drvPath), buildStore->getUri(), totalNarSize);
-
-            auto now1 = std::chrono::steady_clock::now();
-
-            copyPaths(*buildStore, *destStore, outputs, NoRepair, NoCheckSigs);
-
-            auto now2 = std::chrono::steady_clock::now();
-
-            overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
-        }
-    }
-
-    FdSink to { STDOUT_FILENO };
-    WorkerProto::WriteConn wconn {
-        .to = to,
-        // Hardcode latest version because we are deploying hydra
-        // itself atomically
-        .version = PROTOCOL_VERSION,
-    };
-    WorkerProto::write(*evalStore, wconn, buildResult);
-}
-
-int main(int argc, char * * argv)
-{
-    return handleExceptions(argv[0], [&]() {
-        initNix();
-        mainWrapped(argvToStrings(argc, argv));
-    });
-}

src/hydra-build-step/meson.build

@@ -1,14 +0,0 @@
-srcs = files(
-  'hydra-build-step.cc',
-)
-
-hydra_build_step = executable('hydra-build-step',
-  'hydra-build-step.cc',
-  srcs,
-  dependencies: [
-    libhydra_dep,
-    nix_dep,
-    dependency('nix-cmd', required: true)
-  ],
-  install: true,
-)

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

@@ -1,27 +1,330 @@
-#include <math.h>
+#include <algorithm>
+#include <cmath>
+
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
 
 #include "build-result.hh"
+#include "path.hh"
+#include "legacy-ssh-store.hh"
 #include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
 #include "state.hh"
 #include "current-process.hh"
 #include "processes.hh"
 #include "util.hh"
+#include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
+#include "ssh.hh"
 #include "finally.hh"
 #include "url.hh"
-#include "worker-protocol.hh"
 
 using namespace nix;
 
+bool ::Machine::isLocalhost() const
+{
+    return storeUri.params.empty() && std::visit(overloaded {
+        [](const StoreReference::Auto &) {
+            return true;
+        },
+        [](const StoreReference::Specified & s) {
+            return
+               (s.scheme == "local" || s.scheme == "unix") ||
+               ((s.scheme == "ssh" || s.scheme == "ssh-ng") &&
+                s.authority == "localhost");
+        },
+    }, storeUri.variant);
+}
+
 namespace nix::build_remote {
 
-static Path createLogFileDir(const std::string & logDir, const StorePath & drvPath)
+static std::unique_ptr<SSHMaster::Connection> openConnection(
+    ::Machine::ptr machine, SSHMaster & master)
+{
+    Strings command = {"nix-store", "--serve", "--write"};
+    if (machine->isLocalhost()) {
+        command.push_back("--builders");
+        command.push_back("");
+    } else {
+        auto remoteStore = machine->storeUri.params.find("remote-store");
+        if (remoteStore != machine->storeUri.params.end()) {
+            command.push_back("--store");
+            command.push_back(shellEscape(remoteStore->second));
+        }
+    }
+
+    auto ret = master.startCommand(std::move(command), {
+        "-a", "-oBatchMode=yes", "-oConnectTimeout=60", "-oTCPKeepAlive=yes"
+    });
+
+    // XXX: determine the actual max value we can use from /proc.
+
+    // FIXME: Should this be upstreamed into `startCommand` in Nix?
+
+    int pipesize = 1024 * 1024;
+
+    fcntl(ret->in.get(), F_SETPIPE_SZ, &pipesize);
+    fcntl(ret->out.get(), F_SETPIPE_SZ, &pipesize);
+
+    return ret;
+}
+
+
+static void copyClosureTo(
+    ::Machine::Connection & conn,
+    Store & destStore,
+    const StorePathSet & paths,
+    SubstituteFlag useSubstitutes = NoSubstitute)
+{
+    StorePathSet closure;
+    destStore.computeFSClosure(paths, closure);
+
+    /* Send the "query valid paths" command with the "lock" option
+       enabled. This prevents a race where the remote host
+       garbage-collect paths that are already there. Optionally, ask
+       the remote host to substitute missing paths. */
+    // FIXME: substitute output pollutes our build log
+    /* Get back the set of paths that are already valid on the remote
+       host. */
+    auto present = conn.queryValidPaths(
+        destStore, true, closure, useSubstitutes);
+
+    if (present.size() == closure.size()) return;
+
+    auto sorted = destStore.topoSortPaths(closure);
+
+    StorePathSet missing;
+    for (auto i = sorted.rbegin(); i != sorted.rend(); ++i)
+        if (!present.count(*i)) missing.insert(*i);
+
+    printMsg(lvlDebug, "sending %d missing paths", missing.size());
+
+    std::unique_lock<std::timed_mutex> sendLock(conn.machine->state->sendLock,
+        std::chrono::seconds(600));
+
+    conn.to << ServeProto::Command::ImportPaths;
+    destStore.exportPaths(missing, conn.to);
+    conn.to.flush();
+
+    if (readInt(conn.from) != 1)
+        throw Error("remote machine failed to import closure");
+}
+
+
+// FIXME: use Store::topoSortPaths().
+static StorePaths reverseTopoSortPaths(const std::map<StorePath, UnkeyedValidPathInfo> & paths)
+{
+    StorePaths sorted;
+    StorePathSet visited;
+
+    std::function<void(const StorePath & path)> dfsVisit;
+
+    dfsVisit = [&](const StorePath & path) {
+        if (!visited.insert(path).second) return;
+
+        auto info = paths.find(path);
+        auto references = info == paths.end() ? StorePathSet() : info->second.references;
+
+        for (auto & i : references)
+            /* Don't traverse into paths that don't exist.  That can
+               happen due to substitutes for non-existent paths. */
+            if (i != path && paths.count(i))
+                dfsVisit(i);
+
+        sorted.push_back(path);
+    };
+
+    for (auto & i : paths)
+        dfsVisit(i.first);
+
+    return sorted;
+}
+
+static std::pair<Path, AutoCloseFD> openLogFile(const std::string & logDir, const StorePath & drvPath)
 {
     std::string base(drvPath.to_string());
     auto logFile = logDir + "/" + std::string(base, 0, 2) + "/" + std::string(base, 2);
 
     createDirs(dirOf(logFile));
 
-    return logFile;
+    AutoCloseFD logFD = open(logFile.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0666);
+    if (!logFD) throw SysError("creating log file ‘%s’", logFile);
+
+    return {std::move(logFile), std::move(logFD)};
+}
+
+static BasicDerivation sendInputs(
+    State & state,
+    Step & step,
+    Store & localStore,
+    Store & destStore,
+    ::Machine::Connection & conn,
+    unsigned int & overhead,
+    counter & nrStepsWaiting,
+    counter & nrStepsCopyingTo
+)
+{
+    /* Replace the input derivations by their output paths to send a
+       minimal closure to the builder.
+
+       `tryResolve` currently does *not* rewrite input addresses, so it
+       is safe to do this in all cases. (It should probably have a mode
+       to do that, however, but we would not use it here.)
+     */
+    BasicDerivation basicDrv = ({
+        auto maybeBasicDrv = step.drv->tryResolve(destStore, &localStore);
+        if (!maybeBasicDrv)
+            throw Error(
+                "the derivation '%s' can’t be resolved. It’s probably "
+                "missing some outputs",
+                localStore.printStorePath(step.drvPath));
+        *maybeBasicDrv;
+    });
+
+    /* Ensure that the inputs exist in the destination store. This is
+       a no-op for regular stores, but for the binary cache store,
+       this will copy the inputs to the binary cache from the local
+       store. */
+    if (&localStore != &destStore) {
+        copyClosure(localStore, destStore,
+            step.drv->inputSrcs,
+            NoRepair, NoCheckSigs, NoSubstitute);
+    }
+
+    {
+        auto mc1 = std::make_shared<MaintainCount<counter>>(nrStepsWaiting);
+        mc1.reset();
+        MaintainCount<counter> mc2(nrStepsCopyingTo);
+
+        printMsg(lvlDebug, "sending closure of ‘%s’ to ‘%s’",
+            localStore.printStorePath(step.drvPath), conn.machine->storeUri.render());
+
+        auto now1 = std::chrono::steady_clock::now();
+
+        /* Copy the input closure. */
+        if (conn.machine->isLocalhost()) {
+            StorePathSet closure;
+            destStore.computeFSClosure(basicDrv.inputSrcs, closure);
+            copyPaths(destStore, localStore, closure, NoRepair, NoCheckSigs, NoSubstitute);
+        } else {
+            copyClosureTo(conn, destStore, basicDrv.inputSrcs, Substitute);
+        }
+
+        auto now2 = std::chrono::steady_clock::now();
+
+        overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
+    }
+
+    return basicDrv;
+}
+
+static BuildResult performBuild(
+    ::Machine::Connection & conn,
+    Store & localStore,
+    StorePath drvPath,
+    const BasicDerivation & drv,
+    const ServeProto::BuildOptions & options,
+    counter & nrStepsBuilding
+)
+{
+    conn.putBuildDerivationRequest(localStore, drvPath, drv, options);
+
+    BuildResult result;
+
+    time_t startTime, stopTime;
+
+    startTime = time(0);
+    {
+        MaintainCount<counter> mc(nrStepsBuilding);
+        result = ServeProto::Serialise<BuildResult>::read(localStore, conn);
+    }
+    stopTime = time(0);
+
+    if (!result.startTime) {
+        // If the builder gave `startTime = 0`, use our measurements
+        // instead of the builder's.
+        //
+        // Note: this represents the duration of a single round, rather
+        // than all rounds.
+        result.startTime = startTime;
+        result.stopTime = stopTime;
+    }
+
+    // If the protocol was too old to give us `builtOutputs`, initialize
+    // it manually by introspecting the derivation.
+    if (GET_PROTOCOL_MINOR(conn.remoteVersion) < 6)
+    {
+        // 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(localStore);
+        // Since this a `BasicDerivation`, `staticOutputHashes` will not
+        // do any real work.
+        auto outputHashes = staticOutputHashes(localStore, 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 };
+            result.builtOutputs.insert_or_assign(
+                std::move(outputName),
+                Realisation { drvOutput, *outputPath });
+        }
+    }
+
+    return result;
+}
+
+static void copyPathFromRemote(
+    ::Machine::Connection & conn,
+    NarMemberDatas & narMembers,
+    Store & localStore,
+    Store & destStore,
+    const ValidPathInfo & info
+)
+{
+      /* Receive the NAR from the remote and add it to the
+          destination store. Meanwhile, extract all the info from the
+          NAR that getBuildOutput() needs. */
+      auto source2 = sinkToSource([&](Sink & sink)
+      {
+          /* Note: we should only send the command to dump the store
+              path to the remote if the NAR is actually going to get read
+              by the destination store, which won't happen if this path
+              is already valid on the destination store. Since this
+              lambda function only gets executed if someone tries to read
+              from source2, we will send the command from here rather
+              than outside the lambda. */
+          conn.to << ServeProto::Command::DumpStorePath << localStore.printStorePath(info.path);
+          conn.to.flush();
+
+          TeeSource tee(conn.from, sink);
+          extractNarData(tee, localStore.printStorePath(info.path), narMembers);
+      });
+
+      destStore.addToStore(info, *source2, NoRepair, NoCheckSigs);
+}
+
+static void copyPathsFromRemote(
+    ::Machine::Connection & conn,
+    NarMemberDatas & narMembers,
+    Store & localStore,
+    Store & destStore,
+    const std::map<StorePath, UnkeyedValidPathInfo> & infos
+)
+{
+      auto pathsSorted = reverseTopoSortPaths(infos);
+
+      for (auto & path : pathsSorted) {
+          auto & info = infos.find(path)->second;
+          copyPathFromRemote(
+              conn, narMembers, localStore, destStore,
+              ValidPathInfo { path, info });
+      }
+
 }
 
 }
@@ -30,14 +333,11 @@ static Path createLogFileDir(const std::string & logDir, const StorePath & drvPa
 
 void RemoteResult::updateWithBuildResult(const nix::BuildResult & buildResult)
 {
-    // FIXME: make RemoteResult inherit BuildResult.
+    startTime = buildResult.startTime;
+    stopTime = buildResult.stopTime;
     timesBuilt = buildResult.timesBuilt;
     errorMsg = buildResult.errorMsg;
     isNonDeterministic = buildResult.isNonDeterministic;
-    if (buildResult.startTime && buildResult.stopTime) {
-        startTime = buildResult.startTime;
-        stopTime = buildResult.stopTime;
-    }
 
     switch ((BuildResult::Status) buildResult.status) {
         case BuildResult::Built:
@@ -96,44 +396,38 @@ void State::buildRemote(ref<Store> destStore,
 {
     assert(BuildResult::TimedOut == 8);
 
-    result.logFile = build_remote::createLogFileDir(logDir, step->drvPath);
+    auto [logFile, logFD] = build_remote::openLogFile(logDir, step->drvPath);
+    AutoDelete logFileDel(logFile, false);
+    result.logFile = logFile;
 
     try {
 
-        updateStep(ssBuilding);
-        result.startTime = time(0);
+        updateStep(ssConnecting);
 
-        auto buildStoreUrl = machine->completeStoreReference().render();
+        auto storeRef = machine->completeStoreReference();
 
-        Strings args = {
-            localStore->printStorePath(step->drvPath),
-            "--store", destStore->getUri(),
-            "--eval-store", localStore->getUri(),
-            "--build-store", buildStoreUrl,
-            "--max-silent-time", std::to_string(buildOptions.maxSilentTime),
-            "--timeout", std::to_string(buildOptions.buildTimeout),
-            "--max-build-log-size", std::to_string(buildOptions.maxLogSize),
-            "--max-output-size", std::to_string(maxOutputSize),
-            "--repeat", std::to_string(buildOptions.nrRepeats),
-            "--log-file", result.logFile,
-            // FIXME: step->isDeterministic
+        auto * pSpecified = std::get_if<StoreReference::Specified>(&storeRef.variant);
+        if (!pSpecified || pSpecified->scheme != "ssh") {
+            throw Error("Currently, only (legacy-)ssh stores are supported!");
+        }
+
+        LegacySSHStoreConfig storeConfig {
+            pSpecified->scheme,
+            pSpecified->authority,
+            storeRef.params
         };
 
-        // FIXME: set pid for cancellation
+        auto master = storeConfig.createSSHMaster(
+            false, // no SSH master yet
+            logFD.get());
 
-        auto [status, childStdout] = [&]() {
-            MaintainCount<counter> mc(nrStepsBuilding);
-            return runProgram({
-                .program = "hydra-build-step",
-                .args = std::move(args),
-            });
-        }();
+        // FIXME: rewrite to use Store.
+        auto child = build_remote::openConnection(machine, master);
 
-        #if 0
         {
             auto activeStepState(activeStep->state_.lock());
             if (activeStepState->cancelled) throw Error("step cancelled");
-            activeStepState->pid = conn.store->getConnectionPid();
+            activeStepState->pid = child->sshPid;
         }
 
         Finally clearPid([&]() {
@@ -147,32 +441,76 @@ void State::buildRemote(ref<Store> destStore,
                possibility that we end up killing another
                process. Meh. */
         });
-        #endif
 
-        result.stopTime = time(0);
+        ::Machine::Connection conn {
+            {
+                .to = child->in.get(),
+                .from = child->out.get(),
+                /* Handshake. */
+                .remoteVersion = 0xdadbeef, // FIXME avoid dummy initialize
+            },
+            /*.machine =*/ machine,
+        };
 
-        if (!statusOk(status))
-            throw ExecError(status, fmt("hydra-build-step %s with output:\n%s", statusToString(status), stdout));
+        Finally updateStats([&]() {
+            bytesReceived += conn.from.read;
+            bytesSent += conn.to.written;
+        });
+
+        constexpr ServeProto::Version our_version = 0x206;
+
+        try {
+            conn.remoteVersion = decltype(conn)::handshake(
+                conn.to,
+                conn.from,
+                our_version,
+                machine->storeUri.render());
+        } catch (EndOfFile & e) {
+            child->sshPid.wait();
+            std::string s = chomp(readFile(result.logFile));
+            throw Error("cannot connect to ‘%1%’: %2%", machine->storeUri.render(), s);
+        }
 
-        /* The build was executed successfully, so clear the failure
-           count for this machine. */
         {
             auto info(machine->state->connectInfo.lock());
             info->consecutiveFailures = 0;
         }
 
-        StringSource from { childStdout };
-        /* Read the BuildResult from the child. */
-        WorkerProto::ReadConn rconn {
-            .from = from,
-            // Hardcode latest version because we are deploying hydra
-            // itself atomically
-            .version = PROTOCOL_VERSION,
-        };
-        result.overhead += readNum<uint64_t>(rconn.from);
-        auto totalNarSize = readNum<uint64_t>(rconn.from);
-        auto buildResult = WorkerProto::Serialise<BuildResult>::read(*localStore, rconn);
+        /* Gather the inputs. If the remote side is Nix <= 1.9, we have to
+           copy the entire closure of ‘drvPath’, as well as the required
+           outputs of the input derivations. On Nix > 1.9, we only need to
+           copy the immediate sources of the derivation and the required
+           outputs of the input derivations. */
+        updateStep(ssSendingInputs);
+        BasicDerivation resolvedDrv = build_remote::sendInputs(*this, *step, *localStore, *destStore, conn, result.overhead, nrStepsWaiting, nrStepsCopyingTo);
 
+        logFileDel.cancel();
+
+        /* Truncate the log to get rid of messages about substitutions
+            etc. on the remote system. */
+        if (lseek(logFD.get(), SEEK_SET, 0) != 0)
+            throw SysError("seeking to the start of log file ‘%s’", result.logFile);
+
+        if (ftruncate(logFD.get(), 0) == -1)
+            throw SysError("truncating log file ‘%s’", result.logFile);
+
+        logFD = -1;
+
+        /* Do the build. */
+        printMsg(lvlDebug, "building ‘%s’ on ‘%s’",
+            localStore->printStorePath(step->drvPath),
+            machine->storeUri.render());
+
+        updateStep(ssBuilding);
+
+        BuildResult buildResult = build_remote::performBuild(
+            conn,
+            *localStore,
+            step->drvPath,
+            resolvedDrv,
+            buildOptions,
+            nrStepsBuilding
+        );
 
         result.updateWithBuildResult(buildResult);
 
@@ -180,13 +518,6 @@ void State::buildRemote(ref<Store> destStore,
 
         result.errorMsg = "";
 
-        /* If the NAR size limit was exceeded, then hydra-build-step
-           will not have copied the output paths. */
-        if (totalNarSize > maxOutputSize) {
-            result.stepStatus = bsNarSizeLimitExceeded;
-            return;
-        }
-
         /* If the path was substituted or already valid, then we didn't
            get a build log. */
         if (result.isCached) {
@@ -196,6 +527,57 @@ void State::buildRemote(ref<Store> destStore,
             result.logFile = "";
         }
 
+        StorePathSet outputs;
+        for (auto & [_, realisation] : buildResult.builtOutputs)
+            outputs.insert(realisation.outPath);
+
+        /* Copy the output paths. */
+        if (!machine->isLocalhost() || localStore != std::shared_ptr<Store>(destStore)) {
+            updateStep(ssReceivingOutputs);
+
+            MaintainCount<counter> mc(nrStepsCopyingFrom);
+
+            auto now1 = std::chrono::steady_clock::now();
+
+            auto infos = conn.queryPathInfos(*localStore, outputs);
+
+            size_t totalNarSize = 0;
+            for (auto & [_, info] : infos) totalNarSize += info.narSize;
+
+            if (totalNarSize > maxOutputSize) {
+                result.stepStatus = bsNarSizeLimitExceeded;
+                return;
+            }
+
+            /* Copy each path. */
+            printMsg(lvlDebug, "copying outputs of ‘%s’ from ‘%s’ (%d bytes)",
+                localStore->printStorePath(step->drvPath), machine->storeUri.render(), totalNarSize);
+
+            build_remote::copyPathsFromRemote(conn, narMembers, *localStore, *destStore, infos);
+            auto now2 = std::chrono::steady_clock::now();
+
+            result.overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
+        }
+
+        /* Register the outputs of the newly built drv */
+        if (experimentalFeatureSettings.isEnabled(Xp::CaDerivations)) {
+            auto outputHashes = staticOutputHashes(*localStore, *step->drv);
+            for (auto & [outputName, realisation] : buildResult.builtOutputs) {
+                // Register the resolved drv output
+                destStore->registerDrvOutput(realisation);
+
+                // Also register the unresolved one
+                auto unresolvedRealisation = realisation;
+                unresolvedRealisation.signatures.clear();
+                unresolvedRealisation.id.drvHash = outputHashes.at(outputName);
+                destStore->registerDrvOutput(unresolvedRealisation);
+            }
+        }
+
+        /* Shut down the connection. */
+        child->in = -1;
+        child->sshPid.wait();
+
     } catch (Error & e) {
         /* Disable this machine until a certain period of time has
            passed. This period increases on every consecutive

src/hydra-queue-runner/builder.cc

@@ -3,7 +3,6 @@
 #include "state.hh"
 #include "hydra-build-result.hh"
 #include "finally.hh"
-#include "terminal.hh"
 #include "binary-cache-store.hh"
 
 using namespace nix;
@@ -220,7 +219,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
                 result.canRetry = false;
             } else {
                 result.stepStatus = bsAborted;
-                result.errorMsg = filterANSIEscapes(e.msg(), true);
+                result.errorMsg = e.msg();
                 result.canRetry = true;
             }
         }

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

@@ -550,11 +550,12 @@ void State::dumpStatus(Connection & conn)
         {"nrQueuedBuilds", builds.lock()->size()},
         {"nrActiveSteps", activeSteps_.lock()->size()},
         {"nrStepsBuilding", nrStepsBuilding.load()},
-        #if 0
         {"nrStepsCopyingTo", nrStepsCopyingTo.load()},
         {"nrStepsCopyingFrom", nrStepsCopyingFrom.load()},
-        #endif
+        {"nrStepsWaiting", nrStepsWaiting.load()},
         {"nrUnsupportedSteps", nrUnsupportedSteps.load()},
+        {"bytesSent", bytesSent.load()},
+        {"bytesReceived", bytesReceived.load()},
         {"nrBuildsRead", nrBuildsRead.load()},
         {"buildReadTimeMs", buildReadTimeMs.load()},
         {"buildReadTimeAvgMs", nrBuildsRead == 0 ? 0.0 : (float) buildReadTimeMs / nrBuildsRead},

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

@@ -1,6 +1,7 @@
 #include "state.hh"
 #include "hydra-build-result.hh"
 #include "globals.hh"
+#include "parsed-derivations.hh"
 
 #include <cstring>
 
@@ -463,14 +464,17 @@ Step::ptr State::createStep(ref<Store> destStore,
        it's not runnable yet, and other threads won't make it
        runnable while step->created == false. */
     step->drv = std::make_unique<Derivation>(localStore->readDerivation(drvPath));
-    step->parsedDrv = std::make_unique<ParsedDerivation>(drvPath, *step->drv);
+    {
+        auto parsedDrv = ParsedDerivation{drvPath, *step->drv};
+        step->drvOptions = std::make_unique<DerivationOptions>(DerivationOptions::fromParsedDerivation(parsedDrv));
+    }
 
-    step->preferLocalBuild = step->parsedDrv->willBuildLocally(*localStore);
+    step->preferLocalBuild = step->drvOptions->willBuildLocally(*localStore, *step->drv);
     step->isDeterministic = getOr(step->drv->env, "isDetermistic", "0") == "1";
 
     step->systemType = step->drv->platform;
     {
-        StringSet features = step->requiredSystemFeatures = step->parsedDrv->getRequiredSystemFeatures();
+        StringSet features = step->requiredSystemFeatures = step->drvOptions->getRequiredSystemFeatures(*step->drv);
         if (step->preferLocalBuild)
             features.insert("local");
         if (!features.empty()) {

src/hydra-queue-runner/state.hh

@@ -13,7 +13,8 @@
 
 #include "db.hh"
 
-#include "parsed-derivations.hh"
+#include "derivations.hh"
+#include "derivation-options.hh"
 #include "pathlocks.hh"
 #include "pool.hh"
 #include "build-result.hh"
@@ -21,6 +22,8 @@
 #include "sync.hh"
 #include "nar-extractor.hh"
 #include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
+#include "serve-protocol-connection.hh"
 #include "machines.hh"
 
 
@@ -165,7 +168,7 @@ struct Step
 
     nix::StorePath drvPath;
     std::unique_ptr<nix::Derivation> drv;
-    std::unique_ptr<nix::ParsedDerivation> parsedDrv;
+    std::unique_ptr<nix::DerivationOptions> drvOptions;
     std::set<std::string> requiredSystemFeatures;
     bool preferLocalBuild;
     bool isDeterministic;
@@ -288,6 +291,12 @@ struct Machine : nix::Machine
     }
 
     bool isLocalhost() const;
+
+    // A connection to a machine
+    struct Connection : nix::ServeProto::BasicClientConnection {
+        // Backpointer to the machine
+        ptr machine;
+    };
 };
 
 
@@ -352,10 +361,9 @@ private:
     counter nrStepsStarted{0};
     counter nrStepsDone{0};
     counter nrStepsBuilding{0};
-    #if 0
     counter nrStepsCopyingTo{0};
     counter nrStepsCopyingFrom{0};
-    #endif
+    counter nrStepsWaiting{0};
     counter nrUnsupportedSteps{0};
     counter nrRetries{0};
     counter maxNrRetries{0};
@@ -364,6 +372,8 @@ private:
     counter nrQueueWakeups{0};
     counter nrDispatcherWakeups{0};
     counter dispatchTimeMs{0};
+    counter bytesSent{0};
+    counter bytesReceived{0};
     counter nrActiveDbUpdates{0};
 
     /* Specific build to do for --build-one (testing only). */

src/lib/Hydra/View/TT.pm

@@ -6,6 +6,7 @@ use base 'Catalyst::View::TT';
 use Template::Plugin::HTML;
 use Hydra::Helper::Nix;
 use Time::Seconds;
+use Digest::SHA qw(sha1_hex);
 
 __PACKAGE__->config(
     TEMPLATE_EXTENSION => '.tt',
@@ -25,8 +26,14 @@ __PACKAGE__->config(
     makeNameTextForJobset
     relativeDuration
     stripSSHUser
+    metricDivId
     /]);
 
+sub metricDivId {
+    my ($self, $c, $text) = @_;
+    return "metric-" . sha1_hex($text);
+}
+
 sub buildLogExists {
     my ($self, $c, $build) = @_;
     return 1 if defined $c->config->{log_prefix};

src/meson.build

@@ -1,6 +1,5 @@
 # Native code
 subdir('libhydra')
-subdir('hydra-build-step')
 subdir('hydra-evaluator')
 subdir('hydra-queue-runner')
 

src/root/job-metrics-tab.tt

@@ -18,8 +18,7 @@
 
     <h3>Metric: <a [% HTML.attributes(href => c.uri_for('/job' project.name jobset.name job 'metric' metric.name)) %]><tt>[%HTML.escape(metric.name)%]</tt></a></h3>
 
-    [% id = "metric-" _ metric.name;
-       id = id.replace('\.', '_');
+    [% id = metricDivId(metric.name);
        INCLUDE createChart dataUrl=c.uri_for('/job' project.name jobset.name job 'metric' metric.name); %]
 
   [% END %]

src/script/hydra-eval-jobset

@@ -773,6 +773,9 @@ sub checkJobsetWrapped {
     my $jobsetChanged = 0;
     my %buildMap;
 
+    my @jobs;
+    push @jobs, $_ while defined($_ = $jobsIter->());
+
     $db->txn_do(sub {
         my $prevEval = getPrevJobsetEval($db, $jobset, 1);
 
@@ -796,7 +799,7 @@ sub checkJobsetWrapped {
 
         my @jobsWithConstituents;
 
-        while (defined(my $job = $jobsIter->())) {
+        foreach my $job (@jobs) {
             if ($jobsetsJobset) {
                 die "The .jobsets jobset must only have a single job named 'jobsets'"
                     unless $job->{attr} eq "jobsets";

t/evaluator/evaluate-constituents-broken.t

@@ -6,27 +6,55 @@ use Hydra::Helper::Exec;
 
 my $ctx = test_context();
 
-my $jobsetCtx = $ctx->makeJobset(
-    expression => 'constituents-broken.nix',
-);
-my $jobset = $jobsetCtx->{"jobset"};
+subtest "broken constituents expression" => sub {
+    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"
-);
+    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' references non-existent job 'constituentA'/,
+        "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"
-);
+    $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"
+    );
+};
+
+subtest "no matches" => sub {
+    my $jobsetCtx = $ctx->makeJobset(
+        expression => 'constituents-no-matches.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 'non_match_aggregate' references constituent glob pattern 'tests\.\*' with no matches/,
+        "The stderr record includes a relevant error message"
+    );
+
+    $jobset->discard_changes({ '+columns' => {'errormsg' => 'errormsg'} });  # refresh from DB
+    like(
+        $jobset->errormsg,
+        qr/aggregate job ‘non_match_aggregate’ failed with the error: tests\.\*: constituent glob pattern had no matches/,
+        qr/in job ‘non_match_aggregate’:\ntests\.\*: constituent glob pattern had no matches/,
+        "The jobset records a relevant error message"
+    );
+};
 
 done_testing;

t/evaluator/evaluate-constituents-globbing.t

@@ -0,0 +1,138 @@
+use strict;
+use warnings;
+use Setup;
+use Test2::V0;
+use Hydra::Helper::Exec;
+use Data::Dumper;
+
+my $ctx = test_context();
+
+subtest "general glob testing" => sub {
+    my $jobsetCtx = $ctx->makeJobset(
+        expression => 'constituents-glob.nix',
+    );
+    my $jobset = $jobsetCtx->{"jobset"};
+
+    my ($res, $stdout, $stderr) = captureStdoutStderr(60,
+        ("hydra-eval-jobset", $jobsetCtx->{"project"}->name, $jobset->name)
+    );
+    is($res, 0, "hydra-eval-jobset exits zero");
+
+    my $builds = {};
+    for my $build ($jobset->builds) {
+        $builds->{$build->job} = $build;
+    }
+
+    subtest "basic globbing works" => sub {
+        ok(defined $builds->{"ok_aggregate"}, "'ok_aggregate' is part of the jobset evaluation");
+        my @constituents = $builds->{"ok_aggregate"}->constituents->all;
+        is(2, scalar @constituents, "'ok_aggregate' has two constituents");
+
+        my @sortedConstituentNames = sort (map { $_->nixname } @constituents);
+
+        is($sortedConstituentNames[0], "empty-dir-A", "first constituent of 'ok_aggregate' is 'empty-dir-A'");
+        is($sortedConstituentNames[1], "empty-dir-B", "second constituent of 'ok_aggregate' is 'empty-dir-B'");
+    };
+
+    subtest "transitivity is OK" => sub {
+        ok(defined $builds->{"indirect_aggregate"}, "'indirect_aggregate' is part of the jobset evaluation");
+        my @constituents = $builds->{"indirect_aggregate"}->constituents->all;
+        is(1, scalar @constituents, "'indirect_aggregate' has one constituent");
+        is($constituents[0]->nixname, "direct_aggregate", "'indirect_aggregate' has 'direct_aggregate' as single constituent");
+    };
+};
+
+subtest "* selects all except current aggregate" => sub {
+    my $jobsetCtx = $ctx->makeJobset(
+        expression => 'constituents-glob-all.nix',
+    );
+    my $jobset = $jobsetCtx->{"jobset"};
+
+    my ($res, $stdout, $stderr) = captureStdoutStderr(60,
+        ("hydra-eval-jobset", $jobsetCtx->{"project"}->name, $jobset->name)
+    );
+
+    subtest "no eval errors" => sub {
+        ok(utf8::decode($stderr), "Stderr output is UTF8-clean");
+        ok(
+            $stderr !~ "aggregate job ‘ok_aggregate’ has a constituent .* that doesn't correspond to a Hydra build",
+            "Catchall wildcard must not select itself as constituent"
+        );
+
+        $jobset->discard_changes;  # refresh from DB
+        is(
+            $jobset->errormsg,
+            "",
+            "eval-errors non-empty"
+        );
+    };
+
+    my $builds = {};
+    for my $build ($jobset->builds) {
+        $builds->{$build->job} = $build;
+    }
+
+    subtest "two constituents" => sub {
+        ok(defined $builds->{"ok_aggregate"}, "'ok_aggregate' is part of the jobset evaluation");
+        my @constituents = $builds->{"ok_aggregate"}->constituents->all;
+        is(2, scalar @constituents, "'ok_aggregate' has two constituents");
+
+        my @sortedConstituentNames = sort (map { $_->nixname } @constituents);
+
+        is($sortedConstituentNames[0], "empty-dir-A", "first constituent of 'ok_aggregate' is 'empty-dir-A'");
+        is($sortedConstituentNames[1], "empty-dir-B", "second constituent of 'ok_aggregate' is 'empty-dir-B'");
+    };
+};
+
+subtest "trivial cycle check" => sub {
+    my $jobsetCtx = $ctx->makeJobset(
+        expression => 'constituents-cycle.nix',
+    );
+    my $jobset = $jobsetCtx->{"jobset"};
+
+    my ($res, $stdout, $stderr) = captureStdoutStderr(60,
+        ("hydra-eval-jobset", $jobsetCtx->{"project"}->name, $jobset->name)
+    );
+
+    ok(
+        $stderr =~ "Found dependency cycle between jobs 'indirect_aggregate' and 'ok_aggregate'",
+        "Dependency cycle error is on stderr"
+    );
+
+    ok(utf8::decode($stderr), "Stderr output is UTF8-clean");
+
+    $jobset->discard_changes;  # refresh from DB
+    like(
+        $jobset->errormsg,
+        qr/Dependency cycle: indirect_aggregate <-> ok_aggregate/,
+        "eval-errors non-empty"
+    );
+
+    is(0, $jobset->builds->count, "No builds should be scheduled");
+};
+
+subtest "cycle check with globbing" => sub {
+    my $jobsetCtx = $ctx->makeJobset(
+        expression => 'constituents-cycle-glob.nix',
+    );
+    my $jobset = $jobsetCtx->{"jobset"};
+
+    my ($res, $stdout, $stderr) = captureStdoutStderr(60,
+        ("hydra-eval-jobset", $jobsetCtx->{"project"}->name, $jobset->name)
+    );
+
+    ok(utf8::decode($stderr), "Stderr output is UTF8-clean");
+
+    $jobset->discard_changes;  # refresh from DB
+    like(
+        $jobset->errormsg,
+        qr/aggregate job ‘indirect_aggregate’ failed with the error: Dependency cycle: indirect_aggregate <-> packages.constituentA/,
+        "packages.constituentA error missing"
+    );
+
+    # on this branch of Hydra, hydra-eval-jobset fails hard if an aggregate
+    # job is broken.
+    is(0, $jobset->builds->count, "Zero jobs are scheduled");
+};
+
+done_testing;

t/jobs/config.nix

@@ -0,0 +1,14 @@
+rec {
+  path = "/nix/store/l9mg93sgx50y88p5rr6x1vib6j1rjsds-coreutils-9.1/bin";
+
+  mkDerivation = args:
+    derivation ({
+      system = builtins.currentSystem;
+      PATH = path;
+    } // args);
+  mkContentAddressedDerivation = args: mkDerivation ({
+    __contentAddressed = true;
+    outputHashMode = "recursive";
+    outputHashAlgo = "sha256";
+  } // args);
+}

t/jobs/constituents-cycle-glob.nix

@@ -0,0 +1,34 @@
+with import ./config.nix;
+{
+  packages.constituentA = mkDerivation {
+    name = "empty-dir-A";
+    builder = ./empty-dir-builder.sh;
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [ "*_aggregate" ];
+  };
+
+  packages.constituentB = mkDerivation {
+    name = "empty-dir-B";
+    builder = ./empty-dir-builder.sh;
+  };
+
+  ok_aggregate = mkDerivation {
+    name = "direct_aggregate";
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [
+      "packages.*"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+
+  indirect_aggregate = mkDerivation {
+    name = "indirect_aggregate";
+    _hydraAggregate = true;
+    constituents = [
+      "ok_aggregate"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+}

t/jobs/constituents-cycle.nix

@@ -0,0 +1,21 @@
+with import ./config.nix;
+{
+  ok_aggregate = mkDerivation {
+    name = "direct_aggregate";
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [
+      "indirect_aggregate"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+
+  indirect_aggregate = mkDerivation {
+    name = "indirect_aggregate";
+    _hydraAggregate = true;
+    constituents = [
+      "ok_aggregate"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+}

t/jobs/constituents-glob-all.nix

@@ -0,0 +1,22 @@
+with import ./config.nix;
+{
+  packages.constituentA = mkDerivation {
+    name = "empty-dir-A";
+    builder = ./empty-dir-builder.sh;
+  };
+
+  packages.constituentB = mkDerivation {
+    name = "empty-dir-B";
+    builder = ./empty-dir-builder.sh;
+  };
+
+  ok_aggregate = mkDerivation {
+    name = "direct_aggregate";
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [
+      "*"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+}

t/jobs/constituents-glob.nix

@@ -0,0 +1,31 @@
+with import ./config.nix;
+{
+  packages.constituentA = mkDerivation {
+    name = "empty-dir-A";
+    builder = ./empty-dir-builder.sh;
+  };
+
+  packages.constituentB = mkDerivation {
+    name = "empty-dir-B";
+    builder = ./empty-dir-builder.sh;
+  };
+
+  ok_aggregate = mkDerivation {
+    name = "direct_aggregate";
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [
+      "packages.*"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+
+  indirect_aggregate = mkDerivation {
+    name = "indirect_aggregate";
+    _hydraAggregate = true;
+    constituents = [
+      "ok_aggregate"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+}

t/jobs/constituents-no-matches.nix

@@ -0,0 +1,20 @@
+with import ./config.nix;
+{
+  non_match_aggregate = mkDerivation {
+    name = "mixed_aggregate";
+    _hydraAggregate = true;
+    _hydraGlobConstituents = true;
+    constituents = [
+      "tests.*"
+    ];
+    builder = ./empty-dir-builder.sh;
+  };
+
+  # Without a second job no jobset is attempted to be created
+  # (the only job would be broken)
+  # and thus the constituent validation is never reached.
+  dummy = mkDerivation {
+    name = "dummy";
+    builder = ./empty-dir-builder.sh;
+  };
+}

t/jobs/declarative/project.json

@@ -0,0 +1,24 @@
+{
+    "enabled": 1,
+    "hidden": false,
+    "description": "declarative-jobset-example",
+    "nixexprinput": "src",
+    "nixexprpath": "declarative/generator.nix",
+    "checkinterval": 300,
+    "schedulingshares": 100,
+    "enableemail": false,
+    "emailoverride": "",
+    "keepnr": 3,
+    "inputs": {
+        "src": {
+            "type": "path",
+            "value": "/home/ma27/Projects/hydra-cppnix/t/jobs",
+            "emailresponsible": false
+        },
+        "jobspath": {
+            "type": "string",
+            "value": "/home/ma27/Projects/hydra-cppnix/t/jobs",
+            "emailresponsible": false
+        }
+    }
+}

t/meson.build

@@ -28,7 +28,6 @@ testenv.prepend('PERL5LIB',
 )
 testenv.prepend('PATH',
   fs.parent(find_program('nix').full_path()),
-  fs.parent(hydra_build_step.full_path()),
   fs.parent(hydra_evaluator.full_path()),
   fs.parent(hydra_queue_runner.full_path()),
   meson.project_source_root() / 'src/script',

t/queue-runner/constituents.t

@@ -22,11 +22,11 @@ is(nrQueuedBuildsForJobset($jobset), 0, "Evaluating jobs/broken-constituent.nix
 
 like(
     $jobset->errormsg,
-    qr/^"does-not-exist": does not exist$/m,
+    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,
+    qr/^does-not-evaluate: error: assertion 'false' failed/m,
     "Evaluating jobs/broken-constituent.nix should log an error for does-not-evaluate");
 
 done_testing;