From b460a7a38b5316ea4885247693347a1aacdded6f Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Fri, 24 Oct 2025 00:16:15 +0200 Subject: [PATCH] libstore: move curlFileTransfer::TransferItem up we have anonymous namespaces and no other scoping requirements, we can move it out. this will also make it easier to move *other* things out, such as the transport streams and all the shared multi state handling. Change-Id: Iafa1eaeb69e096e98c4667f0531b17d5fe7c78a6 --- lix/libstore/filetransfer.cc | 871 ++++++++++++++++++----------------- 1 file changed, 459 insertions(+), 412 deletions(-) diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index 07f202644..d1268d98a 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -50,468 +50,514 @@ FileTransferSettings fileTransferSettings; static GlobalConfig::Register rFileTransferSettings(&fileTransferSettings); -struct curlFileTransfer : public FileTransfer +namespace { +struct TransferItem { - std::unique_ptr curlm; - - const unsigned int baseRetryTimeMs; - - struct TransferItem + struct DownloadState { - struct DownloadState + bool done = false; + std::exception_ptr exc; + std::string data; + std::optional>> downloadEvent; + + auto wait() { - bool done = false; - std::exception_ptr exc; - std::string data; - std::optional>> downloadEvent; + auto pfp = kj::newPromiseAndCrossThreadFulfiller(); + downloadEvent = std::move(pfp.fulfiller); + return std::move(pfp.promise); + } - auto wait() + void signal() + { + if (downloadEvent) { + (*downloadEvent)->fulfill(); + downloadEvent.reset(); + } + } + }; + + std::string uri; + FileTransferResult result; + Activity act; + std::unique_ptr uploadData; + Sync downloadState; + bool headersDone = false, metadataReturned = false; + kj::Own>> metadataPromise; + std::string statusMsg; + + uint64_t bodySize = 0; + + std::unique_ptr requestHeaders; + std::unique_ptr req; + // buffer to accompany the `req` above + char errbuf[CURL_ERROR_SIZE]; + + inline static const std::set + successfulStatuses{200, 201, 204, 206, 304, 0 /* other protocol */}; + + std::optional httpStatusCode; + + /* Get the scheme for the current curl handle, or none if curl returns NULL. + * Ensures the scheme is always casefolded to lowercase */ + std::optional getCurlScheme() + { + char * scheme_raw = nullptr; + if (curl_easy_getinfo(req.get(), CURLINFO_SCHEME, &scheme_raw) != CURLE_OK) { + throw nix::Error("could not get scheme used from curl handle"); + } + if (scheme_raw) { + return toLower(std::string(scheme_raw)); + } else { + return {}; + } + } + + /* Get the HTTP status code, or 0 for other protocols. */ + long getHTTPStatus() + { + if (httpStatusCode) { + return *httpStatusCode; + } + + long statusCode = 0; + + std::optional scheme = getCurlScheme(); + if (scheme == "http" || scheme == "https") { + if (curl_easy_getinfo(req.get(), CURLINFO_RESPONSE_CODE, &statusCode) != CURLE_OK) { + throw nix::Error("could not get response code from curl handle"); + } + } + + httpStatusCode = statusCode; + return statusCode; + } + + std::string verb() const + { + return uploadData ? "upload" : "download"; + } + + TransferItem( + const std::string & uri, + FileTransferOptions && options, + const Activity * parentAct, + std::optional uploadData, + bool noBody, + curl_off_t writtenToSink, + kj::Own>> metadataPromise, + const std::chrono::milliseconds & connectTimeout + ) + : uri(uri) + , act(logger->startActivity( + lvlTalkative, + actFileTransfer, + fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri), + {uri}, + parentAct + )) + , metadataPromise(std::move(metadataPromise)) + , req(curl_easy_init()) + { + if (req == nullptr) { + throw FileTransferError(FileTransfer::Misc, {}, "could not allocate curl handle"); + } + for (auto it = options.headers.begin(); it != options.headers.end(); ++it) { + if (auto next = curl_slist_append( + requestHeaders.get(), requireCString(fmt("%s: %s", it->first, it->second)) + ); + next != nullptr) { - auto pfp = kj::newPromiseAndCrossThreadFulfiller(); - downloadEvent = std::move(pfp.fulfiller); - return std::move(pfp.promise); - } - - void signal() - { - if (downloadEvent) { - (*downloadEvent)->fulfill(); - downloadEvent.reset(); - } - } - }; - - std::string uri; - FileTransferResult result; - Activity act; - std::unique_ptr uploadData; - Sync downloadState; - bool headersDone = false, metadataReturned = false; - kj::Own>> metadataPromise; - std::string statusMsg; - - uint64_t bodySize = 0; - - std::unique_ptr requestHeaders; - std::unique_ptr req; - // buffer to accompany the `req` above - char errbuf[CURL_ERROR_SIZE]; - - inline static const std::set successfulStatuses {200, 201, 204, 206, 304, 0 /* other protocol */}; - - std::optional httpStatusCode; - - /* Get the scheme for the current curl handle, or none if curl returns NULL. - * Ensures the scheme is always casefolded to lowercase */ - std::optional getCurlScheme() { - char *scheme_raw = nullptr; - if (curl_easy_getinfo(req.get(), CURLINFO_SCHEME, &scheme_raw) != CURLE_OK) { - throw nix::Error("could not get scheme used from curl handle"); - } - if (scheme_raw) { - return toLower(std::string(scheme_raw)); + (void) requestHeaders.release(); // next now owns this pointer + requestHeaders.reset(next); } else { - return {}; + throw FileTransferError( + FileTransfer::Misc, {}, "could not allocate curl request headers" + ); } } - /* Get the HTTP status code, or 0 for other protocols. */ - long getHTTPStatus() - { - if (httpStatusCode) { - return *httpStatusCode; - } - - long statusCode = 0; - - std::optional scheme = getCurlScheme(); - if (scheme == "http" || scheme == "https") { - if (curl_easy_getinfo(req.get(), CURLINFO_RESPONSE_CODE, &statusCode) != CURLE_OK) { - throw nix::Error("could not get response code from curl handle"); - } - } - - httpStatusCode = statusCode; - return statusCode; + if (verbosity >= lvlVomit) { + curl_easy_setopt(req.get(), CURLOPT_VERBOSE, 1); + curl_easy_setopt(req.get(), CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback); } - std::string verb() const - { - return uploadData ? "upload" : "download"; + curl_easy_setopt(req.get(), CURLOPT_URL, uri.c_str()); + curl_easy_setopt(req.get(), CURLOPT_FOLLOWLOCATION, 1L); + curl_easy_setopt(req.get(), CURLOPT_ACCEPT_ENCODING, ""); // all of them! + curl_easy_setopt(req.get(), CURLOPT_MAXREDIRS, 10); + curl_easy_setopt(req.get(), CURLOPT_NOSIGNAL, 1); + curl_easy_setopt( + req.get(), + CURLOPT_USERAGENT, + ("curl/" LIBCURL_VERSION " Lix/" + nixVersion + + (fileTransferSettings.userAgentSuffix != "" + ? " " + fileTransferSettings.userAgentSuffix.get() + : "")) + .c_str() + ); + curl_easy_setopt(req.get(), CURLOPT_PIPEWAIT, 1); + if (fileTransferSettings.enableHttp2) { + curl_easy_setopt(req.get(), CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); + } else { + curl_easy_setopt(req.get(), CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); + } + curl_easy_setopt(req.get(), CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper); + curl_easy_setopt(req.get(), CURLOPT_WRITEDATA, this); + curl_easy_setopt(req.get(), CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper); + curl_easy_setopt(req.get(), CURLOPT_HEADERDATA, this); + + curl_easy_setopt(req.get(), CURLOPT_XFERINFOFUNCTION, progressCallbackWrapper); + curl_easy_setopt(req.get(), CURLOPT_PROGRESSDATA, this); + curl_easy_setopt(req.get(), CURLOPT_NOPROGRESS, 0); + + curl_easy_setopt(req.get(), CURLOPT_ERRORBUFFER, errbuf); + errbuf[0] = 0; + + curl_easy_setopt(req.get(), CURLOPT_PROTOCOLS_STR, "http,https,ftp,ftps"); + + curl_easy_setopt(req.get(), CURLOPT_HTTPHEADER, requestHeaders.get()); + + if (settings.downloadSpeed.get() > 0) { + curl_easy_setopt( + req.get(), + CURLOPT_MAX_RECV_SPEED_LARGE, + (curl_off_t) (settings.downloadSpeed.get() * 1024) + ); } - TransferItem( - const std::string & uri, - FileTransferOptions && options, - const Activity * parentAct, - std::optional uploadData, - bool noBody, - curl_off_t writtenToSink, - kj::Own>> metadataPromise, - const std::chrono::milliseconds & connectTimeout - ) - : uri(uri) - , act(logger->startActivity( - lvlTalkative, - actFileTransfer, - fmt(uploadData ? "uploading '%s'" : "downloading '%s'", uri), - {uri}, - parentAct - )) - , metadataPromise(std::move(metadataPromise)) - , req(curl_easy_init()) - { - if (req == nullptr) { - throw FileTransferError(Misc, {}, "could not allocate curl handle"); - } - for (auto it = options.headers.begin(); it != options.headers.end(); ++it){ - if (auto next = curl_slist_append( - requestHeaders.get(), requireCString(fmt("%s: %s", it->first, it->second)) - ); - next != nullptr) - { - (void) requestHeaders.release(); // next now owns this pointer - requestHeaders.reset(next); - } else { - throw FileTransferError(Misc, {}, "could not allocate curl request headers"); - } - } - - if (verbosity >= lvlVomit) { - curl_easy_setopt(req.get(), CURLOPT_VERBOSE, 1); - curl_easy_setopt(req.get(), CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback); - } - - curl_easy_setopt(req.get(), CURLOPT_URL, uri.c_str()); - curl_easy_setopt(req.get(), CURLOPT_FOLLOWLOCATION, 1L); - curl_easy_setopt(req.get(), CURLOPT_ACCEPT_ENCODING, ""); // all of them! - curl_easy_setopt(req.get(), CURLOPT_MAXREDIRS, 10); - curl_easy_setopt(req.get(), CURLOPT_NOSIGNAL, 1); - curl_easy_setopt(req.get(), CURLOPT_USERAGENT, - ("curl/" LIBCURL_VERSION " Lix/" + nixVersion + - (fileTransferSettings.userAgentSuffix != "" ? " " + fileTransferSettings.userAgentSuffix.get() : "")).c_str()); - curl_easy_setopt(req.get(), CURLOPT_PIPEWAIT, 1); - if (fileTransferSettings.enableHttp2) - curl_easy_setopt(req.get(), CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS); - else - curl_easy_setopt(req.get(), CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1); - curl_easy_setopt(req.get(), CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper); - curl_easy_setopt(req.get(), CURLOPT_WRITEDATA, this); - curl_easy_setopt(req.get(), CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper); - curl_easy_setopt(req.get(), CURLOPT_HEADERDATA, this); - - curl_easy_setopt(req.get(), CURLOPT_XFERINFOFUNCTION, progressCallbackWrapper); - curl_easy_setopt(req.get(), CURLOPT_PROGRESSDATA, this); - curl_easy_setopt(req.get(), CURLOPT_NOPROGRESS, 0); - - curl_easy_setopt(req.get(), CURLOPT_ERRORBUFFER, errbuf); - errbuf[0] = 0; - - curl_easy_setopt(req.get(), CURLOPT_PROTOCOLS_STR, "http,https,ftp,ftps"); - - curl_easy_setopt(req.get(), CURLOPT_HTTPHEADER, requestHeaders.get()); - - if (settings.downloadSpeed.get() > 0) - curl_easy_setopt(req.get(), CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t) (settings.downloadSpeed.get() * 1024)); - - if (noBody) - curl_easy_setopt(req.get(), CURLOPT_NOBODY, 1); - - if (uploadData) { - this->uploadData.reset(fmemopen(const_cast(uploadData->data()), uploadData->size(), "r")); - curl_easy_setopt(req.get(), CURLOPT_UPLOAD, 1L); - curl_easy_setopt(req.get(), CURLOPT_READDATA, this->uploadData.get()); - curl_easy_setopt(req.get(), CURLOPT_INFILESIZE_LARGE, (curl_off_t) uploadData->length()); - } - - if (settings.caFile != "") - curl_easy_setopt(req.get(), CURLOPT_CAINFO, settings.caFile.get().c_str()); - - curl_easy_setopt(req.get(), CURLOPT_CONNECTTIMEOUT_MS, connectTimeout.count()); - - curl_easy_setopt(req.get(), CURLOPT_LOW_SPEED_LIMIT, 1L); - curl_easy_setopt(req.get(), CURLOPT_LOW_SPEED_TIME, fileTransferSettings.stalledDownloadTimeout.get()); - - /* If no file exist in the specified path, curl continues to work - anyway as if netrc support was disabled. */ - curl_easy_setopt(req.get(), CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); - curl_easy_setopt(req.get(), CURLOPT_NETRC, CURL_NETRC_OPTIONAL); - - if (writtenToSink) - curl_easy_setopt(req.get(), CURLOPT_RESUME_FROM_LARGE, writtenToSink); - - if (options.extraSetup) { - options.extraSetup(req.get()); - } + if (noBody) { + curl_easy_setopt(req.get(), CURLOPT_NOBODY, 1); } - bool acceptsRanges() - { - curl_header * h; - if (curl_easy_header(req.get(), "accept-ranges", 0, CURLH_HEADER, -1, &h)) { - // treat any error as the remote not accepting range requests. the only - // interesting local error is out-of-memory, which we can't even handle - return false; - } - - return toLower(trim(h->value)) == "bytes"; + if (uploadData) { + this->uploadData.reset( + fmemopen(const_cast(uploadData->data()), uploadData->size(), "r") + ); + curl_easy_setopt(req.get(), CURLOPT_UPLOAD, 1L); + curl_easy_setopt(req.get(), CURLOPT_READDATA, this->uploadData.get()); + curl_easy_setopt( + req.get(), CURLOPT_INFILESIZE_LARGE, (curl_off_t) uploadData->length() + ); } - void failEx(std::exception_ptr ex) - { - auto state = downloadState.lock(); - assert(!state->done && !state->exc); - if (!metadataReturned) { - metadataPromise->fulfill(ex); - } - state->exc = ex; - state->signal(); + if (settings.caFile != "") { + curl_easy_setopt(req.get(), CURLOPT_CAINFO, settings.caFile.get().c_str()); } - template - void fail(T && e) - { - failEx(std::make_exception_ptr(std::forward(e))); + curl_easy_setopt(req.get(), CURLOPT_CONNECTTIMEOUT_MS, connectTimeout.count()); + + curl_easy_setopt(req.get(), CURLOPT_LOW_SPEED_LIMIT, 1L); + curl_easy_setopt( + req.get(), CURLOPT_LOW_SPEED_TIME, fileTransferSettings.stalledDownloadTimeout.get() + ); + + /* If no file exist in the specified path, curl continues to work + anyway as if netrc support was disabled. */ + curl_easy_setopt(req.get(), CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str()); + curl_easy_setopt(req.get(), CURLOPT_NETRC, CURL_NETRC_OPTIONAL); + + if (writtenToSink) { + curl_easy_setopt(req.get(), CURLOPT_RESUME_FROM_LARGE, writtenToSink); } - void maybeFinishSetup() - { - if (headersDone) { - return; - } + if (options.extraSetup) { + options.extraSetup(req.get()); + } + } - auto status = getHTTPStatus(); - - char * effectiveUriCStr = nullptr; - curl_easy_getinfo(req.get(), CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); - if (effectiveUriCStr) { - result.effectiveUri = effectiveUriCStr; - } - - result.cached = status == 304; - if (successfulStatuses.contains(status)) { - metadataPromise->fulfill(result); - metadataReturned = true; - } - - headersDone = true; + bool acceptsRanges() + { + curl_header * h; + if (curl_easy_header(req.get(), "accept-ranges", 0, CURLH_HEADER, -1, &h)) { + // treat any error as the remote not accepting range requests. the only + // interesting local error is out-of-memory, which we can't even handle + return false; } - std::exception_ptr callbackException; + return toLower(trim(h->value)) == "bytes"; + } - size_t writeCallback(void * contents, size_t size, size_t nmemb) - { - const size_t realSize = size * nmemb; + void failEx(std::exception_ptr ex) + { + auto state = downloadState.lock(); + assert(!state->done && !state->exc); + if (!metadataReturned) { + metadataPromise->fulfill(ex); + } + state->exc = ex; + state->signal(); + } - try { - maybeFinishSetup(); + template + void fail(T && e) + { + failEx(std::make_exception_ptr(std::forward(e))); + } - auto state = downloadState.lock(); - - // when the buffer is full (as determined by a historical magic value) we - // pause the transfer and wait for the receiver to unpause it when ready. - if (successfulStatuses.count(getHTTPStatus()) && state->data.size() > 1024 * 1024) { - return CURL_WRITEFUNC_PAUSE; - } - - state->data.append(static_cast(contents), realSize); - state->signal(); - bodySize += realSize; - return realSize; - } catch (...) { - callbackException = std::current_exception(); - return CURL_WRITEFUNC_ERROR; - } + void maybeFinishSetup() + { + if (headersDone) { + return; } - static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return static_cast(userp)->writeCallback(contents, size, nmemb); + auto status = getHTTPStatus(); + + char * effectiveUriCStr = nullptr; + curl_easy_getinfo(req.get(), CURLINFO_EFFECTIVE_URL, &effectiveUriCStr); + if (effectiveUriCStr) { + result.effectiveUri = effectiveUriCStr; } - size_t headerCallback(void * contents, size_t size, size_t nmemb) + result.cached = status == 304; + if (successfulStatuses.contains(status)) { + metadataPromise->fulfill(result); + metadataReturned = true; + } + + headersDone = true; + } + + std::exception_ptr callbackException; + + size_t writeCallback(void * contents, size_t size, size_t nmemb) + { + const size_t realSize = size * nmemb; + try { - size_t realSize = size * nmemb; - std::string line(static_cast(contents), realSize); - printMsg(lvlVomit, "got header for '%s': %s", uri, trim(line)); + maybeFinishSetup(); - static std::regex statusLine = regex::parse("HTTP/[^ ]+ +[0-9]+(.*)", std::regex::extended | std::regex::icase); - if (std::smatch match; std::regex_match(line, match, statusLine)) { - statusMsg = trim(match.str(1)); - } else { - auto i = line.find(':'); - if (i != std::string::npos) { - std::string name = toLower(trim(line.substr(0, i))); + auto state = downloadState.lock(); - if (name == "etag") { - // NOTE we don't check that the etag hasn't gone *missing*. technically - // this is not an error as long as we get the same data from the remote. - auto etag = trim(line.substr(i + 1)); - result.etag = std::move(etag); - } - - else if (name == "link" || name == "x-amz-meta-link") { - auto value = trim(line.substr(i + 1)); - static std::regex linkRegex = regex::parse("<([^>]*)>; rel=\"immutable\"", std::regex::extended | std::regex::icase); - if (std::smatch match; std::regex_match(value, match, linkRegex)) { - result.immutableUrl = match.str(1); - } else - debug("got invalid link header '%s'", value); - } - } + // when the buffer is full (as determined by a historical magic value) we + // pause the transfer and wait for the receiver to unpause it when ready. + if (successfulStatuses.count(getHTTPStatus()) && state->data.size() > 1024 * 1024) { + return CURL_WRITEFUNC_PAUSE; } + + state->data.append(static_cast(contents), realSize); + state->signal(); + bodySize += realSize; return realSize; } catch (...) { callbackException = std::current_exception(); return CURL_WRITEFUNC_ERROR; } + } - static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) - { - return static_cast(userp)->headerCallback(contents, size, nmemb); - } + static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return static_cast(userp)->writeCallback(contents, size, nmemb); + } - int progressCallback(curl_off_t dltotal, curl_off_t dlnow) - { - try { - if (act.progress(dlnow, dltotal) == Logger::BufferState::NeedsFlush) { - act.getLogger().waitForSpace(); // NOLINT(lix-never-async) - } - } catch (nix::Interrupted &) { - } - return isInterrupted(); - } + size_t headerCallback(void * contents, size_t size, size_t nmemb) + try { + size_t realSize = size * nmemb; + std::string line(static_cast(contents), realSize); + printMsg(lvlVomit, "got header for '%s': %s", uri, trim(line)); - static int progressCallbackWrapper(void * userp, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow) - { - return static_cast(userp)->progressCallback(dltotal, dlnow); - } + static std::regex statusLine = + regex::parse("HTTP/[^ ]+ +[0-9]+(.*)", std::regex::extended | std::regex::icase); + if (std::smatch match; std::regex_match(line, match, statusLine)) { + statusMsg = trim(match.str(1)); + } else { + auto i = line.find(':'); + if (i != std::string::npos) { + std::string name = toLower(trim(line.substr(0, i))); - static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) - { - if (type == CURLINFO_TEXT) - vomit("curl: %s", chomp(std::string(data, size))); - return 0; - } - - void finish(CURLcode code) - { - auto httpStatus = getHTTPStatus(); - - maybeFinishSetup(); - - debug("finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", - verb(), uri, code, httpStatus, bodySize); - - if (callbackException) - failEx(callbackException); - - else if (code == CURLE_OK && successfulStatuses.count(httpStatus)) - { - if (act.progress(bodySize, bodySize) == Logger::BufferState::NeedsFlush) { - act.getLogger().waitForSpace(); // NOLINT(lix-never-async) - } - auto state = downloadState.lock(); - state->done = true; - state->signal(); - } - - else { - // We treat most errors as transient, but won't retry when hopeless - Error err = Transient; - - if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { - // The file is definitely not there - err = NotFound; - } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { - // Don't retry on authentication/authorization failures - err = Forbidden; - } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 && httpStatus != 429) { - // Most 4xx errors are client errors and are probably not worth retrying: - // * 408 means the server timed out waiting for us, so we try again - // * 429 means too many requests, so we retry (with a delay) - err = Misc; - } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { - // Let's treat most 5xx (server) errors as transient, except for a handful: - // * 501 not implemented - // * 505 http version not supported - // * 511 we're behind a captive portal - err = Misc; - } else { - // Don't bother retrying on certain cURL errors either - - // Allow selecting a subset of enum values - #pragma GCC diagnostic push - #pragma GCC diagnostic ignored "-Wswitch-enum" - switch (code) { - case CURLE_FAILED_INIT: - case CURLE_URL_MALFORMAT: - case CURLE_NOT_BUILT_IN: - case CURLE_REMOTE_ACCESS_DENIED: - case CURLE_FILE_COULDNT_READ_FILE: - case CURLE_FUNCTION_NOT_FOUND: - case CURLE_ABORTED_BY_CALLBACK: - case CURLE_BAD_FUNCTION_ARGUMENT: - case CURLE_INTERFACE_FAILED: - case CURLE_UNKNOWN_OPTION: - case CURLE_SSL_CACERT_BADFILE: - case CURLE_TOO_MANY_REDIRECTS: - case CURLE_WRITE_ERROR: - case CURLE_UNSUPPORTED_PROTOCOL: - err = Misc; - break; - default: // Shut up warnings - break; - } - #pragma GCC diagnostic pop + if (name == "etag") { + // NOTE we don't check that the etag hasn't gone *missing*. technically + // this is not an error as long as we get the same data from the remote. + auto etag = trim(line.substr(i + 1)); + result.etag = std::move(etag); } - std::optional response; - if (!successfulStatuses.count(httpStatus)) - response = std::move(downloadState.lock()->data); - - auto textualError = [](const char * errbuf, CURLcode code) -> const char * { - if (errbuf && errbuf[0]) { - return errbuf; + else if (name == "link" || name == "x-amz-meta-link") + { + auto value = trim(line.substr(i + 1)); + static std::regex linkRegex = regex::parse( + "<([^>]*)>; rel=\"immutable\"", std::regex::extended | std::regex::icase + ); + if (std::smatch match; std::regex_match(value, match, linkRegex)) { + result.immutableUrl = match.str(1); } else { - return curl_easy_strerror(code); + debug("got invalid link header '%s'", value); } - }; - auto exc = code == CURLE_ABORTED_BY_CALLBACK && isInterrupted() - ? FileTransferError( - Interrupted, - std::move(response), - "%s of '%s' was interrupted", - verb(), - uri - ) - : httpStatus != 0 - ? FileTransferError( - err, - std::move(response), - "unable to %s '%s': HTTP error %d (%s)%s", - verb(), - uri, - httpStatus, - statusMsg, - code == CURLE_OK - ? "" - : fmt(" (curl error code=%d: %s)", code, textualError(errbuf, code)) - ) - : FileTransferError( - err, - std::move(response), - "unable to %s '%s': %s (curl error code=%d)", - verb(), - uri, - textualError(errbuf, code), - code - ); - - fail(std::move(exc)); + } } } - }; + return realSize; + } catch (...) { + callbackException = std::current_exception(); + return CURL_WRITEFUNC_ERROR; + } + + static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp) + { + return static_cast(userp)->headerCallback(contents, size, nmemb); + } + + int progressCallback(curl_off_t dltotal, curl_off_t dlnow) + { + try { + if (act.progress(dlnow, dltotal) == Logger::BufferState::NeedsFlush) { + act.getLogger().waitForSpace(); // NOLINT(lix-never-async) + } + } catch (nix::Interrupted &) { + } + return isInterrupted(); + } + + static int progressCallbackWrapper( + void * userp, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow + ) + { + return static_cast(userp)->progressCallback(dltotal, dlnow); + } + + static int + debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) + { + if (type == CURLINFO_TEXT) { + vomit("curl: %s", chomp(std::string(data, size))); + } + return 0; + } + + void finish(CURLcode code) + { + auto httpStatus = getHTTPStatus(); + + maybeFinishSetup(); + + debug( + "finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes", + verb(), + uri, + code, + httpStatus, + bodySize + ); + + if (callbackException) { + failEx(callbackException); + } + + else if (code == CURLE_OK && successfulStatuses.count(httpStatus)) + { + if (act.progress(bodySize, bodySize) == Logger::BufferState::NeedsFlush) { + act.getLogger().waitForSpace(); // NOLINT(lix-never-async) + } + auto state = downloadState.lock(); + state->done = true; + state->signal(); + } + + else + { + // We treat most errors as transient, but won't retry when hopeless + auto err = FileTransfer::Transient; + + if (httpStatus == 404 || httpStatus == 410 || code == CURLE_FILE_COULDNT_READ_FILE) { + // The file is definitely not there + err = FileTransfer::NotFound; + } else if (httpStatus == 401 || httpStatus == 403 || httpStatus == 407) { + // Don't retry on authentication/authorization failures + err = FileTransfer::Forbidden; + } else if (httpStatus >= 400 && httpStatus < 500 && httpStatus != 408 + && httpStatus != 429) + { + // Most 4xx errors are client errors and are probably not worth retrying: + // * 408 means the server timed out waiting for us, so we try again + // * 429 means too many requests, so we retry (with a delay) + err = FileTransfer::Misc; + } else if (httpStatus == 501 || httpStatus == 505 || httpStatus == 511) { + // Let's treat most 5xx (server) errors as transient, except for a handful: + // * 501 not implemented + // * 505 http version not supported + // * 511 we're behind a captive portal + err = FileTransfer::Misc; + } else { +// Don't bother retrying on certain cURL errors either + +// Allow selecting a subset of enum values +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wswitch-enum" + switch (code) { + case CURLE_FAILED_INIT: + case CURLE_URL_MALFORMAT: + case CURLE_NOT_BUILT_IN: + case CURLE_REMOTE_ACCESS_DENIED: + case CURLE_FILE_COULDNT_READ_FILE: + case CURLE_FUNCTION_NOT_FOUND: + case CURLE_ABORTED_BY_CALLBACK: + case CURLE_BAD_FUNCTION_ARGUMENT: + case CURLE_INTERFACE_FAILED: + case CURLE_UNKNOWN_OPTION: + case CURLE_SSL_CACERT_BADFILE: + case CURLE_TOO_MANY_REDIRECTS: + case CURLE_WRITE_ERROR: + case CURLE_UNSUPPORTED_PROTOCOL: + err = FileTransfer::Misc; + break; + default: // Shut up warnings + break; + } +#pragma GCC diagnostic pop + } + + std::optional response; + if (!successfulStatuses.count(httpStatus)) { + response = std::move(downloadState.lock()->data); + } + + auto textualError = [](const char * errbuf, CURLcode code) -> const char * { + if (errbuf && errbuf[0]) { + return errbuf; + } else { + return curl_easy_strerror(code); + } + }; + auto exc = code == CURLE_ABORTED_BY_CALLBACK && isInterrupted() + ? FileTransferError( + FileTransfer::Interrupted, + std::move(response), + "%s of '%s' was interrupted", + verb(), + uri + ) + : httpStatus != 0 + ? FileTransferError( + err, + std::move(response), + "unable to %s '%s': HTTP error %d (%s)%s", + verb(), + uri, + httpStatus, + statusMsg, + code == CURLE_OK + ? "" + : fmt(" (curl error code=%d: %s)", code, textualError(errbuf, code)) + ) + : FileTransferError( + err, + std::move(response), + "unable to %s '%s': %s (curl error code=%d)", + verb(), + uri, + textualError(errbuf, code), + code + ); + + fail(std::move(exc)); + } + } +}; + +struct curlFileTransfer : public FileTransfer +{ + std::unique_ptr curlm; + + const unsigned int baseRetryTimeMs; void unpause(const std::shared_ptr & transfer) { @@ -1189,6 +1235,7 @@ struct curlFileTransfer : public FileTransfer return enqueueFileTransfer(uri, std::move(options), std::nullopt, false, context); } }; +} ref makeCurlFileTransfer(std::optional baseRetryTimeMs) {