guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore: remove pre-2.18 protocols

Browse Source 
the old protocols are largely untested, mostly unused, and have design
problems that make the RPC transition a lot harder, if not impossible.
in theory we could ship a transparent protocol-converting proxy that'd
isolate the daemon itself from old protocol versions, but that's a lot
of code to maintain for presumably little gain or even no gain at all.

Change-Id: I4c3f3bb34d39044f6aeb07c10caaf13b8340a220
This commit is contained in:
eldritch horrors
2025-06-02 22:43:24 +00:00
parent 019b17f4e9
commit fca0a30470
19 changed files with 183 additions and 975 deletions
Download Patch File Download Diff File Expand all files Collapse all files
doc/manual/rl-next/remove-old-wires.md
+20
View File
@@ -0,0 +1,20 @@
---
synopsis: Remove support for daemon protocols before 2.18
issues: []
cls: [3249]
significance: significant
category: "Breaking Changes"
credits: [horrors]
---
Support for daemon wire protocols belonging to Nix 2.18 or older have been
removed. This impacts clients connecting to the local daemon socket or any
remote builder configured using the `ssh-ng` protocol. Builders configured
with the `ssh` protocol are still accessible from clients such as Nix 2.3.
Additionally Lix will not be able to connect to an old daemon locally, and
remote build connections to old daemons is likewise limited to `ssh` urls.
We have decided to take this step because the old protocols are very badly
tested (if at all), maintenance overhead is high, and a number of problems
with their design makes it infeasible to remain backwards compatible while
we move Lix to a more modern RPC mechanism with better versioning support.
lix/libstore/daemon.cc
+10 -156
View File
@@ -126,11 +126,7 @@ struct TunnelLogger : public Logger
if (!ex)
to << STDERR_LAST;
else {
if (GET_PROTOCOL_MINOR(clientVersion) >= 26) {
to << STDERR_ERROR << *ex;
} else {
to << STDERR_ERROR << ex->what() << ex->info().status;
}
}
}
@@ -157,31 +153,6 @@ struct TunnelLogger : public Logger
}
};
struct TunnelSink : Sink
{
Sink & to;
TunnelSink(Sink & to) : to(to) { }
void operator () (std::string_view data) override
{
to << STDERR_WRITE << data;
}
};
struct TunnelSource : BufferedSource
{
Source & from;
BufferedSink & to;
TunnelSource(Source & from, BufferedSink & to) : from(from), to(to) { }
size_t readUnbuffered(char * data, size_t len) override
{
to << STDERR_READ << len;
to.flush();
size_t n = readString(data, len, from);
if (n == 0) throw EndOfFile("unexpected end-of-file");
return n;
}
};
struct ClientSettings
{
bool keepFailed;
@@ -284,10 +255,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
case WorkerProto::Op::QueryValidPaths: {
auto paths = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
SubstituteFlag substitute = NoSubstitute;
if (GET_PROTOCOL_MINOR(clientVersion) >= 27) {
substitute = readInt(from) ? Substitute : NoSubstitute;
}
SubstituteFlag substitute = readInt(from) ? Substitute : NoSubstitute;
logger->startWork();
if (substitute) {
@@ -338,9 +306,12 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
break;
}
case WorkerProto::Op::QueryReferrers:
case WorkerProto::Op::QueryValidDerivers:
case WorkerProto::Op::QueryDerivationOutputs: {
throw UnimplementedError("QueryDerivationOutputs is not supported in Lix. This is not used if the declared server protocol is >= 1.21 (Nix 2.4)");
}
case WorkerProto::Op::QueryReferrers:
case WorkerProto::Op::QueryValidDerivers: {
auto path = store->parseStorePath(readString(from));
logger->startWork();
StorePathSet paths;
@@ -356,12 +327,6 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
paths = aio.blockOn(store->queryValidDerivers(path));
break;
}
case WorkerProto::Op::QueryDerivationOutputs: {
// Only sent if server presents proto version <= 1.21
REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
paths = aio.blockOn(store->queryDerivationOutputs(path));
break;
}
default:
abort();
break;
@@ -374,14 +339,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::QueryDerivationOutputNames: {
// Unused in CppNix >= 2.4 (removed in 045b07200c77bf1fe19c0a986aafb531e7e1ba54)
REMOVE_AFTER_DROPPING_PROTO_MINOR(31);
auto path = store->parseStorePath(readString(from));
logger->startWork();
auto names = aio.blockOn(store->readDerivation(path)).outputNames();
logger->stopWork();
to << names;
break;
throw UnimplementedError("QueryDerivationOutputNames is not supported in Lix. This is not used if the declared server protocol is >= 1.31 (Nix 2.4)");
}
case WorkerProto::Op::QueryDerivationOutputMap: {
@@ -403,7 +361,6 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::AddToStore: {
if (GET_PROTOCOL_MINOR(clientVersion) >= 25) {
auto name = readString(from);
auto camStr = readString(from);
auto refs = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
@@ -440,72 +397,6 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
logger->stopWork();
to << WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
} else {
HashType hashAlgo;
std::string baseName;
FileIngestionMethod method;
{
bool fixed;
uint8_t recursive;
std::string hashAlgoRaw;
from >> baseName >> fixed /* obsolete */ >> recursive >> hashAlgoRaw;
if (recursive > (uint8_t) FileIngestionMethod::Recursive)
throw Error("unsupported FileIngestionMethod with value of %i; you may need to upgrade nix-daemon", recursive);
method = FileIngestionMethod { recursive };
/* Compatibility hack. */
if (!fixed) {
hashAlgoRaw = "sha256";
method = FileIngestionMethod::Recursive;
}
hashAlgo = parseHashType(hashAlgoRaw);
}
// Note to future maintainers: do *not* inline this into the
// generator statement as the lambda itself needs to live to the
// end of the generator's lifetime and is otherwise a UAF.
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines): does not outlive the outer function
auto g = [&]() -> WireFormatGenerator {
if (method == FileIngestionMethod::Recursive) {
/* We parse the NAR dump through into `saved` unmodified,
so why all this extra work? We still parse the NAR so
that we aren't sending arbitrary data to `saved`
unwittingly`, and we know when the NAR ends so we don't
consume the rest of `from` and can't parse another
command. (We don't trust `addToStoreFromDump` to not
eagerly consume the entire stream it's given, past the
length of the Nar. */
co_yield copyNAR(from);
} else {
/* Incrementally parse the NAR file, stripping the
metadata, and streaming the sole file we expect into
`saved`. */
auto parser = nar::parse(from);
nar::File * file = nullptr;
while (auto entry = parser.next()) {
file = std::visit(
overloaded{
[](nar::File & f) -> nar::File * { return &f; },
[](auto &) -> nar::File * { throw Error("regular file expected"); },
},
*entry
);
if (file) {
break;
}
}
if (!file) {
throw Error("regular file expected");
}
co_yield std::move(file->contents);
}
};
AsyncGeneratorInputStream dumpSource{g()};
logger->startWork();
auto path = aio.blockOn(store->addToStoreFromDump(dumpSource, baseName, method, hashAlgo));
logger->stopWork();
to << store->printStorePath(path);
}
break;
}
@@ -535,14 +426,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
}
case WorkerProto::Op::AddTextToStore: {
std::string suffix = readString(from);
std::string s = readString(from);
auto refs = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
logger->startWork();
auto path = aio.blockOn(store->addTextToStore(suffix, s, refs, NoRepair));
logger->stopWork();
to << store->printStorePath(path);
break;
throw UnimplementedError("AddTextToStore is not supported in Lix. This is not used if the declared server protocol is >= 1.25 (Nix 2.4)");
}
case WorkerProto::Op::BuildPaths: {
@@ -697,12 +581,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
// Obsolete since 9947f1646a26b339fff2e02b77798e9841fac7f0 (included in CppNix 2.5.0).
case WorkerProto::Op::SyncWithGC: {
// CppNix 2.5.0 is 32
REMOVE_AFTER_DROPPING_PROTO_MINOR(31);
logger->startWork();
logger->stopWork();
to << 1;
break;
throw UnimplementedError("SyncWithGC is not supported in Lix. This is not used if the declared server protocol is >= 1.31 (Nix 2.5)");
}
case WorkerProto::Op::FindRoots: {
@@ -799,13 +678,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
case WorkerProto::Op::QuerySubstitutablePathInfos: {
SubstitutablePathInfos infos;
StorePathCAMap pathsMap = {};
if (GET_PROTOCOL_MINOR(clientVersion) < 22) {
auto paths = WorkerProto::Serialise<StorePathSet>::read(*store, rconn);
for (auto & path : paths)
pathsMap.emplace(path, std::nullopt);
} else
pathsMap = WorkerProto::Serialise<StorePathCAMap>::read(*store, rconn);
StorePathCAMap pathsMap = WorkerProto::Serialise<StorePathCAMap>::read(*store, rconn);
logger->startWork();
aio.blockOn(store->querySubstitutablePathInfos(pathsMap, infos));
logger->stopWork();
@@ -902,7 +775,6 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
if (!trusted)
info.ultimate = false;
if (GET_PROTOCOL_MINOR(clientVersion) >= 23) {
logger->startWork();
{
FramedSource source(from);
@@ -911,21 +783,6 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
dontCheckSigs ? NoCheckSigs : CheckSigs));
}
logger->stopWork();
}
else {
std::unique_ptr<Source> source;
source = std::make_unique<TunnelSource>(from, to);
logger->startWork();
// FIXME: race if addToStore doesn't read source?
AsyncSourceInputStream stream{*source};
aio.blockOn(store->addToStore(info, stream, (RepairFlag) repair,
dontCheckSigs ? NoCheckSigs : CheckSigs));
logger->stopWork();
}
break;
}
@@ -1015,10 +872,8 @@ void processConnection(
readInt(from); // obsolete reserveSpace
if (GET_PROTOCOL_MINOR(clientVersion) >= 33)
to << nixVersion;
if (GET_PROTOCOL_MINOR(clientVersion) >= 35) {
// We and the underlying store both need to trust the client for
// it to be trusted.
auto temp = trusted
@@ -1026,7 +881,6 @@ void processConnection(
: std::optional { NotTrusted };
WorkerProto::WriteConn wconn {clientVersion};
to << WorkerProto::write(*store, wconn, temp);
}
/* Send startup error messages to the client. */
tunnelLogger->startWork();
lix/libstore/remote-store-connection.hh
+2 -2
View File
@@ -88,7 +88,7 @@ struct RemoteStore::Connection
virtual void closeWrite() = 0;
std::exception_ptr processStderr(Sink * sink = 0, Source * source = 0, bool flush = true);
std::exception_ptr processStderr(bool flush = true);
};
/**
@@ -125,7 +125,7 @@ struct RemoteStore::ConnectionHandle
RemoteStore::Connection & operator * () { return *handle; }
RemoteStore::Connection * operator -> () { return &*handle; }
void processStderr(Sink * sink = 0, Source * source = 0, bool flush = true);
void processStderr(bool flush = true);
void withFramedSink(std::function<void(Sink & sink)> fun);
kj::Promise<Result<void>>
lix/libstore/remote-store.cc
+5 -225
View File
@@ -94,17 +94,9 @@ void RemoteStore::initConnection(Connection & conn)
conn.to << false; // obsolete reserveSpace
if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 33) {
conn.to.flush();
conn.daemonNixVersion = readString(conn.from);
}
if (GET_PROTOCOL_MINOR(conn.daemonVersion) >= 35) {
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(*this, conn);
} else {
// We don't know the answer; protocol to old.
conn.remoteTrustsUs = std::nullopt;
}
auto ex = conn.processStderr();
if (ex) std::rethrow_exception(ex);
@@ -165,9 +157,9 @@ RemoteStore::ConnectionHandle::~ConnectionHandle()
}
}
void RemoteStore::ConnectionHandle::processStderr(Sink * sink, Source * source, bool flush)
void RemoteStore::ConnectionHandle::processStderr(bool flush)
{
auto ex = handle->processStderr(sink, source, flush);
auto ex = handle->processStderr(flush);
if (ex) {
daemonException = true;
std::rethrow_exception(ex);
@@ -207,9 +199,7 @@ try {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryValidPaths;
conn->to << WorkerProto::write(*this, *conn, paths);
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) {
conn->to << maybeSubstitute;
}
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
} catch (...) {
@@ -248,12 +238,6 @@ try {
conn->to << WorkerProto::Op::QuerySubstitutablePathInfos;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) < 22) {
StorePathSet paths;
for (auto & path : pathsMap)
paths.insert(path.first);
conn->to << WorkerProto::write(*this, *conn, paths);
} else
conn->to << WorkerProto::write(*this, *conn, pathsMap);
conn.processStderr();
size_t count = readNum<size_t>(conn->from);
@@ -323,25 +307,9 @@ try {
}
kj::Promise<Result<StorePathSet>> RemoteStore::queryDerivationOutputs(const StorePath & path)
try {
if (GET_PROTOCOL_MINOR(TRY_AWAIT(getProtocol())) >= 22) {
co_return TRY_AWAIT(Store::queryDerivationOutputs(path));
}
REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryDerivationOutputs << printStorePath(path);
conn.processStderr();
co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
} catch (...) {
co_return result::current_exception();
}
kj ::Promise<Result<std::map<std::string, StorePath>>>
RemoteStore::queryDerivationOutputMap(const StorePath & path, Store * evalStore_)
try {
if (GET_PROTOCOL_MINOR(TRY_AWAIT(getProtocol())) >= 22) {
if (!evalStore_) {
auto conn(TRY_AWAIT(getConnection()));
conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path);
@@ -371,17 +339,6 @@ try {
}
co_return outputs;
}
} else {
REMOVE_AFTER_DROPPING_PROTO_MINOR(21);
auto & evalStore = evalStore_ ? *evalStore_ : *this;
// Fallback for old daemon versions.
// For floating-CA derivations (and their co-dependencies) this is an
// under-approximation as it only returns the paths that can be inferred
// from the derivation itself (and not the ones that are known because
// the have been built), but as old stores don't handle floating-CA
// derivations this shouldn't matter
co_return TRY_AWAIT(evalStore.queryStaticDerivationOutputMap(path));
}
} catch (...) {
co_return result::current_exception();
}
@@ -411,8 +368,6 @@ try {
std::optional<ConnectionHandle> conn_(TRY_AWAIT(getConnection()));
auto & conn = *conn_;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) {
conn->to
<< WorkerProto::Op::AddToStore
<< name
@@ -431,71 +386,6 @@ try {
co_return make_ref<ValidPathInfo>(
WorkerProto::Serialise<ValidPathInfo>::read(*this, *conn));
}
else {
if (repair) throw Error("repairing is not supported when building through the Nix daemon protocol < 1.25");
auto handlers = overloaded{
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[&](const TextIngestionMethod & thm) -> kj::Promise<Result<void>> {
try {
if (hashType != HashType::SHA256)
throw UnimplementedError("When adding text-hashed data called '%s', only SHA-256 is supported but '%s' was given",
name, printHashType(hashType));
std::string s = TRY_AWAIT(dump.drain());
conn->to << WorkerProto::Op::AddTextToStore << name << s;
conn->to << WorkerProto::write(*this, *conn, references);
conn.processStderr();
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
},
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[&](const FileIngestionMethod & fim) -> kj::Promise<Result<void>> {
try {
conn->to
<< WorkerProto::Op::AddToStore
<< name
<< ((hashType == HashType::SHA256 && fim == FileIngestionMethod::Recursive) ? 0 : 1) /* backwards compatibility hack */
<< (fim == FileIngestionMethod::Recursive ? 1 : 0)
<< printHashType(hashType);
try {
conn->to.written = 0;
connections->incCapacity();
{
Finally cleanup([&]() { connections->decCapacity(); });
if (fim == FileIngestionMethod::Recursive) {
TRY_AWAIT(dump.drainInto(conn->to));
} else {
std::string contents = TRY_AWAIT(dump.drain());
conn->to << dumpString(contents);
}
}
conn.processStderr();
} catch (SysError & e) {
/* Daemon closed while we were sending the path. Probably OOM
or I/O error. */
if (e.errNo == EPIPE)
try {
conn.processStderr();
} catch (EndOfFile & e) { }
throw;
}
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
}
};
TRY_AWAIT(std::visit(handlers, caMethod.raw));
auto path = parseStorePath(readString(conn->from));
// Release our connection to prevent a deadlock in queryPathInfo().
conn_.reset();
co_return TRY_AWAIT(queryPathInfo(path));
}
} catch (...) {
co_return result::current_exception();
}
@@ -534,23 +424,10 @@ try {
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
<< repair << !checkSigs;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 23) {
auto copier = copyNAR(source);
TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) {
return copier->drainInto(sink);
}));
} else {
IndirectAsyncInputStreamToSource is(source);
auto pfp = kj::newPromiseAndCrossThreadFulfiller<void>();
auto thread = std::async(std::launch::async, [&] {
KJ_DEFER(pfp.fulfiller->fulfill());
conn.processStderr(0, &is);
});
co_await pfp.promise.exclusiveJoin(is.feed());
// if the thread stops we're always clear. if the feeder stops early (or
// fails) it'll have thrown an exception, and the thread will stop soon.
thread.get();
}
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -563,7 +440,6 @@ kj::Promise<Result<void>> RemoteStore::addMultipleToStore(
RepairFlag repair,
CheckSigsFlag checkSigs)
try {
if (GET_PROTOCOL_MINOR(TRY_AWAIT(getConnection())->daemonVersion) >= 32) {
auto remoteVersion = TRY_AWAIT(getProtocol());
auto conn(TRY_AWAIT(getConnection()));
@@ -586,12 +462,6 @@ try {
co_return result::current_exception();
}
}));
} else {
for (auto & [pathInfo, pathSource] : pathsToCopy) {
pathInfo.ultimate = false; // duplicated in daemon.cc AddMultipleToStore
TRY_AWAIT(addToStore(pathInfo, *TRY_AWAIT(pathSource()), repair, checkSigs));
}
}
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -662,81 +532,11 @@ try {
std::optional<ConnectionHandle> conn_(TRY_AWAIT(getConnection()));
auto & conn = *conn_;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
conn->to << WorkerProto::Op::BuildPathsWithResults;
conn->to << WorkerProto::write(*this, *conn, paths);
conn->to << buildMode;
conn.processStderr();
co_return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, *conn);
} else {
REMOVE_AFTER_DROPPING_PROTO_MINOR(33);
// Avoid deadlock.
conn_.reset();
// Note: this throws an exception if a build/substitution
// fails, but meh.
TRY_AWAIT(buildPaths(paths, buildMode, evalStore));
std::vector<KeyedBuildResult> results;
for (auto & path : paths) {
auto handlers = overloaded {
[&](const DerivedPath::Opaque & bo) -> kj::Promise<Result<void>> {
try {
results.push_back(KeyedBuildResult {
{
.status = BuildResult::Substituted,
},
/* .path = */ bo,
});
return {result::success()};
} catch (...) {
return {result::current_exception()};
}
},
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
[&](const DerivedPath::Built & bfd) -> kj::Promise<Result<void>> {
try {
KeyedBuildResult res {
{
.status = BuildResult::Built
},
/* .path = */ bfd,
};
OutputPathMap outputs;
auto drvPath = bfd.drvPath.path;
auto drv = TRY_AWAIT(evalStore->readDerivation(drvPath));
const auto outputHashes =
TRY_AWAIT(staticOutputHashes(*evalStore, drv)); // FIXME: expensive
auto built = TRY_AWAIT(resolveDerivedPath(*this, bfd, &*evalStore));
for (auto & [output, outputPath] : built) {
auto outputHash = get(outputHashes, output);
if (!outputHash)
throw Error(
"the derivation '%s' doesn't have an output named '%s'",
printStorePath(drvPath), output);
auto outputId = DrvOutput{ *outputHash, output };
res.builtOutputs.emplace(
output,
Realisation {
.id = outputId,
.outPath = outputPath,
});
}
results.push_back(res);
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
}
};
TRY_AWAIT(std::visit(handlers , path.raw()));
}
co_return results;
}
} catch (...) {
co_return result::current_exception();
}
@@ -975,7 +775,7 @@ static Logger::Fields readFields(Source & from)
}
std::exception_ptr RemoteStore::Connection::processStderr(Sink * sink, Source * source, bool flush)
std::exception_ptr RemoteStore::Connection::processStderr(bool flush)
{
if (flush)
to.flush();
@@ -984,28 +784,8 @@ std::exception_ptr RemoteStore::Connection::processStderr(Sink * sink, Source *
auto msg = readNum<uint64_t>(from);
if (msg == STDERR_WRITE) {
auto s = readString(from);
if (!sink) throw Error("no sink");
(*sink)(s);
}
else if (msg == STDERR_READ) {
if (!source) throw Error("no source");
size_t len = readNum<size_t>(from);
auto buf = std::make_unique<char[]>(len);
to << std::string_view((const char *) buf.get(), source->read(buf.get(), len));
to.flush();
}
else if (msg == STDERR_ERROR) {
if (GET_PROTOCOL_MINOR(daemonVersion) >= 26) {
if (msg == STDERR_ERROR) {
return std::make_exception_ptr(readError(from));
} else {
auto error = readString(from);
unsigned int status = readInt(from);
return std::make_exception_ptr(Error(status, error));
}
}
else if (msg == STDERR_NEXT)
@@ -1048,7 +828,7 @@ RemoteStore::ConnectionHandle::FramedSinkHandler::FramedSinkHandler(
)
: stderrHandler([&]() {
try {
conn.processStderr(nullptr, nullptr, false);
conn.processStderr(false);
} catch (...) {
ex = std::current_exception();
}
lix/libstore/remote-store.hh
-2
View File
@@ -67,8 +67,6 @@ public:
kj::Promise<Result<StorePathSet>> queryValidDerivers(const StorePath & path) override;
kj::Promise<Result<StorePathSet>> queryDerivationOutputs(const StorePath & path) override;
kj::Promise<Result<std::map<std::string, StorePath>>>
queryDerivationOutputMap(const StorePath & path, Store * evalStore = nullptr) override;
kj::Promise<Result<std::optional<StorePath>>>
lix/libstore/worker-protocol.cc
-30
View File
@@ -48,34 +48,12 @@ WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(co
DerivedPath WorkerProto::Serialise<DerivedPath>::read(const Store & store, WorkerProto::ReadConn conn)
{
auto s = readString(conn.from);
if (GET_PROTOCOL_MINOR(conn.version) >= 30) {
return DerivedPath::parseLegacy(store, s);
} else {
return parsePathWithOutputs(store, s).toDerivedPath();
}
}
WireFormatGenerator WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
{
if (GET_PROTOCOL_MINOR(conn.version) >= 30) {
co_yield req.to_string_legacy(store);
} else {
auto sOrDrvPath = StorePathWithOutputs::tryFromDerivedPath(req);
co_yield std::visit(overloaded {
[&](const StorePathWithOutputs & s) -> std::string {
return s.to_string(store);
},
[&](const StorePath & drvPath) -> std::string {
throw Error("trying to request '%s', but daemon protocol %d.%d is too old (< 1.29) to request a derivation file",
store.printStorePath(drvPath),
GET_PROTOCOL_MAJOR(conn.version),
GET_PROTOCOL_MINOR(conn.version));
},
[&](std::monostate) -> std::string {
throw Error("wanted to build a derivation that is itself a build product, but protocols do not support that. Try upgrading the Nix implementation on the other end of this connection");
},
}, sOrDrvPath);
}
}
@@ -101,20 +79,16 @@ BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, Worke
BuildResult res;
res.status = (BuildResult::Status) readInt(conn.from);
conn.from >> res.errorMsg;
if (GET_PROTOCOL_MINOR(conn.version) >= 29) {
conn.from
>> res.timesBuilt
>> res.isNonDeterministic
>> res.startTime
>> res.stopTime;
}
if (GET_PROTOCOL_MINOR(conn.version) >= 28) {
auto builtOutputs = WorkerProto::Serialise<DrvOutputs>::read(store, conn);
for (auto && [output, realisation] : builtOutputs)
res.builtOutputs.insert_or_assign(
std::move(output.outputName),
std::move(realisation));
}
return res;
}
@@ -122,18 +96,14 @@ WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(const Store & sto
{
co_yield res.status;
co_yield res.errorMsg;
if (GET_PROTOCOL_MINOR(conn.version) >= 29) {
co_yield res.timesBuilt;
co_yield res.isNonDeterministic;
co_yield res.startTime;
co_yield res.stopTime;
}
if (GET_PROTOCOL_MINOR(conn.version) >= 28) {
DrvOutputs builtOutputs;
for (auto & [output, realisation] : res.builtOutputs)
builtOutputs.insert_or_assign(realisation.id, realisation);
co_yield WorkerProto::write(store, conn, builtOutputs);
}
}
lix/libstore/worker-protocol.hh
+4 -8
View File
@@ -14,9 +14,7 @@ namespace nix {
// protocol is bad in design and implementation and Lix intends to replace it
// entirely.
#define PROTOCOL_VERSION (1 << 8 | 35)
// Nix 2.3 is protocol 1.21 (see RemoteStore::initConnection for client,
// processConnection for server).
#define MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION 21
#define MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION 35
#define MIN_SUPPORTED_WORKER_PROTO_VERSION (1 << 8 | MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION)
#define GET_PROTOCOL_MAJOR(x) ((x) & 0xff00)
@@ -28,8 +26,6 @@ namespace nix {
#define STDERR_NEXT 0x6f6c6d67
#define STDERR_READ 0x64617461 // data needed from source
#define STDERR_WRITE 0x64617416 // data for sink
#define STDERR_LAST 0x616c7473
#define STDERR_ERROR 0x63787470
#define STDERR_START_ACTIVITY 0x53545254
@@ -145,19 +141,19 @@ enum struct WorkerProto::Op : uint64_t
QueryReferences = 5, // obsolete since 2016, stubbed to error
QueryReferrers = 6,
AddToStore = 7,
AddTextToStore = 8, // obsolete since protocol 1.25, CppNix 2.4. Use WorkerProto::Op::AddToStore
AddTextToStore = 8, // obsolete, removed
BuildPaths = 9,
EnsurePath = 10,
AddTempRoot = 11,
AddIndirectRoot = 12,
SyncWithGC = 13, // obsolete since CppNix 2.5.0
SyncWithGC = 13, // obsolete since CppNix 2.5.0, removed
FindRoots = 14,
ExportPath = 16, // obsolete since 2017, stubbed to error
QueryDeriver = 18, // obsolete since 2016, stubbed to error
SetOptions = 19,
CollectGarbage = 20,
QuerySubstitutablePathInfo = 21,
QueryDerivationOutputs = 22, // obsolete since protocol 1.21, CppNix 2.4
QueryDerivationOutputs = 22, // obsolete, removed
QueryAllValidPaths = 23,
QueryFailedPaths = 24, // obsolete, removed
ClearFailedPaths = 25, // obsolete, removed
lix/libutil/async-io.cc
-51
View File
@@ -80,55 +80,4 @@ kj::Promise<Result<size_t>> AsyncFdInputStream::read(void * buffer, size_t size)
return {result::failure(std::make_exception_ptr(SysError(errno, "read failed")))};
}
}
IndirectAsyncInputStreamToSource::IndirectAsyncInputStreamToSource(AsyncInputStream & source)
: source(source)
, pipe([&] {
auto pfp = kj::newPromiseAndCrossThreadFulfiller<Request>();
return Pipe{std::move(pfp.fulfiller), std::move(pfp.promise)};
}())
{
}
IndirectAsyncInputStreamToSource::~IndirectAsyncInputStreamToSource() noexcept(true)
{
if (pipe.sendRequest->isWaiting()) {
pipe.sendRequest->fulfill(Request{nullptr, 0, {}});
}
}
kj::Promise<void> IndirectAsyncInputStreamToSource::feed()
{
while (true) {
auto req = co_await pipe.nextRequest;
if (req.data == nullptr) {
break;
}
try {
auto got = (co_await source.read(req.data, req.len)).value();
if (req.len != 0 && got == 0) {
auto eof = std::make_exception_ptr(EndOfFile("async input finished"));
req.result.set_exception(eof);
break;
} else {
auto pfp = kj::newPromiseAndCrossThreadFulfiller<Request>();
req.result.set_value(std::pair{got, std::move(pfp.fulfiller)});
pipe.nextRequest = std::move(pfp.promise);
}
} catch (...) {
req.result.set_exception(std::current_exception());
co_return;
}
}
}
size_t IndirectAsyncInputStreamToSource::read(char * data, size_t len)
{
std::promise<std::pair<size_t, kj::Own<kj::CrossThreadPromiseFulfiller<Request>>>> promise;
auto future = promise.get_future();
pipe.sendRequest->fulfill(Request{data, len, std::move(promise)});
auto [result, next] = future.get();
pipe.sendRequest = std::move(next);
return result;
}
}
lix/libutil/async-io.hh
-44
View File
@@ -93,48 +93,4 @@ public:
kj::Promise<Result<size_t>> read(void * buffer, size_t size) override;
};
/**
* Wraps a stream in a source. The returned source must not be used on the
* event loop that created it, otherwise read requests cannot be serviced.
*/
class IndirectAsyncInputStreamToSource : public Source
{
struct Request
{
char * data;
size_t len;
std::promise<std::pair<size_t, kj::Own<kj::CrossThreadPromiseFulfiller<Request>>>> result;
};
struct Pipe
{
// used by the source implementation
kj::Own<kj::CrossThreadPromiseFulfiller<Request>> sendRequest;
// used by the async feeder function
kj::Promise<Request> nextRequest;
};
AsyncInputStream & source;
std::unique_ptr<AsyncInputStream> owned;
Pipe pipe;
public:
explicit IndirectAsyncInputStreamToSource(AsyncInputStream & source);
explicit IndirectAsyncInputStreamToSource(box_ptr<AsyncInputStream> owned)
: IndirectAsyncInputStreamToSource(*owned)
{
this->owned = std::move(owned).take();
}
~IndirectAsyncInputStreamToSource() noexcept(true);
KJ_DISALLOW_COPY_AND_MOVE(IndirectAsyncInputStreamToSource);
/** Feed the source. Must be awaited fully to drain the input stream. */
kj::Promise<void> feed();
size_t read(char * data, size_t len) override;
};
}
tests/nixos/default.nix
-9
View File
@@ -111,15 +111,6 @@ in
# Test our Nix as a client against remotes that are older
remoteBuildsSshNg_remote_2_3 = runNixOSTestFor "x86_64-linux" {
name = "remoteBuildsSshNg_remote_2_3";
imports = [ ./remote-builds-ssh-ng.nix ];
builders.config = { lib, pkgs, ... }: {
imports = [ checkOverrideNixVersion ];
nix.package = lib.mkForce pkgs.nixVersions.nix_2_3;
};
};
remoteBuildsSshNg_remote_2_18 = runNixOSTestFor "x86_64-linux" {
name = "remoteBuildsSshNg_remote_2_18";
imports = [ ./remote-builds-ssh-ng.nix ];
tests/unit/libstore/data/libstore/worker-protocol/build-result-1.27.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/data/libstore/worker-protocol/build-result-1.28.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/data/libstore/worker-protocol/build-result-1.29.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/data/libstore/worker-protocol/derived-path-1.29.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/data/libstore/worker-protocol/derived-path-1.30.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/data/libstore/worker-protocol/keyed-build-result-1.29.bin
BIN
View File
Binary file not shown.
tests/unit/libstore/worker-protocol.cc
-213
View File
@@ -66,55 +66,6 @@ VERSIONED_CHARACTERIZATION_TEST(
},
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
derivedPath_1_29,
"derived-path-1.29",
1 << 8 | 29,
(std::tuple<DerivedPath, DerivedPath, DerivedPath> {
DerivedPath::Opaque {
.path = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-foo" },
},
DerivedPath::Built {
.drvPath = makeConstantStorePath(StorePath {
"g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar.drv",
}),
.outputs = OutputsSpec::All { },
},
DerivedPath::Built {
.drvPath = makeConstantStorePath(StorePath {
"g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar.drv",
}),
.outputs = OutputsSpec::Names { "x", "y" },
},
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
derivedPath_1_30,
"derived-path-1.30",
1 << 8 | 30,
(std::tuple<DerivedPath, DerivedPath, DerivedPath, DerivedPath> {
DerivedPath::Opaque {
.path = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-foo" },
},
DerivedPath::Opaque {
.path = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-foo.drv" },
},
DerivedPath::Built {
.drvPath = makeConstantStorePath(StorePath {
"g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar.drv",
}),
.outputs = OutputsSpec::All { },
},
DerivedPath::Built {
.drvPath = makeConstantStorePath(StorePath {
"g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar.drv",
}),
.outputs = OutputsSpec::Names { "x", "y" },
},
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
drvOutput,
@@ -164,170 +115,6 @@ VERSIONED_CHARACTERIZATION_TEST(
},
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
buildResult_1_27,
"build-result-1.27",
1 << 8 | 27,
({
using namespace std::literals::chrono_literals;
std::tuple<BuildResult, BuildResult, BuildResult> t {
BuildResult {
.status = BuildResult::OutputRejected,
.errorMsg = "no idea why",
},
BuildResult {
.status = BuildResult::NotDeterministic,
.errorMsg = "no idea why",
},
BuildResult {
.status = BuildResult::Built,
},
};
t;
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
buildResult_1_28,
"build-result-1.28",
1 << 8 | 28,
({
using namespace std::literals::chrono_literals;
std::tuple<BuildResult, BuildResult, BuildResult> t {
BuildResult {
.status = BuildResult::OutputRejected,
.errorMsg = "no idea why",
},
BuildResult {
.status = BuildResult::NotDeterministic,
.errorMsg = "no idea why",
},
BuildResult {
.status = BuildResult::Built,
.builtOutputs = {
{
"foo",
{
.id = DrvOutput {
.drvHash = Hash::parseSRI("sha256-b4afnqKCO9oWXgYHb9DeQ2berSwOjS27rSd9TxXDc/U="),
.outputName = "foo",
},
.outPath = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-foo" },
},
},
{
"bar",
{
.id = DrvOutput {
.drvHash = Hash::parseSRI("sha256-b4afnqKCO9oWXgYHb9DeQ2berSwOjS27rSd9TxXDc/U="),
.outputName = "bar",
},
.outPath = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar" },
},
},
},
},
};
t;
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
buildResult_1_29,
"build-result-1.29",
1 << 8 | 29,
({
using namespace std::literals::chrono_literals;
std::tuple<BuildResult, BuildResult, BuildResult> t {
BuildResult {
.status = BuildResult::OutputRejected,
.errorMsg = "no idea why",
},
BuildResult {
.status = BuildResult::NotDeterministic,
.errorMsg = "no idea why",
.timesBuilt = 3,
.isNonDeterministic = true,
.startTime = 30,
.stopTime = 50,
},
BuildResult {
.status = BuildResult::Built,
.timesBuilt = 1,
.builtOutputs = {
{
"foo",
{
.id = DrvOutput {
.drvHash = Hash::parseSRI("sha256-b4afnqKCO9oWXgYHb9DeQ2berSwOjS27rSd9TxXDc/U="),
.outputName = "foo",
},
.outPath = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-foo" },
},
},
{
"bar",
{
.id = DrvOutput {
.drvHash = Hash::parseSRI("sha256-b4afnqKCO9oWXgYHb9DeQ2berSwOjS27rSd9TxXDc/U="),
.outputName = "bar",
},
.outPath = StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar" },
},
},
},
.startTime = 30,
.stopTime = 50,
#if 0
// These fields are not yet serialized.
// FIXME Include in next version of protocol or document
// why they are skipped.
.cpuUser = std::chrono::milliseconds(500s),
.cpuSystem = std::chrono::milliseconds(604s),
#endif
},
};
t;
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
keyedBuildResult_1_29,
"keyed-build-result-1.29",
1 << 8 | 29,
({
using namespace std::literals::chrono_literals;
std::tuple<KeyedBuildResult, KeyedBuildResult/*, KeyedBuildResult*/> t {
KeyedBuildResult {
{
.status = KeyedBuildResult::OutputRejected,
.errorMsg = "no idea why",
},
/* .path = */ DerivedPath::Opaque {
StorePath { "g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-xxx" },
},
},
KeyedBuildResult {
{
.status = KeyedBuildResult::NotDeterministic,
.errorMsg = "no idea why",
.timesBuilt = 3,
.isNonDeterministic = true,
.startTime = 30,
.stopTime = 50,
},
/* .path = */ DerivedPath::Built {
.drvPath = makeConstantStorePath(StorePath {
"g1w7hy3qg1w7hy3qg1w7hy3qg1w7hy3q-bar.drv",
}),
.outputs = OutputsSpec::Names { "out" },
},
},
};
t;
}))
VERSIONED_CHARACTERIZATION_TEST(
WorkerProtoTest,
unkeyedValidPathInfo,
tests/unit/libutil/async-io.cc
-92
View File
@@ -1,92 +0,0 @@
#include "lix/libutil/async-io.hh"
#include "lix/libutil/async.hh"
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/result.hh"
#include <exception>
#include <gtest/gtest.h>
#include <stdexcept>
namespace nix {
TEST(IndirectAsyncInputStreamToSource, basic)
{
struct Stream : AsyncInputStream
{
int round = 0;
kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
{
round++;
if (round <= 10) {
memset(buffer, size, 1);
return {{1}};
} else if (round <= 13) {
memset(buffer, size, size);
return {{size}};
} else {
return {result::success(0)};
}
}
};
AsyncIoRoot aio;
Stream s;
IndirectAsyncInputStreamToSource is(s);
auto user = std::async(std::launch::async, [&]() {
char buf[1026];
// single read
ASSERT_EQ(is.read(buf, 1), 1);
ASSERT_EQ(buf[0], 1);
// read spanning blocks doesn't coalesce
ASSERT_EQ(is.read(buf, 5), 1);
ASSERT_EQ(buf[0], 5);
// coalescing from Source works
ASSERT_NO_THROW(is(buf, 8));
ASSERT_EQ(buf[0], 8);
// next reads fill all sizes
ASSERT_EQ(is.read(buf, 513), 513);
ASSERT_EQ(buf[0], 1);
ASSERT_EQ(is.read(buf, 1025), 1025);
ASSERT_EQ(buf[0], 1);
// zero-size reads don't EOF
ASSERT_EQ(is.read(buf, 0), 0);
// EOF propagates
ASSERT_THROW(is.read(buf, 1025), EndOfFile);
});
is.feed().wait(aio.kj.waitScope);
user.get();
}
TEST(IndirectAsyncInputStreamToSource, errorPropagation)
{
struct Stream : AsyncInputStream
{
int round = 0;
kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
{
return {result::failure(std::make_exception_ptr(std::invalid_argument("foo")))};
}
};
AsyncIoRoot aio;
Stream s;
IndirectAsyncInputStreamToSource is(s);
auto user = std::async(std::launch::async, [&]() {
char buf[1];
ASSERT_THROW(is.read(buf, 1), std::invalid_argument);
});
is.feed().wait(aio.kj.waitScope);
user.get();
}
}
tests/unit/meson.build
-1
View File
@@ -45,7 +45,6 @@ libutil_tests_sources = files(
# keep-sorted start
'libutil/archive.cc',
'libutil/async-collect.cc',
'libutil/async-io.cc',
'libutil/async-semaphore.cc',
'libutil/canon-path.cc',
'libutil/checked-arithmetic.cc',