libstore: asyncify curl interface

Change-Id: I3fc93016b8ac5e59d9062d4f4aead19ae051a680

lix/libfetchers/tarball.cc

@@ -85,7 +85,7 @@ try {
     FileTransferResult res;
     std::string data;
     try {
-        auto [meta, content] = getFileTransfer()->download(url, headers);
+        auto [meta, content] = TRY_AWAIT(getFileTransfer()->download(url, headers));
         res = std::move(meta);
         data = content->drain();
     } catch (FileTransferError & e) {

lix/libstore/builtins/fetchurl.cc

@@ -2,6 +2,7 @@
 #include "lix/libstore/filetransfer.hh"
 #include "lix/libstore/store-api.hh"
 #include "lix/libutil/archive.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/compression.hh"
 #include "lix/libutil/strings.hh"
 
@@ -34,9 +35,11 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData,
        a forked process. */
     auto fileTransfer = makeFileTransfer();
 
-    auto fetch = [&](const std::string & url) {
+    // we also have to run the remainder of this function in a fresh thread so
-
+    // we can have an aio root. the existing root on the current thread is not
-        auto raw = fileTransfer->download(url).second;
+    // 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);
 
@@ -52,21 +55,33 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData,
         }
     };
 
+    std::async(std::launch::async, [&] {
+        AsyncIoRoot aio;
+
         /* Try the hashed mirrors first. */
-    if (getAttr("outputHashMode") == "flat")
+        if (getAttr("outputHashMode") == "flat") {
-        for (auto hashedMirror : settings.hashedMirrors.get())
+            for (auto hashedMirror : settings.hashedMirrors.get()) {
                 try {
-                if (!hashedMirror.ends_with("/")) hashedMirror += '/';
+                    if (!hashedMirror.ends_with("/")) {
+                        hashedMirror += '/';
+                    }
                     std::optional<HashType> ht = parseHashTypeOpt(getAttr("outputHashAlgo"));
                     Hash h = newHashAllowEmpty(getAttr("outputHash"), ht);
-                fetch(hashedMirror + printHashType(h.type) + "/" + h.to_string(Base::Base16, false));
+                    fetch(
+                        aio,
+                        hashedMirror + printHashType(h.type) + "/"
+                            + h.to_string(Base::Base16, false)
+                    );
                     return;
                 } catch (Error & e) {
                     debug(e.what());
                 }
+            }
+        }
 
         /* Otherwise try the specified URL. */
-    fetch(mainUrl);
+        fetch(aio, mainUrl);
+    }).get();
 }
 
 }

lix/libstore/filetransfer.cc

@@ -4,6 +4,7 @@
 #include "lix/libstore/store-api.hh"
 #include "lix/libstore/s3.hh"
 #include "lix/libutil/regex.hh"
+#include "lix/libutil/result.hh"
 #include "lix/libutil/signals.hh"
 #include "lix/libutil/strings.hh"
 #include "lix/libutil/thread-name.hh"
@@ -692,9 +693,13 @@ struct curlFileTransfer : public FileTransfer
     }
 #endif
 
-    void upload(const std::string & uri, std::string data, const Headers & headers) override
+    kj::Promise<Result<void>>
-    {
+    upload(const std::string & uri, std::string data, const Headers & headers) override
-        enqueueFileTransfer(uri, headers, std::move(data), false);
+    try {
+        TRY_AWAIT(enqueueFileTransfer(uri, headers, std::move(data), false));
+        co_return result::success();
+    } catch (...) {
+        co_return result::current_exception();
     }
 
     std::optional<std::pair<FileTransferResult, box_ptr<Source>>> tryEagerTransfers(
@@ -784,20 +789,22 @@ struct curlFileTransfer : public FileTransfer
         return std::nullopt;
     }
 
-    std::pair<FileTransferResult, box_ptr<Source>> enqueueFileTransfer(
+    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>> enqueueFileTransfer(
         const std::string & uri,
         const Headers & headers,
         std::optional<std::string> data,
         bool noBody
     )
-    {
+    try {
         if (auto eager = tryEagerTransfers(uri, headers, data, noBody)) {
-            return std::move(*eager);
+            co_return std::move(*eager);
         }
 
         auto source = make_box_ptr<TransferSource>(*this, uri, headers, std::move(data), noBody);
         source->awaitData();
-        return {source->metadata, std::move(source)};
+        co_return {source->metadata, std::move(source)};
+    } catch (...) {
+        co_return result::current_exception();
     }
 
     struct TransferSource : Source
@@ -972,21 +979,23 @@ struct curlFileTransfer : public FileTransfer
         }
     };
 
-    bool exists(const std::string & uri, const Headers & headers) override
+    kj::Promise<Result<bool>> exists(const std::string & uri, const Headers & headers) override
-    {
     try {
-            enqueueFileTransfer(uri, headers, std::nullopt, true);
+        try {
-            return true;
+            TRY_AWAIT(enqueueFileTransfer(uri, headers, std::nullopt, true));
+            co_return true;
         } catch (FileTransferError & e) {
             /* S3 buckets return 403 if a file doesn't exist and the
                 bucket is unlistable, so treat 403 as 404. */
             if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
-                return false;
+                co_return false;
             throw;
         }
+    } catch (...) {
+        co_return result::current_exception();
     }
 
-    std::pair<FileTransferResult, box_ptr<Source>>
+    kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>>
     download(const std::string & uri, const Headers & headers) override
     {
         return enqueueFileTransfer(uri, headers, std::nullopt, false);

lix/libstore/filetransfer.hh

@@ -4,10 +4,12 @@
 #include "lix/libutil/box_ptr.hh"
 #include "lix/libutil/ref.hh"
 #include "lix/libutil/logging.hh"
+#include "lix/libutil/result.hh"
 #include "lix/libutil/serialise.hh"
 #include "lix/libutil/types.hh"
 #include "lix/libutil/config.hh"
 
+#include <kj/async.h>
 #include <string>
 #include <future>
 
@@ -40,7 +42,7 @@ struct FileTransfer
     /**
      * Upload some data. May throw a FileTransferError exception.
      */
-    virtual void
+    virtual kj::Promise<Result<void>>
     upload(const std::string & uri, std::string data, const Headers & headers = {}) = 0;
 
     /**
@@ -53,7 +55,8 @@ struct FileTransfer
      *
      * S3 objects are downloaded completely to answer this request.
      */
-    virtual bool exists(const std::string & uri, const Headers & headers = {}) = 0;
+    virtual kj::Promise<Result<bool>>
+    exists(const std::string & uri, const Headers & headers = {}) = 0;
 
     /**
      * Download a file, returning its contents through a source. Will not return
@@ -62,7 +65,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 std::pair<FileTransferResult, box_ptr<Source>>
+    virtual kj::Promise<Result<std::pair<FileTransferResult, box_ptr<Source>>>>
     download(const std::string & uri, const Headers & headers = {}) = 0;
 
     enum Error { NotFound, Forbidden, Misc, Transient, Interrupted };

lix/libstore/http-binary-cache-store.cc

@@ -125,7 +125,7 @@ protected:
         checkEnabled();
 
         try {
-            co_return getFileTransfer()->exists(makeURI(path));
+            co_return TRY_AWAIT(getFileTransfer()->exists(makeURI(path)));
         } catch (FileTransferError & e) {
             maybeDisable();
             throw;
@@ -142,7 +142,9 @@ protected:
     try {
         auto data = StreamToSourceAdapter(istream).drain();
         try {
-            getFileTransfer()->upload(makeURI(path), std::move(data), {{"Content-Type", mimeType}});
+            TRY_AWAIT(getFileTransfer()->upload(
+                makeURI(path), std::move(data), {{"Content-Type", mimeType}}
+            ));
         } catch (FileTransferError & e) {
             throw UploadToHTTP(
                 "while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg()
@@ -175,7 +177,9 @@ protected:
                 {
                 }
             };
-            return {make_box_ptr<HttpFile>(getFileTransfer()->download(makeURI(path)).second)};
+            co_return make_box_ptr<HttpFile>(
+                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());
@@ -183,7 +187,7 @@ protected:
             throw;
         }
     } catch (...) {
-        return {result::current_exception()};
+        co_return result::current_exception();
     }
 
     /**

lix/nix/prefetch.cc

@@ -100,7 +100,7 @@ std::tuple<StorePath, Hash> prefetchFile(
 
             FdSink sink(fd.get());
 
-            getFileTransfer()->download(url).second->drainInto(sink);
+            aio.blockOn(getFileTransfer()->download(url)).second->drainInto(sink);
         }
 
         /* Optionally unpack the file. */

lix/nix/upgrade-nix.cc

@@ -287,7 +287,7 @@ struct CmdUpgradeNix : MixDryRun, EvalCommand
         Activity act(*logger, lvlInfo, actUnknown, "querying latest Nix version");
 
         // FIXME: use nixos.org?
-        auto [res, content] = getFileTransfer()->download(storePathsUrl);
+        auto [res, content] = aio().blockOn(getFileTransfer()->download(storePathsUrl));
         auto data = content->drain();
 
         auto evaluator = std::make_unique<Evaluator>(aio(), SearchPath{}, store);

tests/unit/libstore/filetransfer.cc

@@ -1,4 +1,5 @@
 #include "lix/libstore/filetransfer.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/compression.hh"
 #include "lix/libutil/error.hh"
 #include "lix/libutil/signals.hh"
@@ -206,12 +207,16 @@ TEST(FileTransfer, exceptionAbortsDownload)
     struct Done : BaseException
     {};
 
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer();
 
     LambdaSink broken([](auto block) { throw Done(); });
 
     auto [port, srv] = serveHTTP({{"200 ok", "", [](int) { return "foo"; }}});
-    ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->drainInto(broken), Done);
+    ASSERT_THROW(
+        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(broken),
+        Done
+    );
 
     // makeFileTransfer returns a ref<>, which cannot be cleared. since we also
     // can't default-construct it we'll have to overwrite it instead, but we'll
@@ -228,19 +233,21 @@ TEST(FileTransfer, exceptionAbortsDownload)
 TEST(FileTransfer, exceptionAbortsRead)
 {
     auto [port, srv] = serveHTTP("200 ok", "content-length: 0\r\n", [] { return ""; });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer();
     char buf[10] = "";
-    ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->read(buf, 10), EndOfFile);
+    ASSERT_THROW(
+        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10),
+        EndOfFile
+    );
 }
 
 TEST(FileTransfer, NOT_ON_DARWIN(reportsSetupErrors))
 {
     auto [port, srv] = serveHTTP("404 not found", "", [] { return ""; });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer();
-    ASSERT_THROW(
+    ASSERT_THROW(aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))), FileTransferError);
-        ft->download(fmt("http://[::1]:%d/index", port)),
-        FileTransferError
-    );
 }
 
 TEST(FileTransfer, NOT_ON_DARWIN(defersFailures))
@@ -253,8 +260,9 @@ TEST(FileTransfer, NOT_ON_DARWIN(defersFailures))
         // might only do so once its internal buffer has already been filled.)
         return std::string(1024 * 1024, ' ');
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    auto src = ft->download(fmt("http://[::1]:%d/index", port)).second;
+    auto src = aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second;
     ASSERT_THROW(src->drain(), FileTransferError);
 }
 
@@ -264,10 +272,11 @@ TEST(FileTransfer, NOT_ON_DARWIN(handlesContentEncoding))
     std::string compressed = compress("gzip", original);
 
     auto [port, srv] = serveHTTP("200 ok", "content-encoding: gzip\r\n", [&] { return compressed; });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer();
 
     StringSink sink;
-    ft->download(fmt("http://[::1]:%d/index", port)).second->drainInto(sink);
+    aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drainInto(sink);
     EXPECT_EQ(sink.s, original);
 }
 
@@ -289,8 +298,9 @@ TEST(FileTransfer, usesIntermediateLinkHeaders)
          [] { return ""; }},
         {"200 ok", "content-length: 1\r\n", [] { return "a"; }},
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    auto [result, _data] = ft->download(fmt("http://[::1]:%d/first", port));
+    auto [result, _data] = aio.blockOn(ft->download(fmt("http://[::1]:%d/first", port)));
     ASSERT_EQ(result.immutableUrl, "http://foo");
 }
 
@@ -303,9 +313,10 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers)
              return round < 100 ? std::optional(std::string(1'000'000, ' ')) : std::nullopt;
          }},
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    auto [_result1, data1] = ft->download(fmt("http://[::1]:%d", port));
+    auto [_result1, data1] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port)));
-    auto [_result2, data2] = 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) {
         char buf[1000];
         while (size > 0) {
@@ -342,8 +353,9 @@ TEST(FileTransfer, retries)
          [] { return "b"; },
          {"Range: bytes=1-"}},
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    auto [result, data] = ft->download(fmt("http://[::1]:%d", port));
+    auto [result, data] = aio.blockOn(ft->download(fmt("http://[::1]:%d", port)));
     ASSERT_EQ(data->drain(), "ab");
 }
 
@@ -352,8 +364,9 @@ TEST(FileTransfer, doesntRetrySetupForever)
     auto [port, srv] = serveHTTP({
         {"429 try again later", "content-length: 0\r\n", [] { return ""; }},
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    ASSERT_THROW(ft->download(fmt("http://[::1]:%d", port)), FileTransferError);
+    ASSERT_THROW(aio.blockOn(ft->download(fmt("http://[::1]:%d", port))), FileTransferError);
 }
 
 TEST(FileTransfer, doesntRetryTransferForever)
@@ -374,12 +387,16 @@ TEST(FileTransfer, doesntRetryTransferForever)
         );
     }
     auto [port, srv] = serveHTTP(replies);
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    ASSERT_THROW(ft->download(fmt("http://[::1]:%d", port)).second->drain(), FileTransferError);
+    ASSERT_THROW(
+        aio.blockOn(ft->download(fmt("http://[::1]:%d", port))).second->drain(), FileTransferError
+    );
 }
 
 TEST(FileTransfer, doesntRetryUploads)
 {
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
 
     {
@@ -387,14 +404,16 @@ TEST(FileTransfer, doesntRetryUploads)
             {"429 try again later", "", [] { return ""; }},
             {"200 ok", "", [] { return ""; }},
         });
-        ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), ""), FileTransferError);
+        ASSERT_THROW(aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "")), FileTransferError);
     }
     {
         auto [port, srv] = serveHTTP({
             {"429 try again later", "", [] { return ""; }},
             {"200 ok", "", [] { return ""; }},
         });
-        ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), "foo"), FileTransferError);
+        ASSERT_THROW(
+            aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "foo")), FileTransferError
+        );
     }
 }
 
@@ -419,12 +438,16 @@ TEST(FileTransfer, DISABLED_interrupt)
     verbosity = lvlDebug;
     logger = new InterruptingLogger;
 
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
     auto [port, srv] = serveHTTP({
         {"200 ok", "content-length: 10\r\n", [] { return "0123456789"; }},
     });
 
-    ASSERT_THROW(ft->download(fmt("http://[::1]:%d/index", port)).second->drain(), FileTransferError);
+    ASSERT_THROW(
+        aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->drain(),
+        FileTransferError
+    );
 }
 
 TEST(FileTransfer, setupErrorsAreMetadata)
@@ -432,8 +455,9 @@ TEST(FileTransfer, setupErrorsAreMetadata)
     auto [port, srv] = serveHTTP({
         {"404 try again later", "content-length: 1\r\n", [] { return "X"; }},
     });
+    AsyncIoRoot aio;
     auto ft = makeFileTransfer(0);
-    ASSERT_THROW(ft->upload(fmt("http://[::1]:%d", port), ""), FileTransferError);
+    ASSERT_THROW(aio.blockOn(ft->upload(fmt("http://[::1]:%d", port), "")), FileTransferError);
 }
 
 }