From d0a4b55a0ea2941cad056abfa49135d7010e1117 Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Sun, 3 May 2026 16:32:26 +0200 Subject: [PATCH] treewide: pour the rpc foundations pour the foundations for rpc, and let them set. this is very much unstable and must be opted into with explicit store uris (e.g. by setting `NIX_REMOTE=daemon?protocol=any`). the daemon sockets are not enabled by default and must be enabled with the `rpc-sockets` experimental feature. we will not advertise this just yet because in the current state it one has to be *very* dedicated to the rpc cause to deploy this, but once we have some more bits migrated we may want to add release notes and officially as for beta testing. Co-Authored-By: piegames Change-Id: I85a96ccb700b91190c1eb37154bcc6ae1c03401a --- lix/libmain/progress-bar.cc | 4 +- lix/libstore/daemon.capnp | 58 ++++ lix/libstore/daemon.cc | 283 ++++++++++++++++++ lix/libstore/daemon.hh | 4 + lix/libstore/globals.cc | 9 + lix/libstore/globals.hh | 6 + lix/libstore/meson.build | 1 + lix/libstore/remote-store.hh | 4 +- lix/libstore/uds-remote-store.cc | 153 +++++++++- lix/libstore/uds-remote-store.hh | 41 +++ .../experimental-features/rpc-sockets.md | 8 + lix/libutil/meson.build | 1 + lix/nix/daemon.cc | 17 ++ misc/systemd/meson.build | 3 +- tests/functional2/daemon/test_connect.py | 9 +- tests/functional2/daemon/test_rpc_setup.py | 19 ++ tests/functional2/testlib/fixtures/nix.py | 15 +- tests/nixos/default.nix | 2 + 18 files changed, 624 insertions(+), 13 deletions(-) create mode 100644 lix/libstore/daemon.capnp create mode 100644 lix/libutil/experimental-features/rpc-sockets.md create mode 100644 tests/functional2/daemon/test_rpc_setup.py diff --git a/lix/libmain/progress-bar.cc b/lix/libmain/progress-bar.cc index 378ffacd3..54a580a73 100644 --- a/lix/libmain/progress-bar.cc +++ b/lix/libmain/progress-bar.cc @@ -197,8 +197,10 @@ Logger::BufferState ProgressBar::startActivityImpl( if ((type == actFileTransfer && hasAncestor(*state, actCopyPath, parent)) || (type == actFileTransfer && hasAncestor(*state, actQueryPathInfo, parent)) - || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent))) + || (type == actCopyPath && hasAncestor(*state, actSubstitute, parent)) || (s == "daemon connection")) + { i->visible = false; + } update(*state); return BufferState::HasSpace; diff --git a/lix/libstore/daemon.capnp b/lix/libstore/daemon.capnp new file mode 100644 index 000000000..11fc11c39 --- /dev/null +++ b/lix/libstore/daemon.capnp @@ -0,0 +1,58 @@ +@0xd8aa4d286ba6797b; + +# IMPORTANT NOTICE +# +# these definitions are EXPERIMENTAL and come with NO stability guarantees + +using Cxx = import "/capnp/c++.capnp"; +$Cxx.namespace("nix::rpc::daemon"); +$Cxx.allowCancellation; + +using T = import "/lix/libutil/types.capnp"; +using Log = import "/lix/libutil/logging.capnp"; + +struct ProtocolDescription { + id @0 :Text; + description @1 :Text; +} + +interface Bootstrap { + supported @0 () -> (protocols :List(ProtocolDescription)); + request @1 ( + clientInfo :Text, + protocol :Text, + ) -> (result :T.Result(Protocol)); +} + +interface Protocol { + # TODO maybe add information or something +} + +# legacy boot protocol. EXPLICITLY UNSTABLE, this id will change frequently and without notice. +# every change to the experimental tunneling protocol may also change this protocol identifier. +const unstableLegacyTunneled :Text = "lix/legacy/ba3153c5-4153-4d66-91ec-a258c02e9a3c"; + +interface LegacyBoot extends(Protocol) { + enum Trust { + unknown @0; + untrusted @1; + trusted @2; + } + + init @0 ( + logger :Log.LogStream, + replyStream :LegacyStream, + ) -> (result :T.Result(InitResult)); + + struct InitResult { + requestStream @0 :LegacyStream; + trust @1 :Trust; + version @2 :Text; + } +} + +interface LegacyStream { + feed @0 (raw :Data) -> stream; + # must be called before a new op is started, otherwise errors may get lost + sync @1 () -> (result :T.ResultV); +} diff --git a/lix/libstore/daemon.cc b/lix/libstore/daemon.cc index 3da704641..019940c19 100644 --- a/lix/libstore/daemon.cc +++ b/lix/libstore/daemon.cc @@ -1,6 +1,7 @@ #include "lix/libstore/daemon.hh" #include "filetransfer.hh" #include "libutil/async.hh" +#include "libutil/logging-rpc.hh" #include "lix/libutil/async-io.hh" #include "lix/libutil/monitor-fd.hh" #include "lix/libstore/worker-protocol.hh" @@ -18,10 +19,17 @@ #include "lix/libutil/serialise.hh" #include "lix/libutil/strings.hh" #include "lix/libutil/args.hh" +#include "lix/libstore/daemon.capnp.h" +#include "lix/libutil/rpc.hh" +#include "lix/libutil/types-rpc.hh" #include +#include #include #include +#include +#include +#include #include namespace nix::daemon { @@ -978,4 +986,279 @@ void processLegacyConnection( } } +namespace { +using namespace rpc::daemon; + +// Shared state for all legacy protocol implementation structs +struct LegacyState +{ + ref store; + TrustedFlag trusted; + + LegacyState(ref store, TrustedFlag trusted) : store(store), trusted(trusted) {} +}; + +struct RequestStreamImpl final : LegacyStream::Server +{ + ref state; + std::exception_ptr error; + AsyncFdIoStream workerSock; + kj::Promise responseForwarder; + + RequestStreamImpl( + ref state, + kj::Promise error, + LegacyStream::Client callbacks, + AutoCloseFD workerFd + ) + : state(state) + , workerSock(std::move(workerFd)) + , responseForwarder(forwardResponse(callbacks).exclusiveJoin( + error.then([&](std::exception_ptr e) -> kj::Promise { + onError(e); + return kj::NEVER_DONE; + }) + )) + { + } + + void onError(std::exception_ptr e) + { + if (!error) { + error = e; + } + } + + kj::Promise forwardResponse(LegacyStream::Client callbacks) + try { + std::array buf; + while (true) { + if (auto got = TRY_AWAIT(workerSock.read(buf.data(), buf.size())); !got) { + break; + } else { + auto req = callbacks.feedRequest(); + req.initRaw(*got); + std::copy(buf.begin(), buf.begin() + *got, req.getRaw().begin()); + co_await req.send(); + } + } + co_await callbacks.syncRequest().send(); + } catch (...) { + onError(std::current_exception()); + } + + kj::Promise feed(FeedContext context) override + try { + auto bytes = context.getParams().getRaw(); + if (!error) { + TRY_AWAIT(workerSock.writeFull(bytes.begin(), bytes.size())); + } + } catch (...) { + onError(std::current_exception()); + } + + kj::Promise sync(SyncContext context) override + try { + TRY_AWAIT(logger->flush()); + if (error) { + RPC_FILL(context.initResults(), initResult, error); + } else { + context.initResults().initResult().setGood(); + } + } catch (...) { + RPC_FILL(context.initResults(), initResult, std::current_exception()); + } +}; + +struct LegacyBootImpl final : LegacyBoot::Server +{ + ref state; + bool used = false; + + LegacyBootImpl(TrustedFlag trusted, ref store) : state(make_ref(store, trusted)) {} + + kj::Promise init(InitContext context) override + try { + if (used) { + throw Error("connection already initialized"); + } + + auto prevLogger = logger; + logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger()); + + auto args = context.getParams(); + auto result = context.initResults().initResult().initGood(); + // We and the underlying store both need to trust the client for it to be trusted. + if (!state->trusted) { + result.setTrust(LegacyBoot::Trust::UNTRUSTED); + } else if (auto trust = TRY_AWAIT(state->store->isTrustedClient()); trust) { + result.setTrust(*trust ? LegacyBoot::Trust::TRUSTED : LegacyBoot::Trust::UNTRUSTED); + } else { + result.setTrust(LegacyBoot::Trust::UNKNOWN); + } + result.setVersion(PACKAGE_VERSION); + + auto [rpcSock, workerSock] = SocketPair::stream(); + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + + struct Request + { + AutoCloseFD fd; + kj::Own> signal; + + Request(AutoCloseFD fd, kj::Own> fulfiller) + : fd(std::move(fd)) + , signal(std::move(fulfiller)) + { + } + }; + + auto req = make_ref(std::move(rpcSock), std::move(pfp.fulfiller)); + + auto legacyThread = std::async(std::launch::async, [prevLogger, state{state}, req] { + AsyncIoRoot aio; + FdSource from(req->fd.get()); + FdSink to(req->fd.get()); + TunnelLogger logger(to, PROTOCOL_VERSION); + + try { + processLegacyRequests( + aio, prevLogger, &logger, state->store, from, to, state->trusted, PROTOCOL_VERSION + ); + } catch (Error & e) { + req->signal->fulfill(std::current_exception()); + } catch (std::bad_alloc & e) { + req->signal->fulfill(std::make_exception_ptr(Error("Lix daemon out of memory"))); + } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) + // TODO print stack trace to daemon log, maybe crash? + // boost stacktrace has from_current_exception (at a cost) with not-great symbolization, + // cpptrace has a *much* better symbolizer (at unknown cost) + req->signal->fulfill( + std::make_exception_ptr(Error( + "Unexpected exception on the Lix daemon; this is a bug in Lix.\n" + "We would appreciate a report of the circumstances it happened in at " + "https://git.lix.systems/lix-project/lix.\n%s: %s", + Uncolored(boost::core::demangle(typeid(e).name())), + e.what() + )) + ); + } catch (...) { + // TODO print stack trace to daemon log, maybe crash? + req->signal->fulfill( + std::make_exception_ptr(Error( + "Unexpected exception on the Lix daemon; this is a bug in Lix.\n" + "We would appreciate a report of the circumstances it happened in at " + "https://git.lix.systems/lix-project/lix.\n" + )) + ); + } + }); + + result.setRequestStream( + kj::heap( + state, std::move(pfp.promise), args.getReplyStream(), std::move(workerSock) + ) + .attach(std::move(legacyThread)) + ); + used = true; + } catch (...) { + RPC_FILL(context.initResults(), initResult, std::current_exception()); + } +}; + +struct BootstrapImpl final : Bootstrap::Server +{ + struct ProtocolEntry + { + std::string description; + std::function)> factory; + }; + + TrustedFlag trusted; + ref store; + std::map protocols; + bool used = false; + + BootstrapImpl(TrustedFlag trusted, ref store) : trusted(trusted), store(store) + { + if (store->isThreadSafe()) { + protocols.emplace( + rpc::daemon::UNSTABLE_LEGACY_TUNNELED, + ProtocolEntry{"tunneled legacy wire protocol", [](TrustedFlag trusted, ref store) { + return kj::heap(trusted, store); + }} + ); + } + } + + kj::Promise supported(SupportedContext context) override + { + if (!experimentalFeatureSettings.isEnabled(Xp::RpcSockets)) { + kj::throwFatalException( + kj::Exception( + kj::Exception::Type::UNIMPLEMENTED, "main", 0, kj::str("rpc sockets not enabled") + ) + ); + throw Error("rpc sockets not enabled"); + } + + auto result = context.initResults(); + auto protocols = result.initProtocols(this->protocols.size()); + for (auto [i, proto] : enumerate(this->protocols)) { + protocols[i].setId(proto.first); + protocols[i].setDescription(proto.second.description); + } + return kj::READY_NOW; + } + + kj::Promise request(RequestContext context) override + try { + auto id = rpc::to(context.getParams().getProtocol()); + if (used) { + throw Error("connection already initialized"); + } else if (const auto & protocol = get(protocols, id)) { + used = true; + context.initResults().initResult().setGood(protocol->factory(trusted, store)); + } else { + throw Error("unsupported protocol %s", id); + } + return kj::READY_NOW; + } catch (...) { + RPC_FILL(context.initResults(), initResult, std::current_exception()); + return kj::READY_NOW; + } +}; +} + +kj::Promise> +processConnection(ref store, kj::AsyncIoStream & connection, TrustedFlag trusted) +try { + // TODO trace encoders can do neat error info things, use them. we could stuff some serialized + // error struct into the remote trace field instead of using result types and get pipelineable + // calls out of it. needs more investigation to say if it's worth the possible reporting skew. + capnp::TwoPartyServer server{kj::heap(trusted, store)}; + + // NOTE we can't easily disconnect a peer without shutting shutting down the socket connection + // independently of capnp since capnp does not offer such functionality. shutting down sockets + // in this manner is very disruptive and pretty unreliable, so we will have to find some other + // way to disconnect clients. or we just don't do it because the DoS risk is not large anyway. + // + // since we have control over the promise we can have the following await finish early to stop + // processing events, and if we close the socket after that we've dropped the connection. this + // does not guarantee that responses have been sent though, so we can only do this on requests + // received *after* a fatal error response has been *sent*, inflicting per-operation overhead. + { + auto prevLogger = logger; + co_await server.accept(connection).exclusiveJoin(connection.whenWriteDisconnected()); + // NOTE: we do not flush the logger here because the connection is already closed! we only + // delete the non-local logger (if one was set) to ensure all rpc references were dropped. + if (prevLogger != logger) { + std::swap(prevLogger, logger); + delete prevLogger; + } + } + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} } diff --git a/lix/libstore/daemon.hh b/lix/libstore/daemon.hh index 31d3f1314..478b299c2 100644 --- a/lix/libstore/daemon.hh +++ b/lix/libstore/daemon.hh @@ -4,10 +4,14 @@ #include "lix/libutil/async.hh" #include "lix/libutil/serialise.hh" #include "lix/libstore/store-api.hh" +#include namespace nix::daemon { void processLegacyConnection( AsyncIoRoot & aio, ref store, FdSource & from, FdSink & to, TrustedFlag trusted ); + +kj::Promise> +processConnection(ref store, kj::AsyncIoStream & connection, TrustedFlag trusted); } diff --git a/lix/libstore/globals.cc b/lix/libstore/globals.cc index bd75b1495..31b463c41 100644 --- a/lix/libstore/globals.cc +++ b/lix/libstore/globals.cc @@ -68,10 +68,13 @@ Settings::Settings() { if (auto socketDirFromEnv = getEnvNonEmpty("LIX_DAEMON_SOCKET_DIR")) { nixDaemonSockets_ = daemon::supportedProtocols(*socketDirFromEnv); + socketsPath = *socketDirFromEnv; } else if (auto socketPathFromEnv = getEnvNonEmpty("NIX_DAEMON_SOCKET_PATH")) { nixDaemonSockets_ = {{canonPath(*socketPathFromEnv), daemon::Protocol::LEGACY_COMBINED}}; + socketsPath = *socketPathFromEnv; } else { auto baseDir = nixStateDir + DEFAULT_SOCKET_DIR; + socketsPath = baseDir; // this should always match the list of sockets created by daemonLoop and the socket units nixDaemonSockets_ = daemon::supportedProtocols(baseDir); } @@ -532,11 +535,13 @@ void initLibStore() namespace daemon { static inline constexpr auto LEGACY_COMBINED_STR = "legacy-combined"; static inline constexpr auto LEGACY_STR = "legacy"; +static inline constexpr auto LIX_XP_1_STR = "lix-xp-1"; std::list supportedProtocols(std::optional prefix) { const Path base = prefix ? *prefix + "/" : ""; return { + {base + "lix-xp-1/socket", Protocol::RPC_V1}, {base + "socket", Protocol::LEGACY}, }; } @@ -548,6 +553,8 @@ std::string_view daemon::Protocol::id() const return LEGACY_COMBINED_STR; case LEGACY: return LEGACY_STR; + case RPC_V1: + return LIX_XP_1_STR; } } @@ -558,6 +565,8 @@ daemon::Protocol getProtocol(std::string_view protocol, std::optional return {prefix ? Path(*prefix) : "", Protocol::LEGACY_COMBINED}; } else if (protocol == LEGACY_STR) { return {base + "socket", Protocol::LEGACY_COMBINED}; + } else if (protocol == LIX_XP_1_STR) { + return {base + "lix-xp-1/socket", Protocol::RPC_V1}; } else { throw Error("unsupported daemon protocol %s", protocol); } diff --git a/lix/libstore/globals.hh b/lix/libstore/globals.hh index db3e3c2f0..896891133 100644 --- a/lix/libstore/globals.hh +++ b/lix/libstore/globals.hh @@ -18,6 +18,7 @@ struct Protocol enum Type : int { LEGACY_COMBINED, LEGACY, + RPC_V1, } type; /// external identifier of the protocol (eg for `protocol` store parameters) @@ -128,6 +129,11 @@ public: */ Path nixStateDir; + /** + * The directory where sockets are stored. + */ + Path socketsPath; + /** * The directory where system configuration files are stored. */ diff --git a/lix/libstore/meson.build b/lix/libstore/meson.build index 0ec68e41e..ffba26878 100644 --- a/lix/libstore/meson.build +++ b/lix/libstore/meson.build @@ -13,6 +13,7 @@ libstore_rpc = [] libstore_rpc_files = files( # keep-sorted start + 'daemon.capnp', 'types.capnp', # keep-sorted end ) diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh index 84f8157ec..1a81b876f 100644 --- a/lix/libstore/remote-store.hh +++ b/lix/libstore/remote-store.hh @@ -178,7 +178,7 @@ public: struct Connection; - kj::Promise>> openConnectionForDaemonForwarding(); + virtual kj::Promise>> openConnectionForDaemonForwarding(); protected: @@ -186,7 +186,7 @@ protected: kj::Promise>> openAndInitConnection(); - kj::Promise> initConnection(Connection & conn); + virtual kj::Promise> initConnection(Connection & conn); ref> connections; diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index 9d417deb3..91b6cf8fc 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -1,15 +1,23 @@ #include "lix/libstore/uds-remote-store.hh" +#include "daemon.capnp.h" #include "globals.hh" +#include "libstore/daemon.hh" +#include "libutil/logging-rpc.hh" +#include "libutil/rpc.hh" #include "lix/libutil/async.hh" #include "lix/libutil/error.hh" #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/result.hh" +#include "lix/libutil/rpc.hh" +#include "lix/libutil/types-rpc.hh" #include "lix/libutil/strings.hh" #include "lix/libutil/unix-domain-socket.hh" #include "lix/libstore/worker-protocol.hh" #include +#include #include +#include #include #include #include @@ -76,13 +84,10 @@ static bool tryToConnect(AutoCloseFD & sockFD, const daemon::Protocol & socket) } } -kj::Promise>> UDSRemoteStore::openConnection() +kj::Promise>> UDSRemoteStore::openConnection(bool allowRPC) try { auto conn = make_ref(); - /* Connect to a daemon that does the privileged work for us. */ - conn->fd = createUnixDomainSocket(); - std::list candidates; if (path) { @@ -113,8 +118,31 @@ try { } for (const auto & path : candidates) { + if (!allowRPC) { + switch (path.type) { + case daemon::Protocol::LEGACY_COMBINED: + case daemon::Protocol::LEGACY: + break; + case daemon::Protocol::RPC_V1: + continue; + } + } + + /* Connect to a daemon that does the privileged work for us. */ + conn->fd = createUnixDomainSocket(); + if (tryToConnect(conn->fd, path)) { conn->startTime = std::chrono::steady_clock::now(); + + // NOTE we do all this setup *here* instead of in initConnection because we want to + // provide graceful fallback during the transition period to rpc. since clients and + // daemons can be updated independently we can never be sure that the rpc protocols + // we want to use are supported on both sides without checking first, and sadly the + // only place we can do such checks without disturbing fallback behavior is *here*. + if (path.type == daemon::Protocol::RPC_V1 && !TRY_AWAIT(prepareRpcConnection(*conn))) { + continue; + } + co_return conn; } } @@ -127,6 +155,123 @@ try { co_return result::current_exception(); } +kj::Promise>> UDSRemoteStore::openConnectionForDaemonForwarding() +{ + return openConnection(false); +} + +kj::Promise>> UDSRemoteStore::openConnection() +{ + return openConnection(true); +} + +kj::Promise> UDSRemoteStore::prepareRpcConnection(Connection & con) +try { + auto rpcStream = + AIO().lowLevelProvider.wrapSocketFd(con.fd.get(), kj::LowLevelAsyncIoProvider::TAKE_OWNERSHIP); + con.fd.release(); + auto [proxyAsync, proxySync] = SocketPair::stream(); + con.fd = std::move(proxySync); + auto client = make_box_ptr(*rpcStream); + auto bootstrap = client->bootstrap().castAs(); + + // simplistic setup until we have a reasonable state to work with: talk only to + // daemons that support the exact protocol list we support, including the uuid. + { + auto supported = co_await bootstrap.supportedRequest().send(); + auto supportedProtos = supported.getProtocols(); + debug("remote advertised %s", supported.toString().flatten().cStr()); + if (supportedProtos.size() != 1 + && supportedProtos[0].getId() != rpc::daemon::UNSTABLE_LEGACY_TUNNELED) + { + co_return false; + } + } + + con.rpc = std::make_shared(RpcState{ + .rpcStream = std::move(rpcStream), + .proxySock = make_box_ptr(std::move(proxyAsync)), + .client = std::move(client), + .loggerActivity = logger->startActivity(lvlDebug, actUnknown, "daemon connection"), + .requestStream = nullptr, + .forwarder = nullptr, + }); + + auto bootstrapReq = bootstrap.requestRequest(); + bootstrapReq.setClientInfo(PACKAGE_STRING); + bootstrapReq.setProtocol(rpc::daemon::UNSTABLE_LEGACY_TUNNELED); + auto legacyBoot = TRY_AWAIT_RPC(bootstrapReq.send()).castAs(); + auto initReq = legacyBoot.initRequest(); + initReq.setLogger(kj::heap(con.rpc->loggerActivity)); + initReq.setReplyStream(kj::heap(*con.rpc)); + + auto initResp = initReq.send(); + auto initResult = TRY_AWAIT_RPC(initResp); + con.remoteTrustsUs = initResult.getTrust() == rpc::daemon::LegacyBoot::Trust::TRUSTED + ? std::optional{Trusted} + : initResult.getTrust() == rpc::daemon::LegacyBoot::Trust::UNTRUSTED ? std::optional{NotTrusted} + : std::nullopt; + con.daemonVersion = PROTOCOL_VERSION; + con.daemonNixVersion = rpc::to(initResult.getVersion()); + con.store = this; + con.rpc->requestStream = initResult.getRequestStream(); + con.rpc->forwarder = con.rpc->forwardRequests(); + + co_return true; +} catch (std::exception & e) { // NOLINT(lix-foreign-exceptions): just fall back to legacy for now + debug("rpc connection failed: %s", e.what()); + co_return false; +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise UDSRemoteStore::RpcState::forwardRequests() +try { + std::array buf; + while (true) { + if (auto got = TRY_AWAIT(proxySock->read(buf.data(), buf.size())); !got) { + break; + } else { + auto req = requestStream.feedRequest(); + req.initRaw(*got); + std::copy(buf.begin(), buf.begin() + *got, req.getRaw().begin()); + co_await req.send(); + } + } + co_await requestStream.syncRequest().send(); +} catch (...) { + ignoreExceptionExceptInterrupt(); + error = std::current_exception(); +} + +kj::Promise UDSRemoteStore::LegacyStreamProxy::feed(FeedContext context) +try { + if (!state.error) { + auto bytes = context.getParams().getRaw(); + TRY_AWAIT(state.proxySock->writeFull(bytes.begin(), bytes.size())); + } +} catch (...) { + state.error = std::current_exception(); +} + +kj::Promise UDSRemoteStore::LegacyStreamProxy::sync(SyncContext context) +{ + if (state.error) { + RPC_FILL(context.initResults(), initResult, state.error); + } else { + context.initResults().initResult().setGood(); + } + return kj::READY_NOW; +} + +kj::Promise> UDSRemoteStore::initConnection(RemoteStore::Connection & conn) +{ + if (dynamic_cast(conn).rpc) { + return setOptions(conn); + } else { + return RemoteStore::initConnection(conn); + } +} kj::Promise> UDSRemoteStore::addIndirectRoot(const Path & path) try { diff --git a/lix/libstore/uds-remote-store.hh b/lix/libstore/uds-remote-store.hh index a4d04551d..e761dd5d8 100644 --- a/lix/libstore/uds-remote-store.hh +++ b/lix/libstore/uds-remote-store.hh @@ -1,10 +1,18 @@ #pragma once ///@file +#include "libutil/box_ptr.hh" +#include "libutil/logging-rpc.hh" +#include "libutil/logging.hh" #include "lix/libstore/remote-store.hh" #include "lix/libstore/remote-store-connection.hh" #include "lix/libstore/indirect-root-store.hh" #include "lix/libutil/async-io.hh" +#include "lix/libstore/daemon.capnp.h" +#include +#include +#include +#include namespace nix { @@ -28,6 +36,7 @@ struct UDSRemoteStoreConfig : virtual LocalFSStoreConfig, virtual RemoteStoreCon The provided path will be used *unmodified* to locate the combined daemon socket. - `legacy`: legacy wire protocol using a single socket, but the path is used as the base directory for protocol-dependent socket lookup (appending `/socket` to path) + - `lix-xp-1`: experimental RPC protocol. Will use the path as directory of sockets. Also supports the special value `any` to try *all* known protocols using the provided path as the *base* directory for sockets. Unlike `legacy-combined` this will append a @@ -95,10 +104,37 @@ public: kj::Promise> addIndirectRoot(const Path & path) override; private: + struct RpcState + { + kj::Own rpcStream; + box_ptr proxySock; + box_ptr client; + Activity loggerActivity; + rpc::daemon::LegacyStream::Client requestStream; + + kj::Promise forwarder; + std::exception_ptr error; + + kj::Promise forwardRequests(); + }; + + // this class is lifetime-bound to its rpc state; requestStream holds onto one of + // these proxies. ideally we'd use RpcState itself for this, but kj does not have + // shared pointers and cannot provide capabilities through anything except `Own`. + struct LegacyStreamProxy final : rpc::daemon::LegacyStream::Server + { + RpcState & state; + + LegacyStreamProxy(RpcState & state) : state(state) {} + + kj::Promise feed(FeedContext context) override; + kj::Promise sync(SyncContext context) override; + }; struct Connection : RemoteStore::Connection { AutoCloseFD fd; + std::shared_ptr rpc; int getFD() const override { @@ -106,7 +142,12 @@ private: } }; + kj::Promise>> openConnection(bool allowRPC); + + kj::Promise>> openConnectionForDaemonForwarding() override; kj::Promise>> openConnection() override; + kj::Promise> prepareRpcConnection(Connection & con); + kj::Promise> initConnection(RemoteStore::Connection & conn) override; std::optional path; }; diff --git a/lix/libutil/experimental-features/rpc-sockets.md b/lix/libutil/experimental-features/rpc-sockets.md new file mode 100644 index 000000000..e9921cb6d --- /dev/null +++ b/lix/libutil/experimental-features/rpc-sockets.md @@ -0,0 +1,8 @@ +--- +name: rpc-sockets +internalName: RpcSockets +--- + +Enable the experimental RPC sockets. This makes the `lix-xp-1` daemon protocol available for clients. +Note that this feature only enables the *daemon* side of this features; clients always support all of +the protocols (although by default only the legacy sockets are tried without explicit configuration). diff --git a/lix/libutil/meson.build b/lix/libutil/meson.build index a2ca65a8a..1892f6c91 100644 --- a/lix/libutil/meson.build +++ b/lix/libutil/meson.build @@ -167,6 +167,7 @@ experimental_feature_definitions = files( 'experimental-features/pipe-operator.md', 'experimental-features/read-only-local-store.md', 'experimental-features/repl-automation.md', + 'experimental-features/rpc-sockets.md', # keep-sorted end ) diff --git a/lix/nix/daemon.cc b/lix/nix/daemon.cc index bba0b09fb..c8a613fae 100644 --- a/lix/nix/daemon.cc +++ b/lix/nix/daemon.cc @@ -9,8 +9,10 @@ #include "lix/libutil/async-collect.hh" #include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" +#include "lix/libutil/config.hh" #include "lix/libutil/current-process.hh" #include "lix/libutil/error.hh" +#include "lix/libutil/experimental-features.hh" #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/logging.hh" #include "lix/libutil/processes.hh" @@ -441,6 +443,16 @@ try { std::list> sockets; for (auto & socket : settings.nixDaemonSockets()) { + switch (socket.type) { + case daemon::Protocol::LEGACY_COMBINED: + case daemon::Protocol::LEGACY: + break; + case daemon::Protocol::RPC_V1: + if (!experimentalFeatureSettings.isEnabled(Xp::RpcSockets)) { + continue; + } + break; + } createDirs(dirOf(socket.path)); sockets.emplace_back(socket, createUnixDomainSocket(socket.path, 0666)); } @@ -533,6 +545,11 @@ static void daemonInstance( processLegacyConnection(aio, store, from, to, trusted); break; } + case daemon::Protocol::RPC_V1: { + auto stream = AIO().lowLevelProvider.wrapSocketFd(connectionFd); + aio.blockOn(processConnection(store, *stream, trusted)); + break; + } } } diff --git a/misc/systemd/meson.build b/misc/systemd/meson.build index 091f7d1a8..cd7ceb0a6 100644 --- a/misc/systemd/meson.build +++ b/misc/systemd/meson.build @@ -3,7 +3,8 @@ # # specs are (protocol, socket path, unit name, unit description, slice name) lix_daemon_sockets = [ - ['legacy-combined', 'socket', 'nix-daemon', 'legacy', 'legacy'], + ['lix-xp-1', 'lix-xp-1/socket', 'lix-daemon-xp1', 'experimental RPC', 'xp1'], + ['legacy-combined', 'socket', 'nix-daemon', 'legacy', 'legacy'], ] lix_daemon_socket_units = [] diff --git a/tests/functional2/daemon/test_connect.py b/tests/functional2/daemon/test_connect.py index 8b7db91cf..7ed6fdc39 100644 --- a/tests/functional2/daemon/test_connect.py +++ b/tests/functional2/daemon/test_connect.py @@ -27,7 +27,8 @@ def test_connection_order_plain(nix: Nix): def test_connection_order_any(nix: Nix): assert _observe_socket_order(nix, f"unix://{nix.env.dirs.home}?protocol=any") == [ - str(nix.env.dirs.home / "socket") + str(nix.env.dirs.home / "lix-xp-1/socket"), + str(nix.env.dirs.home / "socket"), ] @@ -43,7 +44,11 @@ def test_connection_protocol_any_not_standalone(nix: Nix): assert "unsupported daemon protocol any" in cmd.stderr_s -_protocols: dict[str, str] = {"legacy-combined": ".", "legacy": "socket"} +_protocols: dict[str, str] = { + "lix-xp-1": "lix-xp-1/socket", + "legacy-combined": ".", + "legacy": "socket", +} @pytest.mark.parametrize( diff --git a/tests/functional2/daemon/test_rpc_setup.py b/tests/functional2/daemon/test_rpc_setup.py new file mode 100644 index 000000000..7eba64826 --- /dev/null +++ b/tests/functional2/daemon/test_rpc_setup.py @@ -0,0 +1,19 @@ +import pytest + +from testlib.fixtures.nix import Nix, NixDaemon + + +def test_legacy_sockets_always_appear(nix: Nix, daemon: NixDaemon): + sockets_dir = nix.env.dirs.nix_state_dir / "daemon-socket" + with daemon(nix): + assert sockets_dir.exists() + assert sockets_dir.is_dir() + assert (sockets_dir / "socket").exists() + assert (sockets_dir / "socket").is_socket() + + +@pytest.mark.parametrize("daemon", ["legacy"], indirect=True) +def test_xp_sockets_dont_always_appear(nix: Nix, daemon: NixDaemon): + sockets_dir = nix.env.dirs.nix_state_dir / "daemon-socket" + with daemon(nix): + assert list(sockets_dir.glob("./**")) == [sockets_dir, sockets_dir / "socket"] diff --git a/tests/functional2/testlib/fixtures/nix.py b/tests/functional2/testlib/fixtures/nix.py index 0e2f98f53..1ccd79a21 100644 --- a/tests/functional2/testlib/fixtures/nix.py +++ b/tests/functional2/testlib/fixtures/nix.py @@ -342,10 +342,14 @@ def nix(tmp_path: Path, env: ManagedEnv, logger: logging.Logger) -> Generator[Ni type NixDaemon = Callable[..., contextlib.AbstractAsyncContextManager[Nix]] -type NixDaemonProtocol = Literal["legacy-combined", "legacy"] +# NOTE: the order of items here is important. the daemon fixture requires +# the first item in this list to be the last socket opened by the daemon. +type NixDaemonProtocol = Literal["legacy-combined", "legacy", "lix-xp-1"] daemon_protocols: list[NixDaemonProtocol] = get_args(NixDaemonProtocol.__value__) +_daemon_protocol_xp_features: dict[NixDaemon, list[str]] = {"lix-xp-1": ["rpc-sockets"]} + # paramterize every daemon tests to run using all supported nix protocols @pytest.fixture(params=daemon_protocols) @@ -368,9 +372,11 @@ def daemon(request: pytest.FixtureRequest) -> NixDaemon: daemon.settings["trusted-users"] = [] daemon.settings.store = f"local?root={nix.env.dirs.test_root}" daemon.settings.update(settings) + if requires_features := _daemon_protocol_xp_features.get(protocol): + daemon.settings.add_xp_feature(*requires_features) sockets_dir = Path(daemon.env.dirs.nix_state_dir) / "daemon-socket" - sockets = [sockets_dir / "socket"] + sockets = [sockets_dir / "socket", sockets_dir / "lix-xp-1/socket"] for p in sockets: p.unlink(missing_ok=True) @@ -385,7 +391,10 @@ def daemon(request: pytest.FixtureRequest) -> NixDaemon: daemon.logger.error("daemon exited unexpectedly") # wait for daemon to come up. this may take a while under load. - while not all(s.exists() for s in sockets): + # we wait only for the first socket in the list, expecting that + # it'll be the last one opened by the daemon. this is to ensure + # that we always return correctly regardless of rpc xp settings + while not sockets[0].exists(): if status := proc.wait(0.01): log_daemon_result(status, logging.ERROR) raise RuntimeError("daemon exited during startup") diff --git a/tests/nixos/default.nix b/tests/nixos/default.nix index 15fe79cc8..fa7b4e2df 100644 --- a/tests/nixos/default.nix +++ b/tests/nixos/default.nix @@ -34,7 +34,9 @@ let in { "nix-daemon@" = daemonConfig; + "lix-daemon-lix-xp-1@" = daemonConfig; }; + systemd.sockets."lix-daemon-lix-xp-1".wantedBy = [ "sockets.target" ]; }; _module.args.nixpkgs = nixpkgs; _module.args.system = system;