libstore: pass a log pipe into build-hook

this way we don't have to duplicate build log parsing in the hook.

Change-Id: I1c96b75aea3b4bb747aa0f0cc76c00eace8911c4

lix/legacy/build-remote.cc

@@ -1,4 +1,5 @@
 #include "lix/libstore/path.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/error.hh"
 #include "lix/libutil/file-descriptor.hh"
 #include "lix/libutil/logging.hh"
@@ -183,7 +184,7 @@ struct BuilderConnection
 
     // start the thread that reads ssh stderr and turns it into log items.
     // this future *must* outlive sshStore, otherwise it will never finish
-    std::future<void> startLogThread()
+    std::future<void> startLogThread(int intoFD)
     {
         if (!logPipe.readSide) {
             return {};
@@ -193,47 +194,16 @@ struct BuilderConnection
 
         return std::async(
             std::launch::async,
-            [](AutoCloseFD logFD) {
-                Activity act(*logger, lvlTalkative, actUnknown, "remote builder");
-                std::vector<char> buf(4096);
-                size_t currentLogLinePos = 0;
-                std::string currentLogLine;
+            [](int from, int to) {
+                AsyncIoRoot aio;
 
-                auto flushLine = [&] {
-                    act.result(resBuildLogLine, {currentLogLine});
-                    currentLogLine.clear();
-                    currentLogLinePos = 0;
-                };
+                auto reader = AIO().lowLevelProvider.wrapInputFd(from);
+                auto writer = AIO().lowLevelProvider.wrapOutputFd(to);
 
-                while (true) {
-                    const auto got = ::read(logFD.get(), buf.data(), buf.size());
-                    if (got < 0) {
-                        printError("error reading builder response: %s", strerror(errno));
-                        break;
-                    } else if (got == 0) {
-                        if (!currentLogLine.empty()) {
-                            flushLine();
-                        }
-                        break;
-                    }
-
-                    std::string_view data{buf.data(), size_t(got)};
-
-                    for (auto c : data) {
-                        if (c == '\r') {
-                            currentLogLinePos = 0;
-                        } else if (c == '\n') {
-                            flushLine();
-                        } else {
-                            if (currentLogLinePos >= currentLogLine.size()) {
-                                currentLogLine.resize(currentLogLinePos + 1);
-                            }
-                            currentLogLine[currentLogLinePos++] = c;
-                        }
-                    }
-                }
+                reader->pumpTo(*writer).wait(aio.kj.waitScope);
             },
-            std::move(logPipe.readSide)
+            logPipe.readSide.get(),
+            intoFD
         );
     }
 };
@@ -375,7 +345,7 @@ static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings
 
         auto conn = aio.kj.lowLevelProvider->wrapUnixSocketFd(1);
         capnp::TwoPartyServer srv(kj::heap<Instance>(maxBuildJobs));
-        srv.accept(*conn).wait(aio.kj.waitScope);
+        srv.accept(*conn, 1).wait(aio.kj.waitScope);
         return 0;
     }
 }
@@ -442,7 +412,13 @@ kj::Promise<void> Instance::build(BuildContext context)
 kj::Promise<void> AcceptedBuild::run(RunContext context)
 {
     try {
-        auto logThread = builder.startLogThread();
+        auto builderLogger = context.getParams().getBuildLogger();
+        const int logFD = (co_await builderLogger.getFd()).orDefault(-1);
+        if (logFD < 0) {
+            throw Error("build-hook needs a logFD from the builder to build");
+        }
+
+        auto logThread = builder.startLogThread(logFD);
         KJ_DEFER({
             // drop any existing ssh connection so the log thread can exit
             builder.sshStore = nullptr;

lix/libstore/build/derivation-goal.cc

@@ -1023,6 +1023,20 @@ try {
     co_return result::current_exception();
 }
 
+namespace {
+struct BuildHookLogger final : rpc::build_remote::HookInstance::BuildLogger::Server
+{
+    AutoCloseFD fd;
+
+    BuildHookLogger(AutoCloseFD fd) : fd(std::move(fd)) {}
+
+    kj::Maybe<int> getFd() override
+    {
+        return fd.get();
+    }
+};
+}
+
 kj::Promise<Result<HookResult>> DerivationGoal::tryBuildHook()
 try {
     if (!worker.hook.available || !useDerivation) {
@@ -1039,6 +1053,12 @@ try {
         hook = TRY_AWAIT(HookInstance::create());
     }
 
+    // open a pipe to receive logs directly from the hook
+    Pipe logPipe;
+    logPipe.create();
+    builderOutFD = &logPipe.readSide;
+    KJ_DEFER(builderOutFD = nullptr);
+
     KJ_DEFER(hook = nullptr);
     auto output = handleChildOutput();
 
@@ -1074,6 +1094,7 @@ try {
     /* Tell the hook all the inputs that have to be copied to the
        remote system. */
     RPC_FILL(runReq, initInputs, inputPaths, worker.store);
+    runReq.setBuildLogger(kj::heap<BuildHookLogger>(std::move(logPipe.writeSide)));
 
     /* Tell the hooks the missing outputs that have to be copied back
        from the remote system. */

lix/libstore/build/hook-instance.capnp

@@ -8,10 +8,15 @@ using Types = import "/lix/libutil/types.capnp";
 using StoreTypes = import "/lix/libstore/types.capnp";
 
 interface HookInstance {
+  interface BuildLogger {
+    # only used for fd passing
+  }
+
   interface AcceptedBuild {
     run @0 (
       inputs :List(StoreTypes.StorePath), # actual a set
       wantedOutputs :List(Data), # actually StringSet
+      buildLogger :BuildLogger,
     ) -> (result :Types.ResultV);
   }
 

lix/libstore/build/hook-instance.hh

@@ -6,6 +6,8 @@
 #include "lix/libutil/processes.hh"
 #include "lix/libutil/serialise.hh"
 #include <capnp/rpc-twoparty.h>
+#include <kj/async-io.h>
+#include <utility>
 
 namespace nix {
 
@@ -16,7 +18,7 @@ struct HookInstance
      */
     AutoCloseFD fromHook;
 
-    kj::Own<kj::AsyncIoStream> conn;
+    kj::Own<kj::AsyncCapabilityStream> conn;
     std::optional<capnp::TwoPartyClient> client;
     rpc::build_remote::HookInstance::Client rpc;
 
@@ -32,7 +34,7 @@ struct HookInstance
     HookInstance(AutoCloseFD fromHook, AutoCloseFD rpc, Pid pid)
         : fromHook(std::move(fromHook))
         , conn(AIO().lowLevelProvider.wrapUnixSocketFd(kj::AutoCloseFd(rpc.release())))
-        , client(*this->conn)
+        , client(std::in_place, *this->conn, 1)
         , rpc(client->bootstrap().castAs<rpc::build_remote::HookInstance>())
         , pid(std::move(pid))
     {