libstore/build: block io_uring

Unfortunately, io_uring is totally opaque to seccomp, and while currently there
are no dangerous operations implemented, there is no guarantee that it remains
this way. This means that io_uring should be blocked entirely to ensure that
the sandbox is future-proof. This has not been observed to cause issues in
practice.

Change-Id: I45d3895f95abe1bc103a63969f444c334dbbf50d

boehmgc-coroutine-sp-fallback.diff

@@ -1,54 +0,0 @@
-diff --git a/pthread_stop_world.c b/pthread_stop_world.c
-index 2b45489..0e6d8ef 100644
---- a/pthread_stop_world.c
-+++ b/pthread_stop_world.c
-@@ -776,6 +776,8 @@ STATIC void GC_restart_handler(int sig)
- /* world is stopped.  Should not fail if it isn't.                      */
- GC_INNER void GC_push_all_stacks(void)
- {
-+    size_t stack_limit;
-+    pthread_attr_t pattr;
-     GC_bool found_me = FALSE;
-     size_t nthreads = 0;
-     int i;
-@@ -868,6 +870,40 @@ GC_INNER void GC_push_all_stacks(void)
-             hi = p->altstack + p->altstack_size;
- #         endif
-           /* FIXME: Need to scan the normal stack too, but how ? */
-+        } else {
-+          #ifdef HAVE_PTHREAD_ATTR_GET_NP
-+          if (pthread_attr_init(&pattr) != 0) {
-+            ABORT("GC_push_all_stacks: pthread_attr_init failed!");
-+          }
-+          if (pthread_attr_get_np(p->id, &pattr) != 0) {
-+            ABORT("GC_push_all_stacks: pthread_attr_get_np failed!");
-+          }
-+          #else
-+          if (pthread_getattr_np(p->id, &pattr)) {
-+            ABORT("GC_push_all_stacks: pthread_getattr_np failed!");
-+          }
-+          #endif
-+          if (pthread_attr_getstacksize(&pattr, &stack_limit)) {
-+            ABORT("GC_push_all_stacks: pthread_attr_getstacksize failed!");
-+          }
-+          if (pthread_attr_destroy(&pattr)) {
-+            ABORT("GC_push_all_stacks: pthread_attr_destroy failed!");
-+          }
-+          // When a thread goes into a coroutine, we lose its original sp until
-+          // control flow returns to the thread.
-+          // While in the coroutine, the sp points outside the thread stack,
-+          // so we can detect this and push the entire thread stack instead,
-+          // as an approximation.
-+          // We assume that the coroutine has similarly added its entire stack.
-+          // This could be made accurate by cooperating with the application
-+          // via new functions and/or callbacks.
-+          #ifndef STACK_GROWS_UP
-+            if (lo >= hi || lo < hi - stack_limit) { // sp outside stack
-+              lo = hi - stack_limit;
-+            }
-+          #else
-+          #error "STACK_GROWS_UP not supported in boost_coroutine2 (as of june 2021), so we don't support it in Nix."
-+          #endif
-         }
- #       ifdef STACKPTR_CORRECTOR_AVAILABLE
-           if (GC_sp_corrector != 0)

doc/manual/rl-next/block-io-uring.md

@@ -0,0 +1,12 @@
+---
+synopsis: "Block io_uring in the Linux sandbox"
+cls: 1611
+credits: alois31
+category: Breaking Changes
+---
+
+The io\_uring API has the unfortunate property that it is not possible to selectively decide which operations should be allowed.
+This, together with the fact that new operations are routinely added, makes it a hazard to the proper function of the sandbox.
+
+Therefore, any access to io\_uring has been made unavailable inside the sandbox.
+As such, attempts to execute any system calls forming part of this API will fail with the error `ENOSYS`, as if io\_uring support had not been configured into the kernel.

doc/manual/rl-next/pretty-printing.md

@@ -0,0 +1,58 @@
+---
+synopsis: "Eliminate some pretty-printing surprises"
+cls: [1616, 1617, 1618]
+prs: [11100]
+credits: [alois31, roberth]
+category: Improvements
+---
+
+Some inconsistent and surprising behaviours have been eliminated from the pretty-printing used by the REPL and `nix eval`:
+* Lists and attribute sets that contain only a single item without nested structures are no longer sometimes inappropriately indented in the REPL, depending on internal state of the evaluator.
+* Empty attribute sets and derivations are no longer shown as `«repeated»`, since they are always cheap to print.
+  This matches the existing behaviour of `nix-instantiate` on empty attribute sets.
+  Empty lists were never printed as `«repeated»` already.
+* The REPL by default does not print nested attribute sets and lists, and indicates elided items with an ellipsis.
+  Previously, the ellipsis was printed even when the structure was empty, so that such items do not in fact exist.
+  Since this behaviour was confusing, it does not happen any more.
+
+Before:
+```
+nix-repl> :p let x = 1 + 2; in [ [ x ] [ x ] ]
+[
+  [
+    3
+  ]
+  [ 3 ]
+]
+
+nix-repl> let inherit (import <nixpkgs> { }) hello; in [ hello hello ]
+[
+  «derivation /nix/store/fqs92lzychkm6p37j7fnj4d65nq9fzla-hello-2.12.1.drv»
+  «repeated»
+]
+
+nix-repl> let x = {}; in [ x ]
+[
+  { ... }
+]
+```
+
+After:
+```
+nix-repl> :p let x = 1 + 2; in [ [ x ] [ x ] ]
+[
+  [ 3 ]
+  [ 3 ]
+]
+
+nix-repl> let inherit (import <nixpkgs> { }) hello; in [ hello hello ]
+[
+  «derivation /nix/store/fqs92lzychkm6p37j7fnj4d65nq9fzla-hello-2.12.1.drv»
+  «derivation /nix/store/fqs92lzychkm6p37j7fnj4d65nq9fzla-hello-2.12.1.drv»
+]
+
+nix-repl> let x = {}; in [ x ]
+[
+  { }
+]
+```

flake.nix

@@ -441,7 +441,7 @@
             makeShell pkgs pkgs.stdenv
           ))
           // {
-            default = self.devShells.${system}.native-stdenvPackages;
+            default = self.devShells.${system}.native-clangStdenvPackages;
           }
         );
     };

meson.build

@@ -204,7 +204,7 @@ configdata += {
   'HAVE_BOEHMGC': boehm.found().to_int(),
 }
 
-boost = dependency('boost', required : true, modules : ['context', 'coroutine', 'container'])
+boost = dependency('boost', required : true, modules : ['container'])
 
 # cpuid only makes sense on x86_64
 cpuid_required = is_x64 ? get_option('cpuid') : false

package.nix

@@ -62,15 +62,7 @@
   __forDefaults ? {
     canRunInstalled = stdenv.buildPlatform.canExecute stdenv.hostPlatform;
 
-    boehmgc-nix = (boehmgc.override { enableLargeConfig = true; }).overrideAttrs {
-      patches = [
-        # We do *not* include prev.patches (which doesn't exist in normal pkgs.boehmgc anyway)
-        # because if the caller of this package passed a patched boehm as `boehmgc` instead of
-        # `boehmgc-nix` then this will almost certainly have duplicate patches, which means
-        # the patches won't apply and we'll get a build failure.
-        ./boehmgc-coroutine-sp-fallback.diff
-      ];
-    };
+    boehmgc-nix = boehmgc.override { enableLargeConfig = true; };
 
     editline-lix = editline.overrideAttrs (prev: {
       configureFlags = prev.configureFlags or [ ] ++ [ (lib.enableFeature true "sigstop") ];
@@ -167,7 +159,6 @@ stdenv.mkDerivation (finalAttrs: {
           functionalTestFiles
         ]
         ++ lib.optionals (!finalAttrs.dontBuild || internalApiDocs) [
-          ./boehmgc-coroutine-sp-fallback.diff
           ./doc
           ./misc
           ./src
@@ -463,10 +454,10 @@ stdenv.mkDerivation (finalAttrs: {
               # `bash` from inside `nix develop`, say, because you are using it
               # via direnv, you will by default get bash (unusable edition).
               bashInteractive
+              check-syscalls
               pythonEnv
               # docker image tool
               skopeo
-              check-syscalls
               just
               nixfmt
               # Included above when internalApiDocs is true, but we set that to

src/libexpr/eval.cc

@@ -42,10 +42,6 @@
 #include <gc/gc.h>
 #include <gc/gc_cpp.h>
 
-#include <boost/coroutine2/coroutine.hpp>
-#include <boost/coroutine2/protected_fixedsize_stack.hpp>
-#include <boost/context/stack_context.hpp>
-
 #endif
 
 using json = nlohmann::json;
@@ -192,42 +188,6 @@ static void * oomHandler(size_t requested)
     throw std::bad_alloc();
 }
 
-class BoehmGCStackAllocator : public StackAllocator {
-    boost::coroutines2::protected_fixedsize_stack stack {
-        // We allocate 8 MB, the default max stack size on NixOS.
-        // A smaller stack might be quicker to allocate but reduces the stack
-        // depth available for source filter expressions etc.
-        std::max(boost::context::stack_traits::default_size(), static_cast<std::size_t>(8 * 1024 * 1024))
-    };
-
-    // This is specific to boost::coroutines2::protected_fixedsize_stack.
-    // The stack protection page is included in sctx.size, so we have to
-    // subtract one page size from the stack size.
-    std::size_t pfss_usable_stack_size(boost::context::stack_context &sctx) {
-        return sctx.size - boost::context::stack_traits::page_size();
-    }
-
-  public:
-    boost::context::stack_context allocate() override {
-        auto sctx = stack.allocate();
-
-        // Stacks generally start at a high address and grow to lower addresses.
-        // Architectures that do the opposite are rare; in fact so rare that
-        // boost_routine does not implement it.
-        // So we subtract the stack size.
-        GC_add_roots(static_cast<char *>(sctx.sp) - pfss_usable_stack_size(sctx), sctx.sp);
-        return sctx;
-    }
-
-    void deallocate(boost::context::stack_context sctx) override {
-        GC_remove_roots(static_cast<char *>(sctx.sp) - pfss_usable_stack_size(sctx), sctx.sp);
-        stack.deallocate(sctx);
-    }
-
-};
-
-static BoehmGCStackAllocator boehmGCStackAllocator;
-
 #endif
 
 
@@ -243,23 +203,6 @@ static Symbol getName(const AttrName & name, EvalState & state, Env & env)
     }
 }
 
-#if HAVE_BOEHMGC
-/* Disable GC while this object lives. Used by CoroutineContext.
- *
- * Boehm keeps a count of GC_disable() and GC_enable() calls,
- * and only enables GC when the count matches.
- */
-class BoehmDisableGC {
-public:
-    BoehmDisableGC() {
-        GC_disable();
-    };
-    ~BoehmDisableGC() {
-        GC_enable();
-    };
-};
-#endif
-
 static bool gcInitialised = false;
 
 void initGC()
@@ -281,17 +224,6 @@ void initGC()
 
     GC_set_oom_fn(oomHandler);
 
-    StackAllocator::defaultAllocator = &boehmGCStackAllocator;
-
-
-#if NIX_BOEHM_PATCH_VERSION != 1
-    printTalkative("Unpatched BoehmGC, disabling GC inside coroutines");
-    /* Used to disable GC when entering coroutines on macOS */
-    create_coro_gc_hook = []() -> std::shared_ptr<void> {
-        return std::make_shared<BoehmDisableGC>();
-    };
-#endif
-
     /* Set the initial heap size to something fairly big (25% of
        physical RAM, up to a maximum of 384 MiB) so that in most cases
        we don't need to garbage collect at all.  (Collection has a

src/libstore/binary-cache-store.cc

@@ -329,25 +329,32 @@ std::optional<StorePath> BinaryCacheStore::queryPathFromHashPart(const std::stri
     }
 }
 
-void BinaryCacheStore::narFromPath(const StorePath & storePath, Sink & sink)
+WireFormatGenerator BinaryCacheStore::narFromPath(const StorePath & storePath)
 {
     auto info = queryPathInfo(storePath).cast<const NarInfo>();
 
-    LengthSink narSize;
-    TeeSink tee { sink, narSize };
-
-
     try {
         auto file = getFile(info->url);
-        auto decompressor = makeDecompressionSource(info->compression, *file);
-        decompressor->drainInto(tee);
+        return [](auto info, auto file, auto & stats) -> WireFormatGenerator {
+            std::unique_ptr<char[]> buf(new char[65536]);
+            size_t total = 0;
+            auto decompressor = makeDecompressionSource(info->compression, *file);
+            try {
+                while (true) {
+                    const auto len = decompressor->read(buf.get(), sizeof(buf));
+                    co_yield std::span{buf.get(), len};
+                    total += len;
+                }
+            } catch (EndOfFile &) {
+            }
+
+            stats.narRead++;
+            //stats.narReadCompressedBytes += nar->size(); // FIXME
+            stats.narReadBytes += total;
+        }(std::move(info), std::move(file), stats);
     } catch (NoSuchBinaryCacheFile & e) {
         throw SubstituteGone(std::move(e.info()));
     }
-
-    stats.narRead++;
-    //stats.narReadCompressedBytes += nar->size(); // FIXME
-    stats.narReadBytes += narSize.length;
 }
 
 std::shared_ptr<const ValidPathInfo> BinaryCacheStore::queryPathInfoUncached(const StorePath & storePath)

src/libstore/binary-cache-store.hh

@@ -136,7 +136,7 @@ public:
 
     std::shared_ptr<const Realisation> queryRealisationUncached(const DrvOutput &) override;
 
-    void narFromPath(const StorePath & path, Sink & sink) override;
+    WireFormatGenerator narFromPath(const StorePath & path) override;
 
     ref<FSAccessor> getFSAccessor() override;
 

src/libstore/build/derivation-goal.cc

@@ -1204,11 +1204,9 @@ HookReply DerivationGoal::tryBuildHook()
         throw;
     }
 
-    CommonProto::WriteConn conn { hook->sink };
-
     /* Tell the hook all the inputs that have to be copied to the
        remote system. */
-    CommonProto::write(worker.store, conn, inputPaths);
+    hook->sink << CommonProto::write(worker.store, {}, inputPaths);
 
     /* Tell the hooks the missing outputs that have to be copied back
        from the remote system. */
@@ -1219,7 +1217,7 @@ HookReply DerivationGoal::tryBuildHook()
             if (buildMode != bmCheck && status.known && status.known->isValid()) continue;
             missingOutputs.insert(outputName);
         }
-        CommonProto::write(worker.store, conn, missingOutputs);
+        hook->sink << CommonProto::write(worker.store, {}, missingOutputs);
     }
 
     hook->sink = FdSink();

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

@@ -1084,11 +1084,11 @@ struct RestrictedStore : public virtual RestrictedStoreConfig, public virtual In
         return path;
     }
 
-    void narFromPath(const StorePath & path, Sink & sink) override
+    WireFormatGenerator narFromPath(const StorePath & path) override
     {
         if (!goal.isAllowed(path))
             throw InvalidPath("cannot dump unknown path '%s' in recursive Nix", printStorePath(path));
-        LocalFSStore::narFromPath(path, sink);
+        return LocalFSStore::narFromPath(path);
     }
 
     void ensurePath(const StorePath & path) override
@@ -1363,10 +1363,18 @@ void LocalDerivationGoal::chownToBuilder(const Path & path)
 }
 
 #if HAVE_SECCOMP
+
 void allowSyscall(scmp_filter_ctx ctx, int syscall) {
     if (seccomp_rule_add(ctx, SCMP_ACT_ALLOW, syscall, 0) != 0)
         throw SysError("unable to add seccomp rule");
 }
+
+#define ALLOW_CHMOD_IF_SAFE(ctx, syscall, modePos) \
+    if (seccomp_rule_add(ctx, SCMP_ACT_ALLOW, syscall, 1, SCMP_A##modePos(SCMP_CMP_MASKED_EQ, S_ISUID | S_ISGID, 0)) != 0 || \
+        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), syscall, 1, SCMP_A##modePos(SCMP_CMP_MASKED_EQ, S_ISUID, S_ISUID)) != 0 || \
+        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), syscall, 1, SCMP_A##modePos(SCMP_CMP_MASKED_EQ, S_ISGID, S_ISGID)) != 0) \
+        throw SysError("unable to add seccomp rule");
+
 #endif
 
 void setupSeccomp()
@@ -1415,7 +1423,7 @@ void setupSeccomp()
     // This list is intended for machine consumption.
     // Please keep its format, order and BEGIN/END markers.
     //
-    // Currently, it is up to date with libseccomp 2.5.5 and glibc 2.38.
+    // Currently, it is up to date with libseccomp 2.5.5 and glibc 2.39.
     // Run check-syscalls to determine which new syscalls should be added.
     // New syscalls must be audited and handled in a way that blocks the following dangerous operations:
     // * Creation of non-empty setuid/setgid files
@@ -1495,7 +1503,7 @@ void setupSeccomp()
     allowSyscall(ctx, SCMP_SYS(fchdir));
     // skip fchmod (dangerous)
     // skip fchmodat (dangerous)
-    // skip fchmodat2 (requires glibc 2.39, dangerous)
+    // skip fchmodat2 (dangerous)
     allowSyscall(ctx, SCMP_SYS(fchown));
     allowSyscall(ctx, SCMP_SYS(fchown32));
     allowSyscall(ctx, SCMP_SYS(fchownat));
@@ -1523,11 +1531,11 @@ void setupSeccomp()
     allowSyscall(ctx, SCMP_SYS(ftruncate));
     allowSyscall(ctx, SCMP_SYS(ftruncate64));
     allowSyscall(ctx, SCMP_SYS(futex));
-    // skip futex_requeue (requires glibc 2.39)
+    allowSyscall(ctx, SCMP_SYS(futex_requeue));
     allowSyscall(ctx, SCMP_SYS(futex_time64));
-    // skip futex_wait (requires glibc 2.39)
+    allowSyscall(ctx, SCMP_SYS(futex_wait));
     allowSyscall(ctx, SCMP_SYS(futex_waitv));
-    // skip futex_wake (requires glibc 2.39)
+    allowSyscall(ctx, SCMP_SYS(futex_wake));
     allowSyscall(ctx, SCMP_SYS(futimesat));
     allowSyscall(ctx, SCMP_SYS(getcpu));
     allowSyscall(ctx, SCMP_SYS(getcwd));
@@ -1617,7 +1625,7 @@ void setupSeccomp()
     allowSyscall(ctx, SCMP_SYS(lstat));
     allowSyscall(ctx, SCMP_SYS(lstat64));
     allowSyscall(ctx, SCMP_SYS(madvise));
-    // skip map_shadow_stack (requires glibc 2.39)
+    allowSyscall(ctx, SCMP_SYS(map_shadow_stack));
     allowSyscall(ctx, SCMP_SYS(mbind));
     allowSyscall(ctx, SCMP_SYS(membarrier));
     allowSyscall(ctx, SCMP_SYS(memfd_create));
@@ -1913,16 +1921,10 @@ void setupSeccomp()
 
     // chmod family: prevent adding setuid/setgid bits to existing files.
     // The Nix store does not support setuid/setgid, and even their temporary creation can weaken the security of the sandbox.
-    if (seccomp_rule_add(ctx, SCMP_ACT_ALLOW, SCMP_SYS(chmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISUID | S_ISGID, 0)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(chmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISUID, S_ISUID)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(chmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISGID, S_ISGID)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ALLOW, SCMP_SYS(fchmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISUID | S_ISGID, 0)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(fchmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISUID, S_ISUID)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(fchmod), 1, SCMP_A1(SCMP_CMP_MASKED_EQ, S_ISGID, S_ISGID)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ALLOW, SCMP_SYS(fchmodat), 1, SCMP_A2(SCMP_CMP_MASKED_EQ, S_ISUID | S_ISGID, 0)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(fchmodat), 1, SCMP_A2(SCMP_CMP_MASKED_EQ, S_ISUID, S_ISUID)) != 0 ||
-        seccomp_rule_add(ctx, SCMP_ACT_ERRNO(EPERM), SCMP_SYS(fchmodat), 1, SCMP_A2(SCMP_CMP_MASKED_EQ, S_ISGID, S_ISGID)) != 0)
-        throw SysError("unable to add seccomp rule");
+    ALLOW_CHMOD_IF_SAFE(ctx, SCMP_SYS(chmod), 1);
+    ALLOW_CHMOD_IF_SAFE(ctx, SCMP_SYS(fchmod), 1);
+    ALLOW_CHMOD_IF_SAFE(ctx, SCMP_SYS(fchmodat), 2);
+    ALLOW_CHMOD_IF_SAFE(ctx, SCMP_SYS(fchmodat2), 2);
 
     // setxattr family: prevent creation of extended attributes or ACLs.
     // Not all filesystems support them, and they're incompatible with the NAR format.

src/libstore/common-protocol-impl.hh

@@ -20,9 +20,9 @@ namespace nix {
     { \
         return LengthPrefixedProtoHelper<CommonProto, T >::read(store, conn); \
     } \
-    TEMPLATE void CommonProto::Serialise< T >::write(const Store & store, CommonProto::WriteConn conn, const T & t) \
+    TEMPLATE [[nodiscard]] WireFormatGenerator CommonProto::Serialise< T >::write(const Store & store, CommonProto::WriteConn conn, const T & t) \
     { \
-        LengthPrefixedProtoHelper<CommonProto, T >::write(store, conn, t); \
+        return LengthPrefixedProtoHelper<CommonProto, T >::write(store, conn, t); \
     }
 
 COMMON_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)

src/libstore/common-protocol.cc

@@ -16,9 +16,9 @@ std::string CommonProto::Serialise<std::string>::read(const Store & store, Commo
     return readString(conn.from);
 }
 
-void CommonProto::Serialise<std::string>::write(const Store & store, CommonProto::WriteConn conn, const std::string & str)
+WireFormatGenerator CommonProto::Serialise<std::string>::write(const Store & store, CommonProto::WriteConn conn, const std::string & str)
 {
-    conn.to << str;
+    co_yield str;
 }
 
 
@@ -27,9 +27,9 @@ StorePath CommonProto::Serialise<StorePath>::read(const Store & store, CommonPro
     return store.parseStorePath(readString(conn.from));
 }
 
-void CommonProto::Serialise<StorePath>::write(const Store & store, CommonProto::WriteConn conn, const StorePath & storePath)
+WireFormatGenerator CommonProto::Serialise<StorePath>::write(const Store & store, CommonProto::WriteConn conn, const StorePath & storePath)
 {
-    conn.to << store.printStorePath(storePath);
+    co_yield store.printStorePath(storePath);
 }
 
 
@@ -38,9 +38,9 @@ ContentAddress CommonProto::Serialise<ContentAddress>::read(const Store & store,
     return ContentAddress::parse(readString(conn.from));
 }
 
-void CommonProto::Serialise<ContentAddress>::write(const Store & store, CommonProto::WriteConn conn, const ContentAddress & ca)
+WireFormatGenerator CommonProto::Serialise<ContentAddress>::write(const Store & store, CommonProto::WriteConn conn, const ContentAddress & ca)
 {
-    conn.to << renderContentAddress(ca);
+    co_yield renderContentAddress(ca);
 }
 
 
@@ -53,9 +53,9 @@ Realisation CommonProto::Serialise<Realisation>::read(const Store & store, Commo
     );
 }
 
-void CommonProto::Serialise<Realisation>::write(const Store & store, CommonProto::WriteConn conn, const Realisation & realisation)
+WireFormatGenerator CommonProto::Serialise<Realisation>::write(const Store & store, CommonProto::WriteConn conn, const Realisation & realisation)
 {
-    conn.to << realisation.toJSON().dump();
+    co_yield realisation.toJSON().dump();
 }
 
 
@@ -64,9 +64,9 @@ DrvOutput CommonProto::Serialise<DrvOutput>::read(const Store & store, CommonPro
     return DrvOutput::parse(readString(conn.from));
 }
 
-void CommonProto::Serialise<DrvOutput>::write(const Store & store, CommonProto::WriteConn conn, const DrvOutput & drvOutput)
+WireFormatGenerator CommonProto::Serialise<DrvOutput>::write(const Store & store, CommonProto::WriteConn conn, const DrvOutput & drvOutput)
 {
-    conn.to << drvOutput.to_string();
+    co_yield drvOutput.to_string();
 }
 
 
@@ -76,9 +76,11 @@ std::optional<StorePath> CommonProto::Serialise<std::optional<StorePath>>::read(
     return s == "" ? std::optional<StorePath> {} : store.parseStorePath(s);
 }
 
-void CommonProto::Serialise<std::optional<StorePath>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
+WireFormatGenerator CommonProto::Serialise<std::optional<StorePath>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
 {
-    conn.to << (storePathOpt ? store.printStorePath(*storePathOpt) : "");
+    return [](std::string s) -> WireFormatGenerator {
+        co_yield s;
+    }(storePathOpt ? store.printStorePath(*storePathOpt) : "");
 }
 
 
@@ -87,9 +89,11 @@ std::optional<ContentAddress> CommonProto::Serialise<std::optional<ContentAddres
     return ContentAddress::parseOpt(readString(conn.from));
 }
 
-void CommonProto::Serialise<std::optional<ContentAddress>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
+WireFormatGenerator CommonProto::Serialise<std::optional<ContentAddress>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
 {
-    conn.to << (caOpt ? renderContentAddress(*caOpt) : "");
+    return [](std::string s) -> WireFormatGenerator {
+        co_yield s;
+    }(caOpt ? renderContentAddress(*caOpt) : "");
 }
 
 }

src/libstore/common-protocol.hh

@@ -37,7 +37,6 @@ struct CommonProto
      * canonical serializers below.
      */
     struct WriteConn {
-        Sink & to;
     };
 
     template<typename T>
@@ -48,9 +47,10 @@ struct CommonProto
      * infer the type instead of having to write it down explicitly.
      */
     template<typename T>
-    static void write(const Store & store, WriteConn conn, const T & t)
+    [[nodiscard]]
+    static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
     {
-        CommonProto::Serialise<T>::write(store, conn, t);
+        return CommonProto::Serialise<T>::write(store, conn, t);
     }
 };
 
@@ -58,7 +58,7 @@ struct CommonProto
     struct CommonProto::Serialise< T > \
     { \
         static T read(const Store & store, CommonProto::ReadConn conn); \
-        static void write(const Store & store, CommonProto::WriteConn conn, const T & str); \
+        [[nodiscard]] static WireFormatGenerator write(const Store & store, CommonProto::WriteConn conn, const T & str); \
     }
 
 template<>

src/libstore/daemon.cc

@@ -264,7 +264,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
     Source & from, BufferedSink & to, WorkerProto::Op op)
 {
     WorkerProto::ReadConn rconn{from, clientVersion};
-    WorkerProto::WriteConn wconn{to, clientVersion};
+    WorkerProto::WriteConn wconn{clientVersion};
 
     switch (op) {
 
@@ -291,7 +291,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         }
         auto res = store->queryValidPaths(paths, substitute);
         logger->stopWork();
-        WorkerProto::write(*store, wconn, res);
+        to << WorkerProto::write(*store, wconn, res);
         break;
     }
 
@@ -300,7 +300,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         logger->startWork();
         auto res = store->querySubstitutablePaths(paths);
         logger->stopWork();
-        WorkerProto::write(*store, wconn, res);
+        to << WorkerProto::write(*store, wconn, res);
         break;
     }
 
@@ -365,7 +365,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         #pragma GCC diagnostic pop
 
         logger->stopWork();
-        WorkerProto::write(*store, wconn, paths);
+        to << WorkerProto::write(*store, wconn, paths);
         break;
     }
 
@@ -385,7 +385,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         logger->startWork();
         auto outputs = store->queryPartialDerivationOutputMap(path);
         logger->stopWork();
-        WorkerProto::write(*store, wconn, outputs);
+        to << WorkerProto::write(*store, wconn, outputs);
         break;
     }
 
@@ -432,7 +432,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
             }();
             logger->stopWork();
 
-            WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
+            to << WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
         } else {
             HashType hashAlgo;
             std::string baseName;
@@ -453,7 +453,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
                 hashAlgo = parseHashType(hashAlgoRaw);
             }
 
-            auto dumpSource = sinkToSource([&](Sink & saved) {
+            GeneratorSource dumpSource{[&]() -> WireFormatGenerator {
                 if (method == FileIngestionMethod::Recursive) {
                     /* We parse the NAR dump through into `saved` unmodified,
                        so why all this extra work? We still parse the NAR so
@@ -463,18 +463,35 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
                        command. (We don't trust `addToStoreFromDump` to not
                        eagerly consume the entire stream it's given, past the
                        length of the Nar. */
-                    saved << copyNAR(from);
+                    co_yield copyNAR(from);
                 } else {
                     /* Incrementally parse the NAR file, stripping the
                        metadata, and streaming the sole file we expect into
                        `saved`. */
-                    RetrieveRegularNARSink savedRegular { saved };
-                    parseDump(savedRegular, from);
-                    if (!savedRegular.regular) throw Error("regular file expected");
+                    auto parser = nar::parse(from);
+                    nar::File * file = nullptr;
+                    while (auto entry = parser.next()) {
+                        file = std::visit(
+                            overloaded{
+                                [](nar::MetadataString) -> nar::File * { return nullptr; },
+                                [](nar::MetadataRaw) -> nar::File * { return nullptr; },
+                                [](nar::File & f) -> nar::File * { return &f; },
+                                [](auto &) -> nar::File * { throw Error("regular file expected"); },
+                            },
+                            *entry
+                        );
+                        if (file) {
+                            break;
+                        }
+                    }
+                    if (!file) {
+                        throw Error("regular file expected");
+                    }
+                    co_yield std::move(file->contents);
                 }
-            });
+            }()};
             logger->startWork();
-            auto path = store->addToStoreFromDump(*dumpSource, baseName, method, hashAlgo);
+            auto path = store->addToStoreFromDump(dumpSource, baseName, method, hashAlgo);
             logger->stopWork();
 
             to << store->printStorePath(path);
@@ -548,7 +565,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         auto results = store->buildPathsWithResults(drvs, mode);
         logger->stopWork();
 
-        WorkerProto::write(*store, wconn, results);
+        to << WorkerProto::write(*store, wconn, results);
 
         break;
     }
@@ -626,7 +643,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
 
         auto res = store->buildDerivation(drvPath, drv, buildMode);
         logger->stopWork();
-        WorkerProto::write(*store, wconn, res);
+        to << WorkerProto::write(*store, wconn, res);
         break;
     }
 
@@ -760,7 +777,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         else {
             to << 1
                << (i->second.deriver ? store->printStorePath(*i->second.deriver) : "");
-            WorkerProto::write(*store, wconn, i->second.references);
+            to << WorkerProto::write(*store, wconn, i->second.references);
             to << i->second.downloadSize
                << i->second.narSize;
         }
@@ -783,7 +800,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         for (auto & i : infos) {
             to << store->printStorePath(i.first)
                << (i.second.deriver ? store->printStorePath(*i.second.deriver) : "");
-            WorkerProto::write(*store, wconn, i.second.references);
+            to << WorkerProto::write(*store, wconn, i.second.references);
             to << i.second.downloadSize << i.second.narSize;
         }
         break;
@@ -793,7 +810,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         logger->startWork();
         auto paths = store->queryAllValidPaths();
         logger->stopWork();
-        WorkerProto::write(*store, wconn, paths);
+        to << WorkerProto::write(*store, wconn, paths);
         break;
     }
 
@@ -810,7 +827,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         logger->stopWork();
         if (info) {
             to << 1;
-            WorkerProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
+            to << WorkerProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
         } else {
             to << 0;
         }
@@ -905,9 +922,9 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         uint64_t downloadSize, narSize;
         store->queryMissing(targets, willBuild, willSubstitute, unknown, downloadSize, narSize);
         logger->stopWork();
-        WorkerProto::write(*store, wconn, willBuild);
-        WorkerProto::write(*store, wconn, willSubstitute);
-        WorkerProto::write(*store, wconn, unknown);
+        to << WorkerProto::write(*store, wconn, willBuild);
+        to << WorkerProto::write(*store, wconn, willSubstitute);
+        to << WorkerProto::write(*store, wconn, unknown);
         to << downloadSize << narSize;
         break;
     }
@@ -935,11 +952,11 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
         if (GET_PROTOCOL_MINOR(clientVersion) < 31) {
             std::set<StorePath> outPaths;
             if (info) outPaths.insert(info->outPath);
-            WorkerProto::write(*store, wconn, outPaths);
+            to << WorkerProto::write(*store, wconn, outPaths);
         } else {
             std::set<Realisation> realisations;
             if (info) realisations.insert(*info);
-            WorkerProto::write(*store, wconn, realisations);
+            to << WorkerProto::write(*store, wconn, realisations);
         }
         break;
     }
@@ -1019,8 +1036,8 @@ void processConnection(
         auto temp = trusted
             ? store->isTrustedClient()
             : std::optional { NotTrusted };
-        WorkerProto::WriteConn wconn {to, clientVersion};
-        WorkerProto::write(*store, wconn, temp);
+        WorkerProto::WriteConn wconn {clientVersion};
+        to << WorkerProto::write(*store, wconn, temp);
     }
 
     /* Send startup error messages to the client. */

src/libstore/derivations.cc

@@ -994,8 +994,8 @@ void writeDerivation(Sink & out, const Store & store, const BasicDerivation & dr
             },
         }, i.second.raw);
     }
-    CommonProto::write(store,
-        CommonProto::WriteConn { .to = out },
+    out << CommonProto::write(store,
+        CommonProto::WriteConn {},
         drv.inputSrcs);
     out << drv.platform << drv.builder << drv.args;
     out << drv.env.size();

src/libstore/dummy-store.cc

@@ -63,7 +63,7 @@ struct DummyStore : public virtual DummyStoreConfig, public virtual Store
         RepairFlag repair) override
     { unsupported("addTextToStore"); }
 
-    void narFromPath(const StorePath & path, Sink & sink) override
+    WireFormatGenerator narFromPath(const StorePath & path) override
     { unsupported("narFromPath"); }
 
     std::shared_ptr<const Realisation> queryRealisationUncached(const DrvOutput &) override

src/libstore/export-import.cc

@@ -33,7 +33,7 @@ void Store::exportPath(const StorePath & path, Sink & sink)
     HashSink hashSink(htSHA256);
     TeeSink teeSink(sink, hashSink);
 
-    narFromPath(path, teeSink);
+    teeSink << narFromPath(path);
 
     /* Refuse to export paths that have changed.  This prevents
        filesystem corruption from spreading to other machines.
@@ -46,8 +46,8 @@ void Store::exportPath(const StorePath & path, Sink & sink)
     teeSink
         << exportMagic
         << printStorePath(path);
-    CommonProto::write(*this,
-        CommonProto::WriteConn { .to = teeSink },
+    teeSink << CommonProto::write(*this,
+        CommonProto::WriteConn {},
         info->references);
     teeSink
         << (info->deriver ? printStorePath(*info->deriver) : "")

src/libstore/legacy-ssh-store.cc

@@ -74,7 +74,6 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
         operator ServeProto::WriteConn ()
         {
             return ServeProto::WriteConn {
-                .to = to,
                 .version = remoteVersion,
             };
         }
@@ -185,7 +184,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
                 << printStorePath(info.path)
                 << (info.deriver ? printStorePath(*info.deriver) : "")
                 << info.narHash.to_string(Base16, false);
-            ServeProto::write(*this, *conn, info.references);
+            conn->to << ServeProto::write(*this, *conn, info.references);
             conn->to
                 << info.registrationTime
                 << info.narSize
@@ -214,7 +213,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
             conn->to
                 << exportMagic
                 << printStorePath(info.path);
-            ServeProto::write(*this, *conn, info.references);
+            conn->to << ServeProto::write(*this, *conn, info.references);
             conn->to
                 << (info.deriver ? printStorePath(*info.deriver) : "")
                 << 0
@@ -227,13 +226,15 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
             throw Error("failed to add path '%s' to remote host '%s'", printStorePath(info.path), host);
     }
 
-    void narFromPath(const StorePath & path, Sink & sink) override
+    WireFormatGenerator narFromPath(const StorePath & path) override
     {
         auto conn(connections->get());
 
         conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
         conn->to.flush();
-        sink << copyNAR(conn->from);
+        return [] (auto conn) -> WireFormatGenerator {
+            co_yield copyNAR(conn->from);
+        }(std::move(conn));
     }
 
     std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
@@ -364,7 +365,7 @@ public:
         conn->to
             << ServeProto::Command::QueryClosure
             << includeOutputs;
-        ServeProto::write(*this, *conn, paths);
+        conn->to << ServeProto::write(*this, *conn, paths);
         conn->to.flush();
 
         for (auto & i : ServeProto::Serialise<StorePathSet>::read(*this, *conn))
@@ -380,7 +381,7 @@ public:
             << ServeProto::Command::QueryValidPaths
             << false // lock
             << maybeSubstitute;
-        ServeProto::write(*this, *conn, paths);
+        conn->to << ServeProto::write(*this, *conn, paths);
         conn->to.flush();
 
         return ServeProto::Serialise<StorePathSet>::read(*this, *conn);

src/libstore/length-prefixed-protocol-helper.hh

@@ -7,6 +7,7 @@
  */
 
 #include "types.hh"
+#include "serialise.hh"
 
 namespace nix {
 
@@ -45,7 +46,7 @@ struct LengthPrefixedProtoHelper;
     struct LengthPrefixedProtoHelper< Inner, T > \
     { \
         static T read(const Store & store, typename Inner::ReadConn conn); \
-        static void write(const Store & store, typename Inner::WriteConn conn, const T & str); \
+        [[nodiscard]] static WireFormatGenerator write(const Store & store, typename Inner::WriteConn conn, const T & str); \
     private: \
         template<typename U> using S = typename Inner::template Serialise<U>; \
     }
@@ -78,13 +79,13 @@ LengthPrefixedProtoHelper<Inner, std::vector<T>>::read(
 }
 
 template<class Inner, typename T>
-void
+WireFormatGenerator
 LengthPrefixedProtoHelper<Inner, std::vector<T>>::write(
     const Store & store, typename Inner::WriteConn conn, const std::vector<T> & resSet)
 {
-    conn.to << resSet.size();
+    co_yield resSet.size();
     for (auto & key : resSet) {
-        S<T>::write(store, conn, key);
+        co_yield S<T>::write(store, conn, key);
     }
 }
 
@@ -102,13 +103,13 @@ LengthPrefixedProtoHelper<Inner, std::set<T>>::read(
 }
 
 template<class Inner, typename T>
-void
+WireFormatGenerator
 LengthPrefixedProtoHelper<Inner, std::set<T>>::write(
     const Store & store, typename Inner::WriteConn conn, const std::set<T> & resSet)
 {
-    conn.to << resSet.size();
+    co_yield resSet.size();
     for (auto & key : resSet) {
-        S<T>::write(store, conn, key);
+        co_yield S<T>::write(store, conn, key);
     }
 }
 
@@ -128,14 +129,14 @@ LengthPrefixedProtoHelper<Inner, std::map<K, V>>::read(
 }
 
 template<class Inner, typename K, typename V>
-void
+WireFormatGenerator
 LengthPrefixedProtoHelper<Inner, std::map<K, V>>::write(
     const Store & store, typename Inner::WriteConn conn, const std::map<K, V> & resMap)
 {
-    conn.to << resMap.size();
+    co_yield resMap.size();
     for (auto & i : resMap) {
-        S<K>::write(store, conn, i.first);
-        S<V>::write(store, conn, i.second);
+        co_yield S<K>::write(store, conn, i.first);
+        co_yield S<V>::write(store, conn, i.second);
     }
 }
 
@@ -150,13 +151,24 @@ LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::read(
 }
 
 template<class Inner, typename... Ts>
-void
+WireFormatGenerator
 LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::write(
     const Store & store, typename Inner::WriteConn conn, const std::tuple<Ts...> & res)
 {
-    std::apply([&]<typename... Us>(const Us &... args) {
-        (S<Us>::write(store, conn, args), ...);
-    }, res);
+    auto fullArgs = std::apply(
+        [&](auto &... rest) {
+            return std::tuple<const Store &, typename Inner::WriteConn &, const Ts &...>(
+                std::cref(store), conn, rest...
+            );
+        },
+        res
+    );
+    return std::apply(
+        []<typename... Us>(auto & store, auto conn, const Us &... args) -> WireFormatGenerator {
+            (co_yield S<Us>::write(store, conn, args), ...);
+        },
+        fullArgs
+    );
 }
 
 }

src/libstore/local-fs-store.cc

@@ -78,11 +78,11 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
             std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
 }
 
-void LocalFSStore::narFromPath(const StorePath & path, Sink & sink)
+WireFormatGenerator LocalFSStore::narFromPath(const StorePath & path)
 {
     if (!isValidPath(path))
         throw Error("path '%s' does not exist in store", printStorePath(path));
-    sink << dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()));
+    return dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()));
 }
 
 const std::string LocalFSStore::drvsLogDir = "drvs";

src/libstore/local-fs-store.hh

@@ -42,7 +42,7 @@ public:
 
     LocalFSStore(const Params & params);
 
-    void narFromPath(const StorePath & path, Sink & sink) override;
+    WireFormatGenerator narFromPath(const StorePath & path) override;
     ref<FSAccessor> getFSAccessor() override;
 
     /**

src/libstore/make-content-addressed.cc

@@ -23,7 +23,7 @@ std::map<StorePath, StorePath> makeContentAddressed(
         std::string oldHashPart(path.hashPart());
 
         StringSink sink;
-        srcStore.narFromPath(path, sink);
+        sink << srcStore.narFromPath(path);
 
         StringMap rewrites;
 

src/libstore/remote-fs-accessor.cc

@@ -97,7 +97,7 @@ std::pair<ref<FSAccessor>, Path> RemoteFSAccessor::fetch(const Path & path_, boo
     }
 
     StringSink sink;
-    store->narFromPath(storePath, sink);
+    sink << store->narFromPath(storePath);
     return {addToCache(storePath.hashPart(), std::move(sink.s)), restPath};
 }
 

src/libstore/remote-store-connection.hh

@@ -81,7 +81,7 @@ struct RemoteStore::Connection
      */
     operator WorkerProto::WriteConn ()
     {
-        return WorkerProto::WriteConn {to, daemonVersion};
+        return WorkerProto::WriteConn {daemonVersion};
     }
 
     virtual ~Connection();

src/libstore/remote-store.cc

@@ -203,7 +203,7 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute
 {
     auto conn(getConnection());
     conn->to << WorkerProto::Op::QueryValidPaths;
-    WorkerProto::write(*this, *conn, paths);
+    conn->to << WorkerProto::write(*this, *conn, paths);
     if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) {
         conn->to << maybeSubstitute;
     }
@@ -225,7 +225,7 @@ StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
 {
     auto conn(getConnection());
     conn->to << WorkerProto::Op::QuerySubstitutablePaths;
-    WorkerProto::write(*this, *conn, paths);
+    conn->to << WorkerProto::write(*this, *conn, paths);
     conn.processStderr();
     return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
 }
@@ -243,9 +243,9 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
         StorePathSet paths;
         for (auto & path : pathsMap)
             paths.insert(path.first);
-        WorkerProto::write(*this, *conn, paths);
+        conn->to << WorkerProto::write(*this, *conn, paths);
     } else
-        WorkerProto::write(*this, *conn, pathsMap);
+        conn->to << WorkerProto::write(*this, *conn, pathsMap);
     conn.processStderr();
     size_t count = readNum<size_t>(conn->from);
     for (size_t n = 0; n < count; n++) {
@@ -377,7 +377,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
             << WorkerProto::Op::AddToStore
             << name
             << caMethod.render(hashType);
-        WorkerProto::write(*this, *conn, references);
+        conn->to << WorkerProto::write(*this, *conn, references);
         conn->to << repair;
 
         // The dump source may invoke the store, so we need to make some room.
@@ -402,7 +402,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
                         name, printHashType(hashType));
                 std::string s = dump.drain();
                 conn->to << WorkerProto::Op::AddTextToStore << name << s;
-                WorkerProto::write(*this, *conn, references);
+                conn->to << WorkerProto::write(*this, *conn, references);
                 conn.processStderr();
             },
             [&](const FileIngestionMethod & fim) -> void {
@@ -462,7 +462,7 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
              << printStorePath(info.path)
              << (info.deriver ? printStorePath(*info.deriver) : "")
              << info.narHash.to_string(Base16, false);
-    WorkerProto::write(*this, *conn, info.references);
+    conn->to << WorkerProto::write(*this, *conn, info.references);
     conn->to << info.registrationTime << info.narSize
              << info.ultimate << info.sigs << renderContentAddress(info.ca)
              << repair << !checkSigs;
@@ -485,17 +485,26 @@ void RemoteStore::addMultipleToStore(
 {
     auto remoteVersion = getProtocol();
 
-    auto source = sinkToSource([&](Sink & sink) {
-        sink << pathsToCopy.size();
-        for (auto & [pathInfo, pathSource] : pathsToCopy) {
-            WorkerProto::Serialise<ValidPathInfo>::write(*this,
-                 WorkerProto::WriteConn {sink, remoteVersion},
-                 pathInfo);
-            pathSource->drainInto(sink);
-        }
-    });
+    GeneratorSource source{
+        [](auto self, auto & pathsToCopy, auto remoteVersion) -> WireFormatGenerator {
+            co_yield pathsToCopy.size();
+            for (auto & [pathInfo, pathSource] : pathsToCopy) {
+                co_yield WorkerProto::Serialise<ValidPathInfo>::write(*self,
+                    WorkerProto::WriteConn {remoteVersion},
+                    pathInfo);
+                try {
+                    char buf[65536];
+                    while (true) {
+                        const auto read = pathSource->read(buf, sizeof(buf));
+                        co_yield std::span{buf, read};
+                    }
+                } catch (EndOfFile &) {
+                }
+            }
+        }(this, pathsToCopy, remoteVersion)
+    };
 
-    addMultipleToStore(*source, repair, checkSigs);
+    addMultipleToStore(source, repair, checkSigs);
 }
 
 void RemoteStore::addMultipleToStore(
@@ -536,7 +545,7 @@ void RemoteStore::registerDrvOutput(const Realisation & info)
         conn->to << info.id.to_string();
         conn->to << std::string(info.outPath.to_string());
     } else {
-        WorkerProto::write(*this, *conn, info);
+        conn->to << WorkerProto::write(*this, *conn, info);
     }
     conn.processStderr();
 }
@@ -597,7 +606,7 @@ void RemoteStore::buildPaths(const std::vector<DerivedPath> & drvPaths, BuildMod
 
     auto conn(getConnection());
     conn->to << WorkerProto::Op::BuildPaths;
-    WorkerProto::write(*this, *conn, drvPaths);
+    conn->to << WorkerProto::write(*this, *conn, drvPaths);
     conn->to << buildMode;
     conn.processStderr();
     readInt(conn->from);
@@ -615,7 +624,7 @@ std::vector<KeyedBuildResult> RemoteStore::buildPathsWithResults(
 
     if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
         conn->to << WorkerProto::Op::BuildPathsWithResults;
-        WorkerProto::write(*this, *conn, paths);
+        conn->to << WorkerProto::write(*this, *conn, paths);
         conn->to << buildMode;
         conn.processStderr();
         return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, *conn);
@@ -740,7 +749,7 @@ void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
 
     conn->to
         << WorkerProto::Op::CollectGarbage << options.action;
-    WorkerProto::write(*this, *conn, options.pathsToDelete);
+    conn->to << WorkerProto::write(*this, *conn, options.pathsToDelete);
     conn->to << options.ignoreLiveness
         << options.maxFreed
         /* removed options */
@@ -792,7 +801,7 @@ void RemoteStore::queryMissing(const std::vector<DerivedPath> & targets,
 {
     auto conn(getConnection());
     conn->to << WorkerProto::Op::QueryMissing;
-    WorkerProto::write(*this, *conn, targets);
+    conn->to << WorkerProto::write(*this, *conn, targets);
     conn.processStderr();
     willBuild = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
     willSubstitute = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
@@ -848,12 +857,14 @@ RemoteStore::Connection::~Connection()
     }
 }
 
-void RemoteStore::narFromPath(const StorePath & path, Sink & sink)
+WireFormatGenerator RemoteStore::narFromPath(const StorePath & path)
 {
     auto conn(connections->get());
     conn->to << WorkerProto::Op::NarFromPath << printStorePath(path);
     conn->processStderr();
-    sink << copyNAR(conn->from);
+    return [](auto conn) -> WireFormatGenerator {
+        co_yield copyNAR(conn->from);
+    }(std::move(conn));
 }
 
 ref<FSAccessor> RemoteStore::getFSAccessor()

src/libstore/remote-store.hh

@@ -183,7 +183,7 @@ protected:
 
     virtual ref<FSAccessor> getFSAccessor() override;
 
-    virtual void narFromPath(const StorePath & path, Sink & sink) override;
+    virtual WireFormatGenerator narFromPath(const StorePath & path) override;
 
 private:
 

src/libstore/serve-protocol-impl.hh

@@ -20,9 +20,9 @@ namespace nix {
     { \
         return LengthPrefixedProtoHelper<ServeProto, T >::read(store, conn); \
     } \
-    TEMPLATE void ServeProto::Serialise< T >::write(const Store & store, ServeProto::WriteConn conn, const T & t) \
+    TEMPLATE [[nodiscard]] WireFormatGenerator ServeProto::Serialise< T >::write(const Store & store, ServeProto::WriteConn conn, const T & t) \
     { \
-        LengthPrefixedProtoHelper<ServeProto, T >::write(store, conn, t); \
+        return LengthPrefixedProtoHelper<ServeProto, T >::write(store, conn, t); \
     }
 
 SERVE_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -46,10 +46,11 @@ struct ServeProto::Serialise
         return CommonProto::Serialise<T>::read(store,
             CommonProto::ReadConn { .from = conn.from });
     }
-    static void write(const Store & store, ServeProto::WriteConn conn, const T & t)
+    [[nodiscard]]
+    static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t)
     {
-        CommonProto::Serialise<T>::write(store,
-            CommonProto::WriteConn { .to = conn.to },
+        return CommonProto::Serialise<T>::write(store,
+            CommonProto::WriteConn {},
             t);
     }
 };

src/libstore/serve-protocol.cc

@@ -34,23 +34,22 @@ BuildResult ServeProto::Serialise<BuildResult>::read(const Store & store, ServeP
     return status;
 }
 
-void ServeProto::Serialise<BuildResult>::write(const Store & store, ServeProto::WriteConn conn, const BuildResult & status)
+WireFormatGenerator ServeProto::Serialise<BuildResult>::write(const Store & store, ServeProto::WriteConn conn, const BuildResult & status)
 {
-    conn.to
-        << status.status
-        << status.errorMsg;
+    co_yield status.status;
+    co_yield status.errorMsg;
 
-    if (GET_PROTOCOL_MINOR(conn.version) >= 3)
-        conn.to
-            << status.timesBuilt
-            << status.isNonDeterministic
-            << status.startTime
-            << status.stopTime;
+    if (GET_PROTOCOL_MINOR(conn.version) >= 3) {
+        co_yield status.timesBuilt;
+        co_yield status.isNonDeterministic;
+        co_yield status.startTime;
+        co_yield status.stopTime;
+    }
     if (GET_PROTOCOL_MINOR(conn.version) >= 6) {
         DrvOutputs builtOutputs;
         for (auto & [output, realisation] : status.builtOutputs)
             builtOutputs.insert_or_assign(realisation.id, realisation);
-        ServeProto::write(store, conn, builtOutputs);
+        co_yield ServeProto::write(store, conn, builtOutputs);
     }
 }
 
@@ -80,21 +79,19 @@ UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(const Sto
     return info;
 }
 
-void ServeProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & info)
+WireFormatGenerator ServeProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & info)
 {
-    conn.to
-        << (info.deriver ? store.printStorePath(*info.deriver) : "");
+    co_yield (info.deriver ? store.printStorePath(*info.deriver) : "");
 
-    ServeProto::write(store, conn, info.references);
+    co_yield ServeProto::write(store, conn, info.references);
     // !!! Maybe we want compression?
-    conn.to
-        << info.narSize // downloadSize, lie a little
-        << info.narSize;
-    if (GET_PROTOCOL_MINOR(conn.version) >= 4)
-        conn.to
-            << info.narHash.to_string(Base32, true)
-            << renderContentAddress(info.ca)
-            << info.sigs;
+    co_yield info.narSize; // downloadSize, lie a little
+    co_yield info.narSize;
+    if (GET_PROTOCOL_MINOR(conn.version) >= 4) {
+        co_yield info.narHash.to_string(Base32, true);
+        co_yield renderContentAddress(info.ca);
+        co_yield info.sigs;
+    }
 }
 
 }

src/libstore/serve-protocol.hh

@@ -60,7 +60,6 @@ struct ServeProto
      * canonical serializers below.
      */
     struct WriteConn {
-        Sink & to;
         Version version;
     };
 
@@ -79,7 +78,7 @@ struct ServeProto
 #if 0
     {
         static T read(const Store & store, ReadConn conn);
-        static void write(const Store & store, WriteConn conn, const T & t);
+        static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
     };
 #endif
 
@@ -88,9 +87,10 @@ struct ServeProto
      * infer the type instead of having to write it down explicitly.
      */
     template<typename T>
-    static void write(const Store & store, WriteConn conn, const T & t)
+    [[nodiscard]]
+    static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
     {
-        ServeProto::Serialise<T>::write(store, conn, t);
+        return ServeProto::Serialise<T>::write(store, conn, t);
     }
 };
 
@@ -142,7 +142,7 @@ inline std::ostream & operator << (std::ostream & s, ServeProto::Command op)
     struct ServeProto::Serialise< T > \
     { \
         static T read(const Store & store, ServeProto::ReadConn conn); \
-        static void write(const Store & store, ServeProto::WriteConn conn, const T & t); \
+        [[nodiscard]] static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t); \
     };
 
 template<>

src/libstore/store-api.cc

@@ -335,9 +335,9 @@ void Store::addMultipleToStore(
             info.ultimate = false;
 
             /* Make sure that the Source object is destroyed when
-               we're done. In particular, a SinkToSource object must
-               be destroyed to ensure that the destructors on its
-               stack frame are run; this includes
+               we're done. In particular, a coroutine object must
+               be destroyed to ensure that the destructors in its
+               state are run; this includes
                LegacySSHStore::narFromPath()'s connection lock. */
             auto source = std::move(source_);
 
@@ -1059,16 +1059,19 @@ void copyStorePath(
         info = info2;
     }
 
-    auto source = sinkToSource([&](Sink & sink) {
-        LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
-            total += data.size();
-            act.progress(total, info->narSize);
-        });
-        TeeSink tee { sink, progressSink };
-        srcStore.narFromPath(storePath, tee);
-    });
+    GeneratorSource source{
+        [](auto & act, auto & info, auto & srcStore, auto & storePath) -> WireFormatGenerator {
+            auto nar = srcStore.narFromPath(storePath);
+            uint64_t total = 0;
+            while (auto data = nar.next()) {
+                total += data->size();
+                act.progress(total, info->narSize);
+                co_yield *data;
+            }
+        }(act, info, srcStore, storePath)
+    };
 
-    dstStore.addToStore(*info, *source, repair, checkSigs);
+    dstStore.addToStore(*info, source, repair, checkSigs);
 }
 
 
@@ -1180,31 +1183,34 @@ std::map<StorePath, StorePath> copyPaths(
         ValidPathInfo infoForDst = *info;
         infoForDst.path = storePathForDst;
 
-        auto source =
-            sinkToSource([&srcStore, &dstStore, missingPath = missingPath, info = std::move(info)](Sink & sink) {
-                // We can reasonably assume that the copy will happen whenever we
-                // read the path, so log something about that at that point
-                auto srcUri = srcStore.getUri();
-                auto dstUri = dstStore.getUri();
-                auto storePathS = srcStore.printStorePath(missingPath);
-                Activity act(
-                    *logger,
-                    lvlInfo,
-                    actCopyPath,
-                    makeCopyPathMessage(srcUri, dstUri, storePathS),
-                    {storePathS, srcUri, dstUri}
-                );
-                PushActivity pact(act.id);
+        auto source = [](auto & srcStore, auto & dstStore, auto missingPath, auto info
+                      ) -> WireFormatGenerator {
+            // We can reasonably assume that the copy will happen whenever we
+            // read the path, so log something about that at that point
+            auto srcUri = srcStore.getUri();
+            auto dstUri = dstStore.getUri();
+            auto storePathS = srcStore.printStorePath(missingPath);
+            Activity act(
+                *logger,
+                lvlInfo,
+                actCopyPath,
+                makeCopyPathMessage(srcUri, dstUri, storePathS),
+                {storePathS, srcUri, dstUri}
+            );
+            PushActivity pact(act.id);
 
-                LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
-                    total += data.size();
-                    act.progress(total, info->narSize);
-                });
-                TeeSink tee{sink, progressSink};
-
-                srcStore.narFromPath(missingPath, tee);
-            });
-        pathsToCopy.push_back(std::pair{infoForDst, std::move(source)});
+            auto nar = srcStore.narFromPath(missingPath);
+            uint64_t total = 0;
+            while (auto data = nar.next()) {
+                total += data->size();
+                act.progress(total, info->narSize);
+                co_yield *data;
+            }
+        };
+        pathsToCopy.push_back(std::pair{
+            infoForDst,
+            std::make_unique<GeneratorSource>(source(srcStore, dstStore, missingPath, info))
+        });
     }
 
     dstStore.addMultipleToStore(pathsToCopy, act, repair, checkSigs);

src/libstore/store-api.hh

@@ -577,9 +577,9 @@ public:
     { return registerDrvOutput(output); }
 
     /**
-     * Write a NAR dump of a store path.
+     * Generate a NAR dump of a store path.
      */
-    virtual void narFromPath(const StorePath & path, Sink & sink) = 0;
+    virtual WireFormatGenerator narFromPath(const StorePath & path) = 0;
 
     /**
      * For each path, if it's a derivation, build it.  Building a

src/libstore/uds-remote-store.hh

@@ -38,8 +38,8 @@ public:
     ref<FSAccessor> getFSAccessor() override
     { return LocalFSStore::getFSAccessor(); }
 
-    void narFromPath(const StorePath & path, Sink & sink) override
-    { LocalFSStore::narFromPath(path, sink); }
+    WireFormatGenerator narFromPath(const StorePath & path) override
+    { return LocalFSStore::narFromPath(path); }
 
     /**
      * Implementation of `IndirectRootStore::addIndirectRoot()` which

src/libstore/worker-protocol-impl.hh

@@ -20,9 +20,9 @@ namespace nix {
     { \
         return LengthPrefixedProtoHelper<WorkerProto, T >::read(store, conn); \
     } \
-    TEMPLATE void WorkerProto::Serialise< T >::write(const Store & store, WorkerProto::WriteConn conn, const T & t) \
+    TEMPLATE [[nodiscard]] WireFormatGenerator WorkerProto::Serialise< T >::write(const Store & store, WorkerProto::WriteConn conn, const T & t) \
     { \
-        LengthPrefixedProtoHelper<WorkerProto, T >::write(store, conn, t); \
+        return LengthPrefixedProtoHelper<WorkerProto, T >::write(store, conn, t); \
     }
 
 WORKER_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -46,10 +46,11 @@ struct WorkerProto::Serialise
         return CommonProto::Serialise<T>::read(store,
             CommonProto::ReadConn { .from = conn.from });
     }
-    static void write(const Store & store, WorkerProto::WriteConn conn, const T & t)
+    [[nodiscard]]
+    static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t)
     {
-        CommonProto::Serialise<T>::write(store,
-            CommonProto::WriteConn { .to = conn.to },
+        return CommonProto::Serialise<T>::write(store,
+            CommonProto::WriteConn {},
             t);
     }
 };

src/libstore/worker-protocol.cc

@@ -28,17 +28,17 @@ std::optional<TrustedFlag> WorkerProto::Serialise<std::optional<TrustedFlag>>::r
     }
 }
 
-void WorkerProto::Serialise<std::optional<TrustedFlag>>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
+WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
 {
     if (!optTrusted)
-        conn.to << (uint8_t)0;
+        co_yield (uint8_t)0;
     else {
         switch (*optTrusted) {
         case Trusted:
-            conn.to << (uint8_t)1;
+            co_yield (uint8_t)1;
             break;
         case NotTrusted:
-            conn.to << (uint8_t)2;
+            co_yield (uint8_t)2;
             break;
         default:
             assert(false);
@@ -57,23 +57,23 @@ DerivedPath WorkerProto::Serialise<DerivedPath>::read(const Store & store, Worke
     }
 }
 
-void WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
+WireFormatGenerator WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
 {
     if (GET_PROTOCOL_MINOR(conn.version) >= 30) {
-        conn.to << req.to_string_legacy(store);
+        co_yield req.to_string_legacy(store);
     } else {
         auto sOrDrvPath = StorePathWithOutputs::tryFromDerivedPath(req);
-        std::visit(overloaded {
-            [&](const StorePathWithOutputs & s) {
-                conn.to << s.to_string(store);
+        co_yield std::visit(overloaded {
+            [&](const StorePathWithOutputs & s) -> std::string {
+                return s.to_string(store);
             },
-            [&](const StorePath & drvPath) {
+            [&](const StorePath & drvPath) -> std::string {
                 throw Error("trying to request '%s', but daemon protocol %d.%d is too old (< 1.29) to request a derivation file",
                     store.printStorePath(drvPath),
                     GET_PROTOCOL_MAJOR(conn.version),
                     GET_PROTOCOL_MINOR(conn.version));
             },
-            [&](std::monostate) {
+            [&](std::monostate) -> std::string {
                 throw Error("wanted to build a derivation that is itself a build product, but protocols do not support that. Try upgrading the Nix implementation on the other end of this connection");
             },
         }, sOrDrvPath);
@@ -91,10 +91,10 @@ KeyedBuildResult WorkerProto::Serialise<KeyedBuildResult>::read(const Store & st
     };
 }
 
-void WorkerProto::Serialise<KeyedBuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res)
+WireFormatGenerator WorkerProto::Serialise<KeyedBuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res)
 {
-    WorkerProto::write(store, conn, res.path);
-    WorkerProto::write(store, conn, static_cast<const BuildResult &>(res));
+    co_yield WorkerProto::write(store, conn, res.path);
+    co_yield WorkerProto::write(store, conn, static_cast<const BuildResult &>(res));
 }
 
 
@@ -120,23 +120,21 @@ BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, Worke
     return res;
 }
 
-void WorkerProto::Serialise<BuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res)
+WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res)
 {
-    conn.to
-        << res.status
-        << res.errorMsg;
+    co_yield res.status;
+    co_yield res.errorMsg;
     if (GET_PROTOCOL_MINOR(conn.version) >= 29) {
-        conn.to
-            << res.timesBuilt
-            << res.isNonDeterministic
-            << res.startTime
-            << res.stopTime;
+        co_yield res.timesBuilt;
+        co_yield res.isNonDeterministic;
+        co_yield res.startTime;
+        co_yield res.stopTime;
     }
     if (GET_PROTOCOL_MINOR(conn.version) >= 28) {
         DrvOutputs builtOutputs;
         for (auto & [output, realisation] : res.builtOutputs)
             builtOutputs.insert_or_assign(realisation.id, realisation);
-        WorkerProto::write(store, conn, builtOutputs);
+        co_yield WorkerProto::write(store, conn, builtOutputs);
     }
 }
 
@@ -150,10 +148,10 @@ ValidPathInfo WorkerProto::Serialise<ValidPathInfo>::read(const Store & store, R
     };
 }
 
-void WorkerProto::Serialise<ValidPathInfo>::write(const Store & store, WriteConn conn, const ValidPathInfo & pathInfo)
+WireFormatGenerator WorkerProto::Serialise<ValidPathInfo>::write(const Store & store, WriteConn conn, const ValidPathInfo & pathInfo)
 {
-    WorkerProto::write(store, conn, pathInfo.path);
-    WorkerProto::write(store, conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
+    co_yield WorkerProto::write(store, conn, pathInfo.path);
+    co_yield WorkerProto::write(store, conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
 }
 
 
@@ -173,18 +171,17 @@ UnkeyedValidPathInfo WorkerProto::Serialise<UnkeyedValidPathInfo>::read(const St
     return info;
 }
 
-void WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
+WireFormatGenerator WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
 {
-    conn.to
-        << (pathInfo.deriver ? store.printStorePath(*pathInfo.deriver) : "")
-        << pathInfo.narHash.to_string(Base16, false);
-    WorkerProto::write(store, conn, pathInfo.references);
-    conn.to << pathInfo.registrationTime << pathInfo.narSize;
+    co_yield (pathInfo.deriver ? store.printStorePath(*pathInfo.deriver) : "");
+    co_yield pathInfo.narHash.to_string(Base16, false);
+    co_yield WorkerProto::write(store, conn, pathInfo.references);
+    co_yield pathInfo.registrationTime;
+    co_yield pathInfo.narSize;
 
-    conn.to
-        << pathInfo.ultimate
-        << pathInfo.sigs
-        << renderContentAddress(pathInfo.ca);
+    co_yield pathInfo.ultimate;
+    co_yield pathInfo.sigs;
+    co_yield renderContentAddress(pathInfo.ca);
 }
 
 }

src/libstore/worker-protocol.hh

@@ -87,10 +87,9 @@ struct WorkerProto
      * canonical serializers below.
      */
     struct WriteConn {
-        Sink & to;
         Version version;
 
-        WriteConn(Sink & to, Version version) : to(to), version(version) {
+        explicit WriteConn(Version version) : version(version) {
             assert(version >= MIN_SUPPORTED_WORKER_PROTO_VERSION);
         }
     };
@@ -122,7 +121,7 @@ struct WorkerProto
 #if 0
     {
         static T read(const Store & store, ReadConn conn);
-        static void write(const Store & store, WriteConn conn, const T & t);
+        static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
     };
 #endif
 
@@ -131,9 +130,10 @@ struct WorkerProto
      * infer the type instead of having to write it down explicitly.
      */
     template<typename T>
-    static void write(const Store & store, WriteConn conn, const T & t)
+    [[nodiscard]]
+    static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
     {
-        WorkerProto::Serialise<T>::write(store, conn, t);
+        return WorkerProto::Serialise<T>::write(store, conn, t);
     }
 };
 
@@ -219,7 +219,7 @@ inline std::ostream & operator << (std::ostream & s, WorkerProto::Op op)
     struct WorkerProto::Serialise< T > \
     { \
         static T read(const Store & store, WorkerProto::ReadConn conn); \
-        static void write(const Store & store, WorkerProto::WriteConn conn, const T & t); \
+        [[nodiscard]] static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t); \
     };
 
 template<>

src/libutil/archive.cc

@@ -1,5 +1,6 @@
 #include <cerrno>
 #include <algorithm>
+#include <string_view>
 #include <vector>
 #include <map>
 
@@ -13,6 +14,8 @@
 
 #include "archive.hh"
 #include "file-system.hh"
+#include "finally.hh"
+#include "serialise.hh"
 #include "config.hh"
 #include "logging.hh"
 #include "signals.hh"
@@ -174,31 +177,6 @@ static void skipGeneric(Source & source)
 #endif
 
 
-static WireFormatGenerator parseContents(ParseSink & sink, Source & source, const Path & path)
-{
-    uint64_t size = readLongLong(source);
-    co_yield size;
-
-    sink.preallocateContents(size);
-
-    uint64_t left = size;
-    std::array<char, 65536> buf;
-
-    while (left) {
-        checkInterrupt();
-        auto n = buf.size();
-        if ((uint64_t)n > left) n = left;
-        source(buf.data(), n);
-        co_yield std::span{buf.data(), n};
-        sink.receiveContents({buf.data(), n});
-        left -= n;
-    }
-
-    readPadding(size, source);
-    co_yield SerializingTransform::padding(size);
-}
-
-
 struct CaseInsensitiveCompare
 {
     bool operator() (const std::string & a, const std::string & b) const
@@ -207,129 +185,201 @@ struct CaseInsensitiveCompare
     }
 };
 
-static WireFormatGenerator parse(ParseSink & sink, Source & source, const Path & path)
+namespace nar {
+
+static Generator<Entry> parseObject(Source & source, const Path & path)
 {
-    std::string s;
+#define EXPECT(raw, kind)                              \
+    do {                                               \
+        const auto s = readString(source);             \
+        if (s != raw) {                                \
+            throw badArchive("expected " kind " tag"); \
+        }                                              \
+        co_yield MetadataString{s};                    \
+    } while (0)
 
-    s = readString(source);
-    co_yield s;
-    if (s != "(") throw badArchive("expected open tag");
+    EXPECT("(", "open");
+    EXPECT("type", "type");
 
-    enum { tpUnknown, tpRegular, tpDirectory, tpSymlink } type = tpUnknown;
+    checkInterrupt();
 
-    std::map<Path, int, CaseInsensitiveCompare> names;
+    const auto t = readString(source);
+    co_yield MetadataString{t};
 
-    while (1) {
-        checkInterrupt();
-
-        s = readString(source);
-        co_yield s;
-
-        if (s == ")") {
-            break;
-        }
-
-        else if (s == "type") {
-            if (type != tpUnknown)
-                throw badArchive("multiple type fields");
-            std::string t = readString(source);
-            co_yield t;
-
-            if (t == "regular") {
-                type = tpRegular;
-                sink.createRegularFile(path);
-            }
-
-            else if (t == "directory") {
-                sink.createDirectory(path);
-                type = tpDirectory;
-            }
-
-            else if (t == "symlink") {
-                type = tpSymlink;
-            }
-
-            else throw badArchive("unknown file type " + t);
-
-        }
-
-        else if (s == "contents" && type == tpRegular) {
-            co_yield parseContents(sink, source, path);
-            sink.closeRegularFile();
-        }
-
-        else if (s == "executable" && type == tpRegular) {
+    if (t == "regular") {
+        auto contentsOrFlag = readString(source);
+        co_yield MetadataString{contentsOrFlag};
+        const bool executable = contentsOrFlag == "executable";
+        if (executable) {
             auto s = readString(source);
-            co_yield s;
-            if (s != "") throw badArchive("executable marker has non-empty value");
-            sink.isExecutable();
+            co_yield MetadataString{s};
+            if (s != "") {
+                throw badArchive("executable marker has non-empty value");
+            }
+            contentsOrFlag = readString(source);
+            co_yield MetadataString{contentsOrFlag};
         }
+        if (contentsOrFlag == "contents") {
+            const uint64_t size = readLongLong(source);
+            co_yield MetadataRaw{SerializingTransform()(size)};
+            auto makeReader = [](Source & source, uint64_t & left) -> Generator<Bytes> {
+                std::array<char, 65536> buf;
 
-        else if (s == "entry" && type == tpDirectory) {
-            std::string name, prevName;
-
-            s = readString(source);
-            co_yield s;
-            if (s != "(") throw badArchive("expected open tag");
+                while (left) {
+                    checkInterrupt();
+                    auto n = std::min<uint64_t>(buf.size(), left);
+                    source(buf.data(), n);
+                    co_yield std::span{buf.data(), n};
+                    left -= n;
+                }
+            };
+            auto left = size;
+            co_yield File{path, executable, size, makeReader(source, left)};
+            // we could drain the remainder of the file, but coroutines being interruptible
+            // at any time makes this difficult. for files this is not that hard, but being
+            // consistent with directories is more important than handling the simple case.
+            assert(left == 0);
+            readPadding(size, source);
+            co_yield MetadataRaw{SerializingTransform::padding(size)};
+        } else {
+            throw badArchive("file without contents found: " + path);
+        }
+    } else if (t == "directory") {
+        auto makeReader = [](Source & source, const Path & path, bool & completed
+                          ) -> Generator<Entry> {
+            std::map<Path, int, CaseInsensitiveCompare> names;
+            std::string prevName;
 
             while (1) {
                 checkInterrupt();
 
-                s = readString(source);
-                co_yield s;
-
-                if (s == ")") {
-                    break;
-                } else if (s == "name") {
-                    name = readString(source);
-                    co_yield name;
-                    if (name.empty() || name == "." || name == ".." || name.find('/') != std::string::npos || name.find((char) 0) != std::string::npos)
-                        throw Error("NAR contains invalid file name '%1%'", name);
-                    if (name <= prevName)
-                        throw Error("NAR directory is not sorted");
-                    prevName = name;
-                    if (archiveSettings.useCaseHack) {
-                        auto i = names.find(name);
-                        if (i != names.end()) {
-                            debug("case collision between '%1%' and '%2%'", i->first, name);
-                            name += caseHackSuffix;
-                            name += std::to_string(++i->second);
-                        } else
-                            names[name] = 0;
+                {
+                    const auto s = readString(source);
+                    co_yield MetadataString{s};
+                    if (s == ")") {
+                        completed = true;
+                        co_return;
+                    } else if (s != "entry") {
+                        throw badArchive("expected entry tag");
                     }
-                } else if (s == "node") {
-                    if (name.empty()) throw badArchive("entry name missing");
-                    co_yield parse(sink, source, path + "/" + name);
-                } else
-                    throw badArchive("unknown field " + s);
+                    EXPECT("(", "open");
+                }
+
+                EXPECT("name", "name");
+                auto name = readString(source);
+                co_yield MetadataString{name};
+                if (name.empty() || name == "." || name == ".."
+                    || name.find('/') != std::string::npos
+                    || name.find((char) 0) != std::string::npos)
+                {
+                    throw Error("NAR contains invalid file name '%1%'", name);
+                }
+                if (name <= prevName) {
+                    throw Error("NAR directory is not sorted");
+                }
+                prevName = name;
+                if (archiveSettings.useCaseHack) {
+                    auto i = names.find(name);
+                    if (i != names.end()) {
+                        debug("case collision between '%1%' and '%2%'", i->first, name);
+                        name += caseHackSuffix;
+                        name += std::to_string(++i->second);
+                    } else {
+                        names[name] = 0;
+                    }
+                }
+
+                EXPECT("node", "node");
+                co_yield parseObject(source, path + "/" + name);
+                EXPECT(")", "close");
             }
-        }
-
-        else if (s == "target" && type == tpSymlink) {
-            std::string target = readString(source);
-            co_yield target;
-            sink.createSymlink(path, target);
-        }
-
-        else
-            throw badArchive("unknown field " + s);
+        };
+        bool completed = false;
+        co_yield Directory{path, makeReader(source, path, completed)};
+        // directories may nest, so to drain a directory properly we'd have to add a Finally
+        // argument to the generator to ensure that the draining code is always run. this is
+        // usually not necessary, hard to follow, and rather error-prone on top of all that.
+        assert(completed);
+        // directories are terminated already, don't try to read another ")"
+        co_return;
+    } else if (t == "symlink") {
+        EXPECT("target", "target");
+        std::string target = readString(source);
+        co_yield MetadataString{target};
+        co_yield Symlink{path, target};
+    } else {
+        throw badArchive("unknown file type " + t);
     }
+
+    EXPECT(")", "close");
+
+#undef EXPECT
 }
 
-
-WireFormatGenerator parseAndCopyDump(ParseSink & sink, Source & source)
+Generator<Entry> parse(Source & source)
 {
     std::string version;
     try {
         version = readString(source, narVersionMagic1.size());
-        co_yield version;
+        co_yield MetadataString{version};
     } catch (SerialisationError & e) {
         /* This generally means the integer at the start couldn't be
            decoded.  Ignore and throw the exception below. */
     }
     if (version != narVersionMagic1)
         throw badArchive("input doesn't look like a Nix archive");
-    co_yield parse(sink, source, "");
+    co_yield parseObject(source, "");
+}
+
+}
+
+
+static WireFormatGenerator restore(ParseSink & sink, Generator<nar::Entry> nar)
+{
+    while (auto entry = nar.next()) {
+        co_yield std::visit(
+            overloaded{
+                [](nar::MetadataString m) -> WireFormatGenerator {
+                    co_yield m.data;
+                },
+                [](nar::MetadataRaw r) -> WireFormatGenerator {
+                    co_yield r.raw;
+                },
+                [&](nar::File f) {
+                    return [](auto f, auto & sink) -> WireFormatGenerator {
+                        sink.createRegularFile(f.path);
+                        sink.preallocateContents(f.size);
+                        if (f.executable) {
+                            sink.isExecutable();
+                        }
+                        while (auto block = f.contents.next()) {
+                            sink.receiveContents(std::string_view{block->data(), block->size()});
+                            co_yield *block;
+                        }
+                        sink.closeRegularFile();
+                    }(std::move(f), sink);
+                },
+                [&](nar::Symlink sl) {
+                    return [](auto sl, auto & sink) -> WireFormatGenerator {
+                        sink.createSymlink(sl.path, sl.target);
+                        co_return;
+                    }(std::move(sl), sink);
+                },
+                [&](nar::Directory d) {
+                    return [](auto d, auto & sink) -> WireFormatGenerator {
+                        sink.createDirectory(d.path);
+                        return restore(sink, std::move(d.contents));
+                    }(std::move(d), sink);
+                },
+            },
+            std::move(*entry)
+        );
+    }
+}
+
+WireFormatGenerator parseAndCopyDump(ParseSink & sink, Source & source)
+{
+    return restore(sink, nar::parse(source));
 }
 
 void parseDump(ParseSink & sink, Source & source)

src/libutil/archive.hh

@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "generator.hh"
 #include "types.hh"
 #include "serialise.hh"
 #include "file-system.hh"
@@ -116,6 +117,49 @@ struct RetrieveRegularNARSink : ParseSink
     }
 };
 
+namespace nar {
+
+struct MetadataString;
+struct MetadataRaw;
+struct File;
+struct Symlink;
+struct Directory;
+using Entry = std::variant<MetadataString, MetadataRaw, File, Symlink, Directory>;
+
+struct MetadataString
+{
+    std::string_view data;
+};
+
+struct MetadataRaw
+{
+    Bytes raw;
+};
+
+struct File
+{
+    const Path & path;
+    bool executable;
+    uint64_t size;
+    Generator<Bytes> contents;
+};
+
+struct Symlink
+{
+    const Path & path;
+    const Path & target;
+};
+
+struct Directory
+{
+    const Path & path;
+    Generator<Entry> contents;
+};
+
+Generator<Entry> parse(Source & source);
+
+}
+
 WireFormatGenerator parseAndCopyDump(ParseSink & sink, Source & source);
 void parseDump(ParseSink & sink, Source & source);
 

src/libutil/file-system.cc

@@ -1,4 +1,5 @@
 #include <sys/time.h>
+#include <cstdlib>
 #include <filesystem>
 #include <atomic>
 
@@ -106,6 +107,24 @@ Path canonPath(PathView path, bool resolveSymlinks)
     return s.empty() ? "/" : std::move(s);
 }
 
+Path realPath(Path const & path)
+{
+    // With nullptr, realpath() malloc's and returns a new c-string.
+    char * resolved = realpath(path.c_str(), nullptr);
+    int saved = errno;
+    if (resolved == nullptr) {
+        throw SysError(saved, "cannot get realpath for '%s'", path);
+    }
+
+    Finally const _free([&] { free(resolved); });
+
+    // There's not really a from_raw_parts() for std::string.
+    // The copy is not a big deal.
+    Path ret(resolved);
+
+    return ret;
+}
+
 void chmodPath(const Path & path, mode_t mode)
 {
     if (chmod(path.c_str(), mode) == -1)

src/libutil/file-system.hh

@@ -46,6 +46,22 @@ Path absPath(Path path,
  */
 Path canonPath(PathView path, bool resolveSymlinks = false);
 
+/**
+ * Resolves a file path to a fully absolute path with no symbolic links.
+ *
+ * @param path The path to resolve. If it is relative, it will be resolved relative
+ * to the process's current directory.
+ *
+ * @note This is not a pure function; it performs this resolution by querying
+ * the filesystem.
+ *
+ * @note @ref path sadly must be (a reference to) an owned string, as std::string_view
+ * are not valid C strings...
+ *
+ * @return The fully resolved path.
+ */
+Path realPath(Path const & path);
+
 /**
  * Change the permissions of a path
  * Not called `chmod` as it shadows and could be confused with

src/libutil/serialise.cc

@@ -5,8 +5,6 @@
 #include <cerrno>
 #include <memory>
 
-#include <boost/coroutine2/coroutine.hpp>
-
 
 namespace nix {
 
@@ -149,177 +147,6 @@ size_t StringSource::read(char * data, size_t len)
 }
 
 
-#if BOOST_VERSION >= 106300 && BOOST_VERSION < 106600
-#error Coroutines are broken in this version of Boost!
-#endif
-
-/* A concrete datatype allow virtual dispatch of stack allocation methods. */
-struct VirtualStackAllocator {
-    StackAllocator *allocator = StackAllocator::defaultAllocator;
-
-    boost::context::stack_context allocate() {
-        return allocator->allocate();
-    }
-
-    void deallocate(boost::context::stack_context sctx) {
-        allocator->deallocate(sctx);
-    }
-};
-
-
-/* This class reifies the default boost coroutine stack allocation strategy with
-   a virtual interface. */
-class DefaultStackAllocator : public StackAllocator {
-    boost::coroutines2::default_stack stack;
-
-    boost::context::stack_context allocate() {
-        return stack.allocate();
-    }
-
-    void deallocate(boost::context::stack_context sctx) {
-        stack.deallocate(sctx);
-    }
-};
-
-static DefaultStackAllocator defaultAllocatorSingleton;
-
-StackAllocator *StackAllocator::defaultAllocator = &defaultAllocatorSingleton;
-
-
-std::shared_ptr<void> (*create_coro_gc_hook)() = []() -> std::shared_ptr<void> {
-    return {};
-};
-
-/* This class is used for entry and exit hooks on coroutines */
-class CoroutineContext {
-    /* Disable GC when entering the coroutine without the boehm patch,
-     * since it doesn't find the main thread stack in this case.
-     * std::shared_ptr<void> performs type-erasure, so it will call the right
-     * deleter. */
-    const std::shared_ptr<void> coro_gc_hook = create_coro_gc_hook();
-public:
-    CoroutineContext() {};
-    ~CoroutineContext() {};
-};
-
-std::unique_ptr<FinishSink> sourceToSink(std::function<void(Source &)> fun)
-{
-    struct SourceToSink : FinishSink
-    {
-        typedef boost::coroutines2::coroutine<bool> coro_t;
-
-        std::function<void(Source &)> fun;
-        std::optional<coro_t::push_type> coro;
-
-        SourceToSink(std::function<void(Source &)> fun) : fun(fun)
-        {
-        }
-
-        std::string_view cur;
-
-        void operator () (std::string_view in) override
-        {
-            if (in.empty()) return;
-            cur = in;
-
-            if (!coro) {
-                CoroutineContext ctx;
-                coro = coro_t::push_type(VirtualStackAllocator{}, [&](coro_t::pull_type & yield) {
-                    LambdaSource source([&](char *out, size_t out_len) {
-                        if (cur.empty()) {
-                            yield();
-                            if (yield.get()) {
-                                throw EndOfFile("coroutine exhausted");
-                            }
-                        }
-
-                        size_t n = std::min(cur.size(), out_len);
-                        memcpy(out, cur.data(), n);
-                        cur.remove_prefix(n);
-                        return n;
-                    });
-                    fun(source);
-                });
-            }
-
-            if (!*coro) { abort(); }
-
-            if (!cur.empty()) {
-                CoroutineContext ctx;
-                (*coro)(false);
-            }
-        }
-
-        void finish() override
-        {
-            if (!coro) return;
-            if (!*coro) abort();
-            {
-                CoroutineContext ctx;
-                (*coro)(true);
-            }
-            if (*coro) abort();
-        }
-    };
-
-    return std::make_unique<SourceToSink>(fun);
-}
-
-
-std::unique_ptr<Source> sinkToSource(std::function<void(Sink &)> fun)
-{
-    struct SinkToSource : Source
-    {
-        typedef boost::coroutines2::coroutine<std::string> coro_t;
-
-        std::function<void(Sink &)> fun;
-        std::optional<coro_t::pull_type> coro;
-
-        SinkToSource(std::function<void(Sink &)> fun)
-            : fun(fun)
-        {
-        }
-
-        std::string cur;
-        size_t pos = 0;
-
-        size_t read(char * data, size_t len) override
-        {
-            if (!coro) {
-                CoroutineContext ctx;
-                coro = coro_t::pull_type(VirtualStackAllocator{}, [&](coro_t::push_type & yield) {
-                    LambdaSink sink([&](std::string_view data) {
-                        if (!data.empty()) yield(std::string(data));
-                    });
-                    fun(sink);
-                });
-            }
-
-            if (!*coro) {
-                throw EndOfFile("coroutine has finished");
-            }
-
-            if (pos == cur.size()) {
-                if (!cur.empty()) {
-                    CoroutineContext ctx;
-                    (*coro)();
-                }
-                cur = coro->get();
-                pos = 0;
-            }
-
-            auto n = std::min(cur.size() - pos, len);
-            memcpy(data, cur.data() + pos, n);
-            pos += n;
-
-            return n;
-        }
-    };
-
-    return std::make_unique<SinkToSource>(fun);
-}
-
-
 void writePadding(size_t len, Sink & sink)
 {
     if (len % 8) {

src/libutil/serialise.hh

@@ -362,14 +362,6 @@ private:
     Bytes buf{};
 };
 
-std::unique_ptr<FinishSink> sourceToSink(std::function<void(Source &)> fun);
-
-/**
- * Convert a function that feeds data into a Sink into a Source. The
- * Source executes the function as a coroutine.
- */
-std::unique_ptr<Source> sinkToSource(std::function<void(Sink &)> fun);
-
 inline Sink & operator<<(Sink & sink, Generator<Bytes> && g)
 {
     while (auto buffer = g.next()) {

src/nix-store/nix-store.cc

@@ -763,7 +763,7 @@ static void opVerifyPath(Strings opFlags, Strings opArgs)
         printMsg(lvlTalkative, "checking path '%s'...", store->printStorePath(path));
         auto info = store->queryPathInfo(path);
         HashSink sink(info->narHash.type);
-        store->narFromPath(path, sink);
+        sink << store->narFromPath(path);
         auto current = sink.finish();
         if (current.first != info->narHash) {
             printError("path '%s' was modified! expected hash '%s', got '%s'",
@@ -824,7 +824,6 @@ static void opServe(Strings opFlags, Strings opArgs)
         .version = clientVersion,
     };
     ServeProto::WriteConn wconn {
-        .to = out,
         .version = clientVersion,
     };
 
@@ -880,7 +879,8 @@ static void opServe(Strings opFlags, Strings opArgs)
                     store->substitutePaths(paths);
                 }
 
-                ServeProto::write(*store, wconn, store->queryValidPaths(paths));
+                auto valid = store->queryValidPaths(paths);
+                out << ServeProto::write(*store, wconn, valid);
                 break;
             }
 
@@ -891,7 +891,7 @@ static void opServe(Strings opFlags, Strings opArgs)
                     try {
                         auto info = store->queryPathInfo(i);
                         out << store->printStorePath(info->path);
-                        ServeProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
+                        out << ServeProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
                     } catch (InvalidPath &) {
                     }
                 }
@@ -900,7 +900,7 @@ static void opServe(Strings opFlags, Strings opArgs)
             }
 
             case ServeProto::Command::DumpStorePath:
-                store->narFromPath(store->parseStorePath(readString(in)), out);
+                out << store->narFromPath(store->parseStorePath(readString(in)));
                 break;
 
             case ServeProto::Command::ImportPaths: {
@@ -950,7 +950,7 @@ static void opServe(Strings opFlags, Strings opArgs)
                 MonitorFdHup monitor(in.fd);
                 auto status = store->buildDerivation(drvPath, drv);
 
-                ServeProto::write(*store, wconn, status);
+                out << ServeProto::write(*store, wconn, status);
                 break;
             }
 
@@ -959,7 +959,7 @@ static void opServe(Strings opFlags, Strings opArgs)
                 StorePathSet closure;
                 store->computeFSClosure(ServeProto::Serialise<StorePathSet>::read(*store, rconn),
                     closure, false, includeOutputs);
-                ServeProto::write(*store, wconn, closure);
+                out << ServeProto::write(*store, wconn, closure);
                 break;
             }
 

src/nix/dump-path.cc

@@ -22,7 +22,7 @@ struct CmdDumpPath : StorePathCommand
     {
         logger->pause();
         FdSink sink(STDOUT_FILENO);
-        store->narFromPath(storePath, sink);
+        sink << store->narFromPath(storePath);
         sink.flush();
     }
 };

src/nix/upgrade-nix.cc

@@ -97,12 +97,19 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand
             store->ensurePath(storePath);
         }
 
+        // {profileDir}/bin/nix-env is a symlink to {profileDir}/bin/nix, which *then*
+        // is a symlink to /nix/store/meow-nix/bin/nix.
+        // We want /nix/store/meow-nix/bin/nix-env.
+        Path const oldNixInStore = realPath(canonProfileDir + "/bin/nix");
+        Path const oldNixEnv = dirOf(oldNixInStore) + "/nix-env";
+
+        Path const newNixEnv = store->printStorePath(storePath) + "/bin/nix-env";
+
         {
             Activity act(*logger, lvlInfo, actUnknown, fmt("verifying that '%s' works...", store->printStorePath(storePath)));
-            auto program = store->printStorePath(storePath) + "/bin/nix-env";
-            auto s = runProgram(program, false, {"--version"});
+            auto s = runProgram(newNixEnv, false, {"--version"});
             if (s.find("Nix") == std::string::npos)
-                throw Error("could not verify that '%s' works", program);
+                throw Error("could not verify that '%s' works", newNixEnv);
         }
 
         logger->pause();
@@ -110,23 +117,17 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand
         auto const fullStorePath = store->printStorePath(storePath);
 
         if (pathExists(canonProfileDir + "/manifest.nix")) {
-
-            // {settings.nixBinDir}/nix-env is a symlink to a {settings.nixBinDir}/nix, which *then*
-            // is a symlink to /nix/store/meow-nix/bin/nix. We want /nix/store/meow-nix/bin/nix-env.
-            Path const nixInStore = canonPath(settings.nixBinDir + "/nix-env", true);
-            Path const nixEnvCmd = dirOf(nixInStore) + "/nix-env";
-
-            // First remove the existing Nix, then use that Nix by absolute path to
+            // First remove the existing Nix, then use the *new* Nix by absolute path to
             // install the new one, in case the new and old versions aren't considered
             // to be "the same package" by nix-env's logic (e.g., if their pnames differ).
             Strings removeArgs = {
                 "--uninstall",
-                nixEnvCmd,
+                oldNixEnv,
                 "--profile",
                 this->profileDir,
             };
-            printTalkative("running %s %s", nixEnvCmd, concatStringsSep(" ", removeArgs));
-            runProgram(nixEnvCmd, false, removeArgs);
+            printTalkative("running %s %s", newNixEnv, concatStringsSep(" ", removeArgs));
+            runProgram(newNixEnv, false, removeArgs);
 
             Strings upgradeArgs = {
                 "--profile",
@@ -136,8 +137,8 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand
                 "--no-sandbox",
             };
 
-            printTalkative("running %s %s", nixEnvCmd, concatStringsSep(" ", upgradeArgs));
-            runProgram(nixEnvCmd, false, upgradeArgs);
+            printTalkative("running %s %s", newNixEnv, concatStringsSep(" ", upgradeArgs));
+            runProgram(newNixEnv, false, upgradeArgs);
         } else if (pathExists(canonProfileDir + "/manifest.json")) {
             this->upgradeNewStyleProfile(store, storePath);
         } else {

src/nix/verify.cc

@@ -100,7 +100,7 @@ struct CmdVerify : StorePathsCommand
 
                     auto hashSink = HashSink(info->narHash.type);
 
-                    store->narFromPath(info->path, hashSink);
+                    hashSink << store->narFromPath(info->path);
 
                     auto hash = hashSink.finish();
 

tests/functional/repl.sh

@@ -62,21 +62,31 @@ stripColors () {
 testReplResponseGeneral () {
     local grepMode="$1"; shift
     local commands="$1"; shift
-    local expectedResponse="$1"; shift
-    local response="$(nix repl "$@" <<< "$commands" | stripColors)"
-    echo "$response" | grepQuiet "$grepMode" -s "$expectedResponse" \
-      || fail "repl command set:
+    # Expected response can contain newlines.
+    # grep can't handle multiline patterns, so replace newlines with TEST_NEWLINE
+    # in both expectedResponse and response.
+    # awk ORS always adds a trailing record separator, so we strip it with sed.
+    local expectedResponse="$(printf '%s' "$1" | awk 1 ORS=TEST_NEWLINE | sed 's/TEST_NEWLINE$//')"; shift
+    # We don't need to strip trailing record separator here, since extra data is ok.
+    local response="$(nix repl "$@" <<< "$commands" 2>&1 | stripColors | awk 1 ORS=TEST_NEWLINE)"
+    printf '%s' "$response" | grepQuiet "$grepMode" -s "$expectedResponse" \
+      || fail "$(echo "repl command set:
 
 $commands
 
 does not respond with:
 
+---
 $expectedResponse
+---
 
 but with:
 
+---
 $response
-"
+---
+
+" | sed 's/TEST_NEWLINE/\n/g')"
 }
 
 testReplResponse () {
@@ -179,7 +189,7 @@ testReplResponseNoRegex '
 let x = { y = { a = 1; }; inherit x; }; in x
 ' \
 '{
-  x = { ... };
+  x = «repeated»;
   y = { ... };
 }
 '
@@ -231,6 +241,6 @@ testReplResponseNoRegex '
 ' \
 '{
   x = «repeated»;
-  y = { a = 1 };
+  y = { a = 1; };
 }
 '

tests/nixos/default.nix

@@ -155,4 +155,6 @@ in
   broken-userns = runNixOSTestFor "x86_64-linux" ./broken-userns.nix;
 
   coredumps = runNixOSTestFor "x86_64-linux" ./coredumps;
+
+  io_uring = runNixOSTestFor "x86_64-linux" ./io_uring;
 }

tests/nixos/io_uring/default.nix

@@ -0,0 +1,7 @@
+let
+  inherit (import ../util.nix) mkNixBuildTest;
+in
+mkNixBuildTest {
+  name = "io_uring";
+  expressionFile = ./package.nix;
+}

tests/nixos/io_uring/package.nix

@@ -0,0 +1,19 @@
+{ runCommandCC }:
+runCommandCC "io_uring-is-blocked" { } ''
+  cat > test.c <<EOF
+  #include <errno.h>
+  #include <sys/syscall.h>
+  #include <unistd.h>
+
+  int main() {
+      int res = syscall(SYS_io_uring_setup, 0, NULL);
+      return res == -1 && errno == ENOSYS ? 0 : 1;
+  }
+  EOF
+  "$CC" -o test test.c
+  if ! ./test; then
+    echo "Oh no! io_uring is available!"
+    exit 1
+  fi
+  touch "$out"
+''

tests/nixos/nix-upgrade-nix.nix

@@ -35,9 +35,8 @@ in {
 
     machine.succeed("nix --version >&2")
 
-    # Install Lix into the default profile, overriding /run/current-system/sw/bin/nix,
-    # and thus making Lix think we're not on NixOS.
-    machine.succeed("nix-env --install '${lib.getBin lix}' --profile /nix/var/nix/profiles/default >&2")
+    # Use Lix to install CppNix into the default profile, overriding /run/current-system/sw/bin/nix
+    machine.succeed("nix-env --install '${lib.getBin newNix}' --profile /nix/var/nix/profiles/default")
 
     # Make sure that correctly got inserted into our PATH.
     default_profile_nix_path = machine.succeed("command -v nix")
@@ -45,16 +44,15 @@ in {
     assert default_profile_nix_path.strip() == "/nix/var/nix/profiles/default/bin/nix", \
       f"{default_profile_nix_path.strip()=} != /nix/var/nix/profiles/default/bin/nix"
 
-    # And that it's the Nix we specified.
+    # And that it's the Nix we specified
     default_profile_version = machine.succeed("nix --version")
-    assert "${lixVersion}" in default_profile_version, f"${lixVersion} not in {default_profile_version}"
+    assert "${newNixVersion}" in default_profile_version, f"${newNixVersion} not in {default_profile_version}"
 
-    # Upgrade to a different version of Nix, and make sure that also worked.
-
-    machine.succeed("nix upgrade-nix --store-path ${newNix} >&2")
+    # Now upgrade to Lix, and make sure that worked.
+    machine.succeed("${lib.getExe lix} upgrade-nix --debug --store-path ${lix} 2>&1")
     default_profile_version = machine.succeed("nix --version")
     print(default_profile_version)
-    assert "${newNixVersion}" in default_profile_version, f"${newNixVersion} not in {default_profile_version}"
+    assert "${lixVersion}" in default_profile_version, f"${lixVersion} not in {default_profile_version}"
 
     # Now 'break' this profile -- use nix profile on it so nix-env will no longer work on it.
     machine.succeed(

tests/unit/libstore/common-protocol.cc

@@ -50,9 +50,9 @@ public:
         auto file = goldenMaster(testStem);
 
         StringSink to;
-        CommonProto::write(
+        to << CommonProto::write(
             *store,
-            CommonProto::WriteConn { .to = to },
+            CommonProto::WriteConn {},
             value);
 
         if (testAccept())

tests/unit/libstore/protocol.hh

@@ -56,9 +56,9 @@ public:
         auto file = ProtoTest<Proto, protocolDir>::goldenMaster(testStem);
 
         StringSink to;
-        Proto::write(
+        to << Proto::write(
             *LibStoreTest::store,
-            typename Proto::WriteConn {to, version},
+            typename Proto::WriteConn {version},
             value);
 
         if (testAccept())