libstore: asyncify Store::addMultipleToStore source set

this will let us asyncify narFromPath, in some distant future.

Change-Id: I722718e434ee105d74aa14688829cca6c29ac442

lix/libstore/remote-store.cc

@@ -524,19 +524,25 @@ try {
             << WorkerProto::Op::AddMultipleToStore
             << repair
             << !checkSigs;
-        conn.withFramedSink([&](Sink & sink) {
-            sink << pathsToCopy.size();
-            for (auto & [pathInfo, pathSource] : pathsToCopy) {
-                sink << WorkerProto::Serialise<ValidPathInfo>::write(*this,
-                    WorkerProto::WriteConn {remoteVersion},
-                    pathInfo);
-                pathSource()->drainInto(sink);
+        // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+        TRY_AWAIT(conn.withFramedSinkAsync([&](Sink & sink) -> kj::Promise<Result<void>> {
+            try {
+                sink << pathsToCopy.size();
+                for (auto & [pathInfo, pathSource] : pathsToCopy) {
+                    sink << WorkerProto::Serialise<ValidPathInfo>::write(*this,
+                        WorkerProto::WriteConn {remoteVersion},
+                        pathInfo);
+                    TRY_AWAIT(pathSource())->drainInto(sink);
+                }
+                co_return result::success();
+            } catch (...) {
+                co_return result::current_exception();
             }
-        });
+        }));
     } else {
         for (auto & [pathInfo, pathSource] : pathsToCopy) {
             pathInfo.ultimate = false; // duplicated in daemon.cc AddMultipleToStore
-            TRY_AWAIT(addToStore(pathInfo, *pathSource(), repair, checkSigs));
+            TRY_AWAIT(addToStore(pathInfo, *TRY_AWAIT(pathSource()), repair, checkSigs));
         }
     }
     co_return result::success();

lix/libstore/store-api.cc

@@ -19,6 +19,7 @@
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libutil/users.hh"
 
+#include <kj/async.h>
 #include <mutex>
 #include <nlohmann/json.hpp>
 #include <regex>
@@ -349,7 +350,7 @@ try {
                 MaintainCount<decltype(nrRunning)> mc(nrRunning);
                 showProgress();
                 try {
-                    aio.blockOn(addToStore(info, *source(), repair, checkSigs));
+                    aio.blockOn(addToStore(info, *aio.blockOn(source()), repair, checkSigs));
                 } catch (Error & e) {
                     nrFailed++;
                     if (!settings.keepGoing)
@@ -1277,21 +1278,28 @@ try {
             }
         };
 
-        auto source = [&srcStore, &dstStore, missingPath, info] {
-            // 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);
+        auto source = [](auto & srcStore, auto & dstStore, auto missingPath, auto info
+                      ) -> kj::Promise<Result<box_ptr<Source>>> {
+            try {
+                // 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);
 
-            return make_box_ptr<SinglePathSource>(
-                makeCopyPathMessage(srcUri, dstUri, storePathS),
-                Logger::Fields{storePathS, srcUri, dstUri},
-                info->narSize,
-                srcStore.narFromPath(missingPath)
-            );
+                co_return make_box_ptr<SinglePathSource>(
+                    makeCopyPathMessage(srcUri, dstUri, storePathS),
+                    Logger::Fields{storePathS, srcUri, dstUri},
+                    info->narSize,
+                    srcStore.narFromPath(missingPath)
+                );
+            } catch (...) {
+                co_return result::current_exception();
+            }
         };
-        pathsToCopy.push_back(std::pair{infoForDst, std::move(source)});
+        pathsToCopy.push_back(std::pair{
+            infoForDst, std::bind(source, std::ref(srcStore), std::ref(dstStore), missingPath, info)
+        });
     }
 
     TRY_AWAIT(dstStore.addMultipleToStore(pathsToCopy, act, repair, checkSigs));

lix/libstore/store-api.hh

@@ -506,7 +506,8 @@ public:
      * A list of paths infos along with a source providing the content
      * of the associated store path
      */
-    using PathsSource = std::vector<std::pair<ValidPathInfo, std::function<box_ptr<Source>()>>>;
+    using PathsSource = std::vector<
+        std::pair<ValidPathInfo, std::function<kj::Promise<Result<box_ptr<Source>>>()>>>;
 
     /**
      * Import multiple paths into the store.