libstore: asyncify RemoteStore::Connection::processStderr

we need a wrapper type for the remote exception because our Result type
does not deal well with its good type being the same as its error type.
we could have also return a `Result<Result<void>>` to fix this, but the
wrapper type clarifies via its name where the exception_ptr originates.

Change-Id: Ia6ce67b962cb8d6528b017f4cb682a55d6918939

lix/libstore/remote-store-connection.hh

@@ -4,9 +4,11 @@
 #include "lix/libstore/remote-store.hh"
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libstore/worker-protocol-impl.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/pool.hh"
 #include "lix/libutil/result.hh"
 #include "lix/libutil/serialise.hh"
+#include <kj/async.h>
 #include <tuple>
 #include <type_traits>
 #include <utility>
@@ -97,7 +99,14 @@ struct RemoteStore::Connection
 
     virtual ~Connection();
 
-    std::exception_ptr processStderr();
+    // wrapper type for remote errors because `Result<std::exception_ptr>`
+    // does not work very well and `Result<Result<void>>` is too confusing
+    struct [[nodiscard]] RemoteError
+    {
+        std::exception_ptr e;
+    };
+
+    kj::Promise<Result<RemoteError>> processStderr();
 };
 
 /**
@@ -134,7 +143,7 @@ struct RemoteStore::ConnectionHandle
     RemoteStore::Connection & operator * () { return *handle; }
     RemoteStore::Connection * operator -> () { return &*handle; }
 
-    void processStderr();
+    kj::Promise<Result<void>> processStderr();
 
     kj::Promise<Result<void>>
     withFramedSinkAsync(std::function<kj::Promise<Result<void>>(Sink & sink)> fun);
@@ -154,7 +163,7 @@ struct RemoteStore::ConnectionHandle
         if constexpr (requires { handle->to << std::declval<LastArgT>(); }) {
             ((handle->to << std::forward<Args>(args)), ...);
             handle->to.flush();
-            processStderr();
+            LIX_TRY_AWAIT(processStderr());
         } else {
             using ImmediateArgsIdxs = std::make_index_sequence<sizeof...(Args) - 1>;
             AllArgsT allArgs(std::forward<Args>(args)...);
@@ -180,7 +189,7 @@ private:
     struct FramedSinkHandler
     {
         std::exception_ptr ex;
-        std::packaged_task<void()> stderrHandler;
+        std::packaged_task<void(AsyncIoRoot &)> stderrHandler;
 
         explicit FramedSinkHandler(ConnectionHandle & conn, ThreadPool & handlerThreads);
 

lix/libstore/remote-store.cc

@@ -105,8 +105,10 @@ try {
         conn.daemonNixVersion = readString(conn.from);
         conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(conn);
 
-        auto ex = conn.processStderr();
-        if (ex) std::rethrow_exception(ex);
+        auto ex = TRY_AWAIT(conn.processStderr());
+        if (ex.e) {
+            std::rethrow_exception(ex.e);
+        }
     }
     catch (Error & e) {
         throw Error("cannot open connection to remote store '%s': %s", getUri(), e.what());
@@ -157,8 +159,10 @@ try {
 
     StringSource{command.s}.drainInto(conn.to);
     conn.to.flush();
-    auto ex = conn.processStderr();
-    if (ex) std::rethrow_exception(ex);
+    auto ex = TRY_AWAIT(conn.processStderr());
+    if (ex.e) {
+        std::rethrow_exception(ex.e);
+    }
     co_return result::success();
 } catch (...) {
     co_return result::current_exception();
@@ -172,16 +176,18 @@ RemoteStore::ConnectionHandle::~ConnectionHandle()
     }
 }
 
-void RemoteStore::ConnectionHandle::processStderr()
-{
-    auto ex = handle->processStderr();
-    if (ex) {
+kj::Promise<Result<void>> RemoteStore::ConnectionHandle::processStderr()
+try {
+    auto ex = TRY_AWAIT(handle->processStderr());
+    if (ex.e) {
         daemonException = true;
-        std::rethrow_exception(ex);
+        std::rethrow_exception(ex.e);
     }
+    co_return result::success();
+} catch (...) {
+    co_return result::current_exception();
 }
 
-
 kj::Promise<Result<RemoteStore::ConnectionHandle>> RemoteStore::getConnection()
 try {
     co_return ConnectionHandle(TRY_AWAIT(connections->get()), handlerThreads);
@@ -746,15 +752,14 @@ static Logger::Fields readFields(Source & from)
     return fields;
 }
 
-
-std::exception_ptr RemoteStore::Connection::processStderr()
-{
+kj::Promise<Result<RemoteStore::Connection::RemoteError>> RemoteStore::Connection::processStderr()
+try {
     while (true) {
 
         auto msg = readNum<uint64_t>(from);
 
         if (msg == STDERR_ERROR) {
-            return std::make_exception_ptr(readError(from));
+            co_return RemoteError{std::make_exception_ptr(readError(from))};
         }
 
         else if (msg == STDERR_NEXT)
@@ -789,21 +794,23 @@ std::exception_ptr RemoteStore::Connection::processStderr()
             throw Error("got unknown message type %x from Nix daemon", msg);
     }
 
-    return nullptr;
+    co_return RemoteError{nullptr};
+} catch (...) {
+    co_return result::current_exception();
 }
 
 RemoteStore::ConnectionHandle::FramedSinkHandler::FramedSinkHandler(
     ConnectionHandle & conn, ThreadPool & handlerThreads
 )
-    : stderrHandler([&]() {
+    : stderrHandler([&](AsyncIoRoot & aio) {
         try {
-            conn.processStderr();
+            aio.blockOn(conn.processStderr());
         } catch (...) {
             ex = std::current_exception();
         }
     })
 {
-    handlerThreads.enqueue([&] { stderrHandler(); });
+    handlerThreads.enqueueWithAio([&](AsyncIoRoot & aio) { stderrHandler(aio); });
 }
 
 RemoteStore::ConnectionHandle::FramedSinkHandler::~FramedSinkHandler() noexcept(false)