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

treewide: pour the rpc foundations

Browse Source 
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 <git@piegames.de>
Change-Id: I85a96ccb700b91190c1eb37154bcc6ae1c03401a
This commit is contained in:
eldritch horrors
2026-05-06 10:55:08 +00:00
co-authored by piegames
parent b2c95d41c9
commit d0a4b55a0e
18 changed files with 624 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libmain/progress-bar.cc
+3 -1
View File
@@ -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;
lix/libstore/daemon.capnp
+58
View File
@@ -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);
}
lix/libstore/daemon.cc
+283
View File
@@ -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 <boost/core/demangle.hpp>
#include <capnp/rpc-twoparty.h>
#include <cstdint>
#include <ctime>
#include <kj/encoding.h>
#include <kj/exception.h>
#include <kj/memory.h>
#include <sstream>
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> store;
TrustedFlag trusted;
LegacyState(ref<Store> store, TrustedFlag trusted) : store(store), trusted(trusted) {}
};
struct RequestStreamImpl final : LegacyStream::Server
{
ref<LegacyState> state;
std::exception_ptr error;
AsyncFdIoStream workerSock;
kj::Promise<void> responseForwarder;
RequestStreamImpl(
ref<LegacyState> state,
kj::Promise<std::exception_ptr> error,
LegacyStream::Client callbacks,
AutoCloseFD workerFd
)
: state(state)
, workerSock(std::move(workerFd))
, responseForwarder(forwardResponse(callbacks).exclusiveJoin(
error.then([&](std::exception_ptr e) -> kj::Promise<void> {
onError(e);
return kj::NEVER_DONE;
})
))
{
}
void onError(std::exception_ptr e)
{
if (!error) {
error = e;
}
}
kj::Promise<void> forwardResponse(LegacyStream::Client callbacks)
try {
std::array<char, 8192> 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<void> 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<void> 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<LegacyState> state;
bool used = false;
LegacyBootImpl(TrustedFlag trusted, ref<Store> store) : state(make_ref<LegacyState>(store, trusted)) {}
kj::Promise<void> 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<std::exception_ptr>();
struct Request
{
AutoCloseFD fd;
kj::Own<kj::CrossThreadPromiseFulfiller<std::exception_ptr>> signal;
Request(AutoCloseFD fd, kj::Own<kj::CrossThreadPromiseFulfiller<std::exception_ptr>> fulfiller)
: fd(std::move(fd))
, signal(std::move(fulfiller))
{
}
};
auto req = make_ref<Request>(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<RequestStreamImpl>(
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<rpc::daemon::Protocol::Client(TrustedFlag, ref<Store>)> factory;
};
TrustedFlag trusted;
ref<Store> store;
std::map<kj::StringPtr, ProtocolEntry> protocols;
bool used = false;
BootstrapImpl(TrustedFlag trusted, ref<Store> store) : trusted(trusted), store(store)
{
if (store->isThreadSafe()) {
protocols.emplace(
rpc::daemon::UNSTABLE_LEGACY_TUNNELED,
ProtocolEntry{"tunneled legacy wire protocol", [](TrustedFlag trusted, ref<Store> store) {
return kj::heap<LegacyBootImpl>(trusted, store);
}}
);
}
}
kj::Promise<void> 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<void> request(RequestContext context) override
try {
auto id = rpc::to<std::string>(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<Result<void>>
processConnection(ref<Store> 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<BootstrapImpl>(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();
}
}
lix/libstore/daemon.hh
+4
View File
@@ -4,10 +4,14 @@
#include "lix/libutil/async.hh"
#include "lix/libutil/serialise.hh"
#include "lix/libstore/store-api.hh"
#include <kj/async-io.h>
namespace nix::daemon {
void processLegacyConnection(
AsyncIoRoot & aio, ref<Store> store, FdSource & from, FdSink & to, TrustedFlag trusted
);
kj::Promise<Result<void>>
processConnection(ref<Store> store, kj::AsyncIoStream & connection, TrustedFlag trusted);
}
lix/libstore/globals.cc
+9
View File
@@ -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<daemon::Protocol> supportedProtocols(std::optional<PathView> 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<PathView>
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);
}
lix/libstore/globals.hh
+6
View File
@@ -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.
*/
lix/libstore/meson.build
+1
View File
@@ -13,6 +13,7 @@ libstore_rpc = []
libstore_rpc_files = files(
# keep-sorted start
'daemon.capnp',
'types.capnp',
# keep-sorted end
)
lix/libstore/remote-store.hh
+2 -2
View File
@@ -178,7 +178,7 @@ public:
struct Connection;
kj::Promise<Result<ref<Connection>>> openConnectionForDaemonForwarding();
virtual kj::Promise<Result<ref<Connection>>> openConnectionForDaemonForwarding();
protected:
@@ -186,7 +186,7 @@ protected:
kj::Promise<Result<ref<Connection>>> openAndInitConnection();
kj::Promise<Result<void>> initConnection(Connection & conn);
virtual kj::Promise<Result<void>> initConnection(Connection & conn);
ref<Pool<Connection>> connections;
lix/libstore/uds-remote-store.cc
+149 -4
View File
@@ -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 <algorithm>
#include <capnp/rpc-twoparty.h>
#include <cerrno>
#include <kj/encoding.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/socket.h>
@@ -76,13 +84,10 @@ static bool tryToConnect(AutoCloseFD & sockFD, const daemon::Protocol & socket)
}
}
kj::Promise<Result<ref<RemoteStore::Connection>>> UDSRemoteStore::openConnection()
kj::Promise<Result<ref<RemoteStore::Connection>>> UDSRemoteStore::openConnection(bool allowRPC)
try {
auto conn = make_ref<Connection>();
/* Connect to a daemon that does the privileged work for us. */
conn->fd = createUnixDomainSocket();
std::list<daemon::Protocol> 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<Result<ref<RemoteStore::Connection>>> UDSRemoteStore::openConnectionForDaemonForwarding()
{
return openConnection(false);
}
kj::Promise<Result<ref<RemoteStore::Connection>>> UDSRemoteStore::openConnection()
{
return openConnection(true);
}
kj::Promise<Result<bool>> 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<capnp::TwoPartyClient>(*rpcStream);
auto bootstrap = client->bootstrap().castAs<rpc::daemon::Bootstrap>();
// 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>(RpcState{
.rpcStream = std::move(rpcStream),
.proxySock = make_box_ptr<AsyncFdIoStream>(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<rpc::daemon::LegacyBoot>();
auto initReq = legacyBoot.initRequest();
initReq.setLogger(kj::heap<rpc::log::RpcLoggerServer>(con.rpc->loggerActivity));
initReq.setReplyStream(kj::heap<LegacyStreamProxy>(*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<std::string>(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<void> UDSRemoteStore::RpcState::forwardRequests()
try {
std::array<char, 8192> 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<void> 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<void> 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<Result<void>> UDSRemoteStore::initConnection(RemoteStore::Connection & conn)
{
if (dynamic_cast<Connection &>(conn).rpc) {
return setOptions(conn);
} else {
return RemoteStore::initConnection(conn);
}
}
kj::Promise<Result<void>> UDSRemoteStore::addIndirectRoot(const Path & path)
try {
lix/libstore/uds-remote-store.hh
+41
View File
@@ -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 <capnp/rpc-twoparty.h>
#include <kj/async-io.h>
#include <kj/async.h>
#include <memory>
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<Result<void>> addIndirectRoot(const Path & path) override;
private:
struct RpcState
{
kj::Own<kj::AsyncIoStream> rpcStream;
box_ptr<AsyncFdIoStream> proxySock;
box_ptr<capnp::TwoPartyClient> client;
Activity loggerActivity;
rpc::daemon::LegacyStream::Client requestStream;
kj::Promise<void> forwarder;
std::exception_ptr error;
kj::Promise<void> 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<void> feed(FeedContext context) override;
kj::Promise<void> sync(SyncContext context) override;
};
struct Connection : RemoteStore::Connection
{
AutoCloseFD fd;
std::shared_ptr<RpcState> rpc;
int getFD() const override
{
@@ -106,7 +142,12 @@ private:
}
};
kj::Promise<Result<ref<RemoteStore::Connection>>> openConnection(bool allowRPC);
kj::Promise<Result<ref<RemoteStore::Connection>>> openConnectionForDaemonForwarding() override;
kj::Promise<Result<ref<RemoteStore::Connection>>> openConnection() override;
kj::Promise<Result<bool>> prepareRpcConnection(Connection & con);
kj::Promise<Result<void>> initConnection(RemoteStore::Connection & conn) override;
std::optional<std::string> path;
};
lix/libutil/experimental-features/rpc-sockets.md
+8
View File
@@ -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).
lix/libutil/meson.build
+1
View File
@@ -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
)
lix/nix/daemon.cc
+17
View File
@@ -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<std::pair<daemon::Protocol, AutoCloseFD>> 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;
}
}
}
misc/systemd/meson.build
+2 -1
View File
@@ -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 = []
tests/functional2/daemon/test_connect.py
+7 -2
View File
@@ -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(
tests/functional2/daemon/test_rpc_setup.py
+19
View File
@@ -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"]
tests/functional2/testlib/fixtures/nix.py
+12 -3
View File
@@ -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")
tests/nixos/default.nix
+2
View File
@@ -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;