libstore: return sources from Store::narFromPath

this will make it easier to return async streams instead of sources at
some point in the future. the primary benefactors of the current state
are not greatly inconvenienced by the api change, and would need to be
changed much as they are now once async streams come around either way

Change-Id: I4db9ea8b186f358c239f7863ac8140c500986c2d

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

lix/libstore/binary-cache-store.cc

@@ -317,13 +317,13 @@ std::optional<StorePath> BinaryCacheStore::queryPathFromHashPart(const std::stri
     }
 }
 
-WireFormatGenerator BinaryCacheStore::narFromPath(const StorePath & storePath)
+box_ptr<Source> BinaryCacheStore::narFromPath(const StorePath & storePath)
 {
     auto info = queryPathInfo(storePath).cast<const NarInfo>();
 
     try {
         auto file = getFile(info->url);
-        return [](auto info, auto file, auto & stats) -> WireFormatGenerator {
+        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;
@@ -340,7 +340,7 @@ WireFormatGenerator BinaryCacheStore::narFromPath(const StorePath & storePath)
             stats.narRead++;
             //stats.narReadCompressedBytes += nar->size(); // FIXME
             stats.narReadBytes += total;
-        }(std::move(info), std::move(file), stats);
+        }(std::move(info), std::move(file), stats));
     } catch (NoSuchBinaryCacheFile & e) {
         throw SubstituteGone(std::move(e.info()));
     }

lix/libstore/binary-cache-store.hh

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

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

@@ -1076,7 +1076,7 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
         return path;
     }
 
-    WireFormatGenerator narFromPath(const StorePath & path) override
+    box_ptr<Source> narFromPath(const StorePath & path) override
     {
         if (!goal.isAllowed(path))
             throw InvalidPath("cannot dump unknown path '%s' in recursive Nix", printStorePath(path));

lix/libstore/dummy-store.cc

@@ -65,7 +65,7 @@ struct DummyStore final : public Store
         RepairFlag repair) override
     { unsupported("addTextToStore"); }
 
-    WireFormatGenerator narFromPath(const StorePath & path) override
+    box_ptr<Source> narFromPath(const StorePath & path) override
     { unsupported("narFromPath"); }
 
     std::shared_ptr<const Realisation> queryRealisationUncached(const DrvOutput &) override

lix/libstore/export-import.cc

@@ -28,7 +28,7 @@ void Store::exportPath(const StorePath & path, Sink & sink)
     HashSink hashSink(HashType::SHA256);
     TeeSink teeSink(sink, hashSink);
 
-    teeSink << narFromPath(path);
+    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

@@ -241,15 +241,15 @@ struct LegacySSHStore final : public Store
             throw Error("failed to add path '%s' to remote host '%s'", printStorePath(info.path), host);
     }
 
-    WireFormatGenerator narFromPath(const StorePath & path) override
+    box_ptr<Source> narFromPath(const StorePath & path) override
     {
         auto conn(connections->get());
 
         conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
         conn->to.flush();
-        return [] (auto conn) -> WireFormatGenerator {
+        return make_box_ptr<GeneratorSource>([] (auto conn) -> WireFormatGenerator {
             co_yield copyNAR(conn->from);
-        }(std::move(conn));
+        }(std::move(conn)));
     }
 
     std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override

lix/libstore/local-fs-store.cc

@@ -73,11 +73,11 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
             std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
 }
 
-WireFormatGenerator LocalFSStore::narFromPath(const StorePath & path)
+box_ptr<Source> LocalFSStore::narFromPath(const StorePath & path)
 {
     if (!isValidPath(path))
         throw Error("path '%s' does not exist in store", printStorePath(path));
-    return dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size()));
+    return make_box_ptr<GeneratorSource>(dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size())));
 }
 
 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;
 
-    WireFormatGenerator narFromPath(const StorePath & path) override;
+    box_ptr<Source> narFromPath(const StorePath & path) override;
     ref<FSAccessor> getFSAccessor() override;
 
     /**

lix/libstore/make-content-addressed.cc

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

lix/libstore/remote-fs-accessor.cc

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

lix/libstore/remote-store.cc

@@ -861,14 +861,14 @@ RemoteStore::Connection::~Connection()
     }
 }
 
-WireFormatGenerator RemoteStore::narFromPath(const StorePath & path)
+box_ptr<Source> RemoteStore::narFromPath(const StorePath & path)
 {
     auto conn(connections->get());
     conn->to << WorkerProto::Op::NarFromPath << printStorePath(path);
     conn->processStderr();
-    return [](auto conn) -> WireFormatGenerator {
+    return make_box_ptr<GeneratorSource>([](auto conn) -> WireFormatGenerator {
         co_yield copyNAR(conn->from);
-    }(std::move(conn));
+    }(std::move(conn)));
 }
 
 ref<FSAccessor> RemoteStore::getFSAccessor()

lix/libstore/remote-store.hh

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

lix/libstore/store-api.cc

@@ -1061,6 +1061,10 @@ static std::string makeCopyPathMessage(
 }
 
 
+// buffer size for path copy progress reporting. should be large enough to not cause excessive
+// overhead during copies, but small enough to provide reasonably quick copy progress updates.
+static constexpr unsigned PATH_COPY_BUFSIZE = 65536;
+
 void copyStorePath(
     Store & srcStore,
     Store & dstStore,
@@ -1103,11 +1107,17 @@ void copyStorePath(
     GeneratorSource source{
         [](auto & act, auto & info, auto & srcStore, auto & storePath) -> WireFormatGenerator {
             auto nar = srcStore.narFromPath(storePath);
+            auto buf = std::make_unique<char[]>(PATH_COPY_BUFSIZE);
             uint64_t total = 0;
-            while (auto data = nar.next()) {
-                total += data->size();
-                act.progress(total, info->narSize);
-                co_yield *data;
+            while (true) {
+                try {
+                    auto got = nar->read(buf.get(), PATH_COPY_BUFSIZE);
+                    total += got;
+                    act.progress(total, info->narSize);
+                    co_yield std::span{buf.get(), got};
+                } catch (EndOfFile &) {
+                    break;
+                }
             }
         }(act, info, srcStore, storePath)
     };
@@ -1239,11 +1249,17 @@ std::map<StorePath, StorePath> copyPaths(
             PushActivity pact(act.id);
 
             auto nar = srcStore.narFromPath(missingPath);
+            auto buf = std::make_unique<char[]>(PATH_COPY_BUFSIZE);
             uint64_t total = 0;
-            while (auto data = nar.next()) {
-                total += data->size();
-                act.progress(total, info->narSize);
-                co_yield *data;
+            while (true) {
+                try {
+                    auto got = nar->read(buf.get(), PATH_COPY_BUFSIZE);
+                    total += got;
+                    act.progress(total, info->narSize);
+                    co_yield std::span{buf.get(), got};
+                } catch (EndOfFile &) {
+                    break;
+                }
             }
         };
         pathsToCopy.push_back(std::pair{

lix/libstore/store-api.hh

@@ -580,7 +580,7 @@ public:
     /**
      * Generate a NAR dump of a store path.
      */
-    virtual WireFormatGenerator narFromPath(const StorePath & path) = 0;
+    virtual 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(); }
 
-    WireFormatGenerator narFromPath(const StorePath & path) override
+    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);
-        sink << store->narFromPath(storePath);
+        store->narFromPath(storePath)->drainInto(sink);
         sink.flush();
     }
 };

lix/nix/verify.cc

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