libstore: asyncify Store::addSignatures

Change-Id: If5fac14c9b5c8e7c9605080606f147979cc3ca31

lix/libstore/binary-cache-store.cc

@@ -521,8 +521,9 @@ ref<FSAccessor> BinaryCacheStore::getFSAccessor()
     return make_ref<RemoteFSAccessor>(ref<Store>(shared_from_this()), config().localNarCache);
 }
 
-void BinaryCacheStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
-{
+kj::Promise<Result<void>>
+BinaryCacheStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
+try {
     /* Note: this is inherently racy since there is no locking on
        binary caches. In particular, with S3 this unreliable, even
        when addSignatures() is called sequentially on a path, because
@@ -534,6 +535,9 @@ void BinaryCacheStore::addSignatures(const StorePath & storePath, const StringSe
     narInfo->sigs.insert(sigs.begin(), sigs.end());
 
     writeNarInfo(narInfo);
+    co_return result::success();
+} catch (...) {
+    co_return result::current_exception();
 }
 
 kj::Promise<Result<std::optional<std::string>>>

lix/libstore/binary-cache-store.hh

@@ -153,7 +153,8 @@ public:
 
     ref<FSAccessor> getFSAccessor() override;
 
-    void addSignatures(const StorePath & storePath, const StringSet & sigs) override;
+    kj::Promise<Result<void>>
+    addSignatures(const StorePath & storePath, const StringSet & sigs) override;
 
     kj::Promise<Result<std::optional<std::string>>> getBuildLogExact(const StorePath & path) override;
 

lix/libstore/build/local-derivation-goal.cc

@@ -1242,8 +1242,9 @@ struct RestrictedStore : public virtual IndirectRootStore, public virtual GcStor
         return {result::success()};
     }
 
-    void addSignatures(const StorePath & storePath, const StringSet & sigs) override
-    { unsupported("addSignatures"); }
+    kj::Promise<Result<void>>
+    addSignatures(const StorePath & storePath, const StringSet & sigs) override
+    try { unsupported("addSignatures"); } catch (...) { return {result::current_exception()}; }
 
     kj::Promise<Result<void>> queryMissing(const std::vector<DerivedPath> & targets,
         StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown,

lix/libstore/daemon.cc

@@ -875,7 +875,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
         auto path = store->parseStorePath(readString(from));
         StringSet sigs = readStrings<StringSet>(from);
         logger->startWork();
-        store->addSignatures(path, sigs);
+        aio.blockOn(store->addSignatures(path, sigs));
         logger->stopWork();
         to << 1;
         break;

lix/libstore/local-store.cc

@@ -1810,21 +1810,31 @@ kj::Promise<Result<std::optional<TrustedFlag>>> LocalStore::isTrustedClient()
 }
 
 
-void LocalStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
-{
-    retrySQLite([&]() {
-        auto state = dbPool.get();
+kj::Promise<Result<void>>
+LocalStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
+try {
+    // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines)
+    TRY_AWAIT(retrySQLite([&]() -> kj::Promise<Result<void>> {
+        try {
+            auto state = dbPool.get();
 
-        SQLiteTxn txn = state->db.beginTransaction(SQLiteTxnType::Immediate);
+            SQLiteTxn txn = state->db.beginTransaction(SQLiteTxnType::Immediate);
 
-        auto info = std::const_pointer_cast<ValidPathInfo>(queryPathInfoInternal(*state, storePath));
+            auto info = std::const_pointer_cast<ValidPathInfo>(queryPathInfoInternal(*state, storePath));
 
-        info->sigs.insert(sigs.begin(), sigs.end());
+            info->sigs.insert(sigs.begin(), sigs.end());
 
-        updatePathInfo(*state, *info);
+            updatePathInfo(*state, *info);
 
-        txn.commit();
-    }, always_progresses);
+            txn.commit();
+            co_return result::success();
+        } catch (...) {
+            co_return result::current_exception();
+        }
+    }));
+    co_return result::success();
+} catch (...) {
+    co_return result::current_exception();
 }
 
 

lix/libstore/local-store.hh

@@ -307,7 +307,8 @@ public:
 
     kj::Promise<Result<std::optional<TrustedFlag>>> isTrustedClient() override;
 
-    void addSignatures(const StorePath & storePath, const StringSet & sigs) override;
+    kj::Promise<Result<void>>
+    addSignatures(const StorePath & storePath, const StringSet & sigs) override;
 
     /**
      * If free disk space in /nix/store if below minFree, delete

lix/libstore/remote-store.cc

@@ -896,12 +896,16 @@ try {
 }
 
 
-void RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
-{
+kj::Promise<Result<void>>
+RemoteStore::addSignatures(const StorePath & storePath, const StringSet & sigs)
+try {
     auto conn(getConnection());
     conn->to << WorkerProto::Op::AddSignatures << printStorePath(storePath) << sigs;
     conn.processStderr();
     readInt(conn->from);
+    co_return result::success();
+} catch (...) {
+    co_return result::current_exception();
 }
 
 

lix/libstore/remote-store.hh

@@ -156,7 +156,8 @@ public:
     kj::Promise<Result<void>> repairPath(const StorePath & path) override
     try { unsupported("repairPath"); } catch (...) { return {result::current_exception()}; }
 
-    void addSignatures(const StorePath & storePath, const StringSet & sigs) override;
+    kj::Promise<Result<void>>
+    addSignatures(const StorePath & storePath, const StringSet & sigs) override;
 
     kj::Promise<Result<void>> queryMissing(const std::vector<DerivedPath> & targets,
         StorePathSet & willBuild, StorePathSet & willSubstitute, StorePathSet & unknown,

lix/libstore/store-api.hh

@@ -750,8 +750,9 @@ public:
      * Add signatures to the specified store path. The signatures are
      * not verified.
      */
-    virtual void addSignatures(const StorePath & storePath, const StringSet & sigs)
-    { unsupported("addSignatures"); }
+    virtual kj::Promise<Result<void>>
+    addSignatures(const StorePath & storePath, const StringSet & sigs)
+    try { unsupported("addSignatures"); } catch (...) { return {result::current_exception()}; }
 
     /* Utility functions. */
 

lix/nix/sigs.cc

@@ -1,10 +1,12 @@
 #include "lix/libcmd/command.hh"
 #include "lix/libmain/shared.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/thread-pool.hh"
 #include "lix/libutil/signals.hh"
 
 #include <atomic>
+#include <functional>
 
 using namespace nix;
 
@@ -42,7 +44,7 @@ struct CmdCopySigs : StorePathsCommand
 
         std::atomic<size_t> added{0};
 
-        auto doPath = [&](const Path & storePathS) {
+        auto doPath = [&](AsyncIoRoot & aio, const Path & storePathS) {
 
             checkInterrupt();
 
@@ -71,13 +73,16 @@ struct CmdCopySigs : StorePathsCommand
             }
 
             if (!newSigs.empty()) {
-                store->addSignatures(storePath, newSigs);
+                aio.blockOn(store->addSignatures(storePath, newSigs));
                 added += newSigs.size();
             }
         };
 
-        for (auto & storePath : storePaths)
-            pool.enqueue(std::bind(doPath, store->printStorePath(storePath)));
+        for (auto & storePath : storePaths) {
+            pool.enqueueWithAio(
+                std::bind(doPath, std::placeholders::_1, store->printStorePath(storePath))
+            );
+        }
 
         pool.process();
 
@@ -119,7 +124,7 @@ struct CmdSign : StorePathsCommand
 
         std::atomic<size_t> added{0};
 
-        auto doPath = [&](const Path & storePathS) {
+        auto doPath = [&](AsyncIoRoot & aio, const Path & storePathS) {
 
             checkInterrupt();
 
@@ -133,13 +138,16 @@ struct CmdSign : StorePathsCommand
             assert(!info2.sigs.empty());
 
             if (!info->sigs.count(*info2.sigs.begin())) {
-                store->addSignatures(storePath, info2.sigs);
+                aio.blockOn(store->addSignatures(storePath, info2.sigs));
                 added++;
             }
         };
 
-        for (auto & storePath : storePaths)
-            pool.enqueue(std::bind(doPath, store->printStorePath(storePath)));
+        for (auto & storePath : storePaths) {
+            pool.enqueueWithAio(
+                std::bind(doPath, std::placeholders::_1, store->printStorePath(storePath))
+            );
+        }
 
         pool.process();