632 lines
25 KiB
C++
632 lines
25 KiB
C++
#include <algorithm>
|
||
#include <chrono>
|
||
#include <set>
|
||
#include <map>
|
||
#include <memory>
|
||
#include <optional>
|
||
#include <tuple>
|
||
#include <fstream>
|
||
#include <sstream>
|
||
#include <cstring>
|
||
#include <cerrno>
|
||
#include <sys/socket.h>
|
||
#include <netdb.h>
|
||
#include <netinet/in.h>
|
||
#include <poll.h>
|
||
#include <fcntl.h>
|
||
#include <unistd.h>
|
||
#include <nlohmann/json.hpp>
|
||
#if __APPLE__
|
||
#include <sys/time.h>
|
||
#endif
|
||
|
||
#include "lix/libstore/machines.hh"
|
||
#include "lix/libmain/shared.hh"
|
||
#include "lix/libstore/pathlocks.hh"
|
||
#include "lix/libstore/globals.hh"
|
||
#include "lix/libutil/serialise.hh"
|
||
#include "lix/libstore/build-result.hh"
|
||
#include "lix/libstore/store-api.hh"
|
||
#include "lix/libstore/derivations.hh"
|
||
#include "lix/libutil/strings.hh"
|
||
#include "lix/libstore/local-store.hh"
|
||
#include "lix/libcmd/legacy.hh"
|
||
#include "lix/libutil/experimental-features.hh"
|
||
#include "lix/libutil/hash.hh"
|
||
#include "build-remote.hh"
|
||
|
||
namespace nix {
|
||
|
||
std::string escapeUri(std::string uri)
|
||
{
|
||
std::replace(uri.begin(), uri.end(), '/', '_');
|
||
return uri;
|
||
}
|
||
|
||
static std::string currentLoad;
|
||
|
||
static std::string makeLockFilename(const std::string & storeUri) {
|
||
// We include 48 bytes of escaped URI to give an idea of what the lock
|
||
// is on, then 16 bytes of hash to disambiguate.
|
||
// This avoids issues with the escaped URI being very long and causing
|
||
// path too long errors, while also avoiding any possibility of collision
|
||
// caused by simple truncation.
|
||
auto hash = hashString(HashType::SHA256, storeUri).to_string(Base::Base32, false);
|
||
return escapeUri(storeUri).substr(0, 48) + "-" + hash.substr(0, 16);
|
||
}
|
||
|
||
static AutoCloseFD openSlotLock(const Machine & m, uint64_t slot)
|
||
{
|
||
return openLockFile(fmt("%s/%s-%d", currentLoad, makeLockFilename(m.storeUri), slot), true);
|
||
}
|
||
|
||
static bool allSupportedLocally(Store & store, const std::set<std::string>& requiredFeatures) {
|
||
for (auto & feature : requiredFeatures)
|
||
if (!store.config().systemFeatures.get().count(feature)) return false;
|
||
return true;
|
||
}
|
||
|
||
/* --------------------------------------------------------------------------
|
||
* P1: load- and memory-aware adaptive remote-build selection.
|
||
*
|
||
* All state below is populated ONCE from out-of-band, env-driven config
|
||
* (never from the derivation). Every helper FAILS OPEN: if config is unset,
|
||
* the metrics socket is unreachable/slow/malformed, or the storeUri is
|
||
* unknown, the helpers behave exactly like unpatched Lix
|
||
* (machineHasRoom -> true, liveLoadPenalty -> 0).
|
||
* ------------------------------------------------------------------------ */
|
||
|
||
// A drv that does not match the heavy-crate table is treated as "light":
|
||
// we have NO confident signal that it is memory-heavy, so adaptiveEstPeakRSS
|
||
// returns nullopt and machineHasRoom never filters on account of it. (There is
|
||
// deliberately no numeric light default - an unmatched drv must always permit,
|
||
// so keying it on a free-RAM threshold would wrongly filter light drvs.)
|
||
|
||
// name-substring -> estimated peak RSS in MiB (from LIX_ADAPTIVE_RSS_TABLE).
|
||
static std::map<std::string, uint64_t> adaptiveRssTable;
|
||
// machine storeUri -> "host:port" metrics endpoint (from LIX_ADAPTIVE_METRICS_MAP).
|
||
static std::map<std::string, std::string> adaptiveMetricsMap;
|
||
|
||
struct AdaptiveProbe {
|
||
bool ok = false;
|
||
uint64_t memAvailKb = 0;
|
||
double psiMem = 0, psiIo = 0, psiCpu = 0, load1 = 0, nproc = 0;
|
||
};
|
||
|
||
// In-process TTL cache keyed by storeUri, so selection probes each machine
|
||
// at most once every ~2s regardless of how many drvs stream through.
|
||
static std::map<std::string, std::pair<std::chrono::steady_clock::time_point, AdaptiveProbe>> adaptiveProbeCache;
|
||
|
||
/* Parse the two env-driven config sources once. Any error leaves the tables
|
||
* empty, which degrades to unpatched behavior. */
|
||
static void adaptiveLoadConfig()
|
||
{
|
||
// LIX_ADAPTIVE_RSS_TABLE is a PATH to a JSON object {substring: MiB}.
|
||
try {
|
||
if (auto p = getEnv("LIX_ADAPTIVE_RSS_TABLE")) {
|
||
std::ifstream f(*p);
|
||
if (f) {
|
||
nlohmann::json j;
|
||
f >> j;
|
||
if (j.is_object())
|
||
for (auto & [k, v] : j.items())
|
||
// Per-entry guard: one bad value skips only that entry,
|
||
// it does not discard the whole (otherwise valid) table.
|
||
try {
|
||
if (v.is_number_unsigned() || (v.is_number_integer() && v.get<int64_t>() >= 0))
|
||
adaptiveRssTable[k] = v.get<uint64_t>();
|
||
} catch (...) { continue; }
|
||
}
|
||
}
|
||
} catch (...) { adaptiveRssTable.clear(); }
|
||
|
||
// LIX_ADAPTIVE_METRICS_MAP is an inline JSON object {storeUri: "host:port"}.
|
||
try {
|
||
if (auto m = getEnv("LIX_ADAPTIVE_METRICS_MAP")) {
|
||
auto j = nlohmann::json::parse(*m);
|
||
if (j.is_object())
|
||
for (auto & [k, v] : j.items())
|
||
// Per-entry guard: skip one bad value, keep the rest.
|
||
try {
|
||
if (v.is_string())
|
||
adaptiveMetricsMap[k] = v.get<std::string>();
|
||
} catch (...) { continue; }
|
||
}
|
||
} catch (...) { adaptiveMetricsMap.clear(); }
|
||
}
|
||
|
||
/* Estimated peak RSS (MiB) for a drv, or nullopt when the drv does not match
|
||
* the heavy-crate table. nullopt == "no confident heavy signal". Keyed on the
|
||
* store-path NAME, which is available before readDerivation and never mutates
|
||
* the drv. */
|
||
static std::optional<uint64_t> adaptiveEstPeakRSS(const StorePath & drvPath)
|
||
{
|
||
if (adaptiveRssTable.empty()) return std::nullopt;
|
||
std::string_view name = drvPath.name();
|
||
std::optional<uint64_t> best;
|
||
for (auto & [sub, mib] : adaptiveRssTable)
|
||
if (!sub.empty() && name.find(sub) != std::string_view::npos)
|
||
best = std::max(best.value_or(0), mib);
|
||
return best;
|
||
}
|
||
|
||
/* TCP-connect the metrics endpoint and read one line:
|
||
* "MemAvail_kB psi_mem psi_io psi_cpu load1 nproc"
|
||
* A single ~500ms wall-clock deadline bounds the WHOLE probe (resolve +
|
||
* connect + read) so selection NEVER hangs, regardless of a slow or
|
||
* byte-dribbling peer. The endpoint MUST be a numeric IP:port - resolution is
|
||
* pinned to AI_NUMERICHOST|AI_NUMERICSERV so getaddrinfo never does network
|
||
* I/O (a hostname simply fails fast -> fail-open). Any failure returns an
|
||
* AdaptiveProbe with ok=false. */
|
||
static AdaptiveProbe adaptiveProbeEndpoint(const std::string & hostport)
|
||
{
|
||
AdaptiveProbe r;
|
||
auto colon = hostport.rfind(':');
|
||
if (colon == std::string::npos || colon == 0 || colon + 1 >= hostport.size())
|
||
return r;
|
||
std::string host = hostport.substr(0, colon);
|
||
std::string port = hostport.substr(colon + 1);
|
||
|
||
// Single wall-clock budget for the entire probe.
|
||
auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500);
|
||
auto remainingMs = [&]() -> int {
|
||
auto d = std::chrono::duration_cast<std::chrono::milliseconds>(
|
||
deadline - std::chrono::steady_clock::now()).count();
|
||
return d <= 0 ? 0 : (int) d;
|
||
};
|
||
|
||
struct addrinfo hints;
|
||
memset(&hints, 0, sizeof hints);
|
||
hints.ai_family = AF_UNSPEC;
|
||
hints.ai_socktype = SOCK_STREAM;
|
||
// Numeric-only: no DNS, no resolver blocking. Non-IP endpoint -> fail-open.
|
||
hints.ai_flags = AI_NUMERICHOST | AI_NUMERICSERV;
|
||
struct addrinfo * res = nullptr;
|
||
if (getaddrinfo(host.c_str(), port.c_str(), &hints, &res) != 0 || !res)
|
||
return r;
|
||
|
||
int fd = socket(res->ai_family, res->ai_socktype | SOCK_NONBLOCK, res->ai_protocol);
|
||
if (fd < 0) { freeaddrinfo(res); return r; }
|
||
|
||
int cr = connect(fd, res->ai_addr, res->ai_addrlen);
|
||
if (cr < 0 && errno == EINPROGRESS) {
|
||
struct pollfd pfd;
|
||
pfd.fd = fd;
|
||
pfd.events = POLLOUT;
|
||
if (poll(&pfd, 1, remainingMs()) <= 0) { close(fd); freeaddrinfo(res); return r; }
|
||
int soerr = 0;
|
||
socklen_t sl = sizeof soerr;
|
||
if (getsockopt(fd, SOL_SOCKET, SO_ERROR, &soerr, &sl) < 0 || soerr != 0) {
|
||
close(fd); freeaddrinfo(res); return r;
|
||
}
|
||
} else if (cr < 0) {
|
||
close(fd); freeaddrinfo(res); return r;
|
||
}
|
||
freeaddrinfo(res);
|
||
|
||
/* Read one short line. Keep the socket non-blocking and gate every recv on
|
||
poll(POLLIN) against the shared deadline, so the total read time is
|
||
bounded even if the peer drips one byte at a time. A valid reply is tiny,
|
||
so also cap the number of reads. */
|
||
std::string line;
|
||
char buf[512];
|
||
for (int iter = 0; iter < 16 && line.size() < 4096; ++iter) {
|
||
int rem = remainingMs();
|
||
if (rem == 0) break;
|
||
struct pollfd pfd;
|
||
pfd.fd = fd;
|
||
pfd.events = POLLIN;
|
||
int pr = poll(&pfd, 1, rem);
|
||
if (pr <= 0) break; // timeout or error -> fail-open
|
||
if (!(pfd.revents & POLLIN)) break; // POLLHUP/POLLERR with no data
|
||
ssize_t n = recv(fd, buf, sizeof buf, 0);
|
||
if (n < 0) {
|
||
if (errno == EAGAIN || errno == EWOULDBLOCK || errno == EINTR) continue;
|
||
break;
|
||
}
|
||
if (n == 0) break; // peer closed
|
||
line.append(buf, n);
|
||
if (line.find('\n') != std::string::npos) break;
|
||
}
|
||
close(fd);
|
||
|
||
std::istringstream ss(line);
|
||
AdaptiveProbe tmp;
|
||
if (ss >> tmp.memAvailKb >> tmp.psiMem >> tmp.psiIo >> tmp.psiCpu >> tmp.load1 >> tmp.nproc) {
|
||
tmp.ok = true;
|
||
return tmp;
|
||
}
|
||
return r;
|
||
}
|
||
|
||
/* Cached probe for a machine. Unknown storeUri -> ok=false (fail-open). */
|
||
static AdaptiveProbe adaptiveProbe(const Machine & m)
|
||
{
|
||
auto now = std::chrono::steady_clock::now();
|
||
auto it = adaptiveProbeCache.find(m.storeUri);
|
||
if (it != adaptiveProbeCache.end() && now - it->second.first < std::chrono::seconds(2))
|
||
return it->second.second;
|
||
|
||
AdaptiveProbe r;
|
||
auto mit = adaptiveMetricsMap.find(m.storeUri);
|
||
if (mit != adaptiveMetricsMap.end())
|
||
r = adaptiveProbeEndpoint(mit->second);
|
||
|
||
adaptiveProbeCache[m.storeUri] = { now, r };
|
||
return r;
|
||
}
|
||
|
||
/* OOM guard. Returns TRUE (permit as a candidate) UNLESS we have a confident
|
||
* signal that the drv is heavy AND the machine's free RAM is below the drv's
|
||
* estimated peak RSS. No env, dead socket, or unknown machine -> permit. */
|
||
static bool machineHasRoom(const Machine & m, const StorePath & drvPath)
|
||
{
|
||
auto est = adaptiveEstPeakRSS(drvPath);
|
||
if (!est) return true; // no confident heavy signal
|
||
auto p = adaptiveProbe(m);
|
||
if (!p.ok) return true; // no live signal -> fail open
|
||
uint64_t freeMib = p.memAvailKb / 1024;
|
||
return freeMib >= *est;
|
||
}
|
||
|
||
/* Extra ranking cost from live pressure on a machine; 0 when no signal. */
|
||
static double liveLoadPenalty(const Machine & m)
|
||
{
|
||
auto p = adaptiveProbe(m);
|
||
if (!p.ok) return 0.0;
|
||
double penalty = 0.0;
|
||
penalty += p.psiIo / 10.0; // io-PSI (0..100) -> up to 10
|
||
if (p.nproc > 0) penalty += p.load1 / p.nproc; // load normalized by cores
|
||
return penalty;
|
||
}
|
||
|
||
static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings argv)
|
||
{
|
||
{
|
||
logger = makeJSONLogger(*logger);
|
||
|
||
/* Ensure we don't get any SSH passphrase or host key popups. */
|
||
unsetenv("DISPLAY");
|
||
unsetenv("SSH_ASKPASS");
|
||
|
||
/* If we ever use the common args framework, make sure to
|
||
remove initPlugins below and initialize settings first.
|
||
*/
|
||
if (argv.size() != 1)
|
||
throw UsageError("called without required arguments");
|
||
|
||
verbosity = (Verbosity) std::stoll(argv.front());
|
||
|
||
FdSource source(STDIN_FILENO);
|
||
|
||
/* Read the parent's settings. */
|
||
while (readInt(source)) {
|
||
auto name = readString(source);
|
||
auto value = readString(source);
|
||
settings.set(name, value);
|
||
}
|
||
|
||
auto maxBuildJobs = settings.maxBuildJobs;
|
||
settings.maxBuildJobs.set("1"); // hack to make tests with local?root= work
|
||
|
||
initPlugins();
|
||
|
||
auto store = aio.blockOn(openStore());
|
||
|
||
/* It would be more appropriate to use $XDG_RUNTIME_DIR, since
|
||
that gets cleared on reboot, but it wouldn't work on macOS. */
|
||
auto currentLoadName = "/current-load";
|
||
if (auto localStore = store.try_cast_shared<LocalFSStore>())
|
||
currentLoad = std::string { localStore->config().stateDir } + currentLoadName;
|
||
else
|
||
currentLoad = settings.nixStateDir + currentLoadName;
|
||
|
||
std::shared_ptr<Store> sshStore;
|
||
AutoCloseFD bestSlotLock;
|
||
|
||
auto machines = getMachines();
|
||
debug("got %d remote builders", machines.size());
|
||
|
||
if (machines.empty()) {
|
||
std::cerr << "# decline-permanently\n";
|
||
return 0;
|
||
}
|
||
|
||
std::optional<StorePath> drvPath;
|
||
std::string storeUri;
|
||
|
||
/* P1: parse out-of-band adaptive config once (fail-open on any error). */
|
||
adaptiveLoadConfig();
|
||
|
||
while (true) {
|
||
|
||
try {
|
||
auto s = readString(source);
|
||
if (s != "try") return 0;
|
||
} catch (EndOfFile &) { return 0; }
|
||
|
||
auto amWilling = readInt(source);
|
||
auto neededSystem = readString(source);
|
||
drvPath = store->parseStorePath(readString(source));
|
||
auto requiredFeatures = readStrings<std::set<std::string>>(source);
|
||
|
||
/* It would be possible to build locally after some builds clear out,
|
||
so don't show the warning now: */
|
||
bool couldBuildLocally = maxBuildJobs > 0
|
||
&& ( neededSystem == settings.thisSystem
|
||
|| settings.extraPlatforms.get().count(neededSystem) > 0)
|
||
&& allSupportedLocally(*store, requiredFeatures);
|
||
/* It's possible to build this locally right now: */
|
||
bool canBuildLocally = amWilling && couldBuildLocally;
|
||
|
||
/* Error ignored here, will be caught later */
|
||
mkdir(currentLoad.c_str(), 0777);
|
||
|
||
while (true) {
|
||
bestSlotLock.reset();
|
||
AutoCloseFD lock = openLockFile(currentLoad + "/main-lock", true);
|
||
lockFile(lock.get(), ltWrite);
|
||
|
||
bool rightType = false;
|
||
|
||
Machine * bestMachine = nullptr;
|
||
double bestCost = 0;
|
||
for (auto & m : machines) {
|
||
debug("considering building on remote machine '%s'", m.storeUri);
|
||
|
||
if (m.enabled &&
|
||
m.systemSupported(neededSystem) &&
|
||
m.allSupported(requiredFeatures) &&
|
||
m.mandatoryMet(requiredFeatures) &&
|
||
machineHasRoom(m, *drvPath))
|
||
{
|
||
rightType = true;
|
||
AutoCloseFD free;
|
||
uint64_t load = 0;
|
||
for (uint64_t slot = 0; slot < m.maxJobs; ++slot) {
|
||
auto slotLock = openSlotLock(m, slot);
|
||
if (tryLockFile(slotLock.get(), ltWrite)) {
|
||
if (!free) {
|
||
free = std::move(slotLock);
|
||
}
|
||
} else {
|
||
++load;
|
||
}
|
||
}
|
||
if (!free) {
|
||
continue;
|
||
}
|
||
/* P1: ranking cost folds in live pressure (0 when no
|
||
signal, so this reduces to load / speedFactor). */
|
||
double cost = (double(load) + liveLoadPenalty(m)) / m.speedFactor;
|
||
bool best = false;
|
||
if (!bestSlotLock) {
|
||
best = true;
|
||
} else if (cost < bestCost) {
|
||
best = true;
|
||
} else if (cost == bestCost) {
|
||
if (m.speedFactor > bestMachine->speedFactor) {
|
||
best = true;
|
||
}
|
||
}
|
||
if (best) {
|
||
bestCost = cost;
|
||
bestSlotLock = std::move(free);
|
||
bestMachine = &m;
|
||
}
|
||
}
|
||
}
|
||
|
||
if (!bestSlotLock) {
|
||
if (rightType && !canBuildLocally)
|
||
std::cerr << "# postpone\n";
|
||
else
|
||
{
|
||
// add the template values.
|
||
std::string drvstr;
|
||
if (drvPath.has_value())
|
||
drvstr = drvPath->to_string();
|
||
else
|
||
drvstr = "<unknown>";
|
||
|
||
std::string machinesFormatted;
|
||
|
||
for (auto & m : machines) {
|
||
machinesFormatted += HintFmt(
|
||
"\n([%s], %s, [%s], [%s])",
|
||
concatStringsSep<StringSet>(", ", m.systemTypes),
|
||
m.maxJobs,
|
||
concatStringsSep<StringSet>(", ", m.supportedFeatures),
|
||
concatStringsSep<StringSet>(", ", m.mandatoryFeatures)
|
||
).str();
|
||
}
|
||
|
||
auto error = HintFmt(
|
||
"Failed to find a machine for remote build!\n"
|
||
"derivation: %s\n"
|
||
"required (system, features): (%s, [%s])\n"
|
||
"%s available machines:\n"
|
||
"(systems, maxjobs, supportedFeatures, mandatoryFeatures)%s",
|
||
drvstr,
|
||
neededSystem,
|
||
concatStringsSep<StringSet>(", ", requiredFeatures),
|
||
machines.size(),
|
||
Uncolored(machinesFormatted)
|
||
);
|
||
|
||
printMsg(couldBuildLocally ? lvlChatty : lvlWarn, error.str());
|
||
|
||
std::cerr << "# decline\n";
|
||
}
|
||
break;
|
||
}
|
||
|
||
#if __APPLE__
|
||
futimes(bestSlotLock.get(), nullptr);
|
||
#else
|
||
futimens(bestSlotLock.get(), nullptr);
|
||
#endif
|
||
|
||
lock.reset();
|
||
|
||
try {
|
||
|
||
Activity act(*logger, lvlTalkative, actUnknown, fmt("connecting to '%s'", bestMachine->storeUri));
|
||
|
||
sshStore = aio.blockOn(bestMachine->openStore());
|
||
aio.blockOn(sshStore->connect());
|
||
storeUri = bestMachine->storeUri;
|
||
|
||
} catch (std::exception & e) { // NOLINT(lix-foreign-exceptions)
|
||
auto msg = chomp(drainFD(5, false));
|
||
printError("cannot build on '%s': %s%s",
|
||
bestMachine->storeUri, e.what(),
|
||
msg.empty() ? "" : ": " + msg);
|
||
bestMachine->enabled = false;
|
||
continue;
|
||
}
|
||
|
||
goto connected;
|
||
}
|
||
}
|
||
|
||
connected:
|
||
close(5);
|
||
|
||
assert(sshStore);
|
||
|
||
std::cerr << "# accept\n" << storeUri << "\n";
|
||
|
||
auto inputs = readStrings<PathSet>(source);
|
||
auto wantedOutputs = readStrings<StringSet>(source);
|
||
|
||
auto lockFileName = currentLoad + "/" + makeLockFilename(storeUri) + ".upload-lock";
|
||
|
||
AutoCloseFD uploadLock = openLockFile(lockFileName, true);
|
||
|
||
{
|
||
Activity act(*logger, lvlTalkative, actUnknown, fmt("waiting for the upload lock to '%s'", storeUri));
|
||
|
||
if (!unsafeLockFileSingleThreaded(uploadLock.get(), ltWrite, std::chrono::minutes(15)))
|
||
printError("somebody is hogging the upload lock for '%s', continuing...");
|
||
}
|
||
|
||
auto substitute = settings.buildersUseSubstitutes ? Substitute : NoSubstitute;
|
||
|
||
{
|
||
Activity act(*logger, lvlTalkative, actUnknown, fmt("copying dependencies to '%s'", storeUri));
|
||
aio.blockOn(copyPaths(
|
||
*store,
|
||
*sshStore,
|
||
store->parseStorePathSet(inputs),
|
||
NoRepair,
|
||
NoCheckSigs,
|
||
substitute
|
||
));
|
||
}
|
||
|
||
uploadLock.reset();
|
||
|
||
auto drv = aio.blockOn(store->readDerivation(*drvPath));
|
||
|
||
std::optional<BuildResult> optResult;
|
||
|
||
// If we don't know whether we are trusted (e.g. `ssh://`
|
||
// stores), we assume we are. This is necessary for backwards
|
||
// compat.
|
||
bool trustedOrLegacy = ({
|
||
std::optional trusted = aio.blockOn(sshStore->isTrustedClient());
|
||
!trusted || *trusted;
|
||
});
|
||
|
||
// See the very large comment in `case WorkerProto::Op::BuildDerivation:` in
|
||
// `lix/libstore/daemon.cc` that explains the trust model here.
|
||
//
|
||
// This condition mirrors that: that code enforces the "rules" outlined there;
|
||
// we do the best we can given those "rules".
|
||
if (trustedOrLegacy || drv.type().isCA()) {
|
||
// Hijack the inputs paths of the derivation to include all
|
||
// the paths that come from the `inputDrvs` set. We don’t do
|
||
// that for the derivations whose `inputDrvs` is empty
|
||
// because:
|
||
//
|
||
// 1. It’s not needed
|
||
//
|
||
// 2. Changing the `inputSrcs` set changes the associated
|
||
// output ids, which break CA derivations
|
||
if (!drv.inputDrvs.map.empty())
|
||
drv.inputSrcs = store->parseStorePathSet(inputs);
|
||
optResult = aio.blockOn(sshStore->buildDerivation(*drvPath, (const BasicDerivation &) drv));
|
||
auto & result = *optResult;
|
||
if (!result.success())
|
||
throw Error("build of '%s' on '%s' failed: %s", store->printStorePath(*drvPath), storeUri, result.errorMsg);
|
||
} else {
|
||
aio.blockOn(copyClosure(
|
||
*store, *sshStore, StorePathSet{*drvPath}, NoRepair, NoCheckSigs, substitute
|
||
));
|
||
auto res = aio.blockOn(sshStore->buildPathsWithResults({
|
||
DerivedPath::Built {
|
||
.drvPath = makeConstantStorePathRef(*drvPath),
|
||
.outputs = OutputsSpec::All {},
|
||
}
|
||
}));
|
||
// One path to build should produce exactly one build result
|
||
assert(res.size() == 1);
|
||
optResult = std::move(res[0]);
|
||
}
|
||
|
||
|
||
auto outputHashes = aio.blockOn(staticOutputHashes(*store, drv));
|
||
std::set<Realisation> missingRealisations;
|
||
StorePathSet missingPaths;
|
||
if (experimentalFeatureSettings.isEnabled(Xp::CaDerivations) && !drv.type().hasKnownOutputPaths()) {
|
||
for (auto & outputName : wantedOutputs) {
|
||
auto thisOutputHash = outputHashes.at(outputName);
|
||
auto thisOutputId = DrvOutput{ thisOutputHash, outputName };
|
||
if (!aio.blockOn(store->queryRealisation(thisOutputId))) {
|
||
debug("missing output %s", outputName);
|
||
assert(optResult);
|
||
auto & result = *optResult;
|
||
auto i = result.builtOutputs.find(outputName);
|
||
assert(i != result.builtOutputs.end());
|
||
auto & newRealisation = i->second;
|
||
missingRealisations.insert(newRealisation);
|
||
missingPaths.insert(newRealisation.outPath);
|
||
}
|
||
}
|
||
} else {
|
||
auto outputPaths = drv.outputsAndOptPaths(*store);
|
||
for (auto & [outputName, hopefullyOutputPath] : outputPaths) {
|
||
assert(hopefullyOutputPath.second);
|
||
if (!aio.blockOn(store->isValidPath(*hopefullyOutputPath.second)))
|
||
missingPaths.insert(*hopefullyOutputPath.second);
|
||
}
|
||
}
|
||
|
||
if (!missingPaths.empty()) {
|
||
Activity act(*logger, lvlTalkative, actUnknown, fmt("copying outputs from '%s'", storeUri));
|
||
if (auto localStore = store.try_cast_shared<LocalStore>())
|
||
for (auto & path : missingPaths)
|
||
localStore->locksHeld.insert(store->printStorePath(path)); /* FIXME: ugly */
|
||
aio.blockOn(
|
||
copyPaths(*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
|
||
experimentalFeatureSettings.require(Xp::CaDerivations);
|
||
aio.blockOn(store->registerDrvOutput(realisation));
|
||
}
|
||
|
||
return 0;
|
||
}
|
||
}
|
||
|
||
void registerLegacyBuildRemote() {
|
||
LegacyCommandRegistry::add("build-remote", main_build_remote);
|
||
}
|
||
|
||
}
|