libstore: convert build-hook logging to rpc

Change-Id: I0c20f89de113dce6032c93a32e9fcd43b4478f55

lix/legacy/build-remote.cc

@@ -3,16 +3,15 @@
 #include "lix/libutil/c-calls.hh"
 #include "lix/libutil/error.hh"
 #include "lix/libutil/file-descriptor.hh"
+#include "lix/libutil/logging-rpc.hh"
 #include "lix/libutil/logging.hh"
 #include "lix/libutil/rpc.hh"
-#include "lix/libutil/types-rpc.hh"
+#include "lix/libutil/types-rpc.hh" // IWYU pragma: keep
 #include "lix/libutil/types.hh"
 #include <algorithm>
 #include <capnp/rpc-twoparty.h>
-#include <chrono>
 #include <cstring>
 #include <exception>
-#include <future>
 #include <kj/async.h>
 #include <kj/time.h>
 #include <set>
@@ -47,10 +46,11 @@ namespace {
 struct Instance final : rpc::build_remote::HookInstance::Server
 {
     unsigned int maxBuildJobs;
-    bool initialized = false;
+    bool initialized = false, used = false;
 
     kj::Promise<void> init(InitContext context) override;
 
+    kj::Promise<void> buildImpl(BuildContext context);
     kj::Promise<void> build(BuildContext context) override;
 };
 }
@@ -188,11 +188,10 @@ 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>
+    kj::Promise<Result<void>> startLogThread(std::string buildDescription, std::string drvPath)
-    startLogThread(const std::string & buildDescription, const std::string & drvPath)
+    try {
-    {
         if (!logPipe.readSide) {
-            return {};
+            co_return result::success();
         }
 
         logPipe.writeSide.close();
@@ -200,54 +199,51 @@ struct BuilderConnection
         // NOTE this is very similar to handleBuilderOutput in DerivationGoal, but unlike
         // the derivation goal we do not need to handle EIO from a pty here. we also have
         // no timeouts or limits to keep track of, which makes deduplication less useful.
-        return std::async(
+        auto act = logger->startActivity(
-            std::launch::async,
+            lvlInfo, actBuild, buildDescription, Logger::Fields{drvPath, storeUri, 1, 1}
-            [this, buildDescription, drvPath](int from) {
-                AsyncIoRoot aio;
-
-                auto act = logger->startActivity(
-                    lvlInfo, actBuild, buildDescription, Logger::Fields{drvPath, storeUri, 1, 1}
-                );
-
-                std::map<ActivityId, Activity> activities;
-
-                auto reader = AIO().lowLevelProvider.wrapInputFd(from);
-
-                LogLineSplitter splitter;
-
-                auto flushLine = [&](const std::string & line) {
-                    if (const auto state =
-                            handleJSONLogMessage(line, act, activities, "the derivation builder"))
-                    {
-                        if (state == Logger::BufferState::NeedsFlush) {
-                            aio.blockOn(act.getLogger().flush());
-                        }
-                    } else {
-                        ACTIVITY_RESULT_SYNC(aio, act, resBuildLogLine, line);
-                    }
-                };
-
-                auto buf = kj::heapArray<char>(4096);
-                while (true) {
-                    const auto got = aio.blockOn(reader->tryRead(buf.begin(), 1, buf.size()));
-                    if (got == 0) {
-                        break;
-                    }
-
-                    std::string_view data{buf.begin(), got};
-                    while (!data.empty()) {
-                        if (auto line = splitter.feed(data)) {
-                            flushLine(*line);
-                        }
-                    }
-                }
-
-                if (auto left = splitter.finish(); !left.empty()) {
-                    flushLine(left);
-                }
-            },
-            logPipe.readSide.get()
         );
+
+        std::map<ActivityId, Activity> activities;
+
+        auto reader = AIO().lowLevelProvider.wrapInputFd(logPipe.readSide.get());
+
+        LogLineSplitter splitter;
+
+        auto flushLine = [&](const std::string & line) {
+            if (const auto state =
+                    handleJSONLogMessage(line, act, activities, "the derivation builder"))
+            {
+                return *state;
+            } else {
+                return act.result(resBuildLogLine, line);
+            }
+        };
+
+        auto buf = kj::heapArray<char>(4096);
+        while (true) {
+            const auto got = co_await reader->tryRead(buf.begin(), 1, buf.size());
+            if (got == 0) {
+                break;
+            }
+
+            std::string_view data{buf.begin(), got};
+            while (!data.empty()) {
+                if (auto line = splitter.feed(data)) {
+                    if (flushLine(*line) == Logger::BufferState::NeedsFlush) {
+                        TRY_AWAIT(act.getLogger().flush());
+                    }
+                }
+            }
+        }
+
+        if (auto line = splitter.finish(); !line.empty()) {
+            (void) flushLine(line);
+            TRY_AWAIT(act.getLogger().flush());
+        }
+
+        co_return result::success();
+    } catch (...) {
+        co_return result::current_exception();
     }
 };
 
@@ -256,21 +252,16 @@ struct AcceptedBuild final : rpc::build_remote::HookInstance::AcceptedBuild::Ser
     ref<Store> store;
     StorePath drvPath;
     BuilderConnection builder;
-    rpc::build_remote::HookInstance::BuildLogger::Client buildLogger;
+    bool used = false;
 
-    AcceptedBuild(
+    AcceptedBuild(ref<Store> store, StorePath drvPath, BuilderConnection builder)
-        ref<Store> store,
-        StorePath drvPath,
-        BuilderConnection builder,
-        rpc::build_remote::HookInstance::BuildLogger::Client buildLogger
-    )
         : store(store)
         , drvPath(drvPath)
         , builder(std::move(builder))
-        , buildLogger(std::move(buildLogger))
     {
     }
 
+    kj::Promise<void> runImpl(RunContext context);
     kj::Promise<void> run(RunContext context) override;
 };
 
@@ -365,8 +356,6 @@ try {
 static int main_build_remote(AsyncIoRoot & aio, std::string programName, Strings argv)
 {
     {
-        logger = makeJSONLogger(*logger);
-
         /* Ensure we don't get any SSH passphrase or host key popups. */
         unsetenv("DISPLAY");
         unsetenv("SSH_ASKPASS");
@@ -393,6 +382,8 @@ kj::Promise<void> Instance::init(InitContext context)
             throw Error("build hook can only be initialized once");
         }
 
+        logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger());
+
         /* Read the parent's settings. */
         for (const auto & [name, value] : rpc::to<StringMap>(context.getParams().getSettings())) {
             settings.set(name, value);
@@ -413,85 +404,109 @@ kj::Promise<void> Instance::init(InitContext context)
     return kj::READY_NOW;
 }
 
-kj::Promise<void> Instance::build(BuildContext context)
+kj::Promise<void> Instance::buildImpl(BuildContext context)
 {
-    try {
+    if (!initialized) {
-        if (!initialized) {
+        throw Error("build hook not fully initialized");
-            throw Error("build hook not fully initialized");
+    }
-        }
 
-        // FIXME this does not open a daemon connection for historical reasons.
+    // FIXME this does not open a daemon connection for historical reasons.
-        // we may create a lot of build hook instances, and having each of them
+    // we may create a lot of build hook instances, and having each of them
-        // also create a daemon instance is inefficient and wasteful. in future
+    // also create a daemon instance is inefficient and wasteful. in future
-        // versions of the build hook (where we don't need one hook process per
+    // versions of the build hook (where we don't need one hook process per
-        // build) we should change this to using a daemon connection, ideally a
+    // build) we should change this to using a daemon connection, ideally a
-        // daemon connection provided by the parent via file descriptor passing
+    // daemon connection provided by the parent via file descriptor passing
-        auto store = TRY_AWAIT(openStore(settings.storeUri, {}, AllowDaemon::Disallow));
+    auto store = TRY_AWAIT(openStore(settings.storeUri, {}, AllowDaemon::Disallow));
 
-        /* It would be more appropriate to use $XDG_RUNTIME_DIR, since
+    /* It would be more appropriate to use $XDG_RUNTIME_DIR, since
-           that gets cleared on reboot, but it wouldn't work on macOS. */
+       that gets cleared on reboot, but it wouldn't work on macOS. */
-        auto currentLoadName = "/current-load";
+    auto currentLoadName = "/current-load";
-        if (auto localStore = store.try_cast_shared<LocalFSStore>())
+    if (auto localStore = store.try_cast_shared<LocalFSStore>()) {
-            currentLoad = std::string { localStore->config().stateDir } + currentLoadName;
+        currentLoad = std::string{localStore->config().stateDir} + currentLoadName;
-        else
+    } else {
-            currentLoad = settings.nixStateDir + currentLoadName;
+        currentLoad = settings.nixStateDir + currentLoadName;
+    }
 
-        auto machines = getMachines();
+    auto machines = getMachines();
-        debug("got %d remote builders", machines.size());
+    debug("got %d remote builders", machines.size());
 
-        if (machines.empty()) {
+    if (machines.empty()) {
-            context.getResults().initResult().initGood().setDeclinePermanently();
+        context.getResults().initResult().initGood().setDeclinePermanently();
+        co_return;
+    }
+
+    auto amWilling = context.getParams().getAmWilling();
+    auto neededSystem = rpc::to<std::string>(context.getParams().getNeededSystem());
+    auto drvPath = from(context.getParams().getDrvPath(), *store);
+    auto requiredFeatures =
+        rpc::to<std::set<std::string>>(context.getParams().getRequiredFeatures());
+
+    auto result = TRY_AWAIT(connectToBuilder(
+        store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures
+    ));
+
+    if (auto immediateResponse = std::get_if<BuildRejected>(&result)) {
+        switch (*immediateResponse) {
+        case BuildRejected::Temporarily:
+            context.getResults().initResult().initGood().setPostpone();
+            co_return;
+        case BuildRejected::Permanently:
+            context.getResults().initResult().initGood().setDecline();
             co_return;
         }
-
-        auto amWilling = context.getParams().getAmWilling();
-        auto neededSystem = rpc::to<std::string>(context.getParams().getNeededSystem());
-        auto drvPath = from(context.getParams().getDrvPath(), *store);
-        auto requiredFeatures =
-            rpc::to<std::set<std::string>>(context.getParams().getRequiredFeatures());
-        auto buildLogger = context.getParams().getBuildLogger();
-
-        auto result = TRY_AWAIT(connectToBuilder(
-            store, drvPath, machines, maxBuildJobs, amWilling, neededSystem, requiredFeatures
-        ));
-
-        if (auto immediateResponse = std::get_if<BuildRejected>(&result)) {
-            switch (*immediateResponse) {
-            case BuildRejected::Temporarily:
-                context.getResults().initResult().initGood().setPostpone();
-                co_return;
-            case BuildRejected::Permanently:
-                context.getResults().initResult().initGood().setDecline();
-                co_return;
-            }
-        }
-
-        auto builder = std::get_if<BuilderConnection>(&result);
-        assert(builder);
-
-        auto ac = context.getResults().initResult().initGood().initAccept();
-        ac.setMachine(kj::heap<AcceptedBuild>(store, drvPath, std::move(*builder), buildLogger));
-    } catch (...) {
-        RPC_FILL(context.getResults(), initResult, std::current_exception());
     }
+
+    auto builder = std::get_if<BuilderConnection>(&result);
+    assert(builder);
+
+    auto ac = context.getResults().initResult().initGood().initAccept();
+    ac.setMachine(kj::heap<AcceptedBuild>(store, drvPath, std::move(*builder)));
+}
+
+kj::Promise<void> Instance::build(BuildContext context)
+try {
+    if (used) {
+        throw Error("build hooks can only accept a single job");
+    }
+    used = true; // lock out other rpc calls during processing
+    co_await buildImpl(context);
+    TRY_AWAIT(logger->flush());
+    used = context.getResults().getResult().getGood().isAccept();
+} catch (...) {
+    RPC_FILL(context.getResults(), getResult, std::current_exception());
 }
 
 kj::Promise<void> AcceptedBuild::run(RunContext context)
 {
     try {
-        auto logThread = builder.startLogThread(
+        auto oldLogger = logger;
+        logger = rpc::log::makeRpcLoggerClient(context.getParams().getLogger());
+        TRY_AWAIT(oldLogger->flush());
+        KJ_DEFER({
+            delete logger;
+            logger = oldLogger;
+        });
+
+        if (used) {
+            throw Error("build hooks builds are single-use items");
+        }
+        used = true;
+        co_await runImpl(context);
+        TRY_AWAIT(logger->flush());
+    } catch (...) {
+        RPC_FILL(context.getResults(), getResult, std::current_exception());
+    }
+}
+
+kj::Promise<void> AcceptedBuild::runImpl(RunContext context)
+{
+    try {
+        auto logHandler = builder.startLogThread(
             fmt("%s on '%s'",
                 rpc::to<std::string_view>(context.getParams().getDescription()),
                 builder.storeUri),
             store->printStorePath(drvPath)
         );
-        KJ_DEFER({
-            // drop any existing ssh connection so the log thread can exit
-            builder.sshStore = nullptr;
-            if (logThread.valid()) {
-                logThread.get();
-            }
-        });
 
         auto & sshStore = builder.sshStore;
         auto & storeUri = builder.storeUri;
@@ -602,6 +617,10 @@ kj::Promise<void> AcceptedBuild::run(RunContext context)
             );
         }
 
+        // drop store connection, let log handler process any remaining input
+        builder.sshStore = nullptr;
+        TRY_AWAIT(logHandler);
+
         context.getResults().initResult().setGood();
     } catch (...) {
         RPC_FILL(context.getResults(), initResult, std::current_exception());

lix/libstore/build/derivation-goal.cc

@@ -24,12 +24,14 @@
 #include <boost/outcome/try.hpp>
 #include <capnp/rpc-twoparty.h>
 #include <cstdint>
+#include <exception>
 #include <fstream>
 #include <kj/array.h>
 #include <kj/async-unix.h>
 #include <kj/async.h>
 #include <kj/debug.h>
 #include <kj/exception.h>
+#include <kj/time.h>
 #include <kj/vector.h>
 #include <limits>
 #include <memory>
@@ -798,10 +800,7 @@ int DerivationGoal::getChildStatus()
     return hook->kill();
 }
 
-void DerivationGoal::closeReadPipes()
+void DerivationGoal::closeReadPipes() {}
-{
-    hook->fromHook.reset();
-}
 
 void DerivationGoal::cleanupHookFinally()
 {
@@ -1035,8 +1034,22 @@ try {
 }
 
 namespace {
-struct BuildHookLogger final : rpc::build_remote::HookInstance::BuildLogger::Server
+struct ActivityTrackingHookLogger final : HookInstance::HookLogger
-{};
+{
+    kj::TimePoint & tracker;
+
+    ActivityTrackingHookLogger(const Activity & act, FinishSink * logSink, kj::TimePoint & tracker)
+        : HookLogger(act, logSink)
+        , tracker(tracker)
+    {
+    }
+
+    kj::Promise<void> push(PushContext context) override
+    {
+        tracker = AIO().provider.getTimer().now();
+        return HookLogger::push(context);
+    }
+};
 }
 
 kj::Promise<Result<HookResult>> DerivationGoal::tryBuildHook()
@@ -1052,18 +1065,16 @@ try {
         hook = std::move(worker.hook.instances.front());
         worker.hook.instances.pop_front();
     } else {
-        hook = TRY_AWAIT(HookInstance::create());
+        hook = TRY_AWAIT(HookInstance::create(worker.act));
     }
 
     KJ_DEFER(hook = nullptr);
-    auto output = wrapChildHandler(handleRawChildStream());
 
     auto buildReq = hook->rpc->buildRequest();
     RPC_FILL(buildReq, setAmWilling, slotToken.valid());
     RPC_FILL(buildReq, setNeededSystem, drv->platform);
     RPC_FILL(buildReq, initDrvPath, drvPath, worker.store);
     RPC_FILL(buildReq, initRequiredFeatures, parsedDrv->getRequiredSystemFeatures());
-    buildReq.setBuildLogger(kj::heap<BuildHookLogger>());
     auto buildRespPromise = buildReq.send();
     auto buildResp = TRY_AWAIT_RPC(buildRespPromise);
 
@@ -1085,9 +1096,15 @@ try {
     // the build was accepted by the hook, we can free the slot for another build now
     hookSlot = {};
 
+    /* Create the log file and pipe. */
+    openLogFile();
+
     auto runReq = buildResp.getAccept().getMachine().runRequest();
     /* Tell the hook all the inputs that have to be copied to the
        remote system. */
+    runReq.setLogger(
+        kj::heap<ActivityTrackingHookLogger>(worker.act, logSink.get(), lastChildActivity)
+    );
     RPC_FILL(runReq, initInputs, inputPaths, worker.store);
 
     /* Tell the hooks the missing outputs that have to be copied back
@@ -1103,9 +1120,6 @@ try {
         RPC_FILL(runReq, setDescription, buildDescription());
     }
 
-    /* Create the log file and pipe. */
-    openLogFile();
-
     auto runPromise = runReq.send();
 
     // build via hook is now properly running. wait for it to finish
@@ -1113,21 +1127,30 @@ try {
     buildResult.startTime = time(0); // inexact
     mcRunningBuilds = worker.runningBuilds.addTemporarily(1);
 
-    auto result = co_await runPromise;
+    auto result = TRY_AWAIT(wrapChildHandler(
+        runPromise
+            .then(
+                // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+                [&](auto result) -> kj::Promise<Result<std::optional<WorkResult>>> {
+                    try {
+                        std::shared_ptr<Error> remoteError;
+                        if (result.getResult().isBad()) {
+                            remoteError =
+                                std::make_shared<Error>(from(result.getResult().getBad()));
+                            logErrorInfo(remoteError->info().level, remoteError->info());
+                        }
+                        // close the rpc connection to have the hook exit
+                        hook->rpc = nullptr;
+                        hook->wait();
+                        co_return TRY_AWAIT(buildDone(remoteError));
+                    } catch (...) {
+                        co_return result::current_exception();
+                    }
+                }
+            )
+    ));
 
-    // close the rpc connection to have the hook exit
+    co_return HookResult::Accept{std::move(*result)};
-    hook->rpc = nullptr;
-
-    if (auto error = TRY_AWAIT(output)) {
-        co_return HookResult::Accept{std::move(*error)};
-    }
-
-    std::shared_ptr<Error> remoteError;
-    if (result.getResult().isBad()) {
-        remoteError = std::make_shared<Error>(from(result.getResult().getBad()));
-        logErrorInfo(remoteError->info().level, remoteError->info());
-    }
-    co_return HookResult::Accept{TRY_AWAIT(buildDone(remoteError))};
 } catch (...) {
     co_return result::current_exception();
 }
@@ -1225,65 +1248,6 @@ Goal::WorkResult DerivationGoal::tooMuchLogs()
             getName(), settings.maxLogSize));
 }
 
-kj::Promise<Result<std::optional<Goal::WorkResult>>> DerivationGoal::handleRawChildStream() noexcept
-try {
-    assert(hook);
-
-    std::string currentHookLine;
-
-    AsyncFdIoStream in(AsyncFdIoStream::shared_fd{}, hook->fromHook.get());
-
-    auto buf = kj::heapArray<char>(4096);
-    while (true) {
-        const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size()));
-        if (!got) {
-            co_return std::nullopt;
-        }
-
-        std::string_view data = {buf.begin(), *got};
-        lastChildActivity = AIO().provider.getTimer().now();
-
-        for (auto c : data)
-            if (c == '\n') {
-                auto json = parseJSONMessage(currentHookLine, "the derivation builder");
-                if (json) {
-                    auto s = handleJSONLogMessage(
-                        *json, worker.act, hook->activities, "the derivation builder"
-                    );
-                    // ensure that logs from a builder using `ssh-ng://` as protocol
-                    // are also available to `nix log`.
-                    if (s && logSink) {
-                        const auto type = (*json)["type"];
-                        const auto fields = (*json)["fields"];
-                        if (type == resBuildLogLine) {
-                            const std::string logLine =
-                                (fields.size() > 0 ? fields[0].get<std::string>() : "") + "\n";
-                            (*logSink)(logLine);
-                        } else if (type == resSetPhase && ! fields.is_null()) {
-                            const auto phase = fields[0];
-                            if (! phase.is_null()) {
-                                // nixpkgs' stdenv produces lines in the log to signal
-                                // phase changes.
-                                // We want to get the same lines in case of remote builds.
-                                // The format is:
-                                //   @nix { "action": "setPhase", "phase": "$curPhase" }
-                                const auto logLine = JSON::object({
-                                    {"action", "setPhase"},
-                                    {"phase", phase}
-                                });
-                                (*logSink)("@nix " + logLine.dump(-1, ' ', false, JSON::error_handler_t::replace) + "\n");
-                            }
-                        }
-                    }
-                }
-                currentHookLine.clear();
-            } else
-                currentHookLine += c;
-    }
-} catch (...) {
-    co_return result::current_exception();
-}
-
 kj::Promise<Result<std::optional<Goal::WorkResult>>>
 DerivationGoal::wrapChildHandler(kj::Promise<Result<std::optional<WorkResult>>> handler) noexcept
 {

lix/libstore/build/derivation-goal.hh

@@ -311,7 +311,6 @@ protected:
 
     kj::Promise<Result<std::optional<WorkResult>>>
     wrapChildHandler(kj::Promise<Result<std::optional<WorkResult>>> handler) noexcept;
-    virtual kj::Promise<Result<std::optional<WorkResult>>> handleRawChildStream() noexcept;
     kj::Promise<Result<std::optional<WorkResult>>> monitorForSilence() noexcept;
     WorkResult tooMuchLogs();
 

lix/libstore/build/hook-instance.capnp

@@ -5,15 +5,13 @@ $Cxx.namespace("nix::rpc::build_remote");
 $Cxx.allowCancellation;
 
 using Types = import "/lix/libutil/types.capnp";
+using Log = import "/lix/libutil/logging.capnp";
 using StoreTypes = import "/lix/libstore/types.capnp";
 
 interface HookInstance {
-  interface BuildLogger {
-    # will be used later
-  }
-
   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
@@ -32,13 +30,13 @@ interface HookInstance {
   }
 
   init @0 (
-    settings :Types.Settings
+    logger :Log.LogStream,
+    settings :Types.Settings,
   ) -> (result :Types.ResultV);
   build @1 (
     amWilling :Bool,
     neededSystem :Data,
     drvPath :StoreTypes.StorePath,
     requiredFeatures :List(Data),
-    buildLogger :BuildLogger,
   ) -> (result :Types.Result(BuildResponse));
 }

lix/libstore/build/hook-instance.cc

@@ -4,15 +4,63 @@
 #include "lix/libutil/file-system.hh"
 #include "lix/libstore/globals.hh"
 #include "lix/libstore/build/hook-instance.hh"
+#include "lix/libutil/json.hh"
+#include "lix/libutil/logging.hh"
+#include "lix/libutil/result.hh"
 #include "lix/libutil/rpc.hh"
+#include "lix/libutil/serialise.hh"
 #include "lix/libutil/strings.hh"
+#include "lix/libutil/logging-rpc.hh" // IWYU pragma: keep
 #include "lix/libutil/types-rpc.hh" // IWYU pragma: keep
 #include <kj/memory.h>
 #include <memory>
+#include <string_view>
 
 namespace nix {
 
-kj::Promise<Result<std::unique_ptr<HookInstance>>> HookInstance::create()
+void HookInstance::HookLogger::emitLog(rpc::log::Event::Result::Reader r)
+{
+    auto type = rpc::log::from(r.getType());
+    auto fields = r.getFields();
+    if (!type) {
+        return;
+    }
+
+    // ensure that logs from a builder using `ssh-ng://` as protocol
+    // are also available to `nix log`.
+    if (type == resBuildLogLine) {
+        if (fields.size() > 0 && fields[0].isS()) {
+            (*logSink)(fmt("%s\n", rpc::to<std::string_view>(fields[0].getS())));
+        } else {
+            (*logSink)("\n");
+        }
+    } else if (type == resSetPhase && fields.size() > 0 && fields[0].isS()) {
+        // nixpkgs' stdenv produces lines in the log to signal phase changes.
+        // We want to get the same lines in case of remote builds.
+        // The format is:
+        //   @nix { "action": "setPhase", "phase": "$curPhase" }
+        const auto phase = rpc::to<std::string_view>(fields[0].getS());
+        const auto logLine = JSON::object({{"action", "setPhase"}, {"phase", phase}});
+        (*logSink)("@nix " + logLine.dump(-1, ' ', false, JSON::error_handler_t::replace) + "\n");
+    }
+}
+
+kj::Promise<void> HookInstance::HookLogger::push(PushContext context)
+{
+    try {
+        auto e = context.getParams().getE();
+        if (logSink && e.isResult()) {
+            emitLog(e.getResult());
+        }
+    } catch (std::exception & e) { // NOLINT(lix-foreign-exceptions)
+        printError("error in log processor: %s", e.what());
+        throw; // NOLINT(lix-foreign-exceptions)
+    }
+
+    return RpcLoggerServer::push(context);
+}
+
+kj::Promise<Result<std::unique_ptr<HookInstance>>> HookInstance::create(const Activity & act)
 try {
     debug("starting build hook '%s'", concatStringsSep(" ", settings.buildHook.get()));
 
@@ -32,10 +80,6 @@ try {
 
     args.push_back(std::to_string(verbosity));
 
-    /* Create a pipe to get the output of the child. */
-    Pipe fromHook_;
-    fromHook_.create();
-
     /* Create the communication pipes. */
     auto [selfRPC, hookRPC] = SocketPair::stream();
 
@@ -43,9 +87,6 @@ try {
 
     /* Fork the hook. */
     auto pid = startProcess([&]() {
-        if (dup2(fromHook_.writeSide.get(), STDERR_FILENO) == -1)
-            throw SysError("cannot pipe standard error into log file");
-
         commonExecveingChildInit();
 
         if (chdir("/") == -1) throw SysError("changing into /");
@@ -71,14 +112,13 @@ try {
 
     {
         auto initReq = rpc.initRequest();
+        initReq.setLogger(kj::heap<HookLogger>(act, nullptr));
         RPC_FILL(initReq, initSettings, settings);
         TRY_AWAIT_RPC(initReq.send());
     }
 
     co_return std::make_unique<HookInstance>(
-        std::move(fromHook_.readSide),
+        kj::heap(std::move(rpc)).attach(std::move(conn), std::move(client)), std::move(pid)
-        kj::heap(std::move(rpc)).attach(std::move(conn), std::move(client)),
-        std::move(pid)
     );
 } catch (...) {
     co_return result::current_exception();

lix/libstore/build/hook-instance.hh

@@ -2,9 +2,11 @@
 ///@file
 
 #include "hook-instance.capnp.h"
+#include "lix/libutil/logging-rpc.hh"
 #include "lix/libutil/logging.hh"
 #include "lix/libutil/processes.hh"
 #include "lix/libutil/serialise.hh"
+#include "logging.capnp.h"
 #include <capnp/rpc-twoparty.h>
 #include <kj/async-io.h>
 #include <memory>
@@ -14,22 +16,26 @@ namespace nix {
 
 struct HookInstance
 {
-    /**
+    struct HookLogger : rpc::log::RpcLoggerServer
-     * Pipe for the hook's standard output/error.
+    {
-     */
+        FinishSink * logSink;
-    AutoCloseFD fromHook;
+
+        HookLogger(const Activity & act, FinishSink * logSink)
+            : rpc::log::RpcLoggerServer(act)
+            , logSink(logSink)
+        {
+        }
+
+        void emitLog(rpc::log::Event::Result::Reader r);
+        kj::Promise<void> push(PushContext context) override;
+    };
 
     kj::Own<rpc::build_remote::HookInstance::Client> rpc;
 
-    std::map<ActivityId, Activity> activities;
+    static kj::Promise<Result<std::unique_ptr<HookInstance>>> create(const Activity & act);
 
-    static kj::Promise<Result<std::unique_ptr<HookInstance>>> create();
+    HookInstance(kj::Own<rpc::build_remote::HookInstance::Client> rpc, Pid pid)
-
+        : rpc(std::move(rpc))
-    HookInstance(
-        AutoCloseFD fromHook, kj::Own<rpc::build_remote::HookInstance::Client> rpc, Pid pid
-    )
-        : fromHook(std::move(fromHook))
-        , rpc(std::move(rpc))
         , pidOrStatus(std::move(pid))
     {
     }

lix/libstore/build/local-derivation-goal.cc

@@ -2687,10 +2687,6 @@ StorePath LocalDerivationGoal::makeFallbackPath(const StorePath & path)
 kj::Promise<Result<std::optional<Goal::WorkResult>>>
 LocalDerivationGoal::handleRawChildStream() noexcept
 try {
-    if (hook) {
-        co_return TRY_AWAIT(DerivationGoal::handleRawChildStream());
-    }
-
     AsyncFdIoStream in(AsyncFdIoStream::shared_fd{}, builderOutPTY.get());
 
     std::map<ActivityId, Activity> builderActivities;

lix/libstore/build/local-derivation-goal.hh

@@ -324,7 +324,7 @@ protected:
      */
     virtual Pid startChild(std::function<void()> openSlave);
 
-    kj::Promise<Result<std::optional<WorkResult>>> handleRawChildStream() noexcept override;
+    kj::Promise<Result<std::optional<WorkResult>>> handleRawChildStream() noexcept;
 
     /**
      * Set up the system call filtering required for the sandbox.