diff --git a/lix/legacy/build-remote.cc b/lix/legacy/build-remote.cc index d0fb8f61a..47ddba0ad 100644 --- a/lix/legacy/build-remote.cc +++ b/lix/legacy/build-remote.cc @@ -5,6 +5,7 @@ #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/logging-rpc.hh" #include "lix/libutil/logging.hh" +#include "lix/libutil/result.hh" #include "lix/libutil/rpc.hh" #include "lix/libutil/types-rpc.hh" // IWYU pragma: keep #include "lix/libutil/types.hh" @@ -50,7 +51,7 @@ struct Instance final : rpc::build_remote::HookInstance::Server kj::Promise init(InitContext context) override; - kj::Promise buildImpl(BuildContext context); + kj::Promise> buildImpl(BuildContext context); kj::Promise build(BuildContext context) override; }; } @@ -261,7 +262,7 @@ struct AcceptedBuild final : rpc::build_remote::HookInstance::AcceptedBuild::Ser { } - kj::Promise runImpl(RunContext context); + kj::Promise> runImpl(RunContext context); kj::Promise run(RunContext context) override; }; @@ -391,16 +392,14 @@ kj::Promise Instance::init(InitContext context) initPlugins(); initialized = true; - - context.getResults().initResult().setGood(); } catch (...) { - RPC_FILL(context.getResults(), initResult, std::current_exception()); + rpc::rethrow_as_rpc_error(); } return kj::READY_NOW; } -kj::Promise Instance::buildImpl(BuildContext context) +kj::Promise> Instance::buildImpl(BuildContext context) try { if (!initialized) { throw Error("build hook not fully initialized"); @@ -427,8 +426,8 @@ try { debug("got %d remote builders", machines.size()); if (machines.empty()) { - context.getResults().initResult().initGood().setDeclinePermanently(); - co_return; + context.getResults().initResult().setDeclinePermanently(); + co_return result::success(); } auto amWilling = context.getParams().getAmWilling(); @@ -444,21 +443,23 @@ try { if (auto immediateResponse = std::get_if(&result)) { switch (*immediateResponse) { case BuildRejected::Temporarily: - context.getResults().initResult().initGood().setPostpone(); - co_return; + context.getResults().initResult().setPostpone(); + co_return result::success(); case BuildRejected::Permanently: - context.getResults().initResult().initGood().setDecline(); - co_return; + context.getResults().initResult().setDecline(); + co_return result::success(); } } auto builder = std::get_if(&result); assert(builder); - auto ac = context.getResults().initResult().initGood().initAccept(); + auto ac = context.getResults().initResult().initAccept(); ac.setMachine(kj::heap(store, drvPath, std::move(*builder))); + + co_return result::success(); } catch (...) { - RPC_FILL(context.getResults(), initResult, std::current_exception()); + co_return result::current_exception(); } kj::Promise Instance::build(BuildContext context) @@ -467,11 +468,12 @@ try { throw Error("build hooks can only accept a single job"); } used = true; // lock out other rpc calls during processing - co_await buildImpl(context); + auto result = co_await buildImpl(context); TRY_AWAIT(logger->flush()); - used = context.getResults().getResult().getGood().isAccept(); + used = result.has_value() && context.getResults().getResult().isAccept(); + result.value(); } catch (...) { - RPC_FILL(context.getResults(), getResult, std::current_exception()); + rpc::rethrow_as_rpc_error(); } kj::Promise AcceptedBuild::run(RunContext context) @@ -489,14 +491,15 @@ kj::Promise AcceptedBuild::run(RunContext context) throw Error("build hooks builds are single-use items"); } used = true; - co_await runImpl(context); + auto result = co_await runImpl(context); TRY_AWAIT(logger->flush()); + result.value(); } catch (...) { - RPC_FILL(context.getResults(), getResult, std::current_exception()); + rpc::rethrow_as_rpc_error(); } } -kj::Promise AcceptedBuild::runImpl(RunContext context) +kj::Promise> AcceptedBuild::runImpl(RunContext context) { try { auto logHandler = builder.startLogThread( @@ -618,10 +621,9 @@ kj::Promise AcceptedBuild::runImpl(RunContext context) // drop store connection, let log handler process any remaining input builder.sshStore = nullptr; TRY_AWAIT(logHandler); - - context.getResults().initResult().setGood(); + co_return result::success(); } catch (...) { - RPC_FILL(context.getResults(), initResult, std::current_exception()); + co_return result::current_exception(); } } diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index a3b19cf98..f0f5e312f 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1094,8 +1094,8 @@ try { RPC_FILL(buildReq, setNeededSystem, drv->platform); RPC_FILL(buildReq, initDrvPath, drvPath, worker.store); RPC_FILL(buildReq, initRequiredFeatures, parsedDrv->getRequiredSystemFeatures()); - auto buildRespPromise = buildReq.send(); - auto buildResp = TRY_AWAIT_RPC(buildRespPromise); + auto buildRespV = TRY_AWAIT_RPC(buildReq.send()); + auto buildResp = buildRespV.getResult(); debug("hook reply is '%1%'", buildResp.toString().flatten().cStr()); @@ -1139,7 +1139,7 @@ try { RPC_FILL(runReq, setDescription, buildDescription()); } - auto runPromise = runReq.send(); + auto runPromise = LIX_WRAP_RPC_PROMISE_V1(runReq.send()); // build via hook is now properly running. wait for it to finish actLock.reset(); @@ -1150,8 +1150,8 @@ try { wrapChildHandler(runPromise.then([&](auto result) -> kj::Promise> { try { std::shared_ptr remoteError; - if (result.getResult().isBad()) { - remoteError = std::make_shared(from(result.getResult().getBad())); + if (result.has_error()) { + remoteError = std::make_shared(detail::wrap_exception_as_lix(result.error())); logErrorInfo(remoteError->info().level, remoteError->info()); } // close the rpc connection to have the hook exit diff --git a/lix/libstore/build/hook-instance.capnp b/lix/libstore/build/hook-instance.capnp index 40e60c870..f19cfde31 100644 --- a/lix/libstore/build/hook-instance.capnp +++ b/lix/libstore/build/hook-instance.capnp @@ -8,14 +8,14 @@ using Types = import "/lix/libutil/types.capnp"; using Log = import "/lix/libutil/logging.capnp"; using StoreTypes = import "/lix/libstore/types.capnp"; -interface HookInstance { +interface HookInstance $Types.throws(Types.v1Errors) { interface AcceptedBuild { run @0 ( logger :Log.LogStream, inputs :List(StoreTypes.StorePath), # actual a set wantedOutputs :List(Data), # actually StringSet description :Text, # root activity description for this build - ) -> (result :Types.ResultV); + ); } struct BuildResponse { @@ -32,11 +32,11 @@ interface HookInstance { init @0 ( logger :Log.LogStream, settings :Types.Settings, - ) -> (result :Types.ResultV); + ); build @1 ( amWilling :Bool, neededSystem :Data, drvPath :StoreTypes.StorePath, requiredFeatures :List(Data), - ) -> (result :Types.Result(BuildResponse)); + ) -> (result :BuildResponse); } diff --git a/lix/libstore/daemon.capnp b/lix/libstore/daemon.capnp index 11fc11c39..2c9a9999c 100644 --- a/lix/libstore/daemon.capnp +++ b/lix/libstore/daemon.capnp @@ -21,7 +21,7 @@ interface Bootstrap { request @1 ( clientInfo :Text, protocol :Text, - ) -> (result :T.Result(Protocol)); + ) -> (result :Protocol); } interface Protocol { @@ -32,7 +32,7 @@ interface Protocol { # 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) { +interface LegacyBoot extends(Protocol) $T.throws(T.v1Errors) { enum Trust { unknown @0; untrusted @1; @@ -42,7 +42,7 @@ interface LegacyBoot extends(Protocol) { init @0 ( logger :Log.LogStream, replyStream :LegacyStream, - ) -> (result :T.Result(InitResult)); + ) -> (result :InitResult); struct InitResult { requestStream @0 :LegacyStream; @@ -51,8 +51,8 @@ interface LegacyBoot extends(Protocol) { } } -interface LegacyStream { +interface LegacyStream $T.throws(T.v1Errors) { feed @0 (raw :Data) -> stream; # must be called before a new op is started, otherwise errors may get lost - sync @1 () -> (result :T.ResultV); + sync @1 (); } diff --git a/lix/libstore/daemon.cc b/lix/libstore/daemon.cc index f8748294a..dec3750cf 100644 --- a/lix/libstore/daemon.cc +++ b/lix/libstore/daemon.cc @@ -1040,34 +1040,34 @@ struct RequestStreamImpl final : LegacyStream::Server auto req = callbacks.feedRequest(); req.initRaw(*got); std::copy(buf.begin(), buf.begin() + *got, req.getRaw().begin()); - co_await req.send(); + TRY_AWAIT_RPC(req.send()); } } - co_await callbacks.syncRequest().send(); + TRY_AWAIT_RPC(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())); + if (error) { + std::rethrow_exception(error); } + auto bytes = context.getParams().getRaw(); + TRY_AWAIT(workerSock.writeFull(bytes.begin(), bytes.size())); } catch (...) { onError(std::current_exception()); + rpc::rethrow_as_rpc_error(); } kj::Promise sync(SyncContext context) override try { TRY_AWAIT(logger->flush()); if (error) { - RPC_FILL(context.initResults(), initResult, error); - } else { - context.initResults().initResult().setGood(); + std::rethrow_exception(error); } } catch (...) { - RPC_FILL(context.initResults(), initResult, std::current_exception()); + rpc::rethrow_as_rpc_error(); } }; @@ -1088,7 +1088,7 @@ struct LegacyBootImpl final : LegacyBoot::Server logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger()); auto args = context.getParams(); - auto result = context.initResults().initResult().initGood(); + auto result = context.initResults().initResult(); // We and the underlying store both need to trust the client for it to be trusted. if (!state->trusted) { result.setTrust(LegacyBoot::Trust::UNTRUSTED); @@ -1163,7 +1163,7 @@ struct LegacyBootImpl final : LegacyBoot::Server ); used = true; } catch (...) { - RPC_FILL(context.initResults(), initResult, std::current_exception()); + rpc::rethrow_as_rpc_error(); } }; @@ -1200,7 +1200,6 @@ struct BootstrapImpl final : Bootstrap::Server kj::Exception::Type::UNIMPLEMENTED, "main", 0, kj::str("rpc sockets not enabled") ) ); - throw Error("rpc sockets not enabled"); } auto result = context.initResults(); @@ -1213,20 +1212,25 @@ struct BootstrapImpl final : Bootstrap::Server } kj::Promise request(RequestContext context) override - try { + { auto id = rpc::to(context.getParams().getProtocol()); if (used) { - throw Error("connection already initialized"); + kj::throwFatalException( + kj::Exception( + kj::Exception::Type::FAILED, "main", 0, kj::str("connection already initialized") + ) + ); } else if (const auto & protocol = get(protocols, id)) { used = true; - context.initResults().initResult().setGood(protocol->factory(trusted, store)); + context.initResults().setResult(protocol->factory(trusted, store)); } else { - throw Error("unsupported protocol %s", id); + kj::throwFatalException( + kj::Exception( + kj::Exception::Type::UNIMPLEMENTED, "main", 0, kj::str("unsupported protocol", id) + ) + ); } return kj::READY_NOW; - } catch (...) { - RPC_FILL(context.initResults(), initResult, std::current_exception()); - return kj::READY_NOW; } }; } diff --git a/lix/libstore/uds-remote-store.cc b/lix/libstore/uds-remote-store.cc index 91b6cf8fc..304487336 100644 --- a/lix/libstore/uds-remote-store.cc +++ b/lix/libstore/uds-remote-store.cc @@ -17,6 +17,8 @@ #include #include #include +#include +#include #include #include #include @@ -200,13 +202,14 @@ try { auto bootstrapReq = bootstrap.requestRequest(); bootstrapReq.setClientInfo(PACKAGE_STRING); bootstrapReq.setProtocol(rpc::daemon::UNSTABLE_LEGACY_TUNNELED); - auto legacyBoot = TRY_AWAIT_RPC(bootstrapReq.send()).castAs(); + auto legacyBoot = + TRY_AWAIT_RPC_NOEXCEPT(bootstrapReq.send()).getResult().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); + auto initResp = TRY_AWAIT_RPC(initReq.send()); + auto initResult = initResp.getResult(); con.remoteTrustsUs = initResult.getTrust() == rpc::daemon::LegacyBoot::Trust::TRUSTED ? std::optional{Trusted} : initResult.getTrust() == rpc::daemon::LegacyBoot::Trust::UNTRUSTED ? std::optional{NotTrusted} @@ -235,10 +238,10 @@ try { auto req = requestStream.feedRequest(); req.initRaw(*got); std::copy(buf.begin(), buf.begin() + *got, req.getRaw().begin()); - co_await req.send(); + TRY_AWAIT_RPC(req.send()); } } - co_await requestStream.syncRequest().send(); + TRY_AWAIT_RPC(requestStream.syncRequest().send()); } catch (...) { ignoreExceptionExceptInterrupt(); error = std::current_exception(); @@ -246,22 +249,24 @@ try { 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())); + if (state.error) { + std::rethrow_exception(state.error); } + auto bytes = context.getParams().getRaw(); + TRY_AWAIT(state.proxySock->writeFull(bytes.begin(), bytes.size())); } catch (...) { state.error = std::current_exception(); + rpc::rethrow_as_rpc_error(); } kj::Promise UDSRemoteStore::LegacyStreamProxy::sync(SyncContext context) -{ +try { if (state.error) { - RPC_FILL(context.initResults(), initResult, state.error); - } else { - context.initResults().initResult().setGood(); + std::rethrow_exception(state.error); } return kj::READY_NOW; +} catch (...) { + rpc::rethrow_as_rpc_error(); } kj::Promise> UDSRemoteStore::initConnection(RemoteStore::Connection & conn) diff --git a/lix/libutil/meson.build b/lix/libutil/meson.build index 8acae3919..d5ab3c6f2 100644 --- a/lix/libutil/meson.build +++ b/lix/libutil/meson.build @@ -37,6 +37,7 @@ liblix_sources += files( 'processes.cc', 'references.cc', 'regex.cc', + 'rpc.cc', 'serialise-async.cc', 'serialise.cc', 'shlex.cc', diff --git a/lix/libutil/rpc.cc b/lix/libutil/rpc.cc new file mode 100644 index 000000000..ba9e21cfd --- /dev/null +++ b/lix/libutil/rpc.cc @@ -0,0 +1,35 @@ +#include "libutil/rpc.hh" +#include "libutil/error.hh" +#include "libutil/types-rpc.hh" +#include + +namespace nix::rpc::detail { +std::exception_ptr unwrapErrorRaw(kj::Exception & e, std::source_location loc) +{ + nix::Error fe{e.getDescription().cStr()}; + fe.addAsyncTrace(loc, "RPC call"); + return std::make_exception_ptr(std::move(fe)); +} + +std::exception_ptr unwrapErrorV1(kj::Exception & e, std::source_location loc) +{ + if (auto decoded = error::v1::tryDecode(e.getDescription().cStr())) { + nix::Error fe(std::move(*decoded)); + fe.addAsyncTrace(loc, "RPC call"); + return std::make_exception_ptr(std::move(fe)); + } else { + return unwrapErrorRaw(e, loc); + } +} + +void rethrowAsErrorV1() +{ + try { + throw; // NOLINT(lix-foreign-exceptions) + } catch (const nix::BaseError & e) { + kj::throwFatalException( + kj::Exception(kj::Exception::Type::FAILED, "remote", 0, kj::str(error::v1::encodeLossy(e.info()))) + ); + } +} +} diff --git a/lix/libutil/rpc.hh b/lix/libutil/rpc.hh index 9aa36684c..2d3efb91f 100644 --- a/lix/libutil/rpc.hh +++ b/lix/libutil/rpc.hh @@ -1,17 +1,24 @@ #pragma once ///@file RPC helper functions +#include "lix/libutil/result.hh" #include "lix/libutil/charptr-cast.hh" #include "lix/libutil/rpc-fwd.hh" #include #include #include #include +#include #include +#include #include #include #include +namespace kj { +class Exception; +} + namespace nix::rpc { template @@ -181,15 +188,70 @@ inline void doFill(auto && builder, Inner (Builder::*field)(Init), From && f, au #define LIX_RPC_FILL(fobj, ffield, fsource, ...) \ [&] { ::nix::rpc::detail::doFill(fobj, &decltype(fobj)::ffield, (fsource), ##__VA_ARGS__); }() -#define LIX_TRY_AWAIT_RPC(...) \ - LIX_TRY_AWAIT_CONTEXT_MAP( \ - [] { return "RPC call"; }, \ - ([](auto r) { return ::nix::rpc::from(r.getResult()); }), \ - __VA_ARGS__ \ - ) +namespace detail { +std::exception_ptr unwrapErrorRaw(kj::Exception & e, std::source_location loc); +std::exception_ptr unwrapErrorV1(kj::Exception & e, std::source_location loc); + +[[noreturn]] +void rethrowAsErrorV1(); + +kj::Promise> inline rewrapNoexcept( + kj::Promise && promise, std::source_location loc = std::source_location::current() +) +{ + return promise.then( + []() -> nix::Result { return result::success(); }, + [loc](kj::Exception && e) -> nix::Result { return result::failure(unwrapErrorRaw(e, loc)); } + ); +} + +template +kj::Promise> +rewrapNoexcept(kj::Promise && promise, std::source_location loc = std::source_location::current()) +{ + return promise.then( + [](T result) -> nix::Result { return result; }, + [loc](kj::Exception && e) -> nix::Result { return result::failure(unwrapErrorRaw(e, loc)); } + ); +} + +kj::Promise> inline rewrapV1( + kj::Promise && promise, std::source_location loc = std::source_location::current() +) +{ + return promise.then( + []() -> nix::Result { return result::success(); }, + [loc](kj::Exception && e) -> nix::Result { return result::failure(unwrapErrorV1(e, loc)); } + ); +} + +template +kj::Promise> +rewrapV1(kj::Promise && promise, std::source_location loc = std::source_location::current()) +{ + return promise.then( + [](T result) -> nix::Result { return result; }, + [loc](kj::Exception && e) -> nix::Result { return result::failure(unwrapErrorV1(e, loc)); } + ); +} +} + +#define LIX_WRAP_RPC_PROMISE_NOEXCEPT(...) (::nix::rpc::detail::rewrapNoexcept(__VA_ARGS__)) +#define LIX_WRAP_RPC_PROMISE_V1(...) (::nix::rpc::detail::rewrapV1(__VA_ARGS__)) + +#define LIX_TRY_AWAIT_RPC_NOEXCEPT(...) (LIX_TRY_AWAIT(LIX_WRAP_RPC_PROMISE_NOEXCEPT(__VA_ARGS__))) +#define LIX_TRY_AWAIT_RPC_V1(...) (LIX_TRY_AWAIT(LIX_WRAP_RPC_PROMISE_V1(__VA_ARGS__))) + +[[noreturn]] +inline void rethrow_as_rpc_error() +{ + detail::rethrowAsErrorV1(); +} #ifdef LIX_UR_COMPILER_UWU #define RPC_FILL LIX_RPC_FILL -#define TRY_AWAIT_RPC LIX_TRY_AWAIT_RPC +#define TRY_AWAIT_RPC_NOEXCEPT LIX_TRY_AWAIT_RPC_NOEXCEPT +#define TRY_AWAIT_RPC_V1 LIX_TRY_AWAIT_RPC_V1 +#define TRY_AWAIT_RPC LIX_TRY_AWAIT_RPC_V1 #endif } diff --git a/lix/libutil/types-rpc.hh b/lix/libutil/types-rpc.hh index 946618d57..f04ff487b 100644 --- a/lix/libutil/types-rpc.hh +++ b/lix/libutil/types-rpc.hh @@ -55,82 +55,6 @@ struct Fill } }; -namespace detail { -inline void makeBadResult(auto rb, const std::exception_ptr & e) -{ - try { - std::rethrow_exception(e); - } catch (nix::Error & e) { - LIX_RPC_FILL(rb, initBad, e.info()); - } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions) - LIX_RPC_FILL(rb, initBad, nix::Error("caught non-lix exception: %s", e.what()).info()); - } catch (...) { - LIX_RPC_FILL(rb, initBad, nix::Error("caught non-exception! spooky").info()); - } -} - -template -inline constexpr bool IsResult = false; -template -inline constexpr bool IsResult> = true; - -template -concept ResultReader = IsResult; - -template -concept ResultBuilder = IsResult; -} - -inline nix::Result from(const ResultV::Reader & r, auto &&... args) -{ - if (r.isGood()) { - return result::success(); - } else { - return result::failure(nix::Error(from(r.getBad(), args...))); - } -} - -template<> -struct Fill> -{ - static void fill(ResultV::Builder rb, const nix::Result & r, auto &&...) - { - if (r.has_value()) { - rb.setGood(); - } else { - detail::makeBadResult(rb, r.error()); - } - } -}; - -template<> -struct Fill -{ - static void fill(ResultV::Builder rb, const std::exception_ptr & e, auto &&...) - { - detail::makeBadResult(rb, e); - } -}; - -inline auto from(detail::ResultReader auto r, auto &&... args) -{ - using R = nix::Result; - if (r.isGood()) { - return R(result::success(r.getGood())); - } else { - return R(result::failure(nix::Error(from(r.getBad(), args...)))); - } -} - -template -struct Fill, std::exception_ptr> -{ - static void fill(Result::Builder rb, const std::exception_ptr & e, auto &&...) - { - detail::makeBadResult(rb, e); - } -}; - template<> struct Fill> { diff --git a/lix/libutil/types.capnp b/lix/libutil/types.capnp index 2cdfea42f..ecf1771da 100644 --- a/lix/libutil/types.capnp +++ b/lix/libutil/types.capnp @@ -46,21 +46,6 @@ struct Error { traces @2 :List(Data); } -struct Result(T) { - union { - good @0 :T; - bad @1 :Error; - } -} - -# primitives can't be args to generics, so we need to specialize for primitive here. -struct ResultV { - union { - good @0 :Void; - bad @1 :Error; - } -} - struct Settings { struct Setting { name @0 :Data;