libutil: convert readFile to generator

Change-Id: I5f92b15fd367d46eb047d74ab6e317b4f51a46d3

src/libcmd/built-path.cc

@@ -1,6 +1,7 @@
 #include "built-path.hh"
 #include "derivations.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 

src/libcmd/command.cc

@@ -5,6 +5,7 @@
 #include "nixexpr.hh"
 #include "profiles.hh"
 #include "repl.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 

src/libcmd/installable-attr-path.cc

@@ -16,6 +16,7 @@
 #include "url.hh"
 #include "registry.hh"
 #include "build-result.hh"
+#include "overloaded.hh"
 
 #include <regex>
 #include <queue>

src/libcmd/installable-derived-path.cc

@@ -1,5 +1,6 @@
 #include "installable-derived-path.hh"
 #include "derivations.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libcmd/installable-flake.cc

@@ -17,6 +17,7 @@
 #include "url.hh"
 #include "registry.hh"
 #include "build-result.hh"
+#include "overloaded.hh"
 
 #include <regex>
 #include <queue>

src/libcmd/installables.cc

@@ -20,6 +20,7 @@
 #include "url.hh"
 #include "registry.hh"
 #include "build-result.hh"
+#include "overloaded.hh"
 
 #include <regex>
 #include <queue>

src/libexpr/eval-cache.cc

@@ -3,6 +3,7 @@
 #include "eval.hh"
 #include "eval-inline.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 namespace nix::eval_cache {
 

src/libexpr/eval.cc

@@ -19,6 +19,7 @@
 #include "fetch-to-store.hh"
 #include "flake/flakeref.hh"
 #include "parser-tab.hh"
+#include "overloaded.hh"
 
 #include <algorithm>
 #include <chrono>

src/libexpr/flake/flake.cc

@@ -8,6 +8,7 @@
 #include "fetchers.hh"
 #include "finally.hh"
 #include "fetch-settings.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libexpr/parser-state.hh

@@ -2,6 +2,7 @@
 ///@file
 
 #include "eval.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libexpr/primops.cc

@@ -15,6 +15,7 @@
 #include "value-to-xml.hh"
 #include "primops.hh"
 #include "fetch-to-store.hh"
+#include "overloaded.hh"
 
 #include <boost/container/small_vector.hpp>
 #include <nlohmann/json.hpp>

src/libexpr/primops/context.cc

@@ -2,6 +2,7 @@
 #include "eval-inline.hh"
 #include "derivations.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libexpr/value/context.cc

@@ -1,4 +1,5 @@
 #include "value/context.hh"
+#include "overloaded.hh"
 
 #include <optional>
 

src/libfetchers/path.cc

@@ -130,10 +130,8 @@ struct PathInputScheme : InputScheme
         time_t mtime = 0;
         if (!storePath || storePath->name() != "source" || !store->isValidPath(*storePath)) {
             // FIXME: try to substitute storePath.
-            auto src = sinkToSource([&](Sink & sink) {
-                mtime = dumpPathAndGetMtime(absPath, sink, defaultPathFilter);
-            });
-            storePath = store->addToStoreFromDump(*src, "source");
+            auto src = WireSource{dumpPathAndGetMtime(absPath, mtime, defaultPathFilter)};
+            storePath = store->addToStoreFromDump(src, "source");
         }
         input.attrs.insert_or_assign("lastModified", uint64_t(mtime));
 

src/libfetchers/tarball.cc

@@ -71,7 +71,7 @@ DownloadFileResult downloadFile(
         storePath = std::move(cached->storePath);
     } else {
         StringSink sink;
-        dumpString(res.data, sink);
+        sink << dumpString(res.data);
         auto hash = hashString(htSHA256, res.data);
         ValidPathInfo info {
             *store,

src/libstore/binary-cache-store.cc

@@ -5,6 +5,7 @@
 #include "fs-accessor.hh"
 #include "globals.hh"
 #include "nar-info.hh"
+#include "serialise.hh"
 #include "sync.hh"
 #include "remote-fs-accessor.hh"
 #include "nar-info-disk-cache.hh"
@@ -413,16 +414,14 @@ StorePath BinaryCacheStore::addToStore(
 
     HashSink sink { hashAlgo };
     if (method == FileIngestionMethod::Recursive) {
-        dumpPath(srcPath, sink, filter);
+        sink << dumpPath(srcPath, filter);
     } else {
-        readFile(srcPath, sink);
+        sink << readFileSource(srcPath);
     }
     auto h = sink.finish().first;
 
-    auto source = sinkToSource([&](Sink & sink) {
-        dumpPath(srcPath, sink, filter);
-    });
-    return addToStoreCommon(*source, repair, CheckSigs, [&](HashResult nar) {
+    auto source = WireSource{dumpPath(srcPath, filter)};
+    return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
         ValidPathInfo info {
             *this,
             name,
@@ -455,7 +454,7 @@ StorePath BinaryCacheStore::addTextToStore(
         return path;
 
     StringSink sink;
-    dumpString(s, sink);
+    sink << dumpString(s);
     StringSource source(sink.s);
     return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
         ValidPathInfo info {

src/libstore/build/derivation-goal.cc

@@ -13,6 +13,7 @@
 #include "topo-sort.hh"
 #include "callback.hh"
 #include "local-store.hh" // TODO remove, along with remaining downcasts
+#include "overloaded.hh"
 
 #include <regex>
 #include <queue>

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

@@ -16,6 +16,7 @@
 #include "cgroup.hh"
 #include "personality.hh"
 #include "namespaces.hh"
+#include "overloaded.hh"
 
 #include <regex>
 #include <queue>
@@ -2392,7 +2393,7 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
                 /* FIXME: Is this actually streaming? */
                 auto source = sinkToSource([&](Sink & nextSink) {
                     RewritingSink rsink(rewrites, nextSink);
-                    dumpPath(actualPath, rsink);
+                    rsink << dumpPath(actualPath);
                     rsink.flush();
                 });
                 Path tmpPath = actualPath + ".tmp";
@@ -2453,15 +2454,15 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
             HashModuloSink caSink { outputHash.hashType, oldHashPart };
             std::visit(overloaded {
                 [&](const TextIngestionMethod &) {
-                    readFile(actualPath, caSink);
+                    caSink << readFileSource(actualPath);
                 },
                 [&](const FileIngestionMethod & m2) {
                     switch (m2) {
                     case FileIngestionMethod::Recursive:
-                        dumpPath(actualPath, caSink);
+                        caSink << dumpPath(actualPath);
                         break;
                     case FileIngestionMethod::Flat:
-                        readFile(actualPath, caSink);
+                        caSink << readFileSource(actualPath);
                         break;
                     }
                 },

src/libstore/build/worker.cc

@@ -5,6 +5,7 @@
 #include "local-derivation-goal.hh"
 #include "signals.hh"
 #include "hook-instance.hh"
+#include "overloaded.hh"
 
 #include <poll.h>
 

src/libstore/content-address.cc

@@ -1,6 +1,7 @@
 #include "args.hh"
 #include "content-address.hh"
 #include "split.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/daemon.cc

@@ -13,6 +13,7 @@
 #include "archive.hh"
 #include "derivations.hh"
 #include "args.hh"
+#include "overloaded.hh"
 
 namespace nix::daemon {
 
@@ -161,8 +162,7 @@ struct TunnelSink : Sink
     TunnelSink(Sink & to) : to(to) { }
     void operator () (std::string_view data)
     {
-        to << STDERR_WRITE;
-        writeString(data, to);
+        to << STDERR_WRITE << data;
     }
 };
 
@@ -867,7 +867,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         auto path = store->parseStorePath(readString(from));
         logger->startWork();
         logger->stopWork();
-        dumpPath(store->toRealPath(path), to);
+        to << dumpPath(store->toRealPath(path));
         break;
     }
 

src/libstore/derivations.cc

@@ -8,6 +8,7 @@
 #include "common-protocol.hh"
 #include "common-protocol-impl.hh"
 #include "fs-accessor.hh"
+#include "overloaded.hh"
 #include <boost/container/small_vector.hpp>
 #include <nlohmann/json.hpp>
 

src/libstore/derived-path-map.cc

@@ -1,4 +1,5 @@
 #include "derived-path-map.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/derived-path.cc

@@ -1,5 +1,6 @@
 #include "derived-path.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 

src/libstore/downstream-placeholder.cc

@@ -1,5 +1,6 @@
 #include "downstream-placeholder.hh"
 #include "derivations.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/legacy-ssh-store.cc

@@ -10,6 +10,7 @@
 #include "ssh.hh"
 #include "derivations.hh"
 #include "callback.hh"
+#include "overloaded.hh"
 
 namespace nix {
 
@@ -46,7 +47,6 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
         FdSink to;
         FdSource from;
         ServeProto::Version remoteVersion;
-        bool good = true;
 
         /**
          * Coercion to `ServeProto::ReadConn`. This makes it easy to use the
@@ -97,8 +97,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
         , host(host)
         , connections(make_ref<Pool<Connection>>(
             std::max(1, (int) maxConnections),
-            [this]() { return openConnection(); },
-            [](const ref<Connection> & r) { return r->good; }
+            [this]() { return openConnection(); }
             ))
         , master(
             host,
@@ -199,7 +198,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
             try {
                 copyNAR(source, conn->to);
             } catch (...) {
-                conn->good = false;
+                auto _dropConnDuringUnwind = std::move(conn);
                 throw;
             }
             conn->to.flush();
@@ -212,7 +211,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
             try {
                 copyNAR(source, conn->to);
             } catch (...) {
-                conn->good = false;
+                auto _dropConnDuringUnwind = std::move(conn);
                 throw;
             }
             conn->to

src/libstore/local-binary-cache-store.cc

@@ -71,7 +71,7 @@ protected:
     void getFile(const std::string & path, Sink & sink) override
     {
         try {
-            readFile(binaryCacheDir + "/" + path, sink);
+            sink << readFileSource(binaryCacheDir + "/" + path);
         } catch (SysError & e) {
             if (e.errNo == ENOENT)
                 throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache", path);

src/libstore/local-fs-store.cc

@@ -82,7 +82,7 @@ void LocalFSStore::narFromPath(const StorePath & path, Sink & sink)
 {
     if (!isValidPath(path))
         throw Error("path '%s' is not valid", printStorePath(path));
-    dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()), sink);
+    sink << dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()));
 }
 
 const std::string LocalFSStore::drvsLogDir = "drvs";

src/libstore/local-store.cc

@@ -11,6 +11,7 @@
 #include "signals.hh"
 #include "finally.hh"
 #include "compression.hh"
+#include "overloaded.hh"
 
 #include <iostream>
 #include <algorithm>
@@ -1331,8 +1332,7 @@ StorePath LocalStore::addToStoreFromDump(Source & source0, std::string_view name
 
     if (!inMemory) {
         /* Drain what we pulled so far, and then keep on pulling */
-        StringSource dumpSource { dump };
-        ChainSource bothSource { dumpSource, source };
+        ChainSource bothSource { StringSource{dump}, std::move(source) };
 
         std::tie(tempDir, tempDirFd) = createTempDirInStore();
         delTempDir = std::make_unique<AutoDelete>(tempDir);
@@ -1395,7 +1395,7 @@ StorePath LocalStore::addToStoreFromDump(Source & source0, std::string_view name
             auto narHash = std::pair { hash, size };
             if (method != FileIngestionMethod::Recursive || hashAlgo != htSHA256) {
                 HashSink narSink { htSHA256 };
-                dumpPath(realPath, narSink);
+                narSink << dumpPath(realPath);
                 narHash = narSink.finish();
             }
 
@@ -1450,7 +1450,7 @@ StorePath LocalStore::addTextToStore(
             canonicalisePathMetaData(realPath, {});
 
             StringSink sink;
-            dumpString(s, sink);
+            sink << dumpString(s);
             auto narHash = hashString(htSHA256, sink.s);
 
             optimisePath(realPath, repair);
@@ -1590,7 +1590,7 @@ bool LocalStore::verifyStore(bool checkContents, RepairFlag repair)
 
                 auto hashSink = HashSink(info->narHash.type);
 
-                dumpPath(Store::toRealPath(i), hashSink);
+                hashSink << dumpPath(Store::toRealPath(i));
                 auto current = hashSink.finish();
 
                 if (info->narHash != nullHash && info->narHash != current.first) {
@@ -1886,15 +1886,15 @@ ContentAddress LocalStore::hashCAPath(
     HashModuloSink caSink ( hashType, std::string(pathHash) );
     std::visit(overloaded {
         [&](const TextIngestionMethod &) {
-            readFile(path, caSink);
+            caSink << readFileSource(path);
         },
         [&](const FileIngestionMethod & m2) {
             switch (m2) {
             case FileIngestionMethod::Recursive:
-                dumpPath(path, caSink);
+                caSink << dumpPath(path);
                 break;
             case FileIngestionMethod::Flat:
-                readFile(path, caSink);
+                caSink << readFileSource(path);
                 break;
             }
         },

src/libstore/misc.cc

@@ -8,6 +8,7 @@
 #include "callback.hh"
 #include "closure.hh"
 #include "filetransfer.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/outputs-spec.cc

@@ -5,6 +5,7 @@
 #include "regex-combinators.hh"
 #include "outputs-spec.hh"
 #include "path-regex.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/path-info.cc

@@ -1,5 +1,6 @@
 #include "path-info.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libstore/path-references.cc

@@ -65,7 +65,7 @@ StorePathSet scanForReferences(
     TeeSink sink { refsSink, toTee };
 
     /* Look for the hashes in the NAR dump of the path. */
-    dumpPath(path, sink);
+    sink << dumpPath(path);
 
     return refsSink.getResultPaths();
 }

src/libstore/path-with-outputs.cc

@@ -1,5 +1,6 @@
 #include "path-with-outputs.hh"
 #include "store-api.hh"
+#include "overloaded.hh"
 
 #include <regex>
 

src/libstore/remote-store.cc

@@ -17,6 +17,7 @@
 #include "logging.hh"
 #include "callback.hh"
 #include "filetransfer.hh"
+#include "overloaded.hh"
 #include <nlohmann/json.hpp>
 
 namespace nix {
@@ -39,9 +40,7 @@ RemoteStore::RemoteStore(const Params & params)
             },
             [this](const ref<Connection> & r) {
                 return
-                    r->to.good()
-                    && r->from.good()
-                    && std::chrono::duration_cast<std::chrono::seconds>(
+                    std::chrono::duration_cast<std::chrono::seconds>(
                         std::chrono::steady_clock::now() - r->startTime).count() < maxConnectionAge;
             }
             ))
@@ -155,7 +154,6 @@ void RemoteStore::setOptions(Connection & conn)
 RemoteStore::ConnectionHandle::~ConnectionHandle()
 {
     if (!daemonException && std::uncaught_exceptions()) {
-        handle.markBad();
         debug("closing daemon connection because of an exception");
     }
 }
@@ -181,6 +179,10 @@ void RemoteStore::ConnectionHandle::processStderr(Sink * sink, Source * source,
                     m.find("Derive([") != std::string::npos)
                     throw Error("%s, this might be because the daemon is too old to understand dependencies on dynamic derivations. Check to see if the raw derivation is in the form '%s'", std::move(m), "DrvWithVersion(..)");
             }
+            // the daemon can still handle more requests, so the connection itself
+            // is still valid. the current *handle* however should be considered a
+            // lost cause and abandoned entirely.
+            handle.release();
             throw;
         }
     }
@@ -471,7 +473,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
                             dump.drainInto(conn->to);
                         } else {
                             std::string contents = dump.drain();
-                            dumpString(contents, conn->to);
+                            conn->to << dumpString(contents);
                         }
                     }
                     conn.processStderr();
@@ -940,11 +942,6 @@ std::optional<TrustedFlag> RemoteStore::isTrustedClient()
     return conn->remoteTrustsUs;
 }
 
-void RemoteStore::flushBadConnections()
-{
-    connections->flushBad();
-}
-
 
 RemoteStore::Connection::~Connection()
 {
@@ -1004,7 +1001,7 @@ std::exception_ptr RemoteStore::Connection::processStderr(Sink * sink, Source *
             if (!source) throw Error("no source");
             size_t len = readNum<size_t>(from);
             auto buf = std::make_unique<char[]>(len);
-            writeString({(const char *) buf.get(), source->read(buf.get(), len)}, to);
+            to << std::string_view((const char *) buf.get(), source->read(buf.get(), len));
             to.flush();
         }
 
@@ -1073,27 +1070,15 @@ void RemoteStore::ConnectionHandle::withFramedSink(std::function<void(Sink & sin
 
     Finally joinStderrThread([&]()
     {
-        if (stderrThread.joinable()) {
         stderrThread.join();
         if (ex) {
-                try {
             std::rethrow_exception(ex);
-                } catch (...) {
-                    ignoreException();
-                }
-            }
         }
     });
 
-    {
     FramedSink sink((*this)->to, ex);
     fun(sink);
     sink.flush();
-    }
-
-    stderrThread.join();
-    if (ex)
-        std::rethrow_exception(ex);
 }
 
 }

src/libstore/remote-store.hh

@@ -161,8 +161,6 @@ public:
 
     std::optional<TrustedFlag> isTrustedClient() override;
 
-    void flushBadConnections();
-
     struct Connection;
 
     ref<Connection> openConnectionWrapper();

src/libstore/store-api.cc

@@ -3,6 +3,7 @@
 #include "globals.hh"
 #include "derivations.hh"
 #include "store-api.hh"
+#include "serialise.hh"
 #include "util.hh"
 #include "nar-info-disk-cache.hh"
 #include "thread-pool.hh"
@@ -15,6 +16,7 @@
 // FIXME this should not be here, see TODO below on
 // `addMultipleToStore`.
 #include "worker-protocol.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 #include <regex>
@@ -269,13 +271,12 @@ StorePath Store::addToStore(
     const StorePathSet & references)
 {
     Path srcPath(absPath(_srcPath));
-    auto source = sinkToSource([&](Sink & sink) {
-        if (method == FileIngestionMethod::Recursive)
-            dumpPath(srcPath, sink, filter);
-        else
-            readFile(srcPath, sink);
-    });
-    return addToStoreFromDump(*source, name, method, hashAlgo, repair, references);
+    auto source = WireSource{
+        method == FileIngestionMethod::Recursive
+        ? static_cast<Generator<std::span<const char>, void>>(dumpPath(srcPath, filter))
+        : readFileSource(srcPath)
+    };
+    return addToStoreFromDump(source, name, method, hashAlgo, repair, references);
 }
 
 void Store::addMultipleToStore(
@@ -420,13 +421,11 @@ ValidPathInfo Store::addToStoreSlow(std::string_view name, const Path & srcPath,
     /* Functionally, this means that fileSource will yield the content of
        srcPath. The fact that we use scratchpadSink as a temporary buffer here
        is an implementation detail. */
-    auto fileSource = sinkToSource([&](Sink & scratchpadSink) {
-        dumpPath(srcPath, scratchpadSink);
-    });
+    auto fileSource = WireSource{dumpPath(srcPath)};
 
     /* tapped provides the same data as fileSource, but we also write all the
        information to narSink. */
-    TeeSource tapped { *fileSource, narSink };
+    TeeSource tapped { fileSource, narSink };
 
     ParseSink blank;
     auto & parseSink = method == FileIngestionMethod::Flat
@@ -461,10 +460,8 @@ ValidPathInfo Store::addToStoreSlow(std::string_view name, const Path & srcPath,
     info.narSize = narSize;
 
     if (!isValidPath(info.path)) {
-        auto source = sinkToSource([&](Sink & scratchpadSink) {
-            dumpPath(srcPath, scratchpadSink);
-        });
-        addToStore(info, *source);
+        auto source = WireSource{dumpPath(srcPath)};
+        addToStore(info, source);
     }
 
     return info;

src/libstore/worker-protocol.cc

@@ -7,6 +7,7 @@
 #include "worker-protocol-impl.hh"
 #include "archive.hh"
 #include "path-info.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 

src/libutil/archive.cc

@@ -12,6 +12,7 @@
 #include <fcntl.h>
 
 #include "archive.hh"
+#include "serialise.hh"
 #include "util.hh"
 #include "config.hh"
 #include "signals.hh"
@@ -39,10 +40,10 @@ static GlobalConfig::Register rArchiveSettings(&archiveSettings);
 PathFilter defaultPathFilter = [](const Path &) { return true; };
 
 
-static void dumpContents(const Path & path, off_t size,
-    Sink & sink)
+static WireFormatGenerator dumpContents(const Path & path, off_t size)
 {
-    sink << "contents" << size;
+    co_yield "contents";
+    co_yield size;
 
     AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
     if (!fd) throw SysError("opening file '%1%'", path);
@@ -54,31 +55,35 @@ static void dumpContents(const Path & path, off_t size,
         auto n = std::min(left, buf.size());
         readFull(fd.get(), buf.data(), n);
         left -= n;
-        sink({buf.data(), n});
+        co_yield std::span{buf.data(), n};
     }
 
-    writePadding(size, sink);
+    co_yield SerializingTransform::padding(size);
 }
 
 
-static time_t dump(const Path & path, Sink & sink, PathFilter & filter)
+static WireFormatGenerator dump(const Path & path, time_t & mtime, PathFilter & filter)
 {
     checkInterrupt();
 
     auto st = lstat(path);
-    time_t result = st.st_mtime;
+    mtime = st.st_mtime;
 
-    sink << "(";
+    co_yield "(";
 
     if (S_ISREG(st.st_mode)) {
-        sink << "type" << "regular";
-        if (st.st_mode & S_IXUSR)
-            sink << "executable" << "";
-        dumpContents(path, st.st_size, sink);
+        co_yield "type";
+        co_yield "regular";
+        if (st.st_mode & S_IXUSR) {
+            co_yield "executable";
+            co_yield "";
+        }
+        co_yield dumpContents(path, st.st_size);
     }
 
     else if (S_ISDIR(st.st_mode)) {
-        sink << "type" << "directory";
+        co_yield "type";
+        co_yield "directory";
 
         /* If we're on a case-insensitive system like macOS, undo
            the case hack applied by restorePath(). */
@@ -100,41 +105,55 @@ static time_t dump(const Path & path, Sink & sink, PathFilter & filter)
 
         for (auto & i : unhacked)
             if (filter(path + "/" + i.first)) {
-                sink << "entry" << "(" << "name" << i.first << "node";
-                auto tmp_mtime = dump(path + "/" + i.second, sink, filter);
-                if (tmp_mtime > result) {
-                    result = tmp_mtime;
+                co_yield "entry";
+                co_yield "(";
+                co_yield "name";
+                co_yield i.first;
+                co_yield "node";
+                time_t tmp_mtime;
+                co_yield dump(path + "/" + i.second, tmp_mtime, filter);
+                if (tmp_mtime > mtime) {
+                    mtime = tmp_mtime;
                 }
-                sink << ")";
+                co_yield ")";
             }
     }
 
-    else if (S_ISLNK(st.st_mode))
-        sink << "type" << "symlink" << "target" << readLink(path);
+    else if (S_ISLNK(st.st_mode)) {
+        co_yield "type";
+        co_yield "symlink";
+        co_yield "target";
+        co_yield readLink(path);
+    }
 
     else throw Error("file '%1%' has an unsupported type", path);
 
-    sink << ")";
-
-    return result;
+    co_yield ")";
 }
 
 
-time_t dumpPathAndGetMtime(const Path & path, Sink & sink, PathFilter & filter)
+WireFormatGenerator dumpPathAndGetMtime(const Path & path, time_t & mtime, PathFilter & filter)
 {
-    sink << narVersionMagic1;
-    return dump(path, sink, filter);
+    co_yield narVersionMagic1;
+    co_yield dump(path, mtime, filter);
 }
 
-void dumpPath(const Path & path, Sink & sink, PathFilter & filter)
+WireFormatGenerator dumpPath(const Path & path, PathFilter & filter)
 {
-    dumpPathAndGetMtime(path, sink, filter);
+    time_t ignored;
+    co_yield dumpPathAndGetMtime(path, ignored, filter);
 }
 
 
-void dumpString(std::string_view s, Sink & sink)
+WireFormatGenerator dumpString(std::string_view s)
 {
-    sink << narVersionMagic1 << "(" << "type" << "regular" << "contents" << s << ")";
+    co_yield narVersionMagic1;
+    co_yield "(";
+    co_yield "type";
+    co_yield "regular";
+    co_yield "contents";
+    co_yield s;
+    co_yield ")";
 }
 
 
@@ -391,10 +410,8 @@ void copyNAR(Source & source, Sink & sink)
 
 void copyPath(const Path & from, const Path & to)
 {
-    auto source = sinkToSource([&](Sink & sink) {
-        dumpPath(from, sink);
-    });
-    restorePath(to, *source);
+    auto source = WireSource{dumpPath(from)};
+    restorePath(to, source);
 }
 
 

src/libutil/archive.hh

@@ -56,13 +56,13 @@ namespace nix {
  *   `+` denotes string concatenation.
  * ```
  */
-void dumpPath(const Path & path, Sink & sink,
+WireFormatGenerator dumpPath(const Path & path,
     PathFilter & filter = defaultPathFilter);
 
 /**
  * Same as dumpPath(), but returns the last modified date of the path.
  */
-time_t dumpPathAndGetMtime(const Path & path, Sink & sink,
+WireFormatGenerator dumpPathAndGetMtime(const Path & path, time_t & mtime,
     PathFilter & filter = defaultPathFilter);
 
 /**
@@ -70,7 +70,7 @@ time_t dumpPathAndGetMtime(const Path & path, Sink & sink,
  *
  * @param s Contents of the file.
  */
-void dumpString(std::string_view s, Sink & sink);
+WireFormatGenerator dumpString(std::string_view s);
 
 /**
  * \todo Fix this API, it sucks.

src/libutil/finally.hh

@@ -19,5 +19,5 @@ public:
     Finally(Finally &&other) : fun(std::move(other.fun)) {
         other.movedFrom = true;
     }
-    ~Finally() { if (!movedFrom) fun(); }
+    ~Finally() noexcept(false) { if (!movedFrom) fun(); }
 };

src/libutil/generator.hh

@@ -0,0 +1,192 @@
+#pragma once
+
+#include "overloaded.hh"
+
+#include <coroutine>
+#include <optional>
+#include <utility>
+#include <variant>
+
+namespace nix {
+
+template<typename T, typename Transform = std::identity>
+struct Generator
+{
+    struct promise_type;
+    using handle_type = std::coroutine_handle<promise_type>;
+
+    explicit Generator(handle_type h) : impl{h, h.promise().state} {}
+
+    explicit operator bool()
+    {
+        return bool(impl);
+    }
+    T operator()()
+    {
+        return impl();
+    }
+
+    operator Generator<T, void> &() &
+    {
+        return impl;
+    }
+    operator Generator<T, void>() &&
+    {
+        return std::move(impl);
+    }
+
+private:
+    Generator<T, void> impl;
+};
+
+template<typename T>
+struct Generator<T, void>
+{
+    template<typename, typename>
+    friend struct Generator;
+    template<typename T2, typename Transform>
+    friend struct Generator<T2, Transform>::promise_type;
+
+    struct promise_state;
+
+    struct _link
+    {
+        std::coroutine_handle<> handle{};
+        promise_state * state{};
+    };
+
+    struct promise_state
+    {
+        std::variant<_link, T> value{};
+        std::exception_ptr exception{};
+        _link parent{};
+    };
+
+    // NOTE coroutine handles are LiteralType, own a memory resource (that may
+    // itself own unique resources), and are "typically TriviallyCopyable". we
+    // need to take special care to wrap this into a less footgunny interface,
+    // which mostly means move-only.
+    Generator(Generator && other)
+    {
+        swap(other);
+    }
+
+    Generator & operator=(Generator && other)
+    {
+        Generator(std::move(other)).swap(*this);
+        return *this;
+    }
+
+    ~Generator()
+    {
+        if (h) {
+            h.destroy();
+        }
+    }
+
+    explicit operator bool()
+    {
+        return ensure();
+    }
+
+    T operator()()
+    {
+        ensure();
+        auto result = std::move(*current);
+        current = nullptr;
+        return result;
+    }
+
+protected:
+    std::coroutine_handle<> h{};
+    _link active{};
+    T * current{};
+
+    Generator(std::coroutine_handle<> h, promise_state & state) : h(h), active(h, &state) {}
+
+    void swap(Generator & other)
+    {
+        std::swap(h, other.h);
+        std::swap(active, other.active);
+        std::swap(current, other.current);
+    }
+
+    bool ensure()
+    {
+        while (!current && active.handle) {
+            active.handle.resume();
+            auto & p = *active.state;
+            if (p.exception) {
+                std::rethrow_exception(p.exception);
+            } else if (active.handle.done()) {
+                active = p.parent;
+            } else {
+                std::visit(
+                    overloaded{
+                        [&](_link & inner) {
+                            auto base = inner.state;
+                            while (base->parent.handle) {
+                                base = base->parent.state;
+                            }
+                            base->parent = active;
+                            active = inner;
+                        },
+                        [&](T & value) { current = &value; },
+                    },
+                    p.value
+                );
+            }
+        }
+        return current;
+    }
+};
+
+template<typename T, typename Transform>
+struct Generator<T, Transform>::promise_type
+{
+    Generator<T, void>::promise_state state;
+    Transform convert;
+    std::optional<Generator<T, void>> inner;
+
+    Generator get_return_object()
+    {
+        return Generator(handle_type::from_promise(*this));
+    }
+    std::suspend_always initial_suspend()
+    {
+        return {};
+    }
+    std::suspend_always final_suspend() noexcept
+    {
+        return {};
+    }
+    void unhandled_exception()
+    {
+        state.exception = std::current_exception();
+    }
+
+    template<typename From>
+        requires requires(Transform t, From && f) {
+            {
+                t(std::forward<From>(f))
+            } -> std::convertible_to<T>;
+        }
+    std::suspend_always yield_value(From && from)
+    {
+        state.value = convert(std::forward<From>(from));
+        return {};
+    }
+
+    template<typename From>
+        requires requires(Transform t, From f) { static_cast<Generator<T, void>>(t(std::move(f))); }
+    std::suspend_always yield_value(From from)
+    {
+        inner = static_cast<Generator<T, void>>(convert(std::move(from)));
+        state.value = inner->active;
+        return {};
+    }
+
+    void return_void() {}
+};
+
+}

src/libutil/hash.cc

@@ -325,7 +325,7 @@ Hash hashString(HashType ht, std::string_view s)
 Hash hashFile(HashType ht, const Path & path)
 {
     HashSink sink(ht);
-    readFile(path, sink);
+    sink << readFileSource(path);
     return sink.finish().first;
 }
 
@@ -371,7 +371,7 @@ HashResult hashPath(
     HashType ht, const Path & path, PathFilter & filter)
 {
     HashSink sink(ht);
-    dumpPath(path, sink, filter);
+    sink << dumpPath(path, filter);
     return sink.finish();
 }
 

src/libutil/overloaded.hh

@@ -0,0 +1,17 @@
+#pragma once
+
+namespace nix {
+
+/**
+ * C++17 std::visit boilerplate
+ */
+template<class... Ts>
+struct overloaded : Ts...
+{
+    using Ts::operator()...;
+};
+
+template<class... Ts>
+overloaded(Ts...) -> overloaded<Ts...>;
+
+}

src/libutil/pool.hh

@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include <exception>
 #include <functional>
 #include <limits>
 #include <list>
@@ -102,12 +103,24 @@ public:
     private:
         Pool & pool;
         std::shared_ptr<R> r;
-        bool bad = false;
 
         friend Pool;
 
         Handle(Pool & pool, std::shared_ptr<R> r) : pool(pool), r(r) { }
 
+        void drop(bool stillValid)
+        {
+            {
+                auto state_(pool.state.lock());
+                if (stillValid)
+                    state_->idle.emplace_back(std::move(r));
+                assert(state_->inUse);
+                state_->inUse--;
+            }
+            pool.wakeup.notify_one();
+            r = nullptr;
+        }
+
     public:
         Handle(Handle && h) : pool(h.pool), r(h.r) { h.r.reset(); }
 
@@ -115,25 +128,27 @@ public:
 
         ~Handle()
         {
-            if (!r) return;
-            {
-                auto state_(pool.state.lock());
-                if (!bad)
-                    state_->idle.push_back(ref<R>(r));
-                assert(state_->inUse);
-                state_->inUse--;
+            if (r)
+                drop(std::uncaught_exceptions() == 0);
         }
-            pool.wakeup.notify_one();
+
+        void release()
+        {
+            drop(true);
         }
 
         R * operator -> () { return &*r; }
         R & operator * () { return *r; }
-
-        void markBad() { bad = true; }
     };
 
     Handle get()
     {
+        // we do not want to handle the complexity that comes with allocating
+        // resources during stack unwinding. it would be possible to do this,
+        // but doing so requires more per-handle bookkeeping to properly free
+        // resources allocated during unwinding. that effort is not worth it.
+        assert(std::uncaught_exceptions() == 0);
+
         {
             auto state_(state.lock());
 
@@ -177,16 +192,6 @@ public:
     {
         return state.lock()->max;
     }
-
-    void flushBad()
-    {
-        auto state_(state.lock());
-        std::vector<ref<R>> left;
-        for (auto & p : state_->idle)
-            if (validator(p))
-                left.push_back(p);
-        std::swap(state_->idle, left);
-    }
 };
 
 }

src/libutil/position.cc

@@ -1,4 +1,5 @@
 #include "position.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/libutil/serialise.cc

@@ -52,18 +52,7 @@ FdSink::~FdSink()
 void FdSink::writeUnbuffered(std::string_view data)
 {
     written += data.size();
-    try {
     writeFull(fd, data);
-    } catch (SysError & e) {
-        _good = false;
-        throw;
-    }
-}
-
-
-bool FdSink::good()
-{
-    return _good;
 }
 
 
@@ -128,19 +117,13 @@ size_t FdSource::readUnbuffered(char * data, size_t len)
         checkInterrupt();
         n = ::read(fd, data, len);
     } while (n == -1 && errno == EINTR);
-    if (n == -1) { _good = false; throw SysError("reading from file"); }
-    if (n == 0) { _good = false; throw EndOfFile(std::string(*endOfFileError)); }
+    if (n == -1) { throw SysError("reading from file"); }
+    if (n == 0) { throw EndOfFile(std::string(*endOfFileError)); }
     read += n;
     return n;
 }
 
 
-bool FdSource::good()
-{
-    return _good;
-}
-
-
 size_t StringSource::read(char * data, size_t len)
 {
     if (pos == s.size()) throw EndOfFile("end of string reached");
@@ -332,55 +315,43 @@ void writePadding(size_t len, Sink & sink)
 }
 
 
-void writeString(std::string_view data, Sink & sink)
+WireFormatGenerator SerializingTransform::operator()(std::string_view s)
 {
-    sink << data.size();
-    sink(data);
-    writePadding(data.size(), sink);
+    co_yield s.size();
+    co_yield raw(s.begin(), s.size());
+    if (s.size() % 8) {
+        std::array<char, 8> pad{};
+        co_yield raw(pad.begin(), 8 - s.size() % 8);
+    }
 }
 
-
-Sink & operator << (Sink & sink, std::string_view s)
+WireFormatGenerator SerializingTransform::operator()(const Strings & ss)
 {
-    writeString(s, sink);
-    return sink;
+    co_yield ss.size();
+    for (const auto & s : ss)
+        co_yield std::string_view(s);
 }
 
-
-template<class T> void writeStrings(const T & ss, Sink & sink)
+WireFormatGenerator SerializingTransform::operator()(const StringSet & ss)
 {
-    sink << ss.size();
-    for (auto & i : ss)
-        sink << i;
+    co_yield ss.size();
+    for (const auto & s : ss)
+        co_yield std::string_view(s);
 }
 
-Sink & operator << (Sink & sink, const Strings & s)
-{
-    writeStrings(s, sink);
-    return sink;
-}
-
-Sink & operator << (Sink & sink, const StringSet & s)
-{
-    writeStrings(s, sink);
-    return sink;
-}
-
-Sink & operator << (Sink & sink, const Error & ex)
+WireFormatGenerator SerializingTransform::operator()(const Error & ex)
 {
     auto & info = ex.info();
-    sink
-        << "Error"
-        << info.level
-        << "Error" // removed
-        << info.msg.str()
-        << 0 // FIXME: info.errPos
-        << info.traces.size();
+    co_yield "Error";
+    co_yield info.level;
+    co_yield "Error"; // removed
+    co_yield info.msg.str();
+    co_yield 0; // FIXME: info.errPos
+    co_yield info.traces.size();
     for (auto & trace : info.traces) {
-        sink << 0; // FIXME: trace.pos
-        sink << trace.hint.str();
+        co_yield 0; // FIXME: trace.pos
+        co_yield trace.hint.str();
     }
-    return sink;
 }
 
 
@@ -466,18 +437,4 @@ void StringSink::operator () (std::string_view data)
     s.append(data);
 }
 
-size_t ChainSource::read(char * data, size_t len)
-{
-    if (useSecond) {
-        return source2.read(data, len);
-    } else {
-        try {
-            return source1.read(data, len);
-        } catch (EndOfFile &) {
-            useSecond = true;
-            return this->read(data, len);
-        }
-    }
-}
-
 }

src/libutil/serialise.hh

@@ -1,8 +1,10 @@
 #pragma once
 ///@file
 
+#include <concepts>
 #include <memory>
 
+#include "generator.hh"
 #include "types.hh"
 #include "util.hh"
 
@@ -18,7 +20,6 @@ struct Sink
 {
     virtual ~Sink() { }
     virtual void operator () (std::string_view data) = 0;
-    virtual bool good() { return true; }
 };
 
 /**
@@ -80,8 +81,6 @@ struct Source
      */
     virtual size_t read(char * data, size_t len) = 0;
 
-    virtual bool good() { return true; }
-
     void drainInto(Sink & sink);
 
     std::string drain();
@@ -136,11 +135,6 @@ struct FdSink : BufferedSink
     ~FdSink();
 
     void writeUnbuffered(std::string_view data) override;
-
-    bool good() override;
-
-private:
-    bool _good = true;
 };
 
 
@@ -165,11 +159,8 @@ struct FdSource : BufferedSource
         return *this;
     }
 
-    bool good() override;
 protected:
     size_t readUnbuffered(char * data, size_t len) override;
-private:
-    bool _good = true;
 };
 
 
@@ -317,18 +308,58 @@ struct LambdaSource : Source
 };
 
 /**
- * Chain two sources together so after the first is exhausted, the second is
- * used
+ * Chain a number of sources together, exhausting them all in turn.
  */
+template<typename... Sources>
+    requires (std::derived_from<Sources, Source> && ...)
 struct ChainSource : Source
 {
-    Source & source1, & source2;
-    bool useSecond = false;
-    ChainSource(Source & s1, Source & s2)
-        : source1(s1), source2(s2)
-    { }
+private:
+    std::tuple<Sources...> sources;
+    std::array<Source *, sizeof...(Sources)> ptrs;
+    size_t sourceIdx = 0;
 
-    size_t read(char * data, size_t len) override;
+    template<size_t... N>
+    void fillPtrs(std::index_sequence<N...>)
+    {
+        ((ptrs[N] = &std::get<N>(sources)), ...);
+    }
+
+public:
+    ChainSource(Sources && ... sources)
+        : sources(std::move(sources)...)
+    {
+        fillPtrs(std::index_sequence_for<Sources...>{});
+    }
+
+    ChainSource(ChainSource && other)
+        : sources(std::move(other.sources))
+        , sourceIdx(other.sourceIdx)
+    {
+        fillPtrs(std::index_sequence_for<Sources...>{});
+        other.sourceIdx = sizeof...(Sources);
+    }
+
+    ChainSource & operator=(ChainSource && other)
+    {
+        std::swap(sources, other.sources);
+        // since Sources... are the same the tuple type and offsets
+        // are the same, so pointers remain valid on both sides.
+        std::swap(sourceIdx, other.sourceIdx);
+        return *this;
+    }
+
+    size_t read(char * data, size_t len) override
+    {
+        if (sourceIdx == sizeof...(Sources))
+            throw EndOfFile("reached end of chained sources");
+        try {
+            return ptrs[sourceIdx]->read(data, len);
+        } catch (EndOfFile &) {
+            sourceIdx++;
+            return this->read(data, len);
+        }
+    }
 };
 
 std::unique_ptr<FinishSink> sourceToSink(std::function<void(Source &)> fun);
@@ -343,13 +374,54 @@ std::unique_ptr<Source> sinkToSource(
         throw EndOfFile("coroutine has finished");
     });
 
+struct SerializingTransform;
+using WireFormatGenerator = Generator<std::span<const char>, SerializingTransform>;
 
-void writePadding(size_t len, Sink & sink);
-void writeString(std::string_view s, Sink & sink);
-
-inline Sink & operator << (Sink & sink, uint64_t n)
+template<typename T>
+void drainGenerator(Generator<std::span<const char>, T> g, std::derived_from<Sink> auto & into)
 {
-    unsigned char buf[8];
+    while (g) {
+        auto bit = g();
+        into(std::string_view(bit.data(), bit.size()));
+    }
+}
+
+struct WireSource : Source
+{
+    template<typename F>
+    explicit WireSource(Generator<std::span<const char>, F> g) : g(std::move(g)) {}
+
+    virtual size_t read(char * data, size_t len)
+    {
+        while (!buf.size() && g) {
+            buf = g();
+        }
+        if (!buf.size()) {
+            throw EndOfFile("coroutine has finished");
+        }
+
+        len = std::min(len, buf.size());
+        memcpy(data, buf.data(), len);
+        buf = buf.subspan(len);
+        return len;
+    }
+
+private:
+    Generator<std::span<const char>, void> g;
+    std::span<const char> buf{};
+};
+
+struct SerializingTransform
+{
+    std::array<char, 8> buf;
+
+    static std::span<const char> raw(auto... args)
+    {
+        return std::span<const char>(args...);
+    }
+
+    std::span<const char> operator()(uint64_t n)
+    {
         buf[0] = n & 0xff;
         buf[1] = (n >> 8) & 0xff;
         buf[2] = (n >> 16) & 0xff;
@@ -358,19 +430,76 @@ inline Sink & operator << (Sink & sink, uint64_t n)
         buf[5] = (n >> 40) & 0xff;
         buf[6] = (n >> 48) & 0xff;
         buf[7] = (unsigned char) (n >> 56) & 0xff;
-    sink({(char *) buf, sizeof(buf)});
+        return {buf.begin(), 8};
+    }
+
+    static std::span<const char> padding(size_t unpadded)
+    {
+        return std::span("\0\0\0\0\0\0\0", unpadded % 8 ? 8 - unpadded % 8 : 0);
+    }
+
+    // opt in to generator chaining. without this co_yielding
+    // another generator of any type will cause a type error.
+    template<typename TF>
+    auto operator()(Generator<std::span<const char>, TF> && g)
+    {
+        return std::move(g);
+    }
+
+    // only choose this for *exactly* char spans, do not allow implicit
+    // conversions. this would cause ambiguities with strings literals,
+    // and resolving those with more string-like overloads needs a lot.
+    template<typename Span>
+        requires std::same_as<Span, std::span<char>> || std::same_as<Span, std::span<const char>>
+    std::span<const char> operator()(Span s)
+    {
+        return s;
+    }
+    WireFormatGenerator operator()(std::string_view s);
+    WireFormatGenerator operator()(const Strings & s);
+    WireFormatGenerator operator()(const StringSet & s);
+    WireFormatGenerator operator()(const Error & s);
+};
+
+template<typename Transform>
+inline Sink & operator<<(Sink & sink, Generator<std::span<const char>, Transform> && g)
+{
+    while (g) {
+        auto bit = g();
+        sink(std::string_view(bit.data(), bit.size()));
+    }
     return sink;
 }
 
-Sink & operator << (Sink & in, const Error & ex);
-Sink & operator << (Sink & sink, std::string_view s);
-Sink & operator << (Sink & sink, const Strings & s);
-Sink & operator << (Sink & sink, const StringSet & s);
+void writePadding(size_t len, Sink & sink);
 
+inline Sink & operator<<(Sink & sink, uint64_t u)
+{
+    return sink << [&]() -> WireFormatGenerator { co_yield u; }();
+}
+
+inline Sink & operator<<(Sink & sink, std::string_view s)
+{
+    return sink << [&]() -> WireFormatGenerator { co_yield s; }();
+}
+
+inline Sink & operator<<(Sink & sink, const Strings & s)
+{
+    return sink << [&]() -> WireFormatGenerator { co_yield s; }();
+}
+
+inline Sink & operator<<(Sink & sink, const StringSet & s)
+{
+    return sink << [&]() -> WireFormatGenerator { co_yield s; }();
+}
+
+inline Sink & operator<<(Sink & sink, const Error & ex)
+{
+    return sink << [&]() -> WireFormatGenerator { co_yield ex; }();
+}
 
 MakeError(SerialisationError, Error);
 
-
 template<typename T>
 T readNum(Source & source)
 {

src/libutil/source-path.hh

@@ -99,7 +99,7 @@ struct SourcePath
     void dumpPath(
         Sink & sink,
         PathFilter & filter = defaultPathFilter) const
-    { return nix::dumpPath(path.abs(), sink, filter); }
+    { sink << nix::dumpPath(path.abs(), filter); }
 
     /**
      * Return the location of this path in the "real" filesystem, if

src/libutil/util.cc

@@ -366,12 +366,12 @@ std::string readFile(const Path & path)
 }
 
 
-void readFile(const Path & path, Sink & sink)
+Generator<std::span<const char>> readFileSource(const Path & path)
 {
     AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
     if (!fd)
         throw SysError("opening file '%s'", path);
-    drainFD(fd.get(), sink);
+    co_yield drainFDSource(fd.get());
 }
 
 
@@ -722,12 +722,12 @@ std::string drainFD(int fd, bool block, const size_t reserveSize)
     // the parser needs two extra bytes to append terminating characters, other users will
     // not care very much about the extra memory.
     StringSink sink(reserveSize + 2);
-    drainFD(fd, sink, block);
+    sink << drainFDSource(fd, block);
     return std::move(sink.s);
 }
 
 
-void drainFD(int fd, Sink & sink, bool block)
+Generator<std::span<const char>> drainFDSource(int fd, bool block)
 {
     // silence GCC maybe-uninitialized warning in finally
     int saved = 0;
@@ -756,7 +756,7 @@ void drainFD(int fd, Sink & sink, bool block)
                 throw SysError("reading from file");
         }
         else if (rd == 0) break;
-        else sink({(char *) buf.data(), (size_t) rd});
+        else co_yield std::span{(char *) buf.data(), (size_t) rd};
     }
 }
 
@@ -1281,7 +1281,7 @@ void runProgram2(const RunOptions & options)
     }
 
     if (options.standardOut)
-        drainFD(out.readSide.get(), *options.standardOut);
+        *options.standardOut << drainFDSource(out.readSide.get());
 
     /* Wait for the child to finish. */
     int status = pid.wait();

src/libutil/util.hh

@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "generator.hh"
 #include "types.hh"
 #include "error.hh"
 #include "logging.hh"
@@ -162,7 +163,7 @@ unsigned char getFileType(const Path & path);
  */
 std::string readFile(int fd);
 std::string readFile(const Path & path);
-void readFile(const Path & path, Sink & sink);
+Generator<std::span<const char>> readFileSource(const Path & path);
 
 /**
  * Write a string to a file.
@@ -296,7 +297,7 @@ MakeError(EndOfFile, Error);
  */
 std::string drainFD(int fd, bool block = true, const size_t reserveSize=0);
 
-void drainFD(int fd, Sink & sink, bool block = true);
+Generator<std::span<const char>> drainFDSource(int fd, bool block = true);
 
 /**
  * If cgroups are active, attempt to calculate the number of CPUs available.
@@ -880,13 +881,6 @@ constexpr auto enumerate(T && iterable)
 }
 
 
-/**
- * C++17 std::visit boilerplate
- */
-template<class... Ts> struct overloaded : Ts... { using Ts::operator()...; };
-template<class... Ts> overloaded(Ts...) -> overloaded<Ts...>;
-
-
 std::string showBytes(uint64_t bytes);
 
 

src/nix-store/nix-store.cc

@@ -672,7 +672,7 @@ static void opDump(Strings opFlags, Strings opArgs)
 
     FdSink sink(STDOUT_FILENO);
     std::string path = *opArgs.begin();
-    dumpPath(path, sink);
+    sink << dumpPath(path);
     sink.flush();
 }
 

src/nix/add-to-store.cc

@@ -30,14 +30,14 @@ struct CmdAddToStore : MixDryRun, StoreCommand
         if (!namePart) namePart = baseNameOf(path);
 
         StringSink sink;
-        dumpPath(path, sink);
+        sink << dumpPath(path);
 
         auto narHash = hashString(htSHA256, sink.s);
 
         Hash hash = narHash;
         if (ingestionMethod == FileIngestionMethod::Flat) {
             HashSink hsink(htSHA256);
-            readFile(path, hsink);
+            hsink << readFileSource(path);
             hash = hsink.finish().first;
         }
 

src/nix/app.cc

@@ -8,6 +8,7 @@
 #include "command.hh"
 #include "derivations.hh"
 #include "downstream-placeholder.hh"
+#include "overloaded.hh"
 
 namespace nix {
 

src/nix/build.cc

@@ -4,6 +4,7 @@
 #include "store-api.hh"
 #include "local-fs-store.hh"
 #include "progress-bar.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 

src/nix/dump-path.cc

@@ -56,7 +56,7 @@ struct CmdDumpPath2 : Command
     void run() override
     {
         FdSink sink(STDOUT_FILENO);
-        dumpPath(path, sink);
+        sink << dumpPath(path);
         sink.flush();
     }
 };

src/nix/hash.cc

@@ -85,10 +85,10 @@ struct CmdHashBase : Command
 
             switch (mode) {
             case FileIngestionMethod::Flat:
-                readFile(path, *hashSink);
+                *hashSink << readFileSource(path);
                 break;
             case FileIngestionMethod::Recursive:
-                dumpPath(path, *hashSink);
+                *hashSink << dumpPath(path);
                 break;
             }
 

src/nix/log.cc

@@ -4,6 +4,7 @@
 #include "store-api.hh"
 #include "log-store.hh"
 #include "progress-bar.hh"
+#include "overloaded.hh"
 
 using namespace nix;
 

src/nix/profile.cc

@@ -10,6 +10,7 @@
 #include "../nix-env/user-env.hh"
 #include "profiles.hh"
 #include "names.hh"
+#include "overloaded.hh"
 
 #include <nlohmann/json.hpp>
 #include <regex>
@@ -214,7 +215,7 @@ struct ProfileManifest
 
         /* Add the symlink tree to the store. */
         StringSink sink;
-        dumpPath(tempDir, sink);
+        sink << dumpPath(tempDir);
 
         auto narHash = hashString(htSHA256, sink.s);
 

tests/unit/libutil-support/tests/cli-literate-parser.cc

@@ -2,6 +2,7 @@
 #include "libexpr/print.hh"
 #include "debug-char.hh"
 #include "types.hh"
+#include "overloaded.hh"
 #include "util.hh"
 #include <iostream>
 #include <memory>

tests/unit/libutil/generator.cc

@@ -0,0 +1,141 @@
+#include "generator.hh"
+
+#include <concepts>
+#include <cstdint>
+#include <gtest/gtest.h>
+
+namespace nix {
+
+TEST(Generator, yields)
+{
+    auto g = []() -> Generator<int> {
+        co_yield 1;
+        co_yield 2;
+    }();
+
+    ASSERT_TRUE(bool(g));
+    ASSERT_EQ(g(), 1);
+    ASSERT_EQ(g(), 2);
+    ASSERT_FALSE(bool(g));
+}
+
+TEST(Generator, nests)
+{
+    auto g = []() -> Generator<int> {
+        co_yield 1;
+        co_yield []() -> Generator<int> {
+            co_yield 9;
+            co_yield []() -> Generator<int> {
+                co_yield 99;
+                co_yield 100;
+            }();
+        }();
+
+        auto g2 = []() -> Generator<int> {
+            co_yield []() -> Generator<int> {
+                co_yield 2000;
+                co_yield 2001;
+            }();
+            co_yield 1001;
+        }();
+
+        co_yield g2();
+        co_yield std::move(g2);
+        co_yield 2;
+    }();
+
+    ASSERT_TRUE(bool(g));
+    ASSERT_EQ(g(), 1);
+    ASSERT_EQ(g(), 9);
+    ASSERT_EQ(g(), 99);
+    ASSERT_EQ(g(), 100);
+    ASSERT_EQ(g(), 2000);
+    ASSERT_EQ(g(), 2001);
+    ASSERT_EQ(g(), 1001);
+    ASSERT_EQ(g(), 2);
+    ASSERT_FALSE(bool(g));
+}
+
+TEST(Generator, nestsExceptions)
+{
+    auto g = []() -> Generator<int> {
+        co_yield 1;
+        co_yield []() -> Generator<int> {
+            co_yield 9;
+            throw 1;
+            co_yield 10;
+        }();
+        co_yield 2;
+    }();
+
+    ASSERT_TRUE(bool(g));
+    ASSERT_EQ(g(), 1);
+    ASSERT_EQ(g(), 9);
+    ASSERT_THROW(g(), int);
+}
+
+TEST(Generator, exception)
+{
+    {
+        auto g = []() -> Generator<int> {
+            throw 1;
+            co_return;
+        }();
+
+        ASSERT_THROW(void(bool(g)), int);
+    }
+    {
+        auto g = []() -> Generator<int> {
+            throw 1;
+            co_return;
+        }();
+
+        ASSERT_THROW(g(), int);
+    }
+}
+
+namespace {
+struct Transform
+{
+    int state = 0;
+
+    std::pair<uint32_t, int> operator()(std::integral auto x)
+    {
+        return {x, state++};
+    }
+
+    Generator<std::pair<uint32_t, int>, Transform> operator()(const char *)
+    {
+        co_yield 9;
+        co_yield 19;
+    }
+
+    Generator<std::pair<uint32_t, int>, Transform> operator()(Generator<int> && inner)
+    {
+        return [](auto g) mutable -> Generator<std::pair<uint32_t, int>, Transform> {
+            while (g) {
+                co_yield g();
+            }
+        }(std::move(inner));
+    }
+};
+}
+
+TEST(Generator, transform)
+{
+    auto g = []() -> Generator<std::pair<uint32_t, int>, Transform> {
+        co_yield int32_t(-1);
+        co_yield "";
+        std::cerr << "1\n";
+        co_yield []() -> Generator<int> { co_yield 7; }();
+        co_yield 20;
+    }();
+
+    ASSERT_EQ(g(), (std::pair<unsigned, int>{4294967295, 0}));
+    ASSERT_EQ(g(), (std::pair<unsigned, int>{9, 0}));
+    ASSERT_EQ(g(), (std::pair<unsigned, int>{19, 1}));
+    ASSERT_EQ(g(), (std::pair<unsigned, int>{7, 0}));
+    ASSERT_EQ(g(), (std::pair<unsigned, int>{20, 1}));
+}
+
+}

tests/unit/libutil/local.mk

@@ -18,6 +18,7 @@ libutil-tests_EXTRA_INCLUDES = \
 
 libutil-tests_CXXFLAGS += $(libutil-tests_EXTRA_INCLUDES)
 
-libutil-tests_LIBS = libutil-test-support libutil
+# libexpr is needed for exception serialization tests. sigh.
+libutil-tests_LIBS = libutil-test-support libutil libexpr
 
 libutil-tests_LDFLAGS := -lrapidcheck $(GTEST_LIBS)

tests/unit/libutil/pool.cc

@@ -65,21 +65,6 @@ namespace nix {
         ASSERT_EQ(pool.capacity(), 0);
     }
 
-    TEST(Pool, flushBadDropsOutOfScopeResources) {
-        auto isGood = [](const ref<TestResource> & r) { return false; };
-        auto createResource = []() { return make_ref<TestResource>(); };
-
-        Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource, isGood);
-
-        {
-            auto _r = pool.get();
-            ASSERT_EQ(pool.count(), 1);
-        }
-
-        pool.flushBad();
-        ASSERT_EQ(pool.count(), 0);
-    }
-
     // Test that the resources we allocate are being reused when they are still good.
     TEST(Pool, reuseResource) {
         auto isGood = [](const ref<TestResource> & r) { return true; };
@@ -124,4 +109,19 @@ namespace nix {
             ASSERT_NE(h->num, counter);
         }
     }
+
+    TEST(Pool, throwingOperationDropsResource)
+    {
+        auto createResource = []() { return make_ref<TestResource>(); };
+
+        Pool<TestResource> pool = Pool<TestResource>((size_t)1, createResource);
+
+        ASSERT_THROW({
+            auto _r = pool.get();
+            ASSERT_EQ(pool.count(), 1);
+            throw 1;
+        }, int);
+
+        ASSERT_EQ(pool.count(), 0);
+    }
 }

tests/unit/libutil/serialise.cc

@@ -0,0 +1,260 @@
+#include "serialise.hh"
+#include "error.hh"
+#include "fmt.hh"
+#include "generator.hh"
+#include "libexpr/pos-table.hh"
+#include "ref.hh"
+#include "types.hh"
+#include "util.hh"
+
+#include <concepts>
+#include <initializer_list>
+#include <limits.h>
+#include <gtest/gtest.h>
+
+#include <numeric>
+#include <stdexcept>
+
+namespace nix {
+
+TEST(ChainSource, single)
+{
+    ChainSource s{StringSource{"test"}};
+    ASSERT_EQ(s.drain(), "test");
+}
+
+TEST(ChainSource, multiple)
+{
+    ChainSource s{StringSource{"1"}, StringSource{""}, StringSource{"3"}};
+    ASSERT_EQ(s.drain(), "13");
+}
+
+TEST(ChainSource, chunk)
+{
+    std::string buf(2, ' ');
+    ChainSource s{StringSource{"111"}, StringSource{""}, StringSource{"333"}};
+
+    s(buf.data(), buf.size());
+    ASSERT_EQ(buf, "11");
+    s(buf.data(), buf.size());
+    ASSERT_EQ(buf, "13");
+    s(buf.data(), buf.size());
+    ASSERT_EQ(buf, "33");
+    ASSERT_THROW(s(buf.data(), buf.size()), EndOfFile);
+}
+
+TEST(ChainSource, move)
+{
+    std::string buf(2, ' ');
+    ChainSource s1{StringSource{"111"}, StringSource{""}, StringSource{"333"}};
+
+    s1(buf.data(), buf.size());
+    ASSERT_EQ(buf, "11");
+
+    ChainSource s2 = std::move(s1);
+    ASSERT_THROW(s1(buf.data(), buf.size()), EndOfFile);
+    s2(buf.data(), buf.size());
+    ASSERT_EQ(buf, "13");
+
+    s1 = std::move(s2);
+    ASSERT_THROW(s2(buf.data(), buf.size()), EndOfFile);
+    s1(buf.data(), buf.size());
+    ASSERT_EQ(buf, "33");
+}
+
+static std::string simpleToWire(const auto & val)
+{
+    std::string result;
+    auto g = [&] () -> WireFormatGenerator { co_yield val; }();
+    while (g) {
+        auto bit = g();
+        result.append(bit.data(), bit.size());
+    }
+    return result;
+}
+
+TEST(WireFormatGenerator, uint64_t)
+{
+    auto s = simpleToWire(42);
+    ASSERT_EQ(s, std::string({42, 0, 0, 0, 0, 0, 0, 0}));
+}
+
+TEST(WireFormatGenerator, string_view)
+{
+    auto s = simpleToWire("");
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            0, 0, 0, 0, 0, 0, 0, 0,
+            // data (omitted)
+        })
+    );
+    // clang-format on
+
+    s = simpleToWire("test");
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            4, 0, 0, 0, 0, 0, 0, 0,
+            // data
+            't', 'e', 's', 't',
+            // padding
+            0, 0, 0, 0,
+        })
+    );
+    // clang-format on
+
+    s = simpleToWire("longer string");
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            13, 0, 0, 0, 0, 0, 0, 0,
+            // data
+            'l', 'o', 'n', 'g', 'e', 'r', ' ', 's', 't', 'r', 'i', 'n', 'g',
+            // padding
+            0, 0, 0,
+        })
+    );
+    // clang-format on
+}
+
+TEST(WireFormatGenerator, StringSet)
+{
+    auto s = simpleToWire(StringSet{});
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            0, 0, 0, 0, 0, 0, 0, 0,
+            // data (omitted)
+        })
+    );
+    // clang-format on
+
+    s = simpleToWire(StringSet{"a", ""});
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            2, 0, 0, 0, 0, 0, 0, 0,
+            // data ""
+            0, 0, 0, 0, 0, 0, 0, 0,
+            // data "a"
+            1, 0, 0, 0, 0, 0, 0, 0, 'a', 0, 0, 0, 0, 0, 0, 0,
+        })
+    );
+    // clang-format on
+}
+
+TEST(WireFormatGenerator, Strings)
+{
+    auto s = simpleToWire(Strings{});
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            0, 0, 0, 0, 0, 0, 0, 0,
+            // data (omitted)
+        })
+    );
+    // clang-format on
+
+    s = simpleToWire(Strings{"a", ""});
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            // length
+            2, 0, 0, 0, 0, 0, 0, 0,
+            // data "a"
+            1, 0, 0, 0, 0, 0, 0, 0, 'a', 0, 0, 0, 0, 0, 0, 0,
+            // data ""
+            0, 0, 0, 0, 0, 0, 0, 0,
+        })
+    );
+    // clang-format on
+}
+
+TEST(WireFormatGenerator, Error)
+{
+    PosTable pt;
+    auto o = pt.addOrigin(Pos::String{make_ref<std::string>("test")}, 4);
+
+    auto s = simpleToWire(Error{{
+        .level = lvlInfo,
+        .msg = HintFmt("foo"),
+        .pos = pt[pt.add(o, 1)],
+        .traces = {{.pos = pt[pt.add(o, 2)], .hint = HintFmt("b %1%", "foo")}},
+    }});
+    // NOTE position of the error and all traces are ignored
+    // by the wire format
+    // clang-format off
+    ASSERT_EQ(
+        s,
+        std::string({
+            5, 0, 0, 0, 0, 0, 0, 0, 'E', 'r', 'r', 'o', 'r', 0, 0, 0,
+            3, 0, 0, 0, 0, 0, 0, 0,
+            5, 0, 0, 0, 0, 0, 0, 0, 'E', 'r', 'r', 'o', 'r', 0, 0, 0,
+            3, 0, 0, 0, 0, 0, 0, 0, 'f', 'o', 'o', 0, 0, 0, 0, 0,
+            0, 0, 0, 0, 0, 0, 0, 0,
+            1, 0, 0, 0, 0, 0, 0, 0,
+            0, 0, 0, 0, 0, 0, 0, 0,
+            16, 0, 0, 0, 0, 0, 0, 0,
+            'b', ' ', '\x1b', '[', '3', '5', ';', '1', 'm', 'f', 'o', 'o', '\x1b', '[', '0', 'm',
+        })
+    );
+    // clang-format on
+}
+
+TEST(FullFormatter, foo)
+{
+    auto gen = []() -> Generator<std::span<const char>, SerializingTransform> {
+        std::set<std::string> foo{"a", "longer string", ""};
+        co_yield 42;
+        co_yield foo;
+        co_yield std::string_view("test");
+        co_yield 7;
+    }();
+
+    std::vector<char> full;
+    while (gen) {
+        auto s = gen();
+        full.insert(full.end(), s.begin(), s.end());
+    }
+
+    ASSERT_EQ(
+        full,
+        (std::vector<char>{
+            // clang-format off
+            // 32
+            42, 0, 0, 0, 0, 0, 0, 0,
+            // foo
+            3, 0, 0, 0, 0, 0, 0, 0,
+            /// ""
+            0, 0, 0, 0, 0, 0, 0, 0,
+            /// a
+            1, 0, 0, 0, 0, 0, 0, 0,
+            'a', 0, 0, 0, 0, 0, 0, 0,
+            /// longer string
+            13, 0, 0, 0, 0, 0, 0, 0,
+            'l', 'o', 'n', 'g', 'e', 'r', ' ', 's', 't', 'r', 'i', 'n', 'g', 0, 0, 0,
+            // foo done
+            // test
+            4, 0, 0, 0, 0, 0, 0, 0,
+            't', 'e', 's', 't', 0, 0, 0, 0,
+            // 7
+            7, 0, 0, 0, 0, 0, 0, 0,
+            //clang-format on
+            }));
+}
+
+}