From fca0a30470b7040489feeb2a86bad05bf9b1aa95 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Thu, 29 May 2025 14:20:00 +0200 Subject: [PATCH] libstore: remove pre-2.18 protocols 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 --- doc/manual/rl-next/remove-old-wires.md | 20 + lix/libstore/daemon.cc | 262 +++--------- lix/libstore/remote-store-connection.hh | 4 +- lix/libstore/remote-store.cc | 378 ++++-------------- lix/libstore/remote-store.hh | 2 - lix/libstore/worker-protocol.cc | 70 +--- lix/libstore/worker-protocol.hh | 12 +- lix/libutil/async-io.cc | 51 --- lix/libutil/async-io.hh | 44 -- tests/nixos/default.nix | 9 - .../worker-protocol/build-result-1.27.bin | Bin 80 -> 0 bytes .../worker-protocol/build-result-1.28.bin | Bin 648 -> 0 bytes .../worker-protocol/build-result-1.29.bin | Bin 744 -> 0 bytes .../worker-protocol/derived-path-1.29.bin | Bin 184 -> 0 bytes .../worker-protocol/derived-path-1.30.bin | Bin 248 -> 0 bytes .../keyed-build-result-1.29.bin | Bin 264 -> 0 bytes tests/unit/libstore/worker-protocol.cc | 213 ---------- tests/unit/libutil/async-io.cc | 92 ----- tests/unit/meson.build | 1 - 19 files changed, 183 insertions(+), 975 deletions(-) create mode 100644 doc/manual/rl-next/remove-old-wires.md delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/build-result-1.27.bin delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/build-result-1.28.bin delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/build-result-1.29.bin delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/derived-path-1.29.bin delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/derived-path-1.30.bin delete mode 100644 tests/unit/libstore/data/libstore/worker-protocol/keyed-build-result-1.29.bin delete mode 100644 tests/unit/libutil/async-io.cc diff --git a/doc/manual/rl-next/remove-old-wires.md b/doc/manual/rl-next/remove-old-wires.md new file mode 100644 index 000000000..d0e3777b9 --- /dev/null +++ b/doc/manual/rl-next/remove-old-wires.md @@ -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. diff --git a/lix/libstore/daemon.cc b/lix/libstore/daemon.cc index aceb4da29..8362aca7c 100644 --- a/lix/libstore/daemon.cc +++ b/lix/libstore/daemon.cc @@ -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; - } + to << STDERR_ERROR << *ex; } } @@ -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 case WorkerProto::Op::QueryValidPaths: { auto paths = WorkerProto::Serialise::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 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 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 } 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,109 +361,42 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store } case WorkerProto::Op::AddToStore: { - if (GET_PROTOCOL_MINOR(clientVersion) >= 25) { - auto name = readString(from); - auto camStr = readString(from); - auto refs = WorkerProto::Serialise::read(*store, rconn); - bool repairBool; - from >> repairBool; - auto repair = RepairFlag{repairBool}; + auto name = readString(from); + auto camStr = readString(from); + auto refs = WorkerProto::Serialise::read(*store, rconn); + bool repairBool; + from >> repairBool; + auto repair = RepairFlag{repairBool}; - logger->startWork(); - auto pathInfo = [&]() { - // NB: FramedSource must be out of scope before logger->stopWork(); - auto [contentAddressMethod, hashType_] = ContentAddressMethod::parse(camStr); - auto hashType = hashType_; // work around clang bug - FramedSource source(from); - // TODO this is essentially RemoteStore::addCAToStore. Move it up to Store. - return std::visit(overloaded { - [&](const TextIngestionMethod &) { - 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)); - // We could stream this by changing Store - std::string contents = source.drain(); - auto path = aio.blockOn(store->addTextToStore(name, contents, refs, repair)); - return aio.blockOn(store->queryPathInfo(path)); - }, - [&](const FileIngestionMethod & fim) { - AsyncSourceInputStream stream{source}; - auto path = aio.blockOn( - store->addToStoreFromDump(stream, name, fim, hashType, repair, refs) - ); - return aio.blockOn(store->queryPathInfo(path)); - }, - }, contentAddressMethod.raw); - }(); - logger->stopWork(); + logger->startWork(); + auto pathInfo = [&]() { + // NB: FramedSource must be out of scope before logger->stopWork(); + auto [contentAddressMethod, hashType_] = ContentAddressMethod::parse(camStr); + auto hashType = hashType_; // work around clang bug + FramedSource source(from); + // TODO this is essentially RemoteStore::addCAToStore. Move it up to Store. + return std::visit(overloaded { + [&](const TextIngestionMethod &) { + 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)); + // We could stream this by changing Store + std::string contents = source.drain(); + auto path = aio.blockOn(store->addTextToStore(name, contents, refs, repair)); + return aio.blockOn(store->queryPathInfo(path)); + }, + [&](const FileIngestionMethod & fim) { + AsyncSourceInputStream stream{source}; + auto path = aio.blockOn( + store->addToStoreFromDump(stream, name, fim, hashType, repair, refs) + ); + return aio.blockOn(store->queryPathInfo(path)); + }, + }, contentAddressMethod.raw); + }(); + logger->stopWork(); - to << WorkerProto::Serialise::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); - } + to << WorkerProto::Serialise::write(*store, wconn, *pathInfo); break; } @@ -535,14 +426,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store } case WorkerProto::Op::AddTextToStore: { - std::string suffix = readString(from); - std::string s = readString(from); - auto refs = WorkerProto::Serialise::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 // 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 case WorkerProto::Op::QuerySubstitutablePathInfos: { SubstitutablePathInfos infos; - StorePathCAMap pathsMap = {}; - if (GET_PROTOCOL_MINOR(clientVersion) < 22) { - auto paths = WorkerProto::Serialise::read(*store, rconn); - for (auto & path : paths) - pathsMap.emplace(path, std::nullopt); - } else - pathsMap = WorkerProto::Serialise::read(*store, rconn); + StorePathCAMap pathsMap = WorkerProto::Serialise::read(*store, rconn); logger->startWork(); aio.blockOn(store->querySubstitutablePathInfos(pathsMap, infos)); logger->stopWork(); @@ -902,30 +775,14 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref store if (!trusted) info.ultimate = false; - if (GET_PROTOCOL_MINOR(clientVersion) >= 23) { - logger->startWork(); - { - FramedSource source(from); - AsyncSourceInputStream stream{source}; - aio.blockOn(store->addToStore(info, stream, (RepairFlag) repair, - dontCheckSigs ? NoCheckSigs : CheckSigs)); - } - logger->stopWork(); - } - - else { - std::unique_ptr source; - source = std::make_unique(from, to); - - logger->startWork(); - - // FIXME: race if addToStore doesn't read source? - AsyncSourceInputStream stream{*source}; + logger->startWork(); + { + FramedSource source(from); + AsyncSourceInputStream stream{source}; aio.blockOn(store->addToStore(info, stream, (RepairFlag) repair, dontCheckSigs ? NoCheckSigs : CheckSigs)); - - logger->stopWork(); } + logger->stopWork(); break; } @@ -1015,18 +872,15 @@ void processConnection( readInt(from); // obsolete reserveSpace - if (GET_PROTOCOL_MINOR(clientVersion) >= 33) - to << nixVersion; + 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 - ? aio.blockOn(store->isTrustedClient()) - : std::optional { NotTrusted }; - WorkerProto::WriteConn wconn {clientVersion}; - to << WorkerProto::write(*store, wconn, temp); - } + // We and the underlying store both need to trust the client for + // it to be trusted. + auto temp = trusted + ? aio.blockOn(store->isTrustedClient()) + : std::optional { NotTrusted }; + WorkerProto::WriteConn wconn {clientVersion}; + to << WorkerProto::write(*store, wconn, temp); /* Send startup error messages to the client. */ tunnelLogger->startWork(); diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index f9fade57a..a01c01417 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -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 fun); kj::Promise> diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index fd9d8b241..930cbd7a0 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -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>::read(*this, conn); - } else { - // We don't know the answer; protocol to old. - conn.remoteTrustsUs = std::nullopt; - } + conn.to.flush(); + conn.daemonNixVersion = readString(conn.from); + conn.remoteTrustsUs = WorkerProto::Serialise>::read(*this, conn); 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->to << maybeSubstitute; conn.processStderr(); co_return WorkerProto::Serialise::read(*this, *conn); } catch (...) { @@ -248,13 +238,7 @@ 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->to << WorkerProto::write(*this, *conn, pathsMap); conn.processStderr(); size_t count = readNum(conn->from); for (size_t n = 0; n < count; n++) { @@ -323,64 +307,37 @@ try { } -kj::Promise> 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::read(*this, *conn); -} catch (...) { - co_return result::current_exception(); -} - - kj ::Promise>> 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); - conn.processStderr(); - auto tmp = WorkerProto::Serialise>>::read( - *this, *conn - ); - std::map result; - for (auto & [name, outPath] : tmp) { - if (!outPath) { - throw Error( - "remote responded with unknown outpath for %s^%s", path.to_string(), name - ); - } - result.emplace(std::move(name), std::move(*outPath)); + if (!evalStore_) { + auto conn(TRY_AWAIT(getConnection())); + conn->to << WorkerProto::Op::QueryDerivationOutputMap << printStorePath(path); + conn.processStderr(); + auto tmp = WorkerProto::Serialise>>::read( + *this, *conn + ); + std::map result; + for (auto & [name, outPath] : tmp) { + if (!outPath) { + throw Error( + "remote responded with unknown outpath for %s^%s", path.to_string(), name + ); } - co_return result; - } else { - auto & evalStore = *evalStore_; - auto outputs = TRY_AWAIT(evalStore.queryStaticDerivationOutputMap(path)); - // union with the first branch overriding the statically-known ones - // when non-`std::nullopt`. - for (auto && [outputName, optPath] : - TRY_AWAIT(queryDerivationOutputMap(path, nullptr))) - { - outputs.insert_or_assign(std::move(outputName), std::move(optPath)); - } - co_return outputs; + result.emplace(std::move(name), std::move(*outPath)); } + co_return result; } 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)); + auto & evalStore = *evalStore_; + auto outputs = TRY_AWAIT(evalStore.queryStaticDerivationOutputMap(path)); + // union with the first branch overriding the statically-known ones + // when non-`std::nullopt`. + for (auto && [outputName, optPath] : + TRY_AWAIT(queryDerivationOutputMap(path, nullptr))) + { + outputs.insert_or_assign(std::move(outputName), std::move(optPath)); + } + co_return outputs; } } catch (...) { co_return result::current_exception(); @@ -411,91 +368,24 @@ try { std::optional conn_(TRY_AWAIT(getConnection())); auto & conn = *conn_; - if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 25) { + conn->to + << WorkerProto::Op::AddToStore + << name + << caMethod.render(hashType); + conn->to << WorkerProto::write(*this, *conn, references); + conn->to << repair; - conn->to - << WorkerProto::Op::AddToStore - << name - << caMethod.render(hashType); - conn->to << WorkerProto::write(*this, *conn, references); - conn->to << repair; - - // The dump source may invoke the store, so we need to make some room. - connections->incCapacity(); - { - Finally cleanup([&]() { connections->decCapacity(); }); - TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) { - return dump.drainInto(sink); - })); - } - - co_return make_ref( - WorkerProto::Serialise::read(*this, *conn)); + // The dump source may invoke the store, so we need to make some room. + connections->incCapacity(); + { + Finally cleanup([&]() { connections->decCapacity(); }); + TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) { + return dump.drainInto(sink); + })); } - 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> { - 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> { - 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)); - } + co_return make_ref( + WorkerProto::Serialise::read(*this, *conn)); } 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(); - 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(); - } + auto copier = copyNAR(source); + TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) { + return copier->drainInto(sink); + })); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -563,35 +440,28 @@ kj::Promise> RemoteStore::addMultipleToStore( RepairFlag repair, CheckSigsFlag checkSigs) try { - if (GET_PROTOCOL_MINOR(TRY_AWAIT(getConnection())->daemonVersion) >= 32) { - auto remoteVersion = TRY_AWAIT(getProtocol()); + auto remoteVersion = TRY_AWAIT(getProtocol()); - auto conn(TRY_AWAIT(getConnection())); - conn->to - << WorkerProto::Op::AddMultipleToStore - << repair - << !checkSigs; - // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) - TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) -> kj::Promise> { - try { - sink << pathsToCopy.size(); - for (auto & [pathInfo, pathSource] : pathsToCopy) { - sink << WorkerProto::Serialise::write(*this, - WorkerProto::WriteConn {remoteVersion}, - pathInfo); - TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink)); - } - co_return result::success(); - } catch (...) { - co_return result::current_exception(); + auto conn(TRY_AWAIT(getConnection())); + conn->to + << WorkerProto::Op::AddMultipleToStore + << repair + << !checkSigs; + // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) + TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) -> kj::Promise> { + try { + sink << pathsToCopy.size(); + for (auto & [pathInfo, pathSource] : pathsToCopy) { + sink << WorkerProto::Serialise::write(*this, + WorkerProto::WriteConn {remoteVersion}, + pathInfo); + TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink)); } - })); - } 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(); } - } + })); co_return result::success(); } catch (...) { co_return result::current_exception(); @@ -662,81 +532,11 @@ try { std::optional 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>::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 results; - - for (auto & path : paths) { - auto handlers = overloaded { - [&](const DerivedPath::Opaque & bo) -> kj::Promise> { - 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> { - 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; - } + conn->to << WorkerProto::Op::BuildPathsWithResults; + conn->to << WorkerProto::write(*this, *conn, paths); + conn->to << buildMode; + conn.processStderr(); + co_return WorkerProto::Serialise>::read(*this, *conn); } 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(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(from); - auto buf = std::make_unique(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) { - 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)); - } + if (msg == STDERR_ERROR) { + return std::make_exception_ptr(readError(from)); } 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(); } diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh index 55bac529f..367a6b79a 100644 --- a/lix/libstore/remote-store.hh +++ b/lix/libstore/remote-store.hh @@ -67,8 +67,6 @@ public: kj::Promise> queryValidDerivers(const StorePath & path) override; - kj::Promise> queryDerivationOutputs(const StorePath & path) override; - kj::Promise>> queryDerivationOutputMap(const StorePath & path, Store * evalStore = nullptr) override; kj::Promise>> diff --git a/lix/libstore/worker-protocol.cc b/lix/libstore/worker-protocol.cc index 21e21fbde..d2226eb39 100644 --- a/lix/libstore/worker-protocol.cc +++ b/lix/libstore/worker-protocol.cc @@ -48,34 +48,12 @@ WireFormatGenerator WorkerProto::Serialise>::write(co DerivedPath WorkerProto::Serialise::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(); - } + return DerivedPath::parseLegacy(store, s); } WireFormatGenerator WorkerProto::Serialise::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); - } + co_yield req.to_string_legacy(store); } @@ -101,20 +79,16 @@ BuildResult WorkerProto::Serialise::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::read(store, conn); - for (auto && [output, realisation] : builtOutputs) - res.builtOutputs.insert_or_assign( - std::move(output.outputName), - std::move(realisation)); - } + conn.from + >> res.timesBuilt + >> res.isNonDeterministic + >> res.startTime + >> res.stopTime; + auto builtOutputs = WorkerProto::Serialise::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::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); - } + co_yield res.timesBuilt; + co_yield res.isNonDeterministic; + co_yield res.startTime; + co_yield res.stopTime; + DrvOutputs builtOutputs; + for (auto & [output, realisation] : res.builtOutputs) + builtOutputs.insert_or_assign(realisation.id, realisation); + co_yield WorkerProto::write(store, conn, builtOutputs); } diff --git a/lix/libstore/worker-protocol.hh b/lix/libstore/worker-protocol.hh index fc54fcf8c..1c6c12aed 100644 --- a/lix/libstore/worker-protocol.hh +++ b/lix/libstore/worker-protocol.hh @@ -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 diff --git a/lix/libutil/async-io.cc b/lix/libutil/async-io.cc index 93eb7f9f4..0255097f7 100644 --- a/lix/libutil/async-io.cc +++ b/lix/libutil/async-io.cc @@ -80,55 +80,4 @@ kj::Promise> 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(); - 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 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(); - 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>>> 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; -} } diff --git a/lix/libutil/async-io.hh b/lix/libutil/async-io.hh index dd191bb51..237c28e04 100644 --- a/lix/libutil/async-io.hh +++ b/lix/libutil/async-io.hh @@ -93,48 +93,4 @@ public: kj::Promise> 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>>> result; - }; - - struct Pipe - { - // used by the source implementation - kj::Own> sendRequest; - // used by the async feeder function - kj::Promise nextRequest; - }; - - AsyncInputStream & source; - std::unique_ptr owned; - Pipe pipe; - -public: - explicit IndirectAsyncInputStreamToSource(AsyncInputStream & source); - - explicit IndirectAsyncInputStreamToSource(box_ptr 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 feed(); - - size_t read(char * data, size_t len) override; -}; - } diff --git a/tests/nixos/default.nix b/tests/nixos/default.nix index 5b03adc19..e31f3d082 100644 --- a/tests/nixos/default.nix +++ b/tests/nixos/default.nix @@ -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 ]; diff --git a/tests/unit/libstore/data/libstore/worker-protocol/build-result-1.27.bin b/tests/unit/libstore/data/libstore/worker-protocol/build-result-1.27.bin deleted file mode 100644 index ae684778bc26addba4bf1b3e49cc30edf5f038fa..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 80 gcmZQ&fBf0AiU4H~;_u diff --git a/tests/unit/libstore/data/libstore/worker-protocol/build-result-1.28.bin b/tests/unit/libstore/data/libstore/worker-protocol/build-result-1.28.bin deleted file mode 100644 index 74bcd5cf98b828fb63a931ef7810f7fabc805ca5..0000000000000000000000000000000000000000 GIT binary patch literal 0 HcmV?d00001 literal 648 zcmc&wK?=e!5EQ|aK0%e!+`)6%Si?(*`6_8xaxz$KI6Xx-2t!SGa{k;2f8i+RMR2A;`6>RitBjDY7{lnANJD3OVh8WW_c zkTRG+zDX!Yj%68orEsd#Y$KG=*{FoWL-7`MFAQl%7RmZ0!PYe3jk66aF4r+L$O`tu z&uq-x(J#Q)LAOdzsy>VTJDdcofzX)BfXe~O6CwQ^%woR?}>#sX4il$bfs;n zmv%`YOQ~vvTo;t-%xH@l(n4vSK8bRZQHc`kI^B)Ih0TkNGRhXy8rqZN5BnYj(ieFo zAJ+t*u7l`;??iPt&V)lzi91dfGZFf@g4iVAZN4+jUVRU7o>ol_o!fedeM@Pl_mAVf T^ROZOQyyvZZF!scQ+g3>Sw03m8EMgRZ+ diff --git a/tests/unit/libstore/worker-protocol.cc b/tests/unit/libstore/worker-protocol.cc index 80d0c9a21..0bc32ce69 100644 --- a/tests/unit/libstore/worker-protocol.cc +++ b/tests/unit/libstore/worker-protocol.cc @@ -66,55 +66,6 @@ VERSIONED_CHARACTERIZATION_TEST( }, })) -VERSIONED_CHARACTERIZATION_TEST( - WorkerProtoTest, - derivedPath_1_29, - "derived-path-1.29", - 1 << 8 | 29, - (std::tuple { - 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::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 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 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 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 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, diff --git a/tests/unit/libutil/async-io.cc b/tests/unit/libutil/async-io.cc deleted file mode 100644 index c26d67998..000000000 --- a/tests/unit/libutil/async-io.cc +++ /dev/null @@ -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 -#include -#include - -namespace nix { -TEST(IndirectAsyncInputStreamToSource, basic) -{ - struct Stream : AsyncInputStream - { - int round = 0; - kj::Promise> 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> 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(); -} -} diff --git a/tests/unit/meson.build b/tests/unit/meson.build index 3e24375df..957fe1f5b 100644 --- a/tests/unit/meson.build +++ b/tests/unit/meson.build @@ -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',