diff --git a/lix/libfetchers/path.cc b/lix/libfetchers/path.cc
index c30bb320a..b6ccf723c 100644
--- a/lix/libfetchers/path.cc
+++ b/lix/libfetchers/path.cc
@@ -2,6 +2,7 @@
 #include "lix/libfetchers/builtin-fetchers.hh"
 #include "lix/libstore/store-api.hh"
 #include "lix/libutil/archive.hh"
+#include "lix/libutil/async-io.hh"
 
 namespace nix::fetchers {
 
@@ -132,7 +133,7 @@ struct PathInputScheme : InputScheme
         time_t mtime = 0;
         if (!storePath || storePath->name() != "source" || !store->isValidPath(*storePath)) {
             // FIXME: try to substitute storePath.
-            auto src = GeneratorSource{dumpPathAndGetMtime(absPath, mtime)};
+            auto src = AsyncGeneratorInputStream{dumpPathAndGetMtime(absPath, mtime)};
             storePath = TRY_AWAIT(store->addToStoreFromDump(src, "source"));
         }
         input.attrs.insert_or_assign("lastModified", uint64_t(mtime));
diff --git a/lix/libstore/binary-cache-store.cc b/lix/libstore/binary-cache-store.cc
index 65b4f2802..efdb4eab1 100644
--- a/lix/libstore/binary-cache-store.cc
+++ b/lix/libstore/binary-cache-store.cc
@@ -289,13 +289,18 @@ try {
     co_return result::current_exception();
 }
 
-kj::Promise<Result<StorePath>> BinaryCacheStore::addToStoreFromDump(Source & dump, std::string_view name,
-    FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references)
+kj::Promise<Result<StorePath>> BinaryCacheStore::addToStoreFromDump(
+    AsyncInputStream & dump,
+    std::string_view name,
+    FileIngestionMethod method,
+    HashType hashAlgo,
+    RepairFlag repair,
+    const StorePathSet & references
+)
 try {
     if (method != FileIngestionMethod::Recursive || hashAlgo != HashType::SHA256)
         unsupported("addToStoreFromDump");
-    AsyncSourceInputStream stream{dump};
-    co_return TRY_AWAIT(addToStoreCommon(stream, repair, CheckSigs, [&](HashResult nar) {
+    co_return TRY_AWAIT(addToStoreCommon(dump, repair, CheckSigs, [&](HashResult nar) {
         ValidPathInfo info {
             *this,
             name,
diff --git a/lix/libstore/binary-cache-store.hh b/lix/libstore/binary-cache-store.hh
index 68c20efc2..490106fa5 100644
--- a/lix/libstore/binary-cache-store.hh
+++ b/lix/libstore/binary-cache-store.hh
@@ -118,8 +118,14 @@ public:
     kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & narSource,
         RepairFlag repair, CheckSigsFlag checkSigs) override;
 
-    kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
-        FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references) override;
+    kj::Promise<Result<StorePath>> addToStoreFromDump(
+        AsyncInputStream & dump,
+        std::string_view name,
+        FileIngestionMethod method,
+        HashType hashAlgo,
+        RepairFlag repair,
+        const StorePathSet & references
+    ) override;
 
     kj::Promise<Result<StorePath>> addToStoreRecursive(
         std::string_view name,
diff --git a/lix/libstore/build/local-derivation-goal.cc b/lix/libstore/build/local-derivation-goal.cc
index 3740d8b8a..da1c4b50a 100644
--- a/lix/libstore/build/local-derivation-goal.cc
+++ b/lix/libstore/build/local-derivation-goal.cc
@@ -1,4 +1,5 @@
 #include "lix/libstore/build/local-derivation-goal.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/error.hh"
 #include "lix/libstore/indirect-root-store.hh"
@@ -1119,7 +1120,7 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
     }
 
     kj::Promise<Result<StorePath>> addToStoreFromDump(
-        Source & dump,
+        AsyncInputStream & dump,
         std::string_view name,
         FileIngestionMethod method,
         HashType hashAlgo,
diff --git a/lix/libstore/daemon.cc b/lix/libstore/daemon.cc
index fc1c61255..93ae21696 100644
--- a/lix/libstore/daemon.cc
+++ b/lix/libstore/daemon.cc
@@ -1,4 +1,5 @@
 #include "lix/libstore/daemon.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/monitor-fd.hh"
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libstore/worker-protocol-impl.hh"
@@ -428,7 +429,10 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
                         return store->queryPathInfo(path);
                     },
                     [&](const FileIngestionMethod & fim) {
-                        auto path = aio.blockOn(store->addToStoreFromDump(source, name, fim, hashType, repair, refs));
+                        AsyncSourceInputStream stream{source};
+                        auto path = aio.blockOn(
+                            store->addToStoreFromDump(stream, name, fim, hashType, repair, refs)
+                        );
                         return store->queryPathInfo(path);
                     },
                 }, contentAddressMethod.raw);
@@ -495,7 +499,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
                     co_yield std::move(file->contents);
                 }
             };
-            GeneratorSource dumpSource{g()};
+            AsyncGeneratorInputStream dumpSource{g()};
             logger->startWork();
             auto path = aio.blockOn(store->addToStoreFromDump(dumpSource, baseName, method, hashAlgo));
             logger->stopWork();
diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc
index 083169950..471e0187e 100644
--- a/lix/libstore/local-store.cc
+++ b/lix/libstore/local-store.cc
@@ -6,6 +6,7 @@
 #include "lix/libstore/worker-protocol.hh"
 #include "lix/libstore/derivations.hh"
 #include "lix/libstore/nar-info.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/references.hh"
 #include "lix/libutil/result.hh"
@@ -1287,12 +1288,18 @@ try {
 }
 
 
-kj::Promise<Result<StorePath>> LocalStore::addToStoreFromDump(Source & source0, std::string_view name,
-    FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references)
+kj::Promise<Result<StorePath>> LocalStore::addToStoreFromDump(
+    AsyncInputStream & source0,
+    std::string_view name,
+    FileIngestionMethod method,
+    HashType hashAlgo,
+    RepairFlag repair,
+    const StorePathSet & references
+)
 try {
     /* For computing the store path. */
     auto hashSink = std::make_unique<HashSink>(hashAlgo);
-    TeeSource source { source0, *hashSink };
+    AsyncTeeInputStream source { source0, *hashSink };
 
     /* Read the source path into memory, but only if it's up to
        narBufferSize bytes. If it's larger, write it to a temporary
@@ -1327,9 +1334,8 @@ try {
         Finally cleanup([&]() {
             dump = {dumpBuffer.get(), dump.size() + got};
         });
-        try {
-            got = source.read(dumpBuffer.get() + oldSize, want);
-        } catch (EndOfFile &) {
+        got = TRY_AWAIT(source.read(dumpBuffer.get() + oldSize, want));
+        if (got == 0) {
             inMemory = true;
             break;
         }
@@ -1341,29 +1347,31 @@ try {
     AutoCloseFD tempDirFd;
 
     if (!inMemory) {
-        struct ChainSource : Source
+        struct ChainSource : AsyncInputStream
         {
-            Source & source1, & source2;
+            AsyncInputStream & source1, & source2;
             bool useSecond = false;
-            ChainSource(Source & s1, Source & s2) : source1(s1), source2(s2) {}
+            ChainSource(AsyncInputStream & s1, AsyncInputStream & s2) : source1(s1), source2(s2) {}
 
-            size_t read(char * data, size_t len) override
-            {
+            kj::Promise<Result<size_t>> read(void * data, size_t len) override
+            try {
                 if (useSecond) {
-                    return source2.read(data, len);
+                    co_return TRY_AWAIT(source2.read(data, len));
                 } else {
-                    try {
-                        return source1.read(data, len);
-                    } catch (EndOfFile &) {
+                    if (auto got = TRY_AWAIT(source1.read(data, len))) {
+                        co_return got;
+                    } else {
                         useSecond = true;
-                        return this->read(data, len);
+                        co_return TRY_AWAIT(read(data, len));
                     }
                 }
+            } catch (...) {
+                co_return result::current_exception();
             }
         };
 
         /* Drain what we pulled so far, and then keep on pulling */
-        StringSource dumpSource { dump };
+        AsyncStringInputStream dumpSource { dump };
         ChainSource bothSource { dumpSource, source };
 
         std::tie(tempDir, tempDirFd) = createTempDirInStore();
@@ -1371,9 +1379,9 @@ try {
         tempPath = tempDir + "/x";
 
         if (method == FileIngestionMethod::Recursive)
-            restorePath(tempPath, bothSource);
+            TRY_AWAIT(restorePath(tempPath, bothSource));
         else
-            writeFile(tempPath, bothSource);
+            TRY_AWAIT(writeFile(tempPath, bothSource));
 
         dumpBuffer.reset();
         dump = {};
diff --git a/lix/libstore/local-store.hh b/lix/libstore/local-store.hh
index 4a2d5ebf9..1da2a0416 100644
--- a/lix/libstore/local-store.hh
+++ b/lix/libstore/local-store.hh
@@ -5,6 +5,7 @@
 
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/indirect-root-store.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/sync.hh"
 #include "lix/libutil/types.hh"
 
@@ -210,8 +211,14 @@ public:
     kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & source,
         RepairFlag repair, CheckSigsFlag checkSigs) override;
 
-    kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
-        FileIngestionMethod method, HashType hashAlgo, RepairFlag repair, const StorePathSet & references) override;
+    kj::Promise<Result<StorePath>> addToStoreFromDump(
+        AsyncInputStream & dump,
+        std::string_view name,
+        FileIngestionMethod method,
+        HashType hashAlgo,
+        RepairFlag repair,
+        const StorePathSet & references
+    ) override;
 
     kj::Promise<Result<StorePath>> addTextToStore(
         std::string_view name,
diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc
index d42501c6c..5c37f40d3 100644
--- a/lix/libstore/remote-store.cc
+++ b/lix/libstore/remote-store.cc
@@ -488,11 +488,16 @@ try {
 }
 
 
-kj::Promise<Result<StorePath>> RemoteStore::addToStoreFromDump(Source & dump, std::string_view name,
-      FileIngestionMethod method, HashType hashType, RepairFlag repair, const StorePathSet & references)
+kj::Promise<Result<StorePath>> RemoteStore::addToStoreFromDump(
+    AsyncInputStream & dump,
+    std::string_view name,
+    FileIngestionMethod method,
+    HashType hashType,
+    RepairFlag repair,
+    const StorePathSet & references
+)
 try {
-    AsyncSourceInputStream stream{dump};
-    co_return TRY_AWAIT(addCAToStore(stream, name, method, hashType, references, repair))->path;
+    co_return TRY_AWAIT(addCAToStore(dump, name, method, hashType, references, repair))->path;
 } catch (...) {
     co_return result::current_exception();
 }
diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh
index da64618a7..8c730cd4f 100644
--- a/lix/libstore/remote-store.hh
+++ b/lix/libstore/remote-store.hh
@@ -88,8 +88,14 @@ public:
     /**
      * Add a content-addressable store path. Does not support references. `dump` will be drained.
      */
-    kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
-        FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = HashType::SHA256, RepairFlag repair = NoRepair, const StorePathSet & references = StorePathSet()) override;
+    kj::Promise<Result<StorePath>> addToStoreFromDump(
+        AsyncInputStream & dump,
+        std::string_view name,
+        FileIngestionMethod method = FileIngestionMethod::Recursive,
+        HashType hashAlgo = HashType::SHA256,
+        RepairFlag repair = NoRepair,
+        const StorePathSet & references = StorePathSet()
+    ) override;
 
     kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, Source & nar,
         RepairFlag repair, CheckSigsFlag checkSigs) override;
diff --git a/lix/libstore/store-api.cc b/lix/libstore/store-api.cc
index 7e176b369..ed06c382e 100644
--- a/lix/libstore/store-api.cc
+++ b/lix/libstore/store-api.cc
@@ -3,6 +3,7 @@
 #include "lix/libstore/derivations.hh"
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/nar-info-disk-cache.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/box_ptr.hh"
 #include "lix/libutil/hash.hh"
@@ -286,7 +287,7 @@ kj::Promise<Result<StorePath>> Store::addToStoreRecursive(
     HashType hashAlgo,
     RepairFlag repair)
 try {
-    auto source = GeneratorSource{_source.dump()};
+    auto source = AsyncGeneratorInputStream{_source.dump()};
     co_return TRY_AWAIT(
         addToStoreFromDump(source, name, FileIngestionMethod::Recursive, hashAlgo, repair, {})
     );
@@ -301,7 +302,7 @@ kj::Promise<Result<StorePath>> Store::addToStoreFlat(
     RepairFlag repair)
 try {
     Path srcPath(absPath(_srcPath));
-    auto source = GeneratorSource{readFileSource(srcPath)};
+    auto source = AsyncGeneratorInputStream{readFileSource(srcPath)};
     co_return TRY_AWAIT(
         addToStoreFromDump(source, name, FileIngestionMethod::Flat, hashAlgo, repair, {})
     );
diff --git a/lix/libstore/store-api.hh b/lix/libstore/store-api.hh
index c25f63e1f..3ee46c4ed 100644
--- a/lix/libstore/store-api.hh
+++ b/lix/libstore/store-api.hh
@@ -2,6 +2,7 @@
 ///@file
 
 #include "lix/libutil/archive.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/logging.hh"
 #include "lix/libstore/nar-info.hh"
@@ -555,9 +556,14 @@ public:
      *
      * \todo remove?
      */
-    virtual kj::Promise<Result<StorePath>> addToStoreFromDump(Source & dump, std::string_view name,
-        FileIngestionMethod method = FileIngestionMethod::Recursive, HashType hashAlgo = HashType::SHA256, RepairFlag repair = NoRepair,
-        const StorePathSet & references = StorePathSet())
+    virtual kj::Promise<Result<StorePath>> addToStoreFromDump(
+        AsyncInputStream & dump,
+        std::string_view name,
+        FileIngestionMethod method = FileIngestionMethod::Recursive,
+        HashType hashAlgo = HashType::SHA256,
+        RepairFlag repair = NoRepair,
+        const StorePathSet & references = StorePathSet()
+    )
     try { unsupported("addToStoreFromDump"); } catch (...) { return {result::current_exception()}; }
 
     /**