libstore: asyncify Store::narFromPath

Change-Id: Ia231b83d58adcf52fe1ba107a0eefea8111528df

lix/legacy/nix-store.cc

@@ -791,7 +791,7 @@ static void opVerifyPath(AsyncIoRoot & aio, 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)->drainInto(sink);
+        aio.blockOn(store->narFromPath(path))->drainInto(sink);
         auto current = sink.finish();
         if (current.first != info->narHash) {
             printError("path '%s' was modified! expected hash '%s', got '%s'",
@@ -928,7 +928,8 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
             }
 
             case ServeProto::Command::DumpStorePath:
-                store->narFromPath(store->parseStorePath(readString(in)))->drainInto(out);
+                aio.blockOn(store->narFromPath(store->parseStorePath(readString(in))))
+                    ->drainInto(out);
                 break;
 
             case ServeProto::Command::ImportPaths: {

lix/libstore/binary-cache-store.cc

@@ -350,33 +350,37 @@ try {
     co_return result::current_exception();
 }
 
-box_ptr<Source> BinaryCacheStore::narFromPath(const StorePath & storePath)
-{
+kj::Promise<Result<box_ptr<Source>>> BinaryCacheStore::narFromPath(const StorePath & storePath)
+try {
     auto info = queryPathInfo(storePath).cast<const NarInfo>();
 
     try {
         auto file = getFile(info->url);
-        return make_box_ptr<GeneratorSource>([](auto info, auto file, auto & stats) -> WireFormatGenerator {
-            constexpr size_t buflen = 65536;
-            auto buf = std::make_unique<char []>(buflen);
-            size_t total = 0;
-            auto decompressor = makeDecompressionSource(info->compression, *file);
-            try {
-                while (true) {
-                    const auto len = decompressor->read(buf.get(), buflen);
-                    co_yield std::span{buf.get(), len};
-                    total += len;
+        co_return make_box_ptr<GeneratorSource>(
+            [](auto info, auto file, auto & stats) -> WireFormatGenerator {
+                constexpr size_t buflen = 65536;
+                auto buf = std::make_unique<char[]>(buflen);
+                size_t total = 0;
+                auto decompressor = makeDecompressionSource(info->compression, *file);
+                try {
+                    while (true) {
+                        const auto len = decompressor->read(buf.get(), buflen);
+                        co_yield std::span{buf.get(), len};
+                        total += len;
+                    }
+                } catch (EndOfFile &) {
                 }
-            } catch (EndOfFile &) {
-            }
 
-            stats.narRead++;
-            //stats.narReadCompressedBytes += nar->size(); // FIXME
-            stats.narReadBytes += total;
-        }(std::move(info), std::move(file), stats));
+                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()));
     }
+} catch (...) {
+    co_return result::current_exception();
 }
 
 std::shared_ptr<const ValidPathInfo> BinaryCacheStore::queryPathInfoUncached(const StorePath & storePath)

lix/libstore/binary-cache-store.hh

@@ -150,7 +150,7 @@ public:
     kj::Promise<Result<std::shared_ptr<const Realisation>>>
     queryRealisationUncached(const DrvOutput &) override;
 
-    box_ptr<Source> narFromPath(const StorePath & path) override;
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
 
     ref<FSAccessor> getFSAccessor() override;
 

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

@@ -1142,11 +1142,13 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
         co_return result::current_exception();
     }
 
-    box_ptr<Source> narFromPath(const StorePath & path) override
-    {
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    try {
         if (!goal.isAllowed(path))
             throw InvalidPath("cannot dump unknown path '%s' in recursive Nix", printStorePath(path));
-        return LocalFSStore::narFromPath(path);
+        co_return TRY_AWAIT(LocalFSStore::narFromPath(path));
+    } catch (...) {
+        co_return result::current_exception();
     }
 
     kj::Promise<Result<void>> ensurePath(const StorePath & path) override

lix/libstore/dummy-store.cc

@@ -70,8 +70,8 @@ struct DummyStore final : public Store
         RepairFlag repair) override
     try { unsupported("addTextToStore"); } catch (...) { return {result::current_exception()}; }
 
-    box_ptr<Source> narFromPath(const StorePath & path) override
-    { unsupported("narFromPath"); }
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    try { unsupported("narFromPath"); } catch (...) { return {result::current_exception()}; }
 
     kj::Promise<Result<std::shared_ptr<const Realisation>>>
     queryRealisationUncached(const DrvOutput &) override

lix/libstore/export-import.cc

@@ -32,7 +32,7 @@ try {
     HashSink hashSink(HashType::SHA256);
     TeeSink teeSink(sink, hashSink);
 
-    narFromPath(path)->drainInto(teeSink);
+    TRY_AWAIT(narFromPath(path))->drainInto(teeSink);
 
     /* Refuse to export paths that have changed.  This prevents
        filesystem corruption from spreading to other machines.

lix/libstore/legacy-ssh-store.cc

@@ -248,15 +248,17 @@ struct LegacySSHStore final : public Store
         co_return result::current_exception();
     }
 
-    box_ptr<Source> narFromPath(const StorePath & path) override
-    {
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    try {
         auto conn(connections->get());
 
         conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
         conn->to.flush();
-        return make_box_ptr<GeneratorSource>([] (auto conn) -> WireFormatGenerator {
+        co_return make_box_ptr<GeneratorSource>([] (auto conn) -> WireFormatGenerator {
             co_yield copyNAR(conn->from);
         }(std::move(conn)));
+    } catch (...) {
+        co_return result::current_exception();
     }
 
     kj::Promise<Result<std::optional<StorePath>>>

lix/libstore/local-fs-store.cc

@@ -84,11 +84,15 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
             std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
 }
 
-box_ptr<Source> LocalFSStore::narFromPath(const StorePath & path)
-{
+kj::Promise<Result<box_ptr<Source>>> LocalFSStore::narFromPath(const StorePath & path)
+try {
     if (!isValidPath(path))
         throw Error("path '%s' does not exist in store", printStorePath(path));
-    return make_box_ptr<GeneratorSource>(dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size())));
+    co_return make_box_ptr<GeneratorSource>(
+        dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size()))
+    );
+} catch (...) {
+    co_return result::current_exception();
 }
 
 const std::string LocalFSStore::drvsLogDir = "drvs";

lix/libstore/local-fs-store.hh

@@ -42,7 +42,7 @@ public:
     LocalFSStoreConfig & config() override = 0;
     const LocalFSStoreConfig & config() const override = 0;
 
-    box_ptr<Source> narFromPath(const StorePath & path) override;
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
     ref<FSAccessor> getFSAccessor() override;
 
     /**

lix/libstore/make-content-addressed.cc

@@ -26,7 +26,7 @@ try {
         std::string oldHashPart(path.hashPart());
 
         StringSink sink;
-        srcStore.narFromPath(path)->drainInto(sink);
+        TRY_AWAIT(srcStore.narFromPath(path))->drainInto(sink);
 
         StringMap rewrites;
 

lix/libstore/remote-fs-accessor.cc

@@ -101,7 +101,7 @@ try {
     }
 
     StringSink sink;
-    store->narFromPath(storePath)->drainInto(sink);
+    TRY_AWAIT(store->narFromPath(storePath))->drainInto(sink);
     co_return {TRY_AWAIT(addToCache(storePath.hashPart(), std::move(sink.s))), restPath};
 } catch (...) {
     co_return result::current_exception();

lix/libstore/remote-store.cc

@@ -999,15 +999,17 @@ RemoteStore::Connection::~Connection()
     }
 }
 
-box_ptr<Source> RemoteStore::narFromPath(const StorePath & path)
-{
+kj::Promise<Result<box_ptr<Source>>> RemoteStore::narFromPath(const StorePath & path)
+try {
     auto conn(connections->get());
     conn->to << WorkerProto::Op::NarFromPath << printStorePath(path);
     conn->processStderr();
-    return make_box_ptr<GeneratorSource>([](auto conn) -> WireFormatGenerator {
+    co_return make_box_ptr<GeneratorSource>([](auto conn) -> WireFormatGenerator {
         co_yield copyNAR(conn->from);
     }(std::move(conn)));
-}
+} catch (...) {
+    co_return result::current_exception();}
+
 
 ref<FSAccessor> RemoteStore::getFSAccessor()
 {

lix/libstore/remote-store.hh

@@ -199,7 +199,7 @@ protected:
 
     virtual ref<FSAccessor> getFSAccessor() override;
 
-    virtual box_ptr<Source> narFromPath(const StorePath & path) override;
+    virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
 
 private:
 

lix/libstore/store-api.cc

@@ -1126,7 +1126,7 @@ try {
     CopyPathStream source{
         act,
         info->narSize,
-        make_box_ptr<AsyncSourceInputStream>(srcStore.narFromPath(storePath))
+        make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(storePath)))
     };
     TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs));
     co_return result::success();
@@ -1275,7 +1275,7 @@ try {
                     makeCopyPathMessage(srcUri, dstUri, storePathS),
                     Logger::Fields{storePathS, srcUri, dstUri},
                     info->narSize,
-                    make_box_ptr<AsyncSourceInputStream>(srcStore.narFromPath(missingPath))
+                    make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(missingPath)))
                 );
             } catch (...) {
                 co_return result::current_exception();

lix/libstore/store-api.hh

@@ -606,7 +606,7 @@ public:
     /**
      * Generate a NAR dump of a store path.
      */
-    virtual box_ptr<Source> narFromPath(const StorePath & path) = 0;
+    virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) = 0;
 
     /**
      * For each path, if it's a derivation, build it.  Building a

lix/libstore/uds-remote-store.hh

@@ -42,7 +42,7 @@ public:
     ref<FSAccessor> getFSAccessor() override
     { return LocalFSStore::getFSAccessor(); }
 
-    box_ptr<Source> narFromPath(const StorePath & path) override
+    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
     { return LocalFSStore::narFromPath(path); }
 
     /**

lix/nix/dump-path.cc

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

lix/nix/verify.cc

@@ -1,11 +1,13 @@
 #include "lix/libcmd/command.hh"
 #include "lix/libmain/shared.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/thread-pool.hh"
 #include "lix/libutil/signals.hh"
 #include "lix/libutil/exit.hh"
 
 #include <atomic>
+#include <functional>
 
 using namespace nix;
 
@@ -81,7 +83,7 @@ struct CmdVerify : StorePathsCommand
 
         ThreadPool pool{"Verify pool"};
 
-        auto doPath = [&](const StorePath & storePath) {
+        auto doPath = [&](AsyncIoRoot & aio, const StorePath & storePath) {
             try {
                 checkInterrupt();
 
@@ -99,7 +101,7 @@ struct CmdVerify : StorePathsCommand
 
                     auto hashSink = HashSink(info->narHash.type);
 
-                    store->narFromPath(info->path)->drainInto(hashSink);
+                    aio.blockOn(store->narFromPath(info->path))->drainInto(hashSink);
 
                     auto hash = hashSink.finish();
 
@@ -173,7 +175,7 @@ struct CmdVerify : StorePathsCommand
         };
 
         for (auto & storePath : storePaths)
-            pool.enqueue(std::bind(doPath, storePath));
+            pool.enqueueWithAio(std::bind(doPath, std::placeholders::_1, storePath));
 
         pool.process();