mirror of
https://github.com/privatevoid-net/nix-super.git
synced 2024-11-11 00:36:20 +02:00
690 lines
25 KiB
C++
690 lines
25 KiB
C++
#include "download.hh"
|
||
#include "util.hh"
|
||
#include "globals.hh"
|
||
#include "hash.hh"
|
||
#include "store-api.hh"
|
||
#include "archive.hh"
|
||
#include "s3.hh"
|
||
#include "compression.hh"
|
||
#include "pathlocks.hh"
|
||
|
||
#ifdef ENABLE_S3
|
||
#include <aws/core/client/ClientConfiguration.h>
|
||
#endif
|
||
|
||
#include <unistd.h>
|
||
#include <fcntl.h>
|
||
|
||
#include <curl/curl.h>
|
||
|
||
#include <queue>
|
||
#include <iostream>
|
||
#include <thread>
|
||
#include <cmath>
|
||
#include <random>
|
||
|
||
namespace nix {
|
||
|
||
double getTime()
|
||
{
|
||
struct timeval tv;
|
||
gettimeofday(&tv, 0);
|
||
return tv.tv_sec + (tv.tv_usec / 1000000.0);
|
||
}
|
||
|
||
std::string resolveUri(const std::string & uri)
|
||
{
|
||
if (uri.compare(0, 8, "channel:") == 0)
|
||
return "https://nixos.org/channels/" + std::string(uri, 8) + "/nixexprs.tar.xz";
|
||
else
|
||
return uri;
|
||
}
|
||
|
||
ref<std::string> decodeContent(const std::string & encoding, ref<std::string> data)
|
||
{
|
||
if (encoding == "")
|
||
return data;
|
||
else if (encoding == "br")
|
||
return decompress(encoding, *data);
|
||
else
|
||
throw Error("unsupported Content-Encoding ‘%s’", encoding);
|
||
}
|
||
|
||
struct CurlDownloader : public Downloader
|
||
{
|
||
CURLM * curlm = 0;
|
||
|
||
std::random_device rd;
|
||
std::mt19937 mt19937;
|
||
|
||
bool enableHttp2;
|
||
|
||
struct DownloadItem : public std::enable_shared_from_this<DownloadItem>
|
||
{
|
||
CurlDownloader & downloader;
|
||
DownloadRequest request;
|
||
DownloadResult result;
|
||
Activity act;
|
||
bool done = false; // whether either the success or failure function has been called
|
||
std::function<void(const DownloadResult &)> success;
|
||
std::function<void(std::exception_ptr exc)> failure;
|
||
CURL * req = 0;
|
||
bool active = false; // whether the handle has been added to the multi object
|
||
std::string status;
|
||
|
||
unsigned int attempt = 0;
|
||
|
||
/* Don't start this download until the specified time point
|
||
has been reached. */
|
||
std::chrono::steady_clock::time_point embargo;
|
||
|
||
struct curl_slist * requestHeaders = 0;
|
||
|
||
std::string encoding;
|
||
|
||
DownloadItem(CurlDownloader & downloader, const DownloadRequest & request)
|
||
: downloader(downloader), request(request)
|
||
{
|
||
if (!request.expectedETag.empty())
|
||
requestHeaders = curl_slist_append(requestHeaders, ("If-None-Match: " + request.expectedETag).c_str());
|
||
|
||
logger->event(evDownloadCreated, act, request.uri);
|
||
}
|
||
|
||
~DownloadItem()
|
||
{
|
||
if (req) {
|
||
if (active)
|
||
curl_multi_remove_handle(downloader.curlm, req);
|
||
curl_easy_cleanup(req);
|
||
}
|
||
if (requestHeaders) curl_slist_free_all(requestHeaders);
|
||
try {
|
||
if (!done)
|
||
fail(DownloadError(Interrupted, format("download of ‘%s’ was interrupted") % request.uri));
|
||
} catch (...) {
|
||
ignoreException();
|
||
}
|
||
logger->event(evDownloadDestroyed, act);
|
||
}
|
||
|
||
template<class T>
|
||
void fail(const T & e)
|
||
{
|
||
assert(!done);
|
||
done = true;
|
||
callFailure(failure, std::make_exception_ptr(e));
|
||
}
|
||
|
||
size_t writeCallback(void * contents, size_t size, size_t nmemb)
|
||
{
|
||
size_t realSize = size * nmemb;
|
||
result.data->append((char *) contents, realSize);
|
||
return realSize;
|
||
}
|
||
|
||
static size_t writeCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp)
|
||
{
|
||
return ((DownloadItem *) userp)->writeCallback(contents, size, nmemb);
|
||
}
|
||
|
||
size_t headerCallback(void * contents, size_t size, size_t nmemb)
|
||
{
|
||
size_t realSize = size * nmemb;
|
||
std::string line((char *) contents, realSize);
|
||
printMsg(lvlVomit, format("got header for ‘%s’: %s") % request.uri % trim(line));
|
||
if (line.compare(0, 5, "HTTP/") == 0) { // new response starts
|
||
result.etag = "";
|
||
auto ss = tokenizeString<vector<string>>(line, " ");
|
||
status = ss.size() >= 2 ? ss[1] : "";
|
||
result.data = std::make_shared<std::string>();
|
||
encoding = "";
|
||
} else {
|
||
auto i = line.find(':');
|
||
if (i != string::npos) {
|
||
string name = toLower(trim(string(line, 0, i)));
|
||
if (name == "etag") {
|
||
result.etag = trim(string(line, i + 1));
|
||
/* Hack to work around a GitHub bug: it sends
|
||
ETags, but ignores If-None-Match. So if we get
|
||
the expected ETag on a 200 response, then shut
|
||
down the connection because we already have the
|
||
data. */
|
||
if (result.etag == request.expectedETag && status == "200") {
|
||
debug(format("shutting down on 200 HTTP response with expected ETag"));
|
||
return 0;
|
||
}
|
||
} else if (name == "content-encoding")
|
||
encoding = trim(string(line, i + 1));;
|
||
}
|
||
}
|
||
return realSize;
|
||
}
|
||
|
||
static size_t headerCallbackWrapper(void * contents, size_t size, size_t nmemb, void * userp)
|
||
{
|
||
return ((DownloadItem *) userp)->headerCallback(contents, size, nmemb);
|
||
}
|
||
|
||
int progressCallback(double dltotal, double dlnow)
|
||
{
|
||
logger->event(evDownloadProgress, act, dltotal, dlnow);
|
||
return _isInterrupted;
|
||
}
|
||
|
||
static int progressCallbackWrapper(void * userp, double dltotal, double dlnow, double ultotal, double ulnow)
|
||
{
|
||
return ((DownloadItem *) 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 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, DownloadItem::debugCallback);
|
||
}
|
||
|
||
curl_easy_setopt(req, CURLOPT_URL, request.uri.c_str());
|
||
curl_easy_setopt(req, CURLOPT_FOLLOWLOCATION, 1L);
|
||
curl_easy_setopt(req, CURLOPT_NOSIGNAL, 1);
|
||
curl_easy_setopt(req, CURLOPT_USERAGENT,
|
||
("curl/" LIBCURL_VERSION " Nix/" + nixVersion +
|
||
(settings.userAgentSuffix != "" ? " " + settings.userAgentSuffix.get() : "")).c_str());
|
||
#if LIBCURL_VERSION_NUM >= 0x072b00
|
||
curl_easy_setopt(req, CURLOPT_PIPEWAIT, 1);
|
||
#endif
|
||
#if LIBCURL_VERSION_NUM >= 0x072f00
|
||
if (downloader.enableHttp2)
|
||
curl_easy_setopt(req, CURLOPT_HTTP_VERSION, CURL_HTTP_VERSION_2TLS);
|
||
#endif
|
||
curl_easy_setopt(req, CURLOPT_WRITEFUNCTION, DownloadItem::writeCallbackWrapper);
|
||
curl_easy_setopt(req, CURLOPT_WRITEDATA, this);
|
||
curl_easy_setopt(req, CURLOPT_HEADERFUNCTION, DownloadItem::headerCallbackWrapper);
|
||
curl_easy_setopt(req, CURLOPT_HEADERDATA, this);
|
||
|
||
curl_easy_setopt(req, CURLOPT_PROGRESSFUNCTION, progressCallbackWrapper);
|
||
curl_easy_setopt(req, CURLOPT_PROGRESSDATA, this);
|
||
curl_easy_setopt(req, CURLOPT_NOPROGRESS, 0);
|
||
|
||
curl_easy_setopt(req, CURLOPT_HTTPHEADER, requestHeaders);
|
||
|
||
if (request.head)
|
||
curl_easy_setopt(req, CURLOPT_NOBODY, 1);
|
||
|
||
if (request.verifyTLS) {
|
||
if (settings.caFile != "")
|
||
curl_easy_setopt(req, CURLOPT_CAINFO, settings.caFile.c_str());
|
||
} else {
|
||
curl_easy_setopt(req, CURLOPT_SSL_VERIFYPEER, 0);
|
||
curl_easy_setopt(req, CURLOPT_SSL_VERIFYHOST, 0);
|
||
}
|
||
|
||
curl_easy_setopt(req, CURLOPT_CONNECTTIMEOUT, settings.connectTimeout.get());
|
||
|
||
/* If no file exist in the specified path, curl continues to work
|
||
anyway as if netrc support was disabled. */
|
||
curl_easy_setopt(req, CURLOPT_NETRC_FILE, settings.netrcFile.get().c_str());
|
||
curl_easy_setopt(req, CURLOPT_NETRC, CURL_NETRC_OPTIONAL);
|
||
|
||
result.data = std::make_shared<std::string>();
|
||
}
|
||
|
||
void finish(CURLcode code)
|
||
{
|
||
long httpStatus = 0;
|
||
curl_easy_getinfo(req, CURLINFO_RESPONSE_CODE, &httpStatus);
|
||
|
||
char * effectiveUrlCStr;
|
||
curl_easy_getinfo(req, CURLINFO_EFFECTIVE_URL, &effectiveUrlCStr);
|
||
if (effectiveUrlCStr)
|
||
result.effectiveUrl = effectiveUrlCStr;
|
||
|
||
debug(format("finished download of ‘%s’; curl status = %d, HTTP status = %d, body = %d bytes")
|
||
% request.uri % code % httpStatus % (result.data ? result.data->size() : 0));
|
||
|
||
if (code == CURLE_WRITE_ERROR && result.etag == request.expectedETag) {
|
||
code = CURLE_OK;
|
||
httpStatus = 304;
|
||
}
|
||
|
||
if (code == CURLE_OK &&
|
||
(httpStatus == 200 || httpStatus == 304 || httpStatus == 226 /* FTP */ || httpStatus == 0 /* other protocol */))
|
||
{
|
||
result.cached = httpStatus == 304;
|
||
done = true;
|
||
|
||
try {
|
||
result.data = decodeContent(encoding, ref<std::string>(result.data));
|
||
callSuccess(success, failure, const_cast<const DownloadResult &>(result));
|
||
logger->event(evDownloadSucceeded, act, result.data->size());
|
||
} catch (...) {
|
||
done = true;
|
||
callFailure(failure, std::current_exception());
|
||
}
|
||
} else {
|
||
Error err =
|
||
(httpStatus == 404 || code == CURLE_FILE_COULDNT_READ_FILE) ? NotFound :
|
||
httpStatus == 403 ? Forbidden :
|
||
(httpStatus == 408 || httpStatus == 500 || httpStatus == 503
|
||
|| httpStatus == 504 || httpStatus == 522 || httpStatus == 524
|
||
|| code == CURLE_COULDNT_RESOLVE_HOST
|
||
|| code == CURLE_RECV_ERROR
|
||
|
||
// this seems to occur occasionally for retriable reasons, and shows up in an error like this:
|
||
// curl: (23) Failed writing body (315 != 16366)
|
||
|| code == CURLE_WRITE_ERROR
|
||
|
||
// this is a generic SSL failure that in some cases (e.g., certificate error) is permanent but also appears in transient cases, so we consider it retryable
|
||
|| code == CURLE_SSL_CONNECT_ERROR
|
||
#if LIBCURL_VERSION_NUM >= 0x073200
|
||
|| code == CURLE_HTTP2
|
||
|| code == CURLE_HTTP2_STREAM
|
||
#endif
|
||
) ? Transient :
|
||
Misc;
|
||
|
||
attempt++;
|
||
|
||
auto exc =
|
||
code == CURLE_ABORTED_BY_CALLBACK && _isInterrupted
|
||
? DownloadError(Interrupted, format("download of ‘%s’ was interrupted") % request.uri)
|
||
: httpStatus != 0
|
||
? DownloadError(err, format("unable to download ‘%s’: HTTP error %d (curl error: %s)") % request.uri % httpStatus % curl_easy_strerror(code))
|
||
: DownloadError(err, format("unable to download ‘%s’: %s (%d)") % request.uri % curl_easy_strerror(code) % code);
|
||
|
||
/* If this is a transient error, then maybe retry the
|
||
download after a while. */
|
||
if (err == Transient && attempt < request.tries) {
|
||
int ms = request.baseRetryTimeMs * std::pow(2.0f, attempt - 1 + std::uniform_real_distribution<>(0.0, 0.5)(downloader.mt19937));
|
||
printError(format("warning: %s; retrying in %d ms") % exc.what() % ms);
|
||
embargo = std::chrono::steady_clock::now() + std::chrono::milliseconds(ms);
|
||
downloader.enqueueItem(shared_from_this());
|
||
}
|
||
else
|
||
fail(exc);
|
||
}
|
||
}
|
||
};
|
||
|
||
struct State
|
||
{
|
||
struct EmbargoComparator {
|
||
bool operator() (const std::shared_ptr<DownloadItem> & i1, const std::shared_ptr<DownloadItem> & i2) {
|
||
return i1->embargo > i2->embargo;
|
||
}
|
||
};
|
||
bool quit = false;
|
||
std::priority_queue<std::shared_ptr<DownloadItem>, std::vector<std::shared_ptr<DownloadItem>>, EmbargoComparator> incoming;
|
||
};
|
||
|
||
Sync<State> state_;
|
||
|
||
/* We can't use a std::condition_variable to wake up the curl
|
||
thread, because it only monitors file descriptors. So use a
|
||
pipe instead. */
|
||
Pipe wakeupPipe;
|
||
|
||
std::thread workerThread;
|
||
|
||
CurlDownloader()
|
||
: mt19937(rd())
|
||
{
|
||
static std::once_flag globalInit;
|
||
std::call_once(globalInit, curl_global_init, CURL_GLOBAL_ALL);
|
||
|
||
curlm = curl_multi_init();
|
||
|
||
#if LIBCURL_VERSION_NUM >= 0x072b00 // correct?
|
||
curl_multi_setopt(curlm, CURLMOPT_PIPELINING, CURLPIPE_MULTIPLEX);
|
||
#endif
|
||
curl_multi_setopt(curlm, CURLMOPT_MAX_TOTAL_CONNECTIONS,
|
||
settings.binaryCachesParallelConnections.get());
|
||
|
||
enableHttp2 = settings.enableHttp2;
|
||
|
||
wakeupPipe.create();
|
||
fcntl(wakeupPipe.readSide.get(), F_SETFL, O_NONBLOCK);
|
||
|
||
workerThread = std::thread([&]() { workerThreadEntry(); });
|
||
}
|
||
|
||
~CurlDownloader()
|
||
{
|
||
stopWorkerThread();
|
||
|
||
workerThread.join();
|
||
|
||
if (curlm) curl_multi_cleanup(curlm);
|
||
}
|
||
|
||
void stopWorkerThread()
|
||
{
|
||
/* Signal the worker thread to exit. */
|
||
{
|
||
auto state(state_.lock());
|
||
state->quit = true;
|
||
}
|
||
writeFull(wakeupPipe.writeSide.get(), " ", false);
|
||
}
|
||
|
||
void workerThreadMain()
|
||
{
|
||
/* Cause this thread to be notified on SIGINT. */
|
||
auto callback = createInterruptCallback([&]() {
|
||
stopWorkerThread();
|
||
});
|
||
|
||
std::map<CURL *, std::shared_ptr<DownloadItem>> items;
|
||
|
||
bool quit = false;
|
||
|
||
std::chrono::steady_clock::time_point nextWakeup;
|
||
|
||
while (!quit) {
|
||
checkInterrupt();
|
||
|
||
/* Let curl do its thing. */
|
||
int running;
|
||
CURLMcode mc = curl_multi_perform(curlm, &running);
|
||
if (mc != CURLM_OK)
|
||
throw nix::Error(format("unexpected error from curl_multi_perform(): %s") % curl_multi_strerror(mc));
|
||
|
||
/* Set the promises of any finished requests. */
|
||
CURLMsg * msg;
|
||
int left;
|
||
while ((msg = curl_multi_info_read(curlm, &left))) {
|
||
if (msg->msg == CURLMSG_DONE) {
|
||
auto i = items.find(msg->easy_handle);
|
||
assert(i != items.end());
|
||
i->second->finish(msg->data.result);
|
||
curl_multi_remove_handle(curlm, i->second->req);
|
||
i->second->active = false;
|
||
items.erase(i);
|
||
}
|
||
}
|
||
|
||
/* Wait for activity, including wakeup events. */
|
||
int numfds = 0;
|
||
struct curl_waitfd extraFDs[1];
|
||
extraFDs[0].fd = wakeupPipe.readSide.get();
|
||
extraFDs[0].events = CURL_WAIT_POLLIN;
|
||
extraFDs[0].revents = 0;
|
||
auto sleepTimeMs =
|
||
nextWakeup != std::chrono::steady_clock::time_point()
|
||
? std::max(0, (int) std::chrono::duration_cast<std::chrono::milliseconds>(nextWakeup - std::chrono::steady_clock::now()).count())
|
||
: 1000000000;
|
||
vomit("download thread waiting for %d ms", sleepTimeMs);
|
||
mc = curl_multi_wait(curlm, extraFDs, 1, sleepTimeMs, &numfds);
|
||
if (mc != CURLM_OK)
|
||
throw nix::Error(format("unexpected error from curl_multi_wait(): %s") % curl_multi_strerror(mc));
|
||
|
||
nextWakeup = std::chrono::steady_clock::time_point();
|
||
|
||
/* Add new curl requests from the incoming requests queue,
|
||
except for requests that are embargoed (waiting for a
|
||
retry timeout to expire). */
|
||
if (extraFDs[0].revents & CURL_WAIT_POLLIN) {
|
||
char buf[1024];
|
||
auto res = read(extraFDs[0].fd, buf, sizeof(buf));
|
||
if (res == -1 && errno != EINTR)
|
||
throw SysError("reading curl wakeup socket");
|
||
}
|
||
|
||
std::vector<std::shared_ptr<DownloadItem>> incoming;
|
||
auto now = std::chrono::steady_clock::now();
|
||
|
||
{
|
||
auto state(state_.lock());
|
||
while (!state->incoming.empty()) {
|
||
auto item = state->incoming.top();
|
||
if (item->embargo <= now) {
|
||
incoming.push_back(item);
|
||
state->incoming.pop();
|
||
} else {
|
||
if (nextWakeup == std::chrono::steady_clock::time_point()
|
||
|| item->embargo < nextWakeup)
|
||
nextWakeup = item->embargo;
|
||
break;
|
||
}
|
||
}
|
||
quit = state->quit;
|
||
}
|
||
|
||
for (auto & item : incoming) {
|
||
debug(format("starting download of %s") % item->request.uri);
|
||
item->init();
|
||
curl_multi_add_handle(curlm, item->req);
|
||
item->active = true;
|
||
items[item->req] = item;
|
||
}
|
||
}
|
||
|
||
debug("download thread shutting down");
|
||
}
|
||
|
||
void workerThreadEntry()
|
||
{
|
||
try {
|
||
workerThreadMain();
|
||
} catch (nix::Interrupted & e) {
|
||
} catch (std::exception & e) {
|
||
printError(format("unexpected error in download thread: %s") % e.what());
|
||
}
|
||
|
||
{
|
||
auto state(state_.lock());
|
||
while (!state->incoming.empty()) state->incoming.pop();
|
||
state->quit = true;
|
||
}
|
||
}
|
||
|
||
void enqueueItem(std::shared_ptr<DownloadItem> item)
|
||
{
|
||
{
|
||
auto state(state_.lock());
|
||
if (state->quit)
|
||
throw nix::Error("cannot enqueue download request because the download thread is shutting down");
|
||
state->incoming.push(item);
|
||
}
|
||
writeFull(wakeupPipe.writeSide.get(), " ");
|
||
}
|
||
|
||
void enqueueDownload(const DownloadRequest & request,
|
||
std::function<void(const DownloadResult &)> success,
|
||
std::function<void(std::exception_ptr exc)> failure) override
|
||
{
|
||
/* Ugly hack to support s3:// URIs. */
|
||
if (hasPrefix(request.uri, "s3://")) {
|
||
// FIXME: do this on a worker thread
|
||
sync2async<DownloadResult>(success, failure, [&]() -> DownloadResult {
|
||
#ifdef ENABLE_S3
|
||
S3Helper s3Helper(Aws::Region::US_EAST_1); // FIXME: make configurable
|
||
auto slash = request.uri.find('/', 5);
|
||
if (slash == std::string::npos)
|
||
throw nix::Error("bad S3 URI ‘%s’", request.uri);
|
||
std::string bucketName(request.uri, 5, slash - 5);
|
||
std::string key(request.uri, slash + 1);
|
||
// FIXME: implement ETag
|
||
auto s3Res = s3Helper.getObject(bucketName, key);
|
||
DownloadResult res;
|
||
if (!s3Res.data)
|
||
throw DownloadError(NotFound, fmt("S3 object ‘%s’ does not exist", request.uri));
|
||
res.data = s3Res.data;
|
||
return res;
|
||
#else
|
||
throw nix::Error("cannot download ‘%s’ because Nix is not built with S3 support", request.uri);
|
||
#endif
|
||
});
|
||
return;
|
||
}
|
||
|
||
auto item = std::make_shared<DownloadItem>(*this, request);
|
||
item->success = success;
|
||
item->failure = failure;
|
||
enqueueItem(item);
|
||
}
|
||
};
|
||
|
||
ref<Downloader> getDownloader()
|
||
{
|
||
static std::shared_ptr<Downloader> downloader;
|
||
static std::once_flag downloaderCreated;
|
||
std::call_once(downloaderCreated, [&]() { downloader = makeDownloader(); });
|
||
return ref<Downloader>(downloader);
|
||
}
|
||
|
||
ref<Downloader> makeDownloader()
|
||
{
|
||
return make_ref<CurlDownloader>();
|
||
}
|
||
|
||
std::future<DownloadResult> Downloader::enqueueDownload(const DownloadRequest & request)
|
||
{
|
||
auto promise = std::make_shared<std::promise<DownloadResult>>();
|
||
enqueueDownload(request,
|
||
[promise](const DownloadResult & result) { promise->set_value(result); },
|
||
[promise](std::exception_ptr exc) { promise->set_exception(exc); });
|
||
return promise->get_future();
|
||
}
|
||
|
||
DownloadResult Downloader::download(const DownloadRequest & request)
|
||
{
|
||
return enqueueDownload(request).get();
|
||
}
|
||
|
||
Path Downloader::downloadCached(ref<Store> store, const string & url_, bool unpack, string name, const Hash & expectedHash, string * effectiveUrl)
|
||
{
|
||
auto url = resolveUri(url_);
|
||
|
||
if (name == "") {
|
||
auto p = url.rfind('/');
|
||
if (p != string::npos) name = string(url, p + 1);
|
||
}
|
||
|
||
Path expectedStorePath;
|
||
if (expectedHash) {
|
||
expectedStorePath = store->makeFixedOutputPath(unpack, expectedHash, name);
|
||
if (store->isValidPath(expectedStorePath))
|
||
return expectedStorePath;
|
||
}
|
||
|
||
Path cacheDir = getCacheDir() + "/nix/tarballs";
|
||
createDirs(cacheDir);
|
||
|
||
string urlHash = hashString(htSHA256, url).to_string(Base32, false);
|
||
|
||
Path dataFile = cacheDir + "/" + urlHash + ".info";
|
||
Path fileLink = cacheDir + "/" + urlHash + "-file";
|
||
|
||
PathLocks lock({fileLink}, fmt("waiting for lock on ‘%1%’...", fileLink));
|
||
|
||
Path storePath;
|
||
|
||
string expectedETag;
|
||
|
||
int ttl = settings.tarballTtl;
|
||
bool skip = false;
|
||
|
||
if (pathExists(fileLink) && pathExists(dataFile)) {
|
||
storePath = readLink(fileLink);
|
||
store->addTempRoot(storePath);
|
||
if (store->isValidPath(storePath)) {
|
||
auto ss = tokenizeString<vector<string>>(readFile(dataFile), "\n");
|
||
if (ss.size() >= 3 && ss[0] == url) {
|
||
time_t lastChecked;
|
||
if (string2Int(ss[2], lastChecked) && lastChecked + ttl >= time(0)) {
|
||
skip = true;
|
||
if (effectiveUrl)
|
||
*effectiveUrl = url_;
|
||
} else if (!ss[1].empty()) {
|
||
debug(format("verifying previous ETag ‘%1%’") % ss[1]);
|
||
expectedETag = ss[1];
|
||
}
|
||
}
|
||
} else
|
||
storePath = "";
|
||
}
|
||
|
||
if (!skip) {
|
||
|
||
try {
|
||
DownloadRequest request(url);
|
||
request.expectedETag = expectedETag;
|
||
auto res = download(request);
|
||
if (effectiveUrl)
|
||
*effectiveUrl = res.effectiveUrl;
|
||
|
||
if (!res.cached) {
|
||
ValidPathInfo info;
|
||
StringSink sink;
|
||
dumpString(*res.data, sink);
|
||
Hash hash = hashString(expectedHash ? expectedHash.type : htSHA256, *res.data);
|
||
info.path = store->makeFixedOutputPath(false, hash, name);
|
||
info.narHash = hashString(htSHA256, *sink.s);
|
||
info.narSize = sink.s->size();
|
||
info.ca = makeFixedOutputCA(false, hash);
|
||
store->addToStore(info, sink.s, NoRepair, NoCheckSigs);
|
||
storePath = info.path;
|
||
}
|
||
|
||
assert(!storePath.empty());
|
||
replaceSymlink(storePath, fileLink);
|
||
|
||
writeFile(dataFile, url + "\n" + res.etag + "\n" + std::to_string(time(0)) + "\n");
|
||
} catch (DownloadError & e) {
|
||
if (storePath.empty()) throw;
|
||
printError(format("warning: %1%; using cached result") % e.msg());
|
||
}
|
||
}
|
||
|
||
if (unpack) {
|
||
Path unpackedLink = cacheDir + "/" + baseNameOf(storePath) + "-unpacked";
|
||
PathLocks lock2({unpackedLink}, fmt("waiting for lock on ‘%1%’...", unpackedLink));
|
||
Path unpackedStorePath;
|
||
if (pathExists(unpackedLink)) {
|
||
unpackedStorePath = readLink(unpackedLink);
|
||
store->addTempRoot(unpackedStorePath);
|
||
if (!store->isValidPath(unpackedStorePath))
|
||
unpackedStorePath = "";
|
||
}
|
||
if (unpackedStorePath.empty()) {
|
||
printInfo(format("unpacking ‘%1%’...") % url);
|
||
Path tmpDir = createTempDir();
|
||
AutoDelete autoDelete(tmpDir, true);
|
||
// FIXME: this requires GNU tar for decompression.
|
||
runProgram("tar", true, {"xf", storePath, "-C", tmpDir, "--strip-components", "1"});
|
||
unpackedStorePath = store->addToStore(name, tmpDir, true, htSHA256, defaultPathFilter, NoRepair);
|
||
}
|
||
replaceSymlink(unpackedStorePath, unpackedLink);
|
||
storePath = unpackedStorePath;
|
||
}
|
||
|
||
if (expectedStorePath != "" && storePath != expectedStorePath)
|
||
throw nix::Error("store path mismatch in file downloaded from ‘%s’", url);
|
||
|
||
return storePath;
|
||
}
|
||
|
||
|
||
bool isUri(const string & s)
|
||
{
|
||
if (s.compare(0, 8, "channel:") == 0) return true;
|
||
size_t pos = s.find("://");
|
||
if (pos == string::npos) return false;
|
||
string scheme(s, 0, pos);
|
||
return scheme == "http" || scheme == "https" || scheme == "file" || scheme == "channel" || scheme == "git" || scheme == "s3";
|
||
}
|
||
|
||
|
||
}
|