From 43c5aa207672b9d75da35932693993354ac8900c Mon Sep 17 00:00:00 2001 From: Qyriad Date: Wed, 1 Jul 2026 12:24:52 +0200 Subject: [PATCH] libstore: factor out curlFileTransfer As part of efforts to make debugging our Curl usage easier Change-Id: Ib1f8567daa8b3d3d559a164bc85a8b906a6a6964 --- lix/libstore/curlfiletransfer.cc | 470 +++++++++++++++++++++++++++++ lix/libstore/curlfiletransfer.hh | 170 +++++++++++ lix/libstore/filetransfer.cc | 487 +------------------------------ lix/libstore/meson.build | 2 + 4 files changed, 643 insertions(+), 486 deletions(-) create mode 100644 lix/libstore/curlfiletransfer.cc create mode 100644 lix/libstore/curlfiletransfer.hh diff --git a/lix/libstore/curlfiletransfer.cc b/lix/libstore/curlfiletransfer.cc new file mode 100644 index 000000000..4f3ab8161 --- /dev/null +++ b/lix/libstore/curlfiletransfer.cc @@ -0,0 +1,470 @@ +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include + +#if ENABLE_S3 +#include +#endif + +#include "lix/libstore/curlfiletransfer.hh" +#include "lix/libstore/curlmulti.hh" +#include "lix/libstore/filetransfer.hh" +#include "lix/libstore/s3.hh" +#include "lix/libstore/store-api.hh" +#include "lix/libutil/async-io.hh" +#include "lix/libutil/async.hh" +#include "lix/libutil/box_ptr.hh" +#include "lix/libutil/c-calls.hh" +#include "lix/libutil/error.hh" +#include "lix/libutil/logging.hh" +#include "lix/libutil/result.hh" +#include "lix/libutil/tracepoint.hh" +#include "lix/libutil/types.hh" + +#if ENABLE_DTRACE +#include "trace-probes.gen.hh" +#endif + +namespace nix { + +curlFileTransfer::curlFileTransfer(unsigned int baseRetryTimeMs) + : multi(std::make_shared(baseRetryTimeMs)) +{ +} + +curlFileTransfer::~curlFileTransfer() +{ + multi->stopWorkerThread(); +} + +#if ENABLE_S3 +std::tuple curlFileTransfer::parseS3Uri(std::string uri) +{ + auto [path, params] = splitUriAndParams(uri); + + auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix + if (slash == std::string::npos) + throw nix::Error("bad S3 URI '%s'", path); + + std::string bucketName(path, 5, slash - 5); + std::string key(path, slash + 1); + + return {bucketName, key, params}; +} +#endif + +kj::Promise> curlFileTransfer::upload( + const std::string & uri, + std::string data, + FileTransferOptions options, + const Activity * context +) +try { + TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::move(data), false, context)); + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise>>>> +curlFileTransfer::tryEagerTransfers( + const std::string & uri, + const FileTransferOptions & options, + const std::optional & data, + bool noBody +) +try { + // curl transfers using file:// urls cannot be paused, and are a bit unruly + // in other ways too. since their metadata is trivial and we already have a + // backend for simple file system reads we can use that instead. we'll pass + // uploads to files to curl even so, those will fail in enqueueItem anyway. + // on all other decoding failures we also let curl fail for us a bit later. + // + // note that we use kj to decode the url, not curl. curl uses only the path + // component of the url to determine the file name, but it does note expose + // the decoding method it uses for this. for file:// transfers curl forbids + // only \0 characters in the urldecoded path, not all control characters as + // it does in the public curl_url_get(CURLUPART_PATH, CURLU_URLDECODE) api. + // + // also note: everything weird you see here is for compatibility with curl. + // we can't even fix it because nix-channel relies on this. even reading of + // directories being allowed and returning something (though hopefully it's + // enough to return anything instead of a directory listing like curl does) + if (uri.starts_with("file://") && !data.has_value()) { + if (!uri.starts_with("file:///")) { + throw FileTransferError(NotFound, std::nullopt, "file not found"); + } + auto url = curl_url(); + if (!url) { + throw std::bad_alloc(); + } + KJ_DEFER(curl_url_cleanup(url)); + curl_url_set(url, CURLUPART_URL, requireCString(uri), 0); + char * path = nullptr; + curl_url_get(url, CURLUPART_PATH, &path, 0); + auto decoded = kj::decodeUriComponent(kj::arrayPtr(path, path + strlen(path))); + if (!decoded.hadErrors && decoded.findFirst(0) == nullptr) { + Path fsPath(decoded.cStr(), decoded.size()); + FileTransferResult metadata{.effectiveUri = std::string("file://") + path}; + struct stat st; + AutoCloseFD fd(sys::open(fsPath, O_RDONLY)); + if (!fd || fstat(fd.get(), &st) != 0) { + throw FileTransferError( + NotFound, std::nullopt, "%s: file not found (%s)", fsPath, strerror(errno) + ); + } + if (S_ISDIR(st.st_mode)) { + co_return std::pair{ + std::move(metadata), make_box_ptr("") + }; + } + struct OwningFdStream : AsyncInputStream + { + AutoCloseFD fd; + OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {} + kj::Promise>> + 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) { + if (got == 0) { + return {result::success(std::nullopt)}; + } else { + return {result::success(got)}; + } + } else { + return {result::failure(std::make_exception_ptr(SysError("reading file") + ))}; + } + } + }; + co_return std::pair{ + std::move(metadata), make_box_ptr(std::move(fd)) + }; + } + } + + /* Ugly hack to support s3:// URIs. */ + if (uri.starts_with("s3://")) { + // FIXME: do this on a worker thread +#if ENABLE_S3 + auto [bucketName, key, params] = parseS3Uri(uri); + + std::string profile = getOr(params, "profile", ""); + std::string region = getOr(params, "region", Aws::Region::US_EAST_1); + std::string scheme = getOr(params, "scheme", ""); + std::string endpoint = getOr(params, "endpoint", ""); + + S3Helper s3Helper(profile, region, scheme, endpoint); + + // FIXME: implement ETag + auto s3Res = TRY_AWAIT(s3Helper.getObject(bucketName, key)); + FileTransferResult res; + if (!s3Res.data) + throw FileTransferError(NotFound, "S3 object '%s' does not exist", uri); + struct OwningStringStream : private std::string, AsyncStringInputStream + { + OwningStringStream(std::string data) + : std::string(std::move(data)) + , AsyncStringInputStream(*this) + { + } + }; + co_return std::pair{res, make_box_ptr(std::move(*s3Res.data))}; +#else + throw nix::Error( + "cannot download '%s' because Lix is not built with S3 support", uri + ); +#endif + } + + co_return std::nullopt; +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise>>> +curlFileTransfer::enqueueFileTransfer( + const std::string & uri, + FileTransferOptions && options, + std::optional data, + bool noBody, + const Activity * context +) +try { + if (auto eager = TRY_AWAIT(tryEagerTransfers(uri, options, data, noBody))) { + co_return std::move(*eager); + } + + auto source = make_box_ptr( + *this, uri, std::move(options), std::move(data), noBody, context + ); + TRY_AWAIT(source->init()); + TRY_AWAIT(source->awaitData()); + co_return {source->metadata, std::move(source)}; +} catch (...) { + co_return result::current_exception(); +} + +curlFileTransfer::TransferStream::TransferStream( + curlFileTransfer & parent, + const std::string & uri, + FileTransferOptions && options, + std::optional data, + bool noBody, + const Activity * context +) + : parent(parent.multi) + , uri(uri) + , options(options) + , data(std::move(data)) + , noBody(noBody) + , parentAct(context) + , backoff(backoffTimeouts( + fileTransferSettings.tries, + std::chrono::seconds(fileTransferSettings.maxConnectTimeout.get()), + std::chrono::seconds(fileTransferSettings.initialConnectTimeout.get()), + std::chrono::milliseconds(this->parent->baseRetryTimeMs) + )) +{ +} + +curlFileTransfer::TransferStream::~TransferStream() +{ + // wake up the download thread if it's still going and have it abort + try { + if (transfer) { + parent->cancel(transfer); + } + } catch (...) { + ignoreExceptionInDestructor(); + } +} + +kj::Promise> curlFileTransfer::TransferStream::init() +try { + metadata = TRY_AWAIT(withRetries( + [&]() { + return startTransfer( + uri, std::chrono::seconds(fileTransferSettings.initialConnectTimeout.get()) + ); + }, + [&](const std::chrono::milliseconds & timeout) { + return startTransfer(uri, timeout); + } + )); + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::startTransfer( + const std::string & uri, + const std::chrono::milliseconds & timeout, + curl_off_t offset +) +try { + auto uploadData = data ? std::optional(std::string_view(*data)) : std::nullopt; + auto pfp = kj::newPromiseAndCrossThreadFulfiller>(); + transfer = std::make_shared( + uri, + std::move(options), + parentAct, + uploadData, + noBody, + offset, + std::move(pfp.fulfiller), + timeout + ); + parent->enqueueItem(transfer); + co_return TRY_AWAIT(pfp.promise); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::prepareRetry( + const std::string & context, + const std::chrono::milliseconds & waitTime, + unsigned int attempt +) +try { + if (totalReceived) { + printTaggedWarning( + "%s; retrying from offset %d in %d ms (attempt %d/%d)", + Uncolored(context), + totalReceived, + waitTime.count(), + Uncolored(attempt), + Uncolored(tries) + ); + } else { + printTaggedWarning( + "%s; retrying in %d ms (attempt %d/%d)", + Uncolored(context), + waitTime.count(), + Uncolored(attempt), + Uncolored(tries) + ); + } + + co_await AIO().provider.getTimer().afterDelay(waitTime.count() * kj::MILLISECONDS); + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::restartTransfer(const std::chrono::milliseconds & timeout) +try { + auto onChange = + [&](std::string_view what, std::string_view from, std::string_view to, bool throw_ + ) -> void { + if (!from.empty() && from != to) { + FileTransferError e = FileTransferError( + Misc, + {}, + "uri %s changed %s from %s to %s during transfer", + uri, + what, + from, + to + ); + + if (throw_) { + throw e; + } + + logWarning(e.info()); + } + }; + + // 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 = TRY_AWAIT(startTransfer(uri, timeout, totalReceived)); + + onChange("final destination", metadata.effectiveUri, newMeta.effectiveUri, false); + onChange("ETag", metadata.etag, newMeta.etag, true); + onChange( + "immutable url", + metadata.immutableUrl.value_or(""), + newMeta.immutableUrl.value_or(""), + true + ); + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::waitForData() +try { + /* Grab data if available, otherwise wait for the download + thread to wake us up. */ + std::optional> signal; + + while (buffered.empty()) { + if (signal) { + co_await *signal; + signal.reset(); + } + + auto state(transfer->downloadState.lock()); + + if (!state->data.empty()) { + chunk = std::move(state->data); + buffered = chunk; + totalReceived += chunk.size(); + parent->unpause(transfer); + } else if (state->exc) { + std::rethrow_exception(state->exc); + } else if (state->done) { + co_return false; + } else { + parent->unpause(transfer); + signal = state->wait(); + } + } + + co_return true; +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::restartAndWaitForData(const std::chrono::milliseconds & timeout) +try { + TRY_AWAIT(restartTransfer(timeout)); + co_return TRY_AWAIT(waitForData()); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> curlFileTransfer::TransferStream::awaitData() +try { + co_return TRY_AWAIT(withRetries( + [&] { return waitForData(); }, + [&](const std::chrono::milliseconds & timeout) { + return restartAndWaitForData(timeout); + } + )); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise>> curlFileTransfer::TransferStream::read(void * buffer, size_t len) +try { + TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len)); + + size_t total = 0; + auto data = static_cast(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) { + co_return std::nullopt; + } else { + co_return total; + } +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> +curlFileTransfer::exists(const std::string & uri, FileTransferOptions options, const Activity * context) +try { + try { + TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::nullopt, true, context)); + 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) + co_return false; + throw; + } +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise>>> curlFileTransfer::download( + const std::string & uri, FileTransferOptions options, const Activity * context +) +{ + return enqueueFileTransfer(uri, std::move(options), std::nullopt, false, context); +} + +} diff --git a/lix/libstore/curlfiletransfer.hh b/lix/libstore/curlfiletransfer.hh new file mode 100644 index 000000000..4a1fc64a0 --- /dev/null +++ b/lix/libstore/curlfiletransfer.hh @@ -0,0 +1,170 @@ +#pragma once +///@file + +#include "lix/libstore/filetransfer.hh" +#include "lix/libstore/store-api.hh" +#include "lix/libstore/transferitem.hh" +#include "lix/libutil/async-io.hh" +#include "lix/libutil/backoff.hh" +#include "lix/libutil/box_ptr.hh" +#include "lix/libutil/result.hh" + +#include +#include +#include +#include +#include +#include + +#include + +namespace nix { + +struct CurlMulti; + +struct curlFileTransfer : public FileTransfer +{ + // Types. +public: + template + using Async = kj::Promise>; + + struct TransferStream : AsyncInputStream + { + std::shared_ptr parent; + std::string uri; + FileTransferOptions options; + std::optional data; + bool noBody; + const Activity * parentAct; + + std::shared_ptr transfer; + FileTransferResult metadata; + std::string chunk; + std::string_view buffered; + + const size_t tries = fileTransferSettings.tries; + curl_off_t totalReceived = 0; + + Generator backoff; + + TransferStream( + curlFileTransfer & parent, + const std::string & uri, + FileTransferOptions && options, + std::optional data, + bool noBody, + const Activity * context + ); + + ~TransferStream(); + + kj::Promise> init(); + + inline auto withRetries(auto && initial, auto && retry) -> decltype(initial()) + try { + std::optional retryContext; + BackoffTiming timings; + while (true) { + try { + if (retryContext) { + TRY_AWAIT(prepareRetry(*retryContext, timings.waitTime, timings.attempt)); + co_return TRY_AWAIT(retry(timings.downloadTimeout)); + } else { + co_return TRY_AWAIT(initial()); + } + } catch (FileTransferError & e) { + auto next = backoff.next(); + // If this is a transient error, then maybe retry after a while. after any + // bytes have been received we require range support to proceed, otherwise + // we'd need to start from scratch and discard everything we already have. + if (e.error != Transient || data.has_value() || !next.has_value() + || (totalReceived > 0 && !transfer->acceptsRanges())) + { + throw; + } + retryContext = e.what(); + timings = *next; + } + } + } catch (...) { + co_return result::current_exception(); + } + + kj::Promise> startTransfer( + const std::string & uri, + const std::chrono::milliseconds & timeout, + curl_off_t offset = 0 + ); + + kj::Promise> prepareRetry( + const std::string & context, + const std::chrono::milliseconds & waitTime, + unsigned int attempt + ); + + kj::Promise> restartTransfer(const std::chrono::milliseconds & timeout); + + kj::Promise> waitForData(); + + kj::Promise> restartAndWaitForData(const std::chrono::milliseconds & timeout); + + kj::Promise> awaitData(); + + kj::Promise>> read(void * buffer, size_t len) override; + }; + + // Fields. +public: + std::shared_ptr multi; + + // Specials. +public: + curlFileTransfer(unsigned int baseRetryTimeMs); + ~curlFileTransfer(); + +#if ENABLE_S3 + using S3Uri = std::tuple; + static auto parseS3Uri(std::string url) -> S3Uri; +#endif + + // Overrides. +public: + auto upload( + std::string const & uri, + std::string data, + FileTransferOptions options, + Activity const * context + ) -> Async override; + + auto exists( + std::string const & uri, + FileTransferOptions options, + Activity const * context + ) -> Async override; + + auto download( + std::string const & uri, + FileTransferOptions options, + Activity const * context = nullptr + ) -> Async>> override; + + // Actual API. +public: + auto tryEagerTransfers( + std::string const & url, + FileTransferOptions const & options, + std::optional const & data, + bool noBody + ) -> Async>>>; + + auto enqueueFileTransfer( + std::string const & uri, + FileTransferOptions && options, + std::optional data, + bool noBody, + Activity const * context + ) -> Async>>; +}; + +} diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index 2ab4eabac..53c547598 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -1,5 +1,6 @@ #include "lix/libstore/curlmulti.hh" #include "lix/libstore/filetransfer.hh" +#include "lix/libstore/curlfiletransfer.hh" #include "lix/libutil/file-system.hh" #include "lix/libstore/transferitem.hh" #include "lix/libutil/async-io.hh" @@ -49,492 +50,6 @@ namespace nix { FileTransferSettings fileTransferSettings; - -struct curlFileTransfer : public FileTransfer -{ - std::shared_ptr multi; - - curlFileTransfer(unsigned int baseRetryTimeMs) - : multi(std::make_shared(baseRetryTimeMs)) - { - } - - ~curlFileTransfer() - { - multi->stopWorkerThread(); - } - -#if ENABLE_S3 - static std::tuple parseS3Uri(std::string uri) - { - auto [path, params] = splitUriAndParams(uri); - - auto slash = path.find('/', 5); // 5 is the length of "s3://" prefix - if (slash == std::string::npos) - throw nix::Error("bad S3 URI '%s'", path); - - std::string bucketName(path, 5, slash - 5); - std::string key(path, slash + 1); - - return {bucketName, key, params}; - } -#endif - - kj::Promise> upload( - const std::string & uri, - std::string data, - FileTransferOptions options, - const Activity * context - ) override - try { - TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::move(data), false, context)); - co_return result::success(); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise>>>> - tryEagerTransfers( - const std::string & uri, - const FileTransferOptions & options, - const std::optional & data, - bool noBody - ) - try { - // curl transfers using file:// urls cannot be paused, and are a bit unruly - // in other ways too. since their metadata is trivial and we already have a - // backend for simple file system reads we can use that instead. we'll pass - // uploads to files to curl even so, those will fail in enqueueItem anyway. - // on all other decoding failures we also let curl fail for us a bit later. - // - // note that we use kj to decode the url, not curl. curl uses only the path - // component of the url to determine the file name, but it does note expose - // the decoding method it uses for this. for file:// transfers curl forbids - // only \0 characters in the urldecoded path, not all control characters as - // it does in the public curl_url_get(CURLUPART_PATH, CURLU_URLDECODE) api. - // - // also note: everything weird you see here is for compatibility with curl. - // we can't even fix it because nix-channel relies on this. even reading of - // directories being allowed and returning something (though hopefully it's - // enough to return anything instead of a directory listing like curl does) - if (uri.starts_with("file://") && !data.has_value()) { - if (!uri.starts_with("file:///")) { - throw FileTransferError(NotFound, std::nullopt, "file not found"); - } - auto url = curl_url(); - if (!url) { - throw std::bad_alloc(); - } - KJ_DEFER(curl_url_cleanup(url)); - curl_url_set(url, CURLUPART_URL, requireCString(uri), 0); - char * path = nullptr; - curl_url_get(url, CURLUPART_PATH, &path, 0); - auto decoded = kj::decodeUriComponent(kj::arrayPtr(path, path + strlen(path))); - if (!decoded.hadErrors && decoded.findFirst(0) == nullptr) { - Path fsPath(decoded.cStr(), decoded.size()); - FileTransferResult metadata{.effectiveUri = std::string("file://") + path}; - struct stat st; - AutoCloseFD fd(sys::open(fsPath, O_RDONLY)); - if (!fd || fstat(fd.get(), &st) != 0) { - throw FileTransferError( - NotFound, std::nullopt, "%s: file not found (%s)", fsPath, strerror(errno) - ); - } - if (S_ISDIR(st.st_mode)) { - co_return std::pair{ - std::move(metadata), make_box_ptr("") - }; - } - struct OwningFdStream : AsyncInputStream - { - AutoCloseFD fd; - OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {} - kj::Promise>> - 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) { - if (got == 0) { - return {result::success(std::nullopt)}; - } else { - return {result::success(got)}; - } - } else { - return {result::failure(std::make_exception_ptr(SysError("reading file") - ))}; - } - } - }; - co_return std::pair{ - std::move(metadata), make_box_ptr(std::move(fd)) - }; - } - } - - /* Ugly hack to support s3:// URIs. */ - if (uri.starts_with("s3://")) { - // FIXME: do this on a worker thread -#if ENABLE_S3 - auto [bucketName, key, params] = parseS3Uri(uri); - - std::string profile = getOr(params, "profile", ""); - std::string region = getOr(params, "region", Aws::Region::US_EAST_1); - std::string scheme = getOr(params, "scheme", ""); - std::string endpoint = getOr(params, "endpoint", ""); - - S3Helper s3Helper(profile, region, scheme, endpoint); - - // FIXME: implement ETag - auto s3Res = TRY_AWAIT(s3Helper.getObject(bucketName, key)); - FileTransferResult res; - if (!s3Res.data) - throw FileTransferError(NotFound, "S3 object '%s' does not exist", uri); - struct OwningStringStream : private std::string, AsyncStringInputStream - { - OwningStringStream(std::string data) - : std::string(std::move(data)) - , AsyncStringInputStream(*this) - { - } - }; - co_return std::pair{res, make_box_ptr(std::move(*s3Res.data))}; -#else - throw nix::Error( - "cannot download '%s' because Lix is not built with S3 support", uri - ); -#endif - } - - co_return std::nullopt; - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise>>> - enqueueFileTransfer( - const std::string & uri, - FileTransferOptions && options, - std::optional data, - bool noBody, - const Activity * context - ) - try { - if (auto eager = TRY_AWAIT(tryEagerTransfers(uri, options, data, noBody))) { - co_return std::move(*eager); - } - - auto source = make_box_ptr( - *this, uri, std::move(options), std::move(data), noBody, context - ); - TRY_AWAIT(source->init()); - TRY_AWAIT(source->awaitData()); - co_return {source->metadata, std::move(source)}; - } catch (...) { - co_return result::current_exception(); - } - - struct TransferStream : AsyncInputStream - { - std::shared_ptr parent; - std::string uri; - FileTransferOptions options; - std::optional data; - bool noBody; - const Activity * parentAct; - - std::shared_ptr transfer; - FileTransferResult metadata; - std::string chunk; - std::string_view buffered; - - const size_t tries = fileTransferSettings.tries; - curl_off_t totalReceived = 0; - - Generator backoff; - - TransferStream( - curlFileTransfer & parent, - const std::string & uri, - FileTransferOptions && options, - std::optional data, - bool noBody, - const Activity * context - ) - : parent(parent.multi) - , uri(uri) - , options(options) - , data(std::move(data)) - , noBody(noBody) - , parentAct(context) - , backoff(backoffTimeouts( - fileTransferSettings.tries, - std::chrono::seconds(fileTransferSettings.maxConnectTimeout.get()), - std::chrono::seconds(fileTransferSettings.initialConnectTimeout.get()), - std::chrono::milliseconds(this->parent->baseRetryTimeMs) - )) - { - } - - ~TransferStream() - { - // wake up the download thread if it's still going and have it abort - try { - if (transfer) { - parent->cancel(transfer); - } - } catch (...) { - ignoreExceptionInDestructor(); - } - } - - kj::Promise> init() - try { - metadata = TRY_AWAIT(withRetries( - [&]() { - return startTransfer( - uri, std::chrono::seconds(fileTransferSettings.initialConnectTimeout.get()) - ); - }, - [&](const std::chrono::milliseconds & timeout) { - return startTransfer(uri, timeout); - } - )); - co_return result::success(); - } catch (...) { - co_return result::current_exception(); - } - - auto withRetries(auto && initial, auto && retry) -> decltype(initial()) - try { - std::optional retryContext; - BackoffTiming timings; - while (true) { - try { - if (retryContext) { - TRY_AWAIT(prepareRetry(*retryContext, timings.waitTime, timings.attempt)); - co_return TRY_AWAIT(retry(timings.downloadTimeout)); - } else { - co_return TRY_AWAIT(initial()); - } - } catch (FileTransferError & e) { - auto next = backoff.next(); - // If this is a transient error, then maybe retry after a while. after any - // bytes have been received we require range support to proceed, otherwise - // we'd need to start from scratch and discard everything we already have. - if (e.error != Transient || data.has_value() || !next.has_value() - || (totalReceived > 0 && !transfer->acceptsRanges())) - { - throw; - } - retryContext = e.what(); - timings = *next; - } - } - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> startTransfer( - const std::string & uri, - const std::chrono::milliseconds & timeout, - curl_off_t offset = 0 - ) - try { - auto uploadData = data ? std::optional(std::string_view(*data)) : std::nullopt; - auto pfp = kj::newPromiseAndCrossThreadFulfiller>(); - transfer = std::make_shared( - uri, - std::move(options), - parentAct, - uploadData, - noBody, - offset, - std::move(pfp.fulfiller), - timeout - ); - parent->enqueueItem(transfer); - co_return TRY_AWAIT(pfp.promise); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> prepareRetry( - const std::string & context, - const std::chrono::milliseconds & waitTime, - unsigned int attempt - ) - try { - if (totalReceived) { - printTaggedWarning( - "%s; retrying from offset %d in %d ms (attempt %d/%d)", - Uncolored(context), - totalReceived, - waitTime.count(), - Uncolored(attempt), - Uncolored(tries) - ); - } else { - printTaggedWarning( - "%s; retrying in %d ms (attempt %d/%d)", - Uncolored(context), - waitTime.count(), - Uncolored(attempt), - Uncolored(tries) - ); - } - - co_await AIO().provider.getTimer().afterDelay(waitTime.count() * kj::MILLISECONDS); - co_return result::success(); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> restartTransfer(const std::chrono::milliseconds & timeout) - try { - auto onChange = - [&](std::string_view what, std::string_view from, std::string_view to, bool throw_ - ) -> void { - if (!from.empty() && from != to) { - FileTransferError e = FileTransferError( - Misc, - {}, - "uri %s changed %s from %s to %s during transfer", - uri, - what, - from, - to - ); - - if (throw_) { - throw e; - } - - logWarning(e.info()); - } - }; - - // 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 = TRY_AWAIT(startTransfer(uri, timeout, totalReceived)); - - onChange("final destination", metadata.effectiveUri, newMeta.effectiveUri, false); - onChange("ETag", metadata.etag, newMeta.etag, true); - onChange( - "immutable url", - metadata.immutableUrl.value_or(""), - newMeta.immutableUrl.value_or(""), - true - ); - co_return result::success(); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> waitForData() - try { - /* Grab data if available, otherwise wait for the download - thread to wake us up. */ - std::optional> signal; - - while (buffered.empty()) { - if (signal) { - co_await *signal; - signal.reset(); - } - - auto state(transfer->downloadState.lock()); - - if (!state->data.empty()) { - chunk = std::move(state->data); - buffered = chunk; - totalReceived += chunk.size(); - parent->unpause(transfer); - } else if (state->exc) { - std::rethrow_exception(state->exc); - } else if (state->done) { - co_return false; - } else { - parent->unpause(transfer); - signal = state->wait(); - } - } - - co_return true; - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> restartAndWaitForData(const std::chrono::milliseconds & timeout) - try { - TRY_AWAIT(restartTransfer(timeout)); - co_return TRY_AWAIT(waitForData()); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise> awaitData() - try { - co_return TRY_AWAIT(withRetries( - [&] { return waitForData(); }, - [&](const std::chrono::milliseconds & timeout) { - return restartAndWaitForData(timeout); - } - )); - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise>> read(void * buffer, size_t len) override - try { - TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len)); - - size_t total = 0; - auto data = static_cast(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) { - co_return std::nullopt; - } else { - co_return total; - } - } catch (...) { - co_return result::current_exception(); - } - }; - - kj::Promise> - exists(const std::string & uri, FileTransferOptions options, const Activity * context) override - try { - try { - TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::nullopt, true, context)); - 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) - co_return false; - throw; - } - } catch (...) { - co_return result::current_exception(); - } - - kj::Promise>>> download( - const std::string & uri, FileTransferOptions options, const Activity * context - ) override - { - return enqueueFileTransfer(uri, std::move(options), std::nullopt, false, context); - } -}; - ref makeCurlFileTransfer(std::optional baseRetryTimeMs) { return make_ref(baseRetryTimeMs.value_or(250)); diff --git a/lix/libstore/meson.build b/lix/libstore/meson.build index a6ddaed11..4e3f99e83 100644 --- a/lix/libstore/meson.build +++ b/lix/libstore/meson.build @@ -194,6 +194,7 @@ liblix_sources += files( 'common-protocol.cc', 'content-address.cc', 'crypto.cc', + 'curlfiletransfer.cc', 'curlmulti.cc', 'daemon.cc', 'derivations.cc', @@ -262,6 +263,7 @@ libstore_headers = files( 'common-protocol.hh', 'content-address.hh', 'crypto.hh', + 'curlfiletransfer.hh', 'curlmulti.hh', 'daemon.hh', 'derivations.hh',