fix: Terminate daemons properly on Ctrl-C on macOS

This was an absolute nightmare to diagnose. It turns out there's a
kernel bug: poll with events = POLLHUP will receive an event for NOT
POLLHUP internally in the kernel, delete their event subscription, and
then not receive events for any HUP later. lol! lmao!!

We choose to use plain old EVFILT_READ because the watched fd can be
either a socket or a pipe and it's preferable to eat some spurious
wakeups than have separate paths for those. The alternative is using
EVFILT_SOCK, a private API that's existed for years and which netty
uses for its sockets, but that doesn't work on pipes.

Fixes: https://git.lix.systems/lix-project/lix/issues/729
Change-Id: If72b5d7a39f00320a9acccdbe81121cdb1a04c45

doc/manual/rl-next/ctrl-c-works.md

@@ -0,0 +1,12 @@
+---
+synopsis: "Ctrl-C works correctly on macOS again"
+cls: [3066]
+issues: [fj#729]
+category: Fixes
+credits: [jade]
+---
+
+Due to a kernel bug in macOS's `poll(2)` implementation where it would forget about event subscriptions, our detection of closed connections in the Lix daemon didn't work and left around lingering daemon processes.
+We have rewritten that thread to use `kqueue(2)`, which is what the `poll(2)` implementation uses internally in the macOS kernel, so now Ctrl-C on clients will reliably terminate daemons once more.
+
+This FD close monitoring has had the highest Apple bug ID references per line of code anywhere in the project, and hopefully not using poll anymore will stop us hitting bugs in poll.

lix/libutil/monitor-fd.cc

@@ -1,10 +1,77 @@
 #include "monitor-fd.hh"
+#include "error.hh"
+
+#ifdef __APPLE__
+#include <sys/types.h>
+#include <sys/event.h>
+#endif
 
 namespace nix {
 
+#ifdef __APPLE__
+/**
+ * This custom kqueue usage exists because Apple's poll implementation is
+ * broken and loses event subscriptions if EVFILT_READ fires without matching
+ * the requested `events` in the pollfd.
+ *
+ * We use EVFILT_READ, which causes some spurious wakeups (at most one per write
+ * from the client, in addition to the socket lifecycle events), because the
+ * alternate API, EVFILT_SOCK, doesn't work on pipes, which this is also used
+ * to monitor in certain situations.
+ *
+ * See (EVFILT_SOCK):
+ * https://github.com/netty/netty/blob/64bd2f4eb62c2fb906bc443a2aabf894c8b7dce9/transport-classes-kqueue/src/main/java/io/netty/channel/kqueue/AbstractKQueueChannel.java#L434
+ *
+ * See: https://git.lix.systems/lix-project/lix/issues/729
+ * Apple bug in poll(2): FB17447257, available at https://openradar.appspot.com/FB17447257
+ */
+void MonitorFdHup::runThread(int watchFd, int terminateFd)
+{
+    int kqResult = kqueue();
+    if (kqResult < 0) {
+        throw SysError("MonitorFdHup kqueue");
+    }
+    AutoCloseFD kq{kqResult};
+
+    std::array<struct kevent, 2> kevs;
+
+    // kj uses EVFILT_WRITE for this, but it seems that it causes more spurious
+    // wakeups in our case of doing blocking IO from another thread compared to
+    // EVFILT_READ.
+    //
+    // EVFILT_WRITE and EVFILT_READ (for sockets at least, where I am familiar
+    // with the internals) both go through a common filter which catches EOFs
+    // and generates spurious wakeups for either readable/writable events.
+    EV_SET(&kevs[0], watchFd, EVFILT_READ, EV_ADD | EV_ENABLE | EV_CLEAR, 0, 0, nullptr);
+    EV_SET(&kevs[1], terminateFd, EVFILT_READ, EV_ADD | EV_ENABLE | EV_CLEAR, 0, 0, nullptr);
+
+    int result = kevent(kq.get(), kevs.data(), kevs.size(), nullptr, 0, nullptr);
+    if (result < 0) {
+        throw SysError("MonitorFdHup kevent add");
+    }
+
+    while (!quit) {
+        std::array<struct kevent, 2> newEvents;
+        int numEvents = kevent(kq.get(), nullptr, 0, newEvents.data(), newEvents.size(), nullptr);
+        if (numEvents < 0) {
+            throw SysError("MonitorFdHup kevent watch");
+        }
+
+        assert(size_t(numEvents) <= newEvents.size());
+        for (int i = 0; i < numEvents; ++i) {
+            auto & event = newEvents[i];
+
+            if (event.ident == uintptr_t(watchFd)) {
+                if ((event.flags & EV_EOF) != 0) {
+                    callback();
+                }
+            }
+        }
+    }
+}
+#else
 void MonitorFdHup::runThread(int watchFd, int terminateFd)
 {
-    setCurrentThreadName("MonitorFdHup");
     while (!quit) {
         /* Wait indefinitely until a POLLHUP occurs. */
         struct pollfd fds[2];
@@ -65,12 +132,16 @@ void MonitorFdHup::runThread(int watchFd, int terminateFd)
         usleep(1'000);
     }
 }
+#endif
 
 MonitorFdHup::MonitorFdHup(int fd, std::function<void()> callback) : callback(callback)
 {
     terminatePipe.create();
     int terminateFd = terminatePipe.readSide.get();
-    thread = std::thread([this, fd, terminateFd]() { this->runThread(fd, terminateFd); });
+    thread = std::thread([this, fd, terminateFd]() {
+        setCurrentThreadName("MonitorFdHup");
+        this->runThread(fd, terminateFd);
+    });
 };
 
 MonitorFdHup::~MonitorFdHup()

tests/unit/libutil/monitor-fd.cc

@@ -14,20 +14,37 @@ TEST(MonitorFdHup, works)
 {
     int socks[2];
     int rv = socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
-    if (rv) throw SysError("socketpair");
+    if (rv) {
+        throw SysError("socketpair");
+    }
     auto sock1 = AutoCloseFD{socks[0]};
     auto sock2 = AutoCloseFD{socks[1]};
 
     std::promise<void> called;
 
-    MonitorFdHup monitor(sock1.get(), [&called]() {
-        called.set_value();
-    });
+    MonitorFdHup monitor(sock1.get(), [&called]() { called.set_value(); });
 
     sock2.close();
 
-    // 30 seconds should certainly do it.
-    called.get_future().wait_for(10s);
+    // 10 seconds should certainly do it.
+    auto status = called.get_future().wait_for(10s);
+    ASSERT_EQ(status, std::future_status::ready);
+}
+
+// Ensures that it also works with pipes.
+TEST(MonitorFdHup, works_with_pipes)
+{
+    Pipe pipes{};
+    pipes.create();
+    std::promise<void> called;
+
+    MonitorFdHup monitor(pipes.readSide.get(), [&called]() { called.set_value(); });
+
+    pipes.writeSide.close();
+
+    // 10 seconds should certainly do it.
+    auto status = called.get_future().wait_for(10s);
+    ASSERT_EQ(status, std::future_status::ready);
 }
 
 // Ensures that destroying the MonitorFdHup causes it to actually go away.
@@ -35,14 +52,14 @@ TEST(MonitorFdHup, destroys_safely)
 {
     int socks[2];
     int rv = socketpair(AF_UNIX, SOCK_STREAM, 0, socks);
-    if (rv) throw SysError("socketpair");
+    if (rv) {
+        throw SysError("socketpair");
+    }
     auto sock1 = AutoCloseFD{socks[0]};
     auto sock2 = AutoCloseFD{socks[1]};
 
     {
-        MonitorFdHup monitor(sock1.get(), []() {
-            abort();
-        });
+        MonitorFdHup monitor(sock1.get(), []() { abort(); });
     }
 }