Merge changes Ia8ac6af0,I9f478750,I7d594835 into main

* changes:
  deps: backport capnproto promise nodiscard PR
  libstore: await promises instead of discarding
  libutil: fix some thread pool bugs

lix/libstore/local-store.cc

@@ -1671,7 +1671,7 @@ try {
         StorePathSet referrers; queryReferrers(path, referrers);
         for (auto & i : referrers)
             if (i != path) {
-                verifyPath(i, storePathsInStoreDir, done, validPaths, repair, errors);
+                TRY_AWAIT(verifyPath(i, storePathsInStoreDir, done, validPaths, repair, errors));
                 if (validPaths.count(i))
                     canInvalidate = false;
             }

lix/libutil/thread-pool.cc

@@ -47,23 +47,25 @@ void ThreadPool::enqueueWithAio(const work_t & t)
     if (quit)
         throw ThreadPoolShutDown("cannot enqueue a work item while the thread pool is shutting down");
     state->pending.push(t);
-    if (state->pending.size() > state->workers.size() && state->workers.size() < maxThreads)
+    if (state->active == state->workers.size() && state->workers.size() < maxThreads)
         state->workers.emplace_back(&ThreadPool::doWork, this);
     work.notify_one();
 }
 
 void ThreadPool::process()
 {
-    state_.lock()->draining = true;
+    const auto shouldWait = [&] {
+        auto state(state_.lock());
+        state->draining = true;
+        return state->active > 0 || !state->pending.empty();
+    }();
 
     /* Wait until no more work is pending or active. */
     try {
-        if (auto state(state_.lock()); state->active == 0 && state->pending.empty()) {
-            return;
+        if (shouldWait) {
+            quit.wait(false);
         }
 
-        quit.wait(false);
-
         auto state(state_.lock());
         if (state->exception)
             std::rethrow_exception(state->exception);
@@ -141,6 +143,8 @@ void ThreadPool::doWork()
             /* Wait until a work item is available or we're asked to
                quit. */
             while (true) {
+                if (quit) return;
+
                 if (!state->pending.empty()) break;
 
                 /* If there are no active or pending items, and the

misc/capnproto-promise-nodiscard.patch

@@ -0,0 +1,206 @@
+diff --git a/c++/WORKSPACE b/c++/WORKSPACE
+index d94a279e..4871ead7 100644
+--- a/c++/WORKSPACE
++++ b/c++/WORKSPACE
+@@ -31,6 +31,7 @@ cc_library(
+     name = "zlib",
+     srcs = glob(["*.c"]),
+     hdrs = glob(["*.h"]),
++    includes = ["."],
+     # Temporary workaround for zlib warnings and mac compilation, should no longer be needed with next release https://github.com/madler/zlib/issues/633
+     copts = [
+         "-w",
+diff --git a/c++/src/kj/array.h b/c++/src/kj/array.h
+index 3932f9f4..677c691a 100644
+--- a/c++/src/kj/array.h
++++ b/c++/src/kj/array.h
+@@ -780,7 +780,7 @@ struct CopyConstructArray_<T, Iterator, true, false> {
+
+   static T* apply(T* __restrict__ pos, Iterator start, Iterator end) {
+     // Verify that T can be *implicitly* constructed from the source values.
+-    if (false) implicitCast<T>(kj::mv(*start));
++    if (false) (void)implicitCast<T>(kj::mv(*start));
+
+     if (noexcept(T(kj::mv(*start)))) {
+       while (start != end) {
+diff --git a/c++/src/kj/async-coroutine-test.c++ b/c++/src/kj/async-coroutine-test.c++
+index de767eca..d6ed1359 100644
+--- a/c++/src/kj/async-coroutine-test.c++
++++ b/c++/src/kj/async-coroutine-test.c++
+@@ -288,5 +288,5 @@ KJ_TEST("Exceptions during suspended coroutine frame-unwind propagate via destru
+   WaitScope waitScope(loop);
+
+   auto exception = KJ_ASSERT_NONNULL(kj::runCatchingExceptions([&]() {
+-    deferredThrowCoroutine(kj::NEVER_DONE);
++    (void)deferredThrowCoroutine(kj::NEVER_DONE);
+   }));
+
+   KJ_EXPECT(exception.getDescription() == "thrown during unwind");
+diff --git a/c++/src/kj/async-io-test.c++ b/c++/src/kj/async-io-test.c++
+index e8892b79..dffcbb26 100644
+--- a/c++/src/kj/async-io-test.c++
++++ b/c++/src/kj/async-io-test.c++
+@@ -1577,7 +1577,7 @@ KJ_TEST("Userland pipe pump into zero-limited pipe, no data to pump") {
+   auto pipe2 = newOneWayPipe(uint64_t(0));
+   auto pumpPromise = KJ_ASSERT_NONNULL(pipe2.out->tryPumpFrom(*pipe.in));
+
+-  expectRead(*pipe2.in, "");
++  expectRead(*pipe2.in, "").wait(ws);
+   pipe.out = nullptr;
+   KJ_EXPECT(pumpPromise.wait(ws) == 0);
+ }
+@@ -1590,7 +1590,7 @@ KJ_TEST("Userland pipe pump into zero-limited pipe, data is pumped") {
+   auto pipe2 = newOneWayPipe(uint64_t(0));
+   auto pumpPromise = KJ_ASSERT_NONNULL(pipe2.out->tryPumpFrom(*pipe.in));
+
+-  expectRead(*pipe2.in, "");
++  expectRead(*pipe2.in, "").wait(ws);
+   auto writePromise = pipe.out->write("foo", 3);
+   KJ_EXPECT_THROW_RECOVERABLE_MESSAGE("abortRead() has been called", pumpPromise.wait(ws));
+ }
+diff --git a/c++/src/kj/async.h b/c++/src/kj/async.h
+index 564b5171..d4f2d55c 100644
+--- a/c++/src/kj/async.h
++++ b/c++/src/kj/async.h
+@@ -118,7 +118,7 @@ private:
+ // Promises
+
+ template <typename T>
+-class Promise: protected _::PromiseBase {
++class [[nodiscard]] Promise: protected _::PromiseBase {
+   // The basic primitive of asynchronous computation in KJ.  Similar to "futures", but designed
+   // specifically for event loop concurrency.  Similar to E promises and JavaScript Promises/A.
+   //
+diff --git a/c++/src/kj/common-test.c++ b/c++/src/kj/common-test.c++
+index 97856125..913a6be4 100644
+--- a/c++/src/kj/common-test.c++
++++ b/c++/src/kj/common-test.c++
+@@ -573,7 +573,7 @@ TEST(Common, Downcast) {
+
+   EXPECT_EQ(&bar, &downcast<Bar>(foo));
+ #if defined(KJ_DEBUG) && !KJ_NO_RTTI
+-  KJ_EXPECT_THROW_MESSAGE("Value cannot be downcast", downcast<Baz>(foo));
++  KJ_EXPECT_THROW_MESSAGE("Value cannot be downcast", (void)downcast<Baz>(foo));
+ #endif
+
+ #if KJ_NO_RTTI
+diff --git a/c++/src/kj/compat/http-test.c++ b/c++/src/kj/compat/http-test.c++
+index f10ff8d1..9003099d 100644
+--- a/c++/src/kj/compat/http-test.c++
++++ b/c++/src/kj/compat/http-test.c++
+@@ -6553,7 +6553,7 @@ KJ_TEST("Simple CONNECT Server works") {
+              "\r\n"
+              "hello"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+
+@@ -6628,7 +6628,7 @@ KJ_TEST("CONNECT Server (201 status)") {
+              "\r\n"
+              "hello"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+
+@@ -6706,7 +6706,7 @@ KJ_TEST("CONNECT Server rejected") {
+              "\r\n"
+              "boom"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+
+@@ -6774,7 +6774,7 @@ KJ_TEST("CONNECT Server cancels read") {
+              "HTTP/1.1 200 OK\r\n"
+              "\r\n"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+ }
+@@ -6840,7 +6840,7 @@ KJ_TEST("CONNECT Server cancels write") {
+              "HTTP/1.1 200 OK\r\n"
+              "\r\n"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+ }
+@@ -6913,7 +6913,7 @@ KJ_TEST("CONNECT rejects Transfer-Encoding") {
+              "\r\n"
+              "ERROR: Bad Request"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+ }
+@@ -6947,7 +6947,7 @@ KJ_TEST("CONNECT rejects Content-Length") {
+              "\r\n"
+              "ERROR: Bad Request"_kj).wait(waitScope);
+
+-  expectEnd(*pipe.ends[1]);
++  expectEnd(*pipe.ends[1]).wait(waitScope);
+
+   listenTask.wait(waitScope);
+ }
+diff --git a/c++/src/kj/compat/tls-test.c++ b/c++/src/kj/compat/tls-test.c++
+index dddefa57..52ccc68a 100644
+--- a/c++/src/kj/compat/tls-test.c++
++++ b/c++/src/kj/compat/tls-test.c++
+@@ -1037,15 +1037,15 @@ KJ_TEST("TLS receiver experiences pre-TLS error") {
+   TlsReceiverTest test;
+
+   KJ_LOG(INFO, "Accepting before a bad connect");
+-  auto promise = test.receiver->accept();
++  auto acceptPromise = test.receiver->accept();
+
+   KJ_LOG(INFO, "Disappointing our server");
+-  test.baseReceiver->badConnect();
++  auto connectPromise = test.baseReceiver->badConnect();
+
+   // Can't use KJ_EXPECT_THROW_RECOVERABLE_MESSAGE because wait() that returns a value can't throw
+   // recoverable exceptions. Can't use KJ_EXPECT_THROW_MESSAGE because non-recoverable exceptions
+   // will fork() in -fno-exception which screws up our state.
+-  promise.then([](auto) {
++  acceptPromise.then([](auto) {
+     KJ_FAIL_EXPECT("expected exception");
+   }, [](kj::Exception&& e) {
+     KJ_EXPECT(e.getDescription() == "Pipes are leaky");
+diff --git a/c++/src/kj/test.h b/c++/src/kj/test.h
+index 5acbb00d..de5efec2 100644
+--- a/c++/src/kj/test.h
++++ b/c++/src/kj/test.h
+@@ -92,6 +92,7 @@ private:
+   else KJ_FAIL_EXPECT("failed: expected " #cond, _kjCondition, ##__VA_ARGS__)
+ #endif
+
++// TODO(msvc): cast results to void like non-MSVC versions do
+ #if _MSC_VER && !defined(__clang__)
+ #define KJ_EXPECT_THROW_RECOVERABLE(type, code, ...) \
+   do { \
+@@ -115,7 +116,7 @@ private:
+ #else
+ #define KJ_EXPECT_THROW_RECOVERABLE(type, code, ...) \
+   do { \
+-    KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { code; })) { \
++    KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { (void)({code}); })) { \
+       KJ_EXPECT(e->getType() == ::kj::Exception::Type::type, \
+           "code threw wrong exception type: " #code, *e, ##__VA_ARGS__); \
+     } else { \
+@@ -125,7 +126,7 @@ private:
+
+ #define KJ_EXPECT_THROW_RECOVERABLE_MESSAGE(message, code, ...) \
+   do { \
+-    KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { code; })) { \
++    KJ_IF_MAYBE(e, ::kj::runCatchingExceptions([&]() { (void)({code}); })) { \
+       KJ_EXPECT(::kj::_::hasSubstring(e->getDescription(), message), \
+           "exception description didn't contain expected substring", *e, ##__VA_ARGS__); \
+     } else { \

misc/capnproto.nix

@@ -35,6 +35,11 @@ stdenv.mkDerivation rec {
     sha256 = "sha256-LVdkqVBTeh8JZ1McdVNtRcnFVwEJRNjt0JV2l7RkuO8=";
   };
 
+  patches = [
+    # backport of https://github.com/capnproto/capnproto/pull/1810
+    ./capnproto-promise-nodiscard.patch
+  ];
+
   nativeBuildInputs = [ cmake ];
   propagatedBuildInputs = [
     openssl

tests/unit/libutil/thread-pool.cc

@@ -0,0 +1,111 @@
+#include "lix/libutil/thread-pool.hh"
+#include <atomic>
+#include <exception>
+#include <gtest/gtest.h>
+#include <memory>
+
+static auto onThreadExit(auto fn)
+{
+    auto deferred = kj::defer(std::move(fn));
+    return std::make_shared<decltype(deferred)>(std::move(deferred));
+}
+
+namespace nix {
+
+TEST(ThreadPool, creates_threads)
+{
+    ThreadPool t{"test", 2};
+
+    std::atomic_bool unblockA{false}, unblockB{false};
+    std::atomic_bool started{false};
+
+    t.enqueue([&] {
+        started = true;
+        started.notify_all();
+        unblockA.wait(false);
+        unblockB = true;
+        unblockB.notify_all();
+    });
+    started.wait(false);
+
+    // now no work is pending. the next enqueue should start a
+    // new thread; if it does not we'll deadlock and time out.
+
+    started = false;
+    t.enqueue([&] {
+        started = true;
+        started.notify_all();
+        unblockB.wait(false);
+    });
+    started.wait(false);
+
+    unblockA = true;
+    unblockA.notify_all();
+
+    t.process();
+}
+
+TEST(ThreadPool, early_quit)
+{
+    ThreadPool t{"test", 2};
+    bool ran_anyway = false;
+
+    struct Dead : std::exception {};
+
+    std::atomic_bool unblockA{false}, unblockB{false};
+    std::atomic_bool started{false};
+
+    t.enqueue([&] {
+        started = true;
+        started.notify_all();
+        unblockA.wait(false);
+        thread_local auto _ = onThreadExit([&] {
+            unblockB = true;
+            unblockB.notify_all();
+        });
+        throw Dead{};
+    });
+    started.wait(false);
+
+    started = false;
+    t.enqueue([&] {
+        started = true;
+        started.notify_all();
+        unblockB.wait(false);
+    });
+    started.wait(false);
+
+    // this one should never run. the first thread saw an exception,
+    // and the second thread should have exited early because of it.
+    t.enqueue([&] { ran_anyway = true; });
+
+    unblockA = true;
+    unblockA.notify_all();
+
+    ASSERT_THROW(t.process(), Dead);
+    ASSERT_FALSE(ran_anyway);
+}
+
+TEST(ThreadPool, always_rethrows)
+{
+    ThreadPool t{"test"};
+
+    struct Dead : std::exception {};
+
+    std::atomic_bool flag{false};
+
+    t.enqueue([&] {
+        thread_local auto _ = onThreadExit([&] {
+            flag = true;
+            flag.notify_all();
+        });
+
+        throw Dead{};
+    });
+
+    flag.wait(false);
+
+    ASSERT_THROW(t.process(), Dead);
+}
+
+}

tests/unit/meson.build

@@ -64,6 +64,7 @@ libutil_tests_sources = files(
   'libutil/serialise.cc',
   'libutil/suggestions.cc',
   'libutil/tests.cc',
+  'libutil/thread-pool.cc',
   'libutil/url.cc',
   'libutil/url-name.cc',
   'libutil/xml-writer.cc',