From 67fc8213df9b418eb16ab9f05a6644e83381542e Mon Sep 17 00:00:00 2001
From: Maximilian Bosch <maximilian@mbosch.me>
Date: Tue, 28 Jan 2025 17:56:10 +0100
Subject: [PATCH] Update flake; start aio event loop in queue monitor

---
 flake.lock                              |  8 ++++----
 src/hydra-queue-runner/queue-monitor.cc | 15 +++++++++------
 src/hydra-queue-runner/state.hh         |  5 +++--
 3 files changed, 16 insertions(+), 12 deletions(-)

diff --git a/flake.lock b/flake.lock
index f35fc8feb..f045bf7d4 100644
--- a/flake.lock
+++ b/flake.lock
@@ -48,11 +48,11 @@
         "pre-commit-hooks": "pre-commit-hooks"
       },
       "locked": {
-        "lastModified": 1737770806,
-        "narHash": "sha256-mdtBGOkNySRbQ1zkroCdyOHFqMOZSOJCPxjXTpCaWvc=",
+        "lastModified": 1738053421,
+        "narHash": "sha256-B9HigdAgTs/FxUVcGYIRIxJVPQ+vGp3/G6WRuVKIGds=",
         "ref": "refs/heads/main",
-        "rev": "5a41803f744822377587f784ded0e6a061b39cd4",
-        "revCount": 16688,
+        "rev": "f8a592713d14bf0374950c8d3575eb3795e63987",
+        "revCount": 16698,
         "type": "git",
         "url": "https://git.lix.systems/lix-project/lix"
       },
diff --git a/src/hydra-queue-runner/queue-monitor.cc b/src/hydra-queue-runner/queue-monitor.cc
index b66becb0f..6231449bc 100644
--- a/src/hydra-queue-runner/queue-monitor.cc
+++ b/src/hydra-queue-runner/queue-monitor.cc
@@ -1,5 +1,6 @@
 #include "hydra-build-result.hh"
 #include "lix/libstore/globals.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/thread-pool.hh"
 #include "state.hh"
 
@@ -40,13 +41,15 @@ void State::queueMonitorLoop(Connection & conn)
 
     auto destStore = getDestStore();
 
+    AsyncIoRoot aio;
+
     bool quit = false;
     while (!quit) {
         auto t_before_work = std::chrono::steady_clock::now();
 
         localStore->clearPathInfoCache();
 
-        bool done = getQueuedBuilds(conn, destStore);
+        bool done = getQueuedBuilds(aio, conn, destStore);
 
         if (buildOne && buildOneDone) quit = true;
 
@@ -93,7 +96,7 @@ struct PreviousFailure : public std::exception {
 };
 
 
-bool State::getQueuedBuilds(Connection & conn,
+bool State::getQueuedBuilds(AsyncIoRoot & aio, Connection & conn,
     ref<Store> destStore)
 {
     prom.queue_checks_started.Increment();
@@ -176,7 +179,7 @@ bool State::getQueuedBuilds(Connection & conn,
 
         /* Create steps for this derivation and its dependencies. */
         try {
-            step = createStep(destStore, conn, build, build->drvPath,
+            step = createStep(aio, destStore, conn, build, build->drvPath,
                 build, nullptr, finishedDrvs, newSteps, newRunnable);
         } catch (PreviousFailure & ex) {
 
@@ -434,7 +437,7 @@ std::map<DrvOutput, std::optional<StorePath>> State::getMissingRemotePaths(
 }
 
 
-Step::ptr State::createStep(ref<Store> destStore,
+Step::ptr State::createStep(AsyncIoRoot & aio, ref<Store> destStore,
     Connection & conn, Build::ptr build, const StorePath & drvPath,
     Build::ptr referringBuild, Step::ptr referringStep, std::set<StorePath> & finishedDrvs,
     std::set<Step::ptr> & newSteps, std::set<Step::ptr> & newRunnable)
@@ -571,7 +574,7 @@ Step::ptr State::createStep(ref<Store> destStore,
                         printInfo("substituting output ‘%1%’ of ‘%2%’",
                             localStore->printStorePath(path),
                             localStore->printStorePath(drvPath));
-                        localStore->ensurePath(path);
+                        aio.blockOn(localStore->ensurePath(path));
                         // FIXME: should copy directly from substituter to destStore.
                     }
 
@@ -611,7 +614,7 @@ Step::ptr State::createStep(ref<Store> destStore,
 
     /* Create steps for the dependencies. */
     for (auto & i : step->drv->inputDrvs.map) {
-        auto dep = createStep(destStore, conn, build, i.first, nullptr, step, finishedDrvs, newSteps, newRunnable);
+        auto dep = createStep(aio, destStore, conn, build, i.first, nullptr, step, finishedDrvs, newSteps, newRunnable);
         if (dep) {
             auto step_(step->state.lock());
             step_->deps.insert(dep);
diff --git a/src/hydra-queue-runner/state.hh b/src/hydra-queue-runner/state.hh
index 88fe262f4..9e5f21b6f 100644
--- a/src/hydra-queue-runner/state.hh
+++ b/src/hydra-queue-runner/state.hh
@@ -22,6 +22,7 @@
 #include "lix/libstore/pathlocks.hh"
 #include "lix/libstore/serve-protocol.hh"
 #include "lix/libstore/store-api.hh"
+#include "lix/libutil/async.hh"
 #include "lix/libutil/pool.hh"
 #include "lix/libutil/sync.hh"
 #include "nar-extractor.hh"
@@ -541,7 +542,7 @@ private:
     void queueMonitorLoop(Connection & conn);
 
     /* Check the queue for new builds. */
-    bool getQueuedBuilds(Connection & conn, nix::ref<nix::Store> destStore);
+    bool getQueuedBuilds(nix::AsyncIoRoot & aio, Connection & conn, nix::ref<nix::Store> destStore);
 
     /* Handle cancellation, deletion and priority bumps. */
     void processQueueChange(Connection & conn);
@@ -555,7 +556,7 @@ private:
         nix::ref<nix::Store> destStore,
         const std::map<nix::DrvOutput, std::optional<nix::StorePath>> & paths);
 
-    Step::ptr createStep(nix::ref<nix::Store> store,
+    Step::ptr createStep(nix::AsyncIoRoot & aio, nix::ref<nix::Store> store,
         Connection & conn, Build::ptr build, const nix::StorePath & drvPath,
         Build::ptr referringBuild, Step::ptr referringStep, std::set<nix::StorePath> & finishedDrvs,
         std::set<Step::ptr> & newSteps, std::set<Step::ptr> & newRunnable);