Merge pull request #4477 from NixOS/ca/build-remote

Build ca derivations remotely

src/build-remote/build-remote.cc

@@ -251,7 +251,7 @@ connected:
         std::cerr << "# accept\n" << storeUri << "\n";
 
         auto inputs = readStrings<PathSet>(source);
-        auto outputs = readStrings<PathSet>(source);
+        auto wantedOutputs = readStrings<StringSet>(source);
 
         AutoCloseFD uploadLock = openLockFile(currentLoad + "/" + escapeUri(storeUri) + ".upload-lock", true);
 
@@ -276,6 +276,7 @@ connected:
         uploadLock = -1;
 
         auto drv = store->readDerivation(*drvPath);
+        auto outputHashes = staticOutputHashes(*store, drv);
         drv.inputSrcs = store->parseStorePathSet(inputs);
 
         auto result = sshStore->buildDerivation(*drvPath, drv);
@@ -283,16 +284,42 @@ connected:
         if (!result.success())
             throw Error("build of '%s' on '%s' failed: %s", store->printStorePath(*drvPath), storeUri, result.errorMsg);
 
-        StorePathSet missing;
-        for (auto & path : outputs)
-            if (!store->isValidPath(store->parseStorePath(path))) missing.insert(store->parseStorePath(path));
+        std::set<Realisation> missingRealisations;
+        StorePathSet missingPaths;
+        if (settings.isExperimentalFeatureEnabled("ca-derivations") && !derivationHasKnownOutputPaths(drv.type())) {
+            for (auto & outputName : wantedOutputs) {
+                auto thisOutputHash = outputHashes.at(outputName);
+                auto thisOutputId = DrvOutput{ thisOutputHash, outputName };
+                if (!store->queryRealisation(thisOutputId)) {
+                    debug("missing output %s", outputName);
+                    assert(result.builtOutputs.count(thisOutputId));
+                    auto newRealisation = result.builtOutputs.at(thisOutputId);
+                    missingRealisations.insert(newRealisation);
+                    missingPaths.insert(newRealisation.outPath);
+                }
+            }
+        } else {
+            auto outputPaths = drv.outputsAndOptPaths(*store);
+            for (auto & [outputName, hopefullyOutputPath] : outputPaths) {
+                assert(hopefullyOutputPath.second);
+                if (!store->isValidPath(*hopefullyOutputPath.second))
+                    missingPaths.insert(*hopefullyOutputPath.second);
+            }
+        }
 
-        if (!missing.empty()) {
+        if (!missingPaths.empty()) {
             Activity act(*logger, lvlTalkative, actUnknown, fmt("copying outputs from '%s'", storeUri));
             if (auto localStore = store.dynamic_pointer_cast<LocalStore>())
-                for (auto & i : missing)
-                    localStore->locksHeld.insert(store->printStorePath(i)); /* FIXME: ugly */
-            copyPaths(ref<Store>(sshStore), store, missing, NoRepair, NoCheckSigs, NoSubstitute);
+                for (auto & path : missingPaths)
+                    localStore->locksHeld.insert(store->printStorePath(path)); /* FIXME: ugly */
+            copyPaths(ref<Store>(sshStore), store, missingPaths, NoRepair, NoCheckSigs, NoSubstitute);
+        }
+        // XXX: Should be done as part of `copyPaths`
+        for (auto & realisation : missingRealisations) {
+            // Should hold, because if the feature isn't enabled the set
+            // of missing realisations should be empty
+            settings.requireExperimentalFeature("ca-derivations");
+            store->registerDrvOutput(realisation);
         }
 
         return 0;

src/libstore/build/derivation-goal.cc

@@ -1147,13 +1147,13 @@ HookReply DerivationGoal::tryBuildHook()
     /* Tell the hooks the missing outputs that have to be copied back
        from the remote system. */
     {
-        StorePathSet missingPaths;
-        for (auto & [_, status] : initialOutputs) {
-            if (!status.known) continue;
-            if (buildMode != bmCheck && status.known->isValid()) continue;
-            missingPaths.insert(status.known->path);
+        StringSet missingOutputs;
+        for (auto & [outputName, status] : initialOutputs) {
+            // XXX: Does this include known CA outputs?
+            if (buildMode != bmCheck && status.known && status.known->isValid()) continue;
+            missingOutputs.insert(outputName);
         }
-        worker_proto::write(worker.store, hook->sink, missingPaths);
+        worker_proto::write(worker.store, hook->sink, missingOutputs);
     }
 
     hook->sink = FdSink();
@@ -2988,11 +2988,11 @@ void DerivationGoal::registerOutputs()
      */
     if (hook) {
         bool allValid = true;
-        for (auto & i : drv->outputsAndOptPaths(worker.store)) {
-            if (!i.second.second || !worker.store.isValidPath(*i.second.second))
+        for (auto & [outputName, outputPath] : worker.store.queryPartialDerivationOutputMap(drvPath)) {
+            if (!outputPath || !worker.store.isValidPath(*outputPath))
                 allValid = false;
             else
-                finalOutputs.insert_or_assign(i.first, *i.second.second);
+                finalOutputs.insert_or_assign(outputName, *outputPath);
         }
         if (allValid) return;
     }

src/libstore/build/entry-points.cc

@@ -58,6 +58,26 @@ BuildResult Store::buildDerivation(const StorePath & drvPath, const BasicDerivat
         result.status = BuildResult::MiscFailure;
         result.errorMsg = e.msg();
     }
+    // XXX: Should use `goal->queryPartialDerivationOutputMap()` once it's
+    // extended to return the full realisation for each output
+    auto staticDrvOutputs = drv.outputsAndOptPaths(*this);
+    auto outputHashes = staticOutputHashes(*this, drv);
+    for (auto & [outputName, staticOutput] : staticDrvOutputs) {
+        auto outputId = DrvOutput{outputHashes.at(outputName), outputName};
+        if (staticOutput.second)
+            result.builtOutputs.insert_or_assign(
+                    outputId,
+                    Realisation{ outputId, *staticOutput.second}
+                    );
+        if (settings.isExperimentalFeatureEnabled("ca-derivations") && !derivationHasKnownOutputPaths(drv.type())) {
+            auto realisation = this->queryRealisation(outputId);
+            if (realisation)
+                result.builtOutputs.insert_or_assign(
+                        outputId,
+                        *realisation
+                        );
+        }
+    }
 
     return result;
 }

src/libstore/daemon.cc

@@ -575,6 +575,9 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         auto res = store->buildDerivation(drvPath, drv, buildMode);
         logger->stopWork();
         to << res.status << res.errorMsg;
+        if (GET_PROTOCOL_MINOR(clientVersion) >= 0xc) {
+            worker_proto::write(*store, to, res.builtOutputs);
+        }
         break;
     }
 

src/libstore/derivations.cc

@@ -57,6 +57,17 @@ bool derivationIsFixed(DerivationType dt) {
     assert(false);
 }
 
+bool derivationHasKnownOutputPaths(DerivationType dt) {
+    switch (dt) {
+    case DerivationType::InputAddressed: return true;
+    case DerivationType::CAFixed: return true;
+    case DerivationType::CAFloating: return false;
+    case DerivationType::DeferredInputAddressed: return false;
+    };
+    assert(false);
+}
+
+
 bool derivationIsImpure(DerivationType dt) {
     switch (dt) {
     case DerivationType::InputAddressed: return false;

src/libstore/derivations.hh

@@ -94,6 +94,11 @@ bool derivationIsFixed(DerivationType);
    derivation is controlled separately. Never true for non-CA derivations. */
 bool derivationIsImpure(DerivationType);
 
+/* Does the derivation knows its own output paths?
+ * Only true when there's no floating-ca derivation involved in the closure.
+ */
+bool derivationHasKnownOutputPaths(DerivationType);
+
 struct BasicDerivation
 {
     DerivationOutputs outputs; /* keyed on symbolic IDs */

src/libstore/legacy-ssh-store.cc

@@ -258,7 +258,9 @@ public:
 
         if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 3)
             conn->from >> status.timesBuilt >> status.isNonDeterministic >> status.startTime >> status.stopTime;
-
+        if (GET_PROTOCOL_MINOR(conn->remoteVersion) >= 6) {
+            status.builtOutputs = worker_proto::read(*this, conn->from, Phantom<DrvOutputs> {});
+        }
         return status;
     }
 

src/libstore/realisation.hh

@@ -33,6 +33,8 @@ struct Realisation {
     GENERATE_CMP(Realisation, me->id, me->outPath);
 };
 
+typedef std::map<DrvOutput, Realisation> DrvOutputs;
+
 struct OpaquePath {
     StorePath path;
 

src/libstore/remote-store.cc

@@ -12,6 +12,7 @@
 #include "logging.hh"
 #include "callback.hh"
 #include "filetransfer.hh"
+#include <nlohmann/json.hpp>
 
 namespace nix {
 
@@ -49,6 +50,21 @@ void write(const Store & store, Sink & out, const ContentAddress & ca)
     out << renderContentAddress(ca);
 }
 
+Realisation read(const Store & store, Source & from, Phantom<Realisation> _)
+{
+    std::string rawInput = readString(from);
+    return Realisation::fromJSON(
+        nlohmann::json::parse(rawInput),
+        "remote-protocol"
+    );
+}
+void write(const Store & store, Sink & out, const Realisation & realisation)
+{ out << realisation.toJSON().dump(); }
+
+DrvOutput read(const Store & store, Source & from, Phantom<DrvOutput> _)
+{ return DrvOutput::parse(readString(from)); }
+void write(const Store & store, Sink & out, const DrvOutput & drvOutput)
+{ out << drvOutput.to_string(); }
 
 std::optional<StorePath> read(const Store & store, Source & from, Phantom<std::optional<StorePath>> _)
 {
@@ -664,6 +680,10 @@ BuildResult RemoteStore::buildDerivation(const StorePath & drvPath, const BasicD
     unsigned int status;
     conn->from >> status >> res.errorMsg;
     res.status = (BuildResult::Status) status;
+    if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 0xc) {
+        auto builtOutputs = worker_proto::read(*this, conn->from, Phantom<DrvOutputs> {});
+        res.builtOutputs = builtOutputs;
+    }
     return res;
 }
 

src/libstore/serve-protocol.hh

@@ -5,7 +5,7 @@ namespace nix {
 #define SERVE_MAGIC_1 0x390c9deb
 #define SERVE_MAGIC_2 0x5452eecb
 
-#define SERVE_PROTOCOL_VERSION 0x205
+#define SERVE_PROTOCOL_VERSION 0x206
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)
 

src/libstore/store-api.hh

@@ -162,6 +162,8 @@ struct BuildResult
        non-determinism.) */
     bool isNonDeterministic = false;
 
+    DrvOutputs builtOutputs;
+
     /* The start/stop times of the build (or one of the rounds, if it
        was repeated). */
     time_t startTime = 0, stopTime = 0;

src/libstore/worker-protocol.hh

@@ -9,7 +9,7 @@ namespace nix {
 #define WORKER_MAGIC_1 0x6e697863
 #define WORKER_MAGIC_2 0x6478696f
 
-#define PROTOCOL_VERSION 0x11b
+#define PROTOCOL_VERSION 0x11c
 #define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
 #define GET_PROTOCOL_MINOR(x) ((x) & 0x00ff)
 
@@ -86,6 +86,8 @@ namespace worker_proto {
 MAKE_WORKER_PROTO(, std::string);
 MAKE_WORKER_PROTO(, StorePath);
 MAKE_WORKER_PROTO(, ContentAddress);
+MAKE_WORKER_PROTO(, Realisation);
+MAKE_WORKER_PROTO(, DrvOutput);
 
 MAKE_WORKER_PROTO(template<typename T>, std::set<T>);
 

src/nix-store/nix-store.cc

@@ -905,6 +905,10 @@ static void opServe(Strings opFlags, Strings opArgs)
 
                 if (GET_PROTOCOL_MINOR(clientVersion) >= 3)
                     out << status.timesBuilt << status.isNonDeterministic << status.startTime << status.stopTime;
+                if (GET_PROTOCOL_MINOR(clientVersion >= 5)) {
+                    worker_proto::write(*store, out, status.builtOutputs);
+                }
+
 
                 break;
             }

tests/build-hook-ca.nix

@@ -11,6 +11,7 @@ let
       args = ["sh" "-e" args.builder or (builtins.toFile "builder-${args.name}.sh" "if [ -e .attrs.sh ]; then source .attrs.sh; fi; eval \"$buildCommand\"")];
       outputHashMode = "recursive";
       outputHashAlgo = "sha256";
+      __contentAddressed = true;
     } // removeAttrs args ["builder" "meta"])
     // { meta = args.meta or {}; };
 
@@ -19,7 +20,6 @@ let
     name = "build-remote-input-1";
     buildCommand = "echo FOO > $out";
     requiredSystemFeatures = ["foo"];
-    outputHash = "sha256-FePFYIlMuycIXPZbWi7LGEiMmZSX9FMbaQenWBzm1Sc=";
   };
 
   input2 = mkDerivation {
@@ -27,7 +27,16 @@ let
     name = "build-remote-input-2";
     buildCommand = "echo BAR > $out";
     requiredSystemFeatures = ["bar"];
-    outputHash = "sha256-XArauVH91AVwP9hBBQNlkX9ccuPpSYx9o0zeIHb6e+Q=";
+  };
+
+  input3 = mkDerivation {
+    shell = busybox;
+    name = "build-remote-input-3";
+    buildCommand = ''
+      read x < ${input2}
+      echo $x BAZ > $out
+    '';
+    requiredSystemFeatures = ["baz"];
   };
 
 in
@@ -38,8 +47,7 @@ in
     buildCommand =
       ''
         read x < ${input1}
-        read y < ${input2}
+        read y < ${input3}
         echo "$x $y" > $out
       '';
-    outputHash = "sha256-3YGhlOfbGUm9hiPn2teXXTT8M1NEpDFvfXkxMaJRld0=";
   }

tests/build-remote-content-addressed-fixed.sh

@@ -1,5 +0,0 @@
-source common.sh
-
-file=build-hook-ca.nix
-
-source build-remote.sh

tests/build-remote-content-addressed-floating.sh

@@ -0,0 +1,7 @@
+source common.sh
+
+file=build-hook-ca.nix
+
+sed -i 's/experimental-features .*/& ca-derivations/' "$NIX_CONF_DIR"/nix.conf
+
+source build-remote.sh

tests/local.mk

@@ -17,6 +17,7 @@ nix_tests = \
   linux-sandbox.sh \
   build-dry.sh \
   build-remote-input-addressed.sh \
+  build-remote-content-addressed-floating.sh \
   ssh-relay.sh \
   nar-access.sh \
   structured-attrs.sh \
@@ -42,7 +43,6 @@ nix_tests = \
   build.sh \
   compute-levels.sh
   # parallel.sh
-  # build-remote-content-addressed-fixed.sh \
 
 install-tests += $(foreach x, $(nix_tests), tests/$(x))