Ok = 200,

Created = 201,

NoContent = 204,

PartialContent = 206,

NotModified = 304,

Unauthorized = 401,

Forbidden = 403,

NotFound = 404,

ProxyAuthRequired = 407,

RequestTimeout = 408,

Gone = 410,

TooManyRequests = 429,

NotImplemented = 501,

ServiceUnavailable = 503,

HttpVersionNotSupported = 505,

NetworkAuthRequired = 511,

{

uint32_t backoff = clampedExponential(p.baseMs, p.attempt, p.ceilMs);

// Retry-After is a hard minimum — the server explicitly asked us to wait

// at least this long. ceilMs caps the backoff algorithm, not the server's

// signal: retrying before the server is ready just burns an attempt.

uint32_t floor = p.retryAfterMs.value_or(0);

if (!p.jitter)

return std::chrono::milliseconds(std::max(floor, backoff));

// Jitter spreads retries over [floor, floor+backoff] so that concurrent

// clients receiving the same Retry-After don't all retry simultaneously.

// Saturating add: clamp if floor + backoff would overflow uint32_t.

constexpr auto u32max = std::numeric_limits<uint32_t>::max();

uint32_t ceiling = (backoff > u32max - floor) ? u32max : floor + backoff;

if (ceiling <= floor)

return std::chrono::milliseconds(floor);

return std::chrono::milliseconds(std::uniform_int_distribution<uint32_t>(floor, ceiling)(rng));

{

/* Windows has no notion of a default location for the certificate bundles.

#ifndef _WIN32

for (auto & fn :

{"/etc/ssl/certs/ca-certificates.crt", "/nix/var/nix/profiles/default/etc/ssl/certs/ca-bundle.crt"})

if (pathAccessible(fn))

return fn;

#endif

return std::nullopt;

{

std::optional<AbsolutePath> sslOverride =

getEnvOs(OS_STR("NIX_SSL_CERT_FILE"))

.or_else([] { return getEnvOs(OS_STR("SSL_CERT_FILE")); })

.and_then([](OsString s) -> std::optional<OsString> {

return s.empty() ? std::nullopt : std::optional{std::move(s)};

})

.transform([](OsString s) { return AbsolutePath{std::filesystem::path{std::move(s)}}; });

if (sslOverride)

caFile = *sslOverride;

{

::CURLMcode code;

curlMultiError(::CURLMcode code)

: CloneableError{"unexpected curl multi error: %s", ::curl_multi_strerror(code)}

{

assert(code != CURLM_OK);

}

{

const auto * info = ::curl_version_info(CURLVERSION_NOW);

return info && (info->features & CURL_VERSION_HTTP3);

{

const FileTransferSettings & settings;

const bool http3Supported = curlSupportsHttp3();

curlMulti curlm;

std::random_device rd;

std::mt19937 mt19937;

struct TransferItem : public std::enable_shared_from_this<TransferItem>, public FileTransfer::Item

{

curlFileTransfer & fileTransfer;

FileTransferRequest request;

FileTransferResult result;

std::unique_ptr<Activity> _act;

Callback<FileTransferResult> callback;

CURL * req = 0;

// buffer to accompany the `req` above

char errbuf[CURL_ERROR_SIZE];

std::string statusMsg;

uint32_t attempt = 0;

/* Don't start this download until the specified time point

std::chrono::steady_clock::time_point embargo;

curlSList requestHeaders;

/**

bool done:1 = false;

/**

bool active:1 = false;

/**

bool paused:1 = false;

/**

bool enqueued:1 = false;

/**

bool acceptRanges:1 = false;

/**

bool hasContentEncoding:1 = false;

/**

std::optional<uint32_t> retryAfterMs;

curl_off_t writtenToSink = 0;

std::chrono::steady_clock::time_point startTime = std::chrono::steady_clock::now();

inline static const std::set<long> successfulStatuses{

static_cast<long>(HttpStatus::Ok),

static_cast<long>(HttpStatus::Created),

static_cast<long>(HttpStatus::NoContent),

static_cast<long>(HttpStatus::PartialContent),

static_cast<long>(HttpStatus::NotModified),

0 /* other protocol */};

/* Get the HTTP status code, or 0 for other protocols. */

long getHTTPStatus()

{

long httpStatus = 0;

long protocol = 0;

curl_easy_getinfo(req, CURLINFO_PROTOCOL, &protocol);

if (protocol == CURLPROTO_HTTP || protocol == CURLPROTO_HTTPS)

curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus);

return httpStatus;

}

void appendHeaders(const std::string & header)

{

curlSList tmpSList = curlSList(::curl_slist_append(requestHeaders.get(), requireCString(header)));

if (!tmpSList)

throw std::bad_alloc();

requestHeaders.release();

requestHeaders = std::move(tmpSList);

}

TransferItem(

curlFileTransfer & fileTransfer,

const FileTransferRequest & request,

Callback<FileTransferResult> && callback)

: fileTransfer(fileTransfer)

, request(request)

, callback(std::move(callback))

, finalSink([this](std::string_view data) {

if (errorSink) {

(*errorSink)(data);

}

if (this->request.dataCallback) {

auto httpStatus = getHTTPStatus();

/* Only write data to the sink if this is a

if (successfulStatuses.count(httpStatus)) {

writtenToSink += data.size();

PauseTransfer needsPause = this->request.dataCallback(data);

if (needsPause == PauseTransfer::Yes) {

/* Smuggle the boolean flag into writeCallback. Note that

paused = true;

}

}

} else

this->result.data.append(data);

})

{

result.urls.push_back(request.uri.to_string());

if (!request.expectedETag.empty())

appendHeaders("If-None-Match: " + request.expectedETag);

if (!request.mimeType.empty())

appendHeaders("Content-Type: " + request.mimeType);

for (auto it = request.headers.begin(); it != request.headers.end(); ++it) {

appendHeaders(fmt("%s: %s", it->first, it->second));

}

}

~TransferItem()

{

if (req) {

if (active)

curl_multi_remove_handle(fileTransfer.curlm.get(), req);

curl_easy_cleanup(req);

}

try {

if (!done && enqueued)

failInterruptedOrCancelled();

} catch (...) {

ignoreExceptionInDestructor();

}

}

void failEx(std::exception_ptr ex) noexcept

{

assert(!done);

done = true;

try {

std::rethrow_exception(ex);

} catch (FileTransferError & e) {

/* Already descriptive enough. */

} catch (nix::Error & e) {

/* Add more context to the error message. */

e.addTrace({}, "during %s of '%s'", Uncolored(request.noun()), request.displayUri());

} catch (...) {

/* Can't add more context to the error. */

}

callback.rethrow(ex);

}

template<class T>

void fail(T && e) noexcept

{

failEx(std::make_exception_ptr(std::forward<T>(e)));

}

void failInterruptedOrCancelled()

{

HintFmt fmt("%s of '%s' was interrupted", Uncolored(request.noun()), request.displayUri());

/* Technically, we don't really have per-transfer cancellation currently,

if (getInterrupted())

fail(nix::Interrupted(std::move(fmt)));

else

fail(nix::Cancelled(std::move(fmt)));

}

LambdaSink finalSink;

std::optional<StringSink> errorSink;

std::exception_ptr callbackException;

size_t writeCallback(void * contents, size_t size, size_t nmemb) noexcept

try {

size_t realSize = size * nmemb;

result.bodySize += realSize;

if (!errorSink && !successfulStatuses.count(getHTTPStatus())) {

// In this case we want to construct a TeeSink, to keep

// the response around (which we figure won't be big

// like an actual download should be) to improve error

// messages.

errorSink = StringSink{};

}

finalSink({static_cast<const char *>(contents), realSize});

if (paused) {

/* The callback has signaled that the transfer needs to be

curl_easy_pause(req, CURLPAUSE_RECV);

}

return realSize;

} catch (...) {

callbackException = std::current_exception();

return 0;

}

static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp)

{

return ((TransferItem *) userp)->writeCallback(contents, size, nmemb);

}

void appendCurrentUrl()

{

char * effectiveUriCStr = nullptr;

curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUriCStr);

if (effectiveUriCStr && *result.urls.rbegin() != effectiveUriCStr)

result.urls.push_back(effectiveUriCStr);

}

size_t headerCallback(void * contents, size_t size, size_t nmemb) noexcept

try {

size_t realSize = size * nmemb;

std::string line((char *) contents, realSize);

printMsg(lvlVomit, "got header for '%s': %s", request.displayUri(), trim(line));

static std::regex statusLine("HTTP/[^ ]+ +[0-9]+(.*)", std::regex::extended | std::regex::icase);

if (std::smatch match; std::regex_match(line, match, statusLine)) {

result.etag = "";

result.data.clear();

result.bodySize = 0;

statusMsg = trim(match.str(1));

acceptRanges = false;

hasContentEncoding = false;

retryAfterMs = std::nullopt;

appendCurrentUrl();

} else {

auto i = line.find(':');

if (i != std::string::npos) {

std::string name = toLower(trim(line.substr(0, i)));

if (name == "etag") {

result.etag = trim(line.substr(i + 1));

/* Hack to work around a GitHub bug: it sends

long httpStatus = 0;

curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus);

if (result.etag == request.expectedETag && httpStatus == HttpStatus::Ok) {

debug("shutting down on 200 HTTP response with expected ETag");

return 0;

}

}

else if (name == "content-encoding") {

auto value = toLower(trim(line.substr(i + 1)));

hasContentEncoding = !value.empty() && value != "identity";

}

else if (name == "accept-ranges" && toLower(trim(line.substr(i + 1))) == "bytes")

acceptRanges = true;

else if (name == "link" || name == "x-amz-meta-link") {

auto value = trim(line.substr(i + 1));

static std::regex linkRegex(

"<([^>]*)>; 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);

}

else if (name == "retry-after") {

auto value = trim(line.substr(i + 1));

// RFC 7231 §7.1.3: Retry-After = HTTP-date / delay-seconds.

if (auto seconds = string2Int<uint32_t>(value)) {

retryAfterMs = saturateMs(std::chrono::seconds{*seconds});

} else if (time_t date = curl_getdate(requireCString(value), nullptr); date != -1) {

time_t now = time(nullptr);

retryAfterMs = saturateMs(std::chrono::seconds{date > now ? date - now : 0});

} else {

debug("ignoring unparseable Retry-After header: '%s'", value);

}

}

}

}

return realSize;

} catch (...) {

/* https://curl.se/libcurl/c/CURLOPT_HEADERFUNCTION.html:

callbackException = std::current_exception();

return CURL_WRITEFUNC_ERROR;

}

static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp)

{

return ((TransferItem *) userp)->headerCallback(contents, size, nmemb);

}

/**

Activity & act()

{

if (!_act) {

_act = std::make_unique<Activity>(

*logger,

lvlTalkative,

actFileTransfer,

fmt("%s '%s'", request.verb(/*continuous=*/true), request.displayUri()),

Logger::Fields{request.displayUri()},

request.parentAct);

// Reset the start time to when we actually started the download.

startTime = std::chrono::steady_clock::now();

}

return *_act;

}

int progressCallback(curl_off_t dltotal, curl_off_t dlnow) noexcept

try {

act().progress(dlnow, dltotal);

return getInterrupted();

} catch (nix::Interrupted &) {

assert(getInterrupted());

return 1;

} catch (...) {

/* Something unexpected has happened like logger throwing an exception. */

callbackException = std::current_exception();

return 1;

}

static int progressCallbackWrapper(

void * userp, curl_off_t dltotal, curl_off_t dlnow, curl_off_t ultotal, curl_off_t ulnow)

{

auto & item = *static_cast<TransferItem *>(userp);

auto isUpload = bool(item.request.data);

return item.progressCallback(isUpload ? ultotal : dltotal, isUpload ? ulnow : dlnow);

}

static int debugCallback(CURL * handle, curl_infotype type, char * data, size_t size, void * userptr) noexcept

try {

if (type == CURLINFO_TEXT)

vomit("curl: %s", chomp(std::string(data, size)));

return 0;

} catch (...) {

/* Swallow the exception. Nothing left to do. */

return 0;

}

size_t readCallback(char * buffer, size_t size, size_t nitems) noexcept

try {

auto data = request.data;

return data->source->read(buffer, nitems * size);

} catch (EndOfFile &) {

return 0;

} catch (...) {

callbackException = std::current_exception();

return CURL_READFUNC_ABORT;

}

static size_t readCallbackWrapper(char * buffer, size_t size, size_t nitems, void * userp) noexcept

{

return ((TransferItem *) userp)->readCallback(buffer, size, nitems);

}

#if !defined(_WIN32)

static int cloexec_callback(void *, curl_socket_t curlfd, curlsocktype purpose)

{

unix::closeOnExec(curlfd);

vomit("cloexec set for fd %i", curlfd);

return CURL_SOCKOPT_OK;

}

#endif

size_t seekCallback(curl_off_t offset, int origin) noexcept

try {

auto source = request.data->source;

if (origin == SEEK_SET) {

source->restart();

source->skip(offset);

} else if (origin == SEEK_CUR) {

source->skip(offset);

} else if (origin == SEEK_END) {

NullSink sink{};

source->drainInto(sink);

}

return CURL_SEEKFUNC_OK;

} catch (...) {

callbackException = std::current_exception();

return CURL_SEEKFUNC_FAIL;

}

static size_t seekCallbackWrapper(void * clientp, curl_off_t offset, int origin) noexcept

{

return ((TransferItem *) clientp)->seekCallback(offset, origin);

}

static int resolverCallbackWrapper(void *, void *, void * clientp) noexcept

try {

// Create the `Activity` associated with this download.

((TransferItem *) clientp)->act();

return 0;

} catch (...) {

return 1;

}

void unpause()

{

/* Unpausing an already unpaused transfer is a no-op. */

if (paused) {

curl_easy_pause(req, CURLPAUSE_CONT);

paused = false;

}

}

void init()

{

if (!req)

req = curl_easy_init();

curl_easy_reset(req);

if (verbosity >= lvlVomit) {

curl_easy_setopt(req, CURLOPT_VERBOSE, 1);

curl_easy_setopt(req, CURLOPT_DEBUGFUNCTION, TransferItem::debugCallback);

}

curl_easy_setopt(req, CURLOPT_URL, request.uri.to_string().c_str());

/* Enable transparent decompression for downloads.

if (writtenToSink == 0 && !request.data)

/* Empty string means to enable all supported (that libcurl has

curl_easy_setopt(req, CURLOPT_ACCEPT_ENCODING, "");

curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L);

curl_easy_setopt(req, CURLOPT_MAXREDIRS, 10);

curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1);

curl_easy_setopt(

req,

CURLOPT_USERAGENT,

("curl/" LIBCURL_VERSION " Nix/" + nixVersion

+ (fileTransfer.settings.userAgentSuffix != "" ? " " + fileTransfer.settings.userAgentSuffix.get()

: ""))

.c_str());

curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1);

/* Enable HTTP3 only on user config and linked libcurl have support, o.w. fall back. */

if (fileTransfer.settings.enableHttp3 && fileTransfer.http3Supported)

curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_3);

else if (fileTransfer.settings.enableHttp2)

curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);

else

curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_1_1);

curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, TransferItem::writeCallbackWrapper);

curl_easy_setopt(req, CURLOPT_WRITEDATA, this);

curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, TransferItem::headerCallbackWrapper);

curl_easy_setopt(req, CURLOPT_HEADERDATA, this);

curl_easy_setopt(req, CURLOPT_XFERINFOFUNCTION, progressCallbackWrapper);

curl_easy_setopt(req, CURLOPT_XFERINFODATA, this);

curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0);

curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders.get());

if (fileTransfer.settings.downloadSpeed.get() > 0)

curl_easy_setopt(

req, CURLOPT_MAX_RECV_SPEED_LARGE, (curl_off_t) (fileTransfer.settings.downloadSpeed.get() * 1024));

if (request.method == HttpMethod::Head)

curl_easy_setopt(req, CURLOPT_NOBODY, 1);

if (request.method == HttpMethod::Delete)

curl_easy_setopt(req, CURLOPT_CUSTOMREQUEST, "DELETE");

if (request.data) {

// Restart the source to ensure it's at the beginning.

// This is necessary for retries, where the source was

// already consumed by a previous attempt.

request.data->source->restart();

if (request.method == HttpMethod::Post) {

curl_easy_setopt(req, CURLOPT_POST, 1L);

curl_easy_setopt(req, CURLOPT_POSTFIELDSIZE_LARGE, (curl_off_t) request.data->sizeHint);

} else if (request.method == HttpMethod::Put) {

curl_easy_setopt(req, CURLOPT_UPLOAD, 1L);

curl_easy_setopt(req, CURLOPT_INFILESIZE_LARGE, (curl_off_t) request.data->sizeHint);

} else {

unreachable();

}

curl_easy_setopt(req, CURLOPT_READFUNCTION, readCallbackWrapper);

curl_easy_setopt(req, CURLOPT_READDATA, this);

curl_easy_setopt(req, CURLOPT_SEEKFUNCTION, seekCallbackWrapper);

curl_easy_setopt(req, CURLOPT_SEEKDATA, this);

}

/* Note: libcurl copies string arguments, so temporaries from

if (auto & caFile = fileTransfer.settings.caFile.get())

curl_easy_setopt(req, CURLOPT_CAINFO, caFile->string().c_str());

#ifdef _WIN32

/* Use native windows certificate store when the option is not specified explicitly. */

else

curl_easy_setopt(req, CURLOPT_SSL_OPTIONS, CURLSSLOPT_NATIVE_CA);

#endif

#if !defined(_WIN32)

curl_easy_setopt(req, CURLOPT_SOCKOPTFUNCTION, cloexec_callback);

#endif

curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, fileTransfer.settings.connectTimeout.get());

/* Enable TCP keepalive to detect dead connections and server closures.

curl_easy_setopt(req, CURLOPT_TCP_KEEPALIVE, 1L);

curl_easy_setopt(req, CURLOPT_TCP_KEEPIDLE, 30L);

curl_easy_setopt(req, CURLOPT_TCP_KEEPINTVL, 30L);

/* Don't reuse idle connections older than 90s. */

curl_easy_setopt(req, CURLOPT_MAXAGE_CONN, 90L);

curl_easy_setopt(req, CURLOPT_LOW_SPEED_LIMIT, 1L);

curl_easy_setopt(req, CURLOPT_LOW_SPEED_TIME, fileTransfer.settings.stalledDownloadTimeout.get());

/* If no file exist in the specified path, curl continues to work

curl_easy_setopt(req, CURLOPT_NETRC_FILE, fileTransfer.settings.netrcFile.get().string().c_str());

curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);

if (writtenToSink)

curl_easy_setopt(req, CURLOPT_RESUME_FROM_LARGE, writtenToSink);

/* Note that the underlying strings get copied by libcurl, so the path -> string conversion is ok:

if (request.tlsCert) {

curl_easy_setopt(req, CURLOPT_SSLCERTTYPE, "PEM");

curl_easy_setopt(req, CURLOPT_SSLCERT, request.tlsCert->string().c_str());

}

if (request.tlsKey) {

curl_easy_setopt(req, CURLOPT_SSLKEYTYPE, "PEM");

curl_easy_setopt(req, CURLOPT_SSLKEY, request.tlsKey->string().c_str());

}

curl_easy_setopt(req, CURLOPT_ERRORBUFFER, errbuf);

errbuf[0] = 0;

// Set up username/password authentication if provided

if (request.usernameAuth) {

curl_easy_setopt(req, CURLOPT_USERNAME, request.usernameAuth->username.c_str());

if (request.usernameAuth->password) {

curl_easy_setopt(req, CURLOPT_PASSWORD, request.usernameAuth->password->c_str());

}

}

#if NIX_WITH_AWS_AUTH

// Set up AWS SigV4 signing if this is an S3 request

// Note: AWS SigV4 support guaranteed available (curl >= 7.75.0 checked at build time)

// The username/password (access key ID and secret key) are set via the general

// usernameAuth mechanism above.

if (request.awsSigV4Provider) {

curl_easy_setopt(req, CURLOPT_AWS_SIGV4, request.awsSigV4Provider->c_str());

}

#endif

// This seems to be the earliest libcurl callback that signals that the download is happening, so we can

// call act().

curl_easy_setopt(req, CURLOPT_RESOLVER_START_FUNCTION, resolverCallbackWrapper);

curl_easy_setopt(req, CURLOPT_RESOLVER_START_DATA, this);

result.data.clear();

result.bodySize = 0;

}

void finish(CURLcode code)

{

auto finishTime = std::chrono::steady_clock::now();

auto httpStatus = getHTTPStatus();

debug(

"finished %s of '%s'; curl status = %d, HTTP status = %d, body = %d bytes, duration = %.2f s",

Uncolored(request.noun()),

request.displayUri(),

code,

httpStatus,

result.bodySize,

std::chrono::duration_cast<std::chrono::milliseconds>(finishTime - startTime).count() / 1000.0f);

appendCurrentUrl();

if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) {

code = CURLE_OK;

httpStatus = std::to_underlying(HttpStatus::NotModified);

}

if (callbackException)

failEx(callbackException);

else if (code == CURLE_OK && successfulStatuses.count(httpStatus)) {

result.cached = (httpStatus == HttpStatus::NotModified);

// In 2021, GitHub responds to If-None-Match with 304,

// but omits ETag. We just use the If-None-Match etag

// since 304 implies they are the same.

if (httpStatus == HttpStatus::NotModified && result.etag == "")

result.etag = request.expectedETag;

curl_off_t dlSize = 0;

curl_easy_getinfo(req, CURLINFO_SIZE_DOWNLOAD_T, &dlSize);

act().progress(dlSize, dlSize);

done = true;

callback(std::move(result));

}

else {

// We treat most errors as transient, but won't retry when hopeless

Error err = Transient;

// S3 returns certain retryable errors as HTTP 400/500/503 with XML error codes.

// These take precedence over the generic HTTP status handling below.

// Only parse the response body on status codes where S3 XML errors can appear.

static constexpr std::array<std::string_view, 12> s3RetryableErrors{{

"IncompleteBody", // HTTP 400 - network issue

"InternalError", // HTTP 500 - S3 internal failure

"InternalFailure", // HTTP 500 - alias for InternalError

"InternalServerError", // HTTP 500 - alias for InternalError

"RequestExpired", // HTTP 400 - clock skew / slow upload

"RequestTimeout", // HTTP 400 - stale connection reuse

"RequestTimeTooSkewed", // HTTP 403 - clock drift

"RequestThrottled", // HTTP 400 - throttling variant

"SlowDown", // HTTP 503 - throttling

"ServiceUnavailable", // HTTP 503 - temporary unavailability

"Throttling", // HTTP 400 - throttling variant

"ThrottledException", // HTTP 400 - throttling variant

}};

// S3 error responses have the form <Error><Code>...</Code>...</Error>.

// Require the <Error> root to avoid matching unrelated XML with a <Code> element.

static std::regex s3ErrorCodeRegex("<Error>[^]*<Code>([^<]+)</Code>");

std::smatch s3Match;

bool isS3XmlStatus = httpStatus == 400 || httpStatus == 403 || httpStatus == 500 || httpStatus == 503;

auto s3ErrorCode =

(isS3XmlStatus && errorSink && std::regex_search(errorSink->s, s3Match, s3ErrorCodeRegex))

? s3Match[1].str()

: "";

if (std::find(s3RetryableErrors.begin(), s3RetryableErrors.end(), s3ErrorCode)

!= s3RetryableErrors.end()) {

debug("S3 error '%s', will retry", s3ErrorCode);

} else if (

httpStatus == HttpStatus::NotFound || httpStatus == HttpStatus::Gone

|| code == CURLE_FILE_COULDNT_READ_FILE) {

// The file is definitely not there

err = NotFound;

} else if (httpStatus == HttpStatus::Unauthorized || httpStatus == HttpStatus::ProxyAuthRequired) {

err = Unauthorized;

} else if (httpStatus == HttpStatus::Forbidden) {

// Don't retry on authentication/authorization failures.

// Note: the only reason we treat this differently from 401/407 is S3 returns 403 if a file doesn't

// exist and the bucket is unlistable.

err = Forbidden;

} else if (

httpStatus >= 400 && httpStatus < 500 && httpStatus != HttpStatus::RequestTimeout

&& httpStatus != HttpStatus::TooManyRequests) {

// 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

err = Misc;

} else if (

httpStatus == HttpStatus::NotImplemented || httpStatus == HttpStatus::HttpVersionNotSupported

|| httpStatus == HttpStatus::NetworkAuthRequired) {

// 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 {

#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:

case CURLE_BAD_CONTENT_ENCODING:

err = Misc;

break;

default: // Shut up warnings

break;

}

#pragma GCC diagnostic pop

}

attempt++;

std::optional<std::string> response;

if (errorSink)

response = std::move(errorSink->s);

/* TODO: Also support per-transfer cancellations. */

if (code == CURLE_ABORTED_BY_CALLBACK && getInterrupted()) {

failInterruptedOrCancelled();

return;

}

auto exc = httpStatus != 0

? FileTransferError(

err,

std::move(response),

"unable to %s '%s': HTTP error %d%s",

Uncolored(request.verb()),

request.displayUri(),

httpStatus,

code == CURLE_OK ? "" : fmt(" (curl error: %s)", curl_easy_strerror(code)))

: FileTransferError(

err,

std::move(response),

"unable to %s '%s': %s (%d) %s",

Uncolored(request.verb()),

request.displayUri(),

curl_easy_strerror(code),

code,

errbuf);

maybeRetry(err, httpStatus, std::move(exc));

}

}

/* If this is a transient error, then maybe retry the download

void maybeRetry(FileTransfer::Error err, long httpStatus, FileTransferError && exc)

{

// Resolve effective retry config (request overrides > global settings)

auto effAttempts = request.retryAttempts.value_or(fileTransfer.settings.tries);

auto effBaseMs = request.retryDelayMs.value_or(fileTransfer.settings.retryDelayMs);

auto effRateLimitMs =

request.retryDelayRateLimitedMs.value_or(fileTransfer.settings.retryDelayRateLimitedMs);

auto effMaxMs = request.retryMaxDelayMs.value_or(fileTransfer.settings.retryMaxDelayMs);

// Pick base delay by error class: 429/503 indicate the server is

// rate-limiting or overloaded, so use the longer rate-limit delay.

uint32_t baseMs =

(httpStatus == HttpStatus::TooManyRequests || httpStatus == HttpStatus::ServiceUnavailable)

? effRateLimitMs

: effBaseMs;

auto canRetry = [&] {

if (err != Transient)

return false;

if (attempt >= effAttempts)

return false;

// If we've already streamed bytes to the callback, we can only

// resume via a Range request. That requires the server to accept

// byte ranges AND the response to be uncompressed (the Range

// applies to the encoded stream, but the sink saw decoded bytes).

if (request.dataCallback && writtenToSink != 0)

return acceptRanges && !hasContentEncoding;

return true;

}();

if (!canRetry) {

fail(std::move(exc));

return;

}

auto delay = computeRetryDelayMs(

{

.attempt = attempt,

.baseMs = baseMs,

.ceilMs = effMaxMs,

.retryAfterMs = std::exchange(retryAfterMs, std::nullopt),

.jitter = fileTransfer.settings.retryJitter,

},

fileTransfer.mt19937);

if (writtenToSink) {

warn(

"%s; retrying from offset %d in %d ms (attempt %d/%d)",

exc.message(),

writtenToSink,

delay.count(),

attempt,

effAttempts);

} else {

warn("%s; retrying in %d ms (attempt %d/%d)", exc.message(), delay.count(), attempt, effAttempts);

}

errorSink.reset();

embargo = std::chrono::steady_clock::now() + delay;

try {

fileTransfer.enqueueItem(ref{shared_from_this()});

} catch (const nix::Error & e) {

// If enqueue fails (e.g., during shutdown), fail the transfer properly

// instead of letting the exception propagate, which would leave done=false

// and cause the destructor to attempt a second callback invocation

fail(std::move(exc));

}

}

};

struct State

{

struct EmbargoComparator

{

bool operator()(const ref<TransferItem> & i1, const ref<TransferItem> & i2)

{

return i1->embargo > i2->embargo;

}

};

std::priority_queue<ref<TransferItem>, std::vector<ref<TransferItem>>, EmbargoComparator> incoming;

std::vector<std::weak_ptr<Item>> unpause;

private:

bool quitting = false;

public:

bool work = false;