guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

treewide: drop PushActivity in favor of explicit context

Browse Source 
PushActivity does not work with async code since we have no such thing
as promise-local storage. it will be confusing at best, and completely
wrong at worst, with the current thread-local linking state. if we can
find a way to get promise-local storage we may want to bring this back
though, explicit context passing is rather error-prone. luckily we are
not using parent links for anything important, just to keep the multi-
line activity display from filling up with stuff we're already showing

Change-Id: Ie373d713080a3db811b2d5abd681f78137735e45
This commit is contained in:
eldritch horrors
2025-07-25 13:05:24 +02:00
parent e88a85bd92
commit 57b1c289b5
23 changed files with 316 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/binary-cache-store.cc
+59 -39
View File
@@ -68,20 +68,25 @@ try {
}
kj::Promise<Result<void>> BinaryCacheStore::upsertFile(
const std::string & path, std::string && data, const std::string & mimeType
const std::string & path,
std::string && data,
const std::string & mimeType,
const Activity * context
)
try {
TRY_AWAIT(upsertFile(path, std::make_shared<std::stringstream>(std::move(data)), mimeType));
TRY_AWAIT(
upsertFile(path, std::make_shared<std::stringstream>(std::move(data)), mimeType, context)
);
co_return result::success();
} catch (...) {
co_return result::current_exception();
}
kj::Promise<Result<std::optional<std::string>>>
BinaryCacheStore::getFileContents(const std::string & path)
BinaryCacheStore::getFileContents(const std::string & path, const Activity * context)
try {
try {
co_return TRY_AWAIT(TRY_AWAIT(getFile(path))->drain());
co_return TRY_AWAIT(TRY_AWAIT(getFile(path, context))->drain());
} catch (NoSuchBinaryCacheFile &) {
co_return std::nullopt;
}
@@ -94,11 +99,12 @@ std::string BinaryCacheStore::narInfoFileFor(const StorePath & storePath)
return std::string(storePath.hashPart()) + ".narinfo";
}
kj::Promise<Result<void>> BinaryCacheStore::writeNarInfo(ref<NarInfo> narInfo)
kj::Promise<Result<void>>
BinaryCacheStore::writeNarInfo(ref<NarInfo> narInfo, const Activity * context)
try {
auto narInfoFile = narInfoFileFor(narInfo->path);
TRY_AWAIT(upsertFile(narInfoFile, narInfo->to_string(*this), "text/x-nix-narinfo"));
TRY_AWAIT(upsertFile(narInfoFile, narInfo->to_string(*this), "text/x-nix-narinfo", context));
{
auto state_(co_await state.lock());
@@ -115,8 +121,12 @@ try {
}
kj::Promise<Result<ref<const ValidPathInfo>>> BinaryCacheStore::addToStoreCommon(
AsyncInputStream & narSource, RepairFlag repair, CheckSigsFlag checkSigs,
std::function<ValidPathInfo(HashResult)> mkInfo)
AsyncInputStream & narSource,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context,
std::function<ValidPathInfo(HashResult)> mkInfo
)
try {
auto [fdTemp, fnTemp] = createTempFile();
@@ -173,8 +183,9 @@ try {
reads, but typically they'll already be cached. */
for (auto & ref : info.references)
try {
if (ref != info.path)
TRY_AWAIT(queryPathInfo(ref));
if (ref != info.path) {
TRY_AWAIT(queryPathInfo(ref, context));
}
} catch (InvalidPath &) {
throw Error("cannot add '%s' to the binary cache because the reference '%s' does not exist",
printStorePath(info.path), printStorePath(ref));
@@ -189,9 +200,9 @@ try {
};
try {
TRY_AWAIT(
upsertFile(std::string(info.path.hashPart()) + ".ls", j.dump(), "application/json")
);
TRY_AWAIT(upsertFile(
std::string(info.path.hashPart()) + ".ls", j.dump(), "application/json", context
));
} catch (ForeignException & exc) {
if (exc.is<JSON::exception>()) {
warn(
@@ -273,12 +284,13 @@ try {
}
/* Atomically write the NAR file. */
if (repair || !TRY_AWAIT(fileExists(narInfo->url))) {
if (repair || !TRY_AWAIT(fileExists(narInfo->url, context))) {
stats.narWrite++;
TRY_AWAIT(upsertFile(
narInfo->url,
std::make_shared<std::fstream>(fnTemp, std::ios_base::in | std::ios_base::binary),
"application/x-nix-nar"
"application/x-nix-nar",
context
));
} else {
stats.narWriteAverted++;
@@ -291,7 +303,7 @@ try {
/* Atomically write the NAR info file.*/
if (secretKey) narInfo->sign(*this, *secretKey);
TRY_AWAIT(writeNarInfo(narInfo));
TRY_AWAIT(writeNarInfo(narInfo, context));
stats.narInfoWrite++;
@@ -304,22 +316,24 @@ kj::Promise<Result<void>> BinaryCacheStore::addToStore(
const ValidPathInfo & info,
AsyncInputStream & narSource,
RepairFlag repair,
CheckSigsFlag checkSigs
CheckSigsFlag checkSigs,
const Activity * context
)
try {
if (!repair && TRY_AWAIT(isValidPath(info.path))) {
if (!repair && TRY_AWAIT(isValidPath(info.path, context))) {
// FIXME: copyNAR -> null sink
TRY_AWAIT(narSource.drain());
co_return result::success();
}
TRY_AWAIT(addToStoreCommon(narSource, repair, checkSigs, {[&](HashResult nar) {
/* FIXME reinstate these, once we can correctly do hash modulo sink as
needed. We need to throw here in case we uploaded a corrupted store path. */
// assert(info.narHash == nar.first);
// assert(info.narSize == nar.second);
return info;
}}));
TRY_AWAIT(addToStoreCommon(narSource, repair, checkSigs, context, {[&](HashResult nar) {
/* FIXME reinstate these, once we can correctly do hash modulo
sink as needed. We need to throw here in case we uploaded a
corrupted store path. */
// assert(info.narHash == nar.first);
// assert(info.narSize == nar.second);
return info;
}}));
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -336,7 +350,7 @@ kj::Promise<Result<StorePath>> BinaryCacheStore::addToStoreFromDump(
try {
if (method != FileIngestionMethod::Recursive || hashAlgo != HashType::SHA256)
unsupported("addToStoreFromDump");
co_return TRY_AWAIT(addToStoreCommon(dump, repair, CheckSigs, [&](HashResult nar) {
co_return TRY_AWAIT(addToStoreCommon(dump, repair, CheckSigs, nullptr, [&](HashResult nar) {
ValidPathInfo info {
*this,
name,
@@ -358,12 +372,13 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<bool>> BinaryCacheStore::isValidPathUncached(const StorePath & storePath)
kj::Promise<Result<bool>>
BinaryCacheStore::isValidPathUncached(const StorePath & storePath, const Activity * context)
try {
// FIXME: this only checks whether a .narinfo with a matching hash
// part exists. So ‘f4kb...-foo’ matches ‘f4kb...-bar’, even
// though they shouldn't. Not easily fixed.
co_return TRY_AWAIT(fileExists(narInfoFileFor(storePath)));
co_return TRY_AWAIT(fileExists(narInfoFileFor(storePath), context));
} catch (...) {
co_return result::current_exception();
}
@@ -383,7 +398,7 @@ try {
}
kj::Promise<Result<box_ptr<AsyncInputStream>>>
BinaryCacheStore::narFromPath(const StorePath & storePath)
BinaryCacheStore::narFromPath(const StorePath & storePath, const Activity * context)
try {
struct NarFromPath : AsyncInputStream
{
@@ -416,12 +431,12 @@ try {
}
};
auto info_ = TRY_AWAIT(queryPathInfo(storePath)).try_cast<const NarInfo>();
auto info_ = TRY_AWAIT(queryPathInfo(storePath, context)).try_cast<const NarInfo>();
assert(info_ && "binary cache queryPathInfo didn't return a NarInfo");
auto & info = *info_;
try {
auto file = TRY_AWAIT(getFile(info->url));
auto file = TRY_AWAIT(getFile(info->url, context));
co_return make_box_ptr<NarFromPath>(stats, info->compression, std::move(file));
} catch (NoSuchBinaryCacheFile & e) {
throw SubstituteGone(std::move(e.info()));
@@ -431,17 +446,22 @@ try {
}
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
BinaryCacheStore::queryPathInfoUncached(const StorePath & storePath)
BinaryCacheStore::queryPathInfoUncached(const StorePath & storePath, const Activity * context)
try {
auto uri = getUri();
auto storePathS = printStorePath(storePath);
auto act = std::make_shared<Activity>(*logger, lvlTalkative, actQueryPathInfo,
fmt("querying info about '%s' on '%s'", storePathS, uri), Logger::Fields{storePathS, uri});
PushActivity pact(act->id);
auto act = std::make_shared<Activity>(
*logger,
lvlTalkative,
actQueryPathInfo,
fmt("querying info about '%s' on '%s'", storePathS, uri),
Logger::Fields{storePathS, uri},
context ? context->id : 0
);
auto narInfoFile = narInfoFileFor(storePath);
auto data = TRY_AWAIT(getFileContents(narInfoFile));
auto data = TRY_AWAIT(getFileContents(narInfoFile, act.get()));
if (!data) co_return result::success(nullptr);
@@ -489,7 +509,7 @@ try {
auto h = sink.finish().first;
auto source = AsyncGeneratorInputStream{_source.dump()};
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, nullptr, [&](HashResult nar) {
return makeAddToStoreInfo(nar, *this, FileIngestionMethod::Recursive, name, h);
}))->path;
} catch (...) {
@@ -511,7 +531,7 @@ try {
auto h = sink.finish().first;
auto source = AsyncGeneratorInputStream{dumpPath(srcPath)};
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, nullptr, [&](HashResult nar) {
return makeAddToStoreInfo(nar, *this, FileIngestionMethod::Flat, name, h);
}))->path;
} catch (...) {
@@ -533,7 +553,7 @@ try {
StringSink sink;
sink << dumpString(s);
AsyncStringInputStream source(sink.s);
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
co_return TRY_AWAIT(addToStoreCommon(source, repair, CheckSigs, nullptr, [&](HashResult nar) {
ValidPathInfo info {
*this,
std::string { name },
lix/libstore/binary-cache-store.hh
+30 -14
View File
@@ -73,28 +73,32 @@ public:
BinaryCacheStoreConfig & config() override = 0;
const BinaryCacheStoreConfig & config() const override = 0;
virtual kj::Promise<Result<bool>> fileExists(const std::string & path) = 0;
virtual kj::Promise<Result<bool>>
fileExists(const std::string & path, const Activity * context = nullptr) = 0;
virtual kj::Promise<Result<void>> upsertFile(
const std::string & path,
std::shared_ptr<std::basic_iostream<char>> istream,
const std::string & mimeType
const std::string & mimeType,
const Activity * context
) = 0;
kj::Promise<Result<void>> upsertFile(
const std::string & path,
// FIXME: use std::string_view
std::string && data,
const std::string & mimeType
const std::string & mimeType,
const Activity * context = nullptr
);
/**
* Dump the contents of the specified file to a sink.
*/
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> getFile(const std::string & path) = 0;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>>
getFile(const std::string & path, const Activity * context = nullptr) = 0;
virtual kj::Promise<Result<std::optional<std::string>>> getFileContents(const std::string & path
);
virtual kj::Promise<Result<std::optional<std::string>>>
getFileContents(const std::string & path, const Activity * context = nullptr);
public:
@@ -106,24 +110,35 @@ private:
std::string narInfoFileFor(const StorePath & storePath);
kj::Promise<Result<void>> writeNarInfo(ref<NarInfo> narInfo);
kj::Promise<Result<void>>
writeNarInfo(ref<NarInfo> narInfo, const Activity * context = nullptr);
kj::Promise<Result<ref<const ValidPathInfo>>> addToStoreCommon(
AsyncInputStream & narSource, RepairFlag repair, CheckSigsFlag checkSigs,
std::function<ValidPathInfo(HashResult)> mkInfo);
AsyncInputStream & narSource,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context,
std::function<ValidPathInfo(HashResult)> mkInfo
);
public:
kj::Promise<Result<bool>> isValidPathUncached(const StorePath & path) override;
kj::Promise<Result<bool>>
isValidPathUncached(const StorePath & path, const Activity * context = nullptr) override;
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) override;
queryPathInfoUncached(const StorePath & path, const Activity * context = nullptr) override;
kj::Promise<Result<std::optional<StorePath>>>
queryPathFromHashPart(const std::string & hashPart) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & narSource,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<void>> addToStore(
const ValidPathInfo & info,
AsyncInputStream & narSource,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context
) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
@@ -151,7 +166,8 @@ public:
const StorePathSet & references,
RepairFlag repair) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override;
ref<FSAccessor> getFSAccessor() override;
lix/libstore/build/derivation-goal.cc
+7 -4
View File
@@ -840,10 +840,13 @@ void runPostBuildHook(
if (hook == "")
return;
Activity act(logger, lvlTalkative, actPostBuildHook,
fmt("running post-build-hook '%s'", settings.postBuildHook),
Logger::Fields{store.printStorePath(drvPath)});
PushActivity pact(act.id);
Activity act(
logger,
lvlTalkative,
actPostBuildHook,
fmt("running post-build-hook '%s'", settings.postBuildHook),
Logger::Fields{store.printStorePath(drvPath)}
);
std::map<std::string, std::string> hookEnvironment = getEnv();
auto drvPathPretty = store.printStorePath(drvPath);
lix/libstore/build/substitution-goal.cc
+7 -3
View File
@@ -219,15 +219,19 @@ try {
try {
ReceiveInterrupts receiveInterrupts;
Activity act(*logger, actSubstitute, Logger::Fields{worker.store.printStorePath(storePath), sub->getUri()});
PushActivity pact(act.id);
Activity act(
*logger,
actSubstitute,
Logger::Fields{worker.store.printStorePath(storePath), sub->getUri()}
);
aio.blockOn(copyStorePath(
*sub,
worker.store,
fetchPath,
repair,
sub->config().isTrusted ? NoCheckSigs : CheckSigs
sub->config().isTrusted ? NoCheckSigs : CheckSigs,
&act
));
} catch (const EndOfFile &) {
throw EndOfFile(
lix/libstore/dummy-store.cc
+20 -6
View File
@@ -36,7 +36,7 @@ struct DummyStore final : public Store
}
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) override
queryPathInfoUncached(const StorePath & path, const Activity * context) override
{
return {result::success(nullptr)};
}
@@ -61,9 +61,18 @@ struct DummyStore final : public Store
return {result::current_exception()};
}
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & source,
RepairFlag repair, CheckSigsFlag checkSigs) override
try { unsupported("addToStore"); } catch (...) { return {result::current_exception()}; }
kj::Promise<Result<void>> addToStore(
const ValidPathInfo & info,
AsyncInputStream & source,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context
) override
try {
unsupported("addToStore");
} catch (...) {
return {result::current_exception()};
}
kj::Promise<Result<StorePath>> addTextToStore(
std::string_view name,
@@ -72,8 +81,13 @@ struct DummyStore final : public Store
RepairFlag repair) override
try { unsupported("addTextToStore"); } catch (...) { return {result::current_exception()}; }
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
try { unsupported("narFromPath"); } catch (...) { return {result::current_exception()}; }
kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override
try {
unsupported("narFromPath");
} catch (...) {
return {result::current_exception()};
}
virtual ref<FSAccessor> getFSAccessor() override
{ unsupported("getFSAccessor"); }
lix/libstore/filetransfer.cc
+23 -13
View File
@@ -741,10 +741,14 @@ struct curlFileTransfer : public FileTransfer
}
#endif
kj::Promise<Result<void>>
upload(const std::string & uri, std::string data, FileTransferOptions options) override
kj::Promise<Result<void>> 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));
TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::move(data), false, context));
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -873,15 +877,17 @@ struct curlFileTransfer : public FileTransfer
const std::string & uri,
FileTransferOptions && options,
std::optional<std::string> data,
bool noBody
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<TransferStream>(*this, uri, std::move(options), std::move(data), noBody);
auto source = make_box_ptr<TransferStream>(
*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)};
@@ -896,7 +902,7 @@ struct curlFileTransfer : public FileTransfer
FileTransferOptions options;
std::optional<std::string> data;
bool noBody;
ActivityId parentAct = getCurActivity();
ActivityId parentAct;
std::shared_ptr<TransferItem> transfer;
FileTransferResult metadata;
@@ -912,13 +918,15 @@ struct curlFileTransfer : public FileTransfer
const std::string & uri,
FileTransferOptions && options,
std::optional<std::string> data,
bool noBody
bool noBody,
const Activity * context
)
: parent(parent)
, uri(uri)
, options(options)
, data(std::move(data))
, noBody(noBody)
, parentAct(context ? context->id : 0)
{
}
@@ -1110,10 +1118,11 @@ struct curlFileTransfer : public FileTransfer
}
};
kj::Promise<Result<bool>> exists(const std::string & uri, FileTransferOptions options) override
kj::Promise<Result<bool>>
exists(const std::string & uri, FileTransferOptions options, const Activity * context) override
try {
try {
TRY_AWAIT(enqueueFileTransfer(uri, std::move(options), std::nullopt, true));
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
@@ -1126,10 +1135,11 @@ struct curlFileTransfer : public FileTransfer
co_return result::current_exception();
}
kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>>
download(const std::string & uri, FileTransferOptions options) override
kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>> download(
const std::string & uri, FileTransferOptions options, const Activity * context
) override
{
return enqueueFileTransfer(uri, std::move(options), std::nullopt, false);
return enqueueFileTransfer(uri, std::move(options), std::nullopt, false, context);
}
};
lix/libstore/filetransfer.hh
+16 -6
View File
@@ -56,8 +56,12 @@ struct FileTransfer
/**
* Upload some data. May throw a FileTransferError exception.
*/
virtual kj::Promise<Result<void>>
upload(const std::string & uri, std::string data, FileTransferOptions options = {}) = 0;
virtual kj::Promise<Result<void>> upload(
const std::string & uri,
std::string data,
FileTransferOptions options = {},
const Activity * context = nullptr
) = 0;
/**
* Checks whether the given URI exists. For historical reasons this function
@@ -69,8 +73,11 @@ struct FileTransfer
*
* S3 objects are downloaded completely to answer this request.
*/
virtual kj::Promise<Result<bool>>
exists(const std::string & uri, FileTransferOptions options = {}) = 0;
virtual kj::Promise<Result<bool>> exists(
const std::string & uri,
FileTransferOptions options = {},
const Activity * context = nullptr
) = 0;
/**
* Download a file, returning its contents through a source. Will not return
@@ -79,8 +86,11 @@ struct FileTransfer
* thrown by the returned source. The source will only throw errors detected
* during the transfer itself (decompression errors, connection drops, etc).
*/
virtual kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>>
download(const std::string & uri, FileTransferOptions options = {}) = 0;
virtual kj::Promise<Result<std::pair<FileTransferResult, box_ptr<AsyncInputStream>>>> download(
const std::string & uri,
FileTransferOptions options = {},
const Activity * context = nullptr
) = 0;
enum Error { NotFound, Forbidden, Misc, Transient, Interrupted };
};
lix/libstore/http-binary-cache-store.cc
+9 -6
View File
@@ -82,12 +82,13 @@ void HttpBinaryCacheStore::checkEnabled()
throw SubstituterDisabled("substituter '%s' is disabled", getUri());
}
kj::Promise<Result<bool>> HttpBinaryCacheStore::fileExists(const std::string & path)
kj::Promise<Result<bool>>
HttpBinaryCacheStore::fileExists(const std::string & path, const Activity * context)
try {
checkEnabled();
try {
co_return TRY_AWAIT(getFileTransfer()->exists(makeURI(path), makeOptions()));
co_return TRY_AWAIT(getFileTransfer()->exists(makeURI(path), makeOptions(), context));
} catch (FileTransferError & e) {
maybeDisable();
throw;
@@ -99,13 +100,14 @@ try {
kj::Promise<Result<void>> HttpBinaryCacheStore::upsertFile(
const std::string & path,
std::shared_ptr<std::basic_iostream<char>> istream,
const std::string & mimeType
const std::string & mimeType,
const Activity * context
)
try {
auto data = StreamToSourceAdapter(istream).drain();
try {
TRY_AWAIT(getFileTransfer()->upload(
makeURI(path), std::move(data), makeOptions({{"Content-Type", mimeType}})
makeURI(path), std::move(data), makeOptions({{"Content-Type", mimeType}}), context
));
} catch (FileTransferError & e) {
throw UploadToHTTP("while uploading to HTTP binary cache at '%s': %s", cacheUri, e.msg());
@@ -116,11 +118,12 @@ try {
}
kj::Promise<Result<box_ptr<AsyncInputStream>>>
HttpBinaryCacheStore::getFile(const std::string & path)
HttpBinaryCacheStore::getFile(const std::string & path, const Activity * context)
try {
checkEnabled();
try {
co_return TRY_AWAIT(getFileTransfer()->download(makeURI(path), makeOptions())).second;
co_return TRY_AWAIT(getFileTransfer()->download(makeURI(path), makeOptions(), context))
.second;
} catch (FileTransferError & e) {
if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden) {
throw NoSuchBinaryCacheFile(
lix/libstore/http-binary-cache-store.hh
+4 -3
View File
@@ -80,13 +80,14 @@ protected:
void maybeDisable();
void checkEnabled();
kj::Promise<Result<bool>> fileExists(const std::string & path) override;
kj::Promise<Result<bool>> fileExists(const std::string & path, const Activity * context) override;
kj::Promise<Result<void>> upsertFile(
const std::string & path,
std::shared_ptr<std::basic_iostream<char>> istream,
const std::string & mimeType
const std::string & mimeType,
const Activity * context
) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>> getFile(const std::string & path) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>> getFile(const std::string & path, const Activity * context) override;
std::string makeURI(const std::string & path)
{
lix/libstore/legacy-ssh-store.cc
+10 -4
View File
@@ -277,7 +277,7 @@ struct LegacySSHStore final : public Store
}
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) override
queryPathInfoUncached(const StorePath & path, const Activity * context) override
try {
auto conn(TRY_AWAIT(connections->get()));
@@ -297,8 +297,13 @@ struct LegacySSHStore final : public Store
co_return result::current_exception();
}
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & source,
RepairFlag repair, CheckSigsFlag checkSigs) override
kj::Promise<Result<void>> addToStore(
const ValidPathInfo & info,
AsyncInputStream & source,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context
) override
try {
debug("adding path '%s' to remote host '%s'", printStorePath(info.path), host);
@@ -343,7 +348,8 @@ struct LegacySSHStore final : public Store
co_return result::current_exception();
}
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override
try {
auto conn(TRY_AWAIT(connections->get()));
lix/libstore/local-binary-cache-store.cc
+9 -4
View File
@@ -3,6 +3,7 @@
#include "lix/libstore/globals.hh"
#include "lix/libstore/nar-info-disk-cache.hh"
#include "lix/libutil/async-io.hh"
#include "lix/libutil/logging.hh"
#include "lix/libutil/result.hh"
#include <atomic>
@@ -57,12 +58,14 @@ public:
protected:
kj::Promise<Result<bool>> fileExists(const std::string & path) override;
kj::Promise<Result<bool>>
fileExists(const std::string & path, const Activity * context) override;
kj::Promise<Result<void>> upsertFile(
const std::string & path,
std::shared_ptr<std::basic_iostream<char>> istream,
const std::string & mimeType
const std::string & mimeType,
const Activity * context
) override
try {
auto path2 = binaryCacheDir + "/" + path;
@@ -78,7 +81,8 @@ protected:
return {result::current_exception()};
}
kj::Promise<Result<box_ptr<AsyncInputStream>>> getFile(const std::string & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>>
getFile(const std::string & path, const Activity * context) override
try {
try {
return {
@@ -130,7 +134,8 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<bool>> LocalBinaryCacheStore::fileExists(const std::string & path)
kj::Promise<Result<bool>>
LocalBinaryCacheStore::fileExists(const std::string & path, const Activity * context)
try {
return {pathExists(binaryCacheDir + "/" + path)};
} catch (...) {
lix/libstore/local-fs-store.cc
+4 -2
View File
@@ -85,10 +85,12 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
}
kj::Promise<Result<box_ptr<AsyncInputStream>>> LocalFSStore::narFromPath(const StorePath & path)
kj::Promise<Result<box_ptr<AsyncInputStream>>>
LocalFSStore::narFromPath(const StorePath & path, const Activity * context)
try {
if (!TRY_AWAIT(isValidPath(path)))
if (!TRY_AWAIT(isValidPath(path, context))) {
throw Error("path '%s' does not exist in store", printStorePath(path));
}
co_return make_box_ptr<AsyncGeneratorInputStream>(
dumpPath(getRealStoreDir() + std::string(printStorePath(path), config().storeDir.size()))
);
lix/libstore/local-fs-store.hh
+2 -1
View File
@@ -43,7 +43,8 @@ public:
LocalFSStoreConfig & config() override = 0;
const LocalFSStoreConfig & config() const override = 0;
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override;
kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override;
ref<FSAccessor> getFSAccessor() override;
/**
lix/libstore/local-store.cc
+7 -8
View File
@@ -686,9 +686,8 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
LocalStore::queryPathInfoUncached(const StorePath & path)
LocalStore::queryPathInfoUncached(const StorePath & path, const Activity * context)
try {
co_return TRY_AWAIT(
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
@@ -705,7 +704,6 @@ try {
co_return result::current_exception();
}
std::shared_ptr<const ValidPathInfo> LocalStore::queryPathInfoInternal(DBState & state, const StorePath & path)
{
/* Get the path info. */
@@ -784,8 +782,8 @@ bool LocalStore::isValidPath_(DBState & state, const StorePath & path)
return state.stmts->QueryPathInfo.use()(printStorePath(path)).next();
}
kj::Promise<Result<bool>> LocalStore::isValidPathUncached(const StorePath & path)
kj::Promise<Result<bool>>
LocalStore::isValidPathUncached(const StorePath & path, const Activity * context)
try {
// NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
co_return TRY_AWAIT(retrySQLite([&]() -> kj::Promise<Result<bool>> {
@@ -1088,7 +1086,8 @@ kj::Promise<Result<void>> LocalStore::addToStore(
const ValidPathInfo & info,
AsyncInputStream & source,
RepairFlag repair,
CheckSigsFlag checkSigs
CheckSigsFlag checkSigs,
const Activity * context
)
try {
if (checkSigs && pathInfoIsUntrusted(info))
@@ -1099,7 +1098,7 @@ try {
TRY_AWAIT(addTempRoot(info.path));
if (repair || !TRY_AWAIT(isValidPath(info.path))) {
if (repair || !TRY_AWAIT(isValidPath(info.path, context))) {
std::optional<PathLock> outputLock;
@@ -1111,7 +1110,7 @@ try {
if (!locksHeld.count(printStorePath(info.path)))
outputLock = TRY_AWAIT(lockPathAsync(realPath));
if (repair || !TRY_AWAIT(isValidPath(info.path))) {
if (repair || !TRY_AWAIT(isValidPath(info.path, context))) {
deletePath(realPath);
lix/libstore/local-store.hh
+10 -4
View File
@@ -198,7 +198,8 @@ public:
std::string getUri() override;
kj::Promise<Result<bool>> isValidPathUncached(const StorePath & path) override;
kj::Promise<Result<bool>>
isValidPathUncached(const StorePath & path, const Activity * context) override;
kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
SubstituteFlag maybeSubstitute = NoSubstitute) override;
@@ -206,7 +207,7 @@ public:
kj::Promise<Result<StorePathSet>> queryAllValidPaths() override;
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) override;
queryPathInfoUncached(const StorePath & path, const Activity * context) override;
kj::Promise<Result<void>>
queryReferrers(const StorePath & path, StorePathSet & referrers) override;
@@ -223,8 +224,13 @@ public:
bool pathInfoIsUntrusted(const ValidPathInfo &) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & source,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<void>> addToStore(
const ValidPathInfo & info,
AsyncInputStream & source,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context
) override;
kj::Promise<Result<StorePath>> addToStoreFromDump(
AsyncInputStream & dump,
lix/libstore/remote-store.cc
+7 -6
View File
@@ -207,7 +207,8 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<bool>> RemoteStore::isValidPathUncached(const StorePath & path)
kj::Promise<Result<bool>>
RemoteStore::isValidPathUncached(const StorePath & path, const Activity * context)
try {
auto conn(TRY_AWAIT(getConnection()));
co_return TRY_AWAIT(
@@ -264,9 +265,8 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
RemoteStore::queryPathInfoUncached(const StorePath & path)
RemoteStore::queryPathInfoUncached(const StorePath & path, const Activity * context)
try {
auto conn(TRY_AWAIT(getConnection()));
std::optional<UnkeyedValidPathInfo> pathInfo;
@@ -402,12 +402,12 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<void>> RemoteStore::addToStore(
const ValidPathInfo & info,
AsyncInputStream & source,
RepairFlag repair,
CheckSigsFlag checkSigs
CheckSigsFlag checkSigs,
const Activity * context
)
try {
auto conn(TRY_AWAIT(getConnection()));
@@ -721,7 +721,8 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<box_ptr<AsyncInputStream>>> RemoteStore::narFromPath(const StorePath & path)
kj::Promise<Result<box_ptr<AsyncInputStream>>>
RemoteStore::narFromPath(const StorePath & path, const Activity * context)
try {
struct NarStream : AsyncInputStream
{
lix/libstore/remote-store.hh
+12 -6
View File
@@ -51,7 +51,8 @@ public:
/* Implementations of abstract store API methods. */
kj::Promise<Result<bool>> isValidPathUncached(const StorePath & path) override;
kj::Promise<Result<bool>>
isValidPathUncached(const StorePath & path, const Activity * context) override;
kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
SubstituteFlag maybeSubstitute = NoSubstitute) override;
@@ -59,7 +60,7 @@ public:
kj::Promise<Result<StorePathSet>> queryAllValidPaths() override;
kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) override;
queryPathInfoUncached(const StorePath & path, const Activity * context) override;
kj::Promise<Result<void>>
queryReferrers(const StorePath & path, StorePathSet & referrers) override;
@@ -99,8 +100,13 @@ public:
const StorePathSet & references = StorePathSet()
) override;
kj::Promise<Result<void>> addToStore(const ValidPathInfo & info, AsyncInputStream & nar,
RepairFlag repair, CheckSigsFlag checkSigs) override;
kj::Promise<Result<void>> addToStore(
const ValidPathInfo & info,
AsyncInputStream & nar,
RepairFlag repair,
CheckSigsFlag checkSigs,
const Activity * context
) override;
kj::Promise<Result<void>> addMultipleToStore(
PathsSource & pathsToCopy,
@@ -196,8 +202,8 @@ protected:
virtual ref<FSAccessor> getFSAccessor() override;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path
) override;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override;
private:
lix/libstore/s3-binary-cache-store.cc
+9 -5
View File
@@ -331,9 +331,10 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
fetches the .narinfo file, rather than first checking for its
existence via a HEAD request. Since .narinfos are small, doing
a GET is unlikely to be slower than HEAD. */
kj::Promise<Result<bool>> isValidPathUncached(const StorePath & storePath) override
kj::Promise<Result<bool>>
isValidPathUncached(const StorePath & storePath, const Activity * context) override
try {
TRY_AWAIT(queryPathInfo(storePath));
TRY_AWAIT(queryPathInfo(storePath, context));
co_return true;
} catch (InvalidPath & e) {
co_return false;
@@ -341,7 +342,8 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
co_return result::current_exception();
}
kj::Promise<Result<bool>> fileExists(const std::string & path) override
kj::Promise<Result<bool>>
fileExists(const std::string & path, const Activity * context) override
try {
stats.head++;
@@ -508,7 +510,8 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
kj::Promise<Result<void>> upsertFile(
const std::string & path,
std::shared_ptr<std::basic_iostream<char>> istream,
const std::string & mimeType
const std::string & mimeType,
const Activity * context
) override
try {
auto compress = [&](std::string compression)
@@ -536,7 +539,8 @@ struct S3BinaryCacheStoreImpl : public S3BinaryCacheStore
co_return result::current_exception();
}
kj::Promise<Result<box_ptr<AsyncInputStream>>> getFile(const std::string & path) override
kj::Promise<Result<box_ptr<AsyncInputStream>>>
getFile(const std::string & path, const Activity * context) override
try {
stats.get++;
lix/libstore/store-api.cc
+37 -27
View File
@@ -8,6 +8,7 @@
#include "lix/libutil/box_ptr.hh"
#include "lix/libutil/hash.hh"
#include "lix/libutil/json.hh"
#include "lix/libutil/logging.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/serialise.hh"
#include "lix/libutil/sync.hh"
@@ -25,6 +26,7 @@
#include <functional>
#include <kj/async.h>
#include <memory>
#include <mutex>
#include <regex>
@@ -600,8 +602,7 @@ try {
co_return result::current_exception();
}
kj::Promise<Result<bool>> Store::isValidPath(const StorePath & storePath)
kj::Promise<Result<bool>> Store::isValidPath(const StorePath & storePath, const Activity * context)
try {
{
auto state_(co_await state.lock());
@@ -623,7 +624,7 @@ try {
}
}
bool valid = TRY_AWAIT(isValidPathUncached(storePath));
bool valid = TRY_AWAIT(isValidPathUncached(storePath, context));
if (diskCache && !valid)
// FIXME: handle valid = true case.
@@ -634,12 +635,12 @@ try {
co_return result::current_exception();
}
/* Default implementation for stores that only implement
queryPathInfoUncached(). */
kj::Promise<Result<bool>> Store::isValidPathUncached(const StorePath & path)
kj::Promise<Result<bool>>
Store::isValidPathUncached(const StorePath & path, const Activity * context)
try {
TRY_AWAIT(queryPathInfo(path));
TRY_AWAIT(queryPathInfo(path, context));
co_return true;
} catch (InvalidPath &) {
co_return false;
@@ -663,8 +664,8 @@ static void ensureGoodStorePath(Store * store, const StorePath & expected, const
}
}
kj::Promise<Result<ref<const ValidPathInfo>>> Store::queryPathInfo(const StorePath & storePath)
kj::Promise<Result<ref<const ValidPathInfo>>>
Store::queryPathInfo(const StorePath & storePath, const Activity * context)
try {
auto hashPart = std::string(storePath.hashPart());
@@ -693,7 +694,7 @@ try {
}
}
auto info = TRY_AWAIT(queryPathInfoUncached(storePath));
auto info = TRY_AWAIT(queryPathInfoUncached(storePath, context));
if (info) {
// first, before we cache anything, check that the store gave us valid data.
ensureGoodStorePath(this, storePath, info->path);
@@ -1039,22 +1040,29 @@ kj::Promise<Result<void>> copyStorePath(
Store & dstStore,
const StorePath & storePath,
RepairFlag repair,
CheckSigsFlag checkSigs)
CheckSigsFlag checkSigs,
const Activity * context
)
try {
/* Bail out early (before starting a download from srcStore) if
dstStore already has this path. */
if (!repair && TRY_AWAIT(dstStore.isValidPath(storePath)))
if (!repair && TRY_AWAIT(dstStore.isValidPath(storePath, context))) {
co_return result::success();
}
auto srcUri = srcStore.getUri();
auto dstUri = dstStore.getUri();
auto storePathS = srcStore.printStorePath(storePath);
Activity act(*logger, lvlInfo, actCopyPath,
Activity act(
*logger,
lvlInfo,
actCopyPath,
makeCopyPathMessage(srcUri, dstUri, storePathS),
{storePathS, srcUri, dstUri});
PushActivity pact(act.id);
{storePathS, srcUri, dstUri},
context ? context->id : 0
);
auto info = TRY_AWAIT(srcStore.queryPathInfo(storePath));
auto info = TRY_AWAIT(srcStore.queryPathInfo(storePath, &act));
// recompute store path on the chance dstStore does it differently
if (info->ca && info->references.empty()) {
@@ -1073,8 +1081,8 @@ try {
info = info2;
}
CopyPathStream source{act, info->narSize, TRY_AWAIT(srcStore.narFromPath(storePath))};
TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs));
CopyPathStream source{act, info->narSize, TRY_AWAIT(srcStore.narFromPath(storePath, &act))};
TRY_AWAIT(dstStore.addToStore(*info, source, repair, checkSigs, &act));
co_return result::success();
} catch (...) {
co_return result::current_exception();
@@ -1157,17 +1165,15 @@ try {
struct SinglePathStream : CopyPathStream
{
Activity act;
PushActivity pact{act.id};
std::shared_ptr<Activity> act;
SinglePathStream(
std::string message,
Logger::Fields fields,
const std::shared_ptr<Activity> & act,
size_t expected,
box_ptr<AsyncInputStream> inner
)
: CopyPathStream(act, expected, std::move(inner))
, act(*logger, lvlInfo, actCopyPath, message, fields)
: CopyPathStream(*act, expected, std::move(inner))
, act(act)
{
}
};
@@ -1180,12 +1186,16 @@ try {
auto srcUri = srcStore.getUri();
auto dstUri = dstStore.getUri();
auto storePathS = srcStore.printStorePath(missingPath);
auto act = std::make_shared<Activity>(
*logger,
lvlInfo,
actCopyPath,
makeCopyPathMessage(srcUri, dstUri, storePathS),
Logger::Fields{storePathS, srcUri, dstUri}
);
co_return make_box_ptr<SinglePathStream>(
makeCopyPathMessage(srcUri, dstUri, storePathS),
Logger::Fields{storePathS, srcUri, dstUri},
info->narSize,
TRY_AWAIT(srcStore.narFromPath(missingPath))
act, info->narSize, TRY_AWAIT(srcStore.narFromPath(missingPath, act.get()))
);
} catch (...) {
co_return result::current_exception();
lix/libstore/store-api.hh
+14 -9
View File
@@ -367,11 +367,13 @@ public:
* Check whether a path is valid.
* A path is valid when it exists in the store *now*.
*/
kj::Promise<Result<bool>> isValidPath(const StorePath & path);
kj::Promise<Result<bool>>
isValidPath(const StorePath & path, const Activity * context = nullptr);
protected:
virtual kj::Promise<Result<bool>> isValidPathUncached(const StorePath & path);
virtual kj::Promise<Result<bool>>
isValidPathUncached(const StorePath & path, const Activity * context = nullptr);
public:
@@ -406,8 +408,8 @@ public:
* Query information about a valid path. It is permitted to omit
* the name part of the store path.
*/
kj::Promise<Result<ref<const ValidPathInfo>>> queryPathInfo(const StorePath & path);
kj::Promise<Result<ref<const ValidPathInfo>>>
queryPathInfo(const StorePath & path, const Activity * context = nullptr);
/**
* Check whether the given valid path info is sufficiently attested, by
@@ -432,7 +434,7 @@ protected:
* Note to implementors: should return `nullptr` when the path is not found.
*/
virtual kj::Promise<Result<std::shared_ptr<const ValidPathInfo>>>
queryPathInfoUncached(const StorePath & path) = 0;
queryPathInfoUncached(const StorePath & path, const Activity * context = nullptr) = 0;
public:
@@ -512,7 +514,8 @@ public:
const ValidPathInfo & info,
AsyncInputStream & narSource,
RepairFlag repair = NoRepair,
CheckSigsFlag checkSigs = CheckSigs
CheckSigsFlag checkSigs = CheckSigs,
const Activity * context = nullptr
) = 0;
/**
@@ -591,7 +594,8 @@ public:
/**
* Generate a NAR dump of a store path.
*/
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) = 0;
virtual kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context = nullptr) = 0;
/**
* For each path, if it's a derivation, build it. Building a
@@ -927,8 +931,9 @@ kj::Promise<Result<void>> copyStorePath(
Store & dstStore,
const StorePath & storePath,
RepairFlag repair = NoRepair,
CheckSigsFlag checkSigs = CheckSigs);
CheckSigsFlag checkSigs = CheckSigs,
const Activity * context = nullptr
);
/**
* Copy store paths from one store to another. The paths may be copied
lix/libstore/uds-remote-store.hh
+5 -2
View File
@@ -43,8 +43,11 @@ public:
ref<FSAccessor> getFSAccessor() override
{ return LocalFSStore::getFSAccessor(); }
kj::Promise<Result<box_ptr<AsyncInputStream>>> narFromPath(const StorePath & path) override
{ return LocalFSStore::narFromPath(path); }
kj::Promise<Result<box_ptr<AsyncInputStream>>>
narFromPath(const StorePath & path, const Activity * context) override
{
return LocalFSStore::narFromPath(path, context);
}
/**
* Implementation of `IndirectRootStore::addIndirectRoot()` which
lix/libutil/logging.cc
-11
View File
@@ -19,17 +19,6 @@ LoggerSettings loggerSettings;
static GlobalConfig::Register rLoggerSettings(&loggerSettings);
static thread_local ActivityId curActivity = 0;
ActivityId getCurActivity()
{
return curActivity;
}
void setCurActivity(const ActivityId activityId)
{
curActivity = activityId;
}
Logger * logger = makeSimpleLogger(true);
void Logger::warn(const std::string & msg)
lix/libutil/logging.hh
+15 -38
View File
@@ -178,21 +178,28 @@ struct nop
{ }
};
ActivityId getCurActivity();
void setCurActivity(const ActivityId activityId);
struct Activity
{
Logger & logger;
const ActivityId id;
Activity(Logger & logger, Verbosity lvl, ActivityType type, const std::string & s = "",
const Logger::Fields & fields = {}, ActivityId parent = getCurActivity());
Activity(
Logger & logger,
Verbosity lvl,
ActivityType type,
const std::string & s = "",
const Logger::Fields & fields = {},
ActivityId parent = 0
);
Activity(Logger & logger, ActivityType type,
const Logger::Fields & fields = {}, ActivityId parent = getCurActivity())
: Activity(logger, lvlError, type, "", fields, parent) { };
Activity(
Logger & logger,
ActivityType type,
const Logger::Fields & fields = {},
ActivityId parent = 0
)
: Activity(logger, lvlError, type, "", fields, parent) {};
Activity(const Activity & act) = delete;
@@ -220,36 +227,6 @@ struct Activity
friend class Logger;
};
class PushActivity
{
std::optional<ActivityId> prevAct;
public:
PushActivity(ActivityId act) : prevAct(getCurActivity())
{
setCurActivity(act);
}
PushActivity(PushActivity && other)
{
std::swap(prevAct, other.prevAct);
}
PushActivity & operator=(PushActivity && other)
{
auto tmp(std::move(other));
std::swap(prevAct, tmp.prevAct);
return *this;
}
~PushActivity()
{
if (prevAct) {
setCurActivity(*prevAct);
}
}
};
extern Logger * logger;
Logger * makeSimpleLogger(bool printBuildLogs = true);