libutil: add logging rpc definitions

this is still somewhat experimental and should be considered in flux. we
will have to nail down a logger interface once we start moving the store
protocol to rpc, but until we do that we can use build hooks to test it.

Change-Id: Id20cd346c9520f45871799c31b0af040adde56ef

lix/libstore/filetransfer.cc

@@ -375,7 +375,7 @@ struct curlFileTransfer : public FileTransfer
         {
             try {
                 if (act.progress(dlnow, dltotal) == Logger::BufferState::NeedsFlush) {
-                    act.getLogger().waitForSpace();
+                    act.getLogger().waitForSpace(); // NOLINT(lix-never-async)
                 }
             } catch (nix::Interrupted &) {
             }
@@ -409,7 +409,7 @@ struct curlFileTransfer : public FileTransfer
             else if (code == CURLE_OK && successfulStatuses.count(httpStatus))
             {
                 if (act.progress(bodySize, bodySize) == Logger::BufferState::NeedsFlush) {
-                    act.getLogger().waitForSpace();
+                    act.getLogger().waitForSpace(); // NOLINT(lix-never-async)
                 }
                 auto state = downloadState.lock();
                 state->done = true;

lix/libutil/logging-rpc.cc

@@ -0,0 +1,324 @@
+#include "logging-rpc.hh"
+#include "types-rpc.hh" // IWYU pragma: keep
+#include "sync.hh"
+#include "types.hh"
+#include <cstdlib>
+#include <kj/exception.h>
+
+namespace nix {
+namespace {
+struct RpcLogger : Logger
+{
+    struct Log
+    {
+        Verbosity level;
+        std::string msg;
+    };
+    struct LogEI
+    {
+        ErrorInfo ei;
+    };
+    struct StartActivity
+    {
+        Verbosity level;
+        uint64_t id;
+        ActivityType type;
+        std::string text;
+        uint64_t parent;
+        Logger::Fields fields;
+    };
+    struct StopActivity
+    {
+        uint64_t id;
+    };
+    struct ActivityResult
+    {
+        uint64_t id;
+        ResultType type;
+        Logger::Fields fields;
+    };
+
+    using Event = std::variant<Log, LogEI, StartActivity, StopActivity, ActivityResult>;
+
+    struct Buffer
+    {
+        std::vector<Event> items;
+        /// *very* rough approximation of how much memory our buffer uses. this
+        /// should not be an exact byte count to keep accounting simple, but it
+        /// should still be roughly representative of reality. a small constant
+        /// error factor during average execution is acceptable, expected even.
+        size_t sizeEstimate;
+
+        /// non-null if a remote log operation failed. we'll rethrow it blindly
+        /// every time the buffers are flushed, but not during message enqueue.
+        /// this hopefully avoids crashing due to recursive exception handling.
+        std::exception_ptr failure;
+
+        auto take()
+        {
+            if (failure) {
+                std::rethrow_exception(failure);
+            }
+            sizeEstimate = 0;
+            return std::move(items);
+        }
+    };
+
+    rpc::log::LogStream::Client remote;
+    Sync<Buffer> buffer;
+    Sync<std::list<kj::Own<kj::PromiseFulfiller<void>>>, AsyncMutex> flushReq;
+    kj::Promise<void> flusher;
+
+    RpcLogger(rpc::log::LogStream::Client remote)
+        : remote(remote)
+        , flusher(flushLoop().eagerlyEvaluate([](kj::Exception && e) {
+            if (e.getType() == kj::Exception::Type::DISCONNECTED) {
+                std::cerr << "peer disconnected, exiting with haste\n";
+                std::exit(90);
+            } else {
+                std::cerr << "log flusher failed catastrophically!\n";
+                // rethrow the exception and terminate to cause a core dump and stack trace
+                try {
+                    kj::throwFatalException(std::move(e));
+                } catch (...) {
+                    std::terminate();
+                }
+            }
+        }))
+    {
+    }
+
+    ~RpcLogger() noexcept = default;
+
+    static size_t fieldSize(const Logger::Fields & fields)
+    {
+        size_t size = 0;
+        for (auto & f : fields) {
+            size += sizeof(f) + f.s.size();
+        }
+        return size;
+    }
+
+    BufferState push(size_t extraSize, Event e)
+    {
+        auto buffer = this->buffer.lock();
+        if (buffer->failure) {
+            return BufferState::HasSpace;
+        }
+        buffer->sizeEstimate += sizeof(e) + extraSize;
+        buffer->items.emplace_back(std::move(e));
+        return buffer->sizeEstimate >= 1024 * 1024 ? BufferState::NeedsFlush
+                                                   : BufferState::HasSpace;
+    }
+
+    BufferState log(Verbosity lvl, std::string_view s) override
+    {
+        return push(s.size(), Log{lvl, std::string(s)});
+    }
+
+    BufferState logEI(const ErrorInfo & ei) override
+    {
+        // size is just a guess. errors are usually rare and small
+        return push(1024, LogEI{ei});
+    }
+
+    BufferState startActivityImpl(
+        ActivityId act,
+        Verbosity lvl,
+        ActivityType type,
+        const std::string & s,
+        const Fields & fields,
+        ActivityId parent
+    ) override
+    {
+        return push(fieldSize(fields), StartActivity{lvl, act, type, s, parent, fields});
+    }
+
+    BufferState stopActivityImpl(ActivityId act) override
+    {
+        return push(0, StopActivity{act});
+    }
+
+    BufferState resultImpl(ActivityId act, ResultType type, const Fields & fields) override
+    {
+        return push(fieldSize(fields), ActivityResult{act, type, fields});
+    }
+
+    kj::Promise<Result<void>> flush() override
+    try {
+        auto pfp = kj::newPromiseAndCrossThreadFulfiller<void>();
+        (co_await flushReq.lock())->emplace_back(std::move(pfp.fulfiller));
+        flushReq.notify();
+        co_await pfp.promise;
+        co_return result::success();
+    } catch (...) {
+        co_return result::current_exception();
+    }
+
+    static void fillEventArg(rpc::log::Event::Builder arg, const Event & e)
+    {
+        overloaded handlers{
+            [&](const Log & l) {
+                arg.initLog().setLevel(rpc::Verbosity(l.level));
+                RPC_FILL(arg.getLog(), setMsg, l.msg);
+            },
+            [&](const LogEI & l) {
+                auto ei = arg.initLogEI();
+                RPC_FILL(ei, initInfo, l.ei);
+            },
+            [&](const StartActivity & s) {
+                auto sa = arg.initStartActivity();
+                sa.setLevel(rpc::Verbosity(s.level));
+                sa.setId(s.id);
+                sa.setType(rpc::log::to(s.type));
+                RPC_FILL(sa, setText, s.text);
+                sa.setParent(s.parent);
+                RPC_FILL(sa, initFields, s.fields);
+            },
+            [&](const StopActivity & s) { arg.initStopActivity().setId(s.id); },
+            [&](const ActivityResult & r) {
+                auto ar = arg.initResult();
+                ar.setId(r.id);
+                ar.setType(rpc::log::to(r.type));
+                RPC_FILL(ar, initFields, r.fields);
+            },
+        };
+        std::visit(handlers, e);
+    }
+
+    kj::Promise<void> flushLoop()
+    {
+        auto req = co_await flushReq.lock();
+        std::optional<kj::Exception> failure;
+
+        while (true) {
+            kj::Own<kj::PromiseFulfiller<void>> f;
+            if (req->empty()) {
+                co_await req.waitFor(100 * kj::MILLISECONDS);
+            }
+            if (!req->empty()) {
+                f = std::move(req->front());
+                req->pop_front();
+            }
+
+            try {
+                auto buffer = this->buffer.lock()->take();
+                for (auto & e : buffer) {
+                    auto req = remote.pushRequest();
+                    fillEventArg(req.initE(), e);
+                    co_await req.send();
+                }
+                co_await remote.synchronizeRequest().send();
+                if (f) {
+                    f->fulfill();
+                }
+            } catch (...) {
+                this->buffer.lock()->failure = std::current_exception();
+                failure.emplace(kj::getCaughtExceptionAsKj());
+                if (f) {
+                    f->reject(auto(*failure));
+                }
+                break;
+            }
+        }
+
+        while (true) {
+            while (req->empty()) {
+                co_await req.wait();
+            }
+            req->front()->reject(auto(*failure));
+            req->pop_front();
+        }
+    }
+
+    void waitForSpace(NeverAsync) override
+    {
+        struct Fulfiller final : kj::PromiseFulfiller<void>
+        {
+            std::shared_ptr<std::atomic_flag> flag;
+
+            Fulfiller(std::shared_ptr<std::atomic_flag> flag) : flag(flag) {}
+
+            void fulfill(kj::_::Void && value = {}) override
+            {
+                flag->clear();
+                flag->notify_all();
+            }
+
+            void reject(kj::Exception && exception) override
+            {
+                fulfill();
+            }
+
+            bool isWaiting() override
+            {
+                return flag->test();
+            }
+        };
+
+        auto flag = std::make_shared<std::atomic_flag>(true);
+
+        flushReq.lockSync()->emplace_back(kj::heap<Fulfiller>(flag));
+        flushReq.notify();
+        flag->wait(true);
+    }
+};
+}
+}
+
+namespace nix::rpc::log {
+Logger * makeRpcLoggerClient(LogStream::Client remote)
+{
+    return new RpcLogger(remote);
+}
+
+kj::Promise<void> RpcLoggerServer::push(PushContext context)
+try {
+    auto state = Logger::BufferState::HasSpace;
+
+    auto e = context.getParams().getE();
+    if (e.isStartActivity()) {
+        auto args = e.getStartActivity();
+        activities.emplace(
+            args.getId(),
+            parent.addChild(
+                nix::Verbosity(args.getLevel()),
+                from(args.getType()).value_or(actUnknown),
+                rpc::to<std::string>(args.getText()),
+                rpc::to<Logger::Fields>(args.getFields())
+            )
+        );
+    } else if (e.isStopActivity()) {
+        activities.erase(e.getStopActivity().getId());
+    } else if (e.isResult()) {
+        auto args = e.getResult();
+        if (auto type = rpc::log::from(args.getType())) {
+            if (auto act = get(activities, args.getId())) {
+                state = act->result(*type, rpc::to<Logger::Fields>(args.getFields()));
+            }
+        } else {
+            debug("got unintellegible result message %s", args.toString().flatten().cStr());
+        }
+    } else if (e.isLog()) {
+        state = logger->log(
+            nix::Verbosity(e.getLog().getLevel()), rpc::to<std::string_view>(e.getLog().getMsg())
+        );
+    } else if (e.isLogEI()) {
+        state = logger->logEI(from(e.getLogEI().getInfo()));
+    } else {
+        debug("got unintellegible log message %s", e.toString().flatten().cStr());
+    }
+
+    if (state == Logger::BufferState::NeedsFlush) {
+        TRY_AWAIT(parent.getLogger().flush());
+    }
+} catch (std::exception & e) { // NOLINT(lix-foreign-exceptions)
+    printError("error in log processor: %s", e.what());
+    throw; // NOLINT(lix-foreign-exceptions)
+}
+
+kj::Promise<void> RpcLoggerServer::synchronize(SynchronizeContext context)
+{
+    return kj::READY_NOW;
+}
+}

lix/libutil/logging-rpc.hh

@@ -0,0 +1,137 @@
+#pragma once
+///@file
+
+#include "logging.capnp.h"
+#include "logging.hh"
+#include "rpc.hh" // IWYU pragma: keep
+#include <optional>
+
+namespace nix::rpc::log {
+inline std::optional<nix::ActivityType> from(ActivityType at)
+{
+    // clang-format off
+    switch (at) {
+    case ActivityType::UNKNOWN:         return actUnknown;
+    case ActivityType::COPY_PATH:       return actCopyPath;
+    case ActivityType::FILE_TRANSFER:   return actFileTransfer;
+    case ActivityType::REALISE:         return actRealise;
+    case ActivityType::COPY_PATHS:      return actCopyPaths;
+    case ActivityType::BUILDS:          return actBuilds;
+    case ActivityType::BUILD:           return actBuild;
+    case ActivityType::OPTIMISE_STORE:  return actOptimiseStore;
+    case ActivityType::VERIFY_PATHS:    return actVerifyPaths;
+    case ActivityType::SUBSTITUTE:      return actSubstitute;
+    case ActivityType::QUERY_PATH_INFO: return actQueryPathInfo;
+    case ActivityType::POST_BUILD_HOOK: return actPostBuildHook;
+    case ActivityType::BUILD_WAITING:   return actBuildWaiting;
+    default:                            return std::nullopt;
+    }
+    // clang-format on
+}
+
+inline ActivityType to(const nix::ActivityType & at)
+{
+    // clang-format off
+    switch (at) {
+    case actUnknown:       return log::ActivityType::UNKNOWN;
+    case actCopyPath:      return log::ActivityType::COPY_PATH;
+    case actFileTransfer:  return log::ActivityType::FILE_TRANSFER;
+    case actRealise:       return log::ActivityType::REALISE;
+    case actCopyPaths:     return log::ActivityType::COPY_PATHS;
+    case actBuilds:        return log::ActivityType::BUILDS;
+    case actBuild:         return log::ActivityType::BUILD;
+    case actOptimiseStore: return log::ActivityType::OPTIMISE_STORE;
+    case actVerifyPaths:   return log::ActivityType::VERIFY_PATHS;
+    case actSubstitute:    return log::ActivityType::SUBSTITUTE;
+    case actQueryPathInfo: return log::ActivityType::QUERY_PATH_INFO;
+    case actPostBuildHook: return log::ActivityType::POST_BUILD_HOOK;
+    case actBuildWaiting:  return log::ActivityType::BUILD_WAITING;
+    }
+    // clang-format on
+}
+
+inline std::optional<nix::ResultType> from(ResultType rt)
+{
+    // clang-format off
+    switch (rt) {
+    case ResultType::FILE_LINKED:         return resFileLinked;
+    case ResultType::BUILD_LOG_LINE:      return resBuildLogLine;
+    case ResultType::UNTRUSTED_PATH:      return resUntrustedPath;
+    case ResultType::CORRUPTED_PATH:      return resCorruptedPath;
+    case ResultType::SET_PHASE:           return resSetPhase;
+    case ResultType::PROGRESS:            return resProgress;
+    case ResultType::SET_EXPECTED:        return resSetExpected;
+    case ResultType::POST_BUILD_LOG_LINE: return resPostBuildLogLine;
+    default:                              return std::nullopt;
+    }
+    // clang-format on
+}
+
+inline ResultType to(const nix::ResultType & rt)
+{
+    // clang-format off
+    switch (rt) {
+    case resFileLinked:       return ResultType::FILE_LINKED;
+    case resBuildLogLine:     return ResultType::BUILD_LOG_LINE;
+    case resUntrustedPath:    return ResultType::UNTRUSTED_PATH;
+    case resCorruptedPath:    return ResultType::CORRUPTED_PATH;
+    case resSetPhase:         return ResultType::SET_PHASE;
+    case resProgress:         return ResultType::PROGRESS;
+    case resSetExpected:      return ResultType::SET_EXPECTED;
+    case resPostBuildLogLine: return ResultType::POST_BUILD_LOG_LINE;
+    }
+    // clang-format on
+}
+}
+
+namespace nix::rpc {
+template<>
+struct Convert<log::Event::Field, nix::Logger::Field>
+{
+    static nix::Logger::Field convert(log::Event::Field::Reader r, auto &&...)
+    {
+        if (r.isI()) {
+            return {r.getI()};
+        } else {
+            return {to<std::string>(r.getS())};
+        }
+    }
+};
+
+template<>
+struct Fill<log::Event::Field, Logger::Field>
+{
+    static void fill(log::Event::Field::Builder fb, const Logger::Field & e, auto &&...)
+    {
+        if (e.type == Logger::Field::tInt) {
+            fb.setI(e.i);
+        } else {
+            LIX_RPC_FILL(fb, setS, e.s);
+        }
+    }
+};
+}
+
+namespace nix::rpc::log {
+/**
+ * create an RPC-backed logger. this logger will flush its contents periodically
+ * (current fixed to a 100ms interval) or once the log buffer fills up enough to
+ * warrant immediate traffic (currently fixed to approximately 1 MiB of buffered
+ * log traffic). *any* error caught during rpc calls will terminate the process.
+ */
+Logger * makeRpcLoggerClient(LogStream::Client remote);
+
+class RpcLoggerServer : public LogStream::Server
+{
+private:
+    const Activity & parent;
+    std::map<ActivityId, Activity> activities;
+
+public:
+    RpcLoggerServer(const Activity & parent) : parent(parent) {}
+    virtual ~RpcLoggerServer() noexcept(false) = default;
+
+    kj::Promise<void> push(PushContext context) override;
+    kj::Promise<void> synchronize(SynchronizeContext context) override;
+};
+}

lix/libutil/logging.capnp

@@ -0,0 +1,93 @@
+@0xff79d868797f140a;
+
+# IMPORTANT NOTICE
+#
+# these definitions are EXPERIMENTAL and come with NO stability guarantees
+
+using Cxx = import "/capnp/c++.capnp";
+$Cxx.namespace("nix::rpc::log");
+$Cxx.allowCancellation;
+
+using Types = import "/lix/libutil/types.capnp";
+
+enum ActivityType {
+  unknown @0;
+  copyPath @1;
+  fileTransfer @2;
+  realise @3;
+  copyPaths @4;
+  builds @5;
+  build @6;
+  optimiseStore @7;
+  verifyPaths @8;
+  substitute @9;
+  queryPathInfo @10;
+  postBuildHook @11;
+  buildWaiting @12;
+}
+
+enum ResultType {
+  fileLinked @0;
+  buildLogLine @1;
+  untrustedPath @2;
+  corruptedPath @3;
+  setPhase @4;
+  progress @5;
+  setExpected @6;
+  postBuildLogLine @7;
+}
+
+struct Event {
+  struct Field {
+    union {
+      s @0 :Types.String;
+      i @1 :UInt64;
+    }
+  }
+
+  union {
+    log :group {
+      level @0 :Types.Verbosity;
+      msg @1 :Types.String;
+    }
+
+    logEI :group {
+      info @2 :Types.Error;
+    }
+
+    startActivity :group {
+      level @3 :Types.Verbosity;
+      id @4 :UInt64;
+      type @5 :ActivityType;
+      text @6 :Types.String;
+      parent @7 :UInt64;
+      fields @8 :List(Field);
+    }
+
+    stopActivity :group {
+      id @9 :UInt64;
+    }
+
+    result :group {
+      id @10 :UInt64;
+      type @11 :ResultType;
+      fields @12 :List(Field);
+    }
+  }
+}
+
+# NOTE: these streams do not return any result. loggers are expected to
+# be infallible since once the logger fails the only thing we can still
+# do is panic. reporting any status is impossible, and writing to other
+# streams (like stdout/stderr) should be reserved for emergencies only.
+#
+# we also purposely do not model the c++ api to the loggers here. since
+# the rpc definitions are still experimental and used only within *one*
+# defined version of lix on a single machine we can get away with this.
+
+interface LogStream {
+  push @0 (e :Event) -> stream;
+
+  # flush rpc streams and synchronize capnp flow control/error state
+  synchronize @1 ();
+}

lix/libutil/logging.hh

@@ -167,7 +167,7 @@ public:
         return {result::success()};
     }
 
-    virtual void waitForSpace() {}
+    virtual void waitForSpace(NeverAsync = {}) {}
 
 protected:
     virtual BufferState startActivityImpl(

lix/libutil/meson.build

@@ -25,6 +25,7 @@ libutil_sources = files(
   'hilite.cc',
   'io-buffer.cc',
   'json-utils.cc',
+  'logging-rpc.cc',
   'logging.cc',
   'monitor-fd.cc',
   'mount.cc',
@@ -102,6 +103,7 @@ libutil_headers = files(
   'json-fwd.hh',
   'json.hh',
   'linear-map.hh',
+  'logging-rpc.hh',
   'logging.hh',
   'lru-cache.hh',
   'manually-drop.hh',
@@ -296,33 +298,35 @@ libutil_settings_headers += custom_target(
 
 libutil_rpc_headers = []
 libutil_rpc_sources = []
-libutil_rpc = []
 
-libutil_rpc += custom_target(
-  command : [
-    capnpc_wrapper,
-    '--language=c++',
-    '--src-prefix=@CURRENT_SOURCE_DIR@',
-    '--outdir=@OUTDIR@',
-    '--depfile=@DEPFILE@',
-    '-I@SOURCE_ROOT@',
-    '@INPUT@',
-  ],
-  input : files(
-    # keep-sorted start
-    'types.capnp',
-    # keep-sorted end
-  ),
-  output : [
-    '@PLAINNAME@.h',
-    '@PLAINNAME@.c++',
-  ],
-  install : true,
-  install_dir : [includedir / 'lix/libutil', false],
-  depfile : '@PLAINNAME@.d',
+libutil_rpc_schemas = files(
+  # keep-sorted start
+  'logging.capnp',
+  'types.capnp',
+  # keep-sorted end
 )
 
-foreach rpc : libutil_rpc
+foreach rpc : libutil_rpc_schemas
+  rpc = custom_target(
+    command : [
+      capnpc_wrapper,
+      '--language=c++',
+      '--src-prefix=@CURRENT_SOURCE_DIR@',
+      '--outdir=@OUTDIR@',
+      '--depfile=@DEPFILE@',
+      '-I@SOURCE_ROOT@',
+      '@INPUT@',
+    ],
+    input : rpc,
+    output : [
+      '@PLAINNAME@.h',
+      '@PLAINNAME@.c++',
+    ],
+    install : true,
+    install_dir : [includedir / 'lix/libutil', false],
+    depfile : '@PLAINNAME@.d',
+  )
+
   libutil_rpc_headers += rpc[0]
   libutil_rpc_sources += rpc[1]
 endforeach
@@ -332,6 +336,7 @@ dependencies = [
   brotli,
   cpuid,
   kj,
+  capnp_rpc,
   libarchive,
   nlohmann_json,
   openssl,
@@ -410,6 +415,7 @@ liblixutil = declare_dependency(
     # Everything has to link to kj if it uses libutil internally (ensured by
     # lix-base pkg-config externally)
     kj,
+    capnp_rpc,
     libarchive,
     nlohmann_json,
     libatomic,

lix/libutil/types.capnp

@@ -3,6 +3,9 @@
 using Cxx = import "/capnp/c++.capnp";
 $Cxx.namespace("nix::rpc");
 
+# many of our strings must be nul-safe :(
+using String = Data;
+
 enum Verbosity {
   error @0;
   warn @1;