libstore: asyncify Store::queryValidPaths

Change-Id: I3a3303b288321cd812f4df87c80a676f997e57b9

lix/legacy/nix-env.cc

@@ -1102,7 +1102,7 @@ static void opQuery(Globals & globals, Strings opFlags, Strings opArgs)
                 printMsg(lvlTalkative, "skipping derivation named '%s' which gives an assertion failure", i.queryName(*state));
                 i.setFailed();
             }
-        validPaths = store.queryValidPaths(paths);
+        validPaths = globals.aio.blockOn(store.queryValidPaths(paths));
         substitutablePaths = globals.aio.blockOn(store.querySubstitutablePaths(paths));
     }
 

lix/legacy/nix-store.cc

@@ -897,7 +897,7 @@ static void opServe(AsyncIoRoot & aio, Strings opFlags, Strings opArgs)
                     aio.blockOn(store->substitutePaths(paths));
                 }
 
-                auto valid = store->queryValidPaths(paths);
+                auto valid = aio.blockOn(store->queryValidPaths(paths));
                 out << ServeProto::write(*store, wconn, valid);
                 break;
             }

lix/libstore/daemon.cc

@@ -292,7 +292,7 @@ static void performOp(AsyncIoRoot & aio, TunnelLogger * logger, ref<Store> store
         if (substitute) {
             aio.blockOn(store->substitutePaths(paths));
         }
-        auto res = store->queryValidPaths(paths, substitute);
+        auto res = aio.blockOn(store->queryValidPaths(paths, substitute));
         logger->stopWork();
         to << WorkerProto::write(*store, wconn, res);
         break;

lix/libstore/legacy-ssh-store.cc

@@ -410,9 +410,9 @@ public:
         co_return result::current_exception();
     }
 
-    StorePathSet queryValidPaths(const StorePathSet & paths,
+    kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
         SubstituteFlag maybeSubstitute = NoSubstitute) override
-    {
+    try {
         auto conn(connections->get());
 
         conn->to
@@ -422,7 +422,9 @@ public:
         conn->to << ServeProto::write(*this, *conn, paths);
         conn->to.flush();
 
-        return ServeProto::Serialise<StorePathSet>::read(*this, *conn);
+        co_return ServeProto::Serialise<StorePathSet>::read(*this, *conn);
+    } catch (...) {
+        co_return result::current_exception();
     }
 
     void connect() override

lix/libstore/local-store.cc

@@ -977,12 +977,15 @@ bool LocalStore::isValidPathUncached(const StorePath & path)
 }
 
 
-StorePathSet LocalStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
-{
+kj::Promise<Result<StorePathSet>>
+LocalStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
+try {
     StorePathSet res;
     for (auto & i : paths)
         if (isValidPath(i)) res.insert(i);
-    return res;
+    co_return res;
+} catch (...) {
+    co_return result::current_exception();
 }
 
 
@@ -1085,7 +1088,7 @@ try {
         if (sub->config().storeDir != config_.storeDir) continue;
         if (!sub->config().wantMassQuery) continue;
 
-        auto valid = sub->queryValidPaths(remaining);
+        auto valid = TRY_AWAIT(sub->queryValidPaths(remaining));
 
         StorePathSet remaining2;
         for (auto & path : remaining)

lix/libstore/local-store.hh

@@ -187,7 +187,7 @@ public:
 
     bool isValidPathUncached(const StorePath & path) override;
 
-    StorePathSet queryValidPaths(const StorePathSet & paths,
+    kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
         SubstituteFlag maybeSubstitute = NoSubstitute) override;
 
     StorePathSet queryAllValidPaths() override;

lix/libstore/remote-store.cc

@@ -207,8 +207,9 @@ bool RemoteStore::isValidPathUncached(const StorePath & path)
 }
 
 
-StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
-{
+kj ::Promise<Result<StorePathSet>>
+RemoteStore::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
+try {
     auto conn(getConnection());
     conn->to << WorkerProto::Op::QueryValidPaths;
     conn->to << WorkerProto::write(*this, *conn, paths);
@@ -216,7 +217,9 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute
         conn->to << maybeSubstitute;
     }
     conn.processStderr();
-    return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
+    co_return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
+} catch (...) {
+    co_return result::current_exception();
 }
 
 

lix/libstore/remote-store.hh

@@ -51,7 +51,7 @@ public:
 
     bool isValidPathUncached(const StorePath & path) override;
 
-    StorePathSet queryValidPaths(const StorePathSet & paths,
+    kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
         SubstituteFlag maybeSubstitute = NoSubstitute) override;
 
     StorePathSet queryAllValidPaths() override;

lix/libstore/store-api.cc

@@ -844,8 +844,9 @@ try {
 }
 
 
-StorePathSet Store::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
-{
+kj::Promise<Result<StorePathSet>>
+Store::queryValidPaths(const StorePathSet & paths, SubstituteFlag maybeSubstitute)
+try {
     struct State
     {
         size_t left;
@@ -890,16 +891,18 @@ StorePathSet Store::queryValidPaths(const StorePathSet & paths, SubstituteFlag m
     for (auto & path : paths)
         pool.enqueue(std::bind(doQuery, path));
 
-    pool.process();
+    TRY_AWAIT(pool.processAsync());
 
     while (true) {
         auto state(state_.lock());
         if (!state->left) {
             if (state->exc) std::rethrow_exception(state->exc);
-            return std::move(state->valid);
+            co_return std::move(state->valid);
         }
         state.wait(wakeup);
     }
+} catch (...) {
+    co_return result::current_exception();
 }
 
 
@@ -1220,7 +1223,7 @@ kj::Promise<Result<std::map<StorePath, StorePath>>> copyPaths(
     CheckSigsFlag checkSigs,
     SubstituteFlag substitute)
 try {
-    auto valid = dstStore.queryValidPaths(storePaths, substitute);
+    auto valid = TRY_AWAIT(dstStore.queryValidPaths(storePaths, substitute));
 
     StorePathSet missing;
     for (auto & path : storePaths)

lix/libstore/store-api.hh

@@ -361,7 +361,7 @@ public:
      * Query which of the given paths is valid. Optionally, try to
      * substitute missing paths.
      */
-    virtual StorePathSet queryValidPaths(const StorePathSet & paths,
+    virtual kj::Promise<Result<StorePathSet>> queryValidPaths(const StorePathSet & paths,
         SubstituteFlag maybeSubstitute = NoSubstitute);
 
     /**