diff --git a/lix/libfetchers/tarball.cc b/lix/libfetchers/tarball.cc
index 3adc23ce9..9019ee586 100644
--- a/lix/libfetchers/tarball.cc
+++ b/lix/libfetchers/tarball.cc
@@ -87,7 +87,7 @@ try {
     try {
         auto [meta, content] = TRY_AWAIT(getFileTransfer()->download(url, headers));
         res = std::move(meta);
-        data = content->drain();
+        data = TRY_AWAIT(content->drain());
     } catch (FileTransferError & e) {
         if (cached) {
             warn("%s; using cached version", e.msg());
diff --git a/lix/libstore/builtins/fetchurl.cc b/lix/libstore/builtins/fetchurl.cc
index d90fb0570..07d258c03 100644
--- a/lix/libstore/builtins/fetchurl.cc
+++ b/lix/libstore/builtins/fetchurl.cc
@@ -40,13 +40,14 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData,
     // safe to use because that would badly interfere with the parent process.
     auto fetch = [&](AsyncIoRoot & aio, const std::string & url) {
         auto raw = aio.blockOn(fileTransfer->download(url)).second;
-        auto decompressor = makeDecompressionSource(
-            unpack && mainUrl.ends_with(".xz") ? "xz" : "none", *raw);
+        auto decompressor = makeDecompressionStream(
+            unpack && mainUrl.ends_with(".xz") ? "xz" : "none", std::move(raw)
+        );
 
         if (unpack)
-            restorePath(storePath, *decompressor);
+            aio.blockOn(restorePath(storePath, *decompressor));
         else
-            writeFile(storePath, *decompressor);
+            aio.blockOn(writeFile(storePath, *decompressor));
 
         auto executable = drv.env.find("executable");
         if (executable != drv.env.end() && executable->second == "1") {
diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc
index cc1443320..df7990753 100644
--- a/lix/libstore/filetransfer.cc
+++ b/lix/libstore/filetransfer.cc
@@ -1,4 +1,7 @@
 #include "lix/libstore/filetransfer.hh"
+#include "lix/libutil/async-io.hh"
+#include "lix/libutil/async.hh"
+#include "lix/libutil/error.hh"
 #include "lix/libutil/namespaces.hh"
 #include "lix/libstore/globals.hh"
 #include "lix/libstore/store-api.hh"
@@ -12,7 +15,10 @@
 
 #include <cstddef>
 #include <cstdio>
+#include <exception>
+#include <kj/async.h>
 #include <kj/encoding.h>
+#include <kj/time.h>
 
 #if ENABLE_DTRACE
 #include "trace-probes.gen.hh"
@@ -55,6 +61,22 @@ struct curlFileTransfer : public FileTransfer
             bool done = false;
             std::exception_ptr exc;
             std::string data;
+            std::optional<kj::Own<kj::CrossThreadPromiseFulfiller<void>>> downloadEvent;
+
+            auto wait()
+            {
+                auto pfp = kj::newPromiseAndCrossThreadFulfiller<void>();
+                downloadEvent = std::move(pfp.fulfiller);
+                return std::move(pfp.promise);
+            }
+
+            void signal()
+            {
+                if (downloadEvent) {
+                    (*downloadEvent)->fulfill();
+                    downloadEvent.reset();
+                }
+            }
         };
 
         std::string uri;
@@ -62,9 +84,8 @@ struct curlFileTransfer : public FileTransfer
         Activity act;
         std::unique_ptr<FILE, decltype([](FILE * f) { fclose(f); })> uploadData;
         Sync<DownloadState> downloadState;
-        std::condition_variable downloadEvent;
         bool headersDone = false, metadataReturned = false;
-        std::promise<FileTransferResult> metadataPromise;
+        kj::Own<kj::CrossThreadPromiseFulfiller<Result<FileTransferResult>>> metadataPromise;
         std::string statusMsg;
 
         uint64_t bodySize = 0;
@@ -117,17 +138,23 @@ struct curlFileTransfer : public FileTransfer
             return uploadData ? "upload" : "download";
         }
 
-        TransferItem(const std::string & uri,
+        TransferItem(
+            const std::string & uri,
             const Headers & headers,
             ActivityId parentAct,
             std::optional<std::string_view> uploadData,
             bool noBody,
-            curl_off_t writtenToSink
+            curl_off_t writtenToSink,
+            kj::Own<kj::CrossThreadPromiseFulfiller<Result<FileTransferResult>>> metadataPromise
         )
             : uri(uri)
-            , act(*logger, lvlTalkative, actFileTransfer,
-                fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri),
-                {uri}, parentAct)
+            , act(*logger,
+                  lvlTalkative,
+                  actFileTransfer,
+                  fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri),
+                  {uri},
+                  parentAct)
+            , metadataPromise(std::move(metadataPromise))
             , req(curl_easy_init())
         {
             if (req == nullptr) {
@@ -227,10 +254,10 @@ struct curlFileTransfer : public FileTransfer
             auto state = downloadState.lock();
             assert(!state->done && !state->exc);
             if (!metadataReturned) {
-                metadataPromise.set_exception(ex);
+                metadataPromise->fulfill(ex);
             }
             state->exc = ex;
-            downloadEvent.notify_all();
+            state->signal();
         }
 
         template<class T>
@@ -255,7 +282,7 @@ struct curlFileTransfer : public FileTransfer
 
             result.cached = status == 304;
             if (successfulStatuses.contains(status)) {
-                metadataPromise.set_value(result);
+                metadataPromise->fulfill(result);
                 metadataReturned = true;
             }
 
@@ -280,7 +307,7 @@ struct curlFileTransfer : public FileTransfer
                 }
 
                 state->data.append(static_cast<const char *>(contents), realSize);
-                downloadEvent.notify_all();
+                state->signal();
                 bodySize += realSize;
                 return realSize;
             } catch (...) {
@@ -373,8 +400,9 @@ struct curlFileTransfer : public FileTransfer
             else if (code == CURLE_OK && successfulStatuses.count(httpStatus))
             {
                 act.progress(bodySize, bodySize);
-                downloadState.lock()->done = true;
-                downloadEvent.notify_all();
+                auto state = downloadState.lock();
+                state->done = true;
+                state->signal();
             }
 
             else {
@@ -702,7 +730,7 @@ struct curlFileTransfer : public FileTransfer
         co_return result::current_exception();
     }
 
-    std::optional<std::pair<FileTransferResult, box_ptr<Source>>> tryEagerTransfers(
+    std::optional<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>> tryEagerTransfers(
         const std::string & uri,
         const Headers & headers,
         const std::optional<std::string> & data,
@@ -749,14 +777,25 @@ struct curlFileTransfer : public FileTransfer
                     );
                 }
                 if (S_ISDIR(st.st_mode)) {
-                    return {{std::move(metadata), make_box_ptr<StringSource>("")}};
+                    return {{std::move(metadata), make_box_ptr<AsyncStringInputStream>("")}};
                 }
-                struct OwningFdSource : FdSource
+                struct OwningFdStream : AsyncInputStream
                 {
                     AutoCloseFD fd;
-                    OwningFdSource(AutoCloseFD fd) : FdSource(fd.get()), fd(std::move(fd)) {}
+                    OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {}
+                    kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
+                    {
+                        // NOTE the synchronous implementation used to have a buffer for
+                        // file data, but we cannot be bothered to treat this edge case.
+                        if (const auto got = ::read(fd.get(), buffer, size); got >= 0) {
+                            return {result::success(got)};
+                        } else {
+                            return {result::failure(std::make_exception_ptr(SysError("reading file")
+                            ))};
+                        }
+                    }
                 };
-                return {{std::move(metadata), make_box_ptr<OwningFdSource>(std::move(fd))}};
+                return {{std::move(metadata), make_box_ptr<OwningFdStream>(std::move(fd))}};
             }
         }
 
@@ -778,7 +817,15 @@ struct curlFileTransfer : public FileTransfer
             FileTransferResult res;
             if (!s3Res.data)
                 throw FileTransferError(NotFound, "S3 object '%s' does not exist", uri);
-            return {{res, make_box_ptr<StringSource>(std::move(*s3Res.data))}};
+            struct OwningStringStream : private std::string, AsyncStringInputStream
+            {
+                OwningStringStream(std::string data)
+                    : std::string(std::move(data))
+                    , AsyncStringInputStream(*this)
+                {
+                }
+            };
+            return {{ res, make_box_ptr<OwningStringStream>(std::move(*s3Res.data)) }};
 #else
             throw nix::Error(
                 "cannot download '%s' because Lix is not built with S3 support", uri
@@ -789,7 +836,8 @@ struct curlFileTransfer : public FileTransfer
         return std::nullopt;
     }
 
-    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>> enqueueFileTransfer(
+    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>>
+    enqueueFileTransfer(
         const std::string & uri,
         const Headers & headers,
         std::optional<std::string> data,
@@ -800,15 +848,15 @@ struct curlFileTransfer : public FileTransfer
             co_return std::move(*eager);
         }
 
-        auto source = make_box_ptr<TransferSource>(*this, uri, headers, std::move(data), noBody);
-        source->init();
-        source->awaitData();
+        auto source = make_box_ptr<TransferStream>(*this, uri, headers, std::move(data), noBody);
+        TRY_AWAIT(source->init());
+        TRY_AWAIT(source->awaitData());
         co_return {source->metadata, std::move(source)};
     } catch (...) {
         co_return result::current_exception();
     }
 
-    struct TransferSource : Source
+    struct TransferStream : AsyncInputStream
     {
         curlFileTransfer & parent;
         std::string uri;
@@ -826,7 +874,7 @@ struct curlFileTransfer : public FileTransfer
         const size_t tries = fileTransferSettings.tries;
         curl_off_t totalReceived = 0;
 
-        TransferSource(
+        TransferStream(
             curlFileTransfer & parent,
             const std::string & uri,
             const Headers & headers,
@@ -841,7 +889,7 @@ struct curlFileTransfer : public FileTransfer
         {
         }
 
-        ~TransferSource()
+        ~TransferStream()
         {
             // wake up the download thread if it's still going and have it abort
             try {
@@ -853,22 +901,25 @@ struct curlFileTransfer : public FileTransfer
             }
         }
 
-        void init()
-        {
+        kj::Promise<Result<void>> init()
+        try {
             auto setup = [&] { return startTransfer(uri); };
-            metadata = withRetries(setup, setup);
+            metadata = TRY_AWAIT(withRetries(setup, setup));
+            co_return result::success();
+        } catch (...) {
+            co_return result::current_exception();
         }
 
         auto withRetries(auto && initial, auto && retry) -> decltype(initial())
-        {
+        try {
             std::optional<std::string> retryContext;
             while (true) {
                 try {
                     if (retryContext) {
-                        prepareRetry(*retryContext);
-                        return retry();
+                        TRY_AWAIT(prepareRetry(*retryContext));
+                        co_return TRY_AWAIT(retry());
                     } else {
-                        return initial();
+                        co_return TRY_AWAIT(initial());
                     }
                 } catch (FileTransferError & e) {
                     // If this is a transient error, then maybe retry after a while. after any
@@ -882,16 +933,23 @@ struct curlFileTransfer : public FileTransfer
                     retryContext = e.what();
                 }
             }
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        FileTransferResult startTransfer(const std::string & uri, curl_off_t offset = 0)
-        {
+        kj::Promise<Result<FileTransferResult>>
+        startTransfer(const std::string & uri, curl_off_t offset = 0)
+        try {
             attempt += 1;
             auto uploadData = data ? std::optional(std::string_view(*data)) : std::nullopt;
-            transfer =
-                std::make_shared<TransferItem>(uri, headers, parentAct, uploadData, noBody, offset);
+            auto pfp = kj::newPromiseAndCrossThreadFulfiller<Result<FileTransferResult>>();
+            transfer = std::make_shared<TransferItem>(
+                uri, headers, parentAct, uploadData, noBody, offset, std::move(pfp.fulfiller)
+            );
             parent.enqueueItem(transfer);
-            return transfer->metadataPromise.get_future().get();
+            co_return TRY_AWAIT(pfp.promise);
+        } catch (...) {
+            co_return result::current_exception();
         }
 
         void throwChangedTarget(std::string_view what, std::string_view from, std::string_view to)
@@ -903,8 +961,8 @@ struct curlFileTransfer : public FileTransfer
             }
         }
 
-        void prepareRetry(const std::string & context)
-        {
+        kj::Promise<Result<void>> prepareRetry(const std::string & context)
+        try {
             thread_local std::minstd_rand random{std::random_device{}()};
             std::uniform_real_distribution<> dist(0.0, 0.5);
             int ms = parent.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + dist(random));
@@ -914,16 +972,19 @@ struct curlFileTransfer : public FileTransfer
                 warn("%s; retrying in %d ms (attempt %d/%d)", context, ms, attempt, tries);
             }
 
-            std::this_thread::sleep_for(std::chrono::milliseconds(ms));
+            co_await AIO().provider.getTimer().afterDelay(ms * kj::MILLISECONDS);
+            co_return result::success();
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        void restartTransfer()
-        {
+        kj::Promise<Result<void>> restartTransfer()
+        try {
             // use the effective URI of the previous transfer for retries. this avoids
             // some silent corruption if a redirect changes between starting and retry
             const auto & uri = metadata.effectiveUri.empty() ? this->uri : metadata.effectiveUri;
 
-            auto newMeta = startTransfer(uri, totalReceived);
+            auto newMeta = TRY_AWAIT(startTransfer(uri, totalReceived));
             throwChangedTarget("final destination", metadata.effectiveUri, newMeta.effectiveUri);
             throwChangedTarget("ETag", metadata.etag, newMeta.etag);
             throwChangedTarget(
@@ -931,13 +992,23 @@ struct curlFileTransfer : public FileTransfer
                 metadata.immutableUrl.value_or(""),
                 newMeta.immutableUrl.value_or("")
             );
+            co_return result::success();
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        bool waitForData()
-        {
+        kj::Promise<Result<bool>> waitForData()
+        try {
             /* Grab data if available, otherwise wait for the download
                thread to wake us up. */
+            std::optional<kj::Promise<void>> signal;
+
             while (buffered.empty()) {
+                if (signal) {
+                    co_await *signal;
+                    signal.reset();
+                }
+
                 auto state(transfer->downloadState.lock());
 
                 if (!state->data.empty()) {
@@ -948,46 +1019,51 @@ struct curlFileTransfer : public FileTransfer
                 } else if (state->exc) {
                     std::rethrow_exception(state->exc);
                 } else if (state->done) {
-                    return false;
+                    co_return false;
                 } else {
                     parent.unpause(transfer);
-                    state.wait(transfer->downloadEvent);
+                    signal = state->wait();
                 }
             }
 
-            return true;
+            co_return true;
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        bool restartAndWaitForData()
-        {
-            restartTransfer();
-            return waitForData();
+        kj::Promise<Result<bool>> restartAndWaitForData()
+        try {
+            TRY_AWAIT(restartTransfer());
+            co_return TRY_AWAIT(waitForData());
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        bool awaitData()
-        {
-            return withRetries(
-                [&] { return waitForData(); }, [&] { return restartAndWaitForData(); }
+        kj::Promise<Result<bool>> awaitData()
+        try {
+            co_return TRY_AWAIT(
+                withRetries([&] { return waitForData(); }, [&] { return restartAndWaitForData(); })
             );
+        } catch (...) {
+            co_return result::current_exception();
         }
 
-        size_t read(char * data, size_t len) override
-        {
+        kj::Promise<Result<size_t>> read(void * buffer, size_t len) override
+        try {
             TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len));
 
             size_t total = 0;
-            while (total < len && awaitData()) {
+            auto data = static_cast<char *>(buffer);
+            while (total < len && TRY_AWAIT(awaitData())) {
                 const auto available = std::min(len - total, buffered.size());
                 memcpy(data + total, buffered.data(), available);
                 buffered.remove_prefix(available);
                 total += available;
             }
 
-            if (total == 0) {
-                throw EndOfFile("transfer finished");
-            }
-
-            return total;
+            co_return total;
+        } catch (...) {
+            co_return result::current_exception();
         }
     };
 
@@ -1007,7 +1083,7 @@ struct curlFileTransfer : public FileTransfer
         co_return result::current_exception();
     }
 
-    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>>
+    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>>
     download(const std::string & uri, const Headers & headers) override
     {
         return enqueueFileTransfer(uri, headers, std::nullopt, false);
diff --git a/lix/libstore/filetransfer.hh b/lix/libstore/filetransfer.hh
index 6f2812a8f..42644f7aa 100644
--- a/lix/libstore/filetransfer.hh
+++ b/lix/libstore/filetransfer.hh
@@ -1,6 +1,7 @@
 #pragma once
 ///@file
 
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/box_ptr.hh"
 #include "lix/libutil/ref.hh"
 #include "lix/libutil/logging.hh"
@@ -65,7 +66,7 @@ struct FileTransfer
      * thrown by the returned source. The source will only throw errors detected
      * during the transfer itself (decompression errors, connection drops, etc).
      */
-    virtual kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>>
+    virtual kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>>
     download(const std::string & uri, const Headers & headers = {}) = 0;
 
     enum Error { NotFound, Forbidden, Misc, Transient, Interrupted };
diff --git a/lix/libstore/http-binary-cache-store.cc b/lix/libstore/http-binary-cache-store.cc
index dc13ddf4c..a7c6f0634 100644
--- a/lix/libstore/http-binary-cache-store.cc
+++ b/lix/libstore/http-binary-cache-store.cc
@@ -167,19 +167,7 @@ protected:
     try {
         checkEnabled();
         try {
-            struct HttpFile : AsyncSourceInputStream
-            {
-                box_ptr<Source> source;
-
-                HttpFile(box_ptr<Source> source)
-                    : AsyncSourceInputStream(*source)
-                    , source(std::move(source))
-                {
-                }
-            };
-            co_return make_box_ptr<HttpFile>(
-                TRY_AWAIT(getFileTransfer()->download(makeURI(path))).second
-            );
+            co_return TRY_AWAIT(getFileTransfer()->download(makeURI(path))).second;
         } catch (FileTransferError & e) {
             if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
                 throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
diff --git a/lix/nix/prefetch.cc b/lix/nix/prefetch.cc
index afff1af2b..f8a365234 100644
--- a/lix/nix/prefetch.cc
+++ b/lix/nix/prefetch.cc
@@ -100,7 +100,7 @@ std::tuple<StorePath, Hash> prefetchFile(
 
             FdSink sink(fd.get());
 
-            aio.blockOn(getFileTransfer()->download(url)).second->drainInto(sink);
+            aio.blockOn(aio.blockOn(getFileTransfer()->download(url)).second->drainInto(sink));
         }
 
         /* Optionally unpack the file. */
diff --git a/lix/nix/upgrade-nix.cc b/lix/nix/upgrade-nix.cc
index cd483b400..35a1a964b 100644
--- a/lix/nix/upgrade-nix.cc
+++ b/lix/nix/upgrade-nix.cc
@@ -288,7 +288,7 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand
 
         // FIXME: use nixos.org?
         auto [res, content] = aio().blockOn(getFileTransfer()->download(storePathsUrl));
-        auto data = content->drain();
+        auto data = aio().blockOn(content->drain());
 
         auto evaluator = std::make_unique<Evaluator>(aio(), SearchPath{}, store);
         auto state = evaluator->begin(aio());
diff --git a/tests/unit/libstore/filetransfer.cc b/tests/unit/libstore/filetransfer.cc
index f2a17bbb8..b12baa2be 100644
--- a/tests/unit/libstore/filetransfer.cc
+++ b/tests/unit/libstore/filetransfer.cc
@@ -1,4 +1,5 @@
 #include "lix/libstore/filetransfer.hh"
+#include "lix/libutil/async-io.hh"
 #include "lix/libutil/async.hh"
 #include "lix/libutil/compression.hh"
 #include "lix/libutil/error.hh"
@@ -214,7 +215,9 @@ TEST(FileTransfer, exceptionAbortsDownload)
 
     auto [port, srv] = serveHTTP({{"200 ok", "", [](int) { return "foo"; }}});
     ASSERT_THROW(
-        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken),
+        aio.blockOn(
+            aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken)
+        ),
         Done
     );
 
@@ -236,9 +239,11 @@ TEST(FileTransfer, exceptionAbortsRead)
     AsyncIoRoot aio;
     auto ft = makeFileTransfer();
     char buf[10] = "";
-    ASSERT_THROW(
-        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10),
-        EndOfFile
+    ASSERT_EQ(
+        aio.blockOn(
+            aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10)
+        ),
+        0
     );
 }
 
@@ -263,7 +268,7 @@ TEST(FileTransfer, NOT_ON_DARWIN(defersFailures))
     AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
     auto src = aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second;
-    ASSERT_THROW(src->drain(), FileTransferError);
+    ASSERT_THROW(aio.blockOn(src->drain()), FileTransferError);
 }
 
 TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
@@ -276,7 +281,9 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
     auto ft = makeFileTransfer();
 
     StringSink sink;
-    aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink);
+    aio.blockOn(
+        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink)
+    );
     EXPECT_EQ(sink.s, original);
 }
 
@@ -317,23 +324,29 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers)
     auto ft = makeFileTransfer(0);
     auto [_result1, data1] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port)));
     auto [_result2, data2] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port)));
-    auto drop = [](Source & source, size_t size) {
+    auto drop = [&](AsyncInputStream & source, size_t size) {
         char buf[1000];
+        size_t dropped = 0;
         while (size > 0) {
             auto round = std::min(size, sizeof(buf));
-            source(buf, round);
+            round = aio.blockOn(source.read(buf, round));
+            if (round == 0) {
+                break;
+            }
             size -= round;
+            dropped += round;
         }
+        return dropped;
     };
     // read 10M of each of the 100M, then the rest. neither reader should
     // block the other, nor should it take that long to copy 200MB total.
-    drop(*data1, 10'000'000);
-    drop(*data2, 10'000'000);
-    drop(*data1, 90'000'000);
-    drop(*data2, 90'000'000);
+    ASSERT_EQ(drop(*data1, 10'000'000), 10'000'000);
+    ASSERT_EQ(drop(*data2, 10'000'000), 10'000'000);
+    ASSERT_EQ(drop(*data1, 90'000'000), 90'000'000);
+    ASSERT_EQ(drop(*data2, 90'000'000), 90'000'000);
 
-    ASSERT_THROW(drop(*data1, 1), EndOfFile);
-    ASSERT_THROW(drop(*data2, 1), EndOfFile);
+    ASSERT_EQ(drop(*data1, 1), 0);
+    ASSERT_EQ(drop(*data2, 1), 0);
 }
 
 TEST(FileTransfer, retries)
@@ -356,7 +369,7 @@ TEST(FileTransfer, retries)
     AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
     auto [result, data] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port)));
-    ASSERT_EQ(data->drain(), "ab");
+    ASSERT_EQ(aio.blockOn(data->drain()), "ab");
 }
 
 TEST(FileTransfer, doesntRetrySetupForever)
@@ -390,7 +403,8 @@ TEST(FileTransfer, doesntRetryTransferForever)
     AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
     ASSERT_THROW(
-        aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain(), FileTransferError
+        aio.blockOn(aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain()),
+        FileTransferError
     );
 }
 
@@ -445,7 +459,7 @@ TEST(FileTransfer, DISABLED_interrupt)
     });
 
     ASSERT_THROW(
-        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain(),
+        aio.blockOn(aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain()),
         FileTransferError
     );
 }