diff --git a/lix/legacy/nix-store.cc b/lix/legacy/nix-store.cc
index 3df50684e..56fd52e91 100644
--- a/lix/legacy/nix-store.cc
+++ b/lix/legacy/nix-store.cc
@@ -800,7 +800,7 @@ static void opVerifyPath(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
         printMsg(lvlTalkative, "checking path '%s'...", store->printStorePath(path));
         auto info = aio.blockOn(store->queryPathInfo(path));
         HashSink sink(info->narHash.type);
-        aio.blockOn(store->narFromPath(path))->drainInto(sink);
+        aio.blockOn(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'",
@@ -939,8 +939,8 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
             }
 
             case ServeProto::Command::DumpStorePath:
-                aio.blockOn(store->narFromPath(store->parseStorePath(readString(in))))
-                    ->drainInto(out);
+                aio.blockOn(aio.blockOn(store->narFromPath(store->parseStorePath(readString(in))))
+                                ->drainInto(out));
                 break;
 
             case ServeProto::Command::ImportPaths: {
diff --git a/lix/libstore/binary-cache-store.cc b/lix/libstore/binary-cache-store.cc
index 3cab26714..632d13267 100644
--- a/lix/libstore/binary-cache-store.cc
+++ b/lix/libstore/binary-cache-store.cc
@@ -1,6 +1,8 @@
 #include "lix/libutil/archive.hh"
 #include "lix/libstore/binary-cache-store.hh"
 #include "lix/libutil/async-io.hh"
+#include "lix/libutil/async.hh"
+#include "lix/libutil/box_ptr.hh"
 #include "lix/libutil/compression.hh"
 #include "lix/libstore/derivations.hh"
 #include "lix/libstore/fs-accessor.hh"
@@ -362,34 +364,49 @@ try {
     co_return result::current_exception();
 }
 
-kj::Promise<Result<box_ptr<Source>>> BinaryCacheStore::narFromPath(const StorePath & storePath)
+kj::Promise<Result<box_ptr<AsyncInputStream>>>
+BinaryCacheStore::narFromPath(const StorePath & storePath)
 try {
+    struct NarFromPath : AsyncInputStream
+    {
+        Stats<std::atomic> & stats;
+        box_ptr<Source> file;
+        box_ptr<AsyncInputStream> decompressed;
+        uint64_t total;
+
+        NarFromPath(Stats<std::atomic> & stats, const std::string & method, box_ptr<Source> file)
+            : stats(stats)
+            , file(std::move(file))
+            , decompressed(
+                  makeDecompressionStream(method, make_box_ptr<AsyncSourceInputStream>(*this->file))
+              )
+        {
+        }
+
+        kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+        {
+            return decompressed->read(buffer, size).then([&](auto r) {
+                if (r.has_value()) {
+                    if (r.value() > 0) {
+                        total += r.value();
+                    } else {
+                        stats.narRead++;
+                        // stats.narReadCompressedBytes += nar->size(); // FIXME
+                        stats.narReadBytes += total;
+                    }
+                }
+                return r;
+            });
+        }
+    };
+
     auto info_ = TRY_AWAIT(queryPathInfo(storePath)).try_cast<const NarInfo>();
     assert(info_ && "binary cache queryPathInfo didn't return a NarInfo");
     auto & info = *info_;
 
     try {
         auto file = getFile(info->url);
-        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 &) {
-                }
-
-                stats.narRead++;
-                // stats.narReadCompressedBytes += nar->size(); // FIXME
-                stats.narReadBytes += total;
-            }(std::move(info), std::move(file), stats)
-        );
+        co_return make_box_ptr<NarFromPath>(stats, info->compression, std::move(file));
     } catch (NoSuchBinaryCacheFile & e) {
         throw SubstituteGone(std::move(e.info()));
     }
diff --git a/lix/libstore/binary-cache-store.hh b/lix/libstore/binary-cache-store.hh
index db7424434..054e624c1 100644
--- a/lix/libstore/binary-cache-store.hh
+++ b/lix/libstore/binary-cache-store.hh
@@ -146,7 +146,7 @@ public:
         const StorePathSet & references,
         RepairFlag repair) override;
 
-    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
+    kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
 
     ref<FSAccessor> getFSAccessor() override;
 
diff --git a/lix/libstore/dummy-store.cc b/lix/libstore/dummy-store.cc
index 84c56f298..6fd524d12 100644
--- a/lix/libstore/dummy-store.cc
+++ b/lix/libstore/dummy-store.cc
@@ -1,5 +1,6 @@
 #include "lix/libstore/dummy-store.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async-io.hh"
 
 namespace nix {
 
@@ -71,7 +72,7 @@ struct DummyStore final : public Store
         RepairFlag repair) override
     try { unsupported("addTextToStore"); } catch (...) { return {result::current_exception()}; }
 
-    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
     try { unsupported("narFromPath"); } catch (...) { return {result::current_exception()}; }
 
     virtual ref<FSAccessor> getFSAccessor() override
diff --git a/lix/libstore/export-import.cc b/lix/libstore/export-import.cc
index 8b4a144ae..86ae7684a 100644
--- a/lix/libstore/export-import.cc
+++ b/lix/libstore/export-import.cc
@@ -32,7 +32,7 @@ try {
     HashSink hashSink(HashType::SHA256);
     TeeSink teeSink(sink, hashSink);
 
-    TRY_AWAIT(narFromPath(path))->drainInto(teeSink);
+    TRY_AWAIT(TRY_AWAIT(narFromPath(path))->drainInto(teeSink));
 
     /* Refuse to export paths that have changed.  This prevents
        filesystem corruption from spreading to other machines.
diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc
index f9c34c531..b5805b799 100644
--- a/lix/libstore/legacy-ssh-store.cc
+++ b/lix/libstore/legacy-ssh-store.cc
@@ -1,5 +1,6 @@
 #include "lix/libstore/legacy-ssh-store.hh"
 #include "lix/libutil/archive.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/pool.hh"
 #include "lix/libstore/remote-store.hh"
 #include "lix/libstore/serve-protocol.hh"
@@ -255,13 +256,13 @@ struct LegacySSHStore final : public Store
         co_return result::current_exception();
     }
 
-    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
     try {
         auto conn(TRY_AWAIT(connections->get()));
 
         conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
         conn->to.flush();
-        co_return make_box_ptr<GeneratorSource>([] (auto conn) -> WireFormatGenerator {
+        co_return make_box_ptr<AsyncGeneratorInputStream>([](auto conn) -> WireFormatGenerator {
             co_yield copyNAR(conn->from);
         }(std::move(conn)));
     } catch (...) {
diff --git a/lix/libstore/local-fs-store.cc b/lix/libstore/local-fs-store.cc
index fc56fb6fe..54a6e0c8a 100644
--- a/lix/libstore/local-fs-store.cc
+++ b/lix/libstore/local-fs-store.cc
@@ -3,6 +3,7 @@
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/local-fs-store.hh"
 #include "lix/libstore/globals.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/compression.hh"
 #include "lix/libstore/derivations.hh"
 
@@ -84,11 +85,11 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
             std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
 }
 
-kj::Promise<Result<box_ptr<Source>>> LocalFSStore::narFromPath(const StorePath & path)
+kj::Promise<Result<box_ptr<AsyncInputStream>>> LocalFSStore::narFromPath(const StorePath & path)
 try {
     if (!TRY_AWAIT(isValidPath(path)))
         throw Error("path '%s' does not exist in store", printStorePath(path));
-    co_return make_box_ptr<GeneratorSource>(
+    co_return make_box_ptr<AsyncGeneratorInputStream>(
         dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size()))
     );
 } catch (...) {
diff --git a/lix/libstore/local-fs-store.hh b/lix/libstore/local-fs-store.hh
index 7f4668b45..573ca5140 100644
--- a/lix/libstore/local-fs-store.hh
+++ b/lix/libstore/local-fs-store.hh
@@ -4,6 +4,7 @@
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/gc-store.hh"
 #include "lix/libstore/log-store.hh"
+#include "lix/libutil/async-io.hh"
 
 namespace nix {
 
@@ -42,7 +43,7 @@ public:
     LocalFSStoreConfig & config() override = 0;
     const LocalFSStoreConfig & config() const override = 0;
 
-    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
+    kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
     ref<FSAccessor> getFSAccessor() override;
 
     /**
diff --git a/lix/libstore/make-content-addressed.cc b/lix/libstore/make-content-addressed.cc
index ccfa05d88..4c8222646 100644
--- a/lix/libstore/make-content-addressed.cc
+++ b/lix/libstore/make-content-addressed.cc
@@ -26,7 +26,7 @@ try {
         std::string oldHashPart(path.hashPart());
 
         StringSink sink;
-        TRY_AWAIT(srcStore.narFromPath(path))->drainInto(sink);
+        TRY_AWAIT(TRY_AWAIT(srcStore.narFromPath(path))->drainInto(sink));
 
         StringMap rewrites;
 
diff --git a/lix/libstore/remote-fs-accessor.cc b/lix/libstore/remote-fs-accessor.cc
index 6f9a9a160..b00df732e 100644
--- a/lix/libstore/remote-fs-accessor.cc
+++ b/lix/libstore/remote-fs-accessor.cc
@@ -101,7 +101,7 @@ try {
     }
 
     StringSink sink;
-    TRY_AWAIT(store->narFromPath(storePath))->drainInto(sink);
+    TRY_AWAIT(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();
diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc
index 86a32e5fd..c36a3f130 100644
--- a/lix/libstore/remote-store.cc
+++ b/lix/libstore/remote-store.cc
@@ -719,11 +719,11 @@ RemoteStore::Connection::~Connection()
     }
 }
 
-kj::Promise<Result<box_ptr<Source>>> RemoteStore::narFromPath(const StorePath & path)
+kj::Promise<Result<box_ptr<AsyncInputStream>>> RemoteStore::narFromPath(const StorePath & path)
 try {
     auto conn(TRY_AWAIT(getConnection()));
     TRY_AWAIT(conn.sendCommand(WorkerProto::Op::NarFromPath, printStorePath(path)));
-    co_return make_box_ptr<GeneratorSource>([](auto conn) -> WireFormatGenerator {
+    co_return make_box_ptr<AsyncGeneratorInputStream>([](auto conn) -> WireFormatGenerator {
         co_yield copyNAR(conn->from);
     }(std::move(conn)));
 } catch (...) {
diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh
index b937bb340..e76c6929c 100644
--- a/lix/libstore/remote-store.hh
+++ b/lix/libstore/remote-store.hh
@@ -197,7 +197,8 @@ protected:
 
     virtual ref<FSAccessor> getFSAccessor() override;
 
-    virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override;
+    virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path
+    ) override;
 
 private:
 
diff --git a/lix/libstore/store-api.cc b/lix/libstore/store-api.cc
index c59c7e909..491f11f0a 100644
--- a/lix/libstore/store-api.cc
+++ b/lix/libstore/store-api.cc
@@ -1071,11 +1071,7 @@ try {
         info = info2;
     }
 
-    CopyPathStream source{
-        act,
-        info->narSize,
-        make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(storePath)))
-    };
+    CopyPathStream source{act, info->narSize, TRY_AWAIT(srcStore.narFromPath(storePath))};
     TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs));
     co_return result::success();
 } catch (...) {
@@ -1187,7 +1183,7 @@ try {
                     makeCopyPathMessage(srcUri, dstUri, storePathS),
                     Logger::Fields{storePathS, srcUri, dstUri},
                     info->narSize,
-                    make_box_ptr<AsyncSourceInputStream>(TRY_AWAIT(srcStore.narFromPath(missingPath)))
+                    TRY_AWAIT(srcStore.narFromPath(missingPath))
                 );
             } catch (...) {
                 co_return result::current_exception();
diff --git a/lix/libstore/store-api.hh b/lix/libstore/store-api.hh
index 90be090a0..d8df9ecfb 100644
--- a/lix/libstore/store-api.hh
+++ b/lix/libstore/store-api.hh
@@ -575,7 +575,7 @@ public:
     /**
      * Generate a NAR dump of a store path.
      */
-    virtual kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) = 0;
+    virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) = 0;
 
     /**
      * For each path, if it's a derivation, build it.  Building a
diff --git a/lix/libstore/uds-remote-store.hh b/lix/libstore/uds-remote-store.hh
index 1ae3d3e72..733771a5c 100644
--- a/lix/libstore/uds-remote-store.hh
+++ b/lix/libstore/uds-remote-store.hh
@@ -4,6 +4,7 @@
 #include "lix/libstore/remote-store.hh"
 #include "lix/libstore/remote-store-connection.hh"
 #include "lix/libstore/indirect-root-store.hh"
+#include "lix/libutil/async-io.hh"
 
 namespace nix {
 
@@ -42,7 +43,7 @@ public:
     ref<FSAccessor> getFSAccessor() override
     { return LocalFSStore::getFSAccessor(); }
 
-    kj::Promise<Result<box_ptr<Source>>> narFromPath(const StorePath & path) override
+    kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
     { return LocalFSStore::narFromPath(path); }
 
     /**
diff --git a/lix/nix/dump-path.cc b/lix/nix/dump-path.cc
index 879814e71..a659e5188 100644
--- a/lix/nix/dump-path.cc
+++ b/lix/nix/dump-path.cc
@@ -23,7 +23,7 @@ struct CmdDumpPath : StorePathCommand
     {
         logger->pause();
         FdSink sink(STDOUT_FILENO);
-        aio().blockOn(store->narFromPath(storePath))->drainInto(sink);
+        aio().blockOn(aio().blockOn(store->narFromPath(storePath))->drainInto(sink));
         sink.flush();
     }
 };
diff --git a/lix/nix/verify.cc b/lix/nix/verify.cc
index c4ee44f7a..4ebca2582 100644
--- a/lix/nix/verify.cc
+++ b/lix/nix/verify.cc
@@ -102,7 +102,7 @@ struct CmdVerify : StorePathsCommand
 
                     auto hashSink = HashSink(info->narHash.type);
 
-                    aio.blockOn(store->narFromPath(info->path))->drainInto(hashSink);
+                    aio.blockOn(aio.blockOn(store->narFromPath(info->path))->drainInto(hashSink));
 
                     auto hash = hashSink.finish();