Merge pull request #1349 from NixOS/ca-no-new-col

Allow building content-addressed derivations with hydra, minimally

.gitignore

@@ -38,6 +38,7 @@ t/jobs/declarative/project.json
 hydra-config.h
 hydra-config.h.in
 result
+result-*
 outputs
 config
 stamp-h1

Makefile.am

@@ -1,8 +1,12 @@
-SUBDIRS = src t doc
+SUBDIRS = src doc
+if CAN_DO_CHECK
+  SUBDIRS += t
+endif
+
 BOOTCLEAN_SUBDIRS = $(SUBDIRS)
 DIST_SUBDIRS      = $(SUBDIRS)
-EXTRA_DIST        = hydra-module.nix
+EXTRA_DIST        = nixos-modules/hydra.nix
 
-install-data-local: hydra-module.nix
+install-data-local: nixos-modules/hydra.nix
 	$(INSTALL) -d $(DESTDIR)$(datadir)/nix
-	$(INSTALL_DATA) hydra-module.nix $(DESTDIR)$(datadir)/nix/
+	$(INSTALL_DATA) nixos-modules/hydra.nix $(DESTDIR)$(datadir)/nix/hydra-module.nix

README.md

@@ -80,7 +80,7 @@ $ nix-build
 You can use the provided shell.nix to get a working development environment:
 ```
 $ nix-shell
-$ ./bootstrap
+$ autoreconfPhase
 $ configurePhase # NOTE: not ./configure
 $ make
 ```

bootstrap

@@ -1,2 +0,0 @@
-#! /bin/sh -e
-exec autoreconf -vfi

configure.ac

@@ -53,9 +53,6 @@ PKG_CHECK_MODULES([NIX], [nix-main nix-expr nix-store])
 testPath="$(dirname $(type -p expr))"
 AC_SUBST(testPath)
 
-jobsPath="$(realpath ./t/jobs)"
-AC_SUBST(jobsPath)
-
 CXXFLAGS+=" -include nix/config.h"
 
 AC_CONFIG_FILES([
@@ -71,11 +68,22 @@ AC_CONFIG_FILES([
   src/lib/Makefile
   src/root/Makefile
   src/script/Makefile
-  t/Makefile
-  t/jobs/config.nix
-  t/jobs/declarative/project.json
 ])
 
+# Tests might be filtered out
+AM_CONDITIONAL([CAN_DO_CHECK], [test -f "$srcdir/t/api-test.t"])
+AM_COND_IF(
+  [CAN_DO_CHECK],
+  [
+    jobsPath="$(realpath ./t/jobs)"
+    AC_SUBST(jobsPath)
+    AC_CONFIG_FILES([
+      t/Makefile
+      t/jobs/config.nix
+      t/jobs/declarative/project.json
+    ])
+  ])
+
 AC_CONFIG_COMMANDS([executable-scripts], [])
 
 AC_CONFIG_HEADER([hydra-config.h])

doc/manual/src/configuration.md

@@ -74,6 +74,30 @@ following:
         }
     }
 
+Populating a Cache
+------------------
+
+A common use for Hydra is to pre-build and cache derivations which
+take a long time to build. While it is possible to direcly access the
+Hydra server's store over SSH, a more scalable option is to upload
+built derivations to a remote store like an [S3-compatible object
+store](https://nixos.org/manual/nix/stable/command-ref/new-cli/nix3-help-stores.html#s3-binary-cache-store). Setting
+the `store_uri` parameter will cause Hydra to sign and upload
+derivations as they are built:
+
+```
+store_uri = s3://cache-bucket-name?compression=zstd&parallel-compression=true&write-nar-listing=1&ls-compression=br&log-compression=br&secret-key=/path/to/cache/private/key
+```
+
+This example uses [Zstandard](https://github.com/facebook/zstd)
+compression on derivations to reduce CPU usage on the server, but
+[Brotli](https://brotli.org/) compression for derivation listings and
+build logs because it has better browser support.
+
+See [`nix help
+stores`](https://nixos.org/manual/nix/stable/command-ref/new-cli/nix3-help-stores.html)
+for a description of the store URI format.
+
 Statsd Configuration
 --------------------
 

doc/manual/src/hacking.md

@@ -18,7 +18,7 @@ $ nix-shell
 To build Hydra, you should then do:
 
 ```console
-[nix-shell]$ ./bootstrap
+[nix-shell]$ autoreconfPhase
 [nix-shell]$ configurePhase
 [nix-shell]$ make
 ```

doc/manual/src/projects.md

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

flake.lock

@@ -42,27 +42,27 @@
         "nixpkgs-regression": "nixpkgs-regression"
       },
       "locked": {
-        "lastModified": 1690219894,
-        "narHash": "sha256-QMYAkdtU+g9HlZKtoJ+AI6TbWzzovKGnPZJHfZdclc8=",
+        "lastModified": 1706208340,
+        "narHash": "sha256-wNyHUEIiKKVs6UXrUzhP7RSJQv0A8jckgcuylzftl8k=",
         "owner": "NixOS",
         "repo": "nix",
-        "rev": "a212300a1d9f9c7b0daf19c00c87fc50480f54f4",
+        "rev": "2c4bb93ba5a97e7078896ebc36385ce172960e4e",
         "type": "github"
       },
       "original": {
         "owner": "NixOS",
-        "ref": "2.17.0",
+        "ref": "2.19-maintenance",
         "repo": "nix",
         "type": "github"
       }
     },
     "nixpkgs": {
       "locked": {
-        "lastModified": 1687379288,
-        "narHash": "sha256-cSuwfiqYfeVyqzCRkU9AvLTysmEuSal8nh6CYr+xWog=",
+        "lastModified": 1701615100,
+        "narHash": "sha256-7VI84NGBvlCTduw2aHLVB62NvCiZUlALLqBe5v684Aw=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "ef0bc3976340dab9a4e087a0bcff661a8b2e87f3",
+        "rev": "e9f06adb793d1cca5384907b3b8a4071d5d7cb19",
         "type": "github"
       },
       "original": {
@@ -72,6 +72,22 @@
         "type": "github"
       }
     },
+    "nixpkgs-for-fileset": {
+      "locked": {
+        "lastModified": 1706098335,
+        "narHash": "sha256-r3dWjT8P9/Ah5m5ul4WqIWD8muj5F+/gbCdjiNVBKmU=",
+        "owner": "NixOS",
+        "repo": "nixpkgs",
+        "rev": "a77ab169a83a4175169d78684ddd2e54486ac651",
+        "type": "github"
+      },
+      "original": {
+        "owner": "NixOS",
+        "ref": "nixos-23.11",
+        "repo": "nixpkgs",
+        "type": "github"
+      }
+    },
     "nixpkgs-regression": {
       "locked": {
         "lastModified": 1643052045,
@@ -91,7 +107,8 @@
     "root": {
       "inputs": {
         "nix": "nix",
-        "nixpkgs": "nixpkgs"
+        "nixpkgs": "nixpkgs",
+        "nixpkgs-for-fileset": "nixpkgs-for-fileset"
       }
     }
   },

flake.nix

@@ -2,19 +2,24 @@
   description = "A Nix-based continuous build system";
 
   inputs.nixpkgs.url = "github:NixOS/nixpkgs/nixos-23.05";
-  inputs.nix.url = "github:NixOS/nix/2.17.0";
+  inputs.nix.url = "github:NixOS/nix/2.19-maintenance";
   inputs.nix.inputs.nixpkgs.follows = "nixpkgs";
 
-  outputs = { self, nixpkgs, nix }:
-    let
-      version = "${builtins.readFile ./version.txt}.${builtins.substring 0 8 (self.lastModifiedDate or "19700101")}.${self.shortRev or "DIRTY"}";
+  # TODO get rid of this once https://github.com/NixOS/nix/pull/9546 is
+  # mered and we upgrade or Nix, so the main `nixpkgs` input is at least
+  # 23.11 and has `lib.fileset`.
+  inputs.nixpkgs-for-fileset.url = "github:NixOS/nixpkgs/nixos-23.11";
 
+  outputs = { self, nixpkgs, nix, nixpkgs-for-fileset }:
+    let
       systems = [ "x86_64-linux" "aarch64-linux" ];
       forEachSystem = nixpkgs.lib.genAttrs systems;
 
+      overlayList = [ self.overlays.default nix.overlays.default ];
+
       pkgsBySystem = forEachSystem (system: import nixpkgs {
         inherit system;
-        overlays = [ self.overlays.default nix.overlays.default ];
+        overlays = overlayList;
       });
 
       # NixOS configuration used for VM tests.
@@ -61,197 +66,9 @@
 
         };
 
-        hydra = with final; let
-          perlDeps = buildEnv {
-            name = "hydra-perl-deps";
-            paths = with perlPackages; lib.closePropagation
-              [
-                AuthenSASL
-                CatalystActionREST
-                CatalystAuthenticationStoreDBIxClass
-                CatalystAuthenticationStoreLDAP
-                CatalystDevel
-                CatalystPluginAccessLog
-                CatalystPluginAuthorizationRoles
-                CatalystPluginCaptcha
-                CatalystPluginPrometheusTiny
-                CatalystPluginSessionStateCookie
-                CatalystPluginSessionStoreFastMmap
-                CatalystPluginStackTrace
-                CatalystTraitForRequestProxyBase
-                CatalystViewDownload
-                CatalystViewJSON
-                CatalystViewTT
-                CatalystXRoleApplicator
-                CatalystXScriptServerStarman
-                CryptPassphrase
-                CryptPassphraseArgon2
-                CryptRandPasswd
-                DataDump
-                DateTime
-                DBDPg
-                DBDSQLite
-                DigestSHA1
-                EmailMIME
-                EmailSender
-                FileLibMagic
-                FileSlurper
-                FileWhich
-                final.nix.perl-bindings
-                git
-                IOCompress
-                IPCRun
-                IPCRun3
-                JSON
-                JSONMaybeXS
-                JSONXS
-                ListSomeUtils
-                LWP
-                LWPProtocolHttps
-                ModulePluggable
-                NetAmazonS3
-                NetPrometheus
-                NetStatsd
-                PadWalker
-                ParallelForkManager
-                PerlCriticCommunity
-                PrometheusTinyShared
-                ReadonlyX
-                SetScalar
-                SQLSplitStatement
-                Starman
-                StringCompareConstantTime
-                SysHostnameLong
-                TermSizeAny
-                TermReadKey
-                Test2Harness
-                TestPostgreSQL
-                TextDiff
-                TextTable
-                UUID4Tiny
-                YAML
-                XMLSimple
-              ];
-          };
-
-        in
-        stdenv.mkDerivation {
-
-          name = "hydra-${version}";
-
-          src = self;
-
-          buildInputs =
-            [
-              makeWrapper
-              autoconf
-              automake
-              libtool
-              unzip
-              nukeReferences
-              pkg-config
-              libpqxx
-              top-git
-              mercurial
-              darcs
-              subversion
-              breezy
-              openssl
-              bzip2
-              libxslt
-              final.nix
-              perlDeps
-              perl
-              mdbook
-              pixz
-              boost
-              postgresql_13
-              (if lib.versionAtLeast lib.version "20.03pre"
-              then nlohmann_json
-              else nlohmann_json.override { multipleHeaders = true; })
-              prometheus-cpp
-            ];
-
-          checkInputs = [
-            cacert
-            foreman
-            glibcLocales
-            libressl.nc
-            openldap
-            python3
-          ];
-
-          hydraPath = lib.makeBinPath (
-            [
-              subversion
-              openssh
-              final.nix
-              coreutils
-              findutils
-              pixz
-              gzip
-              bzip2
-              xz
-              gnutar
-              unzip
-              git
-              top-git
-              mercurial
-              darcs
-              gnused
-              breezy
-            ] ++ lib.optionals stdenv.isLinux [ rpm dpkg cdrkit ]
-          );
-
-          OPENLDAP_ROOT = openldap;
-
-          shellHook = ''
-            pushd $(git rev-parse --show-toplevel) >/dev/null
-
-            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
-            export HYDRA_DATA="$(pwd)/.hydra-data"
-            export HYDRA_DBI='dbi:Pg:dbname=hydra;host=localhost;port=64444'
-
-            popd >/dev/null
-          '';
-
-          preConfigure = "autoreconf -vfi";
-
-          NIX_LDFLAGS = [ "-lpthread" ];
-
-          enableParallelBuilding = true;
-
-          doCheck = true;
-
-          preCheck = ''
-            patchShebangs .
-            export LOGNAME=''${LOGNAME:-foo}
-            # set $HOME for bzr so it can create its trace file
-            export HOME=$(mktemp -d)
-          '';
-
-          postInstall = ''
-            mkdir -p $out/nix-support
-
-            for i in $out/bin/*; do
-                read -n 4 chars < $i
-                if [[ $chars =~ ELF ]]; then continue; fi
-                wrapProgram $i \
-                    --prefix PERL5LIB ':' $out/libexec/hydra/lib:$PERL5LIB \
-                    --prefix PATH ':' $out/bin:$hydraPath \
-                    --set HYDRA_RELEASE ${version} \
-                    --set HYDRA_HOME $out/libexec/hydra \
-                    --set NIX_RELEASE ${final.nix.name or "unknown"}
-            done
-          '';
-
-          dontStrip = true;
-
-          meta.description = "Build of Hydra on ${final.stdenv.system}";
-          passthru = { inherit perlDeps; inherit (final) nix; };
+        hydra = final.callPackage ./package.nix {
+          inherit (nixpkgs-for-fileset.lib) fileset;
+          rawSrc = self;
         };
       };
 
@@ -259,9 +76,15 @@
 
         build = forEachSystem (system: packages.${system}.hydra);
 
+        buildNoTests = forEachSystem (system:
+          packages.${system}.hydra.overrideAttrs (_: {
+            doCheck = false;
+          })
+        );
+
         manual = forEachSystem (system:
           let pkgs = pkgsBySystem.${system}; in
-          pkgs.runCommand "hydra-manual-${version}" { }
+          pkgs.runCommand "hydra-manual-${pkgs.hydra.version}" { }
             ''
               mkdir -p $out/share
               cp -prvd ${pkgs.hydra}/share/doc $out/share/
@@ -565,50 +388,8 @@
         default = pkgsBySystem.${system}.hydra;
       });
 
-      nixosModules.hydra = {
-        imports = [ ./hydra-module.nix ];
-        nixpkgs.overlays = [ self.overlays.default nix.overlays.default ];
-      };
-
-      nixosModules.hydraTest = { pkgs, ... }: {
-        imports = [ self.nixosModules.hydra ];
-
-        services.hydra-dev.enable = true;
-        services.hydra-dev.hydraURL = "http://hydra.example.org";
-        services.hydra-dev.notificationSender = "admin@hydra.example.org";
-
-        systemd.services.hydra-send-stats.enable = false;
-
-        services.postgresql.enable = true;
-        services.postgresql.package = pkgs.postgresql_11;
-
-        # The following is to work around the following error from hydra-server:
-        #   [error] Caught exception in engine "Cannot determine local time zone"
-        time.timeZone = "UTC";
-
-        nix.extraOptions = ''
-          allowed-uris = https://github.com/
-        '';
-      };
-
-      nixosModules.hydraProxy = {
-        services.httpd = {
-          enable = true;
-          adminAddr = "hydra-admin@example.org";
-          extraConfig = ''
-            <Proxy *>
-              Order deny,allow
-              Allow from all
-            </Proxy>
-
-            ProxyRequests     Off
-            ProxyPreserveHost On
-            ProxyPass         /apache-errors !
-            ErrorDocument 503 /apache-errors/503.html
-            ProxyPass         /       http://127.0.0.1:3000/ retry=5 disablereuse=on
-            ProxyPassReverse  /       http://127.0.0.1:3000/
-          '';
-        };
+      nixosModules = import ./nixos-modules {
+        overlays = overlayList;
       };
 
       nixosConfigurations.container = nixpkgs.lib.nixosSystem {

nixos-modules/default.nix

@@ -0,0 +1,49 @@
+{ overlays }:
+
+rec {
+  hydra = {
+    imports = [ ./hydra.nix ];
+    nixpkgs = { inherit overlays; };
+  };
+
+  hydraTest = { pkgs, ... }: {
+    imports = [ hydra ];
+
+    services.hydra-dev.enable = true;
+    services.hydra-dev.hydraURL = "http://hydra.example.org";
+    services.hydra-dev.notificationSender = "admin@hydra.example.org";
+
+    systemd.services.hydra-send-stats.enable = false;
+
+    services.postgresql.enable = true;
+    services.postgresql.package = pkgs.postgresql_11;
+
+    # The following is to work around the following error from hydra-server:
+    #   [error] Caught exception in engine "Cannot determine local time zone"
+    time.timeZone = "UTC";
+
+    nix.extraOptions = ''
+      allowed-uris = https://github.com/
+    '';
+  };
+
+  hydraProxy = {
+    services.httpd = {
+      enable = true;
+      adminAddr = "hydra-admin@example.org";
+      extraConfig = ''
+        <Proxy *>
+          Order deny,allow
+          Allow from all
+        </Proxy>
+
+        ProxyRequests     Off
+        ProxyPreserveHost On
+        ProxyPass         /apache-errors !
+        ErrorDocument 503 /apache-errors/503.html
+        ProxyPass         /       http://127.0.0.1:3000/ retry=5 disablereuse=on
+        ProxyPassReverse  /       http://127.0.0.1:3000/
+      '';
+    };
+  };
+}

package.nix

@@ -0,0 +1,272 @@
+{ stdenv
+, lib
+, fileset
+
+, rawSrc
+
+, buildEnv
+
+, perlPackages
+
+, nix
+, git
+
+, makeWrapper
+, autoreconfHook
+, nukeReferences
+, pkg-config
+, mdbook
+
+, unzip
+, libpqxx
+, top-git
+, mercurial
+, darcs
+, subversion
+, breezy
+, openssl
+, bzip2
+, libxslt
+, perl
+, pixz
+, boost
+, postgresql_13
+, nlohmann_json
+, prometheus-cpp
+
+, cacert
+, foreman
+, glibcLocales
+, libressl
+, openldap
+, python3
+
+, openssh
+, coreutils
+, findutils
+, gzip
+, xz
+, gnutar
+, gnused
+
+, rpm
+, dpkg
+, cdrkit
+}:
+
+let
+  perlDeps = buildEnv {
+    name = "hydra-perl-deps";
+    paths = lib.closePropagation
+      ([
+        nix.perl-bindings
+        git
+      ] ++ (with perlPackages; [
+        AuthenSASL
+        CatalystActionREST
+        CatalystAuthenticationStoreDBIxClass
+        CatalystAuthenticationStoreLDAP
+        CatalystDevel
+        CatalystPluginAccessLog
+        CatalystPluginAuthorizationRoles
+        CatalystPluginCaptcha
+        CatalystPluginPrometheusTiny
+        CatalystPluginSessionStateCookie
+        CatalystPluginSessionStoreFastMmap
+        CatalystPluginStackTrace
+        CatalystTraitForRequestProxyBase
+        CatalystViewDownload
+        CatalystViewJSON
+        CatalystViewTT
+        CatalystXRoleApplicator
+        CatalystXScriptServerStarman
+        CryptPassphrase
+        CryptPassphraseArgon2
+        CryptRandPasswd
+        DataDump
+        DateTime
+        DBDPg
+        DBDSQLite
+        DigestSHA1
+        EmailMIME
+        EmailSender
+        FileLibMagic
+        FileSlurper
+        FileWhich
+        IOCompress
+        IPCRun
+        IPCRun3
+        JSON
+        JSONMaybeXS
+        JSONXS
+        ListSomeUtils
+        LWP
+        LWPProtocolHttps
+        ModulePluggable
+        NetAmazonS3
+        NetPrometheus
+        NetStatsd
+        PadWalker
+        ParallelForkManager
+        PerlCriticCommunity
+        PrometheusTinyShared
+        ReadonlyX
+        SetScalar
+        SQLSplitStatement
+        Starman
+        StringCompareConstantTime
+        SysHostnameLong
+        TermSizeAny
+        TermReadKey
+        Test2Harness
+        TestPostgreSQL
+        TextDiff
+        TextTable
+        UUID4Tiny
+        YAML
+        XMLSimple
+      ]));
+  };
+
+  version = "${builtins.readFile ./version.txt}.${builtins.substring 0 8 (rawSrc.lastModifiedDate or "19700101")}.${rawSrc.shortRev or "DIRTY"}";
+in
+stdenv.mkDerivation (finalAttrs: {
+  pname = "hydra";
+  inherit version;
+
+  src = fileset.toSource {
+    root = ./.;
+    fileset = fileset.unions ([
+      ./version.txt
+      ./configure.ac
+      ./Makefile.am
+      ./src
+      ./doc
+      ./nixos-modules/hydra.nix
+      # These are always needed to appease Automake
+      ./t/Makefile.am
+      ./t/jobs/config.nix.in
+      ./t/jobs/declarative/project.json.in
+    ] ++ lib.optionals finalAttrs.doCheck [
+      ./t
+      ./.perlcriticrc
+      ./.yath.rc
+    ]);
+  };
+
+  strictDeps = true;
+
+  nativeBuildInputs = [
+    makeWrapper
+    autoreconfHook
+    nukeReferences
+    pkg-config
+    mdbook
+    nix
+    perlDeps
+    perl
+    unzip
+  ];
+
+  buildInputs = [
+    libpqxx
+    openssl
+    libxslt
+    nix
+    perlDeps
+    perl
+    boost
+    nlohmann_json
+    prometheus-cpp
+  ];
+
+  nativeCheckInputs = [
+    bzip2
+    darcs
+    foreman
+    top-git
+    mercurial
+    subversion
+    breezy
+    openldap
+    postgresql_13
+    pixz
+  ];
+
+  checkInputs = [
+    cacert
+    glibcLocales
+    libressl.nc
+    python3
+  ];
+
+  hydraPath = lib.makeBinPath (
+    [
+      subversion
+      openssh
+      nix
+      coreutils
+      findutils
+      pixz
+      gzip
+      bzip2
+      xz
+      gnutar
+      unzip
+      git
+      top-git
+      mercurial
+      darcs
+      gnused
+      breezy
+    ] ++ lib.optionals stdenv.isLinux [ rpm dpkg cdrkit ]
+  );
+
+  OPENLDAP_ROOT = openldap;
+
+  shellHook = ''
+    pushd $(git rev-parse --show-toplevel) >/dev/null
+
+    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
+    export HYDRA_DATA="$(pwd)/.hydra-data"
+    export HYDRA_DBI='dbi:Pg:dbname=hydra;host=localhost;port=64444'
+
+    popd >/dev/null
+  '';
+
+  NIX_LDFLAGS = [ "-lpthread" ];
+
+  enableParallelBuilding = true;
+
+  doCheck = true;
+
+  preCheck = ''
+    patchShebangs .
+    export LOGNAME=''${LOGNAME:-foo}
+    # set $HOME for bzr so it can create its trace file
+    export HOME=$(mktemp -d)
+  '';
+
+  postInstall = ''
+    mkdir -p $out/nix-support
+
+    for i in $out/bin/*; do
+        read -n 4 chars < $i
+        if [[ $chars =~ ELF ]]; then continue; fi
+        wrapProgram $i \
+            --prefix PERL5LIB ':' $out/libexec/hydra/lib:$PERL5LIB \
+            --prefix PATH ':' $out/bin:$hydraPath \
+            --set HYDRA_RELEASE ${version} \
+            --set HYDRA_HOME $out/libexec/hydra \
+            --set NIX_RELEASE ${nix.name or "unknown"}
+    done
+  '';
+
+  dontStrip = true;
+
+  meta.description = "Build of Hydra on ${stdenv.system}";
+  passthru = { inherit perlDeps nix; };
+})

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

@@ -7,6 +7,9 @@
 #include "store-api.hh"
 #include "eval.hh"
 #include "eval-inline.hh"
+#include "eval-settings.hh"
+#include "signals.hh"
+#include "terminal.hh"
 #include "util.hh"
 #include "get-drvs.hh"
 #include "globals.hh"
@@ -53,7 +56,7 @@ using namespace nix;
 static Path gcRootsDir;
 static size_t maxMemorySize;
 
-struct MyArgs : MixEvalArgs, MixCommonArgs
+struct MyArgs : MixEvalArgs, MixCommonArgs, RootArgs
 {
     Path releaseExpr;
     bool flake = false;
@@ -94,7 +97,7 @@ static std::string queryMetaStrings(EvalState & state, DrvInfo & drv, const std:
     rec = [&](Value & v) {
         state.forceValue(v, noPos);
         if (v.type() == nString)
-            res.push_back(v.string.s);
+            res.emplace_back(v.string_view());
         else if (v.isList())
             for (unsigned int n = 0; n < v.listSize(); ++n)
                 rec(*v.listElems()[n]);
@@ -175,7 +178,11 @@ static void worker(
 
             if (auto drv = getDerivation(state, *v, false)) {
 
-                DrvInfo::Outputs outputs = drv->queryOutputs();
+                // 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")
                     throw EvalError("derivation must have a 'system' attribute");
@@ -208,20 +215,20 @@ static void worker(
                     for (auto & c : context)
                         std::visit(overloaded {
                             [&](const NixStringContextElem::Built & b) {
-                                job["constituents"].push_back(state.store->printStorePath(b.drvPath));
+                                job["constituents"].push_back(b.drvPath->to_string(*state.store));
                             },
                             [&](const NixStringContextElem::Opaque & o) {
                             },
                             [&](const NixStringContextElem::DrvDeep & d) {
                             },
-                        }, c.raw());
+                        }, 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());
+                            job["namedConstituents"].push_back(v->string_view());
                     }
                 }
 
@@ -236,12 +243,17 @@ static void worker(
                 }
 
                 nlohmann::json out;
-                for (auto & j : outputs)
-                    // FIXME: handle CA/impure builds.
-                    if (j.second)
-                        out[j.first] = state.store->printStorePath(*j.second);
+                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);
             }
 
@@ -516,7 +528,7 @@ int main(int argc, char * * argv)
                     auto drvPath2 = store->parseStorePath((std::string) (*job2)["drvPath"]);
                     auto drv2 = store->readDerivation(drvPath2);
                     job["constituents"].push_back(store->printStorePath(drvPath2));
-                    drv.inputDrvs[drvPath2] = {drv2.outputs.begin()->first};
+                    drv.inputDrvs.map[drvPath2].value = {drv2.outputs.begin()->first};
                 }
 
                 if (brokenJobs.empty()) {

src/hydra-evaluator/hydra-evaluator.cc

@@ -2,6 +2,7 @@
 #include "hydra-config.hh"
 #include "pool.hh"
 #include "shared.hh"
+#include "signals.hh"
 
 #include <algorithm>
 #include <thread>

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

@@ -9,27 +9,25 @@
 #include "path.hh"
 #include "serve-protocol.hh"
 #include "state.hh"
+#include "current-process.hh"
+#include "processes.hh"
 #include "util.hh"
-#include "worker-protocol.hh"
-#include "worker-protocol-impl.hh"
+#include "serve-protocol.hh"
+#include "serve-protocol-impl.hh"
+#include "ssh.hh"
 #include "finally.hh"
 #include "url.hh"
 
 using namespace nix;
 
 
-struct Child
-{
-    Pid pid;
-    AutoCloseFD to, from;
-};
-
-
 static void append(Strings & dst, const Strings & src)
 {
     dst.insert(dst.end(), src.begin(), src.end());
 }
 
+namespace nix::build_remote {
+
 static Strings extraStoreArgs(std::string & machine)
 {
     Strings result;
@@ -50,7 +48,7 @@ static Strings extraStoreArgs(std::string & machine)
     return result;
 }
 
-static void openConnection(Machine::ptr machine, Path tmpDir, int stderrFD, Child & child)
+static void openConnection(::Machine::ptr machine, Path tmpDir, int stderrFD, SSHMaster::Connection & child)
 {
     std::string pgmName;
     Pipe to, from;
@@ -80,7 +78,7 @@ static void openConnection(Machine::ptr machine, Path tmpDir, int stderrFD, Chil
         append(argv, extraArgs);
     }
 
-    child.pid = startProcess([&]() {
+    child.sshPid = startProcess([&]() {
         restoreProcessContext();
 
         if (dup2(to.readSide.get(), STDIN_FILENO) == -1)
@@ -100,32 +98,32 @@ static void openConnection(Machine::ptr machine, Path tmpDir, int stderrFD, Chil
     to.readSide = -1;
     from.writeSide = -1;
 
-    child.to = to.writeSide.release();
-    child.from = from.readSide.release();
+    child.in = to.writeSide.release();
+    child.out = from.readSide.release();
 }
 
 
-static void copyClosureTo(std::timed_mutex & sendMutex, Store & destStore,
-    FdSource & from, FdSink & to, const StorePathSet & paths,
-    bool useSubstitutes = false)
+static void copyClosureTo(
+    ::Machine::Connection & conn,
+    Store & destStore,
+    const StorePathSet & paths,
+    SubstituteFlag useSubstitutes = NoSubstitute)
 {
     StorePathSet closure;
     destStore.computeFSClosure(paths, closure);
 
-    WorkerProto::WriteConn wconn { .to = to };
-    WorkerProto::ReadConn rconn { .from = from };
     /* 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
-    to << ServeProto::Command::QueryValidPaths << 1 << useSubstitutes;
-    WorkerProto::write(destStore, wconn, closure);
-    to.flush();
+    conn.to << ServeProto::Command::QueryValidPaths << 1 << useSubstitutes;
+    ServeProto::write(destStore, conn, closure);
+    conn.to.flush();
 
     /* Get back the set of paths that are already valid on the remote
        host. */
-    auto present = WorkerProto::Serialise<StorePathSet>::read(destStore, rconn);
+    auto present = ServeProto::Serialise<StorePathSet>::read(destStore, conn);
 
     if (present.size() == closure.size()) return;
 
@@ -137,20 +135,20 @@ static void copyClosureTo(std::timed_mutex & sendMutex, Store & destStore,
 
     printMsg(lvlDebug, "sending %d missing paths", missing.size());
 
-    std::unique_lock<std::timed_mutex> sendLock(sendMutex,
+    std::unique_lock<std::timed_mutex> sendLock(conn.machine->state->sendLock,
         std::chrono::seconds(600));
 
-    to << ServeProto::Command::ImportPaths;
-    destStore.exportPaths(missing, to);
-    to.flush();
+    conn.to << ServeProto::Command::ImportPaths;
+    destStore.exportPaths(missing, conn.to);
+    conn.to.flush();
 
-    if (readInt(from) != 1)
+    if (readInt(conn.from) != 1)
         throw Error("remote machine failed to import closure");
 }
 
 
 // FIXME: use Store::topoSortPaths().
-StorePaths reverseTopoSortPaths(const std::map<StorePath, ValidPathInfo> & paths)
+static StorePaths reverseTopoSortPaths(const std::map<StorePath, ValidPathInfo> & paths)
 {
     StorePaths sorted;
     StorePathSet visited;
@@ -178,24 +176,332 @@ StorePaths reverseTopoSortPaths(const std::map<StorePath, ValidPathInfo> & paths
     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));
+
+    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)};
+}
+
+/**
+ * @param conn is not fully initialized; it is this functions job to set
+ * the `remoteVersion` field after the handshake is completed.
+ * Therefore, no `ServeProto::Serialize` functions can be used until
+ * that field is set.
+ */
+static void handshake(::Machine::Connection & conn, unsigned int repeats)
+{
+    conn.to << SERVE_MAGIC_1 << 0x206;
+    conn.to.flush();
+
+    unsigned int magic = readInt(conn.from);
+    if (magic != SERVE_MAGIC_2)
+        throw Error("protocol mismatch with ‘nix-store --serve’ on ‘%1%’", conn.machine->sshName);
+    conn.remoteVersion = readInt(conn.from);
+    // Now `conn` is initialized.
+    if (GET_PROTOCOL_MAJOR(conn.remoteVersion) != 0x200)
+        throw Error("unsupported ‘nix-store --serve’ protocol version on ‘%1%’", conn.machine->sshName);
+    if (GET_PROTOCOL_MINOR(conn.remoteVersion) < 3 && repeats > 0)
+        throw Error("machine ‘%1%’ does not support repeating a build; please upgrade it to Nix 1.12", conn.machine->sshName);
+}
+
+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->sshName);
+
+        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 State::BuildOptions & options,
+    counter & nrStepsBuilding
+)
+{
+    conn.to << ServeProto::Command::BuildDerivation << localStore.printStorePath(drvPath);
+    writeDerivation(conn.to, localStore, drv);
+    conn.to << options.maxSilentTime << options.buildTimeout;
+    if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 2)
+        conn.to << options.maxLogSize;
+    if (GET_PROTOCOL_MINOR(conn.remoteVersion) >= 3) {
+        conn.to
+            << options.repeats // == build-repeat
+            << options.enforceDeterminism;
+    }
+    conn.to.flush();
+
+    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 std::map<StorePath, ValidPathInfo> queryPathInfos(
+    ::Machine::Connection & conn,
+    Store & localStore,
+    StorePathSet & outputs,
+    size_t & totalNarSize
+)
+{
+
+    /* Get info about each output path. */
+    std::map<StorePath, ValidPathInfo> infos;
+    conn.to << ServeProto::Command::QueryPathInfos;
+    ServeProto::write(localStore, conn, outputs);
+    conn.to.flush();
+    while (true) {
+        auto storePathS = readString(conn.from);
+        if (storePathS == "") break;
+        auto deriver = readString(conn.from); // deriver
+        auto references = ServeProto::Serialise<StorePathSet>::read(localStore, conn);
+        readLongLong(conn.from); // download size
+        auto narSize = readLongLong(conn.from);
+        auto narHash = Hash::parseAny(readString(conn.from), htSHA256);
+        auto ca = ContentAddress::parseOpt(readString(conn.from));
+        readStrings<StringSet>(conn.from); // sigs
+        ValidPathInfo info(localStore.parseStorePath(storePathS), narHash);
+        assert(outputs.count(info.path));
+        info.references = references;
+        info.narSize = narSize;
+        totalNarSize += info.narSize;
+        info.narHash = narHash;
+        info.ca = ca;
+        if (deriver != "")
+            info.deriver = localStore.parseStorePath(deriver);
+        infos.insert_or_assign(info.path, info);
+    }
+
+    return infos;
+}
+
+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, ValidPathInfo> & infos
+)
+{
+      auto pathsSorted = reverseTopoSortPaths(infos);
+
+      for (auto & path : pathsSorted) {
+          auto & info = infos.find(path)->second;
+          copyPathFromRemote(conn, narMembers, localStore, destStore, info);
+      }
+
+}
+
+}
+
+/* using namespace nix::build_remote; */
+
+void RemoteResult::updateWithBuildResult(const nix::BuildResult & buildResult)
+{
+    startTime = buildResult.startTime;
+    stopTime = buildResult.stopTime;
+    timesBuilt = buildResult.timesBuilt;
+    errorMsg = buildResult.errorMsg;
+    isNonDeterministic = buildResult.isNonDeterministic;
+
+    switch ((BuildResult::Status) buildResult.status) {
+        case BuildResult::Built:
+            stepStatus = bsSuccess;
+            break;
+        case BuildResult::Substituted:
+        case BuildResult::AlreadyValid:
+            stepStatus = bsSuccess;
+            isCached = true;
+            break;
+        case BuildResult::PermanentFailure:
+            stepStatus = bsFailed;
+            canCache = true;
+            errorMsg = "";
+            break;
+        case BuildResult::InputRejected:
+        case BuildResult::OutputRejected:
+            stepStatus = bsFailed;
+            canCache = true;
+            break;
+        case BuildResult::TransientFailure:
+            stepStatus = bsFailed;
+            canRetry = true;
+            errorMsg = "";
+            break;
+        case BuildResult::TimedOut:
+            stepStatus = bsTimedOut;
+            errorMsg = "";
+            break;
+        case BuildResult::MiscFailure:
+            stepStatus = bsAborted;
+            canRetry = true;
+            break;
+        case BuildResult::LogLimitExceeded:
+            stepStatus = bsLogLimitExceeded;
+            break;
+        case BuildResult::NotDeterministic:
+            stepStatus = bsNotDeterministic;
+            canRetry = false;
+            canCache = true;
+            break;
+        default:
+            stepStatus = bsAborted;
+            break;
+    }
+
+}
+
 
 void State::buildRemote(ref<Store> destStore,
-    Machine::ptr machine, Step::ptr step,
-    unsigned int maxSilentTime, unsigned int buildTimeout, unsigned int repeats,
+    ::Machine::ptr machine, Step::ptr step,
+    const BuildOptions & buildOptions,
     RemoteResult & result, std::shared_ptr<ActiveStep> activeStep,
     std::function<void(StepState)> updateStep,
     NarMemberDatas & narMembers)
 {
     assert(BuildResult::TimedOut == 8);
 
-    std::string base(step->drvPath.to_string());
-    result.logFile = logDir + "/" + std::string(base, 0, 2) + "/" + std::string(base, 2);
-    AutoDelete autoDelete(result.logFile, false);
-
-    createDirs(dirOf(result.logFile));
-
-    AutoCloseFD logFD = open(result.logFile.c_str(), O_CREAT | O_TRUNC | O_WRONLY, 0666);
-    if (!logFD) throw SysError("creating log file ‘%s’", result.logFile);
+    auto [logFile, logFD] = build_remote::openLogFile(logDir, step->drvPath);
+    AutoDelete logFileDel(logFile, false);
+    result.logFile = logFile;
 
     nix::Path tmpDir = createTempDir();
     AutoDelete tmpDirDel(tmpDir, true);
@@ -205,13 +511,13 @@ void State::buildRemote(ref<Store> destStore,
         updateStep(ssConnecting);
 
         // FIXME: rewrite to use Store.
-        Child child;
-        openConnection(machine, tmpDir, logFD.get(), child);
+        SSHMaster::Connection child;
+        build_remote::openConnection(machine, tmpDir, logFD.get(), child);
 
         {
             auto activeStepState(activeStep->state_.lock());
             if (activeStepState->cancelled) throw Error("step cancelled");
-            activeStepState->pid = child.pid;
+            activeStepState->pid = child.sshPid;
         }
 
         Finally clearPid([&]() {
@@ -226,34 +532,21 @@ void State::buildRemote(ref<Store> destStore,
                process. Meh. */
         });
 
-        FdSource from(child.from.get());
-        WorkerProto::ReadConn rconn { .from = from };
-        FdSink to(child.to.get());
-        WorkerProto::WriteConn wconn { .to = to };
+        ::Machine::Connection conn {
+            .from = child.out.get(),
+            .to = child.in.get(),
+            .machine = machine,
+        };
 
         Finally updateStats([&]() {
-            bytesReceived += from.read;
-            bytesSent += to.written;
+            bytesReceived += conn.from.read;
+            bytesSent += conn.to.written;
         });
 
-        /* Handshake. */
-        unsigned int remoteVersion;
-
         try {
-            to << SERVE_MAGIC_1 << 0x206;
-            to.flush();
-
-            unsigned int magic = readInt(from);
-            if (magic != SERVE_MAGIC_2)
-                throw Error("protocol mismatch with ‘nix-store --serve’ on ‘%1%’", machine->sshName);
-            remoteVersion = readInt(from);
-            if (GET_PROTOCOL_MAJOR(remoteVersion) != 0x200)
-                throw Error("unsupported ‘nix-store --serve’ protocol version on ‘%1%’", machine->sshName);
-            if (GET_PROTOCOL_MINOR(remoteVersion) < 3 && repeats > 0)
-                throw Error("machine ‘%1%’ does not support repeating a build; please upgrade it to Nix 1.12", machine->sshName);
-
+          build_remote::handshake(conn, buildOptions.repeats);
         } catch (EndOfFile & e) {
-            child.pid.wait();
+            child.sshPid.wait();
             std::string s = chomp(readFile(result.logFile));
             throw Error("cannot connect to ‘%1%’: %2%", machine->sshName, s);
         }
@@ -269,62 +562,12 @@ void State::buildRemote(ref<Store> destStore,
            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);
 
-        StorePathSet inputs;
-        BasicDerivation basicDrv(*step->drv);
-
-        for (auto & p : step->drv->inputSrcs)
-            inputs.insert(p);
-
-        for (auto & input : step->drv->inputDrvs) {
-            auto drv2 = localStore->readDerivation(input.first);
-            for (auto & name : input.second) {
-                if (auto i = get(drv2.outputs, name)) {
-                    auto outPath = i->path(*localStore, drv2.name, name);
-                    inputs.insert(*outPath);
-                    basicDrv.inputSrcs.insert(*outPath);
-                }
-            }
-        }
-
-        /* 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 != std::shared_ptr<Store>(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), machine->sshName);
-
-            auto now1 = std::chrono::steady_clock::now();
-
-            /* Copy the input closure. */
-            if (machine->isLocalhost()) {
-                StorePathSet closure;
-                destStore->computeFSClosure(inputs, closure);
-                copyPaths(*destStore, *localStore, closure, NoRepair, NoCheckSigs, NoSubstitute);
-            } else {
-                copyClosureTo(machine->state->sendLock, *destStore, from, to, inputs, true);
-            }
-
-            auto now2 = std::chrono::steady_clock::now();
-
-            result.overhead += std::chrono::duration_cast<std::chrono::milliseconds>(now2 - now1).count();
-        }
-
-        autoDelete.cancel();
+        logFileDel.cancel();
 
         /* Truncate the log to get rid of messages about substitutions
-           etc. on the remote system. */
+            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);
 
@@ -340,85 +583,17 @@ void State::buildRemote(ref<Store> destStore,
 
         updateStep(ssBuilding);
 
-        to << ServeProto::Command::BuildDerivation << localStore->printStorePath(step->drvPath);
-        writeDerivation(to, *localStore, basicDrv);
-        to << maxSilentTime << buildTimeout;
-        if (GET_PROTOCOL_MINOR(remoteVersion) >= 2)
-            to << maxLogSize;
-        if (GET_PROTOCOL_MINOR(remoteVersion) >= 3) {
-            to << repeats // == build-repeat
-               << step->isDeterministic; // == enforce-determinism
-        }
-        to.flush();
+        BuildResult buildResult = build_remote::performBuild(
+            conn,
+            *localStore,
+            step->drvPath,
+            resolvedDrv,
+            buildOptions,
+            nrStepsBuilding
+        );
 
-        result.startTime = time(0);
-        int res;
-        {
-            MaintainCount<counter> mc(nrStepsBuilding);
-            res = readInt(from);
-        }
-        result.stopTime = time(0);
+        result.updateWithBuildResult(buildResult);
 
-        result.errorMsg = readString(from);
-        if (GET_PROTOCOL_MINOR(remoteVersion) >= 3) {
-            result.timesBuilt = readInt(from);
-            result.isNonDeterministic = readInt(from);
-            auto start = readInt(from);
-            auto stop = readInt(from);
-            if (start && start) {
-                /* Note: this represents the duration of a single
-                    round, rather than all rounds. */
-                result.startTime = start;
-                result.stopTime = stop;
-            }
-        }
-        if (GET_PROTOCOL_MINOR(remoteVersion) >= 6) {
-            WorkerProto::Serialise<DrvOutputs>::read(*localStore, rconn);
-        }
-        switch ((BuildResult::Status) res) {
-            case BuildResult::Built:
-                result.stepStatus = bsSuccess;
-                break;
-            case BuildResult::Substituted:
-            case BuildResult::AlreadyValid:
-                result.stepStatus = bsSuccess;
-                result.isCached = true;
-                break;
-            case BuildResult::PermanentFailure:
-                result.stepStatus = bsFailed;
-                result.canCache = true;
-                result.errorMsg = "";
-                break;
-            case BuildResult::InputRejected:
-            case BuildResult::OutputRejected:
-                result.stepStatus = bsFailed;
-                result.canCache = true;
-                break;
-            case BuildResult::TransientFailure:
-                result.stepStatus = bsFailed;
-                result.canRetry = true;
-                result.errorMsg = "";
-                break;
-            case BuildResult::TimedOut:
-                result.stepStatus = bsTimedOut;
-                result.errorMsg = "";
-                break;
-            case BuildResult::MiscFailure:
-                result.stepStatus = bsAborted;
-                result.canRetry = true;
-                break;
-            case BuildResult::LogLimitExceeded:
-                result.stepStatus = bsLogLimitExceeded;
-                break;
-            case BuildResult::NotDeterministic:
-                result.stepStatus = bsNotDeterministic;
-                result.canRetry = false;
-                result.canCache = true;
-                break;
-            default:
-                result.stepStatus = bsAborted;
-                break;
-        }
         if (result.stepStatus != bsSuccess) return;
 
         result.errorMsg = "";
@@ -432,6 +607,10 @@ 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);
@@ -440,39 +619,8 @@ void State::buildRemote(ref<Store> destStore,
 
             auto now1 = std::chrono::steady_clock::now();
 
-            StorePathSet outputs;
-            for (auto & i : step->drv->outputsAndOptPaths(*localStore)) {
-                if (i.second.second)
-                   outputs.insert(*i.second.second);
-            }
-
-            /* Get info about each output path. */
-            std::map<StorePath, ValidPathInfo> infos;
             size_t totalNarSize = 0;
-            to << ServeProto::Command::QueryPathInfos;
-            WorkerProto::write(*localStore, wconn, outputs);
-            to.flush();
-            while (true) {
-                auto storePathS = readString(from);
-                if (storePathS == "") break;
-                auto deriver = readString(from); // deriver
-                auto references = WorkerProto::Serialise<StorePathSet>::read(*localStore, rconn);
-                readLongLong(from); // download size
-                auto narSize = readLongLong(from);
-                auto narHash = Hash::parseAny(readString(from), htSHA256);
-                auto ca = ContentAddress::parseOpt(readString(from));
-                readStrings<StringSet>(from); // sigs
-                ValidPathInfo info(localStore->parseStorePath(storePathS), narHash);
-                assert(outputs.count(info.path));
-                info.references = references;
-                info.narSize = narSize;
-                totalNarSize += info.narSize;
-                info.narHash = narHash;
-                info.ca = ca;
-                if (deriver != "")
-                    info.deriver = localStore->parseStorePath(deriver);
-                infos.insert_or_assign(info.path, info);
-            }
+            auto infos = build_remote::queryPathInfos(conn, *localStore, outputs, totalNarSize);
 
             if (totalNarSize > maxOutputSize) {
                 result.stepStatus = bsNarSizeLimitExceeded;
@@ -483,41 +631,30 @@ void State::buildRemote(ref<Store> destStore,
             printMsg(lvlDebug, "copying outputs of ‘%s’ from ‘%s’ (%d bytes)",
                 localStore->printStorePath(step->drvPath), machine->sshName, totalNarSize);
 
-            auto pathsSorted = reverseTopoSortPaths(infos);
-
-            for (auto & path : pathsSorted) {
-                auto & info = infos.find(path)->second;
-
-                /* 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. */
-                    to << ServeProto::Command::DumpStorePath << localStore->printStorePath(path);
-                    to.flush();
-
-                    TeeSource tee(from, sink);
-                    extractNarData(tee, localStore->printStorePath(path), narMembers);
-                });
-
-                destStore->addToStore(info, *source2, NoRepair, NoCheckSigs);
-            }
-
+            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.to = -1;
-        child.pid.wait();
+        child.in = -1;
+        child.sshPid.wait();
 
     } catch (Error & e) {
         /* Disable this machine until a certain period of time has

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

@@ -1,7 +1,7 @@
 #include "hydra-build-result.hh"
 #include "store-api.hh"
 #include "util.hh"
-#include "fs-accessor.hh"
+#include "source-accessor.hh"
 
 #include <regex>
 
@@ -11,18 +11,18 @@ using namespace nix;
 BuildOutput getBuildOutput(
     nix::ref<Store> store,
     NarMemberDatas & narMembers,
-    const Derivation & drv)
+    const OutputPathMap derivationOutputs)
 {
     BuildOutput res;
 
     /* Compute the closure size. */
     StorePathSet outputs;
     StorePathSet closure;
-    for (auto & i : drv.outputsAndOptPaths(*store))
-        if (i.second.second) {
-            store->computeFSClosure(*i.second.second, closure);
-            outputs.insert(*i.second.second);
-        }
+    for (auto& [outputName, outputPath] : derivationOutputs) {
+        store->computeFSClosure(outputPath, closure);
+        outputs.insert(outputPath);
+        res.outputs.insert({outputName, outputPath});
+    }
     for (auto & path : closure) {
         auto info = store->queryPathInfo(path);
         res.closureSize += info->narSize;
@@ -63,7 +63,7 @@ BuildOutput getBuildOutput(
 
         auto productsFile = narMembers.find(outputS + "/nix-support/hydra-build-products");
         if (productsFile == narMembers.end() ||
-            productsFile->second.type != FSAccessor::Type::tRegular)
+            productsFile->second.type != SourceAccessor::Type::tRegular)
             continue;
         assert(productsFile->second.contents);
 
@@ -94,7 +94,7 @@ BuildOutput getBuildOutput(
 
             product.name = product.path == store->printStorePath(output) ? "" : baseNameOf(product.path);
 
-            if (file->second.type == FSAccessor::Type::tRegular) {
+            if (file->second.type == SourceAccessor::Type::tRegular) {
                 product.isRegular = true;
                 product.fileSize = file->second.fileSize.value();
                 product.sha256hash = file->second.sha256.value();
@@ -107,17 +107,16 @@ BuildOutput getBuildOutput(
     /* If no build products were explicitly declared, then add all
        outputs as a product of type "nix-build". */
     if (!explicitProducts) {
-        for (auto & [name, output] : drv.outputs) {
+        for (auto & [name, output] : derivationOutputs) {
             BuildProduct product;
-            auto outPath = output.path(*store, drv.name, name);
-            product.path = store->printStorePath(*outPath);
+            product.path = store->printStorePath(output);
             product.type = "nix-build";
             product.subtype = name == "out" ? "" : name;
-            product.name = outPath->name();
+            product.name = output.name();
 
             auto file = narMembers.find(product.path);
             assert(file != narMembers.end());
-            if (file->second.type == FSAccessor::Type::tDirectory)
+            if (file->second.type == SourceAccessor::Type::tDirectory)
                 res.products.push_back(product);
         }
     }
@@ -126,7 +125,7 @@ BuildOutput getBuildOutput(
     for (auto & output : outputs) {
         auto file = narMembers.find(store->printStorePath(output) + "/nix-support/hydra-release-name");
         if (file == narMembers.end() ||
-            file->second.type != FSAccessor::Type::tRegular)
+            file->second.type != SourceAccessor::Type::tRegular)
             continue;
         res.releaseName = trim(file->second.contents.value());
         // FIXME: validate release name
@@ -136,7 +135,7 @@ BuildOutput getBuildOutput(
     for (auto & output : outputs) {
         auto file = narMembers.find(store->printStorePath(output) + "/nix-support/hydra-metrics");
         if (file == narMembers.end() ||
-            file->second.type != FSAccessor::Type::tRegular)
+            file->second.type != SourceAccessor::Type::tRegular)
             continue;
         for (auto & line : tokenizeString<Strings>(file->second.contents.value(), "\n")) {
             auto fields = tokenizeString<std::vector<std::string>>(line);

src/hydra-queue-runner/builder.cc

@@ -98,8 +98,10 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
        it). */
     BuildID buildId;
     std::optional<StorePath> buildDrvPath;
-    unsigned int maxSilentTime, buildTimeout;
-    unsigned int repeats = step->isDeterministic ? 1 : 0;
+    BuildOptions buildOptions;
+    buildOptions.repeats = step->isDeterministic ? 1 : 0;
+    buildOptions.maxLogSize = maxLogSize;
+    buildOptions.enforceDeterminism = step->isDeterministic;
 
     auto conn(dbPool.get());
 
@@ -134,18 +136,18 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
             {
                 auto i = jobsetRepeats.find(std::make_pair(build2->projectName, build2->jobsetName));
                 if (i != jobsetRepeats.end())
-                    repeats = std::max(repeats, i->second);
+                    buildOptions.repeats = std::max(buildOptions.repeats, i->second);
             }
         }
         if (!build) build = *dependents.begin();
 
         buildId = build->id;
         buildDrvPath = build->drvPath;
-        maxSilentTime = build->maxSilentTime;
-        buildTimeout = build->buildTimeout;
+        buildOptions.maxSilentTime = build->maxSilentTime;
+        buildOptions.buildTimeout = build->buildTimeout;
 
         printInfo("performing step ‘%s’ %d times on ‘%s’ (needed by build %d and %d others)",
-            localStore->printStorePath(step->drvPath), repeats + 1, machine->sshName, buildId, (dependents.size() - 1));
+            localStore->printStorePath(step->drvPath), buildOptions.repeats + 1, machine->sshName, buildId, (dependents.size() - 1));
     }
 
     if (!buildOneDone)
@@ -206,7 +208,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         try {
             /* FIXME: referring builds may have conflicting timeouts. */
-            buildRemote(destStore, machine, step, maxSilentTime, buildTimeout, repeats, result, activeStep, updateStep, narMembers);
+            buildRemote(destStore, machine, step, buildOptions, result, activeStep, updateStep, narMembers);
         } catch (Error & e) {
             if (activeStep->state_.lock()->cancelled) {
                 printInfo("marking step %d of build %d as cancelled", stepNr, buildId);
@@ -221,7 +223,7 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         if (result.stepStatus == bsSuccess) {
             updateStep(ssPostProcessing);
-            res = getBuildOutput(destStore, narMembers, *step->drv);
+            res = getBuildOutput(destStore, narMembers, destStore->queryDerivationOutputMap(step->drvPath, &*localStore));
         }
     }
 
@@ -275,9 +277,12 @@ State::StepResult State::doBuildStep(nix::ref<Store> destStore,
 
         assert(stepNr);
 
-        for (auto & i : step->drv->outputsAndOptPaths(*localStore)) {
-            if (i.second.second)
-               addRoot(*i.second.second);
+        for (auto & [outputName, optOutputPath] : destStore->queryPartialDerivationOutputMap(step->drvPath, &*localStore)) {
+            if (!optOutputPath)
+                throw Error(
+                    "Missing output %s for derivation %d which was supposed to have succeeded",
+                    outputName, localStore->printStorePath(step->drvPath));
+            addRoot(*optOutputPath);
         }
 
         /* Register success in the database for all Build objects that
@@ -398,7 +403,7 @@ void State::failStep(
     Step::ptr step,
     BuildID buildId,
     const RemoteResult & result,
-    Machine::ptr machine,
+    ::Machine::ptr machine,
     bool & stepFinished)
 {
     /* Register failure in the database for all Build objects that

src/hydra-queue-runner/dispatcher.cc

@@ -85,12 +85,113 @@ system_time State::doDispatch()
         }
     }
 
+    system_time now = std::chrono::system_clock::now();
+
     /* Start steps until we're out of steps or slots. */
     auto sleepUntil = system_time::max();
     bool keepGoing;
 
+    /* Sort the runnable steps by priority. Priority is establised
+       as follows (in order of precedence):
+
+       - The global priority of the builds that depend on the
+         step. This allows admins to bump a build to the front of
+         the queue.
+
+       - The lowest used scheduling share of the jobsets depending
+         on the step.
+
+       - The local priority of the build, as set via the build's
+         meta.schedulingPriority field. Note that this is not
+         quite correct: the local priority should only be used to
+         establish priority between builds in the same jobset, but
+         here it's used between steps in different jobsets if they
+         happen to have the same lowest used scheduling share. But
+         that's not very likely.
+
+       - The lowest ID of the builds depending on the step;
+         i.e. older builds take priority over new ones.
+
+       FIXME: O(n lg n); obviously, it would be better to keep a
+       runnable queue sorted by priority. */
+    struct StepInfo
+    {
+        Step::ptr step;
+        bool alreadyScheduled = false;
+
+        /* The lowest share used of any jobset depending on this
+           step. */
+        double lowestShareUsed = 1e9;
+
+        /* Info copied from step->state to ensure that the
+           comparator is a partial ordering (see MachineInfo). */
+        int highestGlobalPriority;
+        int highestLocalPriority;
+        BuildID lowestBuildID;
+
+        StepInfo(Step::ptr step, Step::State & step_) : step(step)
+        {
+            for (auto & jobset : step_.jobsets)
+                lowestShareUsed = std::min(lowestShareUsed, jobset->shareUsed());
+            highestGlobalPriority = step_.highestGlobalPriority;
+            highestLocalPriority = step_.highestLocalPriority;
+            lowestBuildID = step_.lowestBuildID;
+        }
+    };
+
+    std::vector<StepInfo> runnableSorted;
+
+    struct RunnablePerType
+    {
+        unsigned int count{0};
+        std::chrono::seconds waitTime{0};
+    };
+
+    std::unordered_map<std::string, RunnablePerType> runnablePerType;
+
+    {
+        auto runnable_(runnable.lock());
+        runnableSorted.reserve(runnable_->size());
+        for (auto i = runnable_->begin(); i != runnable_->end(); ) {
+            auto step = i->lock();
+
+            /* Remove dead steps. */
+            if (!step) {
+                i = runnable_->erase(i);
+                continue;
+            }
+
+            ++i;
+
+            auto & r = runnablePerType[step->systemType];
+            r.count++;
+
+            /* Skip previously failed steps that aren't ready
+               to be retried. */
+            auto step_(step->state.lock());
+            r.waitTime += std::chrono::duration_cast<std::chrono::seconds>(now - step_->runnableSince);
+            if (step_->tries > 0 && step_->after > now) {
+                if (step_->after < sleepUntil)
+                    sleepUntil = step_->after;
+                continue;
+            }
+
+            runnableSorted.emplace_back(step, *step_);
+        }
+    }
+
+    sort(runnableSorted.begin(), runnableSorted.end(),
+        [](const StepInfo & a, const StepInfo & b)
+        {
+            return
+                a.highestGlobalPriority != b.highestGlobalPriority ? a.highestGlobalPriority > b.highestGlobalPriority :
+                a.lowestShareUsed != b.lowestShareUsed ? a.lowestShareUsed < b.lowestShareUsed :
+                a.highestLocalPriority != b.highestLocalPriority ? a.highestLocalPriority > b.highestLocalPriority :
+                a.lowestBuildID < b.lowestBuildID;
+        });
+
     do {
-        system_time now = std::chrono::system_clock::now();
+        now = std::chrono::system_clock::now();
 
         /* Copy the currentJobs field of each machine. This is
            necessary to ensure that the sort comparator below is
@@ -98,7 +199,7 @@ system_time State::doDispatch()
            filter out temporarily disabled machines. */
         struct MachineInfo
         {
-            Machine::ptr machine;
+            ::Machine::ptr machine;
             unsigned long currentJobs;
         };
         std::vector<MachineInfo> machinesSorted;
@@ -130,112 +231,14 @@ system_time State::doDispatch()
         sort(machinesSorted.begin(), machinesSorted.end(),
             [](const MachineInfo & a, const MachineInfo & b) -> bool
             {
-                float ta = std::round(a.currentJobs / a.machine->speedFactor);
-                float tb = std::round(b.currentJobs / b.machine->speedFactor);
+                float ta = std::round(a.currentJobs / a.machine->speedFactorFloat);
+                float tb = std::round(b.currentJobs / b.machine->speedFactorFloat);
                 return
                     ta != tb ? ta < tb :
-                    a.machine->speedFactor != b.machine->speedFactor ? a.machine->speedFactor > b.machine->speedFactor :
+                    a.machine->speedFactorFloat != b.machine->speedFactorFloat ? a.machine->speedFactorFloat > b.machine->speedFactorFloat :
                     a.currentJobs > b.currentJobs;
             });
 
-        /* Sort the runnable steps by priority. Priority is establised
-           as follows (in order of precedence):
-
-           - The global priority of the builds that depend on the
-             step. This allows admins to bump a build to the front of
-             the queue.
-
-           - The lowest used scheduling share of the jobsets depending
-             on the step.
-
-           - The local priority of the build, as set via the build's
-             meta.schedulingPriority field. Note that this is not
-             quite correct: the local priority should only be used to
-             establish priority between builds in the same jobset, but
-             here it's used between steps in different jobsets if they
-             happen to have the same lowest used scheduling share. But
-             that's not very likely.
-
-           - The lowest ID of the builds depending on the step;
-             i.e. older builds take priority over new ones.
-
-           FIXME: O(n lg n); obviously, it would be better to keep a
-           runnable queue sorted by priority. */
-        struct StepInfo
-        {
-            Step::ptr step;
-
-            /* The lowest share used of any jobset depending on this
-               step. */
-            double lowestShareUsed = 1e9;
-
-            /* Info copied from step->state to ensure that the
-               comparator is a partial ordering (see MachineInfo). */
-            int highestGlobalPriority;
-            int highestLocalPriority;
-            BuildID lowestBuildID;
-
-            StepInfo(Step::ptr step, Step::State & step_) : step(step)
-            {
-                for (auto & jobset : step_.jobsets)
-                    lowestShareUsed = std::min(lowestShareUsed, jobset->shareUsed());
-                highestGlobalPriority = step_.highestGlobalPriority;
-                highestLocalPriority = step_.highestLocalPriority;
-                lowestBuildID = step_.lowestBuildID;
-            }
-        };
-
-        std::vector<StepInfo> runnableSorted;
-
-        struct RunnablePerType
-        {
-            unsigned int count{0};
-            std::chrono::seconds waitTime{0};
-        };
-
-        std::unordered_map<std::string, RunnablePerType> runnablePerType;
-
-        {
-            auto runnable_(runnable.lock());
-            runnableSorted.reserve(runnable_->size());
-            for (auto i = runnable_->begin(); i != runnable_->end(); ) {
-                auto step = i->lock();
-
-                /* Remove dead steps. */
-                if (!step) {
-                    i = runnable_->erase(i);
-                    continue;
-                }
-
-                ++i;
-
-                auto & r = runnablePerType[step->systemType];
-                r.count++;
-
-                /* Skip previously failed steps that aren't ready
-                   to be retried. */
-                auto step_(step->state.lock());
-                r.waitTime += std::chrono::duration_cast<std::chrono::seconds>(now - step_->runnableSince);
-                if (step_->tries > 0 && step_->after > now) {
-                    if (step_->after < sleepUntil)
-                        sleepUntil = step_->after;
-                    continue;
-                }
-
-                runnableSorted.emplace_back(step, *step_);
-            }
-        }
-
-        sort(runnableSorted.begin(), runnableSorted.end(),
-            [](const StepInfo & a, const StepInfo & b)
-            {
-                return
-                    a.highestGlobalPriority != b.highestGlobalPriority ? a.highestGlobalPriority > b.highestGlobalPriority :
-                    a.lowestShareUsed != b.lowestShareUsed ? a.lowestShareUsed < b.lowestShareUsed :
-                    a.highestLocalPriority != b.highestLocalPriority ? a.highestLocalPriority > b.highestLocalPriority :
-                    a.lowestBuildID < b.lowestBuildID;
-            });
-
         /* Find a machine with a free slot and find a step to run
            on it. Once we find such a pair, we restart the outer
            loop because the machine sorting will have changed. */
@@ -245,6 +248,8 @@ system_time State::doDispatch()
             if (mi.machine->state->currentJobs >= mi.machine->maxJobs) continue;
 
             for (auto & stepInfo : runnableSorted) {
+                if (stepInfo.alreadyScheduled) continue;
+
                 auto & step(stepInfo.step);
 
                 /* Can this machine do this step? */
@@ -271,6 +276,8 @@ system_time State::doDispatch()
                     r.count--;
                 }
 
+                stepInfo.alreadyScheduled = true;
+
                 /* Make a slot reservation and start a thread to
                    do the build. */
                 auto builderThread = std::thread(&State::builder, this,
@@ -428,7 +435,7 @@ void Jobset::pruneSteps()
 }
 
 
-State::MachineReservation::MachineReservation(State & state, Step::ptr step, Machine::ptr machine)
+State::MachineReservation::MachineReservation(State & state, Step::ptr step, ::Machine::ptr machine)
     : state(state), step(step), machine(machine)
 {
     machine->state->currentJobs++;

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

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

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

@@ -1,6 +1,7 @@
 #include <iostream>
 #include <thread>
 #include <optional>
+#include <type_traits>
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -10,6 +11,7 @@
 
 #include <nlohmann/json.hpp>
 
+#include "signals.hh"
 #include "state.hh"
 #include "hydra-build-result.hh"
 #include "store-api.hh"
@@ -139,23 +141,43 @@ void State::parseMachines(const std::string & contents)
         if (tokens.size() < 3) continue;
         tokens.resize(8);
 
-        auto machine = std::make_shared<Machine>();
-        machine->sshName = tokens[0];
-        machine->systemTypes = tokenizeString<StringSet>(tokens[1], ",");
-        machine->sshKey = tokens[2] == "-" ? std::string("") : tokens[2];
-        if (tokens[3] != "")
-            machine->maxJobs = string2Int<decltype(machine->maxJobs)>(tokens[3]).value();
-        else
-            machine->maxJobs = 1;
-        machine->speedFactor = atof(tokens[4].c_str());
         if (tokens[5] == "-") tokens[5] = "";
-        machine->supportedFeatures = tokenizeString<StringSet>(tokens[5], ",");
+        auto supportedFeatures = tokenizeString<StringSet>(tokens[5], ",");
+
         if (tokens[6] == "-") tokens[6] = "";
-        machine->mandatoryFeatures = tokenizeString<StringSet>(tokens[6], ",");
-        for (auto & f : machine->mandatoryFeatures)
-            machine->supportedFeatures.insert(f);
-        if (tokens[7] != "" && tokens[7] != "-")
-            machine->sshPublicHostKey = base64Decode(tokens[7]);
+        auto mandatoryFeatures = tokenizeString<StringSet>(tokens[6], ",");
+
+        for (auto & f : mandatoryFeatures)
+            supportedFeatures.insert(f);
+
+        using MaxJobs = std::remove_const<decltype(nix::Machine::maxJobs)>::type;
+
+        auto machine = std::make_shared<::Machine>(nix::Machine {
+            // `storeUri`, not yet used
+            "",
+            // `systemTypes`, not yet used
+            {},
+            // `sshKey`
+            tokens[2] == "-" ? "" : tokens[2],
+            // `maxJobs`
+            tokens[3] != ""
+                ? string2Int<MaxJobs>(tokens[3]).value()
+                : 1,
+            // `speedFactor`, not yet used
+            1,
+            // `supportedFeatures`
+            std::move(supportedFeatures),
+            // `mandatoryFeatures`
+            std::move(mandatoryFeatures),
+            // `sshPublicHostKey`
+            tokens[7] != "" && tokens[7] != "-"
+                ? base64Decode(tokens[7])
+                : "",
+        });
+
+        machine->sshName = tokens[0];
+        machine->systemTypesSet = tokenizeString<StringSet>(tokens[1], ",");
+        machine->speedFactorFloat = atof(tokens[4].c_str());
 
         /* Re-use the State object of the previous machine with the
            same name. */
@@ -165,7 +187,7 @@ void State::parseMachines(const std::string & contents)
         else
             printMsg(lvlChatty, "updating machine ‘%1%’", machine->sshName);
         machine->state = i == oldMachines.end()
-            ? std::make_shared<Machine::State>()
+            ? std::make_shared<::Machine::State>()
             : i->second->state;
         newMachines[machine->sshName] = machine;
     }
@@ -174,9 +196,9 @@ void State::parseMachines(const std::string & contents)
         if (newMachines.find(m.first) == newMachines.end()) {
             if (m.second->enabled)
                 printInfo("removing machine ‘%1%’", m.first);
-            /* Add a disabled Machine object to make sure stats are
+            /* Add a disabled ::Machine object to make sure stats are
                maintained. */
-            auto machine = std::make_shared<Machine>(*(m.second));
+            auto machine = std::make_shared<::Machine>(*(m.second));
             machine->enabled = false;
             newMachines[m.first] = machine;
         }
@@ -204,7 +226,7 @@ void State::monitorMachinesFile()
         parseMachines("localhost " +
             (settings.thisSystem == "x86_64-linux" ? "x86_64-linux,i686-linux" : settings.thisSystem.get())
             + " - " + std::to_string(settings.maxBuildJobs) + " 1 "
-            + concatStringsSep(",", settings.systemFeatures.get()));
+            + concatStringsSep(",", StoreConfig::getDefaultSystemFeatures()));
         machinesReadyLock.unlock();
         return;
     }
@@ -311,10 +333,13 @@ unsigned int State::createBuildStep(pqxx::work & txn, time_t startTime, BuildID
 
     if (r.affected_rows() == 0) goto restart;
 
-    for (auto & [name, output] : step->drv->outputs)
+    for (auto & [name, output] : getDestStore()->queryPartialDerivationOutputMap(step->drvPath, &*localStore))
         txn.exec_params0
             ("insert into BuildStepOutputs (build, stepnr, name, path) values ($1, $2, $3, $4)",
-            buildId, stepNr, name, localStore->printStorePath(*output.path(*localStore, step->drv->name, name)));
+            buildId, stepNr, name,
+            output
+                ? std::optional { localStore->printStorePath(*output)}
+                : std::nullopt);
 
     if (status == bsBusy)
         txn.exec(fmt("notify step_started, '%d\t%d'", buildId, stepNr));
@@ -351,11 +376,23 @@ void State::finishBuildStep(pqxx::work & txn, const RemoteResult & result,
     assert(result.logFile.find('\t') == std::string::npos);
     txn.exec(fmt("notify step_finished, '%d\t%d\t%s'",
             buildId, stepNr, result.logFile));
+
+    if (result.stepStatus == bsSuccess) {
+        // Update the corresponding `BuildStepOutputs` row to add the output path
+        auto res = txn.exec_params1("select drvPath from BuildSteps where build = $1 and stepnr = $2", buildId, stepNr);
+        assert(res.size());
+        StorePath drvPath = localStore->parseStorePath(res[0].as<std::string>());
+        // If we've finished building, all the paths should be known
+        for (auto & [name, output] : getDestStore()->queryDerivationOutputMap(drvPath, &*localStore))
+            txn.exec_params0
+                ("update BuildStepOutputs set path = $4 where build = $1 and stepnr = $2 and name = $3",
+                  buildId, stepNr, name, localStore->printStorePath(output));
+    }
 }
 
 
 int State::createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
-    Build::ptr build, const StorePath & drvPath, const std::string & outputName, const StorePath & storePath)
+    Build::ptr build, const StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const StorePath & storePath)
 {
  restart:
     auto stepNr = allocBuildStep(txn, build->id);
@@ -456,6 +493,15 @@ void State::markSucceededBuild(pqxx::work & txn, Build::ptr build,
          res.releaseName != "" ? std::make_optional(res.releaseName) : std::nullopt,
          isCachedBuild ? 1 : 0);
 
+    for (auto & [outputName, outputPath] : res.outputs) {
+        txn.exec_params0
+            ("update BuildOutputs set path = $3 where build = $1 and name = $2",
+             build->id,
+             outputName,
+             localStore->printStorePath(outputPath)
+            );
+    }
+
     txn.exec_params0("delete from BuildProducts where build = $1", build->id);
 
     unsigned int productNr = 1;
@@ -467,7 +513,7 @@ void State::markSucceededBuild(pqxx::work & txn, Build::ptr build,
              product.type,
              product.subtype,
              product.fileSize ? std::make_optional(*product.fileSize) : std::nullopt,
-             product.sha256hash ? std::make_optional(product.sha256hash->to_string(Base16, false)) : std::nullopt,
+             product.sha256hash ? std::make_optional(product.sha256hash->to_string(HashFormat::Base16, false)) : std::nullopt,
              product.path,
              product.name,
              product.defaultPath);
@@ -595,7 +641,7 @@ void State::dumpStatus(Connection & conn)
 
                 json machine = {
                     {"enabled",  m->enabled},
-                    {"systemTypes", m->systemTypes},
+                    {"systemTypes", m->systemTypesSet},
                     {"supportedFeatures", m->supportedFeatures},
                     {"mandatoryFeatures", m->mandatoryFeatures},
                     {"nrStepsDone", s->nrStepsDone.load()},

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

@@ -24,13 +24,13 @@ struct Extractor : ParseSink
 
     void createDirectory(const Path & path) override
     {
-        members.insert_or_assign(prefix + path, NarMemberData { .type = FSAccessor::Type::tDirectory });
+        members.insert_or_assign(prefix + path, NarMemberData { .type = SourceAccessor::Type::tDirectory });
     }
 
     void createRegularFile(const Path & path) override
     {
         curMember = &members.insert_or_assign(prefix + path, NarMemberData {
-            .type = FSAccessor::Type::tRegular,
+            .type = SourceAccessor::Type::tRegular,
             .fileSize = 0,
             .contents = filesToKeep.count(path) ? std::optional("") : std::nullopt,
         }).first->second;
@@ -66,8 +66,14 @@ struct Extractor : ParseSink
 
     void createSymlink(const Path & path, const std::string & target) override
     {
-        members.insert_or_assign(prefix + path, NarMemberData { .type = FSAccessor::Type::tSymlink });
+        members.insert_or_assign(prefix + path, NarMemberData { .type = SourceAccessor::Type::tSymlink });
     }
+
+    void isExecutable() override
+    { }
+
+    void closeRegularFile() override
+    { }
 };
 
 

src/hydra-queue-runner/nar-extractor.hh

@@ -1,13 +1,13 @@
 #pragma once
 
-#include "fs-accessor.hh"
+#include "source-accessor.hh"
 #include "types.hh"
 #include "serialise.hh"
 #include "hash.hh"
 
 struct NarMemberData
 {
-    nix::FSAccessor::Type type;
+    nix::SourceAccessor::Type type;
     std::optional<uint64_t> fileSize;
     std::optional<std::string> contents;
     std::optional<nix::Hash> sha256;

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

@@ -192,15 +192,19 @@ bool State::getQueuedBuilds(Connection & conn,
                 if (!res[0].is_null()) propagatedFrom = res[0].as<BuildID>();
 
                 if (!propagatedFrom) {
-                    for (auto & i : ex.step->drv->outputsAndOptPaths(*localStore)) {
-                        if (i.second.second) {
-                            auto res = txn.exec_params
-                                ("select max(s.build) from BuildSteps s join BuildStepOutputs o on s.build = o.build where path = $1 and startTime != 0 and stopTime != 0 and status = 1",
-                                 localStore->printStorePath(*i.second.second));
-                            if (!res[0][0].is_null()) {
-                                propagatedFrom = res[0][0].as<BuildID>();
-                                break;
-                            }
+                    for (auto & [outputName, optOutputPath] : destStore->queryPartialDerivationOutputMap(ex.step->drvPath, &*localStore)) {
+                        constexpr std::string_view common = "select max(s.build) from BuildSteps s join BuildStepOutputs o on s.build = o.build where startTime != 0 and stopTime != 0 and status = 1";
+                        auto res = optOutputPath
+                            ? txn.exec_params(
+                                std::string { common } + " and path = $1",
+                                localStore->printStorePath(*optOutputPath))
+                            : txn.exec_params(
+                                std::string { common } + " and drvPath = $1 and name = $2",
+                                localStore->printStorePath(ex.step->drvPath),
+                                outputName);
+                        if (!res[0][0].is_null()) {
+                            propagatedFrom = res[0][0].as<BuildID>();
+                            break;
                         }
                     }
                 }
@@ -236,12 +240,10 @@ bool State::getQueuedBuilds(Connection & conn,
         /* If we didn't get a step, it means the step's outputs are
            all valid. So we mark this as a finished, cached build. */
         if (!step) {
-            auto drv = localStore->readDerivation(build->drvPath);
-            BuildOutput res = getBuildOutputCached(conn, destStore, drv);
+            BuildOutput res = getBuildOutputCached(conn, destStore, build->drvPath);
 
-            for (auto & i : drv.outputsAndOptPaths(*localStore))
-                if (i.second.second)
-                    addRoot(*i.second.second);
+            for (auto & i : destStore->queryDerivationOutputMap(build->drvPath, &*localStore))
+                addRoot(i.second);
 
             {
             auto mc = startDbUpdate();
@@ -315,7 +317,7 @@ bool State::getQueuedBuilds(Connection & conn,
         if (std::chrono::system_clock::now() > start + std::chrono::seconds(600)) {
             prom.queue_checks_early_exits.Increment();
             break;
-        } 
+        }
     }
 
     prom.queue_checks_finished.Increment();
@@ -464,10 +466,7 @@ Step::ptr State::createStep(ref<Store> destStore,
 
     step->systemType = step->drv->platform;
     {
-        auto i = step->drv->env.find("requiredSystemFeatures");
-        StringSet features;
-        if (i != step->drv->env.end())
-            features = step->requiredSystemFeatures = tokenizeString<std::set<std::string>>(i->second);
+        StringSet features = step->requiredSystemFeatures = step->parsedDrv->getRequiredSystemFeatures();
         if (step->preferLocalBuild)
             features.insert("local");
         if (!features.empty()) {
@@ -481,26 +480,41 @@ Step::ptr State::createStep(ref<Store> destStore,
         throw PreviousFailure{step};
 
     /* Are all outputs valid? */
+    auto outputHashes = staticOutputHashes(*localStore, *(step->drv));
     bool valid = true;
-    DerivationOutputs missing;
-    for (auto & i : step->drv->outputs)
-        if (!destStore->isValidPath(*i.second.path(*localStore, step->drv->name, i.first))) {
-            valid = false;
-            missing.insert_or_assign(i.first, i.second);
-        }
+    std::map<DrvOutput, std::optional<StorePath>> missing;
+    for (auto & [outputName, maybeOutputPath] : destStore->queryPartialDerivationOutputMap(drvPath, &*localStore)) {
+        auto outputHash = outputHashes.at(outputName);
+        if (maybeOutputPath && destStore->isValidPath(*maybeOutputPath))
+            continue;
+        valid = false;
+        missing.insert({{outputHash, outputName}, maybeOutputPath});
+    }
 
     /* Try to copy the missing paths from the local store or from
        substitutes. */
     if (!missing.empty()) {
 
         size_t avail = 0;
-        for (auto & i : missing) {
-            auto path = i.second.path(*localStore, step->drv->name, i.first);
-            if (/* localStore != destStore && */ localStore->isValidPath(*path))
+        for (auto & [i, pathOpt] : missing) {
+            // If we don't know the output path from the destination
+            // store, see if the local store can tell us.
+            if (/* localStore != destStore && */ !pathOpt && experimentalFeatureSettings.isEnabled(Xp::CaDerivations))
+                if (auto maybeRealisation = localStore->queryRealisation(i))
+                    pathOpt = maybeRealisation->outPath;
+
+            if (!pathOpt) {
+                // No hope of getting the store object if we don't know
+                // the path.
+                continue;
+            }
+            auto & path = *pathOpt;
+
+            if (/* localStore != destStore && */ localStore->isValidPath(path))
                 avail++;
             else if (useSubstitutes) {
                 SubstitutablePathInfos infos;
-                localStore->querySubstitutablePathInfos({{*path, {}}}, infos);
+                localStore->querySubstitutablePathInfos({{path, {}}}, infos);
                 if (infos.size() == 1)
                     avail++;
             }
@@ -508,26 +522,29 @@ Step::ptr State::createStep(ref<Store> destStore,
 
         if (missing.size() == avail) {
             valid = true;
-            for (auto & i : missing) {
-                auto path = i.second.path(*localStore, step->drv->name, i.first);
+            for (auto & [i, pathOpt] : missing) {
+                // If we found everything, then we should know the path
+                // to every missing store object now.
+                assert(pathOpt);
+                auto & path = *pathOpt;
 
                 try {
                     time_t startTime = time(0);
 
-                    if (localStore->isValidPath(*path))
+                    if (localStore->isValidPath(path))
                         printInfo("copying output ‘%1%’ of ‘%2%’ from local store",
-                            localStore->printStorePath(*path),
+                            localStore->printStorePath(path),
                             localStore->printStorePath(drvPath));
                     else {
                         printInfo("substituting output ‘%1%’ of ‘%2%’",
-                            localStore->printStorePath(*path),
+                            localStore->printStorePath(path),
                             localStore->printStorePath(drvPath));
-                        localStore->ensurePath(*path);
+                        localStore->ensurePath(path);
                         // FIXME: should copy directly from substituter to destStore.
                     }
 
                     copyClosure(*localStore, *destStore,
-                        StorePathSet { *path },
+                        StorePathSet { path },
                         NoRepair, CheckSigs, NoSubstitute);
 
                     time_t stopTime = time(0);
@@ -535,13 +552,13 @@ Step::ptr State::createStep(ref<Store> destStore,
                     {
                         auto mc = startDbUpdate();
                         pqxx::work txn(conn);
-                        createSubstitutionStep(txn, startTime, stopTime, build, drvPath, "out", *path);
+                        createSubstitutionStep(txn, startTime, stopTime, build, drvPath, *(step->drv), "out", path);
                         txn.commit();
                     }
 
                 } catch (Error & e) {
                     printError("while copying/substituting output ‘%s’ of ‘%s’: %s",
-                        localStore->printStorePath(*path),
+                        localStore->printStorePath(path),
                         localStore->printStorePath(drvPath),
                         e.what());
                     valid = false;
@@ -561,7 +578,7 @@ Step::ptr State::createStep(ref<Store> destStore,
     printMsg(lvlDebug, "creating build step ‘%1%’", localStore->printStorePath(drvPath));
 
     /* Create steps for the dependencies. */
-    for (auto & i : step->drv->inputDrvs) {
+    for (auto & i : step->drv->inputDrvs.map) {
         auto dep = createStep(destStore, conn, build, i.first, 0, step, finishedDrvs, newSteps, newRunnable);
         if (dep) {
             auto step_(step->state.lock());
@@ -640,17 +657,19 @@ void State::processJobsetSharesChange(Connection & conn)
 }
 
 
-BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store> destStore, const nix::Derivation & drv)
+BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store> destStore, const nix::StorePath & drvPath)
 {
+    auto derivationOutputs = destStore->queryDerivationOutputMap(drvPath, &*localStore);
+
     {
     pqxx::work txn(conn);
 
-    for (auto & [name, output] : drv.outputsAndOptPaths(*localStore)) {
+    for (auto & [name, output] : derivationOutputs) {
         auto r = txn.exec_params
             ("select id, buildStatus, releaseName, closureSize, size from Builds b "
              "join BuildOutputs o on b.id = o.build "
              "where finished = 1 and (buildStatus = 0 or buildStatus = 6) and path = $1",
-             localStore->printStorePath(*output.second));
+             localStore->printStorePath(output));
         if (r.empty()) continue;
         BuildID id = r[0][0].as<BuildID>();
 
@@ -704,5 +723,5 @@ BuildOutput State::getBuildOutputCached(Connection & conn, nix::ref<nix::Store>
     }
 
     NarMemberDatas narMembers;
-    return getBuildOutput(destStore, narMembers, drv);
+    return getBuildOutput(destStore, narMembers, derivationOutputs);
 }

src/hydra-queue-runner/state.hh

@@ -21,6 +21,8 @@
 #include "store-api.hh"
 #include "sync.hh"
 #include "nar-extractor.hh"
+#include "serve-protocol.hh"
+#include "machines.hh"
 
 
 typedef unsigned int BuildID;
@@ -78,6 +80,8 @@ struct RemoteResult
     {
         return stepStatus == bsCachedFailure ? bsFailed : stepStatus;
     }
+
+    void updateWithBuildResult(const nix::BuildResult &);
 };
 
 
@@ -231,17 +235,21 @@ void getDependents(Step::ptr step, std::set<Build::ptr> & builds, std::set<Step:
 void visitDependencies(std::function<void(Step::ptr)> visitor, Step::ptr step);
 
 
-struct Machine
+struct Machine : nix::Machine
 {
     typedef std::shared_ptr<Machine> ptr;
 
-    bool enabled{true};
+    /* TODO Get rid of: `nix::Machine::storeUri` is normalized in a way
+       we are not yet used to, but once we are, we don't need this. */
+    std::string sshName;
 
-    std::string sshName, sshKey;
-    std::set<std::string> systemTypes, supportedFeatures, mandatoryFeatures;
-    unsigned int maxJobs = 1;
-    float speedFactor = 1.0;
-    std::string sshPublicHostKey;
+    /* TODO Get rid once `nix::Machine::systemTypes` is a set not
+       vector. */
+    std::set<std::string> systemTypesSet;
+
+    /* TODO Get rid once `nix::Machine::systemTypes` is a `float` not
+       an `int`. */
+    float speedFactorFloat = 1.0;
 
     struct State {
         typedef std::shared_ptr<State> ptr;
@@ -269,7 +277,7 @@ struct Machine
     {
         /* Check that this machine is of the type required by the
            step. */
-        if (!systemTypes.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform))
+        if (!systemTypesSet.count(step->drv->platform == "builtin" ? nix::settings.thisSystem : step->drv->platform))
             return false;
 
         /* Check that the step requires all mandatory features of this
@@ -297,6 +305,32 @@ struct Machine
         std::regex r("^(ssh://|ssh-ng://)?localhost$");
         return std::regex_search(sshName, r);
     }
+
+    // A connection to a machine
+    struct Connection {
+        nix::FdSource from;
+        nix::FdSink to;
+        nix::ServeProto::Version remoteVersion;
+
+        // Backpointer to the machine
+        ptr machine;
+
+        operator nix::ServeProto::ReadConn ()
+        {
+            return {
+                .from = from,
+                .version = remoteVersion,
+            };
+        }
+
+        operator nix::ServeProto::WriteConn ()
+        {
+            return {
+                .to = to,
+                .version = remoteVersion,
+            };
+        }
+    };
 };
 
 
@@ -459,6 +493,12 @@ private:
 public:
     State(std::optional<std::string> metricsAddrOpt);
 
+    struct BuildOptions {
+        unsigned int maxSilentTime, buildTimeout, repeats;
+        size_t maxLogSize;
+        bool enforceDeterminism;
+    };
+
 private:
 
     nix::MaintainCount<counter> startDbUpdate();
@@ -485,7 +525,7 @@ private:
         const std::string & machine);
 
     int createSubstitutionStep(pqxx::work & txn, time_t startTime, time_t stopTime,
-        Build::ptr build, const nix::StorePath & drvPath, const std::string & outputName, const nix::StorePath & storePath);
+        Build::ptr build, const nix::StorePath & drvPath, const nix::Derivation drv, const std::string & outputName, const nix::StorePath & storePath);
 
     void updateBuild(pqxx::work & txn, Build::ptr build, BuildStatus status);
 
@@ -501,7 +541,7 @@ private:
     void processQueueChange(Connection & conn);
 
     BuildOutput getBuildOutputCached(Connection & conn, nix::ref<nix::Store> destStore,
-        const nix::Derivation & drv);
+        const nix::StorePath & drvPath);
 
     Step::ptr createStep(nix::ref<nix::Store> store,
         Connection & conn, Build::ptr build, const nix::StorePath & drvPath,
@@ -543,8 +583,7 @@ private:
 
     void buildRemote(nix::ref<nix::Store> destStore,
         Machine::ptr machine, Step::ptr step,
-        unsigned int maxSilentTime, unsigned int buildTimeout,
-        unsigned int repeats,
+        const BuildOptions & buildOptions,
         RemoteResult & result, std::shared_ptr<ActiveStep> activeStep,
         std::function<void(StepState)> updateStep,
         NarMemberDatas & narMembers);

src/lib/Hydra/Controller/Build.pm

@@ -78,14 +78,16 @@ sub build_GET {
 
     $c->stash->{template} = 'build.tt';
     $c->stash->{isLocalStore} = isLocalStore();
+    # XXX: If the derivation is content-addressed then this will always return
+    # false because `$_->path` will be empty
     $c->stash->{available} =
         $c->stash->{isLocalStore}
-        ? all { isValidPath($_->path) } $build->buildoutputs->all
+        ? all { $_->path && isValidPath($_->path) } $build->buildoutputs->all
         : 1;
     $c->stash->{drvAvailable} = isValidPath $build->drvpath;
 
     if ($build->finished && $build->iscachedbuild) {
-        my $path = ($build->buildoutputs)[0]->path or die;
+        my $path = ($build->buildoutputs)[0]->path or undef;
         my $cachedBuildStep = findBuildStepByOutPath($self, $c, $path);
         if (defined $cachedBuildStep) {
             $c->stash->{cachedBuild} = $cachedBuildStep->build;

src/lib/Hydra/Controller/Root.pm

@@ -18,6 +18,8 @@ use Net::Prometheus;
 use Types::Standard qw/StrMatch/;
 
 use constant NARINFO_REGEX => qr{^([a-z0-9]{32})\.narinfo$};
+# e.g.: https://hydra.example.com/realisations/sha256:a62128132508a3a32eef651d6467695944763602f226ac630543e947d9feb140!out.doi
+use constant REALISATIONS_REGEX => qr{^(sha256:[a-z0-9]{64}![a-z]+)\.doi$};
 
 # Put this controller at top-level.
 __PACKAGE__->config->{namespace} = '';
@@ -355,6 +357,33 @@ sub nix_cache_info :Path('nix-cache-info') :Args(0) {
 }
 
 
+sub realisations :Path('realisations') :Args(StrMatch[REALISATIONS_REGEX]) {
+    my ($self, $c, $realisation) = @_;
+
+    if (!isLocalStore) {
+        notFound($c, "There is no binary cache here.");
+    }
+
+    else {
+        my ($rawDrvOutput) = $realisation =~ REALISATIONS_REGEX;
+        my $rawRealisation = queryRawRealisation($rawDrvOutput);
+
+        if (!$rawRealisation) {
+            $c->response->status(404);
+            $c->response->content_type('text/plain');
+            $c->stash->{plain}->{data} = "does not exist\n";
+            $c->forward('Hydra::View::Plain');
+            setCacheHeaders($c, 60 * 60);
+            return;
+        }
+
+        $c->response->content_type('text/plain');
+        $c->stash->{plain}->{data} = $rawRealisation;
+        $c->forward('Hydra::View::Plain');
+    }
+}
+
+
 sub narinfo :Path :Args(StrMatch[NARINFO_REGEX]) {
     my ($self, $c, $narinfo) = @_;
 

src/lib/Hydra/Schema/Result/BuildOutputs.pm

@@ -49,7 +49,7 @@ __PACKAGE__->table("buildoutputs");
 =head2 path
 
   data_type: 'text'
-  is_nullable: 0
+  is_nullable: 1
 
 =cut
 
@@ -59,7 +59,7 @@ __PACKAGE__->add_columns(
   "name",
   { data_type => "text", is_nullable => 0 },
   "path",
-  { data_type => "text", is_nullable => 0 },
+  { data_type => "text", is_nullable => 1 },
 );
 
 =head1 PRIMARY KEY
@@ -94,8 +94,8 @@ __PACKAGE__->belongs_to(
 );
 
 
-# Created by DBIx::Class::Schema::Loader v0.07049 @ 2021-08-26 12:02:36
-# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:gU+kZ6A0ISKpaXGRGve8mg
+# Created by DBIx::Class::Schema::Loader v0.07049 @ 2022-06-30 12:02:32
+# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:Jsabm3YTcI7YvCuNdKP5Ng
 
 my %hint = (
     columns => [

src/lib/Hydra/Schema/Result/BuildStepOutputs.pm

@@ -55,7 +55,7 @@ __PACKAGE__->table("buildstepoutputs");
 =head2 path
 
   data_type: 'text'
-  is_nullable: 0
+  is_nullable: 1
 
 =cut
 
@@ -67,7 +67,7 @@ __PACKAGE__->add_columns(
   "name",
   { data_type => "text", is_nullable => 0 },
   "path",
-  { data_type => "text", is_nullable => 0 },
+  { data_type => "text", is_nullable => 1 },
 );
 
 =head1 PRIMARY KEY
@@ -119,8 +119,8 @@ __PACKAGE__->belongs_to(
 );
 
 
-# Created by DBIx::Class::Schema::Loader v0.07049 @ 2021-08-26 12:02:36
-# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:gxp8rOjpRVen4YbIjomHTw
+# Created by DBIx::Class::Schema::Loader v0.07049 @ 2022-06-30 12:02:32
+# DO NOT MODIFY THIS OR ANYTHING ABOVE! md5sum:Bad70CRTt7zb2GGuRoQ++Q
 
 
 # You can replace this text with custom code or comments, and it will be preserved on regeneration

src/libhydra/db.hh

@@ -2,6 +2,7 @@
 
 #include <pqxx/pqxx>
 
+#include "environment-variables.hh"
 #include "util.hh"
 
 

src/libhydra/hydra-config.hh

@@ -2,6 +2,7 @@
 
 #include <map>
 
+#include "file-system.hh"
 #include "util.hh"
 
 struct HydraConfig

src/root/topbar.tt

@@ -134,7 +134,7 @@
     [% WRAPPER makeSubMenu title="Sign in" id="sign-in-menu" align="right" %]
       [% IF c.config.enable_google_login %]
         <script src="https://accounts.google.com/gsi/client" async defer></script>
-        <div id="g_id_onload" data-client_id="[% c.config.google_client_id %]" data-callback="onGoogleSignIn">
+        <div id="g_id_onload" data-client_id="[% c.config.google_client_id %]" data-auto_prompt="false" data-callback="onGoogleSignIn">
         </div>
         <div class="g_id_signin" data-type="standard"></div>
         <div class="dropdown-divider"></div>

src/script/hydra-eval-jobset

@@ -438,13 +438,17 @@ sub checkBuild {
         # new build to be scheduled if the meta.maintainers field is
         # changed?
         if (defined $prevEval) {
+            my $pathOrDrvConstraint = defined $firstOutputPath
+                ? { path => $firstOutputPath }
+                : { drvPath => $drvPath };
+
             my ($prevBuild) = $prevEval->builds->search(
                 # The "project" and "jobset" constraints are
                 # semantically unnecessary (because they're implied by
                 # the eval), but they give a factor 1000 speedup on
                 # the Nixpkgs jobset with PostgreSQL.
                 { jobset_id => $jobset->get_column('id'), job => $jobName,
-                  name => $firstOutputName, path => $firstOutputPath },
+                  name => $firstOutputName, %$pathOrDrvConstraint },
                 { rows => 1, columns => ['id', 'finished'], join => ['buildoutputs'] });
             if (defined $prevBuild) {
                 #print STDERR "    already scheduled/built as build ", $prevBuild->id, "\n";

src/sql/hydra.sql

@@ -247,7 +247,7 @@ create trigger BuildBumped after update on Builds for each row
 create table BuildOutputs (
     build         integer not null,
     name          text not null,
-    path          text not null,
+    path          text,
     primary key   (build, name),
     foreign key   (build) references Builds(id) on delete cascade
 );
@@ -303,7 +303,7 @@ create table BuildStepOutputs (
     build         integer not null,
     stepnr        integer not null,
     name          text not null,
-    path          text not null,
+    path          text,
     primary key   (build, stepnr, name),
     foreign key   (build) references Builds(id) on delete cascade,
     foreign key   (build, stepnr) references BuildSteps(build, stepnr) on delete cascade

src/sql/upgrade-84.sql

@@ -0,0 +1,4 @@
+-- CA derivations do not have statically known output paths. The values
+-- are only filled in after the build runs.
+ALTER TABLE BuildStepOutputs ALTER COLUMN path DROP NOT NULL;
+ALTER TABLE BuildOutputs ALTER COLUMN path DROP NOT NULL;

t/content-addressed/basic.t

@@ -0,0 +1,61 @@
+use feature 'unicode_strings';
+use strict;
+use warnings;
+use Setup;
+
+my %ctx = test_init(
+    nix_config => qq|
+    experimental-features = ca-derivations
+    |,
+);
+
+require Hydra::Schema;
+require Hydra::Model::DB;
+
+use JSON::MaybeXS;
+
+use HTTP::Request::Common;
+use Test2::V0;
+require Catalyst::Test;
+Catalyst::Test->import('Hydra');
+
+my $db = Hydra::Model::DB->new;
+hydra_setup($db);
+
+my $project = $db->resultset('Projects')->create({name => "tests", displayname => "", owner => "root"});
+
+my $jobset = createBaseJobset("content-addressed", "content-addressed.nix", $ctx{jobsdir});
+
+ok(evalSucceeds($jobset), "Evaluating jobs/content-addressed.nix should exit with return code 0");
+is(nrQueuedBuildsForJobset($jobset), 5, "Evaluating jobs/content-addressed.nix should result in 4 builds");
+
+for my $build (queuedBuildsForJobset($jobset)) {
+    ok(runBuild($build), "Build '".$build->job."' from jobs/content-addressed.nix should exit with code 0");
+    my $newbuild = $db->resultset('Builds')->find($build->id);
+    is($newbuild->finished, 1, "Build '".$build->job."' from jobs/content-addressed.nix should be finished.");
+    my $expected = $build->job eq "fails" ? 1 : $build->job =~ /with_failed/ ? 6 : 0;
+    is($newbuild->buildstatus, $expected, "Build '".$build->job."' from jobs/content-addressed.nix should have buildstatus $expected.");
+
+    my $response = request("/build/".$build->id);
+    ok($response->is_success, "The 'build' page for build '".$build->job."' should load properly");
+
+    if ($newbuild->buildstatus == 0) {
+      my $buildOutputs = $newbuild->buildoutputs;
+      for my $output ($newbuild->buildoutputs) {
+        # XXX: This hardcodes /nix/store/.
+        # It's fine because in practice the nix store for the tests will be of
+        # the form `/some/thing/nix/store/`, but it would be cleaner if there
+        # was a way to query Nix for its store dir?
+        like(
+          $output->path, qr|/nix/store/|,
+          "Output '".$output->name."' of build '".$build->job."' should be a valid store path"
+        );
+      }
+    }
+
+}
+
+isnt(<$ctx{deststoredir}/realisations/*>, "", "The destination store should have the realisations of the built derivations registered");
+
+done_testing;
+

t/content-addressed/without-experimental-feature.t

@@ -0,0 +1,28 @@
+use feature 'unicode_strings';
+use strict;
+use warnings;
+use Setup;
+
+my %ctx = test_init();
+
+require Hydra::Schema;
+require Hydra::Model::DB;
+
+use JSON::MaybeXS;
+
+use HTTP::Request::Common;
+use Test2::V0;
+require Catalyst::Test;
+Catalyst::Test->import('Hydra');
+
+my $db = Hydra::Model::DB->new;
+hydra_setup($db);
+
+my $project = $db->resultset('Projects')->create({name => "tests", displayname => "", owner => "root"});
+
+my $jobset = createBaseJobset("content-addressed", "content-addressed.nix", $ctx{jobsdir});
+
+ok(evalSucceeds($jobset),                  "Evaluating jobs/content-addressed.nix without the experimental feature should exit with return code 0");
+is(nrQueuedBuildsForJobset($jobset), 0, "Evaluating jobs/content-addressed.nix without the experimental Nix feature should result in 0 build");
+
+done_testing;

t/jobs/config.nix.in

@@ -6,4 +6,9 @@ rec {
       system = builtins.currentSystem;
       PATH = path;
     } // args);
+  mkContentAddressedDerivation = args: mkDerivation ({
+    __contentAddressed = true;
+    outputHashMode = "recursive";
+    outputHashAlgo = "sha256";
+  } // args);
 }

t/jobs/content-addressed.nix

@@ -0,0 +1,35 @@
+let cfg =  import ./config.nix; in
+rec {
+  empty_dir =
+    cfg.mkContentAddressedDerivation {
+      name = "empty-dir";
+      builder = ./empty-dir-builder.sh;
+    };
+
+  fails =
+    cfg.mkContentAddressedDerivation {
+      name = "fails";
+      builder = ./fail.sh;
+    };
+
+  succeed_with_failed =
+    cfg.mkContentAddressedDerivation {
+      name = "succeed-with-failed";
+      builder = ./succeed-with-failed.sh;
+    };
+
+  caDependingOnCA =
+    cfg.mkContentAddressedDerivation {
+      name = "ca-depending-on-ca";
+      builder = ./dir-with-file-builder.sh;
+      FOO = empty_dir;
+    };
+
+  nonCaDependingOnCA =
+    cfg.mkDerivation {
+      name = "non-ca-depending-on-ca";
+      builder = ./dir-with-file-builder.sh;
+      FOO = empty_dir;
+    };
+}
+

t/jobs/dir-with-file-builder.sh

@@ -0,0 +1,4 @@
+#! /bin/sh
+
+mkdir $out
+echo foo > $out/a-file

t/jobs/empty-dir-builder.sh

@@ -1,6 +1,3 @@
 #! /bin/sh
 
-# Workaround for https://github.com/NixOS/nix/pull/6051
-echo "some output"
-
 mkdir $out

t/lib/HydraTestContext.pm

@@ -39,7 +39,11 @@ use Hydra::Helper::Exec;
 sub new {
     my ($class, %opts) = @_;
 
-    my $dir = File::Temp->newdir();
+    my $deststoredir;
+
+    # Cleanup will be managed by yath. By the default it will be cleaned
+    # up, but can be kept to aid in debugging test failures.
+    my $dir = File::Temp->newdir(CLEANUP => 0);
 
     $ENV{'HYDRA_DATA'} = "$dir/hydra-data";
     mkdir $ENV{'HYDRA_DATA'};
@@ -53,6 +57,7 @@ sub new {
     my $hydra_config = $opts{'hydra_config'} || "";
     $hydra_config = "queue_runner_metrics_address = 127.0.0.1:0\n" . $hydra_config;
     if ($opts{'use_external_destination_store'} // 1) {
+        $deststoredir = "$dir/nix/dest-store";
         $hydra_config = "store_uri = file://$dir/nix/dest-store\n" . $hydra_config;
     }
 
@@ -79,7 +84,8 @@ sub new {
         nix_state_dir => $nix_state_dir,
         nix_log_dir => $nix_log_dir,
         testdir => abs_path(dirname(__FILE__) . "/.."),
-        jobsdir => abs_path(dirname(__FILE__) . "/../jobs")
+        jobsdir => abs_path(dirname(__FILE__) . "/../jobs"),
+        deststoredir => $deststoredir,
     }, $class;
 
     if ($opts{'before_init'}) {

t/queue-runner/notifications.t

@@ -8,7 +8,7 @@ my $binarycachedir = File::Temp->newdir();
 
 my $ctx = test_context(
     nix_config => qq|
-    experimental-features = nix-command
+    experimental-features = nix-command ca-derivations
     substituters = file://${binarycachedir}?trusted=1
     |,
     hydra_config => q|